# 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 """Modèle du SOIR D'ÉLECTION (§89-93) — prêt avant le 5 octobre, dormant d'ici là. Principe : le forecast pré-électoral est le PRIOR. À mesure que le DGEQ publie des résultats partiels : 1. chaque circonscription rapportée fournit un résidu `alr(observé) − alr(attendu)`; 2. le résidu est décomposé hiérarchiquement — erreur NATIONALE partagée (prior N(0, σ_industrie²)), erreur RÉGIONALE (prior N(0, σ_rég²)), reste local — par pondération de précision, chaque observation pesée par sa fraction dépouillée; 3. les circonscriptions NON rapportées sont mises à jour par les erreurs nationale et régionale inférées (transfert hiérarchique, jamais d'extrapolation naïve des premiers bureaux); 4. la simulation est relancée avec les variances RÉDUITES par l'information accumulée → P(victoire) en direct; 5. **calling conservateur** : un siège n'est « déclaré » que si P ≥ 99,5 % ET ≥ 40 % dépouillé (ou P ≥ 99,99 %). Garde anti-biais de dépouillement (§90) : les premiers bureaux ne sont pas aléatoires (vote par anticipation, urbain/rural). Chaque circonscription n'est comparée qu'à SA propre attente et pèse au prorata de son propre dépouillement; un plancher de variance de comptage 1/dépouillé protège les débuts de soirée. """ from __future__ import annotations import json from dataclasses import dataclass from datetime import datetime, timezone import numpy as np from sqlalchemy.orm import Session from ..config import DATA_DIR, settings from .compositions import alr, close, inv_alr from .simulate import SimulationInput, run_simulation STATE_FILE = DATA_DIR / "election_night_state.json" MIN_REPORT = 0.05 # < 5 % dépouillé : la circonscription n'informe pas CALL_PROB = 0.995 CALL_MIN_REPORTED = 0.40 COUNT_NOISE_BASE = 0.06 # sd alr du bruit de comptage à 100 % dépouillé @dataclass class NightUpdate: national_shift: np.ndarray # alr (M,) national_var: float regional_shift: dict[str, np.ndarray] n_informative: int def _expected_district_alr(inp: SimulationInput) -> np.ndarray: """Attente pré-électorale par circonscription (base 2022 + swing national).""" delta_nat = inp.x_mean - alr(inp.baseline_national) return alr(inp.district_baselines) + delta_nat[None, :] def infer_shifts(inp: SimulationInput, reported: list[dict], industry_sd: float) -> NightUpdate: """Décomposition hiérarchique des résidus des circonscriptions rapportées.""" M = len(inp.x_mean) exp_alr = _expected_district_alr(inp) name_idx = {n: i for i, n in enumerate(inp.district_names)} resids, weights, regions = [], [], [] for r in reported: i = name_idx.get(r["district"]) frac = float(r.get("pct_reported", 0)) / 100.0 if i is None or frac < MIN_REPORT: continue obs = close(np.array([r["results"].get(p, 0.3) for p in settings.parties]) / 100.0) resid = alr(obs) - exp_alr[i] # variance de l'observation : local + comptage partiel var = (settings.riding_error_sd ** 2 + settings.regional_error_sd ** 2 + (COUNT_NOISE_BASE ** 2) / max(frac, MIN_REPORT)) resids.append(resid) weights.append(1.0 / var) regions.append(inp.district_regions[i]) if not resids: return NightUpdate(np.zeros(M), industry_sd ** 2, {}, 0) R = np.array(resids); W = np.array(weights) # national : posterior N — prior N(0, σ_ind²), vraisemblance moyenne pondérée prior_prec = 1.0 / industry_sd ** 2 like_prec = W.sum() nat_var = 1.0 / (prior_prec + like_prec) nat_shift = nat_var * (R * W[:, None]).sum(axis=0) # régional : sur les résidus nets, par région, prior N(0, σ_rég²) reg_shift: dict[str, np.ndarray] = {} for reg in set(regions): mask = np.array([g == reg for g in regions]) Rr, Wr = R[mask] - nat_shift[None, :], W[mask] rp = 1.0 / settings.regional_error_sd ** 2 rv = 1.0 / (rp + Wr.sum()) reg_shift[reg] = rv * (Rr * Wr[:, None]).sum(axis=0) return NightUpdate(nat_shift, nat_var, reg_shift, len(resids)) def update(db: Session, reported: list[dict], n_sims: int = 15000) -> dict: """Résultats partiels → forecast en direct + déclarations conservatrices.""" from ..pipeline import _district_inputs from .. import models as Mo run = (db.query(Mo.ForecastRun).filter(Mo.ForecastRun.is_backtest.is_(False)) .order_by(Mo.ForecastRun.as_of.desc(), Mo.ForecastRun.id.desc()).first()) if run is None: raise RuntimeError("Aucun forecast pré-électoral (prior manquant)") result_ref = run.national.get("baseline_national") inp = _district_inputs(db, result_ref) inp.x_mean = np.array(run.national["x_forecast"]) inp.P = np.array(run.national["P_forecast"]) industry_sd = float(run.diagnostics.get("industry_sd", settings.industry_error_sd)) upd = infer_shifts(inp, reported, industry_sd) # état national mis à jour : moyenne décalée, incertitude nationale réduite inp.x_mean = inp.x_mean + upd.national_shift M = len(inp.x_mean) # remplace la part « erreur systémique » du P par la variance posterior inp.P = inp.P + (upd.national_var - industry_sd ** 2) * np.eye(M) inp.P = 0.5 * (inp.P + inp.P.T) ev = np.linalg.eigvalsh(inp.P) if ev.min() < 1e-8: # PSD après réduction inp.P += (1e-8 - ev.min()) * np.eye(M) # chocs régionaux inférés injectés comme scénario régional (moyennes) reg_pp: dict[str, dict[str, float]] = {} for reg, shift in upd.regional_shift.items(): base = close(np.array( [result_ref[p] for p in settings.parties]) / 100.0) delta = (inv_alr(alr(base) + shift) - base) * 100.0 reg_pp[reg] = {p: round(float(delta[i]), 2) for i, p in enumerate(settings.parties) if abs(delta[i]) >= 0.1} sim = run_simulation(inp, n_sims=n_sims, scenario={"regional_pp": reg_pp} if reg_pp else None, rng=np.random.default_rng(20261005)) by_name = {r["district"]: r for r in reported} districts, called = [], 0 for d in sim.districts: rep = by_name.get(d["district"]) frac = float((rep or {}).get("pct_reported", 0)) / 100.0 wp = dict(d["win_probs"]) leader = max(wp, key=wp.get) if rep and frac >= 0.99: # dépouillement complet = décidé obs = rep["results"] leader = max(obs, key=obs.get) wp = {p: (1.0 if p == leader else 0.0) for p in settings.parties} is_called = (wp[leader] >= CALL_PROB and (frac >= CALL_MIN_REPORTED or wp[leader] >= 0.9999)) called += is_called districts.append({"district": d["district"], "region": d["region"], "leader": leader, "win_probs": wp, "pct_reported": round(frac * 100, 1), "called": is_called, "reported_results": (rep or {}).get("results")}) state = { "at": datetime.now(timezone.utc).isoformat(), "prior_run_id": run.id, "n_reported": len(reported), "n_informative": upd.n_informative, "national_shift_pp": {p: round(float( (inv_alr(inp.x_mean)[i] - inv_alr(np.array( run.national["x_forecast"]))[i]) * 100), 2) for i, p in enumerate(settings.parties)}, "regional_shift_pp": reg_pp, "seats": sim.seats, "summary": sim.seat_matrix_summary, "national_vote": sim.national_vote, "districts": districts, "seats_called": called, "method": ("Prior = forecast pré-électoral; erreurs nationale/régionales " "inférées hiérarchiquement des circonscriptions rapportées " "(pondérées par leur dépouillement); déclarations à " f"P ≥ {CALL_PROB:.1%} et ≥ {CALL_MIN_REPORTED:.0%} dépouillé."), } STATE_FILE.write_text(json.dumps(state, ensure_ascii=False)) return state def latest_state() -> dict | None: if STATE_FILE.exists(): return json.loads(STATE_FILE.read_text()) return None