Using Technical Indicators with TWS API

Technical analysis is a very important concept for traders who look to make short-term trades or for full-time day traders seeking to identify potential entry and exit positions in the market. The TWS API provides a powerful way to implement these tools programmatically. While TWS charts offer a wide range of built-in indicators, in this article, I’ll walk you through implementing one of these indicators using the Python TWS API.

The Relative Strength Index (RSI) is one of the most widely used indicators in technical analysis. RSI helps traders identify potential overbought or oversold conditions in the market for a financial instrument. It’s calculated using average gains and losses over a specified period, usually 14 days, but this can vary depending on the timeframe of the position the trader is looking to take.

Before diving into the code, please make sure that you have TWS or IB Gateway running and a good understanding of Python programming and pandas libraries. Optionally, a basic understanding of technical indicators will help you better understand the code we’re about to write.

The implementation of RSI consists of 3 main parts: a custom wrapper class that handles API callbacks, a function that calculates the RSI, and a main application that retrieves data and processes it. Let’s start by importing the necessary libraries:

from ibapi.client import EClient
from ibapi.wrapper import EWrapper
from ibapi.contract import Contract
import pandas as pd
import time
from threading import Thread
import datetime

Next, we create a custom class that inherits from EClient and EWrapper, which handles the requests and responses from the TWS Server. We override the historicalData method which processes historical data bars. The historicalData method returns a bar object with ­­attributes like TIME, OPEN, HIGH, LOW, CLOSE, and VOLUME. For our RSI calculation, we only need TIME and CLOSE. The bar object’s TIME attribute is formatted as YYYYMMDD, but we convert it to a datetime object for better readability.

class IBapi(EWrapper, EClient):
    def __init__(self):
        EClient.__init__(self, self)
        self.data = []
        
    def historicalData(self, reqId, bar):
        # Format the date properly (IB returns dates in a strange format)
        date_str = bar.date
        
        # Handle different date formats from IB
         # YYYYMMDD format
        formatted_date = datetime.datetime.strptime(date_str, "%Y%m%d")
        
        # Append data
        self.data.append([formatted_date, bar.close])
        print(f"Received bar: Date={formatted_date}, Close={bar.close}")

For the next step, we create a function that calculates the RSI. For better readability and convenience, we use the pandas library. Our function takes a pandas dataframe as a parameter, and the second parameter is the time period for the RSI calculation (defaulting to 14 days). We calculate the difference in closing prices and assign that to a new column “change”. Then we create two new columns “gain” and “loss” which contain the positive and negative price changes respectively. Finally, we calculate the mean of gains and losses over the specified time period and calculate RSI using the standard formula.­­­

def calculate_rsi(df, periods=14):
    # Calculate price changes
    df['change'] = df['close'].diff()
    
    # Create gains and losses series
    df['gain'] = df['change'].clip(lower=0)
    df['loss'] = -1 * df['change'].clip(upper=0)
    print(df.head())
    # Calculate average gain and loss over the specified period
    avg_gain = df['gain'].rolling(window=periods).mean()
    avg_loss = df['loss'].rolling(window=periods).mean()
    
    # Calculate RS and RSI
    rs = avg_gain / avg_loss
    df['rsi'] = 100 - (100 / (1 + rs))
    
    
    return df

The main application connects to TWS/IB Gateway, retrieves historical data, and calculates the RSI. This function manages the connection to TWS, retrieves historical data, calculates the RSI, and interprets the result. We run the socket connection in a separate thread to prevent blocking, which is a common pattern when working with the TWS API.

def run_app():
    # Connect to TWS
    app = IBapi()
    app.connect('127.0.0.1', 7497, 0)  # Use 7496 for TWS, 7497 for IB Gateway
    
    # Start the socket in a thread
    api_thread = Thread(target=app.run)
    api_thread.start()
    time.sleep(0.5)  # Give time for connection to establish
    
    # Define the contract for the stock
    contract = Contract()
    contract.symbol = "PLTR"  # Change to your stock symbol
    contract.secType = "STK"
    contract.exchange = "SMART"
    contract.currency = "USD"
    
    # Request historical data
    app.reqHistoricalData(1, contract, "", "15 D", "1 day", "TRADES", 0, 1, False, [])
    time.sleep(0.5)  # Wait for data to be received
    
    # Process the data
    df = pd.DataFrame(app.data, columns=['date', 'close'])
    df = df.sort_values('date')
    df = calculate_rsi(df)
    
    # Display results
    valid_rsi = df.dropna(subset=['rsi'])
    print(f"Recent RSI for {contract.symbol}: {valid_rsi['rsi'].iloc[-1]:.2f}")
    
    
    app.disconnect()

if __name__ == "__main__":
    run_app()

This is a straightforward implementation of RSI using the TWS API. We’ve utilized different libraries to make our code concise and readable. You could enhance this by using other technical analysis libraries like TA-Lib, incorporating this into a trading class to make trades based on RSI signals, changing the time period to fit your trading strategy, or combining RSI with other indicators like moving averages, Bollinger bands, or volume indicators.

Remember that RSI is just one indicator among many that traders utilize to make decisions, and it’s often used in conjunction with other technical tools. Feel free to include print statements in different parts of the code to observe how the dataframe changes during the execution process. This implementation provides a solid foundation for exploring technical analysis with the TWS API, allowing you to build more sophisticated trading tools and strategies.

The complete code should appear as follows:

from ibapi.client import EClient
from ibapi.wrapper import EWrapper
from ibapi.contract import Contract
import pandas as pd
import time
from threading import Thread
import datetime

class IBapi(EWrapper, EClient):
    def __init__(self):
        EClient.__init__(self, self)
        self.data = []
        
    def historicalData(self, reqId, bar):
        # Format the date properly (IB returns dates in a strange format)
        date_str = bar.date
        
        # Handle different date formats from IB
         # YYYYMMDD format
        formatted_date = datetime.datetime.strptime(date_str, "%Y%m%d")
        
        # Append data
        self.data.append([formatted_date, bar.close])
        print(f"Received bar: Date={formatted_date}, Close={bar.close}")

def calculate_rsi(df, periods=14):
    # Calculate price changes
    df['change'] = df['close'].diff()
    
    # Create gains and losses series
    df['gain'] = df['change'].clip(lower=0)
    df['loss'] = -1 * df['change'].clip(upper=0)
    print(df.head())
    # Calculate average gain and loss over the specified period
    avg_gain = df['gain'].rolling(window=periods).mean()
    avg_loss = df['loss'].rolling(window=periods).mean()
    
    # Calculate RS and RSI
    rs = avg_gain / avg_loss
    df['rsi'] = 100 - (100 / (1 + rs))
    
    return df

def run_app():
    # Connect to TWS
    app = IBapi()
    app.connect('127.0.0.1', 7496, 0)  # Use 7496 for TWS, 7497 for IB Gateway
    
    # Start the socket in a thread
    api_thread = Thread(target=app.run)
    api_thread.start()
    time.sleep(0.5)  # Give time for connection to establish
    
    # Define the contract for the stock
    contract = Contract()
    contract.symbol = "PLTR"  # Change to your stock symbol
    contract.secType = "STK"
    contract.exchange = "SMART"
    contract.currency = "USD"
    
    # Request historical data
    app.reqHistoricalData(1, contract, "", "15 D", "1 day", "TRADES", 0, 1, False, [])
    time.sleep(0.5)  # Wait for data to be received
    
    # Process the data
    df = pd.DataFrame(app.data, columns=['date', 'close'])
    df = df.sort_values('date')
    df = calculate_rsi(df)
    
    # Display results
    valid_rsi = df.dropna(subset=['rsi'])
    print(f"Recent RSI for {contract.symbol}: {valid_rsi['rsi'].iloc[-1]:.2f}")
    
    app.disconnect()

if __name__ == "__main__":
    run_app()