# QC Élection Forecast — Plateforme de prévision électorale du Québec 2026 # Auteur : Simon-Pierre Boucher # Contact : contact@spboucher.ai # https://www.qc-election.com """Moteur de sentiment politique — signal AUXILIAIRE, poids nul dans le forecast. Pipeline : document → extraction d'entités (partis, chefs) → stance/sentiment → clustering (déduplication de la même nouvelle) → agrégats + anomalies de volume. Deux scoreurs : * « lexicon » : lexique français + négation + heuristique de cible (une attaque de X contre Y est négative envers Y, pas envers X); * « llm » (optionnel, si MACLUSTR_API_KEY) : classification de stance par LLM. Le sentiment Web n'est PAS un sondage : il est affiché comme signal bruité et n'entre jamais dans le calcul des probabilités électorales. """ from __future__ import annotations import hashlib import re from datetime import date, datetime, timedelta, timezone import numpy as np from sqlalchemy.orm import Session from ..config import settings from .. import models as Mo ENTITY_ALIASES: dict[str, list[str]] = { "CAQ": ["caq", "coalition avenir québec", "christine fréchette", "fréchette"], "PLQ": ["plq", "parti libéral du québec", "libéral", "libéraux", "charles milliard", "milliard"], "PQ": ["parti québécois", " pq ", "st-pierre plamondon", "plamondon", "pspp", "péquiste"], "QS": ["québec solidaire", " qs ", "ruba ghazal", "sol zanetti", "solidaire"], "PCQ": ["parti conservateur du québec", " pcq ", "éric duhaime", "duhaime", "conservateur du québec"], } POS_WORDS = {"gagne", "succès", "hausse", "progresse", "populaire", "appui", "appuis", "victoire", "monte", "remonte", "convainc", "salue", "félicite", "record", "confiance", "promet", "espoir", "solide", "fort", "avance", "favori", "endosse", "rallie", "applaudit", "réussit", "améliore"} NEG_WORDS = {"scandale", "chute", "recul", "critique", "critiqué", "controverse", "démission", "échec", "perd", "baisse", "colère", "crise", "attaque", "accuse", "accusé", "dérape", "conflit", "corruption", "enquête", "polémique", "déçoit", "impopulaire", "fragile", "faible", "menace", "dénonce", "erreur", "gaffe", "tollé", "démissionne", "poursuite"} NEGATIONS = {"ne", "pas", "jamais", "sans", "aucun", "aucune", "ni", "non"} ATTACK_VERBS = {"accuse", "attaque", "dénonce", "critique", "blâme", "reproche", "fustige", "tacle", "éreinte", "qualifie"} STOP = {"le", "la", "les", "de", "des", "du", "un", "une", "et", "en", "au", "aux", "pour", "sur", "dans", "avec", "que", "qui", "est", "sont", "a", "à", "d", "l", "se", "son", "sa", "ses", "plus", "pas", "ne", "il", "elle", "on"} def _tokens(text: str) -> list[str]: return re.findall(r"[a-zàâäéèêëîïôöùûüç'-]+", text.lower()) def cluster_key(title: str) -> str: words = sorted(w for w in _tokens(title) if w not in STOP and len(w) > 3)[:6] return hashlib.sha1(" ".join(words).encode()).hexdigest()[:16] def detect_entities(text: str) -> list[str]: t = " " + text.lower() + " " return [code for code, aliases in ENTITY_ALIASES.items() if any(a in t for a in aliases)] def lexicon_score(text: str, entity: str) -> tuple[float, str]: """Sentiment [-1,1] + stance dirigés vers l'entité (heuristique de cible).""" toks = _tokens(text) aliases = ENTITY_ALIASES[entity] positions = [i for i, tok in enumerate(toks) if any(a.strip().split()[0] in tok for a in aliases if a.strip())] score, n = 0.0, 0 for i, tok in enumerate(toks): val = 1.0 if tok in POS_WORDS else (-1.0 if tok in NEG_WORDS else 0.0) if val == 0.0: continue if any(toks[j] in NEGATIONS for j in range(max(0, i - 3), i)): val = -val # attaque politique : le sentiment négatif vise l'entité APRÈS le verbe if tok in ATTACK_VERBS: after = detect_entities(" ".join(toks[i:i + 12])) if entity not in after: continue # l'attaque vise quelqu'un d'autre dist = min((abs(i - p) for p in positions), default=99) if dist > 25: continue # trop loin de l'entité — probablement une autre cible score += val n += 1 if n == 0: return 0.0, "neutre" s = float(np.clip(score / max(n, 1), -1, 1)) stance = "pro" if s > 0.25 else ("anti" if s < -0.25 else ("neutre" if abs(s) < 0.1 else "ambigu")) return s, stance def llm_score(text: str, entity: str) -> tuple[float, str] | None: """Stance par LLM (MacLustr) — optionnel; retourne None en cas d'échec.""" if not settings.maclustr_api_key: return None import httpx, json prompt = ( "Tu analyses la couverture politique québécoise. Pour le texte suivant, " f"détermine la position (stance) envers {settings.party_names.get(entity, entity)}. " "Attention : sarcasme, négation, citations et attaques dirigées vers un adversaire. " "Réponds UNIQUEMENT en JSON: {\"stance\": \"pro|anti|neutre|ambigu\", " "\"sentiment\": nombre entre -1 et 1}.\n\nTexte: " + text[:1500]) try: r = httpx.post(settings.maclustr_llm_url, headers={"Authorization": f"Bearer {settings.maclustr_api_key}"}, json={"model": settings.maclustr_llm_model, "temperature": 0, "messages": [{"role": "user", "content": prompt}], "max_tokens": 100}, timeout=25) r.raise_for_status() content = r.json()["choices"][0]["message"]["content"] m = re.search(r"\{.*\}", content, re.S) data = json.loads(m.group()) return float(np.clip(data["sentiment"], -1, 1)), str(data["stance"]) except Exception: return None def ingest_feeds(db: Session, feeds: list[str] | None = None) -> dict: import feedparser feeds = feeds or settings.sentiment_feeds added, scored = 0, 0 for url in feeds: src = db.query(Mo.DataSource).filter_by(url=url).first() if src is None: src = Mo.DataSource(name=f"RSS {url.split('/')[2]}", url=url, kind="sentiment") db.add(src); db.flush() try: parsed = feedparser.parse(url) for e in parsed.entries[:60]: link = getattr(e, "link", None) title = getattr(e, "title", "") or "" if not link or not title: continue if db.query(Mo.SentimentDocument).filter_by(url=link).first(): continue summary = re.sub(r"<[^>]+>", " ", getattr(e, "summary", "") or "")[:2000] pub = None if getattr(e, "published_parsed", None): pub = datetime(*e.published_parsed[:6], tzinfo=timezone.utc) doc = Mo.SentimentDocument( source=parsed.feed.get("title", url.split("/")[2]), url=link, title=title, summary=summary, published=pub, cluster_key=cluster_key(title)) text = f"{title}. {summary}" entities = detect_entities(text) for ent in entities: res = llm_score(text, ent) method = "llm" if res is None: res, method = lexicon_score(text, ent), "lexicon" s, stance = res doc.scores.append(Mo.SentimentScore( entity=ent, sentiment=s, stance=stance, method=method)) scored += 1 db.add(doc) added += 1 src.last_fetch = datetime.now(timezone.utc) src.last_status = "ok" except Exception as e: src.last_fetch = datetime.now(timezone.utc) src.last_status = f"error: {e}"[:200] db.commit() detect_volume_anomalies(db) return {"documents_ajoutés": added, "scores": scored} def detect_volume_anomalies(db: Session, z_threshold: float = 2.5) -> int: """Anomalies de volume médiatique par parti (z-score vs 14 jours) → événements.""" today = date.today() created = 0 for party in ENTITY_ALIASES: counts = [] for back in range(15): d = today - timedelta(days=back) n = (db.query(Mo.SentimentScore).join(Mo.SentimentDocument) .filter(Mo.SentimentScore.entity == party, Mo.SentimentDocument.published >= datetime(d.year, d.month, d.day, tzinfo=timezone.utc), Mo.SentimentDocument.published < datetime(d.year, d.month, d.day, tzinfo=timezone.utc) + timedelta(days=1)) .count()) counts.append(n) base = counts[1:] if len(base) < 7 or np.std(base) == 0: continue z = (counts[0] - np.mean(base)) / np.std(base) if z >= z_threshold and counts[0] >= 5: exists = (db.query(Mo.NewsEvent) .filter_by(event_date=today, kind="anomalie-volume").count()) if not exists: db.add(Mo.NewsEvent( event_date=today, kind="anomalie-volume", title=f"Pic de couverture médiatique — {settings.party_names[party]}", description=f"Volume {counts[0]} vs moyenne {np.mean(base):.1f} (z={z:.1f})", parties=[party], importance=min(0.9, 0.4 + z / 10), detected_by="anomaly-detector")) created += 1 db.commit() return created