spb/qwhpi Public

1# =============================================================================2# QWHPI — Quebec Weekly Housing Price Index3# Author  : Simon-Pierre Boucher4# Contact : contact@spboucher.ai5# File    : api/app/services/data.py6# Purpose : Data access layer — canonical parquet lake (source of truth),7#           in-memory cached, vintage-keyed for ETag support.8# =============================================================================9"""Data access for the API.1011Reads the processed Parquet lake (never raw CSVs). Frames are cached in12memory and refreshed when the canonical file's mtime changes (the weekly13scheduler swaps files atomically). ``DATA_DIR`` can be overridden via the14``QWHPI_DATA_DIR`` environment variable (used by docker-compose).15"""1617from __future__ import annotations1819import os20import threading21from pathlib import Path2223import polars as pl2425_REPO_ROOT = Path(__file__).resolve().parents[3]26DATA_DIR = Path(os.environ.get("QWHPI_DATA_DIR", _REPO_ROOT / "data" / "processed"))2728CANONICAL = DATA_DIR / "qhpi_monthly.parquet"29COVERAGE = DATA_DIR / "coverage_matrix_monthly.parquet"30LIQUIDITY = DATA_DIR / "monthly_liquidity.parquet"31ASSESSMENT = DATA_DIR / "assessment_gap_monthly.parquet"32MUNICIPALITIES = DATA_DIR / "municipalities.parquet"33FIRST_RELEASE = DATA_DIR / "first_release_monthly.parquet"3435_lock = threading.Lock()36_cache: dict[str, tuple[float, pl.DataFrame]] = {}373839def _load(path: Path) -> pl.DataFrame:40    mtime = path.stat().st_mtime41    with _lock:42        hit = _cache.get(str(path))43        if hit and hit[0] == mtime:44            return hit[1]45        df = pl.read_parquet(path)46        _cache[str(path)] = (mtime, df)47        return df484950def canonical() -> pl.DataFrame:51    return _load(CANONICAL)525354def coverage() -> pl.DataFrame:55    return _load(COVERAGE)565758def liquidity() -> pl.DataFrame:59    return _load(LIQUIDITY)606162def assessment_gap() -> pl.DataFrame:63    return _load(ASSESSMENT)646566def municipalities() -> pl.DataFrame:67    return _load(MUNICIPALITIES)686970def first_release() -> pl.DataFrame | None:71    return _load(FIRST_RELEASE) if FIRST_RELEASE.exists() else None727374def data_vintage() -> str:75    return str(canonical()["data_vintage"][0])767778def model_version() -> str:79    return str(canonical()["model_version"][0])808182def series(geography: str, property_type: str,83           period_from: str | None = None, period_to: str | None = None) -> pl.DataFrame:84    df = canonical().filter(85        (pl.col("geography_id") == geography)86        & (pl.col("property_type") == property_type)87    )88    if period_from:89        df = df.filter(pl.col("period") >= period_from)90    if period_to and period_to != "latest":91        df = df.filter(pl.col("period") <= period_to)92    return df.sort("period")939495def geography_ids() -> list[str]:96    return canonical()["geography_id"].unique().sort().to_list()97