# 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 """Chargement des CSV normalisés (data/) vers la base — idempotent (upsert).""" from __future__ import annotations from datetime import date, datetime import pandas as pd from sqlalchemy.orm import Session from .config import DATA_DIR, settings from .db import SessionLocal, init_db from . import models as Mo PARTY_COLORS = { # teintes usuelles des partis, ajustées et validées (CVD, contraste) "CAQ": "#0097cf", "PLQ": "#e5484d", "PQ": "#3f7ad6", "QS": "#e06c1f", "PCQ": "#7069d6", "AUT": "#ad8226", } LEADERS = { "CAQ": "Christine Fréchette", "PLQ": "Charles Milliard", "PQ": "Paul St-Pierre Plamondon", "QS": "Ruba Ghazal et Sol Zanetti", "PCQ": "Éric Duhaime", "AUT": None, } RESULT_2022 = {"CAQ": 40.98, "PLQ": 14.37, "QS": 15.43, "PQ": 14.61, "PCQ": 12.91, "AUT": 1.70} RESULT_2018 = {"CAQ": 37.42, "PLQ": 24.82, "QS": 16.10, "PQ": 17.06, "PCQ": 1.46, "AUT": 3.14} # Dates officielles — Élections Québec (electionsquebec.qc.ca) KEY_EVENTS = [ (date(2026, 8, 27), "Début de la période de campagne (préparatifs)", "campagne", 0.7), (date(2026, 8, 29), "Déclenchement officiel de l'élection (décret prévu)", "campagne", 0.9), (date(2026, 9, 16), "Date limite — inscription au vote hors Québec", "échéance", 0.5), (date(2026, 9, 25), "Début du vote par anticipation et du vote au bureau du directeur du scrutin (jusqu'au 1er oct.)", "vote", 0.8), (date(2026, 10, 5), "Jour du scrutin — 127 circonscriptions, ~6,8 M d'électeurs admissibles", "élection", 1.0), ] def get_or_create(db: Session, model, defaults=None, **kw): obj = db.query(model).filter_by(**kw).first() if obj: return obj, False obj = model(**kw, **(defaults or {})) db.add(obj) db.flush() return obj, True def seed_all(db: Session | None = None) -> dict: init_db() own = db is None db = db or SessionLocal() stats = {} try: # Élections el26, _ = get_or_create(db, Mo.Election, name="Élection générale du Québec 2026", defaults=dict(election_date=settings.election_date, total_seats=127, majority_seats=64, is_target=True)) el22, _ = get_or_create(db, Mo.Election, name="Élection générale du Québec 2022", defaults=dict(election_date=date(2022, 10, 3), total_seats=125, majority_seats=63, actual_result=RESULT_2022)) el18, _ = get_or_create(db, Mo.Election, name="Élection générale du Québec 2018", defaults=dict(election_date=date(2018, 10, 1), total_seats=125, majority_seats=63, actual_result=RESULT_2018)) # Partis (mise à jour des chefs/couleurs si déjà présents) for code in settings.parties: party, _ = get_or_create(db, Mo.Party, code=code, defaults=dict(name=settings.party_names[code])) party.name = settings.party_names[code] party.leader = LEADERS.get(code) party.color = PARTY_COLORS.get(code) # Circonscriptions ddf = pd.read_csv(DATA_DIR / "districts_2026.csv") for _, r in ddf.iterrows(): base = {p: float(r[f"b_{p}"]) for p in settings.parties} d, created = get_or_create(db, Mo.District, name=r["district"], defaults=dict(region=r["region"], baseline_shares=base)) d.region = r["region"]; d.baseline_shares = base d.baseline_source = r["baseline_source"] d.incumbent_party = None if pd.isna(r["incumbent_party"]) else r["incumbent_party"] d.incumbent_running = bool(r["incumbent_running"]) d.turnout_2022 = None if pd.isna(r["turnout_2022"]) else float(r["turnout_2022"]) d.is_new_2026 = bool(r["is_new_2026"]) stats["districts"] = len(ddf) # Résultats historiques par circonscription (2022 et 2018 reconstruit) rdf = pd.read_csv(DATA_DIR / "riding_results_2022.csv") for _, r in rdf.iterrows(): for p in settings.parties: if p in r and not pd.isna(r[p]): obj, _ = get_or_create(db, Mo.HistoricalDistrictResult, election_id=el22.id, district_name=r["district"], party=p, defaults=dict(pct=float(r[p]))) obj.pct = float(r[p]) # Purge des doublons hérités de l'ancienne clé (maison, fin, taille) : # un même sondage réingéré quand Wikipédia complétait sa taille. n_dupes = 0 seen_polls: dict[tuple, Mo.Poll] = {} for poll in db.query(Mo.Poll).order_by(Mo.Poll.id.asc()).all(): key = (poll.pollster_id, poll.field_end, poll.election_id) kept = seen_polls.get(key) if kept is None: seen_polls[key] = poll else: # garder la version la plus complète (taille connue, + de partis) a = (kept.sample_size is not None, len(kept.results), -kept.id) b = (poll.sample_size is not None, len(poll.results), -poll.id) if b > a: seen_polls[key], poll = poll, kept db.delete(poll) n_dupes += 1 # même maison, même taille d'échantillon, fins de terrain à ±1 jour : # même sondage daté différemment par deux agrégateurs (EN vs FR) by_house: dict[tuple, list] = {} for poll in db.query(Mo.Poll).order_by(Mo.Poll.field_end.asc()).all(): if poll.sample_size: by_house.setdefault((poll.pollster_id, poll.election_id, poll.sample_size), []).append(poll) for group in by_house.values(): for a, b in zip(group, group[1:]): if abs((b.field_end - a.field_end).days) <= 1 and a in db: db.delete(a) # garder la date la plus récente (FR corrige) n_dupes += 1 if n_dupes: db.flush() stats["dupes_purgés"] = n_dupes # Sondages n_polls, n_updated = 0, 0 for fname, el in (("polls_2026.csv", el26), ("polls_2022.csv", el22)): f = DATA_DIR / fname if not f.exists(): continue pdf = pd.read_csv(f) for (pollster, fe), grp in pdf.groupby(["pollster", "field_end"]): # dédoublonnage : le même sondage peut être publié dans plusieurs # tables/sites (clé = maison + fin de terrain + parti) grp = grp.drop_duplicates(subset="party", keep="first") pol, _ = get_or_create(db, Mo.Pollster, name=pollster) first = grp.iloc[0] ss = None if pd.isna(first["sample_size"]) else int(first["sample_size"]) # correspondance par (maison, fin de terrain ±1 j) SANS la # taille : Wikipédia complète n/MoE après coup, et les # agrégateurs EN/FR datent parfois le même sondage à ±1 jour. fe_d = date.fromisoformat(fe) from datetime import timedelta as _td poll = (db.query(Mo.Poll) .filter(Mo.Poll.pollster_id == pol.id, Mo.Poll.field_end >= fe_d - _td(days=1), Mo.Poll.field_end <= fe_d + _td(days=1)) .first()) if poll is None: poll = Mo.Poll( election_id=el.id, pollster_id=pol.id, field_start=date.fromisoformat(first["field_start"]), field_end=date.fromisoformat(fe), sample_size=ss, population=first["population"], region=first["region"], source_name=first["source_name"], accessed_at=datetime.fromisoformat(first["accessed_at"])) db.add(poll); db.flush() n_polls += 1 # métadonnées raffinées à chaque passage (n, MoE, mode, indécis, URL) if ss is not None: poll.sample_size = ss if not pd.isna(first["moe"]): poll.moe = float(first["moe"]) if "undecided" in first and not pd.isna(first.get("undecided")): poll.undecided = float(first["undecided"]) if first["mode"] and str(first["mode"]) != "nan" and ( poll.mode in (None, "unknown") or poll.mode != first["mode"]): poll.mode = first["mode"] if not pd.isna(first["source_url"]): poll.source_url = first["source_url"] # résultats : création OU correction si Wikipédia a révisé un chiffre total = grp["raw_value"].sum() by_party = {pr.party: pr for pr in poll.results} changed = False for _, row in grp.iterrows(): raw = float(row["raw_value"]) normalized = round(raw * 100.0 / total, 3) pr = by_party.get(row["party"]) if pr is None: poll.results.append(Mo.PollResult( party=row["party"], raw_value=raw, normalized_value=normalized)) elif abs(pr.raw_value - raw) > 1e-9: pr.raw_value = raw pr.normalized_value = normalized changed = True if changed: n_updated += 1 stats["new_polls"] = n_polls stats["updated_polls"] = n_updated # Événements clés for d, title, kind, imp in KEY_EVENTS: get_or_create(db, Mo.NewsEvent, event_date=d, title=title, defaults=dict(kind=kind, importance=imp, detected_by="seed")) # Version du modèle get_or_create(db, Mo.ModelVersion, version="1.0.0", defaults=dict(changelog="Modèle initial : DLM Kalman (alr), " "house effects 2022, Monte Carlo corrélé.")) get_or_create(db, Mo.ModelVersion, version="2.0.0", defaults=dict(changelog=( "« Au-delà des sondages » : votes réels des partielles injectés " "dans le filtre (swing rétréci ×0,6), prior de fondamentaux " "(satisfaction/vote précédent/usure, régression 1998-2022, poids " "décroissant w(t)), ajustement médias borné ±0,35 pp, ensemble " "88/12 avec Polymarket sur P(plus de sièges), veille web continue " "Firecrawl (radar sondages + satisfaction + presse élargie), " "décomposition du forecast par couche, backtest comparatif v2."))) get_or_create(db, Mo.ModelVersion, version="2.1.0", defaults=dict(changelog=( "Signaux web avancés : attention Wikipédia (pageviews partis+chefs, " "littérature Smith & Gustafson 2017) → détecteur de turbulence qui " "gonfle la variance de campagne (×≤1,45, jamais la moyenne); sondage " "synthétique LLM (« silicon sampling », 96 strates poststratifiées, " "ancré dans l'actualité réelle) — EXPÉRIMENTAL, poids nul; page " "quotidienne « Le point du jour » (/aujourdhui, API /api/today) : " "forecast daté, Δ vs veille, journal 24 h."))) get_or_create(db, Mo.ModelVersion, version="3.0.0", defaults=dict(changelog=( "v3 « 127 » Phases A+B : cadre de validation (scoring, ablation " "par variante, flags conservateurs), Social Intelligence Layer " "(acteur Apify maison multi-plateformes + Reddit/Mastodon/Lemmy, " "métriques normalisées attention_z/velocity_z/stance, filtrage de " "coordination, observabilité apify_runs), volatilité de campagne " "unifiée (wiki + social, bornée ≤1,5, variance seulement), Event " "Engine à confirmation croisée, circonscriptions pivots de la " "majorité (méthode des simulations), page /circonscriptions " "(forecast par comté), API v3 (/api/v3/*), data-health admin."))) get_or_create(db, Mo.ModelVersion, version="3.1.0", defaults=dict(changelog=( "Méthodologie ultra avancée — première vague : REPLAY HISTORIQUE " "multi-élections (2007/2008/2012/2014/2018/2022, 462 sondages, " "LOEO strict : house effects, fondamentaux et σ appris sur les " "autres élections; rolling origin J-120→J-1; CRPS + log score); " "modèle d'erreur des sondeurs (erreur d'industrie par élection, " "house effects hiérarchiques rétrécis n/(n+k)); σ_industrie " "APPRIS (LOEO ≈0,21 vs 0,12 manuel) adopté après validation : " "log score −3,80 vs −5,33, CRPS 2,64 vs 2,67, couverture 95 % " "83 % vs 76,5 %, EAM inchangée. Cœur alr paramétré par ères de " "partis (ADQ 2007-2008). Audits : METHODOLOGY_AUDIT, " "STATISTICAL_DEBT, V3_MODEL_ARCHITECTURE, HISTORICAL_REPLAY_PLAN."))) get_or_create(db, Mo.ModelVersion, version="3.2.0", defaults=dict(changelog=( "Release produit (distribution du forecast INCHANGÉE) : feature " "store démographique officiel (classeur Élections Québec, " "Recensement 2021 × 127 circ. 2026, similarité top-10), modèle du " "soir d'élection (mise à jour bayésienne hiérarchique, calling " "conservateur, dormant), scénarios partageables par URL, fiches " "enrichies (sensibilité au swing, chemin de bascule, démographie), " "baselines dans le replay, Course à 64, connecteurs primaires 7 " "firmes avec file de révision, favicon + og:image, feuille de " "route Compas électoral (COMPAS-PROGRAMMES.md)."))) # Sources for name, url, kind in [ ("Wikipédia — sondages 2026", "https://en.wikipedia.org/wiki/2026_Quebec_general_election", "polls"), ("Wikipédia — élection 2022", "https://en.wikipedia.org/wiki/2022_Quebec_general_election", "results"), ]: get_or_create(db, Mo.DataSource, name=name, defaults=dict(url=url, kind=kind)) db.commit() stats["polls_total"] = db.query(Mo.Poll).count() return stats finally: if own: db.close() if __name__ == "__main__": print(seed_all())