# Author: Simon-Pierre Boucher — simon-pierre.boucher@uqo.ca """Leakage-audit suite. Every check returns a ``CheckResult``; ``audit`` runs them all and produces a JSON-able report (``scripts/05_leakage_audit.py`` → ``results/data/leakage_audit.json``). The same functions back ``tests/test_leakage.py`` (synthetic data) so that the paper's leakage claims trace to executed code. Checks ------ 1. ``check_split_windows`` split windows ordered, non-overlapping, contiguous (config). 2. ``check_split_assignment`` every row's split matches its date; partitions disjoint. 3. ``check_group_ordering`` no parcel has a sale in ``train`` dated on/after a sale of the same parcel in a later partition (validation/test/holdout), i.e. no "future" sale of a test parcel is trained on. 4. ``check_prior_sale`` prior_sale_date < date, and equals the latest strictly-earlier sale of the parcel. 5. ``check_comparables`` for a sample of subjects: all comparables dated < cutoff, within the look-back, subject parcel absent, same class when required, ids C01.. in order. 6. ``check_market_context`` market context never uses sales after the cutoff (perturbation test). 7. ``check_feature_matrix_asof`` perturbing (or deleting) every sale dated ≥ D leaves the feature rows of sales dated < D unchanged — the matrix contains no same-day/post-sale information. 8. ``check_feature_matrix_columns`` no target-derived column is used as a feature. 9. ``check_subsample`` stratum counts follow the proportional allocation; the draw is deterministic. """ from __future__ import annotations import json from dataclasses import asdict, dataclass, field from datetime import date, timedelta from typing import Any import numpy as np import polars as pl from ..config import load_config from .analytic_schema import add_prior_sale from .splits import SPLIT_ORDER, assign_splits, check_windows, proportional_allocation, split_windows, stratified_subsample @dataclass class CheckResult: name: str passed: bool n_checked: int = 0 n_failed: int = 0 details: dict[str, Any] = field(default_factory=dict) def to_dict(self) -> dict[str, Any]: return asdict(self) # --------------------------------------------------------------------------------------------------------------- def check_split_windows(cfg: dict[str, Any]) -> CheckResult: try: check_windows(cfg) w = split_windows(cfg) gaps = [] for a, b in zip(SPLIT_ORDER, SPLIT_ORDER[1:]): if (w[b][0] - w[a][1]).days != 1: gaps.append(f"{a}->{b}: {(w[b][0] - w[a][1]).days - 1} day(s) gap") 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}) except ValueError as e: return CheckResult("split_windows", False, len(SPLIT_ORDER), 1, {"error": str(e)}) def check_split_assignment(analytic: pl.DataFrame, splits: pl.DataFrame, cfg: dict[str, Any]) -> CheckResult: w = split_windows(cfg) df = analytic.select("tx_id", "date").join(splits, on="tx_id", how="inner") bad = 0 for name, (a, b) in w.items(): part = df.filter(pl.col("split") == name) bad += int(((part["date"] < a) | (part["date"] > b)).sum()) dup = splits.height - splits["tx_id"].n_unique() missing = analytic.height - df.height ok = bad == 0 and dup == 0 and missing == 0 return CheckResult("split_assignment", ok, df.height, bad + dup + missing, {"out_of_window": bad, "duplicate_tx": dup, "rows_without_split": missing}) def check_group_ordering(analytic: pl.DataFrame, splits: pl.DataFrame) -> CheckResult: """For each parcel: max train date < min date in every later partition.""" rank = {s: i for i, s in enumerate(SPLIT_ORDER)} df = analytic.select("tx_id", "parcel_id", "date").join(splits.select("tx_id", "split"), on="tx_id", how="inner") df = df.with_columns(pl.col("split").replace_strict(rank, default=99, return_dtype=pl.Int8).alias("_r")) agg = df.group_by("parcel_id").agg( pl.col("date").filter(pl.col("_r") == 0).max().alias("max_train"), pl.col("date").filter(pl.col("_r") > 0).min().alias("min_later"), ).drop_nulls() viol = agg.filter(pl.col("max_train") >= pl.col("min_later")) return CheckResult("group_ordering", viol.height == 0, agg.height, viol.height, {"parcels_in_train_and_later": agg.height}) def check_prior_sale(analytic: pl.DataFrame) -> CheckResult: ps = analytic.filter(pl.col("prior_sale_date").is_not_null()) not_earlier = int((ps["prior_sale_date"] >= ps["date"]).sum()) recomputed = add_prior_sale(analytic.drop(["prior_sale_price", "prior_sale_date"])) cmp = analytic.select("tx_id", "prior_sale_price", "prior_sale_date").join( recomputed.select("tx_id", pl.col("prior_sale_price").alias("_p"), pl.col("prior_sale_date").alias("_d")), on="tx_id" ) mismatch = int((cmp["prior_sale_date"].ne_missing(cmp["_d"]) | cmp["prior_sale_price"].ne_missing(cmp["_p"])).sum()) # brute force on a sample of repeat parcels multi = analytic.filter(pl.col("parcel_id").is_duplicated()).sort(["parcel_id", "date"]) bf_bad = 0 bf_n = 0 for _pid, g in list(multi.group_by("parcel_id", maintain_order=True))[:2000]: rows = g.to_dicts() for r in rows: earlier = [q for q in rows if q["date"] < r["date"]] bf_n += 1 if not earlier: bf_bad += r["prior_sale_date"] is not None else: dmax = max(q["date"] for q in earlier) if r["prior_sale_date"] != dmax or r["prior_sale_price"] not in {q["price"] for q in earlier if q["date"] == dmax}: bf_bad += 1 ok = not_earlier == 0 and mismatch == 0 and bf_bad == 0 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}) def check_comparables(analytic: pl.DataFrame, subjects: pl.DataFrame, cfg: dict[str, Any]) -> CheckResult: from ..features.comparables import retrieve_comparables_detailed c = cfg["comparables"] lookback = int(c["lookback_days"]) 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} parcel_of = dict(zip(analytic["tx_id"].to_list(), analytic["parcel_id"].to_list())) date_of = dict(zip(analytic["tx_id"].to_list(), analytic["date"].to_list())) n_comps = [] for r in subjects.iter_rows(named=True): cutoff = r["date"] - timedelta(days=1) cs = retrieve_comparables_detailed(analytic, r, cutoff, cfg) comps = cs.comparables n_comps.append(len(comps)) fails["date_ge_cutoff"] += sum(cp.sale_date >= cutoff for cp in comps) fails["beyond_lookback"] += sum((cutoff - cp.sale_date).days > lookback for cp in comps) fails["class_mismatch"] += sum(cp.property_class != r["property_class"] for cp in comps) if c.get("same_class_required", True) else 0 fails["id_order"] += int([cp.evidence_id for cp in comps] != [f"C{i:02d}" for i in range(1, len(comps) + 1)]) fails["too_many"] += int(len(comps) > int(c["max_comps"])) fails["fewer_than_min_unexpanded"] += int(len(comps) < int(c["min_comps"]) and not cs.expanded) # own parcel: compare against every sale of the subject's parcel (by date+price identity is not unique → use parcel sales set) own = analytic.filter(pl.col("parcel_id") == r["parcel_id"]).select("date", "price") own_set = set(zip(own["date"].to_list(), own["price"].to_list())) # a comparable is from the subject's parcel only if it matches a same-parcel sale AND lies at distance 0 fails["own_parcel"] += sum(((cp.sale_date, cp.price) in own_set and cp.distance_km == 0.0) for cp in comps) n_fail = sum(fails.values()) 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}) def check_market_context(analytic: pl.DataFrame, subjects: pl.DataFrame, cfg: dict[str, Any]) -> CheckResult: """Perturb every sale after the cutoff (price × 10) and recompute: the context must not change.""" from ..features.market import market_context n_fail = 0 for r in subjects.iter_rows(named=True): cutoff = r["date"] - timedelta(days=1) base = market_context(analytic, r, cutoff, cfg).model_dump() pert = analytic.with_columns(pl.when(pl.col("date") > cutoff).then(pl.col("price") * 10).otherwise(pl.col("price")).alias("price")) pert = pert.with_columns(pl.col("price").log().alias("log_price")) alt = market_context(pert, r, cutoff, cfg).model_dump() n_fail += int(base != alt) return CheckResult("market_context_asof", n_fail == 0, subjects.height, n_fail) def check_feature_matrix_asof(analytic: pl.DataFrame, splits: pl.DataFrame, geo: pl.DataFrame | None, cutoff: date) -> CheckResult: """Features of sales dated < cutoff must be invariant to perturbing AND to deleting all sales dated ≥ cutoff.""" from ..features.matrix import build_feature_matrix base = build_feature_matrix(analytic, splits, geo=geo).filter(pl.col("date") < cutoff).sort("tx_id") pert = analytic.with_columns(pl.when(pl.col("date") >= cutoff).then(pl.col("price") * 7).otherwise(pl.col("price")).alias("price")) pert = pert.with_columns(pl.col("price").log().alias("log_price")) alt1 = build_feature_matrix(pert, splits, geo=geo).filter(pl.col("date") < cutoff).sort("tx_id") dele = analytic.filter(pl.col("date") < cutoff) alt2 = build_feature_matrix(dele, splits, geo=geo).filter(pl.col("date") < cutoff).sort("tx_id") n_fail = 0 diffs: list[str] = [] for alt in (alt1, alt2): if alt.height != base.height or alt["tx_id"].to_list() != base["tx_id"].to_list(): n_fail += 1 diffs.append("row set differs") continue for c in base.columns: if c in ("price", "log_price"): continue a, b = base[c], alt[c] if a.dtype == pl.Float64: a, b = a.fill_nan(None), b.fill_nan(None) same = ((a - b).abs() < 1e-9) | (a.is_null() & b.is_null()) else: same = a.eq_missing(b) if not same.all(): n_fail += 1 diffs.append(c) return CheckResult("feature_matrix_asof", n_fail == 0, base.height, n_fail, {"cutoff": cutoff.isoformat(), "columns_changed": sorted(set(diffs))}) FORBIDDEN_FEATURES = {"price", "log_price", "price_tercile", "exclude_reason", "assessment_ratio", "ppm2"} def check_feature_matrix_columns(matrix: pl.DataFrame) -> CheckResult: from ..features.matrix import feature_columns num, cat = feature_columns(matrix) 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"))] return CheckResult("feature_matrix_columns", not bad, len(num) + len(cat), len(bad), {"forbidden_present": bad, "numeric": num, "categorical": cat}) def check_subsample(analytic: pl.DataFrame, splits: pl.DataFrame, subs: pl.DataFrame, cfg: dict[str, Any], n_by_split: dict[str, int] | None = None) -> CheckResult: strata_cols = list(cfg["llm_eval_subsample"]["strata"]) s = cfg["llm_eval_subsample"] n_by_split = n_by_split or {"validation": s["validation_n"], "test": s["test_n"], "holdout": s["holdout_n"]} df = analytic.select(["tx_id", *strata_cols]).join(splits.select("tx_id", "split"), on="tx_id") df = df.with_columns(pl.concat_str([pl.col(c).cast(pl.String) for c in strata_cols], separator="|").alias("stratum")) n_fail = 0 details: dict[str, Any] = {} for split, n in n_by_split.items(): counts = {k: int(v) for k, v in df.filter(pl.col("split") == split).group_by("stratum").len().iter_rows()} expected = proportional_allocation(counts, int(n)) got = {k: int(v) for k, v in subs.filter(pl.col("split") == split).group_by("stratum").len().iter_rows()} 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)} n_fail += len(mism) details[split] = {"n": int(sum(got.values())), "target": int(n), "strata": len(got), "mismatched_strata": mism} # membership + determinism joined = subs.join(df.select("tx_id", pl.col("split").alias("_s"), pl.col("stratum").alias("_st")), on="tx_id", how="left") n_fail += int((joined["split"] != joined["_s"]).sum() + (joined["stratum"] != joined["_st"]).sum() + joined["_s"].null_count()) redraw = stratified_subsample(analytic, splits, cfg, n_by_split=n_by_split) deterministic = redraw.sort(["split", "stratum", "tx_id"]).equals(subs.sort(["split", "stratum", "tx_id"])) n_fail += int(not deterministic) details["deterministic_redraw"] = deterministic return CheckResult("llm_subsample", n_fail == 0, subs.height, n_fail, details) # --------------------------------------------------------------------------------------------------------------- def audit( analytic: pl.DataFrame, splits: pl.DataFrame, subs: pl.DataFrame, matrix: pl.DataFrame | None, geo: pl.DataFrame | None, cfg: dict[str, Any] | None = None, n_subjects: int = 300, n_market_subjects: int = 40, n_by_split: dict[str, int] | None = None, seed: int = 20261027, asof_sample: int | None = 20_000, ) -> dict[str, Any]: """Run every check. ``asof_sample``: the feature-matrix perturbation test is run on a random subset of municipalities' rows (all rows when None) to keep the audit under a few minutes on the real table.""" cfg = cfg or load_config("data") rng = np.random.default_rng(seed) eval_rows = analytic.join(subs.select("tx_id"), on="tx_id", how="inner") if eval_rows.height == 0: eval_rows = analytic.join(splits.filter(pl.col("split") != "train").select("tx_id"), on="tx_id") pick = rng.choice(eval_rows.height, size=min(n_subjects, eval_rows.height), replace=False) subjects = eval_rows[sorted(pick.tolist())] results = [ check_split_windows(cfg), check_split_assignment(analytic, splits, cfg), check_group_ordering(analytic, splits), check_prior_sale(analytic), check_comparables(analytic, subjects, cfg), check_market_context(analytic, subjects.head(n_market_subjects), cfg), ] # feature-matrix as-of test on a subset (keeps all rows of the sampled parcels' municipalities so rolling windows are intact) sub_an = analytic if asof_sample is not None and analytic.height > asof_sample: muns = analytic["mun_code"].unique().sort().to_list() rng2 = np.random.default_rng(seed + 1) keep: list[str] = [] total = 0 for m in rng2.permutation(muns).tolist(): keep.append(m) total += int((analytic["mun_code"] == m).sum()) if total >= asof_sample: break sub_an = analytic.filter(pl.col("mun_code").is_in(keep)) w = split_windows(cfg) results.append(check_feature_matrix_asof(sub_an, splits, geo, cutoff=w["test"][0])) if matrix is not None: results.append(check_feature_matrix_columns(matrix)) results.append(check_subsample(analytic, splits, subs, cfg, n_by_split=n_by_split)) report = { "all_passed": all(r.passed for r in results), "n_rows": analytic.height, "n_subjects_comparables": subjects.height, "asof_rows_tested": sub_an.height, "checks": [r.to_dict() for r in results], } return report def to_json(report: dict[str, Any]) -> str: return json.dumps(report, indent=1, ensure_ascii=False, default=str)