fix: harden peek's audit checks against false positives and blind spots

peek/checks/causality.py

@@ -55,13 +55,57 @@ 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]
             truncated_features = ctx.feature_fn(truncated)
             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]

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}",
                     ))
 

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

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
 

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)

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

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"),

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)

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),