# Author: Simon-Pierre Boucher — simon-pierre.boucher@uqo.ca """Comparable retrieval, market context, geo context and feature matrix (synthetic data).""" from __future__ import annotations from datetime import date, timedelta import numpy as np import polars as pl import pytest from agentic_valuation.features.comparables import retrieve_comparables, retrieve_comparables_detailed from agentic_valuation.features.geo import GeoLayers, geo_context from agentic_valuation.features.market import market_context from agentic_valuation.features.matrix import CATEGORICAL, ID_COLUMNS, build_feature_matrix, feature_columns, municipality_rolling, neighbour_summary from agentic_valuation.schemas import Comparable, GeoContext, MarketContext def _subjects(df: pl.DataFrame, n: int = 60) -> pl.DataFrame: return df.filter(pl.col("date") >= date(2024, 1, 1)).sample(n=n, seed=1) def test_comparables_contract_and_order(syn_analytic: pl.DataFrame, cfg: dict) -> None: for r in _subjects(syn_analytic, 60).iter_rows(named=True): cutoff = r["date"] - timedelta(days=1) cs = retrieve_comparables_detailed(syn_analytic, r, cutoff, cfg) comps = cs.comparables assert all(isinstance(c, Comparable) for c in comps) assert len(comps) <= cfg["comparables"]["max_comps"] assert [c.evidence_id for c in comps] == [f"C{i:02d}" for i in range(1, len(comps) + 1)] assert all(c.sale_date < cutoff for c in comps) assert all((cutoff - c.sale_date).days <= cfg["comparables"]["lookback_days"] for c in comps) assert all(c.property_class == r["property_class"] for c in comps) assert all(c.distance_km <= cs.radius_km + 1e-6 for c in comps) assert all(c.months_before_valuation > 0 for c in comps) if len(comps) < cfg["comparables"]["min_comps"]: assert cs.expanded and cs.notes def test_comparables_exclude_subject_parcel(syn_analytic: pl.DataFrame, cfg: dict) -> None: repeat = syn_analytic.filter(pl.col("prior_sale_date").is_not_null()).head(30) for r in repeat.iter_rows(named=True): cutoff = r["date"] - timedelta(days=1) own = syn_analytic.filter(pl.col("parcel_id") == r["parcel_id"]) own_keys = set(zip(own["date"].to_list(), own["price"].to_list())) for c in retrieve_comparables(syn_analytic, r, cutoff, cfg): assert not ((c.sale_date, c.price) in own_keys and c.distance_km == 0.0) def test_comparables_accept_frame_and_future_blind(syn_analytic: pl.DataFrame, cfg: dict) -> None: r = _subjects(syn_analytic, 1) cutoff = r["date"][0] - timedelta(days=1) a = retrieve_comparables(syn_analytic, r, cutoff, cfg) pert = syn_analytic.with_columns(pl.when(pl.col("date") >= cutoff).then(pl.col("price") * 3).otherwise(pl.col("price")).alias("price")) b = retrieve_comparables(pert, r.row(0, named=True), cutoff, cfg) assert [c.model_dump() for c in a] == [c.model_dump() for c in b] def test_market_context_asof(syn_analytic: pl.DataFrame, cfg: dict) -> None: for r in _subjects(syn_analytic, 10).iter_rows(named=True): cutoff = r["date"] - timedelta(days=1) m = market_context(syn_analytic, r, cutoff, cfg) assert isinstance(m, MarketContext) and m.as_of == cutoff assert all(rec["month"] <= cutoff.strftime("%Y-%m") for rec in m.municipality_price_index) if m.municipality_price_index: assert m.municipality_price_index[0]["index"] == 100.0 assert m.mortgage_rate_5y_fixed is None pert = syn_analytic.with_columns(pl.when(pl.col("date") > cutoff).then(pl.col("price") * 5).otherwise(pl.col("price")).alias("price")) assert market_context(pert, r, cutoff, cfg).model_dump() == m.model_dump() def test_feature_matrix_shape_and_columns(syn_matrix: pl.DataFrame, syn_analytic: pl.DataFrame) -> None: assert syn_matrix.height == syn_analytic.height num, cat = feature_columns(syn_matrix) assert set(cat) == set(CATEGORICAL) assert "price" not in num and "log_price" not in num and "price_tercile" not in num assert {"nb_n_12m", "nb_med_ppm2", "mun_med_ppm2_12m", "prior_sale_log_price", "month_index"} <= set(num) assert syn_matrix["tx_id"].n_unique() == syn_matrix.height for c in num: assert syn_matrix[c].dtype == pl.Float64 def test_neighbour_summary_matches_brute_force(syn_analytic: pl.DataFrame) -> None: nb = neighbour_summary(syn_analytic) sample = syn_analytic.filter(pl.col("date") >= date(2023, 1, 1)).sample(n=40, seed=3) for r in sample.iter_rows(named=True): cand = syn_analytic.filter( (pl.col("property_class") == r["property_class"]) & (pl.col("parcel_id") != r["parcel_id"]) & (pl.col("date") < r["date"]) & (pl.col("date") >= r["date"] - timedelta(days=365)) ).with_columns(((pl.col("x_m") - r["x_m"]) ** 2 + (pl.col("y_m") - r["y_m"]) ** 2).sqrt().alias("d")) cand = cand.filter(pl.col("d") <= 3000.0) got = nb.filter(pl.col("tx_id") == r["tx_id"]).row(0, named=True) assert got["nb_n_12m"] == cand.height if cand.height: assert abs(got["nb_med_log_price"] - float(cand["log_price"].median())) < 1e-9 ppm2 = (cand["price"] / cand["floor_area_m2"]).drop_nulls() assert abs(got["nb_med_ppm2"] - float(ppm2.median())) < 1e-9 else: assert got["nb_med_log_price"] is None or np.isnan(got["nb_med_log_price"]) def test_municipality_rolling_matches_brute_force(syn_analytic: pl.DataFrame) -> None: mr = municipality_rolling(syn_analytic) sample = syn_analytic.filter(pl.col("date") >= date(2022, 6, 1)).sample(n=40, seed=4) for r in sample.iter_rows(named=True): cand = syn_analytic.filter( (pl.col("mun_code") == r["mun_code"]) & (pl.col("property_class") == r["property_class"]) & (pl.col("date") < r["date"]) & (pl.col("date") >= r["date"] - timedelta(days=365)) ) got = mr.filter(pl.col("tx_id") == r["tx_id"]).row(0, named=True) assert got["mun_n_12m"] == cand.height if cand.height: assert abs(got["mun_med_log_price_12m"] - float(cand["log_price"].median())) < 1e-9 def test_feature_matrix_future_blind(syn_analytic: pl.DataFrame, syn_splits: pl.DataFrame) -> None: cutoff = date(2025, 4, 1) base = build_feature_matrix(syn_analytic, syn_splits).filter(pl.col("date") < cutoff).sort("tx_id") trunc = build_feature_matrix(syn_analytic.filter(pl.col("date") < cutoff), syn_splits).sort("tx_id") assert base["tx_id"].to_list() == trunc["tx_id"].to_list() for c in base.columns: a, b = base[c], trunc[c] if a.dtype == pl.Float64: a, b = a.fill_nan(None), b.fill_nan(None) assert (((a - b).abs() < 1e-9) | (a.is_null() & b.is_null())).all(), c else: assert a.eq_missing(b).all(), c def test_geo_context_without_layers_returns_none(syn_analytic: pl.DataFrame) -> None: r = syn_analytic.head(1).row(0, named=True) assert geo_context(r, None) is None cached = pl.DataFrame({"tx_id": [r["tx_id"]], "flood_mapped": [True], "in_flood_zone": [False], "flood_zone_kind": [None], "transit_stops_within_1km": [12], "dauid": ["24660001"], "da_population": [500], "da_population_density": [4000.0], "da_median_age": [40.0], "da_median_income": [45000.0], "da_pct_renters": [60.0], "cbd_name": ["Montréal"], "distance_to_cbd_km": [3.2]}) g = geo_context(r, None, cached=cached) assert isinstance(g, GeoContext) and g.transit_stops_within_1km == 12 and g.in_flood_zone is False and g.distance_to_cbd_km == 3.2 @pytest.mark.skipif(not GeoLayers.available(), reason="external layers not copied") def test_geo_layers_point_queries(syn_analytic: pl.DataFrame) -> None: layers = GeoLayers.load() # Old Montréal: many stops, in a DA, near the Montréal CBD g = geo_context({"tx_id": "x", "lat": 45.5075, "lng": -73.5540}, layers) assert g is not None and g.transit_stops_within_1km > 10 and g.distance_to_cbd_km < 2.0 and g.da_population_density is not None # Gatineau point → CBD Gatineau g2 = geo_context({"tx_id": "y", "lat": 45.4290, "lng": -75.7140}, layers) assert g2 is not None and "Gatineau" in g2.notes[-1] @pytest.mark.real def test_real_feature_matrix_future_blind_sample(real_bundle: dict[str, pl.DataFrame]) -> None: an = real_bundle["analytic"] muns = an["mun_code"].unique().sort().head(15).to_list() sub = an.filter(pl.col("mun_code").is_in(muns)) cutoff = date(2025, 4, 1) base = build_feature_matrix(sub, real_bundle["splits"]).filter(pl.col("date") < cutoff).sort("tx_id") trunc = build_feature_matrix(sub.filter(pl.col("date") < cutoff), real_bundle["splits"]).sort("tx_id") assert base.drop(["price", "log_price"]).equals(trunc.drop(["price", "log_price"]))