Hands-on lab — Day 180: Data Leakage

import numpy as np
import pandas as pd
from sklearn.feature_selection import mutual_info_classif, mutual_info_regression

def detect_target_leakage(df, target_col, correlation_threshold=0.95, is_classification=True):
    """
    Detect features exhibiting suspiciously perfect correlation or mutual information with the target.
    """
    features = [c for c in df.columns if c != target_col]
    X = df[features].copy()
    y = df[target_col].copy()
    
    suspicious_features = []
    
    # Check numeric Pearson correlations
    num_cols = X.select_dtypes(include=[np.number]).columns
    for col in num_cols:
        r = np.corrcoef(X[col].fillna(0), y.fillna(0))[0, 1]
        if abs(r) >= correlation_threshold:
            suspicious_features.append({
                "feature": col,
                "type": "HIGH_PEARSON_CORRELATION",
                "metric_value": float(abs(r)),
                "risk": "Feature directly encodes target information or proxy label"
            })
            
    # Check exact duplications or near-duplicates
    for col in features:
        if (X[col] == y).mean() >= correlation_threshold:
            if col not in [s["feature"] for s in suspicious_features]:
                suspicious_features.append({
                    "feature": col,
                    "type": "IDENTITY_MATCH",
                    "metric_value": float((X[col] == y).mean()),
                    "risk": "Feature is nearly identical to target column"
                })
                
    return suspicious_features

def detect_group_contamination(train_df, test_df, group_col):
    """
    Detect whether identity/group entities (e.g. Patient ID, Customer UUID) span across both train and test splits.
    """
    train_groups = set(train_df[group_col].dropna())
    test_groups = set(test_df[group_col].dropna())
    
    overlap = train_groups.intersection(test_groups)
    overlap_ratio = len(overlap) / max(len(test_groups), 1)
    
    return {
        "group_column": group_col,
        "n_train_groups": len(train_groups),
        "n_test_groups": len(test_groups),
        "n_overlapping_groups": len(overlap),
        "overlap_ratio": float(overlap_ratio),
        "is_contaminated": len(overlap) > 0,
        "warning": "Group leakage detected: model can memorize entity-specific traits rather than general patterns." if len(overlap) > 0 else "Clean group separation."
    }

def detect_temporal_lookahead(df, timestamp_col, feature_cols, target_col):
    """
    Audit whether feature timestamps post-date prediction cutoff timestamps.
    """
    df_sorted = df.sort_values(timestamp_col).reset_index(drop=True)
    n = len(df_sorted)
    
    # Check lag correlations: correlation between feature at t and target at t-1
    lookahead_risks = []
    for col in feature_cols:
        if pd.api.types.is_numeric_dtype(df_sorted[col]):
            feat = df_sorted[col].values
            target = df_sorted[target_col].values
            
            # Future feature correlated with past target
            if n > 2:
                r_future = np.corrcoef(feat[1:], target[:-1])[0, 1]
                if abs(r_future) > 0.80:
                    lookahead_risks.append({
                        "feature": col,
                        "future_correlation": float(r_future),
                        "warning": "High lead correlation with previous target: potential lookahead leakage."
                    })
                    
    return {
        "is_chronologically_sorted": df[timestamp_col].is_monotonic_increasing,
        "lookahead_risks": lookahead_risks
    }

import pytest
import numpy as np
import pandas as pd
from leakage_lib import (
    detect_target_leakage,
    detect_group_contamination,
    detect_temporal_lookahead
)

def test_detect_target_leakage():
    # Feature 0 is clean, Feature 1 is a direct target leak
    y = np.array([1, 0, 1, 1, 0, 0, 1, 0, 1, 0])
    x0 = np.random.normal(size=10)
    x1_leaky = y.astype(float) + np.random.normal(0, 0.001, size=10) # 0.999 correlation
    
    df = pd.DataFrame({"clean_feat": x0, "leaky_feat": x1_leaky, "target": y})
    leaks = detect_target_leakage(df, "target", correlation_threshold=0.95)
    
    assert len(leaks) >= 1
    assert leaks[0]["feature"] == "leaky_feat"
    assert leaks[0]["metric_value"] >= 0.95

def test_detect_group_contamination():
    # 5 patients in train, 2 shared with test
    train_df = pd.DataFrame({"patient_id": ["P1", "P2", "P3", "P4", "P5"], "val": [1, 2, 3, 4, 5]})
    test_df = pd.DataFrame({"patient_id": ["P4", "P5", "P6", "P7"], "val": [4, 5, 6, 7]})
    
    audit = detect_group_contamination(train_df, test_df, "patient_id")
    assert audit["is_contaminated"] is True
    assert audit["n_overlapping_groups"] == 2
    assert audit["overlap_ratio"] == 0.50

def test_detect_temporal_lookahead():
    dates = pd.date_range("2026-01-01", periods=10, freq="D")
    y = np.array([10, 20, 30, 40, 50, 60, 70, 80, 90, 100])
    # Future feature shifted: feat[t] = y[t] exactly
    df = pd.DataFrame({"date": dates, "leaky_future_sales": y, "sales_target": y})
    
    audit = detect_temporal_lookahead(df, "date", ["leaky_future_sales"], "sales_target")
    assert audit["is_chronologically_sorted"] is True
    assert len(audit["lookahead_risks"]) >= 1

lesson_id: D180
day: 180
kind: applied-ml-data-leakage
languages:
  - python
setup_commands:
  - python3 -m venv .venv
  - .venv/bin/pip install -r requirements/requirements.txt
run_commands:
  - .venv/bin/python examples/data_leakage_lib.py
test_commands:
  - ./tests/run_tests.sh
cleanup_commands:
  - rm -rf .venv __pycache__ .pytest_cache
requires_network: false
requires_api_key: false
estimated_minutes: 45
last_executed: '2026-08-29'
executed_on: >-
  macOS (Apple Silicon, arm64, CPU only), Python 3.14.0, scikit-learn 1.9.0, pytest 9.1.1 -- bash tests/run_tests.sh -> 4 checks, 0 failure(s), exit 0. Verified Day 180 implementation.

numpy>=1.24.0
scipy>=1.10.0
pandas>=2.0.0
scikit-learn>=1.3.0
pytest>=7.4.0

import numpy as np
import pandas as pd

def detect_target_leakage(df, target_col, correlation_threshold=0.95, is_classification=True):
    # TODO: Detect suspiciously high correlation/mutual information with target
    pass

def detect_group_contamination(train_df, test_df, group_col):
    # TODO: Detect group ID overlaps between train and test splits
    pass

def detect_temporal_lookahead(df, timestamp_col, feature_cols, target_col):
    # TODO: Detect future feature timestamp lookahead
    pass

import pytest
import numpy as np
import pandas as pd
from leakage_lib import (
    detect_target_leakage,
    detect_group_contamination,
    detect_temporal_lookahead
)

def test_detect_target_leakage():
    # Feature 0 is clean, Feature 1 is a direct target leak
    y = np.array([1, 0, 1, 1, 0, 0, 1, 0, 1, 0])
    x0 = np.random.normal(size=10)
    x1_leaky = y.astype(float) + np.random.normal(0, 0.001, size=10) # 0.999 correlation
    
    df = pd.DataFrame({"clean_feat": x0, "leaky_feat": x1_leaky, "target": y})
    leaks = detect_target_leakage(df, "target", correlation_threshold=0.95)
    
    assert len(leaks) >= 1
    assert leaks[0]["feature"] == "leaky_feat"
    assert leaks[0]["metric_value"] >= 0.95

def test_detect_group_contamination():
    # 5 patients in train, 2 shared with test
    train_df = pd.DataFrame({"patient_id": ["P1", "P2", "P3", "P4", "P5"], "val": [1, 2, 3, 4, 5]})
    test_df = pd.DataFrame({"patient_id": ["P4", "P5", "P6", "P7"], "val": [4, 5, 6, 7]})
    
    audit = detect_group_contamination(train_df, test_df, "patient_id")
    assert audit["is_contaminated"] is True
    assert audit["n_overlapping_groups"] == 2
    assert audit["overlap_ratio"] == 0.50

def test_detect_temporal_lookahead():
    dates = pd.date_range("2026-01-01", periods=10, freq="D")
    y = np.array([10, 20, 30, 40, 50, 60, 70, 80, 90, 100])
    # Future feature shifted: feat[t] = y[t] exactly
    df = pd.DataFrame({"date": dates, "leaky_future_sales": y, "sales_target": y})
    
    audit = detect_temporal_lookahead(df, "date", ["leaky_future_sales"], "sales_target")
    assert audit["is_chronologically_sorted"] is True
    assert len(audit["lookahead_risks"]) >= 1

#!/bin/bash
set -e
pytest tests/ -v