1# Author: Simon-Pierre Boucher — simon-pierre.boucher@uqo.ca2"""Build the analytic layer from the WP 22 clean-transactions snapshot (real data) or from the synthetic table.34Real-data rules (``configs/data.yaml`` → ``sample``):5* keep ``in_headline_sample`` (WP 22 cleaning: residential, roll-matched, non-duplicate, ratio not an outlier, …);6* ``property_class`` ∈ {house, condo, plex}; non-null coordinates and ``parcel_id``; price in [price_min, price_max];7* roll attributes are mapped from ``roll_*`` to the contract names; the **assessment in force at the sale**8 (``assessment_at_sale``, recorded by the provider at the time of the sale) replaces the current roll total;9 the land/building split of the current roll is rescaled to that total (the split of the earlier roll is not in10 the snapshot); the roll reference date is the one of the roll in force at the sale, inferred from the triennial11 cycle (reference date + 18 months = entry into force);12* ``urban_class`` from the MAMH municipal population (≥ 100 000 metro, ≥ 10 000 suburban, else rural).1314Everything downstream (features, agents, evaluation) consumes only the resulting table.15"""16from __future__ import annotations1718import json19import time20from datetime import date21from pathlib import Path22from typing import Any2324import polars as pl2526from ..config import PROCESSED, RAW, RESULTS, SYNTHETIC, load_config27from .analytic_schema import ANALYTIC_COLUMNS, add_derived_columns, coerce_analytic, validate_analytic2829URBAN_METRO_POP = 100_00030URBAN_SUBURBAN_POP = 10_00031ROLL_ENTRY_LAG_MONTHS = 18 # Quebec triennial rolls: market reference date = 1 July, entry into force 18 months later32ROLL_CYCLE_YEARS = 3333435def urban_class_expr(pop_col: str = "mamh_population") -> pl.Expr:36 return (37 pl.when(pl.col(pop_col) >= URBAN_METRO_POP).then(pl.lit("metro"))38 .when(pl.col(pop_col) >= URBAN_SUBURBAN_POP).then(pl.lit("suburban"))39 .otherwise(pl.lit("rural"))40 )414243def roll_reference_at_sale_expr(ref_col: str = "roll_reference_date", date_col: str = "date") -> pl.Expr:44 """Reference date of the roll in force at the sale: step back whole triennial cycles from the snapshot roll's45 reference date until reference + 18 months ≤ sale date. Null when the reference date is unknown."""46 ref = pl.col(ref_col).str.to_date("%Y-%m-%d", strict=False)47 entry = ref.dt.offset_by(f"{ROLL_ENTRY_LAG_MONTHS}mo")48 # number of 3-year cycles to step back (0 when the snapshot roll was already in force at the sale)49 months_short = (entry.dt.year() * 12 + entry.dt.month()) - (pl.col(date_col).dt.year() * 12 + pl.col(date_col).dt.month())50 # entry > sale when months_short > 0, or same month but later day51 later = (months_short > 0) | ((months_short == 0) & (entry.dt.day() > pl.col(date_col).dt.day()))52 k = pl.when(later).then(((months_short - 1) // (ROLL_CYCLE_YEARS * 12)) + 1).otherwise(0).clip(0, None)53 out = ref.dt.offset_by(pl.format("-{}y", k * ROLL_CYCLE_YEARS))54 return out.dt.strftime("%Y-%m-%d")555657def _funnel(raw: pl.DataFrame, cfg: dict[str, Any]) -> tuple[pl.DataFrame, list[dict[str, Any]]]:58 s = cfg["sample"]59 steps: list[tuple[str, pl.Expr]] = [60 ("in_headline_sample", pl.col("in_headline_sample").fill_null(False)),61 ("property_class_in_scope", pl.col("property_class").is_in(s["property_classes"])),62 ("coordinates_present", pl.col("lat").is_not_null() & pl.col("lng").is_not_null()),63 ("parcel_id_present", pl.col("parcel_id").is_not_null()),64 ("price_in_range", (pl.col("price") >= s["price_min"]) & (pl.col("price") <= s["price_max"])),65 ]66 if s.get("require_roll_match", True) and "roll_matched" in raw.columns:67 steps.append(("roll_matched", pl.col("roll_matched").fill_null(False)))68 funnel: list[dict[str, Any]] = [{"step": "raw", "rows": raw.height, "dropped": 0}]69 df = raw70 for name, expr in steps:71 before = df.height72 df = df.filter(expr)73 funnel.append({"step": name, "rows": df.height, "dropped": before - df.height})74 return df, funnel757677def build_from_real(raw: pl.DataFrame, cfg: dict[str, Any] | None = None) -> tuple[pl.DataFrame, dict[str, Any]]:78 """WP 22 clean transactions → analytic table + build summary."""79 cfg = cfg or load_config("data")80 df, funnel = _funnel(raw, cfg)81 scale = pl.when(pl.col("roll_total_value") > 0).then(pl.col("assessment_at_sale") / pl.col("roll_total_value")).otherwise(None)82 df = df.with_columns(83 pl.col("roll_floor_area_m2").alias("floor_area_m2"),84 pl.col("roll_land_area_m2").alias("land_area_m2"),85 pl.col("roll_year_built").cast(pl.Int64).alias("year_built"),86 pl.col("roll_storeys").alias("storeys"),87 pl.col("roll_n_units").cast(pl.Int64).alias("n_units"),88 (pl.col("roll_land_value") * scale).round(0).alias("roll_land_value"),89 (pl.col("roll_building_value") * scale).round(0).alias("roll_building_value"),90 pl.col("roll_total_value").alias("_snapshot_roll_total"),91 pl.coalesce(pl.col("assessment_at_sale"), pl.col("roll_total_value")).alias("roll_total_value"),92 pl.col("roll_reference_date").alias("_snapshot_roll_ref"),93 roll_reference_at_sale_expr().alias("roll_reference_date"),94 urban_class_expr().alias("urban_class"),95 pl.col("subtype").fill_null("unknown"),96 )97 # diagnostic: does the inferred roll cycle agree with the recorded assessment? (k=0 ⇔ assessment == snapshot roll)98 diag_df = df.select(99 (pl.col("roll_reference_date") == pl.col("_snapshot_roll_ref")).alias("k0"),100 (pl.col("roll_total_value") == pl.col("_snapshot_roll_total")).alias("eq_cur"),101 (pl.col("roll_total_value") == pl.col("roll_previous_value")).alias("eq_prev"),102 )103 roll_diag = {104 "share_inferred_current_roll": float(diag_df["k0"].mean()),105 "share_assessment_equals_snapshot_roll": float(diag_df["eq_cur"].mean()),106 "share_assessment_equals_previous_roll": float(diag_df["eq_prev"].mean()),107 "agreement_k0_vs_eq_cur": float((diag_df["k0"] == diag_df["eq_cur"]).mean()),108 "note": "roll_total_value = assessment recorded at the sale (roll in force); reference date inferred from the triennial cycle",109 }110 df = add_derived_columns(df, recompute_xy=False)111 out = coerce_analytic(df)112 validate_analytic(out)113 summary = summarize(out)114 summary["funnel"] = funnel115 summary["roll_as_of_diagnostic"] = roll_diag116 summary["source"] = cfg["snapshot"]["transactions_parquet"]117 return out, summary118119120def build_from_synthetic(syn: pl.DataFrame) -> tuple[pl.DataFrame, dict[str, Any]]:121 """Synthetic generator output (24 columns) → analytic table (x/y projected with pyproj)."""122 df = add_derived_columns(syn, recompute_xy=True)123 out = coerce_analytic(df)124 validate_analytic(out)125 summary = summarize(out)126 summary["source"] = "data/synthetic/synthetic_transactions.parquet (SYNTHETIC)"127 return out, summary128129130def summarize(df: pl.DataFrame) -> dict[str, Any]:131 """Row counts by year / class / region group, missingness, tercile cut points, date range."""132 def counts(col: pl.Expr, name: str) -> dict[str, int]:133 t = df.group_by(col.alias(name)).len().sort(name)134 return {str(k): int(v) for k, v in zip(t[name].to_list(), t["len"].to_list())}135136 terciles = (137 df.group_by("property_class").agg(138 pl.col("price").quantile(1 / 3, interpolation="linear").alias("q33"),139 pl.col("price").quantile(2 / 3, interpolation="linear").alias("q67"),140 ).sort("property_class")141 )142 year_class = df.group_by(pl.col("date").dt.year().alias("year"), "property_class").len().sort(["year", "property_class"])143 return {144 "rows": df.height,145 "parcels": int(df["parcel_id"].n_unique()),146 "date_min": df["date"].min().isoformat(),147 "date_max": df["date"].max().isoformat(),148 "by_year": counts(pl.col("date").dt.year(), "year"),149 "by_property_class": counts(pl.col("property_class"), "property_class"),150 "by_region_group": counts(pl.col("region_group"), "region_group"),151 "by_region_name": counts(pl.col("region_name"), "region_name"),152 "by_urban_class": counts(pl.col("urban_class"), "urban_class"),153 "by_year_class": [{"year": int(y), "property_class": c, "rows": int(n)} for y, c, n in year_class.iter_rows()],154 "missingness_pct": {c: round(100.0 * df[c].null_count() / max(df.height, 1), 3) for c in ANALYTIC_COLUMNS},155 "price_tercile_cuts": {r["property_class"]: [r["q33"], r["q67"]] for r in terciles.iter_rows(named=True)},156 "rows_with_prior_sale": int(df["prior_sale_price"].is_not_null().sum()),157 "price_median": float(df["price"].median()),158 }159160161def run(synthetic: bool = False, raw_path: Path | None = None) -> dict[str, Any]:162 """Script entry point: build, write parquet + summary JSON, return summary with timing."""163 cfg = load_config("data")164 t0 = time.perf_counter()165 if synthetic:166 src = raw_path or SYNTHETIC / "synthetic_transactions.parquet"167 if not src.exists():168 from .synthetic import write_default169170 write_default()171 df, summary = build_from_synthetic(pl.read_parquet(src))172 out = SYNTHETIC / "analytic.parquet"173 summary_path = RESULTS / "data" / "analytic_summary_synthetic.json"174 else:175 src = raw_path or RAW / cfg["snapshot"]["transactions_parquet"].split("data/raw/")[-1]176 raw = pl.read_parquet(src)177 df, summary = build_from_real(raw, cfg)178 out = PROCESSED / "analytic.parquet"179 summary_path = RESULTS / "data" / "analytic_summary.json"180 out.parent.mkdir(parents=True, exist_ok=True)181 summary_path.parent.mkdir(parents=True, exist_ok=True)182 df.write_parquet(out, compression="zstd")183 summary["output"] = str(out.relative_to(out.parents[2]))184 summary["built_at"] = date.today().isoformat()185 summary["seconds"] = round(time.perf_counter() - t0, 2)186 summary_path.write_text(json.dumps(summary, indent=1, ensure_ascii=False))187 return summary188