ibcore-py 0.2.0

Standalone IB Gateway integration — market data snapshots, diagnostics, order placement

ibcore

Standalone Rust crate for integrating with Interactive Brokers TWS and IB Gateway.

Wraps ibapi v3.0 with a clean async API, typed errors, market data snapshots, option chain resolution, and diagnostic event broadcasting.

Why ibcore?

The IB API has quirks. ibcore smooths them over:

Raw ibapi	ibcore
Numeric error codes with inconsistent docs	Typed IbError::FarmDisconnect
Error codes in notice stream	tokio::sync::broadcast of DiagnosticEvent
Manual contract construction	build_option_contract, get_primary_exchange
No structured disconnect detection	is_connection_dead() for reconnect logic
Raw TickType decoding	StockSnapshot / OptionSnapshot with Greeks

ibcore is built from production experience running automated options trading against IBKR. Every reconnect edge case, every data farm quirk, every undocumented behavior discovered in live trading is handled here.

Quick Start

Rust

Prerequisites

• Rust 1.85+ (edition 2024)
• IB Gateway or TWS running with API enabled

Installation

cargo add ibcore

Connect and get a stock snapshot

use ibcore::{IbClient, AccountType};

#[tokio::main]
async fn main() -> Result<(), ibcore::IbError> {
    // Connect to paper trading Gateway on port 4002
    let ib = IbClient::connect(
        "127.0.0.1",
        4002,
        1,                    // client_id
        "delayed",            // market_data_type: "delayed" or "realtime"
        AccountType::Paper,
    )
    .await?;

    // Get a stock snapshot
    let snap = ib.stock_snapshot("SPY").await?;
    println!("SPY last: ${:.2}, bid: ${:.2}, ask: ${:.2}",
        snap.last, snap.bid, snap.ask);

    ib.disconnect().await;
    Ok(())
}

Get an option chain

use ibcore::IbClient;

let chain = ib.fetch_option_chain("SPY").await?;
println!("SPY expirations: {:?}", chain.expirations);
println!("SPY strikes: {:?}", chain.strikes);

Get an option snapshot with Greeks

use ibcore::IbClient;

let snap = ib.option_snapshot(
    "SPY",
    (2025, 7, 17),  // expiry_ymd: (year, month, day)
    570.0,           // strike
    true,            // is_call (true = Call, false = Put)
    0.0,             // _implied_vol (reserved)
    0.0,             // _underlying_price (reserved)
    "CBOE",          // exchange
).await?;
println!(
    "SPY 570C: bid={:.2}, ask={:.2}, delta={:.4}, theta={:.4}, iv={:.4}",
    snap.bid, snap.ask, snap.option_delta, snap.option_theta, snap.option_iv
);

Subscribe to live market data ticks

use futures::StreamExt;

let contract = ibcore::Contract::stock("SPY").build();
let mut stream = ib.tick_stream(&contract).await?;
while let Some(event) = stream.next().await {
    match event? {
        ibcore::TickEvent::Price { tick_type, price } => {
            println!("{tick_type:?}: ${price:.2}");
        }
        ibcore::TickEvent::Greeks { delta, theta, implied_volatility, .. } => {
            println!("delta={delta:.4}, theta={theta:.4}, iv={implied_volatility:.4}");
        }
        _ => {} // ignore Size, String, Generic, etc.
    }
}

Fetch historical OHLCV bars

let contract = ibcore::Contract::stock("SPY").build();
let data = ib.historical_data(
    &contract,
    ibcore::BarSize::Hour,
    ibcore::Duration::days(5),
    ibcore::WhatToShow::Trades,
    ibcore::TradingHours::Regular,
).await?;
println!("Period: {} to {} — {} bars", data.start, data.end, data.bars.len());
for bar in &data.bars {
    println!("O={:.2}, H={:.2}, L={:.2}, C={:.2}, V={:.0}",
        bar.open, bar.high, bar.low, bar.close, bar.volume);
}

Monitor diagnostic events

use tokio::task;

let mut rx = ib.diagnostic_events();

let watcher = task::spawn(async move {
    while let Ok(event) = rx.recv().await {
        match event.farm_status {
            ibcore::FarmState::Inactive => {
                eprintln!("⚠️  Data farm inactive: code {}", event.error_code);
            }
            ibcore::FarmState::Warning => {
                eprintln!("⚠️  Data farm warning: {}", event.error_message);
            }
            _ => {}
        }
    }
});

Python

ibcore is also available as a Python package via PyO3 bindings.

Prerequisites

• Python 3.9+
• Rust toolchain (for maturin build)

Installation

pip install ibcore

Connect and get a stock snapshot

from ibcore import IbClient

ib = IbClient.connect("127.0.0.1", 4002, 1, "delayed", "paper")
snap = ib.stock_snapshot("SPY")
print(f"SPY last: ${snap.last:.2f}, bid: ${snap.bid:.2f}, ask: ${snap.ask:.2f}")
ib.disconnect()

Get an option snapshot with Greeks

snap = ib.option_snapshot("SPY", 2025, 7, 17, 570.0, True, "CBOE")
print(f"delta={snap.option_delta:.4f}, theta={snap.option_theta:.4f}, iv={snap.option_iv:.4f}")

Place an order

order_id = ib.place_order("SPY", "BUY", 100.0, "MKT", None, "SMART")
print(f"Order placed: {order_id}")

Build from source

cd python && pip install maturin && maturin develop --release

CLI Tool

ibcore ships a diagnostic CLI tool for Gateway health checks:

# Build from source
cargo build -p ibkr-diag --release

# Check versions
./target/release/ibkr-diag version

# Run a 10-second diagnostic against local Gateway
./target/release/ibkr-diag diagnose --duration 10

# JSON output for scripting
./target/release/ibkr-diag diagnose --json | jq '.farm_state_counts'

API Overview

IbClient

Method	Returns	Description
connect(host, port, client_id, market_data_type, account_type)	Result<IbClient, IbError>	Connect to IB Gateway with typed config
disconnect()	()	Graceful disconnect
stock_snapshot(symbol)	Result<StockSnapshot, IbError>	One-shot bid/ask/last/volume
option_snapshot(symbol, expiry, strike, right)	Result<OptionSnapshot, IbError>	Bid/ask/last + Greeks (delta, gamma, theta, vega, iv)
fetch_option_chain(symbol)	Result<OptionChainData, IbError>	All expirations + all strikes
positions()	Result<Vec<Position>, IbError>	Current positions
account_summary(tag, currency)	Result<Vec<AccountSummaryResult>, IbError>	NetLiq, GrossPosValue, etc.
pnl(account_id, model_code)	Result<PnL, IbError>	Realized + unrealized P&L
net_liquidation(account_id, currency)	Result<f64, IbError>	Single net liquidation value
diagnostic_events()	broadcast::Receiver<DiagnosticEvent>	Subscribe to structured diagnostic events
server_version()	i32	Gateway server version
tick_stream(contract)	Result<TickStream, IbError>	Subscribe to live market data ticks
place_order(symbol, action, quantity, order_type, limit_price, exchange)	Result<i32, IbError>	Place an order; returns order_id
cancel_order(order_id)	Result<(), IbError>	Cancel an open order
open_orders()	list[dict]	Snapshot of all open orders
order_updates()	OrderUpdateReceiver	Subscribe to live order status changes (blocking iterator)
historical_data(contract, bar_size, duration, what_to_show, trading_hours)	Result<HistoricalData, IbError>	One-shot historical OHLCV bars

DiagnosticEvent

Emitted on every IB error/warning notice:

pub struct DiagnosticEvent {
    pub gateway_version: i32,
    pub error_code: i32,
    pub error_message: String,
    pub error_time: Option<time::OffsetDateTime>,
    pub farm_status: FarmState,          // Ok | Warning | Inactive | Unknown(i32)
    pub connection_state: ConnectionState, // Connected | Disconnected | Reconnecting
    pub account_type: AccountType,       // Live | Paper
    pub os: &'static str,
    pub timestamp: chrono::DateTime<chrono::Utc>,
}

Subscribe with ib.diagnostic_events() and process asynchronously. Buffer size is 1024 events; slow subscribers will miss old events.

IbError

Typed error variants — no raw error codes in your business logic:

Variant	Trigger	Typical IB codes
ConnectionFailed	Can't connect	502, 504, 506
ConnectionReset	TCP reset	507, 100
Timeout	IO timeout	—
MarketData	Data subscription error	10000–10999
FarmDisconnect	Data farm offline	2107
CompetingSession	Live session blocks paper	10197
OrderRejected	Order invalid	200–299
ContractResolution	Contract not found	—
Other	Unclassified	everything else

Python

The IbError exception is available in Python as a hierarchy of subclasses. Each Rust variant maps to a specific Python exception type, all inheriting from IbError:

Python Exception	Rust Variant	Extra Fields
IbConnectionFailedError	ConnectionFailed	—
IbConnectionResetError	ConnectionReset	—
IbMarketDataError	MarketData	—
IbFarmDisconnectError	FarmDisconnect	—
IbOrderRejectedError	OrderRejected	rejection_json: Optional[str]
IbContractResolutionError	ContractResolution	—
IbCompetingSessionError	CompetingSession	—
IbTimeoutError	Timeout	—
IbError (base)	Other	—

All subclasses expose .code (Optional[int]) and .message (str) inherited from the base IbError class.

Backward compat: except IbError: still catches every ibcore exception. You can add more specific handlers for fine-grained control:

try:
    ib.stock_snapshot("SPY")
except IbMarketDataError as e:
    print(f"Market data error {e.code}: {e.message}")
except IbConnectionFailedError:
    reconnect()
except IbError as e:
    print(f"Unknown: {e}")

Helper functions

Function	Description
build_option_contract(symbol, expiry, strike, right)	Construct IB Contract for options
get_primary_exchange(symbol)	Map symbol to exchange (SPY→CBOE, TLT→SMART)
parse_expiry(date_str)	Parse "20260717" to NaiveDate
is_connection_dead(&IbError)	Check if error means the connection is gone
classify_farm(error_code)	Map IB error code to FarmState

Reconnection

ibcore does NOT auto-reconnect. It gives you the tools to decide:

loop {
    match IbClient::connect(host, port, id, md_type, acct_type).await {
        Ok(ib) => {
            // ... run your strategy ...
            // When the connection dies:
            break;
        }
        Err(e) if is_connection_dead(&e) => {
            tracing::warn!("Gateway unreachable, retrying in 5s: {e}");
            tokio::time::sleep(Duration::from_secs(5)).await;
            continue;
        }
        Err(e) => return Err(e), // non-connection error
    }
}

This is deliberate. Your reconnect strategy depends on your trading logic. ibcore tells you WHAT went wrong; you decide HOW to recover.

Setup

IB Gateway (Docker)

docker run -d \
  --name ib-gateway \
  -p 4002:4002 \
  -e TWS_USERID=your_username \
  -e TWS_PASSWORD=your_password \
  -e TRADING_MODE=paper \
  ghcr.io/unusualalpha/ib-gateway-docker:latest

IB Gateway (manual)

1. Download IB Gateway from IBKR
2. Configure: API → Enable ActiveX and Socket Clients → Port 4002
3. Check "Download open orders on connection"
4. Start Gateway and log in

Verify connectivity

# Check Gateway is listening
nc -zv 127.0.0.1 4002

# Run ibcore tests (requires Gateway)
cargo test --lib

Design Philosophy

• Wrap, don't re-implement. ibcore wraps ibapi v3.0, adding ergonomics and safety on top. All wire-protocol details stay in ibapi.
• Type errors, don't propagate codes. IbError is exhaustive. If you match on it, you handle every failure mode the Gateway can produce.
• Observe, don't guess. The DiagnosticEvent broadcast channel lets you monitor Gateway health without polling. Multiple consumers can subscribe independently.
• No business logic. ibcore knows about IBKR, not about trading strategies. Position tracking, portfolio allocation, risk management — those belong in your application layer.

Python Bindings

ibcore ships PyO3 Python bindings in the python/ workspace crate (ibcore-python). Build the shared library with:

# Requires Python 3.7+ and Rust 1.85+
PYO3_PYTHON=$(which python3) cargo build -p ibcore-python --release

The resulting .so file is at target/release/libibcore_python.so.

Usage

import asyncio
from ibcore._ibcore import (
    IbClient, StockSnapshot, OptionSnapshot,
    DiagnosticEvent, FarmState, ConnectionState,
    AccountType, DiagnosticEventReceiver,
)

async def main():
    # Connect to paper trading Gateway
    client = await IbClient.connect(
        "127.0.0.1", 4002, 1, "delayed", "paper",
    )

    # Stock snapshot
    snap = await client.stock_snapshot("SPY")
    print(f"SPY: last={snap.last}, bid={snap.bid}, ask={snap.ask}")

    # Option snapshot
    opt = await client.option_snapshot(
        "SPY", 2026, 7, 17, 570.0, True, "CBOE",
    )
    print(f"SPY 570C: delta={opt.option_delta}, iv={opt.option_iv}")

    # Account data
    nl = await client.net_liquidation("")
    print(f"Net Liq: ${nl:.2f}")

    # Positions
    positions = await client.positions()
    for p in positions:
        print(f"  {p['symbol']}: {p['quantity']} @ {p['avg_cost']}")

    # Diagnostics
    receiver = client.diagnostic_events()
    for event in receiver:
        if event.farm_status == FarmState.WARNING:
            print(f"⚠️  {event.error_message}")

    await client.disconnect()

asyncio.run(main())

Async context manager

async with await IbClient.connect("127.0.0.1", 4002, 1, "delayed", "paper") as ib:
    snap = await ib.stock_snapshot("SPY")
    print(snap)

Order management

Open orders and order status updates are accessible via IbClient.open_orders() and IbClient.order_updates().

from ibcore import IbClient, OrderStatusEvent

ib = IbClient.connect("127.0.0.1", 4002, 1, "delayed", "paper")

# (A) Fetch current open orders
open_orders = ib.open_orders()
for o in open_orders:
    print(f"Order #{o['order_id']}: {o['action']} {o['quantity']} {o['symbol']} @ {o['limit_price']} [{o['status']}]")

# (B) Subscribe to live order updates
receiver = ib.order_updates()
for event in receiver:
    if event.order_id == 0 and event.commission is not None:
        print(f"Commission report: ${event.commission:.2f}")
    elif event.kind == "Filled":
        print(f"Order #{event.order_id} filled: {event.filled_qty} @ ${event.avg_price:.2f}")
    elif event.kind == "Rejected":
        print(f"Order #{event.order_id} rejected: {event.reason}")
    elif event.kind == "Cancelled":
        print(f"Order #{event.order_id} cancelled: {event.reason}")
    # loop exits when receiver is dropped or stream ends (disconnect)

ib.disconnect()

Notes:

• open_orders() returns list[dict] — one dict per open order, with keys order_id, symbol, action, quantity, order_type, limit_price, status, filled_qty, remaining_qty.
• order_updates() returns an OrderUpdateReceiver (iterable). The iterator blocks on next() — use a thread wrapper for timeouts.
• Commission-only events appear as kind="Filled", order_id=0, commission=Some(x.xx). Check event.order_id == 0 and event.commission is not None to identify them. The execution_id field (None for fill-status events, Some(id) for commission-report events) allows matching commissions to their fills by temporal pairing.
• KeyboardInterrupt during next() is delayed until the next event arrives.
• Only one order_updates() subscription is allowed per IB Gateway. A second call returns IbError.

Related

• ibquirk — AI bot that diagnoses IBKR API problems using ibcore's DiagnosticEvents. Join the waitlist.

License

MIT — see LICENSE.

ibcore is not affiliated with Interactive Brokers Group, Inc.