spb/wp16_uqo

1# -*- coding: utf-8 -*-2"""Validation empirique de la méthode des comparables — fonctions communes.3Données : instantané Vrai Prix (transactions ⋈ units) du 2026-10-10.4Toutes les sorties sont versionnées sous out/<version>/.5"""6from __future__ import annotations7import json, math, os, sqlite3, time8from pathlib import Path9import numpy as np10import pandas as pd1112ROOT = Path(os.environ.get("CV_ROOT", Path(__file__).resolve().parents[1]))13DB = Path(os.environ.get("CV_DB", ROOT / "data" / "vraiprix-20261010.db"))14REF = ROOT / "ref"15VERSION = os.environ.get("CV_VERSION", time.strftime("%Y-%m-%d"))16OUT = ROOT / "out" / VERSION17OUT.mkdir(parents=True, exist_ok=True)18(OUT / "fig").mkdir(exist_ok=True)1920# Périodes (séparation chronologique)21TRAIN_END = "2023-12-31"      # développement22VAL_START, VAL_END = "2024-01-01", "2024-12-31"   # choix des configurations23TEST_START, TEST_END = "2025-01-01", "2026-09-29"  # évaluation finale hors échantillon24MONTH0 = "2021-01"            # premier mois des données2526def log(msg: str) -> None:27    print(f"[{time.strftime('%H:%M:%S')}] {msg}", flush=True)2829def connect(path: Path = DB) -> sqlite3.Connection:30    return sqlite3.connect(f"file:{path}?mode=ro", uri=True, timeout=60)3132def _json_default(o):33    if isinstance(o, np.integer): return int(o)34    if isinstance(o, np.floating): return None if np.isnan(o) else float(o)35    if isinstance(o, np.bool_): return bool(o)36    if isinstance(o, np.ndarray): return o.tolist()37    if isinstance(o, pd.Timestamp): return o.strftime("%Y-%m-%d")38    if isinstance(o, float) and math.isnan(o): return None39    return str(o)4041def save_json(obj, name: str) -> Path:42    p = OUT / name; p.parent.mkdir(parents=True, exist_ok=True)43    with open(p, "w", encoding="utf-8") as f:44        json.dump(obj, f, ensure_ascii=False, indent=2, default=_json_default)45    return p4647def load_json(name: str):48    with open(OUT / name, encoding="utf-8") as f:49        return json.load(f)5051def save_parquet(df: pd.DataFrame, name: str) -> Path:52    p = OUT / name; df.to_parquet(p, index=False); return p5354def load_parquet(name: str) -> pd.DataFrame:55    return pd.read_parquet(OUT / name)5657def month_index(dates) -> np.ndarray:58    """Mois entier depuis MONTH0 (2021-01 = 0)."""59    d = pd.to_datetime(dates)60    y0, m0 = map(int, MONTH0.split("-"))61    return ((d.dt.year - y0) * 12 + (d.dt.month - m0)).to_numpy()6263def to_xy_km(lat, lng, lat0=46.8, lng0=-71.2):64    """Projection équirectangulaire locale (km) : erreur < 1 % pour des distances < 50 km."""65    x = (np.asarray(lng, float) - lng0) * 111.32 * np.cos(np.radians(np.asarray(lat, float)))66    y = (np.asarray(lat, float) - lat0) * 110.5767    return x, y6869def load_repertoire() -> pd.DataFrame:70    mun = pd.read_csv(REF / "MUN.csv", dtype=str, encoding="utf-8-sig")71    mun = mun.rename(columns=str.lower)72    out = pd.DataFrame({73        "code_mun": mun["mcode"].str.zfill(5),74        "mun_nom": mun["munnom"],75        "region": mun["regadm"].str.extract(r"\((\d+)\)")[0],76        "region_nom": mun["regadm"].str.replace(r"\s*\(\d+\)", "", regex=True),77        "mrc": mun["mrc"].str.extract(r"\((\d+)\)")[0],78        "mrc_nom": mun["mrc"].str.replace(r"\s*\(\d+\)", "", regex=True),79    })80    return out.drop_duplicates("code_mun")8182# Codages catégoriels (ordre stable)83LIEN = {"Détaché": 0, "Jumelé": 1, "En rangée 1 côté": 2, "En rangée plus de 1 côté": 2, "Intégré": 3}84GENRE = {"De plain-pied": 0, "Unimodulaire": 0, "À étages entiers": 1, "À niveaux décalés": 2, "À étage mansardé": 3}85BTYPE = {"single-story": 0, "full-story": 1, "mansard": 2, "split-level": 3, "single-wide": 4}86