fix: fail closed on IBKR execution readiness

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

Author: Pigbibi

application/execution_service.py

@@ -310,7 +310,14 @@ def _resolve_order_account_id(account_ids=None) -> str | None:
     return normalized[0] if normalized else None
 
 
-MARKET_CURRENCY_CASH_TAG_PRIORITY = ("CashBalance", "TotalCashBalance", "SettledCash")
+MARKET_CURRENCY_CASH_TAG_PRIORITY = (
+    "$LEDGER-CashBalance",
+    "$LEDGER-TotalCashBalance",
+    "CashBalance",
+    "TotalCashBalance",
+    "SettledCash",
+    "TotalCashValue",
+)
 
 
 def _coerce_float(value):
@@ -2251,20 +2258,31 @@ def cash_sweep_sale_quantity_to_fund_buy(max_quantity: int, candidate_symbols: t
         buying_power=buying_power,
     )
 
-    execution_summary["execution_status"] = (
-        "executed"
-        if (
-            execution_summary["orders_submitted"]
-            or execution_summary["orders_filled"]
-            or execution_summary["orders_partially_filled"]
-            or execution_summary["option_orders_submitted"]
-            or execution_summary["option_orders_filled"]
-            or execution_summary["option_orders_partially_filled"]
-        )
-        else "no_op"
+    has_accepted_order = bool(
+        execution_summary["orders_submitted"]
+        or execution_summary["orders_filled"]
+        or execution_summary["orders_partially_filled"]
+        or execution_summary["option_orders_submitted"]
+        or execution_summary["option_orders_filled"]
+        or execution_summary["option_orders_partially_filled"]
     )
-    if execution_summary["execution_status"] == "executed":
+    submission_failure = next(
+        (
+            reason
+            for reason in execution_summary["skipped_reasons"]
+            if reason.startswith(("submit_failed:", "option_submit_failed:"))
+        ),
+        None,
+    )
+    if has_accepted_order:
+        execution_summary["execution_status"] = "executed"
         execution_summary["no_op_reason"] = None
+    elif submission_failure:
+        execution_summary["execution_status"] = "blocked"
+        execution_summary["no_op_reason"] = submission_failure
+        trade_logs.append(translator("failed", reason=submission_failure))
+    else:
+        execution_summary["execution_status"] = "no_op"
     execution_summary["residual_cash_estimate"] = float(max(buying_power, 0.0))
     append_small_account_allocation_drift_notes()
     return _finalize_result(trade_logs, execution_summary, return_summary=return_summary)

application/ibkr_portfolio.py

@@ -8,7 +8,18 @@
 
 from quant_platform_kit.common.models import PortfolioSnapshot, Position
 
-MARKET_CURRENCY_CASH_TAG_PRIORITY = ("CashBalance", "TotalCashBalance", "SettledCash")
+MARKET_CURRENCY_CASH_TAG_PRIORITY = (
+    "$LEDGER-CashBalance",
+    "$LEDGER-TotalCashBalance",
+    "CashBalance",
+    "TotalCashBalance",
+    "SettledCash",
+    "TotalCashValue",
+)
+
+
+class IBKRPortfolioSnapshotUnavailableError(RuntimeError):
+    """Raised when IBKR did not return enough account data for safe execution."""
 
 
 def _normalize_account_ids(account_ids: Iterable[str] | str | None) -> tuple[str, ...]:
@@ -126,11 +137,14 @@ def fetch_portfolio_snapshot(
 
     total_equity = 0.0
     available_funds = None
+    matched_account_value_count = 0
+    matched_market_currency_value_count = 0
     values_by_account_currency: dict[tuple[str | None, str], dict[str, float]] = {}
     for account_value in ib.accountValues():
         account_id = str(getattr(account_value, "account", "") or "").strip() or None
         if not _matches_account(account_id, selected_account_ids):
             continue
+        matched_account_value_count += 1
         value_currency = str(getattr(account_value, "currency", "") or "").strip().upper()
         if value_currency:
             tag_values = values_by_account_currency.setdefault((account_id, value_currency), {})
@@ -139,6 +153,7 @@ def fetch_portfolio_snapshot(
                 tag_values[str(getattr(account_value, "tag", "") or "").strip()] = numeric_value
         if value_currency != market_currency:
             continue
+        matched_market_currency_value_count += 1
         if account_value.tag == "NetLiquidation":
             total_equity += float(account_value.value)
         elif account_value.tag == "AvailableFunds":
@@ -149,6 +164,18 @@ def fetch_portfolio_snapshot(
         values_by_account_currency,
         currency=market_currency,
     )
+    if selected_account_ids and matched_account_value_count == 0:
+        raise IBKRPortfolioSnapshotUnavailableError(
+            "IBKR returned no account values for the configured account selection."
+        )
+    if selected_account_ids and matched_market_currency_value_count == 0:
+        raise IBKRPortfolioSnapshotUnavailableError(
+            f"IBKR returned no {market_currency} account values for the configured account selection."
+        )
+    if cash_only_execution and not positions and market_currency_cash is None:
+        raise IBKRPortfolioSnapshotUnavailableError(
+            f"IBKR cash-only snapshot is missing the {market_currency} cash balance and positions."
+        )
     if cash_only_execution:
         buying_power = float(market_currency_cash or 0.0) if market_currency_cash is not None else 0.0
         position_market_values = {

application/rebalance_service.py

@@ -1102,8 +1102,11 @@ def run_strategy_core(
                 flush=True,
             )
         _record_platform_execution_telemetry(signal_metadata, dict(execution_summary or {}))
+        execution_status = str((execution_summary or {}).get("execution_status") or "").strip().lower()
+        execution_blocked = execution_status in {"blocked", "error", "failed", "failure"}
+        execution_reason = str((execution_summary or {}).get("no_op_reason") or "execution_blocked")
         return StrategyCycleResult(
-            result="OK - executed",
+            result=f"Blocked - {execution_reason}" if execution_blocked else "OK - executed",
             signal_metadata=dict(signal_metadata or {}),
             target_weights=dict(resolved_target_weights or {}),
             execution_summary=dict(execution_summary or {}),

application/runtime_broker_adapters.py

@@ -14,6 +14,10 @@ class IBKRGatewayUnavailableError(ConnectionError):
     """Raised after retryable IBKR gateway connection attempts are exhausted."""
 
 
+class IBKRTradingPermissionError(RuntimeError):
+    """Raised when a live Gateway connection cannot verify order-read access."""
+
+
 @dataclass(frozen=True)
 class IBKRRuntimeBrokerAdapters:
     host_resolver: Any
@@ -85,6 +89,62 @@ def fetch_account_portfolio_snapshot(self, ib):
             return self.fetch_portfolio_snapshot_fn(ib, account_ids=self.account_ids)
         return self.fetch_portfolio_snapshot_fn(ib)
 
+    def validate_trading_permissions(self, ib):
+        if self.dry_run_only or str(self.execution_mode or "").strip().lower() != "live":
+            return
+        request_open_orders = getattr(ib, "reqOpenOrders", None)
+        if not callable(request_open_orders):
+            raise IBKRTradingPermissionError(
+                "IB Gateway live execution cannot verify order-read access."
+            )
+
+        original_raise_request_errors = getattr(ib, "RaiseRequestErrors", False)
+        original_request_timeout = getattr(ib, "RequestTimeout", 0)
+        read_only_errors: list[tuple[Any, str]] = []
+        error_event = getattr(ib, "errorEvent", None)
+        error_handler_registered = False
+
+        def is_read_only_error(message: Any) -> bool:
+            normalized = str(message).lower().replace("-", " ").replace("_", " ")
+            return "read only" in " ".join(normalized.split())
+
+        def capture_api_error(_request_id, error_code, error_message, _contract):
+            if is_read_only_error(error_message):
+                read_only_errors.append((error_code, str(error_message)))
+
+        try:
+            if error_event is not None:
+                error_event += capture_api_error
+                error_handler_registered = True
+            # Read order state only; this never calls placeOrder or cancelOrder.
+            ib.RaiseRequestErrors = True
+            ib.RequestTimeout = self.connect_timeout_seconds
+            request_open_orders()
+            if read_only_errors:
+                raise IBKRTradingPermissionError(
+                    "IB Gateway API is in Read-Only mode; live execution is disabled."
+                )
+        except TimeoutError as exc:
+            if read_only_errors:
+                raise IBKRTradingPermissionError(
+                    "IB Gateway API is in Read-Only mode; live execution is disabled."
+                ) from exc
+            raise
+        except Exception as exc:
+            if is_read_only_error(exc):
+                raise IBKRTradingPermissionError(
+                    "IB Gateway API is in Read-Only mode; live execution is disabled."
+                ) from exc
+            raise IBKRTradingPermissionError(
+                "IB Gateway live execution could not verify order-read access "
+                f"(error_type={type(exc).__name__})."
+            ) from exc
+        finally:
+            if error_handler_registered:
+                error_event -= capture_api_error
+            ib.RaiseRequestErrors = original_raise_request_errors
+            ib.RequestTimeout = original_request_timeout
+
     def connect_ib(self):
         self.ensure_event_loop_fn()
         host = self.host_resolver()
@@ -108,6 +168,7 @@ def connect_ib(self):
                 )
                 try:
                     self.validate_configured_accounts(ib)
+                    self.validate_trading_permissions(ib)
                 except Exception:
                     disconnect_fn = getattr(ib, "disconnect", None)
                     if callable(disconnect_fn):

entrypoints/cloud_run.py

@@ -35,7 +35,12 @@ def is_market_open_now(
     schedule = calendar.schedule(start_date=now_ny.date(), end_date=now_ny.date())
     if len(getattr(schedule, "index", ())) == 0:
         return False
-    return calendar.open_at_time(schedule, now_ny)
+    try:
+        return calendar.open_at_time(schedule, now_ny)
+    except ValueError as exc:
+        if "not covered by the schedule" not in str(exc):
+            raise
+        return False
 
 
 def is_market_open_today(

main.py

@@ -21,6 +21,7 @@
 from application.cycle_result import coerce_strategy_cycle_result
 from application.runtime_broker_adapters import (
     IBKRGatewayUnavailableError,
+    IBKRTradingPermissionError,
     build_runtime_broker_adapters,
 )
 from application.runtime_composer import build_runtime_composer
@@ -74,7 +75,10 @@
     max_workers_from_env,
     timeout_seconds_from_env,
 )
-from application.ibkr_portfolio import fetch_portfolio_snapshot
+from application.ibkr_portfolio import (
+    IBKRPortfolioSnapshotUnavailableError,
+    fetch_portfolio_snapshot,
+)
 from application.execution_service import (
     check_order_submitted as application_check_order_submitted,
     execute_rebalance as application_execute_rebalance,
@@ -1292,28 +1296,53 @@ def _handle_request(
         )
         if notification_delivery_log:
             report_summary["notification_delivery_log"] = notification_delivery_log
+        execution_status = str(report_summary.get("execution_status") or "").strip().lower()
+        execution_blocked = execution_status in {"blocked", "error", "failed", "failure"}
+        if execution_blocked:
+            append_runtime_report_error(
+                report,
+                stage="strategy_execution",
+                message=(
+                    f"Strategy execution status is {execution_status}; "
+                    f"reason={report_summary.get('no_op_reason') or 'unspecified'}"
+                ),
+                error_type="StrategyExecutionBlocked",
+                failure_category="strategy_execution_blocked",
+            )
+        report_diagnostics = {
+            "result": cycle_result.result,
+            "price_source_mode": execution_summary.get("price_source_mode") or reconciliation_record.get("price_source_mode"),
+            "snapshot_price_fallback_symbols": execution_summary.get("snapshot_price_fallback_symbols")
+            or reconciliation_record.get("snapshot_price_fallback_symbols")
+            or [],
+            **({"signal_snapshot": signal_snapshot} if has_signal_snapshot else {}),
+        }
+        if execution_blocked:
+            report_diagnostics["failure_category"] = "strategy_execution_blocked"
         finalize_runtime_report(
             report,
-            status="ok",
+            status="error" if execution_blocked else "ok",
             summary=report_summary,
-            diagnostics={
-                "result": cycle_result.result,
-                "price_source_mode": execution_summary.get("price_source_mode") or reconciliation_record.get("price_source_mode"),
-                "snapshot_price_fallback_symbols": execution_summary.get("snapshot_price_fallback_symbols")
-                or reconciliation_record.get("snapshot_price_fallback_symbols")
-                or [],
-                **({"signal_snapshot": signal_snapshot} if has_signal_snapshot else {}),
-            },
+            diagnostics=report_diagnostics,
             artifacts={
                 "reconciliation_record_path": cycle_result.reconciliation_record_path,
             },
         )
         log_runtime_event(
             log_context,
-            "strategy_cycle_completed",
-            message="Strategy dry-run completed" if dry_run_only_override else "Strategy execution completed",
+            "strategy_cycle_blocked" if execution_blocked else "strategy_cycle_completed",
+            message=(
+                "Strategy execution blocked"
+                if execution_blocked
+                else "Strategy dry-run completed"
+                if dry_run_only_override
+                else "Strategy execution completed"
+            ),
+            severity="ERROR" if execution_blocked else "INFO",
             execution_window=execution_window,
             result=cycle_result.result,
+            execution_status=report_summary.get("execution_status"),
+            no_op_reason=report_summary.get("no_op_reason"),
         )
         return (response_body if dry_run_only_override else cycle_result.result), 200
     except RuntimeDeadlineExceeded as exc:
@@ -1362,6 +1391,40 @@ def _handle_request(
             exc=exc,
         )
         return "Error", 503
+    except (IBKRTradingPermissionError, IBKRPortfolioSnapshotUnavailableError) as exc:
+        failure_category = (
+            "ibkr_trading_permission"
+            if isinstance(exc, IBKRTradingPermissionError)
+            else "ibkr_portfolio_snapshot"
+        )
+        append_runtime_report_error(
+            report,
+            stage=failure_category,
+            message=str(exc),
+            error_type=type(exc).__name__,
+            failure_category=failure_category,
+        )
+        finalize_runtime_report(
+            report,
+            status="error",
+            diagnostics={"failure_category": failure_category},
+        )
+        log_runtime_event(
+            log_context,
+            f"{failure_category}_failed",
+            message="IBKR execution readiness validation failed",
+            severity="ERROR",
+            error_type=type(exc).__name__,
+            error_message=str(exc),
+            failure_category=failure_category,
+        )
+        error_msg = f"🚨 【IBKR 执行条件异常】\n{str(exc)}"
+        _publish_runtime_failure_notification(
+            detailed_text=error_msg,
+            compact_text=error_msg,
+            exc=exc,
+        )
+        return "Error", 503
     except Exception as exc:
         append_runtime_report_error(
             report,

scripts/execution_report_heartbeat.py

@@ -17,6 +17,7 @@
 
 DEFAULT_ACCEPT_STATUSES = {"ok", "skipped", "success", "completed", "no_action"}
 DEFAULT_REJECT_STATUSES = {"error", "failed", "failure", "cancelled", "canceled", "timed_out"}
+DEFAULT_REJECT_EXECUTION_STATUSES = {"blocked", "error", "failed", "failure"}
 DEFAULT_ACCEPT_STAGES = {
     "DRY_RUN_COMPLETED",
     "FUNDING_BLOCKED",
@@ -749,6 +750,12 @@ def _report_status(payload: dict[str, Any]) -> tuple[str, str]:
     return status, stage
 
 
+def _report_execution_status(payload: dict[str, Any]) -> str:
+    summary = payload.get("summary")
+    nested_status = summary.get("execution_status") if isinstance(summary, dict) else None
+    return str(payload.get("execution_status") or nested_status or "").strip()
+
+
 def _payload_service_name(payload: dict[str, Any]) -> str:
     runtime_target = payload.get("runtime_target")
     service = payload.get("service_name")
@@ -806,11 +813,15 @@ def _is_accepted_report(payload: dict[str, Any]) -> tuple[bool, str]:
     status, stage = _report_status(payload)
     status_key = status.lower()
     stage_key = stage.upper()
+    execution_status = _report_execution_status(payload)
+    execution_status_key = execution_status.lower()
     errors = _report_errors(payload)
     if errors and not allow_errors:
         return False, f"errors={len(errors)} status={status or '-'} stage={stage or '-'}"
     if status_key in reject_statuses or stage_key in reject_stages:
         return False, f"rejected status={status or '-'} stage={stage or '-'}"
+    if execution_status_key in DEFAULT_REJECT_EXECUTION_STATUSES:
+        return False, f"rejected execution_status={execution_status}"
     if status_key and status_key in accepted_statuses:
         return True, f"status={status}"
     if stage_key and stage_key in accepted_stages: