File size: 4,327 Bytes
ab83ea6 caea0d1 ab83ea6 caea0d1 ab83ea6 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 | """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: # pragma: no cover
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) # positional per API
# deflated_sharpe_ratio(returns, n_trials, var_sharpe, *, bars_per_year)
# var_sharpe is the variance of per-trial per-bar Sharpes; DSR only
# applies when more than one strategy configuration was tried.
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:
# min_track_record_length(observed_sharpe, target_sharpe, alpha,
# skew, kurtosis) - all positional scalars
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]
# CSCV needs a rectangular matrix; configs over shorter histories are
# aligned to the MOST RECENT n_obs so the CSCV blocks cover the
# tradable present, not the discarded early era.
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 # pandas is excess kurtosis
|