#!/usr/bin/env python3 # ============================================================================= # QWHPI — Quebec Weekly Housing Price Index # Author : Simon-Pierre Boucher # Contact : contact@spboucher.ai # File : engine/scripts/08_figures.py # Purpose : Publication figures — aggregate, regions, cities, median-vs- # hedonic, repeat sales, downsampling, volumes, YoY map, gap. # ============================================================================= """Research figures (Execution Order step 10).""" from __future__ import annotations import sys from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src")) import numpy as np import polars as pl from qwhpi import plotting as P from qwhpi.config import FIGURES_DIR, PROCESSED_DIR, TABLES_DIR, ensure_dirs import matplotlib.dates as mdates # noqa: E402 import matplotlib.pyplot as plt # noqa: E402 def dts(weeks: pl.Series) -> np.ndarray: return weeks.str.to_date("%Y-%m-%d").to_numpy() def cell(c: pl.DataFrame, gid: str, t: str) -> pl.DataFrame: return c.filter((pl.col("geography_id") == gid) & (pl.col("property_type") == t)).sort("week") def main() -> None: ensure_dirs() c = pl.read_parquet(PROCESSED_DIR / "qwhpi_weekly.parquet") # 1 — Quebec aggregate with CI band q = cell(c, "quebec", "all") fig, ax = plt.subplots(figsize=(9, 4.5)) x = dts(q["week"]) ax.fill_between(x, q["lower_95"], q["upper_95"], color=P.BAND, label="95% CI") ax.plot(x, q["index_smoothed"], color=P.SERIES[0], label="QWHPI-QC (smoothed)") ax.plot(x, q["index"], color=P.SERIES[0], alpha=0.35, lw=1, label="raw weekly") ax.set_title("Quebec weekly housing price index — all types (2021 = 100)") ax.legend(loc="upper left") ax.xaxis.set_major_formatter(mdates.DateFormatter("%Y")) P.save(fig, FIGURES_DIR / "quebec_aggregate.png") # 2 — 17 regions, small multiples regs = sorted(c.filter(pl.col("geography_level") == "region")["geography_id"].unique()) fig, axes = plt.subplots(3, 6, figsize=(15, 7), sharex=True, sharey=True) for ax, rid in zip(axes.flat, regs): r = cell(c, rid, "all") ax.plot(dts(r["week"]), r["index_smoothed"], color=P.SERIES[0], lw=1.4) ax.set_title(r["geography_name"][0], fontsize=8) ax.xaxis.set_major_formatter(mdates.DateFormatter("%y")) for ax in axes.flat[len(regs):]: ax.axis("off") fig.suptitle("Regional indexes — all types (2021 = 100)") P.save(fig, FIGURES_DIR / "region_comparison.png") # 3 — Big-4 cities fig, ax = plt.subplots(figsize=(9, 4.5)) for i, (gid, label) in enumerate([("montreal", "Montréal"), ("quebec-city", "Québec"), ("laval", "Laval"), ("gatineau", "Gatineau")]): s = cell(c, gid, "all") ax.plot(dts(s["week"]), s["index_smoothed"], color=P.SERIES[i], label=label) ax.set_title("Major cities — all types (2021 = 100)") ax.legend(loc="upper left") P.save(fig, FIGURES_DIR / "big4_cities.png") # 4 — Condo across cities fig, ax = plt.subplots(figsize=(9, 4.5)) for i, (gid, label) in enumerate([("montreal", "Montréal"), ("quebec-city", "Québec"), ("gatineau", "Gatineau")]): s = cell(c, gid, "condo") ax.plot(dts(s["week"]), s["index_smoothed"], color=P.SERIES[i], label=label) ax.set_title("Condo indexes by city (2021 = 100)") ax.legend(loc="upper left") P.save(fig, FIGURES_DIR / "condo_by_city.png") # 5 — Raw median vs hedonic (province, unifamilial) bi = pl.read_parquet("data/interim/baseline_indexes.parquet") b = bi.filter((pl.col("geography_id") == "quebec") & (pl.col("property_type") == "unifamilial")).sort("week") med = b["raw_median"].to_numpy() base_med = np.nanmean(med[:52]) fig, ax = plt.subplots(figsize=(9, 4.5)) ax.plot(dts(b["week"]), med / base_med * 100, color=P.SERIES[1], lw=1.2, label="Raw weekly median (indexed)") h = cell(c, "quebec", "unifamilial") ax.plot(dts(h["week"]), h["index_smoothed"], color=P.SERIES[0], label="QWHPI (quality-adjusted)") ax.set_title("Composition noise: raw median vs hedonic index — Quebec unifamilial") ax.legend(loc="upper left") P.save(fig, FIGURES_DIR / "raw_median_vs_hedonic.png") # 6 — Weekly volumes vol = pl.read_parquet(PROCESSED_DIR / "weekly_liquidity.parquet") v = vol.filter((pl.col("geography_id") == "quebec") & (pl.col("property_type") == "all")).sort("week") fig, ax = plt.subplots(figsize=(9, 3.2)) ax.bar(dts(v["week"]), v["transactions"], width=6, color=P.SERIES[0], alpha=0.6) ax.set_title("Weekly transaction volume — Quebec (research sample)") P.save(fig, FIGURES_DIR / "volumes.png") # 7 — Downsampling stability ds = pl.read_csv(TABLES_DIR / "downsampling_results.csv") fig, ax = plt.subplots(figsize=(7, 4)) ax.plot(ds["target_tx_per_week"], ds["rmse_log_pct"], "o-", color=P.SERIES[0], label="RMSE vs full-sample path (%)") ax.plot(ds["target_tx_per_week"], (1 - ds["ci95_coverage"]) * 10, "s--", color=P.SERIES[1], label="(1 − CI coverage) × 10") ax.set_xlabel("target transactions per week") ax.set_title("Downsampling experiment — Montréal condo") ax.legend() ax.invert_xaxis() P.save(fig, FIGURES_DIR / "downsampling_stability.png") # 8 — YoY map (choropleth) import geopandas as gpd latest = c.filter((pl.col("geography_level") == "region") & (pl.col("property_type") == "all") & (~pl.col("is_partial_week"))) lw = latest.filter(pl.col("week") == latest["week"].max()) g = gpd.read_file("web/public/regions.geojson") g = g.merge(lw.select("geography_id", "yoy_pct").to_pandas(), on="geography_id") fig, ax = plt.subplots(figsize=(8, 7)) g.plot(column="yoy_pct", cmap="Blues", legend=True, ax=ax, edgecolor="#fcfcfb", linewidth=0.8, legend_kwds={"label": "YoY %", "shrink": 0.6}) ax.set_axis_off() ax.set_title(f"Year-over-year appreciation by region — week of {lw['week'][0]}") P.save(fig, FIGURES_DIR / "yoy_map.png") # 9 — Assessment gap evolution gap = pl.read_parquet(PROCESSED_DIR / "assessment_gap.parquet") fig, ax = plt.subplots(figsize=(9, 4)) for i, t in enumerate(["unifamilial", "condo", "plex"]): s = gap.filter((pl.col("geography_id") == "quebec") & (pl.col("property_type") == t)).sort("week") # 13-week rolling median for readability vals = s["median_ratio"].rolling_median(window_size=13) ax.plot(dts(s["week"]), vals, color=P.SERIES[i], label=t) ax.axhline(1.0, color=P.TEXT2, lw=1, ls="--") ax.set_title("Assessment gap — median sale price / municipal assessment (13-wk median)") ax.legend(loc="upper left") P.save(fig, FIGURES_DIR / "assessment_gap.png") # 10 — Reliability heat table (cells × grade) summary = ( c.group_by(["geography_id", "property_type"]) .agg(pl.col("reliability_grade").mode().first().alias("grade")) ) counts = summary.group_by(["property_type", "grade"]).len().sort(["property_type", "grade"]) fig, ax = plt.subplots(figsize=(7, 3.4)) grades = ["A", "B", "C", "D", "E"] types = ["all", "unifamilial", "condo", "plex"] mat = np.zeros((len(types), len(grades))) for r in counts.iter_rows(named=True): mat[types.index(r["property_type"]), grades.index(r["grade"])] = r["len"] im = ax.imshow(mat, cmap="Blues", aspect="auto") ax.set_xticks(range(len(grades)), grades) ax.set_yticks(range(len(types)), types) for i in range(len(types)): for j in range(len(grades)): if mat[i, j]: ax.text(j, i, int(mat[i, j]), ha="center", va="center", color=P.TEXT if mat[i, j] < mat.max() * 0.6 else "white") ax.set_title("Published series by reliability grade") ax.grid(False) P.save(fig, FIGURES_DIR / "reliability_matrix.png") print("All figures written to outputs/figures/") if __name__ == "__main__": main()