1#!/usr/bin/env python32# Author: Simon-Pierre Boucher — simon-pierre.boucher@uqo.ca3"""Leakage audit on the built data layer → results/data/leakage_audit.json (exit code 1 if any check fails).45 python3 scripts/05_leakage_audit.py [--synthetic] [--subjects 300] [--asof-rows 20000]6"""7from __future__ import annotations89import argparse10import sys11import time1213import polars as pl1415from agentic_valuation.config import PROCESSED, RESULTS, SYNTHETIC, load_config16from agentic_valuation.data.leakage import audit, to_json171819def main() -> None:20 ap = argparse.ArgumentParser(description=__doc__)21 ap.add_argument("--synthetic", action="store_true")22 ap.add_argument("--subjects", type=int, default=300)23 ap.add_argument("--asof-rows", type=int, default=20_000, help="rows (whole municipalities) for the feature-matrix perturbation test")24 args = ap.parse_args()25 base = SYNTHETIC if args.synthetic else PROCESSED26 t0 = time.perf_counter()27 analytic = pl.read_parquet(base / "analytic.parquet")28 splits = pl.read_parquet(base / "splits.parquet")29 subs = pl.read_parquet(base / "llm_subsamples.parquet")30 matrix = pl.read_parquet(base / "feature_matrix.parquet") if (base / "feature_matrix.parquet").exists() else None31 geo = pl.read_parquet(base / "geo_features.parquet") if (base / "geo_features.parquet").exists() else None32 n_by = {"validation": 40, "test": 60, "holdout": 30} if args.synthetic else None33 report = audit(analytic, splits, subs, matrix, geo, load_config("data"), n_subjects=args.subjects, n_by_split=n_by, asof_sample=args.asof_rows)34 report["synthetic"] = args.synthetic35 report["seconds"] = round(time.perf_counter() - t0, 1)36 out = RESULTS / "data" / ("leakage_audit_synthetic.json" if args.synthetic else "leakage_audit.json")37 out.parent.mkdir(parents=True, exist_ok=True)38 out.write_text(to_json(report))39 for c in report["checks"]:40 print(f"[{'PASS' if c['passed'] else 'FAIL'}] {c['name']}: checked={c['n_checked']} failed={c['n_failed']}")41 print(f"all_passed={report['all_passed']} seconds={report['seconds']} → {out}")42 sys.exit(0 if report["all_passed"] else 1)434445if __name__ == "__main__":46 main()47