"""CLI for comprehensive overfitting analysis. Usage: fithia2 overfit-check --config EXPERIMENT_ID_OR_NAME [--initial-equity 10000] fithia2 overfit-check --config 415 --skip mc,permutation fithia2 overfit-check --config 415 --quick """ from __future__ import annotations import argparse import copy import json import sys import time from pathlib import Path import numpy as np from rich import box from rich.console import Console from rich.panel import Panel from rich.progress import Progress, SpinnerColumn, TextColumn, BarColumn, TimeElapsedColumn from rich.table import Table _console = Console(width=120) _CONFIGS_DIR = Path("configs/experiments") _RUNS_DIR = Path("runs") def _resolve_config(name_or_id: str) -> tuple[str, dict]: """Resolve experiment name or ID to config dict.""" from libs.backtest.experiments import resolve_experiment_name resolved = resolve_experiment_name(name_or_id) path = _CONFIGS_DIR / f"{resolved}.json" if not path.exists(): raise FileNotFoundError(f"Config not found: {path}") return resolved, json.loads(path.read_text()) def _run_single_backtest(manifest_path: str, snapshot_id: str | None, split: str, initial_equity: float, noise_std: float | None = None, shuffle_candidates: bool = False) -> float: """Run a backtest and return Sharpe ratio. Optionally inject noise or shuffle.""" import structlog structlog.configure(wrapper_class=structlog.make_filtering_bound_logger(50)) # suppress logs from apps.backtester.run import BacktestRunner, _build_store, _build_merged_snapshot_store from libs.backtest.manifests import load_manifest, resolve_config manifest = load_manifest(manifest_path) config = resolve_config(manifest) if snapshot_id: try: store = _build_store(manifest, config, split, snapshot_dir_override=None) except FileNotFoundError: store = _build_merged_snapshot_store(manifest, config, None) else: store = _build_merged_snapshot_store(manifest, config, None) if noise_std and noise_std > 0: _inject_noise_into_store(store, noise_std) if shuffle_candidates: _shuffle_candidate_dates(store) runner = BacktestRunner( manifest=manifest, config=config, store=store, initial_equity=initial_equity, split_name=split, ) result = runner.run(output_root=None) return result.metrics.sharpe_ratio if result.metrics.sharpe_ratio else 0.0 def _inject_noise_into_store(store, noise_std: float) -> None: """Add multiplicative Gaussian noise to all bar prices in-place.""" rng = np.random.default_rng() for symbol, bars in store._bars.items(): for date, bar in bars.items(): noise = 1 + rng.normal(0, noise_std) for field in ("open", "high", "low", "close"): if field in bar and bar[field] is not None: bar[field] = bar[field] * noise def _shuffle_candidate_dates(store) -> None: """Randomly reassign candidates to different reaction dates (destroys alpha signal).""" rng = np.random.default_rng() all_dates = sorted(store._candidates_by_reaction_date.keys()) all_candidates = [] for d in all_dates: all_candidates.extend(store._candidates_by_reaction_date[d]) rng.shuffle(all_candidates) new_map = {} idx = 0 for d in all_dates: n = len(store._candidates_by_reaction_date[d]) new_map[d] = all_candidates[idx:idx + n] idx += n store._candidates_by_reaction_date = new_map # Also rebuild the internal candidates list store._candidates = {d: rows for d, rows in new_map.items()} def _run_wfv_sharpe(manifest_path: str, initial_equity: float, shuffle: bool = False) -> float: """Run walk-forward and return mean test Sharpe.""" import structlog structlog.configure(wrapper_class=structlog.make_filtering_bound_logger(50)) from apps.backtester.run import _build_merged_snapshot_store, BacktestRunner from libs.backtest.manifests import load_manifest, resolve_config from libs.backtest.splits import generate_walk_forward_windows manifest = load_manifest(manifest_path) config = resolve_config(manifest) store = _build_merged_snapshot_store(manifest, config, None) if shuffle: _shuffle_candidate_dates(store) all_dates = store.all_trading_days(include_reaction_dates=True) if not all_dates: return 0.0 windows = generate_walk_forward_windows(all_dates, train_days=252, test_days=63, step_days=63) if not windows: return 0.0 test_sharpes = [] for window in windows: test_store = store.slice_by_date_range(window.test_start, window.test_end) runner = BacktestRunner( manifest=manifest, config=config, store=test_store, initial_equity=initial_equity, split_name="wfv_test", ) result = runner.run(output_root=None) sr = result.metrics.sharpe_ratio if result.metrics.sharpe_ratio else 0.0 test_sharpes.append(sr) return float(np.mean(test_sharpes)) if test_sharpes else 0.0 def _print_report(report) -> None: """Print rich formatted overfitting report.""" from libs.backtest.overfit import OverfitReport verdict_color = {"PASS": "green", "WARN": "yellow", "FAIL": "red"}.get(report.overall_verdict, "white") _console.print() _console.print(Panel( f"[bold]OVERFITTING ANALYSIS REPORT[/bold]\n" f"Strategy: [cyan]{report.experiment_name}[/cyan]" + (f" (ID {report.experiment_id})" if report.experiment_id else "") + f"\n\nOverall: [{verdict_color} bold]{report.overall_verdict}[/{verdict_color} bold]" f" Score: [bold]{report.overall_score:.0f}/100[/bold]" f" ({report.elapsed_seconds:.0f}s)", box=box.DOUBLE, width=100, )) # Test 1: DSR if report.dsr: d = report.dsr vc = {"PASS": "green", "WARN": "yellow", "FAIL": "red"}[d.verdict] _console.print(f"\n [bold]1. Deflated Sharpe Ratio (DSR)[/bold] [{vc}][{d.verdict}][/{vc}]") _console.print(f" Observed Sharpe: {d.observed_sharpe:.3f}") _console.print(f" Expected null (N={d.n_trials} trials): {d.expected_null_sharpe:.3f}") _console.print(f" DSR p-value: {d.dsr_pvalue:.4f} ({d.dsr_pvalue*100:.1f}%)") _console.print(f" Haircut: {d.sharpe_haircut_pct:.1f}% → surviving Sharpe: {d.deflated_sharpe:.3f}") _console.print(f" MinTRL: {d.min_track_record_days:.0f} days (data: {d.data_days} days)") if d.verdict == "PASS": _console.print(f" [green]→ {d.n_trials}번 시도 보정 후에도 통계적으로 유의미[/green]") else: _console.print(f" [yellow]→ Sharpe가 multiple testing에서 유의미하지 않을 수 있음[/yellow]") # Test 2: PBO if report.pbo: p = report.pbo vc = {"PASS": "green", "WARN": "yellow", "FAIL": "red"}[p.verdict] _console.print(f"\n [bold]2. Probability of Backtest Overfitting (PBO)[/bold] [{vc}][{p.verdict}][/{vc}]") _console.print(f" PBO: {p.pbo_probability:.4f} ({p.pbo_probability*100:.1f}%)") _console.print(f" Strategies: {p.n_strategies} | CSCV splits: {p.n_combinations}") _console.print(f" IS-best OOS rank median: {p.median_oos_rank:.3f} (1.0=best)") _console.print(f" OOS 상위 50%: {p.oos_above_50_pct:.1f}% | OOS Sharpe>0: {p.oos_sharpe_positive_pct:.1f}%") if p.verdict == "PASS": _console.print(f" [green]→ IS 최적화가 OOS 성과를 해칠 확률 {p.pbo_probability*100:.1f}%로 낮음[/green]") else: _console.print(f" [yellow]→ IS 최적 전략이 OOS에서 중앙값 이하일 확률 {p.pbo_probability*100:.0f}%[/yellow]") # Test 3: Parameter Sensitivity if report.param_sensitivity: verdicts = [r.verdict for r in report.param_sensitivity] overall_v = "PASS" if all(v == "PASS" for v in verdicts) else "WARN" if "FAIL" not in verdicts else "FAIL" vc = {"PASS": "green", "WARN": "yellow", "FAIL": "red"}[overall_v] _console.print(f"\n [bold]3. Parameter Sensitivity[/bold] [{vc}][{overall_v}][/{vc}]") for r in report.param_sensitivity: pvc = {"PASS": "green", "WARN": "yellow", "FAIL": "red"}[r.verdict] vals = ", ".join(f"{v:.3f}" for v in r.sharpe_values[:5]) _console.print(f" {r.param_name}: plateau={r.plateau_score:.2f} [{pvc}]{r.verdict}[/{pvc}] Sharpes=[{vals}]") _console.print(f" [dim]→ Plateau > 0.70 = robust, < 0.40 = cliff-edge[/dim]") # Test 4: Monte Carlo if report.monte_carlo: m = report.monte_carlo vc = {"PASS": "green", "WARN": "yellow", "FAIL": "red"}[m.verdict] _console.print(f"\n [bold]4. Monte Carlo Noise Injection[/bold] [{vc}][{m.verdict}][/{vc}]") _console.print(f" Baseline Sharpe: {m.baseline_sharpe:.3f}") _console.print(f" Noise {m.noise_level_pct:.1f}% x {m.n_iterations} iterations:") _console.print(f" Median noised: {m.median_noised_sharpe:.3f} | 5th pctile: {m.p05_sharpe:.3f}") _console.print(f" Degradation: {m.degradation_pct:.1f}%") if m.verdict == "PASS": _console.print(f" [green]→ 가격 노이즈에 대해 수익률이 완만하게 감소 — 견고한 신호[/green]") else: _console.print(f" [yellow]→ 노이즈에 민감 — 특정 가격 패턴에 curve-fit 가능성[/yellow]") # Test 5: Permutation if report.permutation: r = report.permutation vc = {"PASS": "green", "WARN": "yellow", "FAIL": "red"}[r.verdict] _console.print(f"\n [bold]5. Walk-Forward Permutation Test[/bold] [{vc}][{r.verdict}][/{vc}]") _console.print(f" Observed Sharpe: {r.observed_sharpe:.3f}") _console.print(f" Null distribution ({r.n_permutations} permutations):") _console.print(f" Median: {r.null_median_sharpe:.3f} | 95th pctile: {r.null_p95_sharpe:.3f}") _console.print(f" p-value: {r.p_value:.4f}") if r.verdict == "PASS": _console.print(f" [green]→ random 셔플의 {r.p_value*100:.1f}%만 이 Sharpe 이상 — 진짜 alpha[/green]") else: _console.print(f" [yellow]→ p={r.p_value:.3f}: alpha가 통계적으로 유의미하지 않을 수 있음[/yellow]") # Summary _console.print() _console.print(Panel( f"[bold]Summary[/bold]\n" + "\n".join([ f" DSR: {report.dsr.verdict if report.dsr else 'SKIP'}" f" PBO: {report.pbo.verdict if report.pbo else 'SKIP'}" f" Sensitivity: {report.param_sensitivity[0].verdict if report.param_sensitivity else 'SKIP'}" f" MC: {report.monte_carlo.verdict if report.monte_carlo else 'SKIP'}" f" Permutation: {report.permutation.verdict if report.permutation else 'SKIP'}", ]), box=box.ROUNDED, width=100, )) def main() -> None: parser = argparse.ArgumentParser(description="Comprehensive overfitting analysis") parser.add_argument("--config", "-c", required=True, help="Experiment ID or name") parser.add_argument("--initial-equity", type=float, default=10_000) parser.add_argument("--skip", help="Comma-separated tests to skip: dsr,pbo,sensitivity,mc,permutation") parser.add_argument("--quick", action="store_true", help="Quick mode: mc=10, perm=30") parser.add_argument("--mc-iterations", type=int, default=50) parser.add_argument("--mc-noise-pct", type=float, default=0.5) parser.add_argument("--permutations", type=int, default=100) parser.add_argument("--output-json", help="Export report as JSON") args = parser.parse_args() if args.quick: args.mc_iterations = 10 args.permutations = 30 skip = set(args.skip.split(",")) if args.skip else set() t0 = time.time() name, config = _resolve_config(args.config) exp_id = config.get("id") manifest_path = str(_CONFIGS_DIR / f"{name}.json") _console.print(f"\n[bold]Overfitting check: {name}[/bold] (equity=${args.initial_equity:,.0f})") from libs.backtest.overfit import ( DSRResult, PBOResult, MonteCarloResult, PermutationResult, OverfitReport, compute_dsr, compute_pbo, compute_monte_carlo, compute_permutation, compute_param_sensitivity, compute_overall_score, load_equity_curve_returns, load_sibling_returns_matrix, ) report = OverfitReport(experiment_name=name, experiment_id=exp_id, overall_verdict="", overall_score=0) # --- Test 1: DSR --- if "dsr" not in skip: _console.print("\n[dim]Running DSR...[/dim]") matrix, strat_names = load_sibling_returns_matrix(name) if matrix is not None: all_sharpes = [] target_returns = None target_sharpe = 0.0 for i, sname in enumerate(strat_names): rets = matrix[:, i] sr = float(rets.mean() / max(rets.std(ddof=1), 1e-10)) all_sharpes.append(sr) if sname == name.replace("return_max_long_", ""): target_returns = rets target_sharpe = sr if target_returns is None: # Use best as proxy best_idx = int(np.argmax(all_sharpes)) target_returns = matrix[:, best_idx] target_sharpe = all_sharpes[best_idx] report.dsr = compute_dsr(target_returns, target_sharpe, len(all_sharpes), all_sharpes) _console.print(f" DSR = {report.dsr.dsr_pvalue*100:.1f}% [{report.dsr.verdict}]") else: _console.print(" [yellow]Skipped: not enough sibling CW runs[/yellow]") # --- Test 2: PBO --- if "pbo" not in skip: _console.print("[dim]Running PBO/CSCV...[/dim]") matrix, strat_names = load_sibling_returns_matrix(name) if matrix is not None and matrix.shape[1] >= 5: report.pbo = compute_pbo(matrix, n_subsets=10) _console.print(f" PBO = {report.pbo.pbo_probability*100:.1f}% [{report.pbo.verdict}]") else: _console.print(f" [yellow]Skipped: need >= 5 siblings (found {matrix.shape[1] if matrix is not None else 0})[/yellow]") # --- Test 3: Parameter Sensitivity --- if "sensitivity" not in skip: _console.print("[dim]Running parameter sensitivity...[/dim]") # Auto-detect key sizing params and test grid base_risk = config.get("overrides", {}).get("risk", {}).get("per_trade_risk_pct", 0.069) daily_risk = config.get("overrides", {}).get("risk", {}).get("max_daily_new_risk_pct", 0.76) params_to_test = [ ("overrides.risk.per_trade_risk_pct", [base_risk * 0.7, base_risk * 0.85, base_risk, base_risk * 1.15, base_risk * 1.3]), ("overrides.risk.max_daily_new_risk_pct", [daily_risk * 0.5, daily_risk * 0.75, daily_risk, daily_risk * 1.25, daily_risk * 1.5]), ] snapshot_id = config.get("dataset_snapshot_id", "") merged_snap = snapshot_id + "_merged" if not snapshot_id.endswith("_merged") else snapshot_id for param_path, values in params_to_test: def _run_bt(cfg_override, _pp=param_path, _v=values, _mpath=manifest_path, _eq=args.initial_equity, _snap=merged_snap): from libs.backtest.overfit import _set_nested import tempfile cfg = copy.deepcopy(config) for k, v in cfg_override.items() if isinstance(cfg_override, dict) else [(_pp, cfg_override)]: _set_nested(cfg, k if isinstance(cfg_override, dict) else _pp, v if isinstance(cfg_override, dict) else cfg_override) tf = Path(tempfile.mktemp(suffix=".json")) tf.write_text(json.dumps(cfg)) try: sr = _run_single_backtest(str(tf), _snap, "train", _eq) except Exception: sr = 0.0 finally: tf.unlink(missing_ok=True) return sr sharpes = [] for val in values: sr = _run_bt(val) sharpes.append(sr) arr = np.array(sharpes) if arr.mean() > 0: plateau = max(0, 1 - float(arr.std(ddof=1) / arr.mean())) else: plateau = 0 result = compute_param_sensitivity(lambda x: 0, {}, param_path, values) result.sharpe_values = sharpes result.plateau_score = plateau result.verdict = "PASS" if plateau > 0.70 else "WARN" if plateau > 0.40 else "FAIL" report.param_sensitivity.append(result) _console.print(f" {param_path}: plateau={plateau:.2f} [{result.verdict}]") # --- Test 4: Monte Carlo Noise --- if "mc" not in skip: _console.print(f"[dim]Running Monte Carlo noise ({args.mc_iterations} iterations)...[/dim]") snapshot_id = config.get("dataset_snapshot_id", "") merged_snap = snapshot_id + "_merged" if not snapshot_id.endswith("_merged") else snapshot_id baseline_sr = _run_single_backtest(manifest_path, merged_snap, "train", args.initial_equity) _console.print(f" Baseline Sharpe: {baseline_sr:.3f}") noised_sharpes = [] with Progress(SpinnerColumn(), TextColumn("[progress.description]{task.description}"), BarColumn(), "{task.completed}/{task.total}", TimeElapsedColumn(), console=_console) as progress: task = progress.add_task("MC noise", total=args.mc_iterations) for i in range(args.mc_iterations): sr = _run_single_backtest(manifest_path, merged_snap, "train", args.initial_equity, noise_std=args.mc_noise_pct / 100) noised_sharpes.append(sr) progress.update(task, advance=1) arr = np.array(noised_sharpes) median_sr = float(np.median(arr)) p05 = float(np.percentile(arr, 5)) deg = (1 - median_sr / baseline_sr) * 100 if baseline_sr > 0 else 100 report.monte_carlo = MonteCarloResult( baseline_sharpe=baseline_sr, noise_level_pct=args.mc_noise_pct, n_iterations=args.mc_iterations, median_noised_sharpe=median_sr, p05_sharpe=p05, degradation_pct=max(0, deg), all_sharpes=noised_sharpes, ) _console.print(f" Median noised: {median_sr:.3f} (degradation {deg:.1f}%) [{report.monte_carlo.verdict}]") # --- Test 5: Walk-Forward Permutation --- if "permutation" not in skip: _console.print(f"[dim]Running WFV permutation ({args.permutations} iterations)...[/dim]") observed_sr = _run_wfv_sharpe(manifest_path, args.initial_equity, shuffle=False) _console.print(f" Observed WFV Sharpe: {observed_sr:.3f}") null_sharpes = [] with Progress(SpinnerColumn(), TextColumn("[progress.description]{task.description}"), BarColumn(), "{task.completed}/{task.total}", TimeElapsedColumn(), console=_console) as progress: task = progress.add_task("Permutation", total=args.permutations) for i in range(args.permutations): sr = _run_wfv_sharpe(manifest_path, args.initial_equity, shuffle=True) null_sharpes.append(sr) progress.update(task, advance=1) arr = np.array(null_sharpes) p_value = float((arr >= observed_sr).mean()) report.permutation = PermutationResult( observed_sharpe=observed_sr, n_permutations=args.permutations, p_value=p_value, null_median_sharpe=float(np.median(arr)), null_p95_sharpe=float(np.percentile(arr, 95)), ) _console.print(f" p-value: {p_value:.4f} [{report.permutation.verdict}]") # --- Compute overall --- report.elapsed_seconds = time.time() - t0 report.overall_score, report.overall_verdict = compute_overall_score(report) _print_report(report) if args.output_json: import dataclasses out = dataclasses.asdict(report) output_path = Path(args.output_json) output_path.parent.mkdir(parents=True, exist_ok=True) output_path.write_text(json.dumps(out, indent=2, default=str)) _console.print(f"\n[dim]Report saved to {args.output_json}[/dim]") if __name__ == "__main__": main()