Aller au contenu principal
Dernière mise à jour le

Prévision des séries chronologiques avec GluonTS

Ouvrir dans Databricks

Exécutez des prévisions probabilistes de séries temporelles avec GluonTS sur AI Runtime. GluonTS est une bibliothèque Python axée sur les approches de deep learning pour la modélisation de séries chronologiques.

GluonTS fournit une boîte à outils pour la prévision et la détection d'anomalies, avec des implémentations pré-construites de modèles de pointe. Il prend en charge les implémentations PyTorch et MXNet et inclut des composants essentiels tels que les architectures de réseau de neurones, le traitement des fonctionnalités et les métriques d’évaluation.

Le notebook couvre :

  • Chargement et préparation des données de consommation électrique
  • Création de répartitions entraînement/test pour les backtests
  • Formation d'un modèle DeepAR pour la prévision
  • Évaluation des prédictions avec des intervalles de confiance
  • Enregistrement et chargement des points de contrôle de modèle

Se connecter au compute GPU serverless​

Cliquez sur le menu déroulant Connecter et sélectionnez Serverless GPU . Ouvrez le panneau latéral Environnement , définissez Accélérateur sur 1xA10 et sélectionnez AI v5 .

Installez GluonTS et les dépendances​

Installez la bibliothèque GluonTS avec le support PyTorch et wget pour download le dataset.

Python
# install gluonts package
%pip install -q --upgrade gluonts[torch] wget
Python
dbutils.library.restartPython()

Configurez le stockage Unity Catalog pour les points de contrôle du modèle​

Configurez les paramètres Unity Catalog pour stocker les points de contrôle du modèle. Le chemin de point de contrôle utilise un volume Unity Catalog pour conserver l'état du modèle pendant l'entraînement.

Python
# You must have `USE CATALOG` privileges on the catalog, and you must have `USE SCHEMA` privileges on the schema.
# If necessary, change the catalog and schema name here.
dbutils.widgets.text("uc_catalog", "main")
dbutils.widgets.text("uc_schema", "default")
dbutils.widgets.text("uc_model_name", "custom_transformer")
dbutils.widgets.text("uc_volume", "checkpoints")

UC_CATALOG = dbutils.widgets.get("uc_catalog")
UC_SCHEMA = dbutils.widgets.get("uc_schema")
UC_VOLUME = dbutils.widgets.get("uc_volume")
MODEL_NAME = dbutils.widgets.get("uc_model_name")
CHECKPOINT_PATH = f"/Volumes/{UC_CATALOG}/{UC_SCHEMA}/{UC_VOLUME}/{MODEL_NAME}"

print(f"UC_CATALOG: {UC_CATALOG}")
print(f"UC_SCHEMA: {UC_SCHEMA}")
print(f"UC_VOLUME: {UC_VOLUME}")
print(f"CHECKPOINT_PATH: {CHECKPOINT_PATH}")
Python
# show the installed gluonts version
%pip show gluonts

Vérifiez la disponibilité et les ressources du GPU​

Vérifiez que le compute GPU est disponible et affichez les spécifications matérielles.

Python
# show the GPU details
!nvidia-smi
Python
import torch
import psutil

# check that GPU is available on the notebook compute
assert torch.cuda.is_available(), 'You need to use GPU compute for this notebook'
# show GPU, GPU RAM, number of CPUs and total RAM
print(f"""
      Number of GPUs available: {torch.cuda.device_count()}
      Total GPU RAM: {torch.cuda.get_device_properties(0).total_memory / (1024 ** 3):.2f} GB
      Number of CPUs: {psutil.cpu_count()}
      Total RAM: {psutil.virtual_memory().total / (1024 ** 3):.2f} GB
      """)

Importer les bibliothèques requises​

Importez les composants GluonTS pour la gestion des datasets, l'entraînement et l'évaluation des modèles, ainsi que les bibliothèques standard de Data Science.

Python
import os
import json
import zipfile
import matplotlib.pyplot as plt
import wget
import tempfile
import numpy as np
import pandas as pd
import matplotlib.colors as mcolors
from itertools import islice

# GluonTS
from gluonts.dataset.pandas import PandasDataset
from gluonts.dataset.split import DateSplitter
from gluonts.dataset.util import to_pandas
from gluonts.evaluation import Evaluator
from gluonts.dataset.field_names import FieldName
from gluonts.torch import DeepAREstimator

from lightning.pytorch.callbacks import ModelCheckpoint

# setup plt environment
plt.rcParams["axes.grid"] = True
plt.rcParams["figure.figsize"] = (20, 3)
colors = list(mcolors.TABLEAU_COLORS)

Charger le dataset de consommation d'électricité​

Ce Notebook utilise le dataset de consommation d'électricité du repository de l'Université de Californie, Irvine. Le dataset contient des relevés de consommation d'électricité de 370 clients entre 2011 et 2014, avec des valeurs enregistrées toutes les 15 minutes en kW.

Configurez l'URL source du dataset et le nom du fichier pour le download.

Python
data_file_name = 'LD2011_2014.txt'
dataset_url = 'https://archive.ics.uci.edu/static/public/321/electricityloaddiagrams20112014.zip'

Le fichier du dataset est d'environ 800 Mo une fois extrait, ce qui dépasse la limite de 500 Mo pour les fichiers de workspace des notebooks Databricks serverless. Le code suivant utilise un répertoire temporaire pour download et extraire les données, puis les charge dans un DataFrame Pandas.

Python
# download and extract data
# the electricity dataset https://archive.ics.uci.edu/ml/datasets/ElectricityLoadDiagrams20112014 from the repository of the University of California, Irvine

with tempfile.TemporaryDirectory() as tmp_dir_name:
  temp_zip = f'{tmp_dir_name}/ts.zip'
  print(f'Downloading data zip file from: {dataset_url}')
  wget.download(dataset_url, out=temp_zip)

  with zipfile.ZipFile(temp_zip, 'r') as zip_ref:
    print(f'Extracting data to: {tmp_dir_name}')
    data_file_path = zip_ref.extract(data_file_name, tmp_dir_name)
    print(f'Zip extracted to: {data_file_path}')

  print('Loading data into Pandas DataFrame')
  df_raw = pd.read_csv(
    data_file_path,
    sep=';',
    index_col=0,
    decimal=',',
    parse_dates=True,
  )

Prévisualisez les données brutes de consommation électrique par intervalles de 15 minutes.

Rééchantillonner les données par intervalles horaires​

Rééchantillonnez les données par intervalles de 15 minutes à 1 heure pour réduire le nombre de points de données et accélérer l’entraînement.

Python
# see the data
df_raw
Python
# resample to 1h intervals to reduce the number of data points
freq = "1h"
div = 4 # 1 hour contain 4x 15 min intervals, you need to  delete the resampled value by 4
data_kw = df_raw.resample(freq).sum() / div

data_kw

Configurez les paramètres de prédiction​

Définissez l'horizon de prédiction sur 7 jours (168 heures) et la plage de dates d'entraînement à l'aide des données de 2014.

Sélectionnez un sous-ensemble de séries temporelles pour un entraînement plus rapide. Définir USE_FULL_DATASET = True pour s'entraîner sur les 370 séries temporelles.

Python
# predict for 7 days
prediction_days = 7
# 24 hours per day
intervals_per_day = 24
prediction_length = prediction_days * intervals_per_day

# take the last year of data for a sample
start_training_date = pd.Timestamp('2014-01-01')
end_dataset_date = pd.Timestamp('2014-12-31')

print(f"Sampling frequency set to {freq}. Generate predictions for {prediction_length} intervals")
Python
USE_FULL_DATASET = False # By default use only a subset of the time series because training of full dataset can take longer time
SAMPLE_SIZE = 10 # set number of samples in the dataset if you don't use the full dataset
MAX_TS_TO_DISPLAY = 10

# get the full dataset or a random sample of SAMPLE_SIZE
# you can change the selection to include specific time series
# ts_sample = data_kw[['item_id1', 'item_id2']]
ts_sample = data_kw if USE_FULL_DATASET else data_kw[np.random.choice(data_kw.columns.to_list(), size=SAMPLE_SIZE, replace=False)]

Convertir les données au format GluonTS​

Convertir le DataFrame Pandas au format GluonTS et visualiser les séries chronologiques. Voir le Quick start de GluonTS pour plus d'exemples.

Python
# convert to GluonTS format, taking only the data between start_training_date and end_dataset_date
ts_dataset = PandasDataset(
    dict(ts_sample[(ts_sample.index > start_training_date) & (ts_sample.index <= end_dataset_date)].astype(np.float32))
)

# visualize time series in the GluonTS dataset
for i, entry in enumerate(islice(ts_dataset, MAX_TS_TO_DISPLAY)):
    to_pandas(entry).plot(label=entry[FieldName.ITEM_ID], color=colors[i % len(colors)])
    plt.legend()
    plt.tight_layout()
    plt.show()

print(f'The GluonTS dataset contains {len(ts_dataset)} individual time series from {start_training_date} to {end_dataset_date}')

Créer une répartition entraînement/test pour le backtesting​

Divisez le dataset en ensembles d'entraînement et de test à l'aide de fenêtres glissantes. Ceci crée 4 fenêtres de test pour le backtesting des performances du modèle.

Python
# set backtest parameters
NUM_WINDOWS = 4 # number of rolling windows for backtest
# distance between windows, set to:
# < prediction_length for overlapping windows
# = prediction length for adjucent windows
# > prediction_length for non overapping and non-adjucent windows
DISTANCE = prediction_length

# set the training-testing split date
end_training_date = pd.Period(end_dataset_date, freq=freq) - NUM_WINDOWS*prediction_length

# split into train and test datasets using GluonTS's DateSplitter
train_ds, test_template = DateSplitter(date=end_training_date).split(ts_dataset)
test_pairs = test_template.generate_instances(
    prediction_length=prediction_length,
    windows=NUM_WINDOWS,
    distance=DISTANCE,
)

print(f"The dataset is splitted in {len(train_ds)} training datasets and {len(test_pairs)} test pairs. Training end is {end_training_date}")

Train a DeepAR model​

Entraînez un estimateur DeepAR, un modèle de réseau de neurones récurrent pour la prévision probabiliste. Consultez Modèles disponibles dans la documentation GluonTS pour d’autres algorithmes.

Configurez les hyperparamètres du modèle DeepAR et les paramètres d'entraînement. Le modèle utilise une longueur de contexte de 4 fois la longueur de prédiction et enregistre les points de contrôle après chaque époque.

Python
NUM_EPOCHS = 10

os.makedirs(CHECKPOINT_PATH, exist_ok=True)

checkpoint_cb = ModelCheckpoint(
    dirpath=CHECKPOINT_PATH,
    filename="deepar-{epoch:02d}-{step}",
    save_top_k=-1,           # keep all checkpoints
    every_n_epochs=1,        # save after every epoch
    save_on_train_epoch_end=True,
)

# set required model hyperparameters. See GluonTS repository for the full list of hyperparameters
model_hyperparameters = {
    "freq":freq,
    "prediction_length":prediction_length,
    "context_length":4*prediction_length,
}

# set required trainer hyperparameters
trainer_hyperparameters = {
    "accelerator":"auto",
    "max_epochs":NUM_EPOCHS,
    "callbacks":[checkpoint_cb]
}

# create a DeepAR estimator
deepar_estimator = DeepAREstimator(
  **model_hyperparameters,
  trainer_kwargs=trainer_hyperparameters,
)

Entraînez le modèle DeepAR sur le dataset d'entraînement. L'entraînement pendant 10 époques prend environ 60 secondes sur un seul GPU.

Python
# Suppress known compatibility warnings
import warnings
warnings.filterwarnings("ignore", message="Using a non-tuple sequence for multidimensional indexing")

torch.set_float32_matmul_precision('high')
Python
# train the network
# the training for 10 epochs takes about 60 second on a single GPU in this notebook
deepar_predictor = deepar_estimator.train(train_ds)

Générer et visualiser des prédictions​

Utilisez le modèle entraîné pour prédire les 7 prochains jours pour chaque série temporelle. Les visualisations montrent des valeurs prédites avec des intervalles de confiance de 90 % et des valeurs de vérité terrain.

Python
# predict
forecasts = deepar_predictor.predict(test_pairs.input, num_samples=20)

# ground truth
labels = [to_pandas(l) for l in test_pairs.label]

# visualize predictions
for i, forecast in enumerate(islice(forecasts, MAX_TS_TO_DISPLAY)):
    plt.plot(labels[i][-NUM_WINDOWS*prediction_length:].to_timestamp())
    forecast.plot(intervals=(0.9,), show_label=True)
    plt.legend([f"Ground truth: {forecast.item_id}", "predicted median", "90% confidence interval"])
    plt.show()

Évaluer les performances du modèle​

Calculez les métriques d'évaluation à l'aide de l'Evaluator de GluonTS. Les métriques incluent le MASE, le RMSE et les pertes quantiles.

Python
# calculate evaluation metrics
evaluator = Evaluator(quantiles=[0.1, 0.5, 0.9])
agg_metrics, item_metrics = evaluator(
    labels,
    deepar_predictor.predict(test_pairs.input, num_samples=20),
    num_series=len(test_pairs),
)

# metrics per time series
item_metrics.display()

# aggregated metrics
print(json.dumps(agg_metrics, indent=2))

Reprendre l'entraînement à partir du point de contrôle​

Chargez un point de contrôle enregistré et poursuivez l’entraînement pour des époques supplémentaires. Ceci montre comment reprendre l’entraînement à partir d’un état de modèle précédemment enregistré.

Configurez le modèle pour qu'il s'entraîne pendant 10 époques supplémentaires à partir du point de contrôle enregistré à l'époque 9.

Python
# set required model hyperparameters. See GluonTS repository for the full list of hyperparameters
model_hyperparameters = {
    "freq": freq,
    "prediction_length": prediction_length,
    "context_length": 4 * prediction_length,
}

# set required trainer hyperparameters
trainer_hyperparameters = {
    "accelerator": "auto",
    "max_epochs": NUM_EPOCHS + 10, # Train for another 10 epochs
    "callbacks": [checkpoint_cb],
}

# create a DeepAR estimator using the model checkpoint
deepar_estimator = DeepAREstimator(
    **model_hyperparameters,
    trainer_kwargs=trainer_hyperparameters,
)

updated_predictor = deepar_estimator.train(
    training_data=train_ds,
    ckpt_path=f"{CHECKPOINT_PATH}/deepar-epoch=09-step=500.ckpt",
)

Étapes suivantes​

Ce notebook a démontré les bases de la prévision de séries chronologiques avec GluonTS sur le compute GPU Serverless de Databricks. Pour en savoir plus :

Exemple de Notebook​

Prévision de séries temporelles avec GluonTS

Ouvrir le notebook dans un nouvel ongletOuvrir dans Databricks
Sur cette page