fix(ml): borne la fenetre de scoring a l'instant demande, pour que --now rejoue l'historique

`load_recent_from_database` n'avait qu'une borne basse. `build_scoring_frame` repartait donc de
la derniere lecture de toute la table quel que soit `--now` : `target_at` valait toujours
"fin du jeu + 1h", et `_age = instant - derniere_lecture` devenait negatif, ce qui passait le
seuil de peremption sans rien signaler.

Consequence concrete : sur le jeu historique, arrete au 31/12/2024, aucune boucle de rattrapage
ne pouvait produire une prevision dont le realise existe deja. La surveillance de derive livree
par la migration precedente n'aurait donc rien eu a comparer en demonstration.

`until` est desormais obligatoire sur ce chargeur, ce qui interdit de l'oublier, et le mode CSV
filtre symetriquement. En exploitation rien ne change, aucune lecture n'etant posterieure a
l'heure courante.
This commit is contained in:
Johan LEROY
2026-09-22 14:29:03 +02:00
parent 68239371f6
commit cb961ec2c5
7 changed files with 105 additions and 26 deletions
+7 -5
View File
@@ -102,11 +102,13 @@ Trois contraintes de cohérence sont portées par la base et non par le code app
sont vérifiées depuis le code qui écrit par `ml/tests/test_score_integration.py`, sur une vraie sont vérifiées depuis le code qui écrit par `ml/tests/test_score_integration.py`, sur une vraie
base : un double ne prouverait rien d'une contrainte SQL. base : un double ne prouverait rien d'une contrainte SQL.
**Limite connue de `--now`.** L'option décale l'instant de référence, pas la fenêtre de lecture : **`--now` borne la fenêtre des deux côtés.** `load_recent_from_database` exige un `until` autant
`load_recent_from_database` n'a pas de borne haute et `build_scoring_frame` part toujours de la qu'un `since`, et le scoring lui passe l'instant de référence. Sans cette borne haute,
dernière lecture connue. `target_at` vaut donc « dernière lecture du jeu + 1 h » quelle que soit `build_scoring_frame` repartait de la dernière lecture de toute la table quelle que soit la valeur
la valeur passée, et aucune boucle de rattrapage ne peut fabriquer de paires prévu/réalisé sur un demandée : `target_at` valait toujours « fin du jeu + 1 h », et l'âge de la dernière lecture
jeu figé. devenait négatif sans franchir le seuil de péremption. Rejouer le scoring sur des instants passés
produit désormais des prévisions dont le réalisé existe déjà, ce dont la surveillance de dérive a
besoin pour se démontrer sur un jeu figé.
### `model_reference` est un hachage, pas un nom de fichier ### `model_reference` est un hachage, pas un nom de fichier
@@ -98,9 +98,15 @@ modèle change n'est pas une dérive, c'est une régression de réentraînement.
- La CLI sort en code non nul sous `--fail-on-drift` seulement. Par défaut, constater une dérive - La CLI sort en code non nul sous `--fail-on-drift` seulement. Par défaut, constater une dérive
n'est pas un échec d'exécution. n'est pas un échec d'exécution.
## Limite connue ## Effet de bord assumé sur le pipeline
`enervision_ml.score --now` ne rejoue pas l'historique : `load_recent_from_database` n'a pas de La dérive n'a de matière que si des paires prévu/réalisé existent. Or `enervision_ml.score --now`
borne haute, et `build_scoring_frame` part toujours de la dernière lecture connue. Aucune boucle ne rejouait pas l'historique : `load_recent_from_database` n'avait pas de borne haute et
de rattrapage ne peut donc fabriquer de paires prévu/réalisé sur des données figées, et la `build_scoring_frame` repartait de la dernière lecture connue, si bien que `target_at` valait
dérive répond `indetermine` tant que le scoring n'a pas tourné plusieurs fois en exploitation. toujours « fin du jeu + 1 h » et que l'âge de la dernière lecture devenait négatif sans franchir
le seuil de péremption. Sur le jeu historique, figé au 31/12/2024, aucune boucle de rattrapage
n'aurait donc rien produit de vérifiable.
`until` est devenu obligatoire sur ce chargeur, et le scoring lui passe son instant de référence.
Le comportement en exploitation ne change pas, aucune lecture n'étant postérieure à l'heure
courante ; seul le rattrapage sur données passées devient possible.
+13 -6
View File
@@ -80,7 +80,7 @@ _RECENT_READING_QUERY = text(
s.capacity_kw s.capacity_kw
FROM reading r FROM reading r
JOIN site s ON s.site_id = r.site_id JOIN site s ON s.site_id = r.site_id
WHERE r.timestamp >= :since WHERE r.timestamp >= :since AND r.timestamp <= :until
ORDER BY r.site_id, r.timestamp ORDER BY r.site_id, r.timestamp
""" """
) )
@@ -94,14 +94,21 @@ def load_from_database(connection: Connectable) -> pd.DataFrame:
return _typer(frame[OUTPUT_COLUMNS]) return _typer(frame[OUTPUT_COLUMNS])
def load_recent_from_database(connection: Connectable, *, since: datetime) -> pd.DataFrame: def load_recent_from_database(
"""Lit `reading` + `site` depuis `since` seulement, pour le scoring. connection: Connectable, *, since: datetime, until: datetime
) -> pd.DataFrame:
"""Lit `reading` + `site` sur la fenetre `[since, until]`, pour le scoring.
Piege evite : un `SELECT` sans borne sur l'hypertable complete juste pour scorer le prochain Piege evite cote bas : un `SELECT` sans borne sur l'hypertable complete juste pour scorer le
pas horaire serait la meme erreur que celle corrigee sur `GET /readings` (fenetre non prochain pas horaire serait la meme erreur que celle corrigee sur `GET /readings` (fenetre non
plafonnee sur une table pouvant porter des annees d'historique). plafonnee sur une table pouvant porter des annees d'historique).
Piege evite cote haut : `until` est obligatoire, et c'est ce qui donne son sens a `--now`.
Sans lui, `build_scoring_frame` repartait de la derniere lecture de toute la table quel que
soit l'instant demande, donc `target_at` valait toujours "fin du jeu + 1h" et l'age de la
derniere lecture devenait negatif sans que rien ne le signale.
""" """
frame = pd.read_sql(_RECENT_READING_QUERY, connection, params={"since": since}) frame = pd.read_sql(_RECENT_READING_QUERY, connection, params={"since": since, "until": until})
return _typer(frame[OUTPUT_COLUMNS]) return _typer(frame[OUTPUT_COLUMNS])
+3 -2
View File
@@ -204,7 +204,8 @@ def _load_recent_from_csv(csv_path: Path, *, now: datetime | None) -> tuple[pd.D
instant = now or ( instant = now or (
brute["timestamp"].max().to_pydatetime() if not brute.empty else datetime.now(UTC) brute["timestamp"].max().to_pydatetime() if not brute.empty else datetime.now(UTC)
) )
return brute[brute["timestamp"] >= instant - LOOKBACK], instant fenetre = (brute["timestamp"] >= instant - LOOKBACK) & (brute["timestamp"] <= instant)
return brute[fenetre], instant
def _score_frame( def _score_frame(
@@ -240,7 +241,7 @@ def run_scoring(
engine = create_engine(config.database_url()) engine = create_engine(config.database_url())
try: try:
instant = now or datetime.now(UTC) instant = now or datetime.now(UTC)
recent = load_recent_from_database(engine, since=instant - LOOKBACK) recent = load_recent_from_database(engine, since=instant - LOOKBACK, until=instant)
resultats = _score_frame(recent, model_path=model_path, site_id=site_id, instant=instant) resultats = _score_frame(recent, model_path=model_path, site_id=site_id, instant=instant)
reference = model_reference(model_path) reference = model_reference(model_path)
+37 -8
View File
@@ -45,7 +45,9 @@ def test_load_recent_from_database_excludes_readings_before_the_since_bound(
site_id = insere_site(connexion_ml) site_id = insere_site(connexion_ml)
insere_lectures(connexion_ml, site_id, heures=5, fin=ANCRAGE) insere_lectures(connexion_ml, site_id, heures=5, fin=ANCRAGE)
frame = load_recent_from_database(connexion_ml, since=ANCRAGE - timedelta(hours=2)) frame = load_recent_from_database(
connexion_ml, since=ANCRAGE - timedelta(hours=2), until=ANCRAGE
)
assert list(frame["timestamp"]) == [ assert list(frame["timestamp"]) == [
ANCRAGE - timedelta(hours=2), ANCRAGE - timedelta(hours=2),
@@ -60,7 +62,9 @@ def test_load_recent_from_database_includes_a_reading_exactly_at_the_since_bound
site_id = insere_site(connexion_ml) site_id = insere_site(connexion_ml)
insere_lecture(connexion_ml, site_id, instant=ANCRAGE) insere_lecture(connexion_ml, site_id, instant=ANCRAGE)
frame = load_recent_from_database(connexion_ml, since=ANCRAGE) frame = load_recent_from_database(
connexion_ml, since=ANCRAGE, until=ANCRAGE + timedelta(hours=3)
)
assert len(frame) == 1 assert len(frame) == 1
@@ -71,7 +75,9 @@ def test_load_recent_from_database_keeps_timestamps_timezone_aware(
site_id = insere_site(connexion_ml) site_id = insere_site(connexion_ml)
insere_lecture(connexion_ml, site_id, instant=ANCRAGE) insere_lecture(connexion_ml, site_id, instant=ANCRAGE)
frame = load_recent_from_database(connexion_ml, since=ANCRAGE) frame = load_recent_from_database(
connexion_ml, since=ANCRAGE, until=ANCRAGE + timedelta(hours=3)
)
assert frame["timestamp"].dt.tz is not None assert frame["timestamp"].dt.tz is not None
@@ -83,7 +89,9 @@ def test_load_recent_from_database_orders_readings_by_site_then_timestamp(
for decalage in (2, 0, 1): for decalage in (2, 0, 1):
insere_lecture(connexion_ml, site_id, instant=ANCRAGE + timedelta(hours=decalage)) insere_lecture(connexion_ml, site_id, instant=ANCRAGE + timedelta(hours=decalage))
frame = load_recent_from_database(connexion_ml, since=ANCRAGE) frame = load_recent_from_database(
connexion_ml, since=ANCRAGE, until=ANCRAGE + timedelta(hours=3)
)
assert list(frame["timestamp"]) == [ assert list(frame["timestamp"]) == [
ANCRAGE, ANCRAGE,
@@ -95,7 +103,9 @@ def test_load_recent_from_database_orders_readings_by_site_then_timestamp(
def test_load_recent_from_database_returns_the_contract_columns_even_without_any_row( def test_load_recent_from_database_returns_the_contract_columns_even_without_any_row(
connexion_ml: Connection, connexion_ml: Connection,
) -> None: ) -> None:
frame = load_recent_from_database(connexion_ml, since=ANCRAGE + timedelta(days=365)) frame = load_recent_from_database(
connexion_ml, since=ANCRAGE + timedelta(days=365), until=ANCRAGE + timedelta(days=400)
)
assert frame.empty assert frame.empty
assert list(frame.columns) == OUTPUT_COLUMNS assert list(frame.columns) == OUTPUT_COLUMNS
@@ -107,7 +117,9 @@ def test_load_recent_from_database_types_a_fully_null_capacity_kw_as_float64(
site_id = insere_site(connexion_ml, capacity_kw=None) site_id = insere_site(connexion_ml, capacity_kw=None)
insere_lectures(connexion_ml, site_id, heures=3, fin=ANCRAGE) insere_lectures(connexion_ml, site_id, heures=3, fin=ANCRAGE)
frame = load_recent_from_database(connexion_ml, since=ANCRAGE - timedelta(hours=2)) frame = load_recent_from_database(
connexion_ml, since=ANCRAGE - timedelta(hours=2), until=ANCRAGE
)
assert frame["capacity_kw"].dtype == "float64" assert frame["capacity_kw"].dtype == "float64"
assert frame["capacity_kw"].isna().all() assert frame["capacity_kw"].isna().all()
@@ -122,7 +134,9 @@ def test_load_recent_from_database_types_a_null_is_working_hours_as_float64(
connexion_ml, site_id, instant=ANCRAGE + timedelta(hours=1), is_working_hours=True connexion_ml, site_id, instant=ANCRAGE + timedelta(hours=1), is_working_hours=True
) )
frame = load_recent_from_database(connexion_ml, since=ANCRAGE) frame = load_recent_from_database(
connexion_ml, since=ANCRAGE, until=ANCRAGE + timedelta(hours=3)
)
assert frame["is_working_hours"].dtype == "float64" assert frame["is_working_hours"].dtype == "float64"
assert list(frame["is_working_hours"].isna()) == [True, False] assert list(frame["is_working_hours"].isna()) == [True, False]
@@ -147,8 +161,23 @@ def test_both_loaders_produce_the_same_columns_in_the_same_order(
} }
).to_csv(csv_path, index=False) ).to_csv(csv_path, index=False)
depuis_la_base = load_recent_from_database(connexion_ml, since=ANCRAGE - timedelta(hours=1)) depuis_la_base = load_recent_from_database(
connexion_ml, since=ANCRAGE - timedelta(hours=1), until=ANCRAGE
)
depuis_le_csv = load_from_csv(csv_path) depuis_le_csv = load_from_csv(csv_path)
assert list(depuis_la_base.columns) == list(depuis_le_csv.columns) assert list(depuis_la_base.columns) == list(depuis_le_csv.columns)
assert depuis_la_base.dtypes.to_dict() == depuis_le_csv.dtypes.to_dict() assert depuis_la_base.dtypes.to_dict() == depuis_le_csv.dtypes.to_dict()
def test_load_recent_from_database_excludes_readings_after_the_until_bound(
connexion_ml: Connection,
) -> None:
site_id = insere_site(connexion_ml)
insere_lectures(connexion_ml, site_id, heures=5, fin=ANCRAGE + timedelta(hours=4))
frame = load_recent_from_database(
connexion_ml, since=ANCRAGE - timedelta(days=1), until=ANCRAGE
)
assert list(frame["timestamp"]) == [ANCRAGE]
+21
View File
@@ -275,3 +275,24 @@ def test_run_scoring_in_csv_mode_scores_without_touching_a_database(tmp_path: Pa
assert {r.site_id for r in resultats} == {"site-a", "site-b"} assert {r.site_id for r in resultats} == {"site-a", "site-b"}
assert all(r.status == "available" for r in resultats) assert all(r.status == "available" for r in resultats)
assert all(r.predicted_value == 7.0 for r in resultats) assert all(r.predicted_value == 7.0 for r in resultats)
def test_run_scoring_in_csv_mode_targets_the_hour_after_the_reference_instant(
tmp_path: Path,
) -> None:
depart = datetime(2026, 1, 1, tzinfo=UTC)
frame = make_recent("site-a", heures=400, depart=depart)
csv_path = tmp_path / "recent.csv"
frame.to_csv(csv_path, index=False)
model_path = tmp_path / "model.txt"
model_path.write_bytes(b"peu importe le contenu pour ce test")
rattrapage = depart + timedelta(hours=300)
with pytest.MonkeyPatch.context() as monkeypatch:
monkeypatch.setattr(
"enervision_ml.score.lgb.Booster", lambda model_file: FakeBooster(valeur=7.0)
)
resultats = run_scoring(model_path=model_path, csv_path=csv_path, now=rattrapage)
assert [r.target_at for r in resultats] == [rattrapage + timedelta(hours=1)]
+13
View File
@@ -223,3 +223,16 @@ def test_run_scoring_appends_a_second_row_when_it_runs_twice(
ecrites = parc.predictions_ecrites(site_id) ecrites = parc.predictions_ecrites(site_id)
assert len(ecrites) == 2 assert len(ecrites) == 2
assert ecrites[0].target_at == ecrites[1].target_at assert ecrites[0].target_at == ecrites[1].target_at
def test_run_scoring_targets_the_hour_after_the_reference_instant(
parc: Parc, modele_jetable: Path
) -> None:
site_id = parc.site()
parc.lectures(site_id, heures=200, fin=ANCRAGE + timedelta(hours=48))
rattrapage = ANCRAGE
run_scoring(model_path=modele_jetable, now=rattrapage)
ligne = parc.predictions_ecrites(site_id)[0]
assert ligne.target_at == rattrapage + timedelta(hours=1)