strategy-lab/to_explore/pyquantnews/116_AutomateWithIBAPI.ipynb

In [ ]:

import threading
import time
from typing import Dict, Optional

In [ ]:

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

Calculate Donchian Channels for price data¶

This section defines a function to calculate Donchian Channels for given price data over a specified period. It calculates the upper and lower bands and optionally the middle line.

In [ ]:

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. Finally, the function returns the DataFrame with the new columns added.

Create a class to interact with Interactive Brokers API¶

This section defines a TradingApp class that interacts with the Interactive Brokers (IB) API. This class handles connections, data retrieval, and order placement.

In [ ]:

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¶

This section connects the TradingApp to the IB API and requests historical data for a specified stock. It also calculates Donchian Channels for the acquired data.

In [ ]:

app = TradingApp()

In [ ]:

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

In [ ]:

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

In [ ]:

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

In [ ]:

nvda = TradingApp.get_contract("NVDA")

In [ ]:

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

In [ ]:

donchian = donchian_channel(data, period=30)
donchian.tail()

The code creates an instance of the TradingApp class and connects it to the IB API using the specified IP address, port, and client ID. 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¶

This section implements trading logic based on the Donchian Channels. It checks for breakouts and places buy or sell orders accordingly.

In [ ]:

period = 30

In [ ]:

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 continuously request data and check for trading opportunities. 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.

Disconnect the trading app¶

This section disconnects the TradingApp from the IB API once the trading logic is complete.

In [ ]:

app.disconnect()

The code disconnects the TradingApp from the IB API. This ensures that the connection is properly closed once the trading logic is complete. It is important to disconnect to avoid leaving open connections, which can lead to issues with the API or unwanted data usage.

Your next steps¶

Try changing the stock symbol to a different one to see how the Donchian Channels and trading logic perform with other stocks. Experiment with different periods for the Donchian Channels to observe how the results change. Consider modifying the order quantity to suit your trading strategy better.

PyQuant News is where finance practitioners level up with Python for quant finance, algorithmic trading, and market data analysis. Looking to get started? Check out the fastest growing, top-selling course to get started with Python for quant finance. For educational purposes. Not investment advise. Use at your own risk.