"""Shortcut checks for a CSV with a two-class target: the four hunts from the game, run automatically, plus the accuracy baseline. web/checks.js is a line-for-line port and tests/parity.py checks the two agree. analyze(columns, rows, target, time_column=None) -> dict (JSON-friendly) Dataset Doctor (projects/dataset-doctor) profiles a file for missing values, formats and label problems. These checks ask a narrower question: if you trained a model on this file and it scored well, which of the game's shortcuts could be the reason? """ from __future__ import annotations import math import re import numpy as np from .models import accuracy, auc, knn1, logreg_fit, logreg_predict, standardize, standardize_fit from .rng import Mulberry32 MISSING = {"", "na", "n/a", "nan", "null", "none", "-", "?"} NUM_RE = re.compile(r"^[-+]?(\d+\.?\d*|\.\d+)([eE][-+]?\d+)?$") DATE_RE = re.compile(r"^(\d{4})[-/](\d{1,2})[-/](\d{1,2})([ T](\d{1,2}):(\d{2})(:(\d{2}))?)?") TIME_NAME = re.compile(r"(date|time|day|month|year|week|stamp|period)", re.I) YES = {"1", "true", "yes", "y", "t", "positive", "pos"} MAX_ROWS = 3000 def is_missing(v: str) -> bool: return v.strip().lower() in MISSING def parse_num(v: str): s = v.strip().replace(",", "") if NUM_RE.match(s): return float(s) return None def days_from_civil(y: int, m: int, d: int) -> int: """Days since 1970-01-01 (Howard Hinnant's algorithm; integer arithmetic only).""" y -= m <= 2 era = (y if y >= 0 else y - 399) // 400 yoe = y - era * 400 doy = (153 * (m + (-3 if m > 2 else 9)) + 2) // 5 + d - 1 doe = yoe * 365 + yoe // 4 - yoe // 100 + doy return era * 146097 + doe - 719468 def parse_date(v: str): m = DATE_RE.match(v.strip()) if not m: return None y, mo, d = int(m.group(1)), int(m.group(2)), int(m.group(3)) if mo < 1 or mo > 12 or d < 1 or d > 31: return None t = days_from_civil(y, mo, d) if m.group(5): t += (int(m.group(5)) * 60 + int(m.group(6)) + (int(m.group(8)) if m.group(8) else 0) / 60.0) / 1440.0 return t def fmt_pct(x: float) -> str: return "%.1f%%" % (100 * x) def fmt3(x: float) -> str: return "%.3f" % x def profile_column(name: str, values: list) -> dict: present = [v for v in values if not is_missing(v)] n = len(present) nums = [parse_num(v) for v in present] n_num = sum(1 for x in nums if x is not None) dates = [parse_date(v) for v in present] n_date = sum(1 for x in dates if x is not None) if n and n_date >= 0.9 * n: kind, parsed = "date", [None if is_missing(v) else parse_date(v) for v in values] elif n and n_num >= 0.9 * n: kind, parsed = "numeric", [None if is_missing(v) else parse_num(v) for v in values] else: kind, parsed = "category", [None if is_missing(v) else v.strip() for v in values] uniq = len(set(present)) return {"name": name, "type": kind, "missing": len(values) - n, "unique": uniq, "unique_ratio": uniq / n if n else 0.0, "values": parsed} def choose_positive(target_values: list): present = [v.strip() for v in target_values if not is_missing(v)] counts = {} for v in present: counts[v] = counts.get(v, 0) + 1 classes = sorted(counts, key=lambda c: (-counts[c], c)) if len(classes) < 2: raise ValueError("The target column has only one value.") if len(classes) == 2: yes = [c for c in classes if c.lower() in YES] pos = yes[0] if yes else classes[1] # the rarer class when neither looks like "yes" return pos, classes, "two classes" if len(classes) > 20: raise ValueError("The target column has %d different values. Pick a column with two classes." % len(classes)) return classes[0], classes, "most common class against the rest" def id_shaped(p: dict) -> bool: """Only whole numbers or strings can be IDs; measurements with decimals and dates are not.""" if p["type"] == "category": return True if p["type"] == "date": return False return all(x is None or x == int(x) for x in p["values"]) def fill_mean(col: list) -> np.ndarray: vals = [x for x in col if x is not None] mean = sum(vals) / len(vals) if vals else 0.0 return np.array([mean if x is None else x for x in col], dtype=float) def analyze(columns: list, rows: list, target: str, time_column=None, seed: int = 7) -> dict: if target not in columns: raise ValueError("No column called %r." % target) ti = columns.index(target) pos, classes, how = choose_positive([r[ti] for r in rows]) keep = [r for r in rows if not is_missing(r[ti])] rng = Mulberry32(seed) if len(keep) > MAX_ROWS: idx = rng.shuffle(list(range(len(keep))))[:MAX_ROWS] keep = [keep[i] for i in sorted(idx)] n = len(keep) if n < 40: raise ValueError("Only %d rows with a label. The checks need at least 40." % n) y = np.array([1 if r[ti].strip() == pos else 0 for r in keep]) profiles = [profile_column(c, [r[j] for r in keep]) for j, c in enumerate(columns) if j != ti] out = {"rows": len(rows), "used_rows": n, "target": target, "positive": pos, "classes": classes, "positive_how": how, "n_pos": int(y.sum()), "columns": [{k: p[k] for k in ("name", "type", "missing", "unique")} for p in profiles], "checks": []} # --- baseline majority = max(float(y.mean()), 1 - float(y.mean())) share = majority out["checks"].append({ "id": "baseline", "title": "Accuracy will flatter a do-nothing model" if share >= 0.9 else "Accuracy baseline", "severity": "high" if share >= 0.9 else "medium" if share >= 0.75 else "ok", "detail": "The bigger class is " + fmt_pct(share) + " of the rows, so a model that always answers \"" + (pos if y.mean() > 0.5 else "not " + pos) + "\" is already " + fmt_pct(share) + " accurate. Judge any model against that, and look at recall or balanced accuracy too.", "data": {"majority": share, "n_pos": int(y.sum()), "n": n}}) # --- ID hunt id_like = set() for p in profiles: if n - p["missing"] < 20 or p["unique_ratio"] < 0.95 or not id_shaped(p): continue if p["name"] == time_column or TIME_NAME.search(p["name"]): # a day counter is a clock, not an ID continue id_like.add(p["name"]) if p["type"] in ("numeric", "date"): score = fill_mean(p["values"]) else: present = sorted(set(v for v in p["values"] if v is not None)) rank = {v: i for i, v in enumerate(present)} score = np.array([rank[v] if v is not None else -1 for v in p["values"]], dtype=float) a = auc(score, y) dev = abs(a - 0.5) order = np.argsort(score, kind="stable") half = n // 2 r1 = float(y[order[:half]].mean()) r2 = float(y[order[half:]].mean()) sev = "high" if dev >= 0.15 else "medium" if dev >= 0.08 else "low" if sev == "low": detail = ("Every row has its own value, so a model can use it to memorise rows. Its order does not track the label here (AUC " + fmt3(a) + "), so dropping it is enough.") else: detail = ("Every row has its own value, and the order of those values tracks the label: the lower half of the IDs is " + fmt_pct(r1) + " \"" + pos + "\" and the upper half is " + fmt_pct(r2) + ". IDs handed out after sorting or batching by outcome do this. Drop the column; a model will read the answer off it.") out["checks"].append({"id": "id:" + p["name"], "title": "ID order gives away the label" if sev != "low" else "ID-like column: " + p["name"], "column": p["name"], "severity": sev, "detail": detail, "data": {"auc": a, "lower_half": r1, "upper_half": r2, "unique": p["unique"]}}) # --- leak hunt: single-feature AUC perm = rng.shuffle(list(range(n))) fold = np.zeros(n, dtype=int) for k, i in enumerate(perm): fold[i] = k % 2 strengths = [] for p in profiles: if p["name"] in id_like: continue if p["type"] in ("numeric", "date"): score = fill_mean(p["values"]) else: if p["unique"] > 50 or p["unique"] < 2: continue score = np.zeros(n) for f in (0, 1): rate, cnt = {}, {} for i in range(n): if fold[i] != f: v = p["values"][i] rate[v] = rate.get(v, 0) + int(y[i]) cnt[v] = cnt.get(v, 0) + 1 base = float(y[fold != f].mean()) for i in range(n): if fold[i] == f: v = p["values"][i] score[i] = rate[v] / cnt[v] if cnt.get(v) else base a = auc(score, y) strengths.append({"column": p["name"], "auc": max(a, 1 - a), "direction": "higher" if a >= 0.5 else "lower", "type": p["type"]}) strengths.sort(key=lambda s: (-s["auc"], s["column"])) flagged = [s for s in strengths if s["auc"] >= 0.9] crowd = len(flagged) >= 4 for s in flagged: sev = "medium" if crowd else "high" if s["auc"] >= 0.95 else "medium" tail = (" Many columns are this strong here, which points to an easy problem rather than one leaked column." if crowd else " Honest features this strong do exist, but they are rare.") out["checks"].append({ "id": "leak:" + s["column"], "title": "Too good to be true: " + s["column"], "column": s["column"], "severity": sev, "detail": ("On its own, " + s["column"] + " separates the classes with AUC " + fmt3(s["auc"]) + " (" + s["direction"] + " values go with \"" + pos + "\"). Ask when this value gets written down. If it is filled in after the outcome, or computed from it, " "a model will score brilliantly on a held-out set and fail the day it is used." + tail), "data": {"auc": s["auc"], "direction": s["direction"]}}) out["strengths"] = strengths[:12] # --- numeric matrix for the model-based hunts feats = [p for p in profiles if p["type"] in ("numeric", "date") and p["name"] not in id_like] time_name = None if time_column and time_column in columns and time_column != target: time_name = time_column else: dates = [p["name"] for p in profiles if p["type"] == "date"] named = [p["name"] for p in feats if TIME_NAME.search(p["name"])] time_name = (dates + named)[0] if dates or named else None time_vals = None if time_name: tp = next(p for p in profiles if p["name"] == time_name) if tp["type"] in ("numeric", "date"): time_vals = fill_mean(tp["values"]) else: time_name = None X = np.column_stack([fill_mean(p["values"]) for p in feats]) if feats else np.zeros((n, 0)) out["numeric_features"] = [p["name"] for p in feats] out["time_column"] = time_name # --- déjà vu hunt: a random split, then look for test rows that already exist in train if X.shape[1] >= 2: order = rng.shuffle(list(range(n))) ntr = int(0.7 * n) tr, te = np.array(sorted(order[:ntr])), np.array(sorted(order[ntr:])) pred, dist = knn1(X[tr], y[tr], X[te]) d = X.shape[1] exact = dist <= 1e-9 # with only two numeric columns, honest rows sit close together all the time, so only exact copies count near = ((dist <= 0.05 * math.sqrt(d)) & ~exact) if d >= 3 else np.zeros(len(dist), dtype=bool) matched = exact | near acc_match = accuracy(pred[matched], y[te][matched]) if matched.any() else None acc_fresh = accuracy(pred[~matched], y[te][~matched]) if (~matched).any() else None share = float(matched.mean()) gap = (acc_match - acc_fresh) if acc_match is not None and acc_fresh is not None else 0.0 sev = "high" if share >= 0.1 and gap >= 0.1 else "medium" if float(exact.mean()) >= 0.05 or (share >= 0.1 and gap >= 0.05) else "ok" if sev == "ok": detail = ("After a random 70/30 split, " + fmt_pct(share) + " of test rows have a near-copy in the training rows (" + fmt_pct(float(exact.mean())) + " exact). Nothing to worry about.") else: detail = ("After a random 70/30 split, " + fmt_pct(share) + " of test rows have a near-copy in the training rows (" + fmt_pct(float(exact.mean())) + " exact). A 1-nearest-neighbour model gets " + fmt_pct(acc_match) + " of those right against " + fmt_pct(acc_fresh) + " of the genuinely new rows. Repeated records, resampled rows or an upstream join can do this; " "a random split then grades the model on rows it has already seen. Split by record or group, or drop the copies first.") out["checks"].append({"id": "dejavu", "title": "Déjà vu: the test rows are already in training" if sev != "ok" else "Déjà vu check", "severity": sev, "detail": detail, "data": {"share": share, "exact": float(exact.mean()), "acc_matched": acc_match, "acc_fresh": acc_fresh, "n_test": int(len(te)), "dist": np.round(dist, 4).tolist()}}) # --- timestamp hunt if time_name is not None and X.shape[1] >= 1: order = np.argsort(time_vals, kind="stable") bins = 5 rates = [] for k in range(bins): idx = order[k * n // bins:(k + 1) * n // bins] rates.append(float(y[idx].mean())) ntr = int(0.7 * n) shuf = rng.shuffle(list(range(n))) rtr, rte = np.array(sorted(shuf[:ntr])), np.array(sorted(shuf[ntr:])) ttr, tte = order[:ntr], order[ntr:] acc_random = accuracy(logreg_predict(logreg_fit(X[rtr], y[rtr]), X[rte]), y[rte]) acc_time = accuracy(logreg_predict(logreg_fit(X[ttr], y[ttr]), X[tte]), y[tte]) drift = max(rates) - min(rates) drop = acc_random - acc_time pbar = float(y.mean()) se = math.sqrt(pbar * (1 - pbar) / (n // bins)) # a five-way split of a small file wobbles by chance real = drift >= 4 * se sev = "high" if (drift >= 0.3 and real) or drop >= 0.1 else "medium" if (drift >= 0.15 and real) or drop >= 0.05 else "ok" if sev == "ok": detail = ("Ordered by " + time_name + ", the share of \"" + pos + "\" moves between " + fmt_pct(min(rates)) + " and " + fmt_pct(max(rates)) + " across five periods, and a logistic regression scores about the same on a random split (" + fmt_pct(acc_random) + ") and on a train-on-the-past, test-on-the-future split (" + fmt_pct(acc_time) + ").") else: detail = ("Ordered by " + time_name + ", the share of \"" + pos + "\" moves between " + fmt_pct(min(rates)) + " and " + fmt_pct(max(rates)) + " across five periods. A logistic regression on the numeric columns scores " + fmt_pct(acc_random) + " on a random split but " + fmt_pct(acc_time) + " when it trains on the first 70% of the time and is tested on the last 30%. The random split lets it learn " "which period a row is from; the future is a period it has never seen. Report the time-split number.") out["checks"].append({"id": "time", "title": "The calendar is doing the work" if sev != "ok" else "Timestamp check", "column": time_name, "severity": sev, "detail": detail, "data": {"rates": rates, "acc_random": acc_random, "acc_time": acc_time}}) # --- the headline number people would see if X.shape[1] >= 1: order = rng.shuffle(list(range(n))) ntr = int(0.7 * n) tr, te = np.array(sorted(order[:ntr])), np.array(sorted(order[ntr:])) model = logreg_fit(X[tr], y[tr]) out["headline"] = {"acc": accuracy(logreg_predict(model, X[te]), y[te]), "majority": majority, "weights": [{"column": p["name"], "w": float(w)} for p, w in zip(feats, model["w"])]} sev_rank = {"high": 0, "medium": 1, "low": 2, "ok": 3} out["checks"].sort(key=lambda c: sev_rank[c["severity"]]) return out def read_csv(text: str): """A small CSV reader (quotes, commas or tabs) matching web/checks.js parseCSV.""" rows, row, field, i, q = [], [], "", 0, False sep = "\t" if text.split("\n", 1)[0].count("\t") > text.split("\n", 1)[0].count(",") else "," text = text.lstrip("") while i < len(text): c = text[i] if q: if c == '"': if i + 1 < len(text) and text[i + 1] == '"': field += '"' i += 1 else: q = False else: field += c elif c == '"': q = True elif c == sep: row.append(field) field = "" elif c == "\n" or c == "\r": if c == "\r" and i + 1 < len(text) and text[i + 1] == "\n": i += 1 row.append(field) rows.append(row) row, field = [], "" else: field += c i += 1 if field or row: row.append(field) rows.append(row) rows = [r for r in rows if any(x.strip() for x in r)] if not rows: raise ValueError("The file is empty.") header = [h.strip() for h in rows[0]] body = [r + [""] * (len(header) - len(r)) if len(r) < len(header) else r[:len(header)] for r in rows[1:]] return header, body