Automate Trading Strategies with the Powerful IB API

The article “Automate Trading Strategies with the Powerful IB API” was originally on PyQuantNews.

Automate trading strategies with the powerful IB API

Trading in 2024 is overwhelming.

Especially when it comes to automating trading strategies.

Many traders rely on manual processes or basic automation, which often fall short by being too slow or not adaptable enough. I’ve been there myself, trying to juggle automation with using my own intuition.

The good news is you can solve this problem by using Python to automate your trading strategies with the Interactive Brokers API (IB API).

I’ve been using Interactive Brokers for more than 10 years. The IB API is infamous because of how complicated it can be to get up and running.

In today’s newsletter, I put it altogether so you can quickly use the IB API to automate your first trading strategy.

Let’s go!

Automate trading strategies with the powerful IB API

Automating trading strategies is a powerful tool for traders seeking efficiency. Interactive Brokers offers a Python API that automates trading strategies.

This API provides access to market data, account information, and order management.

Professionals leverage the API for algorithmic trading, using predefined rules and models for executing trades. The API’s comprehensive functionality is great for both novice and experienced traders aiming for precise control over their trading strategies.

Let’s see how it works with Python.

Imports and set up

You can download and install the Interactive Brokers API from GitHub. Here’s a step-by-step guide. Once the libraries are installed, you can import them.

import time
import threading
from datetime import datetime
from typing import Dict, Optional
import pandas as pd
import warnings

from ibapi.client import EClient
from ibapi.wrapper import EWrapper
from ibapi.contract import Contract
from ibapi.order import Order
from ibapi.common import BarData

Our strategy will look for breakouts using a popular technical indicator called Donchian Channels.

Here, we define a function to calculate Donchian Channels for given price data over a specified period. It calculates the upper and lower bands and the middle line.

def donchian_channel(df: pd.DataFrame, period: int = 30) -> pd.DataFrame:

    df["upper"] = df["high"].rolling(window=period).max()

    df["lower"] = df["low"].rolling(window=period).min()

    df["mid"] = (df["upper"] + df["lower"]) / 2

    return df

This function takes a DataFrame containing price data and a period for the calculation. It computes the upper band as the highest high over the period and the lower band as the lowest low. The middle line is calculated as the average of the upper and lower bands.

Create a class to interact with Interactive Brokers API

This section defines a TradingApp class that interacts with the IB API. This class handles connections, data retrieval, and order placement. It’s our trading app.

class TradingApp(EClient, EWrapper):

    def __init__(self) -> None:

        EClient.__init__(self, self)
        self.data: Dict[int, pd.DataFrame] = {}
        self.nextOrderId: Optional[int] = None

    def error(self, reqId: int, errorCode: int, errorString: str, advanced: any) -> None:

        print(f"Error: {reqId}, {errorCode}, {errorString}")

    def nextValidId(self, orderId: int) -> None:

        super().nextValidId(orderId)
        self.nextOrderId = orderId

    def get_historical_data(self, reqId: int, contract: Contract) -> pd.DataFrame:

        self.data[reqId] = pd.DataFrame(columns=["time", "high", "low", "close"])
        self.data[reqId].set_index("time", inplace=True)
        self.reqHistoricalData(
            reqId=reqId,
            contract=contract,
            endDateTime="",
            durationStr="1 D",
            barSizeSetting="1 min",
            whatToShow="MIDPOINT",
            useRTH=0,
            formatDate=2,
            keepUpToDate=False,
            chartOptions=[],
        )
        time.sleep(5)
        return self.data[reqId]

    def historicalData(self, reqId: int, bar: BarData) -> None:

        df = self.data[reqId]

        df.loc[
            pd.to_datetime(bar.date, unit="s"), 
            ["high", "low", "close"]
        ] = [bar.high, bar.low, bar.close]

        df = df.astype(float)

        self.data[reqId] = df

    @staticmethod
    def get_contract(symbol: str) -> Contract:

        contract = Contract()
        contract.symbol = symbol
        contract.secType = "STK"
        contract.exchange = "SMART"
        contract.currency = "USD"
        return contract

    def place_order(self, contract: Contract, action: str, order_type: str, quantity: int) -> None:

        order = Order()
        order.action = action
        order.orderType = order_type
        order.totalQuantity = quantity

        self.placeOrder(self.nextOrderId, contract, order)
        self.nextOrderId += 1
        print("Order placed")

The TradingApp class extends EClient and EWrapper to interact with the IB API.

It initializes the client and wrapper components and sets up data storage. The error method handles API errors, while nextValidId sets the next valid order ID. The get_historical_data method requests historical market data for a given contract, storing it in a DataFrame. The historicalData method processes and stores the received data.

The get_contract method creates a stock contract, and the place_order method places trades using the provided contract, action, order type, and quantity.

Connect the trading app and request data

Once we build our trading app, we will connect to it and look for signals.

app = TradingApp()

app.connect("127.0.0.1", 7497, clientId=5)

threading.Thread(target=app.run, daemon=True).start()

while True:
    if isinstance(app.nextOrderId, int):
        print("connected")
        break
    else:
        print("waiting for connection")
        time.sleep(1)


nvda = TradingApp.get_contract("NVDA")

Once the app is connected, you can download historical data.

data = app.get_historical_data(99, nvda)
data.tail()

The code creates an instance of the TradingApp class and connects it to the IB API using the IP address, port, and client ID. Note the port is the default for the IB paper trading account.

It then starts the app on a separate thread to allow code execution to continue. A loop checks for a successful connection by verifying the nextOrderId.

Once connected, it defines a contract for the stock symbol NVDA and requests historical data for the last trading day using a specified request ID. The Donchian Channels are then calculated for the acquired data.

Implement trading logic based on Donchian Channels

Now, we check for breakouts and places buy or sell orders accordingly.

period = 30

while True:

    print("Getting data for contract...")
    data = app.get_historical_data(99, nvda)

    if len(data) < period:
        print(f"There are only {len(data)} bars of data, skipping...")
        continue

    print("Computing the Donchian Channel...")
    donchian = donchian_channel(data, period=period)

    last_price = data.iloc[-1].close

    upper, lower = donchian[["upper", "lower"]].iloc[-1]

    print(f"Check if last price {last_price} is outside the channels {upper} and {lower}")

    if last_price >= upper:
        print("Breakout detected, going long...")
        app.place_order(nvda, "BUY", "MKT", 10)

    elif last_price <= lower:
        print("Breakout detected, going short...")
        app.place_order(nvda, "SELL", "MKT", 10)

The code sets the period for the Donchian Channels to 30. It enters an infinite loop to request data and check for trading opportunities continuously.

It retrieves historical data for the NVDA contract and skips further processing if there is insufficient data. It then calculates the Donchian Channels and gets the last traded price.

The code compares the last price with the upper and lower channels to detect breakouts. If a breakout to the upside is detected, it places a buy market order for 10 shares. If a breakout to the downside is detected, it places a sell market order for 10 shares.

To exit this loop, stop the Python kernel. Once you’re done, disconnect the app.

app.disconnect()

Your next steps

This is a simple example of downloading historical market data, looking for signals, and executing trades with Python. It lacks risk and position management, which you should implement based on your experience and risk tolerance. The biggest area for improvement is checking for existing positions before entering new trades.

Finally, make sure to run this code in your paper trading account. Trading it live will expose you to risk of loss.

Go Source