# ----------------------------------------------------------------------------- # Rent-Ka — Rental listings aggregator (Canada, outside Québec) # Author: Simon-Pierre Boucher — contact@spboucher.ai # building.py : « Passeport de l'immeuble » — entité persistante regroupant les # annonces (actives ET historiques) d'un même bâtiment. # # Clé d'immeuble : la clé d'adresse civique de dedup._parse_address # (« civique|rue|ville », préfixe « adr: ») — la même logique éprouvée que la # déduplication inter-sources ; repli « geo:coord_key » (~11 m) quand l'adresse # n'a pas de numéro civique fiable mais que l'annonce est géocodée. # # Tout indicateur est calculé UNIQUEMENT sur ce que Rent-Ka observe réellement # (annonces collectées par les crawls). Un indicateur sans échantillon # suffisant est omis ou marqué « donnees_insuffisantes » — jamais inventé. # ----------------------------------------------------------------------------- from __future__ import annotations import json import statistics import time from . import db from .dedup import CLASSIFIEDS, PORTALS, _parse_address VERSION = "immeuble-1.0" # seuils du score de rotation (annonces des 12 derniers mois / unités estimées) ROTATION = [(0.5, "faible"), (1.0, "normale"), (2.0, "élevée")] ROTATION_MAX = "très élevée" MIN_OBS_DAYS = 180 # fenêtre d'observation minimale pour classer la rotation MIN_ADS_ROTATION = 3 # nb d'annonces minimal pour classer la rotation MIN_ADS_PRESSURE = 3 # nb minimal par fenêtre pour la pression sur les loyers def building_key(address: str | None, city: str | None, coord_key: str | None) -> tuple[str | None, str | None]: """(clé d'immeuble, numéro d'unité) — adresse d'abord, géo en repli.""" akey, unit = _parse_address(address or "", city or "") if akey: return f"adr:{akey}", unit if coord_key: return f"geo:{coord_key}", unit return None, unit def _median(vals: list[float]) -> float | None: vals = [v for v in vals if v] return round(statistics.median(vals)) if vals else None def _stats_for(members: list[dict], now: float) -> dict: """Statistiques d'un immeuble à partir de ses annonces (obs. Rent-Ka).""" d30, d90, d365, d730 = (now - 30 * 86400, now - 90 * 86400, now - 365 * 86400, now - 730 * 86400) actives = [m for m in members if m["active"]] first_obs = min(m["first_seen"] for m in members if m["first_seen"]) last_obs = max(m["last_seen"] for m in members if m["last_seen"]) obs_days = max(1, round((now - first_obs) / 86400)) units = {m["unit"] for m in members if m["unit"]} # unités estimées : numéros d'app distincts observés, plancher = annonces # actives en ce moment (elles existent forcément toutes en même temps) unites_estimees = max(len(units), len(actives), 1) n_12m = sum(1 for m in members if (m["first_seen"] or 0) >= d365) # loyers médians (annonces actives) med = _median([m["price"] for m in actives]) par_cc: dict[str, int] = {} by_cc: dict[str, list[float]] = {} for m in actives: if m["price"] and m["bedrooms"] is not None: by_cc.setdefault(str(int(m["bedrooms"])), []).append(m["price"]) for cc, vals in sorted(by_cc.items()): v = _median(vals) if v and len(vals) >= 2: par_cc[cc] = v vals_pi2 = [m["price"] / m["area"] for m in actives if m["price"] and m["area"] and m["area"] > 100] pi2 = round(statistics.median(vals_pi2), 2) if vals_pi2 else None # pression sur les loyers : médiane des PREMIERS prix observés des annonces # apparues dans les 12 derniers mois vs celles des 12 mois précédents win_new = [m["first_price"] for m in members if m["first_price"] and (m["first_seen"] or 0) >= d365] win_old = [m["first_price"] for m in members if m["first_price"] and d730 <= (m["first_seen"] or 0) < d365] pression = None if len(win_new) >= MIN_ADS_PRESSURE and len(win_old) >= MIN_ADS_PRESSURE: m_new, m_old = statistics.median(win_new), statistics.median(win_old) if m_old > 0: pression = { "variation_12m": round((m_new - m_old) / m_old, 4), "mediane_12m": round(m_new), "mediane_12_24m": round(m_old), "n_12m": len(win_new), "n_12_24m": len(win_old), "statut": "calculated", } # score de rotation : annonces des 12 derniers mois / unités estimées rotation: dict = {"statut": "donnees_insuffisantes"} if obs_days >= MIN_OBS_DAYS and len(members) >= MIN_ADS_ROTATION: ratio = n_12m / unites_estimees classe = ROTATION_MAX for seuil, nom in ROTATION: if ratio < seuil: classe = nom break rotation = { "classe": classe, "ratio": round(ratio, 2), "annonces_12m": n_12m, "unites_estimees": unites_estimees, "observation_jours": obs_days, "statut": "calculated", "methode": ("listings observed in the last 12 months ÷ estimated " "units (distinct unit numbers, floor = simultaneous " "active listings). Only measures what Rent-Ka " "observes — not a demographic turnover rate."), } sources: dict[str, int] = {} for m in members: sources[m["source"]] = sources.get(m["source"], 0) + 1 gestionnaires = sorted(s for s in sources if s not in PORTALS and s not in CLASSIFIEDS) return { "annonces_total": len(members), "annonces_actives": len(actives), "annonces_30j": sum(1 for m in members if (m["first_seen"] or 0) >= d30), "annonces_90j": sum(1 for m in members if (m["first_seen"] or 0) >= d90), "annonces_12m": n_12m, "unites_identifiees": len(units), "unites_estimees": unites_estimees, "loyer_median": med, "loyer_median_par_cc": par_cc or None, "pi2_median": pi2, "pression_loyers": pression, "rotation": rotation, "gestionnaires": gestionnaires, "sources": sources, "premiere_observation": first_obs, "derniere_observation": last_obs, "unites_actives": sorted( [{"uid": m["uid"], "unit_type": m["unit_type"], "price": m["price"], "bedrooms": m["bedrooms"]} for m in actives if not m["dup_of"]], key=lambda x: (x["price"] is None, x["price"] or 0))[:24], } def rollup(con=None) -> dict: """Recalcule building_key de toutes les annonces + la table buildings. Idempotent, recalcul complet (même philosophie que dedup.run) : les annonces INACTIVES participent — l'historique de l'immeuble est justement ce qui fait la valeur du passeport. """ own = con is None if own: con = db.connect() now = time.time() rows = con.execute( "SELECT uid, source, address, city, coord_key, unit_type, bedrooms," " price, area_sqft, first_seen, last_seen, active, dup_of, building_key" " FROM listings").fetchall() # premier prix observé par annonce (pression sur les loyers) firstp = {r["uid"]: r["p"] for r in con.execute( "SELECT uid, (SELECT price FROM price_log p2 WHERE p2.uid=p1.uid" " AND p2.price IS NOT NULL ORDER BY ts LIMIT 1) p" " FROM (SELECT DISTINCT uid FROM price_log) p1")} groups: dict[str, list[dict]] = {} key_updates: list[tuple[str | None, str]] = [] for r in rows: bkey, unit = building_key(r["address"], r["city"], r["coord_key"]) if bkey != r["building_key"]: key_updates.append((bkey, r["uid"])) if not bkey: continue groups.setdefault(bkey, []).append({ "uid": r["uid"], "source": r["source"], "unit": unit, "unit_type": r["unit_type"], "bedrooms": r["bedrooms"], "price": r["price"], "area": r["area_sqft"], "first_seen": r["first_seen"], "last_seen": r["last_seen"], "active": r["active"], "dup_of": r["dup_of"], "first_price": firstp.get(r["uid"]) or r["price"], "address": r["address"], "city": r["city"], }) if key_updates: con.executemany("UPDATE listings SET building_key=? WHERE uid=?", key_updates) upserts = [] for bkey, members in groups.items(): # doublons inter-sources : un même logement publié sur 3 plateformes ne # compte qu'une fois dans les statistiques de l'immeuble uniq = [m for m in members if not m["dup_of"]] if not uniq: continue stats = _stats_for(uniq, now) # adresse d'affichage : la plus fréquente parmi les membres actifs addrs = [m["address"] for m in uniq if m["address"]] addr = max(set(addrs), key=addrs.count) if addrs else None cities = [m["city"] for m in uniq if m["city"]] city = max(set(cities), key=cities.count) if cities else None # position : coord de n'importe quel membre géocodé (via coord_key) upserts.append((bkey, addr, city, json.dumps(stats, ensure_ascii=False), VERSION, now)) con.executemany( "INSERT INTO buildings (bkey, address, city, stats, version, computed_at)" " VALUES (?,?,?,?,?,?)" " ON CONFLICT(bkey) DO UPDATE SET address=excluded.address," " city=excluded.city, stats=excluded.stats, version=excluded.version," " computed_at=excluded.computed_at", upserts) con.commit() out = {"buildings": len(upserts), "listings_scanned": len(rows), "keys_updated": len(key_updates)} if own: con.close() return out def fiche(bkey: str, con=None) -> dict | None: """Passeport d'un immeuble pour l'API (stats précalculées par rollup).""" own = con is None if own: con = db.connect() row = con.execute( "SELECT bkey, address, city, stats, version, computed_at FROM buildings" " WHERE bkey=?", (bkey,)).fetchone() if own and row is None: con.close() return None if row is None: return None out = {"bkey": row["bkey"], "address": row["address"], "city": row["city"], "version": row["version"], "computed_at": row["computed_at"], **json.loads(row["stats"] or "{}")} if own: con.close() return out if __name__ == "__main__": print(rollup())