From bdff227bec853be498d5265204a996e11994e391 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Sun, 10 May 2026 23:36:35 +0800 Subject: [PATCH] Route IBKR live orders by account group --- README.md | 4 +- application/execution_service.py | 105 +++++++++++++++++++++++-- application/ibkr_order_execution.py | 2 + application/runtime_broker_adapters.py | 9 ++- docs/ibkr_runtime_rollout.md | 2 +- requirements.txt | 4 +- strategy_runtime.py | 7 +- tests/test_execution_service.py | 85 +++++++++++++++++++- tests/test_ibkr_order_execution.py | 15 ++++ 9 files changed, 218 insertions(+), 15 deletions(-) diff --git a/README.md b/README.md index 7f7b2b6..86fa541 100644 --- a/README.md +++ b/README.md @@ -197,7 +197,7 @@ Recommended account-group config payload: } ``` -If you later add live, keep it as a separate live group such as `live-main` with `ib_gateway_mode=live` and its own `account_ids`. +For live multi-account rollout, keep one Cloud Run service per live account group. Each live group should carry exactly one `account_ids` value so portfolio reads, pending/fill guards, and submitted IBKR orders are all routed to that account. See [`docs/examples/ibkr-account-groups.paper.json`](docs/examples/ibkr-account-groups.paper.json) for a ready-to-edit starter example, and [`docs/ibkr_runtime_rollout.md`](docs/ibkr_runtime_rollout.md) for the exact rollout steps to get `ACCOUNT_GROUP=paper` running. @@ -403,6 +403,8 @@ IB_GATEWAY_IP_MODE=internal 仓库里也提供了一个可以直接改的起始样例：[`docs/examples/ibkr-account-groups.paper.json`](docs/examples/ibkr-account-groups.paper.json)。如果你要按 `ACCOUNT_GROUP=paper` 先落地，直接看 [`docs/ibkr_runtime_rollout.md`](docs/ibkr_runtime_rollout.md)。 +实盘多账户建议一个 UID 对应一个 Cloud Run 服务和一个账号组。每个实盘账号组只放一个 `account_ids` 值；运行时会用它过滤持仓、pending/fill 检查，并把同一个 UID 写进 IBKR 订单的 `order.account`。 + 当前行为改成了 fail-fast： - 没有 `STRATEGY_PROFILE` → 启动直接报错 diff --git a/application/execution_service.py b/application/execution_service.py index 78f5c94..d53ec84 100644 --- a/application/execution_service.py +++ b/application/execution_service.py @@ -64,9 +64,44 @@ def check_order_submitted(report, *, translator): return False, f"❌ {translator('failed', reason=status)}" -def get_available_buying_power(ib, fallback_buying_power): +def _normalize_account_ids(account_ids=None) -> tuple[str, ...]: + if account_ids is None: + return () + if isinstance(account_ids, str): + candidates = [account_ids] + else: + candidates = list(account_ids) + normalized = [] + for candidate in candidates: + text = str(candidate or "").strip() + if text: + normalized.append(text) + return tuple(dict.fromkeys(normalized)) + + +def _matches_account(account_id: str | None, selected_account_ids: tuple[str, ...]) -> bool: + if not selected_account_ids: + return True + return str(account_id or "").strip() in selected_account_ids + + +def _resolve_order_account_id(account_ids=None) -> str | None: + normalized = _normalize_account_ids(account_ids) + if len(normalized) > 1: + raise ValueError( + "IBKR live order routing requires a single account_id per runtime service; " + f"got {len(normalized)} account_ids." + ) + return normalized[0] if normalized else None + + +def get_available_buying_power(ib, fallback_buying_power, *, account_ids=None): + selected_account_ids = _normalize_account_ids(account_ids) buying_power = fallback_buying_power for account_value in ib.accountValues(): + account_id = str(getattr(account_value, "account", "") or "").strip() or None + if not _matches_account(account_id, selected_account_ids): + continue if account_value.tag == "AvailableFunds" and account_value.currency == "USD": buying_power = float(account_value.value) return buying_power @@ -101,9 +136,24 @@ def _extract_open_order_status(order_like: Any) -> str: return str(status or "").strip() -def _collect_pending_symbols(ib, symbols: set[str]) -> tuple[str, ...]: +def _extract_open_order_account(order_like: Any) -> str | None: + order = getattr(order_like, "order", None) + for candidate in ( + getattr(order, "account", None), + getattr(order_like, "account", None), + ): + text = str(candidate or "").strip() + if text: + return text + return None + + +def _collect_pending_symbols(ib, symbols: set[str], *, account_ids=None) -> tuple[str, ...]: + selected_account_ids = _normalize_account_ids(account_ids) pending = [] for order_like in _iter_open_orders(ib): + if not _matches_account(_extract_open_order_account(order_like), selected_account_ids): + continue status = _extract_open_order_status(order_like) if status in {"Cancelled", "ApiCancelled", "Inactive", "Filled"}: continue @@ -127,6 +177,19 @@ def _extract_fill_symbol(fill_like: Any) -> str | None: return symbol_text or None +def _extract_fill_account(fill_like: Any) -> str | None: + execution = getattr(fill_like, "execution", None) + for candidate in ( + getattr(execution, "acctNumber", None), + getattr(execution, "account", None), + getattr(fill_like, "account", None), + ): + text = str(candidate or "").strip() + if text: + return text + return None + + def _normalize_date_like(value: Any) -> str | None: if value in {None, ""}: return None @@ -150,11 +213,20 @@ def _extract_fill_date(fill_like: Any) -> str | None: return None -def _collect_same_day_filled_symbols(ib, symbols: set[str], trade_date: str | None) -> tuple[str, ...]: +def _collect_same_day_filled_symbols( + ib, + symbols: set[str], + trade_date: str | None, + *, + account_ids=None, +) -> tuple[str, ...]: if not trade_date: return () + selected_account_ids = _normalize_account_ids(account_ids) matched = [] for fill_like in _iter_fills(ib): + if not _matches_account(_extract_fill_account(fill_like), selected_account_ids): + continue symbol = _extract_fill_symbol(fill_like) if not symbol or symbol not in symbols: continue @@ -425,9 +497,15 @@ def execute_rebalance( safe_haven_symbols = tuple(allocation["safe_haven_symbols"]) safe_haven_symbol = safe_haven_symbols[0] if safe_haven_symbols else None equity = account_values.get("equity", 0) + normalized_account_ids = _normalize_account_ids(account_ids) + order_account_id = _resolve_order_account_id(normalized_account_ids) execution_summary = { "mode": "dry_run" if dry_run_only else "paper", "strategy_profile": strategy_profile, + "account_group": account_group, + "account_ids": list(normalized_account_ids), + "order_account_id": order_account_id, + "service_name": service_name, "trade_date": trade_date, "snapshot_as_of": snapshot_date, "safe_haven_symbol": safe_haven_symbol, @@ -508,7 +586,7 @@ def execute_rebalance( (sum(current_mv.values()) - current_safe_haven_mv) / equity ) - pending_symbols = _collect_pending_symbols(ib, set(all_symbols)) + pending_symbols = _collect_pending_symbols(ib, set(all_symbols), account_ids=normalized_account_ids) if pending_symbols: reason = f"pending_orders_detected:{','.join(pending_symbols)}" execution_summary["execution_status"] = "blocked" @@ -597,6 +675,7 @@ def execute_rebalance( anticipated_buying_power = get_available_buying_power( ib, account_values.get("buying_power", 0), + account_ids=normalized_account_ids, ) has_buy_plan = False for symbol, target in target_mv.items(): @@ -655,7 +734,12 @@ def execute_rebalance( trade_logs.append(translator("failed", reason=reason)) return _finalize_result(trade_logs, execution_summary, return_summary=return_summary) - same_day_filled_symbols = _collect_same_day_filled_symbols(ib, set(all_symbols), trade_date) + same_day_filled_symbols = _collect_same_day_filled_symbols( + ib, + set(all_symbols), + trade_date, + account_ids=normalized_account_ids, + ) if same_day_filled_symbols: reason = f"same_day_fills_detected:{','.join(same_day_filled_symbols)}" execution_summary["execution_status"] = "blocked" @@ -685,7 +769,7 @@ def execute_rebalance( strategy_profile=strategy_profile, account_group=account_group, service_name=service_name, - account_ids=tuple(account_ids or ()), + account_ids=normalized_account_ids, trade_date=trade_date, snapshot_date=snapshot_date, target_hash=target_hash, @@ -754,7 +838,12 @@ def execute_rebalance( continue report = submit_order_intent( ib, - order_intent_cls(symbol=symbol, side="sell", quantity=qty), + order_intent_cls( + symbol=symbol, + side="sell", + quantity=qty, + account_id=order_account_id, + ), ) ok, status_msg = check_order_submitted(report, translator=translator) status = str(getattr(report, "status", "") or "") @@ -784,6 +873,7 @@ def execute_rebalance( buying_power = anticipated_buying_power if not sell_executed else get_available_buying_power( ib, account_values.get("buying_power", 0), + account_ids=normalized_account_ids, ) for symbol, target in target_mv.items(): @@ -830,6 +920,7 @@ def execute_rebalance( order_type="limit", limit_price=limit_price, time_in_force="DAY", + account_id=order_account_id, ), ) ok, status_msg = check_order_submitted(report, translator=translator) diff --git a/application/ibkr_order_execution.py b/application/ibkr_order_execution.py index 711c7a4..64b7633 100644 --- a/application/ibkr_order_execution.py +++ b/application/ibkr_order_execution.py @@ -39,6 +39,7 @@ def submit_order_intent( ib: Any, order_intent: OrderIntent, *, + account_id: str | None = None, wait_seconds: float = 1.0, stock_factory: Callable[..., Any] | None = None, market_order_factory: Callable[..., Any] | None = None, @@ -50,6 +51,7 @@ def submit_order_intent( return _submit_order_intent( ib, intent, + account_id=account_id, wait_seconds=wait_seconds, stock_factory=stock_factory, market_order_factory=_market_order_factory_with_time_in_force( diff --git a/application/runtime_broker_adapters.py b/application/runtime_broker_adapters.py index e11d910..94312be 100644 --- a/application/runtime_broker_adapters.py +++ b/application/runtime_broker_adapters.py @@ -46,6 +46,11 @@ class IBKRRuntimeBrokerAdapters: sleep_fn: Any printer: Any = print + def fetch_account_portfolio_snapshot(self, ib): + if self.account_ids: + return self.fetch_portfolio_snapshot_fn(ib, account_ids=self.account_ids) + return self.fetch_portfolio_snapshot_fn(ib) + def connect_ib(self): self.ensure_event_loop_fn() host = self.host_resolver() @@ -82,7 +87,7 @@ def connect_ib(self): raise last_error def get_current_portfolio(self, ib): - snapshot = self.fetch_portfolio_snapshot_fn(ib) + snapshot = self.fetch_account_portfolio_snapshot(ib) positions = {} for position in snapshot.positions: positions[position.symbol] = { @@ -97,7 +102,7 @@ def get_current_portfolio(self, ib): def build_portfolio_snapshot(self, ib, *, get_current_portfolio_fallback=None): if hasattr(ib, "reqPositions"): - return self.fetch_portfolio_snapshot_fn(ib) + return self.fetch_account_portfolio_snapshot(ib) positions, account_values = get_current_portfolio_fallback(ib) return PortfolioSnapshot( as_of=datetime.now(timezone.utc), diff --git a/docs/ibkr_runtime_rollout.md b/docs/ibkr_runtime_rollout.md index 6d9789f..74b4396 100644 --- a/docs/ibkr_runtime_rollout.md +++ b/docs/ibkr_runtime_rollout.md @@ -87,7 +87,7 @@ cp docs/examples/ibkr-account-groups.paper.json /tmp/ibkr-account-groups.json - `ib_gateway_ip_mode`：推荐 `internal`。 - `ib_client_id`：这个账号组对应的 client id。 - `service_name`：当前只是预留元数据，建议先填成现有 Cloud Run 服务名，后面多账号拆服务时更顺。 -- `account_ids`：当前主要是留档和后续扩展，不是启动必填。 +- `account_ids`：实盘账号组建议只放一个 UID。运行时会用它过滤持仓、pending/fill 检查，并写入 IBKR 订单的 `order.account`；如果一个服务配置多个 UID，实盘下单会因为账户路由不明确而失败。 把 secret 建起来： diff --git a/requirements.txt b/requirements.txt index 7918e27..47fc1f0 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,7 +1,7 @@ flask gunicorn -quant-platform-kit @ git+https://github.com/QuantStrategyLab/QuantPlatformKit.git@8769362096227320bc05c791b5244d4b3e88db50 -us-equity-strategies @ git+https://github.com/QuantStrategyLab/UsEquityStrategies.git@ed55a6af0245323dbed82060e89be96d8f77f756 +quant-platform-kit @ git+https://github.com/QuantStrategyLab/QuantPlatformKit.git@c6b22288e1447e56cd51a93ea9138e4f1110f165 +us-equity-strategies @ git+https://github.com/QuantStrategyLab/UsEquityStrategies.git@dd8aca2bea31196c07fd07efba286266478d29e9 pandas numpy requests diff --git a/strategy_runtime.py b/strategy_runtime.py index aa826ac..57af199 100644 --- a/strategy_runtime.py +++ b/strategy_runtime.py @@ -88,9 +88,14 @@ def _runtime_adapter_with_portfolio( def _fetch_portfolio_snapshot_for_context(self, ib, *, required: bool) -> Any | None: if ib is None and not required: return None + account_ids = tuple(self.runtime_settings.account_ids or ()) if required: + if account_ids: + return fetch_portfolio_snapshot(ib, account_ids=account_ids) return fetch_portfolio_snapshot(ib) try: + if account_ids: + return fetch_portfolio_snapshot(ib, account_ids=account_ids) return fetch_portfolio_snapshot(ib) except Exception as exc: self.logger( @@ -245,7 +250,7 @@ def _evaluate_value_target_strategy( runtime_config = dict(self.runtime_config) runtime_config.setdefault("translator", translator) apply_runtime_policy_to_runtime_config(runtime_config, self.runtime_adapter) - portfolio_snapshot = fetch_portfolio_snapshot(ib) + portfolio_snapshot = self._fetch_portfolio_snapshot_for_context(ib, required=True) market_inputs = self._build_value_target_market_inputs( ib=ib, historical_close_loader=historical_close_loader, diff --git a/tests/test_execution_service.py b/tests/test_execution_service.py index 713587d..6bc4537 100644 --- a/tests/test_execution_service.py +++ b/tests/test_execution_service.py @@ -130,6 +130,89 @@ def fake_fetch_quote_snapshots(_ib, symbols): assert any(log.startswith("buy VOO") for log in trade_logs) +def test_execute_rebalance_routes_order_to_single_account_id(monkeypatch, tmp_path): + class FakeIB: + def openTrades(self): + return [] + + def fills(self): + return [] + + def accountValues(self): + return [ + SimpleNamespace(account="U18308207", tag="AvailableFunds", currency="USD", value="5000"), + SimpleNamespace(account="U15998061", tag="AvailableFunds", currency="USD", value="100"), + ] + + submitted = [] + + def fake_submit_order_intent(_ib, intent): + submitted.append(intent) + return SimpleNamespace(broker_order_id="1", status="Submitted") + + monkeypatch.setattr("application.execution_service.time.sleep", lambda _seconds: None) + + _trade_logs, summary = execute_rebalance( + FakeIB(), + {"TQQQ": 1.0}, + {}, + {"equity": 1000.0, "buying_power": 1000.0}, + fetch_quote_snapshots=lambda _ib, symbols: { + symbol: SimpleNamespace(last_price=100.0) for symbol in symbols + }, + submit_order_intent=fake_submit_order_intent, + order_intent_cls=OrderIntent, + translator=translate, + strategy_symbols=["TQQQ"], + strategy_profile="tqqq_growth_income", + account_group="live-tqqq", + service_name="ibkr-tqqq-live", + account_ids=("U18308207",), + signal_metadata=_signal_metadata({"TQQQ": 1.0}, risk_symbols=("TQQQ",), trade_date="2026-04-01"), + dry_run_only=False, + cash_reserve_ratio=0.0, + rebalance_threshold_ratio=0.02, + limit_buy_premium=1.005, + sell_settle_delay_sec=0, + execution_lock_dir=tmp_path, + return_summary=True, + ) + + assert len(submitted) == 1 + assert submitted[0].account_id == "U18308207" + assert summary["order_account_id"] == "U18308207" + + +def test_execute_rebalance_rejects_multiple_order_account_ids(): + class FakeIB: + def accountValues(self): + return [] + + try: + execute_rebalance( + FakeIB(), + {"TQQQ": 1.0}, + {}, + {"equity": 1000.0, "buying_power": 1000.0}, + fetch_quote_snapshots=lambda _ib, _symbols: {}, + submit_order_intent=lambda *_args, **_kwargs: None, + order_intent_cls=OrderIntent, + translator=translate, + strategy_symbols=["TQQQ"], + signal_metadata=_signal_metadata({"TQQQ": 1.0}, risk_symbols=("TQQQ",)), + account_ids=("U18308207", "U15998061"), + dry_run_only=False, + cash_reserve_ratio=0.0, + rebalance_threshold_ratio=0.02, + limit_buy_premium=1.005, + sell_settle_delay_sec=0, + ) + except ValueError as exc: + assert "single account_id" in str(exc) + else: + raise AssertionError("expected ValueError") + + def test_execute_rebalance_can_submit_fractional_buy_when_quantity_step_allows(monkeypatch, tmp_path): class FakeIB: def openTrades(self): @@ -345,7 +428,7 @@ def fills(self): return [ SimpleNamespace( contract=SimpleNamespace(symbol="VOO"), - execution=SimpleNamespace(time="2026-04-01 10:30:00"), + execution=SimpleNamespace(time="2026-04-01 10:30:00", acctNumber="DU123"), ) ] diff --git a/tests/test_ibkr_order_execution.py b/tests/test_ibkr_order_execution.py index e22214d..d57f7ca 100644 --- a/tests/test_ibkr_order_execution.py +++ b/tests/test_ibkr_order_execution.py @@ -59,3 +59,18 @@ def test_submit_order_intent_preserves_explicit_tif_on_market_orders(): ) assert ib.placed_order.tif == "GTC" + + +def test_submit_order_intent_preserves_account_id(): + ib = FakeIB() + + report = submit_order_intent( + ib, + OrderIntent(symbol="AAPL", side="buy", quantity=3, account_id="U18308207"), + wait_seconds=0, + stock_factory=fake_stock, + market_order_factory=FakeMarketOrder, + ) + + assert ib.placed_order.account == "U18308207" + assert report.raw_payload["account_id"] == "U18308207"