# ----------------------------------------------------------------------------- # Rent-Ka — Rental listings aggregator (Canada, outside Québec) # Author: Simon-Pierre Boucher — contact@spboucher.ai # statsfiche.py : panneaux « enrichissements de la fiche » de l'onglet # Statistiques — tout ce que Rent-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") 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": "Geolocation", "value": _pctof(geo, tot)}, {"label": "KA Scores", "value": _pctof(ks, tot)}, {"label": "Fair value", "value": _pctof(fv, tot)}, {"label": "Building attached", "value": _pctof(bld, tot)}, {"label": "Photos audited 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 Rent-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)") table = {"id": "enr_couches", "title": "Data layers attached to every listing", "columns": ["Layer", "Volume", "Description"], "rows": layers} return { "id": "enrichissement", "title": "Listing enrichment — what Rent-Ka adds to the raw listing", "subtitle": "Every aggregated listing is enriched automatically: " "geocoding, KA Scores, fair value, building passport and " "photo audit.", "kpis": [k for k in kpis if k.get("value") is not None], "breakdowns": [{"id": "enr_couv", "title": "Enrichment coverage (% of active listings)", "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 Rent-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": "Rent-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 : 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 [], } 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), lambda: _panel_gestionnaires(con)): 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