diff --git a/notifications/telegram.py b/notifications/telegram.py index 96808d8..2229728 100644 --- a/notifications/telegram.py +++ b/notifications/telegram.py @@ -103,6 +103,7 @@ "strategy_name_mega_cap_leader_rotation_top50_balanced": "Mega Cap Top50 平衡龙头轮动", "strategy_name_hk_listed_global_etf_rotation": "港股上市全球 ETF 轮动", "strategy_name_hk_high_dividend_low_vol_trend": "港股高股息低波趋势", + "strategy_name_hk_low_vol_dividend_quality": "港股低波股息质量", "strategy_plugin_line": "🧩 插件：{plugin} | 状态：{route} | 提醒：{action}", "strategy_plugin_alert_subject": "🚨 策略插件告警：{plugin} | {route}", "strategy_plugin_alert_title": "🚨 【策略插件告警】", @@ -222,6 +223,7 @@ "strategy_name_mega_cap_leader_rotation_top50_balanced": "Mega Cap Leader Rotation Top50 Balanced", "strategy_name_hk_listed_global_etf_rotation": "HK-listed Global ETF Rotation", "strategy_name_hk_high_dividend_low_vol_trend": "HK High Dividend Low-Volatility Trend", + "strategy_name_hk_low_vol_dividend_quality": "HK Low-Volatility Dividend Quality", "strategy_plugin_line": "🧩 Plugin: {plugin} | status: {route} | notice: {action}", "strategy_plugin_alert_subject": "🚨 Strategy plugin alert: {plugin} | {route}", "strategy_plugin_alert_title": "🚨 【Strategy Plugin Alert】", diff --git a/requirements.txt b/requirements.txt index 8d0e20b..f6710a6 100644 --- a/requirements.txt +++ b/requirements.txt @@ -2,7 +2,7 @@ flask gunicorn quant-platform-kit @ git+https://github.com/QuantStrategyLab/QuantPlatformKit.git@v0.7.36 us-equity-strategies @ git+https://github.com/QuantStrategyLab/UsEquityStrategies.git@v0.7.50 -hk-equity-strategies @ git+https://github.com/QuantStrategyLab/HkEquityStrategies.git@v0.4.2 +hk-equity-strategies @ git+https://github.com/QuantStrategyLab/HkEquityStrategies.git@hk-low-vol-dividend-quality-20260603 pandas requests pytz diff --git a/scripts/print_strategy_switch_env_plan.py b/scripts/print_strategy_switch_env_plan.py index 0ca74ea..882e8d9 100644 --- a/scripts/print_strategy_switch_env_plan.py +++ b/scripts/print_strategy_switch_env_plan.py @@ -80,6 +80,12 @@ def _should_add_local_src(candidate: Path) -> bool: ] +def _feature_snapshot_filenames(profile: str, snapshot_contract_version: str | None) -> tuple[str, str]: + suffix = "factor_snapshot" if ".factor_snapshot." in str(snapshot_contract_version or "") else "feature_snapshot" + snapshot_filename = f"{profile}_{suffix}_latest.csv" + return snapshot_filename, f"{snapshot_filename}.manifest.json" + + def build_switch_plan( profile: str, *, @@ -222,10 +228,12 @@ def build_switch_plan( hints: dict[str, str] = {} if requires_feature_snapshot: - hints["feature_snapshot_filename"] = f"{definition.profile}_feature_snapshot_latest.csv" - hints["feature_snapshot_manifest_filename"] = ( - f"{definition.profile}_feature_snapshot_latest.csv.manifest.json" + snapshot_filename, manifest_filename = _feature_snapshot_filenames( + definition.profile, + runtime_requirements.get("snapshot_contract_version"), ) + hints["feature_snapshot_filename"] = snapshot_filename + hints["feature_snapshot_manifest_filename"] = manifest_filename if artifact_paths.bundled_config_path is not None: hints["bundled_strategy_config_path"] = str(artifact_paths.bundled_config_path) diff --git a/tests/test_runtime_config_support.py b/tests/test_runtime_config_support.py index 4bfda71..0d863cc 100644 --- a/tests/test_runtime_config_support.py +++ b/tests/test_runtime_config_support.py @@ -64,8 +64,9 @@ OPTIONAL_LONGBRIDGE_PROFILES = frozenset({"global_etf_confidence_vol_gate"}) HK_RUNTIME_ENABLED_PROFILES = frozenset( { - "hk_listed_global_etf_rotation", "hk_high_dividend_low_vol_trend", + "hk_listed_global_etf_rotation", + "hk_low_vol_dividend_quality", } ) HK_DISABLED_PROFILES = frozenset( @@ -761,12 +762,17 @@ def test_print_strategy_profile_status_json_matches_registry(self): self.assertFalse(by_profile["mega_cap_leader_rotation_top50_balanced"]["requires_strategy_config_path"]) for profile in ("hk_index_mean_reversion", "hk_etf_regime_rotation", "hk_blue_chip_leader_rotation"): self.assertNotIn(profile, by_profile) - for profile in HK_RUNTIME_ENABLED_PROFILES: + for profile in HK_RUNTIME_ENABLED_PROFILES - {"hk_low_vol_dividend_quality"}: self.assertEqual(by_profile[profile]["profile_group"], "direct_runtime_inputs") self.assertEqual(by_profile[profile]["input_mode"], "market_history") self.assertFalse(by_profile[profile]["requires_snapshot_artifacts"]) self.assertFalse(by_profile[profile]["requires_snapshot_manifest_path"]) self.assertFalse(by_profile[profile]["requires_strategy_config_path"]) + self.assertEqual(by_profile["hk_low_vol_dividend_quality"]["profile_group"], "snapshot_backed") + self.assertEqual(by_profile["hk_low_vol_dividend_quality"]["input_mode"], "feature_snapshot") + self.assertTrue(by_profile["hk_low_vol_dividend_quality"]["requires_snapshot_artifacts"]) + self.assertTrue(by_profile["hk_low_vol_dividend_quality"]["requires_snapshot_manifest_path"]) + self.assertFalse(by_profile["hk_low_vol_dividend_quality"]["requires_strategy_config_path"]) self.assertFalse( by_profile["russell_1000_multi_factor_defensive"]["requires_strategy_config_path"] ) @@ -995,6 +1001,41 @@ def test_print_strategy_switch_env_plan_for_mega_cap_top50_balanced(self): "mega_cap_leader_rotation_top50_balanced_feature_snapshot_latest.csv", ) + def test_print_strategy_switch_env_plan_for_hk_low_vol_dividend_quality(self): + result = subprocess.run( + [ + sys.executable, + str(SWITCH_PLAN_SCRIPT_PATH), + "--profile", + "hk_low_vol_dividend_quality", + "--account-region", + "hk", + "--dry-run-only", + "--json", + ], + check=True, + capture_output=True, + text=True, + ) + + plan = json.loads(result.stdout) + self.assertEqual(plan["canonical_profile"], "hk_low_vol_dividend_quality") + self.assertTrue(plan["enabled"]) + self.assertEqual(plan["profile_group"], "snapshot_backed") + self.assertEqual(plan["input_mode"], "feature_snapshot") + self.assertEqual(plan["snapshot_contract_version"], "hk_low_vol_dividend_quality.factor_snapshot.v1") + self.assertEqual(plan["set_env"]["LONGBRIDGE_DRY_RUN_ONLY"], "true") + self.assertEqual(plan["set_env"]["LONGBRIDGE_FEATURE_SNAPSHOT_PATH"], "") + self.assertEqual(plan["set_env"]["LONGBRIDGE_FEATURE_SNAPSHOT_MANIFEST_PATH"], "") + self.assertEqual( + plan["hints"]["feature_snapshot_filename"], + "hk_low_vol_dividend_quality_factor_snapshot_latest.csv", + ) + self.assertEqual( + plan["hints"]["feature_snapshot_manifest_filename"], + "hk_low_vol_dividend_quality_factor_snapshot_latest.csv.manifest.json", + ) + def test_print_strategy_switch_env_plan_rejects_hk_disabled_profiles(self): for profile in sorted(HK_DISABLED_PROFILES): with self.subTest(profile=profile):