DSR/PBO/PSR/MinBTL statistics

crypto_eval/stats.py

@@ -0,0 +1,126 @@
+"""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]
+    # CPCV test slices differ by a few bars from purge/embargo; CSCV needs
+    # a rectangular (n_configs, n_obs) matrix, so truncate to the shortest.
+    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