spb/qc-election

1# QC Élection Forecast — Plateforme de prévision électorale du Québec 20262# Auteur : Simon-Pierre Boucher3# Contact : contact@spboucher.ai4# https://www.qc-election.com5"""Attention Wikipédia — comportement de recherche d'information (v2.1).67Littérature : les pageviews Wikipédia ajoutent du pouvoir prédictif AU-DELÀ des8sondages et des fondamentaux (Smith & Gustafson, *Public Opinion Quarterly* 2017;9Salem & Stephany 2020 — « challenger's best friend ») : contrairement aux réseaux10sociaux, consulter une page est une recherche d'information privée, sans mécanisme11de rétroaction sociale, et l'attention se porte de façon disproportionnée vers les12partis en MOUVEMENT (gains de votes), surtout les challengers.1314Utilisation (prudente, jamais directionnelle) :15  1. **Signal affiché** : parts d'attention 7 j / 28 j par parti (page du parti +16     pages des chef·fe·s, fr.wikipédia), tendance vs base de 180 j.17  2. **Détecteur de turbulence** : un pic d'attention global (z ≥ seuil) signale18     un moment chaud de campagne → la variance de dérive du forecast est GONFLÉE19     (multiplicateur borné ≤ 1,45). L'attention ne déplace jamais la moyenne —20     un pic peut être bon ou mauvais pour un parti; il annonce du mouvement.2122Source : API REST Wikimedia (pageviews per-article), publique, sans clé.23"""24from __future__ import annotations2526import logging27from datetime import date, timedelta2829import httpx30import numpy as np31from sqlalchemy.orm import Session3233from ..config import settings34from .. import models as Mo3536log = logging.getLogger("wiki-attention")3738WM_API = ("https://wikimedia.org/api/rest_v1/metrics/pageviews/per-article/"39          "fr.wikipedia/all-access/user/{article}/daily/{start}/{end}")40UA = {"User-Agent": "qc-election-forecast/2.1 (https://www.qc-election.com; "41                    "contact@spboucher.ai)"}4243# Pages suivies : parti + chef·fe·s (fr.wikipédia)44ARTICLES: dict[str, list[str]] = {45    "CAQ": ["Coalition_avenir_Québec", "Christine_Fréchette"],46    "PLQ": ["Parti_libéral_du_Québec", "Charles_Milliard"],47    "PQ": ["Parti_québécois", "Paul_St-Pierre_Plamondon"],48    "QS": ["Québec_solidaire", "Ruba_Ghazal", "Sol_Zanetti"],49    "PCQ": ["Parti_conservateur_du_Québec", "Éric_Duhaime"],50}5152BASELINE_DAYS = 18053INDICATOR = "wiki_att_{party}"545556def _fetch_article(article: str, start: date, end: date) -> dict[date, int]:57    from urllib.parse import quote58    url = WM_API.format(article=quote(article, safe=""),59                        start=start.strftime("%Y%m%d"), end=end.strftime("%Y%m%d"))60    r = httpx.get(url, headers=UA, timeout=30)61    if r.status_code == 404:      # article sans données sur la période62        return {}63    r.raise_for_status()64    out = {}65    for item in r.json().get("items", []):66        d = date(int(item["timestamp"][:4]), int(item["timestamp"][4:6]),67                 int(item["timestamp"][6:8]))68        out[d] = int(item["views"])69    return out707172def refresh(db: Session, days: int = BASELINE_DAYS + 40) -> dict:73    """Récupère et upsert les pageviews quotidiens (parti + chefs agrégés)."""74    end = date.today() - timedelta(days=1)   # la veille : journée complète75    # reprise incrémentale : depuis le dernier point stocké (moins 3 j de marge)76    last = (db.query(Mo.Indicator)77            .filter(Mo.Indicator.name.like("wiki_att_%"))78            .order_by(Mo.Indicator.as_of.desc()).first())79    start = max(end - timedelta(days=days),80                (last.as_of - timedelta(days=3)) if last else end - timedelta(days=days))81    added = 082    for party, arts in ARTICLES.items():83        daily: dict[date, int] = {}84        for a in arts:85            try:86                for d, v in _fetch_article(a, start, end).items():87                    daily[d] = daily.get(d, 0) + v88            except Exception as e:89                log.warning("pageviews %s: %s", a, e)90        name = INDICATOR.format(party=party)91        for d, v in daily.items():92            row = db.query(Mo.Indicator).filter_by(name=name, as_of=d).first()93            if row is None:94                db.add(Mo.Indicator(name=name, as_of=d, value=float(v),95                                    source="Wikimedia pageviews (fr)",96                                    source_url="https://wikimedia.org/api/rest_v1/",97                                    method="wikimedia-api",98                                    extra={"articles": arts}))99                added += 1100            else:101                row.value = float(v)102    db.commit()103    return {"points_ajoutés": added, "fenêtre": [start.isoformat(), end.isoformat()]}104105106def signal(db: Session) -> dict:107    """Parts d'attention, tendances et détecteur de turbulence.108109    Retourne aussi `drift_multiplier` ∈ [1, 1,45] : gonfle la variance de dérive110    du forecast quand l'attention totale est anormalement élevée (z ≥ 2)."""111    end = date.today() - timedelta(days=1)112    start = end - timedelta(days=BASELINE_DAYS)113    series: dict[str, dict[date, float]] = {}114    for party in ARTICLES:115        rows = (db.query(Mo.Indicator)116                .filter(Mo.Indicator.name == INDICATOR.format(party=party),117                        Mo.Indicator.as_of >= start).all())118        series[party] = {r.as_of: r.value for r in rows}119    all_days = sorted(set().union(*[set(s) for s in series.values()])) if series else []120    if len(all_days) < 30:121        return {"available": False, "drift_multiplier": 1.0,122                "note": "historique d'attention insuffisant (premier cycle)"}123124    def win_sum(party, n):125        cut = end - timedelta(days=n)126        return sum(v for d, v in series[party].items() if d > cut)127128    parties = list(ARTICLES)129    tot7 = {p: win_sum(p, 7) for p in parties}130    tot28 = {p: win_sum(p, 28) for p in parties}131    base = {p: win_sum(p, BASELINE_DAYS) for p in parties}132    s7, s28, sbase = sum(tot7.values()), sum(tot28.values()), sum(base.values())133134    out_p, z_max = {}, 0.0135    for p in parties:136        vals = np.array([series[p].get(d, 0.0) for d in all_days])137        mu, sd = float(vals[:-7].mean()), float(vals[:-7].std() or 1.0)138        z_recent = float((np.mean(vals[-3:]) - mu) / sd)139        z_max = max(z_max, z_recent)140        share7 = tot7[p] / s7 if s7 else 0.0141        share_base = base[p] / sbase if sbase else 0.0142        out_p[p] = {143            "views_7j": int(tot7[p]),144            "share_7j": round(share7 * 100, 1),145            "share_28j": round((tot28[p] / s28 if s28 else 0) * 100, 1),146            "share_base_180j": round(share_base * 100, 1),147            "delta_share_pp": round((share7 - share_base) * 100, 1),148            "z_3j": round(z_recent, 2),149            "sparkline": [int(v) for v in vals[-28:]],150        }151    # turbulence : pic d'attention global → variance de campagne gonflée (borné)152    mult = float(np.clip(1.0 + 0.15 * max(0.0, z_max - 2.0), 1.0, 1.45))153    return {"available": True, "parties": out_p, "z_max": round(z_max, 2),154            "drift_multiplier": round(mult, 3),155            "as_of": end.isoformat(), "baseline_days": BASELINE_DAYS,156            "note": ("Signal d'attention (jamais directionnel) : un pic global "157                     "gonfle l'incertitude de campagne, ne déplace aucune moyenne."),158            "sparkline_dates": [d.isoformat() for d in all_days[-28:]]}159