1# Author: Simon-Pierre Boucher — simon-pierre.boucher@uqo.ca2"""Repeat sales: a transparent monthly deflator and the resale-implied value of a sale.34Definitions (paper §sec:fundamental, notation of theory.tex §subsec:values)5---------------------------------------------------------------------------6* Monthly index. For municipality m, property class c and calendar month τ, let q_{mcτ} be the median of7 log(price / floor area) over the sales of that cell. Cells with fewer than `min_n` sales (default 5) are replaced by8 the region-group × class cell. The series is smoothed with a centred 3-month rolling median and *chained*: the9 log index is the cumulative sum of month-to-month changes of the smoothed series, each change taken from the10 municipality series when both months are available there and from the region series otherwise. Chaining means that11 a switch between the municipality and the region source never introduces a level jump. The index is an ex-post12 deflator (it uses all sales of the snapshot) and is never shown to any arm.13* Resale-implied value. For a sale (i, t1) at price P1 whose parcel resells at (t2 > t1) for P2,14 log V^R_{t1} = log P2 − [ I(m, c, τ2) − I(m, c, τ1) ],15 i.e. the later price brought back to the first sale date by the index. Under the repeat-sales assumptions of16 Bailey–Muth–Nourse (1963) and Case–Shiller (1989) — unchanged quality, market movement common to the cell — the17 difference log V^R − log P1 is the difference of the two transaction noises, ε_{t2} − ε_{t1}, so an estimate at t118 that is closer to V^R than P1 was captured a component of value that the first transaction's noise obscured.19* Annualised holding return: (log P2 − log P1) × 12 / months_held.2021Selection (documented, never corrected here): parcels that resell within the snapshot window (2021–2026) are not a22random sample — flips, forced sales, renovations and loss-averse listing behaviour (Genesove & Mayer 2001) correlate23the resale decision with ε_{t1}. Holding periods shorter than 12 months are flagged (`short_hold`); renovations are24unobserved because the roll attributes are a single snapshot for both sales.25"""26from __future__ import annotations2728from typing import Any2930import numpy as np31import pandas as pd3233from ..evaluation._common import to_pandas3435MIN_CELL_N = 536SMOOTH_WINDOW = 337SHORT_HOLD_MONTHS = 1238DAYS_PER_MONTH = 365.25 / 1239# rows of the raw snapshot that cannot serve as either leg of a repeat-sales pair (data-quality exclusions of WP 22)40STRUCTURAL_EXCLUSIONS = {"exact_duplicate", "same_parcel_same_date_diff_price", "shared_parcel", "multi_parcel_candidate",41 "land", "non_residential", "built_after_sale"}42INDEX_KEY = ["mun_code", "property_class"]434445# ------------------------------------------------------------------------------------------------ monthly index4647def _month_id(date: pd.Series) -> pd.Series:48 d = pd.to_datetime(date)49 return (d.dt.year * 12 + d.dt.month - 1).astype(int)505152def month_id_to_timestamp(m: np.ndarray | pd.Series) -> pd.Series:53 m = pd.Series(np.asarray(m, dtype=int))54 return pd.to_datetime({"year": m // 12, "month": m % 12 + 1, "day": 1})555657def _centred_rolling_median(wide: np.ndarray, window: int) -> np.ndarray:58 """Centred rolling median along axis 1 ignoring NaN (min 1 valid value); NaN where no neighbour is valid."""59 half = window // 260 n_rows, n_cols = wide.shape61 stack = np.full((window, n_rows, n_cols), np.nan)62 for k, shift in enumerate(range(-half, half + 1)):63 if shift < 0:64 stack[k, :, -shift:] = wide[:, :shift]65 elif shift > 0:66 stack[k, :, :-shift] = wide[:, shift:]67 else:68 stack[k] = wide69 with np.errstate(all="ignore"):70 return np.nanmedian(stack, axis=0)717273def monthly_index(analytic: Any, *, min_n: int = MIN_CELL_N, window: int = SMOOTH_WINDOW) -> pd.DataFrame:74 """Municipality × class monthly chained log-price index (long format).7576 Columns: mun_code, property_class, region_group, month_id, month (Timestamp), n_mun, n_region, raw_mun, raw_region,77 smooth_mun, smooth_region, delta, source ("mun" | "region" | "carry"), log_index (0 at the first month).78 Every (municipality, class) that has at least one sale gets a complete monthly series from the first to the last79 month of the table.80 """81 df = to_pandas(analytic)82 need = INDEX_KEY + ["region_group", "date", "price", "floor_area_m2"]83 missing = [c for c in need if c not in df.columns]84 if missing:85 raise ValueError(f"analytic table is missing {missing}")86 df = df[need].dropna(subset=["floor_area_m2", "price"])87 df = df[(df["floor_area_m2"] > 0) & (df["price"] > 0)].copy()88 if df.empty:89 raise ValueError("no sale with a positive floor area and price")90 df["lp"] = np.log(df["price"].astype(float) / df["floor_area_m2"].astype(float))91 df["month_id"] = _month_id(df["date"])92 months = np.arange(int(df["month_id"].min()), int(df["month_id"].max()) + 1)93 n_months = len(months)9495 # municipality → region group (modal, in case the snapshot has a few inconsistent rows)96 mun_region = df.groupby("mun_code")["region_group"].agg(lambda s: s.value_counts().index[0])9798 def _wide(frame: pd.DataFrame, keys: list[str]) -> tuple[pd.DataFrame, np.ndarray, np.ndarray]:99 g = frame.groupby(keys + ["month_id"])["lp"].agg(["median", "size"])100 med = g["median"].unstack("month_id").reindex(columns=months)101 n = g["size"].unstack("month_id").reindex(columns=months).fillna(0)102 return med, med.to_numpy(dtype=float), n.to_numpy(dtype=float)103104 mun_med, mun_raw, mun_n = _wide(df, INDEX_KEY)105 reg_med, reg_raw, reg_n = _wide(df, ["region_group", "property_class"])106 mun_raw = np.where(mun_n >= min_n, mun_raw, np.nan)107 reg_raw = np.where(reg_n >= min_n, reg_raw, np.nan)108 # the smoother uses the valid neighbours of a month but never *creates* a value for a month below `min_n`109 mun_s = np.where(np.isfinite(mun_raw), _centred_rolling_median(mun_raw, window), np.nan)110 reg_s = np.where(np.isfinite(reg_raw), _centred_rolling_median(reg_raw, window), np.nan)111112 # align each municipality row with its region row113 reg_pos = {k: i for i, k in enumerate(reg_med.index)}114 mun_keys = list(mun_med.index) # (mun_code, property_class)115 rows_reg = np.array([reg_pos.get((mun_region[m], c), -1) for m, c in mun_keys])116 reg_s_al = np.where(rows_reg[:, None] >= 0, reg_s[np.clip(rows_reg, 0, None)], np.nan)117 reg_raw_al = np.where(rows_reg[:, None] >= 0, reg_raw[np.clip(rows_reg, 0, None)], np.nan)118 reg_n_al = np.where(rows_reg[:, None] >= 0, reg_n[np.clip(rows_reg, 0, None)], 0.0)119120 d_mun = np.full_like(mun_s, np.nan)121 d_mun[:, 1:] = mun_s[:, 1:] - mun_s[:, :-1]122 d_reg = np.full_like(reg_s_al, np.nan)123 d_reg[:, 1:] = reg_s_al[:, 1:] - reg_s_al[:, :-1]124 delta = np.where(np.isfinite(d_mun), d_mun, np.where(np.isfinite(d_reg), d_reg, 0.0))125 delta[:, 0] = 0.0126 source = np.where(np.isfinite(d_mun), "mun", np.where(np.isfinite(d_reg), "region", "carry"))127 source[:, 0] = "base"128 log_index = np.cumsum(delta, axis=1)129130 n_rows = len(mun_keys)131 out = pd.DataFrame({132 "level": "mun",133 "mun_code": np.repeat([str(k[0]) for k in mun_keys], n_months),134 "property_class": np.repeat([k[1] for k in mun_keys], n_months),135 "region_group": np.repeat([mun_region[k[0]] for k in mun_keys], n_months),136 "month_id": np.tile(months, n_rows),137 "n_mun": mun_n.ravel(), "n_region": reg_n_al.ravel(),138 "raw_mun": mun_raw.ravel(), "raw_region": reg_raw_al.ravel(),139 "smooth_mun": mun_s.ravel(), "smooth_region": reg_s_al.ravel(),140 "delta": delta.ravel(), "source": source.ravel(), "log_index": log_index.ravel(),141 })142 # region × class series (level="region", keyed by the region group in `mun_code`): the fallback when a143 # municipality × class has no series at all (e.g. every sale of the cell lacks a floor area)144 d_reg_only = np.full_like(reg_s, np.nan)145 d_reg_only[:, 1:] = reg_s[:, 1:] - reg_s[:, :-1]146 d_reg_only = np.where(np.isfinite(d_reg_only), d_reg_only, 0.0)147 d_reg_only[:, 0] = 0.0148 reg_keys = list(reg_med.index)149 reg_out = pd.DataFrame({150 "level": "region",151 "mun_code": np.repeat([str(k[0]) for k in reg_keys], n_months),152 "property_class": np.repeat([k[1] for k in reg_keys], n_months),153 "region_group": np.repeat([k[0] for k in reg_keys], n_months),154 "month_id": np.tile(months, len(reg_keys)),155 "n_mun": np.nan, "n_region": reg_n.ravel(), "raw_mun": np.nan, "raw_region": reg_raw.ravel(),156 "smooth_mun": np.nan, "smooth_region": reg_s.ravel(), "delta": d_reg_only.ravel(),157 "source": np.concatenate([np.full((len(reg_keys), 1), "base"),158 np.where(np.isfinite(reg_s[:, 1:] - reg_s[:, :-1]), "region", "carry")], axis=1).ravel(),159 "log_index": np.cumsum(d_reg_only, axis=1).ravel(),160 })161 out = pd.concat([out, reg_out], ignore_index=True)162 out["month"] = month_id_to_timestamp(out["month_id"]).to_numpy()163 out.attrs["min_n"] = min_n164 out.attrs["window"] = window165 return out166167168def index_change(index: pd.DataFrame, mun_code: Any, property_class: Any, date_from: Any, date_to: Any,169 region_group: Any | None = None) -> np.ndarray:170 """Vectorised I(m, c, τ_to) − I(m, c, τ_from). When the (municipality, class) series does not exist and171 `region_group` is given, the region × class series is used; NaN otherwise (never a silent zero)."""172 key = pd.MultiIndex.from_arrays([index["level"].astype(str), index["mun_code"].astype(str), index["property_class"].astype(str),173 index["month_id"].astype(int)])174 lookup = pd.Series(index["log_index"].to_numpy(dtype=float), index=key)175 mun = pd.Series(mun_code).astype(str).to_numpy()176 cls = pd.Series(property_class).astype(str).to_numpy()177 m_from = _month_id(pd.Series(date_from)).to_numpy()178 m_to = _month_id(pd.Series(date_to)).to_numpy()179 lvl = np.full(len(mun), "mun")180 v_from = lookup.reindex(pd.MultiIndex.from_arrays([lvl, mun, cls, m_from])).to_numpy(dtype=float)181 v_to = lookup.reindex(pd.MultiIndex.from_arrays([lvl, mun, cls, m_to])).to_numpy(dtype=float)182 if region_group is not None:183 reg = pd.Series(region_group).astype(str).to_numpy()184 lvl_r = np.full(len(mun), "region")185 r_from = lookup.reindex(pd.MultiIndex.from_arrays([lvl_r, reg, cls, m_from])).to_numpy(dtype=float)186 r_to = lookup.reindex(pd.MultiIndex.from_arrays([lvl_r, reg, cls, m_to])).to_numpy(dtype=float)187 missing = ~(np.isfinite(v_from) & np.isfinite(v_to))188 v_from = np.where(missing, r_from, v_from)189 v_to = np.where(missing, r_to, v_to)190 return v_to - v_from191192193# ------------------------------------------------------------------------------------------------ pairs194195def resale_pairs(full: Any) -> pd.DataFrame:196 """Consecutive sales of the same parcel in the full snapshot (one row per (first sale, next sale) pair).197198 Rows with a structural data-quality exclusion (duplicates, shared or multi-parcel deeds, land, non-residential,199 built after the sale) and rows without parcel id or positive price cannot be either leg. Pairs on the same date200 are dropped. The later leg keeps its `exclude_reason` (e.g. assessment-ratio outlier) in `resale_exclude_reason`201 and its headline-sample membership in `resale_in_headline` so that users can tighten the sample.202 """203 df = to_pandas(full)204 need = ["tx_id", "parcel_id", "date", "price"]205 missing = [c for c in need if c not in df.columns]206 if missing:207 raise ValueError(f"full table is missing {missing}")208 df = df.dropna(subset=["parcel_id", "price"])209 df = df[df["price"] > 0]210 if "exclude_reason" in df.columns:211 df = df[~df["exclude_reason"].isin(STRUCTURAL_EXCLUSIONS)]212 df = df.assign(date=pd.to_datetime(df["date"])).sort_values(["parcel_id", "date", "tx_id"]).reset_index(drop=True)213 opt = {c: df[c] if c in df.columns else pd.Series(pd.NA, index=df.index) for c in ("property_class", "exclude_reason", "in_headline_sample")}214 nxt = df.groupby("parcel_id", sort=False).shift(-1)215 pairs = pd.DataFrame({216 "tx_id": df["tx_id"].astype(str),217 "parcel_id": df["parcel_id"].astype(str),218 "date": df["date"],219 "price": df["price"].astype(float),220 "property_class": opt["property_class"],221 "resale_tx_id": nxt["tx_id"],222 "resale_date": nxt["date"],223 "resale_price": nxt["price"],224 "resale_property_class": nxt["property_class"] if "property_class" in df.columns else pd.NA,225 "resale_exclude_reason": nxt["exclude_reason"] if "exclude_reason" in df.columns else pd.NA,226 "resale_in_headline": nxt["in_headline_sample"] if "in_headline_sample" in df.columns else pd.NA,227 })228 pairs = pairs.dropna(subset=["resale_tx_id"])229 pairs = pairs[pairs["resale_date"] > pairs["date"]].copy()230 pairs["resale_tx_id"] = pairs["resale_tx_id"].astype(str)231 pairs["resale_price"] = pairs["resale_price"].astype(float)232 return pairs.reset_index(drop=True)233234235def _months_between(d1: pd.Series, d2: pd.Series) -> np.ndarray:236 return (pd.to_datetime(d2) - pd.to_datetime(d1)).dt.days.to_numpy(dtype=float) / DAYS_PER_MONTH237238239def resale_targets(analytic: Any, full: Any | None = None, index: pd.DataFrame | None = None, *, min_n: int = MIN_CELL_N) -> pd.DataFrame:240 """Resale-implied value at the first sale date for every analytic sale with a later resale of the same parcel.241242 Returns one row per analytic `tx_id` that has a later sale: tx_id, resale_tx_id, resale_price, resale_date,243 months_held, index_change, resale_implied_log_value, resale_implied_value, log_price, gap (= implied − log price),244 annualised_log_return, short_hold, class_changed, resale_flagged, mun_code, property_class, region_group, date, price.245 `full` defaults to the analytic table itself (pairs then only use headline sales); `index` defaults to246 `monthly_index(analytic)`.247 """248 a = to_pandas(analytic)249 a["tx_id"] = a["tx_id"].astype(str)250 if index is None:251 index = monthly_index(a, min_n=min_n)252 pairs = resale_pairs(full if full is not None else a)253 keep = ["tx_id", "parcel_id", "date", "price", "mun_code", "property_class", "region_group"]254 keep = [c for c in keep if c in a.columns]255 t = pairs.drop(columns=["date", "price", "property_class", "parcel_id"]).merge(a[keep], on="tx_id", how="inner", validate="one_to_one")256 t["log_price"] = np.log(t["price"].astype(float))257 t["months_held"] = _months_between(t["date"], t["resale_date"])258 t["index_change"] = index_change(index, t["mun_code"], t["property_class"], t["date"], t["resale_date"], t.get("region_group"))259 t["resale_implied_log_value"] = np.log(t["resale_price"]) - t["index_change"]260 t["resale_implied_value"] = np.exp(t["resale_implied_log_value"])261 t["gap"] = t["resale_implied_log_value"] - t["log_price"]262 t["annualised_log_return"] = (np.log(t["resale_price"]) - t["log_price"]) * 12.0 / t["months_held"]263 t["short_hold"] = t["months_held"] < SHORT_HOLD_MONTHS264 t["class_changed"] = t["resale_property_class"].notna() & (t["resale_property_class"].astype("string") != t["property_class"].astype("string"))265 t["resale_flagged"] = t["resale_exclude_reason"].notna() | (t["resale_in_headline"].fillna(True).astype(bool) == False) # noqa: E712266 cols = ["tx_id", "parcel_id", "date", "price", "log_price", "mun_code", "property_class", "region_group",267 "resale_tx_id", "resale_date", "resale_price", "resale_property_class", "resale_exclude_reason", "resale_in_headline",268 "months_held", "index_change", "resale_implied_log_value", "resale_implied_value", "gap", "annualised_log_return",269 "short_hold", "class_changed", "resale_flagged"]270 return t[[c for c in cols if c in t.columns]].sort_values(["date", "tx_id"]).reset_index(drop=True)271272273def prior_targets(analytic: Any, full: Any | None = None, index: pd.DataFrame | None = None, *, min_n: int = MIN_CELL_N) -> pd.DataFrame:274 """Backward direction: for every analytic sale with an *earlier* sale of the same parcel, the earlier price275 brought forward to the sale date by the index (`prior_implied_log_value`). Same selection caveats; unlike276 `prior_sale_price` in the analytic table this uses the full snapshot and the deflator."""277 a = to_pandas(analytic)278 a["tx_id"] = a["tx_id"].astype(str)279 if index is None:280 index = monthly_index(a, min_n=min_n)281 pairs = resale_pairs(full if full is not None else a).rename(columns={282 "tx_id": "prior_tx_id", "date": "prior_date", "price": "prior_price", "property_class": "prior_property_class",283 "resale_tx_id": "tx_id", "exclude_reason": "prior_exclude_reason"})284 pairs = pairs.drop(columns=["resale_date", "resale_price", "resale_property_class", "resale_exclude_reason", "resale_in_headline", "parcel_id"])285 keep = [c for c in ["tx_id", "date", "price", "mun_code", "property_class", "region_group"] if c in a.columns]286 t = pairs.merge(a[keep], on="tx_id", how="inner", validate="one_to_one")287 t["log_price"] = np.log(t["price"].astype(float))288 t["months_held"] = _months_between(t["prior_date"], t["date"])289 t["index_change"] = index_change(index, t["mun_code"], t["property_class"], t["prior_date"], t["date"], t.get("region_group"))290 t["prior_implied_log_value"] = np.log(t["prior_price"]) + t["index_change"]291 t["gap"] = t["prior_implied_log_value"] - t["log_price"]292 t["short_hold"] = t["months_held"] < SHORT_HOLD_MONTHS293 return t.sort_values(["date", "tx_id"]).reset_index(drop=True)294295296# ------------------------------------------------------------------------------------------------ description297298def describe_resale_sample(targets: pd.DataFrame, analytic: Any | None = None) -> dict[str, Any]:299 """Counts, holding periods, returns and the gap between the resale-implied value and the price, overall and by300 class / region group / holding-period bucket. Pure description; no selection correction."""301 t = targets.copy()302 t["hold_bucket"] = pd.cut(t["months_held"], [0, 12, 24, 36, 72], labels=["<12", "12-24", "24-36", "36+"], right=False)303304 def block(g: pd.DataFrame) -> dict[str, float]:305 gap = g["gap"].to_numpy(dtype=float)306 ok = np.isfinite(gap)307 return {308 "n": int(len(g)),309 "share_short_hold": float(g["short_hold"].mean()) if len(g) else np.nan,310 "share_flagged": float(g["resale_flagged"].mean()) if len(g) else np.nan,311 "months_held_p25": float(g["months_held"].quantile(0.25)) if len(g) else np.nan,312 "months_held_median": float(g["months_held"].median()) if len(g) else np.nan,313 "months_held_p75": float(g["months_held"].quantile(0.75)) if len(g) else np.nan,314 "annualised_log_return_median": float(g["annualised_log_return"].median()) if len(g) else np.nan,315 "index_change_median": float(g["index_change"].median()) if len(g) else np.nan,316 "gap_mean": float(gap[ok].mean()) if ok.any() else np.nan,317 "gap_median": float(np.median(gap[ok])) if ok.any() else np.nan,318 "gap_sd": float(gap[ok].std(ddof=1)) if ok.sum() > 1 else np.nan,319 "gap_abs_median": float(np.median(np.abs(gap[ok]))) if ok.any() else np.nan,320 }321322 out: dict[str, Any] = {"overall": block(t), "by_class": {}, "by_region": {}, "by_hold_bucket": {}, "by_split": {}}323 for k, g in t.groupby("property_class"):324 out["by_class"][str(k)] = block(g)325 for k, g in t.groupby("region_group"):326 out["by_region"][str(k)] = block(g)327 for k, g in t.groupby("hold_bucket", observed=True):328 out["by_hold_bucket"][str(k)] = block(g)329 if "split" in t.columns:330 for k, g in t.groupby("split"):331 out["by_split"][str(k)] = block(g)332 if analytic is not None:333 a = to_pandas(analytic)334 out["n_analytic"] = int(len(a))335 out["share_with_resale"] = float(len(t) / len(a)) if len(a) else np.nan336 return out337