"""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, )