feat(backtest): task 3c HK walk-forward orchestrator CLI

scripts/research_hk_proxy_orchestrator_backtest.py

@@ -0,0 +1,56 @@
+#!/usr/bin/env python3
+"""Generic HK orchestrator research entrypoint (task 3c)."""
+
+from __future__ import annotations
+
+import argparse
+import json
+import sys
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parents[1]
+if str(ROOT) not in sys.path:
+    sys.path.insert(0, str(ROOT))
+
+from hk_equity_strategies.backtest.orchestrator_runner import HkEtfRotationBacktestRunner, SUPPORTED_PROFILES  # noqa: E402
+from scripts.run_walk_forward_backtest import run_walk_forward  # noqa: E402
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description="HK orchestrator research backtest.")
+    parser.add_argument("--profile", default="hk_global_etf_tactical_rotation")
+    parser.add_argument("--list-profiles", action="store_true")
+    parser.add_argument("--mode", choices=("single", "walk_forward"), default="walk_forward")
+    parser.add_argument("--synthetic-days", type=int, default=700)
+    parser.add_argument("--json-output", type=Path)
+    args = parser.parse_args()
+
+    if args.list_profiles:
+        print(json.dumps({"profiles": sorted(SUPPORTED_PROFILES)}, indent=2))
+        return 0
+
+    if args.mode == "walk_forward":
+        payload = run_walk_forward(profile=args.profile, synthetic_days=args.synthetic_days)
+    else:
+        runner = HkEtfRotationBacktestRunner(synthetic_days=args.synthetic_days)
+        params = {"min_history_days": 200}
+        result = runner.run(args.profile, params)
+        payload = {
+            "profile": args.profile,
+            "metrics": {
+                "sharpe_ratio": result.sharpe_ratio,
+                "max_drawdown": result.max_drawdown,
+                "cagr": result.cagr,
+            },
+            "source": "HkEtfRotationBacktestRunner",
+        }
+
+    text = json.dumps(payload, indent=2, sort_keys=True, default=str)
+    if args.json_output:
+        args.json_output.write_text(text + "\n")
+    print(text)
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

scripts/run_hk_global_etf_walk_forward_pilot.py

@@ -1,68 +1,29 @@
 #!/usr/bin/env python3
-"""Pilot: run hk_global_etf_tactical_rotation through BacktestOrchestrator.walk_forward()."""
+"""Pilot wrapper — delegates to run_walk_forward_backtest.py (task 3c)."""
 
 from __future__ import annotations
 
 import argparse
 import json
-from datetime import date
+import sys
 from pathlib import Path
-from typing import Any
 
-from hk_equity_strategies.backtest.orchestrator_runner import HkEtfRotationBacktestRunner
-from hk_equity_strategies.strategies.hk_global_etf_tactical_rotation import (
-    DEFAULT_MIN_HISTORY_DAYS,
-    PROFILE_NAME,
-)
+ROOT = Path(__file__).resolve().parents[1]
+if str(ROOT) not in sys.path:
+    sys.path.insert(0, str(ROOT))
 
-DEFAULT_WINDOWS: tuple[tuple[date, date], ...] = (
-    (date(2023, 6, 1), date(2024, 5, 31)),
-    (date(2024, 6, 1), date(2025, 5, 31)),
-)
+from scripts.run_walk_forward_backtest import run_walk_forward  # noqa: E402
 
 
 def main() -> int:
-    parser = argparse.ArgumentParser(description="HK global ETF walk-forward pilot")
+    parser = argparse.ArgumentParser(description="HK global ETF walk-forward pilot (compat wrapper).")
     parser.add_argument("--output", type=Path, default=Path("hk_walk_forward_pilot.json"))
     parser.add_argument("--synthetic-days", type=int, default=700)
     args = parser.parse_args()
-
-    from quant_platform_kit.strategy_lifecycle.backtest_orchestrator import BacktestOrchestrator
-    from quant_platform_kit.strategy_lifecycle.performance_store import PerformanceStore
-
-    runner = HkEtfRotationBacktestRunner(synthetic_days=args.synthetic_days)
-    params: dict[str, Any] = {"min_history_days": DEFAULT_MIN_HISTORY_DAYS}
-    store = PerformanceStore(local_root=args.output.parent / ".wf_store")
-    orchestrator = BacktestOrchestrator(store=store)
-    orchestrator.register_runner("hk_equity", runner)
-
-    baseline = runner.run(PROFILE_NAME, params)
-    results = orchestrator.walk_forward(
-        PROFILE_NAME,
-        domain="hk_equity",
-        params=params,
-        windows=DEFAULT_WINDOWS,
-    )
-    payload = {
-        "profile": PROFILE_NAME,
-        "baseline": {
-            "sharpe_ratio": baseline.sharpe_ratio,
-            "max_drawdown": baseline.max_drawdown,
-            "cagr": baseline.cagr,
-        },
-        "windows": [
-            {
-                "start": item.start_date.isoformat() if item.start_date else None,
-                "end": item.end_date.isoformat() if item.end_date else None,
-                "sharpe_ratio": item.sharpe_ratio,
-                "max_drawdown": item.max_drawdown,
-                "cagr": item.cagr,
-            }
-            for item in results
-        ],
-    }
-    args.output.write_text(json.dumps(payload, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
-    print(json.dumps(payload, ensure_ascii=False, indent=2))
+    payload = run_walk_forward(profile="hk_global_etf_tactical_rotation", synthetic_days=args.synthetic_days)
+    text = json.dumps(payload, ensure_ascii=False, indent=2)
+    args.output.write_text(text + "\n", encoding="utf-8")
+    print(text)
     return 0
 
 

scripts/run_walk_forward_backtest.py

@@ -0,0 +1,101 @@
+#!/usr/bin/env python3
+"""Run walk-forward backtests via QuantPlatformKit BacktestOrchestrator."""
+
+from __future__ import annotations
+
+import argparse
+import json
+from datetime import date
+from pathlib import Path
+from typing import Any
+
+from hk_equity_strategies.backtest.orchestrator_runner import HkEtfRotationBacktestRunner, SUPPORTED_PROFILES
+from hk_equity_strategies.strategies.hk_global_etf_tactical_rotation import DEFAULT_MIN_HISTORY_DAYS
+
+DEFAULT_WINDOWS: tuple[tuple[date, date], ...] = (
+    (date(2023, 6, 1), date(2024, 5, 31)),
+    (date(2024, 6, 1), date(2025, 5, 31)),
+)
+
+PROFILE_DEFAULTS: dict[str, dict[str, Any]] = {
+    "hk_global_etf_tactical_rotation": {"min_history_days": DEFAULT_MIN_HISTORY_DAYS},
+}
+
+
+def _result_payload(item: Any) -> dict[str, Any]:
+    return {
+        "start_date": item.start_date.isoformat() if item.start_date else None,
+        "end_date": item.end_date.isoformat() if item.end_date else None,
+        "sharpe_ratio": item.sharpe_ratio,
+        "max_drawdown": item.max_drawdown,
+        "cagr": item.cagr,
+        "total_return": item.total_return,
+        "observation_count": item.observation_count,
+        "run_id": getattr(item, "run_id", None),
+    }
+
+
+def run_walk_forward(
+    *,
+    profile: str,
+    windows: tuple[tuple[date, date], ...] = DEFAULT_WINDOWS,
+    synthetic_days: int = 700,
+    store_root: Path | None = None,
+) -> dict[str, Any]:
+    from quant_platform_kit.strategy_lifecycle.backtest_orchestrator import BacktestOrchestrator
+    from quant_platform_kit.strategy_lifecycle.performance_store import PerformanceStore
+
+    if profile not in SUPPORTED_PROFILES:
+        raise ValueError(f"unsupported profile={profile!r}; supported={sorted(SUPPORTED_PROFILES)}")
+
+    params = dict(PROFILE_DEFAULTS.get(profile, {"min_history_days": DEFAULT_MIN_HISTORY_DAYS}))
+    runner = HkEtfRotationBacktestRunner(synthetic_days=synthetic_days)
+    store = PerformanceStore(local_root=store_root or Path("/tmp/hk_equity_wf_store"))
+    orchestrator = BacktestOrchestrator(store=store)
+    orchestrator.register_runner("hk_equity", runner)
+
+    baseline = runner.run(profile, params)
+    wf_results = orchestrator.walk_forward(
+        profile,
+        domain="hk_equity",
+        params=params,
+        windows=windows,
+        param_set_id=f"{profile}_wf",
+    )
+    return {
+        "strategy_profile": profile,
+        "domain": "hk_equity",
+        "baseline": _result_payload(baseline),
+        "walk_forward_folds": [_result_payload(item) for item in wf_results],
+        "source": "BacktestOrchestrator.walk_forward",
+    }
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description="HK walk-forward backtest via BacktestOrchestrator.")
+    parser.add_argument("--profile", default="hk_global_etf_tactical_rotation")
+    parser.add_argument("--list-profiles", action="store_true")
+    parser.add_argument("--json-output", type=Path)
+    parser.add_argument("--synthetic-days", type=int, default=700)
+    parser.add_argument("--store-root", type=Path)
+    args = parser.parse_args()
+
+    if args.list_profiles:
+        print(json.dumps({"profiles": sorted(SUPPORTED_PROFILES)}, indent=2))
+        return 0
+
+    payload = run_walk_forward(
+        profile=args.profile,
+        synthetic_days=args.synthetic_days,
+        store_root=args.store_root,
+    )
+    text = json.dumps(payload, indent=2, sort_keys=True, default=str)
+    if args.json_output:
+        args.json_output.parent.mkdir(parents=True, exist_ok=True)
+        args.json_output.write_text(text + "\n")
+    print(text)
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())