feat(backtest): HK global ETF walk-forward pilot

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())

src/hk_equity_strategies/backtest/__init__.py

@@ -0,0 +1 @@
+"""HK equity backtest helpers."""

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",
+]

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"]