spb/qwhpi Public

1# =============================================================================2# QWHPI — Quebec Weekly Housing Price Index3# Author  : Simon-Pierre Boucher4# Contact : contact@spboucher.ai5# File    : engine/tests/test_smoke_pipeline.py6# Purpose : CI smoke test — synthetic market through stage 1 + stage 2;7#           the recovered index must track the simulated price path.8# =============================================================================9"""End-to-end smoke test on a fully synthetic market (no data files)."""1011from __future__ import annotations1213import datetime as dt14import sys15from pathlib import Path1617sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))1819import numpy as np20import polars as pl2122from qwhpi.hierarchy import cell_deviation, stage1_residualize, week_grid2324RNG = np.random.default_rng(3)252627def simulate(n_weeks: int = 60, per_week: int = 150) -> tuple[pl.DataFrame, np.ndarray]:28    """Synthetic province with a known weekly log price path."""29    path = np.cumsum(RNG.normal(0.005, 0.003, n_weeks))  # clear boom30    rows = []31    start = dt.date(2021, 1, 4)32    for w in range(n_weeks):33        week = start + dt.timedelta(weeks=w)34        for i in range(per_week):35            fa = float(np.exp(RNG.normal(4.7, 0.35)))36            age_yrs = int(RNG.integers(0, 90))37            ptype = ["unifamilial", "condo", "plex"][int(RNG.integers(0, 3))]38            base = 12.0 + 0.6 * (np.log(fa) - 4.7) - 0.002 * age_yrs \39                + {"unifamilial": 0.0, "condo": -0.15, "plex": 0.2}[ptype]40            lp = base + path[w] + RNG.normal(0, 0.25)41            rows.append({42                "id": f"{w}-{i}",43                "log_amount": lp,44                "week_str": week.isoformat(),45                "propertyType": ptype,46                "log_fa": np.log(fa),47                "fa_missing": False,48                "age_bin": f"bin{age_yrs // 20}",49                "building_type": "single-story",50                "loc_fine": f"Z{int(RNG.integers(0, 6))}",51                "region_code": "03" if RNG.random() < 0.5 else "06",52                "geo_code": "23027",53                "municipality": "Québec",54            })55    return pl.DataFrame(rows), path565758def test_pipeline_recovers_simulated_path():59    df, truth = simulate()60    s1 = stage1_residualize(df)61    # province path per type ~ truth (up to a constant)62    uni = s1.paths.filter(pl.col("property_type") == "unifamilial").sort("week")63    est = uni["delta"].to_numpy()64    est = est - est.mean()65    tr = truth[-len(est):] - truth[-len(est):].mean()66    corr = np.corrcoef(est, tr)[0, 1]67    assert corr > 0.9, f"stage-1 path corr {corr:.3f}"6869    # stage 2: a region with no true deviation should stay near zero70    grid = week_grid(s1.residuals)71    cell = s1.residuals.filter(72        (pl.col("region_code") == "03")73        & (pl.col("propertyType") == "unifamilial")).rename({"resid": "dev"})74    dev = cell_deviation(cell.select("week_str", "dev"), grid,75                         float(cell["dev"].var()))76    assert np.abs(dev.fit.smoothed).max() < 0.0577