diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index bd60ac8..77536b3 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -100,6 +100,9 @@ jobs: assert resolve_hk_canonical_profile("hk_global_etf_tactical_rotation") == "hk_global_etf_tactical_rotation" PY + - name: Validate production Cloud Run startup + run: uv run --no-sync python scripts/validate_cloud_run_startup.py + - name: Install editable shared repositories run: | set -euo pipefail diff --git a/Dockerfile b/Dockerfile index 5a98c0c..114fd54 100644 --- a/Dockerfile +++ b/Dockerfile @@ -16,6 +16,7 @@ RUN apt-get update \ COPY . . RUN python -m pip install --upgrade pip uv \ && uv sync --frozen --no-dev \ + && python scripts/validate_cloud_run_startup.py \ && apt-get purge -y git \ && apt-get autoremove -y --purge \ && rm -rf /var/lib/apt/lists/* \ diff --git a/application/runtime_broker_adapters.py b/application/runtime_broker_adapters.py index 2aaa097..40626f3 100644 --- a/application/runtime_broker_adapters.py +++ b/application/runtime_broker_adapters.py @@ -10,6 +10,10 @@ from quant_platform_kit.common.models import OrderIntent, PortfolioSnapshot, Position +class IBKRGatewayUnavailableError(ConnectionError): + """Raised after retryable IBKR gateway connection attempts are exhausted.""" + + @dataclass(frozen=True) class IBKRRuntimeBrokerAdapters: host_resolver: Any @@ -51,6 +55,7 @@ class IBKRRuntimeBrokerAdapters: execution_mode: str = "paper" limit_buy_premium_by_symbol: dict[str, float] | None = None printer: Any = print + refresh_host_fn: Any = None def validate_configured_accounts(self, ib): if not self.account_ids: @@ -119,9 +124,23 @@ def connect_ib(self): f"error={exc})", flush=True, ) - if attempt < self.connect_attempts and self.connect_retry_delay_seconds > 0: - self.sleep_fn(self.connect_retry_delay_seconds) - raise last_error + if attempt < self.connect_attempts: + if callable(self.refresh_host_fn): + try: + host = self.refresh_host_fn() + except Exception as refresh_exc: + self.printer( + "IB gateway host refresh failed; retrying last-known-good host " + f"(host={host}, error_type={type(refresh_exc).__name__}, " + f"error={refresh_exc})", + flush=True, + ) + if self.connect_retry_delay_seconds > 0: + self.sleep_fn(self.connect_retry_delay_seconds) + raise IBKRGatewayUnavailableError( + "IB gateway unavailable after " + f"{self.connect_attempts} attempt(s) to {host}:{self.ib_port}: {last_error}" + ) from last_error def get_current_portfolio(self, ib): snapshot = self.fetch_account_portfolio_snapshot(ib) @@ -300,6 +319,7 @@ def build_runtime_broker_adapters( execution_mode: str = "paper", limit_buy_premium_by_symbol: dict[str, float] | None = None, printer=print, + refresh_host_fn=None, ) -> IBKRRuntimeBrokerAdapters: return IBKRRuntimeBrokerAdapters( host_resolver=host_resolver, @@ -341,4 +361,5 @@ def build_runtime_broker_adapters( market_currency=str(market_currency or "USD").upper(), execution_mode=str(execution_mode or "paper").strip().lower().replace("-", "_"), printer=printer, + refresh_host_fn=refresh_host_fn, ) diff --git a/main.py b/main.py index ea5514c..64f9471 100644 --- a/main.py +++ b/main.py @@ -19,7 +19,10 @@ get_compute_discovery = None from application.cycle_result import coerce_strategy_cycle_result -from application.runtime_broker_adapters import build_runtime_broker_adapters +from application.runtime_broker_adapters import ( + IBKRGatewayUnavailableError, + build_runtime_broker_adapters, +) from application.runtime_composer import build_runtime_composer from application.runtime_deadline import ( RuntimeDeadlineExceeded, @@ -141,6 +144,26 @@ def get_ib_host(): return host +def refresh_ib_host(): + """Refresh a zoned gateway address while retaining last-known-good state.""" + global IB_HOST + instance_name = RUNTIME_SETTINGS.ib_gateway_instance_name + zone = RUNTIME_SETTINGS.ib_gateway_zone + if not zone: + return get_ib_host() + + cached_host = IB_HOST + try: + resolved_host = resolve_gce_instance_ip(instance_name, zone) + except Exception as exc: + print(f"GCE host refresh failed for {instance_name}: {exc}", flush=True) + resolved_host = cached_host or instance_name + if cached_host and resolved_host == instance_name: + resolved_host = cached_host + IB_HOST = resolved_host + return resolved_host + + def get_ib_gateway_mode(): return RUNTIME_SETTINGS.ib_gateway_mode @@ -514,6 +537,7 @@ def build_broker_adapters(*, dry_run_only_override: bool | None = None): effective_dry_run_only = RUNTIME_SETTINGS.dry_run_only if dry_run_only_override is None else bool(dry_run_only_override) return build_runtime_broker_adapters( host_resolver=get_ib_host, + refresh_host_fn=refresh_ib_host, ib_port=IB_PORT, ib_client_id=IB_CLIENT_ID, connect_timeout_seconds=IB_CONNECT_TIMEOUT_SECONDS, @@ -1309,21 +1333,27 @@ def _handle_request( }, ) raise - except TimeoutError as exc: + except IBKRGatewayUnavailableError as exc: append_runtime_report_error( report, stage="ibkr_connect", message=str(exc), error_type=type(exc).__name__, + failure_category="ibkr_gateway_unavailable", + ) + finalize_runtime_report( + report, + status="error", + diagnostics={"failure_category": "ibkr_gateway_unavailable"}, ) - finalize_runtime_report(report, status="error") log_runtime_event( log_context, - "ibkr_gateway_connect_timeout", - message="IBKR gateway handshake timed out", + "ibkr_gateway_connect_failed", + message="IBKR gateway connection attempts exhausted", severity="ERROR", error_type=type(exc).__name__, error_message=str(exc), + failure_category="ibkr_gateway_unavailable", ) error_msg = f"🚨 【IBKR 连接异常】\n{str(exc)}" _publish_runtime_failure_notification( @@ -1331,7 +1361,7 @@ def _handle_request( compact_text=error_msg, exc=exc, ) - return "Error", 500 + return "Error", 503 except Exception as exc: append_runtime_report_error( report, diff --git a/scripts/validate_cloud_run_startup.py b/scripts/validate_cloud_run_startup.py new file mode 100644 index 0000000..18ce7da --- /dev/null +++ b/scripts/validate_cloud_run_startup.py @@ -0,0 +1,79 @@ +#!/usr/bin/env python3 +"""Import the production WSGI app with an inert IBKR configuration. + +This check performs no network or broker operations. It catches dependency API +drift, module import failures, and Flask route registration conflicts before a +container can be deployed. +""" + +from __future__ import annotations + +import json +import os +import sys +from pathlib import Path + + +ROOT = Path(__file__).resolve().parents[1] +if str(ROOT) not in sys.path: + sys.path.insert(0, str(ROOT)) + + +def _install_smoke_environment() -> None: + account_group = "startup-smoke" + os.environ["ACCOUNT_GROUP"] = account_group + os.environ["IB_ACCOUNT_GROUP_CONFIG_JSON"] = json.dumps( + { + "groups": { + account_group: { + "ib_gateway_instance_name": "127.0.0.1", + "ib_gateway_mode": "paper", + "ib_client_id": 1, + "account_ids": ["DU000000"], + } + } + }, + separators=(",", ":"), + ) + os.environ["IB_ACCOUNT_GROUP_CONFIG_SECRET_NAME"] = "" + os.environ["RUNTIME_TARGET_JSON"] = json.dumps( + { + "platform_id": "ibkr", + "strategy_profile": "global_etf_rotation", + "dry_run_only": True, + "execution_mode": "paper", + "account_scope": account_group, + }, + separators=(",", ":"), + ) + + +def validate_startup() -> None: + _install_smoke_environment() + + import main + + routes = { + rule.rule: set(rule.methods) - {"HEAD", "OPTIONS"} + for rule in main.app.url_map.iter_rules() + } + required_routes = { + "/health": {"GET"}, + "/run": {"GET", "POST"}, + "/dry-run": {"GET", "POST"}, + "/probe": {"GET", "POST"}, + "/monitor-dispatch": {"GET", "POST"}, + } + missing_or_invalid = { + path: {"expected": sorted(methods), "actual": sorted(routes.get(path, set()))} + for path, methods in required_routes.items() + if routes.get(path) != methods + } + if missing_or_invalid: + raise RuntimeError(f"Cloud Run route contract is invalid: {missing_or_invalid}") + + print(f"Cloud Run startup validation passed: routes={len(routes)}") + + +if __name__ == "__main__": + validate_startup() diff --git a/tests/test_cloud_run_startup_validation.py b/tests/test_cloud_run_startup_validation.py new file mode 100644 index 0000000..c63eabc --- /dev/null +++ b/tests/test_cloud_run_startup_validation.py @@ -0,0 +1,14 @@ +from pathlib import Path + + +ROOT = Path(__file__).resolve().parents[1] + + +def test_production_startup_validation_is_a_ci_and_image_build_gate(): + command = "python scripts/validate_cloud_run_startup.py" + dockerfile = (ROOT / "Dockerfile").read_text() + ci_workflow = (ROOT / ".github" / "workflows" / "ci.yml").read_text() + + assert command in dockerfile + assert "Validate production Cloud Run startup" in ci_workflow + assert f"uv run --no-sync {command}" in ci_workflow diff --git a/tests/test_event_loop.py b/tests/test_event_loop.py index 19a695f..ea1b321 100644 --- a/tests/test_event_loop.py +++ b/tests/test_event_loop.py @@ -138,6 +138,42 @@ def get(self, project, zone, instance): assert module.IB_HOST == "10.0.0.8" +def test_get_ib_host_refreshes_zoned_gateway_ip(strategy_module_factory, monkeypatch): + module = strategy_module_factory( + IB_GATEWAY_ZONE="us-central1-a", + IB_ACCOUNT_GROUP_CONFIG_JSON=( + '{"groups":{"default":{"ib_gateway_instance_name":"ib-gateway",' + '"ib_gateway_mode":"live","ib_client_id":1}}}' + ), + ) + resolved = iter(("10.0.0.8", "10.0.0.9")) + monkeypatch.setattr(module, "resolve_gce_instance_ip", lambda *_args: next(resolved)) + + assert module.get_ib_host() == "10.0.0.8" + assert module.get_ib_host() == "10.0.0.8" + assert module.refresh_ib_host() == "10.0.0.9" + assert module.IB_HOST == "10.0.0.9" + + +def test_refresh_ib_host_keeps_cached_ip_when_gce_lookup_fails(strategy_module_factory, monkeypatch): + module = strategy_module_factory( + IB_GATEWAY_ZONE="us-central1-a", + IB_ACCOUNT_GROUP_CONFIG_JSON=( + '{"groups":{"default":{"ib_gateway_instance_name":"ib-gateway",' + '"ib_gateway_mode":"live","ib_client_id":1}}}' + ), + ) + module.IB_HOST = "10.0.0.8" + monkeypatch.setattr( + module, + "resolve_gce_instance_ip", + lambda *_args: (_ for _ in ()).throw(RuntimeError("compute API unavailable")), + ) + + assert module.refresh_ib_host() == "10.0.0.8" + assert module.IB_HOST == "10.0.0.8" + + def test_ib_gateway_mode_derives_paper_port(strategy_module_factory): module = strategy_module_factory( IB_ACCOUNT_GROUP_CONFIG_JSON=( diff --git a/tests/test_request_handling.py b/tests/test_request_handling.py index 6d4c970..b525f41 100644 --- a/tests/test_request_handling.py +++ b/tests/test_request_handling.py @@ -903,6 +903,79 @@ def test_handle_request_error_persists_machine_readable_report(strategy_module, assert len(observed["messages"]) == 1 +def test_handle_request_classifies_gateway_connection_failure_as_unavailable(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.IBKRGatewayUnavailableError( + "TCP preflight failed: connection refused" + ) + ), + ) + 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_connect" + assert observed["report"]["errors"][0]["failure_category"] == "ibkr_gateway_unavailable" + assert observed["report"]["diagnostics"]["failure_category"] == "ibkr_gateway_unavailable" + assert observed["events"][-1][0] == "ibkr_gateway_connect_failed" + assert observed["events"][-1][1]["failure_category"] == "ibkr_gateway_unavailable" + assert len(observed["notifications"]) == 1 + + +def test_handle_request_does_not_misclassify_unrelated_timeout_as_gateway_failure(strategy_module, monkeypatch): + observed = {"events": []} + + monkeypatch.setattr(strategy_module, "is_market_open_now", lambda **_kwargs: True) + monkeypatch.setattr( + strategy_module, + "run_strategy_core", + lambda **_kwargs: (_ for _ in ()).throw(TimeoutError("snapshot storage timed out")), + ) + 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: None) + + with strategy_module.app.test_request_context("/run", method="POST"): + body, status = strategy_module.handle_request() + + assert status == 500 + assert body == "Error" + assert observed["report"]["errors"][0]["stage"] == "strategy_cycle" + assert observed["events"][-1][0] == "strategy_cycle_failed" + + def test_run_strategy_core_allows_multiple_runs_in_same_process(strategy_module, monkeypatch): observed = {"connect_calls": 0, "disconnect_calls": 0, "messages": []} diff --git a/tests/test_runtime_broker_adapters.py b/tests/test_runtime_broker_adapters.py index c7c6cb1..0441317 100644 --- a/tests/test_runtime_broker_adapters.py +++ b/tests/test_runtime_broker_adapters.py @@ -2,7 +2,10 @@ import pytest -from application.runtime_broker_adapters import build_runtime_broker_adapters +from application.runtime_broker_adapters import ( + IBKRGatewayUnavailableError, + build_runtime_broker_adapters, +) def _build_adapters(*, account_ids=("U1234567",), execution_mode="paper"): @@ -53,6 +56,61 @@ def test_connect_ib_accepts_configured_managed_account(): assert ib.managedAccounts() == ["U1234567"] +def test_connect_ib_resolves_gateway_host_again_for_each_retry(): + observed = {"hosts": [], "refreshes": 0} + current_host = {"value": "10.0.0.8"} + + def resolve_host(): + return current_host["value"] + + def refresh_host(): + observed["refreshes"] += 1 + current_host["value"] = "10.0.0.9" + return current_host["value"] + + def connect(host, *_args, **_kwargs): + observed["hosts"].append(host) + if len(observed["hosts"]) == 1: + raise ConnectionRefusedError("gateway restarting") + return SimpleNamespace(managedAccounts=lambda: ["U1234567"]) + + adapters = _build_adapters() + adapters = adapters.__class__( + **{ + **adapters.__dict__, + "host_resolver": resolve_host, + "refresh_host_fn": refresh_host, + "connect_ib_fn": connect, + "connect_attempts": 2, + } + ) + + adapters.connect_ib() + + assert observed == { + "hosts": ["10.0.0.8", "10.0.0.9"], + "refreshes": 1, + } + + +def test_connect_ib_raises_dedicated_error_after_retries_are_exhausted(): + adapters = _build_adapters() + adapters = adapters.__class__( + **{ + **adapters.__dict__, + "connect_ib_fn": lambda *_args, **_kwargs: (_ for _ in ()).throw( + TimeoutError("handshake timeout") + ), + "connect_attempts": 2, + } + ) + + with pytest.raises(IBKRGatewayUnavailableError, match="unavailable after 2 attempt") as exc_info: + adapters.connect_ib() + + assert isinstance(exc_info.value.__cause__, TimeoutError) + + def test_connect_ib_rejects_configured_account_not_visible_to_gateway_username(): observed = {"disconnects": 0}