How can I request data for two stocks at the same time before running an algorithm?

[Sorais]

I'm trying to code an algorithm that uses two stocks simultaneously. I'm encountering an issue where I can request the data, but can't use both of them simultaneously. I'm wondering if anyone knows how to get around it.

data1 = ib.reqHistoricalData(stock1 ...) data2 = ib.reqHistoricalData(stock2, ...)

This requests data for the two stocks, but then how do I use updateEvent (I assume I have to use this, since it's the only event that looks usable) to run the algorithm? Simply

data1.updateEvent += new_data_received

does not work, since that leaves out the other stock. Adding data2.updateEvent += new_data_received does not work either, since new_data_received returns an error that one of the data does not exist. I tried putting a ib.sleep command between, requesting multiple stocks with the same reqHistoricalData command, or implementing a check to see if both data have arrived before new_data_received continues to run, but none of them led anywhere.

[mattsta]

This is more of a design issue depending on what data management architecture you want to use.

First you have to understand what this does: data1 = ib.reqHistoricalData(stock1 ...) data2 = ib.reqHistoricalData(stock2, ...) because it doesn't really make sense to try and do it that way since the bars generated by reqHistoricalData() (or, more usefully reqRealTimeBars()) only update every 5 seconds while the live millisecond streaming price/quote updates are from reqMktData().

You probably want something more like:

# return last seen prices outside of market hours
self.ib.reqMarketDataType(2)

# for each symbol, first generate a qualified contract and optionally request extra data fields:
# https://interactivebrokers.github.io/tws-api/tick_types.html
self.ib.reqMktData(contract, tickFields)

# you may also want to save the value returned here in a dictionary indexed by name
# of your contract so you can stop/unsubscribe data later too.

You want to subscribe to all your symbols/tickers/instruments updates, then process the updated values in a pendingTickersEvent handler registered like self.ib.pendingTickersEvent += self.tickersUpdate

where tickersUpdate is some method or function attached as in the docs under Events:

pendingTickersEvent (tickers: Set[Ticker]): Emits the set of tickers that have been updated during the last update and for which there are new ticks, tickByTicks or domTicks.

Here's a simple one to get started:

    def tickersUpdate(self, tickers: set[Ticker]) -> None:
        """Note: this event handler will potentially run dozens of times per second"""
        # logger.info("Ticker update: {}", tickers)

        for ticker in tickers:
            c = ticker.contract
            name = (c.localSymbol or c.symbol).replace(" ", "")
            price = (ticker.bid + ticker.ask) / 2

Then your handler for pendingTickersEvent updates receives an object with the updated price details when the IBKR API sends price updates.

...and, as always, your update processing should be quick because all the other updates and data receiving will be blocked while the handlers run too.

You probably don't want to run any trade decisions or execution calls from inside the event handler because order calls can be slow (and event handlers break error reporting due to how they are run from the library, so if you have bugs or crashes in your event handlers you probably don't see them). The best design is to have one client processing quotes and generating "signal triggers" then another client consuming the triggers if this is a full "automated trading system."

If you are just checking live price updates for things like synthetic stops, then you could probably do price tracking and order processing all in the same client though if the logic is fast and simple enough.

[edlang]

You can inspect the bars.contract and merge using that.

Here's a trivial example I was briefly using a while ago:

[...]

	self.ib.barUpdateEvent += self.handle_barUpdateEvent_WatchedStock
        self.logger.info(
            f"added barUpdateEventHandler to realtime bars for {contract.symbol}")

[...]

    def handle_barUpdateEvent_WatchedStock(
            self, bars: BarDataList, hasNewBar: bool):

        if not self.use_merged:
            # Simpler option, just use the bars for the contract.
            if bars.contract == self.contract:
                df = util.df(bars)

                if self.use_macd:
                    self.process_bars_macd(bars, df, hasNewBar)
                else:
                    self.process_bars_v2(bars, df, hasNewBar)

        else:
            if bars.contract == self.idx_contract:
                self.frames[0] = util.df(bars)
            elif bars.contract == self.vol_contract:
                self.frames[1] = util.df(bars)
            elif bars.contract == self.contract:
                self.merged_df = self.frames[0].merge(
                    self.frames[1], on='date', how='inner', suffixes=[
                        "_" + self.idx_contract.symbol, "_" + self.vol_contract.symbol])

                # bars for the main contract.
                contract_df = self.frames[2] = util.df(bars)

                frame = {
                    'date': self.merged_df['date'],
                    'open': (self.merged_df['open' + self.idx_suffix] /
                             self.merged_df['open' + self.vol_suffix]).fillna(0),
                    'high': (self.merged_df['high' + self.idx_suffix] /
                             self.merged_df['high' + self.vol_suffix]).fillna(0),
                    'low': (self.merged_df['low' + self.idx_suffix] /
                            self.merged_df['low' + self.vol_suffix]).fillna(0),
                    'close': (self.merged_df['close' + self.idx_suffix] /
                              self.merged_df['close' + self.vol_suffix]).fillna(0),
                    'volume': (self.merged_df['volume' + self.idx_suffix] /
                               self.merged_df['volume' + self.vol_suffix]).fillna(0).astype(int)
                }

                df = pd.DataFrame(frame)

                if self.use_macd:
                    self.process_bars_macd(bars, df, hasNewBar, contract_df)
                else:
                    self.process_bars_v2(bars, df, hasNewBar, contract_df)

            else:
                return

[jj625]

Hi @Sorais , does the following work for you?

def onBar1Update(bars, bar):
    print('AAPL', bars[-1], bar)

def onBar2Update(bars, bar):
    print('MSFT', bars[-1], bar)

contract1 = Stock('AAPL', 'SMART', 'USD')
contract2 = Stock('MSFT', 'SMART', 'USD')

bars1 = ib.reqHistoricalData(contract1, endDateTime='', durationStr='1 D'
    , barSizeSetting='5 secs', whatToShow='TRADES', useRTH=False, keepUpToDate=True)
bars2 = ib.reqHistoricalData(contract2, endDateTime='', durationStr='1 D'
    , barSizeSetting='5 secs', whatToShow='TRADES', useRTH=False, keepUpToDate=True)
bars1.updateEvent += onBar1Update
bars2.updateEvent += onBar2Update

I can see the callback functions fire every 5 secs. I can also get the bars through bar1 and bar2 variables.