# -----------------------------------------------------------------------------
# Lou-Ka — Agrégateur de logements à louer (province de Québec)
# Auteur : Simon-Pierre Boucher — contact@spboucher.ai
# statsdash.py : tableau de bord statistique — contrat commun Groupe KA v2
#   (voir frontend/src/ka/stats/SPEC.md). Construit le JSON du dashboard à
#   partir de requêtes SQL agrégées (listings, sync_log, price_log) avec un
#   cache mémoire de 5 minutes par clé de période. AUCUNE stat inventée :
#   une section sans donnée réelle est simplement absente du JSON.
#   v2 (2026-08-19) : sparklines KPI, jauges de couverture, série des loyers
#   (area), multi-courbes loyer médian par taille, barres empilées par source,
#   distribution des loyers, heatmap horaire 7×24, tableaux « loyers par
#   taille » et « baisses de loyer », records enrichis.
# -----------------------------------------------------------------------------
from __future__ import annotations

import json
import threading
import time
from datetime import date, datetime, timedelta
from pathlib import Path
from zoneinfo import ZoneInfo

from . import db

TZ = ZoneInfo("America/Toronto")
ROOT = Path(__file__).resolve().parent.parent
SOURCES_PATH = ROOT / "data" / "sources.json"

CACHE_TTL = 300  # secondes
_cache: dict[str, tuple[float, dict]] = {}
_cache_lock = threading.Lock()

# bornes de plausibilité des loyers résidentiels : au-delà, presque toujours
# des erreurs de lecture à la source — exclues des agrégats de prix.
PRIX_MIN, PRIX_MAX = 300, 15000

PERIODS = {
    "auj": ("Aujourd'hui", 0),
    "7j": ("7 jours", 6),
    "30j": ("30 jours", 29),
    "3m": ("3 mois", 89),
    "6m": ("6 mois", 179),
    "12m": ("12 mois", 364),
}

DOW_FR = ["lundi", "mardi", "mercredi", "jeudi", "vendredi", "samedi", "dimanche"]


# ---------------------------------------------------------------- utilitaires
def _day_start_ts(d: date) -> float:
    return datetime(d.year, d.month, d.day, tzinfo=TZ).timestamp()


def _coverage(con) -> tuple[date | None, date | None]:
    """Première et dernière date observées dans la base (first/last_seen)."""
    row = con.execute(
        "SELECT MIN(first_seen) a, MAX(last_seen) b FROM listings"
        " WHERE dup_of IS NULL"
    ).fetchone()
    if row["a"] is None:
        return None, None
    return (datetime.fromtimestamp(row["a"], TZ).date(),
            datetime.fromtimestamp(row["b"], TZ).date())


def resolve_period(period: str, from_: str | None, to_: str | None,
                   cov_min: date, today: date) -> tuple[date, date, str]:
    """Bornes [from, to] (dates locales incluses) + libellé humain."""
    if from_ and to_:
        try:
            a = date.fromisoformat(from_)
            b = date.fromisoformat(to_)
            if a > b:
                a, b = b, a
            return max(a, cov_min), min(b, today), f"{a} → {b}"
        except ValueError:
            pass
    if period == "tout":
        return cov_min, today, "Toute la période"
    if period == "annee":
        return max(date(today.year, 1, 1), cov_min), today, "Année en cours"
    label, back = PERIODS.get(period, PERIODS["30j"])
    return max(today - timedelta(days=back), cov_min), today, label


def _pct(cur: float, prev: float) -> float | None:
    if not prev:
        return None
    return round(100.0 * (cur - prev) / prev, 1)


def _days(a: date, b: date) -> list[date]:
    return [a + timedelta(days=i) for i in range((b - a).days + 1)]


def _fr_int(n: float) -> str:
    return f"{int(round(n)):,}".replace(",", " ")


def _fr_money(n: float) -> str:
    return _fr_int(n) + " $"


def _spark(pts: list[dict], cap: int = 30) -> list[dict] | None:
    """Sous-échantillonne une série pour la sparkline d'un KPI (≤ cap points)."""
    if len(pts) < 2:
        return None
    if len(pts) <= cap:
        return pts
    step = max(1, len(pts) // cap)
    out = pts[::step]
    if out[-1] is not pts[-1]:
        out.append(pts[-1])
    return out


def _source_names() -> dict[str, str]:
    try:
        reg = json.loads(SOURCES_PATH.read_text(encoding="utf-8"))["sources"]
        return {s["id"]: s.get("name") or s["id"] for s in reg}
    except (OSError, ValueError, KeyError):
        return {}


def _pctile(sorted_vals: list[float], frac: float) -> float | None:
    if not sorted_vals:
        return None
    i = min(len(sorted_vals) - 1, int(len(sorted_vals) * frac))
    return sorted_vals[i]


# ------------------------------------------------------------------- dashboard
def compute(period: str = "30j", from_: str | None = None,
            to_: str | None = None) -> dict:
    key = f"{period}|{from_ or ''}|{to_ or ''}"
    now = time.time()
    with _cache_lock:
        hit = _cache.get(key)
        if hit and hit[0] > now:
            return hit[1]
    data = _compute(period, from_, to_)
    with _cache_lock:
        _cache[key] = (now + CACHE_TTL, data)
    return data


def _compute(period: str, from_: str | None, to_: str | None) -> dict:
    con = db.connect()
    try:
        return _build(con, period, from_, to_)
    finally:
        con.close()


# requête réutilisée : premier loyer observé de chaque annonce (journal de
# prix) — c'est le loyer demandé à l'entrée sur le marché, borné plausible.
_FIRST_PRICE_CTE = """
    WITH fp AS (
        SELECT p.uid uid, p.ts ts, p.price price
        FROM price_log p
        JOIN (SELECT uid, MIN(ts) t0 FROM price_log
              WHERE price IS NOT NULL GROUP BY uid) f
          ON f.uid = p.uid AND p.ts = f.t0
        WHERE p.price BETWEEN ? AND ?
    )
"""


def _build(con, period: str, from_: str | None, to_: str | None) -> dict:
    today = datetime.now(TZ).date()
    cov_min, _cov_max = _coverage(con)
    if cov_min is None:  # base vide
        return {"updated": datetime.now(TZ).isoformat(),
                "period": {"from": None, "to": None, "label": "—"},
                "kpis": [], "series": [], "breakdowns": [], "tables": [],
                "records": []}

    d_from, d_to, label = resolve_period(period, from_, to_, cov_min, today)
    ts_from = _day_start_ts(d_from)
    ts_to = _day_start_ts(d_to + timedelta(days=1))  # borne exclusive

    # ---- histogrammes journaliers (2 balayages agrégés, réutilisés partout)
    starts = {r["d"]: r["n"] for r in con.execute(
        "SELECT date(first_seen,'unixepoch','localtime') d, COUNT(*) n"
        " FROM listings WHERE dup_of IS NULL GROUP BY d")}
    ends = {r["d"]: r["n"] for r in con.execute(
        "SELECT date(last_seen,'unixepoch','localtime') d, COUNT(*) n"
        " FROM listings WHERE dup_of IS NULL AND active=0 GROUP BY d")}

    def actives_at(d: date) -> int:
        """Reconstruction : cum(first_seen<=d) − cum(retraits<=d−1)."""
        iso = d.isoformat()
        prev = (d - timedelta(days=1)).isoformat()
        s = sum(n for dd, n in starts.items() if dd <= iso)
        e = sum(n for dd, n in ends.items() if dd <= prev)
        return s - e

    # ---- KPI ------------------------------------------------------------
    actives_now = con.execute(
        "SELECT COUNT(*) n FROM listings WHERE active=1 AND dup_of IS NULL"
    ).fetchone()["n"]
    new_in = sum(n for d, n in starts.items()
                 if d_from.isoformat() <= d <= d_to.isoformat())
    removed_in = sum(n for d, n in ends.items()
                     if d_from.isoformat() <= d <= d_to.isoformat())

    # période précédente de même longueur (uniquement si couverte par la base)
    span = (d_to - d_from).days + 1
    p_from, p_to = d_from - timedelta(days=span), d_from - timedelta(days=1)
    prev_ok = p_from >= cov_min
    pts_prev_from = _day_start_ts(p_from)
    pts_prev_to = _day_start_ts(p_to + timedelta(days=1))
    new_prev = removed_prev = None
    if prev_ok:
        new_prev = sum(n for d, n in starts.items()
                       if p_from.isoformat() <= d <= p_to.isoformat())
        removed_prev = sum(n for d, n in ends.items()
                           if p_from.isoformat() <= d <= p_to.isoformat())

    actives_prev = actives_at(p_to) if d_from - timedelta(days=1) >= cov_min else None

    row = con.execute(
        "SELECT AVG(price) avg_p, COUNT(price) n_p FROM listings"
        " WHERE active=1 AND dup_of IS NULL AND price BETWEEN ? AND ?",
        (PRIX_MIN, PRIX_MAX)).fetchone()
    avg_price, n_priced = row["avg_p"], row["n_p"]
    median_price = None
    if n_priced:
        median_price = con.execute(
            "SELECT price FROM listings WHERE active=1 AND dup_of IS NULL"
            " AND price BETWEEN ? AND ? ORDER BY price LIMIT 1 OFFSET ?",
            (PRIX_MIN, PRIX_MAX, n_priced // 2)).fetchone()["price"]

    connectors = con.execute(
        "SELECT COUNT(DISTINCT source) n FROM sync_log"
        " WHERE ok=1 AND ts>=? AND ts<?", (ts_from, ts_to)).fetchone()["n"]
    cities_n = con.execute(
        "SELECT COUNT(DISTINCT city) n FROM listings"
        " WHERE active=1 AND dup_of IS NULL AND city<>''").fetchone()["n"]

    # séries journalières (réutilisées : séries, sparklines)
    days = _days(d_from, d_to)
    pts_act = [{"t": d.isoformat(), "v": actives_at(d)} for d in days]
    pts_new = [{"t": d.isoformat(), "v": starts.get(d.isoformat(), 0)}
               for d in days]
    pts_rem = [{"t": d.isoformat(), "v": ends.get(d.isoformat(), 0)}
               for d in days]

    # loyer moyen des nouvelles annonces par jour (premier prix observé)
    fp_daily = con.execute(
        _FIRST_PRICE_CTE +
        "SELECT date(ts,'unixepoch','localtime') d, AVG(price) a, COUNT(*) n"
        " FROM fp WHERE ts>=? AND ts<? GROUP BY d ORDER BY d",
        (PRIX_MIN, PRIX_MAX, ts_from, ts_to)).fetchall()
    pts_rent = [{"t": r["d"], "v": round(r["a"])} for r in fp_daily
                if r["n"] >= 5]

    kpis = [
        {"id": "actives", "label": "Annonces actives", "value": actives_now,
         "delta_pct": _pct(actives_now, actives_prev) if actives_prev else None,
         "direction": "up" if (actives_prev and actives_now >= actives_prev) else
                      ("down" if actives_prev else None),
         "spark": _spark(pts_act)},
        {"id": "nouvelles", "label": "Nouvelles annonces (période)",
         "value": new_in,
         "delta_pct": _pct(new_in, new_prev) if prev_ok else None,
         "direction": ("up" if new_in >= (new_prev or 0) else "down") if prev_ok else None,
         "spark": _spark(pts_new)},
        {"id": "retirees", "label": "Annonces retirées (période)",
         "value": removed_in,
         "delta_pct": _pct(removed_in, removed_prev) if prev_ok else None,
         "direction": ("down" if removed_in >= (removed_prev or 0) else "up") if prev_ok else None,
         "spark": _spark(pts_rem)},
    ]
    if avg_price:
        kpis.append({"id": "loyer_moyen", "label": "Loyer moyen (actives)",
                     "value": round(avg_price), "unit": "$",
                     "spark": _spark(pts_rent)})
    if median_price:
        kpis.append({"id": "loyer_median", "label": "Loyer médian (actives)",
                     "value": round(median_price), "unit": "$"})
    # qualité des données (quality.py) : quarantaine
    qual = con.execute(
        "SELECT ROUND(AVG(completeness),1) c, SUM(published=0) q"
        " FROM listings WHERE active=1 AND dup_of IS NULL").fetchone()
    if qual["c"] is not None:
        kpis.append({"id": "quarantaine",
                     "label": "Annonces en quarantaine (qualité)",
                     "value": qual["q"] or 0})
    # juste valeur (fair value) : répartition des classifications
    fv_ok = True
    try:
        fv = con.execute(
            """SELECT SUM(f.verdict='sous') sous, SUM(f.verdict='marche') marche,
                      SUM(f.verdict='sur') sur, COUNT(*) n,
                      ROUND(AVG(f.deviation)*100, 1) dev
               FROM fairvalue f JOIN listings l USING (uid)
               WHERE l.active=1 AND l.published=1 AND l.dup_of IS NULL
                 AND f.verdict IS NOT NULL""").fetchone()
    except Exception:
        fv_ok = False
        fv = None
    if fv_ok and fv and fv["n"]:
        kpis.append({"id": "sous_marche", "label": "Annonces sous le marché",
                     "value": fv["sous"] or 0})
    kpis.append({"id": "connecteurs", "label": "Connecteurs actifs (période)",
                 "value": connectors})
    kpis.append({"id": "villes", "label": "Villes couvertes", "value": cities_n})
    kpis = [{k: v for k, v in kpi.items() if v is not None} for kpi in kpis]

    # ---- jauges de couverture / complétude ---------------------------------
    cov = con.execute(
        """SELECT COUNT(*) n,
                  SUM(lat IS NOT NULL AND lng IS NOT NULL) geo,
                  SUM(price BETWEEN ? AND ?) prix,
                  SUM(images IS NOT NULL AND images NOT IN ('', '[]')) img,
                  SUM(published=1) pub
           FROM listings WHERE active=1 AND dup_of IS NULL""",
        (PRIX_MIN, PRIX_MAX)).fetchone()
    gauges = []
    if cov["n"]:
        n_act = cov["n"]

        def _gauge(gid, lab, num, help_=None):
            g = {"id": gid, "label": lab,
                 "value": round(100.0 * (num or 0) / n_act, 1),
                 "max": 100, "unit": "%"}
            if help_:
                g["help"] = help_
            return g

        gauges = [
            _gauge("geo", "Annonces géolocalisées", cov["geo"]),
            _gauge("prix", "Annonces avec loyer affiché", cov["prix"]),
            _gauge("photos", "Annonces avec photos", cov["img"]),
            _gauge("publiees", "Annonces publiées (qualité OK)", cov["pub"],
                   "Le reste est en quarantaine qualité"),
        ]
        if qual["c"] is not None:
            gauges.append({"id": "completude",
                           "label": "Complétude moyenne des fiches",
                           "value": qual["c"], "max": 100, "unit": "%"})

    # ---- séries temporelles ----------------------------------------------
    series = []
    if len(days) >= 2:
        s_act = {"id": "actives_jour", "title": "Annonces actives par jour",
                 "unit": "annonces", "kind": "line", "points": pts_act}
        s_new = {"id": "nouvelles_jour", "title": "Nouvelles annonces par jour",
                 "unit": "annonces", "kind": "bar", "points": pts_new}
        s_rem = {"id": "retraits_jour", "title": "Annonces retirées par jour",
                 "unit": "annonces", "kind": "line", "points": pts_rem}
        if prev_ok:
            pdays = _days(p_from, p_to)
            s_act["compare"] = [{"t": d.isoformat(), "v": actives_at(d)}
                                for d in pdays]
            s_rem["compare"] = [{"t": d.isoformat(),
                                 "v": ends.get(d.isoformat(), 0)}
                                for d in pdays]
        series = [s_act, s_new, s_rem]
        if len(pts_rent) >= 2:
            series.append({"id": "loyer_entree_jour",
                           "title": "Loyer moyen des nouvelles annonces par jour",
                           "unit": "$", "kind": "area", "points": pts_rent})

    # ---- multi-courbes : loyer médian d'entrée par taille -------------------
    multiseries = []
    if len(days) >= 2:
        top_types = [r["t"] for r in con.execute(
            _FIRST_PRICE_CTE +
            """SELECT l.unit_type t, COUNT(*) n FROM fp
               JOIN listings l ON l.uid=fp.uid AND l.dup_of IS NULL
               WHERE fp.ts>=? AND fp.ts<? AND l.unit_type<>''
               GROUP BY t ORDER BY n DESC LIMIT 4""",
            (PRIX_MIN, PRIX_MAX, ts_from, ts_to)).fetchall()]
        if top_types:
            ph = ",".join("?" * len(top_types))
            rows = con.execute(
                _FIRST_PRICE_CTE +
                f"""SELECT date(fp.ts,'unixepoch','localtime') d,
                           l.unit_type t, fp.price p FROM fp
                    JOIN listings l ON l.uid=fp.uid AND l.dup_of IS NULL
                    WHERE fp.ts>=? AND fp.ts<? AND l.unit_type IN ({ph})""",
                (PRIX_MIN, PRIX_MAX, ts_from, ts_to, *top_types)).fetchall()
            per: dict[str, dict[str, list[float]]] = {t: {} for t in top_types}
            for r in rows:
                per[r["t"]].setdefault(r["d"], []).append(r["p"])
            # axe commun : jours où chaque taille a au moins 3 observations
            common = [d.isoformat() for d in days
                      if all(len(per[t].get(d.isoformat(), [])) >= 3
                             for t in top_types)]
            if len(common) >= 2:
                ms_series = []
                for t in top_types:
                    pts = []
                    for d in common:
                        vals = sorted(per[t][d])
                        pts.append({"t": d, "v": round(vals[len(vals) // 2])})
                    ms_series.append({"label": t, "points": pts})
                multiseries.append({
                    "id": "loyer_taille",
                    "title": "Loyer médian des nouvelles annonces par taille",
                    "unit": "$", "series": ms_series})

    # ---- barres empilées : ajouts par jour et par source --------------------
    names = _source_names()
    stacked = []
    if len(days) >= 2:
        src_tot = con.execute(
            """SELECT source s, COUNT(*) n FROM listings
               WHERE dup_of IS NULL AND first_seen>=? AND first_seen<?
               GROUP BY s ORDER BY n DESC""", (ts_from, ts_to)).fetchall()
        if src_tot:
            top_src = [r["s"] for r in src_tot[:3]]
            others = [r["s"] for r in src_tot[3:]]
            daily_src: dict[str, dict[str, int]] = {}
            for r in con.execute(
                    """SELECT date(first_seen,'unixepoch','localtime') d,
                              source s, COUNT(*) n FROM listings
                       WHERE dup_of IS NULL AND first_seen>=? AND first_seen<?
                       GROUP BY d, s""", (ts_from, ts_to)):
                daily_src.setdefault(r["d"], {})[r["s"]] = r["n"]
            keys = [names.get(s, s) for s in top_src] + (
                ["Autres"] if others else [])
            pts = []
            for d in days:
                by = daily_src.get(d.isoformat(), {})
                vals = [by.get(s, 0) for s in top_src]
                if others:
                    vals.append(sum(by.get(s, 0) for s in others))
                pts.append({"t": d.isoformat(), "values": vals})
            if any(sum(p["values"]) for p in pts):
                stacked.append({"id": "ajouts_source",
                                "title": "Nouvelles annonces par jour et par source",
                                "unit": "ajouts", "keys": keys, "points": pts})

    # ---- répartitions ------------------------------------------------------
    types = [{"label": r["t"] or "Non précisé", "value": r["n"]}
             for r in con.execute(
                 "SELECT unit_type t, COUNT(*) n FROM listings"
                 " WHERE active=1 AND dup_of IS NULL GROUP BY unit_type"
                 " ORDER BY n DESC LIMIT 8")]
    by_source = [{"label": names.get(r["s"], r["s"]), "value": r["n"]}
                 for r in con.execute(
                     "SELECT source s, COUNT(*) n FROM listings"
                     " WHERE active=1 AND dup_of IS NULL GROUP BY source"
                     " ORDER BY n DESC LIMIT 12")]
    breakdowns = []
    if fv_ok and fv and fv["n"]:
        breakdowns.append({"id": "fairvalue",
                           "title": "Position des loyers vs juste valeur estimée",
                           "kind": "donut", "items": [
                               {"label": "Sous le marché", "value": fv["sous"] or 0},
                               {"label": "Dans le marché", "value": fv["marche"] or 0},
                               {"label": "Au-dessus du marché", "value": fv["sur"] or 0}]})
    if types:
        breakdowns.append({"id": "types", "title": "Annonces actives par taille",
                           "kind": "donut", "items": types})
    # nouvelles annonces par taille (période) — deltas vs période précédente
    new_types = con.execute(
        """SELECT COALESCE(NULLIF(unit_type,''),'Non précisé') t, COUNT(*) n
           FROM listings WHERE dup_of IS NULL AND first_seen>=? AND first_seen<?
           GROUP BY t ORDER BY n DESC LIMIT 8""", (ts_from, ts_to)).fetchall()
    if new_types:
        prev_types = {}
        if prev_ok:
            prev_types = {r["t"]: r["n"] for r in con.execute(
                """SELECT COALESCE(NULLIF(unit_type,''),'Non précisé') t,
                          COUNT(*) n FROM listings
                   WHERE dup_of IS NULL AND first_seen>=? AND first_seen<?
                   GROUP BY t""", (pts_prev_from, pts_prev_to))}
        items = []
        for r in new_types:
            it = {"label": r["t"], "value": r["n"]}
            if prev_ok:
                it["delta_pct"] = _pct(r["n"], prev_types.get(r["t"], 0))
            items.append(it)
        breakdowns.append({"id": "nouvelles_taille",
                           "title": "Nouvelles annonces par taille (période)",
                           "kind": "bars", "items": items})
    if by_source:
        breakdowns.append({"id": "sources", "title": "Top sources (annonces actives)",
                           "kind": "bars", "items": by_source})

    # ---- distribution des loyers (annonces actives, tranches de 200 $) ------
    distributions = []
    bins_rows = con.execute(
        """SELECT CAST(price/200 AS INT) b, COUNT(*) n FROM listings
           WHERE active=1 AND dup_of IS NULL AND price BETWEEN ? AND ?
           GROUP BY b ORDER BY b""", (PRIX_MIN, PRIX_MAX)).fetchall()
    if bins_rows:
        LO, HI = 2, 16  # < 400 $ … 3 200 $ et +
        agg: dict[int, int] = {}
        for r in bins_rows:
            b = min(max(r["b"], LO - 1), HI)
            agg[b] = agg.get(b, 0) + r["n"]
        bins = []
        for b in sorted(agg):
            if b == LO - 1:
                lab = f"< {LO * 200} $"
            elif b == HI:
                lab = f"{_fr_int(HI * 200)} $ +"
            else:
                lab = f"{_fr_int(b * 200)}–{_fr_int((b + 1) * 200 - 1)} $"
            bins.append({"label": lab, "value": agg[b]})
        distributions.append({"id": "loyers",
                              "title": "Distribution des loyers (annonces actives)",
                              "unit": "annonces", "bins": bins})

    # ---- géographie --------------------------------------------------------
    geo_items = [{"label": r["c"], "value": r["n"]} for r in con.execute(
        "SELECT city c, COUNT(*) n FROM listings"
        " WHERE active=1 AND dup_of IS NULL AND city<>''"
        " GROUP BY city ORDER BY n DESC LIMIT 14")]
    geo = ({"title": "Top villes (annonces actives)", "items": geo_items}
           if geo_items else None)

    # ---- calendrier de chaleur : 26 dernières semaines d'ajouts --------------
    heat_from = max(cov_min, today - timedelta(days=181))
    heat_cells = [{"date": d, "value": n} for d, n in sorted(starts.items())
                  if d >= heat_from.isoformat()]
    heatmap = ({"title": "Nouvelles annonces par jour (26 dernières semaines)",
                "cells": heat_cells} if heat_cells else None)

    # ---- heatmap horaire 7×24 : heure d'observation des ajouts ---------------
    hourly = None
    hour_rows = con.execute(
        """SELECT CAST(strftime('%w', first_seen,'unixepoch','localtime') AS INT) w,
                  CAST(strftime('%H', first_seen,'unixepoch','localtime') AS INT) h,
                  COUNT(*) n
           FROM listings WHERE dup_of IS NULL AND first_seen>=? AND first_seen<?
           GROUP BY w, h""", (ts_from, ts_to)).fetchall()
    if hour_rows:
        cells = [{"dow": (r["w"] + 6) % 7, "hour": r["h"], "value": r["n"]}
                 for r in hour_rows]  # %w : 0=dim → contrat : 0=lun
        hourly = {"title": "Ajouts d'annonces par heure (période)",
                  "cells": cells}

    # ---- tableaux ----------------------------------------------------------
    tables = []
    top_villes = con.execute(
        """SELECT city, COUNT(*) n, AVG(CASE WHEN price BETWEEN ? AND ? THEN price END) avg_p,
                  SUM(CASE WHEN first_seen>=? AND first_seen<? THEN 1 ELSE 0 END) new_n
           FROM listings WHERE active=1 AND dup_of IS NULL AND city<>''
           GROUP BY city ORDER BY n DESC LIMIT 50""",
        (PRIX_MIN, PRIX_MAX, ts_from, ts_to)).fetchall()
    removed_by_city = {r["city"]: r["n"] for r in con.execute(
        """SELECT city, COUNT(*) n FROM listings
           WHERE active=0 AND dup_of IS NULL AND city<>''
             AND last_seen>=? AND last_seen<? GROUP BY city""",
        (ts_from, ts_to)).fetchall()}

    def _city_median(city: str) -> float | None:
        r = con.execute(
            "SELECT COUNT(*) n FROM listings WHERE active=1 AND dup_of IS NULL"
            " AND city=? AND price BETWEEN ? AND ?",
            (city, PRIX_MIN, PRIX_MAX)).fetchone()
        if not r["n"]:
            return None
        return con.execute(
            "SELECT price FROM listings WHERE active=1 AND dup_of IS NULL"
            " AND city=? AND price BETWEEN ? AND ? ORDER BY price"
            " LIMIT 1 OFFSET ?",
            (city, PRIX_MIN, PRIX_MAX, r["n"] // 2)).fetchone()["price"]

    city_medians: dict[str, float] = {}
    if top_villes:
        rows = []
        for r in top_villes:
            med = _city_median(r["city"])
            if med is not None:
                city_medians[r["city"]] = med
            net = r["new_n"] - removed_by_city.get(r["city"], 0)
            rows.append([
                r["city"], r["n"],
                _fr_money(r["avg_p"]) if r["avg_p"] else "—",
                _fr_money(med) if med is not None else "—",
                r["new_n"], f"{'+' if net >= 0 else ''}{net}"])
        tables.append({"id": "top_villes", "title": "Top villes",
                       "columns": ["Ville", "Annonces actives", "Loyer moyen",
                                   "Loyer médian", "Nouvelles (période)",
                                   "Δ net (période)"],
                       "rows": rows})

    # loyers par taille de logement (moyenne, médiane, P10–P90)
    type_rows = con.execute(
        """SELECT COALESCE(NULLIF(unit_type,''),'Non précisé') t, COUNT(*) n,
                  AVG(price) avg_p FROM listings
           WHERE active=1 AND dup_of IS NULL AND price BETWEEN ? AND ?
           GROUP BY t ORDER BY n DESC LIMIT 12""",
        (PRIX_MIN, PRIX_MAX)).fetchall()
    if type_rows:
        rows = []
        for r in type_rows:
            prices = sorted(x["price"] for x in con.execute(
                """SELECT price FROM listings WHERE active=1 AND dup_of IS NULL
                   AND COALESCE(NULLIF(unit_type,''),'Non précisé')=?
                   AND price BETWEEN ? AND ?""",
                (r["t"], PRIX_MIN, PRIX_MAX)))
            med = _pctile(prices, 0.5)
            p10 = _pctile(prices, 0.10)
            p90 = _pctile(prices, 0.90)
            rows.append([
                r["t"], r["n"], _fr_money(r["avg_p"]),
                _fr_money(med) if med is not None else "—",
                f"{_fr_money(p10)} – {_fr_money(p90)}"
                if p10 is not None and p90 is not None else "—"])
        tables.append({"id": "loyers_taille",
                       "title": "Loyers par taille de logement (actives)",
                       "columns": ["Taille", "Annonces avec loyer",
                                   "Loyer moyen", "Loyer médian",
                                   "Fourchette P10–P90"],
                       "rows": rows})

    top_srcs = con.execute(
        """SELECT source s, COUNT(*) n, AVG(CASE WHEN price BETWEEN ? AND ? THEN price END) avg_p,
                  SUM(CASE WHEN first_seen>=? AND first_seen<? THEN 1 ELSE 0 END) new_n,
                  ROUND(AVG(completeness),0) comp, SUM(published=0) quar
           FROM listings WHERE active=1 AND dup_of IS NULL
           GROUP BY source ORDER BY n DESC LIMIT 50""",
        (PRIX_MIN, PRIX_MAX, ts_from, ts_to)).fetchall()
    last_sync = {r["source"]: r["ts"] for r in con.execute(
        "SELECT source, MAX(ts) ts FROM sync_log WHERE ok=1 GROUP BY source")}
    if top_srcs:
        rows = []
        for r in top_srcs:
            ls = last_sync.get(r["s"])
            rows.append([
                names.get(r["s"], r["s"]), r["n"],
                _fr_money(r["avg_p"]) if r["avg_p"] else "—",
                r["new_n"],
                f"{r['comp']:.0f} %" if r["comp"] is not None else "—",
                r["quar"] or 0,
                datetime.fromtimestamp(ls, TZ).strftime("%Y-%m-%d %H:%M") if ls else "—"])
        tables.append({"id": "top_gestionnaires",
                       "title": "Top gestionnaires & sources",
                       "columns": ["Gestionnaire / source", "Annonces actives",
                                   "Loyer moyen", "Nouvelles (période)",
                                   "Complétude", "Quarantaine",
                                   "Dernière synchro"],
                       "rows": rows})

    # écart moyen au marché par ville (villes à volume suffisant)
    if fv_ok:
        fv_villes = con.execute(
            """SELECT l.city, COUNT(*) n, ROUND(AVG(f.deviation)*100,1) dev,
                      SUM(f.verdict='sous') sous
               FROM fairvalue f JOIN listings l USING (uid)
               WHERE l.active=1 AND l.published=1 AND l.dup_of IS NULL
                 AND f.verdict IS NOT NULL AND l.city<>''
               GROUP BY l.city HAVING n>=50 ORDER BY n DESC LIMIT 25""").fetchall()
        if fv_villes:
            tables.append({
                "id": "fv_villes", "title": "Écart au marché par ville (fair value)",
                "columns": ["Ville", "Annonces évaluées", "Écart moyen",
                            "Sous le marché"],
                "rows": [[r["city"], r["n"],
                          f"{'+' if r['dev'] >= 0 else ''}{r['dev']} %",
                          r["sous"] or 0] for r in fv_villes]})

    # baisses de loyer observées dans la période (journal de prix, plausibles)
    drops = con.execute(
        """SELECT a.uid uid, MAX(a.price-b.price) dp, a.price p0, b.price p1,
                  b.ts ts1
           FROM price_log a JOIN price_log b
             ON a.uid=b.uid AND b.ts>a.ts
            AND a.price BETWEEN ? AND ? AND b.price BETWEEN ? AND ?
            AND b.price<a.price AND b.price>=a.price*0.5
           WHERE a.ts>=? AND b.ts<?
           GROUP BY a.uid ORDER BY dp DESC LIMIT 15""",
        (PRIX_MIN, PRIX_MAX, PRIX_MIN, PRIX_MAX, ts_from, ts_to)).fetchall()
    if drops:
        rows = []
        for r in drops:
            li = con.execute(
                "SELECT city, unit_type FROM listings WHERE uid=?",
                (r["uid"],)).fetchone()
            what = " · ".join(x for x in [li["unit_type"], li["city"]]
                              if x) if li else r["uid"]
            rows.append([
                what or r["uid"], _fr_money(r["p0"]), _fr_money(r["p1"]),
                f"−{_fr_money(r['dp'])}",
                f"{round(100.0 * r['dp'] / r['p0'], 1)} %",
                datetime.fromtimestamp(r["ts1"], TZ).date().isoformat()])
        tables.append({"id": "baisses_loyer",
                       "title": "Plus fortes baisses de loyer (période)",
                       "columns": ["Logement", "Ancien loyer", "Nouveau loyer",
                                   "Baisse", "Baisse %", "Observée le"],
                       "rows": rows})

    # ---- records & faits marquants -----------------------------------------
    records = []
    in_period = {d: n for d, n in starts.items()
                 if d_from.isoformat() <= d <= d_to.isoformat()}
    if in_period:
        best = max(in_period, key=in_period.get)
        records.append({"label": "Jour record d'ajouts",
                        "value": _fr_int(in_period[best]) + " annonces",
                        "date": best})
    rem_period = {d: n for d, n in ends.items()
                  if d_from.isoformat() <= d <= d_to.isoformat()}
    if rem_period:
        worst = max(rem_period, key=rem_period.get)
        records.append({"label": "Jour record de retraits",
                        "value": _fr_int(rem_period[worst]) + " annonces",
                        "date": worst})
    if top_villes:
        ville = max(top_villes, key=lambda r: r["new_n"])
        if ville["new_n"]:
            records.append({"label": "Ville la plus dynamique (nouvelles annonces)",
                            "value": f"{ville['city']} — {ville['new_n']}"})
    # villes la plus chère / la plus abordable (loyer médian, volume suffisant)
    big_meds = {c: m for c, m in city_medians.items()
                if next((r["n"] for r in top_villes if r["city"] == c), 0) >= 100}
    if big_meds:
        chere = max(big_meds, key=big_meds.get)
        abordable = min(big_meds, key=big_meds.get)
        records.append({"label": "Ville la plus chère (loyer médian, ≥ 100 annonces)",
                        "value": f"{chere} — {_fr_money(big_meds[chere])}"})
        if abordable != chere:
            records.append({"label": "Ville la plus abordable (loyer médian, ≥ 100 annonces)",
                            "value": f"{abordable} — {_fr_money(big_meds[abordable])}"})
    if types and types[0]["label"] != "Non précisé":
        records.append({"label": "Taille la plus offerte",
                        "value": f"{types[0]['label']} — "
                                 f"{_fr_int(types[0]['value'])} annonces"})
    # plus forte baisse de loyer observée dans la période (journal de prix)
    if drops:
        d0 = drops[0]
        li = con.execute("SELECT city, unit_type FROM listings WHERE uid=?",
                         (d0["uid"],)).fetchone()
        where = " · ".join(x for x in [li["unit_type"], li["city"]] if x) if li else ""
        records.append({
            "label": "Plus forte baisse de loyer observée"
                     + (f" ({where})" if where else ""),
            "value": f"−{_fr_money(d0['dp'])}"
                     f" ({_fr_money(d0['p0'])} → {_fr_money(d0['p1'])})",
            "date": datetime.fromtimestamp(d0["ts1"], TZ).date().isoformat()})
    # plus forte hausse de loyer observée (plausible : ≤ ×2)
    hike = con.execute(
        """SELECT a.uid uid, MAX(b.price-a.price) dp, a.price p0, b.price p1,
                  b.ts ts1
           FROM price_log a JOIN price_log b
             ON a.uid=b.uid AND b.ts>a.ts
            AND a.price BETWEEN ? AND ? AND b.price BETWEEN ? AND ?
            AND b.price>a.price AND b.price<=a.price*2
           WHERE a.ts>=? AND b.ts<? LIMIT 1""",
        (PRIX_MIN, PRIX_MAX, PRIX_MIN, PRIX_MAX, ts_from, ts_to)).fetchone()
    if hike and hike["dp"]:
        li = con.execute("SELECT city, unit_type FROM listings WHERE uid=?",
                         (hike["uid"],)).fetchone()
        where = " · ".join(x for x in [li["unit_type"], li["city"]] if x) if li else ""
        records.append({
            "label": "Plus forte hausse de loyer observée"
                     + (f" ({where})" if where else ""),
            "value": f"+{_fr_money(hike['dp'])}"
                     f" ({_fr_money(hike['p0'])} → {_fr_money(hike['p1'])})",
            "date": datetime.fromtimestamp(hike["ts1"], TZ).date().isoformat()})
    if hourly:
        peak = max(hourly["cells"], key=lambda c: c["value"])
        records.append({"label": "Heure de pointe des ajouts (période)",
                        "value": f"{DOW_FR[peak['dow']]} {peak['hour']} h — "
                                 f"{_fr_int(peak['value'])} annonces"})
    if top_srcs:
        src = max(top_srcs, key=lambda r: r["new_n"])
        if src["new_n"]:
            records.append({"label": "Source la plus active (nouvelles annonces)",
                            "value": f"{names.get(src['s'], src['s'])} — {src['new_n']}"})

    out = {
        "updated": datetime.now(TZ).isoformat(),
        "period": {"from": d_from.isoformat(), "to": d_to.isoformat(),
                   "label": label},
        "kpis": kpis,
        "series": series,
        "breakdowns": breakdowns,
        "tables": tables,
        "records": records,
    }
    if gauges:
        out["gauges"] = gauges
    if multiseries:
        out["multiseries"] = multiseries
    if stacked:
        out["stacked"] = stacked
    if distributions:
        out["distributions"] = distributions
    if geo:
        out["geo"] = geo
    if heatmap:
        out["heatmap"] = heatmap
    if hourly:
        out["hourly"] = hourly
    try:
        from . import statsextra, statsfiche
        pnls = statsfiche.panels(con) + statsextra.panels(con)
        if pnls:
            out["panels"] = pnls
    except Exception:
        pass
    return out