Previsión de series temporales con GluonTS

Abrir la versión 'notebook' de esta página

Ejecute la previsión de series temporales probabilísticas con GluonTS en tiempo de ejecución de IA. GluonTS es una biblioteca de Python centrada en enfoques basados en aprendizaje profundo para el modelado de series temporales.

GluonTS proporciona un kit de herramientas para la previsión y detección de anomalías, con implementaciones pregeneradas de modelos de última generación. Admite implementaciones de PyTorch y MXNet e incluye componentes esenciales, como arquitecturas de red neuronal, procesamiento de características y métricas de evaluación.

En el cuaderno se describe lo siguiente:

  • Carga y preparación de los datos de consumo de electricidad
  • Creación de divisiones de entrenamiento y prueba para pruebas inversas
  • Entrenamiento de un modelo DeepAR para la previsión
  • Evaluación de predicciones con intervalos de confianza
  • Guardar y cargar puntos de comprobación del modelo

Nota

Este ejemplo requiere el entorno de IA Databricks versión 5 o superior.

Conectar al cómputo de GPU sin servidor

Haga clic en la lista desplegable Conectar y seleccione GPU sin servidor. Abra el panel lateral Entorno , establezca Acelerador en 1xA10 y seleccione AI v5.

Instalación de GluonTS y dependencias

Instale la biblioteca GluonTS con compatibilidad con PyTorch y wget para descargar el conjunto de datos.

# install gluonts package
%pip install -q "gluonts[torch]==0.16.2" wget

dbutils.library.restartPython()

Configuración del almacenamiento del catálogo de Unity para los puntos de control del modelo

Configure los parámetros del catálogo de Unity para almacenar los puntos de control del modelo. La ruta del punto de control usa un volumen del Catálogo Unity para conservar el estado del modelo durante el entrenamiento.

# 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}")

# show the installed gluonts version
%pip show gluonts

Comprobación de la disponibilidad y los recursos de GPU

Compruebe que el proceso de GPU está disponible y muestre las especificaciones de hardware.

# show the GPU details
!nvidia-smi

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
      """)

Importación de bibliotecas necesarias

Importe los componentes de GluonTS para el control de conjuntos de datos, el entrenamiento del modelo y la evaluación, junto con las bibliotecas de ciencia de datos estándar.

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)

Carga del conjunto de datos de consumo de electricidad

En este cuaderno se usa el conjunto de datos de consumo de electricidad del repositorio de la Universidad de California, Irvine. El conjunto de datos contiene lecturas de consumo de electricidad de 370 clientes entre 2011 y 2014, con valores registrados cada 15 minutos en kW.

Configure la dirección URL de origen del conjunto de datos y el nombre de archivo para la descarga.

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

El archivo de conjunto de datos es de aproximadamente 800 MB cuando se extrae, lo que supera el límite de archivos de área de trabajo de 500 MB para cuadernos sin servidor de Databricks. El código siguiente usa un directorio temporal para descargar y extraer los datos y, a continuación, lo carga en un DataFrame de Pandas.

# 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,
  )

Obtenga una vista previa de los datos de consumo de electricidad sin procesar con intervalos de 15 minutos.

Re-muestrear datos en intervalos de una hora

Vuelva a muestrear los datos de intervalos de 15 minutos a intervalos de 1 hora para reducir el número de puntos de datos y acelerar el entrenamiento.

# see the data
df_raw

# 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

Configuración de parámetros de predicción

Establezca el horizonte de predicción en 7 días (168 horas) y defina el intervalo de fechas de entrenamiento con datos de 2014.

Seleccione un subconjunto de series temporales para un entrenamiento más rápido. Establezca USE_FULL_DATASET = True para entrenar con todas las 370 series temporales.

# 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")

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 datos en formato GluonTS

Convierta el DataFrame de Pandas en formato GluonTS y visualice la serie temporal. Consulte el inicio rápido de GluonTS para obtener más ejemplos.

# 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}')

Crear una división de entrenamiento/prueba para backtesting

Divida el conjunto de datos en conjuntos de entrenamiento y pruebas mediante ventanas graduales. Esto crea 4 ventanas de test para evaluar el rendimiento del modelo al realizar backtesting.

# 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}")

Entrenamiento de un modelo DeepAR

Entrene un estimador deepAR, un modelo de red neuronal recurrente para la previsión probabilística. Consulte Modelos disponibles en la documentación de GluonTS para ver otros algoritmos.

Configure los hiperparámetros y los parámetros de entrenamiento del modelo DeepAR. El modelo usa una longitud de contexto de 4 veces la longitud de predicción y guarda los puntos de control después de cada época.

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,
)

Entrene el modelo DeepAR en el conjunto de datos de entrenamiento. El entrenamiento durante 10 épocas tarda aproximadamente 60 segundos en una sola GPU.

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

torch.set_float32_matmul_precision('high')

# 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)

Generación y visualización de predicciones

Use el modelo entrenado para predecir los próximos 7 días para cada serie temporal. Las visualizaciones muestran los valores predichos con intervalos de confianza del 90 % y los valores reales.

# 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()

Evaluación del rendimiento del modelo

Calcule las métricas de evaluación mediante el evaluador de GluonTS. Las métricas incluyen MASE, RMSE y pérdidas de cuantiles.

# 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))

Reanudación del entrenamiento desde el punto de control

Cargue un punto de comprobación guardado y continúe el entrenamiento durante épocas adicionales. Esto muestra cómo reanudar el entrenamiento a partir de un estado de modelo guardado anteriormente.

Configure el modelo para entrenarlo durante 10 épocas adicionales a partir del punto de comprobación guardado en la época 9.

# 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",
)

Pasos siguientes

En este notebook se demuestran los conceptos básicos de la previsión de series temporales con GluonTS en el cómputo de GPU sin servidor de Databricks. Para obtener más información:

Cuaderno de ejemplo

Previsión de series temporales con GluonTS

Obtención del cuaderno

  • Last updated on