בחינה והשוואה של תוצאות בק טסט רשומות

# 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