해시 기반 실행 재사용 및 예측 범위 점검

# completeness.py


```py
"""Registry completeness checks and skip-if-exists logic.

Provides a single entry point for "should I train this config?" decisions
across all model families (linear, gbm, tabular_dl, deep_learning,
latent_factors, causal_dml) and for backtests.

Contract
--------
Every training notebook should guard each config:

    spec = build_training_spec(...)
    status = training_run_status(CASE_STUDY_ID, spec)
    if status.complete and not FORCE_RETRAIN:
        print(f"  {cfg['config_name']}: SKIP — {status.summary()}")
        continue
    if status.partial:
        print(f"  {cfg['config_name']}: RETRAIN — {status.summary()}")
    # ... train and register

Every backtest sweep should guard each variant:

    strategy_spec = build_backtest_spec(...)
    status = backtest_run_status(CASE_STUDY_ID, pred_hash, strategy_spec)
    if status.complete and not FORCE_REBACKTEST:
        print(f"  {variant_name}: SKIP — backtest already complete")
        continue
    # ... run backtest

Rationale
---------
Large sweeps (GBM on 9.2M-row us_equities_panel, nasdaq100 microstructure,
DL families) can take hours. Re-running from scratch after a correctness
fix, partial interruption, or added configs wastes compute. Re-running
training where the training_hash already has complete artifacts is pure
waste — the hash IS the identity. If the hash exists and has all expected
artifacts, the result is reproducible and can be reused.

The only legitimate reasons to retrain:
1. The fix or config change produces a NEW hash (handled automatically).
2. The existing artifacts are corrupt or partially written.
3. FORCE_RETRAIN=True (explicit opt-in for debugging).

Partial state handling
----------------------
If some artifacts exist but not all (e.g., training_runs row but no
predictions.parquet), report the partial state and retrain. We NEVER
silently reuse a partial state because the result would be misleading
(the ic_mean might exist while the predictions are gone).
"""

from __future__ import annotations

import json
from dataclasses import dataclass
from pathlib import Path

from .specs import backtest_hash_from_parts, training_hash_from_spec
from .store import (
    _backtest_dir,
    _case_dir,
    _open_registry,
    _prediction_dir,
)


@dataclass(frozen=True)
class PredictionCoverage:
    """Exact expected-versus-actual prediction coverage evidence.

    ``expected`` is the family's own target. ``_prepare_expected_keys`` builds it from the
    fold inputs the adapter prepared, after the model's own eligibility filtering, so
    ``complete`` here means "this run produced what it set out to produce". It does not and
    cannot mean "this run covered the declared universe": the declared universe never enters
    the comparison. A sequence model that narrows its own target scores ``complete`` here and
    50.9% on :func:`notebook_contracts.measure_prediction_cross_sections`, from the same
    predictions - measured on ``sp500_equity_option_analytics``, where all 140 members ruled
    inadmissible carry a row here reading ``complete`` with ``n_missing = 0``.

    ``status`` is stored, but a stored row can only ever read ``complete``: registration
    raises on a partial evaluation and ``allow_partial`` defaults to False everywhere in
    production, so the non-complete evaluations are refused before the row is written.
    Readers should test the counts rather than the verdict, which is what
    :func:`notebook_contracts.incompletely_registered_predictions` does - though on the
    corpus as it stands that is a change of what is read and not of what can be seen. Every
    count and digest here is computed against the same ``expected_keys``, so they agree with
    ``status`` on all 6,480 rows today; they can disagree with it only on a row written by
    something that stores a shortfall instead of refusing it.
    """

    expected_key_digest: str
    actual_key_digest: str
    n_expected: int
    n_actual: int
    n_duplicates: int
    n_missing: int
    n_extra: int
    n_null: int
    n_non_finite: int
    n_folds_expected: int
    n_folds_actual: int
    schema_json: str
    status: str

    @property
    def complete(self) -> bool:
        return self.status == "complete"

    def as_dict(self) -> dict[str, str | int]:
        return {
            "expected_key_digest": self.expected_key_digest,
            "actual_key_digest": self.actual_key_digest,
            "n_expected": self.n_expected,
            "n_actual": self.n_actual,
            "n_duplicates": self.n_duplicates,
            "n_missing": self.n_missing,
            "n_extra": self.n_extra,
            "n_null": self.n_null,
            "n_non_finite": self.n_non_finite,
            "n_folds_expected": self.n_folds_expected,
            "n_folds_actual": self.n_folds_actual,
            "schema_json": self.schema_json,
            "status": self.status,
        }


# The rendering `_canonical_key_column` implements, stamped onto every digest it produces.
# `k1` is the rendering it implemented before it normalized temporal columns, under which a
# `Date` rendered `2016-01-29` and a `Datetime("ms")` of the same instant rendered
# `2016-01-29 00:00:00.000`, so the same key set digested two ways. Bump `KEY_DIGEST_RENDERING`
# whenever `_canonical_key_column` changes what it emits, and keep the older rendering
# reachable from `_canonical_key_column` so a digest that predates the stamp stays identifiable.
KEY_DIGEST_RENDERING = "k2"
LEGACY_KEY_DIGEST_RENDERING = "k1"


def _prediction_key_columns(frame) -> tuple[str, ...]:
    columns = set(frame.columns)
    entities = [name for name in ("symbol", "product") if name in columns]
    if len(entities) != 1:
        raise ValueError("prediction coverage requires exactly one of symbol or product")
    return (
        entities[0],
        *(("position",) if "position" in columns else ()),
        "timestamp",
        "fold_id",
    )


def _canonical_key_frame(
    frame,
    key_columns: tuple[str, ...] | None = None,
    *,
    rendering: str = KEY_DIGEST_RENDERING,
):
    import polars as pl

    if not isinstance(frame, pl.DataFrame):
        frame = pl.from_pandas(frame)
    if "fold" in frame.columns and "fold_id" not in frame.columns:
        frame = frame.rename({"fold": "fold_id"})
    if key_columns is None:
        key_columns = _prediction_key_columns(frame)
    required = set(key_columns)
    missing = required - set(frame.columns)
    if missing:
        raise ValueError(
            f"prediction coverage requires columns {sorted(required)}; missing {missing}"
        )
    return frame.select(
        *(_canonical_key_column(frame, name, rendering=rendering) for name in key_columns)
    )


def _canonical_key_column(frame, name: str, *, rendering: str = KEY_DIGEST_RENDERING):
    """One key column rendered so two frames from different paths can be joined on it.

    Casting a temporal column straight to String renders whatever dtype it happens to carry:
    a `Date` becomes `2016-01-29` and a `Datetime("ms")` of the same instant becomes
    `2016-01-29 00:00:00.000`. The two never join, so a registry frame meeting a dataset
    frame reports every expected row missing and every actual row extra - a 100% mismatch
    that reads as a data problem and is a dtype problem. Family runners never see it because
    both of their frames come from one source.

    Every temporal dtype is therefore normalized to microsecond UTC first, so the rendering
    is decided by this function rather than by which loader produced the frame.

    *rendering* selects which version of that decision to apply. `k1` is what this function
    did before it normalized, kept only so a stored digest that names no rendering can be
    identified by recomputing it both ways rather than assumed to be one of them.
    """
    import polars as pl

    if name == "fold_id":
        return pl.col(name).cast(pl.Int64)
    if rendering == LEGACY_KEY_DIGEST_RENDERING:
        return pl.col(name).cast(pl.String)
    dtype = frame.schema[name]
    if dtype == pl.Date or isinstance(dtype, pl.Datetime):
        return pl.col(name).cast(pl.Datetime("us")).cast(pl.String).alias(name)
    return pl.col(name).cast(pl.String)


def key_digest_value(digest: str) -> str:
    """*digest* without its rendering prefix: the part two digests are equal on."""
    _, sep, value = str(digest).partition(":")
    return value if sep else str(digest)


def key_digest_rendering(digest: str, *, expected_keys=None) -> str | None:
    """Which rendering produced *digest*, or ``None`` where that cannot be established.

    A digest carrying no prefix was written before the prefix existed, and that is all it
    says: measured across the fleet 2026-09-07, 4,303 of the 4,469 stored digests reproduce
    under the current rendering and 166 - all `us_equities_panel`'s - under `k1`, and every
    one of them is stored bare. Reading a bare digest as `k1` would therefore have declared
    a rendering change on 4,303 rows where there was none, and refuse the next checkpoint
    registered against any of them.

    Pass *expected_keys* to settle it by measurement instead: the key frame is digested
    under each known rendering and the one that reproduces *digest* is the one that produced
    it. Where neither does, the frames hold different keys, which is a different question,
    and this returns ``None`` rather than guessing.
    """
    prefix, sep, _ = str(digest).partition(":")
    if sep:
        return prefix
    if expected_keys is None:
        return None
    for rendering in (KEY_DIGEST_RENDERING, LEGACY_KEY_DIGEST_RENDERING):
        if key_digest_value(coverage_key_digest(expected_keys, rendering=rendering)) == str(digest):
            return rendering
    return None


def coverage_key_digest(expected_keys, *, rendering: str = KEY_DIGEST_RENDERING) -> str:
    """The coverage key digest *expected_keys* takes under *rendering*."""
    frame = _canonical_key_frame(expected_keys, rendering=rendering)
    return _key_digest(frame, tuple(frame.columns), rendering=rendering)


def require_comparable_key_digests(digests, *, what: str, expected_keys=None) -> None:
    """Refuse to treat digests from different renderings as comparable.

    Two digests taken under different renderings are unequal whatever their key sets, and
    nothing about the inequality says so - which is how a rendering change turns into a
    silent mis-grouping rather than an error. A consumer that
    groups predictions by their eligibility digest reports more distinct contracts than
    exist and tells a reader that two checkpoints scored on identical rows are not
    comparable.

    Measured when #1065 was filed: grouping each registry by the stored digest against one
    uniform rendering split one set in sp500_options, one in us_firm_characteristics and two
    in cme_futures, and nothing raised - the guard at the consumer only rejects a group whose
    members disagree on `n_expected`, `n_actual` or `n_folds`, and the split halves agree on
    all three.

    Only renderings that can be established are compared. A digest that names none and that
    *expected_keys* cannot identify makes no claim, and refusing on it would refuse the
    whole fleet as written today.
    """
    renderings = sorted(
        {
            rendering
            for digest in digests
            if digest
            for rendering in (key_digest_rendering(str(digest), expected_keys=expected_keys),)
            if rendering is not None
        }
    )
    if len(renderings) > 1:
        raise ValueError(
            f"{what} spans coverage-key renderings {renderings}, so its digests cannot be "
            "compared: two digests taken under different renderings are unequal whatever "
            "their key sets. Re-register the rows written under the older rendering, or "
            "compare within one rendering."
        )


def _key_digest(frame, key_columns: tuple[str, ...], rendering: str = KEY_DIGEST_RENDERING) -> str:
    from case_studies.utils.artifact_digest import value_digest

    return f"{rendering}:{value_digest(frame, key_columns)}"


def evaluate_prediction_coverage(expected_keys, predictions) -> PredictionCoverage:
    """Compare exact prediction keys and finite scores without mutating storage."""
    import polars as pl

    expected = _canonical_key_frame(expected_keys)
    key_columns = tuple(expected.columns)
    actual = _canonical_key_frame(predictions, key_columns)
    if expected.n_unique(key_columns) != expected.height:
        raise ValueError("expected prediction coverage keys must be unique")

    unique_actual = actual.unique(key_columns)
    n_duplicates = actual.height - unique_actual.height
    n_missing = expected.join(unique_actual, on=key_columns, how="anti").height
    n_extra = unique_actual.join(expected, on=key_columns, how="anti").height

    if not isinstance(predictions, pl.DataFrame):
        predictions = pl.from_pandas(predictions)
    score_col = "y_score" if "y_score" in predictions.columns else "prediction"
    if score_col not in predictions.columns:
        raise ValueError("prediction coverage requires y_score or prediction")
    score = predictions.get_column(score_col).cast(pl.Float64, strict=False)
    n_null = score.null_count()
    n_non_finite = (score.is_not_null() & ~score.is_finite()).sum()
    expected_digest = _key_digest(expected, key_columns)
    actual_digest = _key_digest(unique_actual, key_columns)
    complete = not any((n_duplicates, n_missing, n_extra, n_null, n_non_finite)) and (
        expected_digest == actual_digest
    )
    return PredictionCoverage(
        expected_key_digest=expected_digest,
        actual_key_digest=actual_digest,
        n_expected=expected.height,
        n_actual=actual.height,
        n_duplicates=n_duplicates,
        n_missing=n_missing,
        n_extra=n_extra,
        n_null=n_null,
        n_non_finite=int(n_non_finite),
        n_folds_expected=expected.get_column("fold_id").n_unique(),
        n_folds_actual=actual.get_column("fold_id").n_unique(),
        schema_json=json.dumps(
            {name: str(dtype) for name, dtype in predictions.schema.items()},
            sort_keys=True,
            separators=(",", ":"),
        ),
        status="complete" if complete else "partial",
    )


# ---------------------------------------------------------------------------
# Dataclasses
# ---------------------------------------------------------------------------


@dataclass(frozen=True)
class TrainingRunStatus:
    """Completeness status of a training run in the registry.

    Fields
    ------
    training_hash : str
        Canonical identity hash from the spec.
    exists : bool
        True if the training_runs row exists.
    has_predictions : bool
        True if at least one prediction_sets row exists.
    has_predictions_file : bool
        True if at least one predictions.parquet file exists on disk.
    has_metrics : bool
        True if the prediction has an ic_mean value.
    complete : bool
        True if all required artifacts are present.
    partial : bool
        True if the run exists but some artifacts are missing.
    missing : tuple[str, ...]
        Names of missing artifacts.
    """

    training_hash: str
    exists: bool
    has_predictions: bool
    has_predictions_file: bool
    has_metrics: bool
    missing: tuple[str, ...] = ()

    @property
    def complete(self) -> bool:
        return not self.missing and self.exists

    @property
    def partial(self) -> bool:
        return self.exists and bool(self.missing)

    def summary(self) -> str:
        if not self.exists:
            return f"no training_run for hash {self.training_hash[:12]}"
        if self.complete:
            return f"complete (hash={self.training_hash[:12]})"
        return f"partial (hash={self.training_hash[:12]}, missing: {', '.join(self.missing)})"


@dataclass(frozen=True)
class BacktestRunStatus:
    """Completeness status of a backtest run in the registry."""

    backtest_hash: str
    exists: bool
    has_returns: bool
    has_metrics: bool
    missing: tuple[str, ...] = ()

    @property
    def complete(self) -> bool:
        return not self.missing and self.exists

    @property
    def partial(self) -> bool:
        return self.exists and bool(self.missing)

    def summary(self) -> str:
        if not self.exists:
            return f"no backtest_run for hash {self.backtest_hash[:12]}"
        if self.complete:
            return f"complete (hash={self.backtest_hash[:12]})"
        return f"partial (hash={self.backtest_hash[:12]}, missing: {', '.join(self.missing)})"


# ---------------------------------------------------------------------------
# Training run completeness
# ---------------------------------------------------------------------------


def training_run_status(
    case_study: str,
    spec: dict,
    *,
    require_metrics: bool = True,
    require_predictions_file: bool = True,
    case_dir: Path | None = None,
) -> TrainingRunStatus:
    """Inspect the registry for a training run matching the given spec.

    Parameters
    ----------
    case_study : str
        Case study id.
    spec : dict
        Complete training spec (same structure build_training_spec produces).
    require_metrics : bool
        Whether ic_mean must be non-NULL for the run to count as complete.
        Default True. Causal DML runs are tracked in `causal_runs`, not
        through this path.
    require_predictions_file : bool
        Whether predictions.parquet must exist on disk. Default True.
    case_dir : Path, optional
        Override case study directory.

    Returns
    -------
    TrainingRunStatus
    """
    if case_dir is None:
        case_dir = _case_dir(case_study)

    t_hash = training_hash_from_spec(spec)

    db = _open_registry(case_dir)
    try:
        row = db.execute(
            "SELECT training_hash FROM training_runs WHERE training_hash = ?",
            (t_hash,),
        ).fetchone()
        exists = row is not None
        if not exists:
            return TrainingRunStatus(
                training_hash=t_hash,
                exists=False,
                has_predictions=False,
                has_predictions_file=False,
                has_metrics=False,
                missing=("training_run",),
            )

        # Prediction sets
        pred_hashes = [
            r[0]
            for r in db.execute(
                "SELECT prediction_hash FROM prediction_sets WHERE training_hash = ?",
                (t_hash,),
            ).fetchall()
        ]
        has_predictions = len(pred_hashes) > 0

        # Metrics on the prediction(s)
        has_metrics = False
        if has_predictions:
            # Get any prediction with non-null ic_mean
            q = (
                f"SELECT prediction_hash FROM prediction_metrics "
                f"WHERE prediction_hash IN ({','.join('?' * len(pred_hashes))}) "
                f"AND ic_mean IS NOT NULL"
            )
            m_rows = db.execute(q, tuple(pred_hashes)).fetchall()
            has_metrics = len(m_rows) > 0
    finally:
        db.close()

    # Check predictions.parquet files on disk
    has_predictions_file = False
    if has_predictions:
        for ph in pred_hashes:
            f = _prediction_dir(case_dir, ph) / "predictions.parquet"
            if f.exists():
                has_predictions_file = True
                break

    missing = []
    if not has_predictions:
        missing.append("prediction_sets")
    if require_predictions_file and not has_predictions_file:
        missing.append("predictions.parquet")
    if require_metrics and not has_metrics:
        missing.append("ic_mean")

    return TrainingRunStatus(
        training_hash=t_hash,
        exists=exists,
        has_predictions=has_predictions,
        has_predictions_file=has_predictions_file,
        has_metrics=has_metrics,
        missing=tuple(missing),
    )


def skip_training_if_complete(
    case_study: str,
    spec: dict,
    *,
    force_retrain: bool = False,
    verbose: bool = True,
    **kwargs,
) -> TrainingRunStatus:
    """Convenience wrapper for the "should I train?" decision.

    Returns the status. Caller should check ``status.complete`` and
    ``force_retrain`` to decide whether to skip.

    When ``verbose=True``, prints a one-line status for partial/complete runs
    so interactive runs get visible feedback.

    Example
    -------
        status = skip_training_if_complete(CASE_STUDY_ID, spec,
                                          force_retrain=FORCE_RETRAIN)
        if status.complete and not FORCE_RETRAIN:
            print(f"  {cfg_name}: SKIP ({status.summary()})")
            continue
    """
    status = training_run_status(case_study, spec, **kwargs)
    if verbose:
        if status.complete and not force_retrain:
            return status  # caller prints
        if status.partial:
            print(f"  WARNING: partial run detected, will retrain: {status.summary()}")
    return status


# ---------------------------------------------------------------------------
# Backtest run completeness
# ---------------------------------------------------------------------------


def backtest_run_status(
    case_study: str,
    prediction_hash: str,
    strategy_spec: dict,
    *,
    require_metrics: bool = True,
    require_returns_file: bool = True,
    case_dir: Path | None = None,
) -> BacktestRunStatus:
    """Inspect the registry for a backtest run matching prediction_hash + strategy_spec."""
    if case_dir is None:
        case_dir = _case_dir(case_study)

    b_hash = backtest_hash_from_parts(prediction_hash, strategy_spec)

    db = _open_registry(case_dir)
    try:
        row = db.execute(
            "SELECT backtest_hash FROM backtest_runs WHERE backtest_hash = ?",
            (b_hash,),
        ).fetchone()
        exists = row is not None
        if not exists:
            return BacktestRunStatus(
                backtest_hash=b_hash,
                exists=False,
                has_returns=False,
                has_metrics=False,
                missing=("backtest_run",),
            )

        has_metrics = False
        if require_metrics:
            # A NULL sharpe means "no metric pass has run here" everywhere except one
            # case: a path the engine stopped at ruin registers every ranking metric as
            # NULL on purpose, so that nothing sorts a bankrupt account against a solvent
            # one. Reading that as incomplete would re-run the
            # backtest on every pass and never converge, so `ruin` is what separates
            # them. `ruin` is absent from a registry written before that column existed,
            # in which case a NULL sharpe still means unmeasured.
            has_ruin = "ruin" in {
                row[1] for row in db.execute("PRAGMA table_info(backtest_metrics)").fetchall()
            }
            measured = "sharpe IS NOT NULL OR ruin = 1.0" if has_ruin else "sharpe IS NOT NULL"
            m_row = db.execute(
                f"SELECT sharpe FROM backtest_metrics WHERE backtest_hash = ? AND ({measured})",
                (b_hash,),
            ).fetchone()
            has_metrics = m_row is not None
    finally:
        db.close()

    # Check returns.parquet on disk
    has_returns = (_backtest_dir(case_dir, b_hash) / "daily_returns.parquet").exists()

    missing = []
    if require_returns_file and not has_returns:
        missing.append("daily_returns.parquet")
    if require_metrics and not has_metrics:
        missing.append("sharpe")

    return BacktestRunStatus(
        backtest_hash=b_hash,
        exists=exists,
        has_returns=has_returns,
        has_metrics=has_metrics,
        missing=tuple(missing),
    )


def skip_backtest_if_complete(
    case_study: str,
    prediction_hash: str,
    strategy_spec: dict,
    *,
    force_rebacktest: bool = False,
    verbose: bool = True,
    **kwargs,
) -> BacktestRunStatus:
    """Convenience wrapper for the "should I backtest?" decision.

    Example
    -------
        status = skip_backtest_if_complete(CASE_STUDY_ID, pred_hash, spec,
                                          force_rebacktest=FORCE_REBACKTEST)
        if status.complete and not FORCE_REBACKTEST:
            print(f"  {variant_name}: SKIP ({status.summary()})")
            continue
    """
    status = backtest_run_status(case_study, prediction_hash, strategy_spec, **kwargs)
    if verbose:
        if status.partial:
            print(f"  WARNING: partial backtest detected, will re-run: {status.summary()}")
    return status


__all__ = [
    "TrainingRunStatus",
    "BacktestRunStatus",
    "training_run_status",
    "skip_training_if_complete",
    "backtest_run_status",
    "skip_backtest_if_complete",
]

```