fix: require execution claim before IBKR submission (#477)

Co-authored-by: Codex <noreply@openai.com>

Author: Pigbibi

application/execution_service.py

@@ -1398,6 +1398,7 @@ def execute_rebalance(
     translator,
     strategy_symbols=None,
     signal_metadata=None,
+    acquire_execution_claim=None,
     strategy_profile=None,
     account_group=None,
     service_name=None,
@@ -1423,6 +1424,15 @@ def execute_rebalance(
 ):
     """Execute trades to reach target weights."""
     del target_weights
+    if not dry_run_only:
+        delegate_submit = submit_order_intent
+
+        def submit_claimed_order(ib, order_intent):
+            if acquire_execution_claim is None or not acquire_execution_claim():
+                raise RuntimeError("IBKR execution claim required; refusing broker submission")
+            return delegate_submit(ib, order_intent)
+
+        submit_order_intent = submit_claimed_order
     signal_metadata = signal_metadata or {}
     allocation = _resolve_weight_allocation(signal_metadata)
     target_weights = dict(allocation["targets"])

application/rebalance_service.py

@@ -965,7 +965,6 @@ def run_strategy_core(
         execution_marker_key = _build_execution_marker_key(config=config, signal_metadata=signal_metadata)
         execution_state_store = getattr(config, "execution_state_store", None)
         execution_already_recorded = False
-        execution_claim_acquired = False
         if execution_marker_key and execution_state_store:
             try:
                 execution_already_recorded = bool(execution_state_store.has_marker(execution_marker_key))
@@ -996,24 +995,6 @@ def run_strategy_core(
                         flush=True,
                     )
 
-        if (
-            not execution_already_recorded
-            and execution_marker_key
-            and execution_state_store
-            and bool(getattr(config, "execution_dedup_enabled", False))
-            and not bool(getattr(config, "dry_run_only", False))
-        ):
-            try:
-                execution_claim_acquired = bool(execution_state_store.claim_marker(
-                    execution_marker_key,
-                    metadata={"platform": "ibkr", "strategy_profile": signal_metadata.get("strategy_profile")},
-                ))
-                execution_already_recorded = not execution_claim_acquired
-            except Exception as exc:
-                raise RuntimeError(
-                    f"IBKR execution claim unavailable; refusing broker submission: {type(exc).__name__}"
-                ) from exc
-
         if execution_already_recorded:
             message = _execution_already_recorded_message(config=config, signal_metadata=signal_metadata)
             print(message, flush=True)
@@ -1069,13 +1050,37 @@ def run_strategy_core(
                 reconciliation_record_path=str(record_path),
             )
 
+        execution_claim_attempted = False
+        execution_claim_acquired = False
+
+        def acquire_execution_claim():
+            nonlocal execution_claim_attempted, execution_claim_acquired
+            # No-op cycles never claim; a failed attempt cannot retry on another intent.
+            if not execution_claim_attempted:
+                execution_claim_attempted = True
+                if (
+                    not config.dry_run_only
+                    and config.execution_dedup_enabled
+                    and execution_marker_key
+                    and execution_state_store is not None
+                ):
+                    try:
+                        execution_claim_acquired = bool(execution_state_store.claim_marker(
+                            execution_marker_key,
+                            metadata={"platform": "ibkr", "strategy_profile": signal_metadata.get("strategy_profile")},
+                        ))
+                    except Exception:
+                        raise RuntimeError("IBKR execution claim unavailable; refusing broker submission") from None
+            return execution_claim_acquired
+
         execution_result = runtime.execute_rebalance(
             ib,
             resolved_target_weights,
             positions,
             account_values,
             strategy_symbols=allocation.get("strategy_symbols"),
             signal_metadata=signal_metadata,
+            acquire_execution_claim=acquire_execution_claim,
         )
         if isinstance(execution_result, tuple) and len(execution_result) == 2:
             trade_logs, execution_summary = execution_result

application/runtime_broker_adapters.py

@@ -307,6 +307,7 @@ def execute_rebalance(
         *,
         strategy_symbols=None,
         signal_metadata=None,
+        acquire_execution_claim=None,
     ):
         return self.application_execute_rebalance_fn(
             ib,
@@ -319,6 +320,7 @@ def execute_rebalance(
             translator=self.translator,
             strategy_symbols=strategy_symbols,
             signal_metadata=signal_metadata or {},
+            acquire_execution_claim=acquire_execution_claim,
             strategy_profile=self.strategy_profile,
             account_group=self.account_group,
             service_name=self.service_name,

main.py

@@ -1346,6 +1346,7 @@ def execute_rebalance(
     *,
     strategy_symbols=None,
     signal_metadata=None,
+    acquire_execution_claim=None,
     dry_run_only_override: bool | None = None,
 ):
     return build_broker_adapters(dry_run_only_override=dry_run_only_override).execute_rebalance(
@@ -1355,6 +1356,7 @@ def execute_rebalance(
         account_values,
         strategy_symbols=strategy_symbols,
         signal_metadata=signal_metadata,
+        acquire_execution_claim=acquire_execution_claim,
     )
 
 

tests/test_execution_service.py

@@ -1,5 +1,7 @@
 from types import SimpleNamespace
 
+import pytest
+
 from application.execution_service import check_order_submitted, execute_rebalance, get_available_buying_power
 from notifications.telegram import build_translator
 from quant_platform_kit.common.models import OrderIntent
@@ -159,6 +161,7 @@ def fake_fetch_quote_snapshots(_ib, symbols):
         fetch_quote_snapshots=fake_fetch_quote_snapshots,
         submit_order_intent=fake_submit_order_intent,
         order_intent_cls=OrderIntent,
+        acquire_execution_claim=lambda: True,
         translator=translate,
         strategy_symbols=["VOO", "BIL"],
         strategy_profile="tech_communication_pullback_enhancement",
@@ -188,6 +191,30 @@ def fake_fetch_quote_snapshots(_ib, symbols):
     assert any(log.startswith("buy VOO") for log in trade_logs)
 
 
+@pytest.mark.parametrize("claim", [None, lambda: False])
+def test_execute_rebalance_cannot_submit_without_successful_claim_callback(tmp_path, claim):
+    submitted = []
+    ib = SimpleNamespace(
+        openTrades=lambda: [],
+        accountValues=lambda: [SimpleNamespace(tag="AvailableFunds", currency="USD", value="1000")],
+    )
+    with pytest.raises(RuntimeError, match="execution claim required"):
+        execute_rebalance(
+            ib, {"VOO": 0.5}, {}, {"equity": 1000.0, "buying_power": 1000.0},
+            fetch_quote_snapshots=lambda _ib, symbols: {
+                symbol: SimpleNamespace(last_price=100.0) for symbol in symbols
+            },
+            submit_order_intent=lambda _ib, intent: submitted.append(intent),
+            order_intent_cls=OrderIntent, translator=translate,
+            signal_metadata=_signal_metadata({"VOO": 0.5}, risk_symbols=("VOO",)),
+            acquire_execution_claim=claim, dry_run_only=False,
+            cash_reserve_ratio=0.0, rebalance_threshold_ratio=0.02,
+            limit_buy_premium=1.0, sell_settle_delay_sec=0,
+            execution_lock_dir=tmp_path,
+        )
+    assert submitted == []
+
+
 def test_execute_rebalance_paper_admission_blocks_before_calling_the_broker(tmp_path):
     class FakeIB:
         def openTrades(self):
@@ -273,6 +300,7 @@ def accountValues(self):
             status="Rejected",
         ),
         order_intent_cls=OrderIntent,
+        acquire_execution_claim=lambda: True,
         translator=translate,
         strategy_symbols=["VOO"],
         strategy_profile="tech_communication_pullback_enhancement",
@@ -324,6 +352,7 @@ def fake_submit_order_intent(_ib, intent):
         },
         submit_order_intent=fake_submit_order_intent,
         order_intent_cls=OrderIntent,
+        acquire_execution_claim=lambda: True,
         translator=translate,
         strategy_symbols=["SOXL"],
         strategy_profile="soxl_soxx_trend_income",
@@ -649,6 +678,7 @@ def fake_submit_order_intent(_ib, intent):
         },
         submit_order_intent=fake_submit_order_intent,
         order_intent_cls=OrderIntent,
+        acquire_execution_claim=lambda: True,
         translator=build_translator("zh"),
         strategy_symbols=["SOXL", "SOXX"],
         strategy_profile="soxl_soxx_trend_income",
@@ -708,6 +738,7 @@ def fake_submit_order_intent(_ib, intent):
         },
         submit_order_intent=fake_submit_order_intent,
         order_intent_cls=OrderIntent,
+        acquire_execution_claim=lambda: True,
         translator=build_translator("zh"),
         strategy_symbols=["SOXL", "SOXX"],
         strategy_profile="soxl_soxx_trend_income",
@@ -770,6 +801,7 @@ def fake_submit_order_intent(_ib, intent):
         },
         submit_order_intent=fake_submit_order_intent,
         order_intent_cls=OrderIntent,
+        acquire_execution_claim=lambda: True,
         translator=build_translator("zh"),
         strategy_symbols=["SOXL", "SOXX"],
         strategy_profile="soxl_soxx_trend_income",
@@ -1000,6 +1032,7 @@ def fake_submit_order_intent(_ib, intent):
         },
         submit_order_intent=fake_submit_order_intent,
         order_intent_cls=OrderIntent,
+        acquire_execution_claim=lambda: True,
         translator=translate,
         strategy_symbols=["TQQQ"],
         strategy_profile="tqqq_growth_income",
@@ -1078,6 +1111,7 @@ def fake_submit_order_intent(_ib, intent):
         fetch_quote_snapshots=lambda *_args, **_kwargs: {"VOO": SimpleNamespace(last_price=165.85)},
         submit_order_intent=fake_submit_order_intent,
         order_intent_cls=OrderIntent,
+        acquire_execution_claim=lambda: True,
         translator=translate,
         strategy_symbols=["VOO"],
         strategy_profile="global_etf_rotation",
@@ -1129,6 +1163,7 @@ def fake_submit_order_intent(_ib, intent):
         },
         submit_order_intent=fake_submit_order_intent,
         order_intent_cls=OrderIntent,
+        acquire_execution_claim=lambda: True,
         translator=translate,
         strategy_symbols=["TQQQ", "QQQM"],
         strategy_profile="tqqq_growth_income",
@@ -1457,6 +1492,7 @@ def fake_fetch_quote_snapshots(_ib, symbols):
         fetch_quote_snapshots=fake_fetch_quote_snapshots,
         submit_order_intent=lambda *_args, **_kwargs: SimpleNamespace(broker_order_id="1", status="Submitted"),
         order_intent_cls=OrderIntent,
+        acquire_execution_claim=lambda: True,
         translator=translate,
         strategy_symbols=["VOO", "BOXX"],
         strategy_profile="tech_communication_pullback_enhancement",
@@ -1576,6 +1612,7 @@ def fake_fetch_quote_snapshots(_ib, symbols):
         fetch_quote_snapshots=fake_fetch_quote_snapshots,
         submit_order_intent=lambda *_args, **_kwargs: SimpleNamespace(broker_order_id="1", status="Submitted"),
         order_intent_cls=OrderIntent,
+        acquire_execution_claim=lambda: True,
         translator=translate,
         strategy_symbols=["VOO", "BOXX"],
         strategy_profile="tech_communication_pullback_enhancement",
@@ -1705,6 +1742,7 @@ def fake_submit_order_intent(_ib, intent):
         },
         submit_order_intent=fake_submit_order_intent,
         order_intent_cls=OrderIntent,
+        acquire_execution_claim=lambda: True,
         translator=translate,
         strategy_symbols=["VOO", "BOXX"],
         strategy_profile="tech_communication_pullback_enhancement",
@@ -1901,6 +1939,7 @@ def fake_submit_order_intent(_ib, intent):
         fetch_quote_snapshots=lambda *_args, **_kwargs: {},
         submit_order_intent=fake_submit_order_intent,
         order_intent_cls=OrderIntent,
+        acquire_execution_claim=lambda: True,
         translator=translate,
         strategy_symbols=["VOO"],
         strategy_profile="tech_communication_pullback_enhancement",