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"""Pipeline automatisé : collect → validate → normalize → dedupe → model →6simulate → publish. Chaque étape est isolée : l'échec d'une source n'arrête7pas le reste (journalisé dans pipeline_logs)."""8from __future__ import annotations910import logging11import traceback12from datetime import date, datetime, timezone1314import numpy as np15from sqlalchemy.orm import Session1617from .config import settings18from .db import SessionLocal, init_db19from . import models as Mo20from .modeling import ensemble as ENS21from .modeling import fundamentals as FUND22from .modeling import house_effects as HE23from .modeling import simulate as SIM24from .modeling.forecast import distribution_from, forecast as fc, nowcast as nc25from .modeling.trend import fit_trend2627log = logging.getLogger("pipeline")282930def _log(db: Session, step: str, status: str, message: str = "") -> None:31    db.add(Mo.PipelineLog(step=step, status=status, message=message[:2000]))32    db.commit()333435def load_polls(db: Session, election_id: int, as_of: date | None = None,36               region: str = "QC") -> list[dict]:37    q = (db.query(Mo.Poll).filter(Mo.Poll.election_id == election_id,38                                  Mo.Poll.excluded.is_(False),39                                  Mo.Poll.region == region))40    if as_of:41        q = q.filter(Mo.Poll.field_end <= as_of)42    out = []43    for poll in q.all():44        shares = {r.party: r.normalized_value for r in poll.results}45        out.append({"id": poll.id, "pollster": poll.pollster.name,46                    "field_end": poll.field_end, "sample_size": poll.sample_size,47                    "mode": poll.mode, "shares": shares})48    return out495051def compute_pollster_profiles(db: Session, exclude_after: date | None = None):52    """House effects sur les élections passées à résultat connu."""53    elections = [{"date": e.election_date, "result": e.actual_result}54                 for e in db.query(Mo.Election).filter(Mo.Election.actual_result.isnot(None))55                 if not exclude_after or e.election_date <= exclude_after]56    polls = []57    for e in db.query(Mo.Election).all():58        polls.extend(load_polls(db, e.id))59    profiles = HE.compute_profiles(polls, elections)60    # publication des cotes61    for name, prof in profiles.items():62        pol = db.query(Mo.Pollster).filter_by(name=name).first()63        if not pol:64            continue65        rating = (db.query(Mo.PollsterRating)66                  .filter_by(pollster_id=pol.id, computed_for="2026-general").first())67        if rating is None:68            rating = Mo.PollsterRating(pollster_id=pol.id, computed_for="2026-general")69            db.add(rating)70        rating.n_polls = prof.n_polls71        rating.mae_pp = prof.mae_pp72        rating.house_effects = prof.house_effects73        rating.weight_multiplier = prof.weight_multiplier74        rating.detail = prof.detail75        rating.computed_at = datetime.now(timezone.utc)76    db.commit()77    return profiles787980def _district_inputs(db: Session, baseline_national: dict[str, float]) -> SIM.SimulationInput:81    districts = db.query(Mo.District).order_by(Mo.District.name).all()82    names, baselines, regions, retire, inc_idx = [], [], [], [], []83    for d in districts:84        names.append(d.name)85        baselines.append([d.baseline_shares.get(p, 0.0) / 100.0 for p in settings.parties])86        regions.append(d.region)87        retire.append(0 if d.incumbent_running else 1)88        inc_idx.append(settings.parties.index(d.incumbent_party)89                       if d.incumbent_party in settings.parties else -1)90    return SIM.SimulationInput(91        x_mean=np.zeros(5), P=np.eye(5),92        baseline_national=np.array([baseline_national[p] / 100.0 for p in settings.parties]),93        district_names=names, district_baselines=np.array(baselines),94        district_regions=regions, retirement_flags=np.array(retire),95        incumbent_party_idx=np.array(inc_idx), majority_seats=settings.majority_seats)969798def run_forecast(db: Session, as_of: date | None = None, label: str | None = None,99                 is_backtest: bool = False, n_sims: int | None = None,100                 save: bool = True) -> Mo.ForecastRun | None:101    """Ajuste le modèle v2 (sondages + partielles + fondamentaux + médias),102    simule l'élection cible et persiste un ForecastRun avec la décomposition103    complète de la contribution de chaque couche."""104    as_of = as_of or date.today()105    target = db.query(Mo.Election).filter_by(is_target=True).one()106    days_left = (target.election_date - as_of).days107    profiles = compute_pollster_profiles(db)108    polls = load_polls(db, target.id, as_of=as_of)109110    # --- couche 2 : votes réels des partielles → observations nationales ---111    from .ingest.byelections import pseudo_polls112    bye_polls = pseudo_polls(db, as_of=as_of)113114    trend_a = fit_trend(polls, profiles, as_of)                # sondages seuls115    trend = fit_trend(polls + bye_polls, profiles, as_of) if bye_polls else trend_a116    if trend is None:117        raise RuntimeError("Pas assez de sondages pour ajuster le modèle")118119    # --- volatilité de campagne : attention Wikipédia + pouls social (bornée,120    # variance seulement — v3 §16) ---121    attention: dict = {"available": False, "drift_multiplier": 1.0}122    volatility: dict = {"multiplier": 1.0}123    if not is_backtest:124        try:125            from .ingest.wiki_attention import signal as att_signal126            attention = att_signal(db)127        except Exception:128            pass129        try:130            from .modeling.signals import volatility as VOLA131            volatility = VOLA.compute(db, attention)132        except Exception:133            volatility = {"multiplier": float(attention.get("drift_multiplier", 1.0))}134    drift_mult = float(volatility.get("multiplier", 1.0))135136    # σ_industrie : appris LOEO sur 2007-2022 si le flag l'active (v3.1)137    industry_sd = None138    if settings.use_empirical_industry_error:139        try:140            from .modeling.national.pollster_error import industry_sigma_loeo141            industry_sd = industry_sigma_loeo(None)142        except Exception:143            industry_sd = None144145    now_d = nc(trend, as_of)146    fc_b = fc(trend, as_of, target.election_date,147              drift_multiplier=drift_mult,148              industry_sd=industry_sd)                         # sondages + partielles149150    # --- couche 3 : prior de fondamentaux (poids décroissant vers le scrutin) ---151    result_2022 = (db.query(Mo.Election)152                   .filter(Mo.Election.actual_result.isnot(None))153                   .order_by(Mo.Election.election_date.desc()).first().actual_result)154    fund_diag: dict = {"enabled": settings.fundamentals_enabled}155    x_c, P_c = fc_b.x, fc_b.P156    if settings.fundamentals_enabled:157        from .ingest.firecrawl_watch import latest_satisfaction158        sat, sat_prov = latest_satisfaction(db)159        prior = FUND.compute_prior(as_of, incumbent="CAQ", terms=2,160                                   satisfaction=sat, prev_result=result_2022)161        x_c, P_c, bl = FUND.blend(fc_b.x, fc_b.P, prior, days_left)162        fund_diag.update(prior=prior.detail, blend=bl,163                         satisfaction={"value": sat, **sat_prov})164165    # --- couche 4 : ajustement médias borné ---166    nudge = ENS.media_nudge(db)167    x_d = ENS.apply_nudge(x_c, nudge.get("delta_pp", {}))168    fc_d = distribution_from("forecast", as_of, x_d, P_c)169170    inp = _district_inputs(db, result_2022)171    inp.x_mean, inp.P = fc_d.x, fc_d.P172    sim = SIM.run_simulation(inp, n_sims=n_sims)173174    # --- décomposition : contribution de chaque couche (votes et probabilités) ---175    decomposition = None176    if not is_backtest and trend_a is not None:177        def _probe(x, P, seed):178            inp.x_mean, inp.P = x, P179            s = SIM.run_simulation(inp, n_sims=8000,180                                   rng=np.random.default_rng(seed))181            return {p: s.seats[p]["prob_most"] for p in settings.parties}182        fc_a = fc(trend_a, as_of, target.election_date, drift_multiplier=drift_mult,183                  industry_sd=industry_sd)184        variants = [185            ("sondages", "Sondages seuls (modèle v1)", fc_a.x, fc_a.P),186            ("partielles", "+ votes réels des partielles", fc_b.x, fc_b.P),187            ("fondamentaux", "+ prior de fondamentaux", x_c, P_c),188            ("medias", "+ ajustement médias", x_d, P_c),189        ]190        layers, prev_vote, prev_prob = [], None, None191        for key, lab, x, P in variants:192            d = distribution_from("probe", as_of, x, P)193            vote = {p: d.summary[p]["mean"] for p in settings.parties}194            prob = _probe(x, P, seed=hash(key) % 2 ** 31)195            layers.append({196                "key": key, "label": lab, "vote": vote, "prob_most": prob,197                "delta_vote": ({p: round(vote[p] - prev_vote[p], 2)198                                for p in settings.parties} if prev_vote else None),199                "delta_prob": ({p: round(prob[p] - prev_prob[p], 4)200                                for p in settings.parties} if prev_prob else None)})201            prev_vote, prev_prob = vote, prob202        inp.x_mean, inp.P = fc_d.x, fc_d.P  # restaure l'entrée finale203        decomposition = {"layers": layers, "n_sims_probe": 8000}204205    # --- couche 5 : ensemble avec les marchés prédictifs (probabilités) ---206    market_ens = None207    if not is_backtest:208        try:209            from .ingest.markets import fetch_winner_market210            mk = fetch_winner_market()211            market_ens = ENS.market_blend(212                {p: sim.seats[p]["prob_most"] for p in settings.parties},213                (mk or {}).get("implied_prob_most_seats"))214            if mk:215                market_ens["market_meta"] = {"event": mk.get("event"),216                                             "url": mk.get("url"),217                                             "volume_usd": mk.get("volume_usd"),218                                             "fetched_at": mk.get("fetched_at")}219        except Exception:220            market_ens = None221222    # série lissée compacte pour les graphiques (sous-échantillonnée)223    step = max(1, len(trend.dates) // 400)224    idx = list(range(0, len(trend.dates), step))225    if idx[-1] != len(trend.dates) - 1:226        idx.append(len(trend.dates) - 1)227    series = {228        "dates": [trend.dates[i].isoformat() for i in idx],229        "mean": {p: [round(float(trend.share_mean[i, k] * 100), 2) for i in idx]230                 for k, p in enumerate(settings.parties)},231        "lo": {p: [round(float(trend.share_lo[i, k] * 100), 2) for i in idx]232               for k, p in enumerate(settings.parties)},233        "hi": {p: [round(float(trend.share_hi[i, k] * 100), 2) for i in idx]234               for k, p in enumerate(settings.parties)},235    }236237    # --- circonscriptions pivots (battlegrounds, v3 §24) ---238    battlegrounds = None239    if not is_backtest and settings.enable_tipping_points:240        try:241            inp.x_mean, inp.P = fc_d.x, fc_d.P242            sim_t = SIM.run_simulation(inp, n_sims=settings.tipping_n_sims,243                                       rng=np.random.default_rng(777),244                                       keep_raw=True)245            battlegrounds = SIM.tipping_points(246                sim_t.raw_winners, sim_t.raw_shares, inp.district_names,247                settings.majority_seats, settings.parties)248        except Exception:249            battlegrounds = None250251    # --- distribution prédictive du prochain sondage + surprises (§46-48) ---252    next_poll, poll_surprises = None, []253    if not is_backtest:254        try:255            from .modeling.national.next_poll import predictive, surprise256            next_poll = predictive(now_d.x, now_d.P)257            # surprise des sondages arrivés depuis le run précédent, mesurée258            # contre la prédiction STOCKÉE par ce run-là (jamais rétro-ajustée)259            prev = (db.query(Mo.ForecastRun)260                    .filter(Mo.ForecastRun.is_backtest.is_(False))261                    .order_by(Mo.ForecastRun.run_at.desc()).first())262            prev_pred = ((prev.national.get("beyond") or {}).get("next_poll")263                         if prev else None)264            if prev and prev_pred:265                fresh = (db.query(Mo.Poll)266                         .filter(Mo.Poll.election_id == target.id,267                                 Mo.Poll.accessed_at > prev.run_at,268                                 Mo.Poll.excluded.is_(False)).all())269                for poll in fresh[:6]:270                    shares = {r.party: r.normalized_value for r in poll.results}271                    s = surprise(prev_pred, shares)272                    poll_surprises.append({"pollster": poll.pollster.name,273                                           "field_end": poll.field_end.isoformat(),274                                           **s})275        except Exception:276            next_poll = None277278    beyond = {279        "decomposition": decomposition,280        "web_attention": {k: v for k, v in attention.items() if k != "sparkline_dates"},281        "volatility": volatility,282        "next_poll": next_poll,283        "poll_surprises": poll_surprises,284        "fundamentals": fund_diag,285        "byelections_used": [{"pollster": b["pollster"],286                              "field_end": b["field_end"].isoformat(),287                              "implied_national": b["shares"]} for b in bye_polls],288        "media_adjustment": nudge,289        "market_ensemble": market_ens,290    }291292    run = Mo.ForecastRun(293        as_of=as_of, model_version=settings.model_version, n_polls_used=trend.n_polls,294        n_simulations=sim.n_sims, is_backtest=is_backtest, label=label,295        national={"nowcast": now_d.summary, "forecast": fc_d.summary,296                  "trend_series": series,297                  "x_forecast": [float(v) for v in fc_d.x],298                  "P_forecast": [[float(v) for v in row] for row in fc_d.P],299                  "baseline_national": result_2022,300                  "beyond": beyond},301        seats={"per_party": sim.seats, "summary": sim.seat_matrix_summary,302               "sim_national_vote": sim.national_vote,303               "ensemble": market_ens,304               "battlegrounds": battlegrounds},305        diagnostics={"q_daily_var": trend.q, "loglik": trend.loglik,306                     "days_to_election": days_left,307                     "n_byelections": len(bye_polls),308                     "industry_sd": round(industry_sd, 4) if industry_sd309                     else settings.industry_error_sd,310                     "industry_sd_source": ("loeo-2007-2022" if industry_sd311                                            else "config-manuel"),312                     "fundamentals_weight": (fund_diag.get("blend") or {}).get(313                         "precision_share", 0.0)})314    for d in sim.districts:315        run.district_results.append(Mo.ForecastDistrictResult(316            district_name=d["district"], favorite=d["favorite"], category=d["category"],317            detail=d))318    if save:319        db.add(run)320        db.commit()321    return run322323324def run_pipeline(full_refresh: bool = True) -> dict:325    """Cycle complet. Tolère l'échec individuel des sources."""326    init_db()327    db = SessionLocal()328    report = {}329    try:330        if full_refresh:331            try:332                from .ingest.wikipedia import refresh_all333                report["collect"] = refresh_all()334                _log(db, "collect", "ok", str(report["collect"]))335            except Exception as e:  # la suite continue avec les données en cache336                report["collect"] = f"échec: {e}"337                _log(db, "collect", "error", traceback.format_exc())338            try:339                from .seed import seed_all340                report["seed"] = seed_all(db)341                _log(db, "normalize", "ok", str(report["seed"]))342            except Exception as e:343                report["seed"] = f"échec: {e}"344                _log(db, "normalize", "error", traceback.format_exc())345            try:346                from .ingest.dgeq import refresh_dgeq347                report["dgeq"] = refresh_dgeq(db)348                _log(db, "dgeq", "ok", str(report["dgeq"])[:500])349            except Exception as e:350                report["dgeq"] = f"échec: {e}"351                _log(db, "dgeq", "error", traceback.format_exc())352            try:353                from .ingest.news_rss import ingest_feeds354                report["sentiment"] = ingest_feeds(db)355                _log(db, "sentiment", "ok", str(report["sentiment"]))356            except Exception as e:357                report["sentiment"] = f"échec: {e}"358                _log(db, "sentiment", "error", traceback.format_exc())359            try:360                from .ingest.byelections import ensure_byelections, verify_with_firecrawl361                added = ensure_byelections(db)362                verif = verify_with_firecrawl(db) if settings.firecrawl_api_key else {}363                report["byelections"] = {"ajoutées": added, "vérification": verif}364                _log(db, "byelections", "ok", str(report["byelections"])[:500])365            except Exception as e:366                report["byelections"] = f"échec: {e}"367                _log(db, "byelections", "error", traceback.format_exc())368            try:369                from .ingest.firecrawl_watch import run_watch370                report["veille"] = run_watch(db)371                _log(db, "veille-firecrawl", "ok", str(report["veille"])[:500])372            except Exception as e:373                report["veille"] = f"échec: {e}"374                _log(db, "veille-firecrawl", "error", traceback.format_exc())375            try:376                from .ingest.pollster_reports import run_primary_watch377                report["rapports_primaires"] = run_primary_watch(db)378                _log(db, "rapports-primaires", "ok",379                     str(report["rapports_primaires"])[:500])380            except Exception as e:381                report["rapports_primaires"] = f"échec: {e}"382                _log(db, "rapports-primaires", "error", traceback.format_exc())383            try:384                # source canonique + découverte de comptes : hebdomadaires385                marker = (db.query(Mo.Indicator)386                          .filter_by(name="canonical_checks_last")387                          .order_by(Mo.Indicator.as_of.desc()).first())388                if marker is None or (date.today() - marker.as_of).days >= 7:389                    from .ingest.elections_quebec import run_canonical_checks390                    from .ingest.social_account_discovery import run_discovery391                    report["dgeq_canonique"] = run_canonical_checks(db)392                    report["découverte_comptes"] = run_discovery(db)393                    # validation hebdomadaire AUTOMATIQUE : replay + ablation394                    # régénérés sans intervention (rapports publics à jour)395                    try:396                        from .modeling.validation.historical_replay import run_replay397                        run_replay()398                        _log(db, "replay-hebdo", "ok", "rapport régénéré")399                    except Exception:400                        _log(db, "replay-hebdo", "error", traceback.format_exc())401                    try:402                        from .modeling.validation.ablation import run_ablation403                        run_ablation(db, n_sims=6000)404                        _log(db, "ablation-hebdo", "ok", "rapport régénéré")405                    except Exception:406                        _log(db, "ablation-hebdo", "error", traceback.format_exc())407                    db.add(Mo.Indicator(name="canonical_checks_last",408                                        as_of=date.today(), value=1.0,409                                        method="scheduler"))410                    db.commit()411                    _log(db, "canonique-hebdo", "ok",412                         str({**report.get("dgeq_canonique", {}),413                              **report.get("découverte_comptes", {})})[:500])414            except Exception as e:415                report["canonique"] = f"échec: {e}"416                _log(db, "canonique-hebdo", "error", traceback.format_exc())417            try:418                from .ingest.social_pulse import run_pulse419                report["pouls_social"] = run_pulse(db)420                _log(db, "pouls-social", "ok", str(report["pouls_social"])[:500])421            except Exception as e:422                report["pouls_social"] = f"échec: {e}"423                _log(db, "pouls-social", "error", traceback.format_exc())424            try:425                from .ingest.wiki_attention import refresh as att_refresh426                report["attention"] = att_refresh(db)427                _log(db, "attention-wiki", "ok", str(report["attention"])[:500])428            except Exception as e:429                report["attention"] = f"échec: {e}"430                _log(db, "attention-wiki", "error", traceback.format_exc())431            try:432                from .modeling.signals.event_detection import detect_events433                report["événements"] = detect_events(db)434                _log(db, "event-engine", "ok", str(report["événements"]))435            except Exception as e:436                report["événements"] = f"échec: {e}"437                _log(db, "event-engine", "error", traceback.format_exc())438            try:439                from .modeling.synthetic_poll import run as synth_run440                report["sondage_synthétique"] = synth_run(db)441                _log(db, "sondage-synthetique", "ok",442                     str(report["sondage_synthétique"])[:500])443            except Exception as e:444                report["sondage_synthétique"] = f"échec: {e}"445                _log(db, "sondage-synthetique", "error", traceback.format_exc())446        try:447            run = run_forecast(db)448            report["forecast_run_id"] = run.id449            _log(db, "model+simulate", "ok", f"run {run.id}")450        except Exception as e:451            report["forecast"] = f"échec: {e}"452            _log(db, "model+simulate", "error", traceback.format_exc())453        return report454    finally:455        db.close()456457458if __name__ == "__main__":459    import json460    logging.basicConfig(level=logging.INFO)461    print(json.dumps(run_pipeline(), default=str, indent=2, ensure_ascii=False))462