spb/qwhpi Public
QHPI — Quebec Housing Price Index: quality-adjusted, hierarchically pooled housing price indexes.
Python 63.9%
TypeScript 25.4%
CSS 5.5%
TeX 3.5%
SQL 0.8%
Makefile 0.5%
Dockerfile 0.5%
1#!/usr/bin/env python32# =============================================================================3# QWHPI — Quebec Weekly Housing Price Index4# Author : Simon-Pierre Boucher5# Contact : contact@spboucher.ai6# File : ops/scheduler/refresh.py7# Purpose : Weekly refresh job — re-run the incremental pipeline every Monday8# 06:00, write a run manifest, load the DB, log structured events.9# =============================================================================10"""Weekly refresh scheduler.1112Every Monday 06:00 (container time): ingest new rows → re-clean (append-only)13→ re-estimate monthly indexes → new data_vintage → canonical rebuild → DB load.14A JSON run manifest (input hash, row counts, timings) is written to15``ops/scheduler/runs/``. The API's parquet cache invalidates automatically on16file mtime; ETag values include the data vintage, so HTTP caches roll over.17"""1819from __future__ import annotations2021import json22import logging23import subprocess24import sys25import time26from datetime import datetime, timezone27from pathlib import Path2829import schedule3031WORKSPACE = Path("/workspace")32RUNS_DIR = WORKSPACE / "ops" / "scheduler" / "runs"3334logging.basicConfig(35 level=logging.INFO,36 format='{"ts":"%(asctime)s","level":"%(levelname)s","msg":"%(message)s"}',37)38log = logging.getLogger("qwhpi.scheduler")3940STEPS = [41 "engine/scripts/03_clean.py",42 "engine/scripts/09_monthly.py",43 "engine/scripts/11_monthly_canonical.py",44]454647def run_refresh() -> None:48 started = time.time()49 manifest: dict = {"started_at": datetime.now(timezone.utc).isoformat(),50 "steps": []}51 ok = True52 for step in STEPS:53 t0 = time.time()54 proc = subprocess.run([sys.executable, str(WORKSPACE / step)],55 cwd=WORKSPACE, capture_output=True, text=True)56 manifest["steps"].append({57 "step": step,58 "seconds": round(time.time() - t0, 1),59 "returncode": proc.returncode,60 })61 if proc.returncode != 0:62 log.error("step %s failed: %s", step, proc.stderr[-2000:])63 ok = False64 break65 log.info("step %s ok (%.0fs)", step, time.time() - t0)6667 if ok:68 sys.path.insert(0, str(WORKSPACE / "engine" / "src"))69 try:70 from qwhpi.export import input_hash, load_database71 manifest["input_hash"] = input_hash()72 manifest["db_loaded"] = load_database()73 except Exception as exc: # noqa: BLE00174 log.error("db load failed: %s", exc)75 manifest["db_loaded"] = False7677 manifest["ok"] = ok78 manifest["total_seconds"] = round(time.time() - started, 1)79 RUNS_DIR.mkdir(parents=True, exist_ok=True)80 stamp = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")81 (RUNS_DIR / f"run_{stamp}.json").write_text(json.dumps(manifest, indent=2))82 log.info("refresh finished ok=%s in %.0fs", ok, manifest["total_seconds"])838485if __name__ == "__main__":86 log.info("scheduler up — weekly refresh Mondays 06:00")87 schedule.every().monday.at("06:00").do(run_refresh)88 while True:89 schedule.run_pending()90 time.sleep(60)91