Déployez une prévision simple pour sortir du notebook

Emballez le traitement et le modèle

Nous avons pu jouer avec nos données et nos modèles dans des notebook jupyter. Cela permet d’explorer et de tester rapidement et facilement. Maintenant que notre partie nettoyage des données est maîtrisée et que nous savons quels paramètres donner à notre modèle SARIMA il est temps de le rendre exploitable.

Nous allons nous concentrer sur deux cas d’usage distincts :

• Comment nettoyer les données, entraîner un modèle puis sauvegarder ce modèle

• Comment prédire avec un modèle sauvegardé

Pour ce faire, nous allons passer par des scripts python qui pourront par la suite être interfacés avec une application web, lancés automatiquement avec une exécution planifiée et même pouvoir être mis sous contrôle avec un monitoring simple.

Commençons par le premier cas d’usage avec ce schéma de principe :

Commençons avec une version très simpliste mais fonctionnelle d’un tel script :

from pathlib import Path

import joblib
import pandas as pd
from prophet import Prophet

# Import des fichiers
input_path = Path("data/epidemioscope_raw_daily.csv")
output_path = Path("outputs/prophet_weekly/prophet_model.joblib")
df = pd.read_csv(input_path)

# Cleaning des dates
dates_raw = df["date_raw"].astype(str).str.strip()

dates = pd.to_datetime(dates_raw, format="%Y-%m-%d", errors="coerce")
missing_dates = dates.isna()
dates.loc[missing_dates] = pd.to_datetime(
    dates_raw[missing_dates], format="%d/%m/%Y", errors="coerce"
)
missing_dates = dates.isna()
dates.loc[missing_dates] = pd.to_datetime(
    dates_raw[missing_dates], format="%Y/%m/%d", errors="coerce"
)

df["date"] = dates

# Cleaning des valeurs
df["ili_consultations"] = pd.to_numeric(
    df["ili_consultations"], errors="coerce")

# Création des indexs
df = df.dropna(subset=["date"]).sort_values("date").set_index("date")

daily = df["ili_consultations"].groupby(level=0).mean().sort_index()
daily = daily.asfreq("D")
daily = daily.interpolate(method="time")

# Resampling en weekly
weekly = daily.resample("W-SUN", label="right", closed="right").sum()

df_prophet = weekly.reset_index().rename(
    columns={"date": "ds", "ili_consultations": "y"})

# Train test split chronologique
split_index = int(len(df_prophet) * 0.8)
df_train = df_prophet.iloc[:split_index].copy()
df_test = df_prophet.iloc[split_index:].copy()

# Création du modèle
model = Prophet(
    yearly_seasonality=True,
    weekly_seasonality=False,
    daily_seasonality=False,
)

# Entraîner du modèle
model.fit(df_train)

# Evaluation sur le jeu de test
future = model.make_future_dataframe(
    periods=len(df_test),
    freq="W-SUN",
    include_history=True,
)
forecast = model.predict(future)
forecast_test = forecast[["ds", "yhat"]].merge(df_test, on="ds", how="inner")
mae = (forecast_test["y"] - forecast_test["yhat"]).abs().mean()

print(f"Taille train: {len(df_train)}")
print(f"Taille test: {len(df_test)}")
print(f"MAE test: {mae:.2f}")

# Sauvegarde dnas joblib
output_path.parent.mkdir(parents=True, exist_ok=True)
joblib.dump(model, output_path)

print(f"Modele sauvegarde dans {output_path}")

Vous pouvez retrouver ci-dessous une première version plus orientée production.

"""Pipeline simple de nettoyage, split, entrainement et sauvegarde Prophet.

Le script suit une sequence volontairement simple :
- chargement des donnees brutes,
- nettoyage inspire du notebook P1C2,
- split chronologique train/test,
- entrainement d'un modele Prophet,
- evaluation rapide sur le test,
- sauvegarde du modele avec joblib.
"""

from __future__ import annotations

import argparse
from pathlib import Path

import joblib
import pandas as pd
from prophet import Prophet


DEFAULT_INPUT_PATH = Path("data/epidemioscope_raw_daily.csv")
DEFAULT_OUTPUT_PATH = Path("outputs/prophet_weekly/prophet_model.joblib")


def load_data(input_path: Path) -> pd.DataFrame:
    """Charge le jeu de donnees brut depuis un CSV.

    Args:
        input_path: Chemin du fichier CSV source.

    Returns:
        pd.DataFrame: Donnees brutes lues depuis le fichier.
    """

    return pd.read_csv(input_path)


def clean_data(df: pd.DataFrame) -> pd.DataFrame:
    """Nettoie les donnees et produit une serie hebdomadaire pour Prophet.

    Le nettoyage reprend la logique du notebook P1C2 :
    - parsing multi-format des dates,
    - suppression des dates invalides,
    - moyenne des doublons journaliers,
    - interpolation temporelle sur frequence quotidienne,
    - aggregation hebdomadaire en fin de semaine.

    Args:
        df: DataFrame brut avec au minimum les colonnes `date_raw`
            et `ili_consultations`.

    Returns:
        pd.DataFrame: DataFrame avec les colonnes `ds` et `y`
        attendu par Prophet.
    """

    dates_raw = df["date_raw"].astype(str).str.strip()

    dates = pd.to_datetime(dates_raw, format="%Y-%m-%d", errors="coerce")
    missing_dates = dates.isna()
    dates.loc[missing_dates] = pd.to_datetime(
        dates_raw[missing_dates], format="%d/%m/%Y", errors="coerce"
    )
    missing_dates = dates.isna()
    dates.loc[missing_dates] = pd.to_datetime(
        dates_raw[missing_dates], format="%Y/%m/%d", errors="coerce"
    )

    df = df.copy()
    df["date"] = dates
    df["ili_consultations"] = pd.to_numeric(
        df["ili_consultations"], errors="coerce")

    df = df.dropna(subset=["date"]).sort_values("date").set_index("date")

    daily = df["ili_consultations"].groupby(level=0).mean().sort_index()
    daily = daily.asfreq("D")
    daily = daily.interpolate(method="time")

    weekly = daily.resample("W-SUN", label="right", closed="right").sum()

    return weekly.reset_index().rename(
        columns={"date": "ds", "ili_consultations": "y"}
    )


def split_data(df: pd.DataFrame) -> tuple[pd.DataFrame, pd.DataFrame]:
    """Decoupe les donnees en train/test selon l'ordre temporel.

    Args:
        df: Serie preparee pour Prophet, deja triee chronologiquement.

    Returns:
        tuple[pd.DataFrame, pd.DataFrame]: Le jeu d'entrainement puis
        le jeu de test.
    """

    split_index = int(len(df) * 0.8)
    df_train = df.iloc[:split_index].copy()
    df_test = df.iloc[split_index:].copy()
    return df_train, df_test


def train_model(df: pd.DataFrame) -> Prophet:
    """Entraine un modele Prophet sur le jeu d'entrainement.

    Args:
        df: Historique d'entrainement avec les colonnes `ds` et `y`.

    Returns:
        Prophet: Modele Prophet entrainé.
    """

    model = Prophet(
        yearly_seasonality=True,
        weekly_seasonality=False,
        daily_seasonality=False,
    )
    model.fit(df)
    return model


def evaluate_model(
    model: Prophet,
    df_test: pd.DataFrame,
) -> float:
    """Evalue le modele sur le segment de test avec une MAE simple.

    Args:
        model: Modele Prophet deja entraine.
        df_test: Jeu de test chronologique avec les colonnes `ds` et `y`.

    Returns:
        float: Erreur absolue moyenne sur le jeu de test.
    """

    future = model.make_future_dataframe(
        periods=len(df_test),
        freq="W-SUN",
        include_history=True,
    )
    forecast = model.predict(future)
    forecast_test = forecast[["ds", "yhat"]].merge(
        df_test, on="ds", how="inner")
    return (forecast_test["y"] - forecast_test["yhat"]).abs().mean()


def save_model(model: Prophet, output_path: Path) -> None:
    """Sauvegarde le modele Prophet dans un fichier joblib.

    Args:
        model: Modele Prophet a serialiser.
        output_path: Chemin du fichier de sortie.
    """

    output_path.parent.mkdir(parents=True, exist_ok=True)
    joblib.dump(model, output_path)


def parse_args() -> argparse.Namespace:
    """Parse les arguments de ligne de commande.

    Returns:
        argparse.Namespace: Arguments resolves pour le pipeline.
    """

    parser = argparse.ArgumentParser(
        description="Pipeline simple: load, clean, train Prophet, save model."
    )
    parser.add_argument("--input", type=Path, default=DEFAULT_INPUT_PATH)
    parser.add_argument("--output", type=Path, default=DEFAULT_OUTPUT_PATH)
    return parser.parse_args()


def main() -> int:
    """Execute le pipeline complet de bout en bout.

    Returns:
        int: Code de retour du script.
    """

    args = parse_args()

    df = load_data(args.input)
    df_clean = clean_data(df)
    df_train, df_test = split_data(df_clean)
    model = train_model(df_train)
    mae = evaluate_model(model, df_test)
    save_model(model, args.output)

    print(f"Taille train: {len(df_train)}")
    print(f"Taille test: {len(df_test)}")
    print(f"MAE test: {mae:.2f}")
    print(f"Modele sauvegarde dans {args.output}")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

Continuons maintenant au second cas d’usage qui suit ce schéma de principe :

flowchart LR
    A[Chargement du modèle] --> B[Prédictions]
    B --> C[Enregistrement des prédictions]
   

from pathlib import Path

import joblib


model_path = Path("outputs/prophet_weekly/prophet_model.joblib")
output_path = Path("outputs/prophet_weekly/predictions_simple.csv")
n_future_weeks = 12


model = joblib.load(model_path)

future = model.make_future_dataframe(
    periods=n_future_weeks,
    freq="W-SUN",
    include_history=False,
)

forecast = model.predict(future)

predictions = forecast[["ds", "yhat", "yhat_lower", "yhat_upper"]].copy()

output_path.parent.mkdir(parents=True, exist_ok=True)
predictions.to_csv(output_path, index=False)

print(predictions)
print(f"Predictions sauvegardees dans {output_path}")

Enfin voici la partie “production-ready” :

"""Script simple de chargement d'un modele Prophet et de prediction.

Le script :
- charge un modele Prophet sauvegarde avec joblib,
- genere un horizon futur hebdomadaire,
- calcule les predictions,
- sauvegarde le resultat dans un CSV.
"""

from __future__ import annotations

import argparse
from pathlib import Path

import joblib
import pandas as pd


DEFAULT_MODEL_PATH = Path("outputs/prophet_weekly/prophet_model.joblib")
DEFAULT_OUTPUT_PATH = Path("outputs/prophet_weekly/predictions_simple.csv")
DEFAULT_FUTURE_WEEKS = 12


def load_model(model_path: Path):
    """Charge un modele Prophet serialise avec joblib.

    Args:
        model_path: Chemin du fichier joblib contenant le modele.

    Returns:
        Any: Modele Prophet recharge depuis le disque.
    """

    return joblib.load(model_path)


def make_future_dataframe(model, n_future_weeks: int) -> pd.DataFrame:
    """Construit les dates futures pour la prediction.

    Args:
        model: Modele Prophet deja entraine.
        n_future_weeks: Nombre de semaines futures a predire.

    Returns:
        pd.DataFrame: DataFrame des dates futures au format Prophet.
    """

    return model.make_future_dataframe(
        periods=n_future_weeks,
        freq="W-SUN",
        include_history=False,
    )


def predict(model, future: pd.DataFrame) -> pd.DataFrame:
    """Genere les predictions Prophet.

    Args:
        model: Modele Prophet deja entraine.
        future: DataFrame des dates futures a predire.

    Returns:
        pd.DataFrame: Predictions completes retournees par Prophet.
    """

    return model.predict(future)


def select_prediction_columns(forecast: pd.DataFrame) -> pd.DataFrame:
    """Selectionne les colonnes utiles de prediction.

    Args:
        forecast: Sortie complete produite par Prophet.

    Returns:
        pd.DataFrame: Sous-ensemble avec date, prediction et intervalle.
    """

    return forecast[["ds", "yhat", "yhat_lower", "yhat_upper"]].copy()


def save_predictions(predictions: pd.DataFrame, output_path: Path) -> None:
    """Sauvegarde les predictions dans un fichier CSV.

    Args:
        predictions: DataFrame des predictions a enregistrer.
        output_path: Chemin du fichier CSV de sortie.
    """

    output_path.parent.mkdir(parents=True, exist_ok=True)
    predictions.to_csv(output_path, index=False)


def parse_args() -> argparse.Namespace:
    """Parse les arguments de ligne de commande.

    Returns:
        argparse.Namespace: Arguments resolves pour le script.
    """

    parser = argparse.ArgumentParser(
        description="Load a Prophet model and generate simple predictions."
    )
    parser.add_argument("--model", type=Path, default=DEFAULT_MODEL_PATH)
    parser.add_argument("--output", type=Path, default=DEFAULT_OUTPUT_PATH)
    parser.add_argument(
        "--weeks",
        type=int,
        default=DEFAULT_FUTURE_WEEKS,
        help="Number of future weeks to predict.",
    )
    return parser.parse_args()


def main() -> int:
    """Execute le script de prediction de bout en bout.

    Returns:
        int: Code de retour du script.
    """

    args = parse_args()

    model = load_model(args.model)
    future = make_future_dataframe(model, args.weeks)
    forecast = predict(model, future)
    predictions = select_prediction_columns(forecast)
    save_predictions(predictions, args.output)

    print(predictions)
    print(f"Predictions sauvegardees dans {args.output}")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

Créez une mini-application Streamlit

Ajoutons maintenant une interface pour le premier cas d’usage :

from pathlib import Path

import pandas as pd
import streamlit as st

from pipeline_simple import clean_data, evaluate_model, load_data, save_model, split_data, train_model


st.set_page_config(page_title="Prophet Simple", layout="wide")

default_input_path = Path("data/epidemioscope_raw_daily.csv")
model_output_path = Path("outputs/prophet_weekly/prophet_model_streamlit_simple.joblib")

st.title("Pipeline")
st.caption("Charger des donnees, nettoyer, entrainer Prophet et sauvegarder le modele.")

uploaded_file = st.file_uploader("Charger un fichier CSV brut", type="csv")

if uploaded_file is not None:
    df_raw = pd.read_csv(uploaded_file)
    source_name = uploaded_file.name
else:
    df_raw = load_data(default_input_path)
    source_name = str(default_input_path)

st.subheader("Donnees brutes")
st.write(f"Source: `{source_name}`")
raw_preview = df_raw.head(10).copy()
raw_preview.columns = raw_preview.columns.astype(str)
raw_preview = raw_preview.astype(object)
st.dataframe(raw_preview, use_container_width=True)

if st.button("Lancer le pipeline", type="primary"):
    with st.spinner("Execution du pipeline..."):
        df_prophet = clean_data(df_raw)
        df_train, df_test = split_data(df_prophet)
        model = train_model(df_train)
        mae = evaluate_model(model, df_test)
        components_future = model.make_future_dataframe(
            periods=52,
            freq="W-SUN",
            include_history=True,
        )
        components_forecast = model.predict(components_future)
        save_model(model, model_output_path)

    st.subheader("Pipeline")
    c1, c2, c3 = st.columns(3)
    c1.metric("Points train", len(df_train))
    c2.metric("Points test", len(df_test))
    c3.metric("MAE test", f"{mae:.2f}")

    st.write(f"Modele sauvegarde: `{model_output_path}`")

    st.subheader("Serie nettoyee envoyee a Prophet")
    c4, c5, c6 = st.columns(3)
    c4.metric("Min", f"{df_prophet['y'].min():.2f}")
    c5.metric("Max", f"{df_prophet['y'].max():.2f}")
    c6.metric("Moyenne", f"{df_prophet['y'].mean():.2f}")

    cleaned_preview = df_prophet.tail(20).copy()
    cleaned_preview["ds"] = cleaned_preview["ds"].astype(str)
    cleaned_preview.columns = cleaned_preview.columns.astype(str)
    cleaned_preview = cleaned_preview.astype(object)
    st.dataframe(cleaned_preview, use_container_width=True)

    st.subheader("Forecast Prophet")
    st.pyplot(model.plot(components_forecast))

    st.subheader("Composantes saisonnieres Prophet")
    st.pyplot(model.plot_components(components_forecast))

Ajoutons une deuxième page pour le second cas d’usage :

from pathlib import Path

import streamlit as st

from predict_simple import (
    load_model,
    make_future_dataframe,
    predict,
    save_predictions,
    select_prediction_columns,
)


st.set_page_config(page_title="Prediction Simple", layout="wide")

default_model_path = Path("outputs/prophet_weekly/prophet_model_streamlit_simple.joblib")
default_predictions_path = Path("outputs/prophet_weekly/predictions_streamlit_simple.csv")

st.title("Prediction Prophet")
st.caption("Charger un modele Prophet sauvegarde, generer des previsions futures et visualiser le resultat.")

model_path_text = st.text_input("Chemin du modele joblib", value=str(default_model_path))
future_weeks = st.slider("Nombre de semaines futures a predire", min_value=1, max_value=52, value=12)

model_path = Path(model_path_text)

if st.button("Lancer la prediction", type="primary"):
    if not model_path.exists():
        st.error(f"Modele introuvable: {model_path}")
    else:
        with st.spinner("Generation des predictions..."):
            model = load_model(model_path)
            future = make_future_dataframe(model, future_weeks)
            forecast = predict(model, future)
            predictions = select_prediction_columns(forecast)
            save_predictions(predictions, default_predictions_path)

        st.success("Predictions generees.")
        st.write(f"Modele charge: `{model_path}`")
        st.write(f"Predictions sauvegardees: `{default_predictions_path}`")

        c1, c2 = st.columns(2)
        c1.metric("Nb semaines futures", future_weeks)
        c2.metric("Nb predictions", len(predictions))

        st.subheader("Forecast Prophet")
        st.pyplot(model.plot(forecast))

        st.subheader("Composantes saisonnieres Prophet")
        st.pyplot(model.plot_components(forecast))

        st.subheader("Table des predictions futures")
        predictions_display = predictions.copy()
        predictions_display["ds"] = predictions_display["ds"].astype(str)
        predictions_display.columns = predictions_display.columns.astype(str)
        predictions_display = predictions_display.astype(object)
        st.dataframe(predictions_display, use_container_width=True)

Automatisez un lancement quotidien

Mise en place d’une exécution planifiée (script, tâche cron ou équivalent) pour recalculer la prévision chaque jour sans intervention manuelle.

Une fois les scriptspipeline.py etpredict.py prêts, l’étape suivante consiste à automatiser leur exécution afin de recalculer régulièrement les prévisions sans action manuelle.

L’idée est simple : le système d’exploitation déclenche chaque jour une commande Python qui recharge le modèle ou relance tout le pipeline, puis écrit les résultats dans le dossieroutputs/.

Sous Linux, le plus simple est souvent d’utiliser cron. Il suffit d’éditer la crontab et d’ajouter un exécution, par exemple tous les jours à 05h00 puis 06h00 :

crontab -e
0 5 * * * /usr/bin/python3 /home/user/project/pipeline.py --input /home/user/project/data/epidemioscope_raw_daily.csv --output /home/user/project/outputs/prophet_weekly/prophet_model.joblib >> /home/user/project/logs/train.log 2>&1
0 6 * * * /usr/bin/python3 /home/user/project/predict.py --model /home/user/project/outputs/prophet_weekly/prophet_model.joblib --output /home/user/project/outputs/prophet_weekly/predictions_daily.csv >> /home/user/project/logs/predict.log 2>&1

0 5 * * * signifie tous les jours à 05:00 0 6 * * * signifie tous les jours à 06:00 ; la sortie standard et les erreurs sont enregistrées dans train.log et predict.log

Sur macOS, cron fonctionne mais vous pouvez aussi utiliser l’outil prévu par Apple nommélaunchd. Pour ceci il suffit de créer un fichierxmlen.predictdans~/Library/LaunchAgents/com.user.prophet.predict.plist

<?xml version="1.0" encoding="UTF-8"?>
<!DOCTYPE plist PUBLIC "-//Apple//DTD PLIST 1.0//EN"
 "http://www.apple.com/DTDs/PropertyList-1.0.dtd">
<plist version="1.0">
  <dict>
    <key>Label</key>
    <string>com.user.prophet.predict</string>

    <key>ProgramArguments</key>
    <array>
      <string>/usr/bin/python3</string>
      <string>/Users/user/project/predict_simple.py</string>
      <string>--model</string>
      <string>/Users/user/project/outputs/prophet_weekly/prophet_model.joblib</string>
      <string>--output</string>
      <string>/Users/user/project/outputs/prophet_weekly/predictions_daily.csv</string>
    </array>

    <key>StartCalendarInterval</key>
    <dict>
      <key>Hour</key>
      <integer>6</integer>
      <key>Minute</key>
      <integer>0</integer>
    </dict>

    <key>StandardOutPath</key>
    <string>/Users/user/project/logs/predict.log</string>
    <key>StandardErrorPath</key>
    <string>/Users/user/project/logs/predict.err</string>
  </dict>
</plist>

Les méthodesloadetunloadpermettent de charger ou de décharger le fichier pour l’utiliser :

launchctl load ~/Library/LaunchAgents/com.user.prophet.predict.plist
launchctl unload ~/Library/LaunchAgents/com.user.prophet.predict.plist

L’avantage (ou pas) du planificateur de tâche est l’utilisation d’une interface graphique pour planifier des tâches. Nous allons nous concentrer sur l’outil en ligne de commande qu’estschtasks:

schtasks /create /sc daily /tn "ProphetPrediction" /tr "python C:\project\predict_simple.py --model C:\project\outputs\prophet_weekly\prophet_model.joblib --output C:\project\outputs\prophet_weekly\predictions_daily.csv" /st 06:00 /f

• /create: crée la tâche ;

• /sc daily: exécution quotidienne ;

• /tn : nom de la tâche ;

• /tr : commande à exécuter ;

• /st06:00 : heure de lancement ;

• /f: force l’écrasement si la tâche existe déjà.

En résumé

• Un modèle exploré dans un notebook n’est pas exploitable tel quel : il doit être encapsulé dans des scripts reproductibles (chargement, nettoyage, entraînement, évaluation, sauvegarde)

• La séparation claire entre pipeline d’entraînement et script de prédiction est essentielle pour éviter les dépendances implicites et faciliter la mise en production.

• La sérialisation du modèle (via joblib) permet de découpler entraînement et usage, et donc d’industrialiser les prévisions.

• Une interface simple (Streamlit) permet de rendre le pipeline accessible sans sacrifier la logique métier sous-jacente.

• L’automatisation (cron, launchd, schtasks) transforme un script en système de production en garantissant une exécution régulière sans intervention humaine.

C’est presque la fin de ce cours, merci de l’avoir suivi ! Il vous reste un quiz pour valider vos acquis. Bonne continuation !