1# Author: Simon-Pierre Boucher — simon-pierre.boucher@uqo.ca2"""Leakage-defence suite (synthetic data; ``--real`` re-runs the audit on the processed real data)."""3from __future__ import annotations45from datetime import date67import polars as pl8import pytest910from agentic_valuation.data import leakage as lk11from tests.conftest import SYN_N_BY_SPLIT121314def test_split_windows_non_overlapping(cfg: dict) -> None:15 r = lk.check_split_windows(cfg)16 assert r.passed, r.details17 assert r.details["gaps"] == []181920def test_split_assignment(syn_analytic: pl.DataFrame, syn_splits: pl.DataFrame, cfg: dict) -> None:21 assert lk.check_split_assignment(syn_analytic, syn_splits, cfg).passed222324def test_group_ordering_no_future_sale_in_train(syn_analytic: pl.DataFrame, syn_splits: pl.DataFrame) -> None:25 r = lk.check_group_ordering(syn_analytic, syn_splits)26 assert r.passed and r.details["parcels_in_train_and_later"] > 0272829def test_group_ordering_detects_violation(syn_analytic: pl.DataFrame, syn_splits: pl.DataFrame) -> None:30 # later sale B of a repeat parcel into train while its earlier sale A sits in test → must fail31 b = syn_analytic.filter(pl.col("prior_sale_date").is_not_null() & (pl.col("date") >= date(2025, 4, 1))).head(1)32 a = syn_analytic.filter((pl.col("parcel_id") == b["parcel_id"][0]) & (pl.col("date") == b["prior_sale_date"][0])).head(1)33 bad = syn_splits.with_columns(34 pl.when(pl.col("tx_id") == b["tx_id"][0]).then(pl.lit("train"))35 .when(pl.col("tx_id") == a["tx_id"][0]).then(pl.lit("test")).otherwise(pl.col("split")).alias("split")36 )37 r = lk.check_group_ordering(syn_analytic, bad)38 assert not r.passed and r.n_failed == 1394041def test_prior_sale_strictly_earlier(syn_analytic: pl.DataFrame) -> None:42 r = lk.check_prior_sale(syn_analytic)43 assert r.passed, r.details444546def test_comparables_asof(syn_analytic: pl.DataFrame, syn_subsamples: pl.DataFrame, cfg: dict) -> None:47 subjects = syn_analytic.join(syn_subsamples.select("tx_id"), on="tx_id")48 r = lk.check_comparables(syn_analytic, subjects, cfg)49 assert r.passed, r.details505152def test_market_context_asof(syn_analytic: pl.DataFrame, syn_subsamples: pl.DataFrame, cfg: dict) -> None:53 subjects = syn_analytic.join(syn_subsamples.select("tx_id"), on="tx_id").head(15)54 assert lk.check_market_context(syn_analytic, subjects, cfg).passed555657def test_feature_matrix_asof(syn_analytic: pl.DataFrame, syn_splits: pl.DataFrame) -> None:58 r = lk.check_feature_matrix_asof(syn_analytic, syn_splits, None, cutoff=date(2025, 4, 1))59 assert r.passed, r.details606162def test_feature_matrix_columns(syn_matrix: pl.DataFrame) -> None:63 assert lk.check_feature_matrix_columns(syn_matrix).passed64 bad = syn_matrix.with_columns(pl.col("log_price").alias("price_now"))65 assert not lk.check_feature_matrix_columns(bad).passed666768def test_subsample_strata_and_determinism(syn_analytic: pl.DataFrame, syn_splits: pl.DataFrame, syn_subsamples: pl.DataFrame, cfg: dict) -> None:69 r = lk.check_subsample(syn_analytic, syn_splits, syn_subsamples, cfg, n_by_split=SYN_N_BY_SPLIT)70 assert r.passed, r.details71 assert r.details["deterministic_redraw"] is True727374def test_full_audit_synthetic(syn_analytic: pl.DataFrame, syn_splits: pl.DataFrame, syn_subsamples: pl.DataFrame, syn_matrix: pl.DataFrame, cfg: dict) -> None:75 rep = lk.audit(syn_analytic, syn_splits, syn_subsamples, syn_matrix, None, cfg, n_subjects=50, n_market_subjects=5, n_by_split=SYN_N_BY_SPLIT, asof_sample=None)76 assert rep["all_passed"], [c for c in rep["checks"] if not c["passed"]]777879@pytest.mark.real80def test_real_audit_passes(real_bundle: dict[str, pl.DataFrame], cfg: dict) -> None:81 rep = lk.audit(real_bundle["analytic"], real_bundle["splits"], real_bundle["llm_subsamples"], real_bundle.get("feature_matrix"), real_bundle.get("geo_features"), cfg, n_subjects=200, n_market_subjects=20, asof_sample=20_000)82 assert rep["all_passed"], [c for c in rep["checks"] if not c["passed"]]83