diff --git a/application/execution_service.py b/application/execution_service.py index fc05021..e677507 100644 --- a/application/execution_service.py +++ b/application/execution_service.py @@ -310,7 +310,13 @@ 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", +) def _coerce_float(value): @@ -2251,20 +2257,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) diff --git a/application/ibkr_order_execution.py b/application/ibkr_order_execution.py index 4f5c4f4..a4355c7 100644 --- a/application/ibkr_order_execution.py +++ b/application/ibkr_order_execution.py @@ -52,6 +52,42 @@ def factory(side: str, quantity: float) -> Any: return factory +def probe_order_write_access( + ib: Any, + *, + symbol: str, + account_id: str, + stock_factory: Callable[..., Any] | None = None, + market_order_factory: Callable[..., Any] | None = None, + stock_exchange: str = "SMART", + stock_currency: str = "USD", +) -> Any: + """Verify order-write access with an IBKR what-if order that cannot execute.""" + + what_if_order = getattr(ib, "whatIfOrder", None) + if not callable(what_if_order): + raise RuntimeError("IBKR connection does not support what-if orders") + + normalized_symbol = str(symbol or "").strip().upper() + normalized_account_id = str(account_id or "").strip() + if not normalized_symbol or not normalized_account_id: + raise ValueError("IBKR what-if probe requires a symbol and account_id") + + contract = _stock_factory_for_market( + stock_factory, + exchange=str(stock_exchange or "SMART").upper(), + currency=str(stock_currency or "USD").upper(), + )(normalized_symbol) + order = _market_order_factory_with_time_in_force( + market_order_factory, + time_in_force=DEFAULT_TIME_IN_FORCE, + )("BUY", 1) + order.account = normalized_account_id + order.whatIf = True + order.transmit = True + return what_if_order(contract, order) + + def submit_order_intent( ib: Any, order_intent: OrderIntent, diff --git a/application/ibkr_portfolio.py b/application/ibkr_portfolio.py index ca86382..51911db 100644 --- a/application/ibkr_portfolio.py +++ b/application/ibkr_portfolio.py @@ -8,7 +8,17 @@ 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", +) + + +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 +136,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 +152,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 +163,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 market_currency_cash is None: + raise IBKRPortfolioSnapshotUnavailableError( + f"IBKR cash-only snapshot is missing the {market_currency} cash balance." + ) 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 = { diff --git a/application/rebalance_service.py b/application/rebalance_service.py index 0c56b62..7dca436 100644 --- a/application/rebalance_service.py +++ b/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 {}), diff --git a/application/runtime_broker_adapters.py b/application/runtime_broker_adapters.py index 40626f3..190fc7d 100644 --- a/application/runtime_broker_adapters.py +++ b/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-write access.""" + + @dataclass(frozen=True) class IBKRRuntimeBrokerAdapters: host_resolver: Any @@ -56,6 +60,7 @@ class IBKRRuntimeBrokerAdapters: limit_buy_premium_by_symbol: dict[str, float] | None = None printer: Any = print refresh_host_fn: Any = None + trading_permission_probe_fn: Any = None def validate_configured_accounts(self, ib): if not self.account_ids: @@ -85,6 +90,69 @@ 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 + permission_probe = self.trading_permission_probe_fn + if not callable(permission_probe): + raise IBKRTradingPermissionError( + "IB Gateway live execution cannot verify non-transmitting order-write 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 + # IBKR what-if validation exercises order-write access without creating a live order. + ib.RaiseRequestErrors = True + ib.RequestTimeout = self.connect_timeout_seconds + probe_result = permission_probe(ib) + probe_warning = getattr(probe_result, "warningText", "") + if read_only_errors or is_read_only_error(probe_warning): + 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 (ConnectionError, OSError) as exc: + if read_only_errors or is_read_only_error(exc): + 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 non-transmitting order-write 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 +176,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): @@ -320,6 +389,7 @@ def build_runtime_broker_adapters( limit_buy_premium_by_symbol: dict[str, float] | None = None, printer=print, refresh_host_fn=None, + trading_permission_probe_fn=None, ) -> IBKRRuntimeBrokerAdapters: return IBKRRuntimeBrokerAdapters( host_resolver=host_resolver, @@ -362,4 +432,5 @@ def build_runtime_broker_adapters( execution_mode=str(execution_mode or "paper").strip().lower().replace("-", "_"), printer=printer, refresh_host_fn=refresh_host_fn, + trading_permission_probe_fn=trading_permission_probe_fn, ) diff --git a/entrypoints/cloud_run.py b/entrypoints/cloud_run.py index d66c1b8..ae1a39b 100644 --- a/entrypoints/cloud_run.py +++ b/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( diff --git a/main.py b/main.py index 64f9471..e05f07e 100644 --- a/main.py +++ b/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 @@ -66,7 +67,7 @@ fetch_historical_price_series, fetch_quote_snapshots, ) -from application.ibkr_order_execution import submit_order_intent +from application.ibkr_order_execution import probe_order_write_access, submit_order_intent from application.monitor_dispatcher import ( dispatch_due_monitor_targets, load_monitor_targets, @@ -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, @@ -93,6 +97,21 @@ ensure_event_loop = ibkr_ensure_event_loop NEW_YORK_TZ = ZoneInfo("America/New_York") STRATEGY_RUN_LOCK = threading.Lock() +EXPECTED_EXECUTION_GUARD_REASON_PREFIXES = ( + "pending_orders_detected:", + "same_day_fills_detected:", + "same_day_execution_locked:", +) + + +def _is_execution_failure(execution_status, no_op_reason) -> bool: + status = str(execution_status or "").strip().lower() + if status in {"error", "failed", "failure"}: + return True + if status != "blocked": + return False + reason = str(no_op_reason or "").strip().lower() + return not reason.startswith(EXPECTED_EXECUTION_GUARD_REASON_PREFIXES) def get_project_id(): @@ -524,6 +543,21 @@ def submit_market_order_intent(ib, order_intent, **kwargs): ) +def probe_market_order_write_access(ib): + managed_symbols = resolve_reporting_managed_symbols() + if len(ACCOUNT_IDS) != 1 or not managed_symbols: + raise RuntimeError( + "IBKR what-if permission probe requires one account_id and one managed symbol" + ) + return probe_order_write_access( + ib, + symbol=managed_symbols[0], + account_id=ACCOUNT_IDS[0], + stock_exchange=MARKET_EXCHANGE, + stock_currency=MARKET_CURRENCY, + ) + + def fetch_market_portfolio_snapshot(ib, **kwargs): return fetch_portfolio_snapshot( ib, @@ -538,6 +572,7 @@ def build_broker_adapters(*, dry_run_only_override: bool | None = None): return build_runtime_broker_adapters( host_resolver=get_ib_host, refresh_host_fn=refresh_ib_host, + trading_permission_probe_fn=probe_market_order_write_access, ib_port=IB_PORT, ib_client_id=IB_CLIENT_ID, connect_timeout_seconds=IB_CONNECT_TIMEOUT_SECONDS, @@ -1292,28 +1327,56 @@ 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_failed = _is_execution_failure( + execution_status, + report_summary.get("no_op_reason"), + ) + if execution_failed: + 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_failed: + report_diagnostics["failure_category"] = "strategy_execution_blocked" finalize_runtime_report( report, - status="ok", + status="error" if execution_failed 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_failed else "strategy_cycle_completed", + message=( + "Strategy execution blocked" + if execution_failed + else "Strategy dry-run completed" + if dry_run_only_override + else "Strategy execution completed" + ), + severity="ERROR" if execution_failed 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 +1425,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, diff --git a/scripts/execution_report_heartbeat.py b/scripts/execution_report_heartbeat.py index 4211436..fe6182b 100644 --- a/scripts/execution_report_heartbeat.py +++ b/scripts/execution_report_heartbeat.py @@ -17,6 +17,12 @@ DEFAULT_ACCEPT_STATUSES = {"ok", "skipped", "success", "completed", "no_action"} DEFAULT_REJECT_STATUSES = {"error", "failed", "failure", "cancelled", "canceled", "timed_out"} +DEFAULT_REJECT_EXECUTION_STATUSES = {"error", "failed", "failure"} +EXPECTED_EXECUTION_GUARD_REASON_PREFIXES = ( + "pending_orders_detected:", + "same_day_fills_detected:", + "same_day_execution_locked:", +) DEFAULT_ACCEPT_STAGES = { "DRY_RUN_COMPLETED", "FUNDING_BLOCKED", @@ -749,6 +755,24 @@ 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 _report_no_op_reason(payload: dict[str, Any]) -> str: + summary = payload.get("summary") + nested_reason = summary.get("no_op_reason") if isinstance(summary, dict) else None + return str(payload.get("no_op_reason") or nested_reason or "").strip() + + +def _is_expected_execution_guard(execution_status: str, no_op_reason: str) -> bool: + return execution_status.lower() == "blocked" and no_op_reason.lower().startswith( + EXPECTED_EXECUTION_GUARD_REASON_PREFIXES + ) + + def _payload_service_name(payload: dict[str, Any]) -> str: runtime_target = payload.get("runtime_target") service = payload.get("service_name") @@ -806,11 +830,19 @@ 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() + no_op_reason = _report_no_op_reason(payload) 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 or ( + execution_status_key == "blocked" + and not _is_expected_execution_guard(execution_status, no_op_reason) + ): + 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: diff --git a/tests/test_cloud_run_entrypoint.py b/tests/test_cloud_run_entrypoint.py index 3ddae4e..2331c46 100644 --- a/tests/test_cloud_run_entrypoint.py +++ b/tests/test_cloud_run_entrypoint.py @@ -47,3 +47,18 @@ def test_is_market_open_now_falls_back_to_weekday_when_calendar_import_fails(mon ) assert is_market_open_now() is True + + +def test_is_market_open_now_returns_false_before_regular_session(monkeypatch): + market_timezone = pytz.timezone("America/New_York") + premarket_time = market_timezone.localize(datetime(2026, 4, 6, 2, 0, 0)) + monkeypatch.setattr( + "entrypoints.cloud_run.datetime", + type( + "FakeDatetime", + (), + {"now": staticmethod(lambda _tz: premarket_time)}, + ), + ) + + assert is_market_open_now() is False diff --git a/tests/test_execution_report_heartbeat.py b/tests/test_execution_report_heartbeat.py index eb5d29b..fb15bf8 100644 --- a/tests/test_execution_report_heartbeat.py +++ b/tests/test_execution_report_heartbeat.py @@ -492,6 +492,44 @@ def test_runtime_target_scheduler_does_not_skip_when_lookback_includes_scheduler assert reason is None +def test_report_with_blocked_execution_status_is_rejected_even_when_top_level_is_ok(): + accepted, reason = heartbeat._is_accepted_report( + { + "status": "ok", + "summary": { + "execution_status": "blocked", + "no_op_reason": "no_equity", + }, + } + ) + + assert accepted is False + assert reason == "rejected execution_status=blocked" + + +@pytest.mark.parametrize( + "no_op_reason", + [ + "pending_orders_detected:AAA", + "same_day_fills_detected:AAA", + "same_day_execution_locked:mode=live", + ], +) +def test_report_with_expected_execution_guard_is_accepted_when_top_level_is_ok(no_op_reason): + accepted, reason = heartbeat._is_accepted_report( + { + "status": "ok", + "summary": { + "execution_status": "blocked", + "no_op_reason": no_op_reason, + }, + } + ) + + assert accepted is True + assert reason == "status=ok" + + def test_telegram_token_falls_back_to_secret_manager(monkeypatch): monkeypatch.delenv("TELEGRAM_TOKEN", raising=False) monkeypatch.delenv("TG_TOKEN", raising=False) diff --git a/tests/test_execution_service.py b/tests/test_execution_service.py index acbcef2..8a1f674 100644 --- a/tests/test_execution_service.py +++ b/tests/test_execution_service.py @@ -107,6 +107,27 @@ def accountValues(self): assert get_available_buying_power(FakeIB(), 885.99, currency="USD") == 477.10 +def test_get_available_buying_power_supports_ibkr_ledger_cash_balance(): + class FakeIB: + def accountValues(self): + return [ + SimpleNamespace(tag="AvailableFunds", currency="USD", value="885.99"), + SimpleNamespace(tag="$LEDGER-CashBalance", currency="USD", value="477.10"), + ] + + assert get_available_buying_power(FakeIB(), 885.99, currency="USD") == 477.10 + + +def test_get_available_buying_power_does_not_treat_total_cash_value_as_currency_cash(): + class FakeIB: + def accountValues(self): + return [ + SimpleNamespace(tag="TotalCashValue", currency="USD", value="500.00"), + ] + + assert get_available_buying_power(FakeIB(), 0.0, currency="USD") == 0.0 + + def test_execute_rebalance_submits_limit_buy_for_underweight_position(monkeypatch, tmp_path): class FakeIB: def openTrades(self): @@ -163,6 +184,55 @@ def fake_fetch_quote_snapshots(_ib, symbols): assert any(log.startswith("buy VOO") for log in trade_logs) +def test_execute_rebalance_blocks_when_all_order_submissions_are_rejected(tmp_path): + class FakeIB: + def openTrades(self): + return [] + + def fills(self): + return [] + + def accountValues(self): + return [SimpleNamespace(tag="CashBalance", currency="USD", value="1000")] + + trade_logs, summary = execute_rebalance( + FakeIB(), + {"VOO": 1.0}, + {}, + {"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 *_args, **_kwargs: SimpleNamespace( + broker_order_id=None, + status="Rejected", + ), + order_intent_cls=OrderIntent, + translator=translate, + strategy_symbols=["VOO"], + strategy_profile="tech_communication_pullback_enhancement", + signal_metadata=_signal_metadata( + {"VOO": 1.0}, + risk_symbols=("VOO",), + trade_date="2026-04-01", + snapshot_as_of="2026-03-31", + ), + dry_run_only=False, + cash_reserve_ratio=0.0, + rebalance_threshold_ratio=0.02, + limit_buy_premium=1.005, + sell_settle_delay_sec=0, + execution_lock_dir=tmp_path, + return_summary=True, + ) + + assert summary["execution_status"] == "blocked" + assert summary["no_op_reason"] == "submit_failed:VOO:Rejected" + assert summary["skipped_reasons"] == ["submit_failed:VOO:Rejected"] + assert summary["orders_submitted"] == [] + assert "failed submit_failed:VOO:Rejected" in trade_logs + + def test_execute_rebalance_uses_symbol_specific_limit_buy_premium(monkeypatch, tmp_path): class FakeIB: def openTrades(self): diff --git a/tests/test_ibkr_order_execution.py b/tests/test_ibkr_order_execution.py index 2b7f45b..b84baf9 100644 --- a/tests/test_ibkr_order_execution.py +++ b/tests/test_ibkr_order_execution.py @@ -2,7 +2,7 @@ from quant_platform_kit.common.models import OrderIntent -from application.ibkr_order_execution import submit_order_intent +from application.ibkr_order_execution import probe_order_write_access, submit_order_intent class FakeMarketOrder: @@ -37,6 +37,17 @@ def placeOrder(self, contract, order): ) +class FakeWhatIfIB: + def __init__(self): + self.probed_contract = None + self.probed_order = None + + def whatIfOrder(self, contract, order): + self.probed_contract = contract + self.probed_order = order + return SimpleNamespace(warningText="") + + def fake_stock(symbol, exchange, currency): return SimpleNamespace(symbol=symbol, exchange=exchange, currency=currency) @@ -52,6 +63,23 @@ def fake_option(symbol, expiration, strike, right, exchange, currency): ) +def test_probe_order_write_access_uses_explicit_what_if_order(): + ib = FakeWhatIfIB() + + probe_order_write_access( + ib, + symbol="AAPL", + account_id="U1234567", + stock_factory=fake_stock, + market_order_factory=FakeMarketOrder, + ) + + assert ib.probed_contract.symbol == "AAPL" + assert ib.probed_order.account == "U1234567" + assert ib.probed_order.whatIf is True + assert ib.probed_order.transmit is True + + def test_submit_order_intent_sets_default_day_tif_on_market_orders(): ib = FakeIB() diff --git a/tests/test_ibkr_portfolio.py b/tests/test_ibkr_portfolio.py index b8eb914..7212ed1 100644 --- a/tests/test_ibkr_portfolio.py +++ b/tests/test_ibkr_portfolio.py @@ -1,5 +1,8 @@ from types import SimpleNamespace +import pytest + +from application import ibkr_portfolio from application.ibkr_portfolio import fetch_portfolio_snapshot @@ -44,6 +47,7 @@ def accountValues(self): return [ SimpleNamespace(account="UHK123", currency="HKD", tag="NetLiquidation", value="100000"), SimpleNamespace(account="UHK123", currency="HKD", tag="AvailableFunds", value="80000"), + SimpleNamespace(account="UHK123", currency="HKD", tag="CashBalance", value="0"), SimpleNamespace(account="UHK123", currency="USD", tag="NetLiquidation", value="999"), SimpleNamespace(account="UUS999", currency="HKD", tag="NetLiquidation", value="123"), ] @@ -96,6 +100,59 @@ def accountValues(self): assert snapshot.metadata["available_funds"] == 885.99 +def test_fetch_portfolio_snapshot_supports_ibkr_ledger_cash_balance(): + class LedgerCashIB(FakeIB): + def positions(self): + return [] + + def accountValues(self): + return [ + SimpleNamespace(account="U16608560", currency="USD", tag="NetLiquidation", value="1130"), + SimpleNamespace(account="U16608560", currency="USD", tag="AvailableFunds", value="885.99"), + SimpleNamespace( + account="U16608560", + currency="USD", + tag="$LEDGER-CashBalance", + value="477.10", + ), + SimpleNamespace(account="U16608560", currency="USD", tag="TotalCashValue", value="500.00"), + ] + + snapshot = fetch_portfolio_snapshot( + LedgerCashIB(), + account_ids=("U16608560",), + wait_seconds=0, + currency="USD", + ) + + assert snapshot.total_equity == 477.10 + assert snapshot.buying_power == 477.10 + assert snapshot.metadata["market_currency_cash"] == 477.10 + + +def test_fetch_portfolio_snapshot_rejects_total_cash_value_as_currency_cash(): + class AggregateCashIB(FakeIB): + def positions(self): + return [] + + def accountValues(self): + return [ + SimpleNamespace(account="U16608560", currency="USD", tag="NetLiquidation", value="1130"), + SimpleNamespace(account="U16608560", currency="USD", tag="TotalCashValue", value="500.00"), + ] + + with pytest.raises( + ibkr_portfolio.IBKRPortfolioSnapshotUnavailableError, + match="cash balance", + ): + fetch_portfolio_snapshot( + AggregateCashIB(), + account_ids=("U16608560",), + wait_seconds=0, + currency="USD", + ) + + def test_fetch_portfolio_snapshot_allows_negative_cash_balance(): class NegativeCashIB(FakeIB): def positions(self): @@ -117,3 +174,90 @@ def accountValues(self): assert snapshot.buying_power == -284.0 assert snapshot.metadata["market_currency_cash"] == -284.0 + + +def test_fetch_portfolio_snapshot_rejects_incomplete_cash_only_account_data(): + class MissingCashIB(FakeIB): + def positions(self): + return [] + + def accountValues(self): + return [ + SimpleNamespace( + account="U16608560", + currency="USD", + tag="NetLiquidation", + value="1130", + ) + ] + + with pytest.raises( + ibkr_portfolio.IBKRPortfolioSnapshotUnavailableError, + match="cash balance", + ): + fetch_portfolio_snapshot( + MissingCashIB(), + account_ids=("U16608560",), + wait_seconds=0, + currency="USD", + ) + + +def test_fetch_portfolio_snapshot_rejects_missing_cash_when_positions_exist(): + class MissingCashWithPositionIB(FakeIB): + def positions(self): + return [ + SimpleNamespace( + account="UUS999", + contract=SimpleNamespace(secType="STK", symbol="AAPL", currency="USD"), + position=5, + avgCost=190.0, + ) + ] + + def accountValues(self): + return [ + SimpleNamespace( + account="UUS999", + currency="USD", + tag="NetLiquidation", + value="1130", + ) + ] + + with pytest.raises( + ibkr_portfolio.IBKRPortfolioSnapshotUnavailableError, + match="cash balance", + ): + fetch_portfolio_snapshot( + MissingCashWithPositionIB(), + account_ids=("UUS999",), + wait_seconds=0, + currency="USD", + ) + + +def test_fetch_portfolio_snapshot_allows_explicit_zero_cash_balance(): + class ZeroCashIB(FakeIB): + def positions(self): + return [] + + def accountValues(self): + return [ + SimpleNamespace( + account="U16608560", + currency="USD", + tag="CashBalance", + value="0", + ) + ] + + snapshot = fetch_portfolio_snapshot( + ZeroCashIB(), + account_ids=("U16608560",), + wait_seconds=0, + currency="USD", + ) + + assert snapshot.total_equity == 0.0 + assert snapshot.metadata["market_currency_cash"] == 0.0 diff --git a/tests/test_rebalance_service.py b/tests/test_rebalance_service.py index 1e62c62..a965024 100644 --- a/tests/test_rebalance_service.py +++ b/tests/test_rebalance_service.py @@ -387,6 +387,56 @@ def disconnect(self): assert "目标差异" not in observed["messages"][0] +def test_run_strategy_core_propagates_blocked_execution_status(tmp_path): + class FakeIB: + def isConnected(self): + return True + + def disconnect(self): + return None + + reason = "submit_failed:AAA:Rejected" + result = run_strategy_core( + connect_ib=lambda: FakeIB(), + get_current_portfolio=lambda _ib: ({}, {"equity": 1000.0, "buying_power": 1000.0}), + compute_signals=lambda _ib, _holdings: ( + {"AAA": 1.0}, + "signal", + False, + "breadth=60.0%", + { + "strategy_profile": "tech_communication_pullback_enhancement", + "managed_symbols": ("AAA",), + "trade_date": "2026-04-01", + "snapshot_as_of": "2026-03-31", + "allocation": _weight_allocation({"AAA": 1.0}, risk_symbols=("AAA",)), + }, + ), + execute_rebalance=lambda *_args, **_kwargs: ( + [f"failed {reason}"], + { + "execution_status": "blocked", + "no_op_reason": reason, + "orders_submitted": [], + "orders_filled": [], + "orders_partially_filled": [], + "orders_skipped": [{"symbol": "AAA", "reason": "Rejected"}], + "skipped_reasons": [reason], + }, + ), + send_tg_message=lambda _message: None, + config=IBKRRebalanceConfig( + translator=_build_test_translator(), + separator="---", + notify_no_trade_cycles=False, + reconciliation_output_path=tmp_path / "reconciliation.json", + ), + ) + + assert result.result == f"Blocked - {reason}" + assert result.execution_summary["execution_status"] == "blocked" + + def test_trade_notification_keeps_detailed_logs_out_of_compact_message(): notification = render_trade_notification( dashboard="📌 Strategy portfolio\n - Total assets: $1,000.00", diff --git a/tests/test_request_handling.py b/tests/test_request_handling.py index b525f41..d4cfb76 100644 --- a/tests/test_request_handling.py +++ b/tests/test_request_handling.py @@ -788,6 +788,91 @@ def fake_run_strategy_core(**_kwargs): assert observed["report"]["artifacts"]["reconciliation_record_path"] == "/tmp/reconciliation.json" +def test_handle_request_marks_blocked_cycle_as_report_error_without_http_retry( + strategy_module_factory, + monkeypatch, +): + strategy_module = strategy_module_factory(IBKR_DRY_RUN_ONLY="false") + observed = {"events": []} + + monkeypatch.setattr(strategy_module, "build_run_id", lambda: "run-blocked") + monkeypatch.setattr(strategy_module, "is_market_open_now", lambda **_kwargs: True) + monkeypatch.setattr( + strategy_module, + "run_strategy_core", + lambda **_kwargs: StrategyCycleResult( + result="Blocked - no equity", + execution_summary={ + "execution_status": "blocked", + "no_op_reason": "no_equity", + }, + ), + ) + monkeypatch.setattr( + strategy_module, + "persist_execution_report", + lambda report: observed.setdefault("report", dict(report)) or "/tmp/runtime-report.json", + ) + monkeypatch.setattr( + strategy_module, + "log_runtime_event", + lambda _context, event, **fields: observed["events"].append((event, fields)), + ) + + with strategy_module.app.test_request_context("/run", method="POST"): + body, status = strategy_module.handle_request() + + assert status == 200 + assert body == "Blocked - no equity" + assert observed["report"]["status"] == "error" + assert observed["report"]["summary"]["execution_status"] == "blocked" + assert observed["report"]["diagnostics"]["failure_category"] == "strategy_execution_blocked" + assert observed["events"][-1][0] == "strategy_cycle_blocked" + assert observed["events"][-1][1]["severity"] == "ERROR" + + +def test_handle_request_keeps_expected_execution_guard_as_success( + strategy_module_factory, + monkeypatch, +): + strategy_module = strategy_module_factory(IBKR_DRY_RUN_ONLY="false") + observed = {"events": []} + + monkeypatch.setattr(strategy_module, "build_run_id", lambda: "run-pending-order") + monkeypatch.setattr(strategy_module, "is_market_open_now", lambda **_kwargs: True) + monkeypatch.setattr( + strategy_module, + "run_strategy_core", + lambda **_kwargs: StrategyCycleResult( + result="Blocked - pending order", + execution_summary={ + "execution_status": "blocked", + "no_op_reason": "pending_orders_detected:AAA", + }, + ), + ) + monkeypatch.setattr( + strategy_module, + "persist_execution_report", + lambda report: observed.setdefault("report", dict(report)) or "/tmp/runtime-report.json", + ) + monkeypatch.setattr( + strategy_module, + "log_runtime_event", + lambda _context, event, **fields: observed["events"].append((event, fields)), + ) + + with strategy_module.app.test_request_context("/run", method="POST"): + body, status = strategy_module.handle_request() + + assert status == 200 + assert body == "Blocked - pending order" + assert observed["report"]["status"] == "ok" + assert "failure_category" not in observed["report"]["diagnostics"] + assert observed["events"][-1][0] == "strategy_cycle_completed" + assert observed["events"][-1][1]["severity"] == "INFO" + + def test_cycle_report_summary_counts_dry_run_order_previews(strategy_module): cycle_result = StrategyCycleResult(result="Precheck OK") execution_summary = { @@ -946,6 +1031,51 @@ def test_handle_request_classifies_gateway_connection_failure_as_unavailable(str assert len(observed["notifications"]) == 1 +def test_handle_request_classifies_gateway_read_only_as_trading_permission_failure( + strategy_module, + monkeypatch, +): + observed = {"events": [], "notifications": []} + + monkeypatch.setattr(strategy_module, "is_market_open_now", lambda **_kwargs: True) + monkeypatch.setattr( + strategy_module, + "run_strategy_core", + lambda **_kwargs: (_ for _ in ()).throw( + strategy_module.IBKRTradingPermissionError( + "IB Gateway API is in Read-Only mode; live execution is disabled." + ) + ), + ) + monkeypatch.setattr( + strategy_module, + "persist_execution_report", + lambda report: observed.setdefault("report", dict(report)) or "/tmp/runtime-report.json", + ) + monkeypatch.setattr( + strategy_module, + "log_runtime_event", + lambda _context, event, **fields: observed["events"].append((event, fields)), + ) + monkeypatch.setattr( + strategy_module, + "publish_notification", + lambda **kwargs: observed["notifications"].append(kwargs), + ) + + with strategy_module.app.test_request_context("/run", method="POST"): + body, status = strategy_module.handle_request() + + assert status == 503 + assert body == "Error" + assert observed["report"]["status"] == "error" + assert observed["report"]["errors"][0]["stage"] == "ibkr_trading_permission" + assert observed["report"]["diagnostics"]["failure_category"] == "ibkr_trading_permission" + assert observed["events"][-1][0] == "ibkr_trading_permission_failed" + assert observed["events"][-1][1]["severity"] == "ERROR" + assert len(observed["notifications"]) == 1 + + def test_handle_request_does_not_misclassify_unrelated_timeout_as_gateway_failure(strategy_module, monkeypatch): observed = {"events": []} diff --git a/tests/test_runtime_broker_adapters.py b/tests/test_runtime_broker_adapters.py index 0441317..96e456f 100644 --- a/tests/test_runtime_broker_adapters.py +++ b/tests/test_runtime_broker_adapters.py @@ -4,11 +4,17 @@ from application.runtime_broker_adapters import ( IBKRGatewayUnavailableError, + IBKRTradingPermissionError, build_runtime_broker_adapters, ) -def _build_adapters(*, account_ids=("U1234567",), execution_mode="paper"): +def _build_adapters( + *, + account_ids=("U1234567",), + execution_mode="paper", + trading_permission_probe_fn=None, +): return build_runtime_broker_adapters( host_resolver=lambda: "127.0.0.1", ib_port=4001, @@ -45,6 +51,11 @@ def _build_adapters(*, account_ids=("U1234567",), execution_mode="paper"): strategy_display_name="Test Strategy", sleep_fn=lambda _seconds: None, printer=lambda *_args, **_kwargs: None, + trading_permission_probe_fn=( + trading_permission_probe_fn + if trading_permission_probe_fn is not None + else lambda ib: ib.whatIfOrder(None, None) + ), ) @@ -157,3 +168,187 @@ def disconnect(self): adapters.connect_ib() assert observed["disconnects"] == 1 + + +def test_connect_ib_rejects_live_gateway_read_only_mode_without_retry(): + observed = {"disconnects": 0, "permission_probe_requests": 0} + + class FakeEvent: + def __init__(self): + self.handlers = [] + + def __iadd__(self, handler): + self.handlers.append(handler) + return self + + def __isub__(self, handler): + self.handlers.remove(handler) + return self + + def emit(self, *args): + for handler in tuple(self.handlers): + handler(*args) + + class FakeIB: + RaiseRequestErrors = False + RequestTimeout = 0 + + def __init__(self): + self.errorEvent = FakeEvent() + + def managedAccounts(self): + return ["U1234567"] + + def whatIfOrder(self, _contract, _order): + observed["permission_probe_requests"] += 1 + self.errorEvent.emit(-1, 321, "API is in Read-Only mode", None) + raise TimeoutError("open orders timed out") + + def disconnect(self): + observed["disconnects"] += 1 + + adapters = _build_adapters(account_ids=("U1234567",), execution_mode="live") + adapters = adapters.__class__( + **{ + **adapters.__dict__, + "connect_ib_fn": lambda *_args, **_kwargs: FakeIB(), + "connect_attempts": 3, + } + ) + + with pytest.raises(IBKRTradingPermissionError, match="Read-Only"): + adapters.connect_ib() + + assert observed == { + "disconnects": 1, + "permission_probe_requests": 1, + } + + +def test_connect_ib_retries_when_trading_permission_probe_loses_connection(): + observed = { + "connects": 0, + "disconnects": 0, + "permission_probe_requests": 0, + "refreshes": 0, + } + + class FakeEvent: + def __init__(self): + self.handlers = [] + + def __iadd__(self, handler): + self.handlers.append(handler) + return self + + def __isub__(self, handler): + self.handlers.remove(handler) + return self + + class FakeIB: + RaiseRequestErrors = False + RequestTimeout = 0 + + def __init__(self, attempt): + self.attempt = attempt + self.errorEvent = FakeEvent() + + def managedAccounts(self): + return ["U1234567"] + + def whatIfOrder(self, _contract, _order): + observed["permission_probe_requests"] += 1 + if self.attempt == 1: + raise ConnectionError("gateway connection dropped") + return [] + + def disconnect(self): + observed["disconnects"] += 1 + + def connect(*_args, **_kwargs): + observed["connects"] += 1 + return FakeIB(observed["connects"]) + + def refresh_host(): + observed["refreshes"] += 1 + return "127.0.0.1" + + adapters = _build_adapters(account_ids=("U1234567",), execution_mode="live") + adapters = adapters.__class__( + **{ + **adapters.__dict__, + "connect_ib_fn": connect, + "connect_attempts": 2, + "refresh_host_fn": refresh_host, + } + ) + + assert adapters.connect_ib().managedAccounts() == ["U1234567"] + assert observed == { + "connects": 2, + "disconnects": 1, + "permission_probe_requests": 2, + "refreshes": 1, + } + + +def test_connect_ib_does_not_misclassify_unrelated_321_as_read_only(): + class FakeEvent: + def __init__(self): + self.handlers = [] + + def __iadd__(self, handler): + self.handlers.append(handler) + return self + + def __isub__(self, handler): + self.handlers.remove(handler) + return self + + def emit(self, *args): + for handler in tuple(self.handlers): + handler(*args) + + class FakeIB: + RaiseRequestErrors = False + RequestTimeout = 0 + + def __init__(self): + self.errorEvent = FakeEvent() + + def managedAccounts(self): + return ["U1234567"] + + def whatIfOrder(self, _contract, _order): + self.errorEvent.emit(-1, 321, "Generic validation error", None) + return SimpleNamespace(warningText="") + + adapters = _build_adapters(account_ids=("U1234567",), execution_mode="live") + adapters = adapters.__class__( + **{ + **adapters.__dict__, + "connect_ib_fn": lambda *_args, **_kwargs: FakeIB(), + } + ) + + assert adapters.connect_ib().managedAccounts() == ["U1234567"] + + +def test_connect_ib_skips_trading_permission_probe_for_dry_run(): + class FakeIB: + def managedAccounts(self): + return ["U1234567"] + + def whatIfOrder(self, _contract, _order): + pytest.fail("dry-run connection must not probe trading permissions") + + adapters = _build_adapters(account_ids=("U1234567",), execution_mode="live") + adapters = adapters.__class__( + **{ + **adapters.__dict__, + "connect_ib_fn": lambda *_args, **_kwargs: FakeIB(), + "dry_run_only": True, + } + ) + + assert adapters.connect_ib().managedAccounts() == ["U1234567"]