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"""Historical Replay Engine (§52-54) — la validation PRINCIPALE du modèle.67Rejoue les élections 2007, 2008, 2012, 2014, 2018 (national) et 20228(national + circonscriptions, via le backtest existant) en LEAVE-ONE-ELECTION-OUT9strict :1011  * house effects appris uniquement sur les AUTRES élections;12  * prior de fondamentaux ajusté avec `exclude_year` (l'élection testée ne13    participe jamais à son propre prior);14  * σ_industrie estimé sur les AUTRES élections (`national/pollster_error.py`);15  * rolling origin J-120 → J-1 : seulement l'information ≤ date.1617Scores propres (§55) : EAM/RMSE du vote, couverture 50/90/95, CRPS, log score,18sharpness, biais signé. Rapport : `data/replay_report.json` + `/api/replay`.19"""20from __future__ import annotations2122import json23from datetime import date, timedelta2425import numpy as np2627from ...config import DATA_DIR, settings28from ..compositions import alr, close, inv_alr29from ..forecast import forecast as fc30from ..fundamentals import HISTORY, blend, compute_prior31from ..house_effects import compute_profiles32from ..trend import fit_trend33from .scoring import (crps_from_draws, interval_coverage, log_score_alr,34                      sharpness, vote_errors)3536HORIZONS = [120, 90, 60, 45, 30, 21, 14, 7, 3, 1]373839def _era_prev_result(year: int) -> dict[str, float] | None:40    """Vote de l'élection PRÉCÉDENTE, exprimé dans l'ère de partis de `year`."""41    from ...ingest.wikipedia_historical import ELECTIONS42    PREV = {2007: {"PLQ": 45.99, "PQ": 33.24, "ADQ": 18.18, "QS": 1.06, "AUT": 1.73},43            2008: {"PLQ": 33.08, "PQ": 28.35, "ADQ": 30.84, "QS": 3.64, "AUT": 4.09},44            2012: {"PLQ": 42.08, "PQ": 35.17, "CAQ": 16.37, "QS": 3.78, "AUT": 2.60},45            2014: {"PLQ": 31.20, "PQ": 31.95, "CAQ": 27.05, "QS": 6.03, "AUT": 3.77},46            2018: {"PLQ": 41.52, "PQ": 25.38, "CAQ": 23.05, "QS": 7.63, "AUT": 2.42}}47    return PREV.get(year)484950def _loeo_profiles(test_year: int, parties: list[str]):51    """House effects appris sur toutes les élections rejouables ≠ test_year52    (dans l'ère de partis du test — seuls les partis communs contribuent)."""53    from ...ingest.wikipedia_historical import ELECTIONS, load_polls, to_model_polls54    polls, elections = [], []55    for yr, meta in ELECTIONS.items():56        if yr == test_year:57            continue58        elections.append({"date": meta["date"], "result": meta["result"]})59        polls.extend(to_model_polls(load_polls(yr)))60    return compute_profiles(polls, elections, parties=parties)616263def _baselines_at(polls: list[dict], as_of: date, parties: list[str],64                  actual: dict) -> dict:65    """Règles simples contre lesquelles le modèle DOIT gagner (§calibration) :66    B1 = dernier sondage publié; B2 = moyenne simple des 14 derniers jours67    (repli : 3 derniers sondages)."""68    hist = sorted((p for p in polls if p["field_end"] <= as_of),69                  key=lambda p: p["field_end"])70    if not hist:71        return {}72    def norm(sh):73        tot = sum(sh.get(p, 0.0) for p in parties) or 100.074        return {p: sh.get(p, 0.0) * 100.0 / tot for p in parties}75    last = norm(hist[-1]["shares"])76    recent = [p for p in hist if (as_of - p["field_end"]).days <= 14] or hist[-3:]77    avg = {pt: float(np.mean([norm(p["shares"])[pt] for p in recent]))78           for pt in parties}79    return {80        "dernier_sondage": vote_errors(last, actual, parties)["mae_pp"],81        "moyenne_14j": vote_errors(avg, actual, parties)["mae_pp"],82    }838485def replay_election(year: int, use_fundamentals: bool = True,86                    use_loeo_house: bool = True,87                    industry_sd: float | None = None,88                    horizons: list[int] | None = None) -> dict:89    """Rejoue une élection à tous les horizons; retourne les scores."""90    from ...ingest.wikipedia_historical import ELECTIONS, load_polls, to_model_polls91    meta = ELECTIONS[year]92    parties = meta["parties"]93    eday = meta["date"]94    actual = meta["result"]95    actual_vec = close(np.array([actual[p] for p in parties]) / 100.0)96    polls = to_model_polls(load_polls(year))97    profiles = _loeo_profiles(year, parties) if use_loeo_house else {}98    prev = _era_prev_result(year)99100    rows = []101    rng = np.random.default_rng(year)102    for h in (horizons or HORIZONS):103        as_of = eday - timedelta(days=h)104        trend = fit_trend(polls, profiles, as_of, parties=parties)105        if trend is None:106            rows.append({"horizon": h, "error": "sondages insuffisants"})107            continue108        f = fc(trend, as_of, eday, parties=parties, industry_sd=industry_sd)109        if use_fundamentals and prev is not None:110            hist_row = next((r for r in HISTORY if r[0] == year), None)111            if hist_row:112                _, inc, terms, sat, _, _ = hist_row113                prior = compute_prior(as_of, incumbent=inc, terms=terms,114                                      satisfaction=sat, prev_result=prev,115                                      exclude_year=year, parties=parties)116                x_b, P_b, _ = blend(f.x, f.P, prior, h, parties=parties)117                from ..forecast import distribution_from118                f = distribution_from("forecast", as_of, x_b, P_b,119                                      parties=parties)120        # tirages pour CRPS121        Mx = len(f.x)122        L = np.linalg.cholesky(f.P + 1e-10 * np.eye(Mx))123        draws = inv_alr(f.x + rng.standard_normal((4000, Mx)) @ L.T) * 100.0124        fc_mean = {p: f.summary[p]["mean"] for p in parties}125        rows.append({126            "horizon": h, "as_of": as_of.isoformat(), "n_polls": trend.n_polls,127            **vote_errors(fc_mean, actual, parties),128            **interval_coverage(f.summary, actual, parties),129            "crps_pp": crps_from_draws(draws, actual_vec * 100.0),130            "log_score": log_score_alr(f.x, f.P, alr(actual_vec)),131            "sharpness_pp": sharpness(f.summary, parties),132            "forecast": {p: round(fc_mean[p], 1) for p in parties},133            "baselines_mae_pp": _baselines_at(polls, as_of, parties, actual),134        })135    ok = [r for r in rows if "error" not in r]136    agg = {}137    if ok:138        for k in ("mae_pp", "rmse_pp", "crps_pp", "log_score", "sharpness_pp"):139            agg[k] = round(float(np.mean([r[k] for r in ok])), 3)140        for k in ("coverage_50", "coverage_90", "coverage_95"):141            vals = [r[k] for r in ok if k in r]142            if vals:143                agg[k] = round(float(np.mean(vals)), 3)144        for b in ("dernier_sondage", "moyenne_14j"):145            vals = [r["baselines_mae_pp"][b] for r in ok146                    if r.get("baselines_mae_pp", {}).get(b) is not None]147            if vals:148                agg[f"baseline_{b}_mae_pp"] = round(float(np.mean(vals)), 3)149    return {"year": year, "parties": parties, "actual": actual,150            "horizons": rows, "avg": agg}151152153def run_replay(use_fundamentals: bool = True,154               industry_from_loeo: bool = True) -> dict:155    """Replay complet multi-élections (national). 2022 reste couvert par le156    backtest circonscriptions existant — les deux rapports se complètent."""157    from ...ingest.wikipedia_historical import ELECTIONS158    from ..national.pollster_error import industry_sigma_loeo159    report = {"generated": date.today().isoformat(),160              "model_version": settings.model_version,161              "protocol": ("LOEO strict : house effects, fondamentaux et "162                           "σ_industrie appris sur les autres élections; "163                           "rolling origin J-120→J-1."),164              "elections": {}}165    for yr in sorted(ELECTIONS):166        sd = industry_sigma_loeo(yr) if industry_from_loeo else None167        rep = replay_election(yr, use_fundamentals=use_fundamentals,168                              industry_sd=sd)169        rep["industry_sd_loeo"] = round(sd, 4) if sd is not None else None170        report["elections"][str(yr)] = rep171    # agrégat global (moyenne des moyennes par élection)172    aggs = [e["avg"] for e in report["elections"].values() if e["avg"]]173    if aggs:174        report["overall"] = {k: round(float(np.mean([a[k] for a in aggs if k in a])), 3)175                             for k in aggs[0]}176    (DATA_DIR / "replay_report.json").write_text(177        json.dumps(report, indent=1, ensure_ascii=False))178    return report179