| """Backtest-overfitting statistics: PSR, DSR, PBO, MinBTL. |
| |
| Grounded in the `purgedcv` package (MIT) and López de Prado (2018), |
| ch. 11-12. Accept a strategy only if DSR > 0.95 and PBO <= 0.5. |
| """ |
|
|
| from __future__ import annotations |
|
|
| import numpy as np |
|
|
| try: |
| from purgedcv import ( |
| deflated_sharpe_ratio, |
| probabilistic_sharpe_ratio, |
| probability_of_backtest_overfitting, |
| min_track_record_length, |
| effective_n_trials, |
| ) |
| except ImportError as e: |
| raise ImportError( |
| "purgedcv is required: pip install purgedcv==0.1.11" |
| ) from e |
|
|
| __all__ = ["eval_backtest", "eval_trial_set"] |
|
|
|
|
| def eval_backtest( |
| returns: np.ndarray, |
| n_trials: int, |
| trial_sr_variance: float, |
| benchmark_sr: float = 0.0, |
| periods_per_year: int = 24 * 365, |
| ) -> dict: |
| """Evaluate one OOS return series. |
| |
| returns: per-bar strategy returns (already net of fees/funding). |
| n_trials: number of independent strategy configurations tried. |
| trial_sr_variance: variance of per-trial per-bar Sharpe ratios. |
| """ |
| rets = np.asarray(returns, dtype=float) |
| rets = rets[~np.isnan(rets)] |
| if len(rets) < 30: |
| raise ValueError("need >= 30 return observations") |
| sr_bar = rets.mean() / (rets.std(ddof=1) + 1e-12) |
| sr_annualized = sr_bar * np.sqrt(periods_per_year) |
| sr_benchmark = benchmark_sr / np.sqrt(periods_per_year) |
|
|
| skew = float(pd_skew(rets)) |
| kurt = float(pd_kurt(rets)) |
|
|
| psr = probabilistic_sharpe_ratio(rets, sr_benchmark) |
|
|
| |
| |
| |
| if n_trials > 1: |
| dsr = deflated_sharpe_ratio(rets, n_trials, trial_sr_variance) |
| else: |
| dsr = psr |
|
|
| out = { |
| "n_obs": len(rets), |
| "sharpe_per_bar": sr_bar, |
| "sharpe_annualized": sr_annualized, |
| "skewness": skew, |
| "kurtosis": kurt, |
| "max_drawdown": max_drawdown(rets), |
| "deflated_sharpe": float(dsr) if dsr is not None else None, |
| } |
| try: |
| |
| |
| out["min_track_record_length"] = float( |
| min_track_record_length(sr_annualized, benchmark_sr, 0.95, skew, kurt) |
| ) |
| except Exception: |
| out["min_track_record_length"] = None |
| return out |
|
|
|
|
| def eval_trial_set( |
| trial_returns: list[np.ndarray], |
| ) -> dict: |
| """PBO over a set of trial return series (CSCV). |
| |
| trial_returns: list of per-bar return series, one per strategy config. |
| probability_of_backtest_overfitting takes an (n_configs, n_obs) ndarray |
| and returns a PBOResult dataclass with .pbo / .logits / .slope. |
| n_splits is lowered to 8 (still 70 combos) because 16 splits with |
| purge/embargo is prohibitively slow on CPU. |
| """ |
| series = [np.asarray(r, dtype=float) for r in trial_returns] |
| |
| |
| |
| n_obs = min(len(r) for r in series) |
| series = [r[-n_obs:] for r in series] |
| arr = np.vstack(series) |
| if not np.isfinite(arr).all(): |
| raise ValueError("trial returns contain NaN/inf") |
| result = probability_of_backtest_overfitting(arr, n_splits=8) |
| pbo = float(getattr(result, "pbo", result)) |
| per_trial_sr = np.array([ |
| np.nanmean(r) / (np.nanstd(r, ddof=1) + 1e-12) for r in arr |
| ]) |
| n_eff = effective_n_trials(per_trial_sr) |
| return { |
| "probability_of_backtest_overfitting": pbo, |
| "trial_sr_variance": float(np.var(per_trial_sr)), |
| "effective_n_trials": int(n_eff), |
| } |
|
|
|
|
| def max_drawdown(returns: np.ndarray) -> float: |
| eq = np.cumprod(1.0 + np.asarray(returns)) |
| peak = np.maximum.accumulate(eq) |
| return float(np.min(eq / peak - 1.0)) |
|
|
|
|
| def pd_skew(x: np.ndarray) -> float: |
| import pandas as pd |
| return pd.Series(x).skew() |
|
|
|
|
| def pd_kurt(x: np.ndarray) -> float: |
| import pandas as pd |
| return pd.Series(x).kurt() + 3.0 |
|
|
