1# Author: Simon-Pierre Boucher — simon-pierre.boucher@uqo.ca2"""As-of comparable retrieval (identical for every arm).34Rules (``configs/data.yaml`` → ``comparables``):5* candidate sales have ``date < cutoff`` and ``date >= cutoff − lookback_days`` (strictly before the cutoff);6* the subject's own parcel is excluded (its earlier sale is exposed separately as ``prior_sale_*``);7* same ``property_class`` when ``same_class_required``;8* expanding radius 1 → 3 → 10 km: the first radius with at least ``min_comps`` candidates is used; if even 10 km9 yields fewer than ``min_comps``, the search is extended to 25 km and the result is flagged (``expanded``);10* candidates are ranked by a similarity score (lower = better) combining distance, |Δ log floor area|,11 |Δ year built| and recency; the best ``max_comps`` receive evidence ids C01.. in score order.1213Speed: one KD-tree query on EPSG:32198 coordinates per subject (``AnalyticIndex``), then NumPy masks —14about 1–3 ms per subject on the full 632k-row table.15"""16from __future__ import annotations1718from dataclasses import dataclass, field19from datetime import date20from typing import Any2122import numpy as np23import polars as pl2425from ..config import load_config26from ..data.analytic_schema import PROPERTY_CLASSES27from ..schemas import Comparable28from .index import CLASS_CODE, AnalyticIndex, from_days, get_index, subject_to_dict, to_days2930EXTENDED_RADIUS_KM = 25.031#: similarity-score weights (dimensionless; each term is roughly 0–1 in the typical range)32W_DISTANCE, W_AREA, W_YEAR, W_RECENCY = 1.0, 1.5, 0.5, 0.5333435@dataclass36class ComparableSet:37 comparables: list[Comparable]38 radius_km: float39 n_candidates: int40 expanded: bool41 notes: list[str] = field(default_factory=list)424344def similarity_score(dist_km: np.ndarray, radius_km: float, log_area_diff: np.ndarray, year_diff: np.ndarray, days_before: np.ndarray, lookback_days: int) -> np.ndarray:45 """Lower is more similar. Missing attributes count as a neutral penalty (0.5) rather than being dropped."""46 d = dist_km / max(radius_km, 0.1)47 a = np.where(np.isnan(log_area_diff), 0.5, np.abs(log_area_diff))48 y = np.where(np.isnan(year_diff), 0.5, np.abs(year_diff) / 50.0)49 t = days_before / max(lookback_days, 1)50 return W_DISTANCE * d + W_AREA * a + W_YEAR * y + W_RECENCY * t515253def _candidates(idx: AnalyticIndex, x: float, y: float, radius_km: float, cutoff_days: int, lookback_days: int, cls_code: int | None, parcel_id: str | None) -> np.ndarray:54 cand = idx.within_radius(x, y, radius_km * 1000.0)55 if cand.size == 0:56 return cand57 d = idx.date_days[cand]58 mask = (d < cutoff_days) & (d >= cutoff_days - lookback_days)59 if cls_code is not None:60 mask &= idx.cls[cand] == cls_code61 if parcel_id is not None:62 mask &= idx.parcel_id[cand] != parcel_id63 return cand[mask]646566def retrieve_comparables_detailed(analytic: pl.DataFrame, subject: dict[str, Any] | pl.DataFrame, cutoff: date, cfg: dict[str, Any] | None = None) -> ComparableSet:67 """Full retrieval result (comparables + radius used + flags)."""68 cfg = cfg or load_config("data")69 c = cfg["comparables"] if "comparables" in cfg else cfg70 radii = [float(r) for r in c["radius_km"]]71 lookback = int(c["lookback_days"])72 max_comps, min_comps = int(c["max_comps"]), int(c["min_comps"])73 same_class = bool(c.get("same_class_required", True))7475 s = subject_to_dict(subject)76 idx = get_index(analytic)77 if "x_m" in s and s.get("x_m") is not None:78 x, y = float(s["x_m"]), float(s["y_m"])79 else:80 from ..data.analytic_schema import project_xy8182 xx, yy = project_xy(np.array([s["lat"]]), np.array([s["lng"]]))83 x, y = float(xx[0]), float(yy[0])84 cls_code = CLASS_CODE.get(s.get("property_class")) if same_class else None85 parcel = s.get("parcel_id")86 cutoff_days = to_days(cutoff)87 notes: list[str] = []88 expanded = False89 used_radius = radii[-1]90 cand = np.empty(0, dtype=np.int64)91 for r in radii:92 cand = _candidates(idx, x, y, r, cutoff_days, lookback, cls_code, parcel)93 used_radius = r94 if cand.size >= min_comps:95 break96 if cand.size < min_comps:97 cand = _candidates(idx, x, y, EXTENDED_RADIUS_KM, cutoff_days, lookback, cls_code, parcel)98 used_radius = EXTENDED_RADIUS_KM99 expanded = True100 notes.append(f"fewer than {min_comps} same-class sales within {radii[-1]:g} km in the last {lookback} days; search extended to {EXTENDED_RADIUS_KM:g} km")101 if cand.size < min_comps:102 notes.append(f"only {cand.size} comparable(s) found within {EXTENDED_RADIUS_KM:g} km (thin market)")103 if cand.size == 0:104 return ComparableSet([], used_radius, 0, expanded, notes)105106 dist_km = np.hypot(idx.x[cand] - x, idx.y[cand] - y) / 1000.0107 s_area = s.get("floor_area_m2")108 with np.errstate(all="ignore"):109 log_area_diff = np.log(idx.floor_area[cand]) - np.log(float(s_area)) if s_area else np.full(cand.size, np.nan)110 s_year = s.get("year_built")111 year_diff = (idx.year_built[cand] - float(s_year)) if s_year else np.full(cand.size, np.nan)112 days_before = (cutoff_days - idx.date_days[cand]).astype(float)113 score = similarity_score(dist_km, used_radius, log_area_diff, year_diff, days_before, lookback)114 order = np.lexsort((idx.tx_id[cand].astype(str), score))[:max_comps]115116 comps: list[Comparable] = []117 for rank, j in enumerate(order, start=1):118 i = int(cand[j])119 fa = idx.floor_area[i]120 rt = idx.roll_total[i]121 comps.append(122 Comparable(123 evidence_id=f"C{rank:02d}",124 sale_date=from_days(idx.date_days[i]),125 price=float(idx.price[i]),126 distance_km=round(float(dist_km[j]), 3),127 property_class=PROPERTY_CLASSES[idx.cls[i]],128 subtype=str(idx.subtype[i]) if idx.subtype[i] in ("detached", "semi", "row", "integrated") else "unknown",129 municipality=str(idx.mun_name[i]),130 floor_area_m2=None if np.isnan(fa) else float(fa),131 land_area_m2=None if np.isnan(idx.land_area[i]) else float(idx.land_area[i]),132 year_built=None if np.isnan(idx.year_built[i]) else int(idx.year_built[i]),133 storeys=None if np.isnan(idx.storeys[i]) else float(idx.storeys[i]),134 n_units=None if np.isnan(idx.n_units[i]) else int(idx.n_units[i]),135 roll_total_value=None if np.isnan(rt) else float(rt),136 price_per_m2=None if np.isnan(fa) or fa <= 0 else round(float(idx.price[i] / fa), 1),137 assessment_ratio=None if np.isnan(rt) or rt <= 0 else round(float(idx.price[i] / rt), 4),138 months_before_valuation=round(float(days_before[j]) / 30.4375, 2),139 )140 )141 return ComparableSet(comps, used_radius, int(cand.size), expanded, notes)142143144def retrieve_comparables(analytic: pl.DataFrame, subject: dict[str, Any] | pl.DataFrame, cutoff: date, cfg: dict[str, Any] | None = None) -> list[Comparable]:145 """Comparables C01..C12 for ``subject`` using only sales strictly before ``cutoff`` (see module docstring)."""146 return retrieve_comparables_detailed(analytic, subject, cutoff, cfg).comparables147