[codex] Add LongBridge HK dry-run switch plan

README.md

@@ -59,6 +59,12 @@ Strategy logic, cadence, asset universes, parameters, and research/backtest note
 
 For the HK-equity runtime scope, platform matrix, and env defaults, see [`docs/hk_equity_runtime.md`](docs/hk_equity_runtime.md).
 
+For HK verify-only rollout planning, print the switch plan first instead of changing Cloud Run directly:
+
+```bash
+python scripts/print_strategy_switch_env_plan.py --profile hk_listed_global_etf_rotation --account-region hk --dry-run-only --deployment-selector hk-verify --account-scope hk-verify --service-name longbridge-quant-hk-verify-service --json
+```
+
 ### Notifications
 
 Telegram notifications include structured execution and heartbeat messages, with English and Chinese variants. Strategy-specific signal/status fields come from the selected strategy package profile; LongBridge-specific fields cover order submission, fill/reject/error reporting, account prefix, region, and market scope.
@@ -247,6 +253,12 @@ python3 scripts/print_strategy_profile_status.py
 
 港股运行时范围、平台矩阵和环境变量默认值见 [`docs/hk_equity_runtime.md`](docs/hk_equity_runtime.md)。
 
+港股 verify-only 接入先打印切换计划，不直接改 Cloud Run：
+
+```bash
+python scripts/print_strategy_switch_env_plan.py --profile hk_listed_global_etf_rotation --account-region hk --dry-run-only --deployment-selector hk-verify --account-scope hk-verify --service-name longbridge-quant-hk-verify-service --json
+```
+
 ### 通知格式
 
 Telegram 通知包含结构化的调仓和心跳消息，支持中英文切换。策略相关的信号/状态字段来自当前选择的策略 package profile；LongBridge 侧负责下单、成交/拒单/异常、账户前缀、区域和市场范围字段。

docs/hk_equity_runtime.md

@@ -65,6 +65,31 @@ LONGBRIDGE_SYMBOL_SUFFIX=.HK
 LONGBRIDGE_TRADING_CURRENCY=HKD
 ```
 
+## Dry-run 切换计划
+
+先只生成 verify-only 环境计划，不部署生产 Cloud Run：
+
+```bash
+python scripts/print_strategy_switch_env_plan.py \
+  --profile hk_listed_global_etf_rotation \
+  --account-region hk \
+  --dry-run-only \
+  --deployment-selector hk-verify \
+  --account-scope hk-verify \
+  --service-name longbridge-quant-hk-verify-service \
+  --json
+```
+
+这个命令只打印计划。输出会显式包含：
+
+- `RUNTIME_TARGET_JSON`：`strategy_profile=hk_listed_global_etf_rotation`、`dry_run_only=true`、`execution_mode=paper`。
+- `ACCOUNT_REGION=HK`、`ACCOUNT_PREFIX=HK`、`LONGBRIDGE_DRY_RUN_ONLY=true`。
+- `LONGBRIDGE_MARKET=HK` / `XHKG` / `Asia/Hong_Kong` / `.HK` / `HKD`。
+- `remove_if_present`：清理 snapshot/config 相关环境变量，因为该 profile 直接使用 `market_history`。
+- `dry_run_plan`：检查 HK 行情权限、`.HK` / HKD 映射、整数股和 lot-size、HKD 现金口径、通知和 runtime report。
+
+合并代码或打印计划不会触发生产部署；只有单独执行 Cloud Run env 更新/部署命令才会改变服务配置。
+
 ## 通知和日志
 
 - Telegram 中英文模板新增市场行：市场、交易币种、标的后缀。

scripts/print_strategy_profile_status.py

@@ -10,7 +10,35 @@
 UES_SRC = ROOT.parent / "UsEquityStrategies" / "src"
 HES_SRC = ROOT.parent / "HkEquityStrategies" / "src"
 
+
+def _has_catalog_marker(candidate: Path, package_name: str, marker: str) -> bool:
+    catalog_path = candidate / package_name / "catalog.py"
+    if not catalog_path.exists():
+        return False
+    return marker in catalog_path.read_text(encoding="utf-8")
+
+
+def _should_add_local_src(candidate: Path) -> bool:
+    if candidate == QPK_SRC:
+        return (candidate / "quant_platform_kit" / "common" / "runtime_target.py").exists()
+    if candidate == UES_SRC:
+        return _has_catalog_marker(
+            candidate,
+            "us_equity_strategies",
+            "mega_cap_leader_rotation_top50_balanced",
+        )
+    if candidate == HES_SRC:
+        return _has_catalog_marker(
+            candidate,
+            "hk_equity_strategies",
+            "hk_listed_global_etf_rotation",
+        )
+    return True
+
+
 for candidate in (ROOT, QPK_SRC, UES_SRC, HES_SRC):
+    if not _should_add_local_src(candidate):
+        continue
     candidate_str = str(candidate)
     if candidate_str not in sys.path:
         sys.path.insert(0, candidate_str)

scripts/print_strategy_switch_env_plan.py

@@ -11,14 +11,43 @@
 UES_SRC = ROOT.parent / "UsEquityStrategies" / "src"
 HES_SRC = ROOT.parent / "HkEquityStrategies" / "src"
 
+
+def _has_catalog_marker(candidate: Path, package_name: str, marker: str) -> bool:
+    catalog_path = candidate / package_name / "catalog.py"
+    if not catalog_path.exists():
+        return False
+    return marker in catalog_path.read_text(encoding="utf-8")
+
+
+def _should_add_local_src(candidate: Path) -> bool:
+    if candidate == QPK_SRC:
+        return (candidate / "quant_platform_kit" / "common" / "runtime_target.py").exists()
+    if candidate == UES_SRC:
+        return _has_catalog_marker(
+            candidate,
+            "us_equity_strategies",
+            "mega_cap_leader_rotation_top50_balanced",
+        )
+    if candidate == HES_SRC:
+        return _has_catalog_marker(
+            candidate,
+            "hk_equity_strategies",
+            "hk_listed_global_etf_rotation",
+        )
+    return True
+
+
 for candidate in (ROOT, QPK_SRC, UES_SRC, HES_SRC):
+    if not _should_add_local_src(candidate):
+        continue
     candidate_str = str(candidate)
     if candidate_str not in sys.path:
         sys.path.insert(0, candidate_str)
 
 from quant_platform_kit.common.runtime_target import build_runtime_target  # noqa: E402
 from quant_platform_kit.common.strategies import derive_strategy_artifact_paths  # noqa: E402
 from strategy_registry import (  # noqa: E402
+    HK_EQUITY_DOMAIN,
     LONGBRIDGE_PLATFORM,
     STRATEGY_CATALOG,
     describe_platform_runtime_requirements,
@@ -28,7 +57,39 @@
 )
 
 
-def build_switch_plan(profile: str, *, account_region: str | None = None) -> dict[str, object]:
+LONGBRIDGE_HK_MARKET_ENV: dict[str, str] = {
+    "LONGBRIDGE_MARKET": "HK",
+    "LONGBRIDGE_MARKET_CALENDAR": "XHKG",
+    "LONGBRIDGE_MARKET_TIMEZONE": "Asia/Hong_Kong",
+    "LONGBRIDGE_SYMBOL_SUFFIX": ".HK",
+    "LONGBRIDGE_TRADING_CURRENCY": "HKD",
+}
+
+HK_DRY_RUN_CHECKS = [
+    "Confirm LongBridge ACCOUNT_REGION/ACCOUNT_PREFIX point at the intended HK verify-only runtime identity.",
+    "Confirm HK market-data and trading permissions before evaluating the strategy.",
+    "Load market_history for all HK managed symbols with .HK/HKD mapping.",
+    "Preview orders only; keep LONGBRIDGE_DRY_RUN_ONLY=true until operator approval.",
+    "Verify integer-share sizing and broker lot-size behavior before any live order submission.",
+    "Verify HKD cash, reserved-cash policy, fees, notifications, and runtime report output.",
+]
+
+HK_BLOCKED_DRY_RUN_ACTIONS = [
+    "Do not deploy or update the production Cloud Run service from this plan.",
+    "Do not submit live LongBridge orders while dry_run_only=true.",
+    "Do not remove HK market overrides when testing hk_equity profiles.",
+]
+
+
+def build_switch_plan(
+    profile: str,
+    *,
+    account_region: str | None = None,
+    dry_run_only: bool = False,
+    deployment_selector: str | None = None,
+    account_scope: str | None = None,
+    service_name: str | None = None,
+) -> dict[str, object]:
     definition = resolve_strategy_definition(profile, platform_id=LONGBRIDGE_PLATFORM)
     metadata = resolve_strategy_metadata(definition.profile, platform_id=LONGBRIDGE_PLATFORM)
     status_row = next(
@@ -50,13 +111,19 @@ def build_switch_plan(profile: str, *, account_region: str | None = None) -> dic
     requires_strategy_config_path = bool(runtime_requirements["requires_strategy_config_path"])
     config_source_policy = str(runtime_requirements.get("config_source_policy") or "none")
     normalized_region = (account_region or "").strip().upper()
+    if definition.domain == HK_EQUITY_DOMAIN and not normalized_region:
+        normalized_region = "HK"
+    resolved_service_name = (
+        service_name
+        or (f"longbridge-quant-{normalized_region.lower()}-service" if normalized_region else None)
+    )
     runtime_target = build_runtime_target(
         platform_id=LONGBRIDGE_PLATFORM,
         strategy_profile=definition.profile,
-        dry_run_only=False,
-        deployment_selector=normalized_region or None,
-        account_scope=normalized_region or None,
-        service_name=f"longbridge-quant-{normalized_region.lower()}-service" if normalized_region else None,
+        dry_run_only=dry_run_only,
+        deployment_selector=deployment_selector or normalized_region or None,
+        account_scope=account_scope or normalized_region or None,
+        service_name=resolved_service_name,
     )
 
     set_env: dict[str, str] = {
@@ -87,6 +154,36 @@ def build_switch_plan(profile: str, *, account_region: str | None = None) -> dic
         "Keep ACCOUNT_PREFIX and ACCOUNT_REGION aligned to the current paper, HK, or SG service identity.",
         "For HK-equity deployments set LONGBRIDGE_MARKET=HK, or rely on ACCOUNT_REGION=HK to derive .HK/HKD/XHKG defaults.",
     ]
+    dry_run_plan: dict[str, object] = {}
+    if dry_run_only:
+        set_env["LONGBRIDGE_DRY_RUN_ONLY"] = "true"
+        dry_run_plan = {
+            "dry_run_only": True,
+            "verify_only": True,
+            "checks": [
+                "Confirm runtime_target.execution_mode is paper before applying any env plan.",
+                "Review broker order preview and notifications; do not submit live orders.",
+            ],
+            "blocked_actions": [
+                "Do not deploy or update production Cloud Run from a switch-plan printout alone.",
+                "Do not submit live orders while dry_run_only=true.",
+            ],
+        }
+    if definition.domain == HK_EQUITY_DOMAIN:
+        set_env.update(LONGBRIDGE_HK_MARKET_ENV)
+        set_env["LONGBRIDGE_DRY_RUN_ONLY"] = "true" if dry_run_only else "false"
+        dry_run_plan = {
+            "dry_run_only": dry_run_only,
+            "verify_only": dry_run_only,
+            "market": "HK",
+            "checks": HK_DRY_RUN_CHECKS,
+            "blocked_actions": HK_BLOCKED_DRY_RUN_ACTIONS if dry_run_only else [],
+        }
+        notes.append(
+            "HK-equity switch plans are environment plans only; merge alone must not change production Cloud Run."
+        )
+        if not dry_run_only:
+            notes.append("Use --dry-run-only for first HK runtime validation; live mode requires separate operator approval.")
 
     if not normalized_region:
         notes.append("Pass --account-region PAPER, HK, or SG if you want ACCOUNT_PREFIX/ACCOUNT_REGION placeholders filled in.")
@@ -142,6 +239,7 @@ def build_switch_plan(profile: str, *, account_region: str | None = None) -> dic
         "optional_env": sorted(optional_env),
         "remove_if_present": sorted(set(remove_if_present)),
         "hints": hints,
+        "dry_run_plan": dry_run_plan,
         "notes": notes,
     }
 
@@ -173,6 +271,9 @@ def _print_plan(plan: dict[str, object]) -> None:
         print("\nhints:")
         for key, value in plan["hints"].items():
             print(f"  {key}: {value}")
+    if plan["dry_run_plan"]:
+        print("\ndry_run_plan:")
+        print(json.dumps(plan["dry_run_plan"], indent=2, sort_keys=True))
     if plan["notes"]:
         print("\nnotes:")
         for note in plan["notes"]:
@@ -183,10 +284,21 @@ def main() -> int:
     parser = argparse.ArgumentParser()
     parser.add_argument("--profile", required=True)
     parser.add_argument("--account-region")
+    parser.add_argument("--dry-run-only", action="store_true")
+    parser.add_argument("--deployment-selector")
+    parser.add_argument("--account-scope")
+    parser.add_argument("--service-name")
     parser.add_argument("--json", action="store_true")
     args = parser.parse_args()
 
-    plan = build_switch_plan(args.profile, account_region=args.account_region)
+    plan = build_switch_plan(
+        args.profile,
+        account_region=args.account_region,
+        dry_run_only=args.dry_run_only,
+        deployment_selector=args.deployment_selector,
+        account_scope=args.account_scope,
+        service_name=args.service_name,
+    )
     if args.json:
         print(json.dumps(plan, indent=2, sort_keys=True))
         return 0

tests/test_runtime_config_support.py

@@ -12,7 +12,7 @@
 if str(ROOT) not in sys.path:
     sys.path.insert(0, str(ROOT))
 QPK_SRC = ROOT.parent / "QuantPlatformKit" / "src"
-if str(QPK_SRC) not in sys.path:
+if (QPK_SRC / "quant_platform_kit" / "common" / "runtime_config.py").exists() and str(QPK_SRC) not in sys.path:
     sys.path.insert(0, str(QPK_SRC))
 SCRIPT_PATH = ROOT / "scripts" / "print_strategy_profile_status.py"
 SWITCH_PLAN_SCRIPT_PATH = ROOT / "scripts" / "print_strategy_switch_env_plan.py"
@@ -852,6 +852,60 @@ def test_print_strategy_switch_env_plan_for_global_etf_rotation(self):
         self.assertIn("LONGBRIDGE_TRADING_CURRENCY", plan["optional_env"])
         self.assertIn("LONGBRIDGE_FEATURE_SNAPSHOT_PATH", plan["remove_if_present"])
 
+    def test_print_strategy_switch_env_plan_for_hk_global_etf_dry_run(self):
+        result = subprocess.run(
+            [
+                sys.executable,
+                str(SWITCH_PLAN_SCRIPT_PATH),
+                "--profile",
+                "hk_listed_global_etf_rotation",
+                "--account-region",
+                "hk",
+                "--dry-run-only",
+                "--deployment-selector",
+                "hk-verify",
+                "--account-scope",
+                "hk-verify",
+                "--service-name",
+                "longbridge-quant-hk-verify-service",
+                "--json",
+            ],
+            check=True,
+            capture_output=True,
+            text=True,
+        )
+
+        plan = json.loads(result.stdout)
+        self.assertEqual(plan["platform"], "longbridge")
+        self.assertEqual(plan["canonical_profile"], "hk_listed_global_etf_rotation")
+        self.assertEqual(plan["domain"], HK_EQUITY_DOMAIN)
+        self.assertEqual(plan["set_env"]["ACCOUNT_REGION"], "HK")
+        self.assertEqual(plan["set_env"]["ACCOUNT_PREFIX"], "HK")
+        self.assertEqual(plan["set_env"]["LONGBRIDGE_DRY_RUN_ONLY"], "true")
+        self.assertEqual(plan["set_env"]["LONGBRIDGE_MARKET"], HK_MARKET)
+        self.assertEqual(plan["set_env"]["LONGBRIDGE_MARKET_CALENDAR"], HK_MARKET_CALENDAR)
+        self.assertEqual(plan["set_env"]["LONGBRIDGE_MARKET_TIMEZONE"], HK_MARKET_TIMEZONE)
+        self.assertEqual(plan["set_env"]["LONGBRIDGE_SYMBOL_SUFFIX"], HK_SYMBOL_SUFFIX)
+        self.assertEqual(plan["set_env"]["LONGBRIDGE_TRADING_CURRENCY"], HK_TRADING_CURRENCY)
+        self.assertTrue(plan["runtime_target"]["dry_run_only"])
+        self.assertEqual(plan["runtime_target"]["execution_mode"], "paper")
+        self.assertEqual(plan["runtime_target"]["deployment_selector"], "hk-verify")
+        self.assertEqual(plan["runtime_target"]["account_scope"], "hk-verify")
+        self.assertEqual(plan["runtime_target"]["service_name"], "longbridge-quant-hk-verify-service")
+        runtime_target_env = json.loads(plan["set_env"]["RUNTIME_TARGET_JSON"])
+        self.assertTrue(runtime_target_env["dry_run_only"])
+        self.assertEqual(runtime_target_env["execution_mode"], "paper")
+        self.assertEqual(plan["profile_group"], "direct_runtime_inputs")
+        self.assertEqual(plan["input_mode"], "market_history")
+        self.assertFalse(plan["requires_snapshot_artifacts"])
+        self.assertIn("LONGBRIDGE_FEATURE_SNAPSHOT_PATH", plan["remove_if_present"])
+        self.assertTrue(plan["dry_run_plan"]["dry_run_only"])
+        self.assertTrue(plan["dry_run_plan"]["verify_only"])
+        self.assertTrue(any("lot-size" in check for check in plan["dry_run_plan"]["checks"]))
+        self.assertTrue(
+            any("production Cloud Run" in action for action in plan["dry_run_plan"]["blocked_actions"])
+        )
+
     def test_print_strategy_switch_env_plan_for_russell(self):
         result = subprocess.run(
             [