From 0a15226716d5cd3091cc2d90349ea0936fc1f319 Mon Sep 17 00:00:00 2001 From: Stefan Jansen Date: Mon, 20 Jul 2026 09:17:49 -0400 Subject: [PATCH 1/8] feat: add look-ahead causality auditor (audit_lookahead/assert_causal) Model-class-agnostic look-ahead gate for feature extractors based on future-perturbation invariance: corrupt input rows after a cutoff, re-run the extractor, and assert features at t <= cutoff are unchanged. A measured per-column determinism noise floor separates real leaks from unseeded jitter. - evaluation/causality.py: audit_lookahead, assert_causal, CausalityReport, CausalityError, LeakEvent; nan/shuffle/noise corruptions; auto cutoffs; reuses reporting/ renderers for to_html/to_json/to_markdown. - Exported through api.py and the package __init__; registered in the public API contract. - Synthetic Polars unit tests covering causal/leaky/nondeterministic cases, each corruption strategy, auto cutoffs, single-series keys, and rendering. --- src/ml4t/diagnostic/__init__.py | 13 + src/ml4t/diagnostic/api.py | 10 + src/ml4t/diagnostic/evaluation/causality.py | 613 ++++++++++++++++++++ tests/contracts/public_api_contract.json | 4 + tests/test_evaluation/test_causality.py | 253 ++++++++ 5 files changed, 893 insertions(+) create mode 100644 src/ml4t/diagnostic/evaluation/causality.py create mode 100644 tests/test_evaluation/test_causality.py diff --git a/src/ml4t/diagnostic/__init__.py b/src/ml4t/diagnostic/__init__.py index d0d7afe..3b4a330 100644 --- a/src/ml4t/diagnostic/__init__.py +++ b/src/ml4t/diagnostic/__init__.py @@ -64,6 +64,14 @@ # Main evaluation framework from .evaluation import BarrierAnalysis, EvaluationResult, Evaluator +# Look-ahead / feature causality auditor +from .evaluation.causality import ( + CausalityError, + CausalityReport, + assert_causal, + audit_lookahead, +) + # ValidatedCrossValidation - combines CPCV + DSR in one step from .evaluation.validated_cv import ValidatedCrossValidation @@ -116,6 +124,11 @@ "SignalResult", # Barrier Analysis "BarrierAnalysis", + # Look-ahead / feature causality auditor + "audit_lookahead", + "assert_causal", + "CausalityReport", + "CausalityError", # Configuration (10 primary configs) "DiagnosticConfig", "StatisticalConfig", diff --git a/src/ml4t/diagnostic/api.py b/src/ml4t/diagnostic/api.py index fa15c20..a33ec6c 100644 --- a/src/ml4t/diagnostic/api.py +++ b/src/ml4t/diagnostic/api.py @@ -2,6 +2,12 @@ from ml4t.diagnostic.config import DiagnosticConfig, ValidatedCrossValidationConfig from ml4t.diagnostic.evaluation.barrier_analysis import BarrierAnalysis +from ml4t.diagnostic.evaluation.causality import ( + CausalityError, + CausalityReport, + assert_causal, + audit_lookahead, +) from ml4t.diagnostic.evaluation.feature_diagnostics import ( FeatureDiagnostics, FeatureDiagnosticsResult, @@ -35,6 +41,10 @@ "ValidationResult", "validated_cross_val_score", "BarrierAnalysis", + "audit_lookahead", + "assert_causal", + "CausalityReport", + "CausalityError", "FeatureDiagnostics", "DiagnosticConfig", "FeatureDiagnosticsResult", diff --git a/src/ml4t/diagnostic/evaluation/causality.py b/src/ml4t/diagnostic/evaluation/causality.py new file mode 100644 index 0000000..c49a2cc --- /dev/null +++ b/src/ml4t/diagnostic/evaluation/causality.py @@ -0,0 +1,613 @@ +"""Model-class-agnostic look-ahead auditor for feature extractors. + +Look-ahead bias in a *model-based* feature is invisible to the eye and survives +careful-looking code. There is exactly one property that defines a causal +feature, independent of whether the underlying model is an HMM, a GARCH, a PCA, +an IPCA, an autoencoder, or an SDF: + + The feature value at ``(symbol, t)`` is a deterministic function of inputs at + times ``<= t``. Nothing after ``t`` may change it. + +This module tests that property as a **future-perturbation invariance**, without +reading the model code: destroy the inputs after a cutoff ``T``, re-run the +extractor, and assert the features at ``t <= T`` are unchanged. If any model was +fit on (or filtered over) data past ``T`` and applied backward, the past features +move and the audit fails. One mechanism catches the smoothed-HMM bug, full-sample +PCA loadings, an autoencoder trained on the whole panel, GARCH full-window +variance targeting, and a time-axis z-score - one auditor, every model class. + +Two entry points share one engine: + +- :func:`audit_lookahead` returns a :class:`CausalityReport` (the reader-facing + diagnostic, renderable to HTML/JSON/Markdown). +- :func:`assert_causal` is the CI one-liner; it raises :class:`CausalityError` + (with the report attached) on failure. + +What this deliberately does **not** catch +----------------------------------------- +- **Same-timestamp label leakage** (a feature that uses the contemporaneous + target) survives truncation - it needs a separate label-leakage check. +- **Per-fold train/val seal** - a different invariant, about the CV split rather + than the temporal causality of the extractor. + +Perturbation invariance is the uniform look-ahead gate; it is one layer, not the +whole leakage story. +""" + +from __future__ import annotations + +from collections.abc import Callable, Sequence +from dataclasses import dataclass +from typing import Any + +import numpy as np +import polars as pl + +from ml4t.diagnostic.errors import DiagnosticError +from ml4t.diagnostic.reporting.base import ReportFactory, ReportFormat +from ml4t.diagnostic.results.base import BaseResult + +Extractor = Callable[[pl.DataFrame], pl.DataFrame] + +DEFAULT_CORRUPTIONS: tuple[str, ...] = ("nan", "shuffle", "noise") +DEFAULT_KEYS: tuple[str, ...] = ("symbol", "timestamp") +DEFAULT_QUANTILES: tuple[float, ...] = (0.4, 0.6, 0.8) +_VALID_CORRUPTIONS = frozenset(DEFAULT_CORRUPTIONS) + +_NUMERIC_DTYPES = ( + pl.Float32, + pl.Float64, + pl.Int8, + pl.Int16, + pl.Int32, + pl.Int64, + pl.UInt8, + pl.UInt16, + pl.UInt32, + pl.UInt64, +) + + +# --------------------------------------------------------------------------- +# Exceptions +# --------------------------------------------------------------------------- +class CausalityError(DiagnosticError): + """Raised when a feature extractor leaks future information. + + The full :class:`CausalityReport` is attached as :attr:`report` so callers + (and CI logs) can inspect exactly which columns leaked, where, and by how + much. + + Attributes: + report: The :class:`CausalityReport` that triggered the failure. + """ + + def __init__( + self, + message: str, + report: CausalityReport, + context: dict[str, Any] | None = None, + ) -> None: + super().__init__(message, context=context) + self.report = report + + +# --------------------------------------------------------------------------- +# Result containers +# --------------------------------------------------------------------------- +@dataclass(frozen=True) +class LeakEvent: + """A single detected leak: one column exposed by one cutoff/corruption. + + Attributes: + column: Name of the leaking feature column. + cutoff: Cutoff timestamp whose future was corrupted. + corruption: Corruption strategy that exposed the leak. + first_leak: Earliest pre-cutoff timestamp whose feature value moved. + max_abs_delta: Largest absolute change over pre-cutoff rows. + threshold: Noise-floor-derived threshold the delta had to exceed. + """ + + column: str + cutoff: Any + corruption: str + first_leak: Any + max_abs_delta: float + threshold: float + + +class _CausalityResultSchema(BaseResult): + """Internal :class:`BaseResult` view so the ``reporting/`` renderers apply. + + Holds only JSON-serializable primitives; the leak table is reconstructed as + a Polars DataFrame on demand in :meth:`get_dataframe`. + """ + + analysis_type: str = "lookahead_causality" + is_causal: bool + is_deterministic: bool + noise_floor: float + feature_cols: list[str] + leaking_columns: list[str] + summary_text: str + leak_rows: list[dict[str, Any]] + + def summary(self) -> str: + return self.summary_text + + def list_available_dataframes(self) -> list[str]: + return ["leaks"] if self.leak_rows else [] + + def get_dataframe(self, name: str | None = None) -> pl.DataFrame: + if not self.leak_rows: + return pl.DataFrame() + return pl.DataFrame(self.leak_rows) + + +@dataclass(frozen=True) +class CausalityReport: + """Per-column, per-timestamp look-ahead audit result. + + Attributes: + is_causal: True iff no feature column leaked under any requested + corruption at any cutoff. + leaking_columns: Mapping ``column -> {"first_leak", "max_abs_delta", + "cutoff", "corruption"}`` for the worst (largest-delta) leak observed + for that column. Empty when :attr:`is_causal` is True. + determinism: ``{"is_deterministic": bool, "noise_floor": float}`` measured + by running the extractor twice on the unperturbed frame *before* any + perturbation. ``noise_floor`` is the max per-column noise across + feature columns; a leak is only counted when it rises above this + floor. + leak_events: Every individual leak detected (column x cutoff x + corruption), not just the per-column worst case. + feature_cols: Feature columns that were audited. + cutoffs: Cutoff timestamps used. + corruptions: Corruption strategies applied. + keys: Key columns (never corrupted). + n_rows: Number of rows in the audited frame. + """ + + is_causal: bool + leaking_columns: dict[str, dict[str, Any]] + determinism: dict[str, Any] + leak_events: tuple[LeakEvent, ...] + feature_cols: tuple[str, ...] + cutoffs: tuple[Any, ...] + corruptions: tuple[str, ...] + keys: tuple[str, ...] + n_rows: int = 0 + + # -- human-readable summary ------------------------------------------------ + def summary(self) -> str: + """Return a human-readable audit summary.""" + det = self.determinism + det_line = ( + "deterministic" + if det.get("is_deterministic", True) + else f"NON-deterministic (noise floor = {det.get('noise_floor', 0.0):.3e})" + ) + lines = [ + "Look-Ahead Causality Audit", + f" Feature columns audited : {len(self.feature_cols)} " + f"({', '.join(self.feature_cols) if self.feature_cols else 'none'})", + f" Cutoffs : {len(self.cutoffs)}", + f" Corruptions : {', '.join(self.corruptions)}", + f" Determinism : {det_line}", + f" Verdict : {'CAUSAL' if self.is_causal else 'LEAK DETECTED'}", + ] + if not self.is_causal: + lines.append(" Leaking columns:") + for col, info in self.leaking_columns.items(): + lines.append( + f" - {col}: leaks from {info['first_leak']} at " + f"{info['max_abs_delta']:.4g} under {info['corruption']} " + f"(cutoff {info['cutoff']})" + ) + return "\n".join(lines) + + # -- renderer bridge ------------------------------------------------------- + def _to_schema(self) -> _CausalityResultSchema: + leak_rows = [ + { + "column": ev.column, + "first_leak": str(ev.first_leak), + "cutoff": str(ev.cutoff), + "corruption": ev.corruption, + "max_abs_delta": float(ev.max_abs_delta), + "threshold": float(ev.threshold), + } + for ev in self.leak_events + ] + return _CausalityResultSchema( + is_causal=self.is_causal, + is_deterministic=bool(self.determinism.get("is_deterministic", True)), + noise_floor=float(self.determinism.get("noise_floor", 0.0)), + feature_cols=list(self.feature_cols), + leaking_columns=list(self.leaking_columns.keys()), + summary_text=self.summary(), + leak_rows=leak_rows, + ) + + def to_html(self) -> str: + """Render the report as a standalone HTML document.""" + return ReportFactory.render(self._to_schema(), ReportFormat.HTML) + + def to_markdown(self) -> str: + """Render the report as Markdown.""" + return ReportFactory.render(self._to_schema(), ReportFormat.MARKDOWN) + + def to_json(self) -> str: + """Render the report as a JSON string.""" + return ReportFactory.render(self._to_schema(), ReportFormat.JSON) + + +# --------------------------------------------------------------------------- +# Internal helpers +# --------------------------------------------------------------------------- +def _numeric_cols(frame: pl.DataFrame, cols: Sequence[str]) -> list[str]: + return [c for c in cols if frame.schema[c] in _NUMERIC_DTYPES] + + +def _resolve_cutoffs( + frame: pl.DataFrame, + time_col: str, + cutoffs: str | Sequence[Any], + quantiles: Sequence[float], +) -> list[Any]: + times = frame.select(time_col).unique().sort(time_col).get_column(time_col) + n = len(times) + if isinstance(cutoffs, str): + if cutoffs != "auto": + raise ValueError(f"cutoffs must be 'auto' or a sequence, got {cutoffs!r}") + if n < 2: + return [] + # Inner quantiles of the time axis; drop the max so post-cutoff rows exist. + idxs = sorted({int(round(q * (n - 1))) for q in quantiles}) + idxs = [i for i in idxs if 0 <= i < n - 1] + if not idxs: + idxs = [max(0, n - 2)] + return [times[i] for i in idxs] + resolved = list(cutoffs) + if not resolved: + raise ValueError("cutoffs sequence is empty") + return resolved + + +def _partition(frame: pl.DataFrame, group_cols: Sequence[str]) -> list[pl.DataFrame]: + if not group_cols: + return [frame] + return frame.partition_by(list(group_cols), maintain_order=True) + + +def _corrupt( + frame: pl.DataFrame, + cutoff: Any, + corruption: str, + input_cols: Sequence[str], + group_cols: Sequence[str], + time_col: str, + rng: np.random.Generator, +) -> pl.DataFrame: + """Return a copy of ``frame`` with input columns destroyed for ``t > cutoff``. + + Keys are never touched. ``corruption`` is one of ``nan``, ``shuffle`` or + ``noise``. + """ + future = pl.col(time_col) > cutoff + + if corruption == "nan": + return frame.with_columns( + [pl.when(future).then(None).otherwise(pl.col(c)).alias(c) for c in input_cols] + ) + + if corruption == "shuffle": + pre = frame.filter(~future) + post = frame.filter(future) + if post.is_empty(): + return frame + shuffled_parts: list[pl.DataFrame] = [] + for part in _partition(post, group_cols): + part = part.sort(time_col) + n = len(part) + perm = rng.permutation(n).tolist() + keys_rest = part.drop(list(input_cols)) + reordered_inputs = part.select(list(input_cols))[perm] + shuffled_parts.append(pl.concat([keys_rest, reordered_inputs], how="horizontal")) + post_shuffled = pl.concat(shuffled_parts).select(frame.columns) + return pl.concat([pre, post_shuffled]).sort(list(group_cols) + [time_col]) + + if corruption == "noise": + numeric = _numeric_cols(frame, input_cols) + if not numeric: + # No numeric inputs to resample; fall back to nulling non-numeric. + return frame.with_columns( + [pl.when(future).then(None).otherwise(pl.col(c)).alias(c) for c in input_cols] + ) + if group_cols: + agg = [pl.col(c).mean().alias(f"{c}__mean") for c in numeric] + agg += [pl.col(c).std().alias(f"{c}__std") for c in numeric] + stats = frame.group_by(list(group_cols)).agg(agg) + work = frame.join(stats, on=list(group_cols), how="left") + else: + lits = [] + for c in numeric: + lits.append(pl.lit(frame.get_column(c).mean()).alias(f"{c}__mean")) + lits.append(pl.lit(frame.get_column(c).std()).alias(f"{c}__std")) + work = frame.with_columns(lits) + exprs = [] + for c in numeric: + draw = pl.Series(c + "__z", rng.standard_normal(len(work))) + noisy = pl.col(f"{c}__mean") + draw * pl.col(f"{c}__std").fill_null(0.0) + exprs.append(pl.when(future).then(noisy).otherwise(pl.col(c)).alias(c)) + return work.with_columns(exprs).select(frame.columns) + + raise ValueError(f"unknown corruption strategy: {corruption!r}") + + +def _abs_delta_frame( + base: pl.DataFrame, + other: pl.DataFrame, + col: str, + keys: Sequence[str], + time_col: str, +) -> pl.DataFrame: + """Join two extractor outputs on keys and return (time, delta) for ``col``. + + ``delta`` is the absolute difference for numeric columns, or a large sentinel + where exactly one side is null / the values differ for non-numeric columns. + """ + left = base.select([*keys, col]) + right = other.select([*keys, pl.col(col).alias(col + "__b")]) + joined = left.join(right, on=list(keys), how="inner") + is_numeric = base.schema[col] in _NUMERIC_DTYPES + null_mismatch = pl.col(col).is_null() != pl.col(col + "__b").is_null() + if is_numeric: + raw = (pl.col(col) - pl.col(col + "__b")).abs() + delta = ( + pl.when(null_mismatch).then(float("inf")).otherwise(raw.fill_null(0.0)).alias("__delta") + ) + else: + differs = null_mismatch | (pl.col(col) != pl.col(col + "__b")) + delta = pl.when(differs).then(float("inf")).otherwise(0.0).alias("__delta") + return joined.select(pl.col(time_col), delta) + + +def _column_max_delta(delta_frame: pl.DataFrame) -> float: + val = delta_frame.select(pl.col("__delta").max()).item() + return float(val) if val is not None else 0.0 + + +# --------------------------------------------------------------------------- +# Public API +# --------------------------------------------------------------------------- +def audit_lookahead( + extract: Extractor, + frame: pl.DataFrame, + cutoffs: str | Sequence[Any] = "auto", + corruptions: Sequence[str] = DEFAULT_CORRUPTIONS, + keys: Sequence[str] = DEFAULT_KEYS, + feature_cols: Sequence[str] | None = None, + *, + quantiles: Sequence[float] = DEFAULT_QUANTILES, + seed: int = 0, + atol: float = 1e-9, + noise_multiplier: float = 5.0, +) -> CausalityReport: + """Audit a feature extractor for look-ahead bias via future-perturbation. + + The extractor is called on the unperturbed frame (twice, to measure a + per-column determinism noise floor) and then once per ``(cutoff, corruption)`` + pair on a frame whose *input* columns are destroyed for rows after the cutoff. + A feature column at ``t <= cutoff`` that moves by more than the noise floor is + a leak: its value depended on the future. + + Args: + extract: ``Callable[[pl.DataFrame], pl.DataFrame]`` mapping a + ``(symbol, timestamp, ...)`` panel to a frame carrying ``keys`` and + the feature columns. Output is aligned back to inputs on ``keys``. + frame: The input panel. Must contain all ``keys``. + cutoffs: ``"auto"`` (inner quantiles of the time axis) or an explicit + sequence of cutoff timestamps. + corruptions: Subset of ``("nan", "shuffle", "noise")``. A column is + declared causal only if invariant under *all* requested corruptions. + keys: Key columns. The last key is the time axis; the rest are entity + (e.g. ``symbol``) groups. Keys are never corrupted. + feature_cols: Columns to audit. Defaults to columns present in the + extractor output but absent from ``frame`` (i.e. the derived + features); falls back to all non-key output columns. + quantiles: Inner quantiles used when ``cutoffs="auto"``. + seed: Seed for the corruption RNG (shuffle/noise), for reproducibility. + atol: Absolute floor for the leak threshold when an extractor is exactly + deterministic (noise floor 0). + noise_multiplier: Safety factor applied to the measured per-column noise + floor when judging a leak (``delta > noise_floor * noise_multiplier``). + Values > 1 keep nondeterministic extractors from false-positiving on + their own jitter; the raw noise floor is still reported unscaled. + + Returns: + A :class:`CausalityReport`. + + Raises: + ValueError: If ``keys`` are missing, corruptions are invalid, or the + extractor output cannot be aligned to the input on ``keys``. + """ + keys = tuple(keys) + corruptions = tuple(corruptions) + if not keys: + raise ValueError("keys must be non-empty") + missing_keys = [k for k in keys if k not in frame.columns] + if missing_keys: + raise ValueError(f"frame is missing key columns: {missing_keys}") + bad = [c for c in corruptions if c not in _VALID_CORRUPTIONS] + if bad: + raise ValueError( + f"unknown corruption strategies: {bad}; valid: {sorted(_VALID_CORRUPTIONS)}" + ) + if not corruptions: + raise ValueError("corruptions must be non-empty") + + time_col = keys[-1] + group_cols = list(keys[:-1]) + input_cols = [c for c in frame.columns if c not in keys] + + # -- reference runs (also the determinism probe) -------------------------- + base_out = _align_output(extract(frame), frame, keys) + base_out_2 = _align_output(extract(frame), frame, keys) + + # -- resolve feature columns --------------------------------------------- + if feature_cols is None: + derived = [c for c in base_out.columns if c not in frame.columns] + resolved_features = derived or [c for c in base_out.columns if c not in keys] + else: + resolved_features = list(feature_cols) + missing_feats = [c for c in resolved_features if c not in base_out.columns] + if missing_feats: + raise ValueError(f"extractor output is missing feature columns: {missing_feats}") + + # -- per-column noise floor ---------------------------------------------- + noise_floor: dict[str, float] = {} + for col in resolved_features: + df = _abs_delta_frame(base_out, base_out_2, col, keys, time_col) + noise_floor[col] = _column_max_delta(df) + max_floor = max(noise_floor.values(), default=0.0) + is_deterministic = max_floor <= 0.0 + + thresholds = { + col: (fl * noise_multiplier if fl > 0.0 else atol) for col, fl in noise_floor.items() + } + + # -- perturbation sweep --------------------------------------------------- + resolved_cutoffs = _resolve_cutoffs(frame, time_col, cutoffs, quantiles) + leak_events: list[LeakEvent] = [] + counter = 0 + for cutoff in resolved_cutoffs: + for corruption in corruptions: + counter += 1 + rng = np.random.default_rng(seed + counter) + corrupted = _corrupt(frame, cutoff, corruption, input_cols, group_cols, time_col, rng) + pert_out = _align_output(extract(corrupted), frame, keys) + for col in resolved_features: + df = _abs_delta_frame(base_out, pert_out, col, keys, time_col) + df = df.filter(pl.col(time_col) <= cutoff) + if df.is_empty(): + continue + max_delta = _column_max_delta(df) + threshold = thresholds[col] + if max_delta > threshold: + first = ( + df.filter(pl.col("__delta") > threshold) + .select(pl.col(time_col).min()) + .item() + ) + leak_events.append( + LeakEvent( + column=col, + cutoff=cutoff, + corruption=corruption, + first_leak=first, + max_abs_delta=max_delta, + threshold=threshold, + ) + ) + + # -- aggregate per column (worst leak) ------------------------------------ + leaking_columns: dict[str, dict[str, Any]] = {} + for ev in leak_events: + current = leaking_columns.get(ev.column) + if current is None or ev.max_abs_delta > current["max_abs_delta"]: + leaking_columns[ev.column] = { + "first_leak": ev.first_leak, + "max_abs_delta": ev.max_abs_delta, + "cutoff": ev.cutoff, + "corruption": ev.corruption, + } + + return CausalityReport( + is_causal=len(leaking_columns) == 0, + leaking_columns=leaking_columns, + determinism={"is_deterministic": is_deterministic, "noise_floor": max_floor}, + leak_events=tuple(leak_events), + feature_cols=tuple(resolved_features), + cutoffs=tuple(resolved_cutoffs), + corruptions=corruptions, + keys=keys, + n_rows=len(frame), + ) + + +def assert_causal( + extract: Extractor, + frame: pl.DataFrame, + cutoffs: str | Sequence[Any] = "auto", + corruptions: Sequence[str] = DEFAULT_CORRUPTIONS, + keys: Sequence[str] = DEFAULT_KEYS, + feature_cols: Sequence[str] | None = None, + **kwargs: Any, +) -> CausalityReport: + """Run :func:`audit_lookahead` and raise if any feature leaks. + + Args: + extract: See :func:`audit_lookahead`. + frame: See :func:`audit_lookahead`. + cutoffs: See :func:`audit_lookahead`. + corruptions: See :func:`audit_lookahead`. + keys: See :func:`audit_lookahead`. + feature_cols: See :func:`audit_lookahead`. + **kwargs: Forwarded to :func:`audit_lookahead` (``seed``, ``atol``, + ``noise_multiplier``, ``quantiles``). + + Returns: + The :class:`CausalityReport` on success (all features causal). + + Raises: + CausalityError: If any feature column leaks; the report is attached as + :attr:`CausalityError.report`. + """ + report = audit_lookahead( + extract, + frame, + cutoffs=cutoffs, + corruptions=corruptions, + keys=keys, + feature_cols=feature_cols, + **kwargs, + ) + if not report.is_causal: + cols = ", ".join(report.leaking_columns) + raise CausalityError( + f"Look-ahead leak detected in feature column(s): {cols}", + report=report, + context={"leaking_columns": list(report.leaking_columns)}, + ) + return report + + +def _align_output( + output: pl.DataFrame, + frame: pl.DataFrame, + keys: Sequence[str], +) -> pl.DataFrame: + """Ensure the extractor output carries ``keys`` for join-based comparison. + + If keys are absent but the output is row-aligned with the input frame, the + keys are attached from the input by position. + """ + if all(k in output.columns for k in keys): + return output + if len(output) == len(frame): + return output.with_columns([frame.get_column(k) for k in keys]) + raise ValueError( + "extractor output does not contain key columns " + f"{list(keys)} and is not row-aligned with the input frame " + f"(output rows={len(output)}, input rows={len(frame)})" + ) + + +__all__ = [ + "CausalityError", + "CausalityReport", + "LeakEvent", + "audit_lookahead", + "assert_causal", +] diff --git a/tests/contracts/public_api_contract.json b/tests/contracts/public_api_contract.json index 1759bd4..249fe17 100644 --- a/tests/contracts/public_api_contract.json +++ b/tests/contracts/public_api_contract.json @@ -7,6 +7,10 @@ "ValidationResult", "validated_cross_val_score", "BarrierAnalysis", + "audit_lookahead", + "assert_causal", + "CausalityReport", + "CausalityError", "FeatureDiagnostics", "DiagnosticConfig", "FeatureDiagnosticsResult", diff --git a/tests/test_evaluation/test_causality.py b/tests/test_evaluation/test_causality.py new file mode 100644 index 0000000..da71e49 --- /dev/null +++ b/tests/test_evaluation/test_causality.py @@ -0,0 +1,253 @@ +"""Tests for the look-ahead / feature causality auditor. + +All extractors here are synthetic in-memory Polars functions - no external data. +The panels are built with a fixed seed and the leaky-vs-causal contrast is kept +crisp so the tests genuinely prove the gate discriminates. +""" + +from __future__ import annotations + +import numpy as np +import polars as pl +import pytest + +from ml4t.diagnostic import ( + CausalityError, + CausalityReport, + assert_causal, + audit_lookahead, +) +from ml4t.diagnostic.evaluation.causality import LeakEvent + +SEED = 20260720 + + +def _panel(n_per_symbol: int = 40, symbols: tuple[str, ...] = ("A", "B")) -> pl.DataFrame: + """Build a fixed-seed (symbol, timestamp, x) panel.""" + rng = np.random.default_rng(SEED) + rows: dict[str, list] = {"symbol": [], "timestamp": [], "x": []} + for s in symbols: + # Drift + noise so the whole-series mean differs sharply from any prefix mean. + walk = np.cumsum(rng.standard_normal(n_per_symbol)) + np.arange(n_per_symbol) * 0.3 + for i in range(n_per_symbol): + rows["symbol"].append(s) + rows["timestamp"].append(i) + rows["x"].append(float(walk[i])) + return pl.DataFrame(rows) + + +# --------------------------------------------------------------------------- # +# Synthetic extractors +# --------------------------------------------------------------------------- # +def causal_expanding(df: pl.DataFrame) -> pl.DataFrame: + """Causal: expanding mean within symbol (only uses inputs at t' <= t).""" + return ( + df.sort(["symbol", "timestamp"]) + .with_columns( + (pl.col("x").cum_sum().over("symbol") / (pl.col("timestamp") + 1)).alias("feat") + ) + .select("symbol", "timestamp", "feat") + ) + + +def leaky_full_mean(df: pl.DataFrame) -> pl.DataFrame: + """Leaky: centre each value on the WHOLE-series mean (peeks at the future).""" + return df.with_columns((pl.col("x") - pl.col("x").mean().over("symbol")).alias("feat")).select( + "symbol", "timestamp", "feat" + ) + + +def leaky_last_value(df: pl.DataFrame) -> pl.DataFrame: + """Leaky and order-sensitive: broadcast each symbol's LAST value backward. + + Permutation-based corruption (``shuffle``) catches this because reordering + changes which value lands at the final timestamp. + """ + return ( + df.sort(["symbol", "timestamp"]) + .with_columns(pl.col("x").last().over("symbol").alias("feat")) + .select("symbol", "timestamp", "feat") + ) + + +class _JitterExtractor: + """Nondeterministic-but-causal: expanding mean + fresh per-call jitter.""" + + def __init__(self, scale: float = 1e-5) -> None: + self._rng = np.random.default_rng(SEED) + self._scale = scale + + def __call__(self, df: pl.DataFrame) -> pl.DataFrame: + out = df.sort(["symbol", "timestamp"]).with_columns( + (pl.col("x").cum_sum().over("symbol") / (pl.col("timestamp") + 1)).alias("feat") + ) + jitter = self._rng.normal(0.0, self._scale, len(out)) + return out.with_columns((pl.col("feat") + pl.Series(jitter)).alias("feat")).select( + "symbol", "timestamp", "feat" + ) + + +# --------------------------------------------------------------------------- # +# Core discrimination +# --------------------------------------------------------------------------- # +def test_causal_extractor_passes() -> None: + report = audit_lookahead(causal_expanding, _panel()) + assert isinstance(report, CausalityReport) + assert report.is_causal is True + assert report.leaking_columns == {} + assert report.determinism["is_deterministic"] is True + assert report.determinism["noise_floor"] == 0.0 + assert report.feature_cols == ("feat",) + + +def test_leaky_extractor_is_flagged() -> None: + report = audit_lookahead(leaky_full_mean, _panel()) + assert report.is_causal is False + assert "feat" in report.leaking_columns + info = report.leaking_columns["feat"] + assert info["max_abs_delta"] > 0.0 + assert info["corruption"] in {"nan", "noise"} + # Per-column, per-timestamp evidence is populated. + assert report.leak_events + assert all(isinstance(ev, LeakEvent) for ev in report.leak_events) + assert info["first_leak"] is not None + + +def test_assert_causal_raises_with_report_attached() -> None: + # Passing case returns the report. + ok = assert_causal(causal_expanding, _panel()) + assert ok.is_causal is True + + # Failing case raises with the report attached. + with pytest.raises(CausalityError) as exc: + assert_causal(leaky_full_mean, _panel()) + assert isinstance(exc.value.report, CausalityReport) + assert exc.value.report.is_causal is False + assert "feat" in exc.value.report.leaking_columns + + +# --------------------------------------------------------------------------- # +# Determinism handling +# --------------------------------------------------------------------------- # +def test_nondeterministic_but_causal_is_not_flagged() -> None: + # Robust across seeds: jitter must never masquerade as a leak. + for seed in range(8): + report = audit_lookahead(_JitterExtractor(), _panel(), seed=seed) + assert report.is_causal is True, report.leaking_columns + assert report.determinism["is_deterministic"] is False + assert report.determinism["noise_floor"] > 0.0 + + +def test_nondeterministic_leak_still_caught_above_floor() -> None: + """A real leak on top of jitter must exceed the measured noise floor.""" + + class _JitterLeak: + def __init__(self) -> None: + self._rng = np.random.default_rng(SEED) + + def __call__(self, df: pl.DataFrame) -> pl.DataFrame: + out = df.with_columns((pl.col("x") - pl.col("x").mean().over("symbol")).alias("feat")) + jitter = self._rng.normal(0.0, 1e-5, len(out)) + return out.with_columns((pl.col("feat") + pl.Series(jitter)).alias("feat")).select( + "symbol", "timestamp", "feat" + ) + + report = audit_lookahead(_JitterLeak(), _panel(), seed=1) + assert report.determinism["is_deterministic"] is False + assert report.is_causal is False + assert report.leaking_columns["feat"]["max_abs_delta"] > report.determinism["noise_floor"] + + +# --------------------------------------------------------------------------- # +# Corruption strategies + auto cutoffs +# --------------------------------------------------------------------------- # +@pytest.mark.parametrize("corruption", ["nan", "shuffle", "noise"]) +def test_each_corruption_leaves_causal_extractor_causal(corruption: str) -> None: + report = audit_lookahead(causal_expanding, _panel(), corruptions=(corruption,)) + assert report.is_causal is True + + +@pytest.mark.parametrize( + ("corruption", "extractor"), + [ + ("nan", leaky_full_mean), + ("noise", leaky_full_mean), + ("shuffle", leaky_last_value), + ], +) +def test_each_corruption_catches_a_matching_leak(corruption, extractor) -> None: + report = audit_lookahead(extractor, _panel(), corruptions=(corruption,)) + assert report.is_causal is False + assert report.corruptions == (corruption,) + assert all(ev.corruption == corruption for ev in report.leak_events) + + +def test_auto_cutoffs_picks_inner_quantiles() -> None: + report = audit_lookahead(causal_expanding, _panel(), cutoffs="auto") + # Inner quantiles (0.4, 0.6, 0.8) of timestamps 0..39 -> round(q*39). + assert report.cutoffs == (16, 23, 31) + + +def test_explicit_cutoffs_are_used() -> None: + report = audit_lookahead(leaky_full_mean, _panel(), cutoffs=[10, 20, 30]) + assert report.cutoffs == (10, 20, 30) + assert report.is_causal is False + + +# --------------------------------------------------------------------------- # +# Single-series (no symbol) support +# --------------------------------------------------------------------------- # +def test_single_series_keys() -> None: + rng = np.random.default_rng(SEED) + df = pl.DataFrame( + { + "timestamp": list(range(50)), + "x": [float(v) for v in rng.standard_normal(50)], + } + ) + + def leaky(frame: pl.DataFrame) -> pl.DataFrame: + return frame.with_columns((pl.col("x") - pl.col("x").mean()).alias("feat")).select( + "timestamp", "feat" + ) + + report = audit_lookahead(leaky, df, keys=("timestamp",)) + assert report.is_causal is False + + +# --------------------------------------------------------------------------- # +# Reporting bridge +# --------------------------------------------------------------------------- # +def test_report_renders_all_formats() -> None: + report = audit_lookahead(leaky_full_mean, _panel()) + + html = report.to_html() + assert isinstance(html, str) and " None: + assert "CAUSAL" in audit_lookahead(causal_expanding, _panel()).summary() + assert "LEAK DETECTED" in audit_lookahead(leaky_full_mean, _panel()).summary() + + +# --------------------------------------------------------------------------- # +# Validation +# --------------------------------------------------------------------------- # +def test_missing_keys_raise() -> None: + df = pl.DataFrame({"timestamp": [1, 2, 3], "x": [1.0, 2.0, 3.0]}) + with pytest.raises(ValueError, match="missing key columns"): + audit_lookahead(causal_expanding, df) + + +def test_invalid_corruption_raises() -> None: + with pytest.raises(ValueError, match="unknown corruption"): + audit_lookahead(causal_expanding, _panel(), corruptions=("teleport",)) From 7c5be413db72b12bc974b3113686b346f43866ca Mon Sep 17 00:00:00 2001 From: Stefan Jansen Date: Mon, 20 Jul 2026 09:44:32 -0400 Subject: [PATCH 2/8] fix: harden look-ahead causality audit --- src/ml4t/diagnostic/evaluation/causality.py | 156 +++++++++++++++----- tests/test_evaluation/test_causality.py | 152 +++++++++++++++++++ 2 files changed, 269 insertions(+), 39 deletions(-) diff --git a/src/ml4t/diagnostic/evaluation/causality.py b/src/ml4t/diagnostic/evaluation/causality.py index c49a2cc..cc25aaf 100644 --- a/src/ml4t/diagnostic/evaluation/causality.py +++ b/src/ml4t/diagnostic/evaluation/causality.py @@ -257,11 +257,16 @@ def _resolve_cutoffs( ) -> list[Any]: times = frame.select(time_col).unique().sort(time_col).get_column(time_col) n = len(times) + if n < 2: + raise ValueError("look-ahead audit requires at least two distinct timestamps") if isinstance(cutoffs, str): if cutoffs != "auto": raise ValueError(f"cutoffs must be 'auto' or a sequence, got {cutoffs!r}") - if n < 2: - return [] + if not quantiles: + raise ValueError("quantiles must be non-empty when cutoffs='auto'") + bad_quantiles = [q for q in quantiles if not 0.0 < q < 1.0] + if bad_quantiles: + raise ValueError(f"quantiles must be strictly between 0 and 1, got {bad_quantiles}") # Inner quantiles of the time axis; drop the max so post-cutoff rows exist. idxs = sorted({int(round(q * (n - 1))) for q in quantiles}) idxs = [i for i in idxs if 0 <= i < n - 1] @@ -271,6 +276,14 @@ def _resolve_cutoffs( resolved = list(cutoffs) if not resolved: raise ValueError("cutoffs sequence is empty") + for cutoff in resolved: + has_past = frame.select((pl.col(time_col) <= cutoff).any()).item() + has_future = frame.select((pl.col(time_col) > cutoff).any()).item() + if not has_past or not has_future: + raise ValueError( + f"cutoff {cutoff!r} must have at least one row on each side " + f"({time_col} <= cutoff and {time_col} > cutoff)" + ) return resolved @@ -280,6 +293,26 @@ def _partition(frame: pl.DataFrame, group_cols: Sequence[str]) -> list[pl.DataFr return frame.partition_by(list(group_cols), maintain_order=True) +def _unused_column_name(columns: Sequence[str], base: str) -> str: + name = base + while name in columns: + name += "_" + return name + + +def _validate_unique_keys(frame: pl.DataFrame, keys: Sequence[str], source: str) -> None: + key_frame = frame.select(list(keys)) + null_keys = [key for key in keys if key_frame.get_column(key).null_count()] + if null_keys: + raise ValueError(f"{source} contains null key values in columns: {null_keys}") + duplicate_mask = key_frame.is_duplicated() + if duplicate_mask.any(): + examples = key_frame.filter(duplicate_mask).unique(maintain_order=True).head(5).to_dicts() + raise ValueError( + f"{source} contains duplicate keys for columns {list(keys)}; examples: {examples}" + ) + + def _corrupt( frame: pl.DataFrame, cutoff: Any, @@ -302,20 +335,28 @@ def _corrupt( ) if corruption == "shuffle": - pre = frame.filter(~future) - post = frame.filter(future) + position_col = _unused_column_name(frame.columns, "__ml4t_causality_row_index") + indexed = frame.with_row_index(position_col) + pre = indexed.filter(~future) + post = indexed.filter(future) if post.is_empty(): return frame shuffled_parts: list[pl.DataFrame] = [] for part in _partition(post, group_cols): - part = part.sort(time_col) n = len(part) - perm = rng.permutation(n).tolist() + perm = rng.permutation(n) + if n > 1 and np.array_equal(perm, np.arange(n)): + perm = np.roll(perm, 1) keys_rest = part.drop(list(input_cols)) - reordered_inputs = part.select(list(input_cols))[perm] + reordered_inputs = part.select(list(input_cols))[perm.tolist()] shuffled_parts.append(pl.concat([keys_rest, reordered_inputs], how="horizontal")) - post_shuffled = pl.concat(shuffled_parts).select(frame.columns) - return pl.concat([pre, post_shuffled]).sort(list(group_cols) + [time_col]) + post_shuffled = pl.concat(shuffled_parts).select(indexed.columns) + return ( + pl.concat([pre, post_shuffled]) + .sort(position_col) + .drop(position_col) + .select(frame.columns) + ) if corruption == "noise": numeric = _numeric_cols(frame, input_cols) @@ -324,23 +365,18 @@ def _corrupt( return frame.with_columns( [pl.when(future).then(None).otherwise(pl.col(c)).alias(c) for c in input_cols] ) - if group_cols: - agg = [pl.col(c).mean().alias(f"{c}__mean") for c in numeric] - agg += [pl.col(c).std().alias(f"{c}__std") for c in numeric] - stats = frame.group_by(list(group_cols)).agg(agg) - work = frame.join(stats, on=list(group_cols), how="left") - else: - lits = [] - for c in numeric: - lits.append(pl.lit(frame.get_column(c).mean()).alias(f"{c}__mean")) - lits.append(pl.lit(frame.get_column(c).std()).alias(f"{c}__std")) - work = frame.with_columns(lits) exprs = [] for c in numeric: - draw = pl.Series(c + "__z", rng.standard_normal(len(work))) - noisy = pl.col(f"{c}__mean") + draw * pl.col(f"{c}__std").fill_null(0.0) + if group_cols: + mean = pl.col(c).mean().over(list(group_cols)) + std = pl.col(c).std().over(list(group_cols)) + else: + mean = pl.lit(frame.get_column(c).mean()) + std = pl.lit(frame.get_column(c).std()) + draw = pl.Series(c + "__z", rng.standard_normal(len(frame))) + noisy = mean + draw * std.fill_null(0.0) exprs.append(pl.when(future).then(noisy).otherwise(pl.col(c)).alias(c)) - return work.with_columns(exprs).select(frame.columns) + return frame.with_columns(exprs).select(frame.columns) raise ValueError(f"unknown corruption strategy: {corruption!r}") @@ -357,18 +393,31 @@ def _abs_delta_frame( ``delta`` is the absolute difference for numeric columns, or a large sentinel where exactly one side is null / the values differ for non-numeric columns. """ - left = base.select([*keys, col]) - right = other.select([*keys, pl.col(col).alias(col + "__b")]) - joined = left.join(right, on=list(keys), how="inner") + selected_columns = [*keys, col] + right_col = _unused_column_name(selected_columns, f"{col}__other") + left_present = _unused_column_name([*selected_columns, right_col], "__left_present") + right_present = _unused_column_name( + [*selected_columns, right_col, left_present], "__right_present" + ) + left = base.select(selected_columns).with_columns(pl.lit(True).alias(left_present)) + right = other.select([*keys, pl.col(col).alias(right_col)]).with_columns( + pl.lit(True).alias(right_present) + ) + joined = left.join(right, on=list(keys), how="full", coalesce=True) is_numeric = base.schema[col] in _NUMERIC_DTYPES - null_mismatch = pl.col(col).is_null() != pl.col(col + "__b").is_null() + row_mismatch = pl.col(left_present).is_null() | pl.col(right_present).is_null() + null_mismatch = pl.col(col).is_null() != pl.col(right_col).is_null() if is_numeric: - raw = (pl.col(col) - pl.col(col + "__b")).abs() + nan_mismatch = pl.col(col).is_nan() != pl.col(right_col).is_nan() + raw = (pl.col(col) - pl.col(right_col)).abs().fill_nan(0.0) delta = ( - pl.when(null_mismatch).then(float("inf")).otherwise(raw.fill_null(0.0)).alias("__delta") + pl.when(row_mismatch | null_mismatch | nan_mismatch) + .then(float("inf")) + .otherwise(raw.fill_null(0.0)) + .alias("__delta") ) else: - differs = null_mismatch | (pl.col(col) != pl.col(col + "__b")) + differs = row_mismatch | null_mismatch | (pl.col(col) != pl.col(right_col)) delta = pl.when(differs).then(float("inf")).otherwise(0.0).alias("__delta") return joined.select(pl.col(time_col), delta) @@ -418,8 +467,8 @@ def audit_lookahead( features); falls back to all non-key output columns. quantiles: Inner quantiles used when ``cutoffs="auto"``. seed: Seed for the corruption RNG (shuffle/noise), for reproducibility. - atol: Absolute floor for the leak threshold when an extractor is exactly - deterministic (noise floor 0). + atol: Absolute floor for every leak threshold, including when a small + non-zero determinism noise floor is measured. noise_multiplier: Safety factor applied to the measured per-column noise floor when judging a leak (``delta > noise_floor * noise_multiplier``). Values > 1 keep nondeterministic extractors from false-positiving on @@ -429,8 +478,8 @@ def audit_lookahead( A :class:`CausalityReport`. Raises: - ValueError: If ``keys`` are missing, corruptions are invalid, or the - extractor output cannot be aligned to the input on ``keys``. + ValueError: If validation fails, a requested probe cannot modify future + inputs, or the extractor output cannot be aligned on unique ``keys``. """ keys = tuple(keys) corruptions = tuple(corruptions) @@ -439,6 +488,7 @@ def audit_lookahead( missing_keys = [k for k in keys if k not in frame.columns] if missing_keys: raise ValueError(f"frame is missing key columns: {missing_keys}") + _validate_unique_keys(frame, keys, "frame") bad = [c for c in corruptions if c not in _VALID_CORRUPTIONS] if bad: raise ValueError( @@ -446,14 +496,25 @@ def audit_lookahead( ) if not corruptions: raise ValueError("corruptions must be non-empty") + if not np.isfinite(atol) or atol < 0.0: + raise ValueError(f"atol must be a finite non-negative value, got {atol!r}") + if not np.isfinite(noise_multiplier) or noise_multiplier <= 0.0: + raise ValueError( + f"noise_multiplier must be a finite positive value, got {noise_multiplier!r}" + ) time_col = keys[-1] group_cols = list(keys[:-1]) input_cols = [c for c in frame.columns if c not in keys] + if not input_cols: + raise ValueError("frame must contain at least one non-key input column to corrupt") + resolved_cutoffs = _resolve_cutoffs(frame, time_col, cutoffs, quantiles) # -- reference runs (also the determinism probe) -------------------------- base_out = _align_output(extract(frame), frame, keys) base_out_2 = _align_output(extract(frame), frame, keys) + _validate_unique_keys(base_out, keys, "extractor output") + _validate_unique_keys(base_out_2, keys, "second extractor output") # -- resolve feature columns --------------------------------------------- if feature_cols is None: @@ -461,9 +522,14 @@ def audit_lookahead( resolved_features = derived or [c for c in base_out.columns if c not in keys] else: resolved_features = list(feature_cols) + if not resolved_features: + raise ValueError("extractor output contains no feature columns to audit") missing_feats = [c for c in resolved_features if c not in base_out.columns] if missing_feats: raise ValueError(f"extractor output is missing feature columns: {missing_feats}") + missing_second = [c for c in resolved_features if c not in base_out_2.columns] + if missing_second: + raise ValueError(f"second extractor output is missing feature columns: {missing_second}") # -- per-column noise floor ---------------------------------------------- noise_floor: dict[str, float] = {} @@ -473,12 +539,9 @@ def audit_lookahead( max_floor = max(noise_floor.values(), default=0.0) is_deterministic = max_floor <= 0.0 - thresholds = { - col: (fl * noise_multiplier if fl > 0.0 else atol) for col, fl in noise_floor.items() - } + thresholds = {col: max(fl * noise_multiplier, atol) for col, fl in noise_floor.items()} # -- perturbation sweep --------------------------------------------------- - resolved_cutoffs = _resolve_cutoffs(frame, time_col, cutoffs, quantiles) leak_events: list[LeakEvent] = [] counter = 0 for cutoff in resolved_cutoffs: @@ -486,7 +549,22 @@ def audit_lookahead( counter += 1 rng = np.random.default_rng(seed + counter) corrupted = _corrupt(frame, cutoff, corruption, input_cols, group_cols, time_col, rng) - pert_out = _align_output(extract(corrupted), frame, keys) + if frame.equals(corrupted): + raise ValueError( + f"{corruption} corruption at cutoff {cutoff!r} did not modify any " + "future input values" + ) + pert_out = _align_output(extract(corrupted), corrupted, keys) + _validate_unique_keys( + pert_out, + keys, + f"extractor output for {corruption} corruption at cutoff {cutoff!r}", + ) + missing_perturbed = [c for c in resolved_features if c not in pert_out.columns] + if missing_perturbed: + raise ValueError( + f"perturbed extractor output is missing feature columns: {missing_perturbed}" + ) for col in resolved_features: df = _abs_delta_frame(base_out, pert_out, col, keys, time_col) df = df.filter(pl.col(time_col) <= cutoff) diff --git a/tests/test_evaluation/test_causality.py b/tests/test_evaluation/test_causality.py index da71e49..655610d 100644 --- a/tests/test_evaluation/test_causality.py +++ b/tests/test_evaluation/test_causality.py @@ -194,6 +194,62 @@ def test_explicit_cutoffs_are_used() -> None: assert report.is_causal is False +def test_single_timestamp_cannot_pass_vacuously() -> None: + frame = _panel(n_per_symbol=1) + + with pytest.raises(ValueError, match="at least two distinct timestamps"): + audit_lookahead(causal_expanding, frame) + + +@pytest.mark.parametrize("cutoff", [-1, 39, 40]) +def test_explicit_cutoff_must_split_observed_timestamps(cutoff: int) -> None: + with pytest.raises(ValueError, match="must have at least one row on each side"): + audit_lookahead(causal_expanding, _panel(), cutoffs=[cutoff]) + + +def test_ineffective_corruption_cannot_pass_vacuously() -> None: + with pytest.raises(ValueError, match="shuffle.*did not modify"): + audit_lookahead( + causal_expanding, + _panel(n_per_symbol=3), + cutoffs=[1], + corruptions=("shuffle",), + ) + + +@pytest.mark.parametrize("corruption", ["shuffle", "noise"]) +def test_corruptions_preserve_unsorted_input_order(corruption: str) -> None: + frame = _panel(n_per_symbol=8).reverse() + + def row_position_feature(data: pl.DataFrame) -> pl.DataFrame: + return data.select("symbol", "timestamp").with_columns( + pl.arange(0, len(data), eager=True).cast(pl.Float64).alias("feat") + ) + + report = audit_lookahead( + row_position_feature, + frame, + cutoffs=[3], + corruptions=(corruption,), + ) + assert report.is_causal is True + + +def test_keyless_output_aligns_to_corrupted_input_order() -> None: + frame = _panel(n_per_symbol=8).reverse() + + def keyless_identity(data: pl.DataFrame) -> pl.DataFrame: + return data.select(pl.col("x").alias("feat")) + + report = audit_lookahead( + keyless_identity, + frame, + cutoffs=[3], + corruptions=("shuffle",), + ) + assert report.is_causal is True + + # --------------------------------------------------------------------------- # # Single-series (no symbol) support # --------------------------------------------------------------------------- # @@ -251,3 +307,99 @@ def test_missing_keys_raise() -> None: def test_invalid_corruption_raises() -> None: with pytest.raises(ValueError, match="unknown corruption"): audit_lookahead(causal_expanding, _panel(), corruptions=("teleport",)) + + +def test_duplicate_input_keys_raise() -> None: + frame = pl.concat([_panel(n_per_symbol=4), _panel(n_per_symbol=4).head(1)]) + + with pytest.raises(ValueError, match="frame.*duplicate keys"): + audit_lookahead(causal_expanding, frame) + + +def test_duplicate_output_keys_raise() -> None: + def duplicate_first_row(data: pl.DataFrame) -> pl.DataFrame: + output = causal_expanding(data) + return pl.concat([output, output.head(1)]) + + with pytest.raises(ValueError, match="extractor output.*duplicate keys"): + audit_lookahead(duplicate_first_row, _panel()) + + +def test_atol_is_floor_for_nondeterministic_threshold() -> None: + class TinyDrift: + def __init__(self) -> None: + self.calls = 0 + + def __call__(self, data: pl.DataFrame) -> pl.DataFrame: + initial_offsets = (0.0, 1e-12) + offset = initial_offsets[self.calls] if self.calls < 2 else 1e-10 + self.calls += 1 + return data.select( + "symbol", "timestamp", pl.lit(offset, dtype=pl.Float64).alias("feat") + ) + + report = audit_lookahead( + TinyDrift(), + _panel(n_per_symbol=4), + cutoffs=[1], + corruptions=("nan",), + atol=1e-9, + ) + assert report.is_causal is True + assert report.determinism["noise_floor"] == pytest.approx(1e-12) + + +def test_missing_pre_cutoff_output_rows_are_a_leak() -> None: + def drops_everything_if_future_is_null(data: pl.DataFrame) -> pl.DataFrame: + if data.get_column("x").null_count(): + return pl.DataFrame( + schema={"symbol": pl.String, "timestamp": pl.Int64, "feat": pl.Float64} + ) + return data.select("symbol", "timestamp", pl.col("x").alias("feat")) + + report = audit_lookahead( + drops_everything_if_future_is_null, + _panel(n_per_symbol=4), + cutoffs=[1], + corruptions=("nan",), + ) + assert report.is_causal is False + assert report.leaking_columns["feat"]["max_abs_delta"] == float("inf") + + +def test_matching_nan_features_remain_causal() -> None: + def stable_nan(data: pl.DataFrame) -> pl.DataFrame: + return data.select( + "symbol", + "timestamp", + pl.when(pl.col("timestamp") == 0) + .then(float("nan")) + .otherwise(pl.col("x")) + .alias("feat"), + ) + + report = audit_lookahead( + stable_nan, + _panel(n_per_symbol=4), + cutoffs=[1], + corruptions=("nan",), + ) + assert report.is_causal is True + + +def test_new_nan_in_pre_cutoff_features_is_a_leak() -> None: + def future_null_poisoning(data: pl.DataFrame) -> pl.DataFrame: + if data.get_column("x").null_count(): + feature = pl.lit(float("nan"), dtype=pl.Float64) + else: + feature = pl.col("x") + return data.select("symbol", "timestamp", feature.alias("feat")) + + report = audit_lookahead( + future_null_poisoning, + _panel(n_per_symbol=4), + cutoffs=[1], + corruptions=("nan",), + ) + assert report.is_causal is False + assert report.leaking_columns["feat"]["max_abs_delta"] == float("inf") From f65df900a9688b988b68ace5610ebbb5435846ea Mon Sep 17 00:00:00 2001 From: Stefan Jansen Date: Mon, 20 Jul 2026 09:44:42 -0400 Subject: [PATCH 3/8] chore: normalize integration import ordering --- src/ml4t/diagnostic/integration/__init__.py | 1 - tests/test_integration/test_backtest_profile.py | 2 +- tests/test_integration/test_backtest_result.py | 4 ++-- 3 files changed, 3 insertions(+), 4 deletions(-) diff --git a/src/ml4t/diagnostic/integration/__init__.py b/src/ml4t/diagnostic/integration/__init__.py index b37ea6c..4d62538 100644 --- a/src/ml4t/diagnostic/integration/__init__.py +++ b/src/ml4t/diagnostic/integration/__init__.py @@ -34,7 +34,6 @@ if TYPE_CHECKING: from ml4t.backtest import BacktestResult - from ml4t.diagnostic.evaluation import PortfolioAnalysis diff --git a/tests/test_integration/test_backtest_profile.py b/tests/test_integration/test_backtest_profile.py index b3bc4ce..e525857 100644 --- a/tests/test_integration/test_backtest_profile.py +++ b/tests/test_integration/test_backtest_profile.py @@ -4,9 +4,9 @@ import polars as pl import pytest + from ml4t.backtest import BacktestResult from ml4t.backtest.types import Fill, OrderSide, Trade - from ml4t.diagnostic.integration import analyze_backtest_result diff --git a/tests/test_integration/test_backtest_result.py b/tests/test_integration/test_backtest_result.py index e2f7f26..4e9cbb9 100644 --- a/tests/test_integration/test_backtest_result.py +++ b/tests/test_integration/test_backtest_result.py @@ -10,10 +10,9 @@ import numpy as np import polars as pl import pytest + from ml4t.backtest import BacktestConfig, BacktestResult from ml4t.backtest.types import Trade -from ml4t.specs import FeedSpec - from ml4t.diagnostic.integration import ( BacktestReportMetadata, compute_metrics_from_result, @@ -22,6 +21,7 @@ portfolio_analysis_from_result, profile_from_run_artifacts, ) +from ml4t.specs import FeedSpec def create_sample_result(n_trades: int = 20) -> BacktestResult: From 87878c3b22c05ad87ffbab57d4bba464c7fbf902 Mon Sep 17 00:00:00 2001 From: Stefan Jansen Date: Mon, 20 Jul 2026 09:51:59 -0400 Subject: [PATCH 4/8] fix: preserve causality corruption semantics --- src/ml4t/diagnostic/__init__.py | 2 + src/ml4t/diagnostic/api.py | 2 + src/ml4t/diagnostic/evaluation/causality.py | 32 ++++++++--- tests/contracts/public_api_contract.json | 1 + tests/test_evaluation/test_causality.py | 64 ++++++++++++++++++--- 5 files changed, 85 insertions(+), 16 deletions(-) diff --git a/src/ml4t/diagnostic/__init__.py b/src/ml4t/diagnostic/__init__.py index 3b4a330..62fe9ca 100644 --- a/src/ml4t/diagnostic/__init__.py +++ b/src/ml4t/diagnostic/__init__.py @@ -68,6 +68,7 @@ from .evaluation.causality import ( CausalityError, CausalityReport, + LeakEvent, assert_causal, audit_lookahead, ) @@ -129,6 +130,7 @@ "assert_causal", "CausalityReport", "CausalityError", + "LeakEvent", # Configuration (10 primary configs) "DiagnosticConfig", "StatisticalConfig", diff --git a/src/ml4t/diagnostic/api.py b/src/ml4t/diagnostic/api.py index a33ec6c..50ff7ef 100644 --- a/src/ml4t/diagnostic/api.py +++ b/src/ml4t/diagnostic/api.py @@ -5,6 +5,7 @@ from ml4t.diagnostic.evaluation.causality import ( CausalityError, CausalityReport, + LeakEvent, assert_causal, audit_lookahead, ) @@ -45,6 +46,7 @@ "assert_causal", "CausalityReport", "CausalityError", + "LeakEvent", "FeatureDiagnostics", "DiagnosticConfig", "FeatureDiagnosticsResult", diff --git a/src/ml4t/diagnostic/evaluation/causality.py b/src/ml4t/diagnostic/evaluation/causality.py index cc25aaf..f037c9d 100644 --- a/src/ml4t/diagnostic/evaluation/causality.py +++ b/src/ml4t/diagnostic/evaluation/causality.py @@ -344,8 +344,15 @@ def _corrupt( shuffled_parts: list[pl.DataFrame] = [] for part in _partition(post, group_cols): n = len(part) + if n == 1: + shuffled_parts.append( + part.with_columns( + [pl.lit(None, dtype=part.schema[c]).alias(c) for c in input_cols] + ) + ) + continue perm = rng.permutation(n) - if n > 1 and np.array_equal(perm, np.arange(n)): + if np.array_equal(perm, np.arange(n)): perm = np.roll(perm, 1) keys_rest = part.drop(list(input_cols)) reordered_inputs = part.select(list(input_cols))[perm.tolist()] @@ -360,11 +367,7 @@ def _corrupt( if corruption == "noise": numeric = _numeric_cols(frame, input_cols) - if not numeric: - # No numeric inputs to resample; fall back to nulling non-numeric. - return frame.with_columns( - [pl.when(future).then(None).otherwise(pl.col(c)).alias(c) for c in input_cols] - ) + non_numeric = [c for c in input_cols if c not in numeric] exprs = [] for c in numeric: if group_cols: @@ -375,7 +378,22 @@ def _corrupt( std = pl.lit(frame.get_column(c).std()) draw = pl.Series(c + "__z", rng.standard_normal(len(frame))) noisy = mean + draw * std.fill_null(0.0) - exprs.append(pl.when(future).then(noisy).otherwise(pl.col(c)).alias(c)) + exprs.append( + pl.when(future) + .then(noisy) + .otherwise(pl.col(c)) + .cast(frame.schema[c], strict=False) + .alias(c) + ) + exprs.extend( + [ + pl.when(future) + .then(pl.lit(None, dtype=frame.schema[c])) + .otherwise(pl.col(c)) + .alias(c) + for c in non_numeric + ] + ) return frame.with_columns(exprs).select(frame.columns) raise ValueError(f"unknown corruption strategy: {corruption!r}") diff --git a/tests/contracts/public_api_contract.json b/tests/contracts/public_api_contract.json index 249fe17..0ccb273 100644 --- a/tests/contracts/public_api_contract.json +++ b/tests/contracts/public_api_contract.json @@ -11,6 +11,7 @@ "assert_causal", "CausalityReport", "CausalityError", + "LeakEvent", "FeatureDiagnostics", "DiagnosticConfig", "FeatureDiagnosticsResult", diff --git a/tests/test_evaluation/test_causality.py b/tests/test_evaluation/test_causality.py index 655610d..8d034bd 100644 --- a/tests/test_evaluation/test_causality.py +++ b/tests/test_evaluation/test_causality.py @@ -14,10 +14,10 @@ from ml4t.diagnostic import ( CausalityError, CausalityReport, + LeakEvent, assert_causal, audit_lookahead, ) -from ml4t.diagnostic.evaluation.causality import LeakEvent SEED = 20260720 @@ -207,14 +207,19 @@ def test_explicit_cutoff_must_split_observed_timestamps(cutoff: int) -> None: audit_lookahead(causal_expanding, _panel(), cutoffs=[cutoff]) -def test_ineffective_corruption_cannot_pass_vacuously() -> None: - with pytest.raises(ValueError, match="shuffle.*did not modify"): - audit_lookahead( - causal_expanding, - _panel(n_per_symbol=3), - cutoffs=[1], - corruptions=("shuffle",), - ) +def test_single_future_row_shuffle_uses_value_destroying_fallback() -> None: + report = audit_lookahead( + causal_expanding, + _panel(n_per_symbol=3), + cutoffs=[1], + corruptions=("shuffle",), + ) + assert report.is_causal is True + + +def test_short_panel_with_auto_cutoffs_does_not_fail_on_shuffle() -> None: + report = audit_lookahead(causal_expanding, _panel(n_per_symbol=8)) + assert report.is_causal is True @pytest.mark.parametrize("corruption", ["shuffle", "noise"]) @@ -403,3 +408,44 @@ def future_null_poisoning(data: pl.DataFrame) -> pl.DataFrame: ) assert report.is_causal is False assert report.leaking_columns["feat"]["max_abs_delta"] == float("inf") + + +def test_noise_preserves_integer_input_dtype() -> None: + frame = _panel(n_per_symbol=8).select( + "symbol", "timestamp", pl.col("timestamp").alias("volume") + ) + + def dtype_sensitive(data: pl.DataFrame) -> pl.DataFrame: + value = 1.0 if data.schema["volume"] == pl.Int64 else 2.0 + return data.select("symbol", "timestamp").with_columns(pl.lit(value).alias("feat")) + + report = audit_lookahead( + dtype_sensitive, + frame, + cutoffs=[3], + corruptions=("noise",), + ) + assert report.is_causal is True + + +def test_noise_corrupts_non_numeric_future_inputs() -> None: + frame = _panel(n_per_symbol=8).with_columns( + pl.when(pl.col("timestamp") == 7) + .then(pl.lit("future")) + .otherwise(pl.lit("past")) + .alias("sector") + ) + + def categorical_last(data: pl.DataFrame) -> pl.DataFrame: + return data.select( + "symbol", "timestamp", pl.col("sector").last().over("symbol").alias("feat") + ) + + report = audit_lookahead( + categorical_last, + frame, + cutoffs=[3], + corruptions=("noise",), + ) + assert report.is_causal is False + assert "feat" in report.leaking_columns From a69afb81e10612b165b768e1302ac4c247d874b2 Mon Sep 17 00:00:00 2001 From: Stefan Jansen Date: Mon, 20 Jul 2026 09:59:05 -0400 Subject: [PATCH 5/8] fix: reject vacuous causality audits --- src/ml4t/diagnostic/evaluation/causality.py | 46 ++++++++++++-- tests/test_evaluation/test_causality.py | 66 +++++++++++++++++++++ 2 files changed, 108 insertions(+), 4 deletions(-) diff --git a/src/ml4t/diagnostic/evaluation/causality.py b/src/ml4t/diagnostic/evaluation/causality.py index f037c9d..c82c132 100644 --- a/src/ml4t/diagnostic/evaluation/causality.py +++ b/src/ml4t/diagnostic/evaluation/causality.py @@ -127,6 +127,8 @@ class _CausalityResultSchema(BaseResult): is_causal: bool is_deterministic: bool noise_floor: float + n_effective_probes: int + n_skipped_probes: int feature_cols: list[str] leaking_columns: list[str] summary_text: str @@ -166,6 +168,10 @@ class CausalityReport: corruptions: Corruption strategies applied. keys: Key columns (never corrupted). n_rows: Number of rows in the audited frame. + n_effective_probes: Number of cutoff/corruption pairs that changed at + least one future input value and were evaluated. + n_skipped_probes: Number of requested probes skipped because the + corruption could not change the observed future inputs. """ is_causal: bool @@ -177,6 +183,8 @@ class CausalityReport: corruptions: tuple[str, ...] keys: tuple[str, ...] n_rows: int = 0 + n_effective_probes: int = 0 + n_skipped_probes: int = 0 # -- human-readable summary ------------------------------------------------ def summary(self) -> str: @@ -193,6 +201,8 @@ def summary(self) -> str: f"({', '.join(self.feature_cols) if self.feature_cols else 'none'})", f" Cutoffs : {len(self.cutoffs)}", f" Corruptions : {', '.join(self.corruptions)}", + f" Effective probes : {self.n_effective_probes} " + f"({self.n_skipped_probes} skipped)", f" Determinism : {det_line}", f" Verdict : {'CAUSAL' if self.is_causal else 'LEAK DETECTED'}", ] @@ -223,6 +233,8 @@ def _to_schema(self) -> _CausalityResultSchema: is_causal=self.is_causal, is_deterministic=bool(self.determinism.get("is_deterministic", True)), noise_floor=float(self.determinism.get("noise_floor", 0.0)), + n_effective_probes=self.n_effective_probes, + n_skipped_probes=self.n_skipped_probes, feature_cols=list(self.feature_cols), leaking_columns=list(self.leaking_columns.keys()), summary_text=self.summary(), @@ -531,6 +543,8 @@ def audit_lookahead( # -- reference runs (also the determinism probe) -------------------------- base_out = _align_output(extract(frame), frame, keys) base_out_2 = _align_output(extract(frame), frame, keys) + if base_out.is_empty(): + raise ValueError("extractor output is empty; no feature rows are available to audit") _validate_unique_keys(base_out, keys, "extractor output") _validate_unique_keys(base_out_2, keys, "second extractor output") @@ -542,6 +556,9 @@ def audit_lookahead( resolved_features = list(feature_cols) if not resolved_features: raise ValueError("extractor output contains no feature columns to audit") + key_features = [col for col in resolved_features if col in keys] + if key_features: + raise ValueError(f"feature columns must not overlap key columns, got: {key_features}") missing_feats = [c for c in resolved_features if c not in base_out.columns] if missing_feats: raise ValueError(f"extractor output is missing feature columns: {missing_feats}") @@ -554,6 +571,12 @@ def audit_lookahead( for col in resolved_features: df = _abs_delta_frame(base_out, base_out_2, col, keys, time_col) noise_floor[col] = _column_max_delta(df) + non_finite_floors = [col for col, floor in noise_floor.items() if not np.isfinite(floor)] + if non_finite_floors: + raise ValueError( + "non-finite determinism noise floor for feature columns " + f"{non_finite_floors}; extractor row, null, or NaN output changed between base runs" + ) max_floor = max(noise_floor.values(), default=0.0) is_deterministic = max_floor <= 0.0 @@ -561,6 +584,9 @@ def audit_lookahead( # -- perturbation sweep --------------------------------------------------- leak_events: list[LeakEvent] = [] + comparison_counts = dict.fromkeys(resolved_features, 0) + n_effective_probes = 0 + n_skipped_probes = 0 counter = 0 for cutoff in resolved_cutoffs: for corruption in corruptions: @@ -568,10 +594,9 @@ def audit_lookahead( rng = np.random.default_rng(seed + counter) corrupted = _corrupt(frame, cutoff, corruption, input_cols, group_cols, time_col, rng) if frame.equals(corrupted): - raise ValueError( - f"{corruption} corruption at cutoff {cutoff!r} did not modify any " - "future input values" - ) + n_skipped_probes += 1 + continue + n_effective_probes += 1 pert_out = _align_output(extract(corrupted), corrupted, keys) _validate_unique_keys( pert_out, @@ -588,6 +613,7 @@ def audit_lookahead( df = df.filter(pl.col(time_col) <= cutoff) if df.is_empty(): continue + comparison_counts[col] += len(df) max_delta = _column_max_delta(df) threshold = thresholds[col] if max_delta > threshold: @@ -607,6 +633,16 @@ def audit_lookahead( ) ) + if n_effective_probes == 0: + raise ValueError( + "all requested corruption probes were ineffective for the observed future inputs" + ) + never_compared = [col for col, count in comparison_counts.items() if count == 0] + if never_compared: + raise ValueError( + f"no pre-cutoff comparisons were made for feature columns: {never_compared}" + ) + # -- aggregate per column (worst leak) ------------------------------------ leaking_columns: dict[str, dict[str, Any]] = {} for ev in leak_events: @@ -629,6 +665,8 @@ def audit_lookahead( corruptions=corruptions, keys=keys, n_rows=len(frame), + n_effective_probes=n_effective_probes, + n_skipped_probes=n_skipped_probes, ) diff --git a/tests/test_evaluation/test_causality.py b/tests/test_evaluation/test_causality.py index 8d034bd..f10e713 100644 --- a/tests/test_evaluation/test_causality.py +++ b/tests/test_evaluation/test_causality.py @@ -449,3 +449,69 @@ def categorical_last(data: pl.DataFrame) -> pl.DataFrame: ) assert report.is_causal is False assert "feat" in report.leaking_columns + + +def test_non_finite_determinism_floor_raises() -> None: + class UnstableRows: + def __init__(self) -> None: + self.calls = 0 + + def __call__(self, data: pl.DataFrame) -> pl.DataFrame: + self.calls += 1 + output = data.select("symbol", "timestamp", pl.col("x").alias("feat")) + return output if self.calls == 1 else output.head(len(output) - 1) + + with pytest.raises(ValueError, match="non-finite determinism noise floor.*feat"): + audit_lookahead(UnstableRows(), _panel(n_per_symbol=4)) + + +def test_empty_extractor_output_cannot_pass_vacuously() -> None: + def empty_output(data: pl.DataFrame) -> pl.DataFrame: + return pl.DataFrame(schema={"symbol": pl.String, "timestamp": pl.Int64, "feat": pl.Float64}) + + with pytest.raises(ValueError, match="extractor output is empty"): + audit_lookahead(empty_output, _panel(n_per_symbol=4)) + + +def test_output_without_pre_cutoff_comparisons_cannot_pass() -> None: + def future_only(data: pl.DataFrame) -> pl.DataFrame: + return data.filter(pl.col("timestamp") > 2).select( + "symbol", "timestamp", pl.col("x").alias("feat") + ) + + with pytest.raises(ValueError, match="no pre-cutoff comparisons.*feat"): + audit_lookahead( + future_only, + _panel(n_per_symbol=4), + cutoffs=[1], + corruptions=("nan",), + ) + + +def test_noop_probe_is_skipped_when_another_probe_is_effective() -> None: + frame = _panel(n_per_symbol=4).with_columns(pl.lit(1.0).alias("x")) + report = audit_lookahead( + causal_expanding, + frame, + cutoffs=[1], + corruptions=("shuffle", "nan"), + ) + assert report.is_causal is True + assert report.n_effective_probes == 1 + assert report.n_skipped_probes == 1 + + +def test_all_noop_probes_raise() -> None: + frame = _panel(n_per_symbol=4).with_columns(pl.lit(1.0).alias("x")) + with pytest.raises(ValueError, match="all requested corruption probes were ineffective"): + audit_lookahead( + causal_expanding, + frame, + cutoffs=[1], + corruptions=("shuffle",), + ) + + +def test_feature_columns_cannot_overlap_keys() -> None: + with pytest.raises(ValueError, match="feature columns must not overlap key columns"): + audit_lookahead(causal_expanding, _panel(), feature_cols=("timestamp",)) From 2b09c70af2474e84b9f6c9bcea51dd49dca73d1e Mon Sep 17 00:00:00 2001 From: Stefan Jansen Date: Mon, 20 Jul 2026 10:09:13 -0400 Subject: [PATCH 6/8] fix: preserve pre-cutoff corruption inputs --- src/ml4t/diagnostic/evaluation/AGENTS.md | 1 + src/ml4t/diagnostic/evaluation/__init__.py | 12 +++++++ src/ml4t/diagnostic/evaluation/causality.py | 5 ++- tests/contracts/evaluation_api_surface.json | 5 +++ tests/test_evaluation/test_causality.py | 40 +++++++++++++++++++++ 5 files changed, 60 insertions(+), 3 deletions(-) diff --git a/src/ml4t/diagnostic/evaluation/AGENTS.md b/src/ml4t/diagnostic/evaluation/AGENTS.md index aa0b4e4..64f2964 100644 --- a/src/ml4t/diagnostic/evaluation/AGENTS.md +++ b/src/ml4t/diagnostic/evaluation/AGENTS.md @@ -5,6 +5,7 @@ This package holds the main research and validation workflows. ## Main Entry Points - `ValidatedCrossValidation` for CPCV plus DSR +- `audit_lookahead` and `assert_causal` for model-agnostic feature causality checks - `FeatureDiagnostics` for feature-quality checks - `PortfolioAnalysis` for return and risk analytics - `TradeAnalysis` and `TradeShapAnalyzer` for trade-level diagnostics diff --git a/src/ml4t/diagnostic/evaluation/__init__.py b/src/ml4t/diagnostic/evaluation/__init__.py index 5fcc1f6..c41e074 100644 --- a/src/ml4t/diagnostic/evaluation/__init__.py +++ b/src/ml4t/diagnostic/evaluation/__init__.py @@ -19,6 +19,13 @@ from . import drift, stats # noqa: F401 (module re-export) from .barrier_analysis import BarrierAnalysis # noqa: F401 +from .causality import ( # noqa: F401 + CausalityError, + CausalityReport, + LeakEvent, + assert_causal, + audit_lookahead, +) from .event_analysis import EventStudyAnalysis # noqa: F401 # Factor exposure and attribution @@ -128,6 +135,11 @@ def __getattr__(name: str): "validated_cross_val_score", "ValidationFoldResult", "ValidationResult", + "audit_lookahead", + "assert_causal", + "CausalityReport", + "CausalityError", + "LeakEvent", # Analysis workflows "FeatureDiagnostics", "FeatureDiagnosticsAnalysisResult", diff --git a/src/ml4t/diagnostic/evaluation/causality.py b/src/ml4t/diagnostic/evaluation/causality.py index c82c132..4af3020 100644 --- a/src/ml4t/diagnostic/evaluation/causality.py +++ b/src/ml4t/diagnostic/evaluation/causality.py @@ -392,9 +392,8 @@ def _corrupt( noisy = mean + draw * std.fill_null(0.0) exprs.append( pl.when(future) - .then(noisy) + .then(noisy.cast(frame.schema[c], strict=False)) .otherwise(pl.col(c)) - .cast(frame.schema[c], strict=False) .alias(c) ) exprs.extend( @@ -591,7 +590,7 @@ def audit_lookahead( for cutoff in resolved_cutoffs: for corruption in corruptions: counter += 1 - rng = np.random.default_rng(seed + counter) + rng = np.random.default_rng([seed, counter]) corrupted = _corrupt(frame, cutoff, corruption, input_cols, group_cols, time_col, rng) if frame.equals(corrupted): n_skipped_probes += 1 diff --git a/tests/contracts/evaluation_api_surface.json b/tests/contracts/evaluation_api_surface.json index 35fb610..f966cb5 100644 --- a/tests/contracts/evaluation_api_surface.json +++ b/tests/contracts/evaluation_api_surface.json @@ -10,6 +10,11 @@ "validated_cross_val_score", "ValidationFoldResult", "ValidationResult", + "audit_lookahead", + "assert_causal", + "CausalityReport", + "CausalityError", + "LeakEvent", "FeatureDiagnostics", "FeatureDiagnosticsAnalysisResult", "FeatureDiagnosticsResult", diff --git a/tests/test_evaluation/test_causality.py b/tests/test_evaluation/test_causality.py index f10e713..5e3b6a0 100644 --- a/tests/test_evaluation/test_causality.py +++ b/tests/test_evaluation/test_causality.py @@ -18,6 +18,7 @@ assert_causal, audit_lookahead, ) +from ml4t.diagnostic.evaluation.causality import _corrupt SEED = 20260720 @@ -255,6 +256,45 @@ def keyless_identity(data: pl.DataFrame) -> pl.DataFrame: assert report.is_causal is True +def test_keyless_output_still_detects_a_leak() -> None: + frame = _panel(n_per_symbol=8).reverse() + + def keyless_full_mean(data: pl.DataFrame) -> pl.DataFrame: + return data.select((pl.col("x") - pl.col("x").mean().over("symbol")).alias("feat")) + + report = audit_lookahead( + keyless_full_mean, + frame, + cutoffs=[3], + corruptions=("nan",), + ) + assert report.is_causal is False + + +@pytest.mark.parametrize("corruption", ["nan", "shuffle", "noise"]) +def test_corruption_preserves_pre_cutoff_values_bit_for_bit(corruption: str) -> None: + frame = pl.DataFrame( + { + "symbol": ["A", "A", "A", "A"], + "timestamp": [0, 1, 2, 3], + "large_int": [2**53 + 1, 2**53 + 3, 2**53 + 5, 2**53 + 7], + } + ) + corrupted = _corrupt( + frame, + cutoff=1, + corruption=corruption, + input_cols=("large_int",), + group_cols=("symbol",), + time_col="timestamp", + rng=np.random.default_rng(SEED), + ) + + expected = frame.filter(pl.col("timestamp") <= 1) + actual = corrupted.filter(pl.col("timestamp") <= 1) + assert actual.equals(expected) + + # --------------------------------------------------------------------------- # # Single-series (no symbol) support # --------------------------------------------------------------------------- # From 074c9354abd02acc4c370260d1e666d569d00451 Mon Sep 17 00:00:00 2001 From: Stefan Jansen Date: Mon, 20 Jul 2026 10:20:23 -0400 Subject: [PATCH 7/8] fix: report causality audit coverage gaps --- src/ml4t/diagnostic/evaluation/causality.py | 99 ++++++++++++++++----- tests/test_evaluation/test_causality.py | 72 ++++++++++++++- 2 files changed, 146 insertions(+), 25 deletions(-) diff --git a/src/ml4t/diagnostic/evaluation/causality.py b/src/ml4t/diagnostic/evaluation/causality.py index 4af3020..5ef12ff 100644 --- a/src/ml4t/diagnostic/evaluation/causality.py +++ b/src/ml4t/diagnostic/evaluation/causality.py @@ -129,6 +129,8 @@ class _CausalityResultSchema(BaseResult): noise_floor: float n_effective_probes: int n_skipped_probes: int + skipped_probes: list[dict[str, str]] + uncovered_pairs: list[dict[str, str]] feature_cols: list[str] leaking_columns: list[str] summary_text: str @@ -172,6 +174,11 @@ class CausalityReport: least one future input value and were evaluated. n_skipped_probes: Number of requested probes skipped because the corruption could not change the observed future inputs. + skipped_probes: ``(cutoff, corruption)`` pairs that could not change + the observed future inputs. + uncovered_pairs: ``(cutoff, feature)`` pairs with no comparable + pre-cutoff extractor output. A leaking report may carry coverage + gaps, but a causal report never does. """ is_causal: bool @@ -185,6 +192,8 @@ class CausalityReport: n_rows: int = 0 n_effective_probes: int = 0 n_skipped_probes: int = 0 + skipped_probes: tuple[tuple[Any, str], ...] = () + uncovered_pairs: tuple[tuple[Any, str], ...] = () # -- human-readable summary ------------------------------------------------ def summary(self) -> str: @@ -203,6 +212,7 @@ def summary(self) -> str: f" Corruptions : {', '.join(self.corruptions)}", f" Effective probes : {self.n_effective_probes} " f"({self.n_skipped_probes} skipped)", + f" Coverage gaps : {len(self.uncovered_pairs)}", f" Determinism : {det_line}", f" Verdict : {'CAUSAL' if self.is_causal else 'LEAK DETECTED'}", ] @@ -235,6 +245,13 @@ def _to_schema(self) -> _CausalityResultSchema: noise_floor=float(self.determinism.get("noise_floor", 0.0)), n_effective_probes=self.n_effective_probes, n_skipped_probes=self.n_skipped_probes, + skipped_probes=[ + {"cutoff": str(cutoff), "corruption": corruption} + for cutoff, corruption in self.skipped_probes + ], + uncovered_pairs=[ + {"cutoff": str(cutoff), "column": column} for cutoff, column in self.uncovered_pairs + ], feature_cols=list(self.feature_cols), leaking_columns=list(self.leaking_columns.keys()), summary_text=self.summary(), @@ -438,7 +455,11 @@ def _abs_delta_frame( null_mismatch = pl.col(col).is_null() != pl.col(right_col).is_null() if is_numeric: nan_mismatch = pl.col(col).is_nan() != pl.col(right_col).is_nan() - raw = (pl.col(col) - pl.col(right_col)).abs().fill_nan(0.0) + value_mismatch = pl.col(col) != pl.col(right_col) + raw = ( + (pl.col(col).cast(pl.Float64) - pl.col(right_col).cast(pl.Float64)).abs().fill_nan(0.0) + ) + raw = pl.when(value_mismatch & (raw == 0.0)).then(1.0).otherwise(raw) delta = ( pl.when(row_mismatch | null_mismatch | nan_mismatch) .then(float("inf")) @@ -487,8 +508,9 @@ def audit_lookahead( frame: The input panel. Must contain all ``keys``. cutoffs: ``"auto"`` (inner quantiles of the time axis) or an explicit sequence of cutoff timestamps. - corruptions: Subset of ``("nan", "shuffle", "noise")``. A column is - declared causal only if invariant under *all* requested corruptions. + corruptions: Subset of ``("nan", "shuffle", "noise")``. Ineffective + probes are skipped and reported; every cutoff must retain at least + one effective probe before a causal verdict can be returned. keys: Key columns. The last key is the time axis; the rest are entity (e.g. ``symbol``) groups. Keys are never corrupted. feature_cols: Columns to audit. Defaults to columns present in the @@ -504,11 +526,15 @@ def audit_lookahead( their own jitter; the raw noise floor is still reported unscaled. Returns: - A :class:`CausalityReport`. + A :class:`CausalityReport`. Probe counts and coverage gaps are included + in the report; coverage gaps can coexist with a detected leak but never + with a causal verdict. Raises: - ValueError: If validation fails, a requested probe cannot modify future - inputs, or the extractor output cannot be aligned on unique ``keys``. + ValueError: If validation fails, the extractor output is empty or + unstable in row/null structure, every probe is ineffective, any + cutoff has no effective probe, a feature has no pre-cutoff + comparisons, or output cannot be aligned on unique ``keys``. """ keys = tuple(keys) corruptions = tuple(corruptions) @@ -583,19 +609,25 @@ def audit_lookahead( # -- perturbation sweep --------------------------------------------------- leak_events: list[LeakEvent] = [] - comparison_counts = dict.fromkeys(resolved_features, 0) + comparison_counts = { + (cutoff_index, col): 0 + for cutoff_index in range(len(resolved_cutoffs)) + for col in resolved_features + } + effective_by_cutoff = [0] * len(resolved_cutoffs) + skipped_probes: list[tuple[Any, str]] = [] n_effective_probes = 0 - n_skipped_probes = 0 counter = 0 - for cutoff in resolved_cutoffs: + for cutoff_index, cutoff in enumerate(resolved_cutoffs): for corruption in corruptions: counter += 1 rng = np.random.default_rng([seed, counter]) corrupted = _corrupt(frame, cutoff, corruption, input_cols, group_cols, time_col, rng) if frame.equals(corrupted): - n_skipped_probes += 1 + skipped_probes.append((cutoff, corruption)) continue n_effective_probes += 1 + effective_by_cutoff[cutoff_index] += 1 pert_out = _align_output(extract(corrupted), corrupted, keys) _validate_unique_keys( pert_out, @@ -612,7 +644,7 @@ def audit_lookahead( df = df.filter(pl.col(time_col) <= cutoff) if df.is_empty(): continue - comparison_counts[col] += len(df) + comparison_counts[(cutoff_index, col)] += len(df) max_delta = _column_max_delta(df) threshold = thresholds[col] if max_delta > threshold: @@ -632,16 +664,6 @@ def audit_lookahead( ) ) - if n_effective_probes == 0: - raise ValueError( - "all requested corruption probes were ineffective for the observed future inputs" - ) - never_compared = [col for col, count in comparison_counts.items() if count == 0] - if never_compared: - raise ValueError( - f"no pre-cutoff comparisons were made for feature columns: {never_compared}" - ) - # -- aggregate per column (worst leak) ------------------------------------ leaking_columns: dict[str, dict[str, Any]] = {} for ev in leak_events: @@ -654,6 +676,27 @@ def audit_lookahead( "corruption": ev.corruption, } + uncovered_cutoffs = [ + resolved_cutoffs[index] for index, count in enumerate(effective_by_cutoff) if count == 0 + ] + uncovered_pairs = tuple( + (resolved_cutoffs[cutoff_index], col) + for (cutoff_index, col), count in comparison_counts.items() + if count == 0 + ) + if not leaking_columns: + if n_effective_probes == 0: + raise ValueError( + "all requested corruption probes were ineffective for the observed future inputs" + ) + if uncovered_cutoffs: + raise ValueError(f"no effective corruption probes at cutoffs: {uncovered_cutoffs}") + if uncovered_pairs: + raise ValueError( + "no pre-cutoff comparisons were made for cutoff/feature pairs: " + f"{list(uncovered_pairs)}" + ) + return CausalityReport( is_causal=len(leaking_columns) == 0, leaking_columns=leaking_columns, @@ -665,7 +708,9 @@ def audit_lookahead( keys=keys, n_rows=len(frame), n_effective_probes=n_effective_probes, - n_skipped_probes=n_skipped_probes, + n_skipped_probes=len(skipped_probes), + skipped_probes=tuple(skipped_probes), + uncovered_pairs=uncovered_pairs, ) @@ -708,10 +753,16 @@ def assert_causal( ) if not report.is_causal: cols = ", ".join(report.leaking_columns) + coverage = ( + f"; coverage gaps at {list(report.uncovered_pairs)}" if report.uncovered_pairs else "" + ) raise CausalityError( - f"Look-ahead leak detected in feature column(s): {cols}", + f"Look-ahead leak detected in feature column(s): {cols}{coverage}", report=report, - context={"leaking_columns": list(report.leaking_columns)}, + context={ + "leaking_columns": list(report.leaking_columns), + "uncovered_pairs": list(report.uncovered_pairs), + }, ) return report diff --git a/tests/test_evaluation/test_causality.py b/tests/test_evaluation/test_causality.py index 5e3b6a0..432c2da 100644 --- a/tests/test_evaluation/test_causality.py +++ b/tests/test_evaluation/test_causality.py @@ -18,7 +18,7 @@ assert_causal, audit_lookahead, ) -from ml4t.diagnostic.evaluation.causality import _corrupt +from ml4t.diagnostic.evaluation.causality import _abs_delta_frame, _corrupt SEED = 20260720 @@ -333,6 +333,11 @@ def test_report_renders_all_formats() -> None: payload = json.loads(report.to_json()) assert payload["is_causal"] is False assert "feat" in payload["leaking_columns"] + assert payload["n_effective_probes"] == 9 + assert payload["n_skipped_probes"] == 0 + assert payload["skipped_probes"] == [] + assert payload["uncovered_pairs"] == [] + assert "Effective probes" in report.summary() def test_summary_reflects_verdict() -> None: @@ -528,6 +533,42 @@ def future_only(data: pl.DataFrame) -> pl.DataFrame: ) +def test_detected_leak_is_reported_with_uncovered_cutoff_pair() -> None: + frame = pl.DataFrame( + { + "symbol": ["A"] * 5, + "timestamp": [0, 1, 2, 3, 4], + "x": [1.0, 2.0, 3.0, 4.0, None], + } + ) + + def late_leaky_feature(data: pl.DataFrame) -> pl.DataFrame: + return ( + data.with_columns((pl.col("x") - pl.col("x").mean()).alias("feat")) + .filter(pl.col("timestamp") >= 3) + .select("symbol", "timestamp", "feat") + ) + + report = audit_lookahead( + late_leaky_feature, + frame, + cutoffs=[1, 3], + corruptions=("noise",), + ) + assert report.is_causal is False + assert "feat" in report.leaking_columns + assert report.uncovered_pairs == ((1, "feat"),) + + with pytest.raises(CausalityError) as exc: + assert_causal( + late_leaky_feature, + frame, + cutoffs=[1, 3], + corruptions=("noise",), + ) + assert exc.value.report.uncovered_pairs == ((1, "feat"),) + + def test_noop_probe_is_skipped_when_another_probe_is_effective() -> None: frame = _panel(n_per_symbol=4).with_columns(pl.lit(1.0).alias("x")) report = audit_lookahead( @@ -539,6 +580,28 @@ def test_noop_probe_is_skipped_when_another_probe_is_effective() -> None: assert report.is_causal is True assert report.n_effective_probes == 1 assert report.n_skipped_probes == 1 + assert report.skipped_probes == ((1, "shuffle"),) + + +def test_each_cutoff_requires_an_effective_probe_for_causal_verdict() -> None: + frame = pl.DataFrame( + { + "symbol": ["A"] * 4, + "timestamp": [0, 1, 2, 3], + "x": [1.0, 2.0, 3.0, None], + } + ) + + def identity(data: pl.DataFrame) -> pl.DataFrame: + return data.select("symbol", "timestamp", pl.col("x").alias("feat")) + + with pytest.raises(ValueError, match="no effective corruption probes at cutoffs.*2"): + audit_lookahead( + identity, + frame, + cutoffs=[1, 2], + corruptions=("nan",), + ) def test_all_noop_probes_raise() -> None: @@ -555,3 +618,10 @@ def test_all_noop_probes_raise() -> None: def test_feature_columns_cannot_overlap_keys() -> None: with pytest.raises(ValueError, match="feature columns must not overlap key columns"): audit_lookahead(causal_expanding, _panel(), feature_cols=("timestamp",)) + + +def test_unsigned_feature_delta_does_not_wrap() -> None: + base = pl.DataFrame({"timestamp": [0], "feat": pl.Series([5], dtype=pl.UInt64)}) + other = pl.DataFrame({"timestamp": [0], "feat": pl.Series([7], dtype=pl.UInt64)}) + delta = _abs_delta_frame(base, other, "feat", ("timestamp",), "timestamp") + assert delta.get_column("__delta").item() == 2.0 From 05787e00c49626986b371d39c0b80256e22d0ee8 Mon Sep 17 00:00:00 2001 From: Stefan Jansen Date: Mon, 20 Jul 2026 10:21:17 -0400 Subject: [PATCH 8/8] chore: bump version to 0.1.0b23 --- src/ml4t/diagnostic/__init__.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/ml4t/diagnostic/__init__.py b/src/ml4t/diagnostic/__init__.py index 62fe9ca..aca0697 100644 --- a/src/ml4t/diagnostic/__init__.py +++ b/src/ml4t/diagnostic/__init__.py @@ -31,7 +31,7 @@ exported in __all__. Breaking changes will only occur in major version bumps. """ -__version__ = "0.1.0b22" +__version__ = "0.1.0b23" # Sub-modules for advanced usage from . import (