fix: fail closed IBKR terminal order outcomes

Co-Authored-By: Codex <noreply@openai.com>

Author: Pigbibi

application/broker_reconciliation.py

@@ -17,6 +17,7 @@
 import os
 from typing import Any
 
+from ib_insync import OrderStatus
 from quant_platform_kit.common.broker_reconciliation import (
     BrokerReconciliationEvidence,
     BrokerReconciliationFinding,
@@ -51,6 +52,14 @@ class IBKRReconciliationReadError(RuntimeError):
     "TotalCashBalance",
     "SettledCash",
 )
+_ORDER_EVENT_DIGEST_FIELDS = frozenset(
+    {
+        "order_key",
+        "order_identity",
+        "cumulative_filled_quantity",
+        "status_transitions",
+    }
+)
 
 
 def _text(value: object) -> str:
@@ -120,21 +129,47 @@ def build_ibkr_order_key(
     return "ibkr-order-v1-" + hashlib.sha256(encoded).hexdigest()
 
 
-def normalize_ibkr_order_state(status: object) -> str:
+def normalize_ibkr_order_state(status: object, *, filled_quantity: object = 0.0) -> str:
+    """Map pinned ib_insync order states to local reconciliation states."""
+
     normalized = _text(status)
-    if normalized == "Filled":
+    try:
+        has_filled_quantity = float(filled_quantity or 0.0) > 0.0
+    except (TypeError, ValueError):
+        has_filled_quantity = False
+    if normalized == OrderStatus.Filled:
         return "filled"
-    if normalized in {"PartiallyFilled", "Partial"}:
-        return "partially_filled"
-    if normalized in {"PendingSubmit", "ApiPending", "ApiPendingSubmit", "Submitted", "PreSubmitted"}:
-        return "submitted"
-    if normalized in {"Cancelled", "ApiCancelled"}:
+    if normalized in {OrderStatus.Cancelled, OrderStatus.ApiCancelled}:
         return "cancelled"
-    if normalized in {"Inactive", "Rejected"}:
-        return "rejected"
+    if normalized in OrderStatus.ActiveStates or normalized == OrderStatus.PendingCancel:
+        if has_filled_quantity:
+            return "partially_filled"
+        return "pending_cancel" if normalized == OrderStatus.PendingCancel else "submitted"
+    if normalized == OrderStatus.Inactive:
+        return "inactive"
     return "unknown"
 
 
+def _without_order_event_digest_fields(value: object) -> object:
+    if isinstance(value, Mapping):
+        return {
+            key: _without_order_event_digest_fields(item)
+            for key, item in value.items()
+            if key not in _ORDER_EVENT_DIGEST_FIELDS
+        }
+    if isinstance(value, tuple):
+        return tuple(_without_order_event_digest_fields(item) for item in value)
+    if isinstance(value, list):
+        return [_without_order_event_digest_fields(item) for item in value]
+    return value
+
+
+def calculate_legacy_reconciliation_observation_sha256(value: object) -> str:
+    """Keep frozen reconciliation baselines independent of new order-event metadata."""
+
+    return calculate_broker_observation_sha256(_without_order_event_digest_fields(value))
+
+
 def _number(value: object, *, field_name: str) -> float:
     try:
         return float(value)
@@ -558,8 +593,8 @@ def build_reconciliation_candidate(
     account_scope_sha256 = calculate_broker_observation_sha256(observations.account_scope)
     positions_sha256 = calculate_broker_observation_sha256(observations.positions)
     cash_sha256 = calculate_broker_observation_sha256(observations.cash)
-    open_orders_sha256 = calculate_broker_observation_sha256(observations.open_orders)
-    recent_executions_sha256 = calculate_broker_observation_sha256(observations.recent_executions)
+    open_orders_sha256 = calculate_legacy_reconciliation_observation_sha256(observations.open_orders)
+    recent_executions_sha256 = calculate_legacy_reconciliation_observation_sha256(observations.recent_executions)
     execution_state_store = build_execution_marker_store_from_env(
         platform_env_prefix="IBKR",
         env_reader=env_reader,
@@ -633,6 +668,7 @@ def matches(key: str, actual_digest: str) -> bool:
     "build_ibkr_order_identity",
     "build_ibkr_order_key",
     "build_reconciliation_candidate",
+    "calculate_legacy_reconciliation_observation_sha256",
     "collect_read_only_reconciliation_observations",
     "normalize_account_ids",
     "normalize_ibkr_order_state",

application/execution_service.py

@@ -254,8 +254,12 @@ def check_order_submitted(report, *, translator):
     """Check if order was accepted. DAY orders auto-expire at close if not filled."""
     order_id = report.broker_order_id
     status = report.status
+    order_state = normalize_ibkr_order_state(
+        status,
+        filled_quantity=getattr(report, "filled_quantity", 0.0),
+    )
 
-    if status == "Filled":
+    if order_state == "filled":
         return (
             True,
             translator(
@@ -267,7 +271,7 @@ def check_order_submitted(report, *, translator):
                 order_id=order_id,
             ),
         )
-    if status in {"PartiallyFilled", "Partial"}:
+    if order_state == "partially_filled":
         return (
             True,
             translator(
@@ -280,18 +284,11 @@ def check_order_submitted(report, *, translator):
                 order_id=order_id,
             ),
         )
-    if status in {"PendingSubmit", "ApiPending", "ApiPendingSubmit", "Submitted", "PreSubmitted"}:
+    if order_state in {"submitted", "pending_cancel"}:
         return True, f"✅ {translator('submitted', order_id=order_id, status=status)}"
     return False, f"❌ {translator('failed', reason=status)}"
 
 
-_FILLED_ORDER_STATUSES = frozenset({"Filled"})
-_PARTIALLY_FILLED_ORDER_STATUSES = frozenset({"PartiallyFilled", "Partial"})
-_PENDING_ORDER_STATUSES = frozenset(
-    {"PendingSubmit", "ApiPending", "ApiPendingSubmit", "Submitted", "PreSubmitted"}
-)
-
-
 def _ibkr_client_id(ib: Any) -> object:
     wrapper = getattr(ib, "wrapper", None)
     if wrapper is not None and getattr(wrapper, "clientId", None) is not None:
@@ -318,7 +315,13 @@ def _build_order_event_payload(
         "broker_order_id": order_id,
         "cumulative_filled_quantity": float(getattr(report, "filled_quantity", 0.0) or 0.0),
         "status_transitions": [
-            {"from": "created", "to": normalize_ibkr_order_state(status)},
+            {
+                "from": "created",
+                "to": normalize_ibkr_order_state(
+                    status,
+                    filled_quantity=getattr(report, "filled_quantity", 0.0),
+                ),
+            },
         ],
     }
     try:
@@ -353,14 +356,18 @@ def _record_order_outcome(
     make the rebalance appear complete.
     """
     normalized_status = str(status or "").strip()
+    order_state = normalize_ibkr_order_state(
+        normalized_status,
+        filled_quantity=order_payload.get("cumulative_filled_quantity", 0.0),
+    )
     prefix = "option_orders" if option_order else "orders"
-    if normalized_status in _FILLED_ORDER_STATUSES:
+    if order_state == "filled":
         execution_summary[f"{prefix}_filled"].append(order_payload)
         return "filled"
-    if normalized_status in _PARTIALLY_FILLED_ORDER_STATUSES:
+    if order_state == "partially_filled":
         execution_summary[f"{prefix}_partially_filled"].append(order_payload)
         return "partially_filled"
-    if normalized_status in _PENDING_ORDER_STATUSES:
+    if order_state in {"submitted", "pending_cancel"}:
         execution_summary[f"{prefix}_pending"].append(order_payload)
         return "pending"
     execution_summary[f"{prefix}_skipped"].append(
@@ -1378,10 +1385,11 @@ def _planned_buy_order_quantity(
 
 def _projected_sell_release_value_for_report(report, *, fallback_price=0.0, fallback_quantity=0.0) -> float:
     status = str(getattr(report, "status", "") or "")
-    if status not in {"Filled", "PartiallyFilled", "Partial"}:
-        return 0.0
     filled_quantity = float(getattr(report, "filled_quantity", 0.0) or 0.0)
-    if status == "Filled" and filled_quantity <= 0.0:
+    order_state = normalize_ibkr_order_state(status, filled_quantity=filled_quantity)
+    if order_state not in {"filled", "partially_filled"}:
+        return 0.0
+    if order_state == "filled" and filled_quantity <= 0.0:
         filled_quantity = float(getattr(report, "quantity", 0.0) or fallback_quantity or 0.0)
     if filled_quantity <= 0.0:
         return 0.0

application/rebalance_service.py

@@ -8,6 +8,7 @@
 import re
 
 from application.cycle_result import StrategyCycleResult
+from application.broker_reconciliation import normalize_ibkr_order_state
 from application.runtime_dependencies import IBKRRebalanceConfig, IBKRRebalanceRuntime
 from application.reconciliation_service import (
     build_reconciliation_record,
@@ -778,10 +779,10 @@ def _record_execution_outcome(
     marker_key: str,
     signal_metadata: dict,
     execution_summary,
-) -> None:
+) -> bool:
     store = getattr(config, "execution_state_store", None)
     if not store or not marker_key:
-        return
+        return False
     summary = dict(execution_summary or {})
     orders_by_key = {}
     for collection in (
@@ -799,8 +800,19 @@ def _record_execution_outcome(
         for order in summary.get(collection) or ():
             if isinstance(order, Mapping) and str(order.get("order_key") or "").strip():
                 orders_by_key[str(order["order_key"])] = dict(order)
+    if not orders_by_key:
+        return False
+    if any(
+        normalize_ibkr_order_state(
+            order.get("status"),
+            filled_quantity=order.get("cumulative_filled_quantity", 0.0),
+        )
+        not in {"filled", "cancelled"}
+        for order in orders_by_key.values()
+    ):
+        return False
     try:
-        store.record_outcome(
+        recorded = store.record_outcome(
             marker_key,
             metadata={
                 "schema_version": "ibkr_order_consumer_outcome.v1",
@@ -816,10 +828,10 @@ def _record_execution_outcome(
             },
         )
     except Exception as exc:
-        print(
-            f"Execution outcome write failed\nMarker: {marker_key}\n{type(exc).__name__}: {exc}",
-            flush=True,
-        )
+        raise RuntimeError("IBKR terminal execution outcome unavailable; refusing success") from exc
+    if recorded is not True:
+        raise RuntimeError("IBKR terminal execution outcome unavailable; refusing success")
+    return True
 
 
 def run_strategy_core(

tests/test_broker_reconciliation.py

@@ -9,8 +9,10 @@
     IBKRReconciliationReadError,
     build_ibkr_order_key,
     build_reconciliation_candidate,
+    calculate_legacy_reconciliation_observation_sha256,
     collect_read_only_reconciliation_observations,
 )
+from quant_platform_kit.common.broker_reconciliation import calculate_broker_observation_sha256
 from quant_platform_kit.common.live_continuity import runtime_target_fingerprint
 from quant_platform_kit.common.runtime_target import build_runtime_target
 
@@ -157,6 +159,33 @@ def test_manual_order_key_requires_perm_id() -> None:
         build_ibkr_order_key(account_id="U123", order_id=0)
 
 
+def test_order_event_metadata_does_not_change_legacy_reconciliation_digest() -> None:
+    legacy_open_order = {
+        "account": "U123",
+        "contract": {"symbol": "SOXL"},
+        "perm_id": "9001",
+        "action": "BUY",
+        "order_type": "LMT",
+        "total_quantity": 2.0,
+        "limit_price": 22.0,
+        "aux_price": 0.0,
+        "status": "Submitted",
+        "filled": 0.0,
+        "remaining": 2.0,
+    }
+    enriched_open_order = {
+        **legacy_open_order,
+        "order_key": "ibkr-order-v1-example",
+        "order_identity": {"account_scope_sha256": "digest", "client_id": "7", "order_id": "456"},
+        "cumulative_filled_quantity": 0.0,
+        "status_transitions": [{"from": "created", "to": "submitted"}],
+    }
+
+    assert calculate_legacy_reconciliation_observation_sha256((enriched_open_order,)) == (
+        calculate_broker_observation_sha256((legacy_open_order,))
+    )
+
+
 def test_cash_reconciliation_ignores_dynamic_margin_and_valuation_tags() -> None:
     def fetch_snapshot(_ib, *, dynamic_net_liquidation: float, dynamic_available_funds: float, **_kwargs):
         return SimpleNamespace(
@@ -411,3 +440,99 @@ def configured_env(name, default=None):
 
     assert candidate.permits_active_lkg is True
     assert candidate.recovery_blockers == ()
+
+
+def test_candidate_keeps_legacy_order_digests_after_order_event_wiring(tmp_path) -> None:
+    target = _frozen_runtime_target()
+
+    def empty_env(name, default=None):
+        return str(tmp_path) if name == "IBKR_EXECUTION_STATE_DIR" else default
+
+    legacy_observations = IBKRReconciliationObservations(
+        account_scope={"account_ids": ["U123"]},
+        account_identity_match=True,
+        positions=(),
+        cash=(),
+        open_orders=(
+            {
+                "account": "U123",
+                "contract": {"symbol": "SOXL"},
+                "perm_id": "9001",
+                "status": "Submitted",
+                "filled": 0.0,
+                "remaining": 2.0,
+            },
+        ),
+        recent_executions=(
+            {
+                "account": "U123",
+                "contract": {"symbol": "SOXL"},
+                "order_id": "456",
+                "execution_id": "exec-1",
+                "shares": 1.0,
+                "price": 21.5,
+            },
+        ),
+    )
+    seed = build_reconciliation_candidate(
+        observations=legacy_observations,
+        runtime_target=target,
+        platform_id="ibkr",
+        strategy_profile="soxl_soxx_trend_income",
+        account_group="LIVE",
+        project_id=None,
+        env_reader=empty_env,
+    )
+    expected = {
+        key: seed.evidence.to_dict()[key]
+        for key in (
+            "positions_sha256",
+            "cash_sha256",
+            "open_orders_sha256",
+            "recent_executions_sha256",
+            "local_execution_ledger_sha256",
+        )
+    }
+    enriched_observations = IBKRReconciliationObservations(
+        account_scope=legacy_observations.account_scope,
+        account_identity_match=legacy_observations.account_identity_match,
+        positions=legacy_observations.positions,
+        cash=legacy_observations.cash,
+        open_orders=(
+            {
+                **legacy_observations.open_orders[0],
+                "order_key": "ibkr-order-v1-example",
+                "order_identity": {"account_scope_sha256": "digest", "client_id": "7", "order_id": "456"},
+                "cumulative_filled_quantity": 0.0,
+            },
+        ),
+        recent_executions=(
+            {
+                **legacy_observations.recent_executions[0],
+                "order_key": "ibkr-order-v1-example",
+                "order_identity": {"account_scope_sha256": "digest", "client_id": "7", "order_id": "456"},
+                "cumulative_filled_quantity": 1.0,
+            },
+        ),
+    )
+
+    def configured_env(name, default=None):
+        if name == "IBKR_EXECUTION_STATE_DIR":
+            return str(tmp_path)
+        if name == "IBKR_RECONCILIATION_EXPECTED_DIGESTS_JSON":
+            import json
+
+            return json.dumps(expected)
+        return default
+
+    candidate = build_reconciliation_candidate(
+        observations=enriched_observations,
+        runtime_target=target,
+        platform_id="ibkr",
+        strategy_profile="soxl_soxx_trend_income",
+        account_group="LIVE",
+        project_id=None,
+        env_reader=configured_env,
+    )
+
+    assert candidate.permits_active_lkg is True