spb/food-ka Public

1# -----------------------------------------------------------------------------2# Food-Ka — Agrégateur de produits d'épicerie (province de Québec)3# Auteur : Simon-Pierre Boucher — contact@spboucher.ai4# marketstats.py : agrégats du marché — source unique de /api/stats/detailed5#   Métriques par bannière (prix médian, part de soldes, rabais moyen…),6#   matrice catégorie × bannière, panier comparatif, baisses de prix récentes.7# -----------------------------------------------------------------------------8from __future__ import annotations910import statistics11import time1213from . import db1415# Panier comparatif : produits courants repérés par mots-clés (fr + en).16# Pour chaque bannière on prend le PRIX MÉDIAN des produits correspondants —17# robuste face aux formats extrêmes (mini-format, caisse de grossiste).18BASKET = [19    ("Lait", ["lait", "milk"]),20    ("Œufs", ["oeuf", "œuf", "egg"]),21    ("Pain", ["pain", "bread"]),22    ("Beurre", ["beurre", "butter"]),23    ("Bananes", ["banane", "banana"]),24    ("Poulet", ["poulet", "chicken"]),25    ("Fromage", ["fromage", "cheese"]),26    ("Riz", ["riz", "rice"]),27    ("Pâtes", ["pâtes", "pates", "spaghetti", "pasta"]),28    ("Café", ["café", "cafe", "coffee"]),29]3031_PRICE_SANE = "price IS NOT NULL AND price > 0 AND price <= 2000"323334def _median(values: list[float]) -> float | None:35    return round(statistics.median(values), 2) if values else None363738def compute() -> dict:39    con = db.connect()40    now = time.time()4142    # -- par bannière ----------------------------------------------------------43    by_source: dict[str, dict] = {}44    for r in con.execute(45            f"""SELECT source, COUNT(*) n, SUM(on_sale) sales,46                       AVG(price) avg_price,47                       SUM(CASE WHEN unit_price IS NOT NULL THEN 1 ELSE 0 END) with_unit,48                       SUM(CASE WHEN {_PRICE_SANE} THEN 0 ELSE 1 END) no_price49                FROM products WHERE active=1 GROUP BY source"""):50        by_source[r["source"]] = {51            "source": r["source"], "n": r["n"], "sales": r["sales"] or 0,52            "sale_share": round((r["sales"] or 0) / r["n"], 3) if r["n"] else 0,53            "avg_price": round(r["avg_price"], 2) if r["avg_price"] else None,54            "with_unit_price": r["with_unit"], "no_price": r["no_price"],55        }56    # médiane des prix + rabais moyen (nécessite les valeurs)57    for src, d in by_source.items():58        prices = [r["price"] for r in con.execute(59            f"SELECT price FROM products WHERE active=1 AND source=? AND {_PRICE_SANE}",60            (src,))]61        d["median_price"] = _median(prices)62        discounts = [(r["regular_price"] - r["price"]) / r["regular_price"]63                     for r in con.execute(64                         f"""SELECT price, regular_price FROM products65                             WHERE active=1 AND source=? AND on_sale=166                             AND regular_price IS NOT NULL AND {_PRICE_SANE}67                             AND regular_price > price""", (src,))]68        d["avg_discount_pct"] = round(100 * statistics.mean(discounts), 1) if discounts else None69        d["max_discount_pct"] = round(100 * max(discounts), 1) if discounts else None7071    # -- matrice catégorie × bannière (prix médian) ----------------------------72    matrix: dict[str, dict[str, dict]] = {}73    rows = con.execute(74        f"""SELECT category, source, price FROM products75            WHERE active=1 AND category<>'' AND category<>'Autres' AND {_PRICE_SANE}"""76    ).fetchall()77    bucket: dict[tuple[str, str], list[float]] = {}78    for r in rows:79        bucket.setdefault((r["category"], r["source"]), []).append(r["price"])80    for (cat, src), prices in bucket.items():81        matrix.setdefault(cat, {})[src] = {82            "median_price": _median(prices), "n": len(prices)}8384    # -- panier comparatif ------------------------------------------------------85    basket: list[dict] = []86    for label, keywords in BASKET:87        like = " OR ".join(["name LIKE ?"] * len(keywords))88        args = [f"%{k}%" for k in keywords]89        per_source: dict[str, dict] = {}90        for r in con.execute(91                f"""SELECT source, price FROM products92                    WHERE active=1 AND {_PRICE_SANE} AND ({like})""", args):93            per_source.setdefault(r["source"], []).append(r["price"])94        basket.append({95            "item": label,96            "by_source": {src: {"median_price": _median(v), "n": len(v)}97                          for src, v in per_source.items()},98        })99    # total du panier par bannière (bannières couvrant >= 6 items sur 10)100    basket_totals = []101    for src in by_source:102        items = [b["by_source"][src]["median_price"] for b in basket103                 if src in b["by_source"] and b["by_source"][src]["median_price"]]104        if len(items) >= 6:105            basket_totals.append({"source": src, "items": len(items),106                                  "total": round(sum(items), 2)})107    basket_totals.sort(key=lambda x: x["total"] / max(x["items"], 1))108109    # -- baisses de prix récentes (7 jours) -------------------------------------110    week_ago = now - 7 * 86400111    drops = []112    for r in con.execute(113            """SELECT p.uid, p.name, p.source, p.images, l.ts, l.price new_price,114                      (SELECT l2.price FROM price_log l2115                       WHERE l2.uid = l.uid AND l2.ts < l.ts116                       ORDER BY l2.ts DESC LIMIT 1) old_price117               FROM price_log l JOIN products p ON p.uid = l.uid118               WHERE l.ts >= ? AND p.active = 1119               ORDER BY l.ts DESC LIMIT 400""", (week_ago,)):120        old, new = r["old_price"], r["new_price"]121        if old and new and new < old:122            drops.append({"uid": r["uid"], "name": r["name"], "source": r["source"],123                          "old_price": old, "new_price": new,124                          "drop_pct": round(100 * (old - new) / old, 1)})125    drops.sort(key=lambda d: -d["drop_pct"])126    price_changes_7d = con.execute(127        "SELECT COUNT(*) c FROM price_log WHERE ts >= ?", (week_ago,)).fetchone()["c"]128129    # -- distribution des prix (paliers) ----------------------------------------130    buckets = [(0, 2), (2, 5), (5, 10), (10, 20), (20, 50), (50, 2000)]131    distribution = []132    for lo, hi in buckets:133        c = con.execute(134            f"SELECT COUNT(*) c FROM products WHERE active=1 AND {_PRICE_SANE}"135            " AND price >= ? AND price < ?", (lo, hi)).fetchone()["c"]136        label = f"{lo}–{hi} $" if hi <= 50 else f"{lo} $ +"137        distribution.append({"range": label, "n": c})138139    # -- global ------------------------------------------------------------------140    g = con.execute(141        f"""SELECT COUNT(*) total, SUM(on_sale) on_sale,142                   COUNT(DISTINCT source) sources,143                   COUNT(DISTINCT category) categories,144                   AVG(price) avg_price,145                   COUNT(DISTINCT brand) brands146            FROM products WHERE active=1""").fetchone()147    all_prices = [r["price"] for r in con.execute(148        f"SELECT price FROM products WHERE active=1 AND {_PRICE_SANE}")]149    con.close()150151    return {152        "global": {153            "total": g["total"], "on_sale": g["on_sale"] or 0,154            "sale_share": round((g["on_sale"] or 0) / g["total"], 3) if g["total"] else 0,155            "sources": g["sources"], "categories": g["categories"],156            "brands": g["brands"],157            "avg_price": round(g["avg_price"], 2) if g["avg_price"] else None,158            "median_price": _median(all_prices),159            "price_changes_7d": price_changes_7d,160        },161        "by_source": sorted(by_source.values(), key=lambda d: -d["n"]),162        "category_matrix": matrix,163        "basket": basket,164        "basket_totals": basket_totals,165        "price_drops": drops[:30],166        "price_distribution": distribution,167    }168