# Author: Simon-Pierre Boucher — simon-pierre.boucher@uqo.ca """Repeat-sales deflator and resale-implied values (paper §sec:fundamental; docs/fundamental_value.md §2). Outputs (results/fundamental/): price_index_monthly.parquet municipality × class (and region × class) chained monthly log-price index resale_targets.parquet one row per analytic sale with a later resale of the same parcel (both legs from the full snapshot): resale price/date, months held, index change, implied value prior_targets.parquet backward direction (earlier sale brought forward to the sale date) resale_sample.json / .csv description of the resale sample (counts, holding periods, gaps) by class/region/split paper/tables/tab_resale_sample.tex LaTeX fragment of the description figures/fundamental_resale_gap.pdf distribution of (resale-implied log value − log price) by holding period python3 scripts/50_repeat_sales.py [--min-n 5] [--window 3] """ 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, RAW, RESULTS # noqa: E402 from agentic_valuation.evaluation.tables import escape, num, tabular # noqa: E402 from agentic_valuation.fundamental.repeat_sales import (describe_resale_sample, monthly_index, prior_targets, # noqa: E402 resale_targets) OUT = RESULTS / "fundamental" 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/50_repeat_sales.py — do not edit\n" + tex.rstrip() + "\n", encoding="utf-8") def sample_table(desc: dict) -> tuple[str, pd.DataFrame]: cols = ["n", "share_short_hold", "months_held_median", "annualised_log_return_median", "gap_mean", "gap_sd", "gap_abs_median"] header = ["Sample", "{N}", "{Hold $<$12 m}", "{Months}", "{Ann. return}", "{Mean gap}", "{SD gap}", "{Med. $|$gap$|$}"] rows, recs, mids = [], [], [] def add(label: str, b: dict) -> None: rows.append([escape(label)] + [num(b[c], 0 if c == "n" else (1 if c == "months_held_median" else 3)) for c in cols]) recs.append({"sample": label, **{c: b[c] for c in cols}}) add("All pairs", desc["overall"]) mids.append(len(rows) - 1) for k in ("condo", "house", "plex"): if k in desc["by_class"]: add(k.capitalize(), desc["by_class"][k]) mids.append(len(rows) - 1) for k in ("Montréal CMA", "Québec CMA", "Outaouais", "other"): if k in desc["by_region"]: add(k, desc["by_region"][k]) mids.append(len(rows) - 1) for k in ("<12", "12-24", "24-36", "36+"): if k in desc["by_hold_bucket"]: add(f"Held {k} months", desc["by_hold_bucket"][k]) mids.append(len(rows) - 1) for k in ("train", "validation", "test", "holdout"): if k in desc["by_split"]: add(f"First sale in {k}", desc["by_split"][k]) return tabular(header, rows, "@{}l" + "r" * len(cols) + "@{}", midrules_after=mids), pd.DataFrame(recs) def gap_figure(t: pd.DataFrame, path: Path) -> None: buckets = ["<12", "12-24", "24-36", "36+"] hb = pd.cut(t["months_held"], [0, 12, 24, 36, 72], labels=buckets, right=False) fig, axes = plt.subplots(1, 3, figsize=(7.2, 2.4), sharey=True) for ax, cls, col in zip(axes, ["condo", "house", "plex"], PALETTE): data = [t.loc[(t["property_class"] == cls) & (hb == b), "gap"].dropna().to_numpy() for b in buckets] bp = ax.boxplot(data, whis=(5, 95), showfliers=False, widths=0.55, patch_artist=True, medianprops={"color": INK, "linewidth": 1.1}, whiskerprops={"color": GREY}, capprops={"color": GREY}) for patch in bp["boxes"]: patch.set(facecolor=col, alpha=0.55, edgecolor=col) ax.axhline(0, color=GREY, linewidth=0.7, linestyle="--") ax.set_xticks(range(1, len(buckets) + 1), buckets) ax.set_title(f"{cls} (n = {sum(len(d) for d in data):,})", fontsize=8.5, color=INK, loc="left") ax.set_xlabel("months between the two sales") axes[0].set_ylabel("resale-implied log value − log price") axes[0].set_ylim(-1.0, 1.0) 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("--full", default=str(RAW / "wp22_clean_transactions.parquet"), help="full snapshot (both legs of a pair); '' → analytic only") ap.add_argument("--splits", default=str(PROCESSED / "splits.parquet")) ap.add_argument("--min-n", type=int, default=5) ap.add_argument("--window", type=int, default=3) 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) full = pl.read_parquet(args.full) if args.full and Path(args.full).exists() else None print(f"analytic rows: {len(analytic):,}; full snapshot rows: {len(full) if full is not None else 'n/a (analytic only)'}") index = monthly_index(analytic, min_n=args.min_n, window=args.window) index.to_parquet(out / "price_index_monthly.parquet", index=False) mun_rows = index[index["level"] == "mun"] src = mun_rows["source"].value_counts().to_dict() print(f"index: {mun_rows[['mun_code', 'property_class']].drop_duplicates().shape[0]:,} municipality × class series, " f"{index['month_id'].nunique()} months; monthly changes from municipality {src.get('mun', 0):,}, region {src.get('region', 0):,}, carried {src.get('carry', 0):,} ({time.time() - t0:.1f} s)") targets = resale_targets(analytic, full, index, min_n=args.min_n) prior = prior_targets(analytic, full, index, min_n=args.min_n) sp = Path(args.splits) if sp.exists(): s = pl.read_parquet(sp).select(["tx_id", "split"]).to_pandas() targets = targets.merge(s, on="tx_id", how="left") prior = prior.merge(s, on="tx_id", how="left") targets.to_parquet(out / "resale_targets.parquet", index=False) prior.to_parquet(out / "prior_targets.parquet", index=False) print(f"resale targets: {len(targets):,} analytic sales with a later resale ({len(targets) / len(analytic):.1%}); " f"short holds < 12 m: {targets['short_hold'].mean():.1%}; later leg flagged: {targets['resale_flagged'].mean():.1%}; " f"prior targets: {len(prior):,} ({time.time() - t0:.1f} s)") desc = describe_resale_sample(targets, analytic) desc["parameters"] = {"min_n": args.min_n, "window": args.window, "full_snapshot": bool(full is not None)} desc["index_sources_mun_level"] = {k: int(v) for k, v in src.items()} desc["generated_at"] = datetime.now(timezone.utc).isoformat(timespec="seconds") desc["seconds"] = round(time.time() - t0, 1) (out / "resale_sample.json").write_text(json.dumps(desc, indent=2, ensure_ascii=False), encoding="utf-8") tex, csv = sample_table(desc) csv.to_csv(out / "resale_sample.csv", index=False) write_fragment(tex, PAPER / "tables" / "tab_resale_sample.tex") FIGURES.mkdir(parents=True, exist_ok=True) gap_figure(targets, FIGURES / "fundamental_resale_gap.pdf") ov = desc["overall"] print(f"gap (implied − log price): mean {ov['gap_mean']:+.3f}, sd {ov['gap_sd']:.3f}, median |gap| {ov['gap_abs_median']:.3f}; " f"median holding {ov['months_held_median']:.1f} months; done in {time.time() - t0:.1f} s") return 0 if __name__ == "__main__": sys.exit(main())