spb/immo-ka

1# -----------------------------------------------------------------------------2# Immo-Ka — Agrégateur de maisons à vendre (province de Québec)3# Auteur : Simon-Pierre Boucher — contact@spboucher.ai4# statsfiche.py : panneaux « enrichissements de la fiche » de l'onglet5#   Statistiques — tout ce qu'Immo-Ka ajoute par-dessus l'annonce brute :6#     · couverture des enrichissements (géocodage, quartier, Vrai-Prix,7#       caractéristiques, audit photos) + couches de données branchées ;8#     · Vrai-Prix (écarts prix demandé vs estimation, verdicts, villes) ;9#     · Historique des prix (baisses/hausses) ;10#     · Taux hypothécaires (meilleurs taux réels multibanques) ;11#     · Annuaires (déménageurs et inspecteurs en bâtiment).12#   Même contrat de rendu que statsextra.py (kpis/breakdowns/distributions/13#   tables, rendu générique du kit stats) — un panneau absent si sa source14#   ne répond pas. Cache 30 min.15# -----------------------------------------------------------------------------16from __future__ import annotations1718import json19import sqlite320import time21from pathlib import Path22from statistics import median2324ROOT = Path(__file__).resolve().parent.parent25DATA = ROOT / "data"26_CACHE: dict[str, tuple[float, list]] = {}27_TTL = 1800282930def _ro(name: str) -> sqlite3.Connection | None:31    p = DATA / name32    if not p.exists():33        return None34    con = sqlite3.connect(f"file:{p}?mode=ro", uri=True)35    con.row_factory = sqlite3.Row36    return con373839def _active_where(con) -> str:40    cols = {r[1] for r in con.execute("PRAGMA table_info(listings)")}41    w = "active=1"42    if "dup_hidden" in cols:43        w += " AND (dup_hidden IS NULL OR dup_hidden=0)"44    return w454647def _n1(con, sql: str, args: tuple = ()) -> int:48    return con.execute(sql, args).fetchone()[0] or 0495051def _fr(n: float) -> str:52    return f"{round(n):,}".replace(",", " ")535455def _pctof(part: int, tot: int) -> float:56    return round(100.0 * part / tot, 1) if tot else 0.0575859# --- Panneau : couverture des enrichissements ---------------------------------60def _panel_enrichissement(con) -> dict | None:61    W = _active_where(con)62    tot = _n1(con, f"SELECT COUNT(*) FROM listings WHERE {W}")63    if tot < 100:64        return None65    geo = _n1(con, f"SELECT COUNT(*) FROM listings WHERE {W}"66                   " AND lat IS NOT NULL AND lng IS NOT NULL")67    quart = _n1(con, f"SELECT COUNT(*) FROM listings WHERE {W}"68                     " AND dauid IS NOT NULL AND dauid<>''")69    vp = _n1(con, f"SELECT COUNT(*) FROM listings WHERE {W}"70                  " AND vraiprix LIKE '%estimation%'")71    annee = _n1(con, f"SELECT COUNT(*) FROM listings WHERE {W}"72                     " AND year_built IS NOT NULL AND year_built > 1600")73    sqft = _n1(con, f"SELECT COUNT(*) FROM listings WHERE {W}"74                    " AND area_sqft IS NOT NULL AND area_sqft > 0")75    desc = _n1(con, f"SELECT COUNT(*) FROM listings WHERE {W}"76                    " AND description IS NOT NULL AND LENGTH(description) > 80")7778    kpis = [79        {"id": "enr_tot", "label": "Fiches actives", "value": tot},80        {"id": "enr_vp", "label": "Avec estimation Vrai-Prix",81         "value": _pctof(vp, tot), "unit": "%"},82        {"id": "enr_img", "label": "Photos auditées (cumul)",83         "value": _n1(con, "SELECT COUNT(*) FROM image_audit")},84        {"id": "enr_px", "label": "Points de prix consignés",85         "value": _n1(con, "SELECT COUNT(*) FROM price_log")},86    ]87    bars = [88        {"label": "Géolocalisation", "value": _pctof(geo, tot)},89        {"label": "Quartier (recensement)", "value": _pctof(quart, tot)},90        {"label": "Estimation Vrai-Prix", "value": _pctof(vp, tot)},91        {"label": "Année de construction", "value": _pctof(annee, tot)},92        {"label": "Superficie habitable", "value": _pctof(sqft, tot)},93        {"label": "Description détaillée", "value": _pctof(desc, tot)},94    ]9596    layers: list[list] = []9798    def layer(nom: str, count_fn, desc_: str) -> None:99        try:100            n = count_fn()101            if n:102                layers.append([nom, _fr(n), desc_])103        except Exception:104            pass105106    def _side(db: str, sql: str) -> int:107        c = _ro(db)108        if c is None:109            return 0110        try:111            return c.execute(sql).fetchone()[0] or 0112        finally:113            c.close()114115    layer("Vrai-Prix", lambda: vp, "estimations de valeur marchande sur les fiches actives")116    layer("Historique des prix", lambda: _n1(con, "SELECT COUNT(*) FROM price_log"),117          "points de prix consignés depuis la première capture")118    layer("Audit des photos", lambda: _n1(con, "SELECT COUNT(*) FROM image_audit"),119          "images vérifiées (dimension, poids, disponibilité)")120    layer("Taux hypothécaires", lambda: _side("mortgage.db",121          "SELECT COUNT(*) FROM rate_observations"),122          "produits hypothécaires suivis en continu (multibanques)")123    layer("Qualité de l'air", lambda: _side("air.db",124          "SELECT COUNT(DISTINCT station) FROM air_stats"),125          "stations de mesure (RSQAQ) rattachées aux fiches")126    layer("Zones inondables", lambda: _side("inondation.db", "SELECT COUNT(*) FROM zi"),127          "polygones officiels vérifiés au survol de chaque fiche")128    layer("Commerces à proximité", lambda: _side("commerces.db",129          "SELECT COUNT(*) FROM commerces_cache"),130          "cellules de commerces essentiels en cache")131    layer("Hydro-Québec", lambda: _side("hydro.db", "SELECT COUNT(*) FROM hydro_cache"),132          "estimations de coût d'électricité par adresse")133    layer("Registre des loyers", lambda: _side("rdl.db",134          "SELECT COUNT(*) FROM rdl_housings"),135          "loyers déclarés (contexte locatif des plex)")136137    def _annuaire(f: str) -> int:138        p = DATA / f139        return len(json.loads(p.read_text(encoding="utf-8")).get("movers", [])) if p.exists() else 0140141    layer("Annuaire des déménageurs", lambda: _annuaire("demenageurs.json"),142          "entreprises de déménagement répertoriées (/demenageurs)")143    layer("Annuaire des inspecteurs", lambda: _annuaire("inspecteurs.json"),144          "inspecteurs en bâtiment répertoriés (/inspecteurs)")145146    table = {"id": "enr_couches",147             "title": "Couches de données branchées sur chaque fiche",148             "columns": ["Couche", "Volume", "Description"],149             "rows": layers}150151    return {152        "id": "enrichissement",153        "title": "Enrichissement des fiches — ce qu'Immo-Ka ajoute à l'annonce",154        "subtitle": "Chaque propriété agrégée est enrichie automatiquement : "155                    "géocodage, quartier de recensement, estimation Vrai-Prix, "156                    "audit des photos et couches du territoire.",157        "kpis": [k for k in kpis if k.get("value") is not None],158        "breakdowns": [{"id": "enr_couv",159                        "title": "Couverture des enrichissements (% des fiches actives)",160                        "kind": "bar", "items": bars}],161        "tables": [table] if layers else [],162    }163164165# --- Panneau : Vrai-Prix ---------------------------------------------------------166def _panel_vraiprix(con) -> dict | None:167    W = _active_where(con)168    rows = con.execute(169        f"SELECT price, CAST(json_extract(vraiprix, '$.value') AS REAL) vp, city"170        f" FROM listings WHERE {W} AND vraiprix LIKE '%estimation%'"171        f" AND price > 10000").fetchall()172    deltas: list[tuple[float, str]] = []173    for r in rows:174        if r["vp"] and r["vp"] > 0:175            d = (r["price"] - r["vp"]) / r["vp"] * 100.0176            if -80.0 <= d <= 300.0:177                deltas.append((d, r["city"] or ""))178    if len(deltas) < 100:179        return None180    ds = [d for d, _ in deltas]181    sur = sum(1 for d in ds if d > 10)182    juste = sum(1 for d in ds if -5 <= d <= 10)183    sous = sum(1 for d in ds if d < -5)184185    kpis = [186        {"id": "vp_n", "label": "Propriétés avec estimation", "value": len(ds)},187        {"id": "vp_med", "label": "Écart médian demandé vs estimé",188         "value": round(median(ds), 1), "unit": "%"},189        {"id": "vp_sur", "label": "Au-dessus de l'estimation (> +10 %)",190         "value": _pctof(sur, len(ds)), "unit": "%"},191        {"id": "vp_juste", "label": "Prix juste (−5 % à +10 %)",192         "value": _pctof(juste, len(ds)), "unit": "%"},193    ]194    donut = [{"label": "Au-dessus (> +10 %)", "value": sur},195             {"label": "Prix juste (−5 à +10 %)", "value": juste},196             {"label": "Sous l'estimation (< −5 %)", "value": sous}]197198    bins = []199    lo = -40200    while lo < 80:201        n = sum(1 for d in ds if lo <= d < lo + 10)202        bins.append({"label": f"{lo}", "value": n})203        lo += 10204205    byc: dict[str, list[float]] = {}206    for d, c in deltas:207        if c:208            byc.setdefault(c, []).append(d)209    top = sorted(((c, v) for c, v in byc.items() if len(v) >= 50),210                 key=lambda kv: -median(kv[1]))[:12]211    table = {"id": "vp_villes",212             "title": "Villes où les prix demandés dépassent le plus "213                      "l'estimation (min. 50 fiches)",214             "columns": ["Ville", "Fiches estimées", "Écart médian"],215             "rows": [[c, len(v), f"{median(v):+.1f} %"] for c, v in top]}216217    return {218        "id": "vraiprix",219        "title": "Vrai-Prix — le prix demandé est-il le bon prix ?",220        "subtitle": "Chaque propriété est comparée à son estimation Vrai-Prix "221                    "(modèle d'évaluation du Groupe KA).",222        "kpis": kpis,223        "breakdowns": [{"id": "vp_verdicts",224                        "title": "Prix demandé vs estimation Vrai-Prix",225                        "kind": "donut", "items": donut}],226        "distributions": [{"id": "vp_hist",227                           "title": "Distribution des écarts demandé vs estimé (%)",228                           "unit": "%", "bins": bins}],229        "tables": [table] if top else [],230    }231232233# --- Panneau : historique des prix -----------------------------------------------234def _panel_historique(con) -> dict | None:235    rows = con.execute(236        "SELECT uid, ts, price FROM price_log ORDER BY uid, ts").fetchall()237    if len(rows) < 200:238        return None239    drops: list[float] = []240    hikes: list[float] = []241    changed: set[str] = set()242    drops_abs: list[float] = []243    prev_uid, prev_price = None, None244    for r in rows:245        if r["uid"] == prev_uid and prev_price and r["price"] and r["price"] != prev_price:246            pct = (r["price"] - prev_price) / prev_price * 100.0247            if -60.0 <= pct <= 120.0:248                (drops if pct < 0 else hikes).append(pct)249                if pct < 0:250                    drops_abs.append(prev_price - r["price"])251                changed.add(r["uid"])252        prev_uid, prev_price = r["uid"], r["price"]253254    kpis = [255        {"id": "px_pts", "label": "Points de prix consignés", "value": len(rows)},256        {"id": "px_chg", "label": "Propriétés avec changement de prix",257         "value": len(changed)},258        {"id": "px_drop", "label": "Baisses de prix détectées", "value": len(drops)},259        {"id": "px_dmed", "label": "Baisse médiane",260         "value": round(median(drops_abs)) if drops_abs else None, "unit": "$"},261    ]262    donut = [{"label": "Baisses", "value": len(drops)},263             {"label": "Hausses", "value": len(hikes)}]264    return {265        "id": "historique_prix",266        "title": "Historique des prix — chaque propriété est suivie dans le temps",267        "subtitle": "Immo-Ka consigne le prix demandé à chaque synchronisation : "268                    "les baisses et hausses apparaissent sur la fiche.",269        "kpis": [k for k in kpis if k.get("value") is not None],270        "breakdowns": [{"id": "px_sens", "title": "Changements de prix détectés",271                        "kind": "donut", "items": donut}],272    }273274275# --- Panneau : taux hypothécaires -------------------------------------------------276def _panel_hypotheques() -> dict | None:277    con = _ro("mortgage.db")278    if con is None:279        return None280    try:281        rows = con.execute(282            "SELECT institution, product_name, rate_type, term_months, kind,"283            " rate FROM rate_observations WHERE rate IS NOT NULL"284            " AND (valid_to IS NULL OR valid_to = '')").fetchall()285        if not rows:286            rows = con.execute(287                "SELECT institution, product_name, rate_type, term_months,"288                " kind, rate FROM rate_observations"289                " WHERE rate IS NOT NULL").fetchall()290    finally:291        con.close()292    if len(rows) < 20:293        return None294295    inst = {r["institution"] for r in rows}296    f5 = [r for r in rows if r["rate_type"] == "fixed" and r["term_months"] == 60]297    v5 = [r for r in rows if r["rate_type"] in ("variable", "adjustable")298          and r["term_months"] == 60]299    best_f5 = min(f5, key=lambda r: r["rate"]) if f5 else None300    best_v5 = min(v5, key=lambda r: r["rate"]) if v5 else None301302    kpis = [303        {"id": "mt_prod", "label": "Produits hypothécaires suivis", "value": len(rows)},304        {"id": "mt_inst", "label": "Institutions financières", "value": len(inst)},305        {"id": "mt_f5", "label": "Meilleur 5 ans fixe",306         "value": round(best_f5["rate"], 2) if best_f5 else None, "unit": "%"},307        {"id": "mt_v5", "label": "Meilleur 5 ans variable",308         "value": round(best_v5["rate"], 2) if best_v5 else None, "unit": "%"},309    ]310311    best_by_inst: dict[str, float] = {}312    for r in f5:313        cur = best_by_inst.get(r["institution"])314        if cur is None or r["rate"] < cur:315            best_by_inst[r["institution"]] = r["rate"]316    bars = sorted(({"label": k, "value": round(v, 2)}317                   for k, v in best_by_inst.items()),318                  key=lambda x: x["value"])[:12]319320    byterm: dict[int, list[float]] = {}321    for r in rows:322        if r["rate_type"] == "fixed" and r["term_months"] in (12, 24, 36, 48, 60, 84, 120):323            byterm.setdefault(r["term_months"], []).append(r["rate"])324    tbars = [{"label": f"{t // 12} an{'s' if t >= 24 else ''}",325              "value": round(min(v), 2)}326             for t, v in sorted(byterm.items())]327328    return {329        "id": "hypotheques",330        "title": "Taux hypothécaires — collecte réelle multibanques",331        "subtitle": "Les taux publiés par les institutions sont collectés en "332                    "continu et alimentent le calculateur des fiches "333                    "(/taux-hypothecaires).",334        "kpis": [k for k in kpis if k.get("value") is not None],335        "breakdowns": ([{"id": "mt_inst_bar",336                         "title": "Meilleur taux fixe 5 ans par institution (%)",337                         "kind": "bar", "items": bars}] if bars else [])338                      + ([{"id": "mt_terms",339                           "title": "Meilleur taux fixe par terme (%)",340                           "kind": "bar", "items": tbars}] if tbars else []),341    }342343344# --- Panneau : annuaires (déménageurs + inspecteurs) ------------------------------345def _panel_annuaires() -> dict | None:346    def _load(f: str) -> list[dict]:347        p = DATA / f348        if not p.exists():349            return []350        return json.loads(p.read_text(encoding="utf-8")).get("movers", [])351352    dem = _load("demenageurs.json")353    insp = _load("inspecteurs.json")354    if len(dem) + len(insp) < 50:355        return None356357    def _note(entries: list[dict]) -> float | None:358        rated = [m["rating"] for m in entries if m.get("rating")]359        return round(median(rated), 1) if len(rated) >= 30 else None360361    kpis = [362        {"id": "ann_dem", "label": "Déménageurs répertoriés", "value": len(dem) or None},363        {"id": "ann_insp", "label": "Inspecteurs en bâtiment", "value": len(insp) or None},364        {"id": "ann_dnote", "label": "Note médiane — déménageurs",365         "value": _note(dem), "unit": "/5"},366        {"id": "ann_inote", "label": "Note médiane — inspecteurs",367         "value": _note(insp), "unit": "/5"},368    ]369    byreg: dict[str, int] = {}370    for m in insp:371        byreg[m["region"]] = byreg.get(m["region"], 0) + 1372    bars = sorted(({"label": k, "value": v} for k, v in byreg.items()),373                  key=lambda x: -x["value"])[:12]374    return {375        "id": "annuaires",376        "title": "Annuaires — déménageurs et inspecteurs en bâtiment",377        "subtitle": "Deux annuaires provinciaux compilés par le Groupe KA "378                    "pour accompagner l'achat : /demenageurs et /inspecteurs.",379        "kpis": [k for k in kpis if k.get("value") is not None],380        "breakdowns": [{"id": "ann_insp_reg",381                        "title": "Inspecteurs en bâtiment par région (top 12)",382                        "kind": "bar", "items": bars}] if bars else [],383    }384385386def panels(con: sqlite3.Connection) -> list[dict]:387    """Panneaux « enrichissements de la fiche » (cache 30 min)."""388    hit = _CACHE.get("panels")389    if hit and time.time() - hit[0] < _TTL:390        return hit[1]391    out = []392    for fn in (lambda: _panel_enrichissement(con),393               lambda: _panel_vraiprix(con),394               lambda: _panel_historique(con),395               _panel_hypotheques,396               _panel_annuaires):397        try:398            p = fn()399            if p and (p.get("kpis") or p.get("tables") or p.get("breakdowns")):400                out.append(p)401        except Exception:402            continue403    _CACHE["panels"] = (time.time(), out)404    return out405