diff --git a/.env.example b/.env.example index 07e5892..6636ae6 100644 --- a/.env.example +++ b/.env.example @@ -26,6 +26,16 @@ QSL_NOTIFY_LANG= IBKR_DRY_RUN_ONLY=false IBKR_FORCE_RUN=false +# ── Isolated Paper Command Consumer (default off) ── +# This endpoint only verifies delayed paper commands. It never constructs an +# order-execution adapter and requires all of the following at runtime: +# RUNTIME_TARGET_ENABLED=false, IBKR_DRY_RUN_ONLY=true, +# IBKR_GATEWAY_MODE=paper, RUNTIME_TARGET_JSON.execution_mode=paper, and +# CASH_ONLY_EXECUTION=true. Keep it false until a paper command store and +# paper-only runtime target have been independently verified. +IBKR_PAPER_EXECUTION_COMMAND_CONSUMER_ENABLED=false +IBKR_EXECUTION_COMMAND_CLOUD_URI=gs://your-bucket/paper-execution-commands + # ── Execution Report ── EXECUTION_REPORT_OUTPUT_DIR=/tmp/quant_runtime_reports EXECUTION_REPORT_GCS_URI=gs://your-bucket/execution-reports diff --git a/application/ibkr_portfolio.py b/application/ibkr_portfolio.py index 51911db..a54824f 100644 --- a/application/ibkr_portfolio.py +++ b/application/ibkr_portfolio.py @@ -217,3 +217,124 @@ def fetch_portfolio_snapshot( ), }, ) + + +def fetch_reconciled_paper_portfolio_snapshot( + ib: Any, + *, + account_ids: Iterable[str] | str | None = None, + currency: str = "USD", +) -> PortfolioSnapshot: + """Read a current IBKR portfolio for the isolated paper command consumer. + + ``positions()`` exposes average cost but not a trustworthy current market + value. Delayed-command reconciliation must not mistake cost basis for + live exposure, so this intentionally uses IBKR's read-only ``portfolio`` + snapshot instead. It is separate from :func:`fetch_portfolio_snapshot` + to avoid changing the existing strategy execution path. + """ + + selected_account_ids = _normalize_account_ids(account_ids) + market_currency = str(currency or "USD").strip().upper() + portfolio_fn = getattr(ib, "portfolio", None) + if not callable(portfolio_fn): + raise IBKRPortfolioSnapshotUnavailableError( + "IBKR paper command reconciliation requires the read-only portfolio snapshot API." + ) + + raw_items = [] + account_queries = selected_account_ids or ("",) + try: + for account_id in account_queries: + raw_items.extend(tuple(portfolio_fn(account_id) or ())) + except Exception as exc: + raise IBKRPortfolioSnapshotUnavailableError( + "IBKR paper command reconciliation could not load current portfolio values." + ) from exc + + positions: list[Position] = [] + seen_positions: set[tuple[str | None, str, str, float]] = set() + for item in raw_items: + account_id = str(getattr(item, "account", "") or "").strip() or None + if not _matches_account(account_id, selected_account_ids): + continue + contract = getattr(item, "contract", None) + symbol = str(getattr(contract, "symbol", "") or "").strip().upper() + contract_currency = str(getattr(contract, "currency", "") or "").strip().upper() + if not symbol or contract_currency != market_currency: + raise IBKRPortfolioSnapshotUnavailableError( + "IBKR paper command reconciliation received an incomplete or non-market-currency position." + ) + quantity = _as_float(getattr(item, "position", None)) + market_value = _as_float(getattr(item, "marketValue", None)) + average_cost = _as_float(getattr(item, "averageCost", None)) + if quantity is None or market_value is None or average_cost is None: + raise IBKRPortfolioSnapshotUnavailableError( + "IBKR paper command reconciliation is missing current position market values." + ) + if quantity == 0.0: + continue + dedupe_key = (account_id, symbol, str(getattr(contract, "conId", "") or ""), quantity) + if dedupe_key in seen_positions: + continue + seen_positions.add(dedupe_key) + positions.append( + Position( + symbol=symbol, + quantity=quantity, + market_value=market_value, + average_cost=average_cost, + currency=contract_currency, + ) + ) + + values_by_account_currency: dict[tuple[str | None, str], dict[str, float]] = {} + try: + raw_account_values = tuple(ib.accountValues() or ()) + except Exception as exc: + raise IBKRPortfolioSnapshotUnavailableError( + "IBKR paper command reconciliation could not load account cash values." + ) from exc + for account_value in raw_account_values: + account_id = str(getattr(account_value, "account", "") or "").strip() or None + if not _matches_account(account_id, selected_account_ids): + continue + value_currency = str(getattr(account_value, "currency", "") or "").strip().upper() + numeric_value = _as_float(getattr(account_value, "value", None)) + if value_currency and numeric_value is not None: + values_by_account_currency.setdefault((account_id, value_currency), {})[ + str(getattr(account_value, "tag", "") or "").strip() + ] = numeric_value + market_currency_cash = _cash_value_for_currency( + values_by_account_currency, + currency=market_currency, + ) + if market_currency_cash is None: + raise IBKRPortfolioSnapshotUnavailableError( + f"IBKR paper command reconciliation is missing the {market_currency} cash balance." + ) + total_equity = float(market_currency_cash) + sum(float(position.market_value) for position in positions) + return PortfolioSnapshot( + as_of=datetime.now(timezone.utc), + total_equity=total_equity, + cash_balance=float(market_currency_cash), + buying_power=float(market_currency_cash), + positions=tuple(positions), + metadata={ + "account_ids": selected_account_ids, + "currency": market_currency, + "market_currency_cash": float(market_currency_cash), + "reconciliation_source": "ibkr_portfolio_market_value", + "cash_balances": tuple( + { + "account_id": account_id, + "currency": value_currency, + **tag_values, + } + for (account_id, value_currency), tag_values in sorted( + values_by_account_currency.items(), + key=lambda item: ((item[0][0] or ""), item[0][1]), + ) + ), + }, + ) diff --git a/application/paper_execution_command_consumer.py b/application/paper_execution_command_consumer.py new file mode 100644 index 0000000..3a7740e --- /dev/null +++ b/application/paper_execution_command_consumer.py @@ -0,0 +1,217 @@ +"""IBKR read-only reconciliation adapter for shared paper commands. + +This module deliberately has no order, contract-submission, or execution-port +import. It consumes only normalized portfolio snapshots and quote snapshots +provided by the isolated endpoint after the shared command binding passes. +""" + +from __future__ import annotations + +import math +from collections.abc import Callable, Mapping, Sequence +from datetime import date +from typing import Any + +from quant_platform_kit.common.execution_commands import ExecutionCommand, ExecutionCommandStore +from quant_platform_kit.common.paper_execution_command_consumer import ( + PaperExecutionProposal, + PaperExecutionReconciliation, + consume_due_paper_execution_commands as consume_shared_paper_execution_commands, +) +from quant_platform_kit.common.runtime_command_gate import RuntimeCommandExposureEffect +from quant_platform_kit.common.strategy_release import StrategyReleaseIdentity + + +IBKR_PAPER_EXECUTION_INTENT_SCHEMA_VERSION = "ibkr.paper-execution-intent.v1" +_NOTIONAL_TOLERANCE = 0.01 + + +def resolve_paper_execution_command_consumer_enabled(*, env_reader, dry_run_only: bool) -> bool: + """Enable only the isolated local-dry-run command consumer.""" + + enabled = str( + env_reader("IBKR_PAPER_EXECUTION_COMMAND_CONSUMER_ENABLED", "") or "" + ).strip().lower() in {"1", "true", "t", "yes", "y", "on"} + if enabled and not dry_run_only: + raise RuntimeError("IBKR paper command consumer requires IBKR_DRY_RUN_ONLY=true") + return enabled + + +def _symbol(value: object) -> str: + return str(value or "").strip().upper() + + +def _symbols(value: object) -> set[str]: + if not isinstance(value, (list, tuple, set)): + return set() + return {_symbol(item) for item in value if _symbol(item)} + + +def _finite(value: object, *, field_name: str) -> float: + try: + number = float(value) + except (TypeError, ValueError) as exc: + raise ValueError(f"{field_name} must be numeric") from exc + if not math.isfinite(number): + raise ValueError(f"{field_name} must be finite") + return number + + +def _cash_balance(portfolio: Any) -> float: + metadata = getattr(portfolio, "metadata", {}) + if not isinstance(metadata, Mapping): + raise ValueError("portfolio metadata is unavailable") + return _finite(metadata.get("market_currency_cash"), field_name="portfolio.market_currency_cash") + + +def _reconcile( + command: ExecutionCommand, + *, + portfolio: Any, + quote_loader: Callable[[str], Any], + managed_symbols: Sequence[str], +) -> PaperExecutionReconciliation: + intent = command.intent + if str(intent.get("schema_version") or "") != IBKR_PAPER_EXECUTION_INTENT_SCHEMA_VERSION: + return PaperExecutionReconciliation(proposals=(), integrity_findings=("data_artifact_invalid",)) + if str(intent.get("target_mode") or "") != "value": + return PaperExecutionReconciliation(proposals=(), integrity_findings=("data_artifact_invalid",)) + raw_targets = intent.get("targets") + if not isinstance(raw_targets, Mapping): + return PaperExecutionReconciliation(proposals=(), integrity_findings=("data_artifact_invalid",)) + try: + targets = { + _symbol(symbol): _finite(value, field_name=f"targets[{symbol!r}]") + for symbol, value in raw_targets.items() + if _symbol(symbol) + } + except ValueError: + return PaperExecutionReconciliation(proposals=(), integrity_findings=("data_artifact_invalid",)) + strategy_symbols = _symbols(intent.get("strategy_symbols")) + expected_symbols = {_symbol(symbol) for symbol in managed_symbols if _symbol(symbol)} + if ( + not strategy_symbols + or strategy_symbols != expected_symbols + or set(targets) != strategy_symbols + or any(value < 0.0 for value in targets.values()) + ): + return PaperExecutionReconciliation(proposals=(), integrity_findings=("data_artifact_invalid",)) + + findings: list[str] = [] + quantities: dict[str, float] = {} + current_values: dict[str, float] = {} + for position in tuple(getattr(portfolio, "positions", ()) or ()): + symbol = _symbol(getattr(position, "symbol", "")) + if symbol not in strategy_symbols: + findings.append("position_reconciliation_mismatch") + continue + try: + quantity = _finite(getattr(position, "quantity", None), field_name=f"position[{symbol}].quantity") + recorded_value = _finite( + getattr(position, "market_value", None), + field_name=f"position[{symbol}].market_value", + ) + quote = quote_loader(symbol) + price = _finite(getattr(quote, "last_price", None), field_name=f"quote[{symbol}].last_price") + if price <= 0.0: + raise ValueError("quote price must be positive") + except Exception: + findings.append("position_reconciliation_mismatch") + continue + quote_value = quantity * price + # IBKR cash-only commands need current market-value evidence. A stale + # average-cost snapshot is not enough to classify exposure safely. + tolerance = max(1.0, abs(quote_value) * 0.005) + if quantity < -_NOTIONAL_TOLERANCE or recorded_value < -_NOTIONAL_TOLERANCE: + findings.append("position_reconciliation_mismatch") + if abs(recorded_value - quote_value) > tolerance: + findings.append("position_reconciliation_mismatch") + quantities[symbol] = quantities.get(symbol, 0.0) + quantity + current_values[symbol] = current_values.get(symbol, 0.0) + quote_value + + try: + cash_balance = _cash_balance(portfolio) + total_equity = _finite(getattr(portfolio, "total_equity", None), field_name="portfolio.total_equity") + tolerance = max(1.0, abs(total_equity) * 0.005) + if abs(cash_balance + sum(current_values.values()) - total_equity) > tolerance: + findings.append("position_reconciliation_mismatch") + except ValueError: + findings.append("position_reconciliation_mismatch") + + proposals: list[PaperExecutionProposal] = [] + for symbol in sorted(strategy_symbols): + current_value = current_values.get(symbol, 0.0) + target_value = targets[symbol] + delta_value = target_value - current_value + if abs(delta_value) <= _NOTIONAL_TOLERANCE: + continue + try: + quote = quote_loader(symbol) + price = _finite(getattr(quote, "last_price", None), field_name=f"quote[{symbol}].last_price") + if price <= 0.0: + raise ValueError("quote price must be positive") + except Exception: + findings.append("position_reconciliation_mismatch") + continue + if abs(target_value) < abs(current_value) - _NOTIONAL_TOLERANCE: + effect = RuntimeCommandExposureEffect.REDUCES + elif abs(target_value) > abs(current_value) + _NOTIONAL_TOLERANCE: + effect = RuntimeCommandExposureEffect.INCREASES + else: + effect = RuntimeCommandExposureEffect.NEUTRAL + proposals.append( + PaperExecutionProposal( + symbol=symbol, + exposure_effect=effect, + details={ + "side": "buy" if delta_value > 0.0 else "sell", + "quantity": round(abs(delta_value) / price, 8), + "reference_price": round(price, 8), + "current_value": round(current_value, 8), + "target_value": round(target_value, 8), + "target_notional_delta": round(delta_value, 8), + "current_quantity": round(quantities.get(symbol, 0.0), 8), + }, + ) + ) + return PaperExecutionReconciliation( + proposals=tuple(proposals), + integrity_findings=tuple(dict.fromkeys(findings)), + ) + + +def consume_due_paper_execution_commands( + *, + store: ExecutionCommandStore | None, + as_of_session: date | str, + claimant: str, + portfolio_loader: Callable[[], Any], + quote_loader: Callable[[str], Any], + managed_symbols: Sequence[str], + runtime_release_receipt: Mapping[str, Any] | None, + expected_strategy_release: StrategyReleaseIdentity | Mapping[str, object] | None, + expected_command_binding: Mapping[str, object] | None, +) -> dict[str, object]: + """Consume paper commands after shared release and delivery binding checks.""" + + return consume_shared_paper_execution_commands( + store=store, + as_of_session=as_of_session, + claimant=claimant, + reconcile_command=lambda command: _reconcile( + command, + portfolio=portfolio_loader(), + quote_loader=quote_loader, + managed_symbols=managed_symbols, + ), + runtime_release_receipt=runtime_release_receipt, + expected_strategy_release=expected_strategy_release, + expected_command_binding=expected_command_binding, + ) + + +__all__ = ( + "IBKR_PAPER_EXECUTION_INTENT_SCHEMA_VERSION", + "consume_due_paper_execution_commands", + "resolve_paper_execution_command_consumer_enabled", +) diff --git a/docs/ibkr_paper_execution_command_consumer.md b/docs/ibkr_paper_execution_command_consumer.md new file mode 100644 index 0000000..37db272 --- /dev/null +++ b/docs/ibkr_paper_execution_command_consumer.md @@ -0,0 +1,38 @@ +# IBKR isolated paper command consumer + +`POST /paper-command-consumer` is an explicit, default-disabled verifier for +delayed paper execution commands. It is not connected to `/run`, `/dry-run`, +Cloud Scheduler, or the ordinary rebalance/order path. + +The consumer uses the shared QuantPlatformKit lifecycle: release attestation, +exact platform/account/strategy binding, create-only claim/events, immutable +paper-risk receipt, and enforced runtime gate. It then reads the IBKR +`portfolio()` market-value snapshot and current quotes to create audit-only +proposals. It has no execution-port or order-submission import. + +## Required isolation + +All conditions must be true before it reads the Gateway: + +- `IBKR_PAPER_EXECUTION_COMMAND_CONSUMER_ENABLED=true` +- `RUNTIME_TARGET_ENABLED=false` +- `IBKR_DRY_RUN_ONLY=true` +- `IBKR_GATEWAY_MODE=paper` +- `RUNTIME_TARGET_JSON.execution_mode=paper` +- `CASH_ONLY_EXECUTION=true` +- `IBKR_EXECUTION_COMMAND_CLOUD_URI` (or `IBKR_EXECUTION_COMMAND_DIR`) is set + +The expected command binding is taken from the runtime target, never from the +command: `platform=ibkr`, plus its `account_scope` and `strategy_profile`. +Any mismatch is rejected before the consumer opens the Gateway. + +## Reconciliation behavior + +The consumer fails closed if the selected IBKR account lacks a current market +value, market-currency cash value, valid quote, exactly matching managed symbol +set, or a reconciled cash-plus-positions total. It records a rejection or +reconciliation-required event; it does not infer a quantity, lower leverage, +or submit an order. + +Disable the flag again after a manual verification run. Moving beyond paper +evidence requires separate reviewed release-readiness and live rollout. diff --git a/main.py b/main.py index 18977d3..48c2f56 100644 --- a/main.py +++ b/main.py @@ -49,6 +49,7 @@ publish_strategy_plugin_alerts as dispatch_strategy_plugin_alerts, ) from quant_platform_kit.common.runtime_assembly import build_runtime_assembly +from quant_platform_kit.common.execution_commands import build_execution_command_store_from_env from quant_platform_kit.common.strategy_release import build_runtime_loaded_receipt from quant_platform_kit.common.runtime_reports import ( append_runtime_report_error, @@ -79,6 +80,7 @@ ) from application.ibkr_portfolio import ( IBKRPortfolioSnapshotUnavailableError, + fetch_reconciled_paper_portfolio_snapshot, fetch_portfolio_snapshot, ) from application.execution_service import ( @@ -88,6 +90,10 @@ ) from application.paper_liquidation_service import execute_paper_liquidation from application.paper_execution_admission import resolve_paper_execution_admission_enabled +from application.paper_execution_command_consumer import ( + consume_due_paper_execution_commands, + resolve_paper_execution_command_consumer_enabled, +) from runtime_logging import build_run_id, emit_runtime_log, extract_cloud_trace from runtime_config_support import ( EXECUTION_BACKEND_GATEWAY, @@ -1127,6 +1133,153 @@ def build_portfolio_snapshot(ib): ) +def _paper_command_consumer_session_date() -> str: + return datetime.now(ZoneInfo(MARKET_TIMEZONE)).date().isoformat() + + +def _paper_command_consumer_runtime_is_isolated() -> bool: + """Require a disabled, paper-only runtime before any Gateway read occurs.""" + + runtime_target = getattr(RUNTIME_SETTINGS, "runtime_target", None) + return bool( + runtime_target is not None + and bool(getattr(RUNTIME_SETTINGS, "dry_run_only", False)) + and not bool(getattr(RUNTIME_SETTINGS, "runtime_target_enabled", True)) + and str(getattr(RUNTIME_SETTINGS, "ib_gateway_mode", "") or "").strip().lower() == "paper" + and str(getattr(runtime_target, "execution_mode", "") or "").strip().lower() == "paper" + and bool(CASH_ONLY_EXECUTION) + ) + + +def run_paper_execution_command_consumer() -> dict[str, object]: + """Explicitly reconcile due commands without creating a broker order. + + This isolated route deliberately avoids the normal rebalance and execution + adapters. It opens the Gateway only after command release and delivery + binding checks pass, then reads portfolio and quote evidence to simulate + what would have been submitted. + """ + + if not _paper_command_consumer_runtime_is_isolated(): + raise RuntimeError( + "paper command consumer requires disabled runtime target, IBKR_DRY_RUN_ONLY=true, " + "paper Gateway mode, paper runtime target, and CASH_ONLY_EXECUTION=true" + ) + if not resolve_paper_execution_command_consumer_enabled( + env_reader=os.getenv, + dry_run_only=bool(RUNTIME_SETTINGS.dry_run_only), + ): + raise RuntimeError("paper command consumer is not enabled") + + runtime_target = RUNTIME_SETTINGS.runtime_target + expected_release = getattr(runtime_target, "strategy_release", None) + expected_binding = { + "platform": "ibkr", + "account_scope": str(getattr(runtime_target, "account_scope", "") or "unknown"), + "strategy_profile": str(getattr(runtime_target, "strategy_profile", "") or "unknown"), + } + store = build_execution_command_store_from_env( + platform_env_prefix="IBKR", + env_reader=os.getenv, + project_id=PROJECT_ID, + ) + if not store.cloud_prefix_uri and not store.local_dir: + raise RuntimeError("IBKR paper command consumer requires an execution command store") + + reporting_adapters = build_composer(dry_run_only_override=True).build_reporting_adapters() + log_context = reporting_adapters.build_log_context( + trace_header=request.headers.get("X-Cloud-Trace-Context"), + ) + report = reporting_adapters.build_report(log_context) + ib = None + quote_cache: dict[str, object] = {} + + def _ib(): + nonlocal ib + if ib is None: + # This uses a dry-run-only adapter; its permission validation does + # not perform the normal live what-if order check. + ib = build_broker_adapters(dry_run_only_override=True).connect_ib() + return ib + + def _load_portfolio(): + return fetch_reconciled_paper_portfolio_snapshot( + _ib(), + account_ids=ACCOUNT_IDS, + currency=MARKET_CURRENCY, + ) + + def _load_quote(symbol: str): + normalized = str(symbol or "").strip().upper() + if normalized not in quote_cache: + quotes = fetch_market_quote_snapshots(_ib(), (normalized,)) + quote = quotes.get(normalized) + if quote is None: + raise IBKRPortfolioSnapshotUnavailableError( + "IBKR paper command reconciliation is missing a current quote." + ) + quote_cache[normalized] = quote + return quote_cache[normalized] + + try: + reporting_adapters.log_event( + log_context, + "paper_execution_command_consumer_started", + message="Starting isolated read-only IBKR paper command consumer", + ) + result = consume_due_paper_execution_commands( + store=store, + as_of_session=_paper_command_consumer_session_date(), + claimant=str(os.getenv("K_SERVICE") or "ibkr-paper-command-consumer"), + portfolio_loader=_load_portfolio, + quote_loader=_load_quote, + managed_symbols=resolve_reporting_managed_symbols(), + runtime_release_receipt=build_runtime_loaded_receipt( + strategy_release=expected_release, + ), + expected_strategy_release=expected_release, + expected_command_binding=expected_binding, + ) + finalize_runtime_report( + report, + status="ok" if result.get("status") == "ok" else "skipped", + summary={"paper_execution_command_consumer": result}, + ) + reporting_adapters.log_event( + log_context, + "paper_execution_command_consumer_completed", + message="IBKR paper command consumer completed", + result_status=result.get("status"), + commands_count=len(tuple(result.get("commands") or ())), + ) + return result + except Exception as exc: + append_runtime_report_error( + report, + stage="paper_execution_command_consumer", + message=str(exc), + error_type=type(exc).__name__, + ) + finalize_runtime_report(report, status="error") + reporting_adapters.log_event( + log_context, + "paper_execution_command_consumer_failed", + message="IBKR paper command consumer failed", + severity="ERROR", + error_type=type(exc).__name__, + ) + raise + finally: + try: + report_path = reporting_adapters.persist_execution_report(report) + print(f"execution_report {report_path}", flush=True) + except Exception as persist_exc: + print(f"failed to persist execution report: {persist_exc}", flush=True) + disconnect_fn = getattr(ib, "disconnect", None) + if callable(disconnect_fn): + disconnect_fn() + + def get_market_prices(ib, symbols): return build_broker_adapters().get_market_prices(ib, symbols) @@ -1793,6 +1946,17 @@ def handle_dry_run(): return _handle_dry_run_with_deadline() +@app.route("/paper-command-consumer", methods=["POST"]) +def handle_paper_execution_command_consumer(): + """Manual-only endpoint for paper command reconciliation evidence.""" + + try: + result = run_paper_execution_command_consumer() + except Exception as exc: + return _handle_route_runtime_error(exc, route_label="POST /paper-command-consumer") + return json.dumps(result, ensure_ascii=False), 200, {"Content-Type": "application/json"} + + @app.route("/probe", methods=["POST", "GET"]) def handle_probe(): return _route_with_runtime_error_fallback(_handle_probe) diff --git a/tests/test_paper_execution_command_consumer.py b/tests/test_paper_execution_command_consumer.py new file mode 100644 index 0000000..28da9a9 --- /dev/null +++ b/tests/test_paper_execution_command_consumer.py @@ -0,0 +1,238 @@ +from __future__ import annotations + +import hashlib +import json +from datetime import datetime, timezone +from pathlib import Path +from types import SimpleNamespace + +import pytest + +from application.ibkr_portfolio import ( + IBKRPortfolioSnapshotUnavailableError, + fetch_reconciled_paper_portfolio_snapshot, +) +from application.paper_execution_command_consumer import ( + IBKR_PAPER_EXECUTION_INTENT_SCHEMA_VERSION, + consume_due_paper_execution_commands, + resolve_paper_execution_command_consumer_enabled, +) +from quant_platform_kit.common.execution_commands import ExecutionCommand, ExecutionCommandState, ExecutionCommandStore +from quant_platform_kit.common.models import PortfolioSnapshot, Position, QuoteSnapshot +from quant_platform_kit.common.paper_execution_admission import ( + PAPER_RISK_ADMISSION_RECEIPT_INTENT_FIELD, + build_paper_risk_admission_receipt, +) +from quant_platform_kit.common.strategy_release import build_runtime_loaded_receipt + + +def _release() -> dict[str, str]: + return { + "release_id": "soxl-p2-v3.20260824", + "manifest_sha256": "a" * 64, + "strategy_revision": "soxl-p2-v3", + "config_sha256": "b" * 64, + "risk_policy_sha256": "c" * 64, + "evidence_sha256": "d" * 64, + "plugin_bundle_sha256": "e" * 64, + "effective_session": "2026-08-25", + } + + +def _command(*, platform: str = "ibkr") -> ExecutionCommand: + release = _release() + intent = { + "schema_version": IBKR_PAPER_EXECUTION_INTENT_SCHEMA_VERSION, + "target_mode": "value", + "targets": {"SOXL": 300.0, "BOXX": 100.0}, + "strategy_symbols": ["SOXL", "BOXX"], + "strategy_release": release, + } + decision_digest = hashlib.sha256( + json.dumps(intent, ensure_ascii=True, separators=(",", ":"), sort_keys=True).encode("utf-8") + ).hexdigest() + intent[PAPER_RISK_ADMISSION_RECEIPT_INTENT_FIELD] = build_paper_risk_admission_receipt( + strategy_profile="soxl_soxx_trend_income", + release_id=release["release_id"], + risk_policy_sha256=release["risk_policy_sha256"], + decision_digest=decision_digest, + effective_session="2026-08-25", + disposition="allow_new_risk", + reason_codes=(), + ).to_dict() + return ExecutionCommand.from_decision( + platform=platform, + account_scope="paper", + strategy_profile="soxl_soxx_trend_income", + execution_mode="paper", + signal_date="2026-08-24", + effective_date="2026-08-25", + execution_timing_contract="next_trading_day", + decision_digest=decision_digest, + intent=intent, + ) + + +def _portfolio() -> PortfolioSnapshot: + return PortfolioSnapshot( + as_of=datetime(2026, 8, 25, tzinfo=timezone.utc), + total_equity=1_000.0, + cash_balance=800.0, + buying_power=800.0, + positions=( + Position( + symbol="SOXL", + quantity=20.0, + market_value=200.0, + average_cost=8.0, + currency="USD", + ), + ), + metadata={"market_currency_cash": 800.0}, + ) + + +def _quote(symbol: str) -> QuoteSnapshot: + return QuoteSnapshot( + symbol=symbol, + as_of=datetime(2026, 8, 25, tzinfo=timezone.utc), + last_price=10.0, + ) + + +def _binding() -> dict[str, str]: + return { + "platform": "ibkr", + "account_scope": "paper", + "strategy_profile": "soxl_soxx_trend_income", + } + + +def test_consumer_fills_reconciled_paper_command_without_execution_adapter(tmp_path: Path) -> None: + store = ExecutionCommandStore(local_dir=tmp_path) + command = _command() + assert store.enqueue(command) + + result = consume_due_paper_execution_commands( + store=store, + as_of_session="2026-08-25", + claimant="ibkr-paper-command-consumer", + portfolio_loader=_portfolio, + quote_loader=_quote, + managed_symbols=("SOXL", "BOXX"), + runtime_release_receipt=build_runtime_loaded_receipt(strategy_release=_release()), + expected_strategy_release=_release(), + expected_command_binding=_binding(), + ) + + assert result["status"] == "ok" + assert result["commands"][0]["status"] == "filled" + assert store.current_state(command) is ExecutionCommandState.FILLED + proposals = store.events(command)[1].details["proposals"] + assert [proposal["exposure_effect"] for proposal in proposals] == ["increases", "increases"] + assert all("order" not in proposal["details"] for proposal in proposals) + + +def test_consumer_rejects_cross_platform_command_before_portfolio_read(tmp_path: Path) -> None: + store = ExecutionCommandStore(local_dir=tmp_path) + command = _command(platform="schwab") + assert store.enqueue(command) + reads = {"portfolio": 0, "quote": 0} + + def portfolio_loader(): + reads["portfolio"] += 1 + return _portfolio() + + def quote_loader(symbol: str): + reads["quote"] += 1 + return _quote(symbol) + + result = consume_due_paper_execution_commands( + store=store, + as_of_session="2026-08-25", + claimant="ibkr-paper-command-consumer", + portfolio_loader=portfolio_loader, + quote_loader=quote_loader, + managed_symbols=("SOXL", "BOXX"), + runtime_release_receipt=build_runtime_loaded_receipt(strategy_release=_release()), + expected_strategy_release=_release(), + expected_command_binding=_binding(), + ) + + assert result["commands"][0]["status"] == "rejected" + assert reads == {"portfolio": 0, "quote": 0} + assert store.current_state(command) is ExecutionCommandState.REJECTED + + +def test_consumer_flag_cannot_be_enabled_outside_dry_run() -> None: + with pytest.raises(RuntimeError, match="IBKR_DRY_RUN_ONLY=true"): + resolve_paper_execution_command_consumer_enabled( + env_reader=lambda *_args: "true", + dry_run_only=False, + ) + + +def test_reconciled_snapshot_uses_current_market_value_not_cost_basis() -> None: + class FakeIB: + def portfolio(self, account: str): + assert account == "DU123" + return [ + SimpleNamespace( + account="DU123", + contract=SimpleNamespace(symbol="SOXL", currency="USD", conId=1), + position=20.0, + marketValue=200.0, + averageCost=8.0, + ) + ] + + def accountValues(self): + return [ + SimpleNamespace( + account="DU123", + currency="USD", + tag="CashBalance", + value="800", + ) + ] + + snapshot = fetch_reconciled_paper_portfolio_snapshot( + FakeIB(), + account_ids=("DU123",), + currency="USD", + ) + + assert snapshot.positions[0].market_value == 200.0 + assert snapshot.positions[0].average_cost == 8.0 + assert snapshot.total_equity == 1_000.0 + + +def test_reconciled_snapshot_requires_current_market_value() -> None: + class IncompleteIB: + def portfolio(self, _account: str): + return [ + SimpleNamespace( + account="DU123", + contract=SimpleNamespace(symbol="SOXL", currency="USD", conId=1), + position=20.0, + marketValue=None, + averageCost=8.0, + ) + ] + + def accountValues(self): + return [ + SimpleNamespace( + account="DU123", + currency="USD", + tag="CashBalance", + value="800", + ) + ] + + with pytest.raises(IBKRPortfolioSnapshotUnavailableError, match="market values"): + fetch_reconciled_paper_portfolio_snapshot( + IncompleteIB(), + account_ids=("DU123",), + currency="USD", + ) diff --git a/tests/test_request_handling.py b/tests/test_request_handling.py index d4b0a47..024c159 100644 --- a/tests/test_request_handling.py +++ b/tests/test_request_handling.py @@ -15,6 +15,7 @@ def test_cloud_run_route_contracts_are_registered(strategy_module): assert route_methods(strategy_module) == { "/run": ["GET", "POST"], "/dry-run": ["GET", "POST"], + "/paper-command-consumer": ["POST"], "/probe": ["GET", "POST"], "/monitor-dispatch": ["GET", "POST"], "/health": ["GET"], @@ -23,6 +24,66 @@ def test_cloud_run_route_contracts_are_registered(strategy_module): } +def test_paper_command_consumer_does_not_open_gateway_without_due_command( + strategy_module_factory, + monkeypatch, + tmp_path, +): + strategy_module = strategy_module_factory( + IBKR_DRY_RUN_ONLY="true", + RUNTIME_TARGET_ENABLED="false", + IBKR_PAPER_EXECUTION_COMMAND_CONSUMER_ENABLED="true", + IBKR_EXECUTION_COMMAND_DIR=str(tmp_path / "commands"), + IB_ACCOUNT_GROUP_CONFIG_JSON=( + '{"groups":{"default":{"ib_gateway_instance_name":"127.0.0.1",' + '"ib_gateway_mode":"paper","ib_client_id":1,"account_ids":["DU123"]}}}' + ), + RUNTIME_TARGET_JSON=json.dumps( + { + "platform_id": "ibkr", + "strategy_profile": "global_etf_rotation", + "account_scope": "paper", + "dry_run_only": True, + "execution_mode": "paper", + }, + separators=(",", ":"), + ), + ) + observed = {"gateway": 0} + + class ReportingAdapters: + def build_log_context(self, **_kwargs): + return object() + + def build_report(self, _context): + return {} + + def log_event(self, *_args, **_kwargs): + return None + + def persist_execution_report(self, _report): + return "/tmp/report.json" + + monkeypatch.setattr( + strategy_module, + "build_composer", + lambda **_kwargs: types.SimpleNamespace(build_reporting_adapters=lambda: ReportingAdapters()), + ) + + def fail_if_gateway_opens(**_kwargs): + observed["gateway"] += 1 + raise AssertionError("no due command must not open IBKR Gateway") + + monkeypatch.setattr(strategy_module, "build_broker_adapters", fail_if_gateway_opens) + + with strategy_module.app.test_request_context("/paper-command-consumer", method="POST"): + body, status, _headers = strategy_module.handle_paper_execution_command_consumer() + + assert status == 200 + assert json.loads(body)["status"] == "blocked" + assert observed["gateway"] == 0 + + def test_health_route_returns_ok(strategy_module): with strategy_module.app.test_request_context("/health", method="GET"): body, status = strategy_module.health()