Introduction

If you've started working with the Interactive Brokers API (via the ibapi python package) but you're confused about how to interact with TWS, you're in the right place!

In this post I will present one methodology of reliably asking for and receiving data from the TWS GUI using the ibapi package in your programs.

The methodology consists of the two main sections in this post:

1. Tying together the EClient and EWrapper classes
2. Implementing a strategy to create, read from and destroy queue objects

Before we dive in, though, I'm going to assume that you already have a working connection to TWS from your python program.

If you do not, there are already plenty of great blog posts and examples walking you through how to get setup and connected, including the official docs and this post from @wrighter, so go ahead and get connected & then jump back here for the low-down.

But first, the big picture

Let's first review the high level flow:

The TWS GUI is our source of truth. It is what our python program talks to- which is why the TWS GUI must be running locally for our program to work.

Hence, the one and only API we are using is the API for this local GUI program. Any communication external to our machine is handled by the TWS GUI itself- which is not our concern (as long as you have internet).

Also, the API is the collection of class definitions given to us in the ibapi package- it's not a URL, as you may have assumed (this confused me for a while).

And here's a screenshot of TWS just so we're all on the same page about what's what:

Now that we've got that sorted let's get right into it.

Tying together the EClient and EWrapper classes

As you may have already learned from the official documentation, the ibapi package instructs us to work with two classes, EWrapper and EClient.

The EClient class gives us one-way communication to send messages to TWS. The messages EClient sends are requests for data.

The EWrapper class gives us one-way communication to receive messages from TWS. The messages EWrapper receives are data.

Herein lies the crux of the problem. We can ask for data from our EClient methods, but we're not going to receive a response in those methods. We only receive messages from EWrapper methods.

So how do we coordinate asking for and receiving data across these two different classes?

If you're from the web development world like me, and you're used to requesting and receiving data in the same method call using something like: fetch('https://google.com').then(res => res.json())... this presents a serious paradigm shift (at least until we work some magic 😁).

Part of the answer is to tie the EClient and EWrapper classes together in one instantiated object. This process is given to us by the official docs in this (slightly modified) code snippet:

from ibapi.wrapper import EWrapper
from ibapi.client import EClient

class TestWrapper(EWrapper):
    def __init__(self):
        pass

class TestClient(EClient):
    def __init__(self, wrapper):
        EClient.__init__(self, wrapper)

class TestApp(TestWrapper, TestClient):
    def __init__(self):
        TestWrapper.__init__(self)
        TestClient.__init__(self, wrapper=self)

First you'll see that TestWrapper is just a direct copy of EWrapper (we'll do something useful and different later).

Next, we see that our new client class TestClient is just the good old EClient that we know and love, except that upon instantiation, it must be provided with access to a wrapper object.

And so when we instantiate TestApp via app = TestApp():

• We first instantiate TestWrapper to the app variable
• We then pass that app variable to TestClient as wrapper, and then re-instantiate the app variable as a TestClient instance instead

If you look carefully at the EClient.__init__() definition you'll see that the wrapper argument is attached to the new object as self.wrapper:

class EClient(object):
    def __init__(self, wrapper):
        self.wrapper = wrapper
        ...

So the end result is that we now have an app object:

• That has access to TestClient properties and methods via app.some_client_method() calls
• That has access to TestWrapper properties and methods via app.wrapper.some_wrapper_method() calls

This is basically a clever way to have two object instances in one. app is a full-blown TestClient instance with a full-blown TestWrapper instance attached to the app.wrapper property.

Implementing a strategy to create, read from and destroy queue objects

OK great, now we've tied together our EClient and EWrapper classes via this TestApp class. But... so what?

How do I make a request to TWS for my account summary? And after I do, how do I collect that data? Well now that we've connected our wrapper and client via our app object, we can design a flow to answer both of those questions.

Let's be very explicit and continue with the account summary example.

First, we can request our account summary from TWS via the EClient.reqAccountSummary method. How did I know that? Well, the purpose and description of this and all API-provided methods can be found in the ibapi source itself, or in (once again) the official docs.

Looking at the docstring for reqAccountSummary in EClient, we see:

Call this method to request and keep up to date the data that appears on the TWS Account Window Summary tab. The data is returned by accountSummary().

Now here we must discuss a very important point. accountSummary is an EWrapper method. In fact, it's not just a method but an event! That means that accountSummary will be automatically called when TWS is ready to deliver our requested account summary. This is how all of these wrapper methods/events work (openOrder, tickPrice, etc..)

To be 100% clear:

• We request our account summary from TWS by running reqAccountSummary() (an EClient method)
• TWS will take however much time it needs to gather that data, and when it's ready to send it back to us...
• The accountSummary() method (from EWrapper) will be automatically called, containing the data in its method arguments

Again, we do not manually call accountSummary, it's called automatically!

OK so what happens to the data delivered in accountSummary? Let's look at its definition in EWrapper:

def accountSummary(self, reqId:int, account:str, tag:str, value:str, currency:str):
    self.logAnswer(current_fn_name(), vars())

First let's note the function arguments. We're getting back the following data: reqId, account, tag, value and currency (the :int and :str are just type annotations).

Looking at the function body, there's really nothing going on. And that's intentional. Interactive Brokers expects us to overwrite this accountSummary method in TestWrapper to dictate what we want to do with the returned data. This is why the TestWrapper class definition was empty earlier (but now we're going to fill it in!)

Here is where we implement our data flow strategy. The gist of the methodology revolves around two principles:

1. We control client methods, but we do not control wrapper methods
   • Yes, we control what wrapper methods do (we override EWrapper methods with our own definitions in TestWrapper), but we do not control when they run. TWS will fire specific methods (such as EWrapper.accountSummary) automatically.
   • However, we do control when client methods will run.
2. The client part of our app object knows about the wrapper, but the wrapper does not know about the client
   • When we instantiated TestApp, we gave our TestClient class an instantiated TestWrapper object.
   • As we've already discussed, this means that the client methods and properties on app have access to the wrapper methods/props on app.wrapper, but the reverse is not true. The wrapper methods/props on app.wrapper have no knowledge of the client methods/props on the base app object.

What does this mean for our design? Our client methods should be our command center and contain our conditional checks & flow control. We want this because the client methods/props have supreme visibility across the object and because we control when the methods run. On the other hand our wrapper methods should be dead simple because they have limited visibility across the object and because we do not control when some of them run.

The gameplan to request data from TWS (using a TestClient method)

1. Initialize data storage on the wrapper object (think app.wrapper)
2. Request data from TWS (think EClient.reqAccountSummary)
   • (Auto-run wrapper method puts data into storage -> think TestWrapper.accountSummary)
3. Retrieve target data from storage on the wrapper
4. Delete data storage on the wrapper object
5. Check for errors that may have occurred during this process
6. Return the retrieved data

For all the visual folks:

In our case we want our account summary. So let's tackle these step-by-step and then bring it all together at the end in one TestClient method definition.

Step 1: Initialize data storage on the wrapper object

• First, let's define a method on TestWrapper that will build us an empty queue:

import queue

class TestWrapper(EWrapper):

    def init_accountSummary_queue(self):                                                                              
        self.accountSummary_queue = queue.Queue()

• What's a queue? It's just a first-in-first-out (FIFO) list. Why use this instead of a List or something else? Stay tuned!
• OK great. Now we have a place on the wrapper to put our account summary data. (As a sanity check, if you instantiate app = TestApp() right now, you'd have access to this method from app.wrapper.init_accountSummary_queue)

Step 2: Request data from TWS

• This is achieved by running the correct EClient method. All we have to do here is call reqAccountSummary from EClient (with the appropriate arguments of course)

Step 2b: Auto-run wrapper method puts data into storage

• Here's where we define our custom TestWrapper method overriding the default EWrapper.accountSummary

class TestWrapper(EWrapper):

    def accountSummary(
        self, reqId: int, account: str, tag: str, value: str, currency: str
    ):
        """
        Triggered by EClient.reqAccountSummary()
        """
        if hasattr(self, "accountSummary_queue"):
            self.accountSummary_queue.put(
                {
                    "reqId": reqId,
                    "account": account,
                    "tag": tag,
                    "value": value,
                    "currency": currency,
                }
            )

• Note that (a) we're keeping it simple and just passing the arguments through to our queue in a dictionary and (b) we only do this if the queue already exists (we do not want to accumulate data that we didn't ask for, in case TWS fires events on its own)

Step 3: Retrieve target data from storage on the wrapper

• Here is where queue objects shine, because we can repeatedly try to get something from an (initially) empty queue until it either: (a) gives us something or (b) times out

• Quick proof-of-concept
  

# Init an empty queue
myQ = queue.Queue()

try:
    # Try to remove an item from the queue    
    myItem = myQ.get(timeout=5)
except queue.Empty:
    print("myQ was empty and max time has been reached")

• This will of course raise the queue.Empty exception, but if some data were to magically be put into the queue during the five second timeout window... we would get the data and avoid the exception!

Step 4: Delete data storage on the wrapper object

• Since the wrapper methods run automatically, we don't want our queue objects to accumulate data sent to us from TWS unprompted. We only want data when we ask for it.

• Proof-of-concept
  

class A():
    def __init__(self):
        self.myQ = queue.Queue()

a = A()
del a.myQ

Step 5: Check for errors that may have occurred during this process

• We're going to keep track of errors on the wrapper side, because if you think about it, just like we receive data that we want on TestWrapper, that's also how we receive error messages from TWS too...

class TestWrapper(EWrapper):

    def init_error(self):
        self.my_errors_queue = queue.Queue()

    def is_error(self):
        error_exist = not self.my_errors_queue.empty()
        return error_exist

    def get_error(self, timeout=5):
        if self.is_error():
            try:
                return self.my_errors_queue.get(timeout=timeout)
            except queue.Empty:
                return None
        return None

    def error(self, id, errorCode, errorString):
        errormessage = (
            "IB returns an error with %d errorcode %d that says %s"
            % (
                id,
                errorCode,
                errorString,
            )
        )

Step 6: Return the retrieved data

• Well, this one speaks for itself

Bringing it all together into one TestClient method

class TestClient(EClient):

    def __init__(self, wrapper):
        EClient.__init__(self, wrapper)
        # Maximum timeout we're comfortable with, in seconds
        self.max_wait_time = 5

    def getAccountSummary(self):
        """
        Runs EClient.reqAccountSummary()
        Returns value from EWrapper.accountSummary()
        """
        # [1] Init a queue on the wrapper
        self.wrapper.init_accountSummary_queue()

        # [2] Request data from TWS
        self.reqAccountSummary(
            9001, "All", "TotalCashValue, BuyingPower, AvailableFunds"
        )

        try:
            # [3] Get data from queue (if it shows up) or eventually timeout
            accountSummary = self.wrapper.accountSummary_queue.get(
                timeout=self.max_wait_time
            )
        except queue.Empty:
            print("accountSummary queue was empty or max time reached")
            accountSummary = None

        # [4] Delete queue from wrapper
        del self.wrapper.accountSummary_queue

        # [5] Check for errors
        while self.wrapper.is_error():
            print("Error:")
            print(self.get_error(timeout=self.max_wait_time)

        # [6] Return data
        return accountSummary

I should also mention that you must use threading for this approach to work- otherwise wrapper events could be blocked by currently executing client functions.

To achieve that, put this in your TestApp definition:

from threading import Thread

class TestApp(TestWrapper, TestClient):

    def __init__(self, ...):
        ...
        thread = Thread(target=self.run)
        thread.start()
        setattr(self, "_thread", thread)

And that should do it! Is this the best way to implement the IB API? Probably not, but it seems to be a fairly reliable way to do so (at least in my thus far limited experience).

Let me know what you think of this design in the comments and finally, thanks for reading!