Inaccurate real time data from stock broker

Question

I use Interactive Brokers and have a subscription to their NASDAQ data. I use it to get near real-time data for the stock PLUG.

I used to get streaming 5-sec bars and aggregate them into 1-min bars, but noticed that the data was different than if pulled again a few minutes later for the same time period. To correct this, I replaced the streaming data with historical data every 60 sec for the past minute. This has helped reduce the discrepancy, but not eliminated it. How can I fix this problem?

Here is the data for the last 2 days in case it helps. Also, I use the ib_insync wrapper instead of interacting with the IB API directly.

Thank you!

Old Code:

def onBarUpdate(bars, hasNewBar):
    df_new_bar = pd.DataFrame([[bars[-1].time, bars[-1].open_, bars[-1].high, bars[-1].low, bars[-1].close, bars[-1].count]], columns=['date', 'open', 'high', 'low', 'close', 'volume'])

    # ensure that none of the 5-sec values just received are NaN
    if not df_new_bar.isnull().values.any():
        df_new_bar.set_index('date', inplace=True)
        df_5_sec = pd.concat([df_5_sec, df_new_bar[df_new_bar.index > df_5_sec.index[-1]]], axis='rows')

    # Isolate 5-sec data that is not in df_1_min yet
    df_tmp = df_5_sec[df_5_sec.index > df_1_min.index[-1]]

    if len(df_tmp) >= 12:
        df_tmp2 = df_tmp.resample('1min').agg({"open":"first", "high":"max", "low":"min", "close":"last"})
        df_tmp2['volume'] = df_tmp['volume'].resample('1min').sum()
        df_1_min = pd.concat([df_1_min, df_tmp2[df_tmp2.index > df_1_min.index[-1]]], axis='rows')

def getLiveData():
    ib.reqMarketDataType(1)
    bars = ib.reqRealTimeBars(contract, 5, 'TRADES', False)
    ib.barUpdateEvent += onBarUpdate
    ib.run()

New code:

def getLiveData():
    from datetime import datetime
    ib.reqMarketDataType(1)
    processed = 0

    while 1:
        try:
            seconds = datetime.now().second

            # wait 20 seconds for the data to "settle"
            if seconds <= 20 and processed != 0:
                processed = 0
            elif seconds > 20 and processed != 1:
                bars_hist = ib.reqHistoricalData(
                    contract,
                    endDateTime='',
                    durationStr='100 S',
                    barSizeSetting='5 secs',
                    whatToShow='TRADES',
                    useRTH=False,
                    keepUpToDate=False,
                    formatDate=2)
                df_5_sec_tmp = pd.DataFrame(bars_hist)
                df_5_sec_tmp.set_index('date', inplace=True)
                df_5_sec_tmp.drop(['average', 'barCount'], axis='columns', inplace=True)
                df_tmp = df_5_sec_tmp[df_5_sec_tmp.index > df_1_min.index[-1]]

                df_tmp2 = df_tmp.resample('1min').agg({"open":"first", "high":"max", "low":"min", "close":"last"})
                df_tmp2['volume'] = df_tmp['volume'].resample('1min').sum()
                df_1_min = pd.concat([df_1_min, df_tmp2[df_tmp2.index > df_1_min.index[-1]]], axis='rows')

                ## further algo processing ##

                processed = 1
        except (KeyboardInterrupt, SystemExit) as e:
            logging.exception("Exception occurred {}".format(e))
        finally:
            ib.sleep(4)