You cannot select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
367 lines
13 KiB
Python
367 lines
13 KiB
Python
"""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/<experiment>/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()
|