feat: gate IBKR PAPER admission

application/execution_service.py

@@ -13,6 +13,7 @@
 from typing import Any
 
 import pandas as pd
+from application.paper_execution_admission import evaluate_ibkr_paper_execution_admission
 try:
     from quant_platform_kit.common.cash_sweep import should_sell_cash_sweep_to_fund_whole_share_buy
 except ImportError:  # pragma: no cover - compatibility with older pinned shared wheels
@@ -1416,6 +1417,9 @@ def execute_rebalance(
     execution_lock_dir=None,
     return_summary=False,
     cash_only_execution=True,
+    paper_execution_admission_enabled=False,
+    runtime_release_receipt=None,
+    expected_strategy_release=None,
 ):
     """Execute trades to reach target weights."""
     del target_weights
@@ -1496,6 +1500,7 @@ def record_quote_snapshot(symbol, snapshot) -> None:
         "snapshot_price_fallback_symbols": [],
         "snapshot_price_fallback_count": 0,
         "lock_path": None,
+        "paper_execution_admission": {},
     }
     equity = float(account_values.get("equity", 0) or 0.0)
     cash_only_deleverage_mode = bool(signal_metadata.get("cash_only_deleverage_mode"))
@@ -1726,6 +1731,26 @@ def append_small_account_allocation_drift_notes():
 
     target_hash = _build_target_hash(target_weights)
     execution_summary["target_vs_current"] = _build_target_diff_rows(target_weights, current_mv, equity)
+    if paper_execution_admission_enabled:
+        paper_admission = evaluate_ibkr_paper_execution_admission(
+            signal_metadata=signal_metadata,
+            strategy_profile=strategy_profile,
+            account_scope=account_group,
+            positions=positions,
+            prices=prices,
+            target_market_values=target_mv,
+            option_order_intents=option_order_intents,
+            runtime_release_receipt=runtime_release_receipt,
+            expected_strategy_release=expected_strategy_release,
+        )
+        execution_summary["paper_execution_admission"] = paper_admission
+        if not paper_admission["broker_write_allowed"]:
+            reason = "paper_execution_admission_blocked"
+            execution_summary["execution_status"] = "blocked"
+            execution_summary["no_op_reason"] = reason
+            execution_summary["skipped_reasons"].append(reason)
+            trade_logs.append(translator("failed", reason=reason))
+            return _finalize_result(trade_logs, execution_summary, return_summary=return_summary)
     if equity > 0:
         current_safe_haven_mv = current_mv.get(safe_haven_symbol, 0.0) if safe_haven_symbol else 0.0
         execution_summary["current_safe_haven_weight"] = float(current_safe_haven_mv / equity)

application/paper_execution_admission.py

@@ -0,0 +1,276 @@
+"""Fail-closed PAPER admission for IBKR's ordinary rebalance path.
+
+This adapter deliberately has no broker dependency.  A strategy/control-plane
+producer supplies an immutable QPK ``ExecutionCommand`` in signal metadata;
+this module verifies its embedded deterministic-risk receipt and the current
+runtime release before the normal rebalance service can submit any order.
+
+Exposure is classified from reconciled quantities and the quotes used by this
+cycle.  In particular, a buy/sell label is never treated as evidence that an
+order increases or reduces risk.
+"""
+
+from __future__ import annotations
+
+import math
+from collections.abc import Mapping, Sequence
+from typing import Any
+
+from quant_platform_kit.common.execution_commands import ExecutionCommand
+from quant_platform_kit.common.paper_execution_admission import (
+    PAPER_RISK_ADMISSION_RECEIPT_INTENT_FIELD,
+    PaperRiskAdmissionReceipt,
+    evaluate_paper_execution_admission,
+)
+from quant_platform_kit.common.runtime_command_gate import (
+    RuntimeCommandAction,
+    RuntimeCommandExposureEffect,
+    RuntimeCommandGateEnforcement,
+    RuntimeCommandGatePolicy,
+    evaluate_runtime_command_gate,
+)
+
+
+PAPER_EXECUTION_ADMISSION_SCHEMA_VERSION = "ibkr.paper_execution_admission.v1"
+PAPER_EXECUTION_COMMAND_SIGNAL_FIELD = "paper_execution_command"
+_PAPER_ADMISSION_GATE_POLICY = RuntimeCommandGatePolicy(
+    enforcement=RuntimeCommandGateEnforcement.ENFORCE,
+)
+_EPSILON = 0.01
+
+
+def resolve_paper_execution_admission_enabled(
+    *,
+    env_reader,
+    dry_run_only: bool,
+    execution_mode: object,
+) -> bool:
+    """Resolve the opt-in flag and reject every non-PAPER configuration."""
+
+    raw_value = str(env_reader("IBKR_PAPER_EXECUTION_ADMISSION_ENABLED", "") or "").strip().lower()
+    enabled = raw_value in {"1", "true", "t", "yes", "y", "on"}
+    if not enabled:
+        return False
+    normalized_mode = str(execution_mode or "").strip().lower().replace("-", "_")
+    if dry_run_only or normalized_mode != "paper":
+        raise RuntimeError(
+            "IBKR_PAPER_EXECUTION_ADMISSION_ENABLED is only supported for ordinary execution_mode=paper"
+        )
+    return True
+
+
+def _append_finding(findings: list[str], finding: str) -> None:
+    if finding not in findings:
+        findings.append(finding)
+
+
+def _finite_nonnegative(value: object) -> float | None:
+    try:
+        number = float(value)
+    except (TypeError, ValueError):
+        return None
+    if not math.isfinite(number) or number < 0.0:
+        return None
+    return number
+
+
+def _effective_session(signal_metadata: Mapping[str, object]) -> str | None:
+    value = signal_metadata.get("effective_date") or signal_metadata.get("trade_date")
+    text = str(value or "").strip()
+    return text[:10] or None
+
+
+def _load_command(signal_metadata: Mapping[str, object]) -> tuple[ExecutionCommand | None, list[str]]:
+    raw_command = signal_metadata.get(PAPER_EXECUTION_COMMAND_SIGNAL_FIELD)
+    if not isinstance(raw_command, Mapping):
+        return None, ["paper_risk_admission_receipt_missing"]
+    try:
+        return ExecutionCommand.from_dict(raw_command), []
+    except (TypeError, ValueError):
+        return None, ["command_digest_mismatch"]
+
+
+def _command_contract_findings(
+    command: ExecutionCommand | None,
+    *,
+    strategy_profile: object,
+    account_scope: object,
+    effective_session: str | None,
+) -> list[str]:
+    if command is None:
+        return []
+    findings: list[str] = []
+    if command.platform != "ibkr":
+        _append_finding(findings, "durable_event_history_invalid")
+    if command.execution_mode != "paper":
+        _append_finding(findings, "paper_execution_mode_invalid")
+    if command.strategy_profile != str(strategy_profile or "").strip().lower():
+        _append_finding(findings, "command_digest_mismatch")
+    if command.account_scope != str(account_scope or "").strip().lower():
+        _append_finding(findings, "durable_event_history_invalid")
+    if not effective_session or command.effective_date != effective_session:
+        _append_finding(findings, "signal_timing_invalid")
+    return findings
+
+
+def _exposure_facts(
+    *,
+    positions: Mapping[str, Mapping[str, object]],
+    prices: Mapping[str, object],
+    target_market_values: Mapping[str, object],
+) -> tuple[tuple[dict[str, object], ...], tuple[str, ...]]:
+    """Classify target changes using position quantities and current quotes."""
+
+    normalized_positions = {
+        str(symbol).strip().upper(): details
+        for symbol, details in positions.items()
+        if str(symbol).strip() and isinstance(details, Mapping)
+    }
+    normalized_prices = {
+        str(symbol).strip().upper(): value
+        for symbol, value in prices.items()
+        if str(symbol).strip()
+    }
+    normalized_targets = {
+        str(symbol).strip().upper(): value
+        for symbol, value in target_market_values.items()
+        if str(symbol).strip()
+    }
+    facts: list[dict[str, object]] = []
+    findings: list[str] = []
+    symbols = sorted(set(normalized_positions) | set(normalized_targets))
+    for symbol in symbols:
+        position = normalized_positions.get(symbol) or {}
+        quantity = _finite_nonnegative(position.get("quantity"))
+        target_value = _finite_nonnegative(normalized_targets.get(symbol, 0.0))
+        price = _finite_nonnegative(normalized_prices.get(symbol))
+        if quantity is None or target_value is None or price is None or price <= 0.0:
+            _append_finding(findings, "position_reconciliation_mismatch")
+            continue
+        current_value = quantity * price
+        before_exposure = abs(current_value)
+        after_exposure = abs(target_value)
+        exposure_delta = after_exposure - before_exposure
+        if exposure_delta > _EPSILON:
+            effect = RuntimeCommandExposureEffect.INCREASES
+        elif exposure_delta < -_EPSILON:
+            effect = RuntimeCommandExposureEffect.REDUCES
+        else:
+            effect = RuntimeCommandExposureEffect.NEUTRAL
+        facts.append(
+            {
+                "symbol": symbol,
+                "position_quantity": round(quantity, 8),
+                "quote_price": round(price, 8),
+                "current_market_value": round(current_value, 8),
+                "target_market_value": round(target_value, 8),
+                "exposure_effect": effect.value,
+            }
+        )
+    return tuple(facts), tuple(findings)
+
+
+def evaluate_ibkr_paper_execution_admission(
+    *,
+    signal_metadata: Mapping[str, object] | None,
+    strategy_profile: object,
+    account_scope: object,
+    positions: Mapping[str, Mapping[str, object]],
+    prices: Mapping[str, object],
+    target_market_values: Mapping[str, object],
+    option_order_intents: Sequence[Mapping[str, object]] = (),
+    runtime_release_receipt: Mapping[str, Any] | None,
+    expected_strategy_release: Any = None,
+) -> dict[str, object]:
+    """Return durable audit evidence and block if a PAPER broker write is unsafe.
+
+    The caller must invoke this after it has collected the current portfolio
+    and quotes, but before it invokes any submit adapter.
+    """
+
+    metadata = signal_metadata if isinstance(signal_metadata, Mapping) else {}
+    command, findings = _load_command(metadata)
+    effective_session = _effective_session(metadata)
+    for finding in _command_contract_findings(
+        command,
+        strategy_profile=strategy_profile,
+        account_scope=account_scope,
+        effective_session=effective_session,
+    ):
+        _append_finding(findings, finding)
+
+    paper_receipt: Mapping[str, object] | None = None
+    if command is not None:
+        raw_receipt = command.intent.get(PAPER_RISK_ADMISSION_RECEIPT_INTENT_FIELD)
+        if isinstance(raw_receipt, Mapping):
+            try:
+                paper_receipt = PaperRiskAdmissionReceipt.from_dict(raw_receipt).to_dict()
+            except (TypeError, ValueError):
+                # Invalid untrusted payloads must not be copied into reports.
+                paper_receipt = None
+        admission = evaluate_paper_execution_admission(
+            command=command,
+            expected_strategy_release=expected_strategy_release,
+        )
+        for finding in admission.integrity_findings:
+            _append_finding(findings, finding)
+    else:
+        admission = None
+
+    if option_order_intents:
+        # The ordinary equity target model has no reconciled option valuation
+        # contract yet.  Treating a side label as exposure evidence would be
+        # unsafe, so PAPER admission closes the whole cycle instead.
+        _append_finding(findings, "durable_event_history_invalid")
+
+    facts, fact_findings = _exposure_facts(
+        positions=positions,
+        prices=prices,
+        target_market_values=target_market_values,
+    )
+    for finding in fact_findings:
+        _append_finding(findings, finding)
+
+    effects = [RuntimeCommandExposureEffect(fact["exposure_effect"]) for fact in facts]
+    if not effects:
+        effects = [RuntimeCommandExposureEffect.NEUTRAL]
+    gate_receipts = []
+    for effect in effects:
+        decision = evaluate_runtime_command_gate(
+            action=RuntimeCommandAction.SUBMIT,
+            exposure_effect=effect,
+            command=command,
+            as_of_session=effective_session,
+            runtime_release_receipt=runtime_release_receipt,
+            expected_strategy_release=expected_strategy_release,
+            integrity_findings=findings,
+            policy=_PAPER_ADMISSION_GATE_POLICY,
+        )
+        gate_receipts.append(decision.to_receipt())
+
+    return {
+        "schema_version": PAPER_EXECUTION_ADMISSION_SCHEMA_VERSION,
+        "enabled": True,
+        "command_id": command.command_id if command is not None else None,
+        "decision_digest": command.decision_digest if command is not None else None,
+        "effective_session": effective_session,
+        "risk_admission_receipt": dict(paper_receipt or {}),
+        "risk_admission_receipt_sha256": (
+            admission.receipt_sha256 if admission is not None else None
+        ),
+        "risk_disposition": admission.disposition.value if admission is not None else "halted",
+        "integrity_findings": list(dict.fromkeys(findings)),
+        "exposure_facts": list(facts),
+        "runtime_command_gate_receipts": gate_receipts,
+        "broker_write_allowed": bool(gate_receipts) and all(
+            bool(receipt["broker_write_allowed"]) for receipt in gate_receipts
+        ),
+    }
+
+
+__all__ = [
+    "PAPER_EXECUTION_ADMISSION_SCHEMA_VERSION",
+    "PAPER_EXECUTION_COMMAND_SIGNAL_FIELD",
+    "evaluate_ibkr_paper_execution_admission",
+    "resolve_paper_execution_admission_enabled",
+]

application/reconciliation_service.py

@@ -85,6 +85,7 @@ def build_reconciliation_record(
         "current_safe_haven_weight": execution_summary.get("current_safe_haven_weight"),
         "price_source_mode": execution_summary.get("price_source_mode"),
         "quote_snapshot": execution_summary.get("quote_snapshot") or {},
+        "paper_execution_admission": execution_summary.get("paper_execution_admission") or {},
         "snapshot_price_fallback_used": execution_summary.get("snapshot_price_fallback_used"),
         "snapshot_price_fallback_count": execution_summary.get("snapshot_price_fallback_count"),
         "snapshot_price_fallback_symbols": execution_summary.get("snapshot_price_fallback_symbols") or [],

application/runtime_broker_adapters.py

@@ -61,6 +61,9 @@ class IBKRRuntimeBrokerAdapters:
     printer: Any = print
     refresh_host_fn: Any = None
     trading_permission_probe_fn: Any = None
+    paper_execution_admission_enabled: bool = False
+    runtime_release_receipt: Any = None
+    expected_strategy_release: Any = None
 
     def validate_configured_accounts(self, ib):
         if not self.account_ids:
@@ -294,6 +297,9 @@ def execute_rebalance(
             sell_settle_delay_sec=self.sell_settle_delay_sec,
             return_summary=True,
             cash_only_execution=self.cash_only_execution,
+            paper_execution_admission_enabled=self.paper_execution_admission_enabled,
+            runtime_release_receipt=self.runtime_release_receipt,
+            expected_strategy_release=self.expected_strategy_release,
         )
 
     def format_liquidation_orders(self, orders) -> str:
@@ -390,6 +396,9 @@ def build_runtime_broker_adapters(
     printer=print,
     refresh_host_fn=None,
     trading_permission_probe_fn=None,
+    paper_execution_admission_enabled: bool = False,
+    runtime_release_receipt=None,
+    expected_strategy_release=None,
 ) -> IBKRRuntimeBrokerAdapters:
     return IBKRRuntimeBrokerAdapters(
         host_resolver=host_resolver,
@@ -433,4 +442,7 @@ def build_runtime_broker_adapters(
         printer=printer,
         refresh_host_fn=refresh_host_fn,
         trading_permission_probe_fn=trading_permission_probe_fn,
+        paper_execution_admission_enabled=bool(paper_execution_admission_enabled),
+        runtime_release_receipt=runtime_release_receipt,
+        expected_strategy_release=expected_strategy_release,
     )