"""CLI for synthetic market scenario backtesting and robustness analysis. Usage: fithia2 scenario-test --config return_max_long_v7.70 fithia2 scenario-test --config return_max_long_v7.70 --quick fithia2 scenario-test --config return_max_long_v7.70 --scenario crash_v_recovery fithia2 scenario-test --config return_max_long_v7.70 --group signal fithia2 scenario-test --config return_max_long_v7.70 --baseline return_max_long_v6new.362 fithia2 scenario-test --config return_max_long_v7.70 --initial-equity 10000 fithia2 scenario-test --config return_max_long_v7.70 --save """ from __future__ import annotations import argparse import json import sys import time from pathlib import Path from rich import box from rich.console import Console from rich.panel import Panel from rich.progress import BarColumn, Progress, SpinnerColumn, TextColumn, TimeElapsedColumn from rich.table import Table _console = Console(width=120) _CONFIGS_DIR = Path("configs/experiments") _RUNS_DIR = Path("runs") def _resolve_config_name(name_or_id: str) -> str: """Resolve experiment name or numeric ID to full config name.""" from libs.backtest.experiments import resolve_experiment_name return resolve_experiment_name(name_or_id) def _print_scenario_table(report) -> None: """Print per-scenario metrics table.""" from libs.backtest.scenarios.scenarios import SCENARIO_REGISTRY table = Table( title=f"Scenario Results — {report.experiment_name}", box=box.ROUNDED, width=118, ) table.add_column("Scenario", style="cyan", min_width=22) table.add_column("Sharpe", justify="right", min_width=7) table.add_column("Return%", justify="right", min_width=8) table.add_column("MaxDD%", justify="right", min_width=7) table.add_column("Win%", justify="right", min_width=6) table.add_column("PF", justify="right", min_width=6) table.add_column("Trades", justify="right", min_width=7) table.add_column("Signal", justify="right", min_width=7) table.add_column("Status", min_width=8) for name, result in report.scenario_results.items(): scenario = SCENARIO_REGISTRY.get(name) sig = f"{scenario.signal_strength:.2f}" if scenario else "?" sharpe = result.sharpe_ratio ret = result.total_return_pct dd = result.max_drawdown_pct win = result.win_rate * 100 pf = result.profit_factor trades = result.trade_count # Colour-code by Sharpe if sharpe >= 1.5: sharpe_str = f"[green]{sharpe:.2f}[/green]" elif sharpe >= 0.5: sharpe_str = f"[yellow]{sharpe:.2f}[/yellow]" elif sharpe >= 0.0: sharpe_str = f"[dim]{sharpe:.2f}[/dim]" else: sharpe_str = f"[red]{sharpe:.2f}[/red]" ret_str = f"[green]+{ret:.1f}[/green]" if ret > 0 else f"[red]{ret:.1f}[/red]" dd_str = f"[red]{dd:.1f}[/red]" if dd > 20 else f"{dd:.1f}" # Status flag if name == "no_signal" and sharpe > 0.5: status = "[red bold]OVERFIT[/red bold]" elif name == "no_signal" and sharpe <= 0: status = "[green]OK[/green]" elif name == "strong_signal" and sharpe < 0.5: status = "[yellow]WEAK[/yellow]" else: status = "" table.add_row( name, sharpe_str, ret_str, dd_str, f"{win:.0f}", f"{pf:.2f}", str(trades), sig, status, ) _console.print() _console.print(table) def _print_rrs_panel(report) -> None: """Print Regime Robustness Score panel.""" verdict_color = { "ROBUST": "green", "FRAGILE": "yellow", "OVERFIT": "red", }.get(report.verdict, "white") def _score_bar(score: float) -> str: filled = int(round(score / 5)) empty = 20 - filled bar = "█" * filled + "░" * empty if score >= 70: color = "green" elif score >= 40: color = "yellow" else: color = "red" return f"[{color}]{bar}[/{color}] {score:.0f}/100" lines = [ f"[bold]REGIME ROBUSTNESS SCORE (RRS)[/bold]", f"Strategy: [cyan]{report.experiment_name}[/cyan]", f"", f" Signal Integrity (25%) {_score_bar(report.signal_integrity)}", f" → no_signal Sharpe ≤ 0 confirms event-driven alpha (not price-pattern overfit)", f"", f" Breadth (25%) {_score_bar(report.breadth)}", f" → fraction of scenarios with positive Sharpe", f"", f" Drawdown Resilience(20%) {_score_bar(report.drawdown_resilience)}", f" → worst-case max drawdown across all scenarios", f"", f" Regime Transition (15%) {_score_bar(report.regime_transition)}", f" → performance on regime_switch vs median", f"", f" Stability (15%) {_score_bar(report.stability)}", f" → low Sharpe variance across diverse market conditions", f"", f" [bold]RRS: {_score_bar(report.rrs)}[/bold]", f" [{verdict_color} bold]Verdict: {report.verdict}[/{verdict_color} bold]", ] if report.notes: lines += ["", "[dim]Notes:"] + [f" • {n}" for n in report.notes] + ["[/dim]"] _console.print() _console.print(Panel("\n".join(lines), box=box.DOUBLE, width=100)) def _print_comparison_table(report_a, report_b) -> None: """Print side-by-side comparison of two strategies.""" table = Table(title="Strategy Comparison", box=box.ROUNDED, width=118) table.add_column("Scenario", style="cyan", min_width=22) table.add_column(f"{report_a.experiment_name[:18]} Sharpe", justify="right") table.add_column(f"{report_b.experiment_name[:18]} Sharpe", justify="right") table.add_column("Delta", justify="right") all_names = sorted( set(report_a.scenario_results.keys()) | set(report_b.scenario_results.keys()) ) for name in all_names: s_a = report_a.scenario_results.get(name) s_b = report_b.scenario_results.get(name) sh_a = s_a.sharpe_ratio if s_a else float("nan") sh_b = s_b.sharpe_ratio if s_b else float("nan") def _fmt(v: float) -> str: if v != v: return "[dim]N/A[/dim]" color = "green" if v > 0.5 else ("yellow" if v >= 0 else "red") return f"[{color}]{v:.2f}[/{color}]" delta = sh_b - sh_a if (sh_a == sh_a and sh_b == sh_b) else float("nan") delta_str = ( f"[green]+{delta:.2f}[/green]" if delta > 0.05 else (f"[red]{delta:.2f}[/red]" if delta < -0.05 else f"[dim]{delta:.2f}[/dim]") if delta == delta else "[dim]N/A[/dim]" ) table.add_row(name, _fmt(sh_a), _fmt(sh_b), delta_str) _console.print() _console.print(table) # RRS comparison _console.print( f"\n RRS: {report_a.experiment_name} = [bold]{report_a.rrs:.0f}[/bold] " f"vs {report_b.experiment_name} = [bold]{report_b.rrs:.0f}[/bold]" f" (delta [bold]{report_b.rrs - report_a.rrs:+.0f}[/bold])" ) def main() -> None: parser = argparse.ArgumentParser( prog="fithia2 scenario-test", description=( "Synthetic market scenario backtesting for overfitting detection. " "Tests strategies against generated market conditions not present in historical data." ), ) parser.add_argument( "--config", required=True, help="Experiment config name or numeric ID (e.g. 'return_max_long_v7.70' or '415')", ) parser.add_argument( "--scenario", default=None, help="Run a single scenario by name (e.g. 'crash_v_recovery')", ) parser.add_argument( "--group", default=None, help="Run a scenario group: trend, volatility, regime, signal, structural, quick, all", ) parser.add_argument( "--quick", action="store_true", help="Quick mode: run only steady_bull, steady_bear, no_signal (3 scenarios)", ) parser.add_argument( "--baseline", default=None, help="Optional baseline config to compare against", ) parser.add_argument( "--initial-equity", type=float, default=10_000.0, help="Starting capital for each scenario (default: 10000)", ) parser.add_argument( "--save", action="store_true", help="Save JSON report to runs//scenario_report.json", ) parser.add_argument( "--list", action="store_true", help="List all available scenarios and exit", ) args = parser.parse_args() from libs.backtest.scenarios.scenarios import SCENARIO_REGISTRY, SCENARIO_GROUPS if args.list: _console.print("\n[bold]Available scenarios:[/bold]") for name, sc in SCENARIO_REGISTRY.items(): _console.print(f" [cyan]{name:<25}[/cyan] signal={sc.signal_strength:.2f} {sc.description[:60]}") _console.print("\n[bold]Scenario groups:[/bold]") for g, names in SCENARIO_GROUPS.items(): _console.print(f" [yellow]{g:<15}[/yellow] {', '.join(names)}") return # Resolve which scenarios to run if args.scenario: if args.scenario not in SCENARIO_REGISTRY: _console.print(f"[red]Unknown scenario '{args.scenario}'. Use --list to see options.[/red]") sys.exit(1) scenario_names = [args.scenario] elif args.quick: scenario_names = SCENARIO_GROUPS["quick"] elif args.group: if args.group not in SCENARIO_GROUPS: _console.print(f"[red]Unknown group '{args.group}'. Use --list to see options.[/red]") sys.exit(1) scenario_names = SCENARIO_GROUPS[args.group] else: scenario_names = SCENARIO_GROUPS["all"] try: experiment_name = _resolve_config_name(args.config) except Exception as exc: _console.print(f"[red]Cannot resolve config '{args.config}': {exc}[/red]") sys.exit(1) _console.print() _console.print(Panel( f"[bold]SYNTHETIC SCENARIO TEST[/bold]\n" f"Strategy: [cyan]{experiment_name}[/cyan]\n" f"Scenarios: [yellow]{len(scenario_names)}[/yellow] ({', '.join(scenario_names)})\n" f"Initial equity: ${args.initial_equity:,.0f}", box=box.DOUBLE, width=100, )) t0 = time.time() completed: list[str] = [] with Progress( SpinnerColumn(), TextColumn("[progress.description]{task.description}"), BarColumn(bar_width=30), TextColumn("{task.completed}/{task.total}"), TimeElapsedColumn(), console=_console, ) as progress: task = progress.add_task("Running scenarios...", total=len(scenario_names)) def _cb(name: str) -> None: progress.update(task, description=f"[cyan]{name}[/cyan]") from libs.backtest.scenarios.robustness import run_scenario_test report = run_scenario_test( experiment_name=experiment_name, scenario_names=scenario_names, initial_equity=args.initial_equity, progress_callback=_cb, ) progress.update(task, completed=len(scenario_names), description="Complete") elapsed = time.time() - t0 _print_scenario_table(report) _print_rrs_panel(report) # Optional baseline comparison if args.baseline: try: baseline_name = _resolve_config_name(args.baseline) _console.print(f"\n[dim]Running baseline {baseline_name}...[/dim]") from libs.backtest.scenarios.robustness import run_scenario_test as _rtt baseline_report = _rtt( experiment_name=baseline_name, scenario_names=scenario_names, initial_equity=args.initial_equity, ) _print_comparison_table(baseline_report, report) except Exception as exc: _console.print(f"[yellow]Baseline comparison failed: {exc}[/yellow]") # Save report if args.save: try: save_dir = _RUNS_DIR / experiment_name save_dir.mkdir(parents=True, exist_ok=True) out_path = save_dir / "scenario_report.json" payload = { "experiment_name": experiment_name, "rrs": report.rrs, "verdict": report.verdict, "signal_integrity": report.signal_integrity, "breadth": report.breadth, "drawdown_resilience": report.drawdown_resilience, "regime_transition": report.regime_transition, "stability": report.stability, "scenarios": { name: { "sharpe_ratio": r.sharpe_ratio, "total_return_pct": r.total_return_pct, "max_drawdown_pct": r.max_drawdown_pct, "win_rate": r.win_rate, "profit_factor": r.profit_factor, "trade_count": r.trade_count, } for name, r in report.scenario_results.items() }, "elapsed_seconds": round(elapsed, 1), "notes": report.notes, } out_path.write_text(json.dumps(payload, indent=2)) _console.print(f"\n[dim]Report saved → {out_path}[/dim]") except Exception as exc: _console.print(f"[yellow]Could not save report: {exc}[/yellow]") _console.print(f"\n[dim]Elapsed: {elapsed:.0f}s[/dim]\n") if __name__ == "__main__": main()