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