From ce2150789da0493b99c503a6f2998d63ab299a48 Mon Sep 17 00:00:00 2001 From: I Luk Kim Date: Thu, 26 Mar 2026 20:15:16 -0700 Subject: [PATCH] =?UTF-8?q?Fix=20leaderboard=20performance=20regression=20?= =?UTF-8?q?(60min=20=E2=86=92=2012s)=20and=20clean=20up=20CLI=20help?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Replaces per-experiment rglob with single-pass manifest/metrics indexing and adds lru_cache. Removes rarely-used commands from help display. Co-Authored-By: Claude Opus 4.6 (1M context) --- apps/tracker/cli.py | 1040 +++++++++++++++---- libs/backtest/tracker.py | 2136 ++++++++++++++++++++++++++++++++++++-- 2 files changed, 2935 insertions(+), 241 deletions(-) diff --git a/apps/tracker/cli.py b/apps/tracker/cli.py index 8a55fae..d051a30 100644 --- a/apps/tracker/cli.py +++ b/apps/tracker/cli.py @@ -2,6 +2,8 @@ from __future__ import annotations import argparse +import datetime as dt +import json import sys from pathlib import Path @@ -9,33 +11,223 @@ from rich import box from rich.console import Console from rich.panel import Panel from rich.table import Table -from rich.text import Text - -_console = Console(width=140) from libs.backtest.domain import ( ConfigDelta, JournalEntry, - MetricsBundle, + OverlayWindowSummary, + RobustnessMatrixSummary, SplitResult, + WalkForwardSummary, ) from libs.backtest.tracker import ( + _hydrate_split_result, append_journal_entry, + attach_out_of_time_robustness_summary, + attach_robustness_summary, + attach_walk_forward_summary, build_split_result, check_duplicate, + compute_deployment_score, compute_public_sqs, + compute_public_sqs_v2, compute_promotion_score, + compute_oot_robustness_gate, + compute_overlay_public_sqs, + compute_overlay_stress_sqs, + compute_robustness_gate, + compute_rqs, compute_sqs, compute_sqs_v2, compute_unified_score, + compute_wfqs, + compute_wfqs_v2, + filter_registry_entries, get_next_entry_id, journal_lock, load_journal, + refresh_public_scores, rebuild_registry, scan_runs_for_experiment, + sync_official_manifests, ) from libs.common.time_utils import utc_now +_console = Console(width=140) +_DEFAULT_JOURNAL_DIR = "journal" +_COL_NAME = 38 + + +def _fmt(val: float | None, fmt: str) -> str: + return format(val, fmt) if val is not None else "-" + + +def _ratio(numerator: float | None, denominator: float | None) -> float | None: + if numerator is None or denominator is None or denominator == 0: + return None + return numerator / denominator + + +def _entry_matches_target(entry: JournalEntry, target: str) -> bool: + target_upper = target.upper() + return entry.entry_id == target_upper or entry.experiment_name == target + + +def _ordered_splits(results: dict[str, SplitResult]) -> list[tuple[str, SplitResult]]: + preferred = ["train", "valid", "test"] + ordered: list[tuple[str, SplitResult]] = [] + seen: set[str] = set() + for name in preferred: + if name in results: + ordered.append((name, results[name])) + seen.add(name) + for name in sorted(results.keys()): + if name not in seen: + ordered.append((name, results[name])) + return ordered + + +def _sync_and_rebuild( + journal_path: Path, + registry_path: Path, + leaderboard_path: Path, +) -> object: + sync_official_manifests(journal_path, Path("runs")) + return rebuild_registry(journal_path, registry_path, leaderboard_path) + + +def _load_optional_wfv_summary(path_str: str | None) -> WalkForwardSummary | None: + if not path_str: + return None + path = Path(path_str) + if not path.exists(): + print(f"ERROR: walk-forward summary not found: {path}") + sys.exit(1) + return WalkForwardSummary.model_validate_json(path.read_text()) + + +def _load_optional_robustness_summary(path_str: str | None) -> RobustnessMatrixSummary | None: + if not path_str: + return None + path = Path(path_str) + if not path.exists(): + print(f"ERROR: robustness summary not found: {path}") + sys.exit(1) + return RobustnessMatrixSummary.model_validate_json(path.read_text()) + + +def _build_overlay_window_summary(spec_path: str, summary_path: str) -> OverlayWindowSummary: + spec = json.loads(Path(spec_path).read_text()) + summary = json.loads(Path(summary_path).read_text()) + + start_date = spec.get("start_date") + end_date = spec.get("end_date") + if not start_date or not end_date: + print("ERROR: overlay spec must include start_date and end_date") + sys.exit(1) + + start = dt.date.fromisoformat(start_date) + end = dt.date.fromisoformat(end_date) + initial_equity = float(spec.get("initial_equity", 10_000.0)) + final_equity = summary.get("final_equity") + annualized_return_pct = None + if final_equity is not None and initial_equity > 0: + calendar_days = max((end - start).days, 1) + annualized_return_pct = ((float(final_equity) / initial_equity) ** (365.25 / calendar_days) - 1.0) * 100.0 + + return OverlayWindowSummary( + window_name=summary.get("overlay_name", Path(spec_path).stem), + overlay_name=summary.get("overlay_name", Path(spec_path).stem), + start_date=start, + end_date=end, + initial_equity=initial_equity, + final_equity=summary.get("final_equity"), + return_pct=summary.get("return_pct"), + annualized_return_pct=annualized_return_pct, + max_drawdown_pct=summary.get("max_dd_pct"), + sharpe_ratio=summary.get("sharpe"), + day_count=int(summary.get("day_count") or 0), + books=list(summary.get("books") or []), + allocations=dict(summary.get("allocations") or {}), + regime_day_counts=dict(summary.get("regime_day_counts") or {}), + ) + + +def _score_sort_key(sort_by: str, entry) -> tuple: + if sort_by == "promotion": + return ( + entry.promotion_score is None, + -(entry.promotion_score or 0.0), + -(entry.sqs_score or 0.0), + ) + if sort_by in {"deployment", "dep"}: + return ( + entry.deployment_score is None, + -(entry.deployment_score or 0.0), + entry.wfqs_score is None, + -(entry.wfqs_score or 0.0), + -(entry.rqs_score or 0.0), + -(entry.sqs_score or 0.0), + ) + if sort_by == "wfqs": + return ( + entry.wfqs_score is None, + -(entry.wfqs_score or 0.0), + -(entry.deployment_score or 0.0), + -(entry.rqs_score or 0.0), + -(entry.sqs_score or 0.0), + ) + if sort_by == "rqs": + return ( + entry.rqs_score is None, + -(entry.rqs_score or 0.0), + -(entry.deployment_score or 0.0), + -(entry.sqs_score or 0.0), + ) + if sort_by == "unified": + return ( + entry.unified_score is None, + -(entry.unified_score or 0.0), + -(entry.sqs_score or 0.0), + ) + if sort_by == "sqs2": + return ( + entry.wfqs_v2_score is None, + -(entry.wfqs_v2_score or 0.0), + -(entry.wfqs_score or 0.0), + -(entry.sqs_score or 0.0), + ) + return (-(entry.sqs_score or 0.0),) + + +def _validation_pending_reason(entry: JournalEntry) -> str | None: + missing: list[str] = [] + if entry.walk_forward_summary is None: + missing.append("WFV") + if entry.robustness_matrix_summary is None: + missing.append("robustness") + if entry.out_of_time_robustness_summary is None: + missing.append("OOT robustness") + if not missing: + return None + return "missing " + " + ".join(missing) + + +def _title_for_sort(sort_by: str) -> str: + if sort_by == "promotion": + return "Promotion 기준 내림차순" + if sort_by in {"deployment", "dep"}: + return "Deployment 기준 내림차순" + if sort_by == "wfqs": + return "WFQS 기준 내림차순" + if sort_by == "rqs": + return "RQS 기준 내림차순" + if sort_by == "unified": + return "Unified 기준 내림차순" + if sort_by == "sqs2": + return "WFQS v2 기준 내림차순" + return "SQS 기준 내림차순" + def cmd_record(args: argparse.Namespace) -> None: """Record an experiment to the improvement journal.""" @@ -44,7 +236,6 @@ def cmd_record(args: argparse.Namespace) -> None: registry_path = journal_dir / "experiment_registry.json" leaderboard_path = journal_dir / "LEADERBOARD.md" - # Scan runs runs_dir = Path(args.runs_dir) if not runs_dir.exists(): print(f"ERROR: runs directory not found: {runs_dir}") @@ -55,12 +246,10 @@ def cmd_record(args: argparse.Namespace) -> None: print(f"ERROR: no runs found for experiment '{args.experiment}' in {runs_dir}") sys.exit(1) - # Build results results: dict[str, SplitResult] = {} for split_name, (run_id, metrics) in split_runs.items(): results[split_name] = build_split_result(split_name, run_id, metrics) - # Compute public SQS from valid+test when available, or test split fallback. test_metrics = None for preferred in ["test", "valid", "train"]: if preferred in split_runs: @@ -71,38 +260,77 @@ def cmd_record(args: argparse.Namespace) -> None: sqs_breakdown: dict[str, float] = {} sqs_v2_score = None sqs_v2_breakdown: dict[str, float] = {} - promotion_score = None - promotion_breakdown: dict[str, float] = {} - unified_score = None - unified_breakdown: dict[str, float] = {} - if test_metrics: - legacy_sqs_score, legacy_sqs_breakdown = compute_sqs(test_metrics) + if test_metrics is not None: + sqs_score, sqs_breakdown = compute_sqs(test_metrics) sqs_v2_score, sqs_v2_breakdown = compute_sqs_v2(test_metrics) - sqs_score = legacy_sqs_score - sqs_breakdown = legacy_sqs_breakdown + promotion_score, promotion_breakdown = compute_promotion_score( results.get("test"), results.get("valid"), ) + unified_score, unified_breakdown = compute_unified_score( + results.get("test"), + results.get("valid"), + ) + rqs_score, rqs_breakdown = compute_rqs( + results.get("train"), + results.get("valid"), + results.get("test"), + ) + + walk_forward_summary = _load_optional_wfv_summary(args.walk_forward_summary) + robustness_matrix_summary = _load_optional_robustness_summary(args.robustness_summary) + out_of_time_robustness_summary = _load_optional_robustness_summary( + getattr(args, "out_of_time_robustness_summary", None) + ) + + wfqs_score = None + wfqs_breakdown: dict[str, float] = {} + wfqs_v2_score = None + wfqs_v2_breakdown: dict[str, float] = {} + deployment_score = None + deployment_breakdown: dict[str, float] = {} + if walk_forward_summary is not None: + wfqs_score, wfqs_breakdown = compute_wfqs(walk_forward_summary) + wfqs_v2_score, wfqs_v2_breakdown = compute_wfqs_v2(walk_forward_summary) + deployment_score, deployment_breakdown = compute_deployment_score( + results.get("train"), + results.get("valid"), + results.get("test"), + walk_forward_summary, + rqs_score=rqs_score, + wfqs_score=wfqs_score, + ) + public_sqs, public_breakdown, public_source = compute_public_sqs( + results.get("train"), + results.get("valid"), results.get("test"), + walk_forward_summary=walk_forward_summary, + robustness_matrix_summary=robustness_matrix_summary, + out_of_time_robustness_summary=out_of_time_robustness_summary, + rqs_score=rqs_score, + wfqs_score=wfqs_v2_score, + ) + stress_sqs, stress_breakdown, _ = compute_public_sqs_v2( + results.get("train"), results.get("valid"), + results.get("test"), + walk_forward_summary=walk_forward_summary, + robustness_matrix_summary=robustness_matrix_summary, + out_of_time_robustness_summary=out_of_time_robustness_summary, + rqs_score=rqs_score, + wfqs_v2_score=wfqs_v2_score, ) if public_sqs is not None: sqs_score = public_sqs sqs_breakdown = public_breakdown - unified_score, unified_breakdown = compute_unified_score( - results.get("test"), - results.get("valid"), - ) - # Build config delta config_delta = None if args.baseline: config_delta = ConfigDelta(base_experiment=args.baseline, changes={}) - # Build tags from experiment name - tags = [t for t in args.experiment.replace("-", "_").split("_") if t] + tags = [tag for tag in args.experiment.replace("-", "_").split("_") if tag] with journal_lock(journal_path): dupes = check_duplicate(journal_path, args.experiment) @@ -119,14 +347,27 @@ def cmd_record(args: argparse.Namespace) -> None: hypothesis=args.hypothesis or "", config_delta=config_delta, results=results, + walk_forward_summary=walk_forward_summary, + robustness_matrix_summary=robustness_matrix_summary, + out_of_time_robustness_summary=out_of_time_robustness_summary, sqs_score=sqs_score, sqs_breakdown=sqs_breakdown, + stress_sqs_score=stress_sqs, + stress_sqs_breakdown=stress_breakdown, sqs_v2_score=sqs_v2_score, sqs_v2_breakdown=sqs_v2_breakdown, promotion_score=promotion_score, promotion_breakdown=promotion_breakdown, unified_score=unified_score, unified_breakdown=unified_breakdown, + rqs_score=rqs_score, + rqs_breakdown=rqs_breakdown, + wfqs_score=wfqs_score, + wfqs_breakdown=wfqs_breakdown, + wfqs_v2_score=wfqs_v2_score, + wfqs_v2_breakdown=wfqs_v2_breakdown, + deployment_score=deployment_score, + deployment_breakdown=deployment_breakdown, verdict=args.verdict or "unknown", verdict_reasoning=args.reasoning or "", next_direction=args.next or "", @@ -134,26 +375,92 @@ def cmd_record(args: argparse.Namespace) -> None: ) append_journal_entry(journal_path, entry) - rebuild_registry(journal_path, registry_path, leaderboard_path) - source_label = f", source={public_source}" if public_source else "" - print(f"Recorded {entry_id}: {args.experiment} (SQS={sqs_score}{source_label})") + _sync_and_rebuild(journal_path, registry_path, leaderboard_path) - # Show splits found - for split_name, sr in results.items(): + if sqs_score is None: + print(f"Recorded {entry_id}: {args.experiment} (SQS=pending validation)") + else: + source_label = f", source={public_source}" if public_source else "" + stress_label = f", stress={stress_sqs:.1f}" if stress_sqs is not None else "" + print(f"Recorded {entry_id}: {args.experiment} (SQS={sqs_score}{stress_label}{source_label})") + for split_name, sr in _ordered_splits(results): pf = f"PF={sr.profit_factor:.2f}" if sr.profit_factor is not None else "PF=-" ret = f"Ret={sr.total_return_pct:+.2f}%" if sr.total_return_pct is not None else "Ret=-" print(f" {split_name}: {sr.trade_count} trades, {pf}, {ret}") - print(f"Leaderboard updated: {leaderboard_path}") -_DEFAULT_JOURNAL_DIR = "journal" +def cmd_record_overlay(args: argparse.Namespace) -> None: + """Record an overlay evaluation to the official journal and leaderboard.""" + journal_dir = Path(args.journal_dir) + journal_path = journal_dir / "improvement_journal.jsonl" + registry_path = journal_dir / "experiment_registry.json" + leaderboard_path = journal_dir / "LEADERBOARD.md" -_COL_NAME = 38 # max experiment name width before truncation + overlay_summary = _build_overlay_window_summary(args.spec, args.summary) + overlay_stress_summary = _build_overlay_window_summary(args.stress_spec, args.stress_summary) + sqs_score, sqs_breakdown, source = compute_overlay_public_sqs( + overlay_summary, + overlay_stress_summary, + ) + stress_sqs_score, stress_sqs_breakdown, _ = compute_overlay_stress_sqs( + overlay_summary, + overlay_stress_summary, + ) + if sqs_score is None: + print(f"ERROR: overlay score unavailable ({source})") + sys.exit(1) + spec = json.loads(Path(args.spec).read_text()) + experiment_name = spec.get("overlay_name") or Path(args.spec).stem + tags = ["overlay", *list(getattr(args, "tags", []) or [])] -def _fmt(val: float | None, fmt: str) -> str: - return format(val, fmt) if val is not None else "-" + with journal_lock(journal_path): + dupes = check_duplicate(journal_path, experiment_name) + if dupes and not args.force: + print(f"WARNING: overlay '{experiment_name}' already in journal ({len(dupes)} entries).") + print("Use --force to add anyway.") + sys.exit(1) + + entry_id = get_next_entry_id(journal_path) + entry = JournalEntry( + entry_id=entry_id, + timestamp=utc_now().isoformat(), + experiment_name=experiment_name, + hypothesis=args.hypothesis or "", + overlay_common_window_summary=overlay_summary, + overlay_stress_window_summary=overlay_stress_summary, + sqs_score=sqs_score, + sqs_breakdown=sqs_breakdown, + sqs_v3_score=sqs_score, + sqs_v3_breakdown=sqs_breakdown, + stress_sqs_score=stress_sqs_score, + stress_sqs_breakdown=stress_sqs_breakdown, + verdict=args.verdict or "unknown", + verdict_reasoning=args.reasoning or "", + next_direction=args.next or "", + tags=tags, + ) + append_journal_entry(journal_path, entry) + _sync_and_rebuild(journal_path, registry_path, leaderboard_path) + + stress_label = f", stress={stress_sqs_score:.1f}" if stress_sqs_score is not None else "" + print(f"Recorded {entry_id}: {experiment_name} (SQS={sqs_score:.1f}{stress_label}, source={source})") + print( + " common-window:" + f" Ret={overlay_summary.return_pct:+.2f}%" + f", Ann={overlay_summary.annualized_return_pct:+.2f}%" + f", DD={overlay_summary.max_drawdown_pct:.2f}%" + f", Sharpe={overlay_summary.sharpe_ratio:.2f}" + ) + print( + " stress-window:" + f" Ret={overlay_stress_summary.return_pct:+.2f}%" + f", Ann={overlay_stress_summary.annualized_return_pct:+.2f}%" + f", DD={overlay_stress_summary.max_drawdown_pct:.2f}%" + f", Sharpe={overlay_stress_summary.sharpe_ratio:.2f}" + ) + print(f"Leaderboard updated: {leaderboard_path}") def cmd_leaderboard(args: argparse.Namespace) -> None: @@ -164,26 +471,21 @@ def cmd_leaderboard(args: argparse.Namespace) -> None: leaderboard_path = journal_dir / "LEADERBOARD.md" if not journal_path.exists(): - print("No journal found. Run 'record' first.") - sys.exit(1) + journal_path.parent.mkdir(parents=True, exist_ok=True) + journal_path.touch() - registry = rebuild_registry(journal_path, registry_path, leaderboard_path) + with journal_lock(journal_path): + registry = _sync_and_rebuild(journal_path, registry_path, leaderboard_path) + ranked_source = filter_registry_entries( + registry.entries, + include_retired=getattr(args, "include_retired", False), + ) - top_n = getattr(args, "top", 10) sort_by = getattr(args, "sort", "sqs") - if sort_by == "promotion": - ranked_entries = sorted( - registry.entries, - key=lambda e: (e.promotion_score is None, -(e.promotion_score or 0.0), -e.sqs_score), - ) - title_mode = "Promotion 기준 내림차순" - elif sort_by in {"sqs", "unified"}: - ranked_entries = sorted(registry.entries, key=lambda e: -e.sqs_score) - title_mode = "SQS 기준 내림차순" - else: - ranked_entries = sorted(registry.entries, key=lambda e: -e.sqs_score) - title_mode = "SQS 기준 내림차순" + ranked_entries = sorted(ranked_source, key=lambda entry: _score_sort_key(sort_by, entry)) total = len(ranked_entries) + top_n = getattr(args, "top", 10) + diagnostics = sort_by in {"deployment", "dep", "wfqs", "rqs", "promotion", "unified", "sqs2"} tbl = Table( box=box.SIMPLE_HEAD, @@ -191,40 +493,59 @@ def cmd_leaderboard(args: argparse.Namespace) -> None: header_style="bold yellow", row_styles=["", "dim"], padding=(0, 1), - title=f"[bold cyan]Top {top_n} / {total}[/] [dim]· {title_mode} · T=test V=valid[/]", + title=f"[bold cyan]Top {top_n} / {total}[/] [dim]· {_title_for_sort(sort_by)} · Tr=train V=valid T=test[/]", title_justify="left", expand=False, ) - # Columns: #, Experiment, SQS | Test: PF / Ret% / WR / DD% / N | Valid: PF / Ret% / WR / N - 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) - tbl.add_column("T.PF", justify="right", no_wrap=True, min_width=5) - tbl.add_column("T.Ret%", justify="right", no_wrap=True, min_width=6) - tbl.add_column("T.WR%", justify="right", no_wrap=True, min_width=5) - tbl.add_column("T.DD%", justify="right", no_wrap=True, min_width=5) - tbl.add_column("T.N", justify="right", no_wrap=True, min_width=4) - tbl.add_column("V.PF", justify="right", style="green", no_wrap=True, min_width=5) - tbl.add_column("V.Ret%", justify="right", style="green", no_wrap=True, min_width=6) - tbl.add_column("V.WR%", justify="right", style="green", no_wrap=True, min_width=5) - tbl.add_column("V.N", justify="right", style="green", no_wrap=True, min_width=4) - - for rank, e in enumerate(ranked_entries[:top_n], 1): - name = e.experiment_name + 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) + if diagnostics: + if sort_by == "promotion": + tbl.add_column("Promo", justify="right", style="bright_magenta", no_wrap=True, min_width=5) + else: + tbl.add_column("DEP", justify="right", style="bold green", no_wrap=True, min_width=5) + tbl.add_column("WFQS", justify="right", style="bright_cyan", no_wrap=True, min_width=5) + tbl.add_column("RQS", justify="right", style="bold magenta", no_wrap=True, min_width=5) + tbl.add_column("Tr.Ret%", justify="right", style="bright_blue", no_wrap=True, min_width=7) + tbl.add_column("V.Ret%", justify="right", style="green", no_wrap=True, min_width=6) + tbl.add_column("T.Ret%", justify="right", no_wrap=True, min_width=6) + tbl.add_column("T.Ann%", justify="right", no_wrap=True, min_width=7) + tbl.add_column("T.DD%", justify="right", no_wrap=True, min_width=5) + tbl.add_column("T.Gross%", justify="right", no_wrap=True, min_width=8) + tbl.add_column("T.DIM%", justify="right", no_wrap=True, min_width=6) + tbl.add_column("T.R/G", justify="right", no_wrap=True, min_width=5) + + for rank, entry in enumerate(ranked_entries[:top_n], 1): + name = entry.experiment_name if len(name) > _COL_NAME: name = name[: _COL_NAME - 1] + "…" - pf = _fmt(e.profit_factor, ".2f") - ret = _fmt(e.total_return_pct, "+.1f") - wr = _fmt(e.win_rate * 100 if e.win_rate is not None else None, ".0f") - dd = _fmt(e.max_drawdown_pct, ".1f") - vpf = _fmt(e.valid_profit_factor, ".2f") - vret = _fmt(e.valid_total_return_pct, "+.1f") - vwr = _fmt(e.valid_win_rate * 100 if e.valid_win_rate is not None else None, ".0f") - tbl.add_row( - str(rank), name, f"{e.sqs_score:.1f}", - pf, ret, wr, dd, str(e.trade_count), - vpf, vret, vwr, str(e.valid_trade_count) if e.valid_trade_count else "-", - ) + is_overlay = entry.strategy_family == "overlay" and entry.overlay_common_window_summary is not None + row = [ + str(rank), + name, + _fmt(entry.sqs_score, ".1f"), + ] + if diagnostics: + if sort_by == "promotion": + row.append(_fmt(entry.promotion_score, ".1f")) + else: + row.extend([ + _fmt(entry.deployment_score, ".1f"), + _fmt(entry.wfqs_score, ".1f"), + _fmt(entry.rqs_score, ".1f"), + ]) + row.extend([ + "-" if is_overlay else _fmt(entry.train_total_return_pct, "+.1f"), + "-" if is_overlay else _fmt(entry.valid_total_return_pct, "+.1f"), + _fmt(entry.total_return_pct, "+.1f"), + _fmt(entry.annualized_return_pct, "+.1f"), + _fmt(entry.max_drawdown_pct, ".1f"), + "-" if is_overlay else _fmt(entry.avg_gross_exposure_pct, ".1f"), + "-" if is_overlay else _fmt(entry.days_in_market_pct, ".1f"), + "-" if is_overlay else _fmt(_ratio(entry.total_return_pct, entry.avg_gross_exposure_pct), ".2f"), + ]) + tbl.add_row(*row) _console.print() _console.print(tbl) @@ -235,55 +556,223 @@ def cmd_show(args: argparse.Namespace) -> None: """Show details of a specific journal entry.""" journal_dir = Path(args.journal_dir) journal_path = journal_dir / "improvement_journal.jsonl" - entries = load_journal(journal_path) - - target = args.entry_id.upper() - found = [e for e in entries if e.entry_id == target or e.experiment_name == target] + registry_path = journal_dir / "experiment_registry.json" + leaderboard_path = journal_dir / "LEADERBOARD.md" + with journal_lock(journal_path): + _sync_and_rebuild(journal_path, registry_path, leaderboard_path) + entries = load_journal(journal_path) + target = args.entry_id + found = [entry for entry in entries if _entry_matches_target(entry, target)] if not found: - # Try partial match - found = [e for e in entries if target.lower() in e.experiment_name.lower()] - + found = [entry for entry in entries if target.lower() in entry.experiment_name.lower()] if not found: - print(f"No entry found for: {args.entry_id}") + print(f"No entry found for: {target}") sys.exit(1) - for e in found: - promotion_score = e.promotion_score - promotion_breakdown = e.promotion_breakdown - if promotion_score is None: - promotion_score, promotion_breakdown = compute_promotion_score( - e.results.get("test"), - e.results.get("valid"), - ) + for entry in found: + train_result = _hydrate_split_result(entry.results.get("train")) + valid_result = _hydrate_split_result(entry.results.get("valid")) + test_result = _hydrate_split_result(entry.results.get("test")) + fresh_rqs_score, fresh_rqs_breakdown = compute_rqs( + train_result, + valid_result, + test_result, + ) + fresh_wfqs_score, fresh_wfqs_breakdown = compute_wfqs(entry.walk_forward_summary) + fresh_wfqs_v2_score, _ = compute_wfqs_v2(entry.walk_forward_summary) + fresh_deployment_score, fresh_deployment_breakdown = compute_deployment_score( + train_result, + valid_result, + test_result, + entry.walk_forward_summary, + rqs_score=fresh_rqs_score, + wfqs_score=fresh_wfqs_score, + ) public_sqs, public_breakdown, public_source = compute_public_sqs( - e.results.get("test"), - e.results.get("valid"), + train_result, + valid_result, + test_result, + walk_forward_summary=entry.walk_forward_summary, + robustness_matrix_summary=entry.robustness_matrix_summary, + out_of_time_robustness_summary=entry.out_of_time_robustness_summary, + rqs_score=fresh_rqs_score, + wfqs_score=fresh_wfqs_v2_score, ) + stress_sqs, _, _ = compute_public_sqs_v2( + train_result, + valid_result, + test_result, + walk_forward_summary=entry.walk_forward_summary, + robustness_matrix_summary=entry.robustness_matrix_summary, + out_of_time_robustness_summary=entry.out_of_time_robustness_summary, + rqs_score=fresh_rqs_score, + wfqs_v2_score=fresh_wfqs_v2_score, + ) + if public_sqs is None: + public_breakdown = public_breakdown or {} + public_source = public_source or "pending_validation" + + print(f"\n{entry.entry_id} — {entry.experiment_name}") + print(f" Timestamp: {entry.timestamp}") + print(f" Hypothesis: {entry.hypothesis}") + print(f" Verdict: {entry.verdict}") + pending_reason = _validation_pending_reason(entry) if public_sqs is None: - public_sqs = e.sqs_score - public_breakdown = e.sqs_breakdown - public_source = "stored" - print(f"\n{e.entry_id} — {e.experiment_name}") - print(f" Timestamp: {e.timestamp}") - print(f" Hypothesis: {e.hypothesis}") - print(f" Verdict: {e.verdict}") - print(f" SQS: {public_sqs} {public_breakdown} [{public_source}]") - if promotion_score is not None: - print(f" Promotion: {promotion_score} {promotion_breakdown} [internal]") - if e.config_delta: - print(f" Baseline: {e.config_delta.base_experiment}") - for k, v in e.config_delta.changes.items(): - print(f" {k}: {v}") - for split_name, sr in e.results.items(): - pf = f"PF={sr.profit_factor:.2f}" if sr.profit_factor is not None else "PF=-" - ret = f"Ret={sr.total_return_pct:+.2f}%" if sr.total_return_pct is not None else "Ret=-" - wr = f"WR={sr.win_rate:.1%}" if sr.win_rate is not None else "WR=-" - print(f" {split_name}: {sr.trade_count} trades, {pf}, {ret}, {wr}") - if e.verdict_reasoning: - print(f" Reasoning: {e.verdict_reasoning}") - if e.next_direction: - print(f" Next: {e.next_direction}") + print(f" SQS: pending [{public_source}]") + if pending_reason: + print(f" Validation: {pending_reason}") + else: + print(f" SQS: {public_sqs} [{public_source}]") + if stress_sqs is not None: + print(f" Stress SQS: {stress_sqs}") + + if entry.config_delta: + print(f" Baseline: {entry.config_delta.base_experiment}") + for key, value in entry.config_delta.changes.items(): + print(f" {key}: {value}") + + hydrated_results = { + "train": train_result, + "valid": valid_result, + "test": test_result, + **{ + name: result + for name, result in entry.results.items() + if name not in {"train", "valid", "test"} + }, + } + for split_name, split_result in _ordered_splits(hydrated_results): + pf = f"PF={split_result.profit_factor:.2f}" if split_result.profit_factor is not None else "PF=-" + ret = f"Ret={split_result.total_return_pct:+.2f}%" if split_result.total_return_pct is not None else "Ret=-" + ann = ( + f"Ann={split_result.annualized_return_pct:+.2f}%" + if split_result.annualized_return_pct is not None + else "Ann=-" + ) + dd = f"DD={split_result.max_drawdown_pct:.2f}%" if split_result.max_drawdown_pct is not None else "DD=-" + gross = ( + f"Gross={split_result.avg_gross_exposure_pct:.2f}%" + if split_result.avg_gross_exposure_pct is not None + else "Gross=-" + ) + dim = ( + f"DIM={split_result.days_in_market_pct:.2f}%" + if split_result.days_in_market_pct is not None + else "DIM=-" + ) + ret_on_gross = _ratio(split_result.total_return_pct, split_result.avg_gross_exposure_pct) + rog = f"R/G={ret_on_gross:.2f}" if ret_on_gross is not None else "R/G=-" + print(f" {split_name}: {split_result.trade_count} trades, {pf}, {ret}, {ann}, {dd}, {gross}, {dim}, {rog}") + + if entry.walk_forward_summary is not None: + summary = entry.walk_forward_summary + test = summary.test_aggregate + gap = summary.gap_stats + print( + " WFV: " + f"{summary.fold_count} folds, " + f"mean={_fmt(test.mean_return_pct, '+.2f')}%, " + f"median={_fmt(test.median_return_pct, '+.2f')}%, " + f"worst={_fmt(test.worst_return_pct, '+.2f')}%, " + f"positive={_fmt(test.positive_fold_rate_pct, '.1f')}%" + ) + print( + " WFV Gap: " + f"mean train-test gap={_fmt(gap.mean_train_test_return_gap_pct, '.2f')}%, " + f"worst={_fmt(gap.worst_train_test_return_gap_pct, '.2f')}%" + ) + + if entry.robustness_matrix_summary is not None: + summary = entry.robustness_matrix_summary + h63 = next((item for item in summary.horizon_summaries if item.horizon_days == 63), None) + h252 = next((item for item in summary.horizon_summaries if item.horizon_days == 252), None) + print( + " Robust: " + f"windows={summary.overall_window_count}, " + f"positive={_fmt(summary.overall_positive_window_rate_pct, '.1f')}%, " + f"worst={_fmt(summary.overall_worst_return_pct, '+.2f')}%, " + f"63d med={_fmt(h63.median_return_pct if h63 else None, '+.2f')}%, " + f"252d med={_fmt(h252.median_return_pct if h252 else None, '+.2f')}%" + ) + for horizon in summary.horizon_summaries: + print( + " " + f"{horizon.horizon_days:>3}d: " + f"n={horizon.window_count}, " + f"mean={_fmt(horizon.mean_return_pct, '+.2f')}%, " + f"median={_fmt(horizon.median_return_pct, '+.2f')}%, " + f"worst={_fmt(horizon.worst_return_pct, '+.2f')}%, " + f"positive={_fmt(horizon.positive_window_rate_pct, '.1f')}%, " + f"dd={_fmt(horizon.mean_max_drawdown_pct, '.2f')}%" + ) + if entry.out_of_time_robustness_summary is not None: + summary = entry.out_of_time_robustness_summary + h63 = next((item for item in summary.horizon_summaries if item.horizon_days == 63), None) + h252 = next((item for item in summary.horizon_summaries if item.horizon_days == 252), None) + print( + " OOT Robust: " + f"windows={summary.overall_window_count}, " + f"positive={_fmt(summary.overall_positive_window_rate_pct, '.1f')}%, " + f"worst={_fmt(summary.overall_worst_return_pct, '+.2f')}%, " + f"63d med={_fmt(h63.median_return_pct if h63 else None, '+.2f')}%, " + f"252d med={_fmt(h252.median_return_pct if h252 else None, '+.2f')}%" + ) + + if args.diagnostics: + promotion_score = entry.promotion_score + promotion_breakdown = entry.promotion_breakdown + if promotion_score is None: + promotion_score, promotion_breakdown = compute_promotion_score( + test_result, + valid_result, + ) + unified_score = entry.unified_score + unified_breakdown = entry.unified_breakdown + if unified_score is None: + unified_score, unified_breakdown = compute_unified_score( + test_result, + valid_result, + ) + rqs_score = fresh_rqs_score + rqs_breakdown = fresh_rqs_breakdown + wfqs_score = fresh_wfqs_score + wfqs_breakdown = fresh_wfqs_breakdown + deployment_score = fresh_deployment_score + deployment_breakdown = fresh_deployment_breakdown + robustness_gate_factor, robustness_breakdown = compute_robustness_gate( + entry.robustness_matrix_summary, + ) + oot_gate_factor, oot_breakdown = compute_oot_robustness_gate( + entry.out_of_time_robustness_summary, + ) + print(" Diagnostics:") + print(f" Public SQS: {public_sqs} {public_breakdown}") + if rqs_score is not None: + print(f" RQS: {rqs_score} {rqs_breakdown}") + if wfqs_score is not None: + print(f" WFQS: {wfqs_score} {wfqs_breakdown}") + wfqs_v2_score = entry.wfqs_v2_score + wfqs_v2_breakdown = entry.wfqs_v2_breakdown + if wfqs_v2_score is None: + wfqs_v2_score, wfqs_v2_breakdown = compute_wfqs_v2(entry.walk_forward_summary) + if wfqs_v2_score is not None: + print(f" WFQS v2: {wfqs_v2_score} {wfqs_v2_breakdown}") + if deployment_score is not None: + print(f" Deploy: {deployment_score} {deployment_breakdown}") + if promotion_score is not None: + print(f" Promotion: {promotion_score} {promotion_breakdown}") + if unified_score is not None: + print(f" Unified: {unified_score} {unified_breakdown}") + if entry.robustness_matrix_summary is not None: + print(f" RobustGate: {robustness_gate_factor:.2f} {robustness_breakdown}") + if entry.out_of_time_robustness_summary is not None: + print(f" OOTGate: {oot_gate_factor:.2f} {oot_breakdown}") + + if entry.verdict_reasoning: + print(f" Reasoning: {entry.verdict_reasoning}") + if entry.next_direction: + print(f" Next: {entry.next_direction}") def cmd_check_duplicate(args: argparse.Namespace) -> None: @@ -293,65 +782,196 @@ def cmd_check_duplicate(args: argparse.Namespace) -> None: dupes = check_duplicate(journal_path, args.experiment) if dupes: print(f"FOUND {len(dupes)} existing entries for '{args.experiment}':") - for e in dupes: - print(f" {e.entry_id} ({e.timestamp[:10]}) — SQS={e.sqs_score}, verdict={e.verdict}") + for entry in dupes: + print(f" {entry.entry_id} ({entry.timestamp[:10]}) — SQS={entry.sqs_score}, verdict={entry.verdict}") else: print(f"No existing entries for '{args.experiment}'.") +def cmd_attach_wfv(args: argparse.Namespace) -> None: + """Attach a walk-forward summary to an existing journal entry.""" + journal_dir = Path(args.journal_dir) + journal_path = journal_dir / "improvement_journal.jsonl" + registry_path = journal_dir / "experiment_registry.json" + leaderboard_path = journal_dir / "LEADERBOARD.md" + summary_path = Path(args.summary) + + if not journal_path.exists(): + print("No journal found. Run 'record' first.") + sys.exit(1) + if not summary_path.exists(): + print(f"ERROR: walk-forward summary not found: {summary_path}") + sys.exit(1) + + summary = WalkForwardSummary.model_validate_json(summary_path.read_text()) + with journal_lock(journal_path): + updated = attach_walk_forward_summary(journal_path, args.target, summary) + _sync_and_rebuild(journal_path, registry_path, leaderboard_path) + + print( + f"Attached WFV to {updated.entry_id}: {updated.experiment_name} " + f"(WFQS={_fmt(updated.wfqs_score, '.1f')}, DEP={_fmt(updated.deployment_score, '.1f')}, folds={summary.fold_count})" + ) + print(f"Leaderboard updated: {leaderboard_path}") + + +def cmd_attach_robustness(args: argparse.Namespace) -> None: + """Attach a robustness matrix summary to an existing journal entry.""" + journal_dir = Path(args.journal_dir) + journal_path = journal_dir / "improvement_journal.jsonl" + registry_path = journal_dir / "experiment_registry.json" + leaderboard_path = journal_dir / "LEADERBOARD.md" + summary_path = Path(args.summary) + + if not journal_path.exists(): + print("No journal found. Run 'record' first.") + sys.exit(1) + if not summary_path.exists(): + print(f"ERROR: robustness summary not found: {summary_path}") + sys.exit(1) + + summary = RobustnessMatrixSummary.model_validate_json(summary_path.read_text()) + with journal_lock(journal_path): + updated = attach_robustness_summary(journal_path, args.target, summary) + _sync_and_rebuild(journal_path, registry_path, leaderboard_path) + + gate_factor, _ = compute_robustness_gate(summary) + print( + f"Attached robustness matrix to {updated.entry_id}: {updated.experiment_name} " + f"(windows={summary.overall_window_count}, gate={gate_factor:.2f})" + ) + print(f"Leaderboard updated: {leaderboard_path}") + + +def cmd_attach_oot_robustness(args: argparse.Namespace) -> None: + """Attach out-of-time robustness matrix summary to an existing journal entry.""" + journal_dir = Path(args.journal_dir) + journal_path = journal_dir / "improvement_journal.jsonl" + registry_path = journal_dir / "experiment_registry.json" + leaderboard_path = journal_dir / "LEADERBOARD.md" + summary_path = Path(args.summary) + + if not journal_path.exists(): + print("No journal found. Run 'record' first.") + sys.exit(1) + if not summary_path.exists(): + print(f"ERROR: robustness summary not found: {summary_path}") + sys.exit(1) + + summary = RobustnessMatrixSummary.model_validate_json(summary_path.read_text()) + with journal_lock(journal_path): + updated = attach_out_of_time_robustness_summary(journal_path, args.target, summary) + _sync_and_rebuild(journal_path, registry_path, leaderboard_path) + + gate_factor, _ = compute_oot_robustness_gate(summary) + print( + f"Attached out-of-time robustness to {updated.entry_id}: {updated.experiment_name} " + f"(windows={summary.overall_window_count}, gate={gate_factor:.2f})" + ) + print(f"Leaderboard updated: {leaderboard_path}") + + +def cmd_rescore_public(args: argparse.Namespace) -> None: + """Recompute stored public scores for matching journal entries.""" + journal_dir = Path(args.journal_dir) + journal_path = journal_dir / "improvement_journal.jsonl" + registry_path = journal_dir / "experiment_registry.json" + leaderboard_path = journal_dir / "LEADERBOARD.md" + + prefix = args.prefix or "" + + def _matches(entry: JournalEntry) -> bool: + if prefix and not entry.experiment_name.startswith(prefix): + return False + if args.min_version is None and args.max_version is None: + return True + if not prefix: + return False + suffix = entry.experiment_name[len(prefix):] + try: + version = float(suffix) + except ValueError: + return False + if args.min_version is not None and version < args.min_version: + return False + if args.max_version is not None and version > args.max_version: + return False + return True + + with journal_lock(journal_path): + updated_count = refresh_public_scores(journal_path, selector=_matches) + _sync_and_rebuild(journal_path, registry_path, leaderboard_path) + + print(f"Rescored entries: {updated_count}") + print(f"Leaderboard updated: {leaderboard_path}") + + def _print_help() -> None: _console.print() _console.print(Panel( - "[bold cyan]fithia2[/] — ACE-F Strategy Improvement Tracker\n" - "[dim]백테스트 실험을 기록하고 전략 품질 점수(SQS)로 순위를 매깁니다.[/]", + "[bold cyan]fithia2[/] — ACE-F Strategy Research & Paper Trading\n" + "[dim]백테스트 실험 추적 + Alpaca 페이퍼 트레이딩[/]", border_style="cyan", padding=(0, 2), )) - t = Table(box=box.SIMPLE, show_header=True, header_style="bold yellow", padding=(0, 2)) - t.add_column("Command", style="bold green", no_wrap=True) - t.add_column("Description") - t.add_column("Key Options", style="dim") - - t.add_row( + table = Table(box=box.SIMPLE, show_header=True, header_style="bold yellow", padding=(0, 2)) + table.add_column("Command", style="bold green", no_wrap=True) + table.add_column("Description") + table.add_column("Key Options", style="dim") + table.add_row( "leaderboard [dim]lb[/]", - "SQS 순위표 출력 및 LEADERBOARD.md 재생성", - "-n N --sort sqs|promotion", + "기본 public SQS 순위표 출력 및 LEADERBOARD.md 재생성", + "-n N --sort sqs|rqs|wfqs|deployment|promotion --include-retired", ) - t.add_row( + table.add_row( "record [dim]rec[/]", "실험 결과를 저널에 기록", - "-e NAME -H TEXT -v better|worse|neutral -b BASELINE", + "-e NAME -H TEXT --walk-forward-summary PATH --robustness-summary PATH --out-of-time-robustness-summary PATH", ) - t.add_row( + table.add_row( "show [dim]s[/]", "특정 저널 항목 상세 조회", - "ENTRY_ID (예: IMP-0001 또는 실험명)", + "ENTRY_ID --diagnostics", ) - t.add_row( - "check-duplicate [dim]dup[/]", - "동일 실험명이 이미 기록됐는지 확인", - "-e NAME", + table.add_row( + "paper", + "Alpaca 페이퍼 트레이딩 [dim](fithia2 paper 로 상세 확인)[/]", + "start run positions status trades ...", ) - - _console.print(t) + table.add_row( + "pipeline", + "데이터 파이프라인 실행 [dim](fithia2 pipeline 로 상세 확인)[/]", + "run [dim]--step poller|fetcher|parser|features|labels[/]", + ) + _console.print(table) _console.print( " [dim]공통 옵션:[/] [bold]--journal-dir[/] [dim](기본: journal/)[/]" " [bold]--runs-dir[/] [dim](기본: runs/)[/]\n" ) - _console.print(" [bold]예시[/]") _console.print(" [green]fithia2 leaderboard[/]") - _console.print(" [green]fithia2 leaderboard --top 20[/]") - _console.print(" [green]fithia2 leaderboard --sort sqs[/]") - _console.print(" [green]fithia2 leaderboard --sort promotion[/]") _console.print(" [green]fithia2 record --experiment pead_v2 --hypothesis '...' --verdict better[/]") - _console.print(" [green]fithia2 show IMP-0007[/]") + _console.print(" [green]fithia2 paper[/]") _console.print() def main() -> None: + # 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:] + from apps.paper_trader.cli import main as paper_main + paper_main() + 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:] + from apps.pipeline.cli import main as pipeline_main + pipeline_main() + return + if len(sys.argv) == 1: _print_help() sys.exit(0) @@ -359,47 +979,115 @@ def main() -> None: parser = argparse.ArgumentParser(description="Strategy Improvement Tracker", add_help=True) sub = parser.add_subparsers(dest="command", required=True) - _jdir_kwargs = {"default": _DEFAULT_JOURNAL_DIR, "help": f"Path to journal/ directory (default: {_DEFAULT_JOURNAL_DIR})"} + journal_kwargs = { + "default": _DEFAULT_JOURNAL_DIR, + "help": f"Path to journal/ directory (default: {_DEFAULT_JOURNAL_DIR})", + } - # record (alias: rec) for name in ("record", "rec"): - p = sub.add_parser(name, help="Record an experiment to the journal") - p.add_argument("--journal-dir", **_jdir_kwargs) - p.add_argument("--runs-dir", default="runs", help="Path to runs output directory (default: runs)") - p.add_argument("--experiment", "-e", required=True, help="Experiment name (matches manifest)") - p.add_argument("--hypothesis", "-H", help="What you expected this change to do") - p.add_argument("--baseline", "-b", help="Baseline experiment name for comparison") - p.add_argument("--verdict", "-v", choices=["better", "worse", "neutral", "unknown"], default="unknown") - p.add_argument("--reasoning", "-r", help="Why this verdict") - p.add_argument("--next", "-n", help="Next experiment direction") - p.add_argument("--force", "-f", action="store_true", help="Allow duplicate experiment names") - - # leaderboard (alias: lb) + subparser = sub.add_parser(name, help="Record an experiment to the journal") + subparser.add_argument("--journal-dir", **journal_kwargs) + subparser.add_argument("--runs-dir", default="runs", help="Path to runs output directory (default: runs)") + subparser.add_argument("--experiment", "-e", required=True, help="Experiment name (matches manifest)") + subparser.add_argument("--hypothesis", "-H", help="What you expected this change to do") + subparser.add_argument("--baseline", "-b", help="Baseline experiment name for comparison") + subparser.add_argument("--verdict", "-v", choices=["better", "worse", "neutral", "unknown"], default="unknown") + subparser.add_argument("--reasoning", "-r", help="Why this verdict") + subparser.add_argument("--next", "-n", help="Next experiment direction") + subparser.add_argument("--walk-forward-summary", help="Optional path to walk_forward_summary.json") + subparser.add_argument("--robustness-summary", help="Optional path to robustness_matrix_summary.json") + subparser.add_argument( + "--out-of-time-robustness-summary", + help="Optional path to out-of-time robustness_matrix_summary.json", + ) + subparser.add_argument("--force", "-f", action="store_true", help="Allow duplicate experiment names") + for name in ("leaderboard", "lb"): - p = sub.add_parser(name, help="Show/regenerate the leaderboard") - p.add_argument("--journal-dir", **_jdir_kwargs) - p.add_argument("--top", "-n", type=int, default=10, help="Show top N entries (default: 10)") - p.add_argument("--sort", choices=["sqs", "promotion", "unified"], default="sqs", help="Sort leaderboard by public SQS (default) or internal promotion score") + subparser = sub.add_parser(name, help="Show/regenerate the leaderboard") + subparser.add_argument("--journal-dir", **journal_kwargs) + subparser.add_argument("--top", "-n", type=int, default=10, help="Show top N entries (default: 10)") + subparser.add_argument( + "--sort", + choices=["sqs", "rqs", "wfqs", "deployment", "dep", "promotion", "unified", "sqs2"], + default="sqs", + help="Sort by public SQS (default) or internal diagnostics", + ) + subparser.add_argument( + "--include-retired", + action="store_true", + help="Include retired legacy PEAD / short-core / exact-pocket families", + ) - # show (alias: s) for name in ("show", "s"): - p = sub.add_parser(name, help="Show details of a journal entry") - p.add_argument("--journal-dir", **_jdir_kwargs) - p.add_argument("entry_id", help="Entry ID (IMP-0001) or experiment name") + subparser = sub.add_parser(name, help="Show details of a journal entry") + subparser.add_argument("--journal-dir", **journal_kwargs) + subparser.add_argument("entry_id", help="Entry ID (IMP-0001) or experiment name") + subparser.add_argument("--diagnostics", action="store_true", help="Show internal RQS/WFQS/DEP diagnostics") - # check-duplicate (alias: dup) for name in ("check-duplicate", "dup"): - p = sub.add_parser(name, help="Check if experiment already recorded") - p.add_argument("--journal-dir", **_jdir_kwargs) - p.add_argument("--experiment", "-e", required=True, help="Experiment name to check") + subparser = sub.add_parser(name, help="Check if experiment already recorded") + subparser.add_argument("--journal-dir", **journal_kwargs) + subparser.add_argument("--experiment", "-e", required=True, help="Experiment name to check") - args = parser.parse_args() + for name in ("attach-wfv", "awf"): + subparser = sub.add_parser(name, help="Attach walk-forward summary to an existing journal entry") + subparser.add_argument("--journal-dir", **journal_kwargs) + subparser.add_argument("target", help="Entry ID or experiment name to update") + subparser.add_argument("--summary", required=True, help="Path to walk_forward_summary.json") + + for name in ("attach-robustness", "arb"): + subparser = sub.add_parser(name, help="Attach robustness matrix summary to an existing journal entry") + subparser.add_argument("--journal-dir", **journal_kwargs) + subparser.add_argument("target", help="Entry ID or experiment name to update") + subparser.add_argument("--summary", required=True, help="Path to robustness_matrix_summary.json") + + for name in ("attach-oot-robustness", "aoot"): + subparser = sub.add_parser(name, help="Attach out-of-time robustness summary to an existing journal entry") + subparser.add_argument("--journal-dir", **journal_kwargs) + subparser.add_argument("target", help="Entry ID or experiment name to update") + subparser.add_argument("--summary", required=True, help="Path to robustness_matrix_summary.json") + for name in ("record-overlay", "rovl"): + subparser = sub.add_parser(name, help="Record an official overlay evaluation") + subparser.add_argument("--journal-dir", **journal_kwargs) + subparser.add_argument("--spec", required=True, help="Path to overlay spec JSON for the main window") + subparser.add_argument("--summary", required=True, help="Path to overlay_summary.json for the main window") + subparser.add_argument("--stress-spec", required=True, help="Path to overlay spec JSON for the stress window") + subparser.add_argument("--stress-summary", required=True, help="Path to overlay_summary.json for the stress window") + subparser.add_argument("--hypothesis", default="", help="Short hypothesis / notes") + subparser.add_argument("--verdict", help="Initial verdict label") + subparser.add_argument("--reasoning", help="Verdict reasoning") + subparser.add_argument("--next", help="Next direction") + subparser.add_argument("--force", action="store_true", help="Allow duplicate overlay names") + subparser.add_argument("--tags", nargs="*", default=[], help="Extra tags") + + for name in ("rescore-public", "rsp"): + subparser = sub.add_parser(name, help="Recompute stored public scores for journal entries") + subparser.add_argument("--journal-dir", **journal_kwargs) + subparser.add_argument("--prefix", help="Only rescore experiment names with this prefix") + subparser.add_argument("--min-version", type=float, help="Minimum numeric suffix after --prefix") + subparser.add_argument("--max-version", type=float, help="Maximum numeric suffix after --prefix") + + args = parser.parse_args() dispatch = { - "record": cmd_record, "rec": cmd_record, - "leaderboard": cmd_leaderboard, "lb": cmd_leaderboard, - "show": cmd_show, "s": cmd_show, - "check-duplicate": cmd_check_duplicate, "dup": cmd_check_duplicate, + "record": cmd_record, + "rec": cmd_record, + "leaderboard": cmd_leaderboard, + "lb": cmd_leaderboard, + "show": cmd_show, + "s": cmd_show, + "check-duplicate": cmd_check_duplicate, + "dup": cmd_check_duplicate, + "attach-wfv": cmd_attach_wfv, + "awf": cmd_attach_wfv, + "attach-robustness": cmd_attach_robustness, + "arb": cmd_attach_robustness, + "attach-oot-robustness": cmd_attach_oot_robustness, + "aoot": cmd_attach_oot_robustness, + "record-overlay": cmd_record_overlay, + "rovl": cmd_record_overlay, + "rescore-public": cmd_rescore_public, + "rsp": cmd_rescore_public, } dispatch[args.command](args) diff --git a/libs/backtest/tracker.py b/libs/backtest/tracker.py index e06ee23..380738e 100644 --- a/libs/backtest/tracker.py +++ b/libs/backtest/tracker.py @@ -2,23 +2,33 @@ from __future__ import annotations import contextlib +import datetime as dt import functools import fcntl import json from pathlib import Path -from typing import Any, Iterator +from typing import Any, Callable, Iterator from libs.backtest.domain import ( + CommonWindowSummary, ConfigDelta, + DeploymentScoreWeights, ExperimentRegistry, JournalEntry, MetricsBundle, + RobustnessMatrixSummary, + WalkForwardScoreWeights, + WFQSv2Weights, + ReturnScoreWeights, PromotionScoreWeights, RegistryEntry, + OverlayWindowSummary, SplitResult, SQSWeights, SQSv2Weights, UnifiedScoreWeights, + WalkForwardAggregate, + WalkForwardSummary, ) from libs.common.logging import get_logger from libs.common.time_utils import utc_now @@ -29,6 +39,29 @@ _DEFAULT_WEIGHTS = SQSWeights() _DEFAULT_V2_WEIGHTS = SQSv2Weights() _DEFAULT_PROMOTION_WEIGHTS = PromotionScoreWeights() _DEFAULT_UNIFIED_WEIGHTS = UnifiedScoreWeights() +_DEFAULT_RQS_WEIGHTS = ReturnScoreWeights() +_DEFAULT_WFQS_WEIGHTS = WalkForwardScoreWeights() +_DEFAULT_WFQS_V2_WEIGHTS = WFQSv2Weights() +_DEFAULT_DEPLOYMENT_WEIGHTS = DeploymentScoreWeights() +_DEFAULT_PUBLIC_COMMON_WINDOW_WEIGHT = 0.20 +_RETIRED_STRATEGY_FAMILIES = { + "short_core", + "legacy_pead", + "leveraged_return_max_long", + "exact_pocket_return_max_long", + "named_micro_return_max_long", +} +_EXPERIMENTS_DIR = Path("configs/experiments") +_NAMED_MICRO_ENGINE_TOKENS = ( + "epam_micro", + "nrg_np12", + "glw_np11", + "pl_micro", + "exas_micro", + "aap_micro", + "apld_micro", + "czr_hot_micro", +) # --------------------------------------------------------------------------- @@ -84,6 +117,12 @@ def _calibrate_sqs(value: float | None) -> float | None: return round(max(0.0, value * 0.70 - 7.5), 1) +def _safe_ratio(numerator: float | None, denominator: float | None) -> float | None: + if numerator is None or denominator is None or denominator <= 0: + return None + return numerator / denominator + + def _apply_single_split_penalty(value: float | None) -> float | None: """Discount scores that have no valid/test confirmation pair.""" if value is None: @@ -216,6 +255,7 @@ def build_split_result(split_name: str, run_id: str, metrics: MetricsBundle) -> trade_count=metrics.trade_count, profit_factor=metrics.profit_factor, total_return_pct=metrics.total_return_pct, + annualized_return_pct=metrics.annualized_return_pct, win_rate=metrics.win_rate, max_drawdown_pct=metrics.max_drawdown_pct, sharpe_ratio=metrics.sharpe_ratio, @@ -234,13 +274,39 @@ def _metrics_from_split_result(result: SplitResult | None) -> MetricsBundle | No return MetricsBundle.model_validate(payload) -@functools.lru_cache(maxsize=512) +_metrics_path_index: dict[str, Path] | None = None + + +def _get_metrics_path_index() -> dict[str, Path]: + """Build run_id -> metrics_summary.json path index (single rglob).""" + global _metrics_path_index + if _metrics_path_index is not None: + return _metrics_path_index + _metrics_path_index = {} + runs_root = Path("runs") + for metrics_path in runs_root.rglob("metrics/metrics_summary.json"): + # run_dir is metrics_path.parent.parent (e.g. runs/.../run_id/) + run_dir = metrics_path.parent.parent + run_dir_name = run_dir.name + # Index by directory name as run_id + if run_dir_name not in _metrics_path_index or metrics_path.stat().st_mtime > _metrics_path_index[run_dir_name].stat().st_mtime: + _metrics_path_index[run_dir_name] = metrics_path + return _metrics_path_index + + +@functools.lru_cache(maxsize=2048) def _load_run_metrics_summary(run_id: str) -> dict[str, Any] | None: - metrics_path = Path("runs") / run_id / "metrics" / "metrics_summary.json" + runs_root = Path("runs") + metrics_path = runs_root / run_id / "metrics" / "metrics_summary.json" if not metrics_path.exists(): - return None + idx = _get_metrics_path_index() + found = idx.get(run_id) + if not found: + return None + metrics_path = found summary = json.loads(metrics_path.read_text()) needed_fields = ( + "annualized_return_pct", "avg_gross_exposure_pct", "avg_net_exposure_pct", "days_in_market_pct", @@ -248,7 +314,7 @@ def _load_run_metrics_summary(run_id: str) -> dict[str, Any] | None: if all(summary.get(field) is not None for field in needed_fields): return summary - equity_curve_path = Path("runs") / run_id / "artifacts" / "daily_equity_curve.parquet" + equity_curve_path = metrics_path.parent.parent / "artifacts" / "daily_equity_curve.parquet" if not equity_curve_path.exists(): return summary @@ -288,6 +354,7 @@ def _hydrate_split_result(result: SplitResult | None) -> SplitResult | None: if result is None: return None needed_fields = ( + "annualized_return_pct", "avg_gross_exposure_pct", "avg_net_exposure_pct", "days_in_market_pct", @@ -443,24 +510,1136 @@ def compute_unified_score( def compute_public_sqs( + train_result: SplitResult | None, + valid_result: SplitResult | None, test_result: SplitResult | None, + walk_forward_summary: WalkForwardSummary | None = None, + robustness_matrix_summary: RobustnessMatrixSummary | None = None, + out_of_time_robustness_summary: RobustnessMatrixSummary | None = None, + common_window_summary: CommonWindowSummary | None = None, + rqs_score: float | None = None, + wfqs_score: float | None = None, + deployment_score: float | None = None, +) -> tuple[float | None, dict[str, float], str | None]: + """Return the current public-facing SQS. + + v4 keeps the v3 stress/robustness gates intact, and adds a modest + full-cycle capital-growth term when a comparable common-window run + is available. Missing common-window summaries fall back to v3. + """ + return compute_public_sqs_v4( + train_result, + valid_result, + test_result, + walk_forward_summary=walk_forward_summary, + robustness_matrix_summary=robustness_matrix_summary, + out_of_time_robustness_summary=out_of_time_robustness_summary, + common_window_summary=common_window_summary, + rqs_score=rqs_score, + wfqs_v2_score=wfqs_score, + ) + + +def _compute_return_split_score( + metrics: MetricsBundle, + split_name: str, + weights: ReturnScoreWeights | None = None, +) -> tuple[float, dict[str, float]]: + """Compute a return-max score for a single split.""" + w = weights or _DEFAULT_RQS_WEIGHTS + + return_bands = { + "train": (0.0, 50.0), + "valid": (0.0, 30.0), + "test": (0.0, 35.0), + } + annualized_return_bands = { + "train": (0.0, 25.0), + "valid": (0.0, 120.0), + "test": (0.0, 120.0), + } + dd_bands = { + "train": (20.0, 2.0), + "valid": (12.0, 2.0), + "test": (12.0, 2.0), + } + ret_low, ret_high = return_bands.get(split_name, (0.0, 35.0)) + ann_low, ann_high = annualized_return_bands.get(split_name, (0.0, 100.0)) + dd_low, dd_high = dd_bands.get(split_name, (12.0, 2.0)) + + total_return_score = _normalize(metrics.total_return_pct, low=ret_low, high=ret_high) + annualized_return_score = _normalize(metrics.annualized_return_pct, low=ann_low, high=ann_high) + effective_profit_factor = metrics.profit_factor + if ( + effective_profit_factor is None + and metrics.trade_count > 0 + and metrics.win_rate is not None + and metrics.win_rate >= 0.999 + ): + # Backtest summaries emit ``None`` when there are no losing trades. + # For return-max ranking that should be treated as capped-best, not zero. + effective_profit_factor = 3.0 + pf_score = _normalize(effective_profit_factor, low=1.0, high=3.0) + sharpe_score = _normalize(metrics.sharpe_ratio, low=0.0, high=3.5) + drawdown_score = _normalize_inverse(metrics.max_drawdown_pct, low=dd_low, high=dd_high) + gross_score = _normalize_band( + metrics.avg_gross_exposure_pct, + low_bad=2.0, + low_good=8.0, + high_good=40.0, + high_bad=75.0, + ) + dim_score = _normalize_band( + metrics.days_in_market_pct, + low_bad=5.0, + low_good=15.0, + high_good=60.0, + high_bad=90.0, + ) + return_on_gross_exposure = None + if metrics.total_return_pct is not None and metrics.avg_gross_exposure_pct and metrics.avg_gross_exposure_pct > 0: + return_on_gross_exposure = metrics.total_return_pct / metrics.avg_gross_exposure_pct + return_on_gross_score = _normalize(return_on_gross_exposure, low=0.20, high=2.50) + + split_score = ( + total_return_score * w.split_total_return + + annualized_return_score * w.split_annualized_return + + pf_score * w.split_profitability + + sharpe_score * w.split_sharpe + + drawdown_score * w.split_drawdown + + return_on_gross_score * w.split_return_on_gross + + gross_score * w.split_gross_exposure + + dim_score * w.split_days_in_market + ) + if metrics.trade_count < w.low_trade_penalty_threshold: + split_score *= w.low_trade_penalty_factor + + split_score = round(split_score, 1) + breakdown = { + "total_return": round(total_return_score, 1), + "annualized_return": round(annualized_return_score, 1), + "profitability": round(pf_score, 1), + "sharpe": round(sharpe_score, 1), + "drawdown": round(drawdown_score, 1), + "return_on_gross": round(return_on_gross_score, 1), + "gross_exposure": round(gross_score, 1), + "days_in_market": round(dim_score, 1), + } + return split_score, breakdown + + +def compute_rqs( + train_result: SplitResult | None, valid_result: SplitResult | None, + test_result: SplitResult | None, + weights: ReturnScoreWeights | None = None, +) -> tuple[float | None, dict[str, float]]: + """Compute Return Quality Score for return-max strategy ranking.""" + w = weights or _DEFAULT_RQS_WEIGHTS + + hydrated = { + "train": _hydrate_split_result(train_result), + "valid": _hydrate_split_result(valid_result), + "test": _hydrate_split_result(test_result), + } + split_scores: dict[str, float] = {} + for split_name, result in hydrated.items(): + metrics = _metrics_from_split_result(result) + if metrics is None: + continue + split_score, _ = _compute_return_split_score(metrics, split_name, w) + split_scores[split_name] = split_score + + if "valid" not in split_scores or "test" not in split_scores: + return None, {} + + weighted_sum = 0.0 + weighted_den = 0.0 + if "train" in split_scores: + weighted_sum += split_scores["train"] * w.train_quality + weighted_den += w.train_quality + weighted_sum += split_scores["valid"] * w.valid_quality + weighted_den += w.valid_quality + weighted_sum += split_scores["test"] * w.test_quality + weighted_den += w.test_quality + split_quality = weighted_sum / weighted_den if weighted_den > 0 else 0.0 + + floor_quality = min(split_scores.values()) + gap_quality = _normalize_inverse( + max(split_scores.values()) - min(split_scores.values()), + low=60.0, + high=10.0, + ) + rqs = split_quality * (1.0 - w.floor_quality - w.gap_quality) + floor_quality * w.floor_quality + gap_quality * w.gap_quality + if "train" not in split_scores: + rqs *= w.missing_train_penalty + + breakdown = { + "train_quality": round(split_scores.get("train", 0.0), 1), + "valid_quality": round(split_scores["valid"], 1), + "test_quality": round(split_scores["test"], 1), + "floor_quality": round(floor_quality, 1), + "gap_quality": round(gap_quality, 1), + } + return round(rqs, 1), breakdown + + +def compute_wfqs( + walk_forward_summary: WalkForwardSummary | None, + weights: WalkForwardScoreWeights | None = None, +) -> tuple[float | None, dict[str, float]]: + """Compute a walk-forward robustness score from fold-level aggregates.""" + if walk_forward_summary is None or walk_forward_summary.fold_count <= 0: + return None, {} + + w = weights or _DEFAULT_WFQS_WEIGHTS + test = walk_forward_summary.test_aggregate + gap = walk_forward_summary.gap_stats + + median_return_score = _normalize(test.median_return_pct, low=0.0, high=25.0) + mean_return_score = _normalize(test.mean_return_pct, low=0.0, high=25.0) + worst_return_score = _normalize(test.worst_return_pct, low=-5.0, high=10.0) + positive_fold_rate_score = _normalize(test.positive_fold_rate_pct, low=50.0, high=100.0) + profit_factor_score = _normalize(test.mean_profit_factor, low=1.0, high=4.0) + drawdown_score = _normalize_inverse(test.mean_max_drawdown_pct, low=15.0, high=2.0) + gap_score = _normalize_inverse(gap.mean_train_test_return_gap_pct, low=70.0, high=15.0) + fold_count_score = _normalize(float(walk_forward_summary.fold_count), low=4.0, high=10.0) + + wfqs = ( + median_return_score * w.median_return + + mean_return_score * w.mean_return + + worst_return_score * w.worst_return + + positive_fold_rate_score * w.positive_fold_rate + + profit_factor_score * w.profit_factor + + drawdown_score * w.drawdown + + gap_score * w.train_test_gap + + fold_count_score * w.fold_count + ) + if walk_forward_summary.fold_count < w.low_fold_penalty_threshold: + wfqs *= w.low_fold_penalty_factor + + breakdown = { + "median_return": round(median_return_score, 1), + "mean_return": round(mean_return_score, 1), + "worst_return": round(worst_return_score, 1), + "positive_fold_rate": round(positive_fold_rate_score, 1), + "profit_factor": round(profit_factor_score, 1), + "drawdown": round(drawdown_score, 1), + "train_test_gap": round(gap_score, 1), + "fold_count": round(fold_count_score, 1), + } + return round(wfqs, 1), breakdown + + +# --------------------------------------------------------------------------- +# WFQS v2: multiplicative penalty model +# --------------------------------------------------------------------------- + + +def _gap_penalty(gap_pct: float | None) -> float: + """Multiplicative penalty based on mean train-test gap percentage.""" + if gap_pct is None: + return 1.0 + if gap_pct <= 30.0: + return 1.0 + if gap_pct <= 100.0: + return 1.0 - (gap_pct - 30.0) / (100.0 - 30.0) * 0.4 # 1.0 → 0.6 + if gap_pct <= 300.0: + return 0.6 - (gap_pct - 100.0) / (300.0 - 100.0) * 0.3 # 0.6 → 0.3 + return 0.2 + + +def _fold_variance_penalty(fold_return_cv: float | None) -> float: + """Multiplicative penalty based on CV of fold test returns.""" + if fold_return_cv is None: + return 1.0 + if fold_return_cv <= 0.5: + return 1.0 + if fold_return_cv <= 1.5: + return 1.0 - (fold_return_cv - 0.5) / (1.5 - 0.5) * 0.3 # 1.0 → 0.7 + return 0.6 + + +def _trade_credibility(mean_trades: float | None, mean_win_rate: float | None) -> float: + """Multiplicative penalty for low trade counts or suspiciously high win rates.""" + if mean_trades is None: + return 0.5 + if mean_trades < 5: + factor = 0.5 + elif mean_trades < 10: + factor = 0.7 + elif mean_trades < 20: + factor = 0.85 + else: + factor = 1.0 + if (mean_win_rate or 0.0) > 0.95 and mean_trades < 15: + factor *= 0.8 + return factor + + +def _engine_reliability_penalty(ratio: float | None) -> float: + """Multiplicative penalty based on engine reliability ratio.""" + if ratio is None: + return 1.0 + if ratio >= 0.7: + return 1.0 + if ratio >= 0.3: + return 0.5 + (ratio - 0.3) / (0.7 - 0.3) * 0.5 # 0.5 → 1.0 + return 0.4 + + +def _derive_fold_stats( + walk_forward_summary: WalkForwardSummary, +) -> tuple[float | None, float | None, float | None]: + """Derive mean_trade_count, mean_win_rate, fold_return_cv from folds when aggregate fields are missing.""" + import statistics as _stats + + folds = walk_forward_summary.folds + if not folds: + return None, None, None + + trade_counts = [float(f.test_metrics.trade_count) for f in folds] + win_rates = [f.test_metrics.win_rate for f in folds if f.test_metrics.win_rate is not None] + returns = [f.test_metrics.total_return_pct for f in folds if f.test_metrics.total_return_pct is not None] + + mean_tc = round(_stats.mean(trade_counts), 1) if trade_counts else None + mean_wr = round(_stats.mean(win_rates), 4) if win_rates else None + cv = None + if len(returns) >= 2: + mean_ret = _stats.mean(returns) + if abs(mean_ret) > 1e-9: + cv = round(_stats.stdev(returns) / abs(mean_ret), 3) + return mean_tc, mean_wr, cv + + +def _wfqs_v2_quality_from_aggregate( + aggregate: WalkForwardAggregate, + fold_count: int, + weights: WFQSv2Weights, +) -> float: + """Compute additive WFQS v2 quality score before multiplicative penalties.""" + median_return_score = _normalize(aggregate.median_return_pct, low=0.0, high=25.0) + mean_return_score = _normalize(aggregate.mean_return_pct, low=0.0, high=25.0) + worst_return_score = _normalize(aggregate.worst_return_pct, low=-5.0, high=10.0) + positive_fold_rate_score = _normalize(aggregate.positive_fold_rate_pct, low=50.0, high=100.0) + profit_factor_score = _normalize(aggregate.mean_profit_factor, low=1.0, high=4.0) + drawdown_score = _normalize_inverse(aggregate.mean_max_drawdown_pct, low=15.0, high=2.0) + fold_count_score = _normalize(float(fold_count), low=4.0, high=10.0) + quality = ( + median_return_score * weights.median_return + + mean_return_score * weights.mean_return + + worst_return_score * weights.worst_return + + positive_fold_rate_score * weights.positive_fold_rate + + profit_factor_score * weights.profit_factor + + drawdown_score * weights.drawdown + + fold_count_score * weights.fold_count + ) + if fold_count < weights.low_fold_penalty_threshold: + quality *= weights.low_fold_penalty_factor + return quality + + +def _build_recent_fold_aggregate( + walk_forward_summary: WalkForwardSummary, + lookback_days: int, + min_folds: int, +) -> tuple[WalkForwardAggregate | None, int]: + """Aggregate test-fold quality for folds ending within the recent lookback window.""" + import statistics as _stats + + folds = walk_forward_summary.folds + if not folds: + return None, 0 + + anchor = max(f.test_end for f in folds) + cutoff = anchor - dt.timedelta(days=max(lookback_days - 1, 0)) + recent_folds = [f for f in folds if f.test_end >= cutoff] + if len(recent_folds) < min_folds: + return None, len(recent_folds) + + returns = [f.test_metrics.total_return_pct for f in recent_folds if f.test_metrics.total_return_pct is not None] + pfs = [f.test_metrics.profit_factor for f in recent_folds if f.test_metrics.profit_factor is not None] + dds = [f.test_metrics.max_drawdown_pct for f in recent_folds if f.test_metrics.max_drawdown_pct is not None] + trade_counts = [float(f.test_metrics.trade_count) for f in recent_folds] + win_rates = [f.test_metrics.win_rate for f in recent_folds if f.test_metrics.win_rate is not None] + positive = [r for r in returns if r > 0] + + aggregate = WalkForwardAggregate( + mean_return_pct=round(_stats.mean(returns), 2) if returns else None, + median_return_pct=round(_stats.median(returns), 2) if returns else None, + worst_return_pct=round(min(returns), 2) if returns else None, + positive_fold_rate_pct=round(len(positive) / len(recent_folds) * 100.0, 1) if recent_folds else None, + mean_profit_factor=round(_stats.mean(pfs), 2) if pfs else None, + mean_max_drawdown_pct=round(_stats.mean(dds), 2) if dds else None, + mean_trade_count=round(_stats.mean(trade_counts), 1) if trade_counts else None, + mean_win_rate=round(_stats.mean(win_rates), 4) if win_rates else None, + ) + return aggregate, len(recent_folds) + + +def _public_activity_factor( + test_result: SplitResult | None, + walk_forward_summary: WalkForwardSummary | None, +) -> tuple[float, dict[str, float]]: + """Return a light multiplicative penalty for strategies that barely trade.""" + if test_result is None or walk_forward_summary is None: + return 1.0, {} + + mean_tc = walk_forward_summary.test_aggregate.mean_trade_count + if mean_tc is None: + derived_tc, _, _ = _derive_fold_stats(walk_forward_summary) + mean_tc = derived_tc + + test_trade_score = _normalize(float(test_result.trade_count), low=8.0, high=24.0) + wf_trade_score = _normalize(mean_tc, low=3.0, high=10.0) + dim_score = _normalize(test_result.days_in_market_pct, low=20.0, high=60.0) + + activity_quality = ( + test_trade_score * 0.45 + + wf_trade_score * 0.35 + + dim_score * 0.20 + ) + + if test_result.trade_count < 10: + activity_quality *= 0.85 + if (mean_tc or 0.0) < 4.0: + activity_quality *= 0.85 + + factor = 0.50 + 0.50 * (activity_quality / 100.0) + breakdown = { + "activity_factor": round(factor, 2), + "activity_quality": round(activity_quality, 1), + "activity_test_trades": round(test_trade_score, 1), + "activity_wf_mean_trades": round(wf_trade_score, 1), + "activity_days_in_market": round(dim_score, 1), + } + return factor, breakdown + + +def compute_wfqs_v2( + walk_forward_summary: WalkForwardSummary | None, + weights: WFQSv2Weights | None = None, +) -> tuple[float | None, dict[str, float]]: + """Compute WFQS v2 with multiplicative overfitting penalties.""" + if walk_forward_summary is None or walk_forward_summary.fold_count <= 0: + return None, {} + + w = weights or _DEFAULT_WFQS_V2_WEIGHTS + test = walk_forward_summary.test_aggregate + gap = walk_forward_summary.gap_stats + + # Derive missing stats from folds when aggregate fields are absent + mean_tc = test.mean_trade_count + mean_wr = test.mean_win_rate + fold_cv = gap.fold_return_cv + if mean_tc is None or mean_wr is None or fold_cv is None: + derived_tc, derived_wr, derived_cv = _derive_fold_stats(walk_forward_summary) + if mean_tc is None: + mean_tc = derived_tc + if mean_wr is None: + mean_wr = derived_wr + if fold_cv is None: + fold_cv = derived_cv + + overall_quality = _wfqs_v2_quality_from_aggregate( + test, + walk_forward_summary.fold_count, + w, + ) + median_return_score = _normalize(test.median_return_pct, low=0.0, high=25.0) + mean_return_score = _normalize(test.mean_return_pct, low=0.0, high=25.0) + worst_return_score = _normalize(test.worst_return_pct, low=-5.0, high=10.0) + positive_fold_rate_score = _normalize(test.positive_fold_rate_pct, low=50.0, high=100.0) + profit_factor_score = _normalize(test.mean_profit_factor, low=1.0, high=4.0) + drawdown_score = _normalize_inverse(test.mean_max_drawdown_pct, low=15.0, high=2.0) + fold_count_score = _normalize(float(walk_forward_summary.fold_count), low=4.0, high=10.0) + recent_aggregate, recent_fold_count = _build_recent_fold_aggregate( + walk_forward_summary, + lookback_days=w.recent_lookback_days, + min_folds=w.recent_min_folds, + ) + recent_quality = None + if recent_aggregate is not None: + recent_quality = _wfqs_v2_quality_from_aggregate( + recent_aggregate, + recent_fold_count, + w, + ) + if recent_quality is not None: + base_score = overall_quality * (1.0 - w.recent_fold_quality) + recent_quality * w.recent_fold_quality + else: + base_score = overall_quality + + # Multiplicative penalties + gp = _gap_penalty(gap.mean_train_test_return_gap_pct) + fvp = _fold_variance_penalty(fold_cv) + tc = _trade_credibility(mean_tc, mean_wr) + er = _engine_reliability_penalty(walk_forward_summary.engine_reliability_ratio) + + wfqs_v2 = base_score * gp * fvp * tc * er + + breakdown = { + "base_score": round(base_score, 1), + "median_return": round(median_return_score, 1), + "mean_return": round(mean_return_score, 1), + "worst_return": round(worst_return_score, 1), + "positive_fold_rate": round(positive_fold_rate_score, 1), + "profit_factor": round(profit_factor_score, 1), + "drawdown": round(drawdown_score, 1), + "fold_count": round(fold_count_score, 1), + "overall_quality": round(overall_quality, 1), + "recent_quality": round(recent_quality, 1) if recent_quality is not None else 0.0, + "recent_fold_count": float(recent_fold_count), + "recent_fold_weight": round(w.recent_fold_quality if recent_quality is not None else 0.0, 2), + "gap_penalty": round(gp, 3), + "fold_variance_penalty": round(fvp, 3), + "trade_credibility": round(tc, 3), + "engine_reliability": round(er, 3), + } + return round(wfqs_v2, 1), breakdown + + +def _compute_public_sqs_components( + train_result: SplitResult | None, + valid_result: SplitResult | None, + test_result: SplitResult | None, + walk_forward_summary: WalkForwardSummary | None = None, + robustness_matrix_summary: RobustnessMatrixSummary | None = None, + out_of_time_robustness_summary: RobustnessMatrixSummary | None = None, + rqs_score: float | None = None, + wfqs_v2_score: float | None = None, +) -> tuple[dict[str, Any] | None, dict[str, float], str | None]: + """Resolve shared components for public SQS variants.""" + resolved_rqs = rqs_score + if resolved_rqs is None: + resolved_rqs, _ = compute_rqs(train_result, valid_result, test_result) + if resolved_rqs is None: + return None, {}, None + + missing_requirements: dict[str, float] = {} + if walk_forward_summary is None: + missing_requirements["requires_walk_forward"] = 1.0 + if robustness_matrix_summary is None: + missing_requirements["requires_robustness"] = 1.0 + if out_of_time_robustness_summary is None: + missing_requirements["requires_out_of_time_robustness"] = 1.0 + if missing_requirements: + return None, missing_requirements, "pending_validation" + + resolved_wfqs_v2 = wfqs_v2_score + if resolved_wfqs_v2 is None: + resolved_wfqs_v2, _ = compute_wfqs_v2(walk_forward_summary) + + if resolved_wfqs_v2 is None: + return None, {}, None + + test = walk_forward_summary.test_aggregate + gap = walk_forward_summary.gap_stats + pass_positive = (test.positive_fold_rate_pct or 0.0) >= 70.0 + pass_median = (test.median_return_pct or 0.0) >= 5.0 + pass_worst = (test.worst_return_pct or 0.0) >= -5.0 + pass_gap = (gap.mean_train_test_return_gap_pct or 999.0) <= 35.0 + pass_count = sum([pass_positive, pass_median, pass_worst, pass_gap]) + deployment_gate_factor = {4: 1.00, 3: 0.85, 2: 0.65, 1: 0.40, 0: 0.20}[pass_count] + base_score = (resolved_rqs * 0.45 + resolved_wfqs_v2 * 0.55) * deployment_gate_factor + + gate_factor_rb, gate_breakdown = compute_robustness_gate(robustness_matrix_summary) + oot_gate_factor, oot_gate_breakdown = compute_oot_robustness_gate(out_of_time_robustness_summary) + oot_quality, oot_quality_breakdown = compute_oot_robustness_quality(out_of_time_robustness_summary) + activity_factor, activity_breakdown = _public_activity_factor(test_result, walk_forward_summary) + return ( + { + "base_score": base_score, + "deployment_gate_factor": deployment_gate_factor, + "rb_gate_factor": gate_factor_rb, + "rb_gate_breakdown": gate_breakdown, + "oot_gate_factor": oot_gate_factor, + "oot_gate_breakdown": oot_gate_breakdown, + "oot_quality": oot_quality, + "oot_quality_breakdown": oot_quality_breakdown, + "activity_factor": activity_factor, + "activity_breakdown": activity_breakdown, + }, + {}, + "v3_deployment", + ) + + +def compute_public_sqs_v2( + train_result: SplitResult | None, + valid_result: SplitResult | None, + test_result: SplitResult | None, + walk_forward_summary: WalkForwardSummary | None = None, + robustness_matrix_summary: RobustnessMatrixSummary | None = None, + out_of_time_robustness_summary: RobustnessMatrixSummary | None = None, + rqs_score: float | None = None, + wfqs_v2_score: float | None = None, +) -> tuple[float | None, dict[str, float], str | None]: + """Return legacy stress-adjusted public SQS. + + This keeps the historical multiplicative OOT quality factor that + compresses strong recent strategies when stress-period quality is weaker. + """ + components, breakdown, source = _compute_public_sqs_components( + train_result, + valid_result, + test_result, + walk_forward_summary=walk_forward_summary, + robustness_matrix_summary=robustness_matrix_summary, + out_of_time_robustness_summary=out_of_time_robustness_summary, + rqs_score=rqs_score, + wfqs_v2_score=wfqs_v2_score, + ) + if components is None: + return None, breakdown, source + + oot_quality_factor = 1.0 + if components["oot_quality"] is not None: + oot_quality_factor = 0.30 + 0.70 * (components["oot_quality"] / 100.0) + final_score = ( + components["base_score"] + * components["rb_gate_factor"] + * components["oot_gate_factor"] + * oot_quality_factor + * components["activity_factor"] + ) + resolved_breakdown = { + "base_score": round(components["base_score"], 1), + "gate_factor": round(components["rb_gate_factor"], 2), + **components["rb_gate_breakdown"], + "oot_gate_factor": round(components["oot_gate_factor"], 2), + "oot_quality_factor": round(oot_quality_factor, 2), + **{ + f"oot_{key}": value + for key, value in components["oot_gate_breakdown"].items() + }, + **{ + f"oot_{key}": value + for key, value in components["oot_quality_breakdown"].items() + }, + **components["activity_breakdown"], + } + if components["oot_quality"] is not None: + resolved_breakdown["oot_quality"] = round(components["oot_quality"], 1) + return round(final_score, 1), resolved_breakdown, "v2_deployment+robustness+oot" + + +def compute_public_sqs_v3( + train_result: SplitResult | None, + valid_result: SplitResult | None, + test_result: SplitResult | None, + walk_forward_summary: WalkForwardSummary | None = None, + robustness_matrix_summary: RobustnessMatrixSummary | None = None, + out_of_time_robustness_summary: RobustnessMatrixSummary | None = None, + rqs_score: float | None = None, + wfqs_v2_score: float | None = None, +) -> tuple[float | None, dict[str, float], str | None]: + """Return public SQS v3. + + v3 treats stress OOT as a hard gate for eligibility, but keeps OOT quality + as a diagnostic instead of a multiplicative ranking penalty. + """ + components, breakdown, source = _compute_public_sqs_components( + train_result, + valid_result, + test_result, + walk_forward_summary=walk_forward_summary, + robustness_matrix_summary=robustness_matrix_summary, + out_of_time_robustness_summary=out_of_time_robustness_summary, + rqs_score=rqs_score, + wfqs_v2_score=wfqs_v2_score, + ) + if components is None: + return None, breakdown, source + + final_score = ( + components["base_score"] + * components["rb_gate_factor"] + * components["oot_gate_factor"] + * components["activity_factor"] + ) + resolved_breakdown = { + "base_score": round(components["base_score"], 1), + "gate_factor": round(components["rb_gate_factor"], 2), + **components["rb_gate_breakdown"], + "oot_gate_factor": round(components["oot_gate_factor"], 2), + **{ + f"oot_{key}": value + for key, value in components["oot_gate_breakdown"].items() + }, + **{ + f"oot_{key}": value + for key, value in components["oot_quality_breakdown"].items() + }, + **components["activity_breakdown"], + } + if components["oot_quality"] is not None: + resolved_breakdown["oot_quality"] = round(components["oot_quality"], 1) + return round(final_score, 1), resolved_breakdown, "v3_deployment+robustness+oot_gate" + + +def compute_public_sqs_v4( + train_result: SplitResult | None, + valid_result: SplitResult | None, + test_result: SplitResult | None, + walk_forward_summary: WalkForwardSummary | None = None, + robustness_matrix_summary: RobustnessMatrixSummary | None = None, + out_of_time_robustness_summary: RobustnessMatrixSummary | None = None, + common_window_summary: CommonWindowSummary | None = None, + rqs_score: float | None = None, + wfqs_v2_score: float | None = None, +) -> tuple[float | None, dict[str, float], str | None]: + """Return public SQS v4. + + v4 keeps v3 intact as the deployment/stress backbone, then blends in a + modest full-cycle capital-growth score when a comparable common-window + run is available. If no common-window summary exists, v4 falls back to v3. + """ + v3_score, v3_breakdown, source = compute_public_sqs_v3( + train_result, + valid_result, + test_result, + walk_forward_summary=walk_forward_summary, + robustness_matrix_summary=robustness_matrix_summary, + out_of_time_robustness_summary=out_of_time_robustness_summary, + rqs_score=rqs_score, + wfqs_v2_score=wfqs_v2_score, + ) + if v3_score is None: + return None, v3_breakdown, source + + common_score, common_breakdown = compute_common_window_score(common_window_summary) + if common_score is None: + return v3_score, v3_breakdown, "v4_fallback_v3_missing_common_window" + + weight = _DEFAULT_PUBLIC_COMMON_WINDOW_WEIGHT + final_score = v3_score * (1.0 - weight) + common_score * weight + breakdown = { + "v3_score": round(v3_score, 1), + "common_window_score": round(common_score, 1), + "common_window_weight": round(weight, 2), + **v3_breakdown, + **common_breakdown, + } + return round(final_score, 1), breakdown, "v4_deployment+common_window" + + +def compute_deployment_score( + train_result: SplitResult | None, + valid_result: SplitResult | None, + test_result: SplitResult | None, + walk_forward_summary: WalkForwardSummary | None, + rqs_score: float | None = None, + wfqs_score: float | None = None, + weights: DeploymentScoreWeights | None = None, +) -> tuple[float | None, dict[str, float]]: + """Compute a deployment-oriented score that requires walk-forward confirmation.""" + if walk_forward_summary is None: + return None, {} + + rqs = rqs_score + if rqs is None: + rqs, _ = compute_rqs(train_result, valid_result, test_result) + wfqs = wfqs_score + if wfqs is None: + wfqs, _ = compute_wfqs(walk_forward_summary) + if rqs is None or wfqs is None: + return None, {} + + w = weights or _DEFAULT_DEPLOYMENT_WEIGHTS + test = walk_forward_summary.test_aggregate + gap = walk_forward_summary.gap_stats + + pass_positive = (test.positive_fold_rate_pct or 0.0) >= 70.0 + pass_median = (test.median_return_pct or 0.0) >= 5.0 + pass_worst = (test.worst_return_pct or 0.0) >= -5.0 + pass_gap = (gap.mean_train_test_return_gap_pct or 999.0) <= 35.0 + pass_count = sum([pass_positive, pass_median, pass_worst, pass_gap]) + gate_factor = { + 4: 1.00, + 3: 0.85, + 2: 0.65, + 1: 0.40, + 0: 0.20, + }[pass_count] + + deployment = (rqs * w.rqs_quality + wfqs * w.wfqs_quality) * gate_factor + breakdown = { + "rqs_quality": round(rqs, 1), + "wfqs_quality": round(wfqs, 1), + "gate_factor": round(gate_factor, 2), + "gate_positive_fold_rate": 1.0 if pass_positive else 0.0, + "gate_median_return": 1.0 if pass_median else 0.0, + "gate_worst_return": 1.0 if pass_worst else 0.0, + "gate_train_test_gap": 1.0 if pass_gap else 0.0, + } + return round(deployment, 1), breakdown + + +def compute_common_window_score( + common_window_summary: CommonWindowSummary | None, +) -> tuple[float | None, dict[str, float]]: + """Score a continuous full-cycle run for capital-growth / recycling quality.""" + if common_window_summary is None: + return None, {} + + metrics = common_window_summary.metrics + total_return_score = _normalize(metrics.total_return_pct, low=0.0, high=250.0) + profit_factor_score = _normalize(metrics.profit_factor, low=1.0, high=6.0) + sharpe_score = _normalize(metrics.sharpe_ratio, low=0.0, high=3.0) + drawdown_score = _normalize_inverse(metrics.max_drawdown_pct, low=12.0, high=2.5) + return_on_gross = _safe_ratio(metrics.total_return_pct, metrics.avg_gross_exposure_pct) + return_on_gross_score = _normalize(return_on_gross, low=1.0, high=8.0) + capital_velocity = _safe_ratio(metrics.total_return_pct, metrics.days_in_market_pct) + capital_velocity_score = _normalize(capital_velocity, low=0.4, high=3.0) + + score = ( + total_return_score * 0.35 + + profit_factor_score * 0.10 + + sharpe_score * 0.15 + + drawdown_score * 0.15 + + return_on_gross_score * 0.20 + + capital_velocity_score * 0.05 + ) + if metrics.trade_count < 25: + score *= 0.85 + + breakdown = { + "cw_total_return": round(total_return_score, 1), + "cw_profit_factor": round(profit_factor_score, 1), + "cw_sharpe": round(sharpe_score, 1), + "cw_drawdown": round(drawdown_score, 1), + "cw_return_on_gross": round(return_on_gross_score, 1), + "cw_capital_velocity": round(capital_velocity_score, 1), + "cw_trade_count": float(metrics.trade_count), + } + return round(score, 1), breakdown + + +def compute_overlay_window_score( + overlay_summary: OverlayWindowSummary | None, +) -> tuple[float | None, dict[str, float]]: + """Score an overlay on full-window capital growth and risk efficiency.""" + if overlay_summary is None: + return None, {} + + return_score = _normalize(overlay_summary.return_pct, low=0.0, high=180.0) + ann_score = _normalize(overlay_summary.annualized_return_pct, low=0.0, high=35.0) + sharpe_score = _normalize(overlay_summary.sharpe_ratio, low=0.0, high=3.0) + drawdown_score = _normalize_inverse(overlay_summary.max_drawdown_pct, low=12.0, high=2.5) + activity_score = _normalize(float(overlay_summary.day_count), low=250.0, high=1000.0) + + score = ( + return_score * 0.40 + + ann_score * 0.15 + + sharpe_score * 0.25 + + drawdown_score * 0.15 + + activity_score * 0.05 + ) + breakdown = { + "overlay_return": round(return_score, 1), + "overlay_annualized_return": round(ann_score, 1), + "overlay_sharpe": round(sharpe_score, 1), + "overlay_drawdown": round(drawdown_score, 1), + "overlay_activity": round(activity_score, 1), + "overlay_day_count": float(overlay_summary.day_count), + } + return round(score, 1), breakdown + + +def compute_overlay_stress_gate( + overlay_stress_summary: OverlayWindowSummary | None, +) -> tuple[float, dict[str, float]]: + """Gate overlay scores using a stress-period overlay run.""" + if overlay_stress_summary is None: + return 0.0, {"requires_overlay_stress_window": 1.0} + + pass_return = (overlay_stress_summary.return_pct or 0.0) >= 5.0 + pass_sharpe = (overlay_stress_summary.sharpe_ratio or 0.0) >= 0.50 + pass_drawdown = (overlay_stress_summary.max_drawdown_pct or 999.0) <= 10.0 + pass_count = sum([pass_return, pass_sharpe, pass_drawdown]) + gate_factor = { + 3: 1.00, + 2: 0.85, + 1: 0.65, + 0: 0.40, + }[pass_count] + return gate_factor, { + "overlay_gate_return": 1.0 if pass_return else 0.0, + "overlay_gate_sharpe": 1.0 if pass_sharpe else 0.0, + "overlay_gate_drawdown": 1.0 if pass_drawdown else 0.0, + } + + +def compute_overlay_stress_quality( + overlay_stress_summary: OverlayWindowSummary | None, +) -> tuple[float | None, dict[str, float]]: + """Diagnostic stress quality for overlay entries.""" + if overlay_stress_summary is None: + return None, {} + + return_score = _normalize(overlay_stress_summary.return_pct, low=0.0, high=20.0) + sharpe_score = _normalize(overlay_stress_summary.sharpe_ratio, low=0.0, high=1.5) + drawdown_score = _normalize_inverse(overlay_stress_summary.max_drawdown_pct, low=12.0, high=3.0) + quality = return_score * 0.45 + sharpe_score * 0.35 + drawdown_score * 0.20 + breakdown = { + "overlay_stress_return": round(return_score, 1), + "overlay_stress_sharpe": round(sharpe_score, 1), + "overlay_stress_drawdown": round(drawdown_score, 1), + } + return round(quality, 1), breakdown + + +def compute_overlay_public_sqs( + overlay_common_window_summary: OverlayWindowSummary | None, + overlay_stress_window_summary: OverlayWindowSummary | None, +) -> tuple[float | None, dict[str, float], str | None]: + """Return the official overlay score.""" + base_score, base_breakdown = compute_overlay_window_score(overlay_common_window_summary) + if base_score is None: + return None, {"requires_overlay_common_window": 1.0}, "pending_overlay_common_window" + gate_factor, gate_breakdown = compute_overlay_stress_gate(overlay_stress_window_summary) + if overlay_stress_window_summary is None: + return None, gate_breakdown, "pending_overlay_stress_window" + stress_quality, stress_breakdown = compute_overlay_stress_quality(overlay_stress_window_summary) + final_score = base_score * gate_factor + breakdown = { + "overlay_base_score": round(base_score, 1), + "overlay_gate_factor": round(gate_factor, 2), + **base_breakdown, + **gate_breakdown, + **stress_breakdown, + } + if stress_quality is not None: + breakdown["overlay_stress_quality"] = round(stress_quality, 1) + return round(final_score, 1), breakdown, "overlay_v1_common_window+stress_gate" + + +def compute_overlay_stress_sqs( + overlay_common_window_summary: OverlayWindowSummary | None, + overlay_stress_window_summary: OverlayWindowSummary | None, ) -> tuple[float | None, dict[str, float], str | None]: - """Return the public-facing SQS. + """Legacy overlay score with an extra stress-quality penalty.""" + base_score, base_breakdown = compute_overlay_window_score(overlay_common_window_summary) + if base_score is None: + return None, {"requires_overlay_common_window": 1.0}, "pending_overlay_common_window" + gate_factor, gate_breakdown = compute_overlay_stress_gate(overlay_stress_window_summary) + if overlay_stress_window_summary is None: + return None, gate_breakdown, "pending_overlay_stress_window" + stress_quality, stress_breakdown = compute_overlay_stress_quality(overlay_stress_window_summary) + quality_factor = 1.0 + if stress_quality is not None: + quality_factor = 0.30 + 0.70 * (stress_quality / 100.0) + final_score = base_score * gate_factor * quality_factor + breakdown = { + "overlay_base_score": round(base_score, 1), + "overlay_gate_factor": round(gate_factor, 2), + "overlay_quality_factor": round(quality_factor, 2), + **base_breakdown, + **gate_breakdown, + **stress_breakdown, + } + if stress_quality is not None: + breakdown["overlay_stress_quality"] = round(stress_quality, 1) + return round(final_score, 1), breakdown, "overlay_v1_common_window+stress_quality" + + +def _get_robustness_horizon_summary( + summary: RobustnessMatrixSummary | None, + horizon_days: int, +) -> dict[str, float | int | None] | None: + if summary is None: + return None + for item in summary.horizon_summaries: + if item.horizon_days == horizon_days: + return item.model_dump() + return None + + +def compute_robustness_gate( + robustness_matrix_summary: RobustnessMatrixSummary | None, +) -> tuple[float, dict[str, float]]: + """Return multiplicative gate from the robustness matrix summary.""" + if robustness_matrix_summary is None: + return 1.0, {} + + h63 = _get_robustness_horizon_summary(robustness_matrix_summary, 63) + h252 = _get_robustness_horizon_summary(robustness_matrix_summary, 252) + pass_positive = (robustness_matrix_summary.overall_positive_window_rate_pct or 0.0) >= 65.0 + pass_63 = ((h63 or {}).get("median_return_pct") or 0.0) >= 3.0 + pass_252 = ((h252 or {}).get("median_return_pct") or 0.0) >= 8.0 + pass_worst = (robustness_matrix_summary.overall_worst_return_pct or -999.0) >= -12.0 + pass_count = sum([pass_positive, pass_63, pass_252, pass_worst]) + gate_factor = { + 4: 1.00, + 3: 0.85, + 2: 0.65, + 1: 0.40, + 0: 0.20, + }[pass_count] + breakdown = { + "rb_gate_positive_rate": 1.0 if pass_positive else 0.0, + "rb_gate_63d_median": 1.0 if pass_63 else 0.0, + "rb_gate_252d_median": 1.0 if pass_252 else 0.0, + "rb_gate_worst_return": 1.0 if pass_worst else 0.0, + } + return gate_factor, breakdown + + +def _compute_oot_positive_rate_63plus( + summary: RobustnessMatrixSummary, +) -> float | None: + """63d+ horizon weighted-average positive window rate. + + 21d windows penalise catalyst strategies unfairly (0-trade windows + count as non-positive). 63d+ windows give a fairer survival signal. + """ + qualifying = [h for h in summary.horizon_summaries if h.horizon_days >= 63] + if not qualifying: + return None + total_windows = sum(h.window_count for h in qualifying) + if total_windows == 0: + return None + weighted_positive = sum( + (h.positive_window_rate_pct or 0.0) * h.window_count + for h in qualifying + ) + return round(weighted_positive / total_windows, 1) + + +def compute_oot_robustness_gate( + robustness_matrix_summary: RobustnessMatrixSummary | None, +) -> tuple[float, dict[str, float]]: + """OOT stress-test robustness gate with relaxed thresholds. + + Stress-test thresholds (vs main-period): + - 63d+ positive rate >= 50% (main: overall >= 65%) + - 63d median >= 0.5% (main: >= 3.0%) + - 252d median >= 3.0% (main: >= 8.0%) + - worst return >= -15.0% (main: >= -12.0%) + """ + if robustness_matrix_summary is None: + return 1.0, {} + + h63 = _get_robustness_horizon_summary(robustness_matrix_summary, 63) + h252 = _get_robustness_horizon_summary(robustness_matrix_summary, 252) + positive_rate_63plus = _compute_oot_positive_rate_63plus(robustness_matrix_summary) + pass_positive = (positive_rate_63plus or 0.0) >= 50.0 + pass_63 = ((h63 or {}).get("median_return_pct") or 0.0) >= 0.5 + pass_252 = ((h252 or {}).get("median_return_pct") or 0.0) >= 3.0 + pass_worst = (robustness_matrix_summary.overall_worst_return_pct or -999.0) >= -15.0 + pass_count = sum([pass_positive, pass_63, pass_252, pass_worst]) + gate_factor = { + 4: 1.00, + 3: 0.85, + 2: 0.65, + 1: 0.40, + 0: 0.20, + }[pass_count] + breakdown = { + "rb_gate_positive_rate": 1.0 if pass_positive else 0.0, + "rb_gate_63d_median": 1.0 if pass_63 else 0.0, + "rb_gate_252d_median": 1.0 if pass_252 else 0.0, + "rb_gate_worst_return": 1.0 if pass_worst else 0.0, + } + return gate_factor, breakdown - Prefer the stricter integrated score when both valid/test are available. - Fall back to the best available single-split quality score otherwise, - using the same harsher calibration band. + +def compute_robustness_quality( + robustness_matrix_summary: RobustnessMatrixSummary | None, +) -> tuple[float | None, dict[str, float]]: + """Return a continuous robustness quality score in [0, 100]. + + This complements the coarse pass/fail gate so repaired OOT runs can + meaningfully reorder strategies instead of collapsing into the same gate. """ - integrated_score, integrated_breakdown = compute_unified_score(test_result, valid_result) - if integrated_score is not None: - return integrated_score, integrated_breakdown, "integrated" + if robustness_matrix_summary is None: + return None, {} + + h63 = _get_robustness_horizon_summary(robustness_matrix_summary, 63) + h252 = _get_robustness_horizon_summary(robustness_matrix_summary, 252) + positive_rate_score = _normalize( + robustness_matrix_summary.overall_positive_window_rate_pct, + low=30.0, + high=80.0, + ) + worst_return_score = _normalize( + robustness_matrix_summary.overall_worst_return_pct, + low=-25.0, + high=0.0, + ) + h63_median_score = _normalize( + (h63 or {}).get("median_return_pct"), + low=-5.0, + high=5.0, + ) + h252_median_score = _normalize( + (h252 or {}).get("median_return_pct"), + low=-10.0, + high=15.0, + ) + quality = ( + positive_rate_score * 0.30 + + h63_median_score * 0.25 + + h252_median_score * 0.30 + + worst_return_score * 0.15 + ) + breakdown = { + "rb_quality_positive_rate": round(positive_rate_score, 1), + "rb_quality_63d_median": round(h63_median_score, 1), + "rb_quality_252d_median": round(h252_median_score, 1), + "rb_quality_worst_return": round(worst_return_score, 1), + } + return round(quality, 1), breakdown + - split_score, split_breakdown, split_source = _compute_split_quality_score(test_result) - if split_score is not None: - return _apply_single_split_penalty(_calibrate_sqs(split_score)), split_breakdown, split_source +def compute_oot_robustness_quality( + robustness_matrix_summary: RobustnessMatrixSummary | None, +) -> tuple[float | None, dict[str, float]]: + """OOT stress-test continuous quality score [0, 100]. - return None, {}, None + Normalisation ranges adjusted for stress-test periods (e.g. COVID). + Uses 63d+ positive rate instead of overall (excludes 21d noise). + """ + if robustness_matrix_summary is None: + return None, {} + + h63 = _get_robustness_horizon_summary(robustness_matrix_summary, 63) + h252 = _get_robustness_horizon_summary(robustness_matrix_summary, 252) + positive_rate_63plus = _compute_oot_positive_rate_63plus(robustness_matrix_summary) + positive_rate_score = _normalize( + positive_rate_63plus, + low=20.0, + high=70.0, + ) + worst_return_score = _normalize( + robustness_matrix_summary.overall_worst_return_pct, + low=-30.0, + high=-5.0, + ) + h63_median_score = _normalize( + (h63 or {}).get("median_return_pct"), + low=-8.0, + high=3.0, + ) + h252_median_score = _normalize( + (h252 or {}).get("median_return_pct"), + low=-15.0, + high=10.0, + ) + quality = ( + positive_rate_score * 0.30 + + h63_median_score * 0.25 + + h252_median_score * 0.30 + + worst_return_score * 0.15 + ) + breakdown = { + "rb_quality_positive_rate": round(positive_rate_score, 1), + "rb_quality_63d_median": round(h63_median_score, 1), + "rb_quality_252d_median": round(h252_median_score, 1), + "rb_quality_worst_return": round(worst_return_score, 1), + } + return round(quality, 1), breakdown def compute_config_delta( @@ -523,6 +1702,23 @@ def append_journal_entry(journal_path: Path, entry: JournalEntry) -> None: logger.info("journal_entry_appended", entry_id=entry.entry_id, experiment=entry.experiment_name) +def replace_journal_entry(journal_path: Path, entry: JournalEntry) -> None: + """Replace an existing journal entry by entry_id.""" + entries = load_journal(journal_path) + replaced = False + updated_entries: list[JournalEntry] = [] + for current in entries: + if current.entry_id == entry.entry_id: + updated_entries.append(entry) + replaced = True + else: + updated_entries.append(current) + if not replaced: + raise ValueError(f"Journal entry not found: {entry.entry_id}") + journal_path.write_text("".join(e.model_dump_json() + "\n" for e in updated_entries)) + logger.info("journal_entry_replaced", entry_id=entry.entry_id, experiment=entry.experiment_name) + + def load_journal(journal_path: Path) -> list[JournalEntry]: """Load all journal entries from JSONL file.""" if not journal_path.exists(): @@ -535,6 +1731,621 @@ def load_journal(journal_path: Path) -> list[JournalEntry]: return entries +def refresh_public_scores( + journal_path: Path, + selector: Callable[[JournalEntry], bool] | None = None, +) -> int: + """Recompute stored public scores and score diagnostics for journal entries.""" + entries = load_journal(journal_path) + updated_entries: list[JournalEntry] = [] + updated_count = 0 + + for entry in entries: + if selector is not None and not selector(entry): + updated_entries.append(entry) + continue + + if entry.overlay_common_window_summary is not None: + overlay_sqs, overlay_breakdown, _ = compute_overlay_public_sqs( + entry.overlay_common_window_summary, + entry.overlay_stress_window_summary, + ) + overlay_stress_sqs, overlay_stress_breakdown, _ = compute_overlay_stress_sqs( + entry.overlay_common_window_summary, + entry.overlay_stress_window_summary, + ) + refreshed = entry.model_copy( + update={ + "sqs_score": overlay_sqs, + "sqs_breakdown": overlay_breakdown, + "sqs_v3_score": overlay_sqs, + "sqs_v3_breakdown": overlay_breakdown, + "stress_sqs_score": overlay_stress_sqs, + "stress_sqs_breakdown": overlay_stress_breakdown, + "common_window_score": None, + "common_window_breakdown": {}, + } + ) + if refreshed.model_dump() != entry.model_dump(): + updated_count += 1 + updated_entries.append(refreshed) + continue + + test_result = _hydrate_split_result(entry.results.get("test")) + valid_result = _hydrate_split_result(entry.results.get("valid")) + train_result = _hydrate_split_result(entry.results.get("train")) + test_metrics = _metrics_from_split_result(test_result) + + sqs_v2_score = None + sqs_v2_breakdown: dict[str, float] = {} + if test_metrics is not None: + sqs_v2_score, sqs_v2_breakdown = compute_sqs_v2(test_metrics) + + rqs_score, rqs_breakdown = compute_rqs(train_result, valid_result, test_result) + wfqs_score, wfqs_breakdown = compute_wfqs(entry.walk_forward_summary) + wfqs_v2_score, wfqs_v2_breakdown = compute_wfqs_v2(entry.walk_forward_summary) + deployment_score, deployment_breakdown = compute_deployment_score( + train_result, + valid_result, + test_result, + entry.walk_forward_summary, + rqs_score=rqs_score, + wfqs_score=wfqs_score, + ) + common_window_score, common_window_breakdown = compute_common_window_score( + entry.common_window_summary + ) + sqs_v3_score, sqs_v3_breakdown, _ = compute_public_sqs_v3( + train_result, + valid_result, + test_result, + walk_forward_summary=entry.walk_forward_summary, + robustness_matrix_summary=entry.robustness_matrix_summary, + out_of_time_robustness_summary=entry.out_of_time_robustness_summary, + rqs_score=rqs_score, + wfqs_v2_score=wfqs_v2_score, + ) + sqs_score, sqs_breakdown, _ = compute_public_sqs( + train_result, + valid_result, + test_result, + walk_forward_summary=entry.walk_forward_summary, + 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, + rqs_score=rqs_score, + wfqs_score=wfqs_v2_score, + ) + stress_sqs_score, stress_sqs_breakdown, _ = compute_public_sqs_v2( + train_result, + valid_result, + test_result, + walk_forward_summary=entry.walk_forward_summary, + robustness_matrix_summary=entry.robustness_matrix_summary, + out_of_time_robustness_summary=entry.out_of_time_robustness_summary, + rqs_score=rqs_score, + wfqs_v2_score=wfqs_v2_score, + ) + + refreshed = entry.model_copy( + update={ + "sqs_score": sqs_score, + "sqs_breakdown": sqs_breakdown, + "sqs_v3_score": sqs_v3_score, + "sqs_v3_breakdown": sqs_v3_breakdown, + "stress_sqs_score": stress_sqs_score, + "stress_sqs_breakdown": stress_sqs_breakdown, + "sqs_v2_score": sqs_v2_score, + "sqs_v2_breakdown": sqs_v2_breakdown, + "rqs_score": rqs_score, + "rqs_breakdown": rqs_breakdown, + "wfqs_score": wfqs_score, + "wfqs_breakdown": wfqs_breakdown, + "wfqs_v2_score": wfqs_v2_score, + "wfqs_v2_breakdown": wfqs_v2_breakdown, + "deployment_score": deployment_score, + "deployment_breakdown": deployment_breakdown, + "common_window_score": common_window_score, + "common_window_breakdown": common_window_breakdown, + } + ) + if refreshed.model_dump() != entry.model_dump(): + updated_count += 1 + updated_entries.append(refreshed) + + if updated_count: + journal_path.write_text("".join(item.model_dump_json() + "\n" for item in updated_entries)) + logger.info("journal_public_scores_refreshed", updated_count=updated_count) + + return updated_count + + +def _resolve_journal_target(entries: list[JournalEntry], target: str) -> JournalEntry: + target_upper = target.upper() + by_id = [entry for entry in entries if entry.entry_id.upper() == target_upper] + if by_id: + return by_id[0] + + exact_name = [entry for entry in entries if entry.experiment_name == target] + if len(exact_name) == 1: + return exact_name[0] + if len(exact_name) > 1: + return sorted(exact_name, key=lambda e: e.timestamp)[-1] + + partial = [entry for entry in entries if target.lower() in entry.experiment_name.lower()] + if len(partial) != 1: + raise ValueError(f"Unable to uniquely match journal entry: {target}") + return partial[0] + + +def attach_walk_forward_summary( + journal_path: Path, + target: str, + summary: WalkForwardSummary, +) -> JournalEntry: + """Attach walk-forward validation summary to an existing journal entry.""" + entries = load_journal(journal_path) + selected = _resolve_journal_target(entries, target) + + wfqs_score, wfqs_breakdown = compute_wfqs(summary) + wfqs_v2_score, wfqs_v2_breakdown = compute_wfqs_v2(summary) + deployment_score, deployment_breakdown = compute_deployment_score( + selected.results.get("train"), + selected.results.get("valid"), + selected.results.get("test"), + summary, + ) + updated = selected.model_copy( + update={ + "walk_forward_summary": summary, + "wfqs_score": wfqs_score, + "wfqs_breakdown": wfqs_breakdown, + "wfqs_v2_score": wfqs_v2_score, + "wfqs_v2_breakdown": wfqs_v2_breakdown, + "deployment_score": deployment_score, + "deployment_breakdown": deployment_breakdown, + } + ) + replace_journal_entry(journal_path, updated) + return updated + + +def attach_robustness_summary( + journal_path: Path, + target: str, + summary: RobustnessMatrixSummary, +) -> JournalEntry: + """Attach robustness matrix summary to an existing journal entry.""" + entries = load_journal(journal_path) + selected = _resolve_journal_target(entries, target) + + updated = selected.model_copy( + update={ + "robustness_matrix_summary": summary, + } + ) + replace_journal_entry(journal_path, updated) + return updated + + +def attach_out_of_time_robustness_summary( + journal_path: Path, + target: str, + summary: RobustnessMatrixSummary, +) -> JournalEntry: + """Attach out-of-time robustness summary to an existing journal entry.""" + entries = load_journal(journal_path) + selected = _resolve_journal_target(entries, target) + + updated = selected.model_copy( + update={ + "out_of_time_robustness_summary": summary, + } + ) + replace_journal_entry(journal_path, updated) + return updated + + +def attach_common_window_summary( + journal_path: Path, + target: str, + summary: CommonWindowSummary, +) -> JournalEntry: + """Attach common-window continuous-run summary to an existing journal entry.""" + entries = load_journal(journal_path) + selected = _resolve_journal_target(entries, target) + + common_window_score, common_window_breakdown = compute_common_window_score(summary) + updated = selected.model_copy( + update={ + "common_window_summary": summary, + "common_window_score": common_window_score, + "common_window_breakdown": common_window_breakdown, + } + ) + replace_journal_entry(journal_path, updated) + return updated + + +@functools.lru_cache(maxsize=1024) +def _load_manifest_json(experiment_name: str) -> dict[str, Any] | None: + path = _EXPERIMENTS_DIR / f"{experiment_name}.json" + if not path.exists(): + return None + with path.open("r", encoding="utf-8") as fh: + return json.load(fh) + + +def _manifest_has_exact_pocket_structure(experiment_name: str) -> bool: + manifest = _load_manifest_json(experiment_name) + if manifest is None: + return False + for engine in manifest.get("strategy_engines", []): + engine_id = str(engine.get("engine_id", "")).lower() + if engine.get("enabled", True) and "exact" in engine_id: + return True + return False + + +def _manifest_has_named_micro_structure(experiment_name: str) -> bool: + manifest = _load_manifest_json(experiment_name) + if manifest is None: + return False + for engine in manifest.get("strategy_engines", []): + engine_id = str(engine.get("engine_id", "")).lower() + if any(token in engine_id for token in _NAMED_MICRO_ENGINE_TOKENS): + return True + return False + + +def classify_strategy_family(experiment_name: str, tags: list[str] | None = None) -> str: + """Classify an experiment into a coarse strategy family for leaderboard filtering.""" + lowered_name = experiment_name.lower() + lowered_tags = {tag.lower() for tag in tags or []} + + if lowered_name.startswith("return_book_overlay") or "overlay" in lowered_tags: + return "overlay" + if _manifest_has_exact_pocket_structure(experiment_name): + return "exact_pocket_return_max_long" + if _manifest_has_named_micro_structure(experiment_name): + return "named_micro_return_max_long" + if ( + "exact" in lowered_name + or "combo" in lowered_name + or "exact" in lowered_tags + or "combo" in lowered_tags + ): + return "exact_pocket_return_max_long" + if lowered_name.startswith("return_max_long") and "_bp" in lowered_name: + return "leveraged_return_max_long" + if lowered_name.startswith("return_max_long"): + return "return_max_long" + if "short_core" in lowered_name or "short_core" in lowered_tags: + return "short_core" + if lowered_name.startswith("pead_") or "pead" in lowered_tags: + return "legacy_pead" + return "other" + + +def is_retired_strategy_family(strategy_family: str) -> bool: + """Return True when the strategy family is retired from the default workflow.""" + return strategy_family in _RETIRED_STRATEGY_FAMILIES + + +def filter_registry_entries( + entries: list[RegistryEntry], + *, + include_retired: bool = False, + include_overlays: bool = False, +) -> list[RegistryEntry]: + """Filter registry entries for the default surfaced leaderboard.""" + filtered = list(entries) if include_retired else [entry for entry in entries if not entry.is_retired] + return [ + entry + for entry in filtered + if ( + entry.sqs_score is not None + and ( + (include_overlays and entry.strategy_family == "overlay") + or ( + entry.strategy_family != "overlay" + and entry.trade_count > 0 + and entry.valid_trade_count > 0 + ) + ) + ) + ] + + +def filter_overlay_registry_entries( + entries: list[RegistryEntry], + *, + include_retired: bool = False, +) -> list[RegistryEntry]: + """Filter registry entries for the dedicated overlay leaderboard.""" + filtered = list(entries) if include_retired else [entry for entry in entries if not entry.is_retired] + return [ + entry + for entry in filtered + if ( + entry.sqs_score is not None + and entry.strategy_family == "overlay" + and entry.overlay_common_window_summary is not None + ) + ] + + +def _is_retired_journal_entry(entry: JournalEntry) -> bool: + return is_retired_strategy_family(classify_strategy_family(entry.experiment_name, entry.tags)) + + +def _is_complete_journal_entry(entry: JournalEntry) -> bool: + if entry.overlay_common_window_summary is not None: + return entry.sqs_score is not None + test_result = entry.results.get("test") + valid_result = entry.results.get("valid") + return ( + test_result is not None + and valid_result is not None + and test_result.trade_count > 0 + and valid_result.trade_count > 0 + ) + + +def _load_optional_walk_forward_summary(experiment_name: str) -> WalkForwardSummary | None: + summary_path = Path("runs") / f"{experiment_name}_wfv" / "walk_forward" / "walk_forward_summary.json" + if not summary_path.exists(): + return None + return WalkForwardSummary.model_validate_json(summary_path.read_text()) + + +def _load_optional_robustness_summary(experiment_name: str) -> RobustnessMatrixSummary | None: + summary_path = ( + Path("runs") + / f"{experiment_name}_rm" + / "robustness_matrix" + / "robustness_matrix_summary.json" + ) + if not summary_path.exists(): + return None + return RobustnessMatrixSummary.model_validate_json(summary_path.read_text()) + + +def _load_optional_out_of_time_robustness_summary(experiment_name: str) -> RobustnessMatrixSummary | None: + summary_path = ( + Path("runs") + / f"{experiment_name}_oot_rm" + / "robustness_matrix" + / "robustness_matrix_summary.json" + ) + if not summary_path.exists(): + return None + return RobustnessMatrixSummary.model_validate_json(summary_path.read_text()) + + +def _build_manifest_index( + runs_dir: Path, +) -> dict[str, list[tuple[Path, dict, dict]]]: + """Build experiment_name -> [(run_dir, manifest_data, metadata)] index in one rglob pass.""" + index: dict[str, list[tuple[Path, dict, dict]]] = {} + for manifest_file in sorted(runs_dir.rglob("manifest.json")): + run_path = manifest_file.parent + try: + manifest_data = json.loads(manifest_file.read_text()) + except Exception: + continue + exp_name = manifest_data.get("experiment_name") + if not exp_name: + continue + metadata_file = run_path / "metadata.json" + try: + metadata = json.loads(metadata_file.read_text()) if metadata_file.exists() else {} + except Exception: + metadata = {} + index.setdefault(exp_name, []).append((run_path, manifest_data, metadata)) + return index + + +def _scan_runs_from_index( + run_entries: list[tuple[Path, dict, dict]], +) -> dict[str, tuple[str, "MetricsBundle"]]: + """Resolve split runs from pre-indexed manifest entries (no rglob).""" + from libs.backtest.domain import MetricsBundle + + matches: list[tuple[Path, str, str, str, str | None]] = [] + for run_path, _manifest_data, metadata in run_entries: + run_id = metadata.get("run_id", run_path.name) + started_at = str(metadata.get("started_at") or metadata.get("finished_at") or "") + split_name = metadata.get("split_name") + parts = run_path.name.split("_") + config_hash = parts[-1] if len(parts) >= 2 else "" + matches.append((run_path, run_id, config_hash, started_at, split_name)) + + if not matches: + return {} + + results: dict[str, tuple[str, MetricsBundle]] = {} + has_split_names = any(sn is not None for _, _, _, _, sn in matches) + canonical_splits = {"train", "valid", "test"} + + if has_split_names: + canonical_matches = [m for m in matches if m[4] in canonical_splits] + if not canonical_matches: + return {} + latest_by_split: dict[str, tuple[str, Path, str]] = {} + for run_path, run_id, _, started_at, split_name in canonical_matches: + split = split_name or "unknown" + current = latest_by_split.get(split) + if current is None or started_at >= current[0]: + latest_by_split[split] = (started_at, run_path, run_id) + for split, (_, run_path, run_id) in latest_by_split.items(): + metrics_file = run_path / "metrics" / "metrics_summary.json" + if metrics_file.exists(): + metrics = MetricsBundle.model_validate_json(metrics_file.read_text()) + results[split] = (run_id, metrics) + else: + from collections import defaultdict + groups: dict[str, list[tuple[str, Path, str]]] = defaultdict(list) + for run_path, run_id, config_hash, started_at, _ in matches: + groups[config_hash].append((started_at, run_path, run_id)) + best_hash = max(groups, key=lambda h: max(t[0] for t in groups[h])) + sorted_runs = sorted(groups[best_hash], key=lambda t: t[0]) + split_order = ["train", "valid", "test"] + for i, (_, run_path, run_id) in enumerate(sorted_runs[-3:]): + split = split_order[i] if i < 3 else f"extra_{i}" + metrics_file = run_path / "metrics" / "metrics_summary.json" + if metrics_file.exists(): + metrics = MetricsBundle.model_validate_json(metrics_file.read_text()) + results[split] = (run_id, metrics) + + return results + + +def sync_official_manifests( + journal_path: Path, + runs_dir: Path, + configs_dir: Path | None = None, +) -> list[JournalEntry]: + """Append official manifests with complete runs that are missing from the journal.""" + configs_root = configs_dir or _EXPERIMENTS_DIR + existing_entries = load_journal(journal_path) + existing_names = {entry.experiment_name for entry in existing_entries} + next_index = len(existing_entries) + 1 + synced_entries: list[JournalEntry] = [] + + # Build manifest index once (single rglob) instead of per-experiment + manifest_index = _build_manifest_index(runs_dir) + + for manifest_path in sorted(configs_root.glob("*.json")): + manifest_payload = json.loads(manifest_path.read_text()) + experiment_name = str(manifest_payload.get("experiment_name") or manifest_path.stem) + if experiment_name in existing_names: + continue + + run_entries = manifest_index.get(experiment_name) + if not run_entries: + continue + split_runs = _scan_runs_from_index(run_entries) + if not {"train", "valid", "test"}.issubset(split_runs): + continue + + results: dict[str, SplitResult] = {} + for split_name in ("train", "valid", "test"): + run_id, metrics = split_runs[split_name] + results[split_name] = build_split_result(split_name, run_id, metrics) + + test_metrics = _metrics_from_split_result(results.get("test")) + sqs_score = None + sqs_breakdown: dict[str, float] = {} + sqs_v2_score = None + sqs_v2_breakdown: dict[str, float] = {} + if test_metrics is not None: + sqs_score, sqs_breakdown = compute_sqs(test_metrics) + sqs_v2_score, sqs_v2_breakdown = compute_sqs_v2(test_metrics) + + promotion_score, promotion_breakdown = compute_promotion_score( + results.get("test"), + results.get("valid"), + ) + unified_score, unified_breakdown = compute_unified_score( + results.get("test"), + results.get("valid"), + ) + rqs_score, rqs_breakdown = compute_rqs( + results.get("train"), + results.get("valid"), + results.get("test"), + ) + walk_forward_summary = _load_optional_walk_forward_summary(experiment_name) + wfqs_score, wfqs_breakdown = compute_wfqs(walk_forward_summary) + wfqs_v2_score, wfqs_v2_breakdown = compute_wfqs_v2(walk_forward_summary) + deployment_score, deployment_breakdown = compute_deployment_score( + results.get("train"), + results.get("valid"), + results.get("test"), + walk_forward_summary, + rqs_score=rqs_score, + wfqs_score=wfqs_score, + ) + robustness_matrix_summary = _load_optional_robustness_summary(experiment_name) + out_of_time_robustness_summary = _load_optional_out_of_time_robustness_summary(experiment_name) + public_sqs, public_breakdown, _ = compute_public_sqs( + results.get("train"), + results.get("valid"), + results.get("test"), + walk_forward_summary=walk_forward_summary, + robustness_matrix_summary=robustness_matrix_summary, + out_of_time_robustness_summary=out_of_time_robustness_summary, + rqs_score=rqs_score, + wfqs_score=wfqs_v2_score, + ) + public_sqs_v3, public_breakdown_v3, _ = compute_public_sqs_v3( + results.get("train"), + results.get("valid"), + results.get("test"), + walk_forward_summary=walk_forward_summary, + robustness_matrix_summary=robustness_matrix_summary, + out_of_time_robustness_summary=out_of_time_robustness_summary, + rqs_score=rqs_score, + wfqs_v2_score=wfqs_v2_score, + ) + stress_sqs, stress_breakdown, _ = compute_public_sqs_v2( + results.get("train"), + results.get("valid"), + results.get("test"), + walk_forward_summary=walk_forward_summary, + robustness_matrix_summary=robustness_matrix_summary, + out_of_time_robustness_summary=out_of_time_robustness_summary, + rqs_score=rqs_score, + wfqs_v2_score=wfqs_v2_score, + ) + if public_sqs is not None: + sqs_score = public_sqs + sqs_breakdown = public_breakdown + + timestamp = utc_now().isoformat() + entry = JournalEntry( + entry_id=f"IMP-{next_index:04d}", + timestamp=timestamp, + experiment_name=experiment_name, + hypothesis=str(manifest_payload.get("description") or manifest_payload.get("notes") or ""), + results=results, + walk_forward_summary=walk_forward_summary, + robustness_matrix_summary=robustness_matrix_summary, + out_of_time_robustness_summary=out_of_time_robustness_summary, + sqs_score=sqs_score, + sqs_breakdown=sqs_breakdown, + sqs_v3_score=public_sqs_v3, + sqs_v3_breakdown=public_breakdown_v3, + stress_sqs_score=stress_sqs, + stress_sqs_breakdown=stress_breakdown, + sqs_v2_score=sqs_v2_score, + sqs_v2_breakdown=sqs_v2_breakdown, + promotion_score=promotion_score, + promotion_breakdown=promotion_breakdown, + unified_score=unified_score, + unified_breakdown=unified_breakdown, + rqs_score=rqs_score, + rqs_breakdown=rqs_breakdown, + wfqs_score=wfqs_score, + wfqs_breakdown=wfqs_breakdown, + wfqs_v2_score=wfqs_v2_score, + wfqs_v2_breakdown=wfqs_v2_breakdown, + deployment_score=deployment_score, + deployment_breakdown=deployment_breakdown, + verdict="unknown", + verdict_reasoning="Auto-synced from official manifest.", + next_direction="", + tags=list(manifest_payload.get("tags") or []), + ) + append_journal_entry(journal_path, entry) + synced_entries.append(entry) + existing_names.add(experiment_name) + next_index += 1 + + return synced_entries + + # --------------------------------------------------------------------------- # Registry / Leaderboard # --------------------------------------------------------------------------- @@ -550,21 +2361,115 @@ def rebuild_registry( registry_entries: list[RegistryEntry] = [] for je in entries: + strategy_family = classify_strategy_family(je.experiment_name, je.tags) + is_retired = is_retired_strategy_family(strategy_family) + + if je.overlay_common_window_summary is not None: + computed_overlay_sqs, _, _ = compute_overlay_public_sqs( + je.overlay_common_window_summary, + je.overlay_stress_window_summary, + ) + computed_overlay_stress_sqs, _, _ = compute_overlay_stress_sqs( + je.overlay_common_window_summary, + je.overlay_stress_window_summary, + ) + overlay = je.overlay_common_window_summary + registry_entries.append( + RegistryEntry( + entry_id=je.entry_id, + experiment_name=je.experiment_name, + strategy_family=strategy_family, + is_retired=is_retired, + sqs_score=computed_overlay_sqs, + sqs_v3_score=computed_overlay_sqs, + stress_sqs_score=computed_overlay_stress_sqs, + common_window_score=None, + overlay_common_window_summary=je.overlay_common_window_summary, + overlay_stress_window_summary=je.overlay_stress_window_summary, + total_return_pct=overlay.return_pct, + annualized_return_pct=overlay.annualized_return_pct, + sharpe_ratio=overlay.sharpe_ratio, + max_drawdown_pct=overlay.max_drawdown_pct, + trade_count=0, + timestamp=je.timestamp, + ) + ) + continue + test_result = _hydrate_split_result(je.results.get("test")) valid_result = _hydrate_split_result(je.results.get("valid")) + train_result = _hydrate_split_result(je.results.get("train")) computed_sqs_v2 = None + computed_rqs = None + computed_rqs_breakdown: dict[str, float] = {} + computed_wfqs = None + computed_wfqs_breakdown: dict[str, float] = {} + computed_deployment = None + computed_deployment_breakdown: dict[str, float] = {} test_metrics = _metrics_from_split_result(test_result) if test_metrics is not None: computed_sqs_v2, _ = compute_sqs_v2(test_metrics) computed_promotion_score, _ = compute_promotion_score(test_result, valid_result) computed_unified_score, _ = compute_unified_score(test_result, valid_result) - computed_public_sqs, _, _ = compute_public_sqs(test_result, valid_result) - canonical_sqs = computed_public_sqs if computed_public_sqs is not None else je.sqs_score or 0.0 + computed_rqs, computed_rqs_breakdown = compute_rqs( + train_result, + valid_result, + test_result, + ) + computed_wfqs, computed_wfqs_breakdown = compute_wfqs(je.walk_forward_summary) + computed_wfqs_v2, _ = compute_wfqs_v2(je.walk_forward_summary) + computed_deployment, computed_deployment_breakdown = compute_deployment_score( + train_result, + valid_result, + test_result, + je.walk_forward_summary, + rqs_score=computed_rqs, + wfqs_score=computed_wfqs, + ) + computed_common_window, computed_common_window_breakdown = compute_common_window_score( + je.common_window_summary + ) + computed_public_sqs_v3, _, _ = compute_public_sqs_v3( + train_result, + valid_result, + test_result, + walk_forward_summary=je.walk_forward_summary, + robustness_matrix_summary=je.robustness_matrix_summary, + out_of_time_robustness_summary=je.out_of_time_robustness_summary, + rqs_score=computed_rqs, + wfqs_v2_score=computed_wfqs_v2, + ) + computed_public_sqs, computed_public_breakdown, _ = compute_public_sqs( + train_result, + valid_result, + test_result, + walk_forward_summary=je.walk_forward_summary, + robustness_matrix_summary=je.robustness_matrix_summary, + out_of_time_robustness_summary=je.out_of_time_robustness_summary, + common_window_summary=je.common_window_summary, + rqs_score=computed_rqs, + wfqs_score=computed_wfqs_v2, + ) + computed_stress_sqs, _, _ = compute_public_sqs_v2( + train_result, + valid_result, + test_result, + walk_forward_summary=je.walk_forward_summary, + robustness_matrix_summary=je.robustness_matrix_summary, + out_of_time_robustness_summary=je.out_of_time_robustness_summary, + rqs_score=computed_rqs, + wfqs_v2_score=computed_wfqs_v2, + ) + canonical_sqs = computed_public_sqs registry_entries.append( RegistryEntry( entry_id=je.entry_id, experiment_name=je.experiment_name, + strategy_family=strategy_family, + is_retired=is_retired, sqs_score=canonical_sqs, + sqs_v3_score=computed_public_sqs_v3, + stress_sqs_score=computed_stress_sqs, sqs_v2_score=je.sqs_v2_score if je.sqs_v2_score is not None else computed_sqs_v2, promotion_score=( je.promotion_score if je.promotion_score is not None else computed_promotion_score @@ -572,8 +2477,22 @@ def rebuild_registry( unified_score=( je.unified_score if je.unified_score is not None else computed_unified_score ), + rqs_score=computed_rqs if computed_rqs is not None else je.rqs_score, + wfqs_score=computed_wfqs if computed_wfqs is not None else je.wfqs_score, + wfqs_v2_score=computed_wfqs_v2 if computed_wfqs_v2 is not None else je.wfqs_v2_score, + deployment_score=computed_deployment if computed_deployment is not None else je.deployment_score, + common_window_score=( + computed_common_window if computed_common_window is not None else je.common_window_score + ), + common_window_summary=je.common_window_summary, + walk_forward_summary=je.walk_forward_summary, + robustness_matrix_summary=je.robustness_matrix_summary, + out_of_time_robustness_summary=je.out_of_time_robustness_summary, + train_total_return_pct=train_result.total_return_pct if train_result else None, + train_annualized_return_pct=train_result.annualized_return_pct if train_result else None, profit_factor=test_result.profit_factor if test_result else None, total_return_pct=test_result.total_return_pct if test_result else None, + annualized_return_pct=test_result.annualized_return_pct if test_result else None, win_rate=test_result.win_rate if test_result else None, sharpe_ratio=test_result.sharpe_ratio if test_result else None, max_drawdown_pct=test_result.max_drawdown_pct if test_result else None, @@ -583,6 +2502,7 @@ def rebuild_registry( days_in_market_pct=test_result.days_in_market_pct if test_result else None, valid_profit_factor=valid_result.profit_factor if valid_result else None, valid_total_return_pct=valid_result.total_return_pct if valid_result else None, + valid_annualized_return_pct=valid_result.annualized_return_pct if valid_result else None, valid_win_rate=valid_result.win_rate if valid_result else None, valid_sharpe_ratio=valid_result.sharpe_ratio if valid_result else None, valid_max_drawdown_pct=valid_result.max_drawdown_pct if valid_result else None, @@ -594,10 +2514,16 @@ def rebuild_registry( ) ) - # Sort by canonical SQS descending, then promotion. + # Default registry / markdown ordering is public SQS v3 descending. registry_entries.sort( key=lambda e: ( -(e.sqs_score or 0.0), + e.stress_sqs_score is None, + -(e.stress_sqs_score or 0.0), + e.deployment_score is None, + -(e.deployment_score or 0.0), + e.rqs_score is None, + -(e.rqs_score or 0.0), e.promotion_score is None, -(e.promotion_score or 0.0), ) @@ -612,8 +2538,9 @@ def rebuild_registry( registry_path.parent.mkdir(parents=True, exist_ok=True) registry_path.write_text(registry.model_dump_json(indent=2)) - # Write LEADERBOARD.md + # Write LEADERBOARD.md and OVERLAY_LEADERBOARD.md _write_leaderboard_md(leaderboard_path, registry, entries) + _write_overlay_leaderboard_md(leaderboard_path.parent / "OVERLAY_LEADERBOARD.md", registry, entries) logger.info("registry_rebuilt", count=len(registry_entries)) return registry @@ -624,46 +2551,116 @@ def _write_leaderboard_md( registry: ExperimentRegistry, journal_entries: list[JournalEntry], ) -> None: + visible_entries = filter_registry_entries( + registry.entries, + include_retired=False, + include_overlays=False, + ) + visible_recent = [ + entry + for entry in reversed(journal_entries) + if ( + not _is_retired_journal_entry(entry) + and _is_complete_journal_entry(entry) + and classify_strategy_family(entry.experiment_name, entry.tags) != "overlay" + ) + ][:5] lines: list[str] = [] lines.append("# Strategy Improvement Leaderboard") lines.append(f"_Updated: {registry.updated_at}_\n") - lines.append("| # | Experiment | SQS | [T]PF | [T]Ret% | [T]WR | [T]Sharpe | [T]DD% | [T]N | [T]Gross% | [T]Net% | [T]DIM% | [V]PF | [V]Ret% | [V]WR | [V]Sharpe | [V]DD% | [V]N | [V]Gross% | [V]Net% | [V]DIM% | Date |") - lines.append("|---|-----------|-----|-------|---------|-------|-----------|--------|------|-----------|---------|---------|-------|---------|-------|-----------|--------|------|-----------|---------|---------|------|") + lines.append("_Default view excludes overlay/book-of-books rows, retired legacy PEAD / short-core / exact-pocket families, and incomplete train-only scans. Use `fithia2 lb --overlay-only` for overlays or `fithia2 lb --include-retired` to inspect archived research._\n") + lines.append("_`SQS` below is public SQS v4: v3 deployment/WFV-first ranking plus a modest common-window capital-growth term when available. Stress OOT remains an eligibility gate. Prior v3 values remain in `experiment_registry.json` as `sqs_v3_score`._\n") + lines.append("| # | Experiment | SQS | [Tr]Ret% | [V]Ret% | [T]Ret% | [T]Ann% | [T]DD% | [T]Gross% | [T]DIM% | [T]R/G | Date |") + lines.append("|---|-----------|-----|----------|----------|----------|----------|--------|------------|----------|---------|------|") - for rank, e in enumerate(registry.entries, 1): - pf = f"{e.profit_factor:.2f}" if e.profit_factor is not None else "-" + for rank, e in enumerate(visible_entries, 1): + trret = f"{e.train_total_return_pct:+.1f}" if e.train_total_return_pct is not None else "-" + vret = f"{e.valid_total_return_pct:+.1f}" if e.valid_total_return_pct is not None else "-" ret = f"{e.total_return_pct:+.1f}" if e.total_return_pct is not None else "-" - wr = f"{e.win_rate:.0%}" if e.win_rate is not None else "-" - sharpe = f"{e.sharpe_ratio:.1f}" if e.sharpe_ratio is not None else "-" + ann = f"{e.annualized_return_pct:+.1f}" if e.annualized_return_pct is not None else "-" dd = f"{e.max_drawdown_pct:.1f}" if e.max_drawdown_pct is not None else "-" - tgross = f"{e.avg_gross_exposure_pct:.1f}" if e.avg_gross_exposure_pct is not None else "-" - tnet = f"{e.avg_net_exposure_pct:+.1f}" if e.avg_net_exposure_pct is not None else "-" - tdim = f"{e.days_in_market_pct:.1f}" if e.days_in_market_pct is not None else "-" - vpf = f"{e.valid_profit_factor:.2f}" if e.valid_profit_factor is not None else "-" - vret = f"{e.valid_total_return_pct:+.1f}" if e.valid_total_return_pct is not None else "-" - vwr = f"{e.valid_win_rate:.0%}" if e.valid_win_rate is not None else "-" - vsharpe = f"{e.valid_sharpe_ratio:.1f}" if e.valid_sharpe_ratio is not None else "-" - vdd = f"{e.valid_max_drawdown_pct:.1f}" if e.valid_max_drawdown_pct is not None else "-" - vgross = f"{e.valid_avg_gross_exposure_pct:.1f}" if e.valid_avg_gross_exposure_pct is not None else "-" - vnet = f"{e.valid_avg_net_exposure_pct:+.1f}" if e.valid_avg_net_exposure_pct is not None else "-" - vdim = f"{e.valid_days_in_market_pct:.1f}" if e.valid_days_in_market_pct is not None else "-" + gross = f"{e.avg_gross_exposure_pct:.1f}" if e.avg_gross_exposure_pct is not None else "-" + dim = f"{e.days_in_market_pct:.1f}" if e.days_in_market_pct is not None else "-" + ret_on_gross = "-" + if ( + e.total_return_pct is not None + and e.avg_gross_exposure_pct is not None + and e.avg_gross_exposure_pct > 0 + ): + ret_on_gross = f"{e.total_return_pct / e.avg_gross_exposure_pct:.2f}" ts = e.timestamp[:10] if e.timestamp else "-" lines.append( f"| {rank} | {e.experiment_name} | {e.sqs_score:.1f}" - f" | {pf} | {ret} | {wr} | {sharpe} | {dd} | {e.trade_count} | {tgross} | {tnet} | {tdim}" - f" | {vpf} | {vret} | {vwr} | {vsharpe} | {vdd} | {e.valid_trade_count} | {vgross} | {vnet} | {vdim}" - f" | {ts} |" + f" | {trret} | {vret} | {ret} | {ann} | {dd} | {gross} | {dim} | {ret_on_gross} | {ts} |" ) # Recent entries (last 5) - recent = list(reversed(journal_entries))[:5] - if recent: + if visible_recent: registry_by_id = {entry.entry_id: entry for entry in registry.entries} lines.append("\n## Recent Entries") - for je in recent: + for je in visible_recent: canonical_sqs = registry_by_id.get(je.entry_id).sqs_score if je.entry_id in registry_by_id else je.sqs_score + stress_sqs = ( + registry_by_id.get(je.entry_id).stress_sqs_score + if je.entry_id in registry_by_id + else je.stress_sqs_score + ) lines.append(f"### {je.entry_id} ({je.timestamp[:10]}) \u2014 {je.experiment_name}") lines.append(f"Hypothesis: {je.hypothesis}") + stress_suffix = f", Stress SQS {stress_sqs:.1f}" if stress_sqs is not None else "" + lines.append(f"Verdict: **{je.verdict.upper()}** (SQS {canonical_sqs}{stress_suffix})") + if je.verdict_reasoning: + lines.append(f"Reasoning: {je.verdict_reasoning}") + if je.next_direction: + lines.append(f"Next: {je.next_direction}") + lines.append("") + + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text("\n".join(lines) + "\n") + + +def _write_overlay_leaderboard_md( + path: Path, + registry: ExperimentRegistry, + journal_entries: list[JournalEntry], +) -> None: + visible_entries = filter_overlay_registry_entries(registry.entries, include_retired=False) + visible_recent = [ + entry + for entry in reversed(journal_entries) + if ( + not _is_retired_journal_entry(entry) + and _is_complete_journal_entry(entry) + and classify_strategy_family(entry.experiment_name, entry.tags) == "overlay" + ) + ][:5] + lines: list[str] = [] + lines.append("# Overlay Strategy Leaderboard") + lines.append(f"_Updated: {registry.updated_at}_\n") + lines.append("_This board is separate from the default single-book leaderboard. Overlay rows are book-of-books evaluations and are not directly comparable to single-book `SQS` rows._\n") + lines.append("_`SQS` here is overlay official SQS: common-window overlay score with a stress OOT gate. `T.*` columns are common-window overlay metrics._\n") + lines.append("| # | Overlay | SQS | [T]Ret% | [T]Ann% | [T]DD% | Stress Ret% | Stress DD% | Stress Sharpe | Date |") + lines.append("|---|---------|-----|----------|----------|--------|-------------|------------|---------------|------|") + + for rank, e in enumerate(visible_entries, 1): + stress = e.overlay_stress_window_summary + ts = e.timestamp[:10] if e.timestamp else "-" + lines.append( + f"| {rank} | {e.experiment_name} | {e.sqs_score:.1f}" + f" | {e.total_return_pct:+.1f} | {e.annualized_return_pct:+.1f} | {e.max_drawdown_pct:.1f}" + f" | {stress.return_pct:+.1f if stress and stress.return_pct is not None else '-'}" + f" | {stress.max_drawdown_pct:.1f if stress and stress.max_drawdown_pct is not None else '-'}" + f" | {stress.sharpe_ratio:+.2f if stress and stress.sharpe_ratio is not None else '-'}" + f" | {ts} |" + ) + + if visible_recent: + lines.append("\n## Recent Overlay Entries") + registry_by_id = {entry.entry_id: entry for entry in registry.entries} + for je in visible_recent: + canonical_sqs = registry_by_id.get(je.entry_id).sqs_score if je.entry_id in registry_by_id else je.sqs_score + lines.append(f"### {je.entry_id} ({je.timestamp[:10]}) — {je.experiment_name}") + lines.append(f"Hypothesis: {je.hypothesis}") lines.append(f"Verdict: **{je.verdict.upper()}** (SQS {canonical_sqs})") if je.verdict_reasoning: lines.append(f"Reasoning: {je.verdict_reasoning}") @@ -692,24 +2689,22 @@ def scan_runs_for_experiment( """ from libs.backtest.domain import MetricsBundle - matches: list[tuple[Path, str, str]] = [] # (run_dir, run_id, config_hash) + matches: list[tuple[Path, str, str, str, str | None]] = [] # (run_dir, run_id, config_hash, started_at, split) - for run_path in sorted(runs_dir.iterdir()): - if not run_path.is_dir(): - continue - manifest_file = run_path / "manifest.json" - if not manifest_file.exists(): - continue + for manifest_file in sorted(runs_dir.rglob("manifest.json")): + run_path = manifest_file.parent manifest_data = json.loads(manifest_file.read_text()) if manifest_data.get("experiment_name") != experiment_name: continue metadata_file = run_path / "metadata.json" metadata = json.loads(metadata_file.read_text()) if metadata_file.exists() else {} run_id = metadata.get("run_id", run_path.name) + started_at = str(metadata.get("started_at") or metadata.get("finished_at") or "") + split_name = metadata.get("split_name") # Extract config hash from run_id (last segment after underscore) parts = run_path.name.split("_") config_hash = parts[-1] if len(parts) >= 2 else "" - matches.append((run_path, run_id, config_hash)) + matches.append((run_path, run_id, config_hash, started_at, split_name)) if not matches: return {} @@ -718,18 +2713,29 @@ def scan_runs_for_experiment( # Check for split_name in metadata first has_split_names = False - for run_path, run_id, _ in matches: - metadata_file = run_path / "metadata.json" - if metadata_file.exists(): - metadata = json.loads(metadata_file.read_text()) - if "split_name" in metadata: - has_split_names = True - break + for _, _, _, _, split_name in matches: + if split_name is not None: + has_split_names = True + break + + canonical_splits = {"train", "valid", "test"} if has_split_names: - for run_path, run_id, _ in matches: - metadata = json.loads((run_path / "metadata.json").read_text()) - split = metadata.get("split_name", "unknown") + canonical_matches = [ + (run_path, run_id, config_hash, started_at, split_name) + for run_path, run_id, config_hash, started_at, split_name in matches + if split_name in canonical_splits + ] + if not canonical_matches: + return {} + + latest_by_split: dict[str, tuple[str, Path, str]] = {} + for run_path, run_id, _, started_at, split_name in canonical_matches: + split = split_name or "unknown" + current = latest_by_split.get(split) + if current is None or started_at >= current[0]: + latest_by_split[split] = (started_at, run_path, run_id) + for split, (_, run_path, run_id) in latest_by_split.items(): metrics_file = run_path / "metrics" / "metrics_summary.json" if metrics_file.exists(): metrics = MetricsBundle.model_validate_json(metrics_file.read_text()) @@ -737,14 +2743,14 @@ def scan_runs_for_experiment( else: # Fallback: group by config_hash, then assign train/valid/test by time order from collections import defaultdict - groups: dict[str, list[tuple[Path, str]]] = defaultdict(list) - for run_path, run_id, config_hash in matches: - groups[config_hash].append((run_path, run_id)) + groups: dict[str, list[tuple[str, Path, str]]] = defaultdict(list) + for run_path, run_id, config_hash, started_at, _ in matches: + groups[config_hash].append((started_at, run_path, run_id)) # Use the largest group (most likely the 3-split set) biggest_group = max(groups.values(), key=len) if groups else [] split_names = ["train", "valid", "test"] - for i, (run_path, run_id) in enumerate(biggest_group): + for i, (_, run_path, run_id) in enumerate(sorted(biggest_group)): split = split_names[i] if i < len(split_names) else f"extra_{i}" metrics_file = run_path / "metrics" / "metrics_summary.json" if metrics_file.exists():