Interactive Brokers has always been a popular brokerage with systematic traders. Initially this could partially be attributed to the fact that IB provided an Application Programming Interface (API) that allowed quants to obtain market data and place trades directly in code. Many competing brokerages took some time to develop their own APIs, allowing IB to gain a reasonable early-mover advantage in the retail quant trading space.
While the initial IB API had a reputation for having a complex interface, the situation has changed in recent years with the release of the IB Native Python API library.
In this new series of articles we are going to establish how to interact with the Interactive Brokers API via the 'Native Python' interface, using the ibapi library.
Eventually we will learn how to request market data, define contracts and process orders. This article however will serve to introduce the interface itself and test basic connectivity.
The article assumes you have a working Python virtual environment (such as Anaconda Individual Edition) and have successfully installed the IB Python API into this environment. The installation instructions are operating-system specific. The most up to date instructions can be found on the interactive brokers API site itself.
Overview
The IB API works via an asynchronous 'request-response' model. Messages are sent to the IB server (through TraderWorkstation or IB Gateway) via a client class, while responses (known as 'errors') are handled separately by a wrapper class.
Much of the internal connection handling is abstracted away from the end user with the Python API, allowing minimal necessary 'boilerplate' code to get connected. However, the original legacy mechanism for connecting to IB still partially informs the design of the API. Hence it can be confusing for those not used to object oriented design principles.
While it may not seem clear initially how all of the components fit together, after coding up the following classes, it should begin to make sense as to how the API is constructed.
In order to connect to the IB API there are four major components that will need to be coded.
The first is a derived subclass of the IB API EWrapper class. EWrapper is used to handle all responses ('errors') from the IB server.
The second is a derived subclass of the IB API EClient class. EClient is used to send all messages to the IB server.
The third is a multiply-inherited class of both derived subclasses from EWrapper and EClient, used as the basis of the IB API application, which will tie all communication together.
Finally there will be an if __name__ == "__main__": entrypoint, designed to allow the script to be executed from the command line.
Initial Imports
The first step is to import the necessary library components used within the script.
We will require the IB API EWrapper and EClient classes, which will be described below. We will also need the Thread and Queue standard library components from the threading and queue libraries respectively. Finally we will import datetime to convert a Unix timestamp to a more readable format:
# ib_api_connection.py
import datetime
import queue
import threading
from ibapi.client import EClient
from ibapi.wrapper import EWrapper
We are now in a position to define the IBAPIWrapper class.
IBAPIWrapper Class
The EWrapper class provides an interface for handling responses (described as 'errors') from the IB server. The interface specifies methods that can be implemented in a derived subclass. By inheriting from this class we can override these methods to suit our own particular data processing methodology.
Let's begin by creating the IBAPIWrapper derived subclass of EWrapper and override some of the methods. The full snippet for this component will be presented. Each method will be described in turn:
# ib_api_connection.py
class IBAPIWrapper(EWrapper):
"""
A derived subclass of the IB API EWrapper interface
that allows more straightforward response processing
from the IB Gateway or an instance of TWS.
"""
def init_error(self):
"""
Place all of the error messages from IB into a
Python queue, which can be accessed elsewhere.
"""
error_queue = queue.Queue()
self._errors = error_queue
def is_error(self):
"""
Check the error queue for the presence
of errors.
Returns
-------
`boolean`
Whether the error queue is not empty
"""
return not self._errors.empty()
def get_error(self, timeout=5):
"""
Attempts to retrieve an error from the error queue,
otherwise returns None.
Parameters
----------
timeout : `float`
Time-out after this many seconds.
Returns
-------
`str` or None
A potential error message from the error queue.
"""
if self.is_error():
try:
return self._errors.get(timeout=timeout)
except queue.Empty:
return None
return None
def error(self, id, errorCode, errorString):
"""
Format the error message with appropriate codes and
place the error string onto the error queue.
"""
error_message = (
"IB Error ID (%d), Error Code (%d) with "
"response '%s'" % (id, errorCode, errorString)
)
self._errors.put(error_message)
def init_time(self):
"""
Instantiates a new queue to store the server
time, assigning it to a 'private' instance
variable and also returning it.
Returns
-------
`Queue`
The time queue instance.
"""
time_queue = queue.Queue()
self._time_queue = time_queue
return time_queue
def currentTime(self, server_time):
"""
Takes the time received by the server and
appends it to the class instance time queue.
Parameters
----------
server_time : `str`
The server time message.
"""
self._time_queue.put(server_time)
init_error has the task of creating a Python Queue and attaching it a 'private' instance variable called _errors. This queue will be utilised throughout the class to store IB error messages for later processing.
is_error is a simple method that determines whether the _errors queue is empty or not.
get_error attempts to retrieve an error message from the queue, with a prescribed timeout in seconds. If the queue is empty or the timeout is exceeded then nothing is returned from the method.
error formats the provided error codes, along with the error message, into an appropriate string format that is then placed onto the _errors queue. This method is utilised to provide better debug information on the console when executing code against the API.
These four methods complete the handling of responses ('errors') from Interactive Brokers. It is important to note that there is a lot of internal machinery within the ibapi library that carries out this processing. Much of this work is not directly visible from our derived subclass.
The remaining two methods, init_time and currentTime are utilised to carry out a connection 'sanity check'. A simple way to determine if we have a successful connection is to retrieve the local time on the IB servers.
The two methods simply create a new queue to store the server time messages, as well as place new time messages onto this queue when requested.
This concludes our simple subclass of EWrapper. We now have the ability to process certain responses from the IB server. The next task is to actually send messages to the IB server. For this we need to override the EClient class.
IBAPIClient Class
The IBAPIClient derived subclass of EClient is used to send messages to the IB servers.
It is particularly important to note that the constructor __init__ method of our derived subclass takes a wrapper parameter, which is then passed into the parent EClient constructor. This means that within the IBAPIClient class there is no overriding of the native IB API methods. Instead our wrapper instance (instantiated from IBAPIWrapper) is provided to handle the responses.
# ib_api_connection.py
class IBAPIClient(EClient):
"""
Used to send messages to the IB servers via the API. In this
simple derived subclass of EClient we provide a method called
obtain_server_time to carry out a 'sanity check' for connection
testing.
Parameters
----------
wrapper : `EWrapper` derived subclass
Used to handle the responses sent from IB servers
"""
MAX_WAIT_TIME_SECONDS = 10
def __init__(self, wrapper):
EClient.__init__(self, wrapper)
def obtain_server_time(self):
"""
Requests the current server time from IB then
returns it if available.
Returns
-------
`int`
The server unix timestamp.
"""
# Instantiate a queue to store the server time
time_queue = self.wrapper.init_time()
# Ask IB for the server time using the EClient method
self.reqCurrentTime()
# Try to obtain the latest server time if it exists
# in the queue, otherwise issue a warning
try:
server_time = time_queue.get(
timeout=IBAPIClient.MAX_WAIT_TIME_SECONDS
)
except queue.Empty:
print(
"Time queue was empty or exceeded maximum timeout of "
"%d seconds" % IBAPIClient.MAX_WAIT_TIME_SECONDS
)
server_time = None
# Output all additional errors, if they exist
while self.wrapper.is_error():
print(self.get_error())
return server_time
Within obtain_server_time we first create a queue to hold the timestamp messages from the server. We then call the native EClient method reqCurrentTime to obtain the time from the server.
Subsequently we wrap a call to obtain a value from the time queue in a try...except block. We provide a timeout of ten seconds. If the timeout is exceeded or the queue is empty we set the server time to None.
We run a while loop that checks for any additional responses in the error queue defined within the EWrapper derived subclass. These are printed to the console if they exist.
Finally we return the server time.
The next stage is to create a mechanism for instantiating both IBAPIWrapper and IBAPIClient, as well as actually connecting to the IB server.
IBAPIApp
The final class to be derived within this tutorial is the IBAPIApp class.
This class utilises multiple inheritance to inherit from both the IBAPIWrapper and IBAPIClient classes. At the point of initialisation both of these classes are also initialised. However, note that the IBAPIClient class takes wrapper=self as an initialisation keyword argument, since IBAPIApp is also derived from IBAPIWrapper.
Subsequent to the initialisation of both of the parent classes the connect native method is called with the appropriate connection parameters.
The next section of code initialises the various threads needed for the application. There is a single thread for the client instance and another used to add response messages to the various queues.
Finally the init_error method is called to start listening for IB responses.
# ib_api_connection.py
class IBAPIApp(IBAPIWrapper, IBAPIClient):
"""
The IB API application class creates the instances
of IBAPIWrapper and IBAPIClient, through a multiple
inheritance mechanism.
When the class is initialised it connects to the IB
server. At this stage multiple threads of execution
are generated for the client and wrapper.
Parameters
----------
ipaddress : `str`
The IP address of the TWS client/IB Gateway
portid : `int`
The port to connect to TWS/IB Gateway with
clientid : `int`
An (arbitrary) client ID, that must be a positive integer
"""
def __init__(self, ipaddress, portid, clientid):
IBAPIWrapper.__init__(self)
IBAPIClient.__init__(self, wrapper=self)
# Connects to the IB server with the
# appropriate connection parameters
self.connect(ipaddress, portid, clientid)
# Initialise the threads for various components
thread = threading.Thread(target=self.run)
thread.start()
setattr(self, "_thread", thread)
# Listen for the IB responses
self.init_error()
With the prior three classes now defined we are in a position to create the script entrypoint.
Executing the Code
We first set the connection parameters including the host IP address, the port to connect to TWS/IB Gateway on and an (arbitrary) positive-integer client ID.
Then we instantiate an application instance with the appropriate connection parameters.
We use the application to obtain the server time from IB, then appropriately format the Unix timestamp into a more readable date format using the utcfromtimestamp method of the datetime library.
Finally we disconnect from the IB server and end the program.
# ib_api_connection.py
if __name__ == '__main__':
# Application parameters
host = '127.0.0.1' # Localhost, but change if TWS is running elsewhere
port = 7497 # Change to the appropriate IB TWS account port number
client_id = 1234
print("Launching IB API application...")
# Instantiate the IB API application
app = IBAPIApp(host, port, client_id)
print("Successfully launched IB API application...")
# Obtain the server time via the IB API app
server_time = app.obtain_server_time()
server_time_readable = datetime.datetime.utcfromtimestamp(
server_time
).strftime('%Y-%m-%d %H:%M:%S')
print("Current IB server time: %s" % server_time_readable)
# Disconnect from the IB server
app.disconnect()
print("Disconnected from the IB API application. Finished.")
At this stage we are ready to run ib_api_connection.py. Simply navigate to the directory where you have stored the file, ensure that the virtual environment with ibapi is active and that TWS (or IB Gateway) is loaded and configured correctly, then type the following:
python ib_api_connection.py
You should see output similar to the following:
Launching IB API application...
Successfully launched IB API application...
IB Error ID (-1), Error Code (2104) with response 'Market data farm connection is OK:usfarm'
IB Error ID (-1), Error Code (2106) with response 'HMDS data farm connection is OK:ushmds'
IB Error ID (-1), Error Code (2158) with response 'Sec-def data farm connection is OK:secdefnj'
Current IB server time: 2020-07-29 13:27:18
Disconnected from the IB API application. Finished.
unhandled exception in EReader thread
Traceback (most recent call last):
File "/home/mhallsmoore/venv/qstrader/lib/python3.6/site-packages/ibapi/reader.py", line 34, in run
data = self.conn.recvMsg()
File "/home/mhallsmoore/venv/qstrader/lib/python3.6/site-packages/ibapi/connection.py", line 99, in recvMsg
buf = self._recvAllMsg()
File "/home/mhallsmoore/venv/qstrader/lib/python3.6/site-packages/ibapi/connection.py", line 119, in _recvAllMsg
buf = self.socket.recv(4096)
OSError: [Errno 9] Bad file descriptor
The first set of outputs are IB 'errors' with codes 2104, 2106 and 2158. These are actually responses stating that the connections to various servers are functioning correctly. That is, they are not 'errors'!
The server time is also correctly converted from a Unix timestamp to a more readable format and output. At this stage the application disconnects.
Note however that an OSError exception is raised in the EReader thread. This is an internal issue with the IB API itself and does not as yet have a fix that has made it through to a release. It can be ignored for the purposes of this tutorial.
This now completes the tutorial for connecting to the IB Python Native API. The full code for ib_api_connection.py is as follows:
Full Code
# ib_api_connection.py
import datetime
import queue
import threading
from ibapi.client import EClient
from ibapi.wrapper import EWrapper
class IBAPIWrapper(EWrapper):
"""
A derived subclass of the IB API EWrapper interface
that allows more straightforward response processing
from the IB Gateway or an instance of TWS.
"""
def init_error(self):
"""
Place all of the error messages from IB into a
Python queue, which can be accessed elsewhere.
"""
error_queue = queue.Queue()
self._errors = error_queue
def is_error(self):
"""
Check the error queue for the presence
of errors.
Returns
-------
`boolean`
Whether the error queue is not empty
"""
return not self._errors.empty()
def get_error(self, timeout=5):
"""
Attempts to retrieve an error from the error queue,
otherwise returns None.
Parameters
----------
timeout : `float`
Time-out after this many seconds.
Returns
-------
`str` or None
A potential error message from the error queue.
"""
if self.is_error():
try:
return self._errors.get(timeout=timeout)
except queue.Empty:
return None
return None
def error(self, id, errorCode, errorString):
"""
Format the error message with appropriate codes and
place the error string onto the error queue.
"""
error_message = (
"IB Error ID (%d), Error Code (%d) with "
"response '%s'" % (id, errorCode, errorString)
)
self._errors.put(error_message)
def init_time(self):
"""
Instantiates a new queue to store the server
time, assigning it to a 'private' instance
variable and also returning it.
Returns
-------
`Queue`
The time queue instance.
"""
time_queue = queue.Queue()
self._time_queue = time_queue
return time_queue
def currentTime(self, server_time):
"""
Takes the time received by the server and
appends it to the class instance time queue.
Parameters
----------
server_time : `str`
The server time message.
"""
self._time_queue.put(server_time)
class IBAPIClient(EClient):
"""
Used to send messages to the IB servers via the API. In this
simple derived subclass of EClient we provide a method called
obtain_server_time to carry out a 'sanity check' for connection
testing.
Parameters
----------
wrapper : `EWrapper` derived subclass
Used to handle the responses sent from IB servers
"""
MAX_WAIT_TIME_SECONDS = 10
def __init__(self, wrapper):
EClient.__init__(self, wrapper)
def obtain_server_time(self):
"""
Requests the current server time from IB then
returns it if available.
Returns
-------
`int`
The server unix timestamp.
"""
# Instantiate a queue to store the server time
time_queue = self.wrapper.init_time()
# Ask IB for the server time using the EClient method
self.reqCurrentTime()
# Try to obtain the latest server time if it exists
# in the queue, otherwise issue a warning
try:
server_time = time_queue.get(
timeout=IBAPIClient.MAX_WAIT_TIME_SECONDS
)
except queue.Empty:
print(
"Time queue was empty or exceeded maximum timeout of "
"%d seconds" % IBAPIClient.MAX_WAIT_TIME_SECONDS
)
server_time = None
# Output all additional errors, if they exist
while self.wrapper.is_error():
print(self.get_error())
return server_time
class IBAPIApp(IBAPIWrapper, IBAPIClient):
"""
The IB API application class creates the instances
of IBAPIWrapper and IBAPIClient, through a multiple
inheritance mechanism.
When the class is initialised it connects to the IB
server. At this stage multiple threads of execution
are generated for the client and wrapper.
Parameters
----------
ipaddress : `str`
The IP address of the TWS client/IB Gateway
portid : `int`
The port to connect to TWS/IB Gateway with
clientid : `int`
An (arbitrary) client ID, that must be a positive integer
"""
def __init__(self, ipaddress, portid, clientid):
IBAPIWrapper.__init__(self)
IBAPIClient.__init__(self, wrapper=self)
# Connects to the IB server with the
# appropriate connection parameters
self.connect(ipaddress, portid, clientid)
# Initialise the threads for various components
thread = threading.Thread(target=self.run)
thread.start()
setattr(self, "_thread", thread)
# Listen for the IB responses
self.init_error()
if __name__ == '__main__':
# Application parameters
host = '127.0.0.1' # Localhost, but change if TWS is running elsewhere
port = 7497 # Change to the appropriate IB TWS account port number
client_id = 1234
print("Launching IB API application...")
# Instantiate the IB API application
app = IBAPIApp(host, port, client_id)
print("Successfully launched IB API application...")
# Obtain the server time via the IB API app
server_time = app.obtain_server_time()
server_time_readable = datetime.datetime.utcfromtimestamp(
server_time
).strftime('%Y-%m-%d %H:%M:%S')
print("Current IB server time: %s" % server_time_readable)
# Disconnect from the IB server
app.disconnect()
print("Disconnected from the IB API application. Finished.")
Next Steps
We have successfully connected to the IB server and have checked the connection via the call to obtain the current server time. In the next article we will determine how to retrieve market data for a basic cash equity from the IB API.
References and Ackowledgements
The code presented above is a modified, extended and more heavily commented version of the code found within the following three references as well as the IB API documentation itself.
QuantStart would like to thank the authors of the following articles—Jignesh Davda, Corbin Balzan and Robert Carver—for providing insightful tutorials upon which this article is based.
