feat: add isolated IBKR paper command consumer

.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

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]),
+                )
+            ),
+        },
+    )

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",
+)

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.