diff --git a/application/runtime_reporting_adapters.py b/application/runtime_reporting_adapters.py index 1cd1ca4..96fed96 100644 --- a/application/runtime_reporting_adapters.py +++ b/application/runtime_reporting_adapters.py @@ -95,6 +95,10 @@ def persist_execution_report(self, report: dict[str, Any]) -> str | None: ) if isinstance(persisted, str): return persisted + missing = object() + cloud_uri = getattr(persisted, "cloud_uri", missing) + if cloud_uri is not missing: + return cloud_uri or getattr(persisted, "local_path", None) return getattr(persisted, "gcs_uri", None) or getattr(persisted, "local_path", None) diff --git a/entrypoints/cloud_run.py b/entrypoints/cloud_run.py index 37d6c3e..879a59b 100644 --- a/entrypoints/cloud_run.py +++ b/entrypoints/cloud_run.py @@ -16,6 +16,11 @@ def is_market_open_now(*, calendar_name="NYSE", timezone_name="America/New_York" schedule = calendar.schedule(start_date=now_market.date(), end_date=now_market.date()) if schedule.empty: return False, None - return calendar.open_at_time(schedule, now_market), None + try: + return calendar.open_at_time(schedule, now_market), None + except ValueError as exc: + if "not covered by the schedule" not in str(exc): + raise + return False, None except Exception as exc: return False, exc diff --git a/main.py b/main.py index 2ce21f2..0a175ee 100644 --- a/main.py +++ b/main.py @@ -200,6 +200,7 @@ def _profile_estimate_max_purchase_quantity(t_ctx, symbol, **kwargs): ) SEPARATOR = "━━━━━━━━━━━━━━━━━━" +COMPACT_ERROR_NOTIFICATION_MAX_CHARS = 3500 def _load_limit_buy_premium_by_symbol(*env_names: str) -> dict[str, float]: @@ -239,6 +240,22 @@ def t(key, **kwargs): return build_translator(NOTIFY_LANG)(key, **kwargs) +def _compact_error_notification( + exc: Exception, + *, + title: str | None = None, + prefix: str = "", +) -> str: + detail = " ".join(str(exc).split()) + error_text = type(exc).__name__ + if detail: + error_text = f"{error_text}: {detail}" + message = f"{title or t('error_title')}\n{prefix}{error_text}" + if len(message) <= COMPACT_ERROR_NOTIFICATION_MAX_CHARS: + return message + return message[: COMPACT_ERROR_NOTIFICATION_MAX_CHARS - 1].rstrip() + "…" + + def _split_env_list(value: str | None) -> tuple[str, ...]: return tuple( item.strip() @@ -591,17 +608,18 @@ def run_strategy(*, force_run: bool = False, validation_only: bool = False, vali ) if isinstance(market_open, tuple): market_open, error = market_open - reporting_adapters.log_event( - log_context, - "market_hours_check_failed", - message="Market hours check failed", - severity="WARNING", - error_message=str(error), - market=MARKET, - market_calendar=MARKET_CALENDAR, - market_timezone=MARKET_TIMEZONE, - ) - print(composer.with_prefix(f"Market hours check failed: {error}"), flush=True) + if error is not None: + reporting_adapters.log_event( + log_context, + "market_hours_check_failed", + message="Market hours check failed", + severity="WARNING", + error_message=str(error), + market=MARKET, + market_calendar=MARKET_CALENDAR, + market_timezone=MARKET_TIMEZONE, + ) + print(composer.with_prefix(f"Market hours check failed: {error}"), flush=True) if not market_open and not force_run: reporting_adapters.log_event( log_context, @@ -743,7 +761,7 @@ def run_strategy(*, force_run: bool = False, validation_only: bool = False, vali err = traceback.format_exc() notification_adapters.publish_cycle_notification( detailed_text=f"Strategy error:\n{err}", - compact_text=f"{t('error_title')}\n{err}", + compact_text=_compact_error_notification(exc), ) return False finally: @@ -820,7 +838,11 @@ def run_probe(*, response_body: str = "Probe OK"): if composer is not None: composer.build_notification_adapters().publish_cycle_notification( detailed_text=err, - compact_text=err, + compact_text=_compact_error_notification( + exc, + title=t("health_probe_title"), + prefix=t("health_probe_error_prefix"), + ), ) else: print(err, flush=True) diff --git a/tests/test_cloud_run_entrypoint.py b/tests/test_cloud_run_entrypoint.py new file mode 100644 index 0000000..f0bfb3f --- /dev/null +++ b/tests/test_cloud_run_entrypoint.py @@ -0,0 +1,17 @@ +from entrypoints import cloud_run + + +def test_market_hours_after_close_is_closed_without_calendar_error(monkeypatch): + class FakeSchedule: + empty = False + + class FakeCalendar: + def schedule(self, **_kwargs): + return FakeSchedule() + + def open_at_time(self, _schedule, _now): + raise ValueError("The provided timestamp is not covered by the schedule") + + monkeypatch.setattr(cloud_run.mcal, "get_calendar", lambda _name: FakeCalendar()) + + assert cloud_run.is_market_open_now() == (False, None) diff --git a/tests/test_execution_state.py b/tests/test_execution_state.py index 861cf1f..9a328cf 100644 --- a/tests/test_execution_state.py +++ b/tests/test_execution_state.py @@ -121,19 +121,21 @@ def test_prior_report_match_does_not_treat_blocked_no_action_as_completed(): ) -def test_prior_report_scan_is_scoped_to_signal_month(): +def test_prior_report_scan_is_scoped_to_signal_month(monkeypatch): observed = {} - class FakeClient: - def list_blobs(self, bucket_name, *, prefix): - observed["bucket_name"] = bucket_name - observed["prefix"] = prefix + class FakeObjectStore: + def list(self, prefix_uri): + observed["prefix_uri"] = prefix_uri return () + monkeypatch.setattr( + "quant_platform_kit.cloud.get_object_store", + lambda **_kwargs: FakeObjectStore(), + ) store = ExecutionMarkerStore( local_dir=None, cloud_prefix_uri="gs://bucket/execution-reports", - client_factory=lambda **_kwargs: FakeClient(), ) assert ( @@ -147,10 +149,7 @@ def list_blobs(self, bucket_name, *, prefix): ) is False ) - assert observed == { - "bucket_name": "bucket", - "prefix": ( - "execution-reports/longbridge/" - "russell_top50_leader_rotation/PAPER/2026-06" - ), - } + assert observed["prefix_uri"] == ( + "gs://bucket/execution-reports/longbridge/" + "russell_top50_leader_rotation/PAPER/2026-06" + ) diff --git a/tests/test_qsl_metadata.py b/tests/test_qsl_metadata.py index 319d4c4..6207bb8 100644 --- a/tests/test_qsl_metadata.py +++ b/tests/test_qsl_metadata.py @@ -14,7 +14,7 @@ def test_qsl_metadata_has_runtime_platform_fields() -> None: assert qsl["tier"] == "runtime" assert qsl["upgrade_ring"] == "ring_d" assert qsl.get("repo") == "LongBridgePlatform" - assert qsl["compat"]["bundle"] == "2026.07.3" + assert qsl["compat"]["bundle"] == "2026.07.4" requires = qsl["requires"] assert "quant_platform_kit" in requires assert "us_equity_strategies" in requires diff --git a/tests/test_rebalance_service.py b/tests/test_rebalance_service.py index 2b9add7..ff3c4f4 100644 --- a/tests/test_rebalance_service.py +++ b/tests/test_rebalance_service.py @@ -1454,7 +1454,7 @@ def test_sell_then_buy_skip_is_sent_in_single_summary_message(self): self.assertEqual(len(sent_messages), 1) self.assertIn("🔔 【调仓指令】", sent_messages[0]) self.assertIn("🧭 策略: SOXL/SOXX 半导体趋势收益", sent_messages[0]) - self.assertIn("⏱ 执行时点: 2026-04-21 -> 2026-04-22 (次一交易日执行)", sent_messages[0]) + self.assertNotIn("⏱ 执行时点:", sent_messages[0]) self.assertIn("限价卖出", sent_messages[0]) self.assertIn("买入说明", sent_messages[0]) self.assertIn("SOXX.US", sent_messages[0]) @@ -1486,7 +1486,7 @@ def test_buy_skip_without_orders_is_sent_in_single_heartbeat_message(self): self.assertEqual(len(sent_messages), 1) self.assertIn("💓 【心跳检测】", sent_messages[0]) - self.assertIn("⏱ 执行时点: 2026-04-21 -> 2026-04-22 (次一交易日执行)", sent_messages[0]) + self.assertNotIn("⏱ 执行时点:", sent_messages[0]) self.assertIn("本轮没有可执行订单", sent_messages[0]) self.assertIn("说明", sent_messages[0]) self.assertIn("可投资现金", sent_messages[0]) @@ -2780,7 +2780,8 @@ def test_hybrid_heartbeat_hides_empty_semiconductor_fields_and_shows_benchmark_l self.assertIn("QQQI: $0.00 / 0股", sent_messages[0]) self.assertIn("SPYI: $0.00 / 0股", sent_messages[0]) self.assertNotIn("📈 QQQ 基准\n - QQQ: 588.50\n - MA200: 595.25\n - 退出线: 573.00", sent_messages[0]) - self.assertIn("🎯 信号: 💤 等待信号", sent_messages[0]) + self.assertNotIn("🎯 信号:", sent_messages[0]) + self.assertIn("✅ 无需调仓", sent_messages[0]) self.assertNotIn("账户现金: $0.00 | 可投资现金", sent_messages[0]) self.assertNotIn("TQQQ: $0.00 BOXX", sent_messages[0]) self.assertNotIn("📊 市场状态: ", sent_messages[0]) diff --git a/tests/test_request_handling.py b/tests/test_request_handling.py index 0b6764f..8a447cc 100644 --- a/tests/test_request_handling.py +++ b/tests/test_request_handling.py @@ -71,6 +71,12 @@ def run(self, *args, **kwargs): tg_token=None, tg_chat_id="shared-chat-id", dry_run_only=False, + notification_channel="telegram", + wecom_webhook_url=None, + dingtalk_webhook_url=None, + feishu_webhook_url=None, + serverchan_webhook_url=None, + strategy_metadata=None, strategy_plugin_alert_email_recipients=(), strategy_plugin_alert_email_sender_email=None, strategy_plugin_alert_email_sender_password=None, @@ -89,6 +95,7 @@ def run(self, *args, **kwargs): ) qpk_longbridge_module = types.ModuleType("quant_platform_kit.longbridge") + qpk_longbridge_module.__path__ = [] qpk_longbridge_module.build_contexts = lambda *args, **kwargs: ("quote-context", "trade-context") qpk_longbridge_module.calculate_rotation_indicators = lambda *args, **kwargs: {} qpk_longbridge_module.estimate_max_purchase_quantity = lambda *args, **kwargs: 0 @@ -98,6 +105,12 @@ def run(self, *args, **kwargs): qpk_longbridge_module.fetch_token_from_secret = lambda *args, **kwargs: "token" qpk_longbridge_module.refresh_token_if_needed = lambda *args, **kwargs: "token" qpk_longbridge_module.submit_order = lambda *args, **kwargs: None + qpk_longbridge_market_data_module = types.ModuleType( + "quant_platform_kit.longbridge.market_data" + ) + qpk_longbridge_market_data_module.fetch_lot_sizes = ( + lambda *_args, **_kwargs: {} + ) google_module = types.ModuleType("google") google_module.__path__ = [] @@ -155,11 +168,19 @@ def run(self, *args, **kwargs): us_equity_strategies_module = types.ModuleType("us_equity_strategies") us_equity_strategies_module.__path__ = [] + cash_only_equity_module = types.ModuleType("us_equity_strategies.cash_only_equity") + cash_only_equity_module.normalize_account_state_from_snapshot = ( + lambda snapshot, **_kwargs: snapshot + ) catalog_module = types.ModuleType("us_equity_strategies.catalog") catalog_module.resolve_canonical_profile = lambda profile: profile strategy_registry_module = types.ModuleType("strategy_registry") strategy_registry_module.LONGBRIDGE_PLATFORM = "longbridge" + strategy_registry_module.PLATFORM_CAPABILITY_MATRIX = types.SimpleNamespace( + supported_capabilities=frozenset() + ) + strategy_registry_module.STRATEGY_CATALOG = types.SimpleNamespace(definitions={}) strategy_registry_module.resolve_strategy_definition = lambda profile, **_kwargs: types.SimpleNamespace( profile=profile ) @@ -170,6 +191,7 @@ def run(self, *args, **kwargs): "entrypoints.cloud_run": cloud_run_module, "runtime_config_support": runtime_config_support_module, "quant_platform_kit.longbridge": qpk_longbridge_module, + "quant_platform_kit.longbridge.market_data": qpk_longbridge_market_data_module, "google": google_module, "google.auth": google_auth_module, "google.auth.transport": google_auth_transport_module, @@ -184,6 +206,7 @@ def run(self, *args, **kwargs): "longport": longport_module, "longport.openapi": openapi_module, "us_equity_strategies": us_equity_strategies_module, + "us_equity_strategies.cash_only_equity": cash_only_equity_module, "us_equity_strategies.catalog": catalog_module, "strategy_registry": strategy_registry_module, } @@ -467,7 +490,7 @@ def test_handle_probe_failure_sends_notification(self): class FakeRuntime: def bootstrap(self): - raise RuntimeError("probe failed") + raise RuntimeError("probe failed " + "x" * 5000) class FakeNotifications: def publish_cycle_notification(self, **kwargs): @@ -507,7 +530,12 @@ def attach_strategy_plugin_report(self, *_args, **_kwargs): ["health_probe_received", "health_probe_failed"], ) self.assertEqual(len(observed["notifications"]), 1) - self.assertIn("probe failed", observed["notifications"][0]["detailed_text"]) + notification = observed["notifications"][0] + self.assertIn("probe failed", notification["detailed_text"]) + self.assertIn("Traceback", notification["detailed_text"]) + self.assertNotIn("Traceback", notification["compact_text"]) + self.assertIn("RuntimeError: probe failed", notification["compact_text"]) + self.assertLessEqual(len(notification["compact_text"]), 3500) def test_run_strategy_emits_structured_runtime_events(self): module = load_module() @@ -526,6 +554,73 @@ def test_run_strategy_emits_structured_runtime_events(self): ) self.assertTrue(all(run_id == "run-001" for run_id, _event, _fields in observed)) + def test_run_strategy_market_hours_tuple_without_error_does_not_warn(self): + module = load_module() + observed = [] + + module.emit_runtime_log = ( + lambda context, event, **fields: observed.append((event, fields)) + ) + module.is_market_open_now = lambda **_kwargs: (True, None) + module.run_rebalance_cycle = lambda **_kwargs: None + + self.assertTrue(module.run_strategy()) + self.assertNotIn( + "market_hours_check_failed", + [event for event, _fields in observed], + ) + + def test_run_strategy_error_notification_is_compact_and_bounded(self): + module = load_module() + observed = {} + + class FakeComposer: + def build_reporting_adapters(self): + return types.SimpleNamespace( + start_run=lambda: ( + types.SimpleNamespace(run_id="run-001"), + {"status": "pending"}, + ), + log_event=lambda *args, **kwargs: None, + persist_execution_report=lambda report: types.SimpleNamespace( + local_path="/tmp/report.json" + ), + ) + + def build_notification_adapters(self): + return types.SimpleNamespace( + publish_cycle_notification=lambda **kwargs: observed.update( + kwargs + ) + ) + + def load_strategy_plugin_signals(self, *_args, **_kwargs): + return (), None + + def attach_strategy_plugin_report(self, *_args, **_kwargs): + return None + + def with_prefix(self, message): + return message + + def build_rebalance_runtime(self, **_kwargs): + return types.SimpleNamespace() + + def build_rebalance_config(self, **_kwargs): + return types.SimpleNamespace() + + module.build_composer = lambda *, dry_run_only_override=None: FakeComposer() + module.is_market_open_now = lambda **_kwargs: True + module.run_rebalance_cycle = lambda **_kwargs: (_ for _ in ()).throw( + RuntimeError("x" * 5000) + ) + + self.assertFalse(module.run_strategy()) + self.assertIn("Traceback", observed["detailed_text"]) + self.assertNotIn("Traceback", observed["compact_text"]) + self.assertIn("RuntimeError:", observed["compact_text"]) + self.assertLessEqual(len(observed["compact_text"]), 3500) + def test_run_strategy_sends_escalated_strategy_plugin_alert(self): module = load_module() signal = types.SimpleNamespace( diff --git a/tests/test_runtime_reporting_adapters.py b/tests/test_runtime_reporting_adapters.py index a23ce9e..37e032a 100644 --- a/tests/test_runtime_reporting_adapters.py +++ b/tests/test_runtime_reporting_adapters.py @@ -88,7 +88,7 @@ def test_runtime_reporting_adapters_log_and_persist_route_to_dependencies(): def fake_report_persister(report, **kwargs): observed["persist"] = (report, kwargs) - return types.SimpleNamespace(local_path="/tmp/report.json", gcs_uri=None) + return types.SimpleNamespace(local_path="/tmp/report.json", cloud_uri=None) adapters = build_runtime_reporting_adapters( runtime_assembly=build_runtime_assembly( @@ -122,7 +122,34 @@ def fake_report_persister(report, **kwargs): assert observed["event"][2]["printer"] is adapters.printer assert observed["persist"][1] == { "base_dir": None, - "gcs_prefix_uri": None, - "gcp_project_id": "project-1", + "cloud_prefix_uri": None, + "project_id": "project-1", } assert persisted == "/tmp/report.json" + + +def test_runtime_reporting_adapters_prefers_legacy_gcs_uri_over_local_path(): + adapters = build_runtime_reporting_adapters( + runtime_assembly=build_runtime_assembly( + platform="longbridge", + deploy_target="cloud_run", + service_name="longbridge-platform", + strategy_profile="soxl_soxx_trend_income", + account_scope="HK", + account_region="HK", + project_id="project-1", + ), + strategy_domain="us_equity", + managed_symbols=(), + signal_effective_after_trading_days=1, + run_id_builder=lambda: "run-001", + event_logger=lambda *_args, **_kwargs: {}, + report_builder=lambda **kwargs: kwargs, + report_persister=lambda *_args, **_kwargs: types.SimpleNamespace( + local_path="/tmp/report.json", + gcs_uri="gs://bucket/report.json", + ), + printer=lambda *_args, **_kwargs: None, + ) + + assert adapters.persist_execution_report({}) == "gs://bucket/report.json" diff --git a/tests/test_shared_chat_id_fallback.py b/tests/test_shared_chat_id_fallback.py index 1c5f833..23a85c6 100644 --- a/tests/test_shared_chat_id_fallback.py +++ b/tests/test_shared_chat_id_fallback.py @@ -52,7 +52,7 @@ def run(self, *args, **kwargs): requests_module.post = lambda *args, **kwargs: None cloud_run_module = types.ModuleType("entrypoints.cloud_run") - cloud_run_module.is_market_open_now = lambda: True + cloud_run_module.is_market_open_now = lambda **_kwargs: True runtime_config_support_module = types.ModuleType("runtime_config_support") runtime_config_support_module.load_platform_runtime_settings = lambda **_kwargs: types.SimpleNamespace( @@ -72,6 +72,12 @@ def run(self, *args, **kwargs): tg_token=None, tg_chat_id="shared-chat-id", dry_run_only=False, + notification_channel="telegram", + wecom_webhook_url=None, + dingtalk_webhook_url=None, + feishu_webhook_url=None, + serverchan_webhook_url=None, + strategy_metadata=None, runtime_target=build_runtime_target( platform_id="longbridge", strategy_profile="soxl_soxx_trend_income", @@ -84,6 +90,7 @@ def run(self, *args, **kwargs): ) qpk_longbridge_module = types.ModuleType("quant_platform_kit.longbridge") + qpk_longbridge_module.__path__ = [] qpk_longbridge_module.build_contexts = lambda *args, **kwargs: ("quote-context", "trade-context") qpk_longbridge_module.calculate_rotation_indicators = lambda *args, **kwargs: {} qpk_longbridge_module.estimate_max_purchase_quantity = lambda *args, **kwargs: 0 @@ -93,18 +100,37 @@ def run(self, *args, **kwargs): qpk_longbridge_module.fetch_token_from_secret = lambda *args, **kwargs: "token" qpk_longbridge_module.refresh_token_if_needed = lambda *args, **kwargs: "token" qpk_longbridge_module.submit_order = lambda *args, **kwargs: None + qpk_longbridge_market_data_module = types.ModuleType( + "quant_platform_kit.longbridge.market_data" + ) + qpk_longbridge_market_data_module.fetch_lot_sizes = ( + lambda *_args, **_kwargs: {} + ) google_module = types.ModuleType("google") google_module.__path__ = [] google_auth_module = types.ModuleType("google.auth") google_auth_module.default = lambda *args, **kwargs: (None, None) + google_auth_transport_module = types.ModuleType("google.auth.transport") + google_auth_transport_requests_module = types.ModuleType( + "google.auth.transport.requests" + ) + google_auth_transport_requests_module.Request = type("Request", (), {}) + google_oauth2_module = types.ModuleType("google.oauth2") + google_oauth2_id_token_module = types.ModuleType("google.oauth2.id_token") + google_oauth2_id_token_module.fetch_id_token = ( + lambda *_args, **_kwargs: "id-token" + ) google_cloud_module = types.ModuleType("google.cloud") google_cloud_module.__path__ = [] google_secretmanager_module = types.ModuleType("google.cloud.secretmanager_v1") google_module.auth = google_auth_module + google_auth_module.transport = google_auth_transport_module + google_auth_transport_module.requests = google_auth_transport_requests_module + google_oauth2_module.id_token = google_oauth2_id_token_module google_cloud_module.secretmanager_v1 = google_secretmanager_module pandas_module = types.ModuleType("pandas") @@ -140,11 +166,19 @@ def run(self, *args, **kwargs): us_equity_strategies_module = types.ModuleType("us_equity_strategies") us_equity_strategies_module.__path__ = [] + cash_only_equity_module = types.ModuleType("us_equity_strategies.cash_only_equity") + cash_only_equity_module.normalize_account_state_from_snapshot = ( + lambda snapshot, **_kwargs: snapshot + ) catalog_module = types.ModuleType("us_equity_strategies.catalog") catalog_module.resolve_canonical_profile = lambda profile: profile strategy_registry_module = types.ModuleType("strategy_registry") strategy_registry_module.LONGBRIDGE_PLATFORM = "longbridge" + strategy_registry_module.PLATFORM_CAPABILITY_MATRIX = types.SimpleNamespace( + supported_capabilities=frozenset() + ) + strategy_registry_module.STRATEGY_CATALOG = types.SimpleNamespace(definitions={}) strategy_registry_module.resolve_strategy_definition = lambda profile, **_kwargs: types.SimpleNamespace( profile=profile ) @@ -155,8 +189,13 @@ def run(self, *args, **kwargs): "entrypoints.cloud_run": cloud_run_module, "runtime_config_support": runtime_config_support_module, "quant_platform_kit.longbridge": qpk_longbridge_module, + "quant_platform_kit.longbridge.market_data": qpk_longbridge_market_data_module, "google": google_module, "google.auth": google_auth_module, + "google.auth.transport": google_auth_transport_module, + "google.auth.transport.requests": google_auth_transport_requests_module, + "google.oauth2": google_oauth2_module, + "google.oauth2.id_token": google_oauth2_id_token_module, "google.cloud": google_cloud_module, "google.cloud.secretmanager_v1": google_secretmanager_module, "pandas": pandas_module, @@ -165,6 +204,7 @@ def run(self, *args, **kwargs): "longport": longport_module, "longport.openapi": openapi_module, "us_equity_strategies": us_equity_strategies_module, + "us_equity_strategies.cash_only_equity": cash_only_equity_module, "us_equity_strategies.catalog": catalog_module, "strategy_registry": strategy_registry_module, } diff --git a/tests/test_strategy_runtime.py b/tests/test_strategy_runtime.py index 070104c..3a52ed8 100644 --- a/tests/test_strategy_runtime.py +++ b/tests/test_strategy_runtime.py @@ -566,7 +566,9 @@ def test_feature_snapshot_runtime_loads_mega_cap_top50_snapshot_into_context(sel ) self.assertEqual(entrypoint.ctx.market_data["feature_snapshot"][0]["symbol"], "NVDA") - self.assertIs(entrypoint.ctx.portfolio, portfolio) + self.assertEqual(entrypoint.ctx.portfolio.total_equity, portfolio.total_equity) + self.assertEqual(entrypoint.ctx.portfolio.buying_power, portfolio.buying_power) + self.assertEqual(entrypoint.ctx.portfolio.metadata["consecutive_losses"], 0) self.assertEqual(result.metadata["managed_symbols"], ("NVDA", "META", "BOXX")) self.assertEqual(result.metadata["status_icon"], "👑")