Interactive Brokers

The following code examples for integrating with Interactive Brokers are included with Quantstrip. They can all be downloaded from the Script Templates section on the Quantstrip website.

• ib_connect.py
• ib_objects.py
• ib_send_order.py

IB API vs REST API

Interactive Brokers offers two primary APIs for algorithmic trading and account management: the TWS API (also known as the IB API) and the REST API (Client Portal Web API).

The TWS API is a socket-based interface that connects through either the Trader Workstation (TWS) desktop application or IB Gateway. It provides real-time streaming data, comprehensive order management, and access to historical market data. One of its key advantages is authentication persistence—users only need to log in once per week, making it more convenient for automated trading systems that run continuously.

The REST API (Client Portal Web API) is a modern, web-based interface that uses standard HTTP requests, making it easier to integrate with web applications and services that don't require constant connectivity. However, it requires daily authentication, as users must log in each trading day through the Client Portal gateway. This makes it less suitable for fully automated systems but more accessible for developers familiar with RESTful architecture.

The choice between the two depends on the use case, but for continuously running trading applications the weekly authentication requirements of the TWS API makes it the better choice.

---

Synchronous IB API Wrapper

The TWS API uses an asynchronous, callback-based design where requests and responses are decoupled—you send a request through one method and receive the response later through a separate callback function. While this architecture is powerful for handling high-frequency, real-time data streams, it presents a steep learning curve for developers accustomed to synchronous programming patterns.

For intraday trading strategies operating on a 5-10 minute timeframe where millisecond precision isn't critical, many developers find it beneficial to build a synchronous wrapper around the TWS API. This wrapper can convert the asynchronous callbacks into blocking function calls that wait for and return responses directly, making the code more intuitive and easier to reason about. The wrapper effectively bridges the gap between the API's event-driven architecture and traditional procedural code, simplifying strategy development while maintaining full access to the API's capabilities. For slower-paced trading strategies, the slight overhead introduced by this abstraction is negligible and the improved code clarity is well worth the trade-off.

---

ib_connect

This module provides a synchronous wrapper around Interactive Brokers' asynchronous Python API (ibapi). It uses threading events and thread-safe request IDs to convert IB's callback-based architecture into blocking get_* methods that wait for responses and return structured data. Each method manages its own event synchronization, making the API behave like traditional synchronous function calls while maintaining full access to IB's capabilities.

---

Connection Management

connect_client(host="127.0.0.1", port=7497, client_id=1)

Connect to IB Gateway or TWS. Returns True if connected within 5 seconds, otherwise False.

Parameters:

Parameter	Type	Description
host	str	IB Gateway/TWS hostname (default: "127.0.0.1")
port	int	API port (7497 for paper, 7496 for live)
client_id	int	Unique client identifier

Returns: bool

Example:

from IBKR.ib_connect import IB

ib = IB()
if ib.connect_client(host="127.0.0.1", port=7497, client_id=5):
    print("Connected successfully")
else:
    print("Connection failed")

Automatic Client ID Assignment

The ib_connect module provides a client ID pool manager that automatically assigns an available ID to the connection unless specified as an argument. Once the connection closes, the ID is returned to the pool. This is convenient when many clients connect and disconnect at different times and eliminates the need to keep track of already connected IDs in the client code.

NOTE: Clients collecting information from other clients, for example a centralized client keeping track of all executions, needs to connect with client_id = 0 in order to receive callbacks originating from all other clients.

---

disconnect_client()

Disconnect from IB and clean up the socket thread.

Example:

ib.disconnect_client()

---

is_connected()

Returns True if currently connected to IB.

Returns: bool

Example:

if ib.is_connected():
    print("Still connected")

---

get_client_id_stats()

Get statistics about client ID usage from the connection pool.

Returns: Dictionary with 'available', 'in_use', and 'total' counts

Example:

stats = IB.get_client_id_stats()
print(f"Available IDs: {stats['available']}")
print(f"In use: {stats['in_use']}")

---

Using the Context Manager for Safe Connection Handling

IB Gateway and TWS connections must be closed properly. To do this safely, careful exception handling is required to ensure that the connection is closed if the client job crashes. A convenient alternative to manage this is by using the context manager included in the IBKR template code.

from IBKR.ib_connect import IB

with IB() as ib:
    df = ib.get_historical_data(...)
    price = ib.get_last_price(...)
# Connection automatically closed when exiting context

This syntax instantiates the IB class as the object ib and tries to connect to IB in a single line of code. By default it connects to the paper trading environment. To connect to the production environment instead, simply provide the production port number:

with IB(port=7496) as ib:  # Live trading port
    # Your trading logic here

The context manager automatically: - Assigns a client ID from the pool (or uses one you specify) - Connects to IB Gateway/TWS - Releases the client ID back to the pool on exit - Handles cleanup even if exceptions occur

---

Market Data Methods

get_historical_data(contract, endDateTime="", durationStr="1 D", barSizeSetting="1 min", whatToShow="TRADES", useRTH=1, formatDate=1, keepUpToDate=False, chartOptions=None, timeout=10.0)

Request historical bars. Returns pandas DataFrame with columns: open, high, low, close, volume, indexed by datetime. Returns empty DataFrame on timeout/error.

Parameters:

Parameter	Type	Description
contract	Contract	IB contract object
endDateTime	str	End date/time (empty string = now)
durationStr	str	Duration (e.g., "1 D", "2 W", "1 M")
barSizeSetting	str	Bar size (e.g., "1 min", "5 mins", "1 hour")
whatToShow	str	Data type ("TRADES", "MIDPOINT", "BID", "ASK")
useRTH	int	1 = regular trading hours only, 0 = all hours
formatDate	int	1 = yyyyMMdd HHss, 2 = Unix timestamp
keepUpToDate	bool	Keep subscription active for updates
chartOptions	List	Additional chart options
timeout	float	Timeout in seconds

Returns: pandas.DataFrame

Example:

from ibapi.contract import Contract

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

with IB() as ib:
    df = ib.get_historical_data(
        contract,
        durationStr="5 D",
        barSizeSetting="1 hour",
        whatToShow="TRADES"
    )
    print(df.head())

---

get_last_price(contract, timeout=5.0)

Request market snapshot and return the most recent trade price as float. Returns None if unavailable.

Parameters:

Parameter	Type	Description
contract	Contract	IB contract object
timeout	float	Timeout in seconds

Returns: Optional[float]

Example:

with IB() as ib:
    price = ib.get_last_price(contract)
    if price:
        print(f"Current price: ${price:.2f}")
    else:
        print("Price not available")

---

get_market_snapshot(contract, timeout=5.0)

Request market snapshot. Returns dict with keys: last, bid, ask, last_size, last_tick. Returns empty dict on timeout.

Parameters:

Parameter	Type	Description
contract	Contract	IB contract object
timeout	float	Timeout in seconds

Returns: Dict[str, Any]

Example:

with IB() as ib:
    snapshot = ib.get_market_snapshot(contract)
    print(f"Last: {snapshot.get('last')}")
    print(f"Bid: {snapshot.get('bid')}")
    print(f"Ask: {snapshot.get('ask')}")
    print(f"Last Size: {snapshot.get('last_size')}")

---

get_realtime_bars(contract, whatToShow="TRADES", useRTH=0, timeout=5.0)

Request 5-second realtime bars. Returns DataFrame containing the first bar received with columns: time, open, high, low, close, volume, wap, count. Returns empty DataFrame on timeout.

Parameters:

Parameter	Type	Description
contract	Contract	IB contract object
whatToShow	str	Data type ("TRADES", "MIDPOINT", "BID", "ASK")
useRTH	int	1 = regular trading hours only, 0 = all hours
timeout	float	Timeout in seconds

Returns: pandas.DataFrame

Example:

with IB() as ib:
    bars = ib.get_realtime_bars(contract, whatToShow="TRADES")
    if not bars.empty:
        print(f"Latest bar: {bars.iloc[0]}")

---

get_historical_ticks(contract, startDateTime=None, endDateTime=None, numberOfTicks=1000, whatToShow="TRADES", useRth=1, timeout=10.0)

Request historical tick data. Returns list of tick dicts. Returns empty list on timeout/error.

Parameters:

Parameter	Type	Description
contract	Contract	IB contract object
startDateTime	str	Start date/time
endDateTime	str	End date/time
numberOfTicks	int	Maximum number of ticks to retrieve
whatToShow	str	Tick type ("TRADES", "BID_ASK", "MIDPOINT")
useRth	int	1 = regular trading hours only, 0 = all hours
timeout	float	Timeout in seconds

Returns: List[Dict[str, Any]]

Example:

with IB() as ib:
    ticks = ib.get_historical_ticks(
        contract,
        numberOfTicks=100,
        whatToShow="TRADES"
    )
    print(f"Received {len(ticks)} ticks")

---

Account & Portfolio Methods

get_positions(timeout=10.0)

Request current positions. Returns list of dicts with keys: account, contract, position, avgCost. Returns empty list on timeout/error.

Parameters:

Parameter	Type	Description
timeout	float	Timeout in seconds

Returns: List[Dict[str, Any]]

Example:

with IB() as ib:
    positions = ib.get_positions()
    for pos in positions:
        contract = pos['contract']
        print(f"{contract['symbol']}: {pos['position']} @ ${pos['avgCost']:.2f}")

---

get_portfolio(account="", timeout=10.0)

Subscribe briefly to account updates to obtain portfolio snapshot. Returns list of dicts with keys: contract, position, marketPrice, marketValue, averageCost, unrealizedPNL, realizedPNL, accountName. Returns empty list on timeout.

Parameters:

Parameter	Type	Description
account	str	Account identifier (empty = default)
timeout	float	Timeout in seconds

Returns: List[Dict[str, Any]]

Example:

with IB() as ib:
    portfolio = ib.get_portfolio()
    for item in portfolio:
        contract = item['contract']
        print(f"{contract['symbol']}: P&L ${item['unrealizedPNL']:.2f}")

---

get_account_summary(group="All", tags=None, timeout=10.0)

Request account summary values. Default tags: NetLiquidation, TotalCashValue, AvailableFunds, BuyingPower. Returns DataFrame with columns: reqId, account, tag, value, currency. Returns empty DataFrame on timeout/error.

Parameters:

Parameter	Type	Description
group	str	Account group ("All" for all accounts)
tags	str	Comma-separated list of tags to retrieve
timeout	float	Timeout in seconds

Returns: pandas.DataFrame

Example:

with IB() as ib:
    summary = ib.get_account_summary()
    for idx, row in summary.iterrows():
        print(f"{row['tag']}: {row['value']} {row['currency']}")

---

Order Methods

get_next_order_id(timeout=5.0)

Request the next valid order ID from IB. This is useful when you need to place orders and want to ensure you're using a valid order ID that won't conflict.

Parameters:

Parameter	Type	Description
timeout	float	Timeout in seconds

Returns: Optional[int] - Next valid order ID, or None on timeout

Example:

with IB() as ib:
    order_id = ib.get_next_order_id()
    if order_id:
        print(f"Next order ID: {order_id}")
        # Use this ID when placing orders

---

get_executions(executionFilter=None, timeout=10.0)

Request trade executions. Returns list of dicts containing order details with structure:

{
    'reqId': int,
    'contract': {
        'symbol': str,
        'secType': str,
        'exchange': str,
        'currency': str,
        'localSymbol': str,
        'conId': int,
        'primaryExchange': str
    },
    'execution': {
        'orderId': int,
        'clientId': int,
        'execId': str,
        'time': str,
        'acctNumber': str,
        'exchange': str,
        'side': str,
        'shares': float,
        'price': float,
        'permId': int,
        'liquidation': int,
        'cumQty': float,
        'avgPrice': float,
        'orderRef': str,
        'evRule': str,
        'evMultiplier': float,
        'modelCode': str,
        'lastLiquidity': int
    }
}

Parameters:

Parameter	Type	Description
executionFilter	ExecutionFilter	Filter criteria (None = all executions)
timeout	float	Timeout in seconds

Returns: List[Dict[str, Any]]

Example:

with IB() as ib:
    executions = ib.get_executions()
    for exec in executions:
        contract = exec['contract']
        execution = exec['execution']
        print(f"{contract['symbol']}: {execution['side']} {execution['shares']} @ ${execution['price']}")

---

get_commissions(executionFilter=None, timeout=10.0)

Request commission reports. Returns list of dicts with commission structure:

{
    'execId': str,
    'commission': float,
    'currency': str,
    'realizedPNL': float,
    'yield': float,
    'yieldRedemptionDate': int
}

Note: This calls reqExecutions() which triggers both execDetails and commissionReport callbacks, but only returns the commission data.

Parameters:

Parameter	Type	Description
executionFilter	ExecutionFilter	Filter criteria (None = all)
timeout	float	Timeout in seconds

Returns: List[Dict[str, Any]]

Example:

with IB() as ib:
    commissions = ib.get_commissions()
    total_commission = sum(c['commission'] for c in commissions)
    print(f"Total commissions: ${total_commission:.2f}")

---

get_open_orders(timeout=10.0)

Request open orders for this client. Returns list of dicts with keys: orderId, contract, order, orderState. Returns empty list on timeout/error.

Parameters:

Parameter	Type	Description
timeout	float	Timeout in seconds

Returns: List[Dict[str, Any]]

Example:

with IB() as ib:
    orders = ib.get_open_orders()
    for order in orders:
        contract = order['contract']
        order_info = order['order']
        state = order['orderState']
        print(f"Order {order['orderId']}: {contract['symbol']} "
              f"{order_info['action']} {order_info['totalQuantity']} - {state['status']}")

---

get_order_status(order_id, timeout=5.0)

Retrieve latest known status for a specific order ID. Returns dict with status details or None if unknown.

Parameters:

Parameter	Type	Description
order_id	int	Order identifier
timeout	float	Timeout in seconds

Returns: Optional[Dict[str, Any]]

Example:

with IB() as ib:
    status = ib.get_order_status(order_id=12345)
    if status:
        print(f"Status: {status['status']}")
        print(f"Filled: {status['filled']}")
        print(f"Remaining: {status['remaining']}")

---

Contract & Reference Data Methods

get_contract_details(contract, timeout=10.0)

Request detailed contract specifications. Returns list of dicts containing contractDetails objects. Returns empty list on timeout/error.

Parameters:

Parameter	Type	Description
contract	Contract	IB contract object
timeout	float	Timeout in seconds

Returns: List[Dict[str, Any]]

Example:

with IB() as ib:
    details = ib.get_contract_details(contract)
    for item in details:
        cd = item['contractDetails']
        print(f"Contract: {cd.contract.symbol}")
        print(f"Long Name: {cd.longName}")
        print(f"Min Tick: {cd.minTick}")

---

get_fundamental_data(contract, reportType="ReportSnapshot", timeout=10.0)

Request fundamental data (XML/text format). Returns string data or None on timeout/error.

Parameters:

Parameter	Type	Description
contract	Contract	IB contract object
reportType	str	Report type ("ReportSnapshot", "ReportsFinSummary", etc.)
timeout	float	Timeout in seconds

Returns: Optional[str]

Example:

with IB() as ib:
    fundamentals = ib.get_fundamental_data(contract, reportType="ReportSnapshot")
    if fundamentals:
        print(fundamentals[:500])  # Print first 500 chars

---

get_news_bulletins(timeout=5.0)

Subscribe briefly to news bulletins. Returns list of bulletin dicts with keys: msgId, msgType, message, exchange. Best-effort; may return empty list.

Parameters:

Parameter	Type	Description
timeout	float	Timeout in seconds

Returns: List[Dict[str, Any]]

Example:

with IB() as ib:
    bulletins = ib.get_news_bulletins()
    for bulletin in bulletins:
        print(f"{bulletin['msgType']}: {bulletin['message']}")

---

Complete Usage Example

from IBKR.ib_connect import IB
from ibapi.contract import Contract

# Create contract
contract = Contract()
contract.symbol = "AAPL"
contract.secType = "STK"
contract.exchange = "SMART"
contract.currency = "USD"

# Use as context manager
with IB() as ib:
    # Get historical data
    df = ib.get_historical_data(
        contract,
        durationStr="5 D",
        barSizeSetting="1 hour"
    )
    print(f"Retrieved {len(df)} bars")

    # Get current price
    price = ib.get_last_price(contract)
    print(f"Current price: ${price:.2f}")

    # Get market snapshot
    snapshot = ib.get_market_snapshot(contract)
    print(f"Bid: ${snapshot.get('bid'):.2f}, Ask: ${snapshot.get('ask'):.2f}")

    # Get positions
    positions = ib.get_positions()
    for pos in positions:
        print(f"Position: {pos['contract']['symbol']}, Qty: {pos['position']}")

    # Get account summary
    summary = ib.get_account_summary()
    for idx, row in summary.iterrows():
        if row['tag'] == 'NetLiquidation':
            print(f"Net Liquidation: ${row['value']}")

    # Get executions
    executions = ib.get_executions()
    print(f"Total executions: {len(executions)}")

# Connection automatically closed when exiting context

---

IB Objects

ib_objects.py

This module provides example factory functions to create Interactive Brokers API objects (Contracts and Orders) with sensible defaults for common trading scenarios. It has not been the ambition to wrap all available object types in the IB API, but rather to provide examples of how this can be done for contract and order objects.

---

Contract Function

ib_contract(symbol)

Creates an IB contract object for US stocks and ETFs. Extend this function with additional properties and logic or create additional ones for more advanced contracts.

Parameters:

Parameter	Type	Description
symbol	str	Stock or ETF ticker symbol (e.g., "AAPL", "SPY")

Returns: Contract

Default Settings:

• currency: "USD"
• exchange: "SMART"
• secType: "STK" (stock)

Example:

from IBKR.ib_objects import ib_contract

contract = ib_contract("SPY")
# Ready to use for market data or order placement

# Contract can be customized further
contract.primaryExchange = "ARCA"

---

Order Functions

ib_order(quantity, order_ref='', orderType='MOC')

Creates an example IB order object with automatic buy/sell direction based on quantity sign. Extend this function with additional properties and logic or create additional ones for more advanced order types.

Parameters:

Parameter	Type	Description
quantity	int	Order quantity (positive = BUY, negative = SELL)
order_ref	str	Optional reference string for order tracking
orderType	str	Order type ("MOC", "MKT", "LMT", etc.)

Returns: Order

Default Settings:

• action: "BUY" or "SELL" (determined by quantity sign)
• totalQuantity: Absolute value of quantity
• exchange: "SMART"
• eTradeOnly: False
• firmQuoteOnly: False

Example:

from IBKR.ib_objects import ib_order

# Market on close buy order
buy_order = ib_order(quantity=100, orderType="MOC")

# Market sell order with reference
sell_order = ib_order(quantity=-50, order_ref="close_position", orderType="MKT")

# Limit buy order (note: you'll need to set order.lmtPrice separately)
limit_order = ib_order(quantity=200, orderType="LMT")
limit_order.lmtPrice = 150.00

# Stop order
stop_order = ib_order(quantity=-100, order_ref="stop_loss", orderType="STP")
stop_order.auxPrice = 145.00  # Stop price

---

Sending an Order to IB

ib_send_order.py

This module contains a minimal test client demonstrating order placement through Interactive Brokers with database integration. This module shows a complete workflow: connecting to IB, placing an order, and recording the transaction in a database.

Note: This is a reference implementation. Production usage requires proper error handling, connection management, and strategy configuration.

---

Client Class

Client(ClientBase)

A test client that sends a single order to IB and records it in the database.

Features:

• Inherits from ClientBase (Quantstrip framework)
• Connects to IB Gateway/TWS
• Places market order
• Records order in database
• Updates internal position tracking
• Automatically disconnects and stops after execution

---

Configuration

Default Settings

contract = ib_contract("SPY")
order = ib_order(quantity=100, orderType="MKT")

• Symbol: SPY (S&P 500 ETF)
• Quantity: 100 shares (BUY)
• Order Type: Market order
• Client ID: 5 (for IB connection)

---

Workflow

The client executes the following sequence:

1. Connect to IB: Establishes connection with client_id=5
2. Get Order ID: Retrieves next available order ID from database
3. Place Order: Submits order to Interactive Brokers using placeOrder()
4. Record Order: Inserts order details into database orders table
5. Update Position: Inserts strategy event into database
6. Disconnect: Closes IB connection
7. Stop Client: Terminates the client process

---

Database Integration

Order Record

db_handler.insert_order(
    order_id=order_id,
    strategy_id=1,      # Test Strategy
    broker_id=1,        # IB broker
    account="U1234567",
    symbol="SPY",
    side="BUY",
    order_type="MKT",
    total_quantity=100,
    instrument_type="STK",
    client_reference="test_order"
)

Strategy Event Record

db_handler.insert_strategy_event(
    order_id=order_id,
    position=100,
    event_type="OPEN",
    status="PENDING",
    metadata={"source": "test_client"}
)

---

Prerequisites

Database Setup

A strategy must exist in the database strategy table before running:

-- Example: Ensure strategy_id=1 exists
INSERT INTO strategy (strategy_id, name, description)
VALUES (1, 'Test Strategy', 'Testing order placement');

IB Connection

• IB Gateway or TWS must be running
• API connections must be enabled
• Default port: 7497 (paper trading) or 7496 (live)

---

Usage Example

from ib_send_order import Client

# Initialize and run client
client = Client()
client.start()

Scheduler Behavior:

• Job runs once after 1 second
• Client automatically stops after execution
• Use for testing or one-time order placement

---

Customization

Different Symbol/Quantity

# At module level, before Client definition
contract = ib_contract("AAPL")
order = ib_order(quantity=-50, orderType="MKT")  # Sell 50 shares

Different Strategy

# In job() method
db_handler.insert_order(
    order_id=order_id,
    strategy_id=2,  # Change to your strategy ID
    # ... rest of parameters
)

Continuous Operation

class Client(ClientBase):
    def __init__(self, *args):
        super().__init__()
        # Run every hour instead of once
        self.scheduler.every(1).hours.do(self.job)

    def job(self):
        # ... order placement logic ...
        # Remove self.stop_client() to keep running

---

Error Handling

The client includes try-catch-finally structure:

• Try: Execute order placement and database operations
• Catch: Log errors with full traceback
• Finally: Always disconnect from IB (prevents connection leaks)

try:
    # Order placement logic
except Exception as e:
    logger.info(f"Error: {e}")
    logger.info(traceback.format_exc())
finally:
    ib.disconnect_client()

---

Important Notes

Production Considerations

• This is a test/reference implementation
• Add proper validation and error recovery for production use
• Consider order status confirmation before database insertion
• Implement retry logic for connection failures
• Add position reconciliation checks

Database Requirements

• `strategy_id=1