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"""Moteur de sentiment politique — signal AUXILIAIRE, poids nul dans le forecast.67Pipeline : document → extraction d'entités (partis, chefs) → stance/sentiment8→ clustering (déduplication de la même nouvelle) → agrégats + anomalies de volume.910Deux scoreurs :11  * « lexicon » : lexique français + négation + heuristique de cible12    (une attaque de X contre Y est négative envers Y, pas envers X);13  * « llm » (optionnel, si MACLUSTR_API_KEY) : classification de stance par LLM.1415Le sentiment Web n'est PAS un sondage : il est affiché comme signal bruité et16n'entre jamais dans le calcul des probabilités électorales.17"""18from __future__ import annotations1920import hashlib21import re22from datetime import date, datetime, timedelta, timezone2324import numpy as np25from sqlalchemy.orm import Session2627from ..config import settings28from .. import models as Mo2930ENTITY_ALIASES: dict[str, list[str]] = {31    "CAQ": ["caq", "coalition avenir québec", "christine fréchette", "fréchette"],32    "PLQ": ["plq", "parti libéral du québec", "libéral", "libéraux", "charles milliard",33            "milliard"],34    "PQ": ["parti québécois", " pq ", "st-pierre plamondon", "plamondon", "pspp",35           "péquiste"],36    "QS": ["québec solidaire", " qs ", "ruba ghazal", "sol zanetti", "solidaire"],37    "PCQ": ["parti conservateur du québec", " pcq ", "éric duhaime", "duhaime",38            "conservateur du québec"],39}4041POS_WORDS = {"gagne", "succès", "hausse", "progresse", "populaire", "appui", "appuis",42             "victoire", "monte", "remonte", "convainc", "salue", "félicite", "record",43             "confiance", "promet", "espoir", "solide", "fort", "avance", "favori",44             "endosse", "rallie", "applaudit", "réussit", "améliore"}45NEG_WORDS = {"scandale", "chute", "recul", "critique", "critiqué", "controverse",46             "démission", "échec", "perd", "baisse", "colère", "crise", "attaque",47             "accuse", "accusé", "dérape", "conflit", "corruption", "enquête",48             "polémique", "déçoit", "impopulaire", "fragile", "faible", "menace",49             "dénonce", "erreur", "gaffe", "tollé", "démissionne", "poursuite"}50NEGATIONS = {"ne", "pas", "jamais", "sans", "aucun", "aucune", "ni", "non"}51ATTACK_VERBS = {"accuse", "attaque", "dénonce", "critique", "blâme", "reproche",52                "fustige", "tacle", "éreinte", "qualifie"}5354STOP = {"le", "la", "les", "de", "des", "du", "un", "une", "et", "en", "au", "aux",55        "pour", "sur", "dans", "avec", "que", "qui", "est", "sont", "a", "à", "d",56        "l", "se", "son", "sa", "ses", "plus", "pas", "ne", "il", "elle", "on"}575859def _tokens(text: str) -> list[str]:60    return re.findall(r"[a-zàâäéèêëîïôöùûüç'-]+", text.lower())616263def cluster_key(title: str) -> str:64    words = sorted(w for w in _tokens(title) if w not in STOP and len(w) > 3)[:6]65    return hashlib.sha1(" ".join(words).encode()).hexdigest()[:16]666768def detect_entities(text: str) -> list[str]:69    t = " " + text.lower() + " "70    return [code for code, aliases in ENTITY_ALIASES.items()71            if any(a in t for a in aliases)]727374def lexicon_score(text: str, entity: str) -> tuple[float, str]:75    """Sentiment [-1,1] + stance dirigés vers l'entité (heuristique de cible)."""76    toks = _tokens(text)77    aliases = ENTITY_ALIASES[entity]78    positions = [i for i, tok in enumerate(toks)79                 if any(a.strip().split()[0] in tok for a in aliases if a.strip())]80    score, n = 0.0, 081    for i, tok in enumerate(toks):82        val = 1.0 if tok in POS_WORDS else (-1.0 if tok in NEG_WORDS else 0.0)83        if val == 0.0:84            continue85        if any(toks[j] in NEGATIONS for j in range(max(0, i - 3), i)):86            val = -val87        # attaque politique : le sentiment négatif vise l'entité APRÈS le verbe88        if tok in ATTACK_VERBS:89            after = detect_entities(" ".join(toks[i:i + 12]))90            if entity not in after:91                continue  # l'attaque vise quelqu'un d'autre92        dist = min((abs(i - p) for p in positions), default=99)93        if dist > 25:94            continue  # trop loin de l'entité — probablement une autre cible95        score += val96        n += 197    if n == 0:98        return 0.0, "neutre"99    s = float(np.clip(score / max(n, 1), -1, 1))100    stance = "pro" if s > 0.25 else ("anti" if s < -0.25 else ("neutre" if abs(s) < 0.1 else "ambigu"))101    return s, stance102103104def llm_score(text: str, entity: str) -> tuple[float, str] | None:105    """Stance par LLM (MacLustr) — optionnel; retourne None en cas d'échec."""106    if not settings.maclustr_api_key:107        return None108    import httpx, json109    prompt = (110        "Tu analyses la couverture politique québécoise. Pour le texte suivant, "111        f"détermine la position (stance) envers {settings.party_names.get(entity, entity)}. "112        "Attention : sarcasme, négation, citations et attaques dirigées vers un adversaire. "113        "Réponds UNIQUEMENT en JSON: {\"stance\": \"pro|anti|neutre|ambigu\", "114        "\"sentiment\": nombre entre -1 et 1}.\n\nTexte: " + text[:1500])115    try:116        r = httpx.post(settings.maclustr_llm_url,117                       headers={"Authorization": f"Bearer {settings.maclustr_api_key}"},118                       json={"model": settings.maclustr_llm_model, "temperature": 0,119                             "messages": [{"role": "user", "content": prompt}],120                             "max_tokens": 100},121                       timeout=25)122        r.raise_for_status()123        content = r.json()["choices"][0]["message"]["content"]124        m = re.search(r"\{.*\}", content, re.S)125        data = json.loads(m.group())126        return float(np.clip(data["sentiment"], -1, 1)), str(data["stance"])127    except Exception:128        return None129130131def ingest_feeds(db: Session, feeds: list[str] | None = None) -> dict:132    import feedparser133    feeds = feeds or settings.sentiment_feeds134    added, scored = 0, 0135    for url in feeds:136        src = db.query(Mo.DataSource).filter_by(url=url).first()137        if src is None:138            src = Mo.DataSource(name=f"RSS {url.split('/')[2]}", url=url, kind="sentiment")139            db.add(src); db.flush()140        try:141            parsed = feedparser.parse(url)142            for e in parsed.entries[:60]:143                link = getattr(e, "link", None)144                title = getattr(e, "title", "") or ""145                if not link or not title:146                    continue147                if db.query(Mo.SentimentDocument).filter_by(url=link).first():148                    continue149                summary = re.sub(r"<[^>]+>", " ", getattr(e, "summary", "") or "")[:2000]150                pub = None151                if getattr(e, "published_parsed", None):152                    pub = datetime(*e.published_parsed[:6], tzinfo=timezone.utc)153                doc = Mo.SentimentDocument(154                    source=parsed.feed.get("title", url.split("/")[2]),155                    url=link, title=title, summary=summary,156                    published=pub, cluster_key=cluster_key(title))157                text = f"{title}. {summary}"158                entities = detect_entities(text)159                for ent in entities:160                    res = llm_score(text, ent)161                    method = "llm"162                    if res is None:163                        res, method = lexicon_score(text, ent), "lexicon"164                    s, stance = res165                    doc.scores.append(Mo.SentimentScore(166                        entity=ent, sentiment=s, stance=stance, method=method))167                    scored += 1168                db.add(doc)169                added += 1170            src.last_fetch = datetime.now(timezone.utc)171            src.last_status = "ok"172        except Exception as e:173            src.last_fetch = datetime.now(timezone.utc)174            src.last_status = f"error: {e}"[:200]175        db.commit()176    detect_volume_anomalies(db)177    return {"documents_ajoutés": added, "scores": scored}178179180def detect_volume_anomalies(db: Session, z_threshold: float = 2.5) -> int:181    """Anomalies de volume médiatique par parti (z-score vs 14 jours) → événements."""182    today = date.today()183    created = 0184    for party in ENTITY_ALIASES:185        counts = []186        for back in range(15):187            d = today - timedelta(days=back)188            n = (db.query(Mo.SentimentScore).join(Mo.SentimentDocument)189                 .filter(Mo.SentimentScore.entity == party,190                         Mo.SentimentDocument.published >= datetime(d.year, d.month, d.day, tzinfo=timezone.utc),191                         Mo.SentimentDocument.published < datetime(d.year, d.month, d.day, tzinfo=timezone.utc) + timedelta(days=1))192                 .count())193            counts.append(n)194        base = counts[1:]195        if len(base) < 7 or np.std(base) == 0:196            continue197        z = (counts[0] - np.mean(base)) / np.std(base)198        if z >= z_threshold and counts[0] >= 5:199            exists = (db.query(Mo.NewsEvent)200                      .filter_by(event_date=today, kind="anomalie-volume").count())201            if not exists:202                db.add(Mo.NewsEvent(203                    event_date=today, kind="anomalie-volume",204                    title=f"Pic de couverture médiatique — {settings.party_names[party]}",205                    description=f"Volume {counts[0]} vs moyenne {np.mean(base):.1f} (z={z:.1f})",206                    parties=[party], importance=min(0.9, 0.4 + z / 10),207                    detected_by="anomaly-detector"))208                created += 1209    db.commit()210    return created211