"""Checks that look at the data itself (no target, no second file). Each check adds issue dicts to an IssueList: type, column, severity, title (also "message"), detail, fix (pandas code), optional evidence and an optional machine-readable action that fixes.apply_fixes knows how to carry out. """ import re from .fmt import fmt_num, fmt_pct, plural, row_nums from .jscompat import js_json, js_keys, js_str, py_str, to_fixed, trim, u16key from .stats import levenshtein, pearson, quantile_sorted from .values import PII, SENTINELS, fingerprint, luhn, parse_num _PHONE_NAME = re.compile(r"phone|tel|mobile|cell", re.I | re.A) _DIGIT = re.compile(r"[0-9]") def py_list(a): return "[" + ", ".join(py_str(x) for x in a) + "]" def check_ragged(table, n, issues): if table.ragged_rows: issues.add(type="ragged_rows", column=None, severity="medium" if table.ragged_rows > 0.01 * n else "low", affected=table.ragged_rows / max(1, n), title=plural(table.ragged_rows, "row") + " with the wrong number of fields", detail="Some lines have more or fewer separators than the header. They were padded or cut to fit, which can shift values into the wrong column. Usually an unquoted comma inside a value.", fix="# Check the raw file around the reported lines; quote fields that contain the separator.") # ----------------------------------------------------------------------------- per column def check_column(p, n, target, issues, today=None): """Run every single-column check on profile p.""" name, np_ = p.name, len(p.present_idx) if p.type == "empty": issues.add(type="constant_column", column=name, severity="medium", affected=1, title=name + " is empty", detail="Every value is missing.", fix="df = df.drop(columns=[" + py_str(name) + "])", action={"drop": [name]}) return is_target = target is not None and p is target tokens, sentinel_hits = _check_disguised_missing(p, n, np_, issues) _check_missing(p, n, is_target, tokens, sentinel_hits, issues) _check_types(p, n, np_, issues) if p.type == "date": _check_dates(p, n, np_, issues, today) _check_constant(p, np_, n, issues) if p.is_id and not is_target: issues.add(type="id_column", column=name, severity="low", affected=0, title=name + " looks like an identifier", detail="Unique for (almost) every row. Fine for joining tables, but a model can memorise it. Drop it before training, and use it to check that the same entity isn't in both train and test.", fix="X = df.drop(columns=[" + py_str(name) + "])", action={"drop": [name]}) _check_pii(p, n, np_, issues) if p.type in ("categorical", "text") and p.unique <= 500 and not p.is_id: _check_category_variants(p, n, issues) if p.type == "numeric" and np_ >= 8: _check_outliers(p, n, issues) def _check_disguised_missing(p, n, np_, issues): """Placeholders such as "unknown" or -999 that stand for a missing value.""" name = p.name tokens = [t for t in js_keys(p.missing_tokens) if t != ""] hits = [] if p.type == "numeric": counts, order = {}, [] for i in p.present_idx: x = p.num[i] if x is not None and x in SENTINELS: k = js_str(x) if k not in counts: counts[k] = 0 counts[k] += 1 for k in js_keys(counts): x, c = float(k), counts[k] if c < 2 or c < 0.002 * np_: continue others = [y for y in p.sorted if y != x] if len(others) < 10: continue lo, hi = quantile_sorted(others, 0.01), quantile_sorted(others, 0.99) span = max(1e-9, hi - lo) apart = (x < others[0] and lo >= 0) if x < 0 else (x > hi + 2 * span or (x >= 99 and x > others[-1])) if apart: hits.append({"value": int(x), "count": c}) if tokens or hits: parts, cnt = [], 0 for t in tokens: parts.append('"' + t + '" ×' + js_str(p.missing_tokens[t])) cnt += p.missing_tokens[t] for s in hits: parts.append(fmt_num(s["value"]) + " ×" + js_str(s["count"])) cnt += s["count"] repl = [py_str(t) for t in tokens] + [js_str(s["value"]) for s in hits] issues.add(type="disguised_missing", column=name, severity="medium" if cnt > 0.05 * n else "low", affected=cnt / n, title=name + " hides missing values as " + ", ".join(parts[:3]), detail="These look like real values but mean \"unknown\". Left in, " + ( "a number like " + fmt_num(hits[0]["value"]) + " drags averages and models toward a value nobody has" if hits else "they count as a category of their own") + ". Treat them as missing.", evidence={"values": parts}, fix="df[" + py_str(name) + "] = df[" + py_str(name) + "].replace([" + ", ".join(repl) + "], np.nan)", action={"missing": {"column": name, "tokens": tokens, "numbers": [s["value"] for s in hits]}}) p.sentinels = [s["value"] for s in hits] return tokens, hits def _check_missing(p, n, is_target, tokens, hits, issues): name = p.name sent_count = 0 for s in hits: sent_count += s["count"] miss = (p.missing + sent_count) / n p.missing_frac = miss only_tokens = bool(tokens) and p.missing == sum(p.missing_tokens[t] for t in tokens) if miss >= 0.02 and not is_target and not (only_tokens and miss < 0.05): sev = "high" if miss >= 0.5 else "medium" if miss >= 0.2 else "low" q = py_str(name) if miss >= 0.9: fix = "df = df.drop(columns=[" + q + "])" elif p.type == "numeric": fix = "df[" + py_str(name + "_missing") + "] = df[" + q + "].isna()\ndf[" + q + "] = df[" + q + "].fillna(df[" + q + "].median())" else: fix = "df[" + q + "] = df[" + q + "].fillna(\"(missing)\")" issues.add(type="missing_values", column=name, severity=sev, affected=miss, title=fmt_pct(miss) + " of " + name + " is missing", detail="Almost nothing is left in this column. Dropping it is usually the right call." if miss >= 0.9 else "Decide on purpose: drop the column, fill it (median or most common value), or add a \"was missing\" flag so a model can use the fact that it was blank.", fix=fix, action={"drop": [name]} if miss >= 0.9 else None) def _check_types(p, n, np_, issues): """Numbers stored as text ("$1,049") and numeric columns with stray words.""" name = p.name if p.type == "numeric" and p.n_num_text > 0.05 * np_: ex = [p.raw[i] for i in p.present_idx if parse_num(p.raw[i]) is None][:4] issues.add(type="numeric_text", column=name, severity="medium", affected=p.n_num_text / n, title=name + " stores numbers as text", detail="Values like " + ", ".join('"' + e + '"' for e in ex) + " have currency signs, % or thousands separators, so pandas reads the column as strings and you can't do math on it.", evidence={"values": ex}, fix="df[" + py_str(name) + "] = pd.to_numeric(df[" + py_str(name) + "].astype(str).str.strip().str.replace(r\"^\\((.*)\\)$\", r\"-\\1\", regex=True).str.replace(r\"[$€£¥%,\\s]\", \"\", regex=True), errors=\"coerce\")", action={"numeric": name}) junk_keys = js_keys(p.junk) if p.type in ("numeric", "binary") and junk_keys: jc = 0 for k in junk_keys: jc += p.junk[k] issues.add(type="mixed_types", column=name, severity="medium" if jc > 0.02 * n else "low", affected=jc / n, title=name + " is numeric except for " + plural(jc, "value"), detail="Non-numbers such as " + ", ".join('"' + k + '"' for k in junk_keys[:4]) + " will turn the whole column into text or crash a conversion.", evidence={"values": junk_keys[:8]}, fix="df[" + py_str(name) + "] = pd.to_numeric(df[" + py_str(name) + "], errors=\"coerce\") # non-numbers become NaN", action={"numeric": name}) def _check_dates(p, n, np_, issues, today): name = p.name fmts = js_keys(p.date_formats) meaningful = [f for f in fmts if p.date_formats[f] >= max(2, 0.01 * np_)] if len(meaningful) > 1: fdesc = [(("DD/MM/YYYY" if p.date_day_first else "MM/DD/YYYY") if f == "slash" else f) + " ×" + js_str(p.date_formats[f]) for f in meaningful] issues.add(type="mixed_date_formats", column=name, severity="medium", affected=min(p.date_formats[f] for f in meaningful) / n, title=name + " mixes date formats", detail="Found " + ", ".join(fdesc) + ". Parsers guess one format per column, so some dates get swapped or dropped.", evidence={"values": fdesc}, fix="df[" + py_str(name) + "] = pd.to_datetime(df[" + py_str(name) + "], format=\"mixed\", dayfirst=" + ("True" if p.date_day_first else "False") + ", errors=\"coerce\")", action={"date": name}) slash = p.date_formats.get("slash", 0) if p.slash_ambiguous and slash >= 0.05 * np_: issues.add(type="ambiguous_dates", column=name, severity="low", affected=slash / n, title=name + ": day and month can't be told apart", detail="Every slash date has both parts ≤ 12, so 03/04 could be March 4 or April 3. Assumed month first (US style); check the source.", fix="# confirm whether the source writes MM/DD or DD/MM before parsing") if p.invalid_dates: issues.add(type="invalid_dates", column=name, severity="medium", affected=len(p.invalid_dates) / n, title=plural(len(p.invalid_dates), "impossible date") + " in " + name, detail="Dates such as " + ", ".join('"' + p.raw[i] + '"' for i in p.invalid_dates[:3]) + " don't exist (e.g. month 13 or February 30).", evidence={"rows": row_nums(p.invalid_dates)}, fix="df[" + py_str(name) + "] = pd.to_datetime(df[" + py_str(name) + "], errors=\"coerce\") # impossible dates become NaT") if today: fut = [i for i in p.present_idx if p.iso[i] and u16key(p.iso[i]) > u16key(today)] if fut: latest = sorted((p.iso[i] for i in fut), key=u16key)[-1] issues.add(type="future_dates", column=name, severity="low", affected=len(fut) / n, title=plural(len(fut), "date") + " in " + name + " " + ("is" if len(fut) == 1 else "are") + " in the future", detail="Latest: " + latest + ". Often a typo in the year or a placeholder like 2099-12-31.", evidence={"rows": row_nums(fut)}, fix="# inspect: df[df[" + py_str(name) + "] > pd.Timestamp.today()]") def _check_constant(p, np_, n, issues): name = p.name if p.unique == 1 and p.missing < n: issues.add(type="constant_column", column=name, severity="medium", affected=1, title=name + " has only one value", detail='Every row is "' + p.top[0][0] + '"' + (" (or missing)" if p.missing else "") + ". It carries no information.", fix="df = df.drop(columns=[" + py_str(name) + "])", action={"drop": [name]}) elif p.unique > 1 and p.top[0][1] >= 0.99 * np_ and np_ >= 50: issues.add(type="near_constant", column=name, severity="low", affected=1 - p.top[0][1] / np_, title=name + " is almost always \"" + p.top[0][0] + "\"", detail=fmt_pct(p.top[0][1] / np_) + " of rows share one value. Rarely useful, and the few exceptions can make a model overfit.", fix="# consider: df = df.drop(columns=[" + py_str(name) + "])") def _check_pii(p, n, np_, issues): if p.type not in ("text", "categorical", "numeric"): return name = p.name for rule in PII: hits = checked = 0 for i in p.present_idx[:2000]: v = trim(p.raw[i]) checked += 1 if rule["re"].match(v) and (not rule.get("luhn") or luhn(v)): hits += 1 if checked and hits >= 0.3 * checked and not (rule["kind"] == "phone numbers" and p.type == "numeric" and not _PHONE_NAME.search(name)): issues.add(type="pii", column=name, severity="medium", affected=hits / checked * np_ / n, title=name + " contains " + rule["kind"], detail="Personal data. Remove it or replace it with a salted hash before sharing the file or training on it. It almost never helps a model.", fix="df = df.drop(columns=[" + py_str(name) + "]) # or hash: df[" + py_str(name) + "].map(lambda s: hashlib.sha256((SALT + str(s)).encode()).hexdigest())", action={"drop": [name]}) break def _by_count(counts): return lambda v: (-counts[v], u16key(v)) def _check_category_variants(p, n, issues): """Spellings of the same category ("US", "usa", "United States") and rare one-letter typos.""" name, counts = p.name, p.counts groups = {} for v in js_keys(counts): f = fingerprint(v) if not f: continue groups.setdefault(f, []).append(v) mapping, merged, affected_rows = {}, [], 0 for f in js_keys(groups): vs = groups[f] if len(vs) < 2: continue vs.sort(key=_by_count(counts)) # in place, like the JS canon = vs[0] for v in vs[1:]: mapping[v] = canon affected_rows += counts[v] merged.append(", ".join('"' + v + '" ×' + js_str(counts[v]) for v in vs)) fps = [] for f in js_keys(groups): tot = 0 for v in groups[f]: tot += counts[v] fps.append({"f": f, "n": tot, "canon": sorted(groups[f], key=_by_count(counts))[0], "all": groups[f]}) fps.sort(key=lambda o: (-o["n"], u16key(o["f"]))) if len(fps) <= 300: for a in range(len(fps)): for b in range(a + 1, len(fps)): A, B = fps[a], fps[b] if B["n"] > 0.1 * A["n"] or len(A["f"]) < 4 or len(B["f"]) < 4: continue if _DIGIT.search(A["f"]) or _DIGIT.search(B["f"]): continue maxd = 2 if min(len(A["f"]), len(B["f"])) >= 8 else 1 if levenshtein(A["f"], B["f"], maxd) <= maxd: for v in B["all"]: if not mapping.get(v): mapping[v] = A["canon"] affected_rows += counts[v] merged.append('"' + B["canon"] + '" ×' + js_str(B["n"]) + " → \"" + A["canon"] + '" (likely typo)') B["n"] = 1e18 # don't merge twice if merged: pairs = [py_str(trim(k)) + ": " + py_str(mapping[k]) for k in js_keys(mapping) if trim(k) != mapping[k]][:40] uniq = [s for i, s in enumerate(pairs) if pairs.index(s) == i] issues.add(type="category_variants", column=name, severity="medium" if affected_rows > 0.02 * n else "low", affected=affected_rows / n, title=name + " spells the same thing " + ("in different ways" if len(merged) == 1 else "in different ways (" + js_str(len(merged)) + " groups)"), detail="A model, a group-by or a chart treats every spelling as a different category. Merge them into one.", evidence={"values": merged[:8]}, fix="df[" + py_str(name) + "] = df[" + py_str(name) + "].str.strip().replace({" + ", ".join(uniq) + "})", action={"map": {"column": name, "mapping": mapping}}) p.mapping = mapping def _check_outliers(p, n, issues): """Robust z-score > 5 (on a log scale for long-tailed positive data), ignoring sentinels.""" import math name = p.name xs, idxs = [], [] for i in p.present_idx: x = p.num[i] if x is not None and x not in p.sentinels: xs.append(x) idxs.append(i) if len(xs) < 8 or p.is_id: return s2 = sorted(xs) q10, q50, q90 = quantile_sorted(s2, 0.1), quantile_sorted(s2, 0.5), quantile_sorted(s2, 0.9) use_log = s2[0] >= 0 and (q90 - q50) > 1.5 * (q50 - q10) and q90 > q50 pos_vals = [x for x in s2 if x > 0] unit = quantile_sorted(pos_vals, 0.5) if pos_vals else 1 if not use_log: tx = xs elif s2[0] > 0: tx = [math.log10(x) for x in xs] else: tx = [math.log10(1 + x / unit) for x in xs] ts = sorted(tx) med = quantile_sorted(ts, 0.5) dev = sorted(abs(x - med) for x in tx) mad = quantile_sorted(dev, 0.5) scale = mad * 1.4826 if scale == 0: iqr = quantile_sorted(ts, 0.75) - quantile_sorted(ts, 0.25) scale = iqr / 1.349 if not scale > 0: return out = [] for q in range(len(tx)): z = (tx[q] - med) / scale if abs(z) > 5: out.append({"i": idxs[q], "x": xs[q], "z": z}) if not out: return out.sort(key=lambda o: -abs(o["z"])) frac = len(out) / n med_raw = quantile_sorted(s2, 0.5) rows = row_nums([o["i"] for o in out], 6) issues.add(type="outliers", column=name, severity="medium" if abs(out[0]["z"]) > 10 and frac <= 0.02 else "low", affected=frac, title=plural(len(out), "extreme value") + " in " + name, detail="Far outside the bulk of the data (typical value " + fmt_num(med_raw) + "; most extreme " + fmt_num(out[0]["x"]) + ", row " + js_str(out[0]["i"] + 2) + "). Could be real, a typo (an extra zero) or a unit mix-up. Check them before deciding." + (" Measured on a log scale because the column is long-tailed." if use_log else ""), evidence={"rows": rows, "values": [fmt_num(o["x"]) for o in out[:6]]}, fix="lo, hi = df[" + py_str(name) + "].quantile([0.01, 0.99])\ndf[" + py_str(name) + "] = df[" + py_str(name) + "].clip(lo, hi) # or inspect: df.loc[" + js_json([r - 2 for r in rows]) + "]") # ----------------------------------------------------------------------------- whole table def check_missing_together(prof, n, target, issues): miss_cols = [p for p in prof if p.missing >= 0.02 * n and p.missing < n and p is not target] together = [] for a in range(len(miss_cols)): sa = set(miss_cols[a].missing_idx) for b in range(a + 1, len(miss_cols)): inter = sum(1 for i in miss_cols[b].missing_idx if i in sa) uni = miss_cols[a].missing + miss_cols[b].missing - inter if uni and inter / uni >= 0.9: together.append([miss_cols[a].name, miss_cols[b].name, inter]) if together: cols = [] for t in together: cols.extend(t[:2]) issues.add(type="missing_together", column=None, columns=cols, severity="info", affected=0, title="Some columns are missing together", detail="; ".join(t[0] + " & " + t[1] + " (" + js_str(t[2]) + " rows)" for t in together[:4]) + ". They probably come from the same source; one missingness flag covers both.", fix="# e.g. df[" + py_str(together[0][0] + "_missing") + "] = df[" + py_str(together[0][0]) + "].isna()") def check_duplicates(table, prof, n, issues): """Exact duplicate rows, and rows that differ only in an ID column.""" seen, dup_idx = set(), [] for i, r in enumerate(table.rows): k = "\x01".join(r) if k in seen: dup_idx.append(i) else: seen.add(k) if dup_idx: df = len(dup_idx) / n issues.add(type="duplicate_rows", column=None, severity="high" if df > 0.05 else "medium", affected=df, title=plural(len(dup_idx), "exact duplicate row"), detail="Identical rows, often from a double export or a bad join. They overweight some examples, and if a copy lands in your test split the model gets graded on rows it has already seen.", evidence={"rows": row_nums(dup_idx)}, fix="df = df.drop_duplicates()", action={"dedupe": "all"}) id_cols = [p.index for p in prof if p.is_id] if id_cols and len(id_cols) < len(table.columns): # needs at least one non-ID column cols = table.columns exact = set(dup_idx) seen2, dup2 = set(), [] idset = set(id_cols) for i, r in enumerate(table.rows): k = "\x01".join(v for j, v in enumerate(r) if j not in idset) if k in seen2 and i not in exact: dup2.append(i) seen2.add(k) if dup2: names = [cols[j] for j in id_cols] issues.add(type="near_duplicate_rows", column=None, severity="high" if len(dup2) > 0.05 * n else "medium", affected=len(dup2) / n, title=plural(len(dup2), "row") + " identical except for the ID", detail="Same values in every column but " + ", ".join(names) + ". Usually the same record entered twice under a new ID.", evidence={"rows": row_nums(dup2)}, fix="df = df.drop_duplicates(subset=[c for c in df.columns if c not in " + py_list(names) + "])", action={"dedupe": "noid", "idCols": names}) def check_redundant(prof, n, issues): """Numeric columns correlated at |r| >= 0.999, and categories that map one-to-one.""" num_cols = [p for p in prof if p.type in ("numeric", "binary") and not p.is_id and p.unique > 1][:80] redundant = [] for u in range(len(num_cols)): for v in range(u + 1, len(num_cols)): X, Y = [], [] nu, nv = num_cols[u].num, num_cols[v].num for i in range(n): x1, y1 = nu[i], nv[i] if x1 is not None and y1 is not None: X.append(x1) Y.append(y1) if len(X) < 10: continue r = pearson(X, Y) if abs(r) >= 0.999: redundant.append([num_cols[u].name, num_cols[v].name, r]) cat_cols = [p for p in prof if p.type == "categorical" and p.unique >= 3 and not p.is_id][:60] for c1 in range(len(cat_cols)): for c2 in range(c1 + 1, len(cat_cols)): if cat_cols[c1].unique != cat_cols[c2].unique: continue m1, ok = {}, True ra, rb = cat_cols[c1].raw, cat_cols[c2].raw for i in range(n): va, vb = ra[i], rb[i] if va in m1 and m1[va] != vb: ok = False break m1[va] = vb if ok: redundant.append([cat_cols[c1].name, cat_cols[c2].name, 1]) for rd in redundant: issues.add(type="redundant_columns", column=rd[1], columns=[rd[0], rd[1]], severity="low", affected=0, title=rd[0] + " and " + rd[1] + " carry the same information", detail=("One maps one-to-one onto the other." if rd[2] == 1 else "Correlation " + to_fixed(rd[2], 4) + ".") + " Keep one; duplicates make linear models unstable and feature importances misleading.", fix="df = df.drop(columns=[" + py_str(rd[1]) + "])", action={"drop": [rd[1]]})