spb/crypto-trade-bot

1"""Walk-forward evaluation, parameter sensitivity maps and holdout handling."""23from __future__ import annotations45import itertools6import math7from typing import Any89import numpy as np10import pandas as pd1112from ktrade.backtest.costs import CostModel13from ktrade.backtest.engine import run_backtest14from ktrade.backtest.metrics import compute_metrics15from ktrade.strategy.base import Strategy161718def _score(m: dict[str, Any]) -> float:19    s = m.get("sharpe")20    return s if s is not None and math.isfinite(s) else -1e9212223def param_grid(strategy_cls: type[Strategy], grid: dict[str, list[Any]]) -> list[dict[str, Any]]:24    keys = list(grid)25    combos = []26    for values in itertools.product(*(grid[k] for k in keys)):27        p = dict(zip(keys, values, strict=True))28        try:29            strategy_cls.validate_params(p)30            combos.append(p)31        except ValueError:32            continue33    return combos343536def walk_forward(37    strategy_cls: type[Strategy], df: pd.DataFrame, *, interval_min: int, grid: dict[str, list[Any]], n_folds: int = 4,38    train_frac: float = 0.6, anchored: bool = True, cost_model: CostModel | None = None, initial_cash: float = 10_000.0, embargo_bars: int = 0, **bt_kwargs,39) -> dict[str, Any]:40    """Split the series into n_folds consecutive test windows; for each, pick the grid point with the best IS Sharpe on41    the preceding train window (anchored = growing train, else rolling of fixed length), then evaluate OOS.42    An embargo of `embargo_bars` between train and test avoids leakage through overlapping feature windows."""43    n = len(df)44    combos = param_grid(strategy_cls, grid)45    if n < 300 or not combos:46        return {"folds": [], "error": "not enough data or empty grid", "n_trials": len(combos)}47    first_test = int(n * train_frac)48    test_len = (n - first_test) // n_folds49    folds = []50    oos_returns: list[np.ndarray] = []51    for k in range(n_folds):52        test_start = first_test + k * test_len53        test_end = n if k == n_folds - 1 else test_start + test_len54        train_start = 0 if anchored else max(0, test_start - first_test)55        train_end = max(train_start + 50, test_start - embargo_bars)56        train = df.iloc[train_start:train_end]57        # the test window gets a warm-up prefix (features need history) but metrics are computed on the test range only58        warm = min(train_end, strategy_cls.meta.min_bars * 3)59        test_with_warm = df.iloc[max(0, test_start - warm):test_end]60        best, best_m = None, None61        for p in combos:62            r = run_backtest(strategy_cls(p), train, interval_min=interval_min, initial_cash=initial_cash, cost_model=cost_model, **bt_kwargs)63            m = compute_metrics(r.equity, r.trades, interval_min, initial_cash, r.fees_paid, r.slippage_paid, r.turnover)64            if best_m is None or _score(m) > _score(best_m):65                best, best_m = p, m66        r = run_backtest(strategy_cls(best), test_with_warm, interval_min=interval_min, initial_cash=initial_cash, cost_model=cost_model, **bt_kwargs)67        eq = r.equity[r.equity["ts"] >= int(df.iloc[test_start]["ts"])].reset_index(drop=True)68        if len(eq) > 2:69            eq = eq.assign(equity=eq["equity"] / eq["equity"].iloc[0] * initial_cash)70        m_oos = compute_metrics(eq, [t for t in r.trades if t.ts >= int(df.iloc[test_start]["ts"])], interval_min, initial_cash, 0.0, 0.0, r.turnover) if len(eq) > 2 else {"error": "short"}71        oos_returns.append(np.diff(np.log(np.maximum(eq["equity"].to_numpy(), 1e-12))) if len(eq) > 2 else np.array([]))72        folds.append({73            "fold": k, "train": [int(df.iloc[train_start]["ts"]), int(df.iloc[train_end - 1]["ts"])], "test": [int(df.iloc[test_start]["ts"]), int(df.iloc[test_end - 1]["ts"])],74            "best_params": best, "is_sharpe": best_m.get("sharpe") if best_m else None, "oos_sharpe": m_oos.get("sharpe"), "oos_return": m_oos.get("total_return"), "oos_max_dd": m_oos.get("max_drawdown"),75        })76    all_oos = np.concatenate([x for x in oos_returns if len(x)]) if any(len(x) for x in oos_returns) else np.array([])77    bars_per_year = 365 * 24 * 60 / interval_min78    stitched = {"sharpe": float(all_oos.mean() / all_oos.std(ddof=1) * math.sqrt(bars_per_year)) if len(all_oos) > 10 and all_oos.std(ddof=1) > 0 else None, "total_return": float(math.exp(all_oos.sum()) - 1) if len(all_oos) else None, "bars": int(len(all_oos))}79    return {"folds": folds, "anchored": anchored, "n_trials": len(combos), "grid": grid, "embargo_bars": embargo_bars, "stitched_oos": stitched}808182def sensitivity_map(strategy_cls: type[Strategy], df: pd.DataFrame, *, interval_min: int, x: tuple[str, list[Any]], y: tuple[str, list[Any]], base_params: dict[str, Any], cost_model: CostModel | None = None, initial_cash: float = 10_000.0, metric: str = "sharpe", **bt_kwargs) -> dict[str, Any]:83    xs, ys = x[1], y[1]84    z: list[list[float | None]] = []85    for yv in ys:86        row: list[float | None] = []87        for xv in xs:88            p = {**base_params, x[0]: xv, y[0]: yv}89            try:90                r = run_backtest(strategy_cls(p), df, interval_min=interval_min, initial_cash=initial_cash, cost_model=cost_model, **bt_kwargs)91                m = compute_metrics(r.equity, r.trades, interval_min, initial_cash, r.fees_paid, r.slippage_paid, r.turnover)92                val = m.get(metric)93                row.append(float(val) if val is not None and math.isfinite(val) else None)94            except ValueError:95                row.append(None)96        z.append(row)97    return {"x": {"name": x[0], "values": xs}, "y": {"name": y[0], "values": ys}, "metric": metric, "z": z}98