1# Author: Simon-Pierre Boucher — simon-pierre.boucher@uqo.ca2"""Leakage-audit suite. Every check returns a ``CheckResult``; ``audit`` runs them all and produces a JSON-able3report (``scripts/05_leakage_audit.py`` → ``results/data/leakage_audit.json``). The same functions back4``tests/test_leakage.py`` (synthetic data) so that the paper's leakage claims trace to executed code.56Checks7------81. ``check_split_windows`` split windows ordered, non-overlapping, contiguous (config).92. ``check_split_assignment`` every row's split matches its date; partitions disjoint.103. ``check_group_ordering`` no parcel has a sale in ``train`` dated on/after a sale of the same parcel in a later11 partition (validation/test/holdout), i.e. no "future" sale of a test parcel is trained on.124. ``check_prior_sale`` prior_sale_date < date, and equals the latest strictly-earlier sale of the parcel.135. ``check_comparables`` for a sample of subjects: all comparables dated < cutoff, within the look-back,14 subject parcel absent, same class when required, ids C01.. in order.156. ``check_market_context`` market context never uses sales after the cutoff (perturbation test).167. ``check_feature_matrix_asof`` perturbing (or deleting) every sale dated ≥ D leaves the feature rows of sales17 dated < D unchanged — the matrix contains no same-day/post-sale information.188. ``check_feature_matrix_columns`` no target-derived column is used as a feature.199. ``check_subsample`` stratum counts follow the proportional allocation; the draw is deterministic.20"""21from __future__ import annotations2223import json24from dataclasses import asdict, dataclass, field25from datetime import date, timedelta26from typing import Any2728import numpy as np29import polars as pl3031from ..config import load_config32from .analytic_schema import add_prior_sale33from .splits import SPLIT_ORDER, assign_splits, check_windows, proportional_allocation, split_windows, stratified_subsample343536@dataclass37class CheckResult:38 name: str39 passed: bool40 n_checked: int = 041 n_failed: int = 042 details: dict[str, Any] = field(default_factory=dict)4344 def to_dict(self) -> dict[str, Any]:45 return asdict(self)464748# ---------------------------------------------------------------------------------------------------------------49def check_split_windows(cfg: dict[str, Any]) -> CheckResult:50 try:51 check_windows(cfg)52 w = split_windows(cfg)53 gaps = []54 for a, b in zip(SPLIT_ORDER, SPLIT_ORDER[1:]):55 if (w[b][0] - w[a][1]).days != 1:56 gaps.append(f"{a}->{b}: {(w[b][0] - w[a][1]).days - 1} day(s) gap")57 return CheckResult("split_windows", True, len(SPLIT_ORDER), 0, {"windows": {k: [v[0].isoformat(), v[1].isoformat()] for k, v in w.items()}, "gaps": gaps})58 except ValueError as e:59 return CheckResult("split_windows", False, len(SPLIT_ORDER), 1, {"error": str(e)})606162def check_split_assignment(analytic: pl.DataFrame, splits: pl.DataFrame, cfg: dict[str, Any]) -> CheckResult:63 w = split_windows(cfg)64 df = analytic.select("tx_id", "date").join(splits, on="tx_id", how="inner")65 bad = 066 for name, (a, b) in w.items():67 part = df.filter(pl.col("split") == name)68 bad += int(((part["date"] < a) | (part["date"] > b)).sum())69 dup = splits.height - splits["tx_id"].n_unique()70 missing = analytic.height - df.height71 ok = bad == 0 and dup == 0 and missing == 072 return CheckResult("split_assignment", ok, df.height, bad + dup + missing, {"out_of_window": bad, "duplicate_tx": dup, "rows_without_split": missing})737475def check_group_ordering(analytic: pl.DataFrame, splits: pl.DataFrame) -> CheckResult:76 """For each parcel: max train date < min date in every later partition."""77 rank = {s: i for i, s in enumerate(SPLIT_ORDER)}78 df = analytic.select("tx_id", "parcel_id", "date").join(splits.select("tx_id", "split"), on="tx_id", how="inner")79 df = df.with_columns(pl.col("split").replace_strict(rank, default=99, return_dtype=pl.Int8).alias("_r"))80 agg = df.group_by("parcel_id").agg(81 pl.col("date").filter(pl.col("_r") == 0).max().alias("max_train"),82 pl.col("date").filter(pl.col("_r") > 0).min().alias("min_later"),83 ).drop_nulls()84 viol = agg.filter(pl.col("max_train") >= pl.col("min_later"))85 return CheckResult("group_ordering", viol.height == 0, agg.height, viol.height, {"parcels_in_train_and_later": agg.height})868788def check_prior_sale(analytic: pl.DataFrame) -> CheckResult:89 ps = analytic.filter(pl.col("prior_sale_date").is_not_null())90 not_earlier = int((ps["prior_sale_date"] >= ps["date"]).sum())91 recomputed = add_prior_sale(analytic.drop(["prior_sale_price", "prior_sale_date"]))92 cmp = analytic.select("tx_id", "prior_sale_price", "prior_sale_date").join(93 recomputed.select("tx_id", pl.col("prior_sale_price").alias("_p"), pl.col("prior_sale_date").alias("_d")), on="tx_id"94 )95 mismatch = int((cmp["prior_sale_date"].ne_missing(cmp["_d"]) | cmp["prior_sale_price"].ne_missing(cmp["_p"])).sum())96 # brute force on a sample of repeat parcels97 multi = analytic.filter(pl.col("parcel_id").is_duplicated()).sort(["parcel_id", "date"])98 bf_bad = 099 bf_n = 0100 for _pid, g in list(multi.group_by("parcel_id", maintain_order=True))[:2000]:101 rows = g.to_dicts()102 for r in rows:103 earlier = [q for q in rows if q["date"] < r["date"]]104 bf_n += 1105 if not earlier:106 bf_bad += r["prior_sale_date"] is not None107 else:108 dmax = max(q["date"] for q in earlier)109 if r["prior_sale_date"] != dmax or r["prior_sale_price"] not in {q["price"] for q in earlier if q["date"] == dmax}:110 bf_bad += 1111 ok = not_earlier == 0 and mismatch == 0 and bf_bad == 0112 return CheckResult("prior_sale", ok, analytic.height, not_earlier + mismatch + bf_bad, {"not_strictly_earlier": not_earlier, "recompute_mismatch": mismatch, "brute_force_checked": bf_n, "brute_force_bad": bf_bad})113114115def check_comparables(analytic: pl.DataFrame, subjects: pl.DataFrame, cfg: dict[str, Any]) -> CheckResult:116 from ..features.comparables import retrieve_comparables_detailed117118 c = cfg["comparables"]119 lookback = int(c["lookback_days"])120 fails: dict[str, int] = {"date_ge_cutoff": 0, "beyond_lookback": 0, "own_parcel": 0, "class_mismatch": 0, "id_order": 0, "too_many": 0, "fewer_than_min_unexpanded": 0}121 parcel_of = dict(zip(analytic["tx_id"].to_list(), analytic["parcel_id"].to_list()))122 date_of = dict(zip(analytic["tx_id"].to_list(), analytic["date"].to_list()))123 n_comps = []124 for r in subjects.iter_rows(named=True):125 cutoff = r["date"] - timedelta(days=1)126 cs = retrieve_comparables_detailed(analytic, r, cutoff, cfg)127 comps = cs.comparables128 n_comps.append(len(comps))129 fails["date_ge_cutoff"] += sum(cp.sale_date >= cutoff for cp in comps)130 fails["beyond_lookback"] += sum((cutoff - cp.sale_date).days > lookback for cp in comps)131 fails["class_mismatch"] += sum(cp.property_class != r["property_class"] for cp in comps) if c.get("same_class_required", True) else 0132 fails["id_order"] += int([cp.evidence_id for cp in comps] != [f"C{i:02d}" for i in range(1, len(comps) + 1)])133 fails["too_many"] += int(len(comps) > int(c["max_comps"]))134 fails["fewer_than_min_unexpanded"] += int(len(comps) < int(c["min_comps"]) and not cs.expanded)135 # own parcel: compare against every sale of the subject's parcel (by date+price identity is not unique → use parcel sales set)136 own = analytic.filter(pl.col("parcel_id") == r["parcel_id"]).select("date", "price")137 own_set = set(zip(own["date"].to_list(), own["price"].to_list()))138 # a comparable is from the subject's parcel only if it matches a same-parcel sale AND lies at distance 0139 fails["own_parcel"] += sum(((cp.sale_date, cp.price) in own_set and cp.distance_km == 0.0) for cp in comps)140 n_fail = sum(fails.values())141 return CheckResult("comparables_asof", n_fail == 0, subjects.height, n_fail, {**fails, "mean_n_comps": float(np.mean(n_comps)) if n_comps else None, "min_n_comps": int(min(n_comps)) if n_comps else None})142143144def check_market_context(analytic: pl.DataFrame, subjects: pl.DataFrame, cfg: dict[str, Any]) -> CheckResult:145 """Perturb every sale after the cutoff (price × 10) and recompute: the context must not change."""146 from ..features.market import market_context147148 n_fail = 0149 for r in subjects.iter_rows(named=True):150 cutoff = r["date"] - timedelta(days=1)151 base = market_context(analytic, r, cutoff, cfg).model_dump()152 pert = analytic.with_columns(pl.when(pl.col("date") > cutoff).then(pl.col("price") * 10).otherwise(pl.col("price")).alias("price"))153 pert = pert.with_columns(pl.col("price").log().alias("log_price"))154 alt = market_context(pert, r, cutoff, cfg).model_dump()155 n_fail += int(base != alt)156 return CheckResult("market_context_asof", n_fail == 0, subjects.height, n_fail)157158159def check_feature_matrix_asof(analytic: pl.DataFrame, splits: pl.DataFrame, geo: pl.DataFrame | None, cutoff: date) -> CheckResult:160 """Features of sales dated < cutoff must be invariant to perturbing AND to deleting all sales dated ≥ cutoff."""161 from ..features.matrix import build_feature_matrix162163 base = build_feature_matrix(analytic, splits, geo=geo).filter(pl.col("date") < cutoff).sort("tx_id")164 pert = analytic.with_columns(pl.when(pl.col("date") >= cutoff).then(pl.col("price") * 7).otherwise(pl.col("price")).alias("price"))165 pert = pert.with_columns(pl.col("price").log().alias("log_price"))166 alt1 = build_feature_matrix(pert, splits, geo=geo).filter(pl.col("date") < cutoff).sort("tx_id")167 dele = analytic.filter(pl.col("date") < cutoff)168 alt2 = build_feature_matrix(dele, splits, geo=geo).filter(pl.col("date") < cutoff).sort("tx_id")169 n_fail = 0170 diffs: list[str] = []171 for alt in (alt1, alt2):172 if alt.height != base.height or alt["tx_id"].to_list() != base["tx_id"].to_list():173 n_fail += 1174 diffs.append("row set differs")175 continue176 for c in base.columns:177 if c in ("price", "log_price"):178 continue179 a, b = base[c], alt[c]180 if a.dtype == pl.Float64:181 a, b = a.fill_nan(None), b.fill_nan(None)182 same = ((a - b).abs() < 1e-9) | (a.is_null() & b.is_null())183 else:184 same = a.eq_missing(b)185 if not same.all():186 n_fail += 1187 diffs.append(c)188 return CheckResult("feature_matrix_asof", n_fail == 0, base.height, n_fail, {"cutoff": cutoff.isoformat(), "columns_changed": sorted(set(diffs))})189190191FORBIDDEN_FEATURES = {"price", "log_price", "price_tercile", "exclude_reason", "assessment_ratio", "ppm2"}192193194def check_feature_matrix_columns(matrix: pl.DataFrame) -> CheckResult:195 from ..features.matrix import feature_columns196197 num, cat = feature_columns(matrix)198 bad = [c for c in num + cat if c in FORBIDDEN_FEATURES or c.startswith("price_") or c.endswith("_price") and not c.startswith(("prior_sale", "nb_med", "mun_med"))]199 return CheckResult("feature_matrix_columns", not bad, len(num) + len(cat), len(bad), {"forbidden_present": bad, "numeric": num, "categorical": cat})200201202def check_subsample(analytic: pl.DataFrame, splits: pl.DataFrame, subs: pl.DataFrame, cfg: dict[str, Any], n_by_split: dict[str, int] | None = None) -> CheckResult:203 strata_cols = list(cfg["llm_eval_subsample"]["strata"])204 s = cfg["llm_eval_subsample"]205 n_by_split = n_by_split or {"validation": s["validation_n"], "test": s["test_n"], "holdout": s["holdout_n"]}206 df = analytic.select(["tx_id", *strata_cols]).join(splits.select("tx_id", "split"), on="tx_id")207 df = df.with_columns(pl.concat_str([pl.col(c).cast(pl.String) for c in strata_cols], separator="|").alias("stratum"))208 n_fail = 0209 details: dict[str, Any] = {}210 for split, n in n_by_split.items():211 counts = {k: int(v) for k, v in df.filter(pl.col("split") == split).group_by("stratum").len().iter_rows()}212 expected = proportional_allocation(counts, int(n))213 got = {k: int(v) for k, v in subs.filter(pl.col("split") == split).group_by("stratum").len().iter_rows()}214 mism = {k: (expected.get(k, 0), got.get(k, 0)) for k in set(expected) | set(got) if expected.get(k, 0) != got.get(k, 0)}215 n_fail += len(mism)216 details[split] = {"n": int(sum(got.values())), "target": int(n), "strata": len(got), "mismatched_strata": mism}217 # membership + determinism218 joined = subs.join(df.select("tx_id", pl.col("split").alias("_s"), pl.col("stratum").alias("_st")), on="tx_id", how="left")219 n_fail += int((joined["split"] != joined["_s"]).sum() + (joined["stratum"] != joined["_st"]).sum() + joined["_s"].null_count())220 redraw = stratified_subsample(analytic, splits, cfg, n_by_split=n_by_split)221 deterministic = redraw.sort(["split", "stratum", "tx_id"]).equals(subs.sort(["split", "stratum", "tx_id"]))222 n_fail += int(not deterministic)223 details["deterministic_redraw"] = deterministic224 return CheckResult("llm_subsample", n_fail == 0, subs.height, n_fail, details)225226227# ---------------------------------------------------------------------------------------------------------------228def audit(229 analytic: pl.DataFrame,230 splits: pl.DataFrame,231 subs: pl.DataFrame,232 matrix: pl.DataFrame | None,233 geo: pl.DataFrame | None,234 cfg: dict[str, Any] | None = None,235 n_subjects: int = 300,236 n_market_subjects: int = 40,237 n_by_split: dict[str, int] | None = None,238 seed: int = 20261027,239 asof_sample: int | None = 20_000,240) -> dict[str, Any]:241 """Run every check. ``asof_sample``: the feature-matrix perturbation test is run on a random subset of242 municipalities' rows (all rows when None) to keep the audit under a few minutes on the real table."""243 cfg = cfg or load_config("data")244 rng = np.random.default_rng(seed)245 eval_rows = analytic.join(subs.select("tx_id"), on="tx_id", how="inner")246 if eval_rows.height == 0:247 eval_rows = analytic.join(splits.filter(pl.col("split") != "train").select("tx_id"), on="tx_id")248 pick = rng.choice(eval_rows.height, size=min(n_subjects, eval_rows.height), replace=False)249 subjects = eval_rows[sorted(pick.tolist())]250 results = [251 check_split_windows(cfg),252 check_split_assignment(analytic, splits, cfg),253 check_group_ordering(analytic, splits),254 check_prior_sale(analytic),255 check_comparables(analytic, subjects, cfg),256 check_market_context(analytic, subjects.head(n_market_subjects), cfg),257 ]258 # feature-matrix as-of test on a subset (keeps all rows of the sampled parcels' municipalities so rolling windows are intact)259 sub_an = analytic260 if asof_sample is not None and analytic.height > asof_sample:261 muns = analytic["mun_code"].unique().sort().to_list()262 rng2 = np.random.default_rng(seed + 1)263 keep: list[str] = []264 total = 0265 for m in rng2.permutation(muns).tolist():266 keep.append(m)267 total += int((analytic["mun_code"] == m).sum())268 if total >= asof_sample:269 break270 sub_an = analytic.filter(pl.col("mun_code").is_in(keep))271 w = split_windows(cfg)272 results.append(check_feature_matrix_asof(sub_an, splits, geo, cutoff=w["test"][0]))273 if matrix is not None:274 results.append(check_feature_matrix_columns(matrix))275 results.append(check_subsample(analytic, splits, subs, cfg, n_by_split=n_by_split))276 report = {277 "all_passed": all(r.passed for r in results),278 "n_rows": analytic.height,279 "n_subjects_comparables": subjects.height,280 "asof_rows_tested": sub_an.height,281 "checks": [r.to_dict() for r in results],282 }283 return report284285286def to_json(report: dict[str, Any]) -> str:287 return json.dumps(report, indent=1, ensure_ascii=False, default=str)288