From 9cb91ee846939e936560b164d939d9d79a639e0d Mon Sep 17 00:00:00 2001 From: I Luk Kim Date: Mon, 30 Mar 2026 23:07:17 -0700 Subject: [PATCH] Add synthetic scenario robustness testing system Builds a full synthetic market data pipeline to test strategies against 12 diverse market regimes (bull/bear/crash/chop/rotation/liquidity drought) that may not exist in historical data. Computes Regime Robustness Score (RRS) to detect overfitting and environment-specific fragility. - libs/backtest/scenarios/: price_gen, macro_gen, event_gen, coupling, store_builder, scenarios (12 pre-built), robustness (RRS) - apps/scenario/cli.py: `fithia2 scenario-test` with Rich output - apps/tracker/cli.py: scenario-test command routing - tests/: 83 unit tests across 3 new test files - docs/scenario_test.md: usage guide and result interpretation - docs/research_workflow_and_handoff.md: Step 5.5 scenario test added Fix: no_signal scenario uses drift=0% (was +10%) for fair signal integrity scoring. Fix: synthetic candidates now carry macro_vix/macro_hy_spread from macro_by_date to pass selector engine filters. Co-Authored-By: Claude Sonnet 4.6 --- apps/scenario/__init__.py | 1 + apps/scenario/cli.py | 366 ++++++++++++++ apps/tracker/cli.py | 98 +++- docs/research_workflow_and_handoff.md | 43 +- docs/scenario_test.md | 157 ++++++ libs/backtest/scenarios/__init__.py | 38 ++ libs/backtest/scenarios/coupling.py | 156 ++++++ libs/backtest/scenarios/event_gen.py | 395 +++++++++++++++ libs/backtest/scenarios/macro_gen.py | 456 ++++++++++++++++++ libs/backtest/scenarios/price_gen.py | 242 ++++++++++ libs/backtest/scenarios/robustness.py | 258 ++++++++++ libs/backtest/scenarios/scenarios.py | 332 +++++++++++++ libs/backtest/scenarios/store_builder.py | 146 ++++++ tests/unit/backtest/test_event_gen.py | 245 ++++++++++ .../unit/backtest/test_scenario_robustness.py | 219 +++++++++ tests/unit/backtest/test_store_builder.py | 104 ++++ 16 files changed, 3254 insertions(+), 2 deletions(-) create mode 100644 apps/scenario/__init__.py create mode 100644 apps/scenario/cli.py create mode 100644 docs/scenario_test.md create mode 100644 libs/backtest/scenarios/__init__.py create mode 100644 libs/backtest/scenarios/coupling.py create mode 100644 libs/backtest/scenarios/event_gen.py create mode 100644 libs/backtest/scenarios/macro_gen.py create mode 100644 libs/backtest/scenarios/price_gen.py create mode 100644 libs/backtest/scenarios/robustness.py create mode 100644 libs/backtest/scenarios/scenarios.py create mode 100644 libs/backtest/scenarios/store_builder.py create mode 100644 tests/unit/backtest/test_event_gen.py create mode 100644 tests/unit/backtest/test_scenario_robustness.py create mode 100644 tests/unit/backtest/test_store_builder.py diff --git a/apps/scenario/__init__.py b/apps/scenario/__init__.py new file mode 100644 index 0000000..dbc6594 --- /dev/null +++ b/apps/scenario/__init__.py @@ -0,0 +1 @@ +# Synthetic scenario testing CLI package diff --git a/apps/scenario/cli.py b/apps/scenario/cli.py new file mode 100644 index 0000000..bf2a602 --- /dev/null +++ b/apps/scenario/cli.py @@ -0,0 +1,366 @@ +"""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() diff --git a/apps/tracker/cli.py b/apps/tracker/cli.py index 32c78c6..cfaabad 100644 --- a/apps/tracker/cli.py +++ b/apps/tracker/cli.py @@ -384,6 +384,21 @@ def cmd_leaderboard(args: argparse.Namespace) -> None: title_justify="left", expand=False, ) + # Load experiment IDs from index for display + _exp_id_map: dict[str, str] = {} + try: + import json as _json + _idx = Path("configs/experiments/.index.json") + if _idx.exists(): + _idx_data = _json.loads(_idx.read_text()) + for _ename, _emeta in _idx_data.get("experiments", {}).items(): + eid = _emeta.get("id") + if eid is not None: + _exp_id_map[_ename] = str(eid) + except Exception: + pass + + tbl.add_column("ID", justify="right", style="bold dim", no_wrap=True, min_width=4) tbl.add_column("#", justify="right", style="bold", no_wrap=True, min_width=3) tbl.add_column("Experiment", no_wrap=True, min_width=30) tbl.add_column("SQS", justify="right", style="bold cyan", no_wrap=True, min_width=5) @@ -408,6 +423,7 @@ def cmd_leaderboard(args: argparse.Namespace) -> None: if len(name) > _COL_NAME: name = name[: _COL_NAME - 1] + "…" row = [ + _exp_id_map.get(entry.experiment_name, "—"), str(rank), name, _fmt(entry.sqs_score, ".1f"), @@ -483,6 +499,7 @@ def cmd_show(args: argparse.Namespace) -> None: robustness_matrix_summary=entry.robustness_matrix_summary, out_of_time_robustness_summary=entry.out_of_time_robustness_summary, common_window_summary=entry.common_window_summary, + multi_capital_common_window_summary=entry.multi_capital_common_window_summary, rqs_score=fresh_rqs_score, wfqs_score=fresh_wfqs_v2_score, ) @@ -831,6 +848,11 @@ def _print_help() -> None: "데이터 파이프라인 실행 [dim](fithia2 pipeline 로 상세 확인)[/]", "run [dim]--step poller|fetcher|parser|features|labels[/]", ) + table.add_row( + "exp", + "실험 관리 [dim](fithia2 exp 로 상세 확인)[/]", + "create search tree info diff promote retire validate migrate", + ) _console.print(table) _console.print( @@ -845,6 +867,27 @@ def _print_help() -> None: def main() -> None: + # Delegate `fithia2 exp ...` to the experiment management CLI + if len(sys.argv) >= 2 and sys.argv[1] == "exp": + sys.argv = [sys.argv[0]] + sys.argv[2:] + from apps.experiment.cli import main as exp_main + exp_main() + return + + # Delegate `fithia2 overfit-check ...` to the overfitting analysis CLI + if len(sys.argv) >= 2 and sys.argv[1] == "overfit-check": + sys.argv = [sys.argv[0]] + sys.argv[2:] + from apps.overfit.cli import main as overfit_main + overfit_main() + return + + # Delegate `fithia2 scenario-test ...` to the synthetic scenario test CLI + if len(sys.argv) >= 2 and sys.argv[1] == "scenario-test": + sys.argv = [sys.argv[0]] + sys.argv[2:] + from apps.scenario.cli import main as scenario_main + scenario_main() + return + # Delegate `fithia2 paper ...` to the paper trader CLI if len(sys.argv) >= 2 and sys.argv[1] == "paper": sys.argv = [sys.argv[0]] + sys.argv[2:] @@ -852,6 +895,53 @@ def main() -> None: paper_main() return + # fithia2 refresh [snapshot_id] — refresh Parquet snapshot + if len(sys.argv) >= 2 and sys.argv[1] == "refresh": + import asyncio + from apps.paper_trader.backtest_sim import _refresh_snapshot + from rich.console import Console + console = Console() + + snapshot_id = sys.argv[2] if len(sys.argv) >= 3 else None + if not snapshot_id: + # Auto-detect: use the most common snapshot_id from configs + from pathlib import Path + import json + configs_dir = Path("configs/experiments") + if configs_dir.exists(): + ids: dict[str, int] = {} + for f in configs_dir.glob("*.json"): + try: + data = json.loads(f.read_text()) + sid = data.get("dataset_snapshot_id", "") + if sid: + ids[sid] = ids.get(sid, 0) + 1 + except Exception: + pass + if ids: + snapshot_id = max(ids, key=ids.get) + console.print(f"[dim]Auto-detected snapshot: {snapshot_id}[/]") + if not snapshot_id: + console.print("[red]Usage: fithia2 refresh [/]") + sys.exit(1) + + universe_profile = None + if "midlarge" in snapshot_id: + universe_profile = "midlarge-liquid-long-v1" + elif "midwide" in snapshot_id: + universe_profile = "midwide-liquid-long-v1" + elif "smallcap" in snapshot_id: + universe_profile = "smallcap-liquid-long-v1" + + console.print(f"[bold]Refreshing snapshot:[/] {snapshot_id}") + try: + asyncio.run(_refresh_snapshot(snapshot_id, universe_profile, console=console)) + console.print("[bold green]Done.[/]") + except Exception as exc: + console.print(f"[bold red]Refresh failed: {exc}[/]") + sys.exit(1) + return + # Delegate `fithia2 pipeline ...` to the pipeline CLI if len(sys.argv) >= 2 and sys.argv[1] == "pipeline": sys.argv = [sys.argv[0]] + sys.argv[2:] @@ -859,6 +949,12 @@ def main() -> None: pipeline_main() return + # Delegate `fithia2 web ...` to the web GUI server + if len(sys.argv) >= 2 and sys.argv[1] == "web": + from apps.web.main import run_server + run_server() + return + if len(sys.argv) == 1: _print_help() sys.exit(0) @@ -902,7 +998,7 @@ def main() -> None: subparser.add_argument( "--include-retired", action="store_true", - help="Include retired legacy PEAD / short-core / exact-pocket families", + help="Include archived pre-IMP-0606 research and retired legacy families", ) for name in ("show", "s"): diff --git a/docs/research_workflow_and_handoff.md b/docs/research_workflow_and_handoff.md index 465ae25..80f4809 100644 --- a/docs/research_workflow_and_handoff.md +++ b/docs/research_workflow_and_handoff.md @@ -108,6 +108,12 @@ Official Experiment 조건에 더해: - baseline 이름은 journal entry와 동일하게 사용한다. - paper backtest와 research backtester는 같은 snapshot을 써야 비교가 된다. 경로 해석 우선순위는 `settings.parquet_dir` 다음 `data/datasets/snapshots`다. +- public default `SQS`용 common-window는 현재 **초기 자본 `10,000`** 기준으로 붙인다. +- official common-window 날짜 범위는 현재 `2022-03-02 ~ 2026-03-24`다. +- `25k/100k` common-window는 optional diagnostics로만 본다. +- 최근 `v6new.34x ~ 37x` lineage는 `25k/100k` diagnostics도 일부 backfill돼 있지만, 기본 rank는 `10k`만 쓴다. +- 기본 leaderboard active window는 현재 **`IMP-0606` / `return_max_long_v6new.29` 이후**다. + 그 전 실험은 기록은 남기되 기본 보드에서는 archived로 숨긴다. ### Step 2. Scratch 탐색 @@ -142,11 +148,39 @@ deploy 후보는 아래를 붙인다. - walk-forward validation - robustness matrix - repaired out-of-time robustness - - 현재 기준 snapshot은 + - 기본 stress OOT snapshot은 [`midlarge-liquid-long-v1-oot-2020-2021`](/Users/yirugi/mycloud/personal/workspace/fithia2/data/datasets/snapshots/midlarge-liquid-long-v1-oot-2020-2021/manifest.json) + - 단, `tier2/tier3` feature를 실제 selector/scorer/gate에 쓰는 전략은 + feature-matched OOT snapshot을 써야 한다. + 현재 기준은 + [`midlarge-liquid-long-v1-oot-2020-2021_tier3`](/Users/yirugi/mycloud/personal/workspace/fithia2/data/parquet/midlarge-liquid-long-v1-oot-2020-2021_tier3/manifest.json) 그리고 tracker에 attach한다. +### Step 5.5. Synthetic Scenario Test (선택, deploy 후보 권장) + +합성 시장 데이터를 이용해 역사에 없던 시장 환경에서의 내성을 검증한다. +자세한 해석 가이드는 [`docs/scenario_test.md`](/Users/yirugi/mycloud/personal/workspace/fithia2/docs/scenario_test.md)를 본다. + +```bash +# 빠른 핵심 3개 (2~3초) +fithia2 scenario-test --config --quick + +# 전체 12개 시나리오 (~10초) +fithia2 scenario-test --config + +# 취약 환경 집중 점검 +fithia2 scenario-test --config --group structural +``` + +**RRS 판정 기준**: ≥ 70 ROBUST / 40–69 FRAGILE / < 40 OVERFIT + +deploy 후보 비교 시 유용한 활용: +- 신규 전략의 `sector_rotation`, `liquidity_drought` Sharpe가 baseline보다 개선됐는지 확인 +- `no_signal` Sharpe > 0이면 가격 패턴 과적합 의심 신호 + +--- + ### Step 6. Leaderboard 갱신 정식 manifest + journal + WFV/robustness attach까지 끝난 뒤에 leaderboard를 본다. @@ -314,6 +348,13 @@ OOT가 전부 `0 trade`로 보이면 전략 탓만 하지 말고 snapshot 자체 실제로 symbol 기반 export에서 `market_cap_proxy`, `exchange_proxy`가 비어 있던 버그가 있었고, 이를 고친 뒤에야 repaired OOT가 의미 있는 비교 지표가 됐다. +추가 원칙: +- repaired OOT summary가 전 horizon에서 전부 `0`인 `all-zero sparse no-trade` 패턴이면, + 기본 `SQS`에서는 fail로 깎지 않는다. +- 이 경우는 `bad stress performance`가 아니라 `stress window non-comparable`로 보고, + OOT gate를 중립 처리한다. +- 대신 여전히 journal 설명과 diagnostics에는 `sparse OOT`라는 사실을 남긴다. + ### 10.4 Paper Backtest stale 판정은 manifest 나이가 아니라 coverage로 본다 `fithia2 paper backtest`는 snapshot이 오래됐다는 이유만으로 refresh하면 안 된다. diff --git a/docs/scenario_test.md b/docs/scenario_test.md new file mode 100644 index 0000000..b4abf22 --- /dev/null +++ b/docs/scenario_test.md @@ -0,0 +1,157 @@ +# Scenario Robustness Test — `fithia2 scenario-test` + +합성 시장 데이터를 이용해 다양한 시장 환경에서 전략을 검증하고 과적합/취약점을 탐지하는 도구. + +`overfit-check`가 **실제 데이터 기반 통계 검증**이라면, `scenario-test`는 **역사에 없었던 시장 환경**에 대한 내성 테스트다. +예: "VIX가 지속적으로 30이었다면?", "2022년 하락장이 2배 길었다면?", "유동성이 60% 증발했다면?" + +--- + +## 빠른 사용법 + +```bash +# 3개 핵심 시나리오 (2~3초) +fithia2 scenario-test --config return_max_long_v7.70 --quick + +# 전체 12개 시나리오 (~10초) +fithia2 scenario-test --config return_max_long_v7.70 + +# 특정 시나리오만 +fithia2 scenario-test --config return_max_long_v7.70 --scenario crash_v_recovery + +# 시나리오 그룹 +fithia2 scenario-test --config return_max_long_v7.70 --group structural +fithia2 scenario-test --config return_max_long_v7.70 --group signal +``` + +`--config`에는 실험 이름, ID(숫자), 파일 경로 모두 사용 가능. + +--- + +## 12개 사전 정의 시나리오 + +| 시나리오 | 시장 특성 | VIX | 신호강도 | 목적 | +|----------|----------|-----|---------|------| +| `steady_bull` | drift +15%, vol 14%, 252d | 14 | 0.35 | 기본 수익 환경 기준선 | +| `steady_bear` | drift -20%, vol 22%, 252d | 28 | 0.35 | 장기 하락 + Gate 0 동작 확인 | +| `crash_v_recovery` | 60d 정상 → 20d 폭락 → 172d 회복 | 15→45 | 0.35 | 폭락 복원력 | +| `prolonged_bear` | drift -15%, vol 25%, 504d | 30 | 0.30 | 역사보다 긴 2년 하락 | +| `low_vol_grind` | drift +8%, vol 8% | 11 | 0.20 | 저변동 환경 (ATR 축소) | +| `high_vol_chop` | drift 0%, vol 30% | 32 | 0.25 | 방향 없는 고변동 횡보 | +| `regime_switch` | 60d 상승/하락 4회 교대 | 20 | 0.30 | 빈번한 레짐 전환 | +| `vix_spike` | 정상 + 5회 VIX 급등(40+) | 18 | 0.35 | 스트레스 클러스터 | +| `no_signal` | drift 0%, vol 16%, 신호 없음 | 16 | **0.00** | 가격 패턴 과적합 탐지 | +| `strong_signal` | drift +10%, vol 16%, 강한 신호 | 16 | **0.60** | 이벤트 알파 포착 확인 | +| `sector_rotation` | 분기별 섹터 순환 | 20 | 0.35 | 섹터 집중 리스크 | +| `liquidity_drought` | drift +5%, vol 18%, 거래량 -70% | 22 | 0.30 | 유동성 필터 영향 | + +### 시나리오 그룹 + +| 그룹 | 포함 시나리오 | +|------|-------------| +| `trend` | steady_bull, steady_bear, prolonged_bear | +| `volatility` | low_vol_grind, high_vol_chop, vix_spike | +| `regime` | regime_switch, crash_v_recovery | +| `signal` | no_signal, strong_signal | +| `structural` | sector_rotation, liquidity_drought | +| `quick` | steady_bull, steady_bear, no_signal | +| `all` | 전체 12개 | + +--- + +## Regime Robustness Score (RRS) + +5개 하위 점수의 가중 합산으로 전략의 환경 내성을 0-100으로 표현한다. + +``` +RRS = 0.25 × Signal Integrity + + 0.25 × Breadth + + 0.20 × Drawdown Resilience + + 0.15 × Regime Transition + + 0.15 × Stability +``` + +| 하위 점수 | 가중치 | 계산 방법 | +|-----------|--------|----------| +| **Signal Integrity** | 25% | `no_signal` Sharpe ≤ 0이면 100. `no_signal` ≥ `strong_signal`이면 0. 선형 보간. | +| **Breadth** | 25% | Sharpe > 0인 시나리오 비율 × 100 | +| **Drawdown Resilience** | 20% | `100 × (1 - 최악_DD / 50)`, 0-100 범위 클램핑 | +| **Regime Transition** | 15% | `regime_switch` Sharpe / 중앙값 Sharpe, 100 상한 | +| **Stability** | 15% | `100 × (1 - CV)`, CV = Sharpe의 변동계수 | + +### 판정 기준 + +| RRS | 판정 | 의미 | +|-----|------|------| +| ≥ 70 | **ROBUST** | 다양한 환경에서 안정적 성과 | +| 40–69 | **FRAGILE** | 특정 환경 의존성 있음 | +| < 40 | **OVERFIT** | 가격 패턴 또는 특정 레짐에 과적합 | + +--- + +## 결과 해석 가이드 + +### v7.70 기준 시나리오별 해석 + +``` +시장 환경 Sharpe 운용 판단 +────────────────────────────────────────────── +steady_bull +1.09 ✅ 정상 운용, 풀 사이징 +crash+회복 +0.41 ✅ 폭락 후 회복 구간 운용 가능 +vix_spike +0.30 ✅ VIX 급등 이벤트, 소폭 수익 +prolonged_bear +0.13 ⚠️ 2년 하락장, flat → 사이징 축소 +steady_bear +0.03 ⚠️ 하락장, 간신히 손익분기 → 방어 모드 +────────────────────────────────────────────── +regime_switch -0.56 🔴 방향 빈번 전환 → 쉬어야 함 +low_vol_grind -0.40 🔴 변동성 너무 낮음 → 신호 희박 +high_vol_chop -1.11 🔴 방향 없는 고변동 → 매우 위험 +liquidity_drought -1.10 🔴 거래량 증발 → 진입 자체 불리 +sector_rotation -1.40 🔴 섹터 로테이션 → 최악 환경 +``` + +### 현재 시장 매핑 + +| 실시간 지표 | 해당 시나리오 | 운용 판단 | +|------------|--------------|---------| +| SPY 상승, VIX < 20 | steady_bull | 풀 운용 | +| SPY 횡보, VIX 20~30 | regime_switch | 사이징 50% | +| VIX 35+ 급등 | vix_spike | 소폭 or 대기 | +| SPY SMA 아래 지속 하락 | steady_bear | 방어, 최소 운용 | +| 거래량 급감 (ADV -40%+) | liquidity_drought | 일시 중단 | + +--- + +## 주요 Status 플래그 + +| 플래그 | 조건 | 의미 | +|--------|------|------| +| `OVERFIT` | `no_signal` Sharpe > `strong_signal` Sharpe | 이벤트 신호 없이도 수익 → 가격 패턴 과적합 | +| `WEAK` | `strong_signal` Sharpe < 0 | 강한 알파 환경에서도 손실 → 셀렉션 기준 불일치 | + +--- + +## 전략 비교 활용 + +```bash +# 취약 환경에서 새 전략이 개선됐는지 확인 +fithia2 scenario-test --config return_max_long_v11.x --group structural + +# sector_rotation / liquidity_drought 내성 개선 여부 비교 +# → 두 Sharpe 모두 -1.0 이하면 개선 없음, -0.5 이상이면 유의미한 개선 +``` + +--- + +## 구현 파일 + +| 파일 | 역할 | +|------|------| +| `libs/backtest/scenarios/price_gen.py` | Regime-switching GBM + 점프 확산 OHLCV 생성 | +| `libs/backtest/scenarios/macro_gen.py` | VIX/HY OU 프로세스, SPY/QQQ + 롤링 통계 | +| `libs/backtest/scenarios/event_gen.py` | 합성 이벤트 후보 전체 필드 생성 | +| `libs/backtest/scenarios/coupling.py` | 이벤트-가격 신호 커플링 (signal_strength 제어) | +| `libs/backtest/scenarios/store_builder.py` | SnapshotStore 조립 파이프라인 | +| `libs/backtest/scenarios/scenarios.py` | 12개 사전 정의 시나리오 + 레지스트리 | +| `libs/backtest/scenarios/robustness.py` | RRS 계산 + 판정 로직 | +| `apps/scenario/cli.py` | CLI 진입점, Rich 테이블 + RRS 패널 | +| `apps/tracker/cli.py` | `scenario-test` 서브커맨드 라우팅 | diff --git a/libs/backtest/scenarios/__init__.py b/libs/backtest/scenarios/__init__.py new file mode 100644 index 0000000..efaa3b0 --- /dev/null +++ b/libs/backtest/scenarios/__init__.py @@ -0,0 +1,38 @@ +"""Synthetic market scenario generation for overfitting detection and robustness testing. + +Public API: + from libs.backtest.scenarios import build_synthetic_store, SCENARIO_REGISTRY + from libs.backtest.scenarios.scenarios import ScenarioConfig, SCENARIO_REGISTRY + from libs.backtest.scenarios.robustness import run_scenario_test, RegimeRobustnessReport +""" +from libs.backtest.scenarios.price_gen import PriceRegime, generate_price_paths +from libs.backtest.scenarios.macro_gen import VIXConfig, HYSpreadConfig, generate_macro_data +from libs.backtest.scenarios.event_gen import EventDistribution, generate_events +from libs.backtest.scenarios.coupling import couple_events_to_prices +from libs.backtest.scenarios.scenarios import ScenarioConfig, SCENARIO_REGISTRY, SCENARIO_GROUPS +from libs.backtest.scenarios.store_builder import build_synthetic_store +from libs.backtest.scenarios.robustness import ( + ScenarioResult, + RegimeRobustnessReport, + run_scenario_test, + compute_rrs, +) + +__all__ = [ + "PriceRegime", + "generate_price_paths", + "VIXConfig", + "HYSpreadConfig", + "generate_macro_data", + "EventDistribution", + "generate_events", + "couple_events_to_prices", + "ScenarioConfig", + "SCENARIO_REGISTRY", + "SCENARIO_GROUPS", + "build_synthetic_store", + "ScenarioResult", + "RegimeRobustnessReport", + "run_scenario_test", + "compute_rrs", +] diff --git a/libs/backtest/scenarios/coupling.py b/libs/backtest/scenarios/coupling.py new file mode 100644 index 0000000..b18b7f5 --- /dev/null +++ b/libs/backtest/scenarios/coupling.py @@ -0,0 +1,156 @@ +"""Event-price coupling for synthetic scenario backtesting. + +Controls the signal-to-noise ratio between event quality features and +subsequent price movements. This is the core mechanism for overfitting detection: + + signal_strength=0.0 → pure noise → strategy should return ~0 (false positive check) + signal_strength=0.35 → realistic SNR → strategy captures genuine alpha + signal_strength=0.60 → strong signal → verify strategy responds to alpha + +The coupling injects a drift into bars AFTER the execution date based on the +event's score and reaction features, while preserving OHLCV consistency. +""" +from __future__ import annotations + +import datetime as dt +import math +from typing import Any + +import numpy as np + + +def couple_events_to_prices( + candidates: dict[dt.date, list[dict[str, Any]]], + bars_by_symbol: dict[str, dict[dt.date, dict[str, Any]]], + trading_dates: list[dt.date], + signal_strength: float, + signal_decay_days: int = 10, + false_positive_rate: float = 0.15, + rng: np.random.Generator | None = None, +) -> None: + """Inject signal-driven drift into bars after each event's execution date. + + Modifies bars_by_symbol in-place. Also updates each candidate's + entry_price / entry_price_est / event_close to match the actual + bar close on its reaction_date (so stop prices are consistent). + + Args: + candidates: candidates_by_exec_date dict from generate_events(). + bars_by_symbol: OHLCV bars to modify in-place. + trading_dates: Ordered list of NYSE trading dates. + signal_strength: 0.0 = pure noise, 0.35 = realistic, 0.6 = strong alpha. + signal_decay_days: Days over which signal drift decays to zero. + false_positive_rate: Fraction of qualifying events that produce negative + returns (traps / false positives). + rng: NumPy random generator. + """ + if rng is None: + rng = np.random.default_rng() + + date_to_idx = {d: i for i, d in enumerate(trading_dates)} + + for exec_date, rows in candidates.items(): + exec_idx = date_to_idx.get(exec_date) + if exec_idx is None: + continue + + # reaction_date is the day before execution + react_idx = exec_idx - 1 + if react_idx < 0: + continue + reaction_date = trading_dates[react_idx] + + for row in rows: + symbol = str(row.get("symbol", "")) + sym_bars = bars_by_symbol.get(symbol) + if sym_bars is None: + continue + + # Sync entry_price with actual bar close on reaction_date + react_bar = sym_bars.get(reaction_date) + if react_bar and react_bar.get("close", 0) > 0: + actual_close = float(react_bar["close"]) + row["entry_price"] = round(actual_close, 4) + row["entry_price_est"] = round(actual_close, 4) + row["event_close"] = round(actual_close, 4) + # Recompute ATR based on actual price + atr_pct = float(row.get("atr_14", actual_close * 0.022)) / max(float(row.get("entry_price", actual_close)), 1e-4) + row["atr_14"] = round(actual_close * atr_pct, 4) + + # Skip coupling if signal_strength == 0 (pure noise scenario) + if signal_strength <= 1e-9: + continue + + # Compute expected drift from event features + score = float(row.get("score", 0.5)) + reaction_return = float(row.get("reaction_day_return", 0.0)) + volume_ratio = float(row.get("volume_ratio_20d", 1.5)) + + expected_drift_5d = signal_strength * ( + 0.030 * (score - 0.5) + + 0.020 * reaction_return + + 0.008 * max(0.0, volume_ratio - 1.5) + ) + + # False positive: flip signal direction + if rng.random() < false_positive_rate: + expected_drift_5d = -expected_drift_5d * 0.7 + + if abs(expected_drift_5d) < 1e-6: + continue + + # Distribute drift over signal_decay_days using a decay schedule + total_drift = expected_drift_5d + decay = _compute_decay_schedule(total_drift, signal_decay_days) + + # Inject drift into bars starting at exec_date + 1 (first day we hold) + apply_start_idx = exec_idx + 1 + for k, daily_adj in enumerate(decay): + bar_idx = apply_start_idx + k + if bar_idx >= len(trading_dates): + break + bar_date = trading_dates[bar_idx] + bar = sym_bars.get(bar_date) + if bar is None: + continue + + _apply_drift_to_bar(bar, daily_adj) + + +def _compute_decay_schedule(total_drift: float, decay_days: int) -> list[float]: + """Distribute total_drift over decay_days using exponential decay. + + Returns a list of per-day drift adjustments that sum to total_drift. + """ + if decay_days <= 0: + return [total_drift] + + # Exponential decay weights + weights = [math.exp(-0.5 * k / max(decay_days, 1)) for k in range(decay_days)] + total_weight = sum(weights) + return [total_drift * w / total_weight for w in weights] + + +def _apply_drift_to_bar(bar: dict[str, Any], daily_drift: float) -> None: + """Multiply all OHLCV price fields by (1 + daily_drift), preserving consistency. + + Applies uniform multiplicative adjustment so that OHLC relationships are + maintained exactly. Volume is unchanged. + """ + factor = 1.0 + daily_drift + factor = max(0.5, min(2.0, factor)) # guard against extreme values + + for field in ("open", "high", "low", "close"): + val = bar.get(field) + if val is not None and float(val) > 0: + bar[field] = round(float(val) * factor, 4) + + # Ensure OHLCV consistency after adjustment + o = bar.get("open", 0) + h = bar.get("high", 0) + lo = bar.get("low", 0) + c = bar.get("close", 0) + + if o and h and lo and c: + bar["high"] = round(max(float(h), float(o), float(c)), 4) + bar["low"] = round(max(0.01, min(float(lo), float(o), float(c))), 4) diff --git a/libs/backtest/scenarios/event_gen.py b/libs/backtest/scenarios/event_gen.py new file mode 100644 index 0000000..7a4ecc8 --- /dev/null +++ b/libs/backtest/scenarios/event_gen.py @@ -0,0 +1,395 @@ +"""Synthetic event candidate generation for scenario backtesting. + +Generates event candidate rows that are compatible with SnapshotStore's +candidates_by_exec_date structure. Each row contains all fields required +by build_candidate() in selector.py plus feature fields used by scoring +functions and strategy engine filters. + +Event timing model (post_market/after_close pattern): + - event_date N: Company announces post-market → event_timestamp = N 21:00 UTC + - reaction_date: N (market's first reaction is on the announcement day close) + - execution_date: next_trading_day(N) — trade entered at next open +""" +from __future__ import annotations + +import datetime as dt +import math +from dataclasses import dataclass, field +from typing import Any + +import numpy as np + + +@dataclass +class EventDistribution: + """Statistical distribution parameters for synthetic event feature generation. + + Each feature is described as (mean, std) for truncated-normal sampling, + or as a dict mapping category → probability for categorical features. + All (mean, std) pairs use np.clip to keep values in reasonable ranges. + """ + + # ---- Event arrival ---- + events_per_day_mean: float = 8.0 + """Average number of events per trading day.""" + events_per_day_std: float = 3.0 + """Standard deviation of events per trading day.""" + + # ---- Categorical distributions ---- + event_types: dict[str, float] = field(default_factory=lambda: { + "earnings_release": 0.65, + "guidance_update": 0.15, + "material_contract": 0.08, + "other_material_event": 0.07, + "unknown": 0.05, + }) + event_directions: dict[str, float] = field(default_factory=lambda: { + "bullish": 0.45, + "mixed": 0.25, + "unknown": 0.20, + "bearish": 0.10, + }) + guidance_statuses: dict[str, float] = field(default_factory=lambda: { + "raised": 0.35, + "inline_or_maintained": 0.40, + "not_provided": 0.15, + "lowered": 0.10, + }) + filing_time_buckets: dict[str, float] = field(default_factory=lambda: { + "post_market": 0.60, + "pre_market": 0.35, + "intraday": 0.05, + }) + sectors: dict[str, float] = field(default_factory=lambda: { + "Technology": 0.22, + "Health Care": 0.14, + "Consumer Discretionary": 0.12, + "Financials": 0.13, + "Industrials": 0.11, + "Communication Services": 0.09, + "Consumer Staples": 0.07, + "Energy": 0.05, + "Materials": 0.04, + "Utilities": 0.03, + }) + + # ---- Quality features (scoring inputs) ---- + signal_strength_mean: float = 0.65 + signal_strength_std: float = 0.15 + document_quality_mean: float = 0.70 + document_quality_std: float = 0.12 + parse_confidence_mean: float = 0.80 + parse_confidence_std: float = 0.10 + guidance_direction_score_mean: float = 0.55 + guidance_direction_score_std: float = 0.20 + oneoff_penalty_prob: float = 0.05 + """Probability of a 1-off event penalty (reduces quality score).""" + + # ---- Reaction features ---- + reaction_return_mean: float = 0.025 + """Mean reaction-day return (positive = bullish bias).""" + reaction_return_std: float = 0.055 + volume_ratio_mean: float = 1.8 + volume_ratio_std: float = 0.9 + close_location_mean: float = 0.60 + close_location_std: float = 0.18 + gap_size_mean: float = 0.010 + gap_size_std: float = 0.025 + + # ---- Technical pre-event features ---- + rsi_14_mean: float = 52.0 + rsi_14_std: float = 12.0 + bb_position_mean: float = 0.55 + bb_position_std: float = 0.20 + volatility_20d_mean: float = 0.28 + volatility_20d_std: float = 0.08 + hurst_60d_mean: float = 0.50 + hurst_60d_std: float = 0.07 + entropy_60d_mean: float = 1.45 + entropy_60d_std: float = 0.15 + ou_theta_60d_mean: float = 5.0 + ou_theta_60d_std: float = 2.0 + market_temperature_mean: float = 0.80 + market_temperature_std: float = 0.25 + gravitational_pull_mean: float = 0.0 + gravitational_pull_std: float = 0.03 + sector_momentum_20d_mean: float = 0.005 + sector_momentum_20d_std: float = 0.03 + + # ---- Universe features ---- + avg_dollar_volume_mean: float = 5_000_000.0 + avg_dollar_volume_std: float = 3_000_000.0 + price_mean: float = 85.0 + price_std: float = 40.0 + + # ---- ATR ---- + atr_pct_mean: float = 0.022 + """ATR-14 as a percentage of price.""" + atr_pct_std: float = 0.008 + + +def _sample_categorical(categories: dict[str, float], rng: np.random.Generator) -> str: + """Sample one category weighted by probabilities.""" + keys = list(categories.keys()) + probs = np.array(list(categories.values()), dtype=float) + probs /= probs.sum() + return str(rng.choice(keys, p=probs)) + + +def _clamp_normal(mean: float, std: float, lo: float, hi: float, rng: np.random.Generator) -> float: + """Sample from a clipped normal distribution.""" + return float(np.clip(rng.normal(mean, std), lo, hi)) + + +def generate_events( + trading_dates: list[dt.date], + symbols: list[str], + dist: EventDistribution, + rng: np.random.Generator, + bars_by_symbol: dict[str, dict[dt.date, dict[str, Any]]] | None = None, + max_holding_days_buffer: int = 25, +) -> dict[dt.date, list[dict[str, Any]]]: + """Generate synthetic event candidates distributed across trading_dates. + + Each candidate row contains all fields required by: + - selector.build_candidate() (event_id, event_timestamp, entry_price, etc.) + - scoring functions (signal_strength_score, reaction_day_return, etc.) + - strategy engine filters (pre_event_rsi_14, pre_event_hurst_60d, etc.) + + Events are placed on execution_dates. The corresponding reaction_date is + the previous trading day (post_market filing pattern). + + Args: + trading_dates: Full sequence of NYSE trading dates. + symbols: List of ticker symbols to assign events to. + dist: EventDistribution parameters. + rng: NumPy random generator. + bars_by_symbol: If provided, entry_price is taken from bar close on reaction_date. + max_holding_days_buffer: Days at end of date range excluded from event placement + (so all positions can close before scenario end). + + Returns: + candidates_by_exec_date dict compatible with SnapshotStore. + """ + n = len(trading_dates) + # Reserve the first ~5 days (warm-up) and last N days (holding buffer) + eligible_range_start = 5 + eligible_range_end = max(eligible_range_start + 1, n - max_holding_days_buffer) + + candidates: dict[dt.date, list[dict[str, Any]]] = {} + event_counter = 0 + used_symbols_today: dict[dt.date, set[str]] = {} + + for i in range(eligible_range_start, eligible_range_end): + exec_date = trading_dates[i] + reaction_date = trading_dates[i - 1] # previous trading day + + # Number of events today (Poisson-like) + n_events = max(0, int(round(rng.normal(dist.events_per_day_mean, dist.events_per_day_std)))) + if n_events == 0: + continue + + today_candidates: list[dict[str, Any]] = [] + used_syms = used_symbols_today.setdefault(exec_date, set()) + + # Pick symbols for today's events (without replacement from pool) + available = [s for s in symbols if s not in used_syms] + if not available: + continue + rng.shuffle(available) + n_events = min(n_events, len(available)) + + for j in range(n_events): + symbol = available[j] + used_syms.add(symbol) + event_counter += 1 + + event_type = _sample_categorical(dist.event_types, rng) + event_direction = _sample_categorical(dist.event_directions, rng) + guidance_status = _sample_categorical(dist.guidance_statuses, rng) + filing_bucket = _sample_categorical(dist.filing_time_buckets, rng) + sector = _sample_categorical(dist.sectors, rng) + + # Event date / timestamp: model after-close and same-day patterns. + # post_market / pre_market → event happened on the trading day BEFORE + # reaction_date (after-close pattern: reaction_date > event_date → "after_close"). + # intraday → event happened on reaction_date itself ("same_day"). + if filing_bucket in ("post_market", "pre_market"): + event_date_d = trading_dates[i - 2] # i >= eligible_range_start=5, safe + hour = "21:00:00" if filing_bucket == "post_market" else "07:00:00" + else: + event_date_d = reaction_date # intraday → same_day timing + hour = "14:00:00" + event_timestamp = f"{event_date_d.isoformat()}T{hour}+00:00" + + # Price (try to get from bars, else sample) + if bars_by_symbol and symbol in bars_by_symbol: + bar = bars_by_symbol[symbol].get(reaction_date) + if bar and bar.get("close", 0) > 0: + entry_price = float(bar["close"]) + event_close = entry_price + atr_14 = entry_price * _clamp_normal(dist.atr_pct_mean, dist.atr_pct_std, 0.005, 0.08, rng) + exec_bar = bars_by_symbol[symbol].get(exec_date) + avg_dollar_volume = float(bar.get("volume", 1_000_000)) * entry_price + else: + entry_price = max(5.0, _clamp_normal(dist.price_mean, dist.price_std, 5.0, 500.0, rng)) + event_close = entry_price + atr_14 = entry_price * _clamp_normal(dist.atr_pct_mean, dist.atr_pct_std, 0.005, 0.08, rng) + avg_dollar_volume = max(100_000.0, rng.normal(dist.avg_dollar_volume_mean, dist.avg_dollar_volume_std)) + else: + entry_price = max(5.0, _clamp_normal(dist.price_mean, dist.price_std, 5.0, 500.0, rng)) + event_close = entry_price + atr_14 = entry_price * _clamp_normal(dist.atr_pct_mean, dist.atr_pct_std, 0.005, 0.08, rng) + avg_dollar_volume = max(100_000.0, float(rng.normal(dist.avg_dollar_volume_mean, dist.avg_dollar_volume_std))) + + # Quality features + signal_strength = _clamp_normal(dist.signal_strength_mean, dist.signal_strength_std, 0.0, 1.0, rng) + doc_quality = _clamp_normal(dist.document_quality_mean, dist.document_quality_std, 0.0, 1.0, rng) + parse_conf = _clamp_normal(dist.parse_confidence_mean, dist.parse_confidence_std, 0.0, 1.0, rng) + guidance_dir_score = _clamp_normal(dist.guidance_direction_score_mean, dist.guidance_direction_score_std, 0.0, 1.0, rng) + oneoff_penalty = 1.0 if rng.random() < dist.oneoff_penalty_prob else 0.0 + + # Reaction features (biased by event_direction) + direction_bias = {"bullish": 0.03, "bearish": -0.03, "mixed": 0.005, "unknown": 0.0}.get(event_direction, 0.0) + reaction_return = float(rng.normal(dist.reaction_return_mean + direction_bias, dist.reaction_return_std)) + volume_ratio = max(0.5, float(rng.normal(dist.volume_ratio_mean, dist.volume_ratio_std))) + close_location = _clamp_normal(dist.close_location_mean, dist.close_location_std, 0.0, 1.0, rng) + gap_size = float(rng.normal(dist.gap_size_mean, dist.gap_size_std)) + + # Technical features + rsi_14 = _clamp_normal(dist.rsi_14_mean, dist.rsi_14_std, 5.0, 95.0, rng) + bb_position = _clamp_normal(dist.bb_position_mean, dist.bb_position_std, -0.2, 1.2, rng) + vol_20d = _clamp_normal(dist.volatility_20d_mean, dist.volatility_20d_std, 0.05, 0.8, rng) + hurst_60d = _clamp_normal(dist.hurst_60d_mean, dist.hurst_60d_std, 0.2, 0.8, rng) + entropy_60d = _clamp_normal(dist.entropy_60d_mean, dist.entropy_60d_std, 0.5, 2.0, rng) + ou_theta = _clamp_normal(dist.ou_theta_60d_mean, dist.ou_theta_60d_std, 0.5, 30.0, rng) + mkt_temp = _clamp_normal(dist.market_temperature_mean, dist.market_temperature_std, 0.0, 2.0, rng) + grav_pull = float(rng.normal(dist.gravitational_pull_mean, dist.gravitational_pull_std)) + sector_mom = float(rng.normal(dist.sector_momentum_20d_mean, dist.sector_momentum_20d_std)) + + # Compute score using the same logic as the real scoring system + row_for_scoring: dict[str, Any] = { + "event_type": event_type, + "event_direction": event_direction, + "guidance_status": guidance_status, + "signal_strength_score": signal_strength, + "document_quality_score": doc_quality, + "parse_confidence_overall": parse_conf, + "guidance_direction_score": guidance_dir_score, + "oneoff_penalty": oneoff_penalty, + "reaction_day_return": reaction_return, + "volume_ratio_20d": volume_ratio, + "close_location": close_location, + "gap_size": gap_size, + "pre_event_entropy_60d": entropy_60d, + } + score = _compute_synthetic_score(row_for_scoring) + + row: dict[str, Any] = { + # Identity + "event_id": f"SYNTH::{symbol}::{event_counter:06d}", + "symbol": symbol, + "event_type": event_type, + "event_direction": event_direction, + "guidance_status": guidance_status, + "filing_time_bucket": filing_bucket, + "sector": sector, + # Dates and timestamps + "event_date": event_date_d.isoformat(), + "event_timestamp": event_timestamp, + "reaction_date": reaction_date.isoformat(), + "entry_date": exec_date.isoformat(), + "execution_date": exec_date, + # Pricing + "entry_price": round(entry_price, 4), + "entry_price_est": round(entry_price, 4), + "event_close": round(event_close, 4), + "atr_14": round(atr_14, 4), + "avg_dollar_volume": round(avg_dollar_volume, 2), + "avg_dollar_volume_20d": round(avg_dollar_volume, 2), + "market_cap_proxy": round(max(avg_dollar_volume * 400, 3_000_000_000), 0), + # Score + "score": round(score, 4), + # Quality features + "signal_strength_score": round(signal_strength, 4), + "document_quality_score": round(doc_quality, 4), + "parse_confidence_overall": round(parse_conf, 4), + "guidance_direction_score": round(guidance_dir_score, 4), + "oneoff_penalty": oneoff_penalty, + # Reaction features + "reaction_day_return": round(reaction_return, 4), + "volume_ratio_20d": round(volume_ratio, 4), + "close_location": round(close_location, 4), + "gap_size": round(gap_size, 4), + "reaction_day_low": round(entry_price * (1.0 - abs(reaction_return) * 0.5), 4), + "reaction_day_high": round(entry_price * (1.0 + abs(reaction_return) * 0.5), 4), + "reaction_day_range_pct": round(abs(reaction_return) + abs(gap_size), 4), + "upper_wick_pct": round(max(0.0, float(rng.exponential(0.01))), 4), + # Technical pre-event features + "pre_event_rsi_14": round(rsi_14, 2), + "pre_event_bb_position": round(bb_position, 4), + "pre_event_volatility_20d": round(vol_20d, 4), + "pre_event_obv_slope_20d": float(rng.normal(0, 0.1)), + "pre_event_hurst_60d": round(hurst_60d, 4), + "pre_event_entropy_60d": round(entropy_60d, 4), + "pre_event_short_ratio": max(0.0, float(rng.exponential(0.05))), + "pre_event_sector_momentum_20d": round(sector_mom, 4), + "pre_event_ou_theta_60d": round(ou_theta, 4), + "pre_event_gravitational_pull": round(grav_pull, 4), + "pre_event_market_temperature": round(mkt_temp, 4), + # Fundamental (optional, not always present) + "reported_eps": None, + "estimated_eps": None, + "earnings_beat": None, + "earnings_surprise_pct": None, + # Universe + "exchange_proxy": "NASDAQ" if rng.random() > 0.4 else "NYSE", + "asset_type_proxy": "stock", + } + + today_candidates.append(row) + + if today_candidates: + candidates[exec_date] = today_candidates + + return candidates + + +def _compute_synthetic_score(row: dict[str, Any]) -> float: + """Compute a realistic score using the same weighting as compute_entry_score (v5 base). + + Simplified version that does not require the full scoring module imports, + matching the 3-component structure: event_quality (65%) + reaction (20%) + volume (15%). + """ + # Event quality component (65%) + doc_q = float(row.get("document_quality_score") or 0.65) + sig_s = float(row.get("signal_strength_score") or 0.60) + parse_c = float(row.get("parse_confidence_overall") or 0.75) + guidance = float(row.get("guidance_direction_score") or 0.50) + oneoff = float(row.get("oneoff_penalty") or 0.0) + + base_quality = (doc_q * 0.35 + sig_s * 0.30 + parse_c * 0.20 + guidance * 0.15) + if oneoff: + base_quality *= 0.60 + + # Reaction direction component (20%) + reaction_return = float(row.get("reaction_day_return") or 0.0) + if reaction_return > 0.01: + reaction_score = min(1.0, 0.5 + reaction_return * 5.0) + elif reaction_return < -0.01: + reaction_score = max(0.0, 0.5 + reaction_return * 5.0) + else: + reaction_score = 0.5 + + # Volume conviction component (15%) + vol_ratio = float(row.get("volume_ratio_20d") or 1.0) + volume_score = min(1.0, vol_ratio / 3.0) + + # Low entropy bonus (v13e entropy feature) + entropy = row.get("pre_event_entropy_60d") + entropy_bonus = 0.0 + if entropy is not None and float(entropy) < 1.2: + entropy_bonus = 0.03 * (1.2 - float(entropy)) + + raw = base_quality * 0.65 + reaction_score * 0.20 + volume_score * 0.15 + entropy_bonus + return max(0.0, min(1.0, raw)) diff --git a/libs/backtest/scenarios/macro_gen.py b/libs/backtest/scenarios/macro_gen.py new file mode 100644 index 0000000..21c6cd6 --- /dev/null +++ b/libs/backtest/scenarios/macro_gen.py @@ -0,0 +1,456 @@ +"""Synthetic macro data generation for scenario backtesting. + +Generates the full macro_by_date dict consumed by SnapshotStore, including: + - SPY/QQQ rolling indicators (SMA, momentum, vol, entropy, Hurst, etc.) + - VIX via Ornstein-Uhlenbeck process correlated with market returns + - HY credit spread via OU correlated with VIX + - Proxy ETF close prices (TQQQ, QQQM, SPYM, SGOV) + +The rolling indicators exactly mirror SnapshotStore._fetch_spy_macro._merge_series() +so that allocator Gates (macro regime, VIX scaler, parking momentum) behave correctly. +""" +from __future__ import annotations + +import datetime as dt +import math +from dataclasses import dataclass +from typing import Any + +import numpy as np + + +@dataclass +class VIXConfig: + """Parameters for the VIX Ornstein-Uhlenbeck simulation.""" + + base_level: float = 18.0 + """Long-run mean (θ). Bull ~14, normal ~18, bear ~25, crash ~35+.""" + mean_reversion: float = 5.0 + """Speed of reversion (κ, annualized). Higher = faster mean-reversion.""" + volatility: float = 5.0 + """VIX process volatility (σ, per year).""" + market_corr: float = -0.7 + """Correlation between VIX changes and market log-returns (negative).""" + + +@dataclass +class HYSpreadConfig: + """Parameters for the HY credit spread OU simulation.""" + + base_level: float = 4.0 + """Long-run mean spread in percentage points.""" + mean_reversion: float = 3.0 + """Speed of reversion (annualized).""" + volatility: float = 1.5 + """Spread process volatility (per year).""" + vix_corr: float = 0.6 + """Correlation with VIX level changes.""" + + +# --------------------------------------------------------------------------- +# Internal: rolling indicator computation (mirrors _merge_series in snapshot_store.py) +# --------------------------------------------------------------------------- + +_SMA_PERIODS = (10, 20, 30, 40, 50) +_ROLLING_HIGH_PERIODS = (20, 50, 100) +_MOMENTUM_PERIODS = (5, 10, 20, 50) +_VOL_PERIODS = (15, 20, 30, 50) +_EFFICIENCY_PERIODS = (10, 20) +_DOWNSIDE_VOL_PERIODS = (10, 20) +_ULCER_PERIODS = (10, 20) +_ENTROPY_PERIODS = (10, 20) +_DRAWDOWN_ACCEL_DAYS = 5 + + +def _compute_rolling_indicators( + prefix: str, + bars: dict[dt.date, dict[str, Any]], + result: dict[dt.date, dict[str, Any]], +) -> None: + """Compute all rolling macro indicators for a price series prefix (spy, qqq, etc.). + + Fills result[date][f"{prefix}_*"] in-place, mirroring exactly the keys + produced by SnapshotStore._fetch_spy_macro._merge_series(). + """ + sorted_dates = sorted(bars.keys()) + closes: list[tuple[dt.date, float]] = [ + (d, float(bars[d]["close"])) for d in sorted_dates + ] + + for i, (date, close) in enumerate(closes): + result.setdefault(date, {}) + bar = bars[date] + result[date][f"{prefix}_close"] = close + result[date][f"{prefix}_open"] = float(bar.get("open", close)) + result[date][f"{prefix}_high"] = float(bar.get("high", close)) + result[date][f"{prefix}_low"] = float(bar.get("low", close)) + result[date][f"{prefix}_volume"] = float(bar.get("volume", 0)) + + # ---- SMAs ---- + for period in _SMA_PERIODS: + sma = None + if i >= period - 1: + window = [c for _, c in closes[i - period + 1 : i + 1]] + sma = sum(window) / len(window) + result[date][f"{prefix}_sma_{period}"] = sma + + # ---- Rolling highs (for drawdown gates) ---- + for rh_p in _ROLLING_HIGH_PERIODS: + rh = None + if i >= rh_p - 1: + rh = max(c for _, c in closes[i - rh_p + 1 : i + 1]) + result[date][f"{prefix}_high_{rh_p}"] = rh + + # ---- Momentum / N-day return ---- + for mom_p in _MOMENTUM_PERIODS: + mom = None + if i >= mom_p: + prev = closes[i - mom_p][1] + if prev > 0: + mom = (close - prev) / prev + result[date][f"{prefix}_mom_{mom_p}"] = mom + + # ---- Realized volatility (annualised std of log returns) ---- + for vol_p in _VOL_PERIODS: + vol = None + if i >= vol_p: + log_rets = [ + math.log(closes[j][1] / closes[j - 1][1]) + for j in range(i - vol_p + 1, i + 1) + if closes[j - 1][1] > 0 + ] + if len(log_rets) >= vol_p - 1: + mean_r = sum(log_rets) / len(log_rets) + var_r = sum((r - mean_r) ** 2 for r in log_rets) / len(log_rets) + vol = math.sqrt(var_r * 252) + result[date][f"{prefix}_vol_{vol_p}"] = vol + + # ---- Efficiency ratio (Kaufman) ---- + for eff_p in _EFFICIENCY_PERIODS: + efficiency = None + if i >= eff_p: + net = abs(close - closes[i - eff_p][1]) + gross = sum( + abs(closes[j][1] - closes[j - 1][1]) + for j in range(i - eff_p + 1, i + 1) + ) + efficiency = net / gross if gross > 0 else 0.0 + result[date][f"{prefix}_efficiency_{eff_p}"] = efficiency + + # ---- Downside semi-volatility ---- + for dv_p in _DOWNSIDE_VOL_PERIODS: + downside_vol = None + if i >= dv_p: + neg_sq = [ + min(closes[j][1] / closes[j - 1][1] - 1, 0.0) ** 2 + for j in range(i - dv_p + 1, i + 1) + if closes[j - 1][1] > 0 + ] + if len(neg_sq) >= dv_p - 1: + downside_vol = math.sqrt(sum(neg_sq) / len(neg_sq) * 252) + result[date][f"{prefix}_downside_vol_{dv_p}"] = downside_vol + + # ---- Shannon entropy (market predictability) ---- + for ent_p in _ENTROPY_PERIODS: + entropy = None + if i >= ent_p: + daily_rets = [ + closes[j][1] / closes[j - 1][1] - 1 + for j in range(i - ent_p + 1, i + 1) + if closes[j - 1][1] > 0 + ] + if len(daily_rets) >= ent_p - 1: + n_pos = sum(1 for r in daily_rets if r > 0.001) + n_neg = sum(1 for r in daily_rets if r < -0.001) + n_flat = len(daily_rets) - n_pos - n_neg + entropy = 0.0 + for cnt in (n_pos, n_neg, n_flat): + if cnt > 0: + p = cnt / len(daily_rets) + entropy -= p * math.log2(p) + result[date][f"{prefix}_entropy_{ent_p}"] = entropy + + # ---- Ulcer index + current drawdown ---- + for ulcer_p in _ULCER_PERIODS: + ulcer = None + current_dd = None + if i >= ulcer_p - 1: + window_closes = [c for _, c in closes[i - ulcer_p + 1 : i + 1]] + peak = 0.0 + drawdowns: list[float] = [] + for wc in window_closes: + peak = max(peak, wc) + if peak > 0: + drawdowns.append(wc / peak - 1.0) + if drawdowns: + ulcer = math.sqrt(sum(dd * dd for dd in drawdowns) / len(drawdowns)) + current_dd = abs(drawdowns[-1]) + result[date][f"{prefix}_ulcer_{ulcer_p}"] = ulcer + result[date][f"{prefix}_drawdown_{ulcer_p}"] = current_dd + + # ---- Drawdown acceleration ---- + dd_lb = 20 + dd_accel = None + if i >= dd_lb - 1 + _DRAWDOWN_ACCEL_DAYS: + cur_window = [c for _, c in closes[i - dd_lb + 1 : i + 1]] + prev_i = i - _DRAWDOWN_ACCEL_DAYS + prev_window = [c for _, c in closes[prev_i - dd_lb + 1 : prev_i + 1]] + cur_peak = max(cur_window) if cur_window else 0.0 + prev_peak = max(prev_window) if prev_window else 0.0 + if cur_peak > 0 and prev_peak > 0: + cur_dd = (cur_peak - close) / cur_peak + prev_dd = (prev_peak - closes[prev_i][1]) / prev_peak + dd_accel = cur_dd - prev_dd + result[date][f"{prefix}_drawdown_accel_{_DRAWDOWN_ACCEL_DAYS}"] = dd_accel + + # ---- Hurst exponent (R/S analysis) ---- + hurst_lookback = 60 + hurst = None + if i >= hurst_lookback + 1: + h_rets = [ + (closes[j + 1][1] - closes[j][1]) / closes[j][1] + for j in range(i - hurst_lookback, i) + if closes[j][1] > 0 + ] + if len(h_rets) >= 30: + def _rs(series: list[float]) -> float: + n_ = len(series) + m_ = sum(series) / n_ + devs = [x - m_ for x in series] + cum, s_ = [], 0.0 + for d_ in devs: + s_ += d_ + cum.append(s_) + r_ = max(cum) - min(cum) + std_ = (sum(d_ ** 2 for d_ in devs) / n_) ** 0.5 + return r_ / std_ if std_ > 0 else 0.0 + + win_sizes = [s for s in [8, 12, 16, 24, 32] if s <= len(h_rets) // 2] + if len(win_sizes) >= 2: + log_n, log_rs = [], [] + for w in win_sizes: + chunks = [h_rets[st: st + w] for st in range(0, len(h_rets) - w + 1, w) if len(h_rets[st: st + w]) == w] + rs_vals = [_rs(c) for c in chunks] + if rs_vals: + avg_rs = sum(rs_vals) / len(rs_vals) + if avg_rs > 0: + log_n.append(math.log(w)) + log_rs.append(math.log(avg_rs)) + if len(log_n) >= 2: + n_h = len(log_n) + xm = sum(log_n) / n_h + ym = sum(log_rs) / n_h + num = sum((log_n[k] - xm) * (log_rs[k] - ym) for k in range(n_h)) + den = sum((log_n[k] - xm) ** 2 for k in range(n_h)) + hurst = num / den if den > 0 else 0.5 + result[date][f"{prefix}_hurst_60"] = hurst + + # ---- Lag-1 autocorrelation (Lo, 2004) ---- + ac_lb = 20 + autocorr = None + if i >= ac_lb + 1: + ac_rets = [ + closes[j][1] / closes[j - 1][1] - 1 + for j in range(i - ac_lb, i + 1) + if closes[j - 1][1] > 0 + ] + if len(ac_rets) >= ac_lb: + x_ac, y_ac = ac_rets[:-1], ac_rets[1:] + n_ac = len(x_ac) + mx, my = sum(x_ac) / n_ac, sum(y_ac) / n_ac + cov_xy = sum((x_ac[k] - mx) * (y_ac[k] - my) for k in range(n_ac)) / n_ac + sx = (sum((x_ac[k] - mx) ** 2 for k in range(n_ac)) / n_ac) ** 0.5 + sy = (sum((y_ac[k] - my) ** 2 for k in range(n_ac)) / n_ac) ** 0.5 + if sx > 1e-12 and sy > 1e-12: + autocorr = cov_xy / (sx * sy) + result[date][f"{prefix}_autocorr_20"] = autocorr + + +def _compute_pair_correlation( + left_prefix: str, + right_prefix: str, + output_key: str, + result: dict[dt.date, dict[str, Any]], +) -> None: + """Compute rolling 20-day cross-asset correlation (mirrors snapshot_store logic).""" + corr_lb = 20 + sorted_dates = sorted(result.keys()) + for idx_c, d_c in enumerate(sorted_dates): + corr_val = None + if idx_c >= corr_lb: + left_r, right_r = [], [] + for jj in range(idx_c - corr_lb + 1, idx_c + 1): + d_j = sorted_dates[jj] + d_prev = sorted_dates[jj - 1] + lc = result.get(d_j, {}).get(f"{left_prefix}_close") + lp = result.get(d_prev, {}).get(f"{left_prefix}_close") + rc = result.get(d_j, {}).get(f"{right_prefix}_close") + rp = result.get(d_prev, {}).get(f"{right_prefix}_close") + if all(v and v > 0 for v in [lc, lp, rc, rp]): + left_r.append(lc / lp - 1) + right_r.append(rc / rp - 1) + if len(left_r) >= corr_lb - 2: + n_cr = len(left_r) + ml = sum(left_r) / n_cr + mr = sum(right_r) / n_cr + cov = sum((left_r[k] - ml) * (right_r[k] - mr) for k in range(n_cr)) / n_cr + sl = (sum((left_r[k] - ml) ** 2 for k in range(n_cr)) / n_cr) ** 0.5 + sr = (sum((right_r[k] - mr) ** 2 for k in range(n_cr)) / n_cr) ** 0.5 + if sl > 1e-12 and sr > 1e-12: + corr_val = cov / (sl * sr) + result[d_c][output_key] = corr_val + + +# --------------------------------------------------------------------------- +# VIX and HY spread simulation +# --------------------------------------------------------------------------- + +def _generate_vix_series( + market_log_rets: list[float], + trading_dates: list[dt.date], + config: VIXConfig, + rng: np.random.Generator, +) -> dict[dt.date, float]: + """Generate VIX time series via OU process correlated with market returns. + + dVIX = κ(θ-VIX)dt + σ·dW where dW is partially driven by market return sign. + """ + dt_step = 1.0 / 252 + vix = config.base_level + vix_series: dict[dt.date, float] = {} + + for date, mkt_lr in zip(trading_dates, market_log_rets): + # Approximate market z-score from log return + market_daily_std = config.base_level / 100.0 * math.sqrt(dt_step) + 1e-8 + mkt_z = mkt_lr / market_daily_std + z_idio = rng.standard_normal() + rho = config.market_corr + # Negative market → positive VIX shock + z_combined = rho * (-mkt_z) + math.sqrt(max(0.0, 1 - rho ** 2)) * z_idio + + kappa = config.mean_reversion + theta = config.base_level + sigma = config.volatility + + dVIX = kappa * (theta - vix) * dt_step + sigma * math.sqrt(dt_step) * z_combined + vix = max(5.0, min(90.0, vix + dVIX)) + vix_series[date] = round(vix, 2) + + return vix_series + + +def _generate_hy_series( + vix_series: dict[dt.date, float], + trading_dates: list[dt.date], + config: HYSpreadConfig, + rng: np.random.Generator, +) -> dict[dt.date, float]: + """Generate HY credit spread via OU process correlated with VIX changes.""" + dt_step = 1.0 / 252 + spread = config.base_level + hy_series: dict[dt.date, float] = {} + prev_vix = config.base_level + + for date in trading_dates: + cur_vix = vix_series.get(date, config.base_level) + vix_chg = (cur_vix - prev_vix) / (prev_vix + 1e-8) + + z_idio = rng.standard_normal() + rho = config.vix_corr + z_combined = rho * vix_chg * 5.0 + math.sqrt(max(0.0, 1 - rho ** 2)) * z_idio + + kappa = config.mean_reversion + theta = config.base_level + sigma = config.volatility + + dSpread = kappa * (theta - spread) * dt_step + sigma * math.sqrt(dt_step) * z_combined + spread = max(1.5, min(30.0, spread + dSpread)) + hy_series[date] = round(spread, 3) + prev_vix = cur_vix + + return hy_series + + +# --------------------------------------------------------------------------- +# Main entry point +# --------------------------------------------------------------------------- + +def generate_macro_data( + spy_bars: dict[dt.date, dict[str, Any]], + qqq_bars: dict[dt.date, dict[str, Any]], + vix_config: VIXConfig, + hy_config: HYSpreadConfig, + trading_dates: list[dt.date], + rng: np.random.Generator, + market_log_rets: list[float] | None = None, +) -> dict[dt.date, dict[str, Any]]: + """Generate complete macro_by_date dict for SnapshotStore. + + Includes: + - SPY/QQQ rolling indicators (matches _merge_series keys exactly) + - spy_qqq_corr_20 cross-correlation + - VIXCLS, macro_vix, macro_hy_spread + - Parking ETF proxies: tqqq_close, qqqm_close, spym_close, sgov_close + + Args: + spy_bars: SPY OHLCV bars {date: {open, high, low, close, volume}}. + qqq_bars: QQQ OHLCV bars. + vix_config: VIX OU simulation parameters. + hy_config: HY spread OU simulation parameters. + trading_dates: Ordered list of NYSE trading dates. + rng: NumPy random generator. + market_log_rets: Optional market log-return series for VIX correlation. + + Returns: + macro_by_date dict compatible with SnapshotStore. + """ + result: dict[dt.date, dict[str, Any]] = {} + + # Rolling indicators for SPY and QQQ + _compute_rolling_indicators("spy", spy_bars, result) + _compute_rolling_indicators("qqq", qqq_bars, result) + + # SPY-QQQ cross correlation (regime shift detection) + _compute_pair_correlation("spy", "qqq", "spy_qqq_corr_20", result) + + # Generate VIX and HY spread + if market_log_rets is None: + market_log_rets = [0.0] * len(trading_dates) + + vix_series = _generate_vix_series(market_log_rets, trading_dates, vix_config, rng) + hy_series = _generate_hy_series(vix_series, trading_dates, hy_config, rng) + + # Inject VIX, HY, and parking ETF proxies + sgov_price = 100.0 + sgov_daily_yield = 0.0525 / 252 # ~5.25% money market yield + + for i, date in enumerate(trading_dates): + result.setdefault(date, {}) + + vix = vix_series.get(date) + if vix is not None: + result[date]["VIXCLS"] = vix + result[date]["macro_vix"] = vix + + hy = hy_series.get(date) + if hy is not None: + result[date]["macro_hy_spread"] = hy + + qqq_close = result[date].get("qqq_close") + spy_close = result[date].get("spy_close") + + if qqq_close is not None and qqq_close > 0: + # TQQQ ≈ 3x leveraged QQQ (simplified proxy) + result[date]["tqqq_close"] = round(qqq_close * 3.0 / 415.0 * 55.0, 4) + # QQQM ≈ QQQ (slightly lower price, same ETF) + result[date]["qqqm_close"] = round(qqq_close * 0.99, 4) + + if spy_close is not None and spy_close > 0: + # SPYM ≈ SPY (2x leveraged; simplified as 1.9× SPY level / 480 × 95) + result[date]["spym_close"] = round(spy_close * 0.98, 4) + + # SGOV ≈ T-bill ETF with daily accrual + sgov_price *= (1 + sgov_daily_yield) + result[date]["sgov_close"] = round(sgov_price, 4) + + return result diff --git a/libs/backtest/scenarios/price_gen.py b/libs/backtest/scenarios/price_gen.py new file mode 100644 index 0000000..5b848d6 --- /dev/null +++ b/libs/backtest/scenarios/price_gen.py @@ -0,0 +1,242 @@ +"""Synthetic OHLCV price path generation for scenario backtesting. + +Implements Regime-Switching GBM with optional jump-diffusion (Merton model). +Individual stock paths are generated as market factor + beta + idiosyncratic noise +to produce realistic cross-sectional correlations. +""" +from __future__ import annotations + +import datetime as dt +import math +from dataclasses import dataclass, field +from typing import Any + +import numpy as np + + +@dataclass +class PriceRegime: + """Parameters for a single market regime segment.""" + + annualized_drift: float + """Expected annual return, e.g. 0.15 for bull, -0.20 for bear.""" + annualized_vol: float + """Annual volatility, e.g. 0.15 for calm, 0.35 for stressed.""" + duration_days: int + """Number of trading days this regime lasts.""" + jump_prob: float = 0.0 + """Per-day probability of a Merton jump event.""" + jump_mean: float = 0.0 + """Mean log-jump size. Negative for crash-type regimes.""" + jump_std: float = 0.02 + """Standard deviation of log-jump size.""" + + +def _generate_regime_log_returns( + regimes: list[PriceRegime], + n_days: int, + rng: np.random.Generator, +) -> list[float]: + """Generate n_days market-level daily log returns following the regime sequence.""" + log_rets: list[float] = [] + dt_step = 1.0 / 252 + + for regime in regimes: + drift = (regime.annualized_drift - 0.5 * regime.annualized_vol ** 2) * dt_step + diffusion = regime.annualized_vol * math.sqrt(dt_step) + days_this_regime = min(regime.duration_days, n_days - len(log_rets)) + + for _ in range(days_this_regime): + lr = drift + diffusion * rng.standard_normal() + if regime.jump_prob > 0 and rng.random() < regime.jump_prob: + lr += rng.normal(regime.jump_mean, max(regime.jump_std, 1e-6)) + log_rets.append(lr) + + # Extend with last regime if regimes run short + last = regimes[-1] + drift = (last.annualized_drift - 0.5 * last.annualized_vol ** 2) * dt_step + diffusion = last.annualized_vol * math.sqrt(dt_step) + while len(log_rets) < n_days: + log_rets.append(drift + diffusion * rng.standard_normal()) + + return log_rets[:n_days] + + +def _bars_from_log_returns( + log_rets: list[float], + trading_dates: list[dt.date], + initial_price: float, + intraday_vol_scale: float, + base_volume: int, + rng: np.random.Generator, +) -> dict[dt.date, dict[str, Any]]: + """Build OHLCV bars from a sequence of daily log-returns. + + Ensures OHLCV consistency: low <= min(open, close), high >= max(open, close). + Volume is log-normally distributed and positively correlated with |return|. + """ + bars: dict[dt.date, dict[str, Any]] = {} + prev_close = initial_price + + for date, lr in zip(trading_dates, log_rets): + close = max(prev_close * math.exp(lr), 0.01) + + # Open: prev_close * small gap (mean-zero noise) + gap = rng.normal(0, intraday_vol_scale * 0.5) + open_ = max(prev_close * math.exp(gap), 0.01) + + # Intraday range around open/close extremes + intraday_noise = abs(rng.normal(0, intraday_vol_scale * 0.7)) + hi_raw = max(open_, close) * (1.0 + intraday_noise) + lo_raw = min(open_, close) * max(1.0 - intraday_noise, 0.001) + + high = max(hi_raw, open_, close) + low = min(lo_raw, open_, close) + low = max(low, 0.01) + + # Volume: log-normal, amplified by absolute return + vol_factor = 1.0 + 3.0 * abs(math.exp(lr) - 1) + volume = max(1000, int(rng.lognormal(math.log(base_volume), 0.4) * vol_factor)) + + bars[date] = { + "date": date, + "open": round(open_, 4), + "high": round(high, 4), + "low": round(low, 4), + "close": round(close, 4), + "volume": volume, + } + prev_close = close + + return bars + + +def generate_price_paths( + n_symbols: int, + initial_prices: list[float] | None, + regimes: list[PriceRegime], + trading_dates: list[dt.date], + market_beta_range: tuple[float, float] = (0.6, 1.2), + rng: np.random.Generator | None = None, + tickers: list[str] | None = None, + return_market_log_rets: bool = False, +) -> ( + tuple[dict[str, dict[dt.date, dict[str, Any]]], list[float]] + | dict[str, dict[dt.date, dict[str, Any]]] +): + """Generate correlated OHLCV bars for n_symbols stocks. + + Each stock has a random beta to a shared market factor plus idiosyncratic noise. + Returns bars_by_symbol_date dict compatible with SnapshotStore. + + Args: + n_symbols: Number of synthetic stocks to generate. + initial_prices: Optional list of starting prices (defaults to random $20-$200). + regimes: Sequence of PriceRegime objects defining the market environment. + trading_dates: Ordered list of NYSE trading dates (from calendar.get_trading_days). + market_beta_range: (min, max) range for individual stock betas. + rng: NumPy random generator (seeded for reproducibility). + tickers: Optional list of ticker symbols (auto-generated if None). + return_market_log_rets: If True, also return the market log-return series. + + Returns: + bars_by_symbol_date dict, or (dict, market_log_rets) if return_market_log_rets=True. + """ + if rng is None: + rng = np.random.default_rng() + + n = len(trading_dates) + if tickers is None: + tickers = [f"SYM{i:03d}" for i in range(n_symbols)] + + # Typical vol across all regimes (for intraday range scaling) + avg_vol = sum(r.annualized_vol for r in regimes) / max(len(regimes), 1) + dt_step = 1.0 / 252 + + # Market-level log returns (shared factor) + market_log_rets = _generate_regime_log_returns(regimes, n, rng) + + bars_by_symbol: dict[str, dict[dt.date, dict[str, Any]]] = {} + + for i, ticker in enumerate(tickers[:n_symbols]): + if initial_prices and i < len(initial_prices): + init_price = initial_prices[i] + else: + init_price = float(rng.uniform(20.0, 200.0)) + + beta = float(rng.uniform(*market_beta_range)) + idio_vol = avg_vol * float(rng.uniform(0.3, 0.8)) + + # Build stock log returns: beta * market + idiosyncratic + stock_log_rets: list[float] = [] + for mkt_lr in market_log_rets: + idio = rng.normal(0, idio_vol * math.sqrt(dt_step)) + stock_log_rets.append(beta * mkt_lr + idio) + + intraday_scale = (idio_vol + avg_vol * beta) * math.sqrt(dt_step) * 0.5 + base_vol = int(rng.uniform(500_000, 10_000_000)) + + bars = _bars_from_log_returns( + stock_log_rets, + trading_dates, + initial_price=init_price, + intraday_vol_scale=intraday_scale, + base_volume=base_vol, + rng=rng, + ) + bars_by_symbol[ticker] = bars + + if return_market_log_rets: + return bars_by_symbol, market_log_rets + return bars_by_symbol + + +def generate_market_etf_paths( + regimes: list[PriceRegime], + trading_dates: list[dt.date], + rng: np.random.Generator, + spy_initial: float = 480.0, + qqq_initial: float = 415.0, +) -> tuple[dict[dt.date, dict[str, Any]], dict[dt.date, dict[str, Any]], list[float]]: + """Generate SPY and QQQ synthetic price bars plus market log-returns. + + QQQ has slightly higher vol and beta than SPY to reflect tech concentration. + + Returns: + (spy_bars, qqq_bars, market_log_rets) + """ + n = len(trading_dates) + dt_step = 1.0 / 252 + avg_vol = sum(r.annualized_vol for r in regimes) / max(len(regimes), 1) + + market_log_rets = _generate_regime_log_returns(regimes, n, rng) + + # SPY ≈ market (beta ~1.0, low idio noise) + spy_log_rets: list[float] = [] + for lr in market_log_rets: + spy_log_rets.append(lr + rng.normal(0, avg_vol * 0.05 * math.sqrt(dt_step))) + + spy_bars = _bars_from_log_returns( + spy_log_rets, trading_dates, + initial_price=spy_initial, + intraday_vol_scale=avg_vol * math.sqrt(dt_step) * 0.4, + base_volume=80_000_000, + rng=rng, + ) + + # QQQ ≈ market * 1.1 beta + higher idio noise + qqq_log_rets: list[float] = [] + for lr in market_log_rets: + qqq_log_rets.append( + 1.1 * lr + rng.normal(0, avg_vol * 0.08 * math.sqrt(dt_step)) + ) + + qqq_bars = _bars_from_log_returns( + qqq_log_rets, trading_dates, + initial_price=qqq_initial, + intraday_vol_scale=avg_vol * math.sqrt(dt_step) * 0.45, + base_volume=60_000_000, + rng=rng, + ) + + return spy_bars, qqq_bars, market_log_rets diff --git a/libs/backtest/scenarios/robustness.py b/libs/backtest/scenarios/robustness.py new file mode 100644 index 0000000..497b48f --- /dev/null +++ b/libs/backtest/scenarios/robustness.py @@ -0,0 +1,258 @@ +"""Cross-scenario analysis and Regime Robustness Score (RRS). + +Computes a composite score measuring how well a strategy generalises across +diverse synthetic market conditions. + +RRS = 0.25 * signal_integrity + 0.25 * breadth + 0.20 * drawdown_resilience + + 0.15 * regime_transition + 0.15 * stability + +Verdicts: + RRS >= 70 → ROBUST (strategy generalises well) + RRS 40-69 → FRAGILE (works on some regimes, fails on others) + RRS < 40 → OVERFIT (likely curve-fit to historical data patterns) +""" +from __future__ import annotations + +import statistics +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any + +import numpy as np + +from libs.backtest.domain import MetricsBundle + + +@dataclass +class ScenarioResult: + """Result of running one scenario against a strategy.""" + + scenario_name: str + metrics: MetricsBundle + trade_count: int + sharpe_ratio: float + total_return_pct: float + max_drawdown_pct: float + win_rate: float + profit_factor: float + + +@dataclass +class RegimeRobustnessReport: + """Aggregated cross-scenario robustness report.""" + + experiment_name: str + scenario_results: dict[str, ScenarioResult] + + # ---- Component scores (0-100) ---- + signal_integrity: float + """no_signal Sharpe ≤ 0 → 100. Detects pure price-pattern overfitting.""" + breadth: float + """Fraction of scenarios with positive Sharpe × 100.""" + drawdown_resilience: float + """100 × (1 - worst_dd / 50). Penalises extreme drawdowns.""" + regime_transition: float + """Performance on regime_switch vs median. Tests rapid regime adaptation.""" + stability: float + """100 × (1 - CV(Sharpes)). Low variance = consistent across regimes.""" + + # ---- Overall ---- + rrs: float + """Regime Robustness Score (0-100).""" + verdict: str + """ROBUST / FRAGILE / OVERFIT.""" + + # ---- Metadata ---- + scenarios_run: list[str] = field(default_factory=list) + notes: list[str] = field(default_factory=list) + + +def compute_rrs( + scenario_results: dict[str, ScenarioResult], +) -> tuple[float, float, float, float, float, float]: + """Compute RRS and its five component scores. + + Returns: + (rrs, signal_integrity, breadth, drawdown_resilience, regime_transition, stability) + """ + if not scenario_results: + return 0.0, 0.0, 0.0, 0.0, 0.0, 0.0 + + sharpes = {name: r.sharpe_ratio for name, r in scenario_results.items()} + drawdowns = {name: r.max_drawdown_pct for name, r in scenario_results.items()} + + all_sharpes = list(sharpes.values()) + + # ---- Signal integrity (25%) ---- + # no_signal Sharpe ≤ 0 → 100. ≥ strong_signal Sharpe → 0. + no_sig = sharpes.get("no_signal") + strong_sig = sharpes.get("strong_signal") + + if no_sig is None: + signal_integrity = 50.0 # neutral if no_signal not run + elif no_sig <= 0: + signal_integrity = 100.0 + elif strong_sig is not None and strong_sig > 0: + # Linear interpolation: 0 when no_signal == strong_signal, 100 when no_signal ≤ 0 + signal_integrity = max(0.0, 100.0 * (1.0 - no_sig / strong_sig)) + else: + # strong_signal not run or ≤ 0; penalise proportionally to no_signal Sharpe + signal_integrity = max(0.0, 100.0 - no_sig * 50.0) + signal_integrity = min(100.0, signal_integrity) + + # ---- Breadth (25%) ---- + n_positive = sum(1 for s in all_sharpes if s > 0) + breadth = 100.0 * n_positive / max(len(all_sharpes), 1) + + # ---- Drawdown resilience (20%) ---- + worst_dd = max(drawdowns.values()) if drawdowns else 0.0 + # 0% DD → 100, 50% DD → 0 + drawdown_resilience = max(0.0, min(100.0, 100.0 * (1.0 - worst_dd / 50.0))) + + # ---- Regime transition (15%) ---- + regime_sharpe = sharpes.get("regime_switch") + if regime_sharpe is None: + regime_transition = 50.0 # neutral + else: + median_sharpe = statistics.median(all_sharpes) if all_sharpes else 0.0 + if median_sharpe > 0: + ratio = regime_sharpe / median_sharpe + regime_transition = min(100.0, max(0.0, ratio * 50.0 + 50.0)) + elif regime_sharpe > 0: + regime_transition = 70.0 + else: + regime_transition = max(0.0, 50.0 + regime_sharpe * 25.0) + + # ---- Stability (15%) ---- + if len(all_sharpes) >= 2: + mean_sharpe = statistics.mean(all_sharpes) + std_sharpe = statistics.stdev(all_sharpes) + if abs(mean_sharpe) > 0.01: + cv = std_sharpe / abs(mean_sharpe) + stability = max(0.0, min(100.0, 100.0 * (1.0 - min(cv, 2.0) / 2.0))) + else: + # mean near 0 → consistent but also weak; moderate stability score + stability = max(0.0, 50.0 - std_sharpe * 25.0) + else: + stability = 50.0 + + # ---- RRS composite ---- + rrs = ( + 0.25 * signal_integrity + + 0.25 * breadth + + 0.20 * drawdown_resilience + + 0.15 * regime_transition + + 0.15 * stability + ) + + return rrs, signal_integrity, breadth, drawdown_resilience, regime_transition, stability + + +def _verdict(rrs: float) -> str: + if rrs >= 70.0: + return "ROBUST" + elif rrs >= 40.0: + return "FRAGILE" + else: + return "OVERFIT" + + +def run_scenario_test( + experiment_name: str, + scenario_names: list[str], + initial_equity: float = 10_000.0, + progress_callback: Any | None = None, +) -> RegimeRobustnessReport: + """Run a strategy against the specified synthetic scenarios and return RRS report. + + Args: + experiment_name: Experiment config name (e.g., "return_max_long_v7.70"). + scenario_names: List of scenario names to run (from SCENARIO_REGISTRY). + initial_equity: Starting capital for each scenario backtest. + progress_callback: Optional callable(scenario_name) called before each run. + + Returns: + RegimeRobustnessReport with per-scenario metrics and composite RRS. + """ + import structlog + structlog.configure(wrapper_class=structlog.make_filtering_bound_logger(50)) + + from libs.backtest.experiments import resolve_experiment_name + from libs.backtest.manifests import load_manifest, resolve_config + from libs.backtest.scenarios.scenarios import SCENARIO_REGISTRY + from libs.backtest.scenarios.store_builder import build_synthetic_store + from apps.backtester.run import BacktestRunner + + _CONFIGS_DIR = Path("configs/experiments") + resolved = resolve_experiment_name(experiment_name) + manifest_path = _CONFIGS_DIR / f"{resolved}.json" + manifest = load_manifest(manifest_path) + config = resolve_config(manifest) + + scenario_results: dict[str, ScenarioResult] = {} + notes: list[str] = [] + + for scenario_name in scenario_names: + scenario = SCENARIO_REGISTRY.get(scenario_name) + if scenario is None: + notes.append(f"Unknown scenario '{scenario_name}' — skipped.") + continue + + if progress_callback: + progress_callback(scenario_name) + + try: + store = build_synthetic_store(scenario) + runner = BacktestRunner( + manifest=manifest, + config=config, + store=store, + initial_equity=initial_equity, + split_name=f"scenario_{scenario_name}", + ) + result = runner.run(output_root=None) + m = result.metrics + + scenario_results[scenario_name] = ScenarioResult( + scenario_name=scenario_name, + metrics=m, + trade_count=m.trade_count, + sharpe_ratio=m.sharpe_ratio or 0.0, + total_return_pct=m.total_return_pct or 0.0, + max_drawdown_pct=abs(m.max_drawdown_pct or 0.0), + win_rate=m.win_rate or 0.0, + profit_factor=m.profit_factor or 0.0, + ) + except Exception as exc: + notes.append(f"Scenario '{scenario_name}' failed: {exc}") + # Insert a zero-performance placeholder so scores aren't skewed by missing data + scenario_results[scenario_name] = ScenarioResult( + scenario_name=scenario_name, + metrics=MetricsBundle( + trade_count=0, win_rate=0.0, avg_win_pct=0.0, avg_loss_pct=0.0, + profit_factor=0.0, total_return_pct=0.0, max_drawdown_pct=0.0, + sharpe_ratio=0.0, + ), + trade_count=0, + sharpe_ratio=0.0, + total_return_pct=0.0, + max_drawdown_pct=0.0, + win_rate=0.0, + profit_factor=0.0, + ) + + rrs, si, br, dr, rt, st = compute_rrs(scenario_results) + + return RegimeRobustnessReport( + experiment_name=experiment_name, + scenario_results=scenario_results, + signal_integrity=si, + breadth=br, + drawdown_resilience=dr, + regime_transition=rt, + stability=st, + rrs=rrs, + verdict=_verdict(rrs), + scenarios_run=list(scenario_results.keys()), + notes=notes, + ) diff --git a/libs/backtest/scenarios/scenarios.py b/libs/backtest/scenarios/scenarios.py new file mode 100644 index 0000000..8344ee6 --- /dev/null +++ b/libs/backtest/scenarios/scenarios.py @@ -0,0 +1,332 @@ +"""Pre-built scenario library for synthetic market backtesting. + +Defines 12 scenarios covering diverse market regimes. Each scenario is a +ScenarioConfig that specifies market dynamics, event characteristics, and +signal coupling strength. + +Key scenarios for overfitting detection: + no_signal → signal_strength=0.0 → strategy must return ~0 + strong_signal → signal_strength=0.6 → strategy must capture alpha + +Scenario groups for targeted analysis: + trend → steady_bull, steady_bear, prolonged_bear + volatility → low_vol_grind, high_vol_chop, vix_spike + regime → regime_switch, crash_v_recovery + signal → no_signal, strong_signal + structural → sector_rotation, liquidity_drought + quick → steady_bull, steady_bear, no_signal (fast validation) +""" +from __future__ import annotations + +from dataclasses import dataclass, field + +from libs.backtest.scenarios.price_gen import PriceRegime +from libs.backtest.scenarios.macro_gen import VIXConfig, HYSpreadConfig +from libs.backtest.scenarios.event_gen import EventDistribution + + +@dataclass +class ScenarioConfig: + """Complete specification of a synthetic market scenario.""" + + name: str + description: str + + # Market dynamics + price_regimes: list[PriceRegime] + """Ordered sequence of market regime segments.""" + vix_config: VIXConfig = field(default_factory=VIXConfig) + hy_config: HYSpreadConfig = field(default_factory=HYSpreadConfig) + + # Event characteristics + event_distribution: EventDistribution = field(default_factory=EventDistribution) + + # Signal coupling + signal_strength: float = 0.35 + """0.0=pure noise, ~0.35=realistic SNR, 0.6=strong alpha.""" + signal_decay_days: int = 10 + """Days over which post-event signal drift decays.""" + false_positive_rate: float = 0.15 + """Fraction of qualifying events that produce negative returns (traps).""" + + # Simulation parameters + n_symbols: int = 200 + """Number of unique synthetic tickers in the universe.""" + seed: int | None = None + """Random seed for reproducibility (None = non-deterministic).""" + + +# --------------------------------------------------------------------------- +# Helper: bear-market event distribution (weaker signals, more negative reactions) +# --------------------------------------------------------------------------- + +def _bear_event_dist(reaction_return_mean: float = -0.010) -> EventDistribution: + return EventDistribution( + reaction_return_mean=reaction_return_mean, + reaction_return_std=0.065, + event_directions={ + "bullish": 0.30, + "mixed": 0.30, + "unknown": 0.25, + "bearish": 0.15, + }, + volume_ratio_mean=2.2, + signal_strength_mean=0.58, + market_temperature_mean=1.2, + volatility_20d_mean=0.38, + ) + + +def _stress_event_dist() -> EventDistribution: + return EventDistribution( + reaction_return_mean=0.000, + reaction_return_std=0.080, + event_directions={ + "bullish": 0.35, + "mixed": 0.35, + "unknown": 0.20, + "bearish": 0.10, + }, + volume_ratio_mean=2.5, + signal_strength_mean=0.60, + market_temperature_mean=1.5, + volatility_20d_mean=0.42, + rsi_14_mean=45.0, + bb_position_mean=0.40, + ) + + +# --------------------------------------------------------------------------- +# Pre-built scenarios +# --------------------------------------------------------------------------- + +STEADY_BULL = ScenarioConfig( + name="steady_bull", + description="Sustained bull market: +15% drift, 14% vol. Baseline profitable environment.", + price_regimes=[ + PriceRegime(annualized_drift=0.15, annualized_vol=0.14, duration_days=252), + ], + vix_config=VIXConfig(base_level=14.0, mean_reversion=5.0, volatility=4.0), + hy_config=HYSpreadConfig(base_level=3.5, mean_reversion=3.0, volatility=0.8), + signal_strength=0.35, + seed=1001, +) + +STEADY_BEAR = ScenarioConfig( + name="steady_bear", + description="Sustained bear market: -20% drift, 22% vol. Tests macro regime filter (Gate 0).", + price_regimes=[ + PriceRegime(annualized_drift=-0.20, annualized_vol=0.22, duration_days=252), + ], + vix_config=VIXConfig(base_level=28.0, mean_reversion=4.0, volatility=7.0), + hy_config=HYSpreadConfig(base_level=7.0, mean_reversion=2.5, volatility=2.0), + event_distribution=_bear_event_dist(reaction_return_mean=-0.010), + signal_strength=0.35, + seed=1002, +) + +CRASH_V_RECOVERY = ScenarioConfig( + name="crash_v_recovery", + description=( + "V-shaped crash + recovery: 60d normal → 20d crash (-40%/40% vol) → 172d recovery. " + "Tests kill switch and drawdown protection." + ), + price_regimes=[ + PriceRegime(annualized_drift=0.10, annualized_vol=0.15, duration_days=60), + PriceRegime( + annualized_drift=-0.40, annualized_vol=0.40, duration_days=20, + jump_prob=0.08, jump_mean=-0.05, jump_std=0.03, + ), + PriceRegime(annualized_drift=0.25, annualized_vol=0.20, duration_days=172), + ], + vix_config=VIXConfig(base_level=15.0, mean_reversion=3.0, volatility=8.0), + hy_config=HYSpreadConfig(base_level=4.0, mean_reversion=2.0, volatility=2.5), + signal_strength=0.35, + seed=1003, +) + +PROLONGED_BEAR = ScenarioConfig( + name="prolonged_bear", + description=( + "2-year bear market: -15% drift, 25% vol over 504 trading days. " + "Exceeds duration of any historical bear in training data." + ), + price_regimes=[ + PriceRegime(annualized_drift=-0.15, annualized_vol=0.25, duration_days=504), + ], + vix_config=VIXConfig(base_level=30.0, mean_reversion=3.5, volatility=8.0), + hy_config=HYSpreadConfig(base_level=8.5, mean_reversion=2.0, volatility=2.5), + event_distribution=_bear_event_dist(reaction_return_mean=-0.015), + signal_strength=0.30, + seed=1004, +) + +LOW_VOL_GRIND = ScenarioConfig( + name="low_vol_grind", + description=( + "Low-volatility grind: +8% drift, 8% vol. " + "ATR shrinks → stop distances compress → fewer trades qualify." + ), + price_regimes=[ + PriceRegime(annualized_drift=0.08, annualized_vol=0.08, duration_days=252), + ], + vix_config=VIXConfig(base_level=11.0, mean_reversion=6.0, volatility=2.5), + hy_config=HYSpreadConfig(base_level=2.8, mean_reversion=4.0, volatility=0.5), + event_distribution=EventDistribution( + volatility_20d_mean=0.15, + volatility_20d_std=0.04, + reaction_return_std=0.03, + ), + signal_strength=0.20, + seed=1005, +) + +HIGH_VOL_CHOP = ScenarioConfig( + name="high_vol_chop", + description=( + "High-volatility sideways chop: 0% drift, 30% vol. " + "Whipsaws test stop-loss resilience." + ), + price_regimes=[ + PriceRegime(annualized_drift=0.00, annualized_vol=0.30, duration_days=252), + ], + vix_config=VIXConfig(base_level=32.0, mean_reversion=4.0, volatility=9.0), + hy_config=HYSpreadConfig(base_level=6.5, mean_reversion=2.5, volatility=2.0), + event_distribution=_stress_event_dist(), + signal_strength=0.25, + seed=1006, +) + +REGIME_SWITCH = ScenarioConfig( + name="regime_switch", + description=( + "Rapid regime alternation: 4 × (60d bull / 63d bear) cycles. " + "Tests whether macro gate adapts quickly to changing conditions." + ), + price_regimes=[ + PriceRegime(annualized_drift=0.15, annualized_vol=0.16, duration_days=60), + PriceRegime(annualized_drift=-0.18, annualized_vol=0.24, duration_days=63), + PriceRegime(annualized_drift=0.12, annualized_vol=0.16, duration_days=60), + PriceRegime(annualized_drift=-0.15, annualized_vol=0.22, duration_days=69), + ], + vix_config=VIXConfig(base_level=20.0, mean_reversion=4.0, volatility=8.0), + hy_config=HYSpreadConfig(base_level=5.0, mean_reversion=2.5, volatility=1.8), + signal_strength=0.30, + seed=1007, +) + +VIX_SPIKE = ScenarioConfig( + name="vix_spike", + description=( + "Normal market with 5 random VIX spike weeks (VIX 40+). " + "Tests VIX continuous scaler and kill switch under stress clusters." + ), + price_regimes=[ + PriceRegime(annualized_drift=0.08, annualized_vol=0.16, duration_days=252), + ], + vix_config=VIXConfig(base_level=18.0, mean_reversion=3.0, volatility=12.0), + hy_config=HYSpreadConfig(base_level=4.5, mean_reversion=2.5, volatility=2.0), + signal_strength=0.35, + seed=1008, +) + +NO_SIGNAL = ScenarioConfig( + name="no_signal", + description=( + "Pure noise: 0% drift, 16% vol, signal_strength=0.0. " + "Market is flat so any positive return = OVERFIT to price patterns, not event alpha." + ), + price_regimes=[ + PriceRegime(annualized_drift=0.00, annualized_vol=0.16, duration_days=252), + ], + vix_config=VIXConfig(base_level=16.0, mean_reversion=5.0, volatility=4.5), + hy_config=HYSpreadConfig(base_level=4.0, mean_reversion=3.0, volatility=1.0), + signal_strength=0.00, + seed=1009, +) + +STRONG_SIGNAL = ScenarioConfig( + name="strong_signal", + description=( + "Strong alpha: +10% drift, 16% vol, signal_strength=0.6. " + "Strategy must capture meaningful positive returns." + ), + price_regimes=[ + PriceRegime(annualized_drift=0.10, annualized_vol=0.16, duration_days=252), + ], + vix_config=VIXConfig(base_level=16.0, mean_reversion=5.0, volatility=4.5), + hy_config=HYSpreadConfig(base_level=4.0, mean_reversion=3.0, volatility=1.0), + signal_strength=0.60, + seed=1010, +) + +SECTOR_ROTATION = ScenarioConfig( + name="sector_rotation", + description=( + "Quarterly sector rotation: Tech underperforms while Healthcare/Financials rally. " + "Tests sector concentration limits." + ), + price_regimes=[ + PriceRegime(annualized_drift=0.05, annualized_vol=0.18, duration_days=63), + PriceRegime(annualized_drift=0.12, annualized_vol=0.16, duration_days=63), + PriceRegime(annualized_drift=-0.05, annualized_vol=0.20, duration_days=63), + PriceRegime(annualized_drift=0.08, annualized_vol=0.15, duration_days=63), + ], + vix_config=VIXConfig(base_level=20.0, mean_reversion=4.5, volatility=5.0), + hy_config=HYSpreadConfig(base_level=4.5, mean_reversion=3.0, volatility=1.2), + signal_strength=0.35, + seed=1011, +) + +LIQUIDITY_DROUGHT = ScenarioConfig( + name="liquidity_drought", + description=( + "Liquidity drought: +5% drift, 18% vol, stock volumes drop 60%. " + "Tests ADV fraction limits and position sizing." + ), + price_regimes=[ + PriceRegime(annualized_drift=0.05, annualized_vol=0.18, duration_days=252), + ], + vix_config=VIXConfig(base_level=22.0, mean_reversion=4.0, volatility=5.5), + hy_config=HYSpreadConfig(base_level=5.0, mean_reversion=2.5, volatility=1.5), + event_distribution=EventDistribution( + avg_dollar_volume_mean=1_500_000.0, # 70% lower than default + avg_dollar_volume_std=800_000.0, + volume_ratio_mean=1.2, + ), + signal_strength=0.30, + seed=1012, +) + + +# --------------------------------------------------------------------------- +# Registry and groups +# --------------------------------------------------------------------------- + +SCENARIO_REGISTRY: dict[str, ScenarioConfig] = { + s.name: s + for s in [ + STEADY_BULL, + STEADY_BEAR, + CRASH_V_RECOVERY, + PROLONGED_BEAR, + LOW_VOL_GRIND, + HIGH_VOL_CHOP, + REGIME_SWITCH, + VIX_SPIKE, + NO_SIGNAL, + STRONG_SIGNAL, + SECTOR_ROTATION, + LIQUIDITY_DROUGHT, + ] +} + +SCENARIO_GROUPS: dict[str, list[str]] = { + "trend": ["steady_bull", "steady_bear", "prolonged_bear"], + "volatility": ["low_vol_grind", "high_vol_chop", "vix_spike"], + "regime": ["regime_switch", "crash_v_recovery"], + "signal": ["no_signal", "strong_signal"], + "structural": ["sector_rotation", "liquidity_drought"], + "quick": ["steady_bull", "steady_bear", "no_signal"], + "all": list(SCENARIO_REGISTRY.keys()), +} diff --git a/libs/backtest/scenarios/store_builder.py b/libs/backtest/scenarios/store_builder.py new file mode 100644 index 0000000..f3dd849 --- /dev/null +++ b/libs/backtest/scenarios/store_builder.py @@ -0,0 +1,146 @@ +"""Assembles a synthetic SnapshotStore from a ScenarioConfig. + +Orchestrates the full synthetic data generation pipeline: + 1. Trading dates (real NYSE calendar) + 2. Market ETF price paths (SPY, QQQ) + 3. Individual stock price paths (correlated with market) + 4. Macro indicators (VIX, HY spread, SPY/QQQ rolling stats) + 5. Event candidates (parameterized feature distributions) + 6. Event-price coupling (signal-to-noise injection) + 7. SnapshotStore assembly +""" +from __future__ import annotations + +import datetime as dt +from typing import Any + +import numpy as np + +from libs.backtest.scenarios.coupling import couple_events_to_prices +from libs.backtest.scenarios.event_gen import generate_events +from libs.backtest.scenarios.macro_gen import generate_macro_data +from libs.backtest.scenarios.price_gen import generate_market_etf_paths, generate_price_paths +from libs.backtest.scenarios.scenarios import ScenarioConfig +from libs.backtest.snapshot_store import SnapshotStore + + +# Starting dates for synthetic scenarios (real NYSE calendar) +# Using a date range that's safely after 2020 (no pandemic-era data issues) +_SCENARIO_START_DATE = dt.date(2024, 1, 2) + + +def _get_trading_dates(n_days: int, start: dt.date = _SCENARIO_START_DATE) -> list[dt.date]: + """Return n_days consecutive NYSE trading dates starting from start.""" + from libs.backtest.calendar import get_trading_days + # Request a window of n_days * 1.5 calendar days to account for weekends/holidays + end_estimate = start + dt.timedelta(days=int(n_days * 1.5) + 30) + all_days = get_trading_days(start, end_estimate) + return all_days[:n_days] + + +def _total_regime_days(scenario: ScenarioConfig) -> int: + """Sum of all regime durations in the scenario.""" + return sum(r.duration_days for r in scenario.price_regimes) + + +def build_synthetic_store( + scenario: ScenarioConfig, + rng: np.random.Generator | None = None, +) -> SnapshotStore: + """Assemble a complete synthetic SnapshotStore from a ScenarioConfig. + + The returned store has no connection to real market data. It can be + passed directly to BacktestRunner without any Oracle API or database. + + Args: + scenario: Fully specified synthetic market scenario. + rng: NumPy random generator. If None, uses scenario.seed or random state. + + Returns: + SnapshotStore ready for BacktestRunner.run(). + """ + if rng is None: + seed = scenario.seed + rng = np.random.default_rng(seed) + + # 1. Trading dates + n_days = _total_regime_days(scenario) + 30 # extra buffer for parking tail + trading_dates = _get_trading_dates(n_days) + + # 2. Market ETF paths (SPY, QQQ) + market log-returns for macro correlation + spy_bars, qqq_bars, market_log_rets = generate_market_etf_paths( + regimes=scenario.price_regimes, + trading_dates=trading_dates, + rng=rng, + ) + + # 3. Individual stock paths + n_sym = scenario.n_symbols + tickers = [f"SYN{i:03d}" for i in range(n_sym)] + + bars_by_symbol = generate_price_paths( + n_symbols=n_sym, + initial_prices=None, + regimes=scenario.price_regimes, + trading_dates=trading_dates, + market_beta_range=(0.5, 1.4), + rng=rng, + tickers=tickers, + ) + assert isinstance(bars_by_symbol, dict), "generate_price_paths must return dict" + + # Merge ETF bars in as well (for parking lookups) + bars_by_symbol["SPY"] = spy_bars + bars_by_symbol["QQQ"] = qqq_bars + + # 4. Macro data + macro_by_date = generate_macro_data( + spy_bars=spy_bars, + qqq_bars=qqq_bars, + vix_config=scenario.vix_config, + hy_config=scenario.hy_config, + trading_dates=trading_dates, + rng=rng, + market_log_rets=market_log_rets, + ) + + # 5. Event candidates + candidates_by_exec_date = generate_events( + trading_dates=trading_dates, + symbols=tickers, + dist=scenario.event_distribution, + rng=rng, + bars_by_symbol=bars_by_symbol, + ) + + # 5b. Inject macro_vix / macro_hy_spread into each candidate row. + # selector._row_matches_strategy_engine_filters() reads macro_vix and + # macro_hy_spread directly from the row dict (not from macro_by_date), + # so we must populate them here. + for exec_date, rows in candidates_by_exec_date.items(): + macro = macro_by_date.get(exec_date, {}) + mv = macro.get("macro_vix") + hy = macro.get("macro_hy_spread") + for row in rows: + if mv is not None: + row["macro_vix"] = mv + if hy is not None: + row["macro_hy_spread"] = hy + + # 6. Event-price coupling (signal injection) + couple_events_to_prices( + candidates=candidates_by_exec_date, + bars_by_symbol=bars_by_symbol, + trading_dates=trading_dates, + signal_strength=scenario.signal_strength, + signal_decay_days=scenario.signal_decay_days, + false_positive_rate=scenario.false_positive_rate, + rng=rng, + ) + + # 7. Assemble SnapshotStore + return SnapshotStore( + candidates_by_exec_date=candidates_by_exec_date, + bars_by_symbol_date=bars_by_symbol, + macro_by_date=macro_by_date, + ) diff --git a/tests/unit/backtest/test_event_gen.py b/tests/unit/backtest/test_event_gen.py new file mode 100644 index 0000000..31cffc1 --- /dev/null +++ b/tests/unit/backtest/test_event_gen.py @@ -0,0 +1,245 @@ +"""Unit tests for event_gen.py.""" +import datetime as dt + +import numpy as np +import pytest + +from libs.backtest.scenarios.event_gen import ( + EventDistribution, + _compute_synthetic_score, + _sample_categorical, + generate_events, +) +from libs.backtest.scenarios.price_gen import PriceRegime, generate_price_paths + +_DATES_RAW = [dt.date(2024, 1, 2) + dt.timedelta(days=i) for i in range(400)] +_TRADING_DATES = [d for d in _DATES_RAW if d.weekday() < 5][:252] +_SYMBOLS = [f"SYM{i:03d}" for i in range(30)] + + +@pytest.mark.unit +class TestSampleCategorical: + def test_returns_valid_key(self): + rng = np.random.default_rng(1) + cats = {"a": 0.5, "b": 0.3, "c": 0.2} + result = _sample_categorical(cats, rng) + assert result in cats + + def test_distribution_roughly_correct(self): + rng = np.random.default_rng(42) + cats = {"x": 0.9, "y": 0.1} + counts = {"x": 0, "y": 0} + for _ in range(1000): + k = _sample_categorical(cats, rng) + counts[k] += 1 + # x should appear roughly 90% of the time + assert counts["x"] > 800 + + +@pytest.mark.unit +class TestComputeSyntheticScore: + def test_high_quality_event_has_high_score(self): + row = { + "signal_strength_score": 0.95, + "document_quality_score": 0.95, + "parse_confidence_overall": 0.95, + "guidance_direction_score": 0.90, + "oneoff_penalty": 0.0, + "reaction_day_return": 0.08, + "volume_ratio_20d": 3.0, + "close_location": 0.85, + "gap_size": 0.02, + "pre_event_entropy_60d": 1.0, + } + score = _compute_synthetic_score(row) + assert score > 0.75 + + def test_low_quality_event_has_low_score(self): + row = { + "signal_strength_score": 0.20, + "document_quality_score": 0.20, + "parse_confidence_overall": 0.25, + "guidance_direction_score": 0.10, + "oneoff_penalty": 1.0, + "reaction_day_return": -0.05, + "volume_ratio_20d": 0.5, + "close_location": 0.2, + "gap_size": -0.01, + "pre_event_entropy_60d": 1.8, + } + score = _compute_synthetic_score(row) + assert score < 0.35 + + def test_score_bounded_0_to_1(self): + """Score must always be in [0, 1].""" + rng = np.random.default_rng(7) + for _ in range(50): + row = { + "signal_strength_score": float(rng.uniform(0, 1)), + "document_quality_score": float(rng.uniform(0, 1)), + "parse_confidence_overall": float(rng.uniform(0, 1)), + "guidance_direction_score": float(rng.uniform(0, 1)), + "oneoff_penalty": float(rng.choice([0.0, 1.0])), + "reaction_day_return": float(rng.normal(0, 0.05)), + "volume_ratio_20d": float(rng.uniform(0.3, 4.0)), + "pre_event_entropy_60d": float(rng.uniform(0.5, 2.0)), + } + score = _compute_synthetic_score(row) + assert 0.0 <= score <= 1.0, f"score {score} out of [0, 1]" + + def test_oneoff_penalty_reduces_score(self): + base_row = { + "signal_strength_score": 0.7, + "document_quality_score": 0.7, + "parse_confidence_overall": 0.8, + "guidance_direction_score": 0.6, + "oneoff_penalty": 0.0, + "reaction_day_return": 0.03, + "volume_ratio_20d": 2.0, + "pre_event_entropy_60d": 1.3, + } + penalized_row = {**base_row, "oneoff_penalty": 1.0} + assert _compute_synthetic_score(base_row) > _compute_synthetic_score(penalized_row) + + +@pytest.mark.unit +class TestGenerateEvents: + def test_returns_dict_keyed_by_exec_date(self): + rng = np.random.default_rng(42) + dist = EventDistribution() + candidates = generate_events(_TRADING_DATES, _SYMBOLS, dist, rng) + assert isinstance(candidates, dict) + for d in candidates: + assert isinstance(d, dt.date) + assert d in _TRADING_DATES + + def test_all_exec_dates_within_eligible_range(self): + rng = np.random.default_rng(42) + dist = EventDistribution() + buffer = 25 + candidates = generate_events(_TRADING_DATES, _SYMBOLS, dist, rng, max_holding_days_buffer=buffer) + # First 5 days are warm-up; last buffer days are excluded + eligible_end = _TRADING_DATES[len(_TRADING_DATES) - buffer - 1] + for d in candidates: + assert d <= eligible_end, f"exec_date {d} beyond eligible range" + + def test_required_fields_present(self): + rng = np.random.default_rng(42) + dist = EventDistribution() + candidates = generate_events(_TRADING_DATES, _SYMBOLS, dist, rng) + + required_fields = [ + "event_id", "symbol", "event_type", "event_direction", + "filing_time_bucket", "sector", + "event_timestamp", "reaction_date", "entry_date", "execution_date", + "entry_price", "entry_price_est", "event_close", "atr_14", "avg_dollar_volume", + "score", + "signal_strength_score", "document_quality_score", "parse_confidence_overall", + "guidance_direction_score", "oneoff_penalty", + "reaction_day_return", "volume_ratio_20d", "close_location", "gap_size", + "pre_event_rsi_14", "pre_event_bb_position", "pre_event_volatility_20d", + "pre_event_hurst_60d", "pre_event_entropy_60d", + "pre_event_ou_theta_60d", "pre_event_market_temperature", + ] + + for exec_date, rows in candidates.items(): + for row in rows: + for field in required_fields: + assert field in row, f"Missing field '{field}' in candidate on {exec_date}" + + def test_execution_date_matches_dict_key(self): + rng = np.random.default_rng(42) + dist = EventDistribution() + candidates = generate_events(_TRADING_DATES, _SYMBOLS, dist, rng) + for exec_date, rows in candidates.items(): + for row in rows: + assert row["execution_date"] == exec_date + + def test_reaction_date_is_previous_trading_day(self): + rng = np.random.default_rng(42) + dist = EventDistribution() + candidates = generate_events(_TRADING_DATES, _SYMBOLS, dist, rng) + date_set = set(_TRADING_DATES) + for exec_date, rows in candidates.items(): + exec_idx = _TRADING_DATES.index(exec_date) + expected_reaction = _TRADING_DATES[exec_idx - 1] + for row in rows: + rxn = row["reaction_date"] + # reaction_date stored as isoformat string + rxn_date = dt.date.fromisoformat(rxn) if isinstance(rxn, str) else rxn + assert rxn_date == expected_reaction, f"reaction_date mismatch on {exec_date}" + + def test_event_id_unique(self): + rng = np.random.default_rng(42) + dist = EventDistribution() + candidates = generate_events(_TRADING_DATES, _SYMBOLS, dist, rng) + all_ids = [row["event_id"] for rows in candidates.values() for row in rows] + assert len(all_ids) == len(set(all_ids)), "Duplicate event_ids found" + + def test_no_duplicate_symbol_per_day(self): + rng = np.random.default_rng(42) + dist = EventDistribution() + candidates = generate_events(_TRADING_DATES, _SYMBOLS, dist, rng) + for exec_date, rows in candidates.items(): + syms = [r["symbol"] for r in rows] + assert len(syms) == len(set(syms)), f"Duplicate symbols on {exec_date}" + + def test_prices_positive(self): + rng = np.random.default_rng(42) + dist = EventDistribution() + candidates = generate_events(_TRADING_DATES, _SYMBOLS, dist, rng) + for rows in candidates.values(): + for row in rows: + assert row["entry_price"] > 0, f"Non-positive entry_price: {row['entry_price']}" + assert row["atr_14"] > 0, f"Non-positive atr_14: {row['atr_14']}" + + def test_score_between_0_and_1(self): + rng = np.random.default_rng(42) + dist = EventDistribution() + candidates = generate_events(_TRADING_DATES, _SYMBOLS, dist, rng) + for rows in candidates.values(): + for row in rows: + assert 0.0 <= row["score"] <= 1.0, f"score {row['score']} out of [0, 1]" + + def test_uses_bar_close_when_bars_provided(self): + rng = np.random.default_rng(42) + dist = EventDistribution(events_per_day_mean=2.0, events_per_day_std=0.0) + regimes = [PriceRegime(0.10, 0.15, 252)] + bars_by_symbol = generate_price_paths( + n_symbols=len(_SYMBOLS), + initial_prices=None, + regimes=regimes, + trading_dates=_TRADING_DATES, + rng=np.random.default_rng(1), + tickers=_SYMBOLS, + ) + candidates = generate_events( + _TRADING_DATES, _SYMBOLS, dist, + np.random.default_rng(42), + bars_by_symbol=bars_by_symbol, + ) + # At least some events should be present + assert len(candidates) > 0 + # entry_prices should be positive and reasonable + for rows in candidates.values(): + for row in rows: + assert row["entry_price"] > 0 + + def test_event_timestamp_format(self): + """event_timestamp should be ISO with timezone.""" + rng = np.random.default_rng(42) + dist = EventDistribution() + candidates = generate_events(_TRADING_DATES, _SYMBOLS, dist, rng) + for rows in candidates.values(): + for row in rows: + ts = row["event_timestamp"] + assert "T" in ts, f"event_timestamp not ISO: {ts}" + assert "+00:00" in ts or "Z" in ts, f"event_timestamp missing TZ: {ts}" + + def test_produces_events_with_default_distribution(self): + """Default distribution should produce at least some events.""" + rng = np.random.default_rng(99) + dist = EventDistribution() + candidates = generate_events(_TRADING_DATES, _SYMBOLS, dist, rng) + total = sum(len(v) for v in candidates.values()) + assert total > 50, f"Too few events generated: {total}" diff --git a/tests/unit/backtest/test_scenario_robustness.py b/tests/unit/backtest/test_scenario_robustness.py new file mode 100644 index 0000000..257e13f --- /dev/null +++ b/tests/unit/backtest/test_scenario_robustness.py @@ -0,0 +1,219 @@ +"""Unit tests for scenario robustness.py — RRS computation logic.""" +import pytest + +from libs.backtest.domain import MetricsBundle +from libs.backtest.scenarios.robustness import ( + RegimeRobustnessReport, + ScenarioResult, + compute_rrs, +) + + +def _make_result( + name: str, + sharpe: float = 1.0, + total_return: float = 10.0, + max_dd: float = 10.0, + win_rate: float = 0.55, + profit_factor: float = 1.5, + trade_count: int = 20, +) -> ScenarioResult: + """Helper: construct a minimal ScenarioResult.""" + m = MetricsBundle( + trade_count=trade_count, + win_rate=win_rate, + avg_win_pct=2.0, + avg_loss_pct=-1.0, + profit_factor=profit_factor, + total_return_pct=total_return, + max_drawdown_pct=max_dd, + sharpe_ratio=sharpe, + ) + return ScenarioResult( + scenario_name=name, + metrics=m, + trade_count=trade_count, + sharpe_ratio=sharpe, + total_return_pct=total_return, + max_drawdown_pct=max_dd, + win_rate=win_rate, + profit_factor=profit_factor, + ) + + +@pytest.mark.unit +class TestComputeRRS: + def test_empty_results_returns_zeros(self): + rrs, si, br, dr, rt, st = compute_rrs({}) + assert rrs == 0.0 + assert si == 0.0 + assert br == 0.0 + + def test_signal_integrity_perfect_when_no_signal_sharpe_zero(self): + results = { + "no_signal": _make_result("no_signal", sharpe=0.0), + "strong_signal": _make_result("strong_signal", sharpe=1.5), + } + _, si, _, _, _, _ = compute_rrs(results) + assert si == 100.0 + + def test_signal_integrity_perfect_when_no_signal_sharpe_negative(self): + results = { + "no_signal": _make_result("no_signal", sharpe=-0.5), + "strong_signal": _make_result("strong_signal", sharpe=1.5), + } + _, si, _, _, _, _ = compute_rrs(results) + assert si == 100.0 + + def test_signal_integrity_zero_when_no_signal_equals_strong_signal(self): + results = { + "no_signal": _make_result("no_signal", sharpe=1.5), + "strong_signal": _make_result("strong_signal", sharpe=1.5), + } + _, si, _, _, _, _ = compute_rrs(results) + assert si == pytest.approx(0.0, abs=1.0) + + def test_signal_integrity_neutral_when_no_signal_scenario_missing(self): + results = {"steady_bull": _make_result("steady_bull", sharpe=1.0)} + _, si, _, _, _, _ = compute_rrs(results) + assert si == 50.0 + + def test_breadth_all_positive(self): + results = { + "a": _make_result("a", sharpe=0.5), + "b": _make_result("b", sharpe=1.0), + "c": _make_result("c", sharpe=0.1), + } + _, _, br, _, _, _ = compute_rrs(results) + assert br == pytest.approx(100.0) + + def test_breadth_all_negative(self): + results = { + "a": _make_result("a", sharpe=-0.5), + "b": _make_result("b", sharpe=-1.0), + } + _, _, br, _, _, _ = compute_rrs(results) + assert br == pytest.approx(0.0) + + def test_breadth_half_positive(self): + results = { + "a": _make_result("a", sharpe=1.0), + "b": _make_result("b", sharpe=-1.0), + } + _, _, br, _, _, _ = compute_rrs(results) + assert br == pytest.approx(50.0) + + def test_drawdown_resilience_zero_dd_gives_100(self): + results = {"a": _make_result("a", max_dd=0.0)} + _, _, _, dr, _, _ = compute_rrs(results) + assert dr == pytest.approx(100.0) + + def test_drawdown_resilience_50pct_dd_gives_0(self): + results = {"a": _make_result("a", max_dd=50.0)} + _, _, _, dr, _, _ = compute_rrs(results) + assert dr == pytest.approx(0.0) + + def test_drawdown_resilience_clamped_not_negative(self): + results = {"a": _make_result("a", max_dd=80.0)} + _, _, _, dr, _, _ = compute_rrs(results) + assert dr >= 0.0 + + def test_regime_transition_neutral_when_missing(self): + results = {"steady_bull": _make_result("steady_bull", sharpe=1.0)} + _, _, _, _, rt, _ = compute_rrs(results) + assert rt == 50.0 + + def test_regime_transition_good_when_above_median(self): + results = { + "a": _make_result("a", sharpe=0.5), + "b": _make_result("b", sharpe=0.5), + "regime_switch": _make_result("regime_switch", sharpe=2.0), + } + _, _, _, _, rt, _ = compute_rrs(results) + # regime_switch > median → should be > 50 + assert rt > 50.0 + + def test_stability_perfect_when_all_same_sharpe(self): + results = { + "a": _make_result("a", sharpe=1.0), + "b": _make_result("b", sharpe=1.0), + "c": _make_result("c", sharpe=1.0), + } + _, _, _, _, _, st = compute_rrs(results) + assert st == pytest.approx(100.0) + + def test_stability_lower_when_high_variance(self): + low_var = { + "a": _make_result("a", sharpe=1.0), + "b": _make_result("b", sharpe=1.1), + "c": _make_result("c", sharpe=0.9), + } + high_var = { + "a": _make_result("a", sharpe=3.0), + "b": _make_result("b", sharpe=-0.5), + "c": _make_result("c", sharpe=0.1), + } + _, _, _, _, _, st_low = compute_rrs(low_var) + _, _, _, _, _, st_high = compute_rrs(high_var) + assert st_low > st_high + + def test_rrs_weighted_sum(self): + """RRS = 0.25*si + 0.25*br + 0.20*dr + 0.15*rt + 0.15*st.""" + results = { + "no_signal": _make_result("no_signal", sharpe=-0.1), + "strong_signal": _make_result("strong_signal", sharpe=1.5), + "steady_bull": _make_result("steady_bull", sharpe=1.2), + "steady_bear": _make_result("steady_bear", sharpe=0.3), + "regime_switch": _make_result("regime_switch", sharpe=0.8), + } + rrs, si, br, dr, rt, st = compute_rrs(results) + expected = 0.25 * si + 0.25 * br + 0.20 * dr + 0.15 * rt + 0.15 * st + assert rrs == pytest.approx(expected, rel=1e-6) + + def test_rrs_bounded_0_to_100(self): + results = { + "no_signal": _make_result("no_signal", sharpe=-0.5), + "strong_signal": _make_result("strong_signal", sharpe=2.0), + "a": _make_result("a", sharpe=1.0), + "b": _make_result("b", sharpe=1.5), + } + rrs, _, _, _, _, _ = compute_rrs(results) + assert 0.0 <= rrs <= 100.0 + + +@pytest.mark.unit +class TestVerdicts: + def _make_report(self, rrs: float) -> RegimeRobustnessReport: + from libs.backtest.scenarios.robustness import _verdict + results = {"a": _make_result("a", sharpe=1.0)} + return RegimeRobustnessReport( + experiment_name="test", + scenario_results=results, + signal_integrity=50.0, + breadth=50.0, + drawdown_resilience=50.0, + regime_transition=50.0, + stability=50.0, + rrs=rrs, + verdict=_verdict(rrs), + ) + + def test_robust_verdict(self): + r = self._make_report(75.0) + assert r.verdict == "ROBUST" + + def test_fragile_verdict_lower_bound(self): + r = self._make_report(40.0) + assert r.verdict == "FRAGILE" + + def test_fragile_verdict_upper_bound(self): + r = self._make_report(69.9) + assert r.verdict == "FRAGILE" + + def test_overfit_verdict(self): + r = self._make_report(39.9) + assert r.verdict == "OVERFIT" + + def test_boundary_70_is_robust(self): + r = self._make_report(70.0) + assert r.verdict == "ROBUST" diff --git a/tests/unit/backtest/test_store_builder.py b/tests/unit/backtest/test_store_builder.py new file mode 100644 index 0000000..e65c12b --- /dev/null +++ b/tests/unit/backtest/test_store_builder.py @@ -0,0 +1,104 @@ +"""Unit tests for store_builder.py.""" +import datetime as dt + +import numpy as np +import pytest + +from libs.backtest.scenarios.price_gen import PriceRegime +from libs.backtest.scenarios.scenarios import SCENARIO_REGISTRY, STEADY_BULL, NO_SIGNAL +from libs.backtest.scenarios.store_builder import _get_trading_dates, build_synthetic_store +from libs.backtest.snapshot_store import SnapshotStore + + +@pytest.mark.unit +class TestGetTradingDates: + def test_returns_correct_count(self): + dates = _get_trading_dates(50) + assert len(dates) == 50 + + def test_dates_are_weekdays(self): + dates = _get_trading_dates(30) + for d in dates: + assert d.weekday() < 5, f"Non-weekday in trading dates: {d}" + + def test_dates_are_sorted_ascending(self): + dates = _get_trading_dates(30) + assert dates == sorted(dates) + + def test_starts_on_or_after_start_date(self): + start = dt.date(2024, 1, 2) + dates = _get_trading_dates(20, start=start) + assert dates[0] >= start + + +@pytest.mark.unit +class TestBuildSyntheticStore: + def test_returns_snapshot_store(self): + store = build_synthetic_store(STEADY_BULL, rng=np.random.default_rng(42)) + assert isinstance(store, SnapshotStore) + + def test_store_has_candidates(self): + store = build_synthetic_store(STEADY_BULL, rng=np.random.default_rng(42)) + exec_dates = store.all_execution_dates() + assert len(exec_dates) > 0, "No candidates generated" + + def test_store_has_macro_data(self): + store = build_synthetic_store(STEADY_BULL, rng=np.random.default_rng(42)) + trading_days = store.all_trading_days() + assert len(trading_days) > 0 + + def test_store_has_bar_data(self): + store = build_synthetic_store(STEADY_BULL, rng=np.random.default_rng(42)) + trading_days = store.all_trading_days() + assert len(trading_days) > 0 + + def test_spy_and_qqq_in_bars(self): + store = build_synthetic_store(STEADY_BULL, rng=np.random.default_rng(42)) + trading_days = store.all_trading_days() + some_day = trading_days[10] + assert store.get_bar("SPY", some_day) is not None + assert store.get_bar("QQQ", some_day) is not None + + def test_macro_has_required_keys(self): + store = build_synthetic_store(STEADY_BULL, rng=np.random.default_rng(42)) + trading_days = store.all_trading_days() + # Pick a date past rolling window warm-up + late_date = trading_days[60] + macro = store.get_macro_for_date(late_date) + for key in ("VIXCLS", "macro_vix", "macro_hy_spread", "spy_close", "qqq_close"): + assert key in macro, f"Missing macro key: {key}" + + def test_seed_is_deterministic(self): + store_a = build_synthetic_store(STEADY_BULL, rng=np.random.default_rng(123)) + store_b = build_synthetic_store(STEADY_BULL, rng=np.random.default_rng(123)) + # Same seed → same exec dates + assert store_a.all_execution_dates() == store_b.all_execution_dates() + + def test_no_signal_scenario_produces_store(self): + """no_signal scenario should assemble without error.""" + store = build_synthetic_store(NO_SIGNAL, rng=np.random.default_rng(1)) + assert isinstance(store, SnapshotStore) + + def test_scenario_seed_used_when_rng_none(self): + """If rng=None, scenario.seed is used for determinism.""" + from dataclasses import replace + seeded_scenario = replace(STEADY_BULL, seed=77) + store_a = build_synthetic_store(seeded_scenario, rng=None) + store_b = build_synthetic_store(seeded_scenario, rng=None) + assert store_a.all_execution_dates() == store_b.all_execution_dates() + + def test_candidate_exec_dates_align_with_macro_dates(self): + """All candidate exec_dates should exist in trading days (have macro).""" + store = build_synthetic_store(STEADY_BULL, rng=np.random.default_rng(42)) + macro_trading_days = set(store.all_trading_days()) + for exec_date in store.all_execution_dates(): + assert exec_date in macro_trading_days, f"exec_date {exec_date} not in trading days" + + def test_all_registered_scenarios_can_build(self): + """Smoke-test: every scenario in SCENARIO_REGISTRY builds without error.""" + for name, scenario in SCENARIO_REGISTRY.items(): + try: + store = build_synthetic_store(scenario, rng=np.random.default_rng(0)) + assert isinstance(store, SnapshotStore), f"Bad store for {name}" + except Exception as exc: + pytest.fail(f"build_synthetic_store failed for scenario '{name}': {exc}")