TWS Python API Concurrency Example

Hello, and welcome to this lesson on how to utilize concurrency in the TWS API. We will be discussing how to utilize multiple, concurrent requests in the API to create a rudimentary trading system. Please be aware that this program will be referencing nearly all of the prior lessons discussed in this course and will assume that the underlying behavior of these functions is already understood. This lesson focuses on utilizing these individual pieces together, for a more cohesive system. Therefore, we have already resolved the code beforehand, and instead we can step through each piece instead of our typical format.

It is extremely important for users to operate this script exclusively in a Paper Trading environment. This is not meant as trading advice and is intended for learning purposes only.

As a general statement on the matter, concurrency is the process of handling one or more calculations or operations simultaneously. This would allow us to do something like calculate our account values while retrieving continuous market data and placing trades in the background with simultaneous execution monitoring. 

We’ll start our script with the same standard build we have used throughout our tutorial series. After establishing our imports, I will create two dictionaries to use throughout our program, which are bank and position_ref. We will leave these for now and jump down to after we’ve called our run loop.

I will make a call to reqAccountUpdates, with the “subscribe” parameter set to True, so we can maintain a constant feed of our account data. Our updateAccountValue field will be very limited in scope in this implementation. I am going to exclusively filter out the “TotalCashBalance” key in use with the “BASE” currency value. Since I am only planning to trade in my base currency of USD, I don’t care to review my other values. If your base currency is something else, such as CAD, then you may want to specify “USD” here instead. Within our if statement, I will create another if statement that will disconnect my application if the val value goes below $1 Million USD. This isn’t technically necessary, though this is a very simple failsafe to make sure my code doesn’t run my account into the ground.

Next, I can put a request to app.reqPositions() after our reqAccountUpdates call. We had touched on this in our previous lesson, but this function will return data through EWrapper.position, and take arguments for self, account, contract, position, and avgCost. This will return ongoing position updates for our entirely portfolio, so we can keep track of our account as we trade. Within the EWrapper function, I will include the contract.symbol value as a key, and set it equal to the position value. That way, I know a symbol, like AAPL, will carry a certain position value throughout the day. I will be implementing this later to make sure that my account does not trigger any short position.

Moving along, we can start moving into our symbol discovery. I am going to utilize a market scanner to decide the trending contracts of the day; however, this stage could also utilize an array of pre-defined contracts instead. For my ScannerSubscription object, I will mirror my prior lesson by focusing on major U.S. stock exchanges. This time though, I will utilize the “MOST_ACTIVE” scanCode, so I can monitor the major companies like NVDA, AAPL, or TSLA. I will also only filter the exchanges with average volume above 1000000 and a price above $10 so I can filter out relatively small companies. Then, I can send out a request for our market scanner.

For the EWrapper method, I will actually filter the results with an if statements to only those with a rank value under 5. That way, we can just trade the top 5 stocks on any given day. Within our statement, I’ll first define a new value rankId, which is just the sum of our reqId and rank and have an easy integer to track our market data. Then, we’ll set our rankId to our bank dictionary, and have it reference the contract object. This way, we can equate our market data with our symbol with ease. Now let’s set the contract symbol in our position_ref dictionary to 0. This is to instantiate a position value to further prevent any short position. All of this data can go out now and create a live market data request. I won’t be using it this time, though you are welcome to declare delayed market data if necessary. Finally, I’ll print out my top-ranking contracts, so I have a frame of reference to look back on. I will also cancel my market scanner through the scannerDataEnd function this time as well; however, I will not disconnect the program.

You may have noticed that this was the first time we have printed any details throughout this script so far. The reason for this is because our program is written to work independently of me, so the printing is primarily for our tutorial and could be ignored. The same goes for all future print references, though in a production environment it would be advised to log these types of values to validate in the future.

Before diving into our next methods, let’s recap what we’ve programmed so far to largely happen simultaneously. We first requested account updates to maintain a constant check on our account balance. In my case, I’ve instructed the program to simply disconnect once I drop beneath a certain threshold. After beginning that loop, I’ve built out a request for position updates to prevent my account from shorting. Finally, I’ve created a market scanner subscription to receive the most active instruments and requested market data for them.  Even though the only concurrent requests here are to place market data subscriptions, remember that all of these requests are operating against the run() loop, which would have otherwise locked out our requests after the first had we not implemented threading.  Now we can move on to making requests happen concurrently.

With our requests for market data happening concurrently, I’m going to build out tickPrice to handle my logic. If I was concerned about things like tickSize or tickGeneric values, I could build similar logic because of my bank dictionary’s use of request ID as a tracking metric. Starting out, I will create an initial if statement to check if the “LAST” tick type is already in my bank[reqId] keys. If the value is not there, I’ll simply assign LAST to the current price. Even though the initial run will largely be skipped since the price will be the same, this will save us any errors moving forward. Afterwards, I’ll create a variable bankTick and set it to my existing LAST value. I will also create a reference to the existing contract value we set in the scanenrData function, by calling it bankContract.

After setting our variables, we can move into some order placement logic. I can create a generic order object only containing the tif, quantity, and orderType. In my case, I am only operating with Market orders, though calculating a limit price would be easy enough as well. With an order object set, I can do a quick if/else check for last price differences greater than 5% or less than 6%. These statements will determine a potential action, BUY or SELL, and then submit my affiliated order. I can submit these orders using my nextId method, my bankContract contract, and then our new order object. In addition to the percentage calculation for my SELL logic, I am also going to check if the contract’s position is at least 5, so I don’t short. And to cap off my function, I’ll set my contract’s LAST price to my current price, to reference in the next iteration of comparison.

Now, for additional tracking metrics, I can create a reference to openOrder and execDetails. In my case, I will format them as an easy-to-read format so I can glance at the execution behavior. I am just printing rejected orders from my openOrders response, though it would stand to print all details for something like a LMT order, which may not be expected to execute immediately. As mentioned before, this is technically an optional step as it is solely used for viewing purposes. Users may look to log things like execution prices and order details for end of day review.

Though to recap our order behavior since our last, we are now acknowledging the 5 concurrent market data requests, and on each tick we are calculating whether or not an order should be triggered or not based on our very simple logic. Once the market data fits our criteria, we launch out an order, where we can find updates later in the openOrder or execDetails methods. We are now completely done with this script, and with any luck, we can earn some simulated dollars.

In a relatively short span of time, we were able to construct an entirely automated system to trade throughout the day. There is plenty of room for improvement, and flexibility from other lessons, but this is an excellent scenario to gain some confidence in making concurrent requests and handling multiple actions simultaneously.

Users interested in learning more should look to implement defensive trading logic, such as trading bracket orders rather than our simple Market orders that were discussed in the Complex Order lesson. Alternatively, you may want to create logic around historical trends of our scanned contracts, rather than plucking the top 5 to ensure safer trading. 

This concludes our lesson on utilizing concurrency in the TWS API. Thank you for watching. If you have any questions, please be sure to review our documentation or leave a comment below this video. Thank you for following along throughout the series.

Code Snippet – concurrency.py

from ibapi.client import *
from ibapi.wrapper import *
from ibapi.tag_value import TagValue
import datetime
import time
import threading

port = 7497

bank = {}
position_ref = {}

class TestApp(EClient, EWrapper):
    def __init__(self):
        EClient.__init__(self, self)

    def nextValidId(self, orderId: OrderId):
        self.orderId = orderId
    
    def nextId(self):
        self.orderId += 1
        return self.orderId
    
    def error(self, reqId, errorCode, errorString, advancedOrderReject):
        print(f"reqId: {reqId}, errorCode: {errorCode}, errorString: {errorString}, orderReject: {advancedOrderReject}")

    def updateAccountValue(self, key, val, currency, accountName):
        if key == "TotalCashBalance" and currency == "BASE":
          bank[key] = float(val)

          # If we drop below $1M, disconnect
          if float(val) <1000000:
              self.disconnect()
    
    def position(self, account, contract, position, avgCost):
        position_ref[contract.symbol] = position

    def scannerData(self, reqId, rank, contractDetails, distance, benchmark, projection, legsStr):
        if rank < 5:
            rankId = rank+reqId
            bank[rankId] = {"contract": contractDetails.contract}
            position_ref[contractDetails.contract.symbol] = 0
            app.reqMktData(rankId, contractDetails.contract, "", False, False, [])
            print(f"Rank {rank} Contract: {contractDetails.contract.symbol} @ {contractDetails.contract.exchange}")

    def scannerDataEnd(self, reqId):
        self.cancelScannerSubscription(reqId)
      
    def tickPrice(self, reqId, tickType, price, attrib):
        if "LAST" not in bank[reqId].keys():
            bank[reqId]["LAST"] = price

        bankTick = bank[reqId]["LAST"]
        bankContract = bank[reqId]["contract"]

        order = Order()
        order.tif = "DAY"
        order.totalQuantity = 5
        order.orderType = "MKT"

        # If the new price is more than 5% higher than our previous price point.
        if (bankTick * 1.05) < price:
            order.action = "BUY"
            app.placeOrder(app.nextId(), bankContract, order)
        # If the new price is less than 6% of our previous price point
        elif (bankTick * 0.94) > price and position_ref[bankContract.symbol] >= 5:
            order.action = "SELL"
            app.placeOrder(app.nextId(), bankContract, order)

        bank[reqId]["LAST"] = price
    
    def openOrder(self, orderId, contract, order, orderState):
        if orderState.status == "Rejected":
            print(f"{datetime.datetime.now()} {orderState.status}: ID:{orderId} || {order.action} {order.totalQuantity} {contract.symbol}")
    
    def execDetails(self, reqId, contract, execution):
        print(f"Execution Details: ID:{execution.orderId} || {execution.side} {execution.shares} {contract.symbol} @ {execution.time}")

app = TestApp()
app.connect("localhost", port, 1005)
threading.Thread(target=app.run).start()
time.sleep(1)

app.reqAccountUpdates(True, "")

app.reqPositions()

sub = ScannerSubscription()
sub.instrument = "STK"
sub.locationCode = "STK.US.MAJOR"
sub.scanCode = "MOST_ACTIVE"
scan_options = []
filter_options = [
    TagValue("avgVolumeAbove","1000000"),
    TagValue("priceAbove", '10')
]
app.reqScannerSubscription(app.nextId(), sub, scan_options, filter_options)