spb/wp22_uqo

1#!/usr/bin/env python32# Author: Simon-Pierre Boucher — simon-pierre.boucher@uqo.ca3"""Phase 1 — registration / publication lag measured from earlier data vintages.45Compares four earlier extractions of the same transaction table (ids + dates)6with the final snapshot. For a vintage extracted on day D, a transaction dated7t that is absent from the vintage but present in the final snapshot arrived8with a lag > D - t. This yields an empirical arrival profile (share of9eventual transactions visible after L days) used to simulate real-time vintages.10Also records deletions and price revisions between vintages.11"""12from __future__ import annotations1314import sys15from pathlib import Path1617sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))18import polars as pl1920from sharp_hpi import config as C2122OUT = C.RESULTS / "vintages"23OUT.mkdir(parents=True, exist_ok=True)242526def load_vintage(path: Path) -> pl.DataFrame:27    if path.suffix == ".gz":28        import gzip, io29        with gzip.open(path, "rb") as f:30            df = pl.read_csv(io.BytesIO(f.read()), infer_schema_length=0)31    else:32        df = pl.read_parquet(path)33    return df.select(pl.col("id"), pl.col("date").cast(pl.Utf8).str.slice(0, 10).str.to_date("%Y-%m-%d"), pl.col("amount").cast(pl.Float64, strict=False))343536def main() -> None:37    final = pl.read_parquet(C.RAW_TX, columns=["id", "date", "amount", "property_type"])38    rows, profile, revisions = [], [], []39    for vdate, path in C.VINTAGES.items():40        v = load_vintage(path)41        vd = pl.lit(vdate).str.to_date("%Y-%m-%d")42        in_final = final.join(v.select("id", pl.col("amount").alias("amount_v")), on="id", how="left")43        # eligible: dated on/before the vintage's last observed date44        last_obs = v["date"].max()45        elig = in_final.filter(pl.col("date") <= last_obs)46        present = elig["amount_v"].is_not_null()47        deleted = v.join(final.select("id"), on="id", how="anti")48        rows.append({49            "vintage": vdate, "vintage_rows": v.height, "vintage_last_date": str(last_obs), "vintage_max_lag_days": (pl.Series([vdate]).str.to_date("%Y-%m-%d") - pl.Series([last_obs]))[0].days,50            "final_rows_dated_le_last": elig.height, "present_in_vintage": int(present.sum()), "share_present": float(present.mean()),51            "deleted_since_vintage": deleted.height,52            "amount_revised": int((elig.filter(present)["amount"] != elig.filter(present)["amount_v"]).sum()),53        })54        # arrival profile by lag (days between transaction date and vintage date)55        prof = elig.with_columns(((vd - pl.col("date")).dt.total_days()).alias("lag_days"), pl.col("amount_v").is_not_null().alias("present")) \56            .with_columns((pl.col("lag_days") // 7 * 7).alias("lag_bin")).group_by("lag_bin").agg(pl.len().alias("n_final"), pl.col("present").mean().alias("share_visible")) \57            .with_columns(pl.lit(vdate).alias("vintage")).sort("lag_bin")58        profile.append(prof)59        # by transaction month60        bym = elig.with_columns(pl.col("date").dt.strftime("%Y-%m").alias("month"), pl.col("amount_v").is_not_null().alias("present")).group_by("month").agg(61            pl.len().alias("n_final"), pl.col("present").sum().alias("n_in_vintage"), pl.col("present").mean().alias("share_visible")).with_columns(pl.lit(vdate).alias("vintage")).sort("month")62        revisions.append(bym)63    pl.DataFrame(rows).write_csv(OUT / "vintage_summary.csv")64    pl.concat(profile).write_csv(OUT / "arrival_profile_by_lag.csv")65    pl.concat(revisions).write_csv(OUT / "arrival_by_month.csv")66    print(pl.DataFrame(rows))676869if __name__ == "__main__":70    main()71