ParallelLLC/algorithmic_trading

1"""Walk-forward analysis.2 3Re-tune on a training window, trade the next window blind, roll forward. The4gap between in-sample and out-of-sample Sharpe is the honest estimate of how5much of the backtest was curve-fitting.6"""7 8from __future__ import annotations9 10from typing import Callable, Dict, List, Optional11 12import numpy as np13import pandas as pd14 15from ..engine import run_backtest16from ..strategies import Strategy17from ..types import CostModel18 19__all__ = ["walk_forward", "walk_forward_panel"]20 21 22def walk_forward(23    df: pd.DataFrame,24    strategy: Strategy,25    n_folds: int = 5,26    train_ratio: float = 0.7,27    costs: Optional[CostModel] = None,28    lag: int = 1,29    max_leverage: float = 1.0,30    allow_short: bool = True,31    grid_limit: int = 40,32    progress: Optional[Callable[[float, str], None]] = None,33) -> Dict[str, object]:34    """Roll a train/test split forward ``n_folds`` times.35 36    Each fold picks the best parameters by in-sample Sharpe and reports what37    those parameters then did out of sample. The stitched OOS returns are the38    closest thing to a paper-trading record this repo can produce offline.39    """40    costs = costs or CostModel()41    grid = strategy.grid(limit=grid_limit)42    n = len(df)43 44    if n < 250 or n_folds < 2:45        return {"folds": [], "note": "Not enough history for walk-forward analysis."}46 47    # Each fold is a contiguous train+test block; blocks advance by test length.48    block = int(n / (1 + (n_folds - 1) * (1 - train_ratio)))49    block = min(block, n)50    train_len = int(block * train_ratio)51    test_len = block - train_len52    if train_len < 100 or test_len < 20:53        return {"folds": [], "note": "Not enough history for walk-forward analysis."}54 55    folds: List[Dict[str, object]] = []56    oos_returns: List[pd.Series] = []57 58    for k in range(n_folds):59        start = k * test_len60        train = df.iloc[start : start + train_len]61        test = df.iloc[start + train_len : start + train_len + test_len]62        if len(test) < 20:63            break64 65        best_params, best_sharpe = None, -np.inf66        for params in grid:67            target = strategy.generate(train, params)68            sr = run_backtest(69                train, target, costs=costs, lag=lag,70                max_leverage=max_leverage, allow_short=allow_short,71            ).sharpe72            if sr > best_sharpe:73                best_params, best_sharpe = params, sr74 75        # Generate signals on train+test so indicators are warm at the fold76        # boundary, then evaluate only the test slice.77        combined = df.iloc[start : start + train_len + len(test)]78        target = strategy.generate(combined, best_params).loc[test.index]79        oos = run_backtest(80            test, target, costs=costs, lag=lag,81            max_leverage=max_leverage, allow_short=allow_short,82        )83 84        folds.append(85            {86                "fold": k + 1,87                "train_start": str(train.index[0].date()),88                "train_end": str(train.index[-1].date()),89                "test_start": str(test.index[0].date()),90                "test_end": str(test.index[-1].date()),91                "params": best_params,92                "is_sharpe": float(best_sharpe),93                "oos_sharpe": float(oos.sharpe),94                "oos_return": float(oos.metrics.get("total_return", 0.0)),95                "oos_max_dd": float(oos.metrics.get("max_drawdown", 0.0)),96            }97        )98        oos_returns.append(oos.returns)99 100        if progress is not None:101            progress((k + 1) / n_folds, f"Walk-forward fold {k + 1}/{n_folds}")102 103    if not folds:104        return {"folds": [], "note": "Not enough history for walk-forward analysis."}105 106    is_sharpes = np.array([f["is_sharpe"] for f in folds], dtype=float)107    oos_sharpes = np.array([f["oos_sharpe"] for f in folds], dtype=float)108    stitched = pd.concat(oos_returns) if oos_returns else pd.Series(dtype=float)109    stitched = stitched[~stitched.index.duplicated(keep="first")].sort_index()110 111    mean_is = float(np.mean(is_sharpes))112    mean_oos = float(np.mean(oos_sharpes))113 114    return {115        "folds": folds,116        "mean_is_sharpe": mean_is,117        "mean_oos_sharpe": mean_oos,118        # 1.0 = the edge fully survived; 0.0 = it evaporated out of sample.119        "efficiency": float(mean_oos / mean_is) if mean_is > 1e-9 else 0.0,120        "oos_win_rate": float(np.mean(oos_sharpes > 0)),121        # 1.0 means every fold chose different parameters -- a tuning process122        # that cannot make up its mind is fitting noise.123        "param_instability": float(124            len({str(f["params"]) for f in folds}) / max(len(folds), 1)125        ),126        "oos_returns": stitched,127        "oos_equity": (1.0 + stitched).cumprod() if len(stitched) else stitched,128        "note": "",129    }130 131 132def walk_forward_panel(133    panel,134    strategy,135    n_folds: int = 4,136    train_ratio: float = 0.7,137    costs: Optional[CostModel] = None,138    lag: int = 1,139    gross_leverage: float = 1.0,140    allow_short: bool = True,141    rebalance: str = "M",142    grid_limit: int = 16,143    progress: Optional[Callable[[float, str], None]] = None,144) -> Dict[str, object]:145    """Walk-forward for cross-sectional strategies over a :class:`Panel`.146 147    Same contract as :func:`walk_forward`: tune on the training window, trade148    the next window blind, roll on. Signals are generated over train+test149    together so the indicators are warm at the fold boundary, then evaluated150    only on the test slice -- the panel equivalent of the single-asset path.151    """152    from ..portfolio import rebalance_schedule, run_portfolio_backtest153 154    costs = costs or CostModel()155    grid = strategy.grid(limit=grid_limit)156    n = len(panel)157 158    if n < 250 or n_folds < 2:159        return {"folds": [], "note": "Not enough history for walk-forward analysis."}160 161    block = min(int(n / (1 + (n_folds - 1) * (1 - train_ratio))), n)162    train_len = int(block * train_ratio)163    test_len = block - train_len164    if train_len < 100 or test_len < 20:165        return {"folds": [], "note": "Not enough history for walk-forward analysis."}166 167    folds: List[Dict[str, object]] = []168    oos_returns: List[pd.Series] = []169 170    for k in range(n_folds):171        start = k * test_len172        train = panel.slice(panel.index[start], panel.index[min(start + train_len - 1, n - 1)])173        test_start = start + train_len174        if test_start + 20 > n:175            break176        test_end = min(test_start + test_len, n) - 1177        test = panel.slice(panel.index[test_start], panel.index[test_end])178        if len(test) < 20:179            break180 181        best_params, best_sharpe = None, -np.inf182        for params in grid:183            weights = strategy.generate(train, params)184            sharpe = run_portfolio_backtest(185                train, weights, costs=costs, lag=lag, gross_leverage=gross_leverage,186                allow_short=allow_short, rebalance_on=rebalance_schedule(train.index, rebalance),187            ).sharpe188            if sharpe > best_sharpe:189                best_params, best_sharpe = params, sharpe190 191        combined = panel.slice(panel.index[start], panel.index[test_end])192        weights = strategy.generate(combined, best_params).loc[test.index]193        oos = run_portfolio_backtest(194            test, weights, costs=costs, lag=lag, gross_leverage=gross_leverage,195            allow_short=allow_short, rebalance_on=rebalance_schedule(test.index, rebalance),196        )197 198        folds.append({199            "fold": k + 1,200            "train_start": str(train.index[0].date()),201            "train_end": str(train.index[-1].date()),202            "test_start": str(test.index[0].date()),203            "test_end": str(test.index[-1].date()),204            "params": best_params,205            "is_sharpe": float(best_sharpe),206            "oos_sharpe": float(oos.sharpe),207            "oos_return": float(oos.metrics.get("total_return", 0.0)),208            "oos_max_dd": float(oos.metrics.get("max_drawdown", 0.0)),209        })210        oos_returns.append(oos.returns)211        if progress is not None:212            progress((k + 1) / n_folds, f"Walk-forward fold {k + 1}/{n_folds}")213 214    if not folds:215        return {"folds": [], "note": "Not enough history for walk-forward analysis."}216 217    is_sharpes = np.array([f["is_sharpe"] for f in folds], dtype=float)218    oos_sharpes = np.array([f["oos_sharpe"] for f in folds], dtype=float)219    stitched = pd.concat(oos_returns) if oos_returns else pd.Series(dtype=float)220    stitched = stitched[~stitched.index.duplicated(keep="first")].sort_index()221    mean_is, mean_oos = float(np.mean(is_sharpes)), float(np.mean(oos_sharpes))222 223    return {224        "folds": folds,225        "mean_is_sharpe": mean_is,226        "mean_oos_sharpe": mean_oos,227        "efficiency": float(mean_oos / mean_is) if mean_is > 1e-9 else 0.0,228        "oos_win_rate": float(np.mean(oos_sharpes > 0)),229        "param_instability": float(len({str(f["params"]) for f in folds}) / max(len(folds), 1)),230        "oos_returns": stitched,231        "oos_equity": (1.0 + stitched).cumprod() if len(stitched) else stitched,232        "note": "",233    }234