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"""Social Pulse inter-plateformes (v3 « 127 », §13-§15).67Transforme les publications brutes (`social_posts`) en métriques STANDARDISÉES8et comparables entre plateformes — jamais en intentions de vote :910  * attention_z   : volume 24 h vs base 28 j, par parti (z-score);11  * velocity_z    : accélération (24 h vs moyenne 3 j);12  * breadth       : diversité des contenus (1 − part du texte dominant);13  * cross_platform: nombre de plateformes qui confirment le signal;14  * stance_balance: (pro − anti) / total, pondéré engagement;15  * quality       : part d'attention ORGANIQUE après filtrage de la16                    coordination (§14 : near-duplicates, rafales synchronisées,17                    concentration d'engagement).1819Avertissement affiché partout : les signaux sociaux mesurent l'attention et le20discours en ligne — ils ne constituent PAS un sondage représentatif.21"""22from __future__ import annotations2324import math25from collections import Counter26from datetime import datetime, timedelta, timezone2728import numpy as np29from sqlalchemy.orm import Session3031from ...config import settings32from ... import models as Mo3334BASELINE_DAYS = 28353637def _posts(db: Session, days: int) -> list:38    cutoff = datetime.now(timezone.utc) - timedelta(days=days)39    return (db.query(Mo.SocialPost)40            .filter(Mo.SocialPost.fetched_at >= cutoff).all())414243def coordination_filter(posts: list) -> tuple[list, dict]:44    """Sépare attention organique et coordonnée (heuristiques §14) :45    - grappes de quasi-doublons (même text_hash déjà dédupliqué en amont;46      ici : mêmes 4 premiers mots × ≥3 occurrences);47    - rafales synchronisées (≥5 posts d'une même grappe dans la même heure);48    - concentration d'engagement (>60 % de l'engagement d'un parti porté par49      <5 % des posts → part marquée incertaine, pas rejetée)."""50    by_prefix: dict[str, list] = {}51    for p in posts:52        prefix = " ".join((p.text or "").lower().split()[:4])53        by_prefix.setdefault(prefix, []).append(p)54    organic, coordinated = [], []55    for prefix, group in by_prefix.items():56        if len(group) >= 3:57            hours = Counter((p.fetched_at or datetime.now(timezone.utc))58                            .strftime("%Y%m%d%H") for p in group)59            if max(hours.values()) >= 5 or len(group) >= 6:60                coordinated.extend(group)61                continue62        organic.extend(group)63    diag = {"n_organic": len(organic), "n_coordinated": len(coordinated),64            "organic_share": round(len(organic) / max(1, len(posts)), 3)}65    return organic, diag666768def _party_daily_volumes(posts: list, party: str) -> dict[str, float]:69    out: dict[str, float] = {}70    for p in posts:71        if party in (p.scores or {}):72            d = (p.created_at or p.fetched_at).strftime("%Y-%m-%d")73            out[d] = out.get(d, 0.0) + 1.074    return out757677def pulse_index(db: Session) -> dict:78    """Indice Social Pulse par parti (attention, vélocité, stance, largeur,79    confirmation inter-plateformes, qualité)."""80    posts = _posts(db, BASELINE_DAYS)81    if len(posts) < 30:82        return {"available": False,83                "note": "corpus social insuffisant (premiers cycles)"}84    organic, coord_diag = coordination_filter(posts)85    now = datetime.now(timezone.utc)86    parties = [p for p in settings.parties if p != "AUT"]87    out = {}88    for party in parties:89        mine = [p for p in organic if party in (p.scores or {})]90        vols = _party_daily_volumes(organic, party)91        days = [(now - timedelta(days=i)).strftime("%Y-%m-%d")92                for i in range(BASELINE_DAYS)]93        series = np.array([vols.get(d, 0.0) for d in days])   # [0]=aujourd'hui94        base = series[3:]95        mu, sd = float(base.mean()), float(base.std() or 1.0)96        attention_z = (float(series[0]) - mu) / sd97        recent3 = float(series[:3].mean())98        velocity_z = (float(series[0]) - recent3) / (sd or 1.0)99        # stance pondérée engagement100        wsum = wtot = 0.0101        pro = anti = 0102        platforms = set()103        for p in mine:104            sc = p.scores[party]105            w = 1.0 + math.log1p(max(0.0, p.engagement))106            wsum += (1 if sc["stance"] == "pro" else107                     -1 if sc["stance"] == "anti" else 0) * w108            wtot += w109            pro += sc["stance"] == "pro"110            anti += sc["stance"] == "anti"111            platforms.add(p.platform)112        texts = Counter(" ".join((p.text or "").lower().split()[:4]) for p in mine)113        breadth = round(1.0 - (max(texts.values()) / len(mine) if mine else 0), 3)114        out[party] = {115            "attention_z": round(attention_z, 2),116            "velocity_z": round(velocity_z, 2),117            "volume_24h": int(series[0]), "volume_28j": int(series.sum()),118            "stance_balance": round(wsum / wtot, 3) if wtot else None,119            "pro": pro, "anti": anti,120            "breadth": breadth,121            "cross_platform": len(platforms),122            "platforms": sorted(platforms),123        }124    return {"available": True, "parties": out, "quality": coord_diag,125            "as_of": now.isoformat(),126            "note": ("Les signaux sociaux mesurent l'attention et le discours "127                     "public en ligne — ils ne constituent PAS un sondage "128                     "représentatif. Poids directionnel : nul.")}129