From 725d44cf5ec73a8b811652d94e7f5ed6e33ac773 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Tue, 8 Sep 2026 06:03:23 +0800 Subject: [PATCH] fix: stop order cycle after unknown broker outcome Co-Authored-By: Codex --- application/execution_service.py | 93 +++++++++++-------- application/rebalance_service.py | 73 ++++++++------- tests/test_rebalance_service.py | 155 +++++++++++++++++++++++++++++++ 3 files changed, 250 insertions(+), 71 deletions(-) diff --git a/application/execution_service.py b/application/execution_service.py index 7b5d6f8..2090963 100644 --- a/application/execution_service.py +++ b/application/execution_service.py @@ -1043,6 +1043,7 @@ def execute_rebalance_cycle( small_account_bootstrap_note_keys: set[str] = set() action_done = False sell_submitted = False + submission_halted = False pending_sell_release_symbols: list[str] = [] threshold_value = float(execution["trade_threshold_value"]) limit_order_symbols = set( @@ -1206,7 +1207,16 @@ def append_order_id_suffix(log_message, order_id): suffix = f"[order_id={order_id_text}]" return f"{log_message} {suffix}" + def notify_submission_issue(title, detail): + try: + notify_issue(title, detail) + except Exception: + print("order_notification_failed", flush=True) + def submit_order_via_port(symbol, order_type, side, quantity, log_message, *, submitted_price=None): + nonlocal submission_halted + if submission_halted: + return False if fractional_buy_execution and side == "buy": normalized_quantity = _normalize_buy_quantity( quantity, @@ -1235,59 +1245,52 @@ def submit_order_via_port(symbol, order_type, side, quantity, log_message, *, su try: report = execution_port.submit_order(order_intent) except Exception: - notify_issue( - "Order submit failed", - ( - f"Symbol: {symbol} Side: {side_text} Qty: {quantity} " - f"Type: {order_type} Price: {submitted_price if submitted_price is not None else 'MO'}\n" - f"{traceback.format_exc()}" - ), - ) - return False - - status = str(report.status or "").strip().lower() - if status not in {"submitted", "accepted"}: - detail = report.raw_payload.get("detail", report.status) if isinstance(report.raw_payload, Mapping) else report.status - notify_issue( - "Order submit failed", - ( - f"Symbol: {symbol} Side: {side_text} Qty: {quantity} " - f"Type: {order_type} Price: {submitted_price if submitted_price is not None else 'MO'}\n" - f"Status: {detail}" - ), - ) + report = None + status = str(getattr(report, "status", "") or "").strip().lower() + if status == "rejected": + notify_submission_issue("Order submit failed", "order_rejected") return False - log_with_order_id = append_order_id_suffix(log_message, report.broker_order_id) - pending_log = translator("order_pending_confirmation", detail=log_with_order_id) - print(with_prefix(pending_log), flush=True) - logs.append(pending_log) + known_submission = status in {"submitted", "accepted", "filled", "partially_filled"} + if not known_submission: + submission_halted = True order_payload = { "symbol": str(symbol or "").strip().upper(), "side": str(side or "").strip().lower(), "quantity": float(order_intent.quantity or 0.0), "order_type": str(order_type or "").strip().lower(), - "status": "pending_reconciliation", - "submission_status": report.status, + "status": "pending_reconciliation" if known_submission else "unknown", + "submission_status": status if known_submission else "unknown", } if submitted_price is not None: order_payload["price"] = round(float(submitted_price), 4) if order_type == "limit": order_payload["limit_price"] = round(float(submitted_price), 4) - if report.broker_order_id: - order_payload["broker_order_id"] = report.broker_order_id - submitted_orders.append(order_payload) + broker_order_id = getattr(report, "broker_order_id", None) + if broker_order_id: + order_payload["broker_order_id"] = broker_order_id + if known_submission: + order_payload["filled_quantity"] = report.filled_quantity + if report.average_fill_price is not None: + order_payload["average_fill_price"] = report.average_fill_price + # Preserve the broker fact before notifications or other fallible post-submit work. pending_orders.append(order_payload) + if not known_submission: + notify_submission_issue("Order submit failed", "broker_outcome_unknown_reconciliation_required") + return False + submitted_orders.append(order_payload) if str(side or "").strip().lower() == "sell": submitted_sell_orders.append(order_payload) - if post_submit_order is not None: - try: + try: + log_with_order_id = append_order_id_suffix(log_message, broker_order_id) + pending_log = translator("order_pending_confirmation", detail=log_with_order_id) + logs.append(pending_log) + print(with_prefix(pending_log), flush=True) + if post_submit_order is not None: post_submit_order(trade_context, order_intent, report) - except Exception: - notify_issue( - "Order post-submit hook failed", - f"Symbol: {symbol} Side: {side_text} Qty: {quantity}\n{traceback.format_exc()}", - ) + except Exception: + submission_halted = True + notify_submission_issue("Order post-submit hook failed", "post_submit_processing_failed_reconciliation_required") return True def record_dry_run(symbol, side, quantity, price, *, order_type): @@ -1319,6 +1322,8 @@ def record_dry_run(symbol, side, quantity, price, *, order_type): return True for symbol in strategy_assets: + if submission_halted: + break if _sell_delta_exceeds_floor( current_value=market_values[symbol], target_value=target_values[symbol], @@ -1381,6 +1386,8 @@ def record_dry_run(symbol, side, quantity, price, *, order_type): translator("market_sell", symbol=symbol, qty=quantity_text, price=round(price, 2)), ) + if submission_halted and not submitted: + break if submitted: action_done = True sell_submitted = True @@ -1428,7 +1435,8 @@ def record_dry_run(symbol, side, quantity, price, *, order_type): if small_account_buy_blocked or account_new_risk_buy_blocked: funding_buy_candidates = [] if ( - not sell_submitted + not submission_halted + and not sell_submitted and funding_buy_candidates and cash_sweep_symbol and sellable_quantities.get(cash_sweep_symbol, 0.0) > 0.0 @@ -1494,7 +1502,7 @@ def record_dry_run(symbol, side, quantity, price, *, order_type): sell_submitted = True cash_sweep_sold_this_cycle = True - if sell_submitted: + if sell_submitted and not submission_halted: if dry_run_only and dry_run_sale_proceeds > 0.0: simulated_cash = float(dry_run_sale_proceeds) available_cash = max(0.0, available_cash + simulated_cash) @@ -1611,6 +1619,8 @@ def record_dry_run(symbol, side, quantity, price, *, order_type): if (target_values[symbol] - market_values[symbol]) > threshold_value and abs(target_values[symbol] - market_values[symbol]) > current_min_trade ] + if submission_halted: + buy_candidates = [] buys_blocked_reason: str | None = None if small_account_buy_blocked and buy_candidates: buys_blocked_reason = "small_account_below_recommended_equity" @@ -1690,6 +1700,8 @@ def record_dry_run(symbol, side, quantity, price, *, order_type): buy_candidates = [] for symbol in buy_candidates: + if submission_halted: + break diff = target_values[symbol] - market_values[symbol] price = safe_quote_last_price( market_symbol(symbol), @@ -1837,7 +1849,8 @@ def record_dry_run(symbol, side, quantity, price, *, order_type): allocation.get("small_account_safe_haven_cash_substituted_symbols") ) if ( - not cash_sweep_sold_this_cycle + not submission_halted + and not cash_sweep_sold_this_cycle and cash_sweep_symbol and cash_sweep_symbol in strategy_assets and not small_account_buy_blocked diff --git a/application/rebalance_service.py b/application/rebalance_service.py index f91ddd0..22dfc4f 100644 --- a/application/rebalance_service.py +++ b/application/rebalance_service.py @@ -16,6 +16,7 @@ AccountIdentityPolicy, evaluate_account_identity, ) +from quant_platform_kit.common.models import ExecutionReport from quant_platform_kit.common.port_adapters import CallableExecutionPort from quant_platform_kit.longbridge.market_data import fetch_lot_sizes from application.signal_snapshot import build_signal_snapshot @@ -339,7 +340,7 @@ def _should_record_execution_marker(*, result: ExecutionCycleResult, config: Lon return False if bool(getattr(config, "dry_run_only", False)) and tuple(getattr(result, "dry_run_orders", ()) or ()): return True - return bool(getattr(result, "action_done", False)) + return bool(getattr(result, "action_done", False) or getattr(result, "pending_orders", ())) def _record_execution_marker( @@ -353,7 +354,8 @@ def _record_execution_marker( if not store or not marker_key: return try: - store.record_marker( + record = store.record_outcome if result.pending_orders else store.record_marker + record( marker_key, metadata={ "strategy_profile": getattr(config, "strategy_profile", ""), @@ -362,15 +364,16 @@ def _record_execution_marker( "action_done": bool(getattr(result, "action_done", False)), "dry_run_orders_count": len(tuple(getattr(result, "dry_run_orders", ()) or ())), "pending_orders_count": len(tuple(getattr(result, "pending_orders", ()) or ())), + "pending_orders": list(result.pending_orders), "signal_date": str(dict(getattr(result, "execution", {}) or {}).get("signal_date") or ""), "effective_date": str(dict(getattr(result, "execution", {}) or {}).get("effective_date") or ""), }, ) - except Exception as exc: - notify_issue( - "Execution marker write failed", - f"Marker: {marker_key}\n{type(exc).__name__}: {exc}", - ) + except Exception: + try: + notify_issue("Execution marker write failed", "execution_outcome_persistence_failed") + except Exception: + print("order_notification_failed", flush=True) @@ -545,6 +548,10 @@ def fetch_replanned_state(): def submit_claimed_order(order_intent): nonlocal execution_claim_attempted, execution_claim_acquired + rejected = ExecutionReport( + symbol=order_intent.symbol, side=order_intent.side, + quantity=order_intent.quantity, status="rejected", + ) # Claim only when an order is ready; failed claims are never retried in this cycle. if not execution_claim_attempted: execution_claim_attempted = True @@ -564,24 +571,25 @@ def submit_claimed_order(order_intent): }, ) except Exception: - raise RuntimeError( - "LongBridge account owner fence unavailable; refusing broker submission" - ) from None + return rejected if not owner_claim.allowed: - raise RuntimeError( - "LongBridge account owner fence contested for " - f"account={account_id}; owner={owner_claim.owner_id!r} " - f"contested_by={owner_id!r}" - ) + return rejected try: execution_claim_acquired = bool(execution_state_store.claim_marker( execution_marker_key, - metadata={"platform": "longbridge", "strategy_profile": config.strategy_profile}, + metadata={ + "platform": "longbridge", "strategy_profile": config.strategy_profile, + "order_intent": { + "symbol": order_intent.symbol, "side": order_intent.side, + "quantity": order_intent.quantity, "order_type": order_intent.order_type, + "limit_price": order_intent.limit_price, + }, + }, )) except Exception: - raise RuntimeError("LongBridge execution claim unavailable; refusing broker submission") from None + return rejected if not execution_claim_acquired: - raise RuntimeError("LongBridge execution claim required; refusing broker submission") + return rejected return delegate.submit_order(order_intent) execution_port = CallableExecutionPort(submit_claimed_order) @@ -678,20 +686,23 @@ def submit_claimed_order(order_intent): pending_orders = tuple(getattr(execution_result, "pending_orders", ()) or ()) if pending_orders: - notification_publisher.publish( - notification_renderers.render_rebalance_notification( - execution=execution, - logs=logs, - skip_logs=skip_logs, - note_logs=note_logs, - translator=config.translator, - separator=config.separator, - strategy_display_name=config.strategy_display_name, - dry_run_only=config.dry_run_only, - extra_notification_lines=config.extra_notification_lines, - title_key="pending_order_title", + try: + notification_publisher.publish( + notification_renderers.render_rebalance_notification( + execution=execution, + logs=logs, + skip_logs=skip_logs, + note_logs=note_logs, + translator=config.translator, + separator=config.separator, + strategy_display_name=config.strategy_display_name, + dry_run_only=config.dry_run_only, + extra_notification_lines=config.extra_notification_lines, + title_key="pending_order_title", + ) ) - ) + except Exception: + print("pending_order_notification_failed", flush=True) elif action_done: notification_publisher.publish( notification_renderers.render_rebalance_notification( diff --git a/tests/test_rebalance_service.py b/tests/test_rebalance_service.py index 4857706..8aa6c95 100644 --- a/tests/test_rebalance_service.py +++ b/tests/test_rebalance_service.py @@ -1,3 +1,4 @@ +import json import os import sys import unittest @@ -3422,6 +3423,160 @@ def _run(self, **overrides): runtime=self.runtime, config=replace(self.config, **overrides), ) + def _add_second_buy_target(self): + self.plan["allocation"]["strategy_symbols"] = ("SOXL", "SOXX") + self.plan["allocation"]["risk_symbols"] = ("SOXL", "SOXX") + self.plan["allocation"]["targets"] = {"SOXL": 200.0, "SOXX": 200.0} + self.plan["portfolio"]["market_values"]["SOXX"] = 0.0 + self.plan["portfolio"]["quantities"]["SOXX"] = 0 + self.plan["portfolio"]["sellable_quantities"]["SOXX"] = 0 + + def test_unknown_submission_stops_cycle_and_retains_intent_and_claim(self): + self._add_second_buy_target() + for outcome in ("timeout", "unknown_report", "missing_report"): + with self.subTest(outcome=outcome), TemporaryDirectory() as directory: + self.orders.clear() + self.issues.clear() + store = ExecutionMarkerStore(local_dir=directory) + + def uncertain_submit(intent): + self.orders.append(intent) + if outcome == "timeout": + raise TimeoutError("synthetic-private-provider-detail") + if outcome == "missing_report": + return None + return ExecutionReport( + symbol=intent.symbol, side=intent.side, quantity=intent.quantity, + status="unknown", broker_order_id="synthetic-unknown-order", + ) + + self.runtime = replace(self.runtime, execution_port_factory=lambda _context: CallableExecutionPort(uncertain_submit)) + result = self._run(execution_dedup_enabled=True, execution_state_store=store) + self.assertEqual(len(self.orders), 1) + self.assertFalse(result.action_done) + self.assertEqual(len(result.pending_orders), 1) + order = result.pending_orders[0] + intent = self.orders[0] + self.assertEqual((order["symbol"], order["side"], order["quantity"]), (intent.symbol, intent.side, intent.quantity)) + self.assertEqual(order["status"], "unknown") + self.assertEqual(order["submission_status"], "unknown") + from application.execution_receipt_adapter import attach_cycle_execution_receipt + report = { + "platform": "longbridge", "strategy_profile": self.config.strategy_profile, + "dry_run": False, "runtime_target": {"execution_mode": "paper"}, + "runtime_release_receipt": { + "attestation_state": "self_attested", "strategy_release": {"strategy_revision": "a" * 40}, + }, + } + attach_cycle_execution_receipt(report, result) + self.assertEqual(report["execution_receipt"]["outcome"], "reconciliation_required") + if outcome == "unknown_report": + self.assertEqual(order["broker_order_id"], "synthetic-unknown-order") + self.assertNotIn("synthetic-private-provider-detail", str(self.issues)) + key = rebalance_service._build_execution_marker_key(config=replace(self.config, execution_dedup_enabled=True), execution=self.plan["execution"]) + original_claim = store.read_marker(key) + self.assertEqual(original_claim["state"], "claimed") + self.assertEqual(original_claim["metadata"]["order_intent"]["symbol"], intent.symbol) + outcome_payload = json.loads(store._outcome_local_path(key).read_text()) + self.assertEqual(outcome_payload["metadata"]["pending_orders"], list(result.pending_orders)) + self._run(execution_dedup_enabled=True, execution_state_store=store) + self.assertEqual(len(self.orders), 1) + self.assertEqual(store.read_marker(key), original_claim) + + def test_unknown_sell_stops_remaining_sells_buys_and_refresh(self): + self._add_second_buy_target() + self.plan["allocation"]["targets"] = {"SOXL": 0.0, "SOXX": 200.0} + self.plan["portfolio"]["market_values"]["SOXL"] = 400.0 + self.plan["portfolio"]["quantities"]["SOXL"] = 4 + self.plan["portfolio"]["sellable_quantities"]["SOXL"] = 4 + resolve_plan = Mock(return_value=self.plan) + + def submit(intent): + self.orders.append(intent) + raise TimeoutError("synthetic-private-provider-detail") + + self.runtime = replace( + self.runtime, execution_port_factory=lambda _context: CallableExecutionPort(submit), + resolve_rebalance_plan=resolve_plan, + ) + result = self._run(execution_dedup_enabled=True, execution_state_store=self.store) + self.assertEqual([(order.symbol, order.side) for order in self.orders], [("SOXL.US", "sell")]) + self.assertEqual(result.pending_orders[0]["status"], "unknown") + self.assertEqual(resolve_plan.call_count, 1) + + def test_filled_order_and_claim_survive_outcome_store_failure(self): + self._add_second_buy_target() + + def submit(intent): + self.orders.append(intent) + return ExecutionReport( + symbol=intent.symbol, side=intent.side, quantity=intent.quantity, + status="filled", broker_order_id="synthetic-filled-order", + filled_quantity=intent.quantity, average_fill_price=100.0, + ) + + self.runtime = replace( + self.runtime, execution_port_factory=lambda _context: CallableExecutionPort(submit), + post_submit_order=Mock(side_effect=RuntimeError("synthetic-private-hook-detail")), + ) + with patch.object(ExecutionMarkerStore, "record_outcome", side_effect=RuntimeError("synthetic-private-store-detail")): + result = self._run(execution_dedup_enabled=True, execution_state_store=self.store) + self.assertEqual(len(self.orders), 1) + self.assertTrue(result.action_done) + self.assertEqual(result.pending_orders[0]["submission_status"], "filled") + self.assertEqual(result.pending_orders[0]["filled_quantity"], self.orders[0].quantity) + self.assertEqual(result.pending_orders[0]["average_fill_price"], 100.0) + key = rebalance_service._build_execution_marker_key( + config=replace(self.config, execution_dedup_enabled=True), execution=self.plan["execution"], + ) + self.assertEqual(self.store.read_marker(key)["state"], "claimed") + self.assertNotIn("synthetic-private", str(self.issues)) + self._run(execution_dedup_enabled=True, execution_state_store=self.store) + self.assertEqual(len(self.orders), 1) + + def test_definite_rejection_is_not_unknown_and_allows_next_intent(self): + self._add_second_buy_target() + + def submit(intent): + self.orders.append(intent) + return ExecutionReport( + symbol=intent.symbol, side=intent.side, quantity=intent.quantity, + status="rejected" if len(self.orders) == 1 else "submitted", + broker_order_id=None if len(self.orders) == 1 else "synthetic-accepted", + ) + + self.runtime = replace(self.runtime, execution_port_factory=lambda _context: CallableExecutionPort(submit)) + result = self._run(execution_dedup_enabled=True, execution_state_store=self.store) + self.assertEqual(len(self.orders), 2) + self.assertEqual(len(result.pending_orders), 1) + self.assertEqual(result.pending_orders[0]["broker_order_id"], "synthetic-accepted") + + def test_acknowledged_order_hook_failure_stops_new_submits_without_losing_ack(self): + self._add_second_buy_target() + self.runtime = replace(self.runtime, post_submit_order=Mock(side_effect=RuntimeError("synthetic-private-hook-detail"))) + result = self._run(execution_dedup_enabled=True, execution_state_store=self.store) + self.assertEqual(len(self.orders), 1) + self.assertTrue(result.action_done) + self.assertEqual(result.pending_orders[0]["submission_status"], "submitted") + self.assertEqual(result.pending_orders[0]["broker_order_id"], "synthetic-order") + self.assertNotIn("synthetic-private-hook-detail", str(self.issues)) + + def test_unknown_order_survives_notification_failure(self): + self._add_second_buy_target() + + def submit(intent): + self.orders.append(intent) + raise TimeoutError("synthetic uncertain submission") + + self.runtime = replace( + self.runtime, execution_port_factory=lambda _context: CallableExecutionPort(submit), + notify_issue=Mock(side_effect=RuntimeError("synthetic notification failed")), + notifications=CallableNotificationPort(Mock(side_effect=RuntimeError("synthetic notification failed"))), + ) + result = self._run(execution_dedup_enabled=True, execution_state_store=self.store) + self.assertEqual(len(self.orders), 1) + self.assertEqual(result.pending_orders[0]["status"], "unknown") + def test_missing_claim_prerequisites_never_submit(self): cases = ( {},