Skip to content

Interactive Brokers

Connect to IBKR TWS or Gateway for live trading.

Warning

Requires ib_insync package: pip install ib_insync

ibkr

Interactive Brokers adapter using ib_insync.

Connects to TWS or IB Gateway for live/paper trading. Requires TWS/Gateway running with API connections enabled.

Example 
from adapters import get_adapter

adapter = get_adapter("ibkr", {
    "host": "127.0.0.1",
    "port": 7497,  # 7497 for paper, 7496 for live
    "client_id": 1,
})
adapter.connect()

# Fetch quote
quote = adapter.fetch_quote("AAPL", 185.0)
print(quote)  # {"bid": 186.45, "ask": 186.55, ...}

# Execute order
result = adapter.execute_buy("AAPL", 10, 186.50)
print(result)  # {"success": True, "fill_price": 186.52, ...}

adapter.disconnect()
Note

Install with: pip install ib_insync Or use the optional dependency: pip install tradetracer-executor[ibkr]

IBKRAdapter

Bases: BaseAdapter

Interactive Brokers adapter.

Connects to TWS or IB Gateway via the ib_insync library. Supports both live and paper trading accounts.

Attributes:

Name Type Description
host

TWS/Gateway hostname.
port

TWS/Gateway port (7497 paper, 7496 live).
client_id

Unique client identifier.
Source code in adapters/ibkr.py 
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
class IBKRAdapter(BaseAdapter):
    """
    Interactive Brokers adapter.

    Connects to TWS or IB Gateway via the ib_insync library.
    Supports both live and paper trading accounts.

    Attributes:
        host: TWS/Gateway hostname.
        port: TWS/Gateway port (7497 paper, 7496 live).
        client_id: Unique client identifier.
    """

    def __init__(
        self,
        host: str = "127.0.0.1",
        port: int = 7497,
        client_id: int = 1,
        **kwargs,
    ):
        """
        Initialize IBKR adapter.

        Args:
            host: TWS/Gateway hostname.
            port: TWS/Gateway port. Use 7497 for paper trading,
                  7496 for live trading.
            client_id: Unique client ID. Each connection needs
                       a different client_id.
            **kwargs: Ignored (for compatibility with config loading).
        """
        self.host = host
        self.port = int(port)
        self.client_id = int(client_id)
        self._ib = None

    @classmethod
    def get_config_fields(cls) -> list[ConfigField]:
        """
        Return configuration fields for web UI.

        Returns:
            List with host, port, and client_id fields.
        """
        return [
            {
                "name": "host",
                "label": "TWS/Gateway Host",
                "type": "text",
                "default": "127.0.0.1",
            },
            {
                "name": "port",
                "label": "Port",
                "type": "number",
                "default": 7497,
                "help": "7497 for paper trading, 7496 for live",
            },
            {
                "name": "client_id",
                "label": "Client ID",
                "type": "number",
                "default": 1,
            },
        ]

    def connect(self) -> bool:
        """
        Connect to TWS/Gateway.

        Returns:
            True if connection successful, False otherwise.
        """
        try:
            from ib_insync import IB

            self._ib = IB()
            self._ib.connect(
                host=self.host,
                port=self.port,
                clientId=self.client_id,
                readonly=False,
            )
            return self._ib.isConnected()
        except ImportError:
            print("[ERROR] ib_insync not installed. Run: pip install ib_insync")
            return False
        except Exception as e:
            print(f"[ERROR] IBKR connection failed: {e}")
            return False

    def disconnect(self) -> None:
        """Disconnect from TWS/Gateway."""
        if self._ib and self._ib.isConnected():
            self._ib.disconnect()
        self._ib = None

    def _get_contract(self, symbol: str):
        """Create a Stock contract for a symbol."""
        from ib_insync import Stock

        return Stock(symbol, "SMART", "USD")

    def execute_buy(self, symbol: str, shares: int, price: float) -> FillResult:
        """
        Execute a market buy order.

        Args:
            symbol: Stock symbol (e.g., "AAPL").
            shares: Number of shares to buy.
            price: Reference price (not used for market orders).

        Returns:
            Fill details or error dict.
        """
        return self._execute_order(symbol, "BUY", shares)

    def execute_sell(self, symbol: str, shares: int, price: float) -> FillResult:
        """
        Execute a market sell order.

        Args:
            symbol: Stock symbol (e.g., "AAPL").
            shares: Number of shares to sell.
            price: Reference price (not used for market orders).

        Returns:
            Fill details or error dict.
        """
        return self._execute_order(symbol, "SELL", shares)

    def _execute_order(self, symbol: str, action: str, shares: int) -> FillResult:
        """
        Execute a market order.

        Args:
            symbol: Stock symbol.
            action: "BUY" or "SELL".
            shares: Number of shares.

        Returns:
            Fill details or error dict.
        """
        if not self._ib or not self._ib.isConnected():
            return {"success": False, "error": "Not connected"}

        try:
            from ib_insync import MarketOrder

            contract = self._get_contract(symbol)
            order = MarketOrder(action, shares)

            trade = self._ib.placeOrder(contract, order)

            # Wait for fill (up to 30 seconds)
            timeout = 30
            while not trade.isDone() and timeout > 0:
                self._ib.sleep(1)
                timeout -= 1

            if trade.orderStatus.status == "Filled":
                return {
                    "success": True,
                    "fill_price": trade.orderStatus.avgFillPrice,
                    "fill_shares": int(trade.orderStatus.filled),
                    "commission": self._get_commission(trade),
                }
            else:
                return {
                    "success": False,
                    "error": f"Order not filled: {trade.orderStatus.status}",
                }

        except Exception as e:
            return {"success": False, "error": str(e)}

    def _get_commission(self, trade) -> float:
        """Extract commission from trade fills."""
        total = 0.0
        for fill in trade.fills:
            if fill.commissionReport:
                total += fill.commissionReport.commission or 0.0
        return total

    def fetch_quote(self, symbol: str, eod_last_price: float | None) -> Quote | None:
        """
        Fetch a real-time quote from TWS/Gateway.

        Args:
            symbol: Stock symbol (e.g., "AAPL").
            eod_last_price: EOD close from TradeTracer (unused, real data
                is fetched from the broker).

        Returns:
            Quote dict with OHLCV and bid/ask, or None if unavailable.
        """
        if not self._ib or not self._ib.isConnected():
            return None

        try:
            contract = self._get_contract(symbol)

            # Request market data snapshot
            self._ib.qualifyContracts(contract)
            ticker = self._ib.reqMktData(contract, snapshot=True)

            # Wait for data (up to 5 seconds)
            timeout = 5
            while ticker.last != ticker.last and timeout > 0:  # NaN check
                self._ib.sleep(0.5)
                timeout -= 0.5

            # Check if we have valid data
            if not _is_valid(ticker.last) and not _is_valid(ticker.bid):
                return None

            # Use broker timestamp if available, fall back to local time
            ts = int(time.time())
            if hasattr(ticker, "time") and ticker.time is not None:
                try:
                    ts = int(ticker.time.timestamp())
                except (AttributeError, TypeError):
                    pass

            return {
                "open": _safe_float(ticker.open),
                "high": _safe_float(ticker.high),
                "low": _safe_float(ticker.low),
                "close": _safe_float(ticker.last),
                "volume": _safe_int(ticker.volume),
                "bid": _safe_float(ticker.bid),
                "ask": _safe_float(ticker.ask),
                "time": ts,
            }

        except Exception as e:
            print(f"[WARN] Failed to get quote for {symbol}: {e}")
            return None

    def fetch_bars(self, symbol: str, count: int) -> list[Bar]:
        """
        Fetch historical 1-minute bars from TWS/Gateway.

        Args:
            symbol: Stock symbol (e.g., "AAPL").
            count: Maximum number of bars to return (up to 1000).

        Returns:
            List of Bar dicts ordered oldest-first, or empty list.
        """
        if not self._ib or not self._ib.isConnected():
            return []

        try:
            contract = self._get_contract(symbol)
            self._ib.qualifyContracts(contract)

            # 1000 1-min bars ≈ 2.6 trading days
            raw_bars = self._ib.reqHistoricalData(
                contract,
                endDateTime="",
                durationStr="3 D",
                barSizeSetting="1 min",
                whatToShow="TRADES",
                useRTH=True,
            )

            bars: list[Bar] = []
            for b in raw_bars[-count:]:
                ts = int(b.date.timestamp()) if hasattr(b.date, "timestamp") else int(b.date)
                bars.append({
                    "t": ts,
                    "o": _safe_float(b.open),
                    "h": _safe_float(b.high),
                    "l": _safe_float(b.low),
                    "c": _safe_float(b.close),
                    "v": _safe_int(b.volume),
                })
            return bars

        except Exception as e:
            print(f"[WARN] Failed to get bars for {symbol}: {e}")
            return []

__init__(host='127.0.0.1', port=7497, client_id=1, **kwargs)

Initialize IBKR adapter.

Parameters:

Name Type Description Default
host str

TWS/Gateway hostname.

 '127.0.0.1'
port int

TWS/Gateway port. Use 7497 for paper trading, 7496 for live trading.

 7497
client_id int

Unique client ID. Each connection needs a different client_id.

 1
**kwargs

Ignored (for compatibility with config loading).

 {}
Source code in adapters/ibkr.py 
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
def __init__(
    self,
    host: str = "127.0.0.1",
    port: int = 7497,
    client_id: int = 1,
    **kwargs,
):
    """
    Initialize IBKR adapter.

    Args:
        host: TWS/Gateway hostname.
        port: TWS/Gateway port. Use 7497 for paper trading,
              7496 for live trading.
        client_id: Unique client ID. Each connection needs
                   a different client_id.
        **kwargs: Ignored (for compatibility with config loading).
    """
    self.host = host
    self.port = int(port)
    self.client_id = int(client_id)
    self._ib = None

get_config_fields() classmethod

Return configuration fields for web UI.

Returns:

Type Description
list[ConfigField]

List with host, port, and client_id fields.
Source code in adapters/ibkr.py 
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
@classmethod
def get_config_fields(cls) -> list[ConfigField]:
    """
    Return configuration fields for web UI.

    Returns:
        List with host, port, and client_id fields.
    """
    return [
        {
            "name": "host",
            "label": "TWS/Gateway Host",
            "type": "text",
            "default": "127.0.0.1",
        },
        {
            "name": "port",
            "label": "Port",
            "type": "number",
            "default": 7497,
            "help": "7497 for paper trading, 7496 for live",
        },
        {
            "name": "client_id",
            "label": "Client ID",
            "type": "number",
            "default": 1,
        },
    ]

connect()

Connect to TWS/Gateway.

Returns:

Type Description
bool

True if connection successful, False otherwise.
Source code in adapters/ibkr.py 
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
def connect(self) -> bool:
    """
    Connect to TWS/Gateway.

    Returns:
        True if connection successful, False otherwise.
    """
    try:
        from ib_insync import IB

        self._ib = IB()
        self._ib.connect(
            host=self.host,
            port=self.port,
            clientId=self.client_id,
            readonly=False,
        )
        return self._ib.isConnected()
    except ImportError:
        print("[ERROR] ib_insync not installed. Run: pip install ib_insync")
        return False
    except Exception as e:
        print(f"[ERROR] IBKR connection failed: {e}")
        return False

disconnect()

Disconnect from TWS/Gateway.

Source code in adapters/ibkr.py 
131
132
133
134
135
def disconnect(self) -> None:
    """Disconnect from TWS/Gateway."""
    if self._ib and self._ib.isConnected():
        self._ib.disconnect()
    self._ib = None

execute_buy(symbol, shares, price)

Execute a market buy order.

Parameters:

Name Type Description Default
symbol str

Stock symbol (e.g., "AAPL").

 required
shares int

Number of shares to buy.

 required
price float

Reference price (not used for market orders).

 required

Returns:

Type Description
FillResult

Fill details or error dict.
Source code in adapters/ibkr.py 
143
144
145
146
147
148
149
150
151
152
153
154
155
def execute_buy(self, symbol: str, shares: int, price: float) -> FillResult:
    """
    Execute a market buy order.

    Args:
        symbol: Stock symbol (e.g., "AAPL").
        shares: Number of shares to buy.
        price: Reference price (not used for market orders).

    Returns:
        Fill details or error dict.
    """
    return self._execute_order(symbol, "BUY", shares)

execute_sell(symbol, shares, price)

Execute a market sell order.

Parameters:

Name Type Description Default
symbol str

Stock symbol (e.g., "AAPL").

 required
shares int

Number of shares to sell.

 required
price float

Reference price (not used for market orders).

 required

Returns:

Type Description
FillResult

Fill details or error dict.
Source code in adapters/ibkr.py 
157
158
159
160
161
162
163
164
165
166
167
168
169
def execute_sell(self, symbol: str, shares: int, price: float) -> FillResult:
    """
    Execute a market sell order.

    Args:
        symbol: Stock symbol (e.g., "AAPL").
        shares: Number of shares to sell.
        price: Reference price (not used for market orders).

    Returns:
        Fill details or error dict.
    """
    return self._execute_order(symbol, "SELL", shares)

fetch_quote(symbol, eod_last_price)

Fetch a real-time quote from TWS/Gateway.

Parameters:

Name Type Description Default
symbol str

Stock symbol (e.g., "AAPL").

 required
eod_last_price float | None

EOD close from TradeTracer (unused, real data is fetched from the broker).

 required

Returns:

Type Description
Quote | None

Quote dict with OHLCV and bid/ask, or None if unavailable.
Source code in adapters/ibkr.py 
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
def fetch_quote(self, symbol: str, eod_last_price: float | None) -> Quote | None:
    """
    Fetch a real-time quote from TWS/Gateway.

    Args:
        symbol: Stock symbol (e.g., "AAPL").
        eod_last_price: EOD close from TradeTracer (unused, real data
            is fetched from the broker).

    Returns:
        Quote dict with OHLCV and bid/ask, or None if unavailable.
    """
    if not self._ib or not self._ib.isConnected():
        return None

    try:
        contract = self._get_contract(symbol)

        # Request market data snapshot
        self._ib.qualifyContracts(contract)
        ticker = self._ib.reqMktData(contract, snapshot=True)

        # Wait for data (up to 5 seconds)
        timeout = 5
        while ticker.last != ticker.last and timeout > 0:  # NaN check
            self._ib.sleep(0.5)
            timeout -= 0.5

        # Check if we have valid data
        if not _is_valid(ticker.last) and not _is_valid(ticker.bid):
            return None

        # Use broker timestamp if available, fall back to local time
        ts = int(time.time())
        if hasattr(ticker, "time") and ticker.time is not None:
            try:
                ts = int(ticker.time.timestamp())
            except (AttributeError, TypeError):
                pass

        return {
            "open": _safe_float(ticker.open),
            "high": _safe_float(ticker.high),
            "low": _safe_float(ticker.low),
            "close": _safe_float(ticker.last),
            "volume": _safe_int(ticker.volume),
            "bid": _safe_float(ticker.bid),
            "ask": _safe_float(ticker.ask),
            "time": ts,
        }

    except Exception as e:
        print(f"[WARN] Failed to get quote for {symbol}: {e}")
        return None

fetch_bars(symbol, count)

Fetch historical 1-minute bars from TWS/Gateway.

Parameters:

Name Type Description Default
symbol str

Stock symbol (e.g., "AAPL").

 required
count int

Maximum number of bars to return (up to 1000).

 required

Returns:

Type Description
list[Bar]

List of Bar dicts ordered oldest-first, or empty list.
Source code in adapters/ibkr.py 
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
def fetch_bars(self, symbol: str, count: int) -> list[Bar]:
    """
    Fetch historical 1-minute bars from TWS/Gateway.

    Args:
        symbol: Stock symbol (e.g., "AAPL").
        count: Maximum number of bars to return (up to 1000).

    Returns:
        List of Bar dicts ordered oldest-first, or empty list.
    """
    if not self._ib or not self._ib.isConnected():
        return []

    try:
        contract = self._get_contract(symbol)
        self._ib.qualifyContracts(contract)

        # 1000 1-min bars ≈ 2.6 trading days
        raw_bars = self._ib.reqHistoricalData(
            contract,
            endDateTime="",
            durationStr="3 D",
            barSizeSetting="1 min",
            whatToShow="TRADES",
            useRTH=True,
        )

        bars: list[Bar] = []
        for b in raw_bars[-count:]:
            ts = int(b.date.timestamp()) if hasattr(b.date, "timestamp") else int(b.date)
            bars.append({
                "t": ts,
                "o": _safe_float(b.open),
                "h": _safe_float(b.high),
                "l": _safe_float(b.low),
                "c": _safe_float(b.close),
                "v": _safe_int(b.volume),
            })
        return bars

    except Exception as e:
        print(f"[WARN] Failed to get bars for {symbol}: {e}")
        return []