From 5f6bb03094dd74255437cdfa5eee71a3eb538d60 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Thu, 9 Jul 2026 00:23:41 +0800 Subject: [PATCH] feat(backtest): add HK ETF rotation walk-forward runner pilot Task 3c: HkEtfRotationBacktestRunner + BacktestOrchestrator.walk_forward for hk_global_etf_tactical_rotation (mirrors CN #40 pattern). Co-Authored-By: Claude Co-authored-by: Cursor --- .../run_hk_global_etf_walk_forward_pilot.py | 70 ++++++++ src/hk_equity_strategies/backtest/__init__.py | 1 + .../backtest/etf_rotation_simulator.py | 133 +++++++++++++++ .../backtest/orchestrator_runner.py | 157 ++++++++++++++++++ tests/test_orchestrator_runner.py | 62 +++++++ 5 files changed, 423 insertions(+) create mode 100644 scripts/run_hk_global_etf_walk_forward_pilot.py create mode 100644 src/hk_equity_strategies/backtest/__init__.py create mode 100644 src/hk_equity_strategies/backtest/etf_rotation_simulator.py create mode 100644 src/hk_equity_strategies/backtest/orchestrator_runner.py create mode 100644 tests/test_orchestrator_runner.py diff --git a/scripts/run_hk_global_etf_walk_forward_pilot.py b/scripts/run_hk_global_etf_walk_forward_pilot.py new file mode 100644 index 0000000..aacd0fb --- /dev/null +++ b/scripts/run_hk_global_etf_walk_forward_pilot.py @@ -0,0 +1,70 @@ +#!/usr/bin/env python3 +"""Pilot: run hk_global_etf_tactical_rotation through BacktestOrchestrator.walk_forward().""" + +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 +from hk_equity_strategies.strategies.hk_global_etf_tactical_rotation import ( + DEFAULT_MIN_HISTORY_DAYS, + PROFILE_NAME, +) + +DEFAULT_WINDOWS: tuple[tuple[date, date], ...] = ( + (date(2023, 6, 1), date(2024, 5, 31)), + (date(2024, 6, 1), date(2025, 5, 31)), +) + + +def main() -> int: + parser = argparse.ArgumentParser(description="HK global ETF walk-forward pilot") + 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)) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/src/hk_equity_strategies/backtest/__init__.py b/src/hk_equity_strategies/backtest/__init__.py new file mode 100644 index 0000000..dc54a47 --- /dev/null +++ b/src/hk_equity_strategies/backtest/__init__.py @@ -0,0 +1 @@ +"""HK equity backtest helpers.""" diff --git a/src/hk_equity_strategies/backtest/etf_rotation_simulator.py b/src/hk_equity_strategies/backtest/etf_rotation_simulator.py new file mode 100644 index 0000000..947d35a --- /dev/null +++ b/src/hk_equity_strategies/backtest/etf_rotation_simulator.py @@ -0,0 +1,133 @@ +"""Weight-based ETF rotation backtest for HK orchestrator integration.""" + +from __future__ import annotations + +import math +from dataclasses import dataclass, field +from typing import Any, Callable, Mapping, Sequence + +import pandas as pd + +from hk_equity_strategies.strategies.etf_rotation_core import build_close_matrix, normalize_symbol + +StrategySignalFn = Callable[[Any], tuple[Mapping[str, float], Mapping[str, object]]] + + +@dataclass(frozen=True) +class HkRotationBacktestConfig: + rebalance_frequency: str = "monthly" + min_history_days: int = 260 + cost_bps: float = 10.0 + + +@dataclass +class HkRotationBacktestResult: + daily_returns: pd.Series + metrics: dict[str, float | int] = field(default_factory=dict) + + +def _rebalance_dates(index: pd.DatetimeIndex, *, frequency: str) -> pd.DatetimeIndex: + if frequency == "monthly": + return index.to_series().resample("ME").last().dropna().index + if frequency == "weekly": + return index.to_series().resample("W-FRI").last().dropna().index + raise ValueError("rebalance_frequency must be 'monthly' or 'weekly'") + + +def _history_slice(market_history: pd.DataFrame, as_of: pd.Timestamp) -> pd.DataFrame: + frame = market_history.copy() + frame["date"] = pd.to_datetime(frame["date"], utc=False).dt.tz_localize(None).dt.normalize() + return frame.loc[frame["date"] <= as_of] + + +def _target_weights( + market_history: pd.DataFrame, + close: pd.DataFrame, + *, + signal_fn: StrategySignalFn, + config: HkRotationBacktestConfig, + strategy_kwargs: Mapping[str, Any], +) -> pd.DataFrame: + rows: list[dict[str, Any]] = [] + for target_date in _rebalance_dates(pd.DatetimeIndex(close.index), frequency=config.rebalance_frequency): + position = close.index.searchsorted(target_date, side="right") - 1 + if position < 0: + continue + as_of = pd.Timestamp(close.index[position]) + history = _history_slice(market_history, as_of) + if len(history["date"].drop_duplicates()) < int(config.min_history_days): + weights: dict[str, float] = {} + else: + weights, _metadata = signal_fn(history, **dict(strategy_kwargs)) + rows.append( + {"date": as_of, **{symbol: float(weights.get(symbol, 0.0)) for symbol in close.columns}} + ) + targets = pd.DataFrame(rows).set_index("date") + targets = targets.reindex(close.index, method="ffill").fillna(0.0) + return targets[list(close.columns)].shift(1).fillna(0.0) + + +def compute_backtest_metrics(daily_returns: pd.Series) -> dict[str, float | int]: + returns = daily_returns.dropna() + if returns.empty: + return { + "days": 0, + "annual_return": 0.0, + "max_drawdown": 0.0, + "annual_volatility": 0.0, + "total_return": 0.0, + "sharpe_ratio": 0.0, + } + equity = (1.0 + returns).cumprod() + years = len(returns) / 252.0 + annual_return = float(equity.iloc[-1] ** (1 / years) - 1) if years > 0 else 0.0 + drawdown = equity / equity.cummax() - 1.0 + annual_volatility = float(returns.std(ddof=0) * math.sqrt(252)) + sharpe = annual_return / annual_volatility if annual_volatility > 0 else 0.0 + return { + "days": int(len(returns)), + "annual_return": annual_return, + "max_drawdown": float(drawdown.min()), + "annual_volatility": annual_volatility, + "total_return": float(equity.iloc[-1] - 1.0), + "sharpe_ratio": float(sharpe), + } + + +def run_etf_rotation_backtest( + market_history: pd.DataFrame, + strategy_signal_fn: StrategySignalFn, + *, + config: HkRotationBacktestConfig | None = None, + universe_symbols: Sequence[str] | None = None, + strategy_kwargs: Mapping[str, Any] | None = None, +) -> HkRotationBacktestResult: + settings = config or HkRotationBacktestConfig() + kwargs = dict(strategy_kwargs or {}) + close = build_close_matrix(market_history, universe_symbols=universe_symbols) + if len(close) < int(settings.min_history_days): + raise ValueError( + f"market_history requires at least {int(settings.min_history_days)} overlapping trading days" + ) + + returns = close.pct_change().fillna(0.0) + targets = _target_weights( + market_history, + close, + signal_fn=strategy_signal_fn, + config=settings, + strategy_kwargs=kwargs, + ) + turnover = targets.diff().abs().sum(axis=1).fillna(0.0) + net = (targets * returns).sum(axis=1) - turnover * float(settings.cost_bps) / 10_000.0 + metrics = compute_backtest_metrics(net) + return HkRotationBacktestResult(daily_returns=net, metrics=metrics) + + +__all__ = [ + "HkRotationBacktestConfig", + "HkRotationBacktestResult", + "compute_backtest_metrics", + "normalize_symbol", + "run_etf_rotation_backtest", +] diff --git a/src/hk_equity_strategies/backtest/orchestrator_runner.py b/src/hk_equity_strategies/backtest/orchestrator_runner.py new file mode 100644 index 0000000..340ac68 --- /dev/null +++ b/src/hk_equity_strategies/backtest/orchestrator_runner.py @@ -0,0 +1,157 @@ +"""BacktestRunner adapter for HK ETF rotation strategies.""" + +from __future__ import annotations + +from datetime import date, datetime, timezone +from typing import Any, Mapping + +import pandas as pd + +from hk_equity_strategies.backtest.etf_rotation_simulator import HkRotationBacktestConfig, run_etf_rotation_backtest +from hk_equity_strategies.strategies.hk_global_etf_tactical_rotation import ( + DEFAULT_MIN_HISTORY_DAYS, + DEFAULT_UNIVERSE_SYMBOLS, + PROFILE_NAME as HK_GLOBAL_ETF_TACTICAL_ROTATION_PROFILE, + build_target_weights, + extract_managed_symbols, +) + +try: + from quant_platform_kit.strategy_lifecycle.contracts import BacktestResult +except ImportError: # pragma: no cover + BacktestResult = None # type: ignore[misc, assignment] + + +SUPPORTED_PROFILES = frozenset({HK_GLOBAL_ETF_TACTICAL_ROTATION_PROFILE}) + + +def _synthetic_market_history(*, days: int = 900, start: str = "2022-01-03") -> pd.DataFrame: + dates = pd.bdate_range(start, periods=days) + symbols = tuple(extract_managed_symbols(universe_symbols=DEFAULT_UNIVERSE_SYMBOLS)) + rates = {symbol: 1.00015 + (idx * 0.00004) for idx, symbol in enumerate(symbols)} + rows: list[dict[str, object]] = [] + for symbol in symbols: + price = 12.0 + (hash(symbol) % 9) + rate = rates.get(symbol, 1.00015) + for idx, day in enumerate(dates): + price *= rate + close = price * (1.0 + 0.02 * ((idx % 11) - 5) / 11) + rows.append({"date": day, "symbol": symbol, "close": close}) + return pd.DataFrame(rows) + + +def _slice_history( + market_history: pd.DataFrame, + *, + start_date: date | None, + end_date: date | None, + lookback_days: int = 0, +) -> pd.DataFrame: + frame = market_history.copy() + frame["date"] = pd.to_datetime(frame["date"], utc=False).dt.tz_localize(None).dt.normalize() + if start_date is not None: + effective_start = pd.Timestamp(start_date) - pd.tseries.offsets.BDay(max(int(lookback_days), 0)) + frame = frame[frame["date"] >= effective_start] + if end_date is not None: + frame = frame[frame["date"] <= pd.Timestamp(end_date)] + return frame.sort_values(["date", "symbol"]).reset_index(drop=True) + + +def _signal_fn(history: Any, **kwargs: Any): + return build_target_weights(history, **kwargs) + + +def _metrics_to_backtest_result( + *, + strategy_profile: str, + params: Mapping[str, Any], + metrics: Mapping[str, Any], + start_date: date | None, + end_date: date | None, + run_duration_seconds: float, +) -> Any: + if BacktestResult is None: + raise ImportError("quant_platform_kit is required to build BacktestResult") + annual_return = float(metrics.get("annual_return") or 0.0) + max_drawdown = float(metrics.get("max_drawdown") or 0.0) + calmar = abs(annual_return / max_drawdown) if max_drawdown else None + return BacktestResult( + strategy_profile=strategy_profile, + domain="hk_equity", + param_set_id="", + params=dict(params), + sharpe_ratio=float(metrics.get("sharpe_ratio") or 0.0), + calmar_ratio=calmar, + max_drawdown=max_drawdown, + cagr=annual_return, + volatility=float(metrics.get("annual_volatility") or 0.0), + total_return=float(metrics.get("total_return") or 0.0), + start_date=start_date, + end_date=end_date, + observation_count=int(metrics.get("days") or 0), + source_script="hk_equity_strategies.backtest.orchestrator_runner", + computed_at=datetime.now(timezone.utc).isoformat(), + run_duration_seconds=run_duration_seconds, + ) + + +class HkEtfRotationBacktestRunner: + """Protocol-compatible BacktestRunner for HK global ETF rotation.""" + + def __init__( + self, + *, + market_history: pd.DataFrame | None = None, + synthetic_days: int = 700, + ) -> None: + self._market_history = market_history + self._synthetic_days = int(synthetic_days) + + def run( + self, + strategy_profile: str, + params: Mapping[str, Any], + start_date: date | None = None, + end_date: date | None = None, + ) -> Any: + if strategy_profile not in SUPPORTED_PROFILES: + raise ValueError( + f"Unsupported strategy_profile={strategy_profile!r}; " + f"supported={sorted(SUPPORTED_PROFILES)}" + ) + + min_history_days = int(params.get("min_history_days", DEFAULT_MIN_HISTORY_DAYS)) + history = self._market_history + if history is None: + history = _synthetic_market_history(days=max(self._synthetic_days, min_history_days + 400)) + sliced = _slice_history( + history, + start_date=start_date, + end_date=end_date, + lookback_days=min_history_days + 5, + ) + if sliced.empty: + raise ValueError("No market history rows for requested window") + + started = datetime.now(timezone.utc) + result = run_etf_rotation_backtest( + sliced, + _signal_fn, + config=HkRotationBacktestConfig(min_history_days=min_history_days), + strategy_kwargs={"min_history_days": min_history_days}, + ) + elapsed = (datetime.now(timezone.utc) - started).total_seconds() + eval_frame = sliced + if start_date is not None: + eval_frame = sliced[sliced["date"] >= pd.Timestamp(start_date)] + return _metrics_to_backtest_result( + strategy_profile=strategy_profile, + params=params, + metrics=result.metrics, + start_date=start_date or (eval_frame["date"].min().date() if not eval_frame.empty else None), + end_date=end_date or (eval_frame["date"].max().date() if not eval_frame.empty else None), + run_duration_seconds=elapsed, + ) + + +__all__ = ["SUPPORTED_PROFILES", "HkEtfRotationBacktestRunner"] diff --git a/tests/test_orchestrator_runner.py b/tests/test_orchestrator_runner.py new file mode 100644 index 0000000..896f49c --- /dev/null +++ b/tests/test_orchestrator_runner.py @@ -0,0 +1,62 @@ +from __future__ import annotations + +import tempfile +import unittest +from datetime import date + +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, + PROFILE_NAME, +) + + +class HkEtfRotationBacktestRunnerTests(unittest.TestCase): + def test_supported_profile_includes_global_etf(self) -> None: + self.assertIn(PROFILE_NAME, SUPPORTED_PROFILES) + + def test_run_returns_backtest_result(self) -> None: + runner = HkEtfRotationBacktestRunner(synthetic_days=500) + result = runner.run( + PROFILE_NAME, + {"min_history_days": DEFAULT_MIN_HISTORY_DAYS}, + start_date=date(2023, 6, 1), + end_date=date(2024, 6, 1), + ) + self.assertEqual(result.strategy_profile, PROFILE_NAME) + self.assertEqual(result.domain, "hk_equity") + self.assertIsNotNone(result.sharpe_ratio) + self.assertGreater(result.observation_count, 0) + + def test_unsupported_profile_raises(self) -> None: + runner = HkEtfRotationBacktestRunner(synthetic_days=100) + with self.assertRaises(ValueError): + runner.run("unknown_profile", {}) + + +class WalkForwardPilotTests(unittest.TestCase): + def test_walk_forward_produces_one_result_per_window(self) -> None: + from quant_platform_kit.strategy_lifecycle.backtest_orchestrator import BacktestOrchestrator + from quant_platform_kit.strategy_lifecycle.performance_store import PerformanceStore + from pathlib import Path + + with tempfile.TemporaryDirectory() as tmp: + store = PerformanceStore(local_root=Path(tmp)) + orchestrator = BacktestOrchestrator(store=store) + orchestrator.register_runner("hk_equity", HkEtfRotationBacktestRunner(synthetic_days=700)) + windows = ( + (date(2023, 6, 1), date(2023, 12, 31)), + (date(2024, 1, 1), date(2024, 6, 30)), + ) + results = orchestrator.walk_forward( + PROFILE_NAME, + domain="hk_equity", + params={"min_history_days": DEFAULT_MIN_HISTORY_DAYS}, + windows=windows, + ) + self.assertEqual(len(results), 2) + self.assertTrue(all(item.strategy_profile == PROFILE_NAME for item in results)) + + +if __name__ == "__main__": + unittest.main()