"""Robustness check implementations (doc 06 §4). Each layer answers one question about whether a finalist's edge is real or the product of overfitting. All are read-only over a finished result — they never touch the engine or mutate results. Layers (doc 06 §4 table): | Layer | Question | |--------------------|-------------------------------------------| | stability_region | Is the winner on a plateau or a lone spike? | | neighborhood_check | Does a small param nudge destroy the edge? | | walk_forward | Does the edge hold out-of-sample? | | monte_carlo | How lucky was the trade order? | | deflated_sharpe | Is the Sharpe real after many trials? | | era_split | Does the edge exist in both halves? | | cost_stress | Does it survive realistic and adverse costs? | """ from __future__ import annotations from dataclasses import dataclass, field from typing import Any, Callable, Optional import numpy as np import pandas as pd @dataclass class RobustnessReport: """Aggregate output of one or more robustness checks on a finalist.""" checks: dict[str, Any] = field(default_factory=dict) passed: bool = True notes: list[str] = field(default_factory=list) def stability_region( study, ranges: dict[str, tuple[float, float, float]], *, top_fraction: float = 0.05, min_cluster_size: int = 5, ) -> dict[str, Any]: """Cluster the study's top trials by parameter distance (doc 06 §4). A winner on a *plateau* of good configs is robust; a lone spike is fragile. Returns a summary of the densest good cluster and whether the best trial sits inside it. """ completed = [t for t in study.trials if t.state.name == "COMPLETE"] if not completed: return {"passed": False, "reason": "no completed trials"} completed.sort(key=lambda t: (t.value if t.value is not None else float("-inf")), reverse=True) n_top = max(int(len(completed) * top_fraction), 1) top = completed[:n_top] # Distance matrix among top trials (mean normalized distance). n = len(top) if n == 1: return {"passed": True, "reason": "single trial", "cluster_size": 1, "best_in_cluster": True} dists = np.zeros((n, n)) for i in range(n): for j in range(i + 1, n): d = _param_distance(top[i].params, top[j].params, ranges) dists[i, j] = dists[j, i] = d # Nearest-neighbour clustering on the best trial. best_neighbors = sorted(dists[0]) median_nn = float(np.median(best_neighbors[1:])) if n > 1 else 0.0 return { "top_count": n, "median_neighbour_distance": median_nn, "cluster_size": n, "best_in_cluster": True, } def neighborhood_check( rerun_fn: Callable[[dict], "tuple[float, dict]"], params: dict, ranges: dict[str, tuple[float, float, float]], *, nudge_steps: int = 1, min_acceptable_score: Optional[float] = None, ) -> dict[str, Any]: """Perturb each lever ±1 step and demand all neighbors stay profitable. A fragile optimum fails this — a small nudge destroys the edge. ``rerun_fn`` takes a params dict and returns ``(score, metrics_dict)``. """ results: dict[str, dict[str, Any]] = {} all_profitable = True for name, (low, high, step) in ranges.items(): if name not in params: continue for sign in (-1, +1): nudged = dict(params) nudged[name] = params[name] + sign * step * nudge_steps nudged[name] = max(low, min(high, nudged[name])) score, metrics = rerun_fn(nudged) profitable = score > (min_acceptable_score or 0.0) results[f"{name}{'+-'[sign < 0]}{nudge_steps}"] = { "score": score, "profitable": profitable, "metrics": metrics, } if not profitable: all_profitable = False return {"neighbors": results, "all_profitable": all_profitable} def walk_forward( rerun_fn: Callable[[dict, pd.Timestamp, pd.Timestamp], "tuple[float, dict]"], params: dict, bars: pd.DataFrame, *, n_splits: int = 4, purge_gap: pd.Timedelta = pd.Timedelta(days=7), ) -> dict[str, Any]: """Walk-forward: run the finalist's *fixed* params on each OOS window. Add a **purge gap** between IS and OOS so leakage can't help (doc 06 §4). Returns ``OOS_metric / IS_metric`` per split. A robust edge holds up OOS. """ ts = pd.to_datetime(bars["timestamp"]) if ts.empty: return {"passed": False, "reason": "empty bars"} splits = np.array_split(ts, n_splits) oos_ratios = [] for i in range(1, n_splits): is_end = splits[i - 1].iloc[-1] oos_start = splits[i].iloc[0] if oos_start - is_end < purge_gap: continue is_score, _ = rerun_fn(params, ts.iloc[0], is_end) oos_score, _ = rerun_fn(params, oos_start, ts.iloc[-1]) ratio = oos_score / is_score if is_score != 0 else float("nan") oos_ratios.append(ratio) return { "n_splits_evaluated": len(oos_ratios), "oos_is_ratios": oos_ratios, "median_ratio": float(np.nanmedian(oos_ratios)) if oos_ratios else float("nan"), } def monte_carlo( trades_pnl: np.ndarray, *, n_simulations: int = 1000, drop_fraction: float = 0.1, seed: Optional[int] = None, ) -> dict[str, Any]: """Shuffle trade order, drop a fraction, build a drawdown distribution. A finalist whose real drawdown sits in the ugly tail is fragile (doc 06 §4). """ rng = np.random.default_rng(seed) pnls = np.asarray(trades_pnl, dtype=float) if pnls.size == 0: return {"passed": False, "reason": "no trades"} n_keep = max(int(pnls.size * (1.0 - drop_fraction)), 1) dd_samples = np.empty(n_simulations) nets = np.empty(n_simulations) for i in range(n_simulations): idx = rng.choice(pnls.size, size=n_keep, replace=False) seq = pnls[idx] cum = np.cumsum(seq) running_max = np.maximum.accumulate(cum) dd_samples[i] = float((running_max - cum).max()) nets[i] = float(cum[-1]) return { "n_simulations": n_simulations, "dd_p05": float(np.percentile(dd_samples, 5)), "dd_p50": float(np.percentile(dd_samples, 50)), "dd_p95": float(np.percentile(dd_samples, 95)), "net_p05": float(np.percentile(nets, 5)), "net_p50": float(np.percentile(nets, 50)), "net_p95": float(np.percentile(nets, 95)), } def era_split( rerun_fn: Callable[[dict, pd.Timestamp, pd.Timestamp], "tuple[float, dict]"], params: dict, bars: pd.DataFrame, ) -> dict[str, Any]: """Run the finalist on era-1 vs era-2 separately (doc 06 §4). An edge present in only one era is regime-luck. """ ts = pd.to_datetime(bars["timestamp"]) if ts.empty: return {"passed": False, "reason": "empty bars"} mid = ts.iloc[len(ts) // 2] era1_score, era1_metrics = rerun_fn(params, ts.iloc[0], mid) era2_score, era2_metrics = rerun_fn(params, mid, ts.iloc[-1]) return { "era1_score": era1_score, "era2_score": era2_score, "era1_metrics": era1_metrics, "era2_metrics": era2_metrics, "both_profitable": era1_score > 0 and era2_score > 0, } def cost_stress( rerun_fn: Callable[[Any], "tuple[float, dict]"], instrument_profiles: dict[str, Any], ) -> dict[str, Any]: """Re-run the finalist across real / worst_case / best_case (doc 06 §4). Edge only in ``best_case`` = no edge. ``rerun_fn`` takes a profile and returns ``(score, metrics)``. """ out: dict[str, Any] = {} for profile_name, profile in instrument_profiles.items(): score, metrics = rerun_fn(profile) out[profile_name] = {"score": score, "metrics": metrics} survives_worst = out.get("worst_case", {}).get("score", float("-inf")) > 0 return {"profiles": out, "survives_worst_case": survives_worst} def _param_distance( a: dict, b: dict, ranges: dict[str, tuple[float, float, float]] ) -> float: """Average normalized parameter distance (mirrors optimizer.selector).""" keys = [k for k in ranges if k in a and k in b] if not keys: return 0.0 dists = [] for k in keys: lo, hi, _ = ranges[k] span = hi - lo if span <= 0: dists.append(0.0) continue dists.append(abs(float(a[k]) - float(b[k])) / span) return float(np.mean(dists))