From 2c41ac5afed7798a82b24568c5650f1769310207 Mon Sep 17 00:00:00 2001 From: Avaya Aggarwal Date: Sat, 1 Aug 2026 14:07:22 +0530 Subject: [PATCH] fix: harden peek's audit checks against false positives and blind spots MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - split: fix embargo check being a no-op on datetime time columns (previously gated on np.issubdtype(..., np.number), so it never fired for the normal datetime case); now computed in row-index space. - causality: guard against false CRITICALs from feature_fn contract violations — length-changing feature_fn (e.g. dropna()) and non-deterministic feature_fn now emit a WARNING instead of a bogus leak. - target_leak: emit a WARNING listing non-numeric columns that are skipped by the correlation test, instead of silently ignoring them. - report: CLEAN verdict now names which checks actually ran, and flags when only the shallow target_leak check fired (the common CLI case), so a CLEAN result can't be mistaken for a full audit. - cli: add --json output for CI pipeline use (AuditReport.to_dict() was already there but unreachable from the CLI). Adds 7 new tests covering each fix. Full suite: 87 passed. --- peek/checks/causality.py | 44 ++++++++++++++++++++++++++++++++++ peek/checks/split.py | 24 ++++++++++++------- peek/checks/target_leak.py | 16 ++++++++++++- peek/cli.py | 10 +++++++- peek/report.py | 13 +++++++++- tests/test_peek_causality.py | 35 +++++++++++++++++++++++++++ tests/test_peek_report.py | 23 ++++++++++++++++++ tests/test_peek_split.py | 32 +++++++++++++++++++++++++ tests/test_peek_target_leak.py | 13 ++++++++++ 9 files changed, 198 insertions(+), 12 deletions(-) diff --git a/peek/checks/causality.py b/peek/checks/causality.py index 905a2c9..2bf9c12 100644 --- a/peek/checks/causality.py +++ b/peek/checks/causality.py @@ -55,6 +55,45 @@ def run(self, ctx: AuditContext) -> list[Finding]: if not isinstance(full_features, pd.DataFrame): full_features = pd.DataFrame(full_features) + # Contract check: feature_fn must be length-preserving. If it drops + # or adds rows the probe comparison is undefined. + if len(full_features) != len(df): + return [Finding( + check=self.name, + severity=Severity.WARNING, + message=( + f"feature_fn returned {len(full_features)} rows for a " + f"{len(df)}-row DataFrame — rows were dropped or added. " + "The causality check requires a length-preserving feature_fn " + "(keep NaN rows rather than dropping them)." + ), + )] + + # Non-determinism check: run feature_fn a second time on the full df. + # If outputs differ the function is stochastic and cannot be audited. + full_features_2 = ctx.feature_fn(df) + if not isinstance(full_features_2, pd.DataFrame): + full_features_2 = pd.DataFrame(full_features_2) + numeric_cols = [ + c for c in full_features.columns + if pd.api.types.is_numeric_dtype(full_features[c]) + and c in full_features_2.columns + ] + for col in numeric_cols: + v1 = full_features[col].to_numpy(dtype=float, na_value=np.nan) + v2 = full_features_2[col].to_numpy(dtype=float, na_value=np.nan) + valid = ~(np.isnan(v1) | np.isnan(v2)) + if valid.any() and not np.allclose(v1[valid], v2[valid], atol=ATOL, rtol=RTOL): + return [Finding( + check=self.name, + severity=Severity.WARNING, + message=( + f"feature_fn is non-deterministic: feature '{col}' differs " + "across two runs on the same data. Seed all random steps " + "before using the causality check." + ), + )] + leaking_cols: dict[str, list[int]] = {} for pos in probe_positions: truncated = df.iloc[: pos + 1] @@ -62,6 +101,11 @@ def run(self, ctx: AuditContext) -> list[Finding]: if not isinstance(truncated_features, pd.DataFrame): truncated_features = pd.DataFrame(truncated_features) + # Skip this probe if truncated feature_fn contracted rows — we + # can't safely align without a shared key. + if len(truncated_features) != len(truncated): + continue + common_cols = [c for c in full_features.columns if c in truncated_features.columns] for col in common_cols: full_val = full_features[col].iloc[pos] diff --git a/peek/checks/split.py b/peek/checks/split.py index 67d7de1..11aac84 100644 --- a/peek/checks/split.py +++ b/peek/checks/split.py @@ -63,19 +63,25 @@ def run(self, ctx: AuditContext) -> list[Finding]: ), )) - train_before = train_times[train_times < test_start] - if len(train_before) > 0 and ctx.embargo > 0: - gap = ctx.embargo - boundary = train_before.max() - too_close = train_before[train_before > (test_start - gap)] \ - if np.issubdtype(times.dtype, np.number) else np.array([]) - if len(too_close) > 0: + # Embargo: check in row-index space so this works for any time + # dtype (datetime, int, float). `ctx.embargo` is a number of rows. + if ctx.embargo > 0: + test_start_pos = int(test_idx.min()) + # Train rows that are within `embargo` rows before the test + # window (excluding any already flagged as future-dated). + embargo_violators = train_idx[ + (train_idx < test_start_pos) + & (train_idx >= (test_start_pos - ctx.embargo)) + ] + if len(embargo_violators) > 0: + boundary = times[train_idx[train_idx < test_start_pos].max()] any_issue = True findings.append(Finding( check=self.name, severity=Severity.WARNING, - message=f"fold {fold_i}: {len(too_close)} training row(s) fall inside " - f"the requested embargo gap ({gap}) before the test window", + message=f"fold {fold_i}: {len(embargo_violators)} training row(s) fall " + f"inside the requested embargo gap ({ctx.embargo} rows) before " + "the test window", detail=f"last training timestamp before test: {boundary}", )) diff --git a/peek/checks/target_leak.py b/peek/checks/target_leak.py index 1ba42df..51ea0b5 100644 --- a/peek/checks/target_leak.py +++ b/peek/checks/target_leak.py @@ -36,6 +36,9 @@ def run(self, ctx: AuditContext) -> list[Finding]: if c not in (ctx.target, ctx.time_col) and pd.api.types.is_numeric_dtype(ctx.df[c]) ] + all_feature_cols = [c for c in ctx.df.columns if c not in (ctx.target, ctx.time_col)] + skipped_cols = [c for c in all_feature_cols if c not in feature_cols] + any_leak = False for col in feature_cols: feature = ctx.df[col] @@ -59,10 +62,21 @@ def run(self, ctx: AuditContext) -> list[Finding]: )) break # one finding per feature is enough + if skipped_cols: + findings.append(Finding( + check=self.name, + severity=Severity.WARNING, + message=( + f"{len(skipped_cols)} non-numeric column(s) were not checked for " + "target leakage (correlation test requires numeric features)" + ), + detail=f"skipped: {', '.join(skipped_cols)}", + )) + if not any_leak: findings.append(Finding( check=self.name, severity=Severity.PASS, - message="no feature is a near-exact copy of the (shifted) target", + message="no numeric feature is a near-exact copy of the (shifted) target", )) return findings diff --git a/peek/cli.py b/peek/cli.py index 1b172ca..1e1c5fd 100644 --- a/peek/cli.py +++ b/peek/cli.py @@ -9,6 +9,7 @@ from __future__ import annotations import argparse +import json import sys import pandas as pd @@ -49,7 +50,10 @@ def _run_demo() -> int: def _run_audit(args: argparse.Namespace) -> int: df = pd.read_csv(args.path) report = audit(df, time_col=args.time, target=args.target, horizon=args.horizon) - print(report) + if args.json: + print(json.dumps(report.to_dict(), indent=2)) + else: + print(report) return 1 if report.has_leak else 0 @@ -64,6 +68,10 @@ def build_parser() -> argparse.ArgumentParser: audit_parser.add_argument("--time", required=True, help="Name of the timestamp column") audit_parser.add_argument("--target", required=True, help="Name of the target column") audit_parser.add_argument("--horizon", type=int, default=1, help="Forecast horizon in rows") + audit_parser.add_argument( + "--json", action="store_true", default=False, + help="Output the audit report as JSON (useful for CI pipelines)", + ) return parser diff --git a/peek/report.py b/peek/report.py index d058f88..9d03d06 100644 --- a/peek/report.py +++ b/peek/report.py @@ -105,5 +105,16 @@ def render(self) -> str: elif self.verdict == "SUSPICIOUS": lines.append(f"Verdict: SUSPICIOUS — {n_warning} warning(s), no definitive leak found.") else: - lines.append("Verdict: CLEAN — no leakage detected by the checks that ran.") + ran = ", ".join(self.checks_run) if self.checks_run else "none" + lines.append( + f"Verdict: CLEAN — no leakage detected by the checks that ran ({ran})." + ) + # Warn when the deep checks were never invoked so users don't treat + # the easy CLEAN as a full bill of health. + shallow_only = set(self.checks_run) <= {"target_leak"} + if shallow_only and self.checks_run: + lines.append( + " ↳ Only the target-leak check ran. Pass feature_fn, splits, or " + "pipeline+cv+scorer to enable the causality, split, and shuffle checks." + ) return "\n".join(lines) diff --git a/tests/test_peek_causality.py b/tests/test_peek_causality.py index 5c94cdc..4962bfe 100644 --- a/tests/test_peek_causality.py +++ b/tests/test_peek_causality.py @@ -23,3 +23,38 @@ def test_causality_only_runs_when_feature_fn_given(): df = make_clean_dataset(n=100) report = peek.audit(df, time_col="date", target="target") assert "causality" not in report.checks_run + + +def test_causality_warns_on_length_changing_feature_fn(): + """A feature_fn that drops rows should get a WARNING, not a spurious CRITICAL.""" + df = make_clean_dataset(n=200) + + def dropping_fn(d): + import pandas as pd + features = pd.DataFrame(index=d.index) + features["ma5"] = d["price"].rolling(5).mean() + return features.dropna() # drops first 4 rows — violates length contract + + report = peek.audit(df, time_col="date", target="target", feature_fn=dropping_fn) + causality_findings = [f for f in report.findings if f.check == "causality"] + assert causality_findings + assert causality_findings[0].severity.value == "WARNING" + assert "length-preserving" in causality_findings[0].message + + +def test_causality_warns_on_nondeterministic_feature_fn(): + """A stochastic feature_fn should get a WARNING, not a spurious CRITICAL.""" + import numpy as np + df = make_clean_dataset(n=200) + + def stochastic_fn(d): + import pandas as pd + features = pd.DataFrame(index=d.index) + features["noisy_ma"] = d["price"].rolling(5, min_periods=1).mean() + np.random.randn(len(d)) * 0.01 + return features + + report = peek.audit(df, time_col="date", target="target", feature_fn=stochastic_fn) + causality_findings = [f for f in report.findings if f.check == "causality"] + assert causality_findings + assert causality_findings[0].severity.value == "WARNING" + assert "non-deterministic" in causality_findings[0].message diff --git a/tests/test_peek_report.py b/tests/test_peek_report.py index 343af65..dbcdbe1 100644 --- a/tests/test_peek_report.py +++ b/tests/test_peek_report.py @@ -36,6 +36,29 @@ def test_to_dict_roundtrip_shape(): assert d["findings"][0]["check"] == "target_leak" +def test_clean_verdict_names_checks_run(): + """CLEAN verdict must tell users which checks actually ran.""" + report = AuditReport( + findings=[Finding(check="target_leak", severity=Severity.PASS, message="ok")], + checks_run=["target_leak"], + ) + text = str(report) + assert "target_leak" in text + assert "feature_fn" in text # shallow-only hint must appear + + +def test_clean_verdict_no_shallow_hint_when_deep_checks_ran(): + report = AuditReport( + findings=[ + Finding(check="target_leak", severity=Severity.PASS, message="ok"), + Finding(check="causality", severity=Severity.PASS, message="ok"), + ], + checks_run=["target_leak", "causality"], + ) + text = str(report) + assert "feature_fn" not in text # hint should NOT appear + + def test_render_includes_verdict_text(): report = AuditReport(findings=[ Finding(check="causality", severity=Severity.CRITICAL, message="leak found"), diff --git a/tests/test_peek_split.py b/tests/test_peek_split.py index 04556fe..5b04c8f 100644 --- a/tests/test_peek_split.py +++ b/tests/test_peek_split.py @@ -34,3 +34,35 @@ def test_split_only_runs_when_splits_given(): df = _df() report = peek.audit(df, time_col="date", target="target") assert "split" not in report.checks_run + + +def test_embargo_fires_on_datetime_index(): + """Embargo check must work for datetime time columns (not just numeric).""" + df = _df() # date column is pd.Timestamp — the previously broken case + train_idx = np.arange(0, 50) + test_idx = np.arange(55, len(df)) + # embargo=10: rows 45-54 are within 10 rows of test start (row 55) + # → rows 45-49 in train_idx violate the embargo + report = peek.audit( + df, time_col="date", target="target", + splits=[(train_idx, test_idx)], + embargo=10, + ) + split_findings = [f for f in report.findings if f.check == "split"] + assert any(f.severity.value == "WARNING" for f in split_findings), ( + "embargo check failed to fire on a datetime-indexed dataframe" + ) + + +def test_embargo_does_not_fire_when_gap_is_sufficient(): + df = _df() + train_idx = np.arange(0, 40) + test_idx = np.arange(55, len(df)) + # gap = 15 rows, embargo = 10 → no violation + report = peek.audit( + df, time_col="date", target="target", + splits=[(train_idx, test_idx)], + embargo=10, + ) + split_findings = [f for f in report.findings if f.check == "split"] + assert not any(f.severity.value == "WARNING" for f in split_findings) diff --git a/tests/test_peek_target_leak.py b/tests/test_peek_target_leak.py index 6b3501a..306d17f 100644 --- a/tests/test_peek_target_leak.py +++ b/tests/test_peek_target_leak.py @@ -19,6 +19,19 @@ def test_clean_dataset_target_leak_check_passes(): assert report.verdict == "CLEAN" +def test_non_numeric_features_emit_warning(): + """Categorical columns must get a WARNING instead of being silently skipped.""" + df = pd.DataFrame({ + "date": pd.date_range("2020-01-01", periods=50), + "sector": ["tech"] * 50, # non-numeric + "target": range(50), + }) + report = peek.audit(df, time_col="date", target="target") + target_findings = [f for f in report.findings if f.check == "target_leak"] + messages = " ".join(f.message for f in target_findings) + assert "non-numeric" in messages or "skipped" in " ".join(f.detail for f in target_findings) + + def test_shifted_duplicate_of_target_is_flagged(): df = pd.DataFrame({ "date": pd.date_range("2020-01-01", periods=50),