# ============================================================================= # QWHPI — Quebec Weekly Housing Price Index # Author : Simon-Pierre Boucher # Contact : contact@spboucher.ai # File : api/app/services/stats.py # Purpose : Rich derived metrics per series — peak/drawdown, momentum, # volatility, CAGR, ranks, volumes, assessment gap. # ============================================================================= """Derived series statistics. Everything is computed from the canonical monthly parquet (complete months only). Definitions: - peak / drawdown ....... max of index_smoothed; % below it now - since_2021_pct ........ latest index − 100 (base = 2021 average) - cagr_pct .............. annualized log growth from the first 12-month mean to the latest 12-month mean - volatility_12m_pct .... SD of the last 12 monthly log changes (monthly, %) - momentum_3m_ann_pct ... last 3-month log growth, annualized - yoy_rank .............. rank of YoY among geographies of the same level and property type (1 = fastest) - volume_12m / volume_prev_12m / volume_yoy_pct - dollar_volume_12m ..... Σ transactions × representative value (estimate) - assessment_gap ........ latest median sale/assessment ratio (region & province only) """ from __future__ import annotations import math import numpy as np import polars as pl from app.services import data def _complete(gid: str, ptype: str) -> pl.DataFrame: return data.series(gid, ptype).filter(~pl.col("is_partial_month")) def series_stats(gid: str, ptype: str) -> dict: df = _complete(gid, ptype) if df.height < 14: return {} idx = df["index_smoothed"].to_numpy() periods = df["period"].to_list() li = np.log(idx) peak_i = int(np.argmax(idx)) latest = float(idx[-1]) peak = float(idx[peak_i]) drawdown = (latest / peak - 1) * 100 months_since_peak = len(idx) - 1 - peak_i first12 = float(np.mean(li[:12])) last12 = float(np.mean(li[-12:])) years = (len(li) - 12) / 12 cagr = (math.exp((last12 - first12) / max(years, 1e-9)) - 1) * 100 d = np.diff(li) vol12 = float(np.std(d[-12:])) * 100 mom3 = (math.exp((li[-1] - li[-4]) * 4) - 1) * 100 tx = df["transactions"].to_numpy().astype(float) rep = df["representative_value"].to_numpy() vol_12m = float(np.nansum(tx[-12:])) vol_prev = float(np.nansum(tx[-24:-12])) if len(tx) >= 24 else None volume_yoy = ((vol_12m / vol_prev - 1) * 100) if vol_prev else None dollar_12m = float(np.nansum(tx[-12:] * np.nan_to_num(rep[-12:]))) last = df.row(df.height - 1, named=True) # Rank among same level × type on YoY. level = last["geography_level"] peers = ( data.canonical() .filter((pl.col("geography_level") == level) & (pl.col("property_type") == ptype) & (pl.col("period") == last["period"])) .sort("yoy_pct", descending=True, nulls_last=True) ) ids = peers["geography_id"].to_list() rank = (ids.index(gid) + 1) if gid in ids else None # Assessment gap (province & regions). gap = None gap_df = data.assessment_gap().filter( (pl.col("geography_id") == gid) & (pl.col("property_type") == ptype)) if gap_df.height: gap = float(gap_df.sort("period")["median_ratio"][-1]) hist12 = df.tail(24) return { "period": last["period"], "index": round(latest, 2), "representative_value": last["representative_value"], "lower_95": round(last["lower_95"], 2), "upper_95": round(last["upper_95"], 2), "reliability_grade": last["reliability_grade"], "monthly_pct": last["monthly_pct"], "three_month_pct": last["three_month_pct"], "six_month_pct": last["six_month_pct"], "yoy_pct": last["yoy_pct"], "since_2021_pct": round(latest - 100, 2), "cagr_pct": round(cagr, 2), "peak_index": round(peak, 2), "peak_period": periods[peak_i], "drawdown_pct": round(drawdown, 2), "months_since_peak": months_since_peak, "at_record_high": bool(months_since_peak == 0), "volatility_12m_pct": round(vol12, 2), "momentum_3m_ann_pct": round(mom3, 2), "yoy_rank": rank, "yoy_rank_of": len(ids), "volume_12m": int(vol_12m), "volume_yoy_pct": round(volume_yoy, 2) if volume_yoy is not None else None, "dollar_volume_12m": int(dollar_12m), "assessment_gap": gap, "effective_sample_size": last["effective_sample_size"], "spark": [round(v, 2) for v in hist12["index_smoothed"].to_list()], } def overview(level: str = "region", ptype: str = "all") -> list[dict]: """One rich row per geography of a level — powers the heatmap table.""" c = data.canonical().filter( (pl.col("geography_level") == level) & (pl.col("property_type") == ptype) & (~pl.col("is_partial_month"))) latest_period = c["period"].max() out = [] for gid in c["geography_id"].unique().sort().to_list(): df = c.filter(pl.col("geography_id") == gid).sort("period") if df.height < 14: continue idx = df["index_smoothed"].to_numpy() peak = float(np.max(idx)) last = df.row(df.height - 1, named=True) tx = df["transactions"].to_numpy().astype(float) out.append({ "geography_id": gid, "geography_name": last["geography_name"], "period": last["period"], "index": round(float(idx[-1]), 1), "monthly_pct": last["monthly_pct"], "three_month_pct": last["three_month_pct"], "six_month_pct": last["six_month_pct"], "yoy_pct": last["yoy_pct"], "since_2021_pct": round(float(idx[-1]) - 100, 1), "drawdown_pct": round((float(idx[-1]) / peak - 1) * 100, 2), "at_record_high": bool(np.argmax(idx) == len(idx) - 1), "volume_12m": int(np.nansum(tx[-12:])), "representative_value": last["representative_value"], "reliability_grade": last["reliability_grade"], "spark": [round(v, 2) for v in df["index_smoothed"].tail(24).to_list()], }) return sorted(out, key=lambda r: -(r["yoy_pct"] or -999)), latest_period def overview_rows(level: str = "region", ptype: str = "all") -> dict: rows, latest_period = overview(level, ptype) return {"level": level, "property_type": ptype, "period": latest_period, "rows": rows}