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"""Couches d'ensemble v2 — signaux non-sondages à poids explicites et bornés.671. **Ajustement médias** : le sentiment de presse des 14 derniers jours8   (RSS + veille Firecrawl) déplace le forecast d'au plus ±`media_nudge_pp_max`9   points par parti (défaut 0,35 pp). Un signal bruité obtient un poids10   minuscule mais non nul — c'est un nudge, jamais un moteur.11122. **Ensemble marchés prédictifs** : la probabilité « plus de sièges » publiée13   est une moyenne pondérée modèle/marché (défaut 88 % / 12 %). Le marché14   agrège de l'information hors-sondages (initiés, campagnes locales,15   anticipations) et sa combinaison avec un modèle statistique améliore16   presque toujours le Brier (littérature sur l'agrégation de prévisions).1718Chaque couche est débrayable par configuration et sa contribution exacte est19publiée dans la décomposition du forecast.20"""21from __future__ import annotations2223from datetime import datetime, timedelta, timezone2425import numpy as np26from sqlalchemy.orm import Session2728from ..config import settings29from .. import models as Mo30from .compositions import alr, close, inv_alr313233def media_nudge(db: Session) -> dict:34    """Sentiment 14 j (pondéré par volume, saturé) → délta borné par parti (pp).3536    delta_pp = cap × tanh(sentiment_moyen × 2) × min(1, volume / 60)37    Volume < `media_nudge_min_volume` → aucun effet (signal trop mince)."""38    out = {"enabled": settings.media_nudge_enabled, "delta_pp": {},39           "volume_14j": {}, "sentiment_14j": {}}40    if not settings.media_nudge_enabled:41        return out42    cutoff = datetime.now(timezone.utc) - timedelta(days=14)43    cap = settings.media_nudge_pp_max44    for party in [p for p in settings.parties if p != "AUT"]:45        rows = (db.query(Mo.SentimentScore).join(Mo.SentimentDocument)46                .filter(Mo.SentimentScore.entity == party,47                        Mo.SentimentDocument.fetched_at >= cutoff).all())48        vol = len(rows)49        out["volume_14j"][party] = vol50        if vol < settings.media_nudge_min_volume:51            out["delta_pp"][party] = 0.052            out["sentiment_14j"][party] = None53            continue54        mean_s = float(np.mean([r.sentiment for r in rows]))55        out["sentiment_14j"][party] = round(mean_s, 3)56        out["delta_pp"][party] = round(57            cap * float(np.tanh(mean_s * 2.0)) * min(1.0, vol / 60.0), 3)58    return out596061def apply_nudge(x: np.ndarray, delta_pp: dict[str, float]) -> np.ndarray:62    """Applique le délta (pp de parts) au vecteur d'état alr, à somme conservée."""63    if not any(delta_pp.values()):64        return x65    p0 = inv_alr(x)66    p1 = p0.copy()67    for i, party in enumerate(settings.parties):68        p1[i] += delta_pp.get(party, 0.0) / 100.069    return alr(close(np.clip(p1, 0.001, None)))707172def market_blend(model_probs: dict[str, float],73                 market_probs: dict[str, float] | None) -> dict:74    """P(plus de sièges) finale = (1−w)·modèle + w·marché, renormalisée.75    Sans marché disponible : probabilités du modèle telles quelles."""76    w = settings.market_blend_weight77    if not market_probs or w <= 0:78        return {"weight_market": 0.0, "blended": dict(model_probs),79                "market": market_probs or {}, "model": dict(model_probs)}80    blended = {}81    for p, mv in model_probs.items():82        blended[p] = (1 - w) * mv + w * market_probs.get(p, mv)83    tot = sum(blended.values())84    if tot > 0:85        blended = {p: round(v / tot, 4) for p, v in blended.items()}86    return {"weight_market": w, "blended": blended,87            "market": {p: round(v, 4) for p, v in market_probs.items()},88            "model": {p: round(v, 4) for p, v in model_probs.items()}}89