# Author: Simon-Pierre Boucher — simon-pierre.boucher@uqo.ca """Rent-based proxies: lease matching, median rents, crude gross yields and the user-cost diagnostic (paper §sec:fundamental; docs/fundamental_value.md §3–§4). Outputs (results/fundamental/): leases_matched.parquet cleaned leases with municipality / DA attribution leases_matching.json cleaning funnel + matching counts rents_by_mun_year.csv median rent by municipality × rooms bucket × year gross_yields.parquet one row per condo / plex sale matched to a municipality-year rent cell price_rent.csv region group × class × year: yield distribution + user cost + implied value / price usercost_parameters.csv parameter table (mortgage rate from the Bank of Canada, assumptions labelled) paper/tables/tab_usercost_params.tex figures/fundamental_price_rent.pdf, figures/fundamental_usercost.pdf python3 scripts/52_rents_usercost.py [--no-download] [--condo-bucket all] [--plex-bucket all] """ from __future__ import annotations import argparse import json import sys import time from datetime import datetime, timezone from pathlib import Path import matplotlib matplotlib.use("Agg") import matplotlib.pyplot as plt # noqa: E402 import numpy as np # noqa: E402 import pandas as pd # noqa: E402 import polars as pl # noqa: E402 from agentic_valuation.config import FIGURES, PAPER, PROCESSED, RESULTS # noqa: E402 from agentic_valuation.evaluation.tables import escape, num, tabular # noqa: E402 from agentic_valuation.fundamental.rents import (BOC_CACHE, BOC_SERIES, QUARTIER_DB, RDL_DB, USER_COST_PARAMS, # noqa: E402 annual_mortgage_rate, fetch_boc_mortgage_rate, gross_yields, implied_value_ratio, load_leases, match_leases, median_rents, user_cost) OUT = RESULTS / "fundamental" REGIONS = ["Montréal CMA", "Québec CMA", "Outaouais", "other"] PALETTE = ["#1f5fa8", "#c8541c", "#0b7a5a", "#8b4e8d"] GREY, INK = "#7a7a7a", "#222222" plt.rcParams.update({"font.size": 8.5, "axes.spines.top": False, "axes.spines.right": False, "axes.edgecolor": GREY, "axes.grid": True, "grid.color": "#e3e3e3", "grid.linewidth": 0.5, "axes.axisbelow": True, "pdf.fonttype": 42, "legend.frameon": False, "text.color": INK, "font.family": "sans-serif"}) def write_fragment(tex: str, path: Path) -> None: path.parent.mkdir(parents=True, exist_ok=True) path.write_text("% Author: Simon-Pierre Boucher — simon-pierre.boucher@uqo.ca\n% generated by scripts/52_rents_usercost.py — do not edit\n" + tex.rstrip() + "\n", encoding="utf-8") def fig_price_rent(gy: pd.DataFrame, path: Path) -> None: years = sorted(gy["year"].unique()) fig, axes = plt.subplots(1, 2, figsize=(7.2, 2.6), sharey=True) for ax, cls in zip(axes, ["condo", "plex"]): sub = gy[gy["property_class"] == cls] width = 0.18 for j, reg in enumerate(REGIONS): data = [sub.loc[(sub["region_group"] == reg) & (sub["year"] == y), "gross_yield"].to_numpy() for y in years] pos = [i + (j - 1.5) * width for i in range(len(years))] keep = [k for k, d in enumerate(data) if len(d) >= 30] if not keep: continue bp = ax.boxplot([data[k] for k in keep], positions=[pos[k] for k in keep], widths=width * 0.9, whis=(5, 95), showfliers=False, patch_artist=True, medianprops={"color": INK, "linewidth": 1.0}, whiskerprops={"color": GREY, "linewidth": 0.6}, capprops={"color": GREY, "linewidth": 0.6}) for patch in bp["boxes"]: patch.set(facecolor=PALETTE[j], alpha=0.6, edgecolor=PALETTE[j]) ax.set_xticks(range(len(years)), [str(y) for y in years]) ax.set_title(f"{cls} (n = {len(sub):,})", fontsize=8.5, loc="left", color=INK) ax.set_xlabel("year of sale") axes[0].set_ylabel("gross rental yield (annual rent / price)") axes[0].set_ylim(0, 0.16) handles = [plt.Rectangle((0, 0), 1, 1, facecolor=PALETTE[j], alpha=0.6) for j in range(len(REGIONS))] axes[1].legend(handles, REGIONS, loc="upper right", ncol=2, fontsize=7) fig.tight_layout() fig.savefig(path, bbox_inches="tight", pad_inches=0.02) plt.close(fig) def fig_usercost(pr: pd.DataFrame, g_scen: list[float], path: Path) -> None: fig, axes = plt.subplots(1, 2, figsize=(7.2, 2.6), sharey=True) styles = ["-", "--", ":"] for ax, cls in zip(axes, ["condo", "plex"]): sub = pr[(pr["property_class"] == cls) & (pr["region_group"] == "ALL")].sort_values("year") for k, g in enumerate(g_scen): col = f"implied_ratio_median_g{int(round(g * 100))}" ax.plot(sub["year"], sub[col], linestyle=styles[k], color=PALETTE[k], marker="o", markersize=3, label=f"$\\pi^e$ = {g:.0%}") lo, hi = sub[f"implied_ratio_p25_g{int(round(g_scen[1] * 100))}"], sub[f"implied_ratio_p75_g{int(round(g_scen[1] * 100))}"] ax.fill_between(sub["year"], lo, hi, color=PALETTE[1], alpha=0.12, linewidth=0, label=f"IQR, $\\pi^e$ = {g_scen[1]:.0%}") ax.axhline(1.0, color=GREY, linewidth=0.7, linestyle="--") ax.set_title(f"{cls} — illustrative", fontsize=8.5, loc="left", color=INK) ax.set_xlabel("year of sale") ax.set_xticks(sub["year"].to_numpy()) axes[0].set_ylabel("rent-implied value / price (= gross yield / $u$)") axes[0].legend(loc="upper left", fontsize=7) fig.tight_layout() fig.savefig(path, bbox_inches="tight", pad_inches=0.02) plt.close(fig) def main(argv: list[str] | None = None) -> int: ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) ap.add_argument("--analytic", default=str(PROCESSED / "analytic.parquet")) ap.add_argument("--rdl", default=str(RDL_DB)) ap.add_argument("--quartier", default=str(QUARTIER_DB)) ap.add_argument("--no-download", action="store_true", help="do not call the Bank of Canada API (use the cache or NaN)") ap.add_argument("--condo-bucket", default="all", help="rooms bucket of the leases matched to condo sales (0-1, 2, 3, 4+, all)") ap.add_argument("--plex-bucket", default="all") ap.add_argument("--min-leases", type=int, default=10) ap.add_argument("--out", default=str(OUT)) args = ap.parse_args(argv) out = Path(args.out) out.mkdir(parents=True, exist_ok=True) t0 = time.time() analytic = pl.read_parquet(args.analytic) leases = load_leases(Path(args.rdl)) funnel = dict(leases.attrs["funnel"]) print(f"leases: {funnel}") matched = match_leases(leases, analytic, quartier_db=Path(args.quartier) if args.quartier else None) n_match = int(matched["matched"].sum()) n_da = int(matched["dauid"].notna().sum()) print(f"matched to a municipality (≤ 2 km from a sale): {n_match:,} / {len(matched):,}; in a DA polygon: {n_da:,} ({time.time() - t0:.1f} s)") matched.to_parquet(out / "leases_matched.parquet", index=False) rents = median_rents(matched[matched["matched"]]) rents.to_csv(out / "rents_by_mun_year.csv", index=False) gy = gross_yields(analytic, rents, condo_bucket=args.condo_bucket, plex_bucket=args.plex_bucket, min_leases=args.min_leases) gy.to_parquet(out / "gross_yields.parquet", index=False) print(f"gross yields: {len(gy):,} condo/plex sales matched to a municipality-year rent cell with ≥ {args.min_leases} leases " f"(condo {int((gy['property_class'] == 'condo').sum()):,}, plex {int((gy['property_class'] == 'plex').sum()):,}); " f"median yield condo {gy.loc[gy['property_class'] == 'condo', 'gross_yield'].median():.3f}, plex {gy.loc[gy['property_class'] == 'plex', 'gross_yield'].median():.3f}") # mortgage rate (Bank of Canada Valet, verified series id) — NaN when unavailable, never invented weekly = fetch_boc_mortgage_rate(BOC_CACHE, download=not args.no_download) years = sorted(gy["year"].unique().tolist()) i_year = annual_mortgage_rate(weekly, years) rate_src = f"Bank of Canada Valet {BOC_SERIES} ({weekly.attrs.get('label', '')}), {len(weekly)} weekly obs {weekly['date'].min().date() if len(weekly) else ''}–{weekly['date'].max().date() if len(weekly) else ''}" if len(weekly) else "UNAVAILABLE (download failed, no cache) — rate left NaN" print(f"mortgage rate: {rate_src}; annual means {i_year.round(4).to_dict()}") tau_p, delta, gamma = (USER_COST_PARAMS[k]["value"] for k in ("tau_p", "delta", "gamma")) g_scen = list(USER_COST_PARAMS["g_scenarios"]["value"]) gy = gy.merge(i_year.rename("i").reset_index().rename(columns={"index": "year"}), on="year", how="left") for g in g_scen: u = user_cost(gy["i"], tau_p, delta, gamma, g) gy[f"u_g{int(round(g * 100))}"] = u gy[f"implied_ratio_g{int(round(g * 100))}"] = implied_value_ratio(gy["gross_yield"], u) rows = [] for (reg, cls, y), gsub in list(gy.groupby(["region_group", "property_class", "year"])) + [(("ALL", c, y), s) for (c, y), s in gy.groupby(["property_class", "year"])]: r = {"region_group": reg, "property_class": cls, "year": int(y), "n_sales": int(len(gsub)), "n_mun": int(gsub["mun_code"].nunique()), "price_median": float(gsub["price"].median()), "rent_median": float(gsub["rent_median"].median()), "gross_yield_p25": float(gsub["gross_yield"].quantile(0.25)), "gross_yield_median": float(gsub["gross_yield"].median()), "gross_yield_p75": float(gsub["gross_yield"].quantile(0.75)), "mortgage_rate_5y": float(i_year.get(int(y), np.nan))} for g in g_scen: tag = f"g{int(round(g * 100))}" r[f"user_cost_{tag}"] = float(gsub[f"u_{tag}"].iloc[0]) if len(gsub) else np.nan r[f"implied_ratio_p25_{tag}"] = float(gsub[f"implied_ratio_{tag}"].quantile(0.25)) r[f"implied_ratio_median_{tag}"] = float(gsub[f"implied_ratio_{tag}"].median()) r[f"implied_ratio_p75_{tag}"] = float(gsub[f"implied_ratio_{tag}"].quantile(0.75)) rows.append(r) pr = pd.DataFrame(rows).sort_values(["property_class", "region_group", "year"]).reset_index(drop=True) pr.to_csv(out / "price_rent.csv", index=False) # parameter table prow = [{"parameter": "i (5-year conventional mortgage rate)", "value": "; ".join(f"{y}: {v:.2%}" if np.isfinite(v) else f"{y}: NaN" for y, v in i_year.items()), "status": "measured", "source": rate_src}] for k in ("tau_p", "delta", "gamma"): prow.append({"parameter": USER_COST_PARAMS[k]["label"], "value": f"{USER_COST_PARAMS[k]['value']:.1%}", "status": "assumption", "source": USER_COST_PARAMS[k]["source"]}) prow.append({"parameter": USER_COST_PARAMS["g_scenarios"]["label"], "value": ", ".join(f"{g:.0%}" for g in g_scen), "status": "scenario", "source": USER_COST_PARAMS["g_scenarios"]["source"]}) params = pd.DataFrame(prow) params.to_csv(out / "usercost_parameters.csv", index=False) header = ["Parameter", "Value", "Status"] tex_names = {USER_COST_PARAMS["g_scenarios"]["label"]: r"expected nominal appreciation $\pi^{e}$ (three scenarios)", prow[0]["parameter"]: r"$i$: 5-year conventional mortgage rate (annual mean of weekly observations)", USER_COST_PARAMS["tau_p"]["label"]: r"$\tau_p$: property tax rate (municipal and school, share of market value)", USER_COST_PARAMS["delta"]["label"]: r"$\delta$: maintenance and depreciation", USER_COST_PARAMS["gamma"]["label"]: r"$\gamma$: risk and liquidity premium"} trows = [[tex_names.get(r["parameter"], escape(r["parameter"])), escape(r["value"]), escape(r["status"])] for r in prow] write_fragment(tabular(header, trows, "p{5.6cm}p{6.2cm}l"), PAPER / "tables" / "tab_usercost_params.tex") FIGURES.mkdir(parents=True, exist_ok=True) fig_price_rent(gy, FIGURES / "fundamental_price_rent.pdf") fig_usercost(pr, g_scen, FIGURES / "fundamental_usercost.pdf") summary = {"funnel": funnel, "n_matched_municipality": n_match, "n_in_da_polygon": n_da, "n_unmatched": int(len(matched) - n_match), "rooms_bucket_counts": matched.loc[matched["matched"], "rooms_bucket"].value_counts().to_dict(), "start_date_present_share": float(matched.loc[matched["matched"], "start_date_present"].mean()), "region_counts": matched.loc[matched["matched"], "region_group"].value_counts().to_dict(), "year_counts": {int(k): int(v) for k, v in matched.loc[matched["matched"], "year"].value_counts().sort_index().items()}, "n_rent_cells": int(len(rents)), "n_gross_yield_sales": int(len(gy)), "condo_bucket": args.condo_bucket, "plex_bucket": args.plex_bucket, "min_leases": args.min_leases, "mortgage_rate_source": rate_src, "mortgage_rate_annual": {int(k): (None if not np.isfinite(v) else float(v)) for k, v in i_year.items()}, "user_cost_params": {k: v["value"] for k, v in USER_COST_PARAMS.items()}, "generated_at": datetime.now(timezone.utc).isoformat(timespec="seconds"), "seconds": round(time.time() - t0, 1)} (out / "leases_matching.json").write_text(json.dumps(summary, indent=2, ensure_ascii=False, default=str), encoding="utf-8") print(f"done in {time.time() - t0:.1f} s") return 0 if __name__ == "__main__": sys.exit(main())