# -----------------------------------------------------------------------------
# Immo-Ka — Agrégateur de propriétés à vendre (province de Québec)
# Auteur : Simon-Pierre Boucher — contact@spboucher.ai
# stats.py : tableau de bord analytique /api/stats/dashboard + rapport PDF
#   /api/stats/report (module Stats commun Groupe KA v2 — voir
#   frontend/src/ka/stats/SPEC.md). Toutes les valeurs viennent de la base
#   (listings, price_log, sync_log) — AUCUNE statistique inventée : une
#   mesure indisponible est simplement omise (le front affiche un état vide).
#   v2 : sparklines KPI, jauges (géolocalisation, photos, publiable),
#   multi-courbes (prix médian par type / grande ville), barres empilées
#   (nouvelles inscriptions par bannière), distributions (prix, superficie,
#   année de construction), heatmap horaire 7×24, tableaux quarantaine &
#   bannières, records enrichis.
# -----------------------------------------------------------------------------
from __future__ import annotations

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

from . import db

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

# Position vs estimation Vrai-Prix (mêmes seuils que le fair value Lou-Ka) :
# sous le marché si écart <= -8 %, au-dessus si >= +8 % ; les écarts hors
# (-50 %, +100 %) sont presque toujours des erreurs de lecture -> ignorés.
FV_SEUIL_SOUS = -0.08
FV_SEUIL_SUR = 0.08
FV_DEV_BOUNDS = (-0.50, 1.00)
# bornes de plausibilité des prix de vente résidentiels (journal de prix) :
# les écarts extrêmes sont des erreurs de source, pas de vraies baisses.
PRICE_MIN, PRICE_MAX = 25_000, 50_000_000

# Même règle de visibilité que le reste de l'API (web.DEDUP_CLAUSE) :
# doublons de sous-agences masqués + « Prix sur demande » exclus.
VISIBLE = " AND dup_hidden=0 AND published=1"

PERIOD_LABELS = {
    "auj": "Aujourd'hui", "7j": "7 jours", "30j": "30 jours",
    "3m": "3 mois", "6m": "6 mois", "12m": "12 mois",
    "annee": "Année en cours", "tout": "Toute la période",
}

# --- cache serveur (>= 5 min par période, contrat SPEC) -----------------------
_CACHE: dict[str, tuple[float, dict]] = {}
_CACHE_TTL = 300
_CACHE_LOCK = threading.Lock()


# --- utilitaires --------------------------------------------------------------
def _today() -> date:
    return datetime.now(TZ).date()


def _iso(d: date) -> str:
    return d.isoformat()


def _parse(d: str) -> date | None:
    try:
        return date.fromisoformat(d[:10])
    except (ValueError, TypeError):
        return None


def _epoch(d: date) -> float:
    """Minuit local (heure de l'Est) du jour donné, en epoch."""
    return datetime(d.year, d.month, d.day, tzinfo=TZ).timestamp()


def resolve_period(period: str | None, frm: str | None, to: str | None,
                   data_start: date) -> tuple[date, date, str]:
    today = _today()
    f, t = _parse(frm or ""), _parse(to or "")
    if f and t:
        if t < f:
            f, t = t, f
        return f, t, f"{_iso(f)} → {_iso(t)}"
    p = (period or "30j").lower()
    spans = {"7j": 6, "30j": 29, "3m": 89, "6m": 181, "12m": 364}
    if p == "auj":
        return today, today, PERIOD_LABELS["auj"]
    if p == "annee":
        return date(today.year, 1, 1), today, PERIOD_LABELS["annee"]
    if p == "tout":
        return data_start, today, PERIOD_LABELS["tout"]
    days = spans.get(p, 29)
    label = PERIOD_LABELS.get(p, PERIOD_LABELS["30j"])
    return today - timedelta(days=days), today, label


def _fold(s: str) -> str:
    return "".join(c for c in unicodedata.normalize("NFKD", s.lower().strip())
                   if not unicodedata.combining(c))


def _median(vals: list[float]) -> float | None:
    # défensif : la DB peut contenir des prix NULL — on les écarte
    vals = [v for v in vals if isinstance(v, (int, float))]
    return statistics.median(vals) if vals else None


def _fmt_money(v: float) -> str:
    return f"{round(v):,}".replace(",", " ") + " $"


def _fmt_pct(cur: float, prev: float) -> float | None:
    if prev <= 0:
        return None
    return round((cur - prev) / prev * 100.0, 1)


def _daterange(a: date, b: date):
    d = a
    while d <= b:
        yield d
        d += timedelta(days=1)


def _source_names() -> dict[str, str]:
    """id -> nom lisible depuis data/sources.json (repli : id brut)."""
    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, TypeError):
        return {}


# Familles de connecteurs (mêmes règles que web._FRANCHISES — dupliquées ici
# pour éviter l'import circulaire web ⇄ stats)
_FAMILLES = [
    ("RE/MAX", lambda s: s == "remax_quebec" or s.startswith("remax_ag_")),
    ("Via Capitale", lambda s: s == "via_capitale" or s.startswith("via_ag_")),
    ("Century 21", lambda s: s == "century21" or s.startswith("c21_ag_")),
    ("Royal LePage", lambda s: s == "royal_lepage"),
    ("Groupe Sutton", lambda s: s == "sutton"),
    ("Keller Williams", lambda s: s.startswith("kw_")),
    ("DuProprio", lambda s: s == "duproprio"),
    ("Vendre.ca", lambda s: s == "vendre_ag_ca"),
]


def _famille_of(source: str, names: dict[str, str]) -> str:
    for name, match in _FAMILLES:
        if match(source):
            return name
    return names.get(source, source)


def _downsample(pts: list[dict], keep: int = 40) -> list[dict]:
    """Réduit une série de points {t,v} à <= keep points (sparklines)."""
    if len(pts) <= keep:
        return pts
    step = (len(pts) - 1) / (keep - 1)
    return [pts[round(i * step)] for i in range(keep)]


# Libellés lisibles des motifs de quarantaine/anomalies (quality.py)
_MOTIFS_QUALITE = {
    "quarantaine": "Sous le seuil de publication",
    "sans_image": "Sans image (secours affiché)",
    "prix_hors_bornes": "Prix hors bornes",
    "superficie_improbable": "Superficie improbable",
    "terrain_improbable": "Terrain improbable",
    "chambres_improbables": "Chambres improbables",
    "sdb_improbables": "Salles de bain improbables",
    "chambres_vs_type": "Chambres vs type incohérents",
    "annee_invalide": "Année de construction invalide",
    "prix_pi2_extreme": "Prix au pi² extrême",
}


# --- calcul du tableau de bord ------------------------------------------------
def _compute(frm_q: str | None, to_q: str | None, period: str | None) -> dict:
    con = db.connect()
    try:
        return _compute_con(con, frm_q, to_q, period)
    finally:
        con.close()


def _compute_con(con, frm_q, to_q, period) -> dict:
    today = _today()
    row = con.execute("SELECT MIN(first_seen) m FROM listings").fetchone()
    data_start = (datetime.fromtimestamp(row["m"], TZ).date()
                  if row and row["m"] else today)

    frm, to, label = resolve_period(period, frm_q, to_q, data_start)
    to = min(to, today)
    # fenêtre observée : la collecte a commencé le data_start — les séries
    # sont bornées à ce qui a réellement été mesuré (rien d'extrapolé).
    s_frm = max(frm, data_start)
    s_to = max(to, s_frm)
    ep_frm, ep_to = _epoch(s_frm), _epoch(s_to + timedelta(days=1))
    ndays = (s_to - s_frm).days + 1
    # période précédente de même longueur (pour les deltas)
    p_frm, p_to = s_frm - timedelta(days=ndays), s_frm - timedelta(days=1)
    # deltas seulement si la période précédente a été observée EN ENTIER —
    # comparer à une fenêtre tronquée fausserait les variations.
    prev_ok = p_frm >= data_start
    ep_pfrm, ep_pto = _epoch(p_frm), _epoch(p_to + timedelta(days=1))

    # ---- reconstruction « annonces actives par jour » (événements) ----------
    actives_by_day: dict[str, int] = {}
    deltas: dict[date, int] = {}
    for r in con.execute(
            "SELECT date(first_seen,'unixepoch','localtime') fs,"
            " date(last_seen,'unixepoch','localtime') ls, active"
            " FROM listings WHERE 1=1" + VISIBLE):
        d0 = _parse(r["fs"])
        if d0 is None:
            continue
        deltas[d0] = deltas.get(d0, 0) + 1
        if not r["active"]:
            d1 = (_parse(r["ls"]) or d0) + timedelta(days=1)
            deltas[d1] = deltas.get(d1, 0) - 1
    run = 0
    for d in _daterange(data_start, today):
        run += deltas.get(d, 0)
        actives_by_day[_iso(d)] = run

    # ---- KPI -----------------------------------------------------------------
    snap = con.execute(
        "SELECT COUNT(*) n, AVG(price) avg_p,"
        " COUNT(DISTINCT NULLIF(city,'')) cities"
        " FROM listings WHERE active=1" + VISIBLE).fetchone()
    prices = [r["price"] for r in con.execute(
        "SELECT price FROM listings WHERE active=1" + VISIBLE)]
    med_price = _median(prices)
    ppm2 = [r["v"] for r in con.execute(
        "SELECT price/(area_sqft/" + str(SQFT_PER_M2) + ") v FROM listings"
        " WHERE active=1 AND area_sqft>=200" + VISIBLE)]
    med_ppm2 = _median(ppm2)
    n_ppm2 = len(ppm2)

    new_cur = con.execute(
        "SELECT COUNT(*) n FROM listings WHERE first_seen>=? AND first_seen<?"
        + VISIBLE, (ep_frm, ep_to)).fetchone()["n"]
    new_prev = con.execute(
        "SELECT COUNT(*) n FROM listings WHERE first_seen>=? AND first_seen<?"
        + VISIBLE, (ep_pfrm, ep_pto)).fetchone()["n"] if prev_ok else 0
    gone_cur = con.execute(
        "SELECT COUNT(*) n FROM listings WHERE active=0 AND last_seen>=?"
        " AND last_seen<?" + VISIBLE, (ep_frm, ep_to)).fetchone()["n"]
    gone_prev = con.execute(
        "SELECT COUNT(*) n FROM listings WHERE active=0 AND last_seen>=?"
        " AND last_seen<?" + VISIBLE, (ep_pfrm, ep_pto)).fetchone()["n"] if prev_ok else 0
    conn_cur = con.execute(
        "SELECT COUNT(DISTINCT source) n FROM sync_log WHERE ok=1 AND ts>=?"
        " AND ts<?", (ep_frm, ep_to)).fetchone()["n"]

    act_now = snap["n"]
    act_prev = actives_by_day.get(_iso(p_to)) if prev_ok else None

    def kpi(id_, lbl, val, unit="", dpct=None):
        k = {"id": id_, "label": lbl, "value": val, "unit": unit}
        if dpct is not None:
            k["delta_pct"] = dpct
            k["direction"] = "up" if dpct >= 0 else "down"
        return k

    kpis = [
        kpi("actives", "Annonces actives", act_now, "",
            _fmt_pct(act_now, act_prev) if act_prev else None),
        kpi("nouvelles", "Nouvelles annonces (période)", new_cur, "",
            _fmt_pct(new_cur, new_prev) if prev_ok and new_prev else None),
        kpi("retirees", "Vendues / retirées (période)", gone_cur, "",
            _fmt_pct(gone_cur, gone_prev) if prev_ok and gone_prev else None),
    ]
    if snap["avg_p"]:
        kpis.append(kpi("prix_moyen", "Prix moyen demandé",
                        round(snap["avg_p"]), "$"))
    if med_price:
        kpis.append(kpi("prix_median", "Prix médian demandé",
                        round(med_price), "$"))
    if med_ppm2 and n_ppm2 >= 100:
        kpis.append(kpi("prix_m2",
                        f"Prix médian au m² ({n_ppm2:,} annonces avec superficie)".replace(",", " "),
                        round(med_ppm2), "$/m²"))
    # prix au pi² déclaré à la source (details.prix_pi2) — médiane
    ppi2 = [r["v"] for r in con.execute(
        "SELECT CAST(json_extract(details,'$.prix_pi2') AS REAL) v"
        " FROM listings WHERE active=1" + VISIBLE +
        " AND CAST(json_extract(details,'$.prix_pi2') AS REAL)"
        " BETWEEN 30 AND 10000")]
    med_ppi2 = _median(ppi2)
    if med_ppi2 and len(ppi2) >= 100:
        kpis.append(kpi(
            "prix_pi2",
            f"Prix médian au pi² ({len(ppi2):,} annonces le déclarant)".replace(",", " "),
            round(med_ppi2), "$/pi²"))
    # jours sur le marché (annonces actives) — médiane depuis first_seen
    now_ts = time.time()
    dom = [max((now_ts - r["fs"]) / 86400.0, 0.0) for r in con.execute(
        "SELECT first_seen fs FROM listings WHERE active=1" + VISIBLE)]
    med_dom = _median(dom)
    if med_dom is not None and dom:
        kpis.append(kpi("jours_marche", "Jours sur le marché (médiane, actives)",
                        round(med_dom, 1), "j"))
    # baisses de prix observées dans la période (journal price_log) —
    # une entrée par annonce, bornes de plausibilité (voir en tête de fichier)
    drops = con.execute(
        """SELECT l.city city, MAX(p1.price - p2.price) amt,
                  date(MAX(p2.ts),'unixepoch','localtime') dt
           FROM price_log p1
           JOIN price_log p2 ON p2.uid = p1.uid AND p2.ts > p1.ts
           JOIN listings l ON l.uid = p1.uid
           WHERE p2.ts>=? AND p2.ts<? AND p2.price < p1.price
             AND p1.price BETWEEN ? AND ? AND p2.price BETWEEN ? AND ?
             AND p2.price >= p1.price * 0.5 AND l.dup_hidden=0 AND l.published=1
           GROUP BY l.uid ORDER BY amt DESC""",
        (ep_frm, ep_to, PRICE_MIN, PRICE_MAX, PRICE_MIN, PRICE_MAX)).fetchall()
    kpis.append(kpi("baisses_prix", "Baisses de prix observées (période)",
                    len(drops)))
    # qualité des données (quality.py) : score moyen + quarantaine
    qual = con.execute(
        "SELECT ROUND(AVG(quality_score),1) c FROM listings WHERE active=1"
        + VISIBLE).fetchone()
    quar = con.execute(
        "SELECT COUNT(*) n FROM listings"
        " WHERE active=1 AND dup_hidden=0 AND published=0").fetchone()["n"]
    if qual["c"] is not None:
        kpis.append(kpi("qualite", "Score de qualité moyen des fiches",
                        qual["c"], "/100"))
        kpis.append(kpi("quarantaine", "Annonces en quarantaine (qualité)",
                        quar))
    # position des prix vs estimation Vrai-Prix (juste valeur) — un seul
    # balayage réutilisé par le KPI, l'anneau et le tableau par ville
    b_lo, b_hi = FV_DEV_BOUNDS
    fv_rows = con.execute(
        "SELECT city, (price - CAST(json_extract(vraiprix,'$.value') AS REAL))"
        " / CAST(json_extract(vraiprix,'$.value') AS REAL) dev"
        " FROM listings WHERE active=1" + VISIBLE +
        " AND CAST(json_extract(vraiprix,'$.value') AS REAL) > 0"
        " AND price BETWEEN ? AND ?", (PRICE_MIN, PRICE_MAX)).fetchall()
    fv_sous = fv_marche = fv_sur = 0
    fv_city: dict[str, list[float]] = {}
    for r in fv_rows:
        dev = r["dev"]
        if dev is None or not (b_lo < dev < b_hi):
            continue
        if dev <= FV_SEUIL_SOUS:
            fv_sous += 1
        elif dev >= FV_SEUIL_SUR:
            fv_sur += 1
        else:
            fv_marche += 1
        if r["city"]:
            fv_city.setdefault(r["city"], []).append(dev)
    fv_n = fv_sous + fv_marche + fv_sur
    if fv_n:
        kpis.append(kpi("sous_marche", "Annonces sous l'estimation Vrai-Prix",
                        fv_sous))
    kpis.append(kpi("villes", "Villes couvertes", snap["cities"]))
    kpis.append(kpi("connecteurs", "Connecteurs actifs (période)", conn_cur))
    # indice de tension : retraits / nouvelles entrées (mesuré, pas modélisé)
    if new_cur >= 50:
        kpis.append(kpi("tension", "Tension — retraits / nouvelles",
                        round(100.0 * gone_cur / new_cur, 1), "%"))

    # ---- jauges (v2) : couvertures mesurées sur les annonces publiées --------
    gauges: list[dict] = []
    if act_now:
        g_geo = con.execute(
            "SELECT COUNT(*) n FROM listings WHERE active=1"
            " AND lat IS NOT NULL AND lng IS NOT NULL" + VISIBLE).fetchone()["n"]
        g_photo = con.execute(
            "SELECT COUNT(*) n FROM listings WHERE active=1"
            " AND images IS NOT NULL AND images<>'' AND images<>'[]'"
            + VISIBLE).fetchone()["n"]
        g_vp = con.execute(
            "SELECT COUNT(*) n FROM listings WHERE active=1" + VISIBLE +
            " AND CAST(json_extract(vraiprix,'$.value') AS REAL) > 0"
        ).fetchone()["n"]
        gauges.append({"id": "geoloc", "label": "Fiches géolocalisées",
                       "value": round(100.0 * g_geo / act_now, 1),
                       "max": 100, "unit": "%"})
        gauges.append({"id": "photos", "label": "Fiches avec photos",
                       "value": round(100.0 * g_photo / act_now, 1),
                       "max": 100, "unit": "%"})
        if g_vp:
            gauges.append({"id": "vraiprix",
                           "label": "Fiches avec estimation Vrai-Prix",
                           "value": round(100.0 * g_vp / act_now, 1),
                           "max": 100, "unit": "%"})
    act_all = con.execute(
        "SELECT COUNT(*) n FROM listings WHERE active=1 AND dup_hidden=0"
    ).fetchone()["n"]
    if act_all:
        gauges.append({"id": "publiable",
                       "label": "Hors quarantaine (qualité publiable)",
                       "value": round(100.0 * (act_all - quar) / act_all, 1),
                       "max": 100, "unit": "%"})
    if qual["c"] is not None:
        gauges.append({"id": "completude",
                       "label": "Complétude moyenne des fiches (0–100)",
                       "value": qual["c"], "max": 100})

    # ---- séries quotidiennes ---------------------------------------------------
    days = [_iso(d) for d in _daterange(s_frm, s_to)]
    new_by_day = {r["d"]: r["n"] for r in con.execute(
        "SELECT date(first_seen,'unixepoch','localtime') d, COUNT(*) n"
        " FROM listings WHERE first_seen>=? AND first_seen<?" + VISIBLE +
        " GROUP BY d", (ep_frm, ep_to))}
    gone_by_day = {r["d"]: r["n"] for r in con.execute(
        "SELECT date(last_seen,'unixepoch','localtime') d, COUNT(*) n"
        " FROM listings WHERE active=0 AND last_seen>=? AND last_seen<?"
        + VISIBLE + " GROUP BY d", (ep_frm, ep_to))}
    series = []
    if len(days) >= 2:
        series = [
            {"id": "actives", "title": "Annonces actives par jour",
             "unit": "annonces", "kind": "line",
             "points": [{"t": d, "v": actives_by_day.get(d, 0)} for d in days]},
            {"id": "nouvelles", "title": "Nouvelles annonces par jour",
             "unit": "annonces", "kind": "bar",
             "points": [{"t": d, "v": new_by_day.get(d, 0)} for d in days]},
            {"id": "retraits", "title": "Retraits (vendues / retirées) par jour",
             "unit": "annonces", "kind": "bar",
             "points": [{"t": d, "v": gone_by_day.get(d, 0)} for d in days]},
        ]
        # prix médian demandé des nouvelles inscriptions (jours à >= 3 entrées
        # seulement — rien d'interpolé, l'axe saute les jours creux)
        med_day: list[dict] = []
        day_prices: dict[str, list[float]] = {}
        for r in con.execute(
                "SELECT date(first_seen,'unixepoch','localtime') d, price"
                " FROM listings WHERE first_seen>=? AND first_seen<?" + VISIBLE +
                " AND price BETWEEN ? AND ?",
                (ep_frm, ep_to, PRICE_MIN, PRICE_MAX)):
            day_prices.setdefault(r["d"], []).append(r["price"])
        for d in days:
            ps = day_prices.get(d)
            if ps and len(ps) >= 3:
                med_day.append({"t": d, "v": round(statistics.median(ps))})
        if len(med_day) >= 5:
            series.append({
                "id": "prix_median_nouvelles",
                "title": "Prix médian demandé des nouvelles inscriptions"
                         " (jours à ≥ 3 entrées)",
                "unit": "$", "kind": "area", "points": med_day})
        # comparaison période précédente (même longueur) : seulement si elle
        # a réellement été observée en entier (rien d'extrapolé)
        if prev_ok:
            pdays = [_iso(d) for d in _daterange(p_frm, p_to)]
            series[0]["compare"] = [{"t": d, "v": actives_by_day.get(d, 0)}
                                    for d in pdays]
            cmp_new = {r["d"]: r["n"] for r in con.execute(
                "SELECT date(first_seen,'unixepoch','localtime') d, COUNT(*) n"
                " FROM listings WHERE first_seen>=? AND first_seen<?" + VISIBLE +
                " GROUP BY d", (ep_pfrm, ep_pto))}
            series[1]["compare"] = [{"t": d, "v": cmp_new.get(d, 0)}
                                    for d in pdays]

    # ---- sparklines des KPI (v2) — mêmes données que les séries ---------------
    if len(days) >= 2:
        conn_day = {r["d"]: r["n"] for r in con.execute(
            "SELECT date(ts,'unixepoch','localtime') d,"
            " COUNT(DISTINCT source) n FROM sync_log"
            " WHERE ok=1 AND ts>=? AND ts<? GROUP BY d", (ep_frm, ep_to))}
        sparks = {
            "actives": [{"t": d, "v": actives_by_day.get(d, 0)} for d in days],
            "nouvelles": [{"t": d, "v": new_by_day.get(d, 0)} for d in days],
            "retirees": [{"t": d, "v": gone_by_day.get(d, 0)} for d in days],
            "connecteurs": [{"t": d, "v": conn_day.get(d, 0)} for d in days],
        }
        for k in kpis:
            sp = sparks.get(k["id"])
            if sp and any(p["v"] for p in sp):
                k["spark"] = _downsample(sp)

    # ---- multi-courbes (v2) : prix médian des nouvelles inscriptions ----------
    # Buckets quotidiens (<= 45 j) ou hebdomadaires ; un bucket n'est gardé que
    # si CHAQUE groupe y compte >= 3 inscriptions (axes alignés, rien d'inventé).
    def _multiserie(id_, title, group_sql, top_n):
        weekly = ndays > 45
        bucket_sql = ("strftime('%Y-%m-%d', first_seen, 'unixepoch',"
                      " 'localtime', 'weekday 1', '-6 days')" if weekly else
                      "date(first_seen,'unixepoch','localtime')")
        rows = con.execute(
            f"SELECT {group_sql} g, {bucket_sql} b, price FROM listings"
            " WHERE first_seen>=? AND first_seen<?" + VISIBLE +
            f" AND price BETWEEN ? AND ? AND {group_sql} <> ''",
            (ep_frm, ep_to, PRICE_MIN, PRICE_MAX)).fetchall()
        vol: dict[str, int] = {}
        data: dict[str, dict[str, list[float]]] = {}
        for r in rows:
            if "/" in r["g"]:       # libellés composites de sources ("Laval /
                continue            # North Shore") — bruit, pas une vraie ville
            vol[r["g"]] = vol.get(r["g"], 0) + 1
            data.setdefault(r["g"], {}).setdefault(r["b"], []).append(r["price"])
        groups = [g for g, _ in sorted(vol.items(), key=lambda kv: -kv[1])[:top_n]]
        if len(groups) < 2:
            return None
        buckets = sorted({b for g in groups for b in data[g]
                          if all(len(data[gg].get(b, [])) >= 3 for gg in groups)})
        if len(buckets) < 4:
            return None
        return {"id": id_, "title": title + (" (semaines)" if weekly else ""),
                "unit": "$",
                "series": [{"label": g, "points": [
                    {"t": b, "v": round(statistics.median(data[g][b]))}
                    for b in buckets]} for g in groups]}

    multiseries = []
    ms_type = _multiserie(
        "prix_type", "Prix médian des nouvelles inscriptions par type",
        "property_type", 3)
    if ms_type:
        multiseries.append(ms_type)
    ms_ville = _multiserie(
        "prix_ville", "Prix médian des nouvelles inscriptions — grandes villes",
        "city", 4)
    if ms_ville:
        multiseries.append(ms_ville)

    # ---- barres empilées (v2) : nouvelles inscriptions par bannière -----------
    names = _source_names()
    stacked = []
    if len(days) >= 2:
        weekly_st = ndays > 60
        bucket_st = ("strftime('%Y-%m-%d', first_seen, 'unixepoch',"
                     " 'localtime', 'weekday 1', '-6 days')" if weekly_st else
                     "date(first_seen,'unixepoch','localtime')")
        fam_day: dict[str, dict[str, int]] = {}
        fam_tot: dict[str, int] = {}
        for r in con.execute(
                f"SELECT source s, {bucket_st} b, COUNT(*) n FROM listings"
                " WHERE first_seen>=? AND first_seen<?" + VISIBLE +
                " GROUP BY source, b", (ep_frm, ep_to)):
            fam = _famille_of(r["s"], names)
            fam_day.setdefault(fam, {})
            fam_day[fam][r["b"]] = fam_day[fam].get(r["b"], 0) + r["n"]
            fam_tot[fam] = fam_tot.get(fam, 0) + r["n"]
        if fam_tot:
            top_fams = [f for f, _ in
                        sorted(fam_tot.items(), key=lambda kv: -kv[1])[:5]]
            others = [f for f in fam_day if f not in top_fams]
            keys = top_fams + (["Autres"] if others else [])
            buckets_st = sorted({b for d_ in fam_day.values() for b in d_})
            pts = []
            for b in buckets_st:
                vals = [fam_day[f].get(b, 0) for f in top_fams]
                if others:
                    vals.append(sum(fam_day[f].get(b, 0) for f in others))
                pts.append({"t": b, "values": vals})
            if len(pts) >= 2:
                stacked.append({
                    "id": "ajouts_bannieres",
                    "title": "Nouvelles inscriptions par bannière"
                             + (" (semaines)" if weekly_st else ""),
                    "unit": "inscriptions", "keys": keys, "points": pts})

    # ---- distributions (v2) : prix, superficie, année de construction ---------
    distributions = []
    price_bins = [("< 100 k$", 0, 100e3)] + [
        (f"{i}00–{i+1}00 k$", i * 100e3, (i + 1) * 100e3) for i in range(1, 10)
    ] + [("1–1,5 M$", 1e6, 1.5e6), ("1,5–2 M$", 1.5e6, 2e6),
         ("2 M$ +", 2e6, None)]
    bins_p = []
    for lbl, lo, hi in price_bins:
        q = ("SELECT COUNT(*) n FROM listings WHERE active=1 AND price>=?"
             + VISIBLE)
        args: list = [lo]
        if hi is not None:
            q += " AND price<?"
            args.append(hi)
        bins_p.append({"label": lbl,
                       "value": con.execute(q, args).fetchone()["n"]})
    if sum(b["value"] for b in bins_p):
        distributions.append({
            "id": "prix", "unit": "annonces",
            "title": "Distribution des prix demandés (annonces actives)",
            "bins": bins_p})
    area_bins = [("< 500", 100, 500), ("500–1 000", 500, 1000),
                 ("1 000–1 500", 1000, 1500), ("1 500–2 000", 1500, 2000),
                 ("2 000–2 500", 2000, 2500), ("2 500–3 000", 2500, 3000),
                 ("3 000–4 000", 3000, 4000), ("4 000 +", 4000, 20000)]
    bins_a = [{"label": f"{lbl} pi²",
               "value": con.execute(
                   "SELECT COUNT(*) n FROM listings WHERE active=1"
                   " AND area_sqft>=? AND area_sqft<?" + VISIBLE,
                   (lo, hi)).fetchone()["n"]}
              for lbl, lo, hi in area_bins]
    if sum(b["value"] for b in bins_a) >= 100:
        distributions.append({
            "id": "superficie", "unit": "annonces",
            "title": "Distribution des superficies habitables (renseignées)",
            "bins": bins_a})
    yr_now = today.year
    year_bins = ([("< 1900", 1600, 1900), ("1900–1949", 1900, 1950)] +
                 [(f"{d}–{d+9}", d, d + 10) for d in range(1950, 2020, 10)] +
                 [("2020 +", 2020, yr_now + 2)])
    bins_y = [{"label": lbl,
               "value": con.execute(
                   "SELECT COUNT(*) n FROM listings WHERE active=1"
                   " AND year_built>=? AND year_built<?" + VISIBLE,
                   (lo, hi)).fetchone()["n"]}
              for lbl, lo, hi in year_bins]
    if sum(b["value"] for b in bins_y) >= 100:
        distributions.append({
            "id": "annee", "unit": "annonces",
            "title": "Distribution des années de construction (renseignées)",
            "bins": bins_y})

    # ---- heatmap horaire (v2) : détection des nouvelles annonces (7×24) -------
    # first_seen = moment où la synchronisation a détecté l'annonce — c'est le
    # rythme réel d'alimentation de la plateforme (8 dernières semaines).
    h56 = _epoch(max(data_start, s_to - timedelta(days=55)))
    hourly_cells = [
        {"dow": (int(r["w"]) + 6) % 7, "hour": int(r["h"]), "value": r["n"]}
        for r in con.execute(
            "SELECT strftime('%w', first_seen,'unixepoch','localtime') w,"
            " strftime('%H', first_seen,'unixepoch','localtime') h, COUNT(*) n"
            " FROM listings WHERE first_seen>=? AND first_seen<?" + VISIBLE +
            " GROUP BY w, h", (h56, _epoch(s_to + timedelta(days=1))))]
    hourly = ({"title": "Détection de nouvelles annonces par heure"
                        " (8 dernières semaines)", "cells": hourly_cells}
              if len(hourly_cells) >= 12 else None)

    # ---- répartitions (photo des annonces actives) -----------------------------
    types = [{"label": r["t"] or "Autre / non précisé", "value": r["n"]}
             for r in con.execute(
                 "SELECT property_type t, COUNT(*) n FROM listings"
                 " WHERE active=1" + VISIBLE +
                 " GROUP BY property_type ORDER BY n DESC LIMIT 9")]
    ranges = [("Moins de 200 k$", 0, 200e3), ("200 – 300 k$", 200e3, 300e3),
              ("300 – 400 k$", 300e3, 400e3), ("400 – 500 k$", 400e3, 500e3),
              ("500 – 750 k$", 500e3, 750e3), ("750 k$ – 1 M$", 750e3, 1e6),
              ("1 – 2 M$", 1e6, 2e6), ("2 M$ et plus", 2e6, None)]
    price_items = []
    for lbl, lo, hi in ranges:
        q = "SELECT COUNT(*) n FROM listings WHERE active=1 AND price>=?" + VISIBLE
        args: list = [lo]
        if hi is not None:
            q += " AND price<?"
            args.append(hi)
        price_items.append({"label": lbl,
                            "value": con.execute(q, args).fetchone()["n"]})
    # nouvelles inscriptions de la période par fourchette + delta honnête
    # (vs période précédente entièrement observée seulement)
    new_price_items = []
    for lbl, lo, hi in ranges:
        base = ("SELECT COUNT(*) n FROM listings WHERE first_seen>=?"
                " AND first_seen<? AND price>=?" + VISIBLE)
        args_c: list = [ep_frm, ep_to, lo]
        args_p: list = [ep_pfrm, ep_pto, lo]
        if hi is not None:
            base += " AND price<?"
            args_c.append(hi)
            args_p.append(hi)
        n_c = con.execute(base, args_c).fetchone()["n"]
        item = {"label": lbl, "value": n_c}
        if prev_ok:
            n_p = con.execute(base, args_p).fetchone()["n"]
            if n_p:
                item["delta_pct"] = _fmt_pct(n_c, n_p)
        new_price_items.append(item)
    beds = [{"label": ("8 chambres et +" if r["b"] >= 8
                       else f"{int(r['b'])} chambre" + ("s" if r["b"] > 1 else "")),
             "value": r["n"]}
            for r in con.execute(
                "SELECT MIN(bedrooms,8) b, COUNT(*) n FROM listings"
                " WHERE active=1 AND bedrooms IS NOT NULL" + VISIBLE +
                " GROUP BY MIN(bedrooms,8) ORDER BY b")]
    names = _source_names()
    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" + VISIBLE +
                     " GROUP BY source ORDER BY n DESC LIMIT 12")]
    breakdowns = []
    if fv_n:
        breakdowns.append({
            "id": "fairvalue",
            "title": "Position des prix demandés vs estimation Vrai-Prix",
            "kind": "donut", "items": [
                {"label": "Sous le marché", "value": fv_sous},
                {"label": "Dans le marché", "value": fv_marche},
                {"label": "Au-dessus du marché", "value": fv_sur}]})
    breakdowns += [
        {"id": "types", "title": "Répartition par type de propriété",
         "kind": "donut", "items": types},
        {"id": "prix", "title": "Répartition par fourchette de prix demandé",
         "kind": "bar", "items": price_items},
    ]
    if sum(i["value"] for i in new_price_items):
        breakdowns.append({
            "id": "prix_nouvelles",
            "title": "Nouvelles inscriptions par fourchette de prix (période)",
            "kind": "bar", "items": new_price_items})
    if beds:
        breakdowns.append({"id": "chambres",
                           "title": "Répartition par nombre de chambres (renseignées)",
                           "kind": "bar", "items": beds})
    if by_source:
        breakdowns.append({"id": "sources",
                           "title": "Top sources (annonces actives)",
                           "kind": "bar", "items": by_source})

    # ---- géographie : par région (fusion accents/casse, libellé le + fréquent)
    reg_counts: dict[str, dict[str, int]] = {}
    for r in con.execute(
            "SELECT region, COUNT(*) n FROM listings WHERE active=1"
            " AND region<>''" + VISIBLE + " GROUP BY region"):
        raw = (r["region"] or "").strip()
        key = _fold(raw)
        if not key or key.isdigit():
            continue
        reg_counts.setdefault(key, {})[raw] = reg_counts.get(key, {}).get(raw, 0) + r["n"]
    geo_items = []
    for key, variants in reg_counts.items():
        best_variant = max(variants, key=variants.get)
        geo_items.append({"label": best_variant, "value": sum(variants.values())})
    geo_items.sort(key=lambda x: -x["value"])
    geo = ({"title": "Annonces actives par région", "items": geo_items[:14]}
           if geo_items else None)

    # ---- heatmap : nouvelles annonces par jour (26 dernières semaines max) ----
    h_frm = max(data_start, s_to - timedelta(days=181))
    hm = [{"date": r["d"], "value": r["n"]} for r in con.execute(
        "SELECT date(first_seen,'unixepoch','localtime') d, COUNT(*) n"
        " FROM listings WHERE first_seen>=? AND first_seen<?" + VISIBLE +
        " GROUP BY d", (_epoch(h_frm), _epoch(s_to + timedelta(days=1))))]
    heatmap = ({"title": "Nouvelles annonces par jour", "cells": hm}
               if len(hm) >= 2 else None)

    # ---- tableaux ---------------------------------------------------------------
    # Top villes : actives, prix moyen/médian, nouvelles sur la période + delta
    city_prices: dict[str, list[float]] = {}
    for r in con.execute(
            "SELECT city, price FROM listings WHERE active=1 AND city<>''"
            + VISIBLE):
        city_prices.setdefault(r["city"], []).append(r["price"])
    new_city = {r["city"]: r["n"] for r in con.execute(
        "SELECT city, COUNT(*) n FROM listings WHERE city<>''"
        " AND first_seen>=? AND first_seen<?" + VISIBLE + " GROUP BY city",
        (ep_frm, ep_to))}
    new_city_prev = {r["city"]: r["n"] for r in con.execute(
        "SELECT city, COUNT(*) n FROM listings WHERE city<>''"
        " AND first_seen>=? AND first_seen<?" + VISIBLE + " GROUP BY city",
        (ep_pfrm, ep_pto))} if prev_ok else {}
    gone_city = {r["city"]: r["n"] for r in con.execute(
        "SELECT city, COUNT(*) n FROM listings WHERE active=0 AND city<>''"
        " AND last_seen>=? AND last_seen<?" + VISIBLE + " GROUP BY city",
        (ep_frm, ep_to))}
    top = sorted(city_prices.items(), key=lambda kv: -len(kv[1]))[:50]
    top_rows = []
    for city, ps in top:
        n_new = new_city.get(city, 0)
        n_prev = new_city_prev.get(city, 0)
        net = n_new - gone_city.get(city, 0)
        d = _fmt_pct(n_new, n_prev) if prev_ok and n_prev else None
        pn = [v for v in ps if isinstance(v, (int, float))]  # prix NULL écartés
        top_rows.append([
            city, len(ps),
            _fmt_money(sum(pn) / len(pn)) if pn else "—",
            _fmt_money(statistics.median(pn)) if pn else "—", n_new,
            f"{'+' if net >= 0 else ''}{net}",
            (f"{'+' if d >= 0 else ''}{str(d).replace('.', ',')} %"
             if d is not None else "—"),
        ])
    tables = [{
        "id": "top_villes", "title": "Top villes",
        "columns": ["Ville", "Actives", "Prix moyen", "Prix médian",
                    "Nouvelles (période)", "Δ net (période)", "Var. nouvelles"],
        "rows": top_rows,
    }]
    # Top sources : actives, prix moyen, nouvelles, qualité, quarantaine, synchro
    top_srcs = con.execute(
        """SELECT source s, COUNT(*) n,
                  AVG(CASE WHEN price>0 THEN price END) avg_p,
                  SUM(CASE WHEN first_seen>=? AND first_seen<? THEN 1 ELSE 0 END) new_n,
                  ROUND(AVG(quality_score),0) qual
           FROM listings WHERE active=1""" + VISIBLE +
        " GROUP BY source ORDER BY n DESC LIMIT 50", (ep_frm, ep_to)).fetchall()
    quar_src = {r["s"]: r["n"] for r in con.execute(
        "SELECT source s, COUNT(*) n FROM listings"
        " WHERE active=1 AND dup_hidden=0 AND published=0 GROUP BY source")}
    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:
        src_rows = []
        for r in top_srcs:
            ls = last_sync.get(r["s"])
            src_rows.append([
                names.get(r["s"], r["s"]), r["n"],
                _fmt_money(r["avg_p"]) if r["avg_p"] else "—",
                r["new_n"],
                f"{r['qual']:.0f} /100" if r["qual"] is not None else "—",
                quar_src.get(r["s"], 0),
                (datetime.fromtimestamp(ls, TZ).strftime("%Y-%m-%d %H:%M")
                 if ls else "—")])
        tables.append({
            "id": "top_sources", "title": "Top sources & courtiers",
            "columns": ["Source", "Annonces actives", "Prix moyen",
                        "Nouvelles (période)", "Qualité", "Quarantaine",
                        "Dernière synchro"],
            "rows": src_rows})
    # Délai de présence (retirées de la période) par ville
    dur_city: dict[str, list[float]] = {}
    for r in con.execute(
            "SELECT city, (last_seen-first_seen)/86400.0 d FROM listings"
            " WHERE active=0 AND city<>'' AND last_seen>=? AND last_seen<?"
            + VISIBLE, (ep_frm, ep_to)):
        dur_city.setdefault(r["city"], []).append(max(r["d"], 0.0))
    dur_rows = []
    for city, ds in sorted(dur_city.items(), key=lambda kv: -len(kv[1]))[:50]:
        if len(ds) < 3:
            continue
        dur_rows.append([
            city, len(ds),
            str(round(sum(ds) / len(ds), 1)).replace(".", ","),
            str(round(statistics.median(ds), 1)).replace(".", ","),
        ])
    if dur_rows:
        tables.append({
            "id": "delai_villes",
            "title": "Délai de présence avant retrait, par ville (période)",
            "columns": ["Ville", "Retirées", "Délai moyen (j)", "Délai médian (j)"],
            "rows": dur_rows,
        })
    # Écart moyen à l'estimation Vrai-Prix par ville (volume suffisant)
    fv_rows_city = []
    for city, devs in sorted(fv_city.items(), key=lambda kv: -len(kv[1]))[:25]:
        if len(devs) < 50:
            continue
        dev_pct = round(100.0 * sum(devs) / len(devs), 1)
        fv_rows_city.append([
            city, len(devs),
            f"{'+' if dev_pct >= 0 else ''}{str(dev_pct).replace('.', ',')} %",
            sum(1 for d in devs if d <= FV_SEUIL_SOUS)])
    if fv_rows_city:
        tables.append({
            "id": "fv_villes",
            "title": "Écart à l'estimation Vrai-Prix par ville",
            "columns": ["Ville", "Annonces évaluées", "Écart moyen",
                        "Sous le marché"],
            "rows": fv_rows_city})
    # Couche qualité : anomalies & quarantaine par motif (quality.py)
    anomalies: dict[str, int] = {}
    for r in con.execute(
            "SELECT quality_issues FROM listings WHERE active=1"
            " AND dup_hidden=0 AND quality_issues IS NOT NULL"):
        try:
            for issue in json.loads(r["quality_issues"]):
                key = issue.split(":")[0]
                anomalies[key] = anomalies.get(key, 0) + 1
        except ValueError:
            continue
    if anomalies and act_all:
        tables.append({
            "id": "quarantaine_motifs",
            "title": "Couche qualité — anomalies et quarantaine par motif",
            "columns": ["Motif", "Annonces touchées", "% des actives"],
            "rows": [[_MOTIFS_QUALITE.get(k, k), v,
                      str(round(100.0 * v / act_all, 1)).replace(".", ",") + " %"]
                     for k, v in sorted(anomalies.items(), key=lambda kv: -kv[1])],
        })
    # Bannières / familles de connecteurs : volume, nouvelles, prix, fraîcheur
    fam_info: dict[str, dict] = {}
    for r in con.execute(
            "SELECT source s, price p FROM listings WHERE active=1" + VISIBLE):
        e = fam_info.setdefault(_famille_of(r["s"], names),
                                {"srcs": set(), "prices": [], "new": 0,
                                 "sync": None})
        e["srcs"].add(r["s"])
        e["prices"].append(r["p"])
    for r in con.execute(
            "SELECT source s, COUNT(*) n FROM listings WHERE first_seen>=?"
            " AND first_seen<?" + VISIBLE + " GROUP BY source",
            (ep_frm, ep_to)):
        fam = _famille_of(r["s"], names)
        if fam in fam_info:
            fam_info[fam]["new"] += r["n"]
    for src, ts in last_sync.items():
        fam = _famille_of(src, names)
        if fam in fam_info:
            e = fam_info[fam]
            e["sync"] = max(e["sync"] or 0, ts)
    if fam_info:
        fam_rows = []
        for fam, e in sorted(fam_info.items(),
                             key=lambda kv: -len(kv[1]["prices"]))[:30]:
            fam_rows.append([
                fam, len(e["srcs"]), len(e["prices"]), e["new"],
                (_fmt_money(statistics.median(pn))
                 if (pn := [v for v in e["prices"]
                            if isinstance(v, (int, float))]) else "—"),
                (datetime.fromtimestamp(e["sync"], TZ).strftime("%Y-%m-%d %H:%M")
                 if e["sync"] else "—")])
        tables.append({
            "id": "familles",
            "title": "Bannières & familles de connecteurs",
            "columns": ["Bannière / famille", "Connecteurs", "Annonces actives",
                        "Nouvelles (période)", "Prix médian",
                        "Dernière synchro"],
            "rows": fam_rows})

    # ---- records & faits marquants ---------------------------------------------
    records = []
    if new_by_day:
        best = max(new_by_day.items(), key=lambda kv: kv[1])
        records.append({"label": "Jour record de nouvelles annonces",
                        "value": f"{best[1]:,} annonces".replace(",", " "),
                        "date": best[0]})
    if gone_by_day:
        worst = max(gone_by_day.items(), key=lambda kv: kv[1])
        records.append({"label": "Jour record de retraits",
                        "value": f"{worst[1]:,} annonces".replace(",", " "),
                        "date": worst[0]})
    fast = con.execute(
        "SELECT city, address, (last_seen-first_seen)/86400.0 d,"
        " date(last_seen,'unixepoch','localtime') dt FROM listings"
        " WHERE active=0 AND last_seen>=? AND last_seen<?"
        " AND last_seen-first_seen>=3600"           # >= 1 h : écarte les artefacts de sync
        + VISIBLE + " ORDER BY (last_seen-first_seen) ASC LIMIT 1",
        (ep_frm, ep_to)).fetchone()
    if fast:
        d = fast["d"]
        val = (f"{round(d * 24, 1)} h" if d < 1 else f"{round(d, 1)} j").replace(".", ",")
        records.append({"label": "Retrait le plus rapide (mise en ligne → retrait)",
                        "value": val + (f" · {fast['city']}" if fast["city"] else ""),
                        "date": fast["dt"]})
    if drops:  # balayage price_log fait plus haut (KPI baisses_prix)
        drop = drops[0]
        records.append({"label": "Plus forte baisse de prix demandé",
                        "value": "−" + _fmt_money(drop["amt"]) +
                                 (f" · {drop['city']}" if drop["city"] else ""),
                        "date": drop["dt"]})
    if new_city:
        c, n = max(new_city.items(), key=lambda kv: kv[1])
        records.append({"label": "Ville la plus active (nouvelles annonces)",
                        "value": f"{c} — {n:,} annonces".replace(",", " ")})
    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'])}"
                                     f" — {src['new_n']:,}".replace(",", " ")})
    top_price = con.execute(
        "SELECT city, price FROM listings WHERE active=1" + VISIBLE +
        " AND price BETWEEN ? AND ? ORDER BY price DESC LIMIT 1",
        (PRICE_MIN, PRICE_MAX)).fetchone()
    if top_price:
        records.append({"label": "Inscription active la plus chère",
                        "value": _fmt_money(top_price["price"]) +
                                 (f" · {top_price['city']}"
                                  if top_price["city"] else "")})
    med_cities = {c: statistics.median(pn) for c, ps in city_prices.items()
                  if len(pn := [v for v in ps
                                if isinstance(v, (int, float))]) >= 30}
    if med_cities:
        c_hi = max(med_cities, key=med_cities.get)
        c_lo = min(med_cities, key=med_cities.get)
        records.append({"label": "Ville la plus chère (prix médian, ≥ 30 annonces)",
                        "value": f"{c_hi} — {_fmt_money(med_cities[c_hi])}"})
        records.append({"label": "Ville la plus abordable (prix médian, ≥ 30 annonces)",
                        "value": f"{c_lo} — {_fmt_money(med_cities[c_lo])}"})
    big_area = con.execute(
        "SELECT city, area_sqft a FROM listings WHERE active=1" + VISIBLE +
        " AND area_sqft BETWEEN 100 AND 50000"
        " ORDER BY area_sqft DESC LIMIT 1").fetchone()
    if big_area:
        records.append({"label": "Plus grande superficie habitable (plausible)",
                        "value": f"{round(big_area['a']):,} pi²".replace(",", " ") +
                                 (f" · {big_area['city']}"
                                  if big_area["city"] else "")})
    if fam_info:
        fam_big = max(fam_info.items(), key=lambda kv: len(kv[1]["srcs"]))
        if len(fam_big[1]["srcs"]) > 1:
            records.append({"label": "Bannière au plus grand réseau agrégé",
                            "value": f"{fam_big[0]} — "
                                     f"{len(fam_big[1]['srcs'])} connecteurs"})

    out = {
        "updated": datetime.now(TZ).isoformat(timespec="seconds"),
        "period": {"from": _iso(frm), "to": _iso(to), "label": label,
                   "observed_from": _iso(data_start)},
        "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


def dashboard(period: str | None = None, frm: str | None = None,
              to: str | None = None) -> dict:
    key = f"{period or ''}|{frm or ''}|{to or ''}"
    now = time.time()
    with _CACHE_LOCK:
        hit = _CACHE.get(key)
        if hit and now - hit[0] < _CACHE_TTL:
            return hit[1]
    data = _compute(frm, to, period)
    with _CACHE_LOCK:
        _CACHE[key] = (time.time(), data)
        # garder le cache borné
        if len(_CACHE) > 64:
            for k in sorted(_CACHE, key=lambda k: _CACHE[k][0])[:32]:
                _CACHE.pop(k, None)
    return data