spb/qwhpi Public

1# =============================================================================2# QWHPI — Quebec Weekly Housing Price Index3# Author  : Simon-Pierre Boucher4# Contact : contact@spboucher.ai5# File    : api/app/routers/maps.py6# Purpose : Choropleth data endpoint — one metric across geographies.7# =============================================================================8"""Map data endpoints (region choropleth + assessment gap)."""910from __future__ import annotations1112import polars as pl13from fastapi import APIRouter, HTTPException, Query, Request, Response1415from app.deps import etag_for16from app.schemas import MapCell, MapResponse17from app.services import data1819router = APIRouter()2021METRICS = {22    "yoy": "yoy_pct",23    "six_month": "six_month_pct",24    "three_month": "three_month_pct",25    "index_level": "index_smoothed",26}272829@router.get("/map", response_model=MapResponse)30def map_data(31    request: Request,32    response: Response,33    metric: str = Query("yoy", description=f"one of {sorted(METRICS)} or assessment_gap"),34    level: str = Query("region", pattern="^(region|municipality)$"),35    type: str = Query("all"),36    period: str | None = None,37):38    """One metric per geography for choropleth rendering."""39    c = data.canonical().filter(40        (pl.col("geography_level") == level) & (pl.col("property_type") == type))41    if period is None:42        period = c.filter(~pl.col("is_partial_month"))["period"].max()43    if metric == "assessment_gap":44        gap = data.assessment_gap().filter(45            (pl.col("property_type") == type) & (pl.col("period") == period)46            & pl.col("geography_id").str.starts_with("region-" if level == "region" else ""))47        c_wk = c.filter(pl.col("period") == period).select(48            "geography_id", "geography_name", "reliability_grade", "transactions")49        j = c_wk.join(gap, on="geography_id", how="left")50        cells = [MapCell(geography_id=r["geography_id"],51                         geography_name=r["geography_name"],52                         value=r["median_ratio"],53                         reliability_grade=r["reliability_grade"],54                         transactions=r["transactions"])55                 for r in j.iter_rows(named=True)]56    else:57        if metric not in METRICS:58            raise HTTPException(422, f"metric must be one of {sorted(METRICS)} or assessment_gap")59        col = METRICS[metric]60        wk = c.filter(pl.col("period") == period)61        if wk.height == 0:62            raise HTTPException(404, f"no data for period {period}")63        cells = [MapCell(geography_id=r["geography_id"],64                         geography_name=r["geography_name"],65                         value=r[col],66                         reliability_grade=r["reliability_grade"],67                         transactions=r["transactions"])68                 for r in wk.iter_rows(named=True)]69    if etag_for(request, response, metric, level, type, str(period)):70        return Response(status_code=304)71    return MapResponse(metric=metric, level=level, property_type=type,72                       period=str(period), cells=cells)73