spb/qc-election
Public
Python 66.6%
HTML 24.8%
CSS 4.9%
JavaScript 3.6%
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"""Tests des couches v2 « au-delà des sondages » (fonctions pures)."""6from datetime import date78import numpy as np910from app.config import settings11from app.modeling import ensemble as ENS12from app.modeling import fundamentals as FUND13from app.modeling.compositions import alr, close, inv_alr141516def test_fundamentals_prior_is_valid_composition():17 prior = FUND.compute_prior(date(2026, 8, 29))18 assert prior.mean_shares.shape == (6,)19 assert abs(prior.mean_shares.sum() - 1.0) < 1e-920 assert (prior.mean_shares > 0).all()21 # covariance symétrique définie positive22 assert np.allclose(prior.P, prior.P.T)23 assert (np.linalg.eigvalsh(prior.P) > 0).all()24 # le prior doit être diffus (sd alr large)25 assert prior.detail["sd_alr"] >= 0.3262728def test_fundamentals_regression_sane():29 prior = FUND.compute_prior(date(2026, 8, 29), satisfaction=35.0)30 reg = prior.detail["regression"]31 # la satisfaction doit prédire positivement le vote du sortant32 assert reg["b_satisfaction"] > 033 assert reg["n_elections"] == 834 # erreur LOO honnête, ni nulle ni délirante35 assert 1.5 < reg["loo_rmse_pp"] < 12.036 # sortant plausible pour satisfaction 35 % après 2 mandats37 assert 15.0 < prior.detail["prior_pct"]["CAQ"] < 45.0383940def test_blend_weight_schedule():41 assert FUND.blend_weight(0) == 0.042 assert FUND.blend_weight(-3) == 0.043 w30, w365, w540 = (FUND.blend_weight(d) for d in (30, 365, 540))44 assert 0 < w30 < w365 < w540 <= settings.fundamentals_max_weight45 assert w30 < 0.05 # à J-30, les fondamentaux ne pèsent presque plus464748def test_blend_limits():49 prior = FUND.compute_prior(date(2026, 8, 29))50 x_p = alr(close(np.array([0.15, 0.10, 0.20, 0.35, 0.15, 0.05])))51 P_p = np.eye(5) * 0.0152 # jour du vote : le blend est l'identité53 x0, P0, d0 = FUND.blend(x_p, P_p, prior, 0)54 assert np.allclose(x0, x_p) and np.allclose(P0, P_p)55 # loin du vote : tiré vers le prior, variance réduite, poids publié56 x1, P1, d1 = FUND.blend(x_p, P_p, prior, 540)57 assert 0 < d1["precision_share"] < 158 assert np.trace(P1) < np.trace(P_p)59 dist_before = np.linalg.norm(x_p - prior.x)60 assert np.linalg.norm(x1 - prior.x) < dist_before616263def test_byelection_implied_national_roundtrip():64 """Si la partielle reproduit exactement le résultat local 2022, l'observation65 implicite doit être le national 2022 (swing nul)."""66 from app.ingest.byelections import RESULT_2022, _riding_202267 base = _riding_2022("Jean-Talon")68 assert base is not None and abs(sum(base.values()) - 100) < 269 nat = close(np.array([RESULT_2022[p] for p in settings.parties]) / 100.0)70 b = close(np.array([base[p] for p in settings.parties]) / 100.0)71 implied = inv_alr(alr(nat) + settings.byelection_swing_shrink * (alr(b) - alr(b)))72 assert np.allclose(implied, nat, atol=1e-9)737475def test_byelection_swing_shrunk():76 """Un balayage local doit produire un swing national rétréci, pas intégral."""77 from app.ingest.byelections import RESULT_2022, _riding_202278 base = _riding_2022("Terrebonne")79 nat = close(np.array([RESULT_2022[p] for p in settings.parties]) / 100.0)80 b = close(np.array([base[p] for p in settings.parties]) / 100.0)81 surge = b.copy()82 i_pq = settings.parties.index("PQ")83 surge[i_pq] *= 2.5 # le PQ explose localement84 surge = close(surge)85 full = inv_alr(alr(nat) + (alr(surge) - alr(b)))86 shrunk = inv_alr(alr(nat) + settings.byelection_swing_shrink * (alr(surge) - alr(b)))87 assert nat[i_pq] < shrunk[i_pq] < full[i_pq]888990def test_media_nudge_cap_and_sum():91 x = alr(close(np.array([0.15, 0.10, 0.20, 0.35, 0.15, 0.05])))92 delta = {"PQ": settings.media_nudge_pp_max, "CAQ": -settings.media_nudge_pp_max}93 x2 = ENS.apply_nudge(x, delta)94 p2 = inv_alr(x2)95 assert abs(p2.sum() - 1.0) < 1e-996 moved = (p2 - inv_alr(x)) * 10097 assert abs(moved).max() <= settings.media_nudge_pp_max + 0.1598 # délta nul → état inchangé (aucun bruit numérique injecté)99 assert np.allclose(ENS.apply_nudge(x, {"PQ": 0.0}), x)100101102def test_market_blend():103 model = {"PQ": 0.70, "CAQ": 0.10, "PLQ": 0.12, "QS": 0.04, "PCQ": 0.04, "AUT": 0.0}104 market = {"PQ": 0.80, "CAQ": 0.05, "PLQ": 0.10, "QS": 0.03, "PCQ": 0.02}105 out = ENS.market_blend(model, market)106 assert out["weight_market"] == settings.market_blend_weight107 assert abs(sum(out["blended"].values()) - 1.0) < 1e-6108 assert model["PQ"] < out["blended"]["PQ"] < market["PQ"]109 # sans marché : identité110 same = ENS.market_blend(model, None)111 assert same["blended"] == model and same["weight_market"] == 0.0112113114def test_synthetic_poststratify():115 from app.modeling.synthetic_poll import poststratify, strata116 st = strata()117 assert len(st) == 96118 assert abs(sum(s["weight"] for s in st) - 1.0) < 1e-9119 cells = [{**s, "shares": {"CAQ": 20, "PLQ": 20, "PQ": 30, "QS": 10, "PCQ": 15, "AUT": 5}}120 for s in st]121 agg = poststratify(cells)122 assert abs(sum(agg.values()) - 100.0) < 0.5123 assert abs(agg["PQ"] - 30.0) < 0.2124 # strates non normalisées → renormalisées avant agrégation125 cells[0]["shares"] = {"CAQ": 40, "PLQ": 40, "PQ": 60, "QS": 20, "PCQ": 30, "AUT": 10}126 agg2 = poststratify(cells)127 assert abs(sum(agg2.values()) - 100.0) < 0.5128129130def test_forecast_drift_multiplier_widens_only():131 from datetime import date132 from app.modeling.forecast import forecast133 from app.modeling.trend import TrendResult134 x = np.zeros(5); P = np.eye(5) * 0.005135 tr = TrendResult(dates=[date(2026, 8, 1)], share_mean=np.zeros((1, 6)),136 share_lo=np.zeros((1, 6)), share_hi=np.zeros((1, 6)),137 x=x, P=P, q=1e-4, loglik=0.0, n_polls=10)138 f1 = forecast(tr, date(2026, 8, 30), date(2026, 10, 5))139 f2 = forecast(tr, date(2026, 8, 30), date(2026, 10, 5), drift_multiplier=1.45)140 assert np.allclose(f1.x, f2.x) # moyenne intacte141 assert np.trace(f2.P) > np.trace(f1.P) # variance élargie142143144def test_scoring_functions():145 from app.modeling.validation import scoring as SC146 parties = ["CAQ", "PLQ", "PQ"]147 fc = {"CAQ": 40.0, "PLQ": 15.0, "PQ": 15.0}148 actual = {"CAQ": 41.0, "PLQ": 14.4, "PQ": 14.6}149 ve = SC.vote_errors(fc, actual, parties)150 assert 0 < ve["mae_pp"] < 1.0 and ve["bias_pp"]["CAQ"] == -1.0151 summary = {p: {"p25": fc[p] - 1, "p75": fc[p] + 1, "p05": fc[p] - 2,152 "p95": fc[p] + 2, "lo95": fc[p] - 3, "hi95": fc[p] + 3,153 "mean": fc[p]} for p in parties}154 cov = SC.interval_coverage(summary, actual, parties)155 assert cov["coverage_95"] == 1.0156 probs = np.array([0.1] * 20 + [0.9] * 20)157 outs = np.array([0] * 18 + [1] * 2 + [1] * 18 + [0] * 2)158 bins = SC.calibration_bins(probs, outs)159 assert len(bins) == 2 and abs(bins[0]["observed"] - 0.1) < 0.01160161162def test_volatility_bounds():163 from app.modeling.signals import volatility as V164 from app.config import settings165 # sans DB sociale : seul wiki agit; z énorme → plafonné à la borne166 class FakeDB: pass167 saved = settings.enable_social_volatility168 object.__setattr__(settings, "enable_social_volatility", False)169 try:170 out = V.compute(None, {"z_max": 50.0})171 assert 1.0 <= out["multiplier"] <= settings.volatility_multiplier_max172 assert out["multiplier"] == settings.volatility_multiplier_max173 calm = V.compute(None, {"z_max": 0.5})174 assert calm["multiplier"] == 1.0175 finally:176 object.__setattr__(settings, "enable_social_volatility", saved)177178179def test_tipping_points_math():180 from app.modeling.simulate import tipping_points181 parties = ["A", "B"]182 # 3 circonscriptions, majorité à 3 : le 3e siège le plus sûr (le plus183 # serré, c2) est TOUJOURS celui qui donne la majorité → pivot certain.184 S, D = 200, 3185 rng = np.random.default_rng(7)186 shares = np.zeros((S, D, 2))187 shares[..., 0] = 0.55 + rng.normal(0, 0.008, (S, D))188 shares[:, 2, 0] = 0.51189 shares[..., 1] = 1 - shares[..., 0]190 winners = shares.argmax(axis=-1)191 tips = tipping_points(winners, shares, ["c0", "c1", "c2"], 3, parties)192 assert tips["n_sims_with_majority"] == S193 top = tips["majority_tipping"][0]194 assert top["district"] == "c2" and top["prob_majority_tipping"] > 0.9195 # majorité à 2 : pivot = 2e marge la plus sûre → c0 ou c1, jamais c2196 tips2 = tipping_points(winners, shares, ["c0", "c1", "c2"], 2, parties)197 names = {t["district"] for t in tips2["majority_tipping"]}198 assert "c2" not in names and names <= {"c0", "c1"}199200201def test_ablation_overrides_restore():202 from app.modeling.validation.ablation import _overrides203 from app.config import settings204 before = settings.fundamentals_enabled205 with _overrides({"fundamentals_enabled": not before}):206 assert settings.fundamentals_enabled == (not before)207 assert settings.fundamentals_enabled == before208209210def test_coordination_filter():211 from app.modeling.signals.social import coordination_filter212 from datetime import datetime, timezone213 class P:214 def __init__(self, text):215 self.text = text216 self.fetched_at = datetime(2026, 8, 30, 12, tzinfo=timezone.utc)217 organic = [P(f"opinion unique numéro {i} sur la campagne") for i in range(10)]218 spam = [P("Votez maintenant contre le gouvernement corrompu!") for _ in range(6)]219 org, diag = coordination_filter(organic + spam)220 assert diag["n_coordinated"] == 6 and diag["n_organic"] == 10221222223def test_crps_and_log_score():224 from app.modeling.validation.scoring import crps_from_draws, log_score_alr225 rng = np.random.default_rng(3)226 actual = np.array([30.0, 25.0, 20.0])227 tight = rng.normal(actual, 1.0, (4000, 3))228 wide = rng.normal(actual, 6.0, (4000, 3))229 assert crps_from_draws(tight, actual) < crps_from_draws(wide, actual)230 biased = rng.normal(actual + 8.0, 1.0, (4000, 3))231 assert crps_from_draws(tight, actual) < crps_from_draws(biased, actual)232 x = np.zeros(3); P = np.eye(3) * 0.02233 assert log_score_alr(x, P, np.zeros(3)) > log_score_alr(x, P, np.full(3, 0.5))234235236def test_industry_sigma_bounds():237 import pathlib238 from app.config import DATA_DIR239 if not (DATA_DIR / "historical" / "polls_2018.csv").exists():240 return # cache absent (premier déploiement) — le pipeline le créera241 from app.modeling.national.pollster_error import (SIGMA_CAP, SIGMA_FLOOR,242 industry_sigma_loeo)243 for yr in (None, 2018, 2022):244 s = industry_sigma_loeo(yr)245 assert SIGMA_FLOOR <= s <= SIGMA_CAP246247248def test_replay_smoke_2018():249 from app.config import DATA_DIR250 if not (DATA_DIR / "historical" / "polls_2018.csv").exists():251 return252 from app.modeling.validation.historical_replay import replay_election253 rep = replay_election(2018, horizons=[30, 7])254 assert rep["avg"] and rep["avg"]["mae_pp"] < 12255 for r in rep["horizons"]:256 assert "error" not in r257 fcst = r["forecast"]258 assert abs(sum(fcst.values()) - 100) < 3 # composition ≈ 100 %259 # LOEO : le prior de fondamentaux exclut 2018 (pas de fuite)260 # l'ère 2018 n'a pas de PCQ : le modèle tourne à 5 partis261 assert "PCQ" not in rep["parties"]262263264def test_era_parties_trend():265 """Le cœur alr fonctionne avec une ère à 5 partis (ADQ, sans CAQ/PCQ)."""266 from datetime import date, timedelta267 from app.modeling.trend import fit_trend268 parties = ["ADQ", "PLQ", "PQ", "QS", "AUT"]269 polls = [{"pollster": f"Maison{i%3}", "field_end": date(2007, 1, 1) + timedelta(days=i * 7),270 "sample_size": 1000, "mode": "unknown",271 "shares": {"ADQ": 30.0 + i * 0.1, "PLQ": 33.0, "PQ": 28.0, "QS": 4.0, "AUT": 5.0}}272 for i in range(8)]273 tr = fit_trend(polls, {}, date(2007, 3, 1), parties=parties)274 assert tr is not None and tr.x.shape == (4,)275 assert abs(tr.share_mean[-1].sum() - 1.0) < 1e-9276277278def test_primary_report_extraction():279 from app.modeling.validation.scoring import vote_errors # noqa (import sanity)280 from app.ingest.pollster_reports import extract_toplines281 md = """Sondage Léger — Intentions de vote au Québec282 Réalisé du 22 au 26 septembre 2026 auprès de 1024 répondants.283 Intentions de vote : le PQ obtient 31 %, la CAQ 22 %, le Parti libéral 21 %,284 le Parti conservateur du Québec 14 % et Québec solidaire 9 %."""285 ext = extract_toplines(md)286 assert ext["shares"] == {"PQ": 31.0, "CAQ": 22.0, "PLQ": 21.0,287 "PCQ": 14.0, "QS": 9.0}288 assert str(ext["field_end"]) == "2026-09-26"289 assert ext["sample_size"] == 1024290 assert ext["confidence"] >= 0.9291 # rapport pauvre → confiance basse, jamais de création silencieuse292 bad = extract_toplines("Communiqué sans chiffres pertinents.")293 assert bad["confidence"] < 0.7 and not bad["shares"]294295296def test_invariants_simulation():297 """Invariants §qualité : 127 sièges/simulation, probs [0,1], somme=1."""298 from app.modeling.simulate import SimulationInput, run_simulation299 D, K = 127, 6300 rng = np.random.default_rng(11)301 baselines = rng.dirichlet(np.ones(K) * 8, D)302 inp = SimulationInput(303 x_mean=np.zeros(K - 1), P=np.eye(K - 1) * 0.02,304 baseline_national=np.full(K, 1 / K),305 district_names=[f"c{i}" for i in range(D)],306 district_baselines=baselines,307 district_regions=[f"r{i % 13}" for i in range(D)],308 retirement_flags=np.zeros(D, dtype=int),309 incumbent_party_idx=np.full(D, -1), majority_seats=64)310 sim = run_simulation(inp, n_sims=2000, keep_raw=True)311 counts = np.zeros(2000)312 for k in range(K):313 counts += (sim.raw_winners == k).sum(axis=1) * 0 + 0 # noqa314 assert sim.raw_winners.shape == (2000, D)315 per_sim = np.ones((2000, D)).sum(axis=1)316 assert (per_sim == D).all() # 127 sièges par simulation317 for d in sim.districts:318 s = sum(d["win_probs"].values())319 assert abs(s - 1.0) < 1e-6 # P(victoire) somme à 1320 assert all(0.0 <= v <= 1.0 for v in d["win_probs"].values())321 assert abs(sum(sim.national_vote.values()) - 100.0) < 0.5322323324def test_election_night_infers_national_shift():325 """Un vrai décalage national doit être retrouvé par la décomposition326 hiérarchique, et un décalage nul ne doit rien inventer."""327 from app.modeling.election_night import infer_shifts328 from app.modeling.simulate import SimulationInput329 from app.modeling.compositions import alr, close, inv_alr330 rng = np.random.default_rng(5)331 D, K = 60, 6332 baselines = rng.dirichlet(np.ones(K) * 10, D)333 nat = np.full(K, 1 / K)334 inp = SimulationInput(335 x_mean=alr(close(nat)), P=np.eye(K - 1) * 0.01,336 baseline_national=nat,337 district_names=[f"c{i}" for i in range(D)],338 district_baselines=baselines,339 district_regions=[f"r{i % 6}" for i in range(D)],340 retirement_flags=np.zeros(D, dtype=int),341 incumbent_party_idx=np.full(D, -1), majority_seats=31)342 # vérité : PQ (idx 3) +3 pp national — rapporté dans 30 circonscriptions343 true_shift_pp = np.zeros(K); true_shift_pp[3] = 3.0344 reported = []345 for i in range(30):346 obs = close(np.clip(baselines[i] + true_shift_pp / 100.0, 0.001, None))347 reported.append({"district": f"c{i}", "pct_reported": 100.0,348 "results": {p: float(obs[j] * 100)349 for j, p in enumerate(350 ["CAQ", "PCQ", "PLQ", "PQ", "QS", "AUT"])}})351 from app.config import settings352 upd = infer_shifts(inp, reported, industry_sd=0.22)353 shifted = inv_alr(inp.x_mean + upd.national_shift)354 dpq = (shifted[3] - nat[3]) * 100355 assert 1.5 < dpq < 4.5 # décalage PQ retrouvé (~3 pp, rétréci OK)356 assert upd.national_var < 0.22 ** 2 # l'incertitude nationale a diminué357 # décalage nul → rien d'inventé358 rep0 = [{**r, "results": {p: float(baselines[i][j] * 100)359 for j, p in enumerate(360 ["CAQ", "PCQ", "PLQ", "PQ", "QS", "AUT"])}}361 for i, r in enumerate(reported)]362 upd0 = infer_shifts(inp, rep0, industry_sd=0.22)363 assert np.abs(inv_alr(inp.x_mean + upd0.national_shift) - nat).max() * 100 < 0.5364 # sous le seuil de dépouillement : aucune information365 updm = infer_shifts(inp, [{**reported[0], "pct_reported": 2.0}], 0.22)366 assert updm.n_informative == 0367368369def test_demographics_feature_store():370 from app.config import DATA_DIR371 from app.ingest.eq_socioeconomic import slug_of372 assert slug_of("Anjou–Louis-Riel") == "anjou-louis-riel"373 assert slug_of("Sainte-Marie–Saint-Jacques") == "sainte-marie-saint-jacques"374 assert slug_of("D'Arcy-McGee") == "darcy-mcgee"375 f = DATA_DIR / "population" / "riding_2026_demographics.csv"376 if not f.exists():377 return # premier déploiement — construit par admin/demographics378 import pandas as pd379 df = pd.read_csv(f)380 assert len(df) == 127381 assert df["population"].between(10000, 130000).all()382 assert df["pct_francais"].between(1, 100).all()383 assert df["pct_locataires"].between(5, 95).all()384