Repository Public
1"""Robustness analyses: parameter sensitivity, walk-forward, bootstrap, Monte Carlo, stress windows, look-ahead check.23Everything that compares configurations takes an `Experiment` and therefore only sees train + validation data.4"""5from __future__ import annotations67import math89import numpy as np10import pandas as pd1112from .data.core import bars_per_year, interval_seconds13from .engine import BacktestConfig, BacktestResult, run_backtest14from .metrics import _num, max_drawdown, sharpe_ratio15from .research import Experiment, TestSplitError, params_key16from .strategy import Strategy171819# ------------------------------------------------------------------------------------- sensitivity / selection2021def parameter_sensitivity(exp: Experiment, strategy: Strategy, grid: list[dict]) -> list[dict]:22 """Run every configuration of `grid` on the development range; report train and validation figures.2324 A configuration that makes the strategy fail is reported with its error instead of stopping the analysis.25 """26 N = bars_per_year(exp.config.interval)27 rows = []28 for p in grid:29 row: dict = {"params": p}30 try:31 res = exp.trial(strategy, p)32 except TestSplitError:33 raise34 except Exception as e: # noqa: BLE001 - a bad corner of the grid is a finding, not a crash35 row["error"] = f"{type(e).__name__}: {e}"36 rows.append(row)37 continue38 for split in ("train", "validation"):39 r = exp.split_returns(res, split)40 row[f"{split}_sharpe"] = _num(sharpe_ratio(r, N))41 row[f"{split}_return"] = _num(np.prod(1.0 + r) - 1.0)42 start, end = exp.splits.window("validation")43 t = res.trades44 row["validation_trades"] = int(((t["ts"] >= start) & (t["ts"] < end)).sum()) if len(t) else 045 rows.append(row)46 return rows474849def select_params(rows: list[dict], metric: str = "validation_sharpe") -> dict | None:50 """Best configuration of a sensitivity table according to a VALIDATION (or train) metric."""51 if not metric.startswith(("validation_", "train_")):52 raise ValueError("selection metrics are train_* or validation_* only")53 ok = [r for r in rows if r.get(metric) is not None]54 return max(ok, key=lambda r: r[metric])["params"] if ok else None555657def walk_forward(exp: Experiment, strategy: Strategy, grid: list[dict], n_folds: int = 4) -> list[dict]:58 """Rolling walk-forward inside the development range.5960 The range is cut into n_folds + 2 equal chunks. Fold f selects, among `grid`, the configuration with the best61 Sharpe ratio on chunks [f, f+2) (in-sample), then that configuration is backtested from cash on chunk f+262 (out-of-sample, with earlier bars as lookback history only). In-sample figures are read from the63 development-range run of each configuration (one run per configuration, shared with the sensitivity table).64 """65 index = next(iter(exp.dev_data().values())).index66 c = len(index) // (n_folds + 2)67 if c < 10:68 return []69 N = bars_per_year(exp.config.interval)70 runs = []71 for p in grid:72 try:73 runs.append((p, exp.trial(strategy, p).returns))74 except TestSplitError:75 raise76 except Exception: # noqa: BLE001 - already reported by parameter_sensitivity77 continue78 folds = []79 for f in range(n_folds):80 is_a, is_b, oos_b = index[f * c], index[(f + 2) * c], index[min((f + 3) * c, len(index) - 1)]81 best, best_sr = None, -math.inf82 for p, r in runs:83 sr = sharpe_ratio(r[(r.index >= is_a) & (r.index < is_b)].to_numpy(), N)84 if math.isfinite(sr) and sr > best_sr:85 best, best_sr = p, sr86 row = {"in_sample": [is_a.isoformat(), is_b.isoformat()], "out_of_sample": [is_b.isoformat(), oos_b.isoformat()],87 "params": best, "in_sample_sharpe": _num(best_sr) if best is not None else None}88 if best is not None:89 res = exp.run_window(strategy, is_b, oos_b, best)90 r = res.returns.to_numpy()91 row["out_of_sample_sharpe"] = _num(sharpe_ratio(r, N))92 row["out_of_sample_return"] = _num(res.equity.iloc[-1] / res.equity.iloc[0] - 1.0)93 row["out_of_sample_max_drawdown"] = _num(max_drawdown(res.equity.to_numpy()))94 row["out_of_sample_trades"] = int(len(res.trades))95 folds.append(row)96 return folds979899# -------------------------------------------------------------------------------------------- resampling100101def _stationary_indices(n: int, n_samples: int, mean_block: float, rng: np.random.Generator) -> np.ndarray:102 """Politis-Romano stationary bootstrap: blocks of geometric length (mean `mean_block`), wrapping around."""103 starts = rng.integers(0, n, size=(n_samples, n))104 new_block = rng.random((n_samples, n)) < 1.0 / mean_block105 new_block[:, 0] = True106 pos = np.arange(n)[None, :]107 last_start = np.maximum.accumulate(np.where(new_block, pos, 0), axis=1)108 origin = np.take_along_axis(starts, last_start, axis=1)109 return (origin + pos - last_start) % n110111112def default_block(n: int, interval: str) -> int:113 """Mean block length: one day of bars, at least n^(1/3), at most n/10."""114 per_day = max(1, 86400 // interval_seconds(interval))115 return int(max(1, min(max(per_day, round(n ** (1 / 3))), max(1, n // 10))))116117118def bootstrap_sharpe_ci(returns, interval: str, *, n_samples: int = 2000, mean_block: int | None = None,119 level: float = 0.95, seed: int = 0) -> dict:120 """Stationary block bootstrap percentile confidence interval of the annualised Sharpe ratio."""121 r = np.asarray(returns, dtype="float64")122 n = len(r)123 if n < 20 or not r.std() > 0:124 return {"ci": [None, None], "level": level, "n_samples": 0, "mean_block": None}125 block = mean_block or default_block(n, interval)126 rng = np.random.default_rng(seed)127 sharpes = []128 chunk = max(1, min(n_samples, 4_000_000 // n))129 for lo in range(0, n_samples, chunk):130 x = r[_stationary_indices(n, min(chunk, n_samples - lo), block, rng)]131 sd = x.std(axis=1, ddof=1)132 with np.errstate(divide="ignore", invalid="ignore"):133 sharpes.append(np.where(sd > 0, x.mean(axis=1) / sd, np.nan))134 s = np.concatenate(sharpes) * math.sqrt(bars_per_year(interval))135 alpha = (1.0 - level) / 2.0136 lo_, hi_ = np.nanpercentile(s, [100 * alpha, 100 * (1 - alpha)])137 return {"ci": [_num(lo_), _num(hi_)], "level": level, "n_samples": int(n_samples), "mean_block": int(block),138 "method": "stationary bootstrap (Politis & Romano 1994), percentile interval"}139140141def monte_carlo(returns, lot_pnl, initial_equity: float, *, n_samples: int = 2000, seed: int = 0) -> dict:142 """Two resampling experiments on one backtest.143144 bar_returns: bars drawn with replacement (i.i.d., which destroys serial dependence) and compounded: spread of145 the total return and of the maximum drawdown that the same set of bar returns can produce in another order.146 lots: closed-lot P&Ls drawn with replacement and summed, relative to the initial equity: how much the result147 depends on a few trades.148 """149 rng = np.random.default_rng(seed)150 r = np.asarray(returns, dtype="float64")151 out: dict = {"n_samples": int(n_samples), "bar_returns": None, "lots": None}152 if len(r) >= 20:153 tot, mdd = [], []154 chunk = max(1, min(n_samples, 4_000_000 // len(r)))155 for lo in range(0, n_samples, chunk):156 x = r[rng.integers(0, len(r), size=(min(chunk, n_samples - lo), len(r)))]157 eq = np.cumprod(1.0 + x, axis=1)158 tot.append(eq[:, -1] - 1.0)159 peak = np.maximum(np.maximum.accumulate(eq, axis=1), 1.0)160 mdd.append((1.0 - eq / peak).max(axis=1))161 tot, mdd = np.concatenate(tot), np.concatenate(mdd)162 out["bar_returns"] = {163 "total_return_p05": _num(np.percentile(tot, 5)), "total_return_p50": _num(np.percentile(tot, 50)),164 "total_return_p95": _num(np.percentile(tot, 95)), "probability_of_loss": _num((tot < 0).mean()),165 "max_drawdown_p50": _num(np.percentile(mdd, 50)), "max_drawdown_p95": _num(np.percentile(mdd, 95)),166 }167 pnl = np.asarray(lot_pnl, dtype="float64")168 if len(pnl) >= 5:169 s = pnl[rng.integers(0, len(pnl), size=(n_samples, len(pnl)))].sum(axis=1) / initial_equity170 out["lots"] = {171 "n_lots": int(len(pnl)), "pnl_fraction_p05": _num(np.percentile(s, 5)),172 "pnl_fraction_p50": _num(np.percentile(s, 50)), "pnl_fraction_p95": _num(np.percentile(s, 95)),173 "probability_of_loss": _num((s < 0).mean()),174 }175 return out176177178# ------------------------------------------------------------------------------------------------ stress179180def drawdown_episodes(equity: pd.Series, top: int = 3) -> list[dict]:181 """Largest non-overlapping peak-to-trough declines.182183 The maximum drawdown of the series is taken first; the search then continues separately in the bars before184 its peak and in the bars after its trough, and so on, until `top` windows are found.185 """186 eq = equity.to_numpy(dtype="float64")187188 def deepest(a: int, b: int):189 if b - a < 2:190 return None191 seg = eq[a:b]192 dd = 1.0 - seg / np.maximum.accumulate(seg)193 t = int(np.argmax(dd))194 if not dd[t] > 0:195 return None196 return float(dd[t]), a + int(np.argmax(seg[:t + 1])), a + t, a, b197198 candidates = [c for c in [deepest(0, len(eq))] if c]199 episodes = []200 while candidates and len(episodes) < top:201 candidates.sort()202 depth, p, t, a, b = candidates.pop()203 episodes.append({"peak": equity.index[p], "trough": equity.index[t], "depth": depth})204 candidates += [c for c in (deepest(a, p), deepest(t + 1, b)) if c]205 return episodes206207208def stress_windows(strategy_equity: pd.Series, benchmark_equity: pd.Series, top: int = 3) -> list[dict]:209 """Strategy behaviour during the benchmark's largest drawdowns (same split, same bars)."""210 out = []211 for e in drawdown_episodes(benchmark_equity, top):212 s = strategy_equity.loc[e["peak"]:e["trough"]]213 out.append({214 "window": [e["peak"].isoformat(), e["trough"].isoformat()],215 "benchmark_return": _num(-e["depth"]),216 "strategy_return": _num(s.iloc[-1] / s.iloc[0] - 1.0),217 "strategy_max_drawdown": _num(max_drawdown(s.to_numpy())),218 })219 return out220221222# ---------------------------------------------------------------------------------------------- look-ahead223224def lookahead_check(strategy: Strategy, data: dict[str, pd.DataFrame], config: BacktestConfig, *,225 params: dict | None = None, start=None, end=None, cuts=(0.5, 0.8),226 max_bars: int = 1500, runner=run_backtest) -> dict:227 """Truncation test: decisions and equity up to bar t must not change when every bar after t is removed.228229 A base run is made on (at most) the last `max_bars` bars of [start, end). For each cut, the data are truncated230 after the cut bar and the run is repeated; the target weights decided up to the cut (the decision taken ON the231 cut bar included) and the equity up to the cut must be identical. This fails if the engine hands future bars232 to the strategy or if fills use information beyond the fill bar. It cannot see a strategy that obtains future233 data from outside its inputs (a file, the network): that is what the purity guard and the static check are for.234 `runner` is the backtest function (replaced by a deliberately leaky one in the tests of this check).235 """236 index = next(iter(data.values())).index237 a = 0 if start is None else int(index.searchsorted(pd.Timestamp(start), side="left"))238 b = len(index) if end is None else int(index.searchsorted(pd.Timestamp(end), side="left"))239 a = max(a, b - max_bars)240 if b - a < 10:241 return {"status": "fail", "detail": "window too short for the look-ahead check"}242 try:243 base = runner(strategy, data, config, start=index[a], end=None if b >= len(index) else index[b],244 params=params)245 for c in cuts:246 cut = a + max(3, int((b - a) * c))247 trunc = {s: df.iloc[:cut + 1] for s, df in data.items()}248 part = runner(strategy, trunc, config, start=index[a], params=params)249 eq_full = base.equity.loc[part.equity.index].to_numpy()250 if not np.allclose(eq_full, part.equity.to_numpy(), rtol=1e-10, atol=1e-10):251 return {"status": "fail", "detail": f"equity before {index[cut]} changes when later bars are removed"}252 d_full = base.decisions.loc[base.decisions.index <= index[cut]]253 if not (d_full.index.equals(part.decisions.index)254 and np.allclose(d_full.to_numpy(), part.decisions.to_numpy(), rtol=0, atol=1e-12)):255 return {"status": "fail",256 "detail": f"decisions before {index[cut]} change when later bars are removed"}257 except Exception as e: # noqa: BLE001258 return {"status": "fail", "detail": f"{type(e).__name__}: {e}"}259 return {"status": "pass", "detail": f"truncation test on {b - a} bars, cuts at {list(cuts)}, "260 f"{len(base.decisions)} decisions compared"}261262263__all__ = [264 "parameter_sensitivity", "select_params", "walk_forward", "bootstrap_sharpe_ci", "monte_carlo",265 "drawdown_episodes", "stress_windows", "lookahead_check", "default_block", "params_key", "BacktestResult",266]267