fix: fail closed IBKR terminal order outcomes

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

Author: Pigbibi

.github/workflows/ci.yml

@@ -134,4 +134,4 @@ jobs:
       - name: Run unit tests
         run: |
           set -euo pipefail
-          PYTHONPATH=. PYTEST_DISABLE_PLUGIN_AUTOLOAD=1 uv run --no-sync python -m pytest -q tests --ignore=tests/test_request_handling.py --ignore=tests/test_event_loop.py --ignore=tests/test_monitor_dispatcher.py --ignore=tests/test_notifications.py --ignore=tests/test_connect_timeout_alert.py || true
+          PYTHONPATH=. PYTEST_DISABLE_PLUGIN_AUTOLOAD=1 uv run --no-sync python -m pytest -q tests --ignore=tests/test_request_handling.py --ignore=tests/test_event_loop.py --ignore=tests/test_monitor_dispatcher.py --ignore=tests/test_notifications.py --ignore=tests/test_connect_timeout_alert.py

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/__init__.py

@@ -0,0 +1 @@
+"""Local test package to prevent third-party ``tests`` namespace shadowing."""