Диагностика неопределённости при выборе и проверке моделей

# notebook_render.py


```py
"""Rendering helpers for per-CS ``*_model_analysis.py`` and chapter insight notebooks.

Composes with :mod:`case_studies.utils.analytics` (registry queries) to produce
uncertainty-aware tables and figures. All helpers return either polars
DataFrames or matplotlib Figures.

Tables (return polars):
    - :func:`holdout_decay_table` — val IC ± CI vs holdout IC ± CI per family
    - :func:`selection_adjusted_leader_table` — DSR / PBO / RC / k-variant per leader

Figures (return matplotlib Figure):
    - :func:`headline_forest_plot` — forest plot of IC ± CI sorted by point estimate
    - :func:`fold_heatmap_with_ci` — fold × family IC heatmap with significance shading

Axes overlays (mutate caller-supplied ax):
    - :func:`regime_coverage_strip` — color-coded strip below fold-IC distribution

Classification-aware helpers (filled at Phase C start):
    - :func:`classification_triple` — AUC ± CI, accuracy ± CI, IC-on-continuous ± CI
    - :func:`cross_task_matrix` — {regression model, classification model} × {IC, AUC} 2×2
"""

from __future__ import annotations

import matplotlib.pyplot as plt
import numpy as np
import polars as pl
from matplotlib.axes import Axes
from matplotlib.figure import Figure

from case_studies.utils.analytics import (
    PRIMARY_LABELS,
    _query,
    registry_path,
)
from case_studies.utils.conformal import (
    DEFAULT_MIN_CALIBRATION_N,
    sizing_conformal_lag,
    walk_forward_conformal_coverage,
)
from case_studies.utils.notebook_contracts import (
    degenerate_prediction_sql,
    full_coverage_prediction_sql,
)
from utils.style import COLORS

# ---------------------------------------------------------------------------
# Tables
# ---------------------------------------------------------------------------


def holdout_decay_table(
    case_study: str,
    *,
    label: str | None = None,
    families: list[str] | None = None,
) -> pl.DataFrame:
    """Validation vs holdout IC with 95% CIs and decay for the rank-1 leader.

    By design (Ch16 selection workflow) each case study has at most one
    holdout-retrained model — the signal-stage rank-1 leader. This table
    reports its val→holdout decay with a row per family: families that were
    not selected at rank-1 show null holdout columns.

    Always loads fresh from ``prediction_sets`` — no pre-computed artifact.
    Prefers ``ic_mean_daily`` + HAC CI; falls back to legacy ``ic_mean`` where
    the daily-pooled backfill hasn't run (currently: all holdout splits).

    Returns columns:
        family, config_name, label,
        val_ic, val_ci_lo, val_ci_hi, val_ic_source,
        ho_ic,  ho_ci_lo,  ho_ci_hi,  ho_ic_source,
        decay_pp, decay_pct
    """
    label = label or PRIMARY_LABELS[case_study]
    db = registry_path(case_study)
    if not db.exists():
        return _empty_frame(_HOLDOUT_DECAY_SCHEMA)

    family_clause = ""
    params: list = [label]
    if families:
        placeholders = ",".join("?" * len(families))
        family_clause = f"AND t.family IN ({placeholders})"
        params.extend(families)

    # Prefer daily-pooled IC + HAC CI; fall back to legacy ic_mean where the
    # daily-pooled backfill hasn't run (notably: holdout splits, as of 2026-04-30).
    sql = f"""
        SELECT
            t.family,
            t.config_name,
            t.label,
            p.split,
            COALESCE(pm.ic_mean_daily, pm.ic_mean) AS ic,
            pm.ic_ci_lo,
            pm.ic_ci_hi,
            pm.ic_n_days,
            CASE WHEN pm.ic_mean_daily IS NOT NULL THEN 'daily_hac' ELSE 'fold_mean' END AS ic_source
        FROM training_runs t
        JOIN prediction_sets p ON t.training_hash = p.training_hash
        JOIN prediction_metrics pm ON p.prediction_hash = pm.prediction_hash
        WHERE t.label = ?
          AND p.split IN ('validation', 'holdout')
          {family_clause}
          AND COALESCE(pm.ic_mean_daily, pm.ic_mean) IS NOT NULL
    """
    rows = _query(db, sql, tuple(params))
    if rows.is_empty():
        return _empty_frame(_HOLDOUT_DECAY_SCHEMA)

    # Holdout retrains are at most one per family (the signal-stage rank-1
    # leader). For those families the row's config_name and val_ic must come
    # from the SAME config that was retrained — not the validation IC rank-1
    # config, which need not be the same model. Joining only on `family`
    # would mix two different configs into a single row (e.g. an ETF case
    # where validation IC rank-1 = nlinear and the holdout retrain target =
    # lstm_h64) and mis-attribute lstm_h64's holdout IC to nlinear.
    ho_leaders = (
        rows.filter(pl.col("split") == "holdout")
        .sort("ic", descending=True, nulls_last=True)
        .group_by("family")
        .first()
        .select(
            "family",
            "config_name",
            pl.col("ic").alias("ho_ic"),
            pl.col("ic_ci_lo").alias("ho_ci_lo"),
            pl.col("ic_ci_hi").alias("ho_ci_hi"),
            pl.col("ic_source").alias("ho_ic_source"),
        )
    )

    # For families with a holdout retrain, pull val IC for the same (family, config).
    # Multiple validation predictions can exist for one config (e.g. DL checkpoints);
    # take the highest-IC one to mirror the validation rank-1 selection logic.
    val_rows = rows.filter(pl.col("split") == "validation")
    val_for_ho = (
        val_rows.join(
            ho_leaders.select("family", "config_name"),
            on=["family", "config_name"],
            how="inner",
        )
        .sort("ic", descending=True, nulls_last=True)
        .group_by(["family", "config_name"])
        .first()
        .select(
            "family",
            "config_name",
            "label",
            pl.col("ic").alias("val_ic"),
            pl.col("ic_ci_lo").alias("val_ci_lo"),
            pl.col("ic_ci_hi").alias("val_ci_hi"),
            pl.col("ic_source").alias("val_ic_source"),
        )
    )

    # Families without a holdout retrain still surface their validation IC
    # rank-1 leader; ho_* columns will be null after the left-join below.
    ho_families = ho_leaders.select("family")
    val_no_ho = (
        val_rows.join(ho_families, on="family", how="anti")
        .sort("ic", descending=True, nulls_last=True)
        .group_by("family")
        .first()
        .select(
            "family",
            "config_name",
            "label",
            pl.col("ic").alias("val_ic"),
            pl.col("ic_ci_lo").alias("val_ci_lo"),
            pl.col("ic_ci_hi").alias("val_ci_hi"),
            pl.col("ic_source").alias("val_ic_source"),
        )
    )

    val = pl.concat([val_for_ho, val_no_ho])
    out = val.join(ho_leaders.drop("config_name"), on="family", how="left")
    out = out.with_columns(
        decay_pp=(pl.col("ho_ic") - pl.col("val_ic")),
        decay_pct=pl.when(pl.col("val_ic").abs() > 0)
        .then((pl.col("ho_ic") - pl.col("val_ic")) / pl.col("val_ic").abs() * 100.0)
        .otherwise(None),
    )
    return out.sort("val_ic", descending=True, nulls_last=True)


# The columns each of the three registry readers below promises, with the dtypes a
# populated result carries. An empty result is returned under its own schema rather than
# as a bare `pl.DataFrame()`, because a caller cannot write `.select("family", ...)`
# against a frame with no columns: it raises ColumnNotFoundError, which is a different
# failure from "the stage that fills this has not run" and reads as a defect in the
# notebook rather than as an empty stage. Every case study's model-analysis notebook
# reaches all three before its backtesting stage has run, so the empty case is the one a
# reader meets first. Each function returns the same schema on every one of its empty
# paths, so a caller cannot see columns on one and not the other.
_SELECTION_ADJUSTED_SCHEMA: dict[str, pl.DataType] = {
    "family": pl.String,
    "config_name": pl.String,
    "label": pl.String,
    "sharpe": pl.Float64,
    "sharpe_ci95_lo": pl.Float64,
    "sharpe_ci95_hi": pl.Float64,
    "psr_pvalue": pl.Float64,
    "dsr": pl.Float64,
    "dsr_pvalue": pl.Float64,
    "expected_max_sharpe": pl.Float64,
    "dsr_mp": pl.Float64,
    "dsr_mp_pvalue": pl.Float64,
    "dsr_raw": pl.Float64,
    "dsr_raw_pvalue": pl.Float64,
    "n_trials_effective_er": pl.Float64,
    "n_trials_effective_mp": pl.Float64,
    "ras_leader": pl.Float64,
    "ras_pvalue": pl.Float64,
    "reality_check_pvalue": pl.Float64,
    "pbo": pl.Float64,
    "k_variants": pl.Int64,
}


_HOLDOUT_DECAY_SCHEMA: dict[str, pl.DataType] = {
    "family": pl.String,
    "config_name": pl.String,
    "label": pl.String,
    "val_ic": pl.Float64,
    "val_ci_lo": pl.Float64,
    "val_ci_hi": pl.Float64,
    "val_ic_source": pl.String,
    "ho_ic": pl.Float64,
    "ho_ci_lo": pl.Float64,
    "ho_ci_hi": pl.Float64,
    "ho_ic_source": pl.String,
    "decay_pp": pl.Float64,
    "decay_pct": pl.Float64,
}

_CONFORMAL_COVERAGE_SCHEMA: dict[str, pl.DataType] = {
    "family": pl.String,
    "config_name": pl.String,
    "nominal_level": pl.Float64,
    "empirical_coverage": pl.Float64,
    "mean_interval_width_frac_std": pl.Float64,
    "n_test": pl.Int64,
    "n_uncalibrated": pl.Int64,
}


def _empty_frame(schema: dict[str, pl.DataType]) -> pl.DataFrame:
    return pl.DataFrame(schema=schema)


def selection_adjusted_leader_table(
    case_study: str,
    *,
    stage: str = "signal",
    label: str | None = None,
    prediction_hashes: set[str] | None = None,
) -> pl.DataFrame:
    """Per-family rank-1 backtest with selection-adjusted statistics.

    Joins ``backtest_metrics`` with ``training_runs`` and LEFT JOINs the
    persisted ``cohort_metrics`` (cohort_type='family') for the
    leader-hash. The legacy column names (``dsr``, ``dsr_pvalue``,
    ``expected_max_sharpe``) carry the **effective-rank (ER) DSR** — the
    library maintainer's recommended default. ``dsr_mp`` and ``dsr_raw``
    are surfaced alongside for sensitivity. Non-leader family rows have
    NULL selection-bias columns. When ``prediction_hashes`` is supplied,
    leaders are selected only from that population.

    Returns columns:
        family, config_name, label,
        sharpe, sharpe_ci95_lo, sharpe_ci95_hi,
        psr_pvalue, dsr, dsr_pvalue, expected_max_sharpe,
        dsr_mp, dsr_mp_pvalue, dsr_raw, dsr_raw_pvalue,
        n_trials_effective_er, n_trials_effective_mp,
        ras_leader, ras_pvalue,
        reality_check_pvalue, pbo, k_variants
    """
    db = registry_path(case_study)
    if not db.exists():
        return _empty_frame(_SELECTION_ADJUSTED_SCHEMA)

    label_clause = ""
    params: list = [stage]
    if label:
        label_clause = "AND t.label = ?"
        params.append(label)

    sql = f"""
        SELECT
            b.prediction_hash,
            t.family,
            t.config_name,
            t.label,
            bm.sharpe,
            bm.sharpe_ci95_lo,
            bm.sharpe_ci95_hi,
            bm.psr_pvalue,
            cm.dsr_er                 AS dsr,
            cm.dsr_er_pvalue          AS dsr_pvalue,
            cm.expected_max_sharpe_er AS expected_max_sharpe,
            cm.dsr_mp,
            cm.dsr_mp_pvalue,
            cm.dsr_raw,
            cm.dsr_raw_pvalue,
            cm.n_trials_effective_er,
            cm.n_trials_effective_mp,
            cm.ras_leader,
            cm.ras_pvalue,
            cm.reality_check_pvalue,
            cm.pbo,
            cm.k_variants
        FROM backtest_runs b
        JOIN backtest_metrics bm ON b.backtest_hash = bm.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
        LEFT JOIN cohort_metrics cm
               ON cm.cohort_type = 'family'
              AND cm.stage = b.stage
              AND cm.label = t.label
              AND cm.family = t.family
              AND cm.leader_hash = b.backtest_hash
        WHERE b.stage = ?
          {label_clause}
          AND bm.sharpe IS NOT NULL
          {degenerate_prediction_sql("p.prediction_hash")}
          {full_coverage_prediction_sql("p", "t", "pm")}
    """
    rows = _query(db, sql, tuple(params))
    if rows.is_empty():
        return _empty_frame(_SELECTION_ADJUSTED_SCHEMA)
    if prediction_hashes is not None:
        rows = rows.filter(pl.col("prediction_hash").is_in(prediction_hashes))
        if rows.is_empty():
            return _empty_frame(_SELECTION_ADJUSTED_SCHEMA)

    # Force Float64 dtype on numeric columns that can come back as all-NULL
    # under the LEFT JOIN (polars infers Null dtype otherwise, which breaks
    # downstream ``.round()`` calls).
    float_cols = [
        "dsr",
        "dsr_pvalue",
        "expected_max_sharpe",
        "dsr_mp",
        "dsr_mp_pvalue",
        "dsr_raw",
        "dsr_raw_pvalue",
        "n_trials_effective_er",
        "n_trials_effective_mp",
        "ras_leader",
        "ras_pvalue",
        "reality_check_pvalue",
        "pbo",
    ]
    casts = [pl.col(c).cast(pl.Float64) for c in float_cols if c in rows.columns]
    if casts:
        rows = rows.with_columns(casts)

    leaders = rows.sort("sharpe", descending=True, nulls_last=True).group_by("family").first()
    return leaders.drop("prediction_hash").sort("sharpe", descending=True, nulls_last=True)


# ---------------------------------------------------------------------------
# Figures
# ---------------------------------------------------------------------------


def headline_forest_plot(
    df: pl.DataFrame,
    *,
    ic_col: str = "ic_mean_daily",
    ci_lo_col: str = "ic_ci_lo",
    ci_hi_col: str = "ic_ci_hi",
    label_col: str = "config_name",
    family_col: str = "family",
    task_type_col: str | None = None,
    title: str = "",
    figsize: tuple[float, float] = (8.0, None),  # type: ignore[assignment]
) -> Figure:
    """Forest plot of point estimates with 95% CIs, sorted by point estimate.

    CI bars that include zero are drawn in muted gray; non-zero-crossing CIs
    use the family color from :data:`utils.style.COLORS`. The zero line is
    drawn as a dashed reference.

    Parameters
    ----------
    df : pl.DataFrame
        Must contain ``ic_col``, ``ci_lo_col``, ``ci_hi_col``, ``label_col``,
        ``family_col``. Optional ``task_type_col`` adds a task-type tag.
    """
    required = {ic_col, ci_lo_col, ci_hi_col, label_col, family_col}
    missing = required - set(df.columns)
    if missing:
        raise ValueError(f"forest plot missing columns: {missing}")

    sorted_df = df.sort(ic_col, descending=False, nulls_last=True).drop_nulls(ic_col)
    n = sorted_df.height
    if n == 0:
        raise ValueError("forest plot received empty (or fully-null) data")

    height = figsize[1] if figsize[1] is not None else max(2.5, 0.32 * n + 1.0)
    fig, ax = plt.subplots(figsize=(figsize[0], height), constrained_layout=True)

    ic = sorted_df[ic_col].to_numpy()
    lo = sorted_df[ci_lo_col].to_numpy()
    hi = sorted_df[ci_hi_col].to_numpy()
    families = sorted_df[family_col].to_list()
    labels = sorted_df[label_col].to_list()
    if task_type_col and task_type_col in sorted_df.columns:
        task_types = sorted_df[task_type_col].to_list()
        labels = [f"{lbl}  [{tt}]" if tt else lbl for lbl, tt in zip(labels, task_types)]

    family_palette = {
        "linear": COLORS.get("blue", "C0"),
        "gbm": COLORS.get("orange", "C1"),
        "deep_learning": COLORS.get("green", "C2"),
        "tabular_dl": COLORS.get("purple", "C3"),
        "latent_factors": COLORS.get("red", "C4"),
        "causal": COLORS.get("brown", "C5"),
        "benchmark": COLORS.get("gray", "C7"),
    }

    y_positions = np.arange(n)
    for i, (point, lo_i, hi_i, fam) in enumerate(zip(ic, lo, hi, families)):
        # numpy float arrays carry NaN for nulls — np.isfinite catches both
        # None-cast-to-NaN and explicit NaNs; bare ``is not None`` would
        # always be True after ``to_numpy()`` and let NaN values reach plot.
        ci_valid = bool(np.isfinite(lo_i) and np.isfinite(hi_i))
        crosses_zero = ci_valid and lo_i <= 0 <= hi_i
        color = "#999999" if (not ci_valid or crosses_zero) else family_palette.get(fam, "#444444")
        if ci_valid:
            ax.plot([lo_i, hi_i], [i, i], color=color, linewidth=2.0, alpha=0.85)
        if np.isfinite(point):
            ax.plot(point, i, marker="o", color=color, markersize=6, zorder=3)

    ax.axvline(0.0, color="black", linestyle="--", linewidth=0.8, alpha=0.5)
    ax.set_yticks(y_positions)
    ax.set_yticklabels(labels, fontsize=8)
    ax.set_xlabel("Information Coefficient (daily-pooled, 95% HAC CI)")
    if title:
        ax.set_title(title)
    ax.grid(True, axis="x", linestyle=":", alpha=0.3)
    return fig


def fold_heatmap_with_ci(
    case_study: str,
    label: str | None = None,
    *,
    families: list[str] | None = None,
    significance_threshold: float = 0.05,
    title: str = "",
) -> Figure:
    """Heatmap of fold IC × family with cells where p-value > threshold dimmed.

    Pulls from ``fold_metrics`` joined to ``training_runs``. Each (family, fold)
    cell shows the rank-1 config IC for that (family, fold). Cells whose HAC
    t-statistic gives p > ``significance_threshold`` are rendered in gray.

    Returns
    -------
    matplotlib.figure.Figure
    """
    label = label or PRIMARY_LABELS[case_study]
    db = registry_path(case_study)
    if not db.exists():
        raise FileNotFoundError(f"no registry for {case_study}")

    family_clause = ""
    params: list = [label]
    if families:
        placeholders = ",".join("?" * len(families))
        family_clause = f"AND t.family IN ({placeholders})"
        params.extend(families)

    sql = f"""
        SELECT
            t.family,
            t.config_name,
            fm.fold_id,
            fm.ic,
            fm.ic_std,
            fm.n_entities
        FROM training_runs t
        JOIN prediction_sets p ON t.training_hash = p.training_hash
        JOIN fold_metrics fm ON p.prediction_hash = fm.prediction_hash
        WHERE t.label = ?
          AND p.split = 'validation'
          {family_clause}
          AND fm.ic IS NOT NULL
          {degenerate_prediction_sql("p.prediction_hash")}
    """
    df = _query(db, sql, tuple(params))
    if df.is_empty():
        raise ValueError(f"no fold metrics for {case_study} / {label}")

    leaders = (
        df.group_by(["family", "config_name"])
        .agg(pl.col("ic").mean().alias("avg_ic"))
        .sort("avg_ic", descending=True, nulls_last=True)
        .group_by("family")
        .first()
        .select("family", "config_name")
    )
    df = df.join(leaders, on=["family", "config_name"], how="inner")

    # Approx |t| = |ic| / SE(ic), with SE = ic_std / sqrt(n_entities). Folds with
    # missing ic_std or n_entities fall back to a non-significant cell.
    df = df.with_columns(
        t_approx=pl.when((pl.col("ic_std") > 0) & (pl.col("n_entities") > 0))
        .then(pl.col("ic").abs() / (pl.col("ic_std") / pl.col("n_entities").sqrt()))
        .otherwise(0.0)
    )

    pivot = df.pivot(values="ic", index="family", on="fold_id", aggregate_function="first")
    family_order = pivot["family"].to_list()
    fold_cols = [c for c in pivot.columns if c != "family"]
    fold_cols.sort(key=lambda c: int(c) if str(c).isdigit() else c)
    matrix = pivot.select(fold_cols).to_numpy()

    sig_pivot = df.pivot(
        values="t_approx", index="family", on="fold_id", aggregate_function="first"
    ).select(fold_cols)
    t_matrix = sig_pivot.to_numpy()
    with np.errstate(invalid="ignore"):
        p_matrix = 2.0 * (1.0 - _phi(np.abs(t_matrix)))
    significant = p_matrix <= significance_threshold

    fig, ax = plt.subplots(
        figsize=(max(6.0, 0.5 * len(fold_cols) + 2.0), 0.5 * len(family_order) + 1.5),
        constrained_layout=True,
    )
    vmax = float(np.nanmax(np.abs(matrix))) if np.isfinite(matrix).any() else 0.05
    cmap = plt.get_cmap("RdBu_r")
    for i, fam in enumerate(family_order):
        for j, fold in enumerate(fold_cols):
            value = matrix[i, j]
            sig = significant[i, j] if significant.shape == matrix.shape else False
            if value is None or np.isnan(value):
                continue
            color = cmap(0.5 + 0.5 * (value / vmax) if vmax > 0 else 0.5)
            if not sig:
                color = (0.85, 0.85, 0.85, 1.0)
            ax.add_patch(plt.Rectangle((j, i), 1, 1, facecolor=color, edgecolor="white"))
            ax.text(
                j + 0.5,
                i + 0.5,
                f"{value:+.2f}",
                ha="center",
                va="center",
                fontsize=7,
                color="black",
            )

    ax.set_xlim(0, len(fold_cols))
    ax.set_ylim(0, len(family_order))
    ax.invert_yaxis()
    ax.set_xticks(np.arange(len(fold_cols)) + 0.5)
    ax.set_xticklabels([str(c) for c in fold_cols])
    ax.set_yticks(np.arange(len(family_order)) + 0.5)
    ax.set_yticklabels(family_order)
    ax.set_xlabel("Fold")
    if title:
        ax.set_title(title)
    return fig


def _phi(x: np.ndarray) -> np.ndarray:
    """Standard normal CDF (kept private; avoid scipy dependency)."""
    # Abramowitz & Stegun 26.2.17 erf approximation; max abs error 1.5e-7
    a1, a2, a3, a4, a5, p = (
        0.254829592,
        -0.284496736,
        1.421413741,
        -1.453152027,
        1.061405429,
        0.3275911,
    )
    sign = np.where(x < 0, -1.0, 1.0)
    ax = np.abs(x) / np.sqrt(2.0)
    t = 1.0 / (1.0 + p * ax)
    y = 1.0 - (((((a5 * t + a4) * t) + a3) * t + a2) * t + a1) * t * np.exp(-ax * ax)
    return 0.5 * (1.0 + sign * y)


# ---------------------------------------------------------------------------
# Axes overlays
# ---------------------------------------------------------------------------


def regime_coverage_strip(
    ax: Axes,
    fold_dates: list[tuple[str, str]],
    regime_lookup: dict[str, str],
    *,
    palette: dict[str, str] | None = None,
    strip_height_frac: float = 0.06,
) -> None:
    """Draw a color-coded regime strip below an existing fold-IC distribution.

    Mutates ``ax`` in place. Adds a thin horizontal strip below the existing
    plot area, color-coded by which regime each fold falls into.

    Parameters
    ----------
    ax : matplotlib Axes
        The existing fold-IC axes (e.g. boxplot or strip plot of fold ICs).
    fold_dates : list of (start_iso, end_iso)
        One entry per fold, in the same order as the x-axis.
    regime_lookup : dict[str, str]
        Maps a fold key (e.g. ``"2020-Q1"`` or fold start date) to a regime label.
    palette : dict[str, str], optional
        Maps regime labels to colors. Defaults to a categorical palette.
    """
    n_folds = len(fold_dates)
    if n_folds == 0:
        return

    palette = palette or {
        "calm": COLORS.get("blue", "#1f77b4"),
        "stress": COLORS.get("red", "#d62728"),
        "drift": COLORS.get("orange", "#ff7f0e"),
        "structural_break": COLORS.get("purple", "#9467bd"),
    }

    y0, y1 = ax.get_ylim()
    span = y1 - y0
    strip_top = y0
    strip_bot = y0 - strip_height_frac * span

    for i, (start, end) in enumerate(fold_dates):
        regime = regime_lookup.get(start, regime_lookup.get(end, "calm"))
        color = palette.get(regime, "#cccccc")
        ax.add_patch(
            plt.Rectangle(
                (i + 0.5, strip_bot),
                1.0,
                strip_top - strip_bot,
                facecolor=color,
                edgecolor="white",
                linewidth=0.5,
                clip_on=False,
            )
        )

    ax.set_ylim(strip_bot - 0.02 * span, y1)


# ---------------------------------------------------------------------------
# Classification-aware (Phase C)
# ---------------------------------------------------------------------------


def classification_triple(
    case_study: str,
    classification_label: str,
    regression_label: str,
    *,
    families: list[str] | None = None,
    n_boot: int = 2000,
    block_length: int | None = None,
) -> pl.DataFrame:
    """AUC ± CI, accuracy ± CI, and IC-on-continuous ± CI per family rank-1 config.

    Loads the rank-1 prediction set per family for ``classification_label``,
    pulls AUC and accuracy from ``prediction_metrics``, and computes the IC-vs-
    continuous-return on the matched continuous label.

    Returns columns:
        family, config_name, classification_label, regression_label,
        auc, auc_ci_lo, auc_ci_hi,
        accuracy, accuracy_ci_lo, accuracy_ci_hi,
        ic_continuous, ic_ci_lo, ic_ci_hi

    Status: stub — block-bootstrap CI for AUC/accuracy not yet wired.
    Implemented at the start of Phase C (us_firm / sp500_eo / crypto).
    """
    raise NotImplementedError(
        "classification_triple is a Phase-C helper; implementation is queued for the "
        "us_firm / sp500_eo / crypto_perps walkthrough."
    )


def cross_task_matrix(
    case_study: str,
    regression_label: str,
    classification_label: str,
    *,
    families: list[str] | None = None,
) -> pl.DataFrame:
    """{regression model, classification model} × {IC vs continuous, AUC vs binary}.

    Empirical curiosity table — does the regression model accidentally classify
    well? Does the classification model's continuous score correlate with the
    continuous return?

    Coverage caveat: only feasible for families that trained on both label
    types. Confirmed coverage: ``linear`` and ``gbm``. ``latent_factors`` has 1
    classification run for ``us_firm_characteristics`` only. ``deep_learning``
    has zero classification runs across all CSs.

    Returns columns:
        family, config_name,
        ic_from_regression_model,   ic_reg_ci_lo,   ic_reg_ci_hi,
        ic_from_classification_model, ic_cls_ci_lo, ic_cls_ci_hi,
        auc_from_regression_model,  auc_reg_ci_lo, auc_reg_ci_hi,
        auc_from_classification_model, auc_cls_ci_lo, auc_cls_ci_hi

    Status: stub — implementation queued for Phase C.
    """
    raise NotImplementedError(
        "cross_task_matrix is a Phase-C helper; implementation is queued for the "
        "us_firm / sp500_eo / crypto_perps walkthrough."
    )


# ---------------------------------------------------------------------------
# Conformal coverage diagnostic (spine v2 §7)
# ---------------------------------------------------------------------------


def conformal_coverage_diagnostic(
    case_study: str,
    label: str | None = None,
    *,
    levels: tuple[float, ...] = (0.80, 0.90, 0.95),
    families: list[str] | None = None,
    embargo_steps: int | None = None,
    min_calibration_n: int = DEFAULT_MIN_CALIBRATION_N,
) -> pl.DataFrame:
    """Per-family realised coverage of the widths that size positions.

    One row per family and nominal level, measured by
    :func:`~case_studies.utils.conformal.walk_forward_conformal_coverage` on the estimator
    `conformal_weighted` allocates with. It is a diagnostic of residual dispersion, not a
    guarantee: nothing in the allocation path reads an interval or a coverage level.

    **Which configuration each row describes.** The family's rank-1 validation config by
    ``ic_mean_daily``, over the configs with the longest IC history. That is a model-level
    ranking and not the pipeline's - every selection stage ranks on validation backtest Sharpe -
    and it is used here because this diagnostic runs in the model-analysis notebook, before any
    backtest exists to rank. So a row names the family's IC leader, in the ``family`` and
    ``config_name`` columns, and says nothing about which configuration the funnel went on to
    select.

    ``embargo_steps`` defaults to :func:`~case_studies.utils.conformal.sizing_conformal_lag`
    for this case study and label, which is the reviewed horizon floored at one step.

    Returns columns:
        family, config_name, nominal_level,
        empirical_coverage, mean_interval_width_frac_std, n_test, n_uncalibrated
    """
    label = label or PRIMARY_LABELS[case_study]
    db = registry_path(case_study)
    if not db.exists():
        return _empty_frame(_CONFORMAL_COVERAGE_SCHEMA)

    family_clause = ""
    params: list = [label]
    if families:
        placeholders = ",".join("?" * len(families))
        family_clause = f"AND t.family IN ({placeholders})"
        params.extend(families)

    sql = f"""
        SELECT
            t.family,
            t.config_name,
            p.prediction_hash,
            pm.ic_mean_daily,
            pm.ic_n_days
        FROM training_runs t
        JOIN prediction_sets p ON t.training_hash = p.training_hash
        JOIN prediction_metrics pm ON p.prediction_hash = pm.prediction_hash
        WHERE t.label = ?
          AND p.split = 'validation'
          AND pm.ic_mean_daily IS NOT NULL
          {family_clause}
    """
    rows = _query(db, sql, tuple(params))
    if rows.is_empty():
        return _empty_frame(_CONFORMAL_COVERAGE_SCHEMA)

    leaders = (
        rows.with_columns(pl.col("ic_n_days").max().over("family").alias("_family_days"))
        .filter(
            pl.col("ic_n_days").is_not_null(),
            pl.col("ic_n_days") == pl.col("_family_days"),
        )
        .sort("ic_mean_daily", descending=True, nulls_last=True)
        .group_by("family")
        .first()
    )

    # Resolved here rather than at the top: a registry with no validation rows yet returns the
    # empty frame above, and a case study reaches that state before it has a reviewed horizon.
    if embargo_steps is None:
        embargo_steps = sizing_conformal_lag(case_study, label)

    pred_dir = db.parent / "predictions"
    out_rows: list[dict] = []
    for fam, cfg, p_hash in zip(
        leaders["family"].to_list(),
        leaders["config_name"].to_list(),
        leaders["prediction_hash"].to_list(),
    ):
        pq = pred_dir / p_hash / "predictions.parquet"
        if not pq.exists():
            continue
        try:
            coverage_rows = walk_forward_conformal_coverage(
                pl.read_parquet(pq),
                levels=levels,
                embargo_steps=embargo_steps,
                min_calibration_n=min_calibration_n,
            )
        except ValueError:
            continue
        out_rows.extend({"family": fam, "config_name": cfg, **row} for row in coverage_rows)

    if not out_rows:
        return _empty_frame(_CONFORMAL_COVERAGE_SCHEMA)
    return pl.DataFrame(out_rows).sort(["family", "nominal_level"])


__all__ = [
    "holdout_decay_table",
    "selection_adjusted_leader_table",
    "headline_forest_plot",
    "fold_heatmap_with_ci",
    "regime_coverage_strip",
    "classification_triple",
    "cross_task_matrix",
    "conformal_coverage_diagnostic",
]

```