# ----------------------------------------------------------------------------- # Lou-Ka — Agrégateur de logements à louer (province de Québec) # Auteur : Simon-Pierre Boucher — contact@spboucher.ai # statsfiche.py : panneaux « enrichissements de la fiche » de l'onglet # Statistiques — tout ce que Lou-Ka ajoute par-dessus l'annonce brute : # · couverture des enrichissements (géocodage, quartier, KA Scores, # juste valeur, immeuble, photos) + couches de données branchées ; # · KA Scores (piliers, distribution, villes) ; # · Juste valeur (verdicts, écarts, villes surchauffées) ; # · Historique des prix (baisses/hausses) + vie des annonces ; # · Court terme (chalets et hébergements, CITQ, notes) ; # · Gestionnaires (fiches Google, notes) — masqué tant que < 10 notés ; # · Annuaire des déménageurs du Québec. # 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 ACTIVE = "active=1 AND published=1 AND dup_of IS NULL" 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 _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: tot = _n1(con, f"SELECT COUNT(*) FROM listings WHERE {ACTIVE}") if tot < 100: return None geo = _n1(con, f"SELECT COUNT(*) FROM listings WHERE {ACTIVE} " "AND lat IS NOT NULL AND lng IS NOT NULL") quart = _n1(con, f"SELECT COUNT(*) FROM listings WHERE {ACTIVE} " "AND dauid IS NOT NULL AND dauid<>''") ks = _n1(con, "SELECT COUNT(*) FROM listings l JOIN kascores k" " ON l.coord_key=k.coord_key WHERE l.active=1" " AND l.published=1 AND l.dup_of IS NULL") fv = _n1(con, "SELECT COUNT(*) FROM fairvalue f JOIN listings l ON f.uid=l.uid " "WHERE l.active=1 AND l.published=1 AND l.dup_of IS NULL") bld = _n1(con, f"SELECT COUNT(*) FROM listings WHERE {ACTIVE} " "AND building_key IS NOT NULL AND building_key<>''") imgok = _n1(con, f"SELECT COUNT(*) FROM listings WHERE {ACTIVE} AND images_ok=1") compl = con.execute( f"SELECT ROUND(AVG(completeness),1) FROM listings WHERE {ACTIVE}" ).fetchone()[0] kpis = [ {"id": "enr_tot", "label": "Fiches actives publiées", "value": tot}, {"id": "enr_compl", "label": "Complétude moyenne", "value": compl, "unit": "%"}, {"id": "enr_bld", "label": "Immeubles au passeport", "value": _n1(con, "SELECT COUNT(*) FROM buildings")}, {"id": "enr_img", "label": "Photos vérifiées (cumul)", "value": _n1(con, "SELECT COUNT(*) FROM image_checks")}, ] bars = [ {"label": "Géolocalisation", "value": _pctof(geo, tot)}, {"label": "Quartier (recensement)", "value": _pctof(quart, tot)}, {"label": "KA Scores", "value": _pctof(ks, tot)}, {"label": "Juste valeur", "value": _pctof(fv, tot)}, {"label": "Immeuble rattaché", "value": _pctof(bld, tot)}, {"label": "Photos auditées OK", "value": _pctof(imgok, tot)}, ] # couches de données branchées sur la fiche (chaque ligne est optionnelle) 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 layer("KA Scores", lambda: _n1(con, "SELECT COUNT(*) FROM kascores"), "emplacements notés (marche, transit, vélo, calme, services)") layer("Juste valeur", lambda: _n1(con, "SELECT COUNT(*) FROM fairvalue"), "évaluations de loyer (modèle comparables Lou-Ka)") layer("Historique des prix", lambda: _n1(con, "SELECT COUNT(*) FROM price_log"), "points de prix consignés depuis la première capture") layer("Immeubles", lambda: _n1(con, "SELECT COUNT(*) FROM buildings"), "passeports d'immeuble (historique par adresse)") 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("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("Dossiers TAL", lambda: _side("tal.db", "SELECT COUNT(*) FROM tal_lookup"), "adresses vérifiées au Tribunal administratif du logement") layer("Registre des loyers", lambda: _side("rdl.db", "SELECT COUNT(*) FROM rdl_housings"), "loyers réellement déclarés (Vivre en ville)") 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 que Lou-Ka ajoute à l'annonce", "subtitle": "Chaque annonce agrégée est enrichie automatiquement : " "géocodage, quartier de recensement, KA Scores, juste " "valeur, passeport d'immeuble, 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 : KA Scores ------------------------------------------------------- def _panel_kascores(con) -> dict | None: rows = con.execute( "SELECT k.walk, k.transit, k.bike, k.calme, k.services, k.global, l.city" " FROM listings l JOIN kascores k ON l.coord_key=k.coord_key" " WHERE l.active=1 AND l.published=1 AND l.dup_of IS NULL" " AND k.global IS NOT NULL").fetchall() if len(rows) < 100: return None glob = [r["global"] for r in rows] kpis = [ {"id": "ks_n", "label": "Fiches avec KA Scores", "value": len(rows)}, {"id": "ks_med", "label": "Score global médian", "value": round(median(glob), 1), "unit": "/100"}, {"id": "ks_70", "label": "Score ≥ 70 (excellents secteurs)", "value": _pctof(sum(1 for g in glob if g >= 70), len(glob)), "unit": "%"}, {"id": "ks_loc", "label": "Emplacements notés (cumul)", "value": _n1(con, "SELECT COUNT(*) FROM kascores")}, ] piliers = [("walk", "Marche"), ("transit", "Transport"), ("bike", "Vélo"), ("calme", "Calme"), ("services", "Services")] bars = [] for col, label in piliers: vals = [r[col] for r in rows if r[col] is not None] if vals: bars.append({"label": label, "value": round(median(vals), 1)}) bins = [] for lo in range(0, 100, 10): n = sum(1 for g in glob if lo <= g < lo + 10 or (lo == 90 and g == 100)) bins.append({"label": str(lo), "value": n}) byc: dict[str, list[float]] = {} for r in rows: if r["city"]: byc.setdefault(r["city"], []).append(r["global"]) top = sorted(((c, v) for c, v in byc.items() if len(v) >= 30), key=lambda kv: -median(kv[1]))[:12] table = {"id": "ks_villes", "title": "Meilleurs scores globaux par ville (min. 30 fiches)", "columns": ["Ville", "Fiches notées", "Score global médian"], "rows": [[c, len(v), f"{median(v):.1f}"] for c, v in top]} return { "id": "kascores", "title": "KA Scores — la qualité du secteur, chiffrée", "subtitle": "Cinq piliers calculés sur données ouvertes (OSM, GTFS…) " "pour chaque emplacement : marche, transport, vélo, calme " "et services.", "kpis": kpis, "breakdowns": [{"id": "ks_piliers", "title": "Score médian par pilier (fiches actives)", "kind": "bar", "items": bars}] if bars else [], "distributions": [{"id": "ks_hist", "title": "Distribution du score global", "unit": "pts", "bins": bins}], "tables": [table] if top else [], } # --- Panneau : juste valeur ---------------------------------------------------- def _panel_fairvalue(con) -> dict | None: rows = con.execute( "SELECT f.deviation, f.verdict, l.city FROM fairvalue f" " JOIN listings l ON f.uid=l.uid" " WHERE l.active=1 AND l.published=1 AND l.dup_of IS NULL" " AND f.deviation IS NOT NULL").fetchall() if len(rows) < 100: return None devs = [r["deviation"] for r in rows] # heuristique d'unité : fraction (0.08) vs pourcentage (8.0) scale = 100.0 if median(abs(d) for d in devs) < 1.5 else 1.0 devs_pct = [d * scale for d in devs] labels = {"marche": "Prix du marché", "sous": "Sous le marché", "sur": "Au-dessus du marché"} counts: dict[str, int] = {} for r in rows: if r["verdict"]: counts[labels.get(r["verdict"], r["verdict"])] = \ counts.get(labels.get(r["verdict"], r["verdict"]), 0) + 1 kpis = [ {"id": "fv_n", "label": "Loyers évalués (fiches actives)", "value": len(rows)}, {"id": "fv_med", "label": "Écart médian au loyer estimé", "value": round(median(devs_pct), 1), "unit": "%"}, {"id": "fv_sur", "label": "Au-dessus du marché", "value": _pctof(counts.get("Au-dessus du marché", 0), len(rows)), "unit": "%"}, {"id": "fv_sous", "label": "Sous le marché (aubaines)", "value": _pctof(counts.get("Sous le marché", 0), len(rows)), "unit": "%"}, ] bins = [] lo = -40 while lo < 60: n = sum(1 for d in devs_pct if lo <= d < lo + 10) bins.append({"label": f"{lo}", "value": n}) lo += 10 byc: dict[str, list[float]] = {} for r, d in zip(rows, devs_pct): if r["city"]: byc.setdefault(r["city"], []).append(d) top = sorted(((c, v) for c, v in byc.items() if len(v) >= 30), key=lambda kv: -median(kv[1]))[:12] table = {"id": "fv_villes", "title": "Villes où les loyers affichés dépassent le plus la juste " "valeur (min. 30 fiches)", "columns": ["Ville", "Fiches évaluées", "Écart médian"], "rows": [[c, len(v), f"{median(v):+.1f} %"] for c, v in top]} return { "id": "juste_valeur", "title": "Juste valeur — le loyer affiché est-il le bon prix ?", "subtitle": "Chaque fiche est comparée aux logements semblables du " "même secteur (modèle de comparables Lou-Ka).", "kpis": kpis, "breakdowns": [{"id": "fv_verdicts", "title": "Verdicts de juste valeur", "kind": "donut", "items": [{"label": k, "value": v} for k, v in sorted(counts.items(), key=lambda kv: -kv[1])]}] if counts else [], "distributions": [{"id": "fv_hist", "title": "Distribution des écarts au loyer estimé (%)", "unit": "%", "bins": bins}], "tables": [table] if top else [], } # --- Panneau : historique des prix & vie des annonces -------------------------- 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 npts = len(rows) drops: list[float] = [] hikes: list[float] = [] changed: set[str] = set() 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) changed.add(r["uid"]) prev_uid, prev_price = r["uid"], r["price"] kpis = [ {"id": "px_pts", "label": "Points de prix consignés", "value": npts}, {"id": "px_chg", "label": "Annonces avec changement de prix", "value": len(changed)}, {"id": "px_drop", "label": "Baisses de loyer détectées", "value": len(drops)}, {"id": "px_dmed", "label": "Baisse médiane", "value": round(median(drops), 1) if drops else None, "unit": "%"}, ] donut = [{"label": "Baisses", "value": len(drops)}, {"label": "Hausses", "value": len(hikes)}] ev_labels = {"disparition": "Disparition", "reapparition": "Réapparition", "photos": "Photos modifiées", "description": "Description modifiée", "dispo": "Disponibilité modifiée", "inclusions": "Inclusions modifiées", "superficie": "Superficie modifiée"} evs = [{"label": ev_labels.get(r["event"], r["event"]), "value": r["n"]} for r in con.execute( "SELECT event, COUNT(*) n FROM listing_events GROUP BY event" " ORDER BY n DESC")] return { "id": "historique_prix", "title": "Historique des prix — chaque loyer est suivi dans le temps", "subtitle": "Lou-Ka consigne le loyer de chaque annonce à chaque " "synchronisation : baisses, hausses et événements de vie " "de l'annonce apparaissent sur la fiche.", "kpis": [k for k in kpis if k.get("value") is not None], "breakdowns": ([{"id": "px_sens", "title": "Changements de loyer détectés", "kind": "donut", "items": donut}] + ([{"id": "px_events", "title": "Événements de vie des annonces (photos, " "description, disponibilité…)", "kind": "bar", "items": evs}] if evs else [])), } # --- Panneau : court terme ------------------------------------------------------ def _panel_court_terme() -> dict | None: con = _ro("louka_ct.db") if con is None: return None try: tot, nsrc = con.execute( "SELECT COUNT(*), COUNT(DISTINCT source) FROM st_listings" " WHERE active=1").fetchone() if not tot or tot < 100: return None prices = [r[0] for r in con.execute( "SELECT price_night FROM st_listings WHERE active=1" " AND price_night > 20 AND price_night < 10000")] citq = con.execute("SELECT COUNT(*) FROM st_listings WHERE active=1" " AND citq IS NOT NULL AND citq<>''").fetchone()[0] rated = [r[0] for r in con.execute( "SELECT rating FROM st_listings WHERE active=1 AND rating IS NOT NULL")] byreg = {} for r in con.execute( "SELECT region, price_night FROM st_listings WHERE active=1" " AND region IS NOT NULL AND region<>'' AND price_night > 20" " AND price_night < 10000"): byreg.setdefault(r[0], []).append(r[1]) finally: con.close() bars = sorted(([{"label": k, "value": round(median(v))} for k, v in byreg.items() if len(v) >= 100]), key=lambda x: -x["value"])[:12] kpis = [ {"id": "ct_n", "label": "Hébergements court terme actifs", "value": tot}, {"id": "ct_src", "label": "Plateformes agrégées", "value": nsrc}, {"id": "ct_px", "label": "Prix médian par nuit", "value": round(median(prices)) if prices else None, "unit": "$"}, {"id": "ct_citq", "label": "Avec n° d'enregistrement CITQ", "value": _pctof(citq, tot), "unit": "%"}, {"id": "ct_note", "label": "Note médiane des voyageurs", "value": round(median(rated), 2) if len(rated) >= 50 else None, "unit": "/5"}, ] return { "id": "court_terme", "title": "Court terme — chalets et hébergements à la nuit", "subtitle": "La section Court terme agrège les plateformes de location " "de chalets et d'hébergements partout au Québec.", "kpis": [k for k in kpis if k.get("value") is not None], "breakdowns": [{"id": "ct_reg", "title": "Prix médian par nuit selon la région " "(min. 100 annonces)", "kind": "bar", "items": bars}] if bars else [], } # --- Panneau : gestionnaires (masqué tant que < 10 fiches Google notées) -------- def _panel_gestionnaires(con) -> dict | None: rows = con.execute( "SELECT name, gmaps_rating r, gmaps_reviews nrev FROM managers" " WHERE gmaps_rating IS NOT NULL").fetchall() if len(rows) < 10: return None notes = [r["r"] for r in rows] kpis = [ {"id": "mg_n", "label": "Gestionnaires répertoriés", "value": _n1(con, "SELECT COUNT(*) FROM managers")}, {"id": "mg_fiche", "label": "Avec fiche Google notée", "value": len(rows)}, {"id": "mg_med", "label": "Note Google médiane", "value": round(median(notes), 2), "unit": "/5"}, {"id": "mg_avis", "label": "Avis cumulés", "value": sum(r["nrev"] or 0 for r in rows)}, ] top = sorted((r for r in rows if (r["nrev"] or 0) >= 20), key=lambda r: -r["r"])[:12] table = {"id": "mg_top", "title": "Gestionnaires les mieux notés (min. 20 avis)", "columns": ["Gestionnaire", "Note Google", "Avis"], "rows": [[r["name"], f"{r['r']:.1f}", r["nrev"]] for r in top]} return { "id": "gestionnaires", "title": "Gestionnaires immobiliers — réputation Google", "subtitle": "Chaque gestionnaire agrégé est rapproché de sa fiche " "Google Maps : note et avis apparaissent sur ses annonces.", "kpis": kpis, "tables": [table] if top else [], } # --- Panneau : annuaire des déménageurs ----------------------------------------- def _panel_demenageurs() -> dict | None: p = DATA / "demenageurs.json" if not p.exists(): return None doc = json.loads(p.read_text(encoding="utf-8")) movers = doc.get("movers", []) if len(movers) < 50: return None rated = [m["rating"] for m in movers if m.get("rating")] byreg: dict[str, int] = {} for m in movers: 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] kpis = [ {"id": "dem_n", "label": "Déménageurs répertoriés", "value": len(movers)}, {"id": "dem_reg", "label": "Régions couvertes", "value": len(byreg)}, {"id": "dem_web", "label": "Avec site web", "value": _pctof(sum(1 for m in movers if m.get("website")), len(movers)), "unit": "%"}, {"id": "dem_note", "label": "Note Google médiane", "value": round(median(rated), 1) if len(rated) >= 30 else None, "unit": "/5"}, ] return { "id": "annuaire_demenageurs", "title": "Annuaire des déménageurs du Québec", "subtitle": "L'annuaire complet des entreprises de déménagement de la " "province, présenté sur /demenageurs.", "kpis": [k for k in kpis if k.get("value") is not None], "breakdowns": [{"id": "dem_reg_bar", "title": "Déménageurs par région (top 12)", "kind": "bar", "items": bars}], } 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_kascores(con), lambda: _panel_fairvalue(con), lambda: _panel_historique(con), _panel_court_terme, lambda: _panel_gestionnaires(con), _panel_demenageurs): 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