ibapi nextValidId not always invoked

Question

i built a small ibapi python app to run some strategy, while taking care to stay connected to tws.

if tws is inactive, the python app will start and wait, and will connect to tws when tws will start, but here's my problem:

nextValidId won't be invoked

here is my code. i repeat, nextValidId will be invoked only if this app connects upon it's start to a tws session that was active before my app connects.

as you can see, in a tws session that wasn't active before this script starts, nextValidId won't get invoked at all, not even manually by reqIds.

from ibapi.client import EClient
from ibapi.wrapper import EWrapper
from ibapi.common import MarketDataTypeEnum
from ibapi.errors import *

connection_errors = (
    CONNECT_FAIL, UPDATE_TWS, NOT_CONNECTED, UNKNOWN_ID, UNSUPPORTED_VERSION, 
    #BAD_LENGTH, BAD_MESSAGE, 
    SOCKET_EXCEPTION, FAIL_CREATE_SOCK, SSL_FAIL, 
)
connection_error_codes = [error.code() for error in connection_errors]


import threading
import time
import logging


class IBapi(EWrapper, EClient):

#   connection_lost = True
    
    clientid = None
    hostname = None
    portno = None
    onConnected = None
    
    MarketDataType = None
    
    def __init__(
            self, clientid, hostname='127.0.0.1', portno=7497, 
            MarketDataType=None, onConnected=None, ka_interval=3):
        
        self.clientid = clientid
        self.hostname = hostname
        self.portno = portno
        self.ka_interval = ka_interval
        
        EClient.__init__(self, self)
        
        self.onConnected = onConnected
        self.MarketDataType = MarketDataType or MarketDataTypeEnum.DELAYED
        
        self.host_connect()
        
        # Initialise the threads for various components
        thread = threading.Thread(target=self.run)
        thread.start()
        setattr(self, "_thread", thread)
        
        thread = threading.Thread(target=self.keepAlive)
        thread.start()
        setattr(self, "_thread_ka", thread)

    def host_connect(self):
        """Connects to TWS with the appropriate connection parameters"""
        if not self.hostname or not self.portno:
            logging.error(f'hostname {self.hostname} or portno {self.portno} not [yet] defined')
            return
        super().connect(self.hostname, self.portno, self.clientid)

    def error(self, reqId, errorCode, errorString):
        """disconnect to handle communications errors"""
        # clean the connection status
        if errorCode in connection_error_codes and \
                self.connState not in (EClient.CONNECTING,):
            logging.error(
                f'reset on connection_error {errorCode} "{errorString}"???')
#           self.connection_lost = True
            self.disconnect()
        return super().error(reqId, errorCode, errorString)

    def keepAlive(self):
        data_lock = threading.Lock()
        while self.ka_interval:
            time.sleep(self.ka_interval)
#           isConnected = self.isConnected()
            connState = None
            with data_lock:
                connState = self.connState
#               connection_lost = self.connection_lost
            isConnected = connState == EClient.CONNECTED
            logging.error(f'is connected: {isConnected}')
            if not isConnected:
                isConnecting = connState == EClient.CONNECTING
                if not isConnecting:
                    logging.error(f"let's connect")
                    self.host_connect()
                else:
                    logging.error(f'already connecting')
            else:
                logging.error(f'requesting CurrentTime for keepAlive')
#               if connection_lost:
#                   logging.error('reconnecting. should auto invoke nextValidId')
#                   self.reqIds(1)
#                   self.host_connected()
                self.reqCurrentTime()
                self.reqIds(1)

    def host_connected(self):
#       if self.connection_lost:
#           self.connection_lost = False
            self.reqMarketDataType(self.MarketDataType)
            self.reqPositions()

    def nextValidId(self, orderId):
        print('====================================================')
        logging.error(f'The next valid order id is: {orderId}')
        print('====================================================')
        super().nextValidId(orderId)
        self.nextorderId = orderId
        self.host_connected()

port_TWS_Live = 7496
port_IBGateway_Live = 4001
port_TWS_Simulated = 7497
port_IBGateway_Simulated = 4002

def main():
    logging.basicConfig(
        format='%(levelname)s:%(asctime)s:%(message)s', level=logging.WARN)
    logging.info('Started')
    
    logging.debug('This message should appear on the console')
    
    app = IBapi(
        1234, 
        #portno=port_TWS_Live, 
        portno=port_TWS_Simulated, 
    )
    
    logging.info('Finished')

if __name__ == '__main__':
    main()

i'm open to any interesting suggestions.

thanks in advance, alex

Answer 1

Interesting. I bet you are getting the ids, check the api logs. For something like

13:05:15:387 -> --- 9-1-1007-

means 9 (code for NextValidId) and 1007 is the ID for that Client# so far.

I get the same thing if using your reconnect logic due to the reqIds but it never reaches the nextValidId callback. So the reader thread (which is stopped on disconnect()) doesn't work.

I tried just moving the self.run thread to the host_connect() but it still doesn't work.

I put EClient__init__ in the host_connect() and now it works. I don't know why, too much code to go through.

def host_connect(self):
    """Connects to TWS with the appropriate connection parameters"""
    if not self.hostname or not self.portno:
        logging.error(f'hostname {self.hostname} or portno {self.portno} not [yet] defined')
        return
    
    # don't know exactly why this works, all the reader, connection, decoder, etc. need a callback
    EClient.__init__(self, self) 
    super().connect(self.hostname, self.portno, self.clientid)

    #moved from init
    thread = threading.Thread(target=self.run)
    thread.start()
    setattr(self, "_thread", thread)