feat(ml): initialise le pipeline d'entrainement LightGBM (ADR 0005)

ml/enervision_ml/train.py

@@ -0,0 +1,245 @@
+"""Entrainement du modele LightGBM de prevision de consommation energetique.
+
+CLI autonome, sur le meme gabarit que `apps/backend/app/etl/historical_import.py`
+(argparse, connexion directe a la base). Cf. `docs/ML-START.md`, section 1.
+
+    uv run python -m enervision_ml.train --csv ../ml/data/all_sites_combined.csv
+    uv run python -m enervision_ml.train  # lit ML_DATABASE_URL
+
+Le modele entraine est ecrit en fichier (`Booster.save_model()`) et suivi par MLflow (parametres,
+metriques, artefact). La base ne stocke jamais le modele lui-meme, seulement une reference vers
+lui (`prediction.model_reference`, pose par le futur service de scoring - hors perimetre ici).
+"""
+
+import argparse
+from pathlib import Path
+from typing import Any
+
+import lightgbm as lgb
+import mlflow
+import mlflow.lightgbm
+import pandas as pd
+from sqlalchemy import create_engine
+
+from enervision_ml import config
+from enervision_ml.baseline import seasonal_persistence_predictions
+from enervision_ml.data import load_from_csv, load_from_database
+from enervision_ml.features import TARGET_COLUMN, build_features, feature_columns
+from enervision_ml.metrics import regression_metrics
+
+CATEGORICAL_FEATURES = ["site_type"]
+
+LIGHTGBM_PARAMS: dict[str, Any] = {
+    "objective": "regression",
+    "metric": "mae",
+    "learning_rate": 0.05,
+    "num_leaves": 63,
+    "min_data_in_leaf": 50,
+    "feature_fraction": 0.8,
+    "bagging_fraction": 0.8,
+    "bagging_freq": 1,
+    "verbosity": -1,
+}
+
+NUM_BOOST_ROUND = 1000
+EARLY_STOPPING_ROUNDS = 50
+DEFAULT_TEST_FRACTION = 0.15
+
+
+def load_raw_frame(csv_path: Path | None) -> pd.DataFrame:
+    """Lit les lectures brutes, depuis le CSV de demarrage ou depuis PostgreSQL."""
+    if csv_path is not None:
+        return load_from_csv(csv_path)
+
+    engine = create_engine(config.database_url())
+    try:
+        return load_from_database(engine)
+    finally:
+        engine.dispose()
+
+
+def chronological_split(
+    features: pd.DataFrame, test_fraction: float
+) -> tuple[pd.DataFrame, pd.DataFrame]:
+    """Coupe par date de coupure, jamais par tirage aleatoire de lignes.
+
+    Une coupure aleatoire laisserait des lignes d'apres la coupure "voir" des lignes d'avant via
+    leurs lags/moyennes glissantes, une fuite qui masquerait un surapprentissage a l'evaluation.
+    """
+    coupure = features["timestamp"].quantile(1 - test_fraction)
+    entrainement = features[features["timestamp"] < coupure]
+    validation = features[features["timestamp"] >= coupure]
+    return entrainement, validation
+
+
+def prepare_dataset(frame: pd.DataFrame, columns: list[str]) -> tuple[pd.DataFrame, pd.Series]:
+    typee = frame.copy()
+    typee["site_type"] = typee["site_type"].astype("category")
+    return typee[columns], typee[TARGET_COLUMN]
+
+
+def train(
+    *,
+    csv_path: Path | None,
+    model_output: Path,
+    test_fraction: float = DEFAULT_TEST_FRACTION,
+    tracking_uri: str | None = None,
+) -> tuple[dict[str, float], dict[str, float]]:
+    """Execute le pipeline complet et rend (metriques du modele, metriques de la baseline)."""
+    raw = load_raw_frame(csv_path)
+    features = build_features(raw)
+    columns = feature_columns()
+
+    # Les premieres 168h par site n'ont pas de lag hebdomadaire complet : ni entrainables, ni
+    # comparables a la baseline saisonniere qui en depend.
+    utilisable = features.dropna(subset=[TARGET_COLUMN, f"{TARGET_COLUMN}_lag_168h"])
+
+    entrainement, validation = chronological_split(utilisable, test_fraction)
+    if entrainement.empty or validation.empty:
+        raise ValueError(
+            "Fenetre d'entrainement ou de validation vide : jeu de donnees trop court pour "
+            f"test_fraction={test_fraction}."
+        )
+
+    X_train, y_train = prepare_dataset(entrainement, columns)
+    X_valid, y_valid = prepare_dataset(validation, columns)
+
+    train_set = lgb.Dataset(
+        X_train,
+        label=y_train,
+        categorical_feature=CATEGORICAL_FEATURES,
+        free_raw_data=False,
+    )
+    valid_set = lgb.Dataset(
+        X_valid,
+        label=y_valid,
+        reference=train_set,
+        categorical_feature=CATEGORICAL_FEATURES,
+        free_raw_data=False,
+    )
+
+    booster = lgb.train(
+        LIGHTGBM_PARAMS,
+        train_set,
+        num_boost_round=NUM_BOOST_ROUND,
+        valid_sets=[valid_set],
+        callbacks=[
+            lgb.early_stopping(EARLY_STOPPING_ROUNDS, verbose=False),
+            lgb.log_evaluation(period=0),
+        ],
+    )
+
+    predictions = pd.Series(
+        booster.predict(X_valid, num_iteration=booster.best_iteration),
+        index=X_valid.index,
+    )
+    model_metrics = regression_metrics(y_valid, predictions)
+    baseline_metrics = regression_metrics(y_valid, seasonal_persistence_predictions(validation))
+
+    model_output.parent.mkdir(parents=True, exist_ok=True)
+    booster.save_model(str(model_output))
+
+    _log_to_mlflow(
+        tracking_uri=tracking_uri,
+        booster=booster,
+        model_metrics=model_metrics,
+        baseline_metrics=baseline_metrics,
+        n_train=len(X_train),
+        n_valid=len(X_valid),
+        test_fraction=test_fraction,
+        model_output=model_output,
+    )
+
+    return model_metrics, baseline_metrics
+
+
+def _log_to_mlflow(
+    *,
+    tracking_uri: str | None,
+    booster: lgb.Booster,
+    model_metrics: dict[str, float],
+    baseline_metrics: dict[str, float],
+    n_train: int,
+    n_valid: int,
+    test_fraction: float,
+    model_output: Path,
+) -> None:
+    uri = tracking_uri or config.mlflow_tracking_uri()
+    if uri is not None:
+        mlflow.set_tracking_uri(uri)
+    mlflow.set_experiment(config.MLFLOW_EXPERIMENT_NAME)
+
+    with mlflow.start_run():
+        mlflow.log_params(
+            {
+                **LIGHTGBM_PARAMS,
+                "num_boost_round": booster.best_iteration or NUM_BOOST_ROUND,
+                "test_fraction": test_fraction,
+                "n_train": n_train,
+                "n_valid": n_valid,
+            }
+        )
+        mlflow.log_metrics({f"model_{cle}": valeur for cle, valeur in model_metrics.items()})
+        mlflow.log_metrics({f"baseline_{cle}": valeur for cle, valeur in baseline_metrics.items()})
+        mlflow.lightgbm.log_model(booster, name="model")
+        mlflow.log_artifact(str(model_output))
+
+
+def parse_args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser(description="Entrainement du modele LightGBM EnerVision")
+
+    parser.add_argument(
+        "--csv",
+        type=Path,
+        default=None,
+        help=(
+            "Chemin vers le CSV historique (chemin de demarrage). Omis, lit ML_DATABASE_URL "
+            "et se connecte directement a PostgreSQL (reading + site)."
+        ),
+    )
+    parser.add_argument(
+        "--model-output",
+        type=Path,
+        default=Path("models/lightgbm-consumption.txt"),
+        help="Chemin d'ecriture du modele entraine. Defaut : models/lightgbm-consumption.txt.",
+    )
+    parser.add_argument(
+        "--test-fraction",
+        type=float,
+        default=DEFAULT_TEST_FRACTION,
+        help=(
+            "Part la plus recente de l'historique reservee a la validation. "
+            f"Defaut : {DEFAULT_TEST_FRACTION}."
+        ),
+    )
+    parser.add_argument(
+        "--mlflow-tracking-uri",
+        default=None,
+        help="Surcharge MLFLOW_TRACKING_URI. Omis, magasin SQLite local (./mlflow.db).",
+    )
+
+    return parser.parse_args()
+
+
+def main() -> None:
+    args = parse_args()
+
+    model_metrics, baseline_metrics = train(
+        csv_path=args.csv,
+        model_output=args.model_output,
+        test_fraction=args.test_fraction,
+        tracking_uri=args.mlflow_tracking_uri,
+    )
+
+    print("Modele LightGBM :", model_metrics)
+    print("Baseline saisonniere (t-168h) :", baseline_metrics)
+
+    if model_metrics["mae"] < baseline_metrics["mae"]:
+        gain = (1 - model_metrics["mae"] / baseline_metrics["mae"]) * 100
+        print(f"LightGBM bat la baseline de {gain:.1f}% de MAE.")
+    else:
+        print("LightGBM ne bat pas la baseline saisonniere sur ce decoupage.")
+
+
+if __name__ == "__main__":
+    main()