From 0e2ff8ea330ca889451f81b513203babb4dde3c8 Mon Sep 17 00:00:00 2001 From: Johnson Tam Date: Wed, 12 Aug 2026 16:16:50 +0800 Subject: [PATCH 1/3] fix: keep IBKR sessions usable across API workers and threads `ib_insync` binds a connection to both the process that opened it (TWS and IB Gateway allow one session per client id) and the asyncio event loop of the thread that created it. The API served IBKR from gunicorn workers that shared neither: each process kept its own in-memory session map and defaulted to client id 1, and each gthread request thread created a fresh event loop. A second worker's connect was therefore answered with "Error 326: client id is already in use" followed by a 20s TimeoutError, and requests alternated between a connected and a disconnected view of the same account as they were balanced across workers. Calls that did reach a live client from another thread ran against an idle loop. Alpaca is unaffected because it is a stateless REST broker with no session and no client id. - `ibkr_trading/session.py`: pin each connection to a dedicated thread whose event loop stays alive, and marshal every call onto it, which also serialises access to the socket. A process-wide registry keyed by (host, port, client_id) reuses the live session so the process cannot answer its own connect with Error 326. - `ibkr_trading/config.py`: resolve connection settings from the stored credential instead of querying credential tables from the route, per MODULE_BOUNDARIES. The UI's 127.0.0.1 placeholder is unreachable from a container, so it falls back to the saved host; an explicit host or port is always honoured, and a blank client id no longer raises. The credential's `ibkr_client_id` belongs to the order session (default 7), so the UI session never borrows it. - `routes/ibkr.py`: reconnect on demand from the stored credential, so every endpoint works regardless of which worker serves it. A failed connect from inside a container now explains the loopback trap instead of only suggesting that TWS may be down. - `ibkr_trading/account.py`: `IB.accountSummary()` returns tag rows whose values are strings paired with a currency, but the broker-account UI reads flat numeric fields -- the shape the Alpaca route already returns -- so the account panel rendered every metric as "--" while showing the account id. Expose those flat fields alongside the existing `summary` map, which stays for agent/MCP callers. Missing tags are omitted rather than zeroed, since a margin requirement of 0 reads very differently from "unknown", and a failed summary read no longer reports success with an empty payload. - `live_trading/account_snapshot.py`: read IBKR over one short-lived session instead of the crypto path's swap+spot pair (which collides on the same client id), and serialise snapshots so concurrent refreshes cannot answer each other with Error 326. How to test: - `pytest tests/test_ibkr_session_and_account.py` covers session reuse, thread affinity, failed-connect cleanup, credential merge precedence, and summary flattening. - Verified end to end against a live IB Gateway paper account: connect, repeated connect, account, positions, orders, status and disconnect all succeed, two concurrent account snapshots return without warnings, and no Error 326 remains in the logs. The account panel renders net liquidation, cash, buying power and both margin requirements. Backward compatibility: response fields are additive -- `summary` and `account` are unchanged. `GET /api/ibkr/account` now returns 400 instead of 200 when the summary read fails; the OpenAPI contract already documents 400 for this operation, and regenerating the spec produces no IBKR diff. Co-Authored-By: Claude Opus 5 (1M context) --- backend_api_python/app/routes/ibkr.py | 88 ++++--- .../app/services/ibkr_trading/account.py | 82 +++++++ .../app/services/ibkr_trading/config.py | 112 +++++++++ .../app/services/ibkr_trading/session.py | 153 +++++++++++++ .../services/live_trading/account_snapshot.py | 65 ++++++ .../tests/test_ibkr_session_and_account.py | 216 ++++++++++++++++++ 6 files changed, 689 insertions(+), 27 deletions(-) create mode 100644 backend_api_python/app/services/ibkr_trading/account.py create mode 100644 backend_api_python/app/services/ibkr_trading/config.py create mode 100644 backend_api_python/app/services/ibkr_trading/session.py create mode 100644 backend_api_python/tests/test_ibkr_session_and_account.py diff --git a/backend_api_python/app/routes/ibkr.py b/backend_api_python/app/routes/ibkr.py index d723e9c66..ea67eae3b 100644 --- a/backend_api_python/app/routes/ibkr.py +++ b/backend_api_python/app/routes/ibkr.py @@ -8,13 +8,17 @@ through someone else's IBKR/TWS account. """ -from flask import jsonify, request +import os + +from flask import jsonify, request, g from app.openapi.blueprint import HumanBlueprint as Blueprint from app.utils.auth import login_required from app.utils.logger import get_logger from app.utils.broker_session import BrokerSessionRegistry -from app.services.ibkr_trading import IBKRClient, IBKRConfig +from app.services.ibkr_trading.account import flatten_account_summary +from app.services.ibkr_trading.config import build_ibkr_config +from app.services.ibkr_trading.session import get_or_create_session logger = get_logger(__name__) @@ -35,10 +39,40 @@ def _placeholder_status(): } +def _connect_error_message(config) -> str: + """Explain a failed connect, including the containerised-loopback trap.""" + target = f"{config.host}:{config.port}" + if config.host in ("127.0.0.1", "localhost", "::1") and os.path.exists("/.dockerenv"): + return ( + f"Cannot reach TWS/Gateway at {target}. The backend runs in a container, " + "where 127.0.0.1 is the container itself. Use the host address " + "(host.docker.internal on Docker Desktop) and save it on the IBKR credential." + ) + return ( + f"Cannot reach TWS/Gateway at {target}. Check that it is running and that " + "its API port accepts connections from this host." + ) + + def _require_connected_client(): + """Return a live session, reconnecting from stored credentials if needed. + + The session lives in process memory, so a request served by a different + worker than the one that handled ``POST /connect`` would otherwise report + "Not connected". Reconnecting on demand keeps every endpoint usable + regardless of which worker picks it up. + """ client = _sessions.get() - if client is None or not client.connected: + if client is not None and client.connected: + return client, None + + try: + client = get_or_create_session(build_ibkr_config({}, user_id=g.user_id)) + except Exception as exc: + logger.info(f"IBKR auto-connect failed: {exc}") return None, (jsonify({"success": False, "error": "Not connected to IBKR"}), 400) + + _sessions.set(client) return client, None @@ -79,31 +113,24 @@ def connect(): """ try: data = request.get_json() or {} + config = build_ibkr_config(data, user_id=g.user_id) - config = IBKRConfig( - host=data.get('host', '127.0.0.1'), - port=int(data.get('port', 7497)), - client_id=int(data.get('clientId', 1)), - account=data.get('account', ''), - readonly=data.get('readonly', False), - ) - - client = IBKRClient(config) - success = client.connect() - - if success: - _sessions.set(client) - return jsonify({ - "success": True, - "message": "Connected successfully", - "data": client.get_connection_status() - }) - else: + try: + client = get_or_create_session(config) + except ConnectionError as exc: + logger.warning(f"IBKR connect rejected: {exc}") return jsonify({ "success": False, - "error": "Connection failed. Please check if TWS/Gateway is running." + "error": _connect_error_message(config), }), 400 + _sessions.set(client) + return jsonify({ + "success": True, + "message": "Connected successfully", + "data": client.get_connection_status() + }) + except ImportError: return jsonify({ "success": False, @@ -146,10 +173,17 @@ def get_account(): if err is not None: return err - return jsonify({ - "success": True, - "data": client.get_account_summary() - }) + data = client.get_account_summary() + if not isinstance(data, dict) or not data.get("success"): + error = (data or {}).get("error") if isinstance(data, dict) else "" + return jsonify({ + "success": False, + "error": error or "Failed to read the IBKR account summary", + }), 400 + + # The UI reads flat numeric fields; ``summary`` stays for agent/MCP callers. + data.update(flatten_account_summary(data.get("summary"))) + return jsonify({"success": True, "data": data}) except Exception as e: logger.error(f"Get account info failed: {e}") return jsonify({ diff --git a/backend_api_python/app/services/ibkr_trading/account.py b/backend_api_python/app/services/ibkr_trading/account.py new file mode 100644 index 000000000..10933eed2 --- /dev/null +++ b/backend_api_python/app/services/ibkr_trading/account.py @@ -0,0 +1,82 @@ +"""Normalization of IBKR account summary tags. + +``IB.accountSummary()`` returns tag rows (``NetLiquidation``, ``BuyingPower``, +...) whose values are strings paired with a currency. The broker-account UI +reads flat numeric fields off the account payload -- the same shape the Alpaca +route already returns -- so the nested tag map alone renders as "--". + +This module maps the tags the UI needs onto flat fields. The original +``summary`` map stays in the response for agent/MCP consumers. +""" + +from __future__ import annotations + +from typing import Any, Dict, Mapping, Optional, Tuple + +# Flat field -> IB tags, in priority order. +_FLAT_FIELDS: Tuple[Tuple[str, Tuple[str, ...]], ...] = ( + ("net_liquidation", ("NetLiquidation",)), + ("total_cash_value", ("TotalCashValue",)), + ("buying_power", ("BuyingPower",)), + ("init_margin_req", ("InitMarginReq", "FullInitMarginReq")), + ("maint_margin_req", ("MaintMarginReq", "FullMaintMarginReq")), + ("available_funds", ("AvailableFunds", "FullAvailableFunds")), + ("excess_liquidity", ("ExcessLiquidity", "FullExcessLiquidity")), + ("equity_with_loan", ("EquityWithLoanValue",)), + ("gross_position_value", ("GrossPositionValue",)), +) + +# Currency is read from a monetary tag; descriptive tags carry an empty one. +_CURRENCY_TAGS: Tuple[str, ...] = ("NetLiquidation", "TotalCashValue", "BuyingPower") + + +def _tag_value(summary: Mapping[str, Any], tag: str) -> Optional[float]: + row = summary.get(tag) + raw = row.get("value") if isinstance(row, Mapping) else row + if raw is None or (isinstance(raw, str) and not raw.strip()): + return None + try: + return float(raw) + except (TypeError, ValueError): + return None + + +def _tag_currency(summary: Mapping[str, Any], tag: str) -> str: + row = summary.get(tag) + if not isinstance(row, Mapping): + return "" + return str(row.get("currency") or "").strip() + + +def flatten_account_summary(summary: Any) -> Dict[str, Any]: + """Return the flat numeric fields the broker-account UI reads. + + Tags that IBKR did not report are omitted rather than zeroed, so the UI + shows "--" for genuinely missing values instead of a misleading 0. + """ + if not isinstance(summary, Mapping): + return {} + + flat: Dict[str, Any] = {} + for field, tags in _FLAT_FIELDS: + for tag in tags: + value = _tag_value(summary, tag) + if value is not None: + flat[field] = value + break + + if not flat: + # Currency belongs to an amount; on its own it tells the caller nothing. + return flat + + for tag in _CURRENCY_TAGS: + currency = _tag_currency(summary, tag) + if currency: + flat["currency"] = currency + flat["account_currency"] = currency + break + + return flat + + +__all__ = ["flatten_account_summary"] diff --git a/backend_api_python/app/services/ibkr_trading/config.py b/backend_api_python/app/services/ibkr_trading/config.py new file mode 100644 index 000000000..605c414a3 --- /dev/null +++ b/backend_api_python/app/services/ibkr_trading/config.py @@ -0,0 +1,112 @@ +"""IBKR connection settings resolved from stored exchange credentials. + +The HTTP layer must not query credential tables directly (see +``docs/architecture/MODULE_BOUNDARIES.md``). This module owns that lookup and +returns a ready-to-use :class:`IBKRConfig`. + +The UI sends ``127.0.0.1`` by default, which inside a container points at the +container itself and can never reach a TWS/IB Gateway on the operator's desktop. +The saved credential therefore wins whenever the request carries no explicit +host, so a single click on "Connect" works from a Docker deployment. +""" + +from __future__ import annotations + +from typing import Any, Dict, Optional + +from app.services.ibkr_trading.client import IBKRConfig +from app.utils.db import get_db_connection +from app.utils.logger import get_logger + +logger = get_logger(__name__) + +DEFAULT_HOST = "127.0.0.1" +DEFAULT_PORT = 7497 +DEFAULT_CLIENT_ID = 1 + + +def load_saved_ibkr_config(user_id: int) -> Dict[str, Any]: + """Return the user's most recent stored IBKR credential settings. + + Returns an empty dict when the user has no IBKR credential or the lookup + fails; callers fall back to request values and defaults. + """ + try: + with get_db_connection() as db: + cursor = db.cursor() + cursor.execute( + "SELECT id FROM qd_exchange_credentials " + "WHERE user_id = %s AND exchange_id = 'ibkr' " + "ORDER BY id DESC LIMIT 1", + (int(user_id),), + ) + row = cursor.fetchone() + cursor.close() + if not row or not row.get("id"): + return {} + + from app.services.exchange_execution import resolve_exchange_config + + return resolve_exchange_config( + {"credential_id": int(row["id"])}, user_id=int(user_id) + ) or {} + except Exception as exc: + logger.warning("Failed to load saved IBKR credential: %s", exc) + return {} + + +def _int_or_none(value: Any) -> Optional[int]: + if value is None or (isinstance(value, str) and not value.strip()): + return None + try: + return int(value) + except (TypeError, ValueError): + return None + + +def _text_or_none(value: Any) -> Optional[str]: + if value is None: + return None + text = str(value).strip() + return text or None + + +def build_ibkr_config(request_data: Dict[str, Any], *, user_id: int) -> IBKRConfig: + """Merge request values over the stored credential into an IBKRConfig. + + Explicit request values always win. Anything the request omits (or leaves + at the UI's ``127.0.0.1`` placeholder host) comes from the stored + credential, then from the IB defaults. + """ + data = request_data if isinstance(request_data, dict) else {} + + host = _text_or_none(data.get("host")) + port = _int_or_none(data.get("port")) + client_id = _int_or_none(data.get("clientId")) + account = _text_or_none(data.get("account")) + + # A localhost host is treated as "unset": it is the UI placeholder and is + # unreachable from inside a container. + if host in ("127.0.0.1", "localhost", "::1"): + host = None + + # ``ibkr_client_id`` on the credential belongs to the strategy/order session + # (default 7). Borrowing it here would make the UI evict live orders, so the + # UI session only ever uses the request value or DEFAULT_CLIENT_ID. + if host is None or port is None or account is None: + saved = load_saved_ibkr_config(user_id) + if saved: + host = host or _text_or_none(saved.get("ibkr_host")) + port = port or _int_or_none(saved.get("ibkr_port")) + account = account or _text_or_none(saved.get("ibkr_account")) + + return IBKRConfig( + host=host or DEFAULT_HOST, + port=port or DEFAULT_PORT, + client_id=client_id if client_id is not None else DEFAULT_CLIENT_ID, + account=account or "", + readonly=bool(data.get("readonly", False)), + ) + + +__all__ = ["build_ibkr_config", "load_saved_ibkr_config"] diff --git a/backend_api_python/app/services/ibkr_trading/session.py b/backend_api_python/app/services/ibkr_trading/session.py new file mode 100644 index 000000000..b5956982d --- /dev/null +++ b/backend_api_python/app/services/ibkr_trading/session.py @@ -0,0 +1,153 @@ +"""Thread-affine IBKR session ownership. + +``ib_insync`` binds a connection to the asyncio event loop of the thread that +created it. The HTTP API serves requests from a pool of gthread workers, so a +client connected on request thread A cannot be driven from request thread B: +the call runs against a different (idle) loop and blocks until it times out. + +:class:`IBKRSession` therefore pins one :class:`~app.services.ibkr_trading.client.IBKRClient` +to a dedicated single-thread executor whose thread owns a permanent event loop. +Every call is marshalled onto that thread, which also serialises access so two +concurrent requests cannot interleave on the same socket. + +A process-wide registry keyed by ``(host, port, client_id)`` guarantees the +process never competes with itself for a client id -- TWS/IB Gateway allows one +session per client id and answers a second one with +``Error 326: client id is already in use``. +""" + +from __future__ import annotations + +import asyncio +import threading +from concurrent.futures import ThreadPoolExecutor +from typing import Any, Callable, Dict, Optional, Tuple + +from app.services.ibkr_trading.client import IBKRClient, IBKRConfig +from app.utils.logger import get_logger + +logger = get_logger(__name__) + +# Attributes that only read local state and are safe to serve inline. +_DIRECT_ATTRS = frozenset({"config", "connected", "get_connection_status"}) + +# Upper bound for a marshalled IBKR call. IBKRConfig.timeout (20s) covers the +# connect handshake; the rest of the calls are request/response round trips. +_CALL_TIMEOUT_SEC = 60.0 + +_registry: Dict[Tuple[str, int, int], "IBKRSession"] = {} +_registry_lock = threading.RLock() + + +def _session_key(config: IBKRConfig) -> Tuple[str, int, int]: + return (str(config.host).strip(), int(config.port), int(config.client_id)) + + +class IBKRSession: + """An :class:`IBKRClient` pinned to its own thread and event loop. + + Attribute access is proxied to the wrapped client: plain state reads are + served inline, every method call is submitted to the owning thread. + """ + + def __init__(self, config: IBKRConfig): + self._config = config + self._key = _session_key(config) + self._client = IBKRClient(config) + self._executor = ThreadPoolExecutor( + max_workers=1, + thread_name_prefix=f"ibkr-{self._key[0]}-{self._key[1]}-{self._key[2]}", + ) + self._executor.submit(_install_event_loop).result(timeout=10) + + # -- lifecycle ------------------------------------------------------- + + def connect(self) -> bool: + return self._submit(self._client.connect) + + def disconnect(self) -> None: + try: + self._submit(self._client.disconnect) + finally: + self._executor.shutdown(wait=False) + _forget(self._key, self) + + @property + def connected(self) -> bool: + return self._client.connected + + # -- proxy ----------------------------------------------------------- + + def __getattr__(self, name: str) -> Any: + # Only reached for attributes not defined on IBKRSession itself. + # Private names are never proxied, so a half-built instance raises + # AttributeError instead of recursing on ``self._client``. + if name.startswith("_"): + raise AttributeError(name) + attr = getattr(self._client, name) + if name in _DIRECT_ATTRS or not callable(attr): + return attr + + def _marshalled(*args: Any, **kwargs: Any) -> Any: + return self._submit(lambda: attr(*args, **kwargs)) + + _marshalled.__name__ = name + return _marshalled + + def _submit(self, fn: Callable[[], Any]) -> Any: + return self._executor.submit(fn).result(timeout=_CALL_TIMEOUT_SEC) + + +def _install_event_loop() -> None: + """Give the session thread a permanent event loop for ib_insync.""" + asyncio.set_event_loop(asyncio.new_event_loop()) + + +def _forget(key: Tuple[str, int, int], session: "IBKRSession") -> None: + with _registry_lock: + if _registry.get(key) is session: + _registry.pop(key, None) + + +def get_or_create_session(config: IBKRConfig) -> IBKRSession: + """Return the process-wide session for ``config``, connecting if needed. + + Re-using the live session for a ``(host, port, client_id)`` triple is what + keeps the process from answering its own connection with Error 326. A dead + session is replaced. + """ + key = _session_key(config) + with _registry_lock: + existing = _registry.get(key) + if existing is not None: + if existing.connected: + return existing + try: + existing.disconnect() + except Exception as exc: # pragma: no cover - defensive + logger.debug("Stale IBKR session disconnect raised: %s", exc) + _registry.pop(key, None) + + session = IBKRSession(config) + _registry[key] = session + + if not session.connect(): + _forget(key, session) + session.disconnect() + raise ConnectionError( + "Failed to connect to IBKR TWS/Gateway at " + f"{config.host}:{config.port} (clientId={config.client_id})." + ) + return session + + +def find_session(config: IBKRConfig) -> Optional[IBKRSession]: + """Return the live session for ``config`` without connecting.""" + with _registry_lock: + session = _registry.get(_session_key(config)) + if session is not None and session.connected: + return session + return None + + +__all__ = ["IBKRSession", "get_or_create_session", "find_session"] diff --git a/backend_api_python/app/services/live_trading/account_snapshot.py b/backend_api_python/app/services/live_trading/account_snapshot.py index 21ed98d58..32bbb2290 100644 --- a/backend_api_python/app/services/live_trading/account_snapshot.py +++ b/backend_api_python/app/services/live_trading/account_snapshot.py @@ -5,6 +5,7 @@ from __future__ import annotations +import threading import time from typing import Any, Dict, List, Optional, Tuple @@ -16,6 +17,10 @@ logger = get_logger(__name__) +# TWS/IB Gateway allows one session per client id; two snapshots running at the +# same time would answer each other with "Error 326: client id already in use". +_IBKR_SNAPSHOT_LOCK = threading.Lock() + def _user_facing_exchange_error(exc: Exception, *, context: str) -> str: """Map raw exchange exception to a short UI message.""" @@ -480,6 +485,62 @@ def _fetch_binance_snapshot( return swap_pos, spot_pos, orders +def _fetch_ibkr_snapshot( + exchange_config: Dict[str, Any], + errors: List[str], +) -> Tuple[List[Dict[str, Any]], List[Dict[str, Any]]]: + """Read IBKR positions and open orders over a single short-lived session. + + IBKR allows one session per client id, so the crypto path (which opens a + swap and a spot client with the same credentials) would answer itself with + "Error 326: client id already in use" and leak the losing session. The lock + keeps concurrent snapshot requests from doing the same to each other. + """ + positions_out: List[Dict[str, Any]] = [] + orders_out: List[Dict[str, Any]] = [] + with _IBKR_SNAPSHOT_LOCK: + client = None + try: + client = create_client(exchange_config, market_type="spot") + for p in client.get_positions() or []: + try: + qty = float(p.get("quantity") or 0.0) + except (TypeError, ValueError): + qty = 0.0 + if abs(qty) <= 0: + continue + positions_out.append({ + "symbol": str(p.get("symbol") or p.get("ib_symbol") or "").strip(), + "side": "long" if qty > 0 else "short", + "size": abs(qty), + "entry_price": float(p.get("avgCost") or 0.0), + "market_type": "spot", + "inst_id": str(p.get("ib_symbol") or "").strip(), + }) + for o in client.get_open_orders() or []: + lmt = o.get("limitPrice") + try: + lmt_f = float(lmt) if lmt not in (None, "") else None + except (TypeError, ValueError): + lmt_f = None + orders_out.append({ + "symbol": str(o.get("symbol") or "").strip(), + "side": str(o.get("action") or "").strip().lower(), + "price": lmt_f, + "size": float(o.get("quantity") or 0.0), + "status": str(o.get("status") or "").strip(), + }) + except Exception as e: + _append_snapshot_error(errors, e, context="IBKR 账户连接") + finally: + if client is not None: + try: + client.disconnect() + except Exception: + pass + return positions_out, orders_out + + def fetch_account_snapshot(*, user_id: int, credential_id: int) -> Dict[str, Any]: """Live fetch swap/spot legs + open orders for one credential.""" cred = int(credential_id or 0) @@ -540,6 +601,10 @@ def fetch_account_snapshot(*, user_id: int, credential_id: int) -> Dict[str, Any swap_all.extend(sp) spot_all.extend(st) orders_all.extend(od) + elif exchange_id == "ibkr": + st, od = _fetch_ibkr_snapshot(exchange_config, errors) + spot_all.extend(st) + orders_all.extend(od) else: try: client = create_client(exchange_config, market_type="swap") diff --git a/backend_api_python/tests/test_ibkr_session_and_account.py b/backend_api_python/tests/test_ibkr_session_and_account.py new file mode 100644 index 000000000..04ea9d997 --- /dev/null +++ b/backend_api_python/tests/test_ibkr_session_and_account.py @@ -0,0 +1,216 @@ +"""Regression tests for IBKR session ownership and account payload shape. + +Two failures motivated these: + +* every process defaulted to client id 1 and kept its session in local memory, + so a second API worker's connect was answered with + "Error 326: client id already in use"; +* ``ib_insync`` binds a connection to the event loop of the creating thread, so + a call issued from another request thread ran against an idle loop. +""" + +import threading +from typing import Any, Dict + +import pytest + +from app.services.ibkr_trading import config as ibkr_config +from app.services.ibkr_trading import session as ibkr_session +from app.services.ibkr_trading.account import flatten_account_summary +from app.services.ibkr_trading.client import IBKRConfig +from app.services.ibkr_trading.config import build_ibkr_config + + +class _FakeIBKRClient: + """Stand-in that records which thread each call ran on.""" + + connect_ok = True + + def __init__(self, config: IBKRConfig): + self.config = config + self._connected = False + self.call_threads: set = set() + + def connect(self) -> bool: + self.call_threads.add(threading.current_thread().name) + self._connected = _FakeIBKRClient.connect_ok + return self._connected + + def disconnect(self) -> None: + self._connected = False + + @property + def connected(self) -> bool: + return self._connected + + def get_positions(self): + self.call_threads.add(threading.current_thread().name) + return [{"symbol": "AAPL"}] + + +@pytest.fixture +def fake_ib(monkeypatch): + _FakeIBKRClient.connect_ok = True + monkeypatch.setattr(ibkr_session, "IBKRClient", _FakeIBKRClient) + monkeypatch.setattr(ibkr_session, "_registry", {}) + return _FakeIBKRClient + + +def _config(client_id: int = 1) -> IBKRConfig: + return IBKRConfig(host="gateway.test", port=4002, client_id=client_id) + + +def test_second_acquire_reuses_the_live_session(fake_ib): + """A repeated connect must not open a second socket on the same client id.""" + first = ibkr_session.get_or_create_session(_config()) + second = ibkr_session.get_or_create_session(_config()) + + assert second is first + assert first.connected is True + + +def test_a_different_client_id_gets_its_own_session(fake_ib): + ui_session = ibkr_session.get_or_create_session(_config(client_id=1)) + order_session = ibkr_session.get_or_create_session(_config(client_id=7)) + + assert order_session is not ui_session + + +def test_calls_run_on_the_session_thread_not_the_caller(fake_ib): + """Every IB call must be marshalled onto the thread that owns the loop.""" + session = ibkr_session.get_or_create_session(_config()) + caller_threads = set() + + def issue_call(): + caller_threads.add(threading.current_thread().name) + session.get_positions() + + threads = [threading.Thread(target=issue_call, name=f"request-{i}") for i in range(4)] + for thread in threads: + thread.start() + for thread in threads: + thread.join(timeout=10) + assert not thread.is_alive() + + used = session._client.call_threads + assert len(used) == 1, f"IB calls leaked across threads: {used}" + assert not (used & caller_threads) + + +def test_failed_connect_raises_and_leaves_no_registry_entry(fake_ib): + fake_ib.connect_ok = False + + with pytest.raises(ConnectionError): + ibkr_session.get_or_create_session(_config()) + + assert ibkr_session.find_session(_config()) is None + + +def test_dead_session_is_replaced(fake_ib): + stale = ibkr_session.get_or_create_session(_config()) + stale._client._connected = False + + assert ibkr_session.find_session(_config()) is None + assert ibkr_session.get_or_create_session(_config()) is not stale + + +SAVED_CREDENTIAL: Dict[str, Any] = { + "ibkr_host": "host.docker.internal", + "ibkr_port": 4002, + "ibkr_client_id": 7, + "ibkr_account": "DU1234567", +} + + +@pytest.fixture +def saved_credential(monkeypatch): + monkeypatch.setattr( + ibkr_config, "load_saved_ibkr_config", lambda user_id: dict(SAVED_CREDENTIAL) + ) + + +def test_ui_placeholder_host_falls_back_to_the_saved_host(saved_credential): + """127.0.0.1 is the UI default and is unreachable from inside a container.""" + resolved = build_ibkr_config({"host": "127.0.0.1", "port": 4002, "clientId": 1}, user_id=1) + + assert resolved.host == "host.docker.internal" + assert resolved.account == "DU1234567" + + +def test_explicit_host_and_default_looking_port_are_honoured(saved_credential): + resolved = build_ibkr_config({"host": "192.168.1.9", "port": 7497, "clientId": 3}, user_id=1) + + assert (resolved.host, resolved.port, resolved.client_id) == ("192.168.1.9", 7497, 3) + + +def test_blank_client_id_falls_back_instead_of_raising(saved_credential): + resolved = build_ibkr_config({"host": "", "port": "", "clientId": ""}, user_id=1) + + assert resolved.client_id == ibkr_config.DEFAULT_CLIENT_ID + assert (resolved.host, resolved.port) == ("host.docker.internal", 4002) + + +def test_ui_session_never_borrows_the_credential_order_client_id(saved_credential): + """ibkr_client_id belongs to the order session; sharing it evicts live orders.""" + resolved = build_ibkr_config({}, user_id=1) + + assert resolved.client_id == ibkr_config.DEFAULT_CLIENT_ID + assert resolved.client_id != SAVED_CREDENTIAL["ibkr_client_id"] + + +def test_missing_credential_falls_back_to_ib_defaults(monkeypatch): + monkeypatch.setattr(ibkr_config, "load_saved_ibkr_config", lambda user_id: {}) + + resolved = build_ibkr_config({}, user_id=1) + + assert (resolved.host, resolved.port, resolved.client_id) == ( + ibkr_config.DEFAULT_HOST, + ibkr_config.DEFAULT_PORT, + ibkr_config.DEFAULT_CLIENT_ID, + ) + + +def test_flatten_exposes_the_fields_the_account_ui_reads(): + summary = { + "AccountType": {"value": "INDIVIDUAL", "currency": ""}, + "NetLiquidation": {"value": "179817.69", "currency": "HKD"}, + "TotalCashValue": {"value": "181397.32", "currency": "HKD"}, + "BuyingPower": {"value": "627661.25", "currency": "HKD"}, + "InitMarginReq": {"value": "85668.50", "currency": "HKD"}, + "MaintMarginReq": {"value": "73394.64", "currency": "HKD"}, + } + + flat = flatten_account_summary(summary) + + assert flat["net_liquidation"] == pytest.approx(179817.69) + assert flat["total_cash_value"] == pytest.approx(181397.32) + assert flat["buying_power"] == pytest.approx(627661.25) + assert flat["init_margin_req"] == pytest.approx(85668.50) + assert flat["maint_margin_req"] == pytest.approx(73394.64) + # Descriptive tags carry no currency, so it must come from a monetary tag. + assert flat["currency"] == "HKD" + assert flat["account_currency"] == "HKD" + + +def test_flatten_falls_back_to_the_full_margin_tags(): + flat = flatten_account_summary( + {"FullInitMarginReq": {"value": "10", "currency": "USD"}, + "FullMaintMarginReq": {"value": "5", "currency": "USD"}} + ) + + assert flat["init_margin_req"] == pytest.approx(10) + assert flat["maint_margin_req"] == pytest.approx(5) + + +def test_flatten_omits_missing_tags_rather_than_reporting_zero(): + """A margin requirement shown as 0 reads very differently from "unknown".""" + flat = flatten_account_summary({"NetLiquidation": {"value": "100", "currency": "USD"}}) + + assert flat["net_liquidation"] == pytest.approx(100) + assert "maint_margin_req" not in flat + assert "buying_power" not in flat + + +def test_flatten_ignores_unusable_input(): + assert flatten_account_summary(None) == {} + assert flatten_account_summary({"NetLiquidation": {"value": "", "currency": "USD"}}) == {} From 8adc892fa2d7fc2feeee03a6a8520094951b6d5f Mon Sep 17 00:00:00 2001 From: Johnson Tam Date: Wed, 12 Aug 2026 16:38:18 +0800 Subject: [PATCH 2/3] fix: report IBKR futures entry price and stop order trigger correctly Verified against a live micro gold future (MGCV6, multiplier 10) held with a protective stop. `Position.avgCost` is reported per contract, i.e. already multiplied by the contract multiplier, so using it as the entry price showed a 4,433.30 fill as 44,332.96. Report `avgPrice` per unit alongside the unchanged `avgCost`, and use it for the account snapshot's entry price. Stocks have a multiplier of 1 and are unaffected. A stop order carries its trigger in `auxPrice` and leaves `lmtPrice` at 0, so a protective stop rendered with no price at all. Report `stopPrice` and a `price` field holding whichever of the two governs the order. Futures also share a symbol across expiries, so carry `localSymbol` and `lastTradeDate` and prefer the former when naming a position or order. Expose the order id under the `id` key order tables key rows on, so an open order can be identified and cancelled. Co-Authored-By: Claude Opus 5 (1M context) --- .../app/services/ibkr_trading/client.py | 46 +++++++++- .../services/live_trading/account_snapshot.py | 26 ++++-- .../tests/test_ibkr_session_and_account.py | 87 ++++++++++++++++++- 3 files changed, 147 insertions(+), 12 deletions(-) diff --git a/backend_api_python/app/services/ibkr_trading/client.py b/backend_api_python/app/services/ibkr_trading/client.py index e7fa2994a..a5963732c 100644 --- a/backend_api_python/app/services/ibkr_trading/client.py +++ b/backend_api_python/app/services/ibkr_trading/client.py @@ -52,6 +52,24 @@ def _ensure_ib_insync(): return ib_insync +def _contract_multiplier(contract) -> float: + """Contract multiplier, defaulting to 1 for stocks and malformed values.""" + try: + multiplier = float(getattr(contract, "multiplier", None) or 1) + except (TypeError, ValueError): + return 1.0 + return multiplier if multiplier > 0 else 1.0 + + +def _positive_or_none(value) -> Optional[float]: + """IB reports an unset price as 0.0; treat that as absent.""" + try: + number = float(value) + except (TypeError, ValueError): + return None + return number if number > 0 else None + + @dataclass class IBKRConfig: """IBKR connection configuration.""" @@ -576,16 +594,26 @@ def get_positions(self) -> List[Dict[str, Any]]: for pos in positions: contract = pos.contract exchange = contract.exchange or contract.primaryExchange or "SMART" - + multiplier = _contract_multiplier(contract) + avg_cost = float(pos.avgCost) + result.append({ "symbol": format_display_symbol(contract.symbol, exchange), "ib_symbol": contract.symbol, + # Futures share a symbol across expiries; localSymbol (MGCV6) + # is what identifies the contract actually held. + "localSymbol": contract.localSymbol or "", + "lastTradeDate": contract.lastTradeDateOrContractMonth or "", "secType": contract.secType, "exchange": exchange, "currency": contract.currency, + "multiplier": multiplier, "quantity": float(pos.position), - "avgCost": float(pos.avgCost), - "marketValue": float(pos.position) * float(pos.avgCost), + # IB reports avgCost per contract, i.e. already multiplied. + "avgCost": avg_cost, + # Per unit, so it is comparable to quotes and stop prices. + "avgPrice": avg_cost / multiplier, + "marketValue": float(pos.position) * avg_cost, }) return result @@ -612,15 +640,25 @@ def get_open_orders(self) -> List[Dict[str, Any]]: order = trade.order contract = trade.contract status = trade.orderStatus - + limit_price = _positive_or_none(getattr(order, "lmtPrice", None)) + # Stop and stop-limit orders carry their trigger in auxPrice; + # lmtPrice is 0 for a plain STP, which reads as "no price". + stop_price = _positive_or_none(getattr(order, "auxPrice", None)) + result.append({ + # ``id`` is what order tables key rows and cancel actions on. + "id": order.orderId, "orderId": order.orderId, "permId": getattr(order, "permId", 0), "symbol": contract.symbol, + "localSymbol": contract.localSymbol or "", "action": order.action, "quantity": float(order.totalQuantity), "orderType": order.orderType, "limitPrice": getattr(order, 'lmtPrice', None), + "stopPrice": stop_price, + # The price that actually governs this order. + "price": limit_price if limit_price is not None else stop_price, "status": status.status, "filled": float(status.filled or 0), "remaining": float(status.remaining or 0), diff --git a/backend_api_python/app/services/live_trading/account_snapshot.py b/backend_api_python/app/services/live_trading/account_snapshot.py index 32bbb2290..a82383a84 100644 --- a/backend_api_python/app/services/live_trading/account_snapshot.py +++ b/backend_api_python/app/services/live_trading/account_snapshot.py @@ -509,26 +509,38 @@ def _fetch_ibkr_snapshot( qty = 0.0 if abs(qty) <= 0: continue + # avgPrice is per unit; avgCost is per contract and would show a + # futures entry inflated by the multiplier. + entry = p.get("avgPrice") + if entry in (None, ""): + entry = p.get("avgCost") positions_out.append({ - "symbol": str(p.get("symbol") or p.get("ib_symbol") or "").strip(), + # localSymbol distinguishes futures expiries (MGCV6 vs MGCZ6). + "symbol": str( + p.get("localSymbol") or p.get("symbol") or p.get("ib_symbol") or "" + ).strip(), "side": "long" if qty > 0 else "short", "size": abs(qty), - "entry_price": float(p.get("avgCost") or 0.0), + "entry_price": float(entry or 0.0), "market_type": "spot", "inst_id": str(p.get("ib_symbol") or "").strip(), }) for o in client.get_open_orders() or []: - lmt = o.get("limitPrice") + # A stop order keeps its trigger in stopPrice, not limitPrice. + price = o.get("price") + if price in (None, ""): + price = o.get("limitPrice") try: - lmt_f = float(lmt) if lmt not in (None, "") else None + price_f = float(price) if price not in (None, "") else None except (TypeError, ValueError): - lmt_f = None + price_f = None orders_out.append({ - "symbol": str(o.get("symbol") or "").strip(), + "symbol": str(o.get("localSymbol") or o.get("symbol") or "").strip(), "side": str(o.get("action") or "").strip().lower(), - "price": lmt_f, + "price": price_f, "size": float(o.get("quantity") or 0.0), "status": str(o.get("status") or "").strip(), + "exchange_order_id": str(o.get("orderId") or "").strip(), }) except Exception as e: _append_snapshot_error(errors, e, context="IBKR 账户连接") diff --git a/backend_api_python/tests/test_ibkr_session_and_account.py b/backend_api_python/tests/test_ibkr_session_and_account.py index 04ea9d997..1f6d85dcb 100644 --- a/backend_api_python/tests/test_ibkr_session_and_account.py +++ b/backend_api_python/tests/test_ibkr_session_and_account.py @@ -10,6 +10,7 @@ """ import threading +from types import SimpleNamespace from typing import Any, Dict import pytest @@ -17,7 +18,7 @@ from app.services.ibkr_trading import config as ibkr_config from app.services.ibkr_trading import session as ibkr_session from app.services.ibkr_trading.account import flatten_account_summary -from app.services.ibkr_trading.client import IBKRConfig +from app.services.ibkr_trading.client import IBKRClient, IBKRConfig from app.services.ibkr_trading.config import build_ibkr_config @@ -214,3 +215,87 @@ def test_flatten_omits_missing_tags_rather_than_reporting_zero(): def test_flatten_ignores_unusable_input(): assert flatten_account_summary(None) == {} assert flatten_account_summary({"NetLiquidation": {"value": "", "currency": "USD"}}) == {} + + +class _FakeContract(SimpleNamespace): + pass + + +def _futures_position(): + """A real MGC micro gold future: multiplier 10, avgCost already multiplied.""" + contract = _FakeContract( + symbol="MGC", localSymbol="MGCV6", secType="FUT", exchange="", + primaryExchange="", currency="USD", multiplier="10", + lastTradeDateOrContractMonth="20261028", + ) + return SimpleNamespace(contract=contract, position=1.0, avgCost=44332.96) + + +def _stock_position(): + contract = _FakeContract( + symbol="AAPL", localSymbol="AAPL", secType="STK", exchange="SMART", + primaryExchange="NASDAQ", currency="USD", multiplier="", + lastTradeDateOrContractMonth="", + ) + return SimpleNamespace(contract=contract, position=10.0, avgCost=205.25) + + +def _client_with_positions(positions): + client = object.__new__(IBKRClient) + client._ib = SimpleNamespace(positions=lambda account: positions) + client._account = "DU1" + client._ensure_connected = lambda: None + return client + + +def test_futures_average_price_is_reported_per_unit(): + """avgCost is per contract; showing it as the entry price inflates it 10x.""" + row = _client_with_positions([_futures_position()]).get_positions()[0] + + assert row["multiplier"] == 10 + assert row["avgCost"] == pytest.approx(44332.96) + assert row["avgPrice"] == pytest.approx(4433.296) + assert row["localSymbol"] == "MGCV6" + assert row["lastTradeDate"] == "20261028" + + +def test_stock_average_price_is_unchanged_by_the_multiplier(): + row = _client_with_positions([_stock_position()]).get_positions()[0] + + assert row["multiplier"] == 1 + assert row["avgPrice"] == pytest.approx(205.25) + assert row["avgCost"] == pytest.approx(205.25) + + +def _client_with_order(order_type, lmt_price, aux_price): + order = SimpleNamespace( + orderId=40, permId=515101143, action="SELL", totalQuantity=1.0, + orderType=order_type, lmtPrice=lmt_price, auxPrice=aux_price, + ) + trade = SimpleNamespace( + order=order, + contract=_FakeContract(symbol="MGC", localSymbol="MGCV6"), + orderStatus=SimpleNamespace( + status="PreSubmitted", filled=0, remaining=1, avgFillPrice=0 + ), + ) + client = object.__new__(IBKRClient) + client._ib = SimpleNamespace(openTrades=lambda: [trade]) + client._ensure_connected = lambda: None + return client + + +def test_stop_order_reports_its_trigger_price(): + """A protective stop keeps its price in auxPrice; lmtPrice is 0.""" + row = _client_with_order("STP", 0.0, 4350.0).get_open_orders()[0] + + assert row["stopPrice"] == pytest.approx(4350.0) + assert row["price"] == pytest.approx(4350.0) + assert row["id"] == 40 + + +def test_limit_order_still_reports_its_limit_price(): + row = _client_with_order("LMT", 4400.0, 0.0).get_open_orders()[0] + + assert row["price"] == pytest.approx(4400.0) + assert row["stopPrice"] is None From d1f3e459268f95d8ee8961a18c5679fece7e8dcb Mon Sep 17 00:00:00 2001 From: Johnson Tam Date: Wed, 12 Aug 2026 16:47:41 +0800 Subject: [PATCH 3/3] fix: use the per-unit entry price when mirroring IBKR positions `pending_order_position_sync` reads `IBKRClient.get_positions()` and fed `avgCost` into the L1 account mirror, so a futures entry was recorded inflated by the contract multiplier. Extract `position_entry_price()` into `ibkr_trading/account.py` and use it from both the account snapshot and the position sync, so the two paths cannot drift again. The reconciliation key deliberately stays on `symbol`/`ib_symbol` -- it has to match the symbol the strategy stored -- so only the price changes. Alpaca's `avgCost` is `avg_entry_price` and is already per unit, so that branch is untouched. Co-Authored-By: Claude Opus 5 (1M context) --- .../app/services/ibkr_trading/account.py | 23 ++++++++++++++++++- .../services/live_trading/account_snapshot.py | 10 ++++---- .../services/pending_order_position_sync.py | 12 ++++++---- .../tests/test_ibkr_session_and_account.py | 22 +++++++++++++++++- 4 files changed, 55 insertions(+), 12 deletions(-) diff --git a/backend_api_python/app/services/ibkr_trading/account.py b/backend_api_python/app/services/ibkr_trading/account.py index 10933eed2..c34baedec 100644 --- a/backend_api_python/app/services/ibkr_trading/account.py +++ b/backend_api_python/app/services/ibkr_trading/account.py @@ -48,6 +48,27 @@ def _tag_currency(summary: Mapping[str, Any], tag: str) -> str: return str(row.get("currency") or "").strip() +def position_entry_price(position: Mapping[str, Any]) -> float: + """Per-unit entry price for a position row from ``IBKRClient.get_positions``. + + ``avgCost`` is reported per contract, i.e. already multiplied, so a futures + position read through it is inflated by the contract multiplier. Callers + that compare against quotes, stops or fills want ``avgPrice``; ``avgCost`` + remains the fallback for rows produced before it existed. + """ + if not isinstance(position, Mapping): + return 0.0 + for field in ("avgPrice", "avgCost"): + raw = position.get(field) + if raw in (None, ""): + continue + try: + return float(raw) + except (TypeError, ValueError): + continue + return 0.0 + + def flatten_account_summary(summary: Any) -> Dict[str, Any]: """Return the flat numeric fields the broker-account UI reads. @@ -79,4 +100,4 @@ def flatten_account_summary(summary: Any) -> Dict[str, Any]: return flat -__all__ = ["flatten_account_summary"] +__all__ = ["flatten_account_summary", "position_entry_price"] diff --git a/backend_api_python/app/services/live_trading/account_snapshot.py b/backend_api_python/app/services/live_trading/account_snapshot.py index a82383a84..4c185d5aa 100644 --- a/backend_api_python/app/services/live_trading/account_snapshot.py +++ b/backend_api_python/app/services/live_trading/account_snapshot.py @@ -10,6 +10,9 @@ from typing import Any, Dict, List, Optional, Tuple from app.services.exchange_execution import resolve_exchange_config +from app.services.ibkr_trading.account import ( + position_entry_price as ibkr_position_entry_price, +) from app.services.live_trading.factory import create_client from app.services.live_trading.records import normalize_strategy_symbol from app.services.live_trading.spot_wallet_snapshot import list_spot_wallet_positions @@ -509,11 +512,6 @@ def _fetch_ibkr_snapshot( qty = 0.0 if abs(qty) <= 0: continue - # avgPrice is per unit; avgCost is per contract and would show a - # futures entry inflated by the multiplier. - entry = p.get("avgPrice") - if entry in (None, ""): - entry = p.get("avgCost") positions_out.append({ # localSymbol distinguishes futures expiries (MGCV6 vs MGCZ6). "symbol": str( @@ -521,7 +519,7 @@ def _fetch_ibkr_snapshot( ).strip(), "side": "long" if qty > 0 else "short", "size": abs(qty), - "entry_price": float(entry or 0.0), + "entry_price": ibkr_position_entry_price(p), "market_type": "spot", "inst_id": str(p.get("ib_symbol") or "").strip(), }) diff --git a/backend_api_python/app/services/pending_order_position_sync.py b/backend_api_python/app/services/pending_order_position_sync.py index 881df8fa7..19fabdb00 100644 --- a/backend_api_python/app/services/pending_order_position_sync.py +++ b/backend_api_python/app/services/pending_order_position_sync.py @@ -11,6 +11,9 @@ resolve_exchange_config, safe_exchange_config_for_log, ) +from app.services.ibkr_trading.account import ( + position_entry_price as ibkr_position_entry_price, +) from app.services.live_trading.account_positions import ( account_legs_from_exchange_maps, sync_account_positions, @@ -478,15 +481,16 @@ def _sync_positions_best_effort(self, target_strategy_id: Optional[int] = None) for p in positions: if not isinstance(p, dict): continue + # Reconciliation keys off the strategy's own + # symbol, so this must stay symbol/ib_symbol. sym = str(p.get("symbol") or p.get("ib_symbol") or "").strip() try: qty = float(p.get("quantity") or 0.0) except Exception: qty = 0.0 - try: - avg = float(p.get("avgCost") or 0.0) - except Exception: - avg = 0.0 + # avgCost is per contract; reading a futures + # entry through it inflates it by the multiplier. + avg = ibkr_position_entry_price(p) if not sym or abs(qty) <= 0: continue side = "long" if qty > 0 else "short" diff --git a/backend_api_python/tests/test_ibkr_session_and_account.py b/backend_api_python/tests/test_ibkr_session_and_account.py index 1f6d85dcb..c36ceac7d 100644 --- a/backend_api_python/tests/test_ibkr_session_and_account.py +++ b/backend_api_python/tests/test_ibkr_session_and_account.py @@ -17,7 +17,10 @@ from app.services.ibkr_trading import config as ibkr_config from app.services.ibkr_trading import session as ibkr_session -from app.services.ibkr_trading.account import flatten_account_summary +from app.services.ibkr_trading.account import ( + flatten_account_summary, + position_entry_price, +) from app.services.ibkr_trading.client import IBKRClient, IBKRConfig from app.services.ibkr_trading.config import build_ibkr_config @@ -299,3 +302,20 @@ def test_limit_order_still_reports_its_limit_price(): assert row["price"] == pytest.approx(4400.0) assert row["stopPrice"] is None + + +def test_entry_price_prefers_the_per_unit_average(): + """Reconciliation and account mirrors must not use the per-contract cost.""" + row = _client_with_positions([_futures_position()]).get_positions()[0] + + assert position_entry_price(row) == pytest.approx(4433.296) + + +def test_entry_price_falls_back_to_avg_cost_for_rows_without_avg_price(): + assert position_entry_price({"avgCost": 205.25}) == pytest.approx(205.25) + + +def test_entry_price_ignores_unusable_rows(): + assert position_entry_price(None) == 0.0 + assert position_entry_price({}) == 0.0 + assert position_entry_price({"avgPrice": "", "avgCost": "n/a"}) == 0.0