1# Author: Simon-Pierre Boucher — simon-pierre.boucher@uqo.ca2"""Fundamental-value triangulation: index chaining, resale arithmetic, yields, user-cost identity, lease cleaning."""3from __future__ import annotations45import sqlite367import numpy as np8import pandas as pd9import pytest1011from agentic_valuation.fundamental.repeat_sales import (describe_resale_sample, index_change, monthly_index, prior_targets,12 resale_pairs, resale_targets)13from agentic_valuation.fundamental.rents import (annual_mortgage_rate, gross_yields, implied_value_ratio, load_leases,14 median_rents, rooms_bucket, user_cost)151617def toy_panel(growth: float = 0.01, months: int = 12, n_big: int = 20, n_small: int = 2) -> pd.DataFrame:18 """Two municipalities of one region, same class, log price per m² growing by `growth` per month.19 `big` has n_big sales per month (always ≥ 5), `small` has n_small (< 5, must fall back to the region)."""20 rows = []21 rng = np.random.default_rng(0)22 for m in range(months):23 date = pd.Timestamp(2021, 1, 1) + pd.DateOffset(months=m)24 for mun, n, base in (("big", n_big, 8.0), ("small", n_small, 7.5)):25 for k in range(n):26 area = 100.0 + 10 * (k % 3)27 lp = base + growth * m28 rows.append({"tx_id": f"{mun}-{m}-{k}", "parcel_id": f"{mun}-p{m}-{k}", "mun_code": mun, "region_group": "R",29 "property_class": "house", "date": date + pd.Timedelta(days=int(rng.integers(0, 27))),30 "floor_area_m2": area, "price": float(np.exp(lp) * area)})31 return pd.DataFrame(rows)323334def test_index_chaining_recovers_constant_growth_and_falls_back_to_region():35 panel = toy_panel(growth=0.01, months=12)36 idx = monthly_index(panel, min_n=5, window=3)37 big = idx[(idx["level"] == "mun") & (idx["mun_code"] == "big")].sort_values("month_id")38 small = idx[(idx["level"] == "mun") & (idx["mun_code"] == "small")].sort_values("month_id")39 assert len(big) == 12 and big["log_index"].iloc[0] == 0.040 # deterministic growth: every interior monthly change of the chained log index is 0.01, from the municipality41 # source; the first and last months of the grid carry the edge effect of the centred window (two-point median)42 d = np.diff(big["log_index"].to_numpy())43 np.testing.assert_allclose(d[1:-1], 0.01, atol=1e-9)44 np.testing.assert_allclose([d[0], d[-1]], [0.005, 0.005], atol=1e-9)45 assert set(big["source"].iloc[1:]) == {"mun"}46 # thin municipality: every change comes from the region series (22 sales/month in the region), no level jump47 assert set(small["source"].iloc[1:]) == {"region"}48 np.testing.assert_allclose(small["log_index"].to_numpy(), big["log_index"].to_numpy(), atol=1e-9)49 # region-level series are present for the fallback of index_change50 assert (idx["level"] == "region").sum() == 1251 ch = index_change(idx, ["big", "small", "absent"], ["house"] * 3, [pd.Timestamp(2021, 2, 15)] * 3, [pd.Timestamp(2021, 8, 3)] * 3,52 region_group=["R", "R", "R"])53 np.testing.assert_allclose(ch, [0.06, 0.06, 0.06], atol=1e-9)54 ch_nofallback = index_change(idx, ["absent"], ["house"], [pd.Timestamp(2021, 2, 15)], [pd.Timestamp(2021, 8, 3)])55 assert np.isnan(ch_nofallback[0])565758def test_index_switch_between_sources_has_no_level_jump():59 panel = toy_panel(growth=0.0, months=6, n_big=20)60 # municipality `mid` with 8 sales in months 0-2 and 2 sales in months 3-5, at a level very different from the region61 rows = []62 for m in range(6):63 date = pd.Timestamp(2021, 1, 1) + pd.DateOffset(months=m)64 for k in range(8 if m < 3 else 2):65 rows.append({"tx_id": f"mid-{m}-{k}", "parcel_id": f"mid-{m}-{k}", "mun_code": "mid", "region_group": "R",66 "property_class": "house", "date": date, "floor_area_m2": 100.0, "price": float(np.exp(9.0) * 100)})67 panel = pd.concat([panel, pd.DataFrame(rows)], ignore_index=True)68 idx = monthly_index(panel, min_n=5, window=3)69 mid = idx[(idx["level"] == "mun") & (idx["mun_code"] == "mid")].sort_values("month_id")70 assert list(mid["source"]) == ["base", "mun", "mun", "region", "region", "region"]71 # flat markets everywhere → the chained index stays flat despite the switch of source (no jump of 9 − 8)72 np.testing.assert_allclose(mid["log_index"].to_numpy(), 0.0, atol=1e-9)737475def test_resale_target_arithmetic():76 panel = toy_panel(growth=0.01, months=12)77 idx = monthly_index(panel, min_n=5, window=3)78 # add a resale of parcel big-p1-0: first sale 2021-02-10 at 100 000, resale 2021-08-10 at 130 000 (interior months)79 first = panel[panel["parcel_id"] == "big-p1-0"].iloc[0]80 panel.loc[panel["parcel_id"] == "big-p1-0", "price"] = 100_000.081 panel.loc[panel["parcel_id"] == "big-p1-0", "date"] = pd.Timestamp(2021, 2, 10)82 resale = first.copy()83 resale["tx_id"], resale["date"], resale["price"] = "big-resale", pd.Timestamp(2021, 8, 10), 130_000.084 full = pd.concat([panel, resale.to_frame().T], ignore_index=True)85 full["price"] = full["price"].astype(float)86 pairs = resale_pairs(full)87 assert len(pairs) == 1 and pairs["tx_id"].iloc[0] == "big-1-0" and pairs["resale_tx_id"].iloc[0] == "big-resale"88 t = resale_targets(panel, full, idx)89 assert len(t) == 190 r = t.iloc[0]91 assert r["resale_tx_id"] == "big-resale" and r["resale_price"] == 130_000.092 assert abs(r["months_held"] - 181 / (365.25 / 12)) < 1e-993 np.testing.assert_allclose(r["index_change"], 0.06, atol=1e-9) # Feb → Aug = 6 × 0.0194 np.testing.assert_allclose(r["resale_implied_log_value"], np.log(130_000.0) - 0.06, atol=1e-9)95 np.testing.assert_allclose(r["gap"], np.log(1.3) - 0.06, atol=1e-9)96 np.testing.assert_allclose(r["annualised_log_return"], np.log(1.3) * 12 / r["months_held"], atol=1e-9)97 assert bool(r["short_hold"]) is True and bool(r["resale_flagged"]) is False98 # backward direction: the resale row is not in `panel` (analytic) so prior_targets is empty; add it and check99 analytic2 = full.copy()100 p = prior_targets(analytic2, full, idx)101 assert len(p) == 1102 np.testing.assert_allclose(p["prior_implied_log_value"].iloc[0], np.log(100_000.0) + 0.06, atol=1e-9)103 desc = describe_resale_sample(t, panel)104 assert desc["overall"]["n"] == 1 and desc["by_class"]["house"]["share_short_hold"] == 1.0105106107def test_resale_pairs_drop_structural_exclusions_and_same_date():108 full = pd.DataFrame({109 "tx_id": ["a", "b", "c", "d"], "parcel_id": ["p", "p", "p", "p"],110 "date": pd.to_datetime(["2021-01-01", "2021-01-01", "2022-01-01", "2023-01-01"]),111 "price": [100.0, 100.0, 120.0, 150.0], "exclude_reason": [None, "exact_duplicate", None, "assessment_ratio_outlier"],112 "in_headline_sample": [True, False, True, False],113 })114 pairs = resale_pairs(full)115 assert list(pairs["tx_id"]) == ["a", "c"] and list(pairs["resale_tx_id"]) == ["c", "d"]116 assert pairs["resale_exclude_reason"].iloc[1] == "assessment_ratio_outlier"117118119def test_gross_yield_and_user_cost_identity():120 sales = pd.DataFrame({121 "tx_id": ["s1", "s2", "s3"], "date": pd.to_datetime(["2023-05-01", "2023-06-01", "2023-06-01"]),122 "mun_code": ["66023", "66023", "66023"], "mun_name": ["Montréal"] * 3, "region_group": ["Montréal CMA"] * 3,123 "property_class": ["condo", "plex", "house"], "price": [300_000.0, 900_000.0, 500_000.0], "n_units": [1, 3, 1],124 })125 rents = pd.DataFrame({"mun_code": ["66023"], "year": [2023], "rooms_bucket": ["all"], "n_leases": [50], "rent_median": [1250.0]})126 gy = gross_yields(sales, rents, min_leases=10)127 assert list(gy["tx_id"]) == ["s1", "s2"] # houses are not in the yield classes128 np.testing.assert_allclose(gy["gross_yield"].to_numpy(), [12 * 1250 / 300_000, 12 * 1250 * 3 / 900_000])129 assert gross_yields(sales, rents, min_leases=60).empty # cell too thin130 u = user_cost(0.05, 0.01, 0.025, 0.02, 0.02)131 np.testing.assert_allclose(u, 0.085)132 # identity: when R/P = u the rent-implied value equals the price133 np.testing.assert_allclose(implied_value_ratio(0.085, u), 1.0)134 np.testing.assert_allclose(implied_value_ratio(np.array([0.085, 0.17]), u), [1.0, 2.0])135 np.testing.assert_allclose(user_cost(np.array([0.04, 0.06]), 0.01, 0.025, 0.02, np.array([0.0, 0.04])), [0.095, 0.075])136137138def test_rooms_bucket_and_annual_rate():139 b = rooms_bucket(pd.Series([0, 1, 2, 3, 4, 7, np.nan]))140 assert list(b[:6]) == ["0-1", "0-1", "2", "3", "4+", "4+"] and pd.isna(b.iloc[6])141 weekly = pd.DataFrame({"date": pd.to_datetime(["2023-01-04", "2023-06-07", "2024-02-14"]), "rate_pct": [6.0, 7.0, 6.5]})142 r = annual_mortgage_rate(weekly, [2023, 2024, 2025])143 np.testing.assert_allclose(r.loc[2023], 0.065)144 np.testing.assert_allclose(r.loc[2024], 0.065)145 assert np.isnan(r.loc[2025]) # never filled146 assert annual_mortgage_rate(pd.DataFrame(columns=["date", "rate_pct"]), [2023]).isna().all()147148149def test_load_leases_cleaning_funnel(tmp_path):150 db = tmp_path / "rdl.db"151 con = sqlite3.connect(db)152 con.execute("CREATE TABLE rdl_housings (id INTEGER PRIMARY KEY, full_address TEXT, street_number TEXT, apartment_number TEXT, "153 "street_name TEXT, city TEXT, zip TEXT, lat REAL, lng REAL, price REAL, rooms INTEGER, year INTEGER, start_date TEXT, "154 "type_of_accomodation TEXT, heating_included INTEGER, electricity_included INTEGER, furnishing_included INTEGER, "155 "parking_included INTEGER, animal_allowed INTEGER, address_slug TEXT, updated_at TEXT, fetched_at TEXT)")156 rows = [157 (1, "1 rue A", "1", "", "rue A", "Montréal", "H1H1H1", 45.5, -73.6, 1200, 2, 2023, "2023-07-01", "apartement", 0, 0, 0, 0, 0, "1-rue-a", "", ""),158 (2, "1 rue A", "1", "", "rue A", "Montréal", "H1H1H1", 45.5, -73.6, 1200, 2, 2023, "", "apartement", 0, 0, 0, 0, 0, "1-rue-a", "", ""), # duplicate159 (3, "2 rue B", "2", "", "rue B", "Montréal", "H1H1H1", None, None, 900, 1, 2023, "", "apartement", 0, 0, 0, 0, 0, "2-rue-b", "", ""), # no coords160 (4, "3 rue C", "3", "", "rue C", "Montréal", "H1H1H1", 45.5, -73.6, 5, 1, 2023, "", "apartement", 0, 0, 0, 0, 0, "3-rue-c", "", ""), # rent 5 $161 (5, "4 rue D", "4", "", "rue D", "Montréal", "H1H1H1", 45.5, -73.6, 1500, 3, 2923, "", "apartement", 0, 0, 0, 0, 0, "4-rue-d", "", ""), # year typo162 (6, "5 rue E", "5", "", "rue E", "Montréal", "H1H1H1", 45.5, -73.6, 1500, None, 2024, "", "apartement", 0, 0, 0, 0, 0, "5-rue-e", "", ""), # rooms null163 ]164 con.executemany("INSERT INTO rdl_housings VALUES (" + ",".join("?" * 22) + ")", rows)165 con.commit()166 con.close()167 df = load_leases(db)168 assert len(df) == 1 and df["rooms_bucket"].iloc[0] == "2" and bool(df["start_date_present"].iloc[0])169 assert df.attrs["funnel"] == {"raw": 6, "coordinates_present": 5, "rent_in_range": 4, "year_in_range": 3, "rooms_recorded": 2, "deduplicated": 1}170 df["mun_code"] = "66023"171 r = median_rents(df)172 assert set(r["rooms_bucket"]) == {"2", "all"} and (r["rent_median"] == 1200).all()173174175@pytest.mark.real176def test_real_resale_targets_cover_prior_sale_information():177 """On the real data every analytic row with `prior_sale_price` must appear as the later leg of a pair."""178 import polars as pl179180 from agentic_valuation.config import PROCESSED, RAW181182 a = pl.read_parquet(PROCESSED / "analytic.parquet")183 full = pl.read_parquet(RAW / "wp22_clean_transactions.parquet")184 idx = monthly_index(a)185 p = prior_targets(a, full, idx)186 with_prior = a.filter(pl.col("prior_sale_price").is_not_null())["tx_id"].to_list()187 assert len(set(with_prior) - set(p["tx_id"])) / max(len(with_prior), 1) < 0.02188