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"""Modèle du SOIR D'ÉLECTION (§89-93) — prêt avant le 5 octobre, dormant d'ici là.67Principe : le forecast pré-électoral est le PRIOR. À mesure que le DGEQ publie8des résultats partiels :910  1. chaque circonscription rapportée fournit un résidu11     `alr(observé) − alr(attendu)`;12  2. le résidu est décomposé hiérarchiquement — erreur NATIONALE partagée13     (prior N(0, σ_industrie²)), erreur RÉGIONALE (prior N(0, σ_rég²)),14     reste local — par pondération de précision, chaque observation pesée par15     sa fraction dépouillée;16  3. les circonscriptions NON rapportées sont mises à jour par les erreurs17     nationale et régionale inférées (transfert hiérarchique, jamais18     d'extrapolation naïve des premiers bureaux);19  4. la simulation est relancée avec les variances RÉDUITES par l'information20     accumulée → P(victoire) en direct;21  5. **calling conservateur** : un siège n'est « déclaré » que si22     P ≥ 99,5 % ET ≥ 40 % dépouillé (ou P ≥ 99,99 %).2324Garde anti-biais de dépouillement (§90) : les premiers bureaux ne sont pas25aléatoires (vote par anticipation, urbain/rural). Chaque circonscription n'est26comparée qu'à SA propre attente et pèse au prorata de son propre dépouillement;27un plancher de variance de comptage 1/dépouillé protège les débuts de soirée.28"""29from __future__ import annotations3031import json32from dataclasses import dataclass33from datetime import datetime, timezone3435import numpy as np36from sqlalchemy.orm import Session3738from ..config import DATA_DIR, settings39from .compositions import alr, close, inv_alr40from .simulate import SimulationInput, run_simulation4142STATE_FILE = DATA_DIR / "election_night_state.json"43MIN_REPORT = 0.05          # < 5 % dépouillé : la circonscription n'informe pas44CALL_PROB = 0.99545CALL_MIN_REPORTED = 0.4046COUNT_NOISE_BASE = 0.06    # sd alr du bruit de comptage à 100 % dépouillé474849@dataclass50class NightUpdate:51    national_shift: np.ndarray          # alr (M,)52    national_var: float53    regional_shift: dict[str, np.ndarray]54    n_informative: int555657def _expected_district_alr(inp: SimulationInput) -> np.ndarray:58    """Attente pré-électorale par circonscription (base 2022 + swing national)."""59    delta_nat = inp.x_mean - alr(inp.baseline_national)60    return alr(inp.district_baselines) + delta_nat[None, :]616263def infer_shifts(inp: SimulationInput, reported: list[dict],64                 industry_sd: float) -> NightUpdate:65    """Décomposition hiérarchique des résidus des circonscriptions rapportées."""66    M = len(inp.x_mean)67    exp_alr = _expected_district_alr(inp)68    name_idx = {n: i for i, n in enumerate(inp.district_names)}69    resids, weights, regions = [], [], []70    for r in reported:71        i = name_idx.get(r["district"])72        frac = float(r.get("pct_reported", 0)) / 100.073        if i is None or frac < MIN_REPORT:74            continue75        obs = close(np.array([r["results"].get(p, 0.3)76                              for p in settings.parties]) / 100.0)77        resid = alr(obs) - exp_alr[i]78        # variance de l'observation : local + comptage partiel79        var = (settings.riding_error_sd ** 280               + settings.regional_error_sd ** 281               + (COUNT_NOISE_BASE ** 2) / max(frac, MIN_REPORT))82        resids.append(resid)83        weights.append(1.0 / var)84        regions.append(inp.district_regions[i])85    if not resids:86        return NightUpdate(np.zeros(M), industry_sd ** 2, {}, 0)87    R = np.array(resids); W = np.array(weights)88    # national : posterior N — prior N(0, σ_ind²), vraisemblance moyenne pondérée89    prior_prec = 1.0 / industry_sd ** 290    like_prec = W.sum()91    nat_var = 1.0 / (prior_prec + like_prec)92    nat_shift = nat_var * (R * W[:, None]).sum(axis=0)93    # régional : sur les résidus nets, par région, prior N(0, σ_rég²)94    reg_shift: dict[str, np.ndarray] = {}95    for reg in set(regions):96        mask = np.array([g == reg for g in regions])97        Rr, Wr = R[mask] - nat_shift[None, :], W[mask]98        rp = 1.0 / settings.regional_error_sd ** 299        rv = 1.0 / (rp + Wr.sum())100        reg_shift[reg] = rv * (Rr * Wr[:, None]).sum(axis=0)101    return NightUpdate(nat_shift, nat_var, reg_shift, len(resids))102103104def update(db: Session, reported: list[dict], n_sims: int = 15000) -> dict:105    """Résultats partiels → forecast en direct + déclarations conservatrices."""106    from ..pipeline import _district_inputs107    from .. import models as Mo108    run = (db.query(Mo.ForecastRun).filter(Mo.ForecastRun.is_backtest.is_(False))109           .order_by(Mo.ForecastRun.as_of.desc(), Mo.ForecastRun.id.desc()).first())110    if run is None:111        raise RuntimeError("Aucun forecast pré-électoral (prior manquant)")112    result_ref = run.national.get("baseline_national")113    inp = _district_inputs(db, result_ref)114    inp.x_mean = np.array(run.national["x_forecast"])115    inp.P = np.array(run.national["P_forecast"])116    industry_sd = float(run.diagnostics.get("industry_sd",117                                            settings.industry_error_sd))118119    upd = infer_shifts(inp, reported, industry_sd)120121    # état national mis à jour : moyenne décalée, incertitude nationale réduite122    inp.x_mean = inp.x_mean + upd.national_shift123    M = len(inp.x_mean)124    # remplace la part « erreur systémique » du P par la variance posterior125    inp.P = inp.P + (upd.national_var - industry_sd ** 2) * np.eye(M)126    inp.P = 0.5 * (inp.P + inp.P.T)127    ev = np.linalg.eigvalsh(inp.P)128    if ev.min() < 1e-8:                       # PSD après réduction129        inp.P += (1e-8 - ev.min()) * np.eye(M)130131    # chocs régionaux inférés injectés comme scénario régional (moyennes)132    reg_pp: dict[str, dict[str, float]] = {}133    for reg, shift in upd.regional_shift.items():134        base = close(np.array(135            [result_ref[p] for p in settings.parties]) / 100.0)136        delta = (inv_alr(alr(base) + shift) - base) * 100.0137        reg_pp[reg] = {p: round(float(delta[i]), 2)138                       for i, p in enumerate(settings.parties)139                       if abs(delta[i]) >= 0.1}140    sim = run_simulation(inp, n_sims=n_sims,141                         scenario={"regional_pp": reg_pp} if reg_pp else None,142                         rng=np.random.default_rng(20261005))143144    by_name = {r["district"]: r for r in reported}145    districts, called = [], 0146    for d in sim.districts:147        rep = by_name.get(d["district"])148        frac = float((rep or {}).get("pct_reported", 0)) / 100.0149        wp = dict(d["win_probs"])150        leader = max(wp, key=wp.get)151        if rep and frac >= 0.99:              # dépouillement complet = décidé152            obs = rep["results"]153            leader = max(obs, key=obs.get)154            wp = {p: (1.0 if p == leader else 0.0) for p in settings.parties}155        is_called = (wp[leader] >= CALL_PROB and156                     (frac >= CALL_MIN_REPORTED or wp[leader] >= 0.9999))157        called += is_called158        districts.append({"district": d["district"], "region": d["region"],159                          "leader": leader, "win_probs": wp,160                          "pct_reported": round(frac * 100, 1),161                          "called": is_called,162                          "reported_results": (rep or {}).get("results")})163    state = {164        "at": datetime.now(timezone.utc).isoformat(),165        "prior_run_id": run.id,166        "n_reported": len(reported), "n_informative": upd.n_informative,167        "national_shift_pp": {p: round(float(168            (inv_alr(inp.x_mean)[i] - inv_alr(np.array(169                run.national["x_forecast"]))[i]) * 100), 2)170            for i, p in enumerate(settings.parties)},171        "regional_shift_pp": reg_pp,172        "seats": sim.seats, "summary": sim.seat_matrix_summary,173        "national_vote": sim.national_vote,174        "districts": districts, "seats_called": called,175        "method": ("Prior = forecast pré-électoral; erreurs nationale/régionales "176                   "inférées hiérarchiquement des circonscriptions rapportées "177                   "(pondérées par leur dépouillement); déclarations à "178                   f"P ≥ {CALL_PROB:.1%} et ≥ {CALL_MIN_REPORTED:.0%} dépouillé."),179    }180    STATE_FILE.write_text(json.dumps(state, ensure_ascii=False))181    return state182183184def latest_state() -> dict | None:185    if STATE_FILE.exists():186        return json.loads(STATE_FILE.read_text())187    return None188