fix(ml,backend,frontend): borne la peremption des predictions et isole les erreurs par flux
This commit is contained in:
Dorian
@@ -119,6 +119,11 @@ def _typer(frame: pd.DataFrame) -> pd.DataFrame:
|
||||
LightGBM refuse ("pandas dtypes must be int, float or bool"). `pd.to_numeric` corrige aussi
|
||||
n'importe quelle autre colonne mesuree entierement absente sur une fenetre de scoring, pas
|
||||
seulement `capacity_kw`.
|
||||
|
||||
Piege additionnel : `NUMERIC_COLUMNS` inclut `consumption_kwh`, la cible du modele, pas
|
||||
seulement des variables explicatives. Une valeur non numerique y devient donc silencieusement
|
||||
`NaN` aussi bien a l'entrainement (ou `train.py` l'exclura ensuite via son `dropna`) qu'au
|
||||
scoring -- ce n'est pas un effet de bord limite aux colonnes mesurees.
|
||||
"""
|
||||
typee = frame.copy()
|
||||
for colonne in NUMERIC_COLUMNS:
|
||||
|
||||
+71
-34
@@ -30,6 +30,13 @@ from enervision_ml.features import TARGET_COLUMN, WEATHER_COLUMNS, build_feature
|
||||
# Marge au-dessus des 168h necessaires au lag hebdomadaire, pour absorber les trous de mesure.
|
||||
LOOKBACK = timedelta(days=21)
|
||||
|
||||
# Au-dela de ce seuil, la derniere lecture d'un site est trop vieille pour que "l'heure
|
||||
# suivante" ait un sens operationnel : ce n'est plus une prevision a un pas, c'est un site dont
|
||||
# l'ingestion s'est probablement arretee. Sans cette borne, `build_scoring_frame` produirait
|
||||
# quand meme un `target_at` (derniere lecture + 1h), et rien en aval (ni l'API, ni le dashboard)
|
||||
# ne distingue une prevision fraiche d'une prevision vieille de plusieurs jours.
|
||||
MAX_STALENESS = timedelta(hours=24)
|
||||
|
||||
TARGET_METRIC = "consumption_kwh"
|
||||
PERIOD_MINUTES = 60
|
||||
LAG_168H_COLUMN = f"{TARGET_COLUMN}_lag_168h"
|
||||
@@ -38,6 +45,14 @@ INSUFFICIENT_DATA_REASON = (
|
||||
)
|
||||
|
||||
|
||||
def _stale_reason(age: pd.Timedelta) -> str:
|
||||
return (
|
||||
f"Dernière lecture vieille de {age.total_seconds() / 3600:.0f}h "
|
||||
f"(seuil {MAX_STALENESS.total_seconds() / 3600:.0f}h) : ingestion probablement "
|
||||
"arrêtée pour ce site."
|
||||
)
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class ScoredSite:
|
||||
site_id: str
|
||||
@@ -86,10 +101,31 @@ def build_scoring_frame(recent: pd.DataFrame, *, site_id: str | None = None) ->
|
||||
return features.groupby("site_id", as_index=False, sort=False).tail(1).reset_index(drop=True)
|
||||
|
||||
|
||||
def score(booster: lgb.Booster, scoring_frame: pd.DataFrame) -> list[ScoredSite]:
|
||||
def score(
|
||||
booster: lgb.Booster, scoring_frame: pd.DataFrame, *, instant: datetime
|
||||
) -> list[ScoredSite]:
|
||||
resultats: list[ScoredSite] = []
|
||||
|
||||
insuffisants = scoring_frame[scoring_frame[LAG_168H_COLUMN].isna()]
|
||||
# `timestamp` de la ligne de scoring vaut derniere lecture + 1h (cf. `build_scoring_frame`) :
|
||||
# on en deduit l'age de cette derniere lecture par rapport a `instant`.
|
||||
travail = scoring_frame.copy()
|
||||
travail["_age"] = instant - (travail["timestamp"] - pd.Timedelta(hours=1))
|
||||
|
||||
perimes = travail[travail["_age"] > MAX_STALENESS]
|
||||
for enregistrement in _records(perimes):
|
||||
resultats.append(
|
||||
ScoredSite(
|
||||
site_id=enregistrement["site_id"],
|
||||
target_at=enregistrement["timestamp"].to_pydatetime(),
|
||||
status="insufficient_data",
|
||||
predicted_value=None,
|
||||
failure_reason=_stale_reason(enregistrement["_age"]),
|
||||
)
|
||||
)
|
||||
|
||||
a_jour = travail[travail["_age"] <= MAX_STALENESS]
|
||||
|
||||
insuffisants = a_jour[a_jour[LAG_168H_COLUMN].isna()]
|
||||
for enregistrement in _records(insuffisants):
|
||||
resultats.append(
|
||||
ScoredSite(
|
||||
@@ -101,7 +137,7 @@ def score(booster: lgb.Booster, scoring_frame: pd.DataFrame) -> list[ScoredSite]
|
||||
)
|
||||
)
|
||||
|
||||
suffisants = scoring_frame[scoring_frame[LAG_168H_COLUMN].notna()]
|
||||
suffisants = a_jour[a_jour[LAG_168H_COLUMN].notna()]
|
||||
if not suffisants.empty:
|
||||
typee = suffisants.copy()
|
||||
typee["site_type"] = typee["site_type"].astype("category")
|
||||
@@ -163,20 +199,23 @@ def write_predictions(
|
||||
connection.execute(_INSERT_PREDICTION, lignes)
|
||||
|
||||
|
||||
def _load_recent(*, csv_path: Path | None, now: datetime | None) -> tuple[pd.DataFrame, datetime]:
|
||||
if csv_path is not None:
|
||||
brute = load_from_csv(csv_path)
|
||||
instant = now or (
|
||||
brute["timestamp"].max().to_pydatetime() if not brute.empty else datetime.now(UTC)
|
||||
)
|
||||
return brute[brute["timestamp"] >= instant - LOOKBACK], instant
|
||||
def _load_recent_from_csv(csv_path: Path, *, now: datetime | None) -> tuple[pd.DataFrame, datetime]:
|
||||
brute = load_from_csv(csv_path)
|
||||
instant = now or (
|
||||
brute["timestamp"].max().to_pydatetime() if not brute.empty else datetime.now(UTC)
|
||||
)
|
||||
return brute[brute["timestamp"] >= instant - LOOKBACK], instant
|
||||
|
||||
instant = now or datetime.now(UTC)
|
||||
engine = create_engine(config.database_url())
|
||||
try:
|
||||
return load_recent_from_database(engine, since=instant - LOOKBACK), instant
|
||||
finally:
|
||||
engine.dispose()
|
||||
|
||||
def _score_frame(
|
||||
recent: pd.DataFrame, *, model_path: Path, site_id: str | None, instant: datetime
|
||||
) -> list[ScoredSite]:
|
||||
scoring_frame = build_scoring_frame(recent, site_id=site_id)
|
||||
if scoring_frame.empty:
|
||||
return []
|
||||
|
||||
booster = lgb.Booster(model_file=str(model_path))
|
||||
return score(booster, scoring_frame, instant=instant)
|
||||
|
||||
|
||||
def run_scoring(
|
||||
@@ -190,29 +229,27 @@ def run_scoring(
|
||||
|
||||
En mode `--csv`, rien n'est ecrit : c'est un instantane historique fige (l'heure "future"
|
||||
calculee n'existe dans aucune base reelle), utile pour valider le pipeline sans base
|
||||
joignable, cf. `ml/README.md`.
|
||||
joignable, cf. `ml/README.md`. `site_id` n'est filtre qu'une fois, dans
|
||||
`build_scoring_frame` : le filtrer aussi ici serait redondant.
|
||||
"""
|
||||
recent, _instant = _load_recent(csv_path=csv_path, now=now)
|
||||
if site_id is not None:
|
||||
recent = recent[recent["site_id"] == site_id]
|
||||
if csv_path is not None:
|
||||
recent, instant = _load_recent_from_csv(csv_path, now=now)
|
||||
return _score_frame(recent, model_path=model_path, site_id=site_id, instant=instant)
|
||||
|
||||
scoring_frame = build_scoring_frame(recent, site_id=site_id)
|
||||
if scoring_frame.empty:
|
||||
return []
|
||||
# Un seul engine pour la lecture et l'ecriture de ce run, plutot qu'un par etape.
|
||||
engine = create_engine(config.database_url())
|
||||
try:
|
||||
instant = now or datetime.now(UTC)
|
||||
recent = load_recent_from_database(engine, since=instant - LOOKBACK)
|
||||
resultats = _score_frame(recent, model_path=model_path, site_id=site_id, instant=instant)
|
||||
|
||||
booster = lgb.Booster(model_file=str(model_path))
|
||||
resultats = score(booster, scoring_frame)
|
||||
|
||||
if csv_path is None:
|
||||
reference = model_reference(model_path)
|
||||
engine = create_engine(config.database_url())
|
||||
try:
|
||||
with engine.begin() as connection:
|
||||
write_predictions(connection, resultats, reference=reference)
|
||||
finally:
|
||||
engine.dispose()
|
||||
with engine.begin() as connection:
|
||||
write_predictions(connection, resultats, reference=reference)
|
||||
|
||||
return resultats
|
||||
return resultats
|
||||
finally:
|
||||
engine.dispose()
|
||||
|
||||
|
||||
def parse_args() -> argparse.Namespace:
|
||||
|
||||
Reference in New Issue
Block a user