yirugi
/
fithia2
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Remove overlay backtesting and scoring

Browse Source
main
I Luk Kim 5 months ago
parent 1d9507540b
commit 76581ead04
27 changed files with 66 additions and 4160 deletions
Show Stats Download Patch File Download Diff File

318
apps/paper_trader/backtest_sim.py
Unescape Escape View File

@ -351,310 +351,6 @@ async def _refresh_snapshot(
raise raise
# ── Overlay backtest support ──────────────────────────────────────────
def _is_overlay_config(config_path: str) -> bool:
"""Return True if config_path is an overlay spec (has 'books' key)."""
import json
try:
data = json.loads(Path(config_path).read_text())
return "books" in data and "allocations" in data
except Exception:
return False
def _resolve_book_experiment_config(book: dict) -> str | None:
"""Resolve the experiment config path for an overlay book entry."""
# Explicit field
explicit = book.get("experiment_config")
if explicit and Path(explicit).exists():
return explicit
# Infer from equity_csv filename
csv_path = book.get("equity_csv", "")
if csv_path:
name = Path(csv_path).stem # e.g. "return_max_long_v6.221_equity"
# Strip common suffixes
for suffix in ("_equity", "_train", "_valid", "_test"):
if name.endswith(suffix):
name = name[: -len(suffix)]
break
candidate = f"configs/experiments/{name}.json"
if Path(candidate).exists():
return candidate
return None
def _overlay_books_are_runnable(overlay_config_path: str) -> bool:
"""Check if all books in an overlay config have resolvable experiment configs."""
import json
try:
spec = json.loads(Path(overlay_config_path).read_text())
for book in spec.get("books", []):
csv_path = book.get("equity_csv")
if csv_path and Path(csv_path).exists():
continue
if _resolve_book_experiment_config(book) is None:
return False
return True
except Exception:
return False
def _rebase_equity_slice(
df,
*,
initial_equity: float,
):
"""Recompute equity within a requested window so the first kept day starts flat."""
df = df.sort_values("date").copy()
df["daily_return"] = df["equity"].astype(float).pct_change().fillna(0.0)
equity = float(initial_equity)
rebased: list[float] = []
for ret in df["daily_return"].astype(float):
equity *= 1.0 + float(ret)
rebased.append(equity)
df["equity"] = rebased
return df[["date", "equity", "daily_return"]]
def _summarize_book_curve(df, *, initial_equity: float) -> dict[str, float]:
"""Return a paper-backtest-like summary from a rebased equity curve."""
returns = df["daily_return"].astype(float)
final_equity = float(df["equity"].iloc[-1])
return_pct = (final_equity / float(initial_equity) - 1.0) * 100.0
peak = float(initial_equity)
max_dd_pct = 0.0
for equity in df["equity"].astype(float):
peak = max(peak, float(equity))
drawdown_pct = (peak - float(equity)) / peak * 100.0 if peak > 0 else 0.0
max_dd_pct = max(max_dd_pct, drawdown_pct)
if len(returns) >= 2 and float(returns.std()) > 0:
sharpe = float(returns.mean() / returns.std() * math.sqrt(252.0))
else:
sharpe = 0.0
return {
"return_pct": return_pct,
"final_equity": final_equity,
"max_dd_pct": max_dd_pct,
"trade_count": 0,
"win_rate": 0.0,
"sharpe": sharpe,
}
def _load_overlay_book_curve_from_spec(
book: dict,
*,
capital: float,
start_date: dt.date,
end_date: dt.date,
):
"""Load a frozen overlay input curve from equity_csv and rebase it to the requested window."""
from libs.backtest.overlay import load_equity_curve_csv
csv_path = book.get("equity_csv")
if not csv_path or not Path(csv_path).exists():
return None
df = load_equity_curve_csv(csv_path)
df = df[(df["date"] >= start_date) & (df["date"] <= end_date)].copy()
if df.empty:
return None
rebased = _rebase_equity_slice(df, initial_equity=capital)
summary = _summarize_book_curve(rebased, initial_equity=capital)
return {
"curve": rebased,
"summary": summary,
"source": "equity_csv",
}
def run_overlay_backtest_sync(
overlay_config_path: str,
capital: float,
start_date: dt.date,
end_date: dt.date,
console=None,
) -> dict[str, Any]:
"""Run an overlay backtest: execute each book strategy, then combine by regime."""
import json
import pandas as pd
from libs.backtest.overlay import build_overlay_curve, summarize_overlay_curve
spec = json.loads(Path(overlay_config_path).read_text())
overlay_name = spec.get("overlay_name", Path(overlay_config_path).stem)
allocations = spec["allocations"]
# ── Run each book strategy ────────────────────────────────────────
book_results: list[dict[str, Any]] = []
curves: dict[str, pd.DataFrame] = {}
replay_mode = "frozen_equity_csv"
for book in spec["books"]:
label = book["label"]
loaded = _load_overlay_book_curve_from_spec(
book,
capital=capital,
start_date=start_date,
end_date=end_date,
)
if loaded is not None:
if console:
source_name = Path(book["equity_csv"]).stem
console.print(f" [dim]Book '{label}':[/] {source_name} [dim](frozen equity_csv)[/]")
curves[label] = loaded["curve"]
book_results.append(
{
"label": label,
"result": {
"session_name": f"{overlay_name}__{label}",
"summary": loaded["summary"],
"equity_curve": [
{"date": row.date, "equity": row.equity}
for row in loaded["curve"].itertuples(index=False)
],
"trades": [],
},
"source": loaded["source"],
}
)
continue
replay_mode = "rerun_books"
exp_config = _resolve_book_experiment_config(book)
if exp_config is None:
raise ValueError(
f"Overlay '{overlay_name}': book '{label}' has neither a usable equity_csv nor a resolvable experiment config. "
f"Add 'equity_csv' or 'experiment_config' to the book entry."
)
if console:
console.print(f" [dim]Book '{label}':[/] {Path(exp_config).stem} [dim](rerun)[/]")
result = run_backtest_session_sync(
session_name=f"{overlay_name}__{label}",
config_path=exp_config,
initial_equity=capital,
start_date=start_date,
end_date=end_date,
)
book_results.append({"label": label, "result": result, "source": "rerun"})
eq = result.get("equity_curve", [])
if eq:
df = pd.DataFrame(eq)
df["date"] = pd.to_datetime(df["date"]).dt.date
df["equity"] = df["equity"].astype(float)
curves[label] = _rebase_equity_slice(df[["date", "equity"]], initial_equity=capital)
if not curves:
raise ValueError(f"Overlay '{overlay_name}': no book produced equity curves")
# ── Compute regime for each trading day ───────────────────────────
regimes = _compute_overlay_regimes(spec, start_date, end_date)
# ── Combine using overlay logic ───────────────────────────────────
overlay_curve = build_overlay_curve(
curves=curves,
allocations=allocations,
regimes_by_date=regimes,
initial_equity=capital,
)
summary = summarize_overlay_curve(overlay_curve, initial_equity=capital)
# Convert overlay equity curve to standard format
equity_curve = [
{"date": row.date, "equity": row.overlay_equity}
for row in overlay_curve.itertuples(index=False)
]
# Aggregate trade count across books
total_trades = sum(
br["result"]["summary"]["trade_count"] for br in book_results
)
return {
"session_name": overlay_name,
"config_path": overlay_config_path,
"initial_equity": capital,
"is_overlay": True,
"overlay_replay_mode": replay_mode,
"equity_curve": equity_curve,
"trades": [],
"book_results": book_results,
"allocations": allocations,
"regime_day_counts": summary.get("regime_day_counts", {}),
"summary": {
"return_pct": summary["return_pct"],
"final_equity": summary["final_equity"],
"max_dd_pct": summary["max_dd_pct"],
"trade_count": total_trades,
"win_rate": 0.0,
"sharpe": summary["sharpe"],
},
}
def _compute_overlay_regimes(
spec: dict,
start_date: dt.date,
end_date: dt.date,
) -> dict[dt.date, str]:
"""Compute macro regime for each trading day using the regime_source config.
Uses _build_merged_snapshot_store to get a full-period store with macro data,
covering the paper backtest date range (not just the original snapshot period).
"""
from apps.backtester.run import _build_merged_snapshot_store, load_manifest, resolve_config
from libs.backtest.allocator import _macro_regime_state
from libs.backtest.overlay import load_merged_store_from_snapshot_dir
from libs.common.config import get_settings
regime_source = spec.get("regime_source", {})
config_path = regime_source.get("config_path")
if not config_path:
return {}
manifest = load_manifest(config_path)
config = resolve_config(manifest)
raw_snapshot_dir = regime_source.get("snapshot_dir")
if raw_snapshot_dir and (
(Path(raw_snapshot_dir) / "train.parquet").exists()
or (Path(raw_snapshot_dir) / "test.parquet").exists()
):
settings = get_settings()
store = load_merged_store_from_snapshot_dir(
raw_snapshot_dir,
oracle_url=settings.stock_oracle_url,
db_dsn=settings.postgres_dsn,
)
else:
try:
store = _build_merged_snapshot_store(
manifest,
config,
snapshot_dir_override=raw_snapshot_dir,
)
except FileNotFoundError:
store = _build_merged_snapshot_store(manifest, config, snapshot_dir_override=None)
store = store.slice_by_date_range(start_date, end_date)
regimes: dict[dt.date, str] = {}
for date in store.all_trading_days():
regimes[date] = _macro_regime_state(config, store.get_macro_for_date(date))
return regimes
def run_backtest( def run_backtest(
configs: list[str], configs: list[str],
capital: float, capital: float,
@ -691,8 +387,6 @@ def run_backtest(
# Check if snapshots need refresh (async pipeline, run before sync backtest) # Check if snapshots need refresh (async pipeline, run before sync backtest)
for config_path in configs: for config_path in configs:
if _is_overlay_config(config_path):
continue # overlay books handle their own snapshots
from apps.backtester.run import load_manifest, resolve_config from apps.backtester.run import load_manifest, resolve_config
manifest = load_manifest(config_path) manifest = load_manifest(config_path)
config = resolve_config(manifest) config = resolve_config(manifest)
@ -723,18 +417,6 @@ def run_backtest(
results = [] results = []
for config_path in configs: for config_path in configs:
session_name = Path(config_path).stem session_name = Path(config_path).stem
if _is_overlay_config(config_path):
if console:
console.print(f"\n[bold magenta]Running overlay:[/] {session_name}")
result = run_overlay_backtest_sync(
overlay_config_path=config_path,
capital=capital,
start_date=start_date,
end_date=end_date,
console=console,
)
else:
if console: if console:
console.print(f"\n[bold cyan]Running:[/] {session_name}") console.print(f"\n[bold cyan]Running:[/] {session_name}")
result = run_backtest_session_sync( result = run_backtest_session_sync(

21
apps/paper_trader/cli.py
Unescape Escape View File

@ -311,7 +311,6 @@ def _resolve_rank_configs(start: int, end: int) -> list[str]:
"""Load strategies ranked start..end from leaderboard by SQS score. """Load strategies ranked start..end from leaderboard by SQS score.
start/end are 1-based inclusive. e.g. (1, 5) = top 5, (20, 40) = rank 20-40. start/end are 1-based inclusive. e.g. (1, 5) = top 5, (20, 40) = rank 20-40.
Overlays are excluded; use --overlay to run them explicitly.
""" """
import json import json
@ -322,7 +321,6 @@ def _resolve_rank_configs(start: int, end: int) -> list[str]:
registry = json.loads(registry_path.read_text()) registry = json.loads(registry_path.read_text())
ranked: list[str] = [] ranked: list[str] = []
skipped_overlays: list[str] = []
entries = sorted( entries = sorted(
( (
e for e in registry.get("entries", []) e for e in registry.get("entries", [])
@ -334,10 +332,6 @@ def _resolve_rank_configs(start: int, end: int) -> list[str]:
) )
for e in entries: for e in entries:
is_overlay = e.get("strategy_family") == "overlay" or e.get("overlay_common_window_summary") is not None
if is_overlay:
skipped_overlays.append(e["experiment_name"])
continue
if e.get("trade_count", 0) <= 0 or e.get("valid_trade_count", 0) <= 0: if e.get("trade_count", 0) <= 0 or e.get("valid_trade_count", 0) <= 0:
continue continue
name = e["experiment_name"] name = e["experiment_name"]
@ -345,15 +339,6 @@ def _resolve_rank_configs(start: int, end: int) -> list[str]:
if Path(cfg_path).exists(): if Path(cfg_path).exists():
ranked.append(cfg_path) ranked.append(cfg_path)
if skipped_overlays:
labels = ", ".join(skipped_overlays[:5])
if len(skipped_overlays) > 5:
labels += ", ..."
_console.print(
"[yellow]Skipping overlay leaderboard entries for `--top/--rank` "
f"(use `--overlay` to run them explicitly): {labels}[/]"
)
# 1-based inclusive slice # 1-based inclusive slice
return ranked[start - 1 : end] return ranked[start - 1 : end]
@ -363,8 +348,6 @@ def cmd_backtest(args: argparse.Namespace) -> None:
import datetime as dt import datetime as dt
configs = args.configs or [] configs = args.configs or []
if args.overlays:
configs.extend(args.overlays)
if args.top: if args.top:
configs = _resolve_rank_configs(1, args.top) + configs configs = _resolve_rank_configs(1, args.top) + configs
if args.rank: if args.rank:
@ -378,7 +361,7 @@ def cmd_backtest(args: argparse.Namespace) -> None:
_console.print("[red]ERROR: --rank format: N or START-END (e.g. 5 or 20-40)[/]") _console.print("[red]ERROR: --rank format: N or START-END (e.g. 5 or 20-40)[/]")
sys.exit(1) sys.exit(1)
if not configs: if not configs:
_console.print("[red]ERROR: Specify --config, --overlay, --top, or --rank[/]") _console.print("[red]ERROR: Specify --config, --top, or --rank[/]")
sys.exit(1) sys.exit(1)
for cfg in configs: for cfg in configs:
@ -590,8 +573,6 @@ def main() -> None:
p.add_argument("--config", "-c", action="append", p.add_argument("--config", "-c", action="append",
dest="configs", metavar="PATH", dest="configs", metavar="PATH",
help="Config path (repeat for multiple strategies)") help="Config path (repeat for multiple strategies)")
p.add_argument("--overlay", action="append", dest="overlays", metavar="PATH",
help="Overlay config path (repeat for multiple)")
p.add_argument("--top", "-t", type=int, default=None, metavar="N", p.add_argument("--top", "-t", type=int, default=None, metavar="N",
help="Use top N strategies from leaderboard (by SQS score)") help="Use top N strategies from leaderboard (by SQS score)")
p.add_argument("--rank", default=None, metavar="START-END", p.add_argument("--rank", default=None, metavar="START-END",

71
apps/paper_trader/reporter.py
Unescape Escape View File

@ -309,54 +309,6 @@ def print_sessions(sessions: list[SessionRow]) -> None:
# Run summary # Run summary
# ------------------------------------------------------------------ # # ------------------------------------------------------------------ #
def _print_overlay_detail(r: dict) -> None:
"""Print overlay regime allocation + per-book summary."""
name = r["session_name"]
# Regime day counts
regime_counts = r.get("regime_day_counts", {})
if regime_counts:
regime_tbl = Table(
box=box.SIMPLE_HEAD, show_header=True, header_style="bold yellow",
padding=(0, 1), title=f"[bold magenta]Regime Days — {name}[/]", title_justify="left",
)
regime_tbl.add_column("Regime", style="bold")
regime_tbl.add_column("Days", justify="right")
regime_tbl.add_column("Allocation", no_wrap=True)
allocations = r.get("allocations", {})
for regime, count in sorted(regime_counts.items()):
alloc = allocations.get(regime, {})
alloc_str = " ".join(f"{k}={v:.0%}" for k, v in alloc.items())
regime_tbl.add_row(regime, str(count), alloc_str)
_console.print(regime_tbl)
# Per-book summary
book_results = r.get("book_results", [])
if book_results:
book_tbl = Table(
box=box.SIMPLE_HEAD, show_header=True, header_style="bold yellow",
padding=(0, 1), title=f"[bold magenta]Books — {name}[/]", title_justify="left",
)
book_tbl.add_column("Book", style="bold")
book_tbl.add_column("Return", justify="right")
book_tbl.add_column("MaxDD", justify="right")
book_tbl.add_column("Trades", justify="right")
book_tbl.add_column("WinRate", justify="right")
book_tbl.add_column("Sharpe", justify="right")
for br in book_results:
bs = br["result"]["summary"]
ret_color = "green" if bs["return_pct"] >= 0 else "red"
book_tbl.add_row(
br["label"],
f"[{ret_color}]{bs['return_pct']:+.2f}%[/{ret_color}]",
f"[red]-{bs['max_dd_pct']:.2f}%[/]",
str(bs["trade_count"]),
f"{bs['win_rate']:.0f}%",
f"{bs['sharpe']:+.2f}",
)
_console.print(book_tbl)
def print_backtest_results(results: list[dict], output_dir: str | None = None, show_trades: bool = True) -> None: def print_backtest_results(results: list[dict], output_dir: str | None = None, show_trades: bool = True) -> None:
"""Print equity curve comparison, summary table, and per-strategy trade logs.""" """Print equity curve comparison, summary table, and per-strategy trade logs."""
import csv import csv
@ -386,8 +338,6 @@ def print_backtest_results(results: list[dict], output_dir: str | None = None, s
s = r["summary"] s = r["summary"]
ret_color = "green" if s["return_pct"] >= 0 else "red" ret_color = "green" if s["return_pct"] >= 0 else "red"
name = r["session_name"] name = r["session_name"]
if r.get("is_overlay"):
name = f"{name} [overlay]"
sum_tbl.add_row( sum_tbl.add_row(
name, name,
f"[{ret_color}]{s['return_pct']:+.2f}%[/{ret_color}]", f"[{ret_color}]{s['return_pct']:+.2f}%[/{ret_color}]",
@ -399,12 +349,6 @@ def print_backtest_results(results: list[dict], output_dir: str | None = None, s
_console.print(sum_tbl) _console.print(sum_tbl)
# ── Overlay detail sections ───────────────────────────────────────────
for r in results:
if not r.get("is_overlay"):
continue
_print_overlay_detail(r)
# ── Per-strategy trade logs ──────────────────────────────────────────── # ── Per-strategy trade logs ────────────────────────────────────────────
if not show_trades: if not show_trades:
if output_dir: if output_dir:
@ -520,8 +464,23 @@ def print_run_summary(summary: dict) -> None:
_console.print(f"[dim]{date}: {status}[/]") _console.print(f"[dim]{date}: {status}[/]")
return return
if status == "kill_switch_active":
_console.print(f"[bold red]{date}: KILL SWITCH ACTIVE — all trading halted[/]")
return
_console.print(f"\nProcessing [bold]{date}[/]...") _console.print(f"\nProcessing [bold]{date}[/]...")
# Reconciliation report
recon = summary.get("reconciliation")
if recon and recon.has_issues:
_console.print(" [bold yellow]RECONCILIATION:[/]")
for sym in recon.orphaned_alpaca:
_console.print(f" [yellow]ORPHANED[/] {sym} — on Alpaca but no local state")
for sym in recon.ghost_local:
_console.print(f" [yellow]GHOST[/] {sym} — local state but no Alpaca position (auto-closed)")
if recon.stale_orders_cancelled:
_console.print(f" [dim]Stale orders cancelled: {len(recon.stale_orders_cancelled)}[/]")
exits = summary.get("exits", []) exits = summary.get("exits", [])
entries = summary.get("entries", []) entries = summary.get("entries", [])
rejected = summary.get("rejected", []) rejected = summary.get("rejected", [])

125
apps/tools/evaluate_book_overlay.py
Unescape Escape View File

@ -1,125 +0,0 @@
#!/usr/bin/env python3
from __future__ import annotations
import argparse
import datetime as dt
import json
from pathlib import Path
from typing import Any
from apps.backtester.run import _build_merged_snapshot_store
from libs.backtest.allocator import _macro_regime_state
from libs.backtest.manifests import load_manifest, resolve_config
from libs.backtest.overlay import (
build_overlay_curve,
load_equity_curve_csv,
load_merged_store_from_snapshot_dir,
summarize_overlay_curve,
)
from libs.common.config import get_settings
def _parse_date(value: str | None) -> dt.date | None:
if not value:
return None
return dt.date.fromisoformat(value)
def _compute_regimes(
*,
snapshot_dir: str | Path,
split: str,
config_path: str | Path,
start_date: dt.date | None,
end_date: dt.date | None,
) -> dict[dt.date, str]:
del split # overlay regimes should cover the full requested window, not a single split
manifest = load_manifest(config_path)
config = resolve_config(manifest, config_root=".")
raw_snapshot_dir = Path(snapshot_dir)
if (raw_snapshot_dir / "train.parquet").exists() or (raw_snapshot_dir / "test.parquet").exists():
settings = get_settings()
store = load_merged_store_from_snapshot_dir(
raw_snapshot_dir,
oracle_url=settings.stock_oracle_url,
db_dsn=settings.postgres_dsn,
)
else:
try:
store = _build_merged_snapshot_store(
manifest,
config,
snapshot_dir_override=str(raw_snapshot_dir),
)
except FileNotFoundError:
store = _build_merged_snapshot_store(
manifest,
config,
snapshot_dir_override=None,
)
if start_date or end_date:
lower = start_date or dt.date.min
upper = end_date or dt.date.max
store = store.slice_by_date_range(lower, upper)
regimes: dict[dt.date, str] = {}
for date in store.all_trading_days():
regimes[date] = _macro_regime_state(config, store.get_macro_for_date(date))
return regimes
def _load_spec(path: str | Path) -> dict[str, Any]:
return json.loads(Path(path).read_text())
def main() -> None:
parser = argparse.ArgumentParser(description="Evaluate a regime-switched overlay from book equity curves")
parser.add_argument("--spec", required=True, help="Path to overlay spec JSON")
parser.add_argument("--output-dir", required=True, help="Directory to write overlay outputs")
args = parser.parse_args()
spec = _load_spec(args.spec)
initial_equity = float(spec.get("initial_equity", 10_000.0))
curves = {
book["label"]: load_equity_curve_csv(book["equity_csv"])
for book in spec["books"]
}
regime_source = spec["regime_source"]
start_date = _parse_date(spec.get("start_date"))
end_date = _parse_date(spec.get("end_date"))
regimes = _compute_regimes(
snapshot_dir=regime_source["snapshot_dir"],
split=regime_source.get("split", "train"),
config_path=regime_source["config_path"],
start_date=start_date,
end_date=end_date,
)
curve = build_overlay_curve(
curves=curves,
allocations=spec["allocations"],
regimes_by_date=regimes,
initial_equity=initial_equity,
)
if start_date:
curve = curve[curve["date"] >= start_date]
if end_date:
curve = curve[curve["date"] <= end_date]
summary = summarize_overlay_curve(curve, initial_equity=initial_equity)
summary["overlay_name"] = spec.get("overlay_name", Path(args.spec).stem)
summary["books"] = [book["label"] for book in spec["books"]]
summary["allocations"] = spec["allocations"]
out_dir = Path(args.output_dir)
out_dir.mkdir(parents=True, exist_ok=True)
curve.assign(date=curve["date"].astype(str)).to_csv(out_dir / "overlay_equity.csv", index=False)
(out_dir / "overlay_summary.json").write_text(json.dumps(summary, indent=2) + "\n")
print(json.dumps(summary, indent=2))
if __name__ == "__main__":
main()

168
apps/tracker/cli.py
Unescape Escape View File

@ -2,8 +2,6 @@
from __future__ import annotations from __future__ import annotations
import argparse import argparse
import datetime as dt
import json
import sys import sys
from pathlib import Path from pathlib import Path
@ -15,7 +13,6 @@ from rich.table import Table
from libs.backtest.domain import ( from libs.backtest.domain import (
ConfigDelta, ConfigDelta,
JournalEntry, JournalEntry,
OverlayWindowSummary,
RobustnessMatrixSummary, RobustnessMatrixSummary,
SplitResult, SplitResult,
WalkForwardSummary, WalkForwardSummary,
@ -33,8 +30,6 @@ from libs.backtest.tracker import (
compute_public_sqs_v2, compute_public_sqs_v2,
compute_promotion_score, compute_promotion_score,
compute_oot_robustness_gate, compute_oot_robustness_gate,
compute_overlay_public_sqs,
compute_overlay_stress_sqs,
compute_robustness_gate, compute_robustness_gate,
compute_rqs, compute_rqs,
compute_sqs, compute_sqs,
@ -42,7 +37,6 @@ from libs.backtest.tracker import (
compute_unified_score, compute_unified_score,
compute_wfqs, compute_wfqs,
compute_wfqs_v2, compute_wfqs_v2,
filter_overlay_registry_entries,
filter_registry_entries, filter_registry_entries,
get_next_entry_id, get_next_entry_id,
journal_lock, journal_lock,
@ -117,43 +111,6 @@ def _load_optional_robustness_summary(path_str: str | None) -> RobustnessMatrixS
return RobustnessMatrixSummary.model_validate_json(path.read_text()) 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: def _score_sort_key(sort_by: str, entry) -> tuple:
if sort_by == "promotion": if sort_by == "promotion":
return ( return (
@ -391,86 +348,12 @@ def cmd_record(args: argparse.Namespace) -> None:
print(f"Leaderboard updated: {leaderboard_path}") print(f"Leaderboard updated: {leaderboard_path}")
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"
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 [])]
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: def cmd_leaderboard(args: argparse.Namespace) -> None:
"""Show or regenerate the leaderboard.""" """Show or regenerate the leaderboard."""
journal_dir = Path(args.journal_dir) journal_dir = Path(args.journal_dir)
journal_path = journal_dir / "improvement_journal.jsonl" journal_path = journal_dir / "improvement_journal.jsonl"
registry_path = journal_dir / "experiment_registry.json" registry_path = journal_dir / "experiment_registry.json"
leaderboard_path = journal_dir / "LEADERBOARD.md" leaderboard_path = journal_dir / "LEADERBOARD.md"
overlay_leaderboard_path = journal_dir / "OVERLAY_LEADERBOARD.md"
if not journal_path.exists(): if not journal_path.exists():
journal_path.parent.mkdir(parents=True, exist_ok=True) journal_path.parent.mkdir(parents=True, exist_ok=True)
@ -478,18 +361,9 @@ def cmd_leaderboard(args: argparse.Namespace) -> None:
with journal_lock(journal_path): with journal_lock(journal_path):
registry = _sync_and_rebuild(journal_path, registry_path, leaderboard_path) registry = _sync_and_rebuild(journal_path, registry_path, leaderboard_path)
overlay_only = getattr(args, "overlay_only", False)
if overlay_only:
ranked_source = filter_overlay_registry_entries(
registry.entries,
include_retired=getattr(args, "include_retired", False),
)
displayed_leaderboard_path = overlay_leaderboard_path
else:
ranked_source = filter_registry_entries( ranked_source = filter_registry_entries(
registry.entries, registry.entries,
include_retired=getattr(args, "include_retired", False), include_retired=getattr(args, "include_retired", False),
include_overlays=False,
) )
displayed_leaderboard_path = leaderboard_path displayed_leaderboard_path = leaderboard_path
@ -497,12 +371,8 @@ def cmd_leaderboard(args: argparse.Namespace) -> None:
ranked_entries = sorted(ranked_source, key=lambda entry: _score_sort_key(sort_by, entry)) ranked_entries = sorted(ranked_source, key=lambda entry: _score_sort_key(sort_by, entry))
total = len(ranked_entries) total = len(ranked_entries)
top_n = getattr(args, "top", 10) top_n = getattr(args, "top", 10)
diagnostics = (not overlay_only) and sort_by in {"deployment", "dep", "wfqs", "rqs", "promotion", "unified", "sqs2"} diagnostics = sort_by in {"deployment", "dep", "wfqs", "rqs", "promotion", "unified", "sqs2"}
title = ( title = f"[bold cyan]Top {top_n} / {total}[/] [dim]· {_title_for_sort(sort_by)} · Tr=train V=valid T=test[/]"
f"[bold cyan]Top {top_n} / {total}[/] [dim]· overlay official SQS · Common=full-window Stress=OOT[/]"
if overlay_only
else f"[bold cyan]Top {top_n} / {total}[/] [dim]· {_title_for_sort(sort_by)} · Tr=train V=valid T=test[/]"
)
tbl = Table( tbl = Table(
box=box.SIMPLE_HEAD, box=box.SIMPLE_HEAD,
@ -537,7 +407,6 @@ def cmd_leaderboard(args: argparse.Namespace) -> None:
name = entry.experiment_name name = entry.experiment_name
if len(name) > _COL_NAME: if len(name) > _COL_NAME:
name = name[: _COL_NAME - 1] + "…" name = name[: _COL_NAME - 1] + "…"
is_overlay = entry.strategy_family == "overlay" and entry.overlay_common_window_summary is not None
row = [ row = [
str(rank), str(rank),
name, name,
@ -553,14 +422,14 @@ def cmd_leaderboard(args: argparse.Namespace) -> None:
_fmt(entry.rqs_score, ".1f"), _fmt(entry.rqs_score, ".1f"),
]) ])
row.extend([ row.extend([
"-" if is_overlay else _fmt(entry.train_total_return_pct, "+.1f"), _fmt(entry.train_total_return_pct, "+.1f"),
"-" if is_overlay else _fmt(entry.valid_total_return_pct, "+.1f"), _fmt(entry.valid_total_return_pct, "+.1f"),
_fmt(entry.total_return_pct, "+.1f"), _fmt(entry.total_return_pct, "+.1f"),
_fmt(entry.annualized_return_pct, "+.1f"), _fmt(entry.annualized_return_pct, "+.1f"),
_fmt(entry.max_drawdown_pct, ".1f"), _fmt(entry.max_drawdown_pct, ".1f"),
"-" if is_overlay else _fmt(entry.avg_gross_exposure_pct, ".1f"), _fmt(entry.avg_gross_exposure_pct, ".1f"),
"-" if is_overlay else _fmt(entry.days_in_market_pct, ".1f"), _fmt(entry.days_in_market_pct, ".1f"),
"-" if is_overlay else _fmt(_ratio(entry.total_return_pct, entry.avg_gross_exposure_pct), ".2f"), _fmt(_ratio(entry.total_return_pct, entry.avg_gross_exposure_pct), ".2f"),
]) ])
tbl.add_row(*row) tbl.add_row(*row)
@ -1034,13 +903,6 @@ def main() -> None:
action="store_true", action="store_true",
help="Include retired legacy PEAD / short-core / exact-pocket families", help="Include retired legacy PEAD / short-core / exact-pocket families",
) )
subparser.add_argument(
"--overlay",
"--overlay-only",
dest="overlay_only",
action="store_true",
help="Show the separate overlay/book-of-books leaderboard instead of the default single-book leaderboard",
)
for name in ("show", "s"): for name in ("show", "s"):
subparser = sub.add_parser(name, help="Show details of a journal entry") subparser = sub.add_parser(name, help="Show details of a journal entry")
@ -1071,20 +933,6 @@ def main() -> None:
subparser.add_argument("target", help="Entry ID or experiment name to update") 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") 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"): for name in ("rescore-public", "rsp"):
subparser = sub.add_parser(name, help="Recompute stored public scores for journal entries") subparser = sub.add_parser(name, help="Recompute stored public scores for journal entries")
subparser.add_argument("--journal-dir", **journal_kwargs) subparser.add_argument("--journal-dir", **journal_kwargs)
@ -1108,8 +956,6 @@ def main() -> None:
"arb": cmd_attach_robustness, "arb": cmd_attach_robustness,
"attach-oot-robustness": cmd_attach_oot_robustness, "attach-oot-robustness": cmd_attach_oot_robustness,
"aoot": cmd_attach_oot_robustness, "aoot": cmd_attach_oot_robustness,
"record-overlay": cmd_record_overlay,
"rovl": cmd_record_overlay,
"rescore-public": cmd_rescore_public, "rescore-public": cmd_rescore_public,
"rsp": cmd_rescore_public, "rsp": cmd_rescore_public,
} }

39
configs/overlays/return_book_overlay_v1.json
Unescape Escape View File

@ -1,39 +0,0 @@
{
"overlay_name": "return_book_overlay_v1",
"initial_equity": 10000,
"start_date": "2022-03-03",
"end_date": "2026-03-13",
"regime_source": {
"config_path": "configs/experiments/return_max_long_v6.221.json",
"snapshot_dir": "data/parquet/midlarge-liquid-long-v1_bucketfix_full_audit_mom_v2",
"split": "train"
},
"books": [
{
"label": "core",
"equity_csv": "runs/book_overlay_v1_inputs/return_max_long_v6.221_equity.csv"
},
{
"label": "mom",
"equity_csv": "runs/book_overlay_v1_inputs/return_max_long_v6new.54_equity.csv"
}
],
"allocations": {
"risk_on": {
"core": 0.0,
"mom": 1.0
},
"neutral": {
"core": 0.0,
"mom": 1.0
},
"risk_off": {
"core": 1.0,
"mom": 0.0
},
"unknown": {
"core": 1.0,
"mom": 0.0
}
}
}

39
configs/overlays/return_book_overlay_v1b.json
Unescape Escape View File

@ -1,39 +0,0 @@
{
"overlay_name": "return_book_overlay_v1b",
"initial_equity": 10000,
"start_date": "2022-03-03",
"end_date": "2026-03-13",
"regime_source": {
"config_path": "configs/experiments/return_max_long_v6.221.json",
"snapshot_dir": "data/parquet/midlarge-liquid-long-v1_bucketfix_full_audit_mom_v2",
"split": "train"
},
"books": [
{
"label": "core",
"equity_csv": "runs/book_overlay_v1_inputs/return_max_long_v6.221_equity.csv"
},
{
"label": "mom",
"equity_csv": "runs/book_overlay_v1_inputs/return_max_long_v6new.54_equity.csv"
}
],
"allocations": {
"risk_on": {
"core": 0.0,
"mom": 1.0
},
"neutral": {
"core": 0.0,
"mom": 1.0
},
"risk_off": {
"core": 1.0,
"mom": 0.0
},
"unknown": {
"core": 0.0,
"mom": 1.0
}
}
}

39
configs/overlays/return_book_overlay_v1c.json
Unescape Escape View File

@ -1,39 +0,0 @@
{
"overlay_name": "return_book_overlay_v1c",
"initial_equity": 10000,
"start_date": "2022-03-03",
"end_date": "2026-03-13",
"regime_source": {
"config_path": "configs/experiments/return_max_long_v6.221.json",
"snapshot_dir": "data/parquet/midlarge-liquid-long-v1_bucketfix_full_audit_mom_v2",
"split": "train"
},
"books": [
{
"label": "core",
"equity_csv": "runs/book_overlay_v1_inputs/return_max_long_v6.221_equity.csv"
},
{
"label": "mom",
"equity_csv": "runs/book_overlay_v1_inputs/return_max_long_v6new.54_equity.csv"
}
],
"allocations": {
"risk_on": {
"core": 0.0,
"mom": 1.0
},
"neutral": {
"core": 0.0,
"mom": 1.0
},
"risk_off": {
"core": 1.0,
"mom": 0.0
},
"unknown": {
"core": 0.5,
"mom": 0.5
}
}
}

39
configs/overlays/return_book_overlay_v2.json
Unescape Escape View File

@ -1,39 +0,0 @@
{
"initial_equity": 10000,
"start_date": "2022-03-03",
"end_date": "2026-03-13",
"regime_source": {
"config_path": "configs/experiments/return_max_long_v6.114.json",
"snapshot_dir": "data/parquet/midlarge-liquid-long-v1_bucketfix_full_audit_mom_v2",
"split": "train"
},
"books": [
{
"label": "core",
"equity_csv": "runs/fallback_book_compare/return_max_long_v6.114_equity.csv"
},
{
"label": "mom",
"equity_csv": "runs/book_overlay_v1_inputs/return_max_long_v6new.54_equity.csv"
}
],
"overlay_name": "return_book_overlay_v2",
"allocations": {
"risk_on": {
"core": 0,
"mom": 1
},
"neutral": {
"core": 0,
"mom": 1
},
"risk_off": {
"core": 1,
"mom": 0
},
"unknown": {
"core": 1,
"mom": 0
}
}
}

39
configs/overlays/return_book_overlay_v2b.json
Unescape Escape View File

@ -1,39 +0,0 @@
{
"initial_equity": 10000,
"start_date": "2022-03-03",
"end_date": "2026-03-13",
"regime_source": {
"config_path": "configs/experiments/return_max_long_v6.114.json",
"snapshot_dir": "data/parquet/midlarge-liquid-long-v1_bucketfix_full_audit_mom_v2",
"split": "train"
},
"books": [
{
"label": "core",
"equity_csv": "runs/fallback_book_compare/return_max_long_v6.114_equity.csv"
},
{
"label": "mom",
"equity_csv": "runs/book_overlay_v1_inputs/return_max_long_v6new.54_equity.csv"
}
],
"overlay_name": "return_book_overlay_v2b",
"allocations": {
"risk_on": {
"core": 0,
"mom": 1
},
"neutral": {
"core": 0,
"mom": 1
},
"risk_off": {
"core": 1,
"mom": 0
},
"unknown": {
"core": 0,
"mom": 1
}
}
}

29
configs/overlays/return_book_overlay_v3.json
Unescape Escape View File

@ -1,29 +0,0 @@
{
"overlay_name": "return_book_overlay_v3",
"initial_equity": 10000,
"start_date": "2022-03-03",
"end_date": "2026-03-13",
"regime_source": {
"config_path": "configs/experiments/return_max_long_v6.221.json",
"snapshot_dir": "data/parquet/midlarge-liquid-long-v1_bucketfix_full_audit_mom_v2",
"split": "train"
},
"books": [
{
"label": "core",
"equity_csv": "runs/book_overlay_v1_inputs/return_max_long_v6.221_equity.csv",
"experiment_config": "configs/experiments/return_max_long_v6.221.json"
},
{
"label": "mom",
"equity_csv": "runs/book_overlay_v1_inputs/return_max_long_v6new.54_equity.csv",
"experiment_config": "configs/experiments/return_max_long_v6new.54.json"
}
],
"allocations": {
"risk_on": {"core": 0.5, "mom": 0.5},
"neutral": {"core": 0.0, "mom": 1.0},
"risk_off": {"core": 1.0, "mom": 0.0},
"unknown": {"core": 1.0, "mom": 0.0}
}
}

39
configs/overlays/return_book_overlay_v3_oot.json
Unescape Escape View File

@ -1,39 +0,0 @@
{
"overlay_name": "return_book_overlay_v3_oot",
"initial_equity": 10000,
"start_date": "2020-01-02",
"end_date": "2021-12-31",
"regime_source": {
"config_path": "configs/experiments/return_max_long_v6.221.json",
"snapshot_dir": "data/datasets/snapshots/midlarge-liquid-long-v1-oot-2020-2021",
"split": "train"
},
"books": [
{
"label": "core",
"equity_csv": "runs/book_overlay_oot_inputs/csv/core_equity.csv"
},
{
"label": "mom",
"equity_csv": "runs/book_overlay_oot_inputs/csv/mom_equity.csv"
}
],
"allocations": {
"risk_on": {
"core": 0.5,
"mom": 0.5
},
"neutral": {
"core": 0.0,
"mom": 1.0
},
"risk_off": {
"core": 1.0,
"mom": 0.0
},
"unknown": {
"core": 1.0,
"mom": 0.0
}
}
}

29
configs/overlays/return_book_overlay_v3b.json
Unescape Escape View File

@ -1,29 +0,0 @@
{
"overlay_name": "return_book_overlay_v3b",
"initial_equity": 10000,
"start_date": "2022-03-03",
"end_date": "2026-03-13",
"regime_source": {
"config_path": "configs/experiments/return_max_long_v6.221.json",
"snapshot_dir": "data/parquet/midlarge-liquid-long-v1_bucketfix_full_audit_mom_v2",
"split": "train"
},
"books": [
{
"label": "core",
"equity_csv": "runs/book_overlay_v1_inputs/return_max_long_v6.221_equity.csv",
"experiment_config": "configs/experiments/return_max_long_v6.221.json"
},
{
"label": "mom",
"equity_csv": "runs/book_overlay_v1_inputs/return_max_long_v6new.54_equity.csv",
"experiment_config": "configs/experiments/return_max_long_v6new.54.json"
}
],
"allocations": {
"risk_on": {"core": 0.5, "mom": 0.5},
"neutral": {"core": 0.5, "mom": 0.5},
"risk_off": {"core": 1.0, "mom": 0.0},
"unknown": {"core": 1.0, "mom": 0.0}
}
}

39
configs/overlays/return_book_overlay_v3b_oot.json
Unescape Escape View File

@ -1,39 +0,0 @@
{
"overlay_name": "return_book_overlay_v3b_oot",
"initial_equity": 10000,
"start_date": "2020-01-02",
"end_date": "2021-12-31",
"regime_source": {
"config_path": "configs/experiments/return_max_long_v6.221.json",
"snapshot_dir": "data/datasets/snapshots/midlarge-liquid-long-v1-oot-2020-2021",
"split": "train"
},
"books": [
{
"label": "core",
"equity_csv": "runs/book_overlay_oot_inputs/csv/core_equity.csv"
},
{
"label": "mom",
"equity_csv": "runs/book_overlay_oot_inputs/csv/mom_equity.csv"
}
],
"allocations": {
"risk_on": {
"core": 0.5,
"mom": 0.5
},
"neutral": {
"core": 0.5,
"mom": 0.5
},
"risk_off": {
"core": 1.0,
"mom": 0.0
},
"unknown": {
"core": 1.0,
"mom": 0.0
}
}
}

40
docs/public_sqs_calculation.md
Unescape Escape View File

@ -8,7 +8,6 @@ split-only 점수와 final leaderboard 점수가 다를 수 있는지를 정리
- `SQS v4`: `SQS v3` + common-window capital-growth blend - `SQS v4`: `SQS v3` + common-window capital-growth blend
- `sqs_v3_score`: deployment/WFV-first primary rank - `sqs_v3_score`: deployment/WFV-first primary rank
- `stress_sqs_score`: 예전 public score. stress OOT quality를 추가 패널티로 곱한 legacy score - `stress_sqs_score`: 예전 public score. stress OOT quality를 추가 패널티로 곱한 legacy score
- overlay entry는 split/WFV 기반 `SQS v4` 대신 **overlay 전용 official score**를 쓴다.
핵심 결론부터 말하면: 핵심 결론부터 말하면:
@ -44,42 +43,13 @@ public leaderboard / registry rebuild도 같은 함수를 쓴다:
즉, 수동 계산과 leaderboard 계산은 같은 코드 경로를 따라야 한다. 즉, 수동 계산과 leaderboard 계산은 같은 코드 경로를 따라야 한다.
## 1.5 Overlay는 어떻게 공식 점수화되나 ## 1.5 Overlay 경로는 retired 상태다
overlay는 train/valid/test/WFV 구조가 없으므로 single-book `SQS v4`를 그대로 쓸 수 없다. overlay/book-of-books 평가는 코드베이스에서 제거됐다.
대신 아래 두 summary를 붙여서 공식 점수를 계산한다. - 기본 leaderboard는 single-book 전략만 포함한다.
- journal에 남아 있는 과거 overlay entry는 historical record로만 취급한다.
- `overlay_common_window_summary` - registry rebuild와 `fithia2 lb`는 overlay entry를 건너뛴다.
- `overlay_stress_window_summary`
계산 함수는:
- [`compute_overlay_public_sqs`](/Users/yirugi/mycloud/personal/workspace/fithia2/libs/backtest/tracker.py#L1356)
- [`compute_overlay_stress_sqs`](/Users/yirugi/mycloud/personal/workspace/fithia2/libs/backtest/tracker.py#L1381)
공식 overlay 점수는:
```text
overlay_window_score * overlay_stress_gate
```
legacy overlay stress 점수는:
```text
overlay_window_score * overlay_stress_gate * overlay_quality_factor
```
즉 overlay도 stress window는 ranking 보조가 아니라 **공식 통과 게이트**에 가깝게 다룬다.
추가 원칙:
- overlay official score는 spec의 `books[].equity_csv`를 **frozen input**으로 본다.
- [`fithia2 paper backtest --overlay`](/Users/yirugi/mycloud/personal/workspace/fithia2/apps/paper_trader/cli.py)도 같은 frozen `equity_csv`를 우선 replay해야 한다.
- 그래서 overlay leaderboard 숫자와 paper overlay 숫자가 다르면 먼저
`equity_csv`, `regime_source`, `start_date`, `end_date`가 같은지 확인한다.
- stress OOT overlay는 full-window csv를 재사용하면 안 된다.
OOT 전용 `equity_csv`와 OOT snapshot `regime_source.snapshot_dir`를 같이 고정해야 한다.
## 2. 흔한 오해 ## 2. 흔한 오해

49
docs/research_workflow_and_handoff.md
Unescape Escape View File

@ -224,52 +224,13 @@ python apps/tracker/cli.py attach-oot-robustness \
--summary runs/<experiment>_oot_rm/robustness_matrix/robustness_matrix_summary.json --summary runs/<experiment>_oot_rm/robustness_matrix/robustness_matrix_summary.json
``` ```
### Book Overlay Evaluation ### Retired Overlay Path
single-book manifest를 억지로 섞지 말고, 별도 book을 각각 먼저 고정 기간으로 돌린 뒤 book-of-books overlay 실험은 코드베이스에서 제거됐다.
overlay를 따로 평가한다.
1. 같은 기간, 같은 초기 자본으로 각 book의 equity curve를 만든다. - 공식 leaderboard와 paper backtest는 이제 single-book 전략만 지원한다.
2. [`configs/overlays`](/Users/yirugi/mycloud/personal/workspace/fithia2/configs/overlays) 아래 spec에 - 과거 overlay journal entry는 재현성 기록으로만 남고, registry/leaderboard 재빌드에는 포함되지 않는다.
regime별 자본 배분을 적는다. - 새 연구는 [`configs/experiments`](/Users/yirugi/mycloud/personal/workspace/fithia2/configs/experiments) 아래 single-book manifest 기준으로 진행한다.
3. [`apps/tools/evaluate_book_overlay.py`](/Users/yirugi/mycloud/personal/workspace/fithia2/apps/tools/evaluate_book_overlay.py)로
overlay equity와 summary를 만든다.
4. full-window spec/summary와 stress-window spec/summary를 모두 만든 뒤
[`apps/tracker/cli.py`](/Users/yirugi/mycloud/personal/workspace/fithia2/apps/tracker/cli.py)
`record-overlay`로 공식 journal/leaderboard에 등록한다.
주의:
- overlay spec의 `books[].equity_csv`는 **재현 기준 입력**이다.
- 공식 평가와 [`fithia2 paper backtest --overlay`](/Users/yirugi/mycloud/personal/workspace/fithia2/apps/paper_trader/cli.py)는
둘 다 이 frozen curve를 우선 replay해야 한다.
- OOT overlay는 full-window csv를 재사용하면 안 된다.
`runs/book_overlay_oot_inputs/...`처럼 OOT 전용 book curve를 따로 만든다.
- `regime_source.snapshot_dir`가 explicit snapshot directory라면 evaluator는 그 디렉터리를 그대로 merged load해야 한다.
예시:
```bash
fithia2 paper backtest \
--config configs/experiments/return_max_long_v6.221.json \
--config configs/experiments/return_max_long_v6new.54.json \
--capital 10000 \
--start 2022-03-03 \
--end 2026-03-13 \
--output runs/book_overlay_v1_inputs \
--no-trades
python apps/tools/evaluate_book_overlay.py \
--spec configs/overlays/return_book_overlay_v1.json \
--output-dir runs/return_book_overlay_v1_eval
python apps/tracker/cli.py record-overlay \
--spec configs/overlays/return_book_overlay_v3.json \
--summary runs/return_book_overlay_v3_eval/overlay_summary.json \
--stress-spec configs/overlays/return_book_overlay_v3_oot.json \
--stress-summary runs/return_book_overlay_v3_oot_eval/overlay_summary.json \
--hypothesis "Official overlay candidate"
```
## 8. 빠른 무결성 점검 ## 8. 빠른 무결성 점검

4
journal/LEADERBOARD.md
Unescape Escape View File

@ -1,7 +1,7 @@
# Strategy Improvement Leaderboard # Strategy Improvement Leaderboard
_Updated: 2026-03-27T03:43:33.985346+00:00_ _Updated: 2026-03-27T04:05:32.124207+00:00_
_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` for overlays or `fithia2 lb --include-retired` to inspect archived research._ _Default view excludes retired legacy PEAD / short-core / exact-pocket families and incomplete train-only scans. Use `fithia2 lb --include-retired` to inspect archived research._
_`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`._ _`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`._

36
journal/OVERLAY_LEADERBOARD.md
Unescape Escape View File

@ -1,36 +0,0 @@
# Overlay Strategy Leaderboard
_Updated: 2026-03-27T03:43:33.985346+00:00_
_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._
_`SQS` here is overlay official SQS: common-window overlay score with a stress OOT gate. `T.*` columns are common-window overlay metrics._
| # | Overlay | SQS | [T]Ret% | [T]Ann% | [T]DD% | Stress Ret% | Stress DD% | Stress Sharpe | Date |
|---|---------|-----|----------|----------|--------|-------------|------------|---------------|------|
| 1 | return_book_overlay_v4 | 91.4 | +185.3 | +29.7 | 5.0 | +7.4 | 5.8 | +0.80 | 2026-03-27 |
| 2 | return_book_overlay_v4b | 91.4 | +187.2 | +30.0 | 5.0 | +7.4 | 5.8 | +0.80 | 2026-03-27 |
| 3 | return_book_overlay_v3 | 90.2 | +175.6 | +28.6 | 4.5 | +6.0 | 5.4 | +0.67 | 2026-03-26 |
| 4 | return_book_overlay_v3b | 87.7 | +169.5 | +27.9 | 5.0 | +7.4 | 5.8 | +0.80 | 2026-03-26 |
| 5 | return_book_overlay_v4c | 60.7 | +232.5 | +34.8 | 5.2 | +3.1 | 3.3 | +0.43 | 2026-03-27 |
## Recent Overlay Entries
### IMP-0794 (2026-03-27) — return_book_overlay_v4b
Hypothesis: Core v6.221 plus higher-ranked v6new.207 momentum book with balanced 50/50 neutral allocation.
Verdict: **UNKNOWN** (SQS 91.4)
### IMP-0793 (2026-03-27) — return_book_overlay_v4c
Hypothesis: Core v6.221 plus top-return v6new.275 momentum book with aggressive 75/25 risk-on and full-neutral momentum allocation.
Verdict: **UNKNOWN** (SQS 60.7)
### IMP-0792 (2026-03-27) — return_book_overlay_v4
Hypothesis: Core v6.221 plus higher-ranked v6new.196 momentum book with balanced 50/50 neutral allocation.
Verdict: **UNKNOWN** (SQS 91.4)
### IMP-0738 (2026-03-26) — return_book_overlay_v3b
Hypothesis: Balanced overlay variant with 50/50 neutral allocation between v6.221 and v6new.54.
Verdict: **UNKNOWN** (SQS 87.7)
### IMP-0737 (2026-03-26) — return_book_overlay_v3
Hypothesis: Book-of-books overlay using v6.221 as stress fallback and v6new.54 as normal-regime book.
Verdict: **UNKNOWN** (SQS 90.2)

2207
journal/experiment_registry.json
View File

File diff suppressed because it is too large Load Diff

23
libs/backtest/domain.py
Unescape Escape View File

@ -797,25 +797,6 @@ class CommonWindowSummary(BaseModel):
metrics: MetricsBundle metrics: MetricsBundle
class OverlayWindowSummary(BaseModel):
"""Compact overlay run summary for full-window / stress-window evaluation."""
window_name: str = "overlay_window"
overlay_name: str = ""
start_date: dt.date
end_date: dt.date
initial_equity: float = 10_000.0
final_equity: float | None = None
return_pct: float | None = None
annualized_return_pct: float | None = None
max_drawdown_pct: float | None = None
sharpe_ratio: float | None = None
day_count: int = 0
books: list[str] = Field(default_factory=list)
allocations: dict[str, dict[str, float]] = Field(default_factory=dict)
regime_day_counts: dict[str, int] = Field(default_factory=dict)
class ConfigDelta(BaseModel): class ConfigDelta(BaseModel):
"""Records what changed from a baseline experiment.""" """Records what changed from a baseline experiment."""
@ -836,8 +817,6 @@ class JournalEntry(BaseModel):
robustness_matrix_summary: RobustnessMatrixSummary | None = None robustness_matrix_summary: RobustnessMatrixSummary | None = None
out_of_time_robustness_summary: RobustnessMatrixSummary | None = None out_of_time_robustness_summary: RobustnessMatrixSummary | None = None
common_window_summary: CommonWindowSummary | None = None common_window_summary: CommonWindowSummary | None = None
overlay_common_window_summary: OverlayWindowSummary | None = None
overlay_stress_window_summary: OverlayWindowSummary | None = None
sqs_score: float | None = None sqs_score: float | None = None
sqs_breakdown: dict[str, float] = Field(default_factory=dict) sqs_breakdown: dict[str, float] = Field(default_factory=dict)
sqs_v3_score: float | None = None sqs_v3_score: float | None = None
@ -885,8 +864,6 @@ class RegistryEntry(BaseModel):
deployment_score: float | None = None deployment_score: float | None = None
common_window_score: float | None = None common_window_score: float | None = None
common_window_summary: CommonWindowSummary | None = None common_window_summary: CommonWindowSummary | None = None
overlay_common_window_summary: OverlayWindowSummary | None = None
overlay_stress_window_summary: OverlayWindowSummary | None = None
walk_forward_summary: WalkForwardSummary | None = None walk_forward_summary: WalkForwardSummary | None = None
robustness_matrix_summary: RobustnessMatrixSummary | None = None robustness_matrix_summary: RobustnessMatrixSummary | None = None
out_of_time_robustness_summary: RobustnessMatrixSummary | None = None out_of_time_robustness_summary: RobustnessMatrixSummary | None = None

193
libs/backtest/overlay.py
Unescape Escape View File

@ -1,193 +0,0 @@
"""Evaluate deterministic book overlays from daily equity curves."""
from __future__ import annotations
import datetime as dt
import math
from pathlib import Path
from typing import Any
import pandas as pd
def load_merged_store_from_snapshot_dir(
snapshot_dir: str | Path,
*,
oracle_url: str,
db_dsn: str,
):
"""Load and merge train/valid/test splits from an explicit snapshot directory."""
from libs.backtest.snapshot_store import SnapshotStore
base = Path(snapshot_dir)
stores = []
for split in ("train", "valid", "test"):
if not (base / f"{split}.parquet").exists():
continue
stores.append(
SnapshotStore.load(
snapshot_dir=base,
split_name=split,
oracle_url=oracle_url,
db_dsn=db_dsn,
)
)
if not stores:
raise FileNotFoundError(f"No snapshot splits found under {base}")
merged_candidates: dict[dt.date, dict[tuple[Any, ...], dict[str, Any]]] = {}
merged_bars: dict[str, dict[dt.date, dict[str, Any]]] = {}
merged_macro: dict[dt.date, dict[str, Any]] = {}
for store in stores:
for exec_date in store.all_execution_dates():
bucket = merged_candidates.setdefault(exec_date, {})
for candidate in store.get_candidates_for_date(exec_date):
dedupe_key = (
candidate.get("event_id"),
candidate.get("symbol"),
candidate.get("execution_date"),
candidate.get("reaction_date"),
)
bucket.setdefault(dedupe_key, candidate)
for symbol, bars in store._bars.items():
merged_bars.setdefault(symbol, {}).update(bars)
for macro_date, macro_values in store._macro.items():
merged_macro.setdefault(macro_date, {}).update(macro_values)
from libs.backtest.snapshot_store import SnapshotStore
return SnapshotStore(
candidates_by_exec_date={
date: list(rows.values())
for date, rows in merged_candidates.items()
},
bars_by_symbol_date=merged_bars,
macro_by_date=merged_macro,
)
def load_equity_curve_csv(path: str | Path) -> pd.DataFrame:
"""Load a paper backtest equity CSV into a normalized daily returns frame."""
csv_path = Path(path)
df = pd.read_csv(csv_path, parse_dates=["date"])
required = {"date", "equity"}
missing = required.difference(df.columns)
if missing:
raise ValueError(f"Missing columns in {csv_path}: {sorted(missing)}")
if df.empty:
raise ValueError(f"Equity CSV has no rows: {csv_path}")
df = df.sort_values("date").copy()
df["date"] = pd.to_datetime(df["date"]).dt.date
df["equity"] = df["equity"].astype(float)
df["daily_return"] = df["equity"].pct_change().fillna(0.0)
return df[["date", "equity", "daily_return"]]
def validate_allocations(
allocations: dict[str, dict[str, float]],
labels: set[str],
*,
tolerance: float = 1e-6,
) -> None:
"""Ensure each regime allocation references known labels and sums to 1."""
if not allocations:
raise ValueError("allocations must not be empty")
if "unknown" not in allocations:
raise ValueError("allocations must include an 'unknown' regime")
for regime, weights in allocations.items():
unknown = set(weights).difference(labels)
if unknown:
raise ValueError(
f"Allocation for regime '{regime}' references unknown labels: {sorted(unknown)}"
)
total = sum(float(weight) for weight in weights.values())
if abs(total - 1.0) > tolerance:
raise ValueError(
f"Allocation for regime '{regime}' must sum to 1.0, got {total:.6f}"
)
def build_overlay_curve(
*,
curves: dict[str, pd.DataFrame],
allocations: dict[str, dict[str, float]],
regimes_by_date: dict[dt.date, str],
initial_equity: float = 10_000.0,
) -> pd.DataFrame:
"""Combine per-book daily returns into a single overlay equity curve."""
labels = set(curves)
if not labels:
raise ValueError("curves must not be empty")
validate_allocations(allocations, labels)
merged: pd.DataFrame | None = None
for label, df in curves.items():
renamed = df.rename(
columns={
"equity": f"equity_{label}",
"daily_return": f"daily_return_{label}",
}
)
frame = renamed[["date", f"daily_return_{label}"]]
merged = frame if merged is None else merged.merge(frame, on="date", how="inner")
if merged is None or merged.empty:
raise ValueError("No overlapping dates across curves")
merged = merged.sort_values("date").copy()
merged["regime"] = merged["date"].map(regimes_by_date).fillna("unknown")
overlay_returns: list[float] = []
for row in merged.itertuples(index=False):
weights = allocations.get(row.regime, allocations["unknown"])
ret = 0.0
for label in labels:
ret += float(weights.get(label, 0.0)) * float(getattr(row, f"daily_return_{label}"))
overlay_returns.append(ret)
merged["overlay_return"] = overlay_returns
equity = initial_equity
overlay_equity: list[float] = []
for ret in overlay_returns:
equity *= 1.0 + float(ret)
overlay_equity.append(equity)
merged["overlay_equity"] = overlay_equity
return merged[["date", "regime", "overlay_return", "overlay_equity"]]
def summarize_overlay_curve(curve: pd.DataFrame, *, initial_equity: float) -> dict[str, Any]:
"""Return total return, drawdown, Sharpe, and regime counts for an overlay."""
if curve.empty:
raise ValueError("curve must not be empty")
final_equity = float(curve["overlay_equity"].iloc[-1])
total_return_pct = (final_equity / float(initial_equity) - 1.0) * 100.0
peak = float(initial_equity)
max_drawdown_pct = 0.0
for equity in curve["overlay_equity"]:
peak = max(peak, float(equity))
drawdown_pct = (peak - float(equity)) / peak * 100.0 if peak > 0 else 0.0
max_drawdown_pct = max(max_drawdown_pct, drawdown_pct)
rets = curve["overlay_return"].astype(float)
if len(rets) >= 2 and float(rets.std()) > 0:
sharpe = float(rets.mean() / rets.std() * math.sqrt(252.0))
else:
sharpe = 0.0
regime_counts = {
str(regime): int(count)
for regime, count in curve["regime"].value_counts().sort_index().items()
}
return {
"return_pct": total_return_pct,
"max_dd_pct": max_drawdown_pct,
"sharpe": sharpe,
"final_equity": final_equity,
"day_count": int(len(curve)),
"regime_day_counts": regime_counts,
}

277
libs/backtest/tracker.py
Unescape Escape View File

@ -22,7 +22,6 @@ from libs.backtest.domain import (
ReturnScoreWeights, ReturnScoreWeights,
PromotionScoreWeights, PromotionScoreWeights,
RegistryEntry, RegistryEntry,
OverlayWindowSummary,
SplitResult, SplitResult,
SQSWeights, SQSWeights,
SQSv2Weights, SQSv2Weights,
@ -1316,134 +1315,6 @@ def compute_common_window_score(
return round(score, 1), breakdown 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]:
"""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( def _get_robustness_horizon_summary(
summary: RobustnessMatrixSummary | None, summary: RobustnessMatrixSummary | None,
horizon_days: int, horizon_days: int,
@ -1745,32 +1616,6 @@ def refresh_public_scores(
updated_entries.append(entry) updated_entries.append(entry)
continue 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")) test_result = _hydrate_split_result(entry.results.get("test"))
valid_result = _hydrate_split_result(entry.results.get("valid")) valid_result = _hydrate_split_result(entry.results.get("valid"))
train_result = _hydrate_split_result(entry.results.get("train")) train_result = _hydrate_split_result(entry.results.get("train"))
@ -2036,7 +1881,6 @@ def filter_registry_entries(
entries: list[RegistryEntry], entries: list[RegistryEntry],
*, *,
include_retired: bool = False, include_retired: bool = False,
include_overlays: bool = False,
) -> list[RegistryEntry]: ) -> list[RegistryEntry]:
"""Filter registry entries for the default surfaced leaderboard.""" """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] filtered = list(entries) if include_retired else [entry for entry in entries if not entry.is_retired]
@ -2045,33 +1889,10 @@ def filter_registry_entries(
for entry in filtered for entry in filtered
if ( if (
entry.sqs_score is not None entry.sqs_score is not None
and ( and entry.strategy_family != "overlay"
(include_overlays and entry.strategy_family == "overlay")
or (
entry.strategy_family != "overlay"
and entry.trade_count > 0 and entry.trade_count > 0
and entry.valid_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
)
] ]
@ -2080,8 +1901,6 @@ def _is_retired_journal_entry(entry: JournalEntry) -> bool:
def _is_complete_journal_entry(entry: JournalEntry) -> bool: 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") test_result = entry.results.get("test")
valid_result = entry.results.get("valid") valid_result = entry.results.get("valid")
return ( return (
@ -2365,37 +2184,7 @@ def rebuild_registry(
for je in entries: for je in entries:
strategy_family = classify_strategy_family(je.experiment_name, je.tags) strategy_family = classify_strategy_family(je.experiment_name, je.tags)
is_retired = is_retired_strategy_family(strategy_family) is_retired = is_retired_strategy_family(strategy_family)
if strategy_family == "overlay":
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 continue
test_result = _hydrate_split_result(je.results.get("test")) test_result = _hydrate_split_result(je.results.get("test"))
@ -2540,9 +2329,8 @@ def rebuild_registry(
registry_path.parent.mkdir(parents=True, exist_ok=True) registry_path.parent.mkdir(parents=True, exist_ok=True)
registry_path.write_text(registry.model_dump_json(indent=2)) registry_path.write_text(registry.model_dump_json(indent=2))
# Write LEADERBOARD.md and OVERLAY_LEADERBOARD.md # Write LEADERBOARD.md
_write_leaderboard_md(leaderboard_path, registry, entries) _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)) logger.info("registry_rebuilt", count=len(registry_entries))
return registry return registry
@ -2556,7 +2344,6 @@ def _write_leaderboard_md(
visible_entries = filter_registry_entries( visible_entries = filter_registry_entries(
registry.entries, registry.entries,
include_retired=False, include_retired=False,
include_overlays=False,
) )
visible_recent = [ visible_recent = [
entry entry
@ -2570,7 +2357,7 @@ def _write_leaderboard_md(
lines: list[str] = [] lines: list[str] = []
lines.append("# Strategy Improvement Leaderboard") lines.append("# Strategy Improvement Leaderboard")
lines.append(f"_Updated: {registry.updated_at}_\n") lines.append(f"_Updated: {registry.updated_at}_\n")
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` for overlays or `fithia2 lb --include-retired` to inspect archived research._\n") lines.append("_Default view excludes retired legacy PEAD / short-core / exact-pocket families and incomplete train-only scans. Use `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("_`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("| # | Experiment | SQS | [Tr]Ret% | [V]Ret% | [T]Ret% | [T]Ann% | [T]DD% | [T]Gross% | [T]DIM% | [T]R/G | Date |")
lines.append("|---|-----------|-----|----------|----------|----------|----------|--------|------------|----------|---------|------|") lines.append("|---|-----------|-----|----------|----------|----------|----------|--------|------------|----------|---------|------|")
@ -2621,62 +2408,6 @@ def _write_leaderboard_md(
path.write_text("\n".join(lines) + "\n") 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 "-"
stress_ret = f"{stress.return_pct:+.1f}" if stress and stress.return_pct is not None else "-"
stress_dd = f"{stress.max_drawdown_pct:.1f}" if stress and stress.max_drawdown_pct is not None else "-"
stress_sharpe = f"{stress.sharpe_ratio:+.2f}" if stress and stress.sharpe_ratio is not None 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_ret}"
f" | {stress_dd}"
f" | {stress_sharpe}"
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}")
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")
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
# Run scanning helpers (for CLI record command) # Run scanning helpers (for CLI record command)
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------

76
tests/unit/backtest/test_overlay.py
Unescape Escape View File

@ -1,76 +0,0 @@
from __future__ import annotations
import datetime as dt
import pandas as pd
from libs.backtest.overlay import build_overlay_curve, summarize_overlay_curve, validate_allocations
def test_validate_allocations_requires_unknown_and_unit_sum() -> None:
try:
validate_allocations({"risk_on": {"core": 1.0}}, {"core"})
except ValueError as exc:
assert "unknown" in str(exc)
else:
raise AssertionError("expected ValueError")
try:
validate_allocations({"unknown": {"core": 0.8}}, {"core"})
except ValueError as exc:
assert "sum to 1.0" in str(exc)
else:
raise AssertionError("expected ValueError")
def test_build_overlay_curve_switches_by_regime() -> None:
dates = [dt.date(2024, 1, 2), dt.date(2024, 1, 3), dt.date(2024, 1, 4)]
core = pd.DataFrame(
{
"date": dates,
"equity": [100.0, 110.0, 104.5],
"daily_return": [0.0, 0.10, -0.05],
}
)
mom = pd.DataFrame(
{
"date": dates,
"equity": [100.0, 105.0, 110.25],
"daily_return": [0.0, 0.05, 0.05],
}
)
regimes = {
dates[0]: "risk_on",
dates[1]: "risk_off",
dates[2]: "risk_on",
}
allocations = {
"risk_on": {"core": 0.0, "mom": 1.0},
"risk_off": {"core": 1.0, "mom": 0.0},
"unknown": {"core": 1.0, "mom": 0.0},
}
curve = build_overlay_curve(
curves={"core": core, "mom": mom},
allocations=allocations,
regimes_by_date=regimes,
initial_equity=100.0,
)
assert curve["overlay_return"].round(6).tolist() == [0.0, 0.10, 0.05]
assert round(float(curve["overlay_equity"].iloc[-1]), 4) == 115.5
def test_summarize_overlay_curve_reports_return_drawdown_and_regimes() -> None:
curve = pd.DataFrame(
{
"date": [dt.date(2024, 1, 2), dt.date(2024, 1, 3), dt.date(2024, 1, 4)],
"regime": ["risk_on", "risk_off", "risk_on"],
"overlay_return": [0.0, -0.10, 0.05],
"overlay_equity": [100.0, 90.0, 94.5],
}
)
summary = summarize_overlay_curve(curve, initial_equity=100.0)
assert round(summary["return_pct"], 4) == -5.5
assert round(summary["max_dd_pct"], 4) == 10.0
assert summary["regime_day_counts"] == {"risk_off": 1, "risk_on": 2}

94
tests/unit/backtest/test_tracker.py
Unescape Escape View File

@ -13,7 +13,6 @@ from libs.backtest.domain import (
CommonWindowSummary, CommonWindowSummary,
JournalEntry, JournalEntry,
MetricsBundle, MetricsBundle,
OverlayWindowSummary,
RobustnessHorizonSummary, RobustnessHorizonSummary,
RobustnessMatrixSummary, RobustnessMatrixSummary,
SplitResult, SplitResult,
@ -44,8 +43,6 @@ from libs.backtest.tracker import (
compute_public_sqs_v3, compute_public_sqs_v3,
compute_public_sqs_v4, compute_public_sqs_v4,
compute_common_window_score, compute_common_window_score,
compute_overlay_public_sqs,
compute_overlay_stress_sqs,
compute_promotion_score, compute_promotion_score,
_compute_oot_positive_rate_63plus, _compute_oot_positive_rate_63plus,
compute_oot_robustness_gate, compute_oot_robustness_gate,
@ -58,7 +55,6 @@ from libs.backtest.tracker import (
compute_unified_split_quality, compute_unified_split_quality,
compute_wfqs, compute_wfqs,
compute_wfqs_v2, compute_wfqs_v2,
filter_overlay_registry_entries,
filter_registry_entries, filter_registry_entries,
get_next_entry_id, get_next_entry_id,
journal_lock, journal_lock,
@ -900,61 +896,6 @@ class TestCommonWindowScore:
assert v4_strong > v4_weak assert v4_strong > v4_weak
class TestOverlayPublicScore:
def test_rewards_strong_overlay_with_positive_stress_window(self):
overlay = OverlayWindowSummary(
overlay_name="return_book_overlay_v3",
start_date=dt.date(2022, 3, 3),
end_date=dt.date(2026, 3, 13),
return_pct=145.0,
annualized_return_pct=26.0,
max_drawdown_pct=4.5,
sharpe_ratio=2.62,
day_count=1014,
)
stress = OverlayWindowSummary(
overlay_name="return_book_overlay_v3",
start_date=dt.date(2020, 1, 2),
end_date=dt.date(2021, 12, 31),
return_pct=5.95,
annualized_return_pct=2.9,
max_drawdown_pct=5.37,
sharpe_ratio=0.67,
day_count=508,
)
public_score, breakdown, source = compute_overlay_public_sqs(overlay, stress)
stress_score, _, _ = compute_overlay_stress_sqs(overlay, stress)
assert source == "overlay_v1_common_window+stress_gate"
assert public_score is not None and public_score > 0
assert stress_score is not None and stress_score <= public_score
assert breakdown["overlay_gate_factor"] == 1.0
def test_penalizes_overlay_with_flat_negative_stress_window(self):
overlay = OverlayWindowSummary(
overlay_name="return_book_overlay_v1",
start_date=dt.date(2022, 3, 3),
end_date=dt.date(2026, 3, 13),
return_pct=153.89,
annualized_return_pct=27.0,
max_drawdown_pct=4.52,
sharpe_ratio=2.72,
day_count=1014,
)
weak_stress = OverlayWindowSummary(
overlay_name="return_book_overlay_v1",
start_date=dt.date(2020, 1, 2),
end_date=dt.date(2021, 12, 31),
return_pct=-1.26,
annualized_return_pct=-0.6,
max_drawdown_pct=3.15,
sharpe_ratio=-0.16,
day_count=508,
)
public_score, breakdown, _ = compute_overlay_public_sqs(overlay, weak_stress)
assert public_score is not None
assert breakdown["overlay_gate_factor"] == 0.65
class TestComputeRQS: class TestComputeRQS:
def test_requires_valid_and_test(self): def test_requires_valid_and_test(self):
score, breakdown = compute_rqs(None, None, None) score, breakdown = compute_rqs(None, None, None)
@ -1934,56 +1875,27 @@ class TestRebuildRegistry:
assert "| # |" in lb_text assert "| # |" in lb_text
assert "| # | Experiment | SQS | [Tr]Ret% | [V]Ret% | [T]Ret% | [T]Ann% | [T]DD% | [T]Gross% | [T]DIM% | [T]R/G | Date |" in lb_text assert "| # | Experiment | SQS | [Tr]Ret% | [V]Ret% | [T]Ret% | [T]Ann% | [T]DD% | [T]Gross% | [T]DIM% | [T]R/G | Date |" in lb_text
def test_registry_and_leaderboard_split_overlay_entry(self, tmp_path): def test_registry_skips_legacy_overlay_entries(self, tmp_path):
journal_path = tmp_path / "journal.jsonl" journal_path = tmp_path / "journal.jsonl"
registry_path = tmp_path / "registry.json" registry_path = tmp_path / "registry.json"
leaderboard_path = tmp_path / "LEADERBOARD.md" leaderboard_path = tmp_path / "LEADERBOARD.md"
overlay_leaderboard_path = tmp_path / "OVERLAY_LEADERBOARD.md"
entry = JournalEntry( entry = JournalEntry(
entry_id="IMP-0001", entry_id="IMP-0001",
timestamp="2026-03-25T22:00:00+00:00", timestamp="2026-03-25T22:00:00+00:00",
experiment_name="return_book_overlay_v3", experiment_name="return_book_overlay_v3",
hypothesis="overlay", hypothesis="overlay",
overlay_common_window_summary=OverlayWindowSummary(
overlay_name="return_book_overlay_v3",
start_date=dt.date(2022, 3, 3),
end_date=dt.date(2026, 3, 13),
return_pct=145.0,
annualized_return_pct=26.0,
max_drawdown_pct=4.5,
sharpe_ratio=2.62,
day_count=1014,
),
overlay_stress_window_summary=OverlayWindowSummary(
overlay_name="return_book_overlay_v3",
start_date=dt.date(2020, 1, 2),
end_date=dt.date(2021, 12, 31),
return_pct=5.95,
annualized_return_pct=2.9,
max_drawdown_pct=5.37,
sharpe_ratio=0.67,
day_count=508,
),
tags=["overlay"], tags=["overlay"],
) )
append_journal_entry(journal_path, entry) append_journal_entry(journal_path, entry)
registry = rebuild_registry(journal_path, registry_path, leaderboard_path) registry = rebuild_registry(journal_path, registry_path, leaderboard_path)
visible = filter_registry_entries(registry.entries) visible = filter_registry_entries(registry.entries)
overlay_visible = filter_overlay_registry_entries(registry.entries)
assert len(visible) == 0 assert len(visible) == 0
assert len(overlay_visible) == 1 assert registry.entries == []
assert overlay_visible[0].strategy_family == "overlay"
assert overlay_visible[0].sqs_score is not None
assert overlay_visible[0].total_return_pct == 145.0
assert overlay_visible[0].annualized_return_pct == 26.0
lb_text = leaderboard_path.read_text() lb_text = leaderboard_path.read_text()
overlay_lb_text = overlay_leaderboard_path.read_text()
assert "return_book_overlay_v3" not in lb_text assert "return_book_overlay_v3" not in lb_text
assert "Default view excludes overlay/book-of-books rows" in lb_text assert "Default view excludes retired legacy PEAD" in lb_text
assert "return_book_overlay_v3" in overlay_lb_text
assert "Overlay Strategy Leaderboard" in overlay_lb_text
def test_registry_preserves_walk_forward_summary(self, tmp_path): def test_registry_preserves_walk_forward_summary(self, tmp_path):
journal_path = tmp_path / "journal.jsonl" journal_path = tmp_path / "journal.jsonl"

17
tests/unit/paper_trader/test_cli.py
Unescape Escape View File

@ -8,7 +8,7 @@ from rich.console import Console
from apps.paper_trader import cli from apps.paper_trader import cli
def test_resolve_rank_configs_skips_overlay_rows_and_warns(tmp_path: Path, monkeypatch) -> None: def test_resolve_rank_configs_selects_runnable_single_book_rows(tmp_path: Path, monkeypatch) -> None:
monkeypatch.chdir(tmp_path) monkeypatch.chdir(tmp_path)
journal_dir = tmp_path / "journal" journal_dir = tmp_path / "journal"
journal_dir.mkdir(parents=True) journal_dir.mkdir(parents=True)
@ -19,17 +19,16 @@ def test_resolve_rank_configs_skips_overlay_rows_and_warns(tmp_path: Path, monke
registry = { registry = {
"entries": [ "entries": [
{ {
"experiment_name": "return_book_overlay_v3", "experiment_name": "return_max_long_missing",
"strategy_family": "overlay", "strategy_family": "return_max_long",
"sqs_score": 81.6, "sqs_score": 81.6,
"trade_count": 0, "trade_count": 10,
"valid_trade_count": 0, "valid_trade_count": 3,
"overlay_common_window_summary": {"return_pct": 145.0},
"is_retired": False, "is_retired": False,
}, },
{ {
"experiment_name": "return_max_long_demo", "experiment_name": "return_max_long_demo",
"strategy_family": "pead", "strategy_family": "return_max_long",
"sqs_score": 70.0, "sqs_score": 70.0,
"trade_count": 10, "trade_count": 10,
"valid_trade_count": 3, "valid_trade_count": 3,
@ -46,6 +45,4 @@ def test_resolve_rank_configs_skips_overlay_rows_and_warns(tmp_path: Path, monke
configs = cli._resolve_rank_configs(1, 1) configs = cli._resolve_rank_configs(1, 1)
assert configs == ["configs/experiments/return_max_long_demo.json"] assert configs == ["configs/experiments/return_max_long_demo.json"]
output = console.export_text() assert console.export_text() == ""
assert "Skipping overlay leaderboard entries for `--top/--rank`" in output
assert "return_book_overlay_v3" in output

75
tests/unit/paper_trader/test_overlay_backtest.py
Unescape Escape View File

@ -1,75 +0,0 @@
from __future__ import annotations
import datetime as dt
import json
from pathlib import Path
import pandas as pd
from apps.paper_trader.backtest_sim import run_overlay_backtest_sync
def _write_equity_csv(path: Path, rows: list[tuple[str, float]]) -> None:
df = pd.DataFrame(rows, columns=["date", "equity"])
path.parent.mkdir(parents=True, exist_ok=True)
df.to_csv(path, index=False)
def test_run_overlay_backtest_sync_replays_frozen_equity_csv(tmp_path: Path) -> None:
core_csv = tmp_path / "core.csv"
mom_csv = tmp_path / "mom.csv"
_write_equity_csv(core_csv, [("2025-01-02", 100.0), ("2025-01-03", 110.0), ("2025-01-06", 121.0)])
_write_equity_csv(mom_csv, [("2025-01-02", 100.0), ("2025-01-03", 90.0), ("2025-01-06", 81.0)])
spec = {
"overlay_name": "overlay_demo",
"books": [
{"label": "core", "equity_csv": str(core_csv)},
{"label": "mom", "equity_csv": str(mom_csv)},
],
"allocations": {
"unknown": {"core": 0.5, "mom": 0.5},
},
}
spec_path = tmp_path / "overlay.json"
spec_path.write_text(json.dumps(spec))
result = run_overlay_backtest_sync(
overlay_config_path=str(spec_path),
capital=100.0,
start_date=dt.date(2025, 1, 2),
end_date=dt.date(2025, 1, 6),
)
assert result["is_overlay"] is True
assert result["overlay_replay_mode"] == "frozen_equity_csv"
assert round(result["summary"]["return_pct"], 4) == 0.0
assert result["summary"]["trade_count"] == 0
def test_run_overlay_backtest_sync_rebases_window_from_frozen_curves(tmp_path: Path) -> None:
core_csv = tmp_path / "core.csv"
_write_equity_csv(core_csv, [("2025-01-02", 100.0), ("2025-01-03", 200.0), ("2025-01-06", 400.0)])
spec = {
"overlay_name": "overlay_demo",
"books": [
{"label": "core", "equity_csv": str(core_csv)},
],
"allocations": {
"unknown": {"core": 1.0},
},
}
spec_path = tmp_path / "overlay.json"
spec_path.write_text(json.dumps(spec))
result = run_overlay_backtest_sync(
overlay_config_path=str(spec_path),
capital=100.0,
start_date=dt.date(2025, 1, 3),
end_date=dt.date(2025, 1, 6),
)
assert result["overlay_replay_mode"] == "frozen_equity_csv"
assert round(result["summary"]["return_pct"], 4) == 100.0
assert round(result["summary"]["final_equity"], 4) == 200.0

69
tests/unit/tools/test_evaluate_book_overlay.py
Unescape Escape View File

@ -1,69 +0,0 @@
from __future__ import annotations
import datetime as dt
from apps.tools import evaluate_book_overlay as mod
class _FakeStore:
def __init__(self) -> None:
self._dates = [dt.date(2025, 1, 2), dt.date(2025, 1, 3)]
def slice_by_date_range(self, start: dt.date, end: dt.date):
self._dates = [d for d in self._dates if start <= d <= end]
return self
def all_trading_days(self):
return list(self._dates)
def get_macro_for_date(self, date: dt.date):
return {"state": f"regime-{date.isoformat()}"}
def test_compute_regimes_uses_merged_store(monkeypatch) -> None:
fake_store = _FakeStore()
monkeypatch.setattr(mod, "load_manifest", lambda path: object())
monkeypatch.setattr(mod, "resolve_config", lambda manifest, config_root=".": object())
monkeypatch.setattr(mod, "_build_merged_snapshot_store", lambda manifest, config, snapshot_dir_override=None: fake_store)
monkeypatch.setattr(mod, "_macro_regime_state", lambda config, macro: macro["state"])
regimes = mod._compute_regimes(
snapshot_dir="data/parquet/example_snapshot",
split="train",
config_path="configs/experiments/example.json",
start_date=dt.date(2025, 1, 2),
end_date=dt.date(2025, 1, 3),
)
assert regimes == {
dt.date(2025, 1, 2): "regime-2025-01-02",
dt.date(2025, 1, 3): "regime-2025-01-03",
}
def test_compute_regimes_prefers_explicit_snapshot_dir(monkeypatch, tmp_path) -> None:
snapshot_dir = tmp_path / "snap"
snapshot_dir.mkdir()
(snapshot_dir / "train.parquet").write_text("stub")
fake_store = _FakeStore()
monkeypatch.setattr(mod, "load_manifest", lambda path: object())
monkeypatch.setattr(mod, "resolve_config", lambda manifest, config_root=".": object())
monkeypatch.setattr(mod, "load_merged_store_from_snapshot_dir", lambda snapshot_dir, oracle_url, db_dsn: fake_store)
monkeypatch.setattr(mod, "_build_merged_snapshot_store", lambda *args, **kwargs: (_ for _ in ()).throw(AssertionError("should not be called")))
monkeypatch.setattr(mod, "_macro_regime_state", lambda config, macro: macro["state"])
regimes = mod._compute_regimes(
snapshot_dir=snapshot_dir,
split="train",
config_path="configs/experiments/example.json",
start_date=dt.date(2025, 1, 2),
end_date=dt.date(2025, 1, 3),
)
assert regimes == {
dt.date(2025, 1, 2): "regime-2025-01-02",
dt.date(2025, 1, 3): "regime-2025-01-03",
}
Write Preview
Loading…
Cancel
Save