# ----------------------------------------------------------------------------- # Immo-Ka — Agrégateur de maisons à vendre (province de Québec) # Auteur : Simon-Pierre Boucher — contact@spboucher.ai # statsfiche.py : panneaux « enrichissements de la fiche » de l'onglet # Statistiques — tout ce qu'Immo-Ka ajoute par-dessus l'annonce brute : # · couverture des enrichissements (géocodage, quartier, Vrai-Prix, # caractéristiques, audit photos) + couches de données branchées ; # · Vrai-Prix (écarts prix demandé vs estimation, verdicts, villes) ; # · Historique des prix (baisses/hausses) ; # · Taux hypothécaires (meilleurs taux réels multibanques) ; # · Annuaires (déménageurs et inspecteurs en bâtiment). # Même contrat de rendu que statsextra.py (kpis/breakdowns/distributions/ # tables, rendu générique du kit stats) — un panneau absent si sa source # ne répond pas. Cache 30 min. # ----------------------------------------------------------------------------- from __future__ import annotations import json import sqlite3 import time from pathlib import Path from statistics import median ROOT = Path(__file__).resolve().parent.parent DATA = ROOT / "data" _CACHE: dict[str, tuple[float, list]] = {} _TTL = 1800 def _ro(name: str) -> sqlite3.Connection | None: p = DATA / name if not p.exists(): return None con = sqlite3.connect(f"file:{p}?mode=ro", uri=True) con.row_factory = sqlite3.Row return con def _active_where(con) -> str: cols = {r[1] for r in con.execute("PRAGMA table_info(listings)")} w = "active=1" if "dup_hidden" in cols: w += " AND (dup_hidden IS NULL OR dup_hidden=0)" return w def _n1(con, sql: str, args: tuple = ()) -> int: return con.execute(sql, args).fetchone()[0] or 0 def _fr(n: float) -> str: return f"{round(n):,}".replace(",", " ") def _pctof(part: int, tot: int) -> float: return round(100.0 * part / tot, 1) if tot else 0.0 # --- Panneau : couverture des enrichissements --------------------------------- def _panel_enrichissement(con) -> dict | None: W = _active_where(con) tot = _n1(con, f"SELECT COUNT(*) FROM listings WHERE {W}") if tot < 100: return None geo = _n1(con, f"SELECT COUNT(*) FROM listings WHERE {W}" " AND lat IS NOT NULL AND lng IS NOT NULL") quart = _n1(con, f"SELECT COUNT(*) FROM listings WHERE {W}" " AND dauid IS NOT NULL AND dauid<>''") vp = _n1(con, f"SELECT COUNT(*) FROM listings WHERE {W}" " AND vraiprix LIKE '%estimation%'") annee = _n1(con, f"SELECT COUNT(*) FROM listings WHERE {W}" " AND year_built IS NOT NULL AND year_built > 1600") sqft = _n1(con, f"SELECT COUNT(*) FROM listings WHERE {W}" " AND area_sqft IS NOT NULL AND area_sqft > 0") desc = _n1(con, f"SELECT COUNT(*) FROM listings WHERE {W}" " AND description IS NOT NULL AND LENGTH(description) > 80") kpis = [ {"id": "enr_tot", "label": "Fiches actives", "value": tot}, {"id": "enr_vp", "label": "Avec estimation Vrai-Prix", "value": _pctof(vp, tot), "unit": "%"}, {"id": "enr_img", "label": "Photos auditées (cumul)", "value": _n1(con, "SELECT COUNT(*) FROM image_audit")}, {"id": "enr_px", "label": "Points de prix consignés", "value": _n1(con, "SELECT COUNT(*) FROM price_log")}, ] bars = [ {"label": "Géolocalisation", "value": _pctof(geo, tot)}, {"label": "Quartier (recensement)", "value": _pctof(quart, tot)}, {"label": "Estimation Vrai-Prix", "value": _pctof(vp, tot)}, {"label": "Année de construction", "value": _pctof(annee, tot)}, {"label": "Superficie habitable", "value": _pctof(sqft, tot)}, {"label": "Description détaillée", "value": _pctof(desc, tot)}, ] layers: list[list] = [] def layer(nom: str, count_fn, desc_: str) -> None: try: n = count_fn() if n: layers.append([nom, _fr(n), desc_]) except Exception: pass def _side(db: str, sql: str) -> int: c = _ro(db) if c is None: return 0 try: return c.execute(sql).fetchone()[0] or 0 finally: c.close() layer("Vrai-Prix", lambda: vp, "estimations de valeur marchande sur les fiches actives") layer("Historique des prix", lambda: _n1(con, "SELECT COUNT(*) FROM price_log"), "points de prix consignés depuis la première capture") layer("Audit des photos", lambda: _n1(con, "SELECT COUNT(*) FROM image_audit"), "images vérifiées (dimension, poids, disponibilité)") layer("Taux hypothécaires", lambda: _side("mortgage.db", "SELECT COUNT(*) FROM rate_observations"), "produits hypothécaires suivis en continu (multibanques)") layer("Qualité de l'air", lambda: _side("air.db", "SELECT COUNT(DISTINCT station) FROM air_stats"), "stations de mesure (RSQAQ) rattachées aux fiches") layer("Zones inondables", lambda: _side("inondation.db", "SELECT COUNT(*) FROM zi"), "polygones officiels vérifiés au survol de chaque fiche") layer("Commerces à proximité", lambda: _side("commerces.db", "SELECT COUNT(*) FROM commerces_cache"), "cellules de commerces essentiels en cache") layer("Hydro-Québec", lambda: _side("hydro.db", "SELECT COUNT(*) FROM hydro_cache"), "estimations de coût d'électricité par adresse") layer("Registre des loyers", lambda: _side("rdl.db", "SELECT COUNT(*) FROM rdl_housings"), "loyers déclarés (contexte locatif des plex)") def _annuaire(f: str) -> int: p = DATA / f return len(json.loads(p.read_text(encoding="utf-8")).get("movers", [])) if p.exists() else 0 layer("Annuaire des déménageurs", lambda: _annuaire("demenageurs.json"), "entreprises de déménagement répertoriées (/demenageurs)") layer("Annuaire des inspecteurs", lambda: _annuaire("inspecteurs.json"), "inspecteurs en bâtiment répertoriés (/inspecteurs)") table = {"id": "enr_couches", "title": "Couches de données branchées sur chaque fiche", "columns": ["Couche", "Volume", "Description"], "rows": layers} return { "id": "enrichissement", "title": "Enrichissement des fiches — ce qu'Immo-Ka ajoute à l'annonce", "subtitle": "Chaque propriété agrégée est enrichie automatiquement : " "géocodage, quartier de recensement, estimation Vrai-Prix, " "audit des photos et couches du territoire.", "kpis": [k for k in kpis if k.get("value") is not None], "breakdowns": [{"id": "enr_couv", "title": "Couverture des enrichissements (% des fiches actives)", "kind": "bar", "items": bars}], "tables": [table] if layers else [], } # --- Panneau : Vrai-Prix --------------------------------------------------------- def _panel_vraiprix(con) -> dict | None: W = _active_where(con) rows = con.execute( f"SELECT price, CAST(json_extract(vraiprix, '$.value') AS REAL) vp, city" f" FROM listings WHERE {W} AND vraiprix LIKE '%estimation%'" f" AND price > 10000").fetchall() deltas: list[tuple[float, str]] = [] for r in rows: if r["vp"] and r["vp"] > 0: d = (r["price"] - r["vp"]) / r["vp"] * 100.0 if -80.0 <= d <= 300.0: deltas.append((d, r["city"] or "")) if len(deltas) < 100: return None ds = [d for d, _ in deltas] sur = sum(1 for d in ds if d > 10) juste = sum(1 for d in ds if -5 <= d <= 10) sous = sum(1 for d in ds if d < -5) kpis = [ {"id": "vp_n", "label": "Propriétés avec estimation", "value": len(ds)}, {"id": "vp_med", "label": "Écart médian demandé vs estimé", "value": round(median(ds), 1), "unit": "%"}, {"id": "vp_sur", "label": "Au-dessus de l'estimation (> +10 %)", "value": _pctof(sur, len(ds)), "unit": "%"}, {"id": "vp_juste", "label": "Prix juste (−5 % à +10 %)", "value": _pctof(juste, len(ds)), "unit": "%"}, ] donut = [{"label": "Au-dessus (> +10 %)", "value": sur}, {"label": "Prix juste (−5 à +10 %)", "value": juste}, {"label": "Sous l'estimation (< −5 %)", "value": sous}] bins = [] lo = -40 while lo < 80: n = sum(1 for d in ds if lo <= d < lo + 10) bins.append({"label": f"{lo}", "value": n}) lo += 10 byc: dict[str, list[float]] = {} for d, c in deltas: if c: byc.setdefault(c, []).append(d) top = sorted(((c, v) for c, v in byc.items() if len(v) >= 50), key=lambda kv: -median(kv[1]))[:12] table = {"id": "vp_villes", "title": "Villes où les prix demandés dépassent le plus " "l'estimation (min. 50 fiches)", "columns": ["Ville", "Fiches estimées", "Écart médian"], "rows": [[c, len(v), f"{median(v):+.1f} %"] for c, v in top]} return { "id": "vraiprix", "title": "Vrai-Prix — le prix demandé est-il le bon prix ?", "subtitle": "Chaque propriété est comparée à son estimation Vrai-Prix " "(modèle d'évaluation du Groupe KA).", "kpis": kpis, "breakdowns": [{"id": "vp_verdicts", "title": "Prix demandé vs estimation Vrai-Prix", "kind": "donut", "items": donut}], "distributions": [{"id": "vp_hist", "title": "Distribution des écarts demandé vs estimé (%)", "unit": "%", "bins": bins}], "tables": [table] if top else [], } # --- Panneau : historique des prix ----------------------------------------------- def _panel_historique(con) -> dict | None: rows = con.execute( "SELECT uid, ts, price FROM price_log ORDER BY uid, ts").fetchall() if len(rows) < 200: return None drops: list[float] = [] hikes: list[float] = [] changed: set[str] = set() drops_abs: list[float] = [] prev_uid, prev_price = None, None for r in rows: if r["uid"] == prev_uid and prev_price and r["price"] and r["price"] != prev_price: pct = (r["price"] - prev_price) / prev_price * 100.0 if -60.0 <= pct <= 120.0: (drops if pct < 0 else hikes).append(pct) if pct < 0: drops_abs.append(prev_price - r["price"]) changed.add(r["uid"]) prev_uid, prev_price = r["uid"], r["price"] kpis = [ {"id": "px_pts", "label": "Points de prix consignés", "value": len(rows)}, {"id": "px_chg", "label": "Propriétés avec changement de prix", "value": len(changed)}, {"id": "px_drop", "label": "Baisses de prix détectées", "value": len(drops)}, {"id": "px_dmed", "label": "Baisse médiane", "value": round(median(drops_abs)) if drops_abs else None, "unit": "$"}, ] donut = [{"label": "Baisses", "value": len(drops)}, {"label": "Hausses", "value": len(hikes)}] return { "id": "historique_prix", "title": "Historique des prix — chaque propriété est suivie dans le temps", "subtitle": "Immo-Ka consigne le prix demandé à chaque synchronisation : " "les baisses et hausses apparaissent sur la fiche.", "kpis": [k for k in kpis if k.get("value") is not None], "breakdowns": [{"id": "px_sens", "title": "Changements de prix détectés", "kind": "donut", "items": donut}], } # --- Panneau : taux hypothécaires ------------------------------------------------- def _panel_hypotheques() -> dict | None: con = _ro("mortgage.db") if con is None: return None try: rows = con.execute( "SELECT institution, product_name, rate_type, term_months, kind," " rate FROM rate_observations WHERE rate IS NOT NULL" " AND (valid_to IS NULL OR valid_to = '')").fetchall() if not rows: rows = con.execute( "SELECT institution, product_name, rate_type, term_months," " kind, rate FROM rate_observations" " WHERE rate IS NOT NULL").fetchall() finally: con.close() if len(rows) < 20: return None inst = {r["institution"] for r in rows} f5 = [r for r in rows if r["rate_type"] == "fixed" and r["term_months"] == 60] v5 = [r for r in rows if r["rate_type"] in ("variable", "adjustable") and r["term_months"] == 60] best_f5 = min(f5, key=lambda r: r["rate"]) if f5 else None best_v5 = min(v5, key=lambda r: r["rate"]) if v5 else None kpis = [ {"id": "mt_prod", "label": "Produits hypothécaires suivis", "value": len(rows)}, {"id": "mt_inst", "label": "Institutions financières", "value": len(inst)}, {"id": "mt_f5", "label": "Meilleur 5 ans fixe", "value": round(best_f5["rate"], 2) if best_f5 else None, "unit": "%"}, {"id": "mt_v5", "label": "Meilleur 5 ans variable", "value": round(best_v5["rate"], 2) if best_v5 else None, "unit": "%"}, ] best_by_inst: dict[str, float] = {} for r in f5: cur = best_by_inst.get(r["institution"]) if cur is None or r["rate"] < cur: best_by_inst[r["institution"]] = r["rate"] bars = sorted(({"label": k, "value": round(v, 2)} for k, v in best_by_inst.items()), key=lambda x: x["value"])[:12] byterm: dict[int, list[float]] = {} for r in rows: if r["rate_type"] == "fixed" and r["term_months"] in (12, 24, 36, 48, 60, 84, 120): byterm.setdefault(r["term_months"], []).append(r["rate"]) tbars = [{"label": f"{t // 12} an{'s' if t >= 24 else ''}", "value": round(min(v), 2)} for t, v in sorted(byterm.items())] return { "id": "hypotheques", "title": "Taux hypothécaires — collecte réelle multibanques", "subtitle": "Les taux publiés par les institutions sont collectés en " "continu et alimentent le calculateur des fiches " "(/taux-hypothecaires).", "kpis": [k for k in kpis if k.get("value") is not None], "breakdowns": ([{"id": "mt_inst_bar", "title": "Meilleur taux fixe 5 ans par institution (%)", "kind": "bar", "items": bars}] if bars else []) + ([{"id": "mt_terms", "title": "Meilleur taux fixe par terme (%)", "kind": "bar", "items": tbars}] if tbars else []), } # --- Panneau : annuaires (déménageurs + inspecteurs) ------------------------------ def _panel_annuaires() -> dict | None: def _load(f: str) -> list[dict]: p = DATA / f if not p.exists(): return [] return json.loads(p.read_text(encoding="utf-8")).get("movers", []) dem = _load("demenageurs.json") insp = _load("inspecteurs.json") if len(dem) + len(insp) < 50: return None def _note(entries: list[dict]) -> float | None: rated = [m["rating"] for m in entries if m.get("rating")] return round(median(rated), 1) if len(rated) >= 30 else None kpis = [ {"id": "ann_dem", "label": "Déménageurs répertoriés", "value": len(dem) or None}, {"id": "ann_insp", "label": "Inspecteurs en bâtiment", "value": len(insp) or None}, {"id": "ann_dnote", "label": "Note médiane — déménageurs", "value": _note(dem), "unit": "/5"}, {"id": "ann_inote", "label": "Note médiane — inspecteurs", "value": _note(insp), "unit": "/5"}, ] byreg: dict[str, int] = {} for m in insp: byreg[m["region"]] = byreg.get(m["region"], 0) + 1 bars = sorted(({"label": k, "value": v} for k, v in byreg.items()), key=lambda x: -x["value"])[:12] return { "id": "annuaires", "title": "Annuaires — déménageurs et inspecteurs en bâtiment", "subtitle": "Deux annuaires provinciaux compilés par le Groupe KA " "pour accompagner l'achat : /demenageurs et /inspecteurs.", "kpis": [k for k in kpis if k.get("value") is not None], "breakdowns": [{"id": "ann_insp_reg", "title": "Inspecteurs en bâtiment par région (top 12)", "kind": "bar", "items": bars}] if bars else [], } def panels(con: sqlite3.Connection) -> list[dict]: """Panneaux « enrichissements de la fiche » (cache 30 min).""" hit = _CACHE.get("panels") if hit and time.time() - hit[0] < _TTL: return hit[1] out = [] for fn in (lambda: _panel_enrichissement(con), lambda: _panel_vraiprix(con), lambda: _panel_historique(con), _panel_hypotheques, _panel_annuaires): try: p = fn() if p and (p.get("kpis") or p.get("tables") or p.get("breakdowns")): out.append(p) except Exception: continue _CACHE["panels"] = (time.time(), out) return out