#!/usr/bin/env python3 # ============================================================================= # QWHPI — Quebec Weekly Housing Price Index # Author : Simon-Pierre Boucher # Contact : contact@spboucher.ai # File : engine/scripts/01_profile.py # Purpose : Step 1 audit — schema, descriptives, missingness, outlier scan, # coverage, duplicates, and weekly liquidity matrices. No index yet. # ============================================================================= """Full dataset audit (Execution Order step 1). Idempotent: re-running overwrites its own outputs. Writes tables to ``outputs/tables/`` and a human-readable report to ``outputs/reports/audit_report.md``. The raw CSV is never modified. """ from __future__ import annotations import sys from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src")) import polars as pl from qwhpi.config import REPORTS_DIR, TABLES_DIR, ensure_dirs from qwhpi.ingest import load_raw pl.Config.set_tbl_rows(50) REPORT_LINES: list[str] = [] def note(line: str = "") -> None: REPORT_LINES.append(line) print(line) def save(df: pl.DataFrame, name: str) -> None: path = TABLES_DIR / f"{name}.csv" df.write_csv(path) note(f" -> table saved: outputs/tables/{name}.csv ({df.height} rows)") def main() -> None: ensure_dirs() df = load_raw() note("# QWHPI Data Audit Report") note("") note("Author: Simon-Pierre Boucher — contact@spboucher.ai") note("") # ------------------------------------------------------------------ # # 1. Schema & basic shape # ------------------------------------------------------------------ # note("## 1. Schema and shape") note("") note(f"- Rows: {df.height:,}") note(f"- Columns: {df.width}") note(f"- Date range: {df['date'].min()} -> {df['date'].max()}") n_weeks = df["week"].n_unique() note(f"- Distinct Monday-labeled weeks with >=1 transaction: {n_weeks}") note(f"- Distinct cities (raw text field): {df['city'].n_unique():,}") note(f"- Distinct ids: {df['id'].n_unique():,} (duplicated ids: {df.height - df['id'].n_unique():,})") note("") schema_tbl = pl.DataFrame( {"column": df.columns, "dtype": [str(t) for t in df.dtypes]} ) save(schema_tbl, "audit_schema") # ------------------------------------------------------------------ # # 2. Missingness # ------------------------------------------------------------------ # note("") note("## 2. Missingness") note("") nulls = df.null_count().transpose(include_header=True, header_name="column", column_names=["n_missing"]) nulls = nulls.with_columns( (pl.col("n_missing") / df.height * 100).round(2).alias("pct_missing") ).sort("n_missing", descending=True) save(nulls, "audit_missingness") for row in nulls.filter(pl.col("n_missing") > 0).iter_rows(named=True): note(f"- `{row['column']}`: {row['n_missing']:,} missing ({row['pct_missing']}%)") # Sentinel-style missingness that null counts miss. note("") note("Sentinel / degenerate values (not encoded as null):") sentinels = pl.DataFrame({ "check": [ "amount <= 0", "amount in (1, 2)", "amount < 5000", "floorArea <= 0", "floorArea > 2000 m2", "yearBuilt < 1600", "yearBuilt > 2026", "yearBuilt == 0", "previousValue <= 0", "totalArValue <= 0", "lat/lng outside Quebec bbox (44.9-62.7N, -79.8--57W)", ], "n": [ df.filter(pl.col("amount") <= 0).height, df.filter(pl.col("amount").is_in([1, 2])).height, df.filter(pl.col("amount") < 5000).height, df.filter(pl.col("floorArea") <= 0).height, df.filter(pl.col("floorArea") > 2000).height, df.filter(pl.col("yearBuilt") < 1600).height, df.filter(pl.col("yearBuilt") > 2026).height, df.filter(pl.col("yearBuilt") == 0).height, df.filter(pl.col("previousValue") <= 0).height, df.filter(pl.col("totalArValue") <= 0).height, df.filter( (pl.col("lat") < 44.9) | (pl.col("lat") > 62.7) | (pl.col("lng") < -79.8) | (pl.col("lng") > -57.0) ).height, ], }) save(sentinels, "audit_sentinel_values") for row in sentinels.iter_rows(named=True): note(f"- {row['check']}: {row['n']:,}") # ------------------------------------------------------------------ # # 3. Categorical distributions # ------------------------------------------------------------------ # note("") note("## 3. Categorical distributions") for col in ("propertyType", "buildingType", "ownerType"): note("") note(f"### {col}") dist = ( df.group_by(col).len().sort("len", descending=True) .with_columns((pl.col("len") / df.height * 100).round(2).alias("pct")) ) save(dist, f"audit_dist_{col}") for row in dist.head(15).iter_rows(named=True): note(f"- {row[col]}: {row['len']:,} ({row['pct']}%)") # ------------------------------------------------------------------ # # 4. Price descriptives (by property type) # ------------------------------------------------------------------ # note("") note("## 4. Amount descriptives by propertyType (amount > 0)") note("") pos = df.filter(pl.col("amount") > 0) desc = ( pos.group_by("propertyType") .agg( n=pl.len(), p01=pl.col("amount").quantile(0.01), p05=pl.col("amount").quantile(0.05), p25=pl.col("amount").quantile(0.25), median=pl.col("amount").median(), p75=pl.col("amount").quantile(0.75), p95=pl.col("amount").quantile(0.95), p99=pl.col("amount").quantile(0.99), max=pl.col("amount").max(), mean=pl.col("amount").mean().round(0), ) .sort("n", descending=True) ) save(desc, "audit_amount_by_type") for row in desc.iter_rows(named=True): note( f"- {row['propertyType']}: n={row['n']:,} | p01=${row['p01']:,.0f} " f"| median=${row['median']:,.0f} | p99=${row['p99']:,.0f} | max=${row['max']:,.0f}" ) # Same for hedonic characteristics. char_desc = ( df.group_by("propertyType") .agg( n=pl.len(), floorArea_med=pl.col("floorArea").median(), floorArea_p99=pl.col("floorArea").quantile(0.99), yearBuilt_med=pl.col("yearBuilt").median(), yearBuilt_min=pl.col("yearBuilt").min(), floorArea_miss_pct=(pl.col("floorArea").is_null().mean() * 100).round(2), yearBuilt_miss_pct=(pl.col("yearBuilt").is_null().mean() * 100).round(2), buildingType_miss_pct=(pl.col("buildingType").is_null().mean() * 100).round(2), ) .sort("n", descending=True) ) save(char_desc, "audit_characteristics_by_type") # ------------------------------------------------------------------ # # 5. Duplicates # ------------------------------------------------------------------ # note("") note("## 5. Duplicate scan") note("") full_dupes = df.height - df.unique(subset=[c for c in df.columns if c not in ("id",)]).height addr_date = df.height - df.unique(subset=["street", "city", "date"]).height addr_date_amt = df.height - df.unique(subset=["street", "city", "date", "amount"]).height note(f"- Exact duplicates ignoring id: {full_dupes:,}") note(f"- Same street+city+date (candidate multi-unit or true dupes): {addr_date:,}") note(f"- Same street+city+date+amount (strong duplicate candidates): {addr_date_amt:,}") # Repeat-sales potential: same street+city appearing on different dates. repeat = ( df.group_by(["street", "city"]).agg(n_dates=pl.col("date").n_unique()) .filter(pl.col("n_dates") >= 2) ) note(f"- Addresses (street+city) with >=2 distinct sale dates: {repeat.height:,}") # ------------------------------------------------------------------ # # 6. Weekly coverage & liquidity # ------------------------------------------------------------------ # note("") note("## 6. Weekly coverage and liquidity") note("") weekly = df.group_by("week").len().sort("week") first, last = weekly["week"].min(), weekly["week"].max() grid = pl.DataFrame({"week": pl.date_range(first, last, "1w", eager=True)}) weekly_full = grid.join(weekly, on="week", how="left").fill_null(0) zero_weeks = weekly_full.filter(pl.col("len") == 0) note(f"- Continuous Monday grid: {grid.height} weeks ({first} -> {last})") note(f"- Zero-transaction weeks on the grid: {zero_weeks.height}") note(f"- Median tx/week (province): {weekly_full['len'].median():,.0f}") note(f"- Min tx/week: {weekly_full['len'].min():,} | Max: {weekly_full['len'].max():,}") tail = weekly_full.tail(6) note(f"- Last 6 weeks (partial-week check): " + ", ".join(f"{r['week']}={r['len']}" for r in tail.iter_rows(named=True))) save(weekly_full.rename({"len": "transactions"}), "audit_weekly_volume_province") # Liquidity matrix: median weekly tx for top cities x property type. top_cities = ( df.group_by("city").len().sort("len", descending=True).head(25)["city"].to_list() ) liq = ( df.filter(pl.col("city").is_in(top_cities)) .group_by(["city", "propertyType", "week"]).len() .group_by(["city", "propertyType"]) .agg( median_wk=pl.col("len").median(), mean_wk=pl.col("len").mean().round(1), weeks_present=pl.len(), ) .with_columns( pl.col("weeks_present").truediv(grid.height).mul(100).round(1).alias("pct_weeks_present") ) .sort(["city", "propertyType"]) ) save(liq, "audit_weekly_liquidity_topcities") pivot = ( liq.pivot(values="median_wk", index="city", on="propertyType") .sort("unifamilial", descending=True, nulls_last=True) ) save(pivot, "audit_liquidity_matrix") note("") note("Median weekly transactions, top cities (rows) x type (cols): see audit_liquidity_matrix.csv") for row in pivot.head(12).iter_rows(named=True): note(f"- {row}") # City name hygiene: how many city strings, top mass share. city_dist = df.group_by("city").len().sort("len", descending=True) top50_share = city_dist.head(50)["len"].sum() / df.height * 100 note("") note(f"- Top 50 city strings cover {top50_share:.1f}% of transactions") save(city_dist.head(200), "audit_city_top200") # ------------------------------------------------------------------ # # 7. indéterminé investigation (preliminary) # ------------------------------------------------------------------ # note("") note("## 7. `indéterminé` property type — preliminary look") note("") ind = df.filter(pl.col("propertyType") == "indéterminé") note(f"- Count: {ind.height:,} ({ind.height / df.height * 100:.1f}%)") if ind.height: note(f"- Median amount: ${ind['amount'].median():,.0f} " f"(vs unifamilial ${df.filter(pl.col('propertyType') == 'unifamilial')['amount'].median():,.0f})") note(f"- floorArea missing: {ind['floorArea'].is_null().mean() * 100:.1f}% | " f"yearBuilt missing: {ind['yearBuilt'].is_null().mean() * 100:.1f}% | " f"buildingType missing: {ind['buildingType'].is_null().mean() * 100:.1f}%") bt = ind.group_by("buildingType").len().sort("len", descending=True).head(8) note("- Top buildingType values within indéterminé: " + ", ".join(f"{r['buildingType']}={r['len']:,}" for r in bt.iter_rows(named=True))) ot = ind.group_by("ownerType").len().sort("len", descending=True).head(5) note("- ownerType within indéterminé: " + ", ".join(f"{r['ownerType']}={r['len']:,}" for r in ot.iter_rows(named=True))) # ------------------------------------------------------------------ # # Write report # ------------------------------------------------------------------ # report_path = REPORTS_DIR / "audit_report.md" header = ( "\n\n" ) report_path.write_text(header + "\n".join(REPORT_LINES) + "\n", encoding="utf-8") print(f"\nReport written to {report_path}") if __name__ == "__main__": main()