From 54b71f93314456e3ad88b38a9b94c906170d61a0 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Sun, 10 May 2026 13:32:05 +0800 Subject: [PATCH] Add crisis plugin trigger rehearsal tests --- notifications/telegram.py | 2 + tests/test_notifications.py | 2 + tests/test_request_handling.py | 87 ++++++++++++++++++++++++++++ tests/test_runtime_config_support.py | 65 +++++++++------------ 4 files changed, 119 insertions(+), 37 deletions(-) diff --git a/notifications/telegram.py b/notifications/telegram.py index ec6a372..4f86284 100644 --- a/notifications/telegram.py +++ b/notifications/telegram.py @@ -90,6 +90,7 @@ "strategy_name_tqqq_growth_income": "TQQQ 增长收益", "strategy_name_soxl_soxx_trend_income": "SOXL/SOXX 半导体趋势收益", "strategy_name_global_etf_rotation": "全球 ETF 轮动", + "strategy_name_global_etf_confidence_vol_gate": "全球 ETF 置信波动门控", "strategy_name_russell_1000_multi_factor_defensive": "罗素1000多因子", "strategy_name_tech_communication_pullback_enhancement": "科技通信回调增强", "strategy_name_qqq_tech_enhancement": "科技通信回调增强", @@ -170,6 +171,7 @@ "strategy_name_tqqq_growth_income": "TQQQ Growth Income", "strategy_name_soxl_soxx_trend_income": "SOXL/SOXX Semiconductor Trend Income", "strategy_name_global_etf_rotation": "Global ETF Rotation", + "strategy_name_global_etf_confidence_vol_gate": "Global ETF Confidence Vol Gate", "strategy_name_russell_1000_multi_factor_defensive": "Russell 1000 Multi-Factor", "strategy_name_tech_communication_pullback_enhancement": "Tech/Communication Pullback Enhancement", "strategy_name_qqq_tech_enhancement": "Tech/Communication Pullback Enhancement", diff --git a/tests/test_notifications.py b/tests/test_notifications.py index d9d5886..4e022eb 100644 --- a/tests/test_notifications.py +++ b/tests/test_notifications.py @@ -61,6 +61,8 @@ def test_supported_strategy_profiles_have_translated_names(self): zh_name = build_strategy_display_name(build_translator("zh")) en_name = build_strategy_display_name(build_translator("en")) + self.assertEqual(zh_name("global_etf_confidence_vol_gate"), "全球 ETF 置信波动门控") + self.assertEqual(en_name("global_etf_confidence_vol_gate"), "Global ETF Confidence Vol Gate") for profile in SUPPORTED_STRATEGY_PROFILES: self.assertNotEqual(zh_name(profile), profile) self.assertNotEqual(en_name(profile), profile) diff --git a/tests/test_request_handling.py b/tests/test_request_handling.py index 38f6dcf..bec7868 100644 --- a/tests/test_request_handling.py +++ b/tests/test_request_handling.py @@ -294,6 +294,78 @@ def fake_run_strategy_core(client, now_ny, *, strategy_plugin_signals=()): self.assertEqual(plugin_summary["canonical_route"], "no_action") self.assertEqual(plugin_summary["suggested_action"], "monitor") + def test_handle_schwab_rehearses_triggered_shadow_plugin_report(self): + with tempfile.TemporaryDirectory() as temp_dir: + signal_path = Path(temp_dir) / "latest_signal.json" + signal_path.write_text( + json.dumps( + { + "strategy": "tqqq_growth_income", + "plugin": "crisis_response_shadow", + "mode": "shadow", + "configured_mode": "shadow", + "effective_mode": "shadow", + "schema_version": "crisis_response_shadow.v1", + "as_of": "2008-03-10", + "canonical_route": "true_crisis", + "suggested_action": "defend", + "would_trade_if_enabled": True, + "execution_controls": { + "broker_order_allowed": False, + "live_allocation_mutation_allowed": False, + "repository_broker_write_allowed": False, + "repository_allocation_mutation_allowed": False, + }, + } + ), + encoding="utf-8", + ) + mount_config = json.dumps( + { + "strategy_plugins": [ + { + "strategy": "tqqq_growth_income", + "plugin": "crisis_response_shadow", + "signal_path": str(signal_path), + "enabled": True, + "expected_mode": "shadow", + } + ] + } + ) + module = load_module(strategy_plugin_mounts_json=mount_config) + observed = {} + + module.get_client_from_secret = lambda *args, **kwargs: object() + module.is_market_open_today = lambda: True + module.persist_execution_report = ( + lambda report: observed.setdefault("report", report) or "/tmp/report.json" + ) + + def fake_run_strategy_core(client, now_ny, *, strategy_plugin_signals=()): + observed["signals"] = strategy_plugin_signals + self.assertEqual(len(strategy_plugin_signals), 1) + signal = strategy_plugin_signals[0] + self.assertTrue(signal.would_trade_if_enabled) + self.assertEqual(signal.canonical_route, "true_crisis") + self.assertEqual(signal.suggested_action, "defend") + self.assertFalse(signal.execution_controls["broker_order_allowed"]) + self.assertFalse(signal.execution_controls["live_allocation_mutation_allowed"]) + + module.run_strategy_core = fake_run_strategy_core + + with module.app.test_request_context("/", method="POST"): + body, status = module.handle_schwab() + + self.assertEqual(status, 200) + self.assertEqual(body, "OK") + plugin_summary = observed["report"]["summary"]["strategy_plugins"][0] + self.assertEqual(plugin_summary["canonical_route"], "true_crisis") + self.assertEqual(plugin_summary["suggested_action"], "defend") + self.assertTrue(plugin_summary["would_trade_if_enabled"]) + self.assertFalse(plugin_summary["execution_controls"]["broker_order_allowed"]) + self.assertFalse(plugin_summary["execution_controls"]["live_allocation_mutation_allowed"]) + def test_strategy_plugin_notification_line_uses_i18n(self): module = load_module(notify_lang="zh") signal = types.SimpleNamespace( @@ -311,6 +383,21 @@ def test_strategy_plugin_notification_line_uses_i18n(self): self.assertIn("路由：不操作", lines[0]) self.assertIn("建议：仅观察", lines[0]) + def test_strategy_plugin_notification_line_renders_triggered_shadow_signal(self): + module = load_module(notify_lang="zh") + signal = types.SimpleNamespace( + plugin="crisis_response_shadow", + effective_mode="shadow", + canonical_route="true_crisis", + suggested_action="defend", + ) + + lines = module.build_strategy_plugin_notification_lines((signal,)) + + self.assertEqual(len(lines), 1) + self.assertIn("路由：真危机", lines[0]) + self.assertIn("建议：防守", lines[0]) + def test_handle_schwab_reports_plugin_config_error_without_blocking_strategy(self): mount_config = json.dumps( { diff --git a/tests/test_runtime_config_support.py b/tests/test_runtime_config_support.py index e598165..00e02cf 100644 --- a/tests/test_runtime_config_support.py +++ b/tests/test_runtime_config_support.py @@ -34,6 +34,22 @@ SAMPLE_STRATEGY_PROFILE = "tqqq_growth_income" +BASE_SCHWAB_PROFILES = frozenset( + { + SAMPLE_STRATEGY_PROFILE, + "global_etf_rotation", + "mega_cap_leader_rotation_top50_balanced", + "russell_1000_multi_factor_defensive", + "soxl_soxx_trend_income", + "tech_communication_pullback_enhancement", + } +) +OPTIONAL_SCHWAB_PROFILES = frozenset({"global_etf_confidence_vol_gate"}) + + +def expected_schwab_profiles(actual_profiles) -> frozenset[str]: + actual = frozenset(actual_profiles) + return BASE_SCHWAB_PROFILES | (OPTIONAL_SCHWAB_PROFILES & actual) class RuntimeConfigSupportTests(unittest.TestCase): @@ -71,34 +87,12 @@ def test_rejects_unknown_strategy_profile(self): load_platform_runtime_settings() def test_platform_supported_profiles_are_filtered_by_registry(self): - self.assertEqual( - get_supported_profiles_for_platform(SCHWAB_PLATFORM), - frozenset( - { - SAMPLE_STRATEGY_PROFILE, - "global_etf_rotation", - "mega_cap_leader_rotation_top50_balanced", - "russell_1000_multi_factor_defensive", - "soxl_soxx_trend_income", - "tech_communication_pullback_enhancement", - } - ), - ) + profiles = get_supported_profiles_for_platform(SCHWAB_PLATFORM) + self.assertEqual(profiles, expected_schwab_profiles(profiles)) def test_platform_eligible_profiles_are_exposed_by_capability_matrix(self): - self.assertEqual( - get_eligible_profiles_for_platform(SCHWAB_PLATFORM), - frozenset( - { - SAMPLE_STRATEGY_PROFILE, - "global_etf_rotation", - "mega_cap_leader_rotation_top50_balanced", - "russell_1000_multi_factor_defensive", - "soxl_soxx_trend_income", - "tech_communication_pullback_enhancement", - } - ), - ) + profiles = get_eligible_profiles_for_platform(SCHWAB_PLATFORM) + self.assertEqual(profiles, expected_schwab_profiles(profiles)) def test_rejects_human_readable_alias(self): with patch.dict(os.environ, {"STRATEGY_PROFILE": "qqq_tqqq_growth_income"}, clear=True): @@ -188,17 +182,7 @@ def test_platform_profile_status_matrix_matches_current_schwab_rollout(self): rows = get_platform_profile_status_matrix() by_profile = {row["canonical_profile"]: row for row in rows} - self.assertEqual( - set(by_profile), - { - "tqqq_growth_income", - "global_etf_rotation", - "mega_cap_leader_rotation_top50_balanced", - "russell_1000_multi_factor_defensive", - "soxl_soxx_trend_income", - "tech_communication_pullback_enhancement", - }, - ) + self.assertEqual(set(by_profile), expected_schwab_profiles(by_profile)) self.assertEqual( by_profile["tqqq_growth_income"], { @@ -216,6 +200,13 @@ def test_platform_profile_status_matrix_matches_current_schwab_rollout(self): ) self.assertTrue(by_profile["global_etf_rotation"]["eligible"]) self.assertTrue(by_profile["global_etf_rotation"]["enabled"]) + if "global_etf_confidence_vol_gate" in by_profile: + self.assertEqual( + by_profile["global_etf_confidence_vol_gate"]["display_name"], + "Global ETF Confidence Vol Gate", + ) + self.assertTrue(by_profile["global_etf_confidence_vol_gate"]["eligible"]) + self.assertTrue(by_profile["global_etf_confidence_vol_gate"]["enabled"]) self.assertEqual( by_profile["russell_1000_multi_factor_defensive"]["display_name"], "Russell 1000 Multi-Factor",