ParallelLLC/algorithmic_trading

1"""The Lab: one call that runs a backtest and then tries to disprove it.2 3This is the module both the Gradio Space and the CLI drive. Keeping the whole4pipeline here means the app and the command line can never disagree about what5a Reality Score means.6"""7 8from __future__ import annotations9 10import logging11from dataclasses import dataclass, field12from typing import Callable, Dict, List, Optional13 14import numpy as np15import pandas as pd16 17from .data import load_ohlcv18from .engine import run_backtest19from .metrics import infer_periods_per_year20from .strategies import Strategy, get_strategy, list_strategies21from .types import BacktestResult, CostModel, MarketData22from .validation.deflated_sharpe import deflated_sharpe_ratio, min_track_record_length23from .validation.pbo import probability_of_backtest_overfitting24from .validation.permutation import PermutationResult, permutation_test25from .validation.walkforward import walk_forward26from .verdict import reality_score27 28logger = logging.getLogger(__name__)29 30__all__ = ["LabConfig", "LabReport", "run_lab", "run_arena"]31 32ProgressFn = Optional[Callable[[float, str], None]]33 34 35@dataclass36class LabConfig:37    symbol: str = "SPY"38    start: str = "2015-01-01"39    end: Optional[str] = None40    interval: str = "1d"41    source: str = "yahoo"42 43    strategy: str = "sma_cross"44    params: Dict[str, float] = field(default_factory=dict)45 46    commission_bps: float = 1.047    slippage_bps: float = 2.048    short_borrow_bps: float = 50.049    lag: int = 150    allow_short: bool = True51    max_leverage: float = 1.052    capital: float = 100_000.053 54    n_permutations: int = 25055    permutation_method: str = "permute"56    block_size: int = 2057    wf_folds: int = 558    pbo_splits: int = 859    grid_limit: int = 4060    seed: int = 061 62    def costs(self, multiplier: float = 1.0) -> CostModel:63        return CostModel(64            commission_bps=self.commission_bps * multiplier,65            slippage_bps=self.slippage_bps * multiplier,66            short_borrow_bps=self.short_borrow_bps * multiplier,67        )68 69 70@dataclass71class LabReport:72    config: LabConfig73    market: MarketData74    strategy: Strategy75    params: Dict[str, float]76    backtest: BacktestResult77    permutation: Optional[PermutationResult] = None78    dsr: Dict[str, float] = field(default_factory=dict)79    pbo: Dict[str, object] = field(default_factory=dict)80    walkforward: Dict[str, object] = field(default_factory=dict)81    trials: Dict[str, object] = field(default_factory=dict)82    verdict: Dict[str, object] = field(default_factory=dict)83    cost_stress: Dict[str, float] = field(default_factory=dict)84    benchmark_correlation: float = float("nan")85 86 87def _trial_matrix(88    df: pd.DataFrame,89    strategy: Strategy,90    cfg: LabConfig,91    progress: ProgressFn = None,92) -> tuple[np.ndarray, List[float], List[str]]:93    """Backtest every parameter combination a researcher would plausibly try.94 95    The resulting ``T x N`` return matrix feeds both the Deflated Sharpe (how96    many variants were tried, and how spread out were they) and PBO.97    """98    grid = strategy.grid(limit=cfg.grid_limit)99    costs = cfg.costs()100    columns, sharpes, labels = [], [], []101 102    for i, params in enumerate(grid):103        target = strategy.generate(df, params)104        result = run_backtest(105            df, target, costs=costs, lag=cfg.lag,106            max_leverage=cfg.max_leverage, allow_short=cfg.allow_short,107        )108        columns.append(result.returns.to_numpy(dtype=float))109        sharpes.append(result.sharpe)110        labels.append(", ".join(f"{k}={v}" for k, v in params.items()) or "default")111        if progress is not None and i % 5 == 0:112            progress((i + 1) / max(len(grid), 1), f"Variant {i + 1}/{len(grid)}")113 114    matrix = np.column_stack(columns) if columns else np.zeros((len(df), 0))115    return matrix, sharpes, labels116 117 118def run_lab(cfg: LabConfig, progress: ProgressFn = None) -> LabReport:119    """Run the full honesty pipeline for one strategy on one symbol."""120 121    def step(fraction: float, message: str) -> None:122        if progress is not None:123            progress(min(max(fraction, 0.0), 1.0), message)124 125    step(0.02, "Loading market data")126    market = load_ohlcv(cfg.symbol, cfg.start, cfg.end, cfg.interval, cfg.source)127    df = market.df128    if len(df) < 120:129        raise ValueError(130            f"Only {len(df)} bars available for {cfg.symbol}. "131            "Widen the date range — anything shorter cannot be validated."132        )133 134    strategy = get_strategy(cfg.strategy)135    params = strategy.clean(cfg.params)136    ppy = infer_periods_per_year(df.index)137 138    step(0.10, "Running the backtest")139    target = strategy.generate(df, params)140    backtest = run_backtest(141        df,142        target,143        costs=cfg.costs(),144        lag=cfg.lag,145        max_leverage=cfg.max_leverage,146        allow_short=cfg.allow_short,147        initial_capital=cfg.capital,148        periods_per_year=ppy,149        meta={"symbol": market.symbol, "strategy": strategy.key, "params": params},150    )151 152    step(0.16, "Stress-testing costs")153    stressed = run_backtest(154        df, target, costs=cfg.costs(3.0), lag=cfg.lag,155        max_leverage=cfg.max_leverage, allow_short=cfg.allow_short,156        periods_per_year=ppy,157    )158    base_sharpe = backtest.sharpe159    cost_stress_ratio = float(stressed.sharpe / base_sharpe) if base_sharpe > 1e-9 else 0.0160    cost_stress = {161        "sharpe_1x": base_sharpe,162        "sharpe_3x": stressed.sharpe,163        "ratio": cost_stress_ratio,164        "return_3x": float(stressed.metrics.get("total_return", 0.0)),165    }166 167    step(0.22, "Backtesting every parameter variant")168    matrix, trial_sharpes, labels = _trial_matrix(169        df, strategy, cfg, lambda f, m: step(0.22 + 0.18 * f, m)170    )171    n_trials = max(len(trial_sharpes), 1)172 173    step(0.42, "Deflating the Sharpe ratio for selection bias")174    dsr = deflated_sharpe_ratio(175        backtest.returns.to_numpy(dtype=float),176        sharpe_annual=base_sharpe,177        periods_per_year=ppy,178        n_trials=n_trials,179        trial_sharpes=trial_sharpes if n_trials > 1 else None,180    )181    mtrl = min_track_record_length(182        dsr["sr_per_period"], dsr["n_obs"], dsr["skew"], dsr["kurtosis"],183        benchmark=dsr["threshold_sr_per_period"],184    )185    dsr["min_track_record_bars"] = mtrl186    dsr["min_track_record_years"] = float(mtrl / ppy) if np.isfinite(mtrl) else float("inf")187 188    step(0.46, "Measuring backtest overfitting")189    pbo = probability_of_backtest_overfitting(matrix, n_splits=cfg.pbo_splits, labels=labels)190 191    step(0.50, "Shuffling the market")192    permutation = None193    if cfg.n_permutations > 0:194        permutation = permutation_test(195            df,196            lambda frame: strategy.generate(frame, params),197            n_permutations=cfg.n_permutations,198            method=cfg.permutation_method,199            block=cfg.block_size,200            costs=cfg.costs(),201            lag=cfg.lag,202            max_leverage=cfg.max_leverage,203            allow_short=cfg.allow_short,204            seed=cfg.seed,205            observed=base_sharpe,206            progress=lambda f, m: step(0.50 + 0.32 * f, m),207        )208 209    step(0.84, "Walking the strategy forward")210    wf = walk_forward(211        df, strategy, n_folds=cfg.wf_folds, costs=cfg.costs(), lag=cfg.lag,212        max_leverage=cfg.max_leverage, allow_short=cfg.allow_short,213        grid_limit=min(cfg.grid_limit, 24),214        progress=lambda f, m: step(0.84 + 0.12 * f, m),215    )216 217    bench_corr = float(218        pd.Series(backtest.returns).corr(backtest.benchmark_equity.pct_change().fillna(0.0))219    )220 221    step(0.98, "Grading")222    verdict = reality_score(223        metrics=backtest.metrics,224        benchmark_metrics=backtest.benchmark_metrics,225        p_value=permutation.p_value if permutation else None,226        dsr=dsr.get("dsr"),227        pbo=pbo.get("pbo"),228        wf_efficiency=wf.get("efficiency"),229        wf_win_rate=wf.get("oos_win_rate"),230        cost_stress_ratio=cost_stress_ratio,231        benchmark_correlation=bench_corr,232    )233 234    step(1.0, "Done")235    return LabReport(236        config=cfg,237        market=market,238        strategy=strategy,239        params=params,240        backtest=backtest,241        permutation=permutation,242        dsr=dsr,243        pbo=pbo,244        walkforward=wf,245        trials={"n": n_trials, "sharpes": trial_sharpes, "labels": labels, "matrix_shape": matrix.shape},246        verdict=verdict,247        cost_stress=cost_stress,248        benchmark_correlation=bench_corr,249    )250 251 252def run_arena(253    cfg: LabConfig,254    strategy_keys: Optional[List[str]] = None,255    n_permutations: int = 120,256    progress: ProgressFn = None,257) -> tuple[pd.DataFrame, MarketData, Dict[str, BacktestResult]]:258    """Race every strategy on the same market, ranked by evidence not returns.259 260    Buy & hold and the coin flip stay in the field on purpose: a leaderboard261    without a control group is marketing, not measurement.262    """263    market = load_ohlcv(cfg.symbol, cfg.start, cfg.end, cfg.interval, cfg.source)264    df = market.df265    ppy = infer_periods_per_year(df.index)266    costs = cfg.costs()267 268    keys = strategy_keys or [s.key for s in list_strategies()]269    rows, curves = [], {}270 271    for i, key in enumerate(keys):272        strategy = get_strategy(key)273        params = strategy.defaults()274        target = strategy.generate(df, params)275        result = run_backtest(276            df, target, costs=costs, lag=cfg.lag, max_leverage=cfg.max_leverage,277            allow_short=cfg.allow_short, initial_capital=cfg.capital, periods_per_year=ppy,278        )279        curves[key] = result280 281        p_value = None282        if n_permutations > 0:283            p_value = permutation_test(284                df,285                lambda frame, s=strategy, p=params: s.generate(frame, p),286                n_permutations=n_permutations,287                method=cfg.permutation_method,288                block=cfg.block_size,289                costs=costs,290                lag=cfg.lag,291                max_leverage=cfg.max_leverage,292                allow_short=cfg.allow_short,293                seed=cfg.seed,294                observed=result.sharpe,295            ).p_value296 297        grid_size = len(strategy.grid(limit=cfg.grid_limit))298        dsr = deflated_sharpe_ratio(299            result.returns.to_numpy(dtype=float),300            sharpe_annual=result.sharpe,301            periods_per_year=ppy,302            n_trials=grid_size,303        )304 305        rows.append(306            {307                "Strategy": strategy.name,308                "key": key,309                "Family": strategy.family,310                "Return": result.metrics.get("total_return", 0.0),311                "CAGR": result.metrics.get("cagr", 0.0),312                "Sharpe": result.sharpe,313                "MaxDD": result.metrics.get("max_drawdown", 0.0),314                "Trades": int(result.metrics.get("n_trades", 0)),315                "p-value": p_value if p_value is not None else float("nan"),316                "DSR": dsr["dsr"],317            }318        )319        if progress is not None:320            progress((i + 1) / len(keys), f"{strategy.name} ({i + 1}/{len(keys)})")321 322    table = pd.DataFrame(rows)323    if not table.empty:324        # Rank by evidence: a high Sharpe with a p-value of 0.4 is not a win.325        table["Evidence"] = (1.0 - table["p-value"].fillna(0.5)) * table["DSR"]326        table = table.sort_values("Evidence", ascending=False).reset_index(drop=True)327        table.insert(0, "#", table.index + 1)328    return table, market, curves329