#!/usr/bin/env python3
"""Scan a training table for target leakage: features that know the answer.
Checks (codes, severity):
LD001 high single-feature out-of-fold AUC > --auc (default 0.9) / |Spearman| > 0.95 for numeric targets
LD002 high missingness of a column predicts the target (is-missing AUC > 0.85): filled after the event
LD003 high a datetime column is later than the prediction time (--time-col) in > 1% of rows
LD004 medium column name suggests an outcome or post-event field (status, reason, refund, chargeback,
cancel, closed, resolution, outcome, result, final, settled, approved...)
LD005 medium row order or a numeric ID predicts the target (AUC of the row index / ID > 0.65):
data sorted by outcome, or IDs assigned after labelling
LD006 medium a feature is (almost) a deterministic function of the target: one value per class
(categorical) or |Spearman| > 0.98 (numeric target)
LD007 low identical feature rows with different labels or duplicated rows (label noise / joins)
LD008 info a feature has much stronger signal in the oldest than the newest third of rows
(--time-col): backfilled or overwritten history
Usage:
python leakage_scan.py data.csv --target churned
python leakage_scan.py data.csv --target churned --time-col score_date --drop customer_id
python leakage_scan.py data.csv --target price --json
Exit codes: 0 no high findings, 1 high findings, 2 usage or read error.
"""
import argparse
import csv
import json
import math
import re
import sys
import zlib
from collections import Counter, defaultdict
MISSING = {"", "na", "n/a", "nan", "null", "none", "?"}
OUTCOME_NAMES = re.compile(r"(status|reason|refund|chargeback|cancel|closed|close_date|resolution|resolved|outcome|"
r"result|final|settled|approved|denied|decision|disposition|churn|default|paid_off|"
r"written_off|collection|_after|post_|label|target|response)", re.I)
DATE_RE = re.compile(r"^\d{4}[-/]\d{1,2}[-/]\d{1,2}")
def num(v):
try:
x = float(v)
return x if math.isfinite(x) else None
except (TypeError, ValueError):
return None
def ranks(xs):
order = sorted(range(len(xs)), key=xs.__getitem__)
r = [0.0] * len(xs)
i = 0
while i < len(order):
j = i
while j + 1 < len(order) and xs[order[j + 1]] == xs[order[i]]:
j += 1
for k in range(i, j + 1):
r[order[k]] = (i + j) / 2 + 1
i = j + 1
return r
def auc(s, y):
pos = sum(y)
neg = len(y) - pos
if not pos or not neg:
return None
r = ranks(s)
return (sum(ri for ri, t in zip(r, y) if t) - pos * (pos + 1) / 2) / (pos * neg)
def spearman(a, b):
ra, rb = ranks(a), ranks(b)
n = len(a)
ma, mb = sum(ra) / n, sum(rb) / n
va = math.sqrt(sum((x - ma) ** 2 for x in ra))
vb = math.sqrt(sum((x - mb) ** 2 for x in rb))
return 0.0 if not va or not vb else sum((x - ma) * (y - mb) for x, y in zip(ra, rb)) / (va * vb)
def oof_level_scores(vals, y, prior):
folds = [zlib.crc32(str(i).encode()) % 2 for i in range(len(vals))]
out = [prior] * len(vals)
for f in (0, 1):
st = defaultdict(lambda: [0.0, 0])
for v, t, k in zip(vals, y, folds):
if k != f:
st[v][0] += t
st[v][1] += 1
for i, (v, k) in enumerate(zip(vals, folds)):
if k == f:
s, n = st.get(v, (0.0, 0))
out[i] = (s + 5 * prior) / (n + 5)
return out
def signal(vals, y, binary, prior):
"""(strength 0..1, description) for one column, out-of-fold for categoricals."""
idx = [i for i, v in enumerate(vals) if v.lower() not in MISSING]
if len(idx) < 20:
return None, ""
xs = [num(vals[i]) for i in idx]
ys = [y[i] for i in idx]
numeric = sum(x is not None for x in xs) / len(xs) > 0.95
if numeric:
s = [x if x is not None else 0.0 for x in xs]
else:
s = oof_level_scores([vals[i] for i in idx], ys, prior)
if binary:
a = auc(s, ys)
return (None, "") if a is None else (max(a, 1 - a), f"AUC {max(a, 1 - a):.3f}")
rho = spearman(s, ys)
return abs(rho), f"spearman {rho:+.3f}"
def main():
ap = argparse.ArgumentParser(description="Scan a training table for target leakage.", epilog=__doc__,
formatter_class=argparse.RawDescriptionHelpFormatter)
ap.add_argument("csv")
ap.add_argument("--target", required=True)
ap.add_argument("--time-col", help="prediction-time column (scoring date / event time)")
ap.add_argument("--drop", nargs="*", default=[], help="columns to skip (keys you already know about)")
ap.add_argument("--auc", type=float, default=0.9, help="single-feature AUC threshold for LD001 (default 0.9)")
ap.add_argument("--json", action="store_true")
a = ap.parse_args()
try:
with (sys.stdin if a.csv == "-" else open(a.csv, newline="", encoding="utf-8-sig")) as fh:
reader = csv.DictReader(fh)
header = reader.fieldnames or []
rows = [r for r in reader if (r.get(a.target) or "").strip().lower() not in MISSING]
except OSError as e:
print(f"error: {e}", file=sys.stderr)
return 2
if a.target not in header or len(rows) < 30:
print(f"error: target {a.target!r} missing or fewer than 30 labelled rows", file=sys.stderr)
return 2
if a.time_col and a.time_col not in header:
print(f"error: time column {a.time_col!r} not found", file=sys.stderr)
return 2
tv = [r[a.target].strip() for r in rows]
classes = Counter(tv)
binary = len(classes) == 2
if binary:
pos = min(classes, key=classes.get)
y = [1.0 if v == pos else 0.0 for v in tv]
elif all(num(v) is not None for v in tv):
y = [float(v) for v in tv]
else:
print("error: target must be binary or numeric", file=sys.stderr)
return 2
prior = sum(y) / len(y)
findings = []
def add(code, sev, col, msg):
findings.append({"code": code, "severity": sev, "column": col, "message": msg})
feats = [c for c in header if c not in (a.target, a.time_col) and c not in a.drop]
# LD005: row order
if binary:
ra = auc(list(range(len(rows))), y)
if ra is not None and max(ra, 1 - ra) > 0.65:
add("LD005", "medium", "(row order)", f"row position predicts the target (AUC {max(ra, 1 - ra):.2f}): the file "
"is sorted by outcome or time; shuffle or split by time deliberately")
for c in feats:
vals = [(r.get(c) or "").strip() for r in rows]
present = [v for v in vals if v.lower() not in MISSING]
if not present:
continue
miss = 1 - len(present) / len(vals)
if OUTCOME_NAMES.search(c):
add("LD004", "medium", c, "name suggests an outcome or post-event field; confirm it is known at prediction time")
if binary and 0.01 < miss < 0.99:
m = auc([1.0 if v.lower() in MISSING else 0.0 for v in vals], y)
if m is not None and max(m, 1 - m) > 0.85:
add("LD002", "high", c, f"missingness predicts the target (is-missing AUC {max(m, 1 - m):.3f}): the field "
"is filled because the outcome happened")
is_date = sum(bool(DATE_RE.match(v)) for v in present) / len(present) > 0.9
if is_date and a.time_col:
later = sum(1 for r, v in zip(rows, vals) if v and DATE_RE.match(v) and v[:19] > (r.get(a.time_col) or "")[:19])
share = later / len(present)
if share > 0.01:
add("LD003", "high", c, f"{share:.0%} of values are later than {a.time_col}: information from the future")
continue
if is_date:
continue
sig, desc = signal(vals, y, binary, prior)
if sig is None:
continue
nums = [num(v) for v in present]
numeric = sum(x is not None for x in nums) / len(nums) > 0.95
if binary and sig > a.auc:
add("LD001", "high", c, f"single feature {desc} > {a.auc}: almost certainly leakage or a proxy of the label")
elif not binary and sig > 0.95:
add("LD001", "high", c, f"single feature {desc}: nearly the target itself")
if not numeric and binary:
by_level = defaultdict(set)
for v, t in zip(vals, y):
if v.lower() not in MISSING:
by_level[v].add(t)
pure = sum(1 for s in by_level.values() if len(s) == 1)
if len(by_level) > 1 and pure == len(by_level) and len(by_level) <= 0.5 * len(present):
add("LD006", "medium", c, f"every level of {c} maps to a single class: deterministic of the target")
if numeric and not binary and sig > 0.98:
add("LD006", "medium", c, "near-perfect rank correlation with the target: derived from it?")
if numeric and binary and re.search(r"(^id$|_id$|^id_|number|_no$)", c, re.I):
if sig > 0.65:
add("LD005", "medium", c, f"identifier predicts the target ({desc}): IDs assigned in outcome order")
if a.time_col and sig > 0.65:
order = sorted(range(len(rows)), key=lambda i: rows[i].get(a.time_col, ""))
third = len(order) // 3
old, new = order[:third], order[-third:]
s_old, _ = signal([vals[i] for i in old], [y[i] for i in old], binary, prior)
s_new, _ = signal([vals[i] for i in new], [y[i] for i in new], binary, prior)
if s_old and s_new and s_old - s_new > 0.15:
add("LD008", "info", c, f"signal {s_old:.2f} in the oldest third vs {s_new:.2f} in the newest: "
"history may be backfilled or overwritten")
seen = defaultdict(set)
dup = 0
for r, t in zip(rows, tv):
key = tuple((r.get(c) or "").strip() for c in feats)
if key in seen:
dup += 1
seen[key].add(t)
conflict = sum(1 for s in seen.values() if len(s) > 1)
if conflict:
add("LD007", "low", "(rows)", f"{conflict} identical feature vectors carry different labels (label noise, or "
"the distinguishing information is missing)")
if dup:
add("LD007", "low", "(rows)", f"{dup} duplicated feature rows: duplicates across train/test inflate scores")
order = {"high": 0, "medium": 1, "low": 2, "info": 3}
findings.sort(key=lambda f: (order[f["severity"]], f["code"], f["column"]))
high = sum(f["severity"] == "high" for f in findings)
if a.json:
print(json.dumps({"rows": len(rows), "target": a.target, "findings": findings}, indent=2))
return 1 if high else 0
print(f"Leakage scan of {len(feats)} feature(s) on {len(rows)} labelled rows (target {a.target!r})")
if not findings:
print("No leakage signals found. Still confirm every source's write time with its owner.")
for f in findings:
print(f"{f['code']} [{f['severity']}] {f['column']}: {f['message']}")
print(f"\n{len(findings)} finding(s), {high} high")
return 1 if high else 0
if __name__ == "__main__":
sys.exit(main())
anomaly-detection-model-builder
Build and evaluate anomaly detection for tabular, sensor and transaction data with few or no labels: define the anomaly type (point, contextual, collective), choose detectors (robust z-score/MAD and IQR rules, Mahalanobis with robust covariance, k-NN distance, LOF, Isolation Forest, one-class SVM, PCA reconstruction / autoencoders, forecasting residuals for time series), combine them by rank, explain each alert by its deviating features, set thresholds from an alert budget or the score distribution, and evaluate with precision@k, PR-AUC and recall on known incidents, or detector agreement when no labels exist. Includes pure-Python detectors (robust z, Mahalanobis, kNN, Isolation Forest, ensemble) and an evaluation script.
catboost-model-builder
Build CatBoost models that use its strengths correctly: raw categorical columns as cat_features (ordered target statistics, no manual encoding), text_features, Pool construction with NaN-safe categories, ordered boosting and has_time for temporal data, symmetric (oblivious) trees, overfitting detector with eval_set and use_best_model, losses for classification, regression (Quantile, Poisson, Tweedie) and ranking (YetiRank), class weights, monotone constraints, feature importance (PredictionValuesChange, LossFunctionChange) and SHAP values, GPU training and model export. Plans cat/text/numeric columns from a CSV and generates a runnable training script, and checks parameter sets for the conflicts CatBoost rejects (iterations vs n_estimators, Bayesian bootstrap with subsample, multiple class-weight options, depth > 16, use_best_model without eval_set).
categorical-encoding-selector
Choose and implement the right encoding for every categorical column given the model family and the data: one-hot (with infrequent-level grouping and handle_unknown), ordinal with an explicit order, cross-fitted target encoding (scikit-learn TargetEncoder, CatBoost ordered statistics, category_encoders inside CV), frequency/count encoding, feature hashing, native categorical support in LightGBM, HistGradientBoosting, XGBoost and CatBoost, prefix levels for codes (zip3, NAICS-2), and embeddings. Measures cardinality, rare-level share, unseen-level rate on a time-ordered holdout and target signal per column, emits the ColumnTransformer, and quantifies target-encoding leakage (in-sample vs out-of-fold vs holdout AUC).
classification-model-selector
Choose the right classification algorithm for tabular, text or mixed data and prove the choice with a fair bake-off: logistic regression (L2, L1, elastic net), linear SVM, naive Bayes, k-NN, RBF SVM, decision trees, random forest, extra trees, HistGradientBoosting, LightGBM, XGBoost, CatBoost, explainable boosting machines (EBM/GAM), points scorecards and MLPs, matched to data size, width, categorical and high-cardinality share, missingness, class imbalance, text, and constraints such as interpretability, monotonic constraints, calibrated probabilities, latency budgets and tiny deployable models. Ranks candidates with starting hyperparameters and cautions, then generates a bake-off script running every candidate on identical CV folds with per-fold metrics, fit time and prediction latency.
clustering-algorithm-selector
Choose the right clustering algorithm and check that clustering makes sense at all: cluster tendency (Hopkins statistic with a correlated-features caveat, nearest-neighbour isolation), then k-means / MiniBatchKMeans, Gaussian mixtures with BIC, agglomerative (Ward, average, complete, single) with dendrograms, DBSCAN, HDBSCAN, OPTICS, spectral clustering, BIRCH, mean shift, k-prototypes and k-modes for mixed or categorical data, and Gower distance, matched to data size, expected shapes, noise, varying density, known k, soft memberships, hierarchy and streaming. Covers preprocessing (scaling, encoding, dimensionality, distance metrics), validation (internal indices, bootstrap stability, external labels, business profiling) and assigning new points.
collaborative-filtering-implementer
Implement collaborative filtering recommenders that beat popularity: item-based kNN with cosine, Jaccard or asymmetric similarity, shrinkage, neighbourhood size, positive-only or mean-centred rating signals and 'because you liked X' explanations; implicit ALS with confidence weighting, BPR pairwise ranking and the closed-form EASE model in plain NumPy, tuned on a temporal validation split by recall@k and NDCG@k and refit on all data; plus user-kNN, explicit matrix factorisation, the implicit/LightFM libraries, hyperparameter ranges, cold-start fallbacks, scaling with sparse matrices and approximate nearest neighbours. Produces top-k lists in the format of the recommendation evaluator.