[codex] Add LongBridge HK low-vol dividend switch plan

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】",

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

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)
 

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"], "<required>")
+        self.assertEqual(plan["set_env"]["LONGBRIDGE_FEATURE_SNAPSHOT_MANIFEST_PATH"], "<required>")
+        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):