Explora y compara resultados de backtests
Código Machine Learning for Trading
Resumen
Este documento describe una interfaz de usuario para examinar resultados de backtests almacenados en un registro. Permite resumir el número de ejecuciones, inspeccionar especificaciones y ejecuciones individuales, ordenar resultados, comparar métodos de asignación, consultar curvas de concentración y seguir la evolución de las estrategias. La interfaz lee la base de datos del registro y devuelve tablas estructuradas y registros detallados para que los cuadernos de investigación puedan explorar ejecuciones anteriores sin depender de archivos de resultados separados.
La implementación aborda problemas de interpretación que pueden distorsionar una clasificación: se agrupan las filas duplicadas con identificadores distintos pero los mismos valores visibles para el lector, y se incluyen dimensiones clave de la estrategia, como el punto de control y la concentración, para poder distinguir configuraciones realmente diferentes. También usa esquemas explícitos para los resultados de consultas vacíos y gestiona especificaciones ausentes o mal formadas, reduciendo los errores confusos durante la exploración. Son salvaguardas para el acceso y la presentación de datos, no un método para seleccionar estrategias rentables. Las clasificaciones siguen reflejando los experimentos y métricas ya registrados, y los resultados requieren un análisis adecuado fuera de muestra y una evaluación de la incertidumbre antes de respaldar conclusiones.
Ideas clave
- El explorador ofrece una interfaz uniforme de solo lectura para las ejecuciones de backtests registradas y sus especificaciones.
- Las filas de la clasificación deben mostrar las dimensiones de la estrategia que explican las diferencias entre configuraciones.
- Las filas idénticas en todos los campos mostrados no necesitan ocupar varias posiciones en una tabla ordenada.
- Los esquemas explícitos facilitan la gestión de resultados de consultas vacíos en análisis posteriores.
- Explorar el registro organiza la evidencia existente, pero no demuestra que una estrategia vaya a rendir bien fuera de muestra.
Etiquetas
Texto completo
# backtest_explorer.py
```py
"""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 collections.abc import Iterable
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,
full_coverage_prediction_sql,
)
from case_studies.utils.sweep_config import top_n_cap
from case_studies.utils.uncertainty import STAGE_SEQUENCE, cohort_member_digest
# 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,
# A configuration publishes a prediction set per checkpoint, and those
# checkpoints rank separately, so without these two a ten-row table can print
# one configuration six times at six Sharpes with nothing saying what differs.
# Measured on us_firm_characteristics: six of the ten best signal-stage rows
# for fwd_ret_1m were gbm/leaves_7_mse at top_k 50, iterations 200 to 450.
"checkpoint_kind": pl.Utf8,
"checkpoint_value": pl.Int64,
"label": pl.Utf8,
"signal_method": pl.Utf8,
# The entry-scheme sweep varies concentration and nothing else, so without
# this the ten-row top table reads as one strategy repeated at different
# Sharpes.
"top_k": pl.Int64,
"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,
"ic_mean_daily": pl.Float64,
"ic_ci_lo": pl.Float64,
"ic_ci_hi": pl.Float64,
"ic_n_days": pl.Float64,
}
def _drop_rows_a_reader_cannot_tell_apart(df: pl.DataFrame) -> pl.DataFrame:
"""Collapse leaderboard rows that are identical in every column but the hash.
A configuration keeps its results when a spec field is added that it never set:
the identity is new, the numbers are not. Measured 2026-09-14 across the nine
registries, keying on prediction and on the strategy view a reader sees:
``us_firm_characteristics`` holds 80 of 240 allocation configurations and 912 of
2,276 signal configurations twice, ``etfs`` 424 of 1,767 and
``crypto_perps_funding`` 273 of 2,109 - 1,689 pairs in total, every pair agreeing
on Sharpe. One pair diffed: the two specs differ in
``account.lock_notional_update_mode`` unset against ``position_legs`` and
``position_sizing.share_rounding`` unset against ``nearest``, both previously
implicit defaults that the config schema made explicit. So a schema change
re-keyed the identity and moved nothing measurable.
``best`` ranks over every generation in the registry, so both rows compete and a
ten-row table can show five configurations. This is the same defect
``signal_method`` alone had, one level down: there the
displayed columns could not separate rows that genuinely differed, here the rows
do not differ at all.
The rule is the narrowest one that fixes it: drop a row only when every column a
caller receives, except ``backtest_hash``, equals one already kept. Such a row
carries nothing a reader could have read off it, so no information is lost, and
the first row of any group survives under the query's existing
``sharpe DESC, backtest_hash ASC`` ordering - which is why nothing that selects on
``best`` can change its answer, only the repeats below it disappear.
"""
return df.unique(
subset=[c for c in df.columns if c != "backtest_hash"],
keep="first",
maintain_order=True,
)
# Canonical schema for BacktestExplorer.specs() output. Declared rather than inferred:
# polars types an empty column as Null, and `.str.json_path_match` on a Null series raises
# SchemaError instead of returning an empty result. A notebook whose registry holds no rows
# for the stage it asks about would then fail on the dtype, several cells after the fact,
# reporting a schema problem for what is actually an empty sweep
# .
_SPEC_SCHEMA: dict[str, pl.DataType] = {
"backtest_hash": pl.Utf8,
"stage": pl.Utf8,
"spec_json": pl.Utf8,
}
# ---------------------------------------------------------------------------
# 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()
# Scan every row for the schema, not the default first 100. A metric added after a
# registry was first written is NULL on every earlier row, so a query returning more
# than 100 of those before the first real value types the column `Null` and then
# raises `could not append value: 0.0 of type: f64` on it. `ruin` did exactly that to
# `etfs` once #834 started recording it: 180 NULLs, then a float. The cost is one
# extra pass over rows already in memory.
return pl.DataFrame([dict(r) for r in rows], infer_schema_length=None)
finally:
db.close()
def _has_metric_column(self, column: str) -> bool:
"""Whether ``backtest_metrics`` carries this column in this registry.
A registry written before a metric existed has no column for it, and
selecting one that is not there is a hard SQLite error rather than a NULL.
"""
db = sqlite3.connect(str(self._db_path))
try:
return column in {
row[1] for row in db.execute("PRAGMA table_info(backtest_metrics)").fetchall()
}
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))
def specs(self, stages: str | Iterable[str]) -> pl.DataFrame:
"""Every backtest hash and its spec JSON for ``stages``, schema-stable when empty.
``best()`` reports the metrics and the model that produced a run, not the strategy
dimensions a sweep varies - the entry scheme's ``top_k``, the allocator. Those live
in the spec, so a notebook that varies them reads them back here and joins on
``backtest_hash`` rather than opening ``registry.db`` itself.
The columns of ``_SPEC_SCHEMA`` are returned whether or not any row matched, so a
caller's ``json_path_match`` sees String and yields an empty frame rather than
raising on a Null column.
"""
wanted = [stages] if isinstance(stages, str) else list(stages)
if not wanted:
return pl.DataFrame(schema=_SPEC_SCHEMA)
placeholders = ", ".join("?" * len(wanted))
df = self._query(
"SELECT backtest_hash, stage, spec_json FROM backtest_runs "
f"WHERE stage IN ({placeholders})",
tuple(wanted),
)
if df.is_empty():
return pl.DataFrame(schema=_SPEC_SCHEMA)
return df.select(
[pl.col(name).cast(dtype).alias(name) for name, dtype in _SPEC_SCHEMA.items()]
)
# -----------------------------------------------------------------
# best: top backtests at a stage
# -----------------------------------------------------------------
def _refuse_if_the_coverage_bar_emptied_it(
self,
stage: str,
label: str | None,
prediction_hashes: tuple[str, ...] | list[str] | None,
) -> None:
"""Raise when backtests exist at this stage but none of them is a complete run.
`full_coverage_prediction_sql` keeps only rows whose `ic_n_days` equals the maximum
for their `(split, family, label)`, and that maximum is the population's, not the
backtested subset's. That is deliberate: the bar is how a *complete* run is
identified - `reference/CASE_STUDY_PIPELINE.md` section 10, "a result counts only
when `n_null == 0` and `ic_n_days == max`" - and it exists because an incomplete run
manufactures a false leader. Computing the maximum over only the rows that happen to
have backtests would let an incomplete run set its own bar and win whenever no
complete run was backtested, which is the exact failure the clause was written for.
So an empty result here is a true statement: no complete run in this population has
a backtest at this stage. It is not, however, the same statement as "there are no
backtests", and returning an empty frame says the second. Under a preview reduction
the two diverge - etfs measured a scoped population of 6 predictions and 1 backtest
whose maximum coverage belonged to a prediction nothing backtested - and a notebook
that reads the empty frame as "nothing ran" reports on a comparison that was never
possible. Refuse instead, and name the gap.
"""
# Every filter the real query applies EXCEPT the coverage clause, so an empty
# result there and a non-empty result here isolates the coverage clause as the
# only thing that removed the rows. Dropping one - the family exclusions, or the
# zero-trade filter - would report a run excluded for another reason as a coverage
# failure and send the reader after the wrong thing. `_filter_active_models` runs
# after this point in `best` and cannot be the cause of an empty query result.
rows = self._query(
f"""
SELECT t.family, t.label, pm.ic_n_days, p.prediction_hash
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)
{" AND t.label = ?" if label else ""}
{
" AND p.prediction_hash IN (" + ", ".join("?" for _ in prediction_hashes) + ")"
if prediction_hashes
else ""
}
""",
(
stage,
*excluded_family_sql(self.case_study, "t.family")[1],
*([label] if label else []),
*(prediction_hashes or ()),
),
)
if rows.is_empty():
return
detail = ", ".join(
f"{family}/{row_label} {hash_} covers {days if days is not None else 'no'} days"
for family, row_label, days, hash_ in rows.rows()
)
raise RuntimeError(
f"every backtest at stage {stage!r} sits below its family's coverage bar, so the "
f"complete-run filter admits none of them: {detail}. The bar is the population's "
f"maximum ic_n_days, which is what makes a run complete - an incomplete run "
f"compared as complete manufactures a false leader. Backtest a maximum-coverage "
f"prediction, or say in the notebook that this population supports no comparison."
)
def best(
self,
stage: str = "signal",
*,
top_n: int = 10,
metric: str = "sharpe",
label: str | None = None,
prediction_hashes: tuple[str, ...] | list[str] | None = None,
) -> pl.DataFrame:
"""Top-N backtests at a given stage, ranked by ``metric``.
``top_n=0`` returns every matching backtest, the reading
``sweep_config.top_n_cap`` gives the number everywhere else. Counting a whole
cohort used to mean asking for more rows than any cohort could hold, through a
``_UNBOUNDED_COHORT`` sentinel of a million, because 0 truncated to nothing.
Returns
-------
pl.DataFrame
Columns: backtest_hash, prediction_hash, source, family,
config_name, checkpoint_kind, checkpoint_value, label,
signal_method, top_k, universe_filter, exit_at_max_days, sharpe,
cagr, max_drawdown, total_return, volatility, ic_mean,
ic_mean_daily, ic_ci_lo, ic_ci_hi, ic_n_days
"""
cap = top_n_cap(top_n)
filter_sql = ""
filter_params: list[str] = []
coverage_params: list[str] = []
coverage_sql = full_coverage_prediction_sql("p", "t", "pm")
if label:
filter_sql += " AND t.label = ?"
filter_params.append(label)
if prediction_hashes:
placeholders = ", ".join("?" for _ in prediction_hashes)
filter_sql += f" AND p.prediction_hash IN ({placeholders})"
filter_params.extend(prediction_hashes)
coverage_sql = full_coverage_prediction_sql(
"p",
"t",
"pm",
population_subquery="SELECT value FROM json_each(?)",
)
coverage_params.append(json.dumps(list(prediction_hashes)))
df = self._query(
f"""
SELECT
b.backtest_hash,
b.prediction_hash,
b.spec_json,
b.stage,
t.family,
t.config_name,
p.checkpoint_kind,
p.checkpoint_value,
t.label,
bm.sharpe,
bm.cagr,
bm.max_drawdown,
bm.total_return,
bm.volatility,
pm.ic_mean,
pm.ic_mean_daily,
pm.ic_ci_lo,
pm.ic_ci_hi,
pm.ic_n_days
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]}
{coverage_sql}
AND bm.sharpe IS NOT NULL
AND (bm.num_trades IS NULL OR bm.num_trades > 0)
{filter_sql}
ORDER BY bm.sharpe DESC, b.backtest_hash ASC
""",
(
stage,
*excluded_family_sql(self.case_study, "t.family")[1],
*coverage_params,
*filter_params,
),
)
if df.is_empty():
self._refuse_if_the_coverage_bar_emptied_it(stage, label, prediction_hashes)
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
]
# Every entry scheme in a baseline sweep is `equal_weight_top_k` and varies
# only `top_k`, so `signal_method` alone made nine of the ten rows read as
# the same strategy. `top_k` sits one key across from `method` in the spec
# this block already parses.
top_k = [strategy_view(sp).get("signal", {}).get("top_k") for sp in parsed]
df = df.with_columns(
pl.Series("signal_method", methods),
pl.Series("top_k", top_k, dtype=pl.Int64),
pl.Series("universe_filter", universe_filters),
pl.Series("exit_at_max_days", exit_at_max_days, dtype=pl.Int64),
)
ranked = _drop_rows_a_reader_cannot_tell_apart(
df.select(
"backtest_hash",
"prediction_hash",
"source",
"family",
"config_name",
"checkpoint_kind",
"checkpoint_value",
"label",
"signal_method",
"top_k",
"universe_filter",
"exit_at_max_days",
"sharpe",
"cagr",
"max_drawdown",
"total_return",
"volatility",
"ic_mean",
"ic_mean_daily",
"ic_ci_lo",
"ic_ci_hi",
"ic_n_days",
)
)
return ranked if cap is None else ranked.head(cap)
# -----------------------------------------------------------------
# compare_families: model family comparison at a stage
# -----------------------------------------------------------------
def compare_families(
self,
stage: str = "signal",
*,
prediction_hashes: tuple[str, ...] | list[str] | None = None,
exclude_insolvent: bool = False,
) -> pl.DataFrame:
"""Compare model families by backtest Sharpe at a given stage.
Parameters
----------
prediction_hashes : sequence of str, optional
Restrict the comparison to these prediction identities. The coverage bar is
then taken within them, matching ``best`` and ``compare_allocators``: a family
whose full-coverage rows are outside the population must be compared on its own
terms rather than dropped against a bar nothing in the population can meet.
An empty sequence is a population with no members and compares nothing.
exclude_insolvent : bool, default False
Compute the Sharpe statistics over the runs that stayed solvent,
and report how many did not in an added ``insolvent`` column. A run
whose ``max_drawdown`` reached -100% took its equity to zero or
past it; there is no capital left to earn a later return, so every
subsequent period is arithmetic on zero or a negative balance and
the Sharpe is a number without a meaning. Leaving those in distorts
a median and can top a maximum.
The count is reported rather than the runs silently dropped,
because dropping them alone would rank a family by its survivors: a
family that went to zero in nine runs out of ten would show the
Sharpe of the tenth and nothing to say so. ``n`` counts the solvent
runs the statistics are computed over and ``insolvent`` the rest,
so a reader can see what the median is conditional on. A family
with no solvent run has null statistics and sorts last.
A run with no recorded drawdown is counted under ``unknown`` and
kept out of the statistics. It cannot be shown to have stayed
solvent, but neither did it fail: putting it under ``insolvent``
would report a bankruptcy that was never measured. ``n``,
``insolvent`` and ``unknown`` together are every run of the family:
the query does not drop a run for having no Sharpe when this is on,
because a bankrupt run can carry a null one and dropping it would take
a wiped-out family off the table entirely. The Sharpe statistics skip
any solvent run without a recorded Sharpe, so they can rest on fewer
than ``n`` runs: these columns count runs, not measurements.
Off by default, so a caller that has already reported on the full
population keeps reporting on it, with the same columns as before.
Returns
-------
pl.DataFrame
Columns: family, n, sharpe_median, sharpe_max, sharpe_q75,
pct_positive - and ``insolvent``, ``unknown`` when
``exclude_insolvent``.
"""
scope_sql = ""
scope_params: tuple[str, ...] = ()
coverage_sql = full_coverage_prediction_sql("p", "t", "pm")
if prediction_hashes is not None:
placeholders = ", ".join("?" for _ in prediction_hashes)
scope_sql = f" AND p.prediction_hash IN ({placeholders})"
coverage_sql = full_coverage_prediction_sql(
"p",
"t",
"pm",
population_subquery="SELECT value FROM json_each(?)",
)
scope_params = (json.dumps(list(prediction_hashes)), *prediction_hashes)
# A run that went bankrupt can carry a null Sharpe, and dropping those in SQL would
# remove it before `insolvent` counts it - so a family that went bankrupt in every
# run would leave the comparison entirely, which is the survivorship bias this
# option exists to expose. Under the flag the rows are kept, and the Sharpe
# aggregates skip the nulls themselves.
sharpe_sql = "" if exclude_insolvent else "AND bm.sharpe IS NOT NULL"
df = self._query(
f"""
SELECT
t.family,
bm.sharpe,
bm.max_drawdown
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
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]}
{coverage_sql}
{sharpe_sql}
AND (bm.num_trades IS NULL OR bm.num_trades > 0)
{scope_sql}
""",
(stage, *excluded_family_sql(self.case_study, "t.family")[1], *scope_params),
)
if df.is_empty():
return df
df = self._filter_active_models(df)
if df.is_empty():
return df
if not exclude_insolvent:
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)
)
# Three outcomes, counted separately, because a run with no recorded drawdown
# is not a bankruptcy: reporting it under `insolvent` would put a failure on
# the record that was never measured. Each comparison is null for that run, so
# the sums skip it and it lands only in `unknown`. n + insolvent + unknown is
# every run of the family.
solvent = pl.col("max_drawdown") > -1.0
# Filtering needs a mask with no nulls; the counts above do not, and take the
# null-skipping behaviour instead.
solvent_mask = solvent.fill_null(False)
return (
df.group_by("family")
.agg(
n=solvent.sum(),
insolvent=(pl.col("max_drawdown") <= -1.0).sum(),
unknown=pl.col("max_drawdown").is_null().sum(),
sharpe_median=pl.col("sharpe").filter(solvent_mask).median(),
sharpe_max=pl.col("sharpe").filter(solvent_mask).max(),
sharpe_q75=pl.col("sharpe").filter(solvent_mask).quantile(0.75),
# `mean` rather than `sum() / len()`: a family whose runs all went
# insolvent divides by zero, and mean over an empty selection is null,
# which is what "no solvent run to report a rate over" means.
pct_positive=(pl.col("sharpe").filter(solvent_mask) > 0).mean() * 100,
)
.sort("sharpe_median", descending=True, nulls_last=True)
)
# -----------------------------------------------------------------
# compare_allocators: allocation method comparison
# -----------------------------------------------------------------
def compare_allocators(
self,
*,
prediction_hash: str | None = None,
stages: tuple[str, ...] = ("allocation",),
label: str | None = None,
prediction_hashes: tuple[str, ...] | list[str] | None = None,
) -> 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.
label : str, optional
Restrict the comparison to one target label. Publication notebooks
should pass their active label so accumulated sibling-label rows do
not change the reported allocator ranking.
prediction_hashes : sequence of str, optional
Restrict the comparison to prediction sets advanced by the current
funnel. This excludes accumulated rows from earlier sweeps.
Returns
-------
pl.DataFrame
Columns: allocator, n, ruined, avg_sharpe, best_sharpe, avg_max_dd.
``n`` counts the runs with a rankable Sharpe and every statistic is
taken over those; ``ruined`` counts the runs the engine stopped at zero
equity, which carry no Sharpe to rank.
"""
if not stages:
return pl.DataFrame()
placeholders = ", ".join("?" for _ in stages)
# A run the engine stopped at ruin carries a null Sharpe by design
# . Dropping those in SQL, as this query used to,
# would take an allocator that went bankrupt in every run off the table
# entirely and report the survivors as the whole population. The rows are
# kept and counted under `ruined`; the Sharpe and drawdown statistics are
# taken over the solvent runs only, so `avg_max_dd` is a mean of drawdowns
# that were actually measured.
ruin_select = "bm.ruin" if self._has_metric_column("ruin") else "NULL AS ruin"
coverage_params: tuple[str, ...] = ()
coverage_sql = full_coverage_prediction_sql("p", "t", "pm")
if prediction_hashes:
coverage_sql = full_coverage_prediction_sql(
"p",
"t",
"pm",
population_subquery="SELECT value FROM json_each(?)",
)
coverage_params = (json.dumps(list(prediction_hashes)),)
sql = f"""
SELECT
b.spec_json,
t.family,
t.config_name,
bm.sharpe,
bm.max_drawdown,
{ruin_select}
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 prediction_metrics pm ON p.prediction_hash = pm.prediction_hash
JOIN backtest_metrics bm ON bm.backtest_hash = b.backtest_hash
WHERE b.stage IN ({placeholders})
AND p.split != 'holdout'
{excluded_family_sql(self.case_study, "t.family")[0]}
{coverage_sql}
AND (bm.num_trades IS NULL OR bm.num_trades > 0)
"""
params: tuple = (
*stages,
*excluded_family_sql(self.case_study, "t.family")[1],
*coverage_params,
)
if prediction_hash:
sql += " AND b.prediction_hash LIKE ?"
params = (*params, prediction_hash + "%")
if label:
sql += " AND t.label = ?"
params = (*params, label)
if prediction_hashes:
hash_placeholders = ", ".join("?" for _ in prediction_hashes)
sql += f" AND p.prediction_hash IN ({hash_placeholders})"
params = (*params, *prediction_hashes)
df = self._query(sql, params)
if df.is_empty():
return df
df = self._filter_active_models(df)
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
# A run whose drawdown passed -100% crossed zero equity, so it is bankrupt
# whether or not the engine that produced it recorded `ruin` - registries
# written before that column carry the evidence only in the drawdown.
ruined = (pl.col("ruin") == 1.0) | (pl.col("max_drawdown") <= -1.0)
rankable = pl.col("sharpe").is_not_null() & ~ruined.fill_null(False)
return (
df.group_by("allocator")
.agg(
n=rankable.sum(),
ruined=ruined.fill_null(False).sum(),
avg_sharpe=pl.col("sharpe").filter(rankable).mean(),
best_sharpe=pl.col("sharpe").filter(rankable).max(),
avg_max_dd=pl.col("max_drawdown").filter(rankable).mean(),
)
.sort("avg_sharpe", descending=True, nulls_last=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, b.backtest_hash ASC
""",
tuple(params),
)
if df.is_empty():
return df
# Take best Sharpe per stage
stage_order = {s: i for i, s in enumerate(STAGE_SEQUENCE)}
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,
prediction_hashes: tuple[str, ...] | list[str] | None = None,
) -> 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).
A deflated Sharpe is only meaningful against the trial count of the
selection actually performed, so when ``prediction_hashes`` scopes
the ranking these columns are reported only where the persisted
cohort *is* the scoped cohort. ``k_variants`` is the count the stored
correction was computed over and ``k_variants_scoped`` is the count
in hand; where they disagree the selection-bias columns are null and
both counts are kept, so a reader sees which population the missing
correction belonged to rather than a leader that appears to need
none. The counts are built by the same eligibility rule, since the
scoped cohort is measured with ``best`` itself. 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,
k_variants_scoped, 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, prediction_hashes=prediction_hashes)
if top.is_empty():
return pl.DataFrame()
# The scoped cohort's size per (family, label), measured through `best` so that
# coverage, excluded families and the tradeless-backtest rule are applied exactly
# once. `top_n` bounds the rows displayed, not the cohort selected from, so this
# is a second unbounded read rather than a group-by over `top`.
scoped_k: dict[tuple[str, str], int] = {}
scoped_digest: dict[tuple[str, str], str] = {}
if prediction_hashes:
cohort = self.best(stage=stage, top_n=0, prediction_hashes=prediction_hashes)
if not cohort.is_empty():
grouped = cohort.group_by("family", "label").agg(
n=pl.len(), members=pl.col("backtest_hash")
)
for row in grouped.iter_rows(named=True):
key = (row["family"], row["label"])
scoped_k[key] = row["n"]
scoped_digest[key] = cohort_member_digest(row["members"])
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, member_digest,
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
key = (r["family"], r["label"])
k_scoped = scoped_k.get(key) if prediction_hashes else None
# Applies only where the stored cohort IS the one that was ranked, established
# on the members and not on how many there are: swapping a retired member for a
# live one leaves `k_variants` untouched, and the correction would then be
# reported against a cohort it was not computed over. `member_digest` is null on
# rows written before it was persisted; those still fall back to the count, which
# at least catches the scoping that strictly removes variants, and stop doing so
# the first time their producer re-runs.
applies = is_leader and (
not prediction_hashes
or (
cmr["member_digest"] == scoped_digest.get(key)
if cmr["member_digest"]
else cmr["k_variants"] == k_scoped
)
)
dsr_er_p = cmr["dsr_er_pvalue"] if applies 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 applies else None,
"expected_max_sharpe": _round(cmr["expected_max_sharpe_er"])
if applies
else None,
"dsr_pvalue": _round(dsr_er_p),
"significant": (dsr_er_p is not None and dsr_er_p < 0.05) if applies else None,
"is_best": is_leader,
"dsr_mp": _round(cmr["dsr_mp"]) if applies else None,
"dsr_mp_pvalue": _round(cmr["dsr_mp_pvalue"]) if applies else None,
"dsr_raw": _round(cmr["dsr_raw"]) if applies else None,
"dsr_raw_pvalue": _round(cmr["dsr_raw_pvalue"]) if applies else None,
"k_variants": cmr["k_variants"] if is_leader else None,
"k_variants_scoped": k_scoped,
"n_trials_effective_er": _round(cmr["n_trials_effective_er"], 1)
if applies
else None,
"n_trials_effective_mp": _round(cmr["n_trials_effective_mp"], 1)
if applies
else None,
"ras_leader": _round(cmr["ras_leader"]) if applies else None,
"ras_pvalue": _round(cmr["ras_pvalue"]) if applies else None,
"reality_check_pvalue": _round(cmr["reality_check_pvalue"])
if applies
else None,
"pbo": _round(cmr["pbo"]) if applies 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,
backtest_hashes: Iterable[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.
backtest_hashes : iterable of str, optional
When provided, restrict to exactly these cost rows. A prediction is
not a strategy: several configurations share one prediction set, and
a superseded generation stays in the registry under the same
prediction hash as the one that replaced it - on
us_firm_characteristics the retired ``walk_forward_v2`` conformal
sweep and its ``walk_forward_v3`` replacement both do. Scoping by
prediction then draws two generations as one curve. Pass the hashes
the sweep registered when the curve must describe one strategy.
Returns
-------
pl.DataFrame
Columns: cost_bps, sharpe, max_drawdown, allocator
"""
pred_clause = "" if prediction_hash is None else " AND b.prediction_hash = ?"
params: tuple = () if prediction_hash is None else (prediction_hash,)
hash_clause = ""
if backtest_hashes is not None:
selected = tuple(dict.fromkeys(backtest_hashes))
if not selected:
# An empty selection is an empty curve, not an unscoped one. Falling through
# to no clause would return every cost row in the registry.
return pl.DataFrame()
hash_clause = f" AND b.backtest_hash IN ({', '.join('?' for _ in selected)})"
params = params + selected
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}{hash_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,
prediction_hashes: tuple[str, ...] | list[str] | None = None,
parents: Iterable[tuple[str, str, object]] | None = None,
) -> pl.DataFrame:
"""Load risk overlay results and compute impact vs the baseline each one modified.
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.
prediction_hashes : sequence of str, optional
Restrict to overlays on these predictions. An overlay is a change to
one allocation, so pooling overlays from predictions the caller did
not sweep measures rules against strategies it never built.
parents : sequence of (prediction_hash, allocator, top_k), optional
Restrict to overlays sitting on these allocation specifications. A
prediction has one allocation *stage* and several allocation *parents*,
distinguished by allocator and ``top_k``; scoping on the prediction
alone admits overlays on combinations the caller did not advance, and
those can rank among the reported leaders. ``top_k`` is compared as
written in the spec, so ``None`` matches a spec that carries no
``top_k`` rather than matching everything.
Returns
-------
pl.DataFrame
Columns: risk_name, risk_type, sharpe, max_drawdown, num_trades,
risk_triggers, allocator, prediction_hash, baseline_sharpe,
sharpe_delta
Each overlay's ``baseline_sharpe`` is the no-overlay Sharpe of the
allocation it was applied to, matched on ``(prediction_hash,
allocator)``. One registry-wide maximum would measure most overlays
against a strategy they did not modify, so an overlay could appear to
destroy Sharpe purely because a different prediction allocated better.
"""
pred_clause = ""
pred_params: tuple = ()
if prediction_hash is not None:
pred_clause = " AND b.prediction_hash = ?"
pred_params = (prediction_hash,)
elif prediction_hashes:
pred_clause = " AND b.prediction_hash IN (SELECT value FROM json_each(?))"
pred_params = (json.dumps(list(prediction_hashes)),)
# Absent from a registry written before the engine counted triggers, where
# the question this column answers simply cannot be answered.
triggers_select = (
"bm.risk_triggers"
if self._has_metric_column("risk_triggers")
else "NULL AS risk_triggers"
)
df = self._query(
f"""
SELECT
b.spec_json,
b.prediction_hash,
t.family,
t.config_name,
bm.sharpe,
bm.max_drawdown,
bm.num_trades,
{triggers_select}
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
Se muestra íntegramente con atribución según la licencia de la fuente. Licencia: MIT
Este resumen lo redactó el agente de investigación de Stratmill a partir del original; no es una copia de la fuente.