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..c34baedec --- /dev/null +++ b/backend_api_python/app/services/ibkr_trading/account.py @@ -0,0 +1,103 @@ +"""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 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. + + 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", "position_entry_price"] 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/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..4c185d5aa 100644 --- a/backend_api_python/app/services/live_trading/account_snapshot.py +++ b/backend_api_python/app/services/live_trading/account_snapshot.py @@ -5,10 +5,14 @@ from __future__ import annotations +import threading import time 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 @@ -16,6 +20,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 +488,69 @@ 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({ + # 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": ibkr_position_entry_price(p), + "market_type": "spot", + "inst_id": str(p.get("ib_symbol") or "").strip(), + }) + for o in client.get_open_orders() or []: + # A stop order keeps its trigger in stopPrice, not limitPrice. + price = o.get("price") + if price in (None, ""): + price = o.get("limitPrice") + try: + price_f = float(price) if price not in (None, "") else None + except (TypeError, ValueError): + price_f = None + orders_out.append({ + "symbol": str(o.get("localSymbol") or o.get("symbol") or "").strip(), + "side": str(o.get("action") or "").strip().lower(), + "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 账户连接") + 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 +611,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/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 new file mode 100644 index 000000000..c36ceac7d --- /dev/null +++ b/backend_api_python/tests/test_ibkr_session_and_account.py @@ -0,0 +1,321 @@ +"""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 types import SimpleNamespace +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, + position_entry_price, +) +from app.services.ibkr_trading.client import IBKRClient, 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"}}) == {} + + +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 + + +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