wehub-resource-sync
1329 lines
49 KiB
Python
1329 lines
49 KiB
Python
"""Reader-facing API for exploring backtest results from the registry.
|
|
|
|
Usage::
|
|
|
|
from case_studies.utils.backtest_explorer import BacktestExplorer
|
|
|
|
explorer = BacktestExplorer("etfs")
|
|
explorer.summary()
|
|
explorer.best(stage="signal", top_n=5)
|
|
explorer.compare_allocators()
|
|
explorer.inspect("backtest_hash_abc")
|
|
explorer.progression("prediction_hash_xyz")
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import sqlite3
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import numpy as np
|
|
import polars as pl
|
|
|
|
from case_studies.utils.backtest_presets import cost_view, strategy_view
|
|
from case_studies.utils.notebook_contracts import excluded_family_sql, filter_active_model_rows
|
|
|
|
# Sentinel distinguishing "no filter" from "match exit_at_max_days IS NULL".
|
|
_UNSET = object()
|
|
|
|
# Canonical schema for BacktestExplorer.best() output. Used to construct
|
|
# schema-stable empty DataFrames so downstream `.select("source", ...)`
|
|
# surfaces "(no matching rows)" instead of a cryptic ColumnNotFoundError.
|
|
_BEST_SCHEMA: dict[str, pl.DataType] = {
|
|
"backtest_hash": pl.Utf8,
|
|
"prediction_hash": pl.Utf8,
|
|
"source": pl.Utf8,
|
|
"family": pl.Utf8,
|
|
"config_name": pl.Utf8,
|
|
"label": pl.Utf8,
|
|
"signal_method": pl.Utf8,
|
|
"universe_filter": pl.Utf8,
|
|
"exit_at_max_days": pl.Int64,
|
|
"sharpe": pl.Float64,
|
|
"cagr": pl.Float64,
|
|
"max_drawdown": pl.Float64,
|
|
"total_return": pl.Float64,
|
|
"volatility": pl.Float64,
|
|
"ic_mean": pl.Float64,
|
|
}
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Result containers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@dataclass
|
|
class BacktestDetail:
|
|
"""Full detail for a single backtest run."""
|
|
|
|
backtest_hash: str
|
|
prediction_hash: str
|
|
stage: str | None
|
|
spec: dict
|
|
metrics: dict[str, float]
|
|
daily_returns_path: Path | None
|
|
trades_path: Path | None
|
|
weights_path: Path | None
|
|
source: str | None = None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Spec-string parsing helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _parse_spec(spec_str: str | None) -> dict | None:
|
|
"""Return the parsed JSON spec as a dict, or None if not parseable.
|
|
|
|
Failure modes that return None:
|
|
1. ``spec_str is None`` (NULL in the registry)
|
|
2. ``spec_str == ""`` (empty string)
|
|
3. ``spec_str`` is malformed JSON (truncated writes)
|
|
4. The parsed value is not a dict (e.g. ``"42"`` → int, ``"null"``
|
|
→ None, ``"[1, 2]"`` → list) — callers that feed this into
|
|
``strategy_view`` or ``cost_view`` rely on dict-shaped input
|
|
|
|
Callers that don't care about distinguishing "not parseable" from
|
|
"empty dict" can write ``_parse_spec(s) or {}`` to get a
|
|
guaranteed-dict. Callers that DO need to distinguish (e.g.
|
|
``compare_allocators`` which returns "unknown" for unparseable
|
|
rows and "equal_weight" for a successfully-parsed spec missing
|
|
the allocation method) check ``is None`` explicitly.
|
|
|
|
Replaces five copies of the same parse-and-default pattern.
|
|
"""
|
|
if not spec_str:
|
|
return None
|
|
# Note: only catching JSONDecodeError here, not TypeError. The
|
|
# original sites caught both defensively, but json.loads only raises
|
|
# TypeError on non-string input, which the ``if not spec_str`` guard
|
|
# plus the str|None type annotation make unreachable from well-typed
|
|
# callers. The ``(A, B)`` tuple form is also what ruff 0.15 on a
|
|
# py314 target rewrites to the Python-2-style comma form, which then
|
|
# fails on Python 3.12 CI.
|
|
try:
|
|
value = json.loads(spec_str)
|
|
except json.JSONDecodeError:
|
|
return None
|
|
return value if isinstance(value, dict) else None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Explorer
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class BacktestExplorer:
|
|
"""High-level reader API for querying backtest results.
|
|
|
|
All data is read from ``registry.db`` — no JSON files needed.
|
|
"""
|
|
|
|
def __init__(self, case_study: str, *, case_dir: Path | None = None):
|
|
from utils.paths import get_case_study_dir
|
|
|
|
self.case_study = case_study
|
|
self.case_dir = case_dir or get_case_study_dir(case_study)
|
|
self._db_path = self.case_dir / "run_log" / "registry.db"
|
|
if not self._db_path.exists():
|
|
raise FileNotFoundError(f"No registry.db found for '{case_study}' at {self._db_path}")
|
|
|
|
# -- helpers --
|
|
|
|
def _query(self, sql: str, params: tuple = ()) -> pl.DataFrame:
|
|
db = sqlite3.connect(str(self._db_path))
|
|
db.row_factory = sqlite3.Row
|
|
try:
|
|
rows = db.execute(sql, params).fetchall()
|
|
if not rows:
|
|
return pl.DataFrame()
|
|
return pl.DataFrame([dict(r) for r in rows])
|
|
finally:
|
|
db.close()
|
|
|
|
def _backtest_dir(self, b_hash: str) -> Path:
|
|
return self.case_dir / "run_log" / "backtest" / b_hash
|
|
|
|
def _filter_active_models(self, df: pl.DataFrame) -> pl.DataFrame:
|
|
return filter_active_model_rows(df, self.case_study)
|
|
|
|
# -----------------------------------------------------------------
|
|
# summary: what has been run?
|
|
# -----------------------------------------------------------------
|
|
|
|
def summary(self) -> dict[str, int]:
|
|
"""Count of backtest runs by stage.
|
|
|
|
Returns
|
|
-------
|
|
dict[str, int]
|
|
e.g. {"signal": 3336, "allocation": 329, ...}
|
|
"""
|
|
df = self._query(
|
|
"SELECT stage, COUNT(*) AS n FROM backtest_runs GROUP BY stage ORDER BY n DESC"
|
|
)
|
|
if df.is_empty():
|
|
return {}
|
|
return dict(zip(df["stage"].to_list(), df["n"].to_list(), strict=False))
|
|
|
|
# -----------------------------------------------------------------
|
|
# best: top backtests at a stage
|
|
# -----------------------------------------------------------------
|
|
|
|
def best(
|
|
self,
|
|
stage: str = "signal",
|
|
*,
|
|
top_n: int = 10,
|
|
metric: str = "sharpe",
|
|
) -> pl.DataFrame:
|
|
"""Top-N backtests at a given stage, ranked by ``metric``.
|
|
|
|
Returns
|
|
-------
|
|
pl.DataFrame
|
|
Columns: backtest_hash, prediction_hash, source, family,
|
|
config_name, label, signal_method, sharpe, cagr, max_drawdown,
|
|
total_return, volatility, ic_mean
|
|
"""
|
|
df = self._query(
|
|
f"""
|
|
SELECT
|
|
b.backtest_hash,
|
|
b.prediction_hash,
|
|
b.spec_json,
|
|
b.stage,
|
|
t.family,
|
|
t.config_name,
|
|
t.label,
|
|
bm.sharpe,
|
|
bm.cagr,
|
|
bm.max_drawdown,
|
|
bm.total_return,
|
|
bm.volatility,
|
|
pm.ic_mean
|
|
FROM backtest_runs b
|
|
JOIN prediction_sets p ON b.prediction_hash = p.prediction_hash
|
|
JOIN training_runs t ON p.training_hash = t.training_hash
|
|
LEFT JOIN backtest_metrics bm ON bm.backtest_hash = b.backtest_hash
|
|
LEFT JOIN prediction_metrics pm ON p.prediction_hash = pm.prediction_hash
|
|
WHERE b.stage = ?
|
|
AND p.split != 'holdout'
|
|
{excluded_family_sql(self.case_study, "t.family")[0]}
|
|
AND bm.sharpe IS NOT NULL
|
|
AND (bm.num_trades IS NULL OR bm.num_trades > 0)
|
|
ORDER BY bm.sharpe DESC
|
|
LIMIT ?
|
|
""",
|
|
(stage, *excluded_family_sql(self.case_study, "t.family")[1], top_n),
|
|
)
|
|
if df.is_empty():
|
|
return pl.DataFrame(schema=_BEST_SCHEMA)
|
|
df = self._filter_active_models(df)
|
|
if df.is_empty():
|
|
return pl.DataFrame(schema=_BEST_SCHEMA)
|
|
|
|
# Build source and extract signal_method from spec
|
|
df = df.with_columns(
|
|
(
|
|
pl.col("family") + pl.lit("/") + pl.col("config_name").fill_null(pl.lit("default"))
|
|
).alias("source"),
|
|
)
|
|
|
|
# Extract signal_method, universe_filter, exit_at_max_days from spec_json.
|
|
# The (universe_filter, exit_at_max_days) pair identifies the execution
|
|
# regime in the O'Donovan-Yu (2025) cost-mitigation cascade for
|
|
# sp500_options:
|
|
# - Rung-1 (naive round-trip): universe_filter="full", exit_at_max_days=10
|
|
# - Rung-2 (HTM, full): universe_filter="full", exit_at_max_days=None
|
|
# - Rung-3 (HTM, liquid q20): universe_filter="liquid", exit_at_max_days=None
|
|
# Both Rung-1 and Rung-2 carry universe_filter="full"; pinning the
|
|
# chapter-wide rank-1 to the HTM baseline therefore needs both fields.
|
|
# Other case studies default to ("full", None) and are unaffected.
|
|
parsed = [_parse_spec(s) or {} for s in df["spec_json"].to_list()]
|
|
methods = [strategy_view(sp).get("signal", {}).get("method", "") for sp in parsed]
|
|
universe_filters = [
|
|
strategy_view(sp).get("signal", {}).get("universe_filter", "full") or "full"
|
|
for sp in parsed
|
|
]
|
|
exit_at_max_days = [
|
|
strategy_view(sp).get("signal", {}).get("exit_at_max_days") for sp in parsed
|
|
]
|
|
df = df.with_columns(
|
|
pl.Series("signal_method", methods),
|
|
pl.Series("universe_filter", universe_filters),
|
|
pl.Series("exit_at_max_days", exit_at_max_days, dtype=pl.Int64),
|
|
)
|
|
|
|
return df.select(
|
|
"backtest_hash",
|
|
"prediction_hash",
|
|
"source",
|
|
"family",
|
|
"config_name",
|
|
"label",
|
|
"signal_method",
|
|
"universe_filter",
|
|
"exit_at_max_days",
|
|
"sharpe",
|
|
"cagr",
|
|
"max_drawdown",
|
|
"total_return",
|
|
"volatility",
|
|
"ic_mean",
|
|
)
|
|
|
|
# -----------------------------------------------------------------
|
|
# compare_families: model family comparison at a stage
|
|
# -----------------------------------------------------------------
|
|
|
|
def compare_families(self, stage: str = "signal") -> pl.DataFrame:
|
|
"""Compare model families by backtest Sharpe at a given stage.
|
|
|
|
Returns
|
|
-------
|
|
pl.DataFrame
|
|
Columns: family, n, sharpe_median, sharpe_max, sharpe_q75,
|
|
pct_positive
|
|
"""
|
|
df = self._query(
|
|
f"""
|
|
SELECT
|
|
t.family,
|
|
bm.sharpe
|
|
FROM backtest_metrics bm
|
|
JOIN backtest_runs b ON bm.backtest_hash = b.backtest_hash
|
|
JOIN prediction_sets p ON b.prediction_hash = p.prediction_hash
|
|
JOIN training_runs t ON p.training_hash = t.training_hash
|
|
WHERE b.stage = ?
|
|
AND p.split != 'holdout'
|
|
{excluded_family_sql(self.case_study, "t.family")[0]}
|
|
AND bm.sharpe IS NOT NULL
|
|
AND (bm.num_trades IS NULL OR bm.num_trades > 0)
|
|
""",
|
|
(stage, *excluded_family_sql(self.case_study, "t.family")[1]),
|
|
)
|
|
if df.is_empty():
|
|
return df
|
|
df = self._filter_active_models(df)
|
|
if df.is_empty():
|
|
return df
|
|
|
|
return (
|
|
df.group_by("family")
|
|
.agg(
|
|
n=pl.len(),
|
|
sharpe_median=pl.col("sharpe").median(),
|
|
sharpe_max=pl.col("sharpe").max(),
|
|
sharpe_q75=pl.col("sharpe").quantile(0.75),
|
|
pct_positive=((pl.col("sharpe") > 0).sum() / pl.len() * 100),
|
|
)
|
|
.sort("sharpe_median", descending=True)
|
|
)
|
|
|
|
# -----------------------------------------------------------------
|
|
# compare_allocators: allocation method comparison
|
|
# -----------------------------------------------------------------
|
|
|
|
def compare_allocators(
|
|
self,
|
|
*,
|
|
prediction_hash: str | None = None,
|
|
stages: tuple[str, ...] = ("allocation",),
|
|
) -> pl.DataFrame:
|
|
"""Compare allocation methods from the allocation stage.
|
|
|
|
Parameters
|
|
----------
|
|
prediction_hash : str, optional
|
|
If provided, restrict the comparison to backtests carrying this
|
|
prediction_hash (full or prefix match). Used by Ch20 to align the
|
|
allocator-heatmap pool to the spine rank-1 carrier so Figure 20.7
|
|
and Table 20.6 read off the same prediction.
|
|
stages : tuple of str, default ``("allocation",)``
|
|
Which backtest stages to include. Ch20 Figure 20.14 / Table 20.6
|
|
isolate the allocator layer and read off ``"allocation"`` only; the
|
|
risk overlay (ch19) is a downstream layer covered in §20.7, so
|
|
folding it in here would credit the allocator with the overlay's
|
|
work. Pass ``"risk_overlay"`` explicitly only for cross-stage views.
|
|
|
|
Returns
|
|
-------
|
|
pl.DataFrame
|
|
Columns: allocator, n, avg_sharpe, best_sharpe, avg_max_dd
|
|
"""
|
|
if not stages:
|
|
return pl.DataFrame()
|
|
placeholders = ", ".join("?" for _ in stages)
|
|
sql = f"""
|
|
SELECT
|
|
b.spec_json,
|
|
t.family,
|
|
t.config_name,
|
|
bm.sharpe,
|
|
bm.max_drawdown
|
|
FROM backtest_runs b
|
|
JOIN prediction_sets p ON b.prediction_hash = p.prediction_hash
|
|
JOIN training_runs t ON p.training_hash = t.training_hash
|
|
JOIN backtest_metrics bm ON bm.backtest_hash = b.backtest_hash
|
|
WHERE b.stage IN ({placeholders})
|
|
AND bm.sharpe IS NOT NULL
|
|
AND (bm.num_trades IS NULL OR bm.num_trades > 0)
|
|
"""
|
|
params: tuple = tuple(stages)
|
|
if prediction_hash:
|
|
sql += " AND b.prediction_hash LIKE ?"
|
|
params = (*params, prediction_hash + "%")
|
|
df = self._query(sql, params)
|
|
if df.is_empty():
|
|
return df
|
|
|
|
# Extract allocator from spec_json. Unparseable spec → "unknown";
|
|
# missing allocation key → "unknown" so risk_overlay rows whose
|
|
# spec carries only the risk overlay (no explicit allocator) are
|
|
# not silently bucketed under equal_weight (Ch20 Figure 20.7 / Table
|
|
# 20.6 pin allocator-method semantics to the spec, not to an engine
|
|
# default).
|
|
def _allocator_from(s: str | None) -> str:
|
|
spec = _parse_spec(s)
|
|
if spec is None:
|
|
return "unknown"
|
|
method = strategy_view(spec).get("allocation", {}).get("method")
|
|
return method if method else "unknown"
|
|
|
|
allocators = [_allocator_from(s) for s in df["spec_json"].to_list()]
|
|
df = df.with_columns(pl.Series("allocator", allocators))
|
|
df = df.filter(pl.col("allocator") != "unknown")
|
|
if df.is_empty():
|
|
return df
|
|
|
|
return (
|
|
df.group_by("allocator")
|
|
.agg(
|
|
n=pl.len(),
|
|
avg_sharpe=pl.col("sharpe").mean(),
|
|
best_sharpe=pl.col("sharpe").max(),
|
|
avg_max_dd=pl.col("max_drawdown").mean(),
|
|
)
|
|
.sort("avg_sharpe", descending=True)
|
|
)
|
|
|
|
# -----------------------------------------------------------------
|
|
# inspect: full detail for one backtest
|
|
# -----------------------------------------------------------------
|
|
|
|
def inspect(self, backtest_hash: str) -> BacktestDetail:
|
|
"""Load full details for a single backtest run.
|
|
|
|
Parameters
|
|
----------
|
|
backtest_hash : str
|
|
Full or prefix of the backtest hash.
|
|
|
|
Returns
|
|
-------
|
|
BacktestDetail
|
|
"""
|
|
# Support prefix matching
|
|
df = self._query(
|
|
"""
|
|
SELECT
|
|
b.backtest_hash,
|
|
b.prediction_hash,
|
|
b.stage,
|
|
b.spec_json,
|
|
t.family,
|
|
t.config_name
|
|
FROM backtest_runs b
|
|
JOIN prediction_sets p ON b.prediction_hash = p.prediction_hash
|
|
JOIN training_runs t ON p.training_hash = t.training_hash
|
|
WHERE b.backtest_hash LIKE ?
|
|
LIMIT 1
|
|
""",
|
|
(backtest_hash + "%",),
|
|
)
|
|
if df.is_empty():
|
|
raise KeyError(f"No backtest found matching '{backtest_hash}'")
|
|
|
|
row = df.row(0, named=True)
|
|
b_hash = row["backtest_hash"]
|
|
|
|
# Load metrics (wide format — each column is a metric)
|
|
metrics_df = self._query(
|
|
"SELECT * FROM backtest_metrics WHERE backtest_hash = ?",
|
|
(b_hash,),
|
|
)
|
|
metrics = {}
|
|
if not metrics_df.is_empty():
|
|
row_dict = metrics_df.row(0, named=True)
|
|
metrics = {
|
|
k: v
|
|
for k, v in row_dict.items()
|
|
if k not in ("backtest_hash", "computed_at") and v is not None
|
|
}
|
|
|
|
# Parse spec
|
|
spec = {}
|
|
if row["spec_json"]:
|
|
import contextlib
|
|
|
|
with contextlib.suppress(json.JSONDecodeError, TypeError):
|
|
spec = json.loads(row["spec_json"])
|
|
|
|
# File paths
|
|
bt_dir = self._backtest_dir(b_hash)
|
|
returns_path = bt_dir / "daily_returns.parquet"
|
|
trades_path = bt_dir / "trades.parquet"
|
|
weights_path = bt_dir / "weights.parquet"
|
|
|
|
source = None
|
|
if row.get("family"):
|
|
config = row.get("config_name") or "default"
|
|
source = f"{row['family']}/{config}"
|
|
|
|
return BacktestDetail(
|
|
backtest_hash=b_hash,
|
|
prediction_hash=row["prediction_hash"],
|
|
stage=row["stage"],
|
|
spec=spec,
|
|
metrics=metrics,
|
|
daily_returns_path=returns_path if returns_path.exists() else None,
|
|
trades_path=trades_path if trades_path.exists() else None,
|
|
weights_path=weights_path if weights_path.exists() else None,
|
|
source=source,
|
|
)
|
|
|
|
# -----------------------------------------------------------------
|
|
# progression: Sharpe across stages for a prediction
|
|
# -----------------------------------------------------------------
|
|
|
|
def progression(
|
|
self,
|
|
prediction_hash: str,
|
|
*,
|
|
universe_filter: str | None | object = _UNSET,
|
|
exit_at_max_days: int | None | object = _UNSET,
|
|
) -> pl.DataFrame:
|
|
"""Show Sharpe progression across stages for a given prediction.
|
|
|
|
Finds the best backtest at each stage for this prediction hash
|
|
and shows how performance changes as allocation, costs, and risk
|
|
overlays are added.
|
|
|
|
Parameters
|
|
----------
|
|
prediction_hash : str
|
|
Prediction set to trace through the pipeline.
|
|
universe_filter : str, None, or _UNSET, optional
|
|
If set to a string, restrict to backtests whose
|
|
``strategy.signal.universe_filter`` matches (defaulting null
|
|
spec entries to ``"full"`` so case studies without an explicit
|
|
universe_filter still match). If left at ``_UNSET`` (default),
|
|
no filter is applied. Used to scope sp500_options to its full
|
|
vs. liquid execution regime.
|
|
exit_at_max_days : int, None, or _UNSET, optional
|
|
If set to ``None`` explicitly, restrict to backtests whose
|
|
spec has no ``exit_at_max_days`` set (HTM regime). If set to
|
|
an integer, match exactly. If left at ``_UNSET`` (default),
|
|
no filter is applied. Together with ``universe_filter`` this
|
|
pins sp500_options to a specific cascade rung across all
|
|
stages, not just the signal stage.
|
|
|
|
Returns
|
|
-------
|
|
pl.DataFrame
|
|
Columns: stage, sharpe, cagr, max_drawdown, backtest_hash
|
|
"""
|
|
clauses = [
|
|
"b.prediction_hash = ?",
|
|
"b.stage IS NOT NULL",
|
|
"bm.sharpe IS NOT NULL",
|
|
"(bm.num_trades IS NULL OR bm.num_trades > 0)",
|
|
]
|
|
params: list[object] = [prediction_hash]
|
|
if universe_filter is not _UNSET:
|
|
clauses.append(
|
|
"COALESCE(json_extract(b.spec_json, '$.strategy.signal.universe_filter'), 'full') = ?"
|
|
)
|
|
params.append(universe_filter)
|
|
if exit_at_max_days is not _UNSET:
|
|
if exit_at_max_days is None:
|
|
clauses.append(
|
|
"json_extract(b.spec_json, '$.strategy.signal.exit_at_max_days') IS NULL"
|
|
)
|
|
else:
|
|
clauses.append(
|
|
"json_extract(b.spec_json, '$.strategy.signal.exit_at_max_days') = ?"
|
|
)
|
|
params.append(exit_at_max_days)
|
|
where_sql = " AND ".join(clauses)
|
|
df = self._query(
|
|
f"""
|
|
SELECT
|
|
b.stage,
|
|
b.backtest_hash,
|
|
bm.sharpe,
|
|
bm.cagr,
|
|
bm.max_drawdown
|
|
FROM backtest_runs b
|
|
JOIN backtest_metrics bm ON bm.backtest_hash = b.backtest_hash
|
|
WHERE {where_sql}
|
|
ORDER BY bm.sharpe DESC
|
|
""",
|
|
tuple(params),
|
|
)
|
|
if df.is_empty():
|
|
return df
|
|
|
|
# Take best Sharpe per stage
|
|
stage_order = {"signal": 0, "allocation": 1, "cost_sensitivity": 2, "risk_overlay": 3}
|
|
best_per_stage = df.sort("sharpe", descending=True).group_by("stage").first()
|
|
|
|
# Sort by pipeline order
|
|
return (
|
|
best_per_stage.with_columns(
|
|
pl.col("stage").replace_strict(stage_order, default=99).alias("_order")
|
|
)
|
|
.sort("_order")
|
|
.drop("_order")
|
|
)
|
|
|
|
# -----------------------------------------------------------------
|
|
# deflated_sharpe: DSR from registry data
|
|
# -----------------------------------------------------------------
|
|
|
|
def deflated_sharpe(
|
|
self,
|
|
stage: str = "signal",
|
|
*,
|
|
top_n: int = 20,
|
|
periods_per_year: int = 252,
|
|
) -> pl.DataFrame:
|
|
"""Per-variant Sharpe with selection-bias DSR for family leaders.
|
|
|
|
Per-variant PSR (single-strategy probability of skill, no
|
|
multiple-testing correction) is computed on the fly from
|
|
``daily_returns.parquet``.
|
|
|
|
Selection-bias DSR / RAS / Reality Check / PBO come from the
|
|
persisted ``cohort_metrics`` table (cohort_type='family',
|
|
leader_hash=backtest_hash). Backward-compatible columns
|
|
``deflated_sharpe``, ``expected_max_sharpe``, ``dsr_pvalue``,
|
|
``significant`` carry the **effective-rank (ER) DSR** — the
|
|
library maintainer's recommended default. ``dsr_mp`` and
|
|
``dsr_raw`` are surfaced alongside for sensitivity. Rows that
|
|
are not the family leader for their ``(stage, label, family)``
|
|
have NULL selection-bias columns.
|
|
|
|
Returns
|
|
-------
|
|
pl.DataFrame
|
|
Columns: source, sharpe, psr_pvalue, deflated_sharpe,
|
|
expected_max_sharpe, dsr_pvalue, significant, is_best,
|
|
dsr_mp, dsr_mp_pvalue, dsr_raw, dsr_raw_pvalue, k_variants,
|
|
n_trials_effective_er, n_trials_effective_mp, ras_leader,
|
|
ras_pvalue, reality_check_pvalue, pbo, family, label.
|
|
"""
|
|
from ml4t.diagnostic.evaluation.stats import deflated_sharpe_ratio
|
|
|
|
top = self.best(stage=stage, top_n=top_n)
|
|
if top.is_empty():
|
|
return pl.DataFrame()
|
|
|
|
per_variant_psr: dict[str, float | None] = {}
|
|
for row in top.iter_rows(named=True):
|
|
b_hash = row["backtest_hash"]
|
|
returns_path = self._backtest_dir(b_hash) / "daily_returns.parquet"
|
|
if not returns_path.exists():
|
|
per_variant_psr[b_hash] = None
|
|
continue
|
|
ret_df = pl.read_parquet(returns_path)
|
|
if "daily_return" not in ret_df.columns:
|
|
per_variant_psr[b_hash] = None
|
|
continue
|
|
arr = ret_df["daily_return"].to_numpy()
|
|
if np.std(arr, ddof=1) <= 1e-10:
|
|
per_variant_psr[b_hash] = None
|
|
continue
|
|
try:
|
|
psr = deflated_sharpe_ratio([arr], periods_per_year=periods_per_year)
|
|
per_variant_psr[b_hash] = float(psr.p_value)
|
|
except Exception: # pragma: no cover
|
|
per_variant_psr[b_hash] = None
|
|
|
|
hashes = top["backtest_hash"].to_list()
|
|
placeholders = ",".join("?" for _ in hashes)
|
|
cm = self._query(
|
|
f"""
|
|
SELECT leader_hash, k_variants,
|
|
n_trials_effective_mp, n_trials_effective_er,
|
|
dsr_raw, dsr_raw_pvalue,
|
|
dsr_mp, dsr_mp_pvalue,
|
|
dsr_er, dsr_er_pvalue, expected_max_sharpe_er,
|
|
ras_leader, ras_pvalue,
|
|
reality_check_pvalue, pbo
|
|
FROM cohort_metrics
|
|
WHERE cohort_type = 'family' AND stage = ?
|
|
AND leader_hash IN ({placeholders})
|
|
""",
|
|
(stage, *hashes),
|
|
)
|
|
cm_by_hash: dict[str, dict] = {}
|
|
if not cm.is_empty():
|
|
for r in cm.iter_rows(named=True):
|
|
cm_by_hash[r["leader_hash"]] = r
|
|
|
|
def _round(x, n=4):
|
|
return round(x, n) if x is not None else None
|
|
|
|
rows = []
|
|
for r in top.iter_rows(named=True):
|
|
b_hash = r["backtest_hash"]
|
|
cmr = cm_by_hash.get(b_hash)
|
|
is_leader = cmr is not None
|
|
dsr_er_p = cmr["dsr_er_pvalue"] if is_leader else None
|
|
rows.append(
|
|
{
|
|
"source": r["source"],
|
|
"family": r["family"],
|
|
"label": r["label"],
|
|
"sharpe": _round(r["sharpe"]),
|
|
"psr_pvalue": _round(per_variant_psr.get(b_hash)),
|
|
"deflated_sharpe": _round(cmr["dsr_er"]) if is_leader else None,
|
|
"expected_max_sharpe": _round(cmr["expected_max_sharpe_er"])
|
|
if is_leader
|
|
else None,
|
|
"dsr_pvalue": _round(dsr_er_p),
|
|
"significant": (dsr_er_p is not None and dsr_er_p < 0.05)
|
|
if is_leader
|
|
else None,
|
|
"is_best": is_leader,
|
|
"dsr_mp": _round(cmr["dsr_mp"]) if is_leader else None,
|
|
"dsr_mp_pvalue": _round(cmr["dsr_mp_pvalue"]) if is_leader else None,
|
|
"dsr_raw": _round(cmr["dsr_raw"]) if is_leader else None,
|
|
"dsr_raw_pvalue": _round(cmr["dsr_raw_pvalue"]) if is_leader else None,
|
|
"k_variants": cmr["k_variants"] if is_leader else None,
|
|
"n_trials_effective_er": _round(cmr["n_trials_effective_er"], 1)
|
|
if is_leader
|
|
else None,
|
|
"n_trials_effective_mp": _round(cmr["n_trials_effective_mp"], 1)
|
|
if is_leader
|
|
else None,
|
|
"ras_leader": _round(cmr["ras_leader"]) if is_leader else None,
|
|
"ras_pvalue": _round(cmr["ras_pvalue"]) if is_leader else None,
|
|
"reality_check_pvalue": _round(cmr["reality_check_pvalue"])
|
|
if is_leader
|
|
else None,
|
|
"pbo": _round(cmr["pbo"]) if is_leader else None,
|
|
}
|
|
)
|
|
return pl.DataFrame(rows).sort("sharpe", descending=True, nulls_last=True)
|
|
|
|
# -----------------------------------------------------------------
|
|
# cost_sensitivity: breakeven analysis from registry
|
|
# -----------------------------------------------------------------
|
|
|
|
def cost_sensitivity(self, *, prediction_hash: str | None = None) -> pl.DataFrame:
|
|
"""Load cost sensitivity results from the cost_sensitivity stage.
|
|
|
|
Only the bps (``commission.model='percentage'``) regime is returned;
|
|
per-share rows have ``commission.rate=0`` and ``slippage.rate=0`` so
|
|
their derived ``cost_bps`` is mechanically 0 and would otherwise pile
|
|
up on the bps-axis origin. Notebooks rendering both regimes must
|
|
query the registry directly (see ``etfs/16_costs.py::load_cost_rows``
|
|
for the pattern).
|
|
|
|
Parameters
|
|
----------
|
|
prediction_hash : str, optional
|
|
When provided, restrict to cost rows on this prediction. Case
|
|
studies with a pinned carrier (e.g. nasdaq's cost-feasible
|
|
ensemble) must scope to the carrier so the full-universe
|
|
cost-defeat demonstration rows do not pool into the headline.
|
|
|
|
Returns
|
|
-------
|
|
pl.DataFrame
|
|
Columns: cost_bps, sharpe, max_drawdown, allocator
|
|
"""
|
|
pred_clause = "" if prediction_hash is None else " AND b.prediction_hash = ?"
|
|
params = () if prediction_hash is None else (prediction_hash,)
|
|
df = self._query(
|
|
f"""
|
|
SELECT
|
|
b.spec_json,
|
|
bm.sharpe,
|
|
bm.max_drawdown
|
|
FROM backtest_runs b
|
|
JOIN backtest_metrics bm ON bm.backtest_hash = b.backtest_hash
|
|
WHERE b.stage = 'cost_sensitivity'
|
|
AND bm.sharpe IS NOT NULL
|
|
AND (bm.num_trades IS NULL OR bm.num_trades > 0)
|
|
AND json_extract(b.spec_json, '$.backtest_config.commission.model') = 'percentage'
|
|
{pred_clause}
|
|
""",
|
|
params,
|
|
)
|
|
if df.is_empty():
|
|
return df
|
|
df = self._filter_active_models(df)
|
|
if df.is_empty():
|
|
return df
|
|
|
|
# Extract cost_bps and allocator from spec
|
|
rows = []
|
|
for spec_str, sharpe, max_dd in zip(
|
|
df["spec_json"].to_list(),
|
|
df["sharpe"].to_list(),
|
|
df["max_drawdown"].to_list(),
|
|
strict=False,
|
|
):
|
|
spec = _parse_spec(spec_str) or {}
|
|
costs = cost_view(spec)
|
|
cost_bps = costs.get("commission_bps", 0) + costs.get("slippage_bps", 0)
|
|
allocator = strategy_view(spec).get("allocation", {}).get("method", "equal_weight")
|
|
rows.append(
|
|
{
|
|
"cost_bps": cost_bps,
|
|
"sharpe": sharpe,
|
|
"max_drawdown": max_dd,
|
|
"allocator": allocator,
|
|
}
|
|
)
|
|
|
|
return pl.DataFrame(rows).sort("cost_bps")
|
|
|
|
# -----------------------------------------------------------------
|
|
# risk_impact: risk overlay comparison from registry
|
|
# -----------------------------------------------------------------
|
|
|
|
def risk_impact(self, *, prediction_hash: str | None = None) -> pl.DataFrame:
|
|
"""Load risk overlay results and compute impact vs baseline.
|
|
|
|
Parameters
|
|
----------
|
|
prediction_hash : str, optional
|
|
When provided, restrict to risk-overlay rows on this prediction.
|
|
Case studies with a pinned carrier (e.g. nasdaq's cost-feasible
|
|
ensemble) must scope to the carrier so the full-universe overlay
|
|
demonstration rows do not pool into the headline.
|
|
|
|
Returns
|
|
-------
|
|
pl.DataFrame
|
|
Columns: risk_name, risk_type, sharpe, max_drawdown,
|
|
num_trades, baseline_sharpe, sharpe_delta
|
|
"""
|
|
pred_clause = "" if prediction_hash is None else " AND b.prediction_hash = ?"
|
|
pred_params = () if prediction_hash is None else (prediction_hash,)
|
|
df = self._query(
|
|
f"""
|
|
SELECT
|
|
b.spec_json,
|
|
t.family,
|
|
t.config_name,
|
|
bm.sharpe,
|
|
bm.max_drawdown,
|
|
bm.num_trades
|
|
FROM backtest_runs b
|
|
JOIN prediction_sets p ON b.prediction_hash = p.prediction_hash
|
|
JOIN training_runs t ON p.training_hash = t.training_hash
|
|
JOIN backtest_metrics bm ON bm.backtest_hash = b.backtest_hash
|
|
WHERE b.stage = 'risk_overlay'
|
|
{excluded_family_sql(self.case_study, "t.family")[0]}
|
|
AND bm.sharpe IS NOT NULL
|
|
AND (bm.num_trades IS NULL OR bm.num_trades > 0)
|
|
{pred_clause}
|
|
""",
|
|
tuple(excluded_family_sql(self.case_study, "t.family")[1]) + pred_params,
|
|
)
|
|
if df.is_empty():
|
|
return df
|
|
df = self._filter_active_models(df)
|
|
if df.is_empty():
|
|
return df
|
|
|
|
rows = []
|
|
for spec_str, sharpe, max_dd, trades in zip(
|
|
df["spec_json"].to_list(),
|
|
df["sharpe"].to_list(),
|
|
df["max_drawdown"].to_list(),
|
|
df["num_trades"].to_list(),
|
|
strict=False,
|
|
):
|
|
spec = _parse_spec(spec_str) or {}
|
|
risk = strategy_view(spec).get("risk", {})
|
|
risk_name = risk.get("name", "unknown")
|
|
|
|
# Determine risk type
|
|
pos_rules = risk.get("position_rules", [])
|
|
port_limits = risk.get("portfolio_limits", [])
|
|
if risk_name == "baseline":
|
|
risk_type = "baseline"
|
|
elif pos_rules:
|
|
risk_type = pos_rules[0].get("type", "unknown")
|
|
elif port_limits:
|
|
risk_type = port_limits[0].get("type", "unknown")
|
|
else:
|
|
risk_type = "unknown"
|
|
|
|
rows.append(
|
|
{
|
|
"risk_name": risk_name,
|
|
"risk_type": risk_type,
|
|
"sharpe": sharpe,
|
|
"max_drawdown": max_dd,
|
|
"num_trades": trades,
|
|
}
|
|
)
|
|
|
|
result = pl.DataFrame(rows)
|
|
|
|
# Compute baseline: the no-overlay Sharpe the overlays are measured
|
|
# against. Registry-wide (unpinned) this is the best allocation-stage
|
|
# Sharpe — the normal pipeline where overlays sit on an allocator. When
|
|
# the comparison is pinned to a carrier prediction (e.g. nasdaq's
|
|
# signal-stage slot ensemble, which has no allocation stage), the
|
|
# baseline is the carrier's own no-overlay Sharpe over its signal and
|
|
# allocation rows; the registry-wide allocation max would otherwise
|
|
# return an unrelated full-universe strategy.
|
|
baseline_stage_clause = (
|
|
"b.stage = 'allocation'"
|
|
if prediction_hash is None
|
|
else "b.stage IN ('signal', 'allocation')"
|
|
)
|
|
baseline_df = self._query(
|
|
f"""
|
|
SELECT
|
|
bm.sharpe,
|
|
t.family,
|
|
t.config_name
|
|
FROM backtest_metrics bm
|
|
JOIN backtest_runs b ON bm.backtest_hash = b.backtest_hash
|
|
JOIN prediction_sets p ON b.prediction_hash = p.prediction_hash
|
|
JOIN training_runs t ON p.training_hash = t.training_hash
|
|
WHERE {baseline_stage_clause}
|
|
{excluded_family_sql(self.case_study, "t.family")[0]}
|
|
{pred_clause}
|
|
""",
|
|
tuple(excluded_family_sql(self.case_study, "t.family")[1]) + pred_params,
|
|
)
|
|
baseline_df = self._filter_active_models(baseline_df)
|
|
baseline_sharpe = baseline_df["sharpe"].max() if not baseline_df.is_empty() else None
|
|
|
|
if baseline_sharpe is not None:
|
|
result = result.with_columns(
|
|
pl.lit(baseline_sharpe).alias("baseline_sharpe"),
|
|
(pl.col("sharpe") - baseline_sharpe).alias("sharpe_delta"),
|
|
)
|
|
else:
|
|
result = result.with_columns(
|
|
pl.lit(None).cast(pl.Float64).alias("baseline_sharpe"),
|
|
pl.lit(None).cast(pl.Float64).alias("sharpe_delta"),
|
|
)
|
|
|
|
return result.sort("sharpe", descending=True)
|
|
|
|
# -----------------------------------------------------------------
|
|
# fold_performance: per-fold backtest metrics
|
|
# -----------------------------------------------------------------
|
|
|
|
def fold_performance(self, backtest_hash: str) -> pl.DataFrame:
|
|
"""Per-fold backtest metrics (Sharpe, max_dd, etc.) for one backtest.
|
|
|
|
Parameters
|
|
----------
|
|
backtest_hash : str
|
|
Full or prefix of the backtest hash.
|
|
|
|
Returns
|
|
-------
|
|
pl.DataFrame
|
|
Columns: fold_id, sharpe, cagr, max_drawdown, volatility,
|
|
total_return, n_days, ...
|
|
"""
|
|
df = self._query(
|
|
"""
|
|
SELECT *
|
|
FROM backtest_fold_metrics
|
|
WHERE backtest_hash LIKE ?
|
|
ORDER BY fold_id
|
|
""",
|
|
(backtest_hash + "%",),
|
|
)
|
|
if df.is_empty():
|
|
return df
|
|
|
|
# Drop internal columns, already in wide format
|
|
drop_cols = [c for c in ["backtest_hash", "computed_at"] if c in df.columns]
|
|
if drop_cols:
|
|
df = df.drop(drop_cols)
|
|
return df.sort("fold_id")
|
|
|
|
# -----------------------------------------------------------------
|
|
# ic_sharpe_scatter: IC vs backtest Sharpe per fold
|
|
# -----------------------------------------------------------------
|
|
|
|
def ic_sharpe_scatter(
|
|
self,
|
|
stage: str = "signal",
|
|
*,
|
|
top_n: int = 10,
|
|
) -> pl.DataFrame:
|
|
"""Join prediction IC per fold with backtest Sharpe per fold.
|
|
|
|
This enables the empirical fundamental law test: plotting IC
|
|
against realized Sharpe at the fold level to see whether
|
|
better predictions produce better portfolio returns.
|
|
|
|
Parameters
|
|
----------
|
|
stage : str
|
|
Pipeline stage to filter backtests.
|
|
top_n : int
|
|
Number of top backtests (by headline Sharpe) to include.
|
|
|
|
Returns
|
|
-------
|
|
pl.DataFrame
|
|
Columns: source, fold_id, ic, sharpe, cagr, max_drawdown
|
|
"""
|
|
# Get top backtests at this stage
|
|
top = self.best(stage=stage, top_n=top_n)
|
|
if top.is_empty():
|
|
return pl.DataFrame()
|
|
|
|
rows = []
|
|
for row in top.iter_rows(named=True):
|
|
b_hash = row["backtest_hash"]
|
|
p_hash = row["prediction_hash"]
|
|
source = row.get("source", b_hash[:8])
|
|
|
|
# Get backtest fold metrics (wide format)
|
|
bt_wide = self._query(
|
|
"""
|
|
SELECT *
|
|
FROM backtest_fold_metrics
|
|
WHERE backtest_hash = ?
|
|
""",
|
|
(b_hash,),
|
|
)
|
|
if bt_wide.is_empty():
|
|
continue
|
|
|
|
# Drop internal columns
|
|
drop_cols = [c for c in ["backtest_hash", "computed_at"] if c in bt_wide.columns]
|
|
if drop_cols:
|
|
bt_wide = bt_wide.drop(drop_cols)
|
|
|
|
# Get prediction fold metrics (IC) — wide format
|
|
pred_folds = self._query(
|
|
"""
|
|
SELECT fold_id, ic
|
|
FROM fold_metrics
|
|
WHERE prediction_hash = ?
|
|
""",
|
|
(p_hash,),
|
|
)
|
|
|
|
if pred_folds.is_empty():
|
|
continue
|
|
|
|
ic_df = pred_folds.select("fold_id", "ic")
|
|
|
|
joined = bt_wide.join(ic_df, on="fold_id", how="left")
|
|
joined = joined.with_columns(pl.lit(source).alias("source"))
|
|
rows.append(joined)
|
|
|
|
if not rows:
|
|
return pl.DataFrame()
|
|
|
|
result = pl.concat(rows, how="diagonal")
|
|
|
|
# Select core columns (others available but less critical)
|
|
cols = ["source", "fold_id"]
|
|
for c in ["ic", "sharpe", "cagr", "max_drawdown", "volatility", "n_days"]:
|
|
if c in result.columns:
|
|
cols.append(c)
|
|
|
|
return result.select(cols).sort("source", "fold_id")
|
|
|
|
# -----------------------------------------------------------------
|
|
# backfill_fold_metrics: compute fold metrics for existing backtests
|
|
# -----------------------------------------------------------------
|
|
|
|
def backfill_fold_metrics(
|
|
self,
|
|
stage: str = "signal",
|
|
*,
|
|
label: str = "",
|
|
limit: int = 0,
|
|
) -> int:
|
|
"""Compute and store fold metrics for existing backtests.
|
|
|
|
Finds backtests at the given stage that have daily_returns.parquet
|
|
but no entries in backtest_fold_metrics, then computes and registers
|
|
fold-level performance metrics.
|
|
|
|
Parameters
|
|
----------
|
|
stage : str
|
|
Pipeline stage to backfill.
|
|
label : str
|
|
Label name for fold boundary computation.
|
|
limit : int
|
|
Max backtests to process (0 = all).
|
|
|
|
Returns
|
|
-------
|
|
int
|
|
Number of backtests backfilled.
|
|
"""
|
|
from case_studies.utils.registry import (
|
|
compute_backtest_fold_metrics,
|
|
register_backtest_fold_metrics,
|
|
)
|
|
|
|
# Find backtests without fold metrics
|
|
df = self._query(
|
|
"""
|
|
SELECT b.backtest_hash
|
|
FROM backtest_runs b
|
|
WHERE b.stage = ?
|
|
AND b.backtest_hash NOT IN (
|
|
SELECT DISTINCT backtest_hash FROM backtest_fold_metrics
|
|
)
|
|
""",
|
|
(stage,),
|
|
)
|
|
if df.is_empty():
|
|
return 0
|
|
|
|
hashes = df["backtest_hash"].to_list()
|
|
|
|
count = 0
|
|
for b_hash in hashes:
|
|
if limit > 0 and count >= limit:
|
|
break
|
|
|
|
returns_path = self._backtest_dir(b_hash) / "daily_returns.parquet"
|
|
if not returns_path.exists():
|
|
continue
|
|
|
|
daily_ret = pl.read_parquet(returns_path)
|
|
fold_m = compute_backtest_fold_metrics(daily_ret, self.case_study, label=label)
|
|
if fold_m:
|
|
register_backtest_fold_metrics(self.case_study, b_hash, fold_m)
|
|
count += 1
|
|
|
|
return count
|
|
|
|
# -----------------------------------------------------------------
|
|
# search_context: distribution stats for a stage
|
|
# -----------------------------------------------------------------
|
|
|
|
def search_context(self, stage: str = "signal") -> dict[str, Any]:
|
|
"""Distribution statistics for all backtests at a stage.
|
|
|
|
Quantifies search risk: how exceptional is the champion relative
|
|
to the full sweep?
|
|
|
|
Returns
|
|
-------
|
|
dict
|
|
Keys: total, median_sharpe, mean_sharpe, p90_sharpe,
|
|
champion_sharpe, champion_source, champion_percentile,
|
|
pct_positive
|
|
"""
|
|
df = self._query(
|
|
"""
|
|
SELECT
|
|
bm.sharpe,
|
|
t.family || '/' || COALESCE(t.config_name, 'default') AS source
|
|
FROM backtest_metrics bm
|
|
JOIN backtest_runs b ON bm.backtest_hash = b.backtest_hash
|
|
JOIN prediction_sets p ON b.prediction_hash = p.prediction_hash
|
|
JOIN training_runs t ON p.training_hash = t.training_hash
|
|
WHERE b.stage = ?
|
|
AND p.split != 'holdout'
|
|
AND bm.sharpe IS NOT NULL
|
|
AND (bm.num_trades IS NULL OR bm.num_trades > 0)
|
|
""",
|
|
(stage,),
|
|
)
|
|
if df.is_empty():
|
|
return {}
|
|
|
|
sharpes = df["sharpe"].to_numpy()
|
|
best_idx = int(np.argmax(sharpes))
|
|
|
|
return {
|
|
"total": len(sharpes),
|
|
"median_sharpe": float(np.median(sharpes)),
|
|
"mean_sharpe": float(np.mean(sharpes)),
|
|
"p90_sharpe": float(np.percentile(sharpes, 90)),
|
|
"champion_sharpe": float(sharpes[best_idx]),
|
|
"champion_source": df["source"][best_idx],
|
|
"champion_percentile": float((sharpes <= sharpes[best_idx]).sum() / len(sharpes) * 100),
|
|
"pct_positive": float((sharpes > 0).sum() / len(sharpes) * 100),
|
|
}
|
|
|
|
# -----------------------------------------------------------------
|
|
# champion_lineage: locked path through all stages
|
|
# -----------------------------------------------------------------
|
|
|
|
def champion_lineage(self, prediction_hash: str) -> dict[str, dict]:
|
|
"""Locked path through signal -> allocation -> cost -> risk.
|
|
|
|
For each stage, returns the BEST backtest for this specific
|
|
``prediction_hash`` with spec annotations (allocator, top_k,
|
|
cost_bps, risk_type).
|
|
|
|
Returns
|
|
-------
|
|
dict[str, dict]
|
|
Keyed by stage. Each value has: sharpe, cagr, max_drawdown,
|
|
backtest_hash, plus stage-specific fields.
|
|
"""
|
|
df = self._query(
|
|
"""
|
|
SELECT
|
|
b.stage,
|
|
b.backtest_hash,
|
|
b.spec_json,
|
|
bm.sharpe,
|
|
bm.cagr,
|
|
bm.max_drawdown,
|
|
bm.volatility,
|
|
bm.total_return
|
|
FROM backtest_runs b
|
|
JOIN backtest_metrics bm ON bm.backtest_hash = b.backtest_hash
|
|
WHERE b.prediction_hash = ?
|
|
AND b.stage IS NOT NULL
|
|
AND bm.sharpe IS NOT NULL
|
|
AND (bm.num_trades IS NULL OR bm.num_trades > 0)
|
|
ORDER BY bm.sharpe DESC
|
|
""",
|
|
(prediction_hash,),
|
|
)
|
|
if df.is_empty():
|
|
return {}
|
|
|
|
result: dict[str, dict] = {}
|
|
stage_order = ["signal", "allocation", "cost_sensitivity", "risk_overlay"]
|
|
|
|
for stage_name in stage_order:
|
|
stage_df = df.filter(pl.col("stage") == stage_name)
|
|
if stage_df.is_empty():
|
|
continue
|
|
|
|
row = stage_df.row(0, named=True)
|
|
entry: dict[str, Any] = {
|
|
"sharpe": row["sharpe"],
|
|
"cagr": row["cagr"],
|
|
"max_drawdown": row["max_drawdown"],
|
|
"volatility": row["volatility"],
|
|
"total_return": row["total_return"],
|
|
"backtest_hash": row["backtest_hash"],
|
|
}
|
|
|
|
# Extract stage-specific annotations from spec_json
|
|
spec = {}
|
|
if row["spec_json"]:
|
|
import contextlib
|
|
|
|
with contextlib.suppress(json.JSONDecodeError, TypeError):
|
|
spec = json.loads(row["spec_json"])
|
|
strategy = strategy_view(spec)
|
|
|
|
if stage_name == "signal":
|
|
entry["signal_method"] = strategy.get("signal", {}).get("method", "")
|
|
entry["top_k"] = strategy.get("signal", {}).get("top_k", None)
|
|
elif stage_name == "allocation":
|
|
entry["allocator"] = strategy.get("allocation", {}).get("method", "")
|
|
entry["top_k"] = strategy.get("signal", {}).get("top_k", None)
|
|
elif stage_name == "cost_sensitivity":
|
|
costs = cost_view(spec)
|
|
entry["cost_bps"] = costs.get("commission_bps", 0) + costs.get("slippage_bps", 0)
|
|
entry["allocator"] = strategy.get("allocation", {}).get("method", "")
|
|
elif stage_name == "risk_overlay":
|
|
risk = strategy.get("risk", {})
|
|
entry["risk_name"] = risk.get("name", "")
|
|
pos_rules = risk.get("position_rules", [])
|
|
entry["risk_type"] = pos_rules[0].get("type", "") if pos_rules else ""
|
|
|
|
result[stage_name] = entry
|
|
|
|
return result
|
|
|
|
# -----------------------------------------------------------------
|
|
# concentration_curve: Sharpe vs top_k at allocation stage
|
|
# -----------------------------------------------------------------
|
|
|
|
def concentration_curve(self, prediction_hash: str) -> pl.DataFrame:
|
|
"""Sharpe vs top_k for a given prediction at allocation stage.
|
|
|
|
Shows how portfolio concentration affects performance — typically
|
|
more actionable than allocator comparison alone.
|
|
|
|
Returns
|
|
-------
|
|
pl.DataFrame
|
|
Columns: top_k, allocator, sharpe, max_drawdown, cagr
|
|
"""
|
|
df = self._query(
|
|
"""
|
|
SELECT
|
|
b.spec_json,
|
|
bm.sharpe,
|
|
bm.max_drawdown,
|
|
bm.cagr
|
|
FROM backtest_runs b
|
|
JOIN backtest_metrics bm ON bm.backtest_hash = b.backtest_hash
|
|
WHERE b.prediction_hash = ?
|
|
AND b.stage = 'allocation'
|
|
AND bm.sharpe IS NOT NULL
|
|
AND (bm.num_trades IS NULL OR bm.num_trades > 0)
|
|
""",
|
|
(prediction_hash,),
|
|
)
|
|
if df.is_empty():
|
|
return df
|
|
|
|
rows = []
|
|
for spec_str, sharpe, max_dd, cagr in zip(
|
|
df["spec_json"].to_list(),
|
|
df["sharpe"].to_list(),
|
|
df["max_drawdown"].to_list(),
|
|
df["cagr"].to_list(),
|
|
strict=False,
|
|
):
|
|
spec = _parse_spec(spec_str) or {}
|
|
strategy = strategy_view(spec)
|
|
top_k = strategy.get("signal", {}).get("top_k", None)
|
|
allocator = strategy.get("allocation", {}).get("method", "equal_weight")
|
|
rows.append(
|
|
{
|
|
"top_k": top_k,
|
|
"allocator": allocator,
|
|
"sharpe": sharpe,
|
|
"max_drawdown": max_dd,
|
|
"cagr": cagr,
|
|
}
|
|
)
|
|
|
|
return pl.DataFrame(rows).sort("top_k")
|
|
|
|
# -----------------------------------------------------------------
|
|
# repr
|
|
# -----------------------------------------------------------------
|
|
|
|
def __repr__(self) -> str:
|
|
counts = self.summary()
|
|
total = sum(counts.values())
|
|
parts = ", ".join(f"{k}={v}" for k, v in sorted(counts.items()))
|
|
return f"BacktestExplorer('{self.case_study}', {total} runs: {parts})"
|