Sensible Interactive Broker API

When 2021 first started, I spent some time looking into Interactive Broker API yet again as part of my 2020 new year resolution to learn trading. Part of the reason this 2020 resolution didn’t quite work out is how badly designed these IB APIs are. Yet, there are not that many brokers who offer programmatic APIs as well as full featured trading services, so most people who are interested in the automated trading are stuck with Interactive Broker.

So I dedicated a few days figuring out exactly how the native IB API works, and how to build a clean and robust layers of APIs on the top of the native ones. Is this really meaningful work? Let’s take a look at what the problem is.

NOTE: the code snippets below will use Python, and is available on Github here: https://github.com/dayfine/ibapi-rxpy

Say I want an API to get all the positions in the broker account. Normally I would expect this API to look like:

def get_positions(account_id: str) -> List[Position]:

It takes the ID of the account, which might be the account number, and returns all the positions in the account. Note the basic assumptions of this API:

1. You only need to make one call
2. The call is assumed to be synchronous, which would allow users to write straightforward client code with this API (including mocking the call in unit tests)

Well, unfortunately, the native interactive broker API doesn’t work like this. This is primarily because IB uses a socket style design for their API. To get position data using native IB API, you will actually need to:

1. Create a subscription to get position data
2. Collect incoming messages for position data
3. Wait for a special message that indicates the end of the data stream
4. Cancel the subscription created in step 1

The code might look like this (and I know it’s terrible; just bear with me):

import ibapi
from overrides import overrides
positions = []
positions_ready = False
# Create a Wrapper class that implements IB's EWrapper interface for 
# handling incoming messages.
class MyWrapper(ibapi.wrapper.EWrapper):
   @overrides
   def position(self, *args):
      # The args here are really long          
      positions.append(handle_position(*args))
   @overrides
   def positionEnd(self, *args):
      #  Somehow indicate positions is finalized        
      global positions_ready
      positions_ready = False
wrapper = MyWrapper()
eclient = ibapi.client.EClient(wrapper)
eclient.reqPositions()
while not positions_ready:
    continue
eclient.cancelPositions()
print(positions)

As you can see, there are many obnoxious things, in order to do this using IB’s native API:

1. We have to work with two classes and four different methods just to get the positions. One of the classes is an interface with long lists of methods each with a long list of arguments (omitted here) that we need to implement.
2. The message passing requires maintaining some state in order to building up the result data among different calls. Here I used global variables to keep the example simple. What you will find in many examples elsewhere is that people create a single class that inherits both ibapi.EWrapper and ibapi.EClient, and maintaining the state as members in that class, which makes the class very hard to follow.
3. There is no native tool to handle the async calls: reqPositions doesn’t even take a callback. You will literally write your own code to track the state of the data stream and cancel the subscription after receiving all data. This might not seem to be such a big problem in this example, since there is only subscription, but in another case, say requesting price data for many different stocks, leaving subscriptions uncanceled would definitely cause issues

These issues make it hard to write clean and robust user codes and are prone to lead to bad design. That being said, the problem being dealt with here is actually neither new nor unique, and there is a variety of mature and extensive solution for dealing with async data like this. Here I will present a solution I implemented:

The architecture of the native IB API is similar to that of a message broker. Although to the users, dealing with this level of details is obnoxious (I just want to fucking get my positions!). One familiar tool that I pick to help this situation is the Reactive Extension, and specifically for Python: rxpy. I won’t spend too much time on how Rx works here, which you can find enough tutorials online.

Using Rx, the code to get positions would become:

import dataclasses
import enum
import ibapi
from rx import operators as _
from rx.subject import ReplaySubject

class MsgType(enum.Enum):
    POSITION = enum.auto()
    POSITION_END = enum.auto()

@dataclasses.dataclass
class IbApiMessage(object):
    type: MsgType
    payload: Any
class MyWrapper(ibapi.wrapper.EWrapper):
    def __init__(self, subject: Subject):
        self._subject = subject
    @overrides
    def position(self, *args):
        self._subject.on_next(
            IbApiMessage(type=MsgType.POSITION, payload=*args))
    @overrides
    def positionEnd(self, *args):
        self._subject.on_next(
            IbApiMessage(type=MsgType.POSITION_END, payload=*args))
class Client:
    def __init__(self):
        self._subject = ReplaySubject()
        self._wrapper = MyWrapper(self._subject)
        self._eclient = ibapi.client.EClient(self._wrapper)
    def get_positions(self):
        self._eclient.reqPositions()
        with ExitStack() as stack:
            stack.callback(self._eclient.cancelPositions)
            observable = self._subject.pipe(
                _.filter(lambda m: m.type == MsgType.POSITION 
                                or m.type == MsgType.POSITION_END),
                _.take_while(lambda m:
                                   m.type != MsgType.POSITION_END),
                _.map(handle_position),
                _.reduce(lambda positions, pos: [*positions, pos], []))
            return observable.run()

Again, the last part should look familiar if you have experience with Rx and / or declarative programming.

Note how:

1. state and global variables are eliminated and now handled as data streams and Rx operators on the data streams
2. The Wrapper class no longer does any nontrivial work and now simply dispatches data to the global subject (message queue)
3. Now we have a single get_positions API that encapsulate the four low level methods native IB API provides
4. Details about how to unpack the message, i.e. in the handle_position function is omitted here as it does not impact the program structure much

With the lower level calls put into where they belong, we finally have a set of APIs that are clean and easy to test. So far I implemented a small set of features on Github for my personal exploration. If you want to contribute to the project, feel free to reach out. Also, feel free to share any ideas and feedbacks you might have, e.g. have you also used IB API before and how was your experience dealing it?

Happy New Year!