GluonTS - feat_dynamic_real example

gluonts_example_feat_dynamic_real.py

from gluonts.dataset.common import ListDataset
from gluonts.model.deepar import DeepAREstimator
from gluonts.evaluation import make_evaluation_predictions, Evaluator
from gluonts.mx import Trainer

from itertools import tee
import pandas as pd
import random
import json

####################################################

# Settings
N_PERIODS = 100
PREDICTION_LENGTH = 5
FREQUENCY = 'M'
NUM_SAMPLES = 10000
EPOCHS = 10

# Set the future dynamic features
# Choose 0 or 1. When dynamic_real == 1, target == 3. When dynamic_real == 0, target == 10
FUTURE_DYNAMIC_REAL = 1

####################################################
print('\n', '*' * 80, '\n', '1. Make Datasets', '\n')

# Make datasets
FEAT_DYNAMIC_REAL = [random.randint(0,1) for x in range(N_PERIODS)] + [FUTURE_DYNAMIC_REAL] * PREDICTION_LENGTH # 100 items past + 5 items future = 105 items
TARGET = [3 if x == 1 else 10 for x in FEAT_DYNAMIC_REAL][:N_PERIODS] # 100 items
START = pd.Timestamp("2000-01-01", freq=FREQUENCY)

ds_train = ListDataset(
    [{
        "item_id": "item_1",
        "start": START,
        "target": TARGET[:N_PERIODS - PREDICTION_LENGTH], # 95 items
        "feat_dynamic_real": [FEAT_DYNAMIC_REAL[:N_PERIODS - PREDICTION_LENGTH]], # 95 items
    }],
    freq=FREQUENCY
)

ds_current = ListDataset(
    [{
        "item_id": "item_1",
        "start": START,
        "target": TARGET[:N_PERIODS], # 100 items
        "feat_dynamic_real": [FEAT_DYNAMIC_REAL[:N_PERIODS]], # 100 items
    }],
    freq=FREQUENCY
)

ds_future = ListDataset(
    [{
        "item_id": "item_1",
        "start": START,
        "target": TARGET[:N_PERIODS], # 100 items
        "feat_dynamic_real": [FEAT_DYNAMIC_REAL[:N_PERIODS + PREDICTION_LENGTH]], # 105 items
    }],
    freq=FREQUENCY
)

# Log datasets
print(f"\n* Dataset metadata: target is {N_PERIODS} long, feat_dynamic_real is {N_PERIODS+PREDICTION_LENGTH} long. Prediction length is {PREDICTION_LENGTH}.")
print('\nds_train\n', ds_train)
print('\nds_current\n', ds_current)
print('\nds_future\n', ds_future)

# Make estimator
estimator = DeepAREstimator(
    freq=FREQUENCY,
    prediction_length=PREDICTION_LENGTH,
    use_feat_dynamic_real=True,
    num_parallel_samples=NUM_SAMPLES,
    trainer=Trainer(
        epochs=EPOCHS,
    )
)

####################################################
print('\n', '*' * 80, '\n', '2a. Train / Test Split', '\n')

# Using train/test split on past data
# No future dynamic features are needed for this

# Train model and make predictions
predictor = estimator.train(ds_train)
forecast_it, ts_it = make_evaluation_predictions(
    dataset=ds_current, 
    predictor=predictor,
    num_samples=NUM_SAMPLES
)

# Make a copy of the iterators
ts_it, targets = tee(ts_it) 
forecast_it, predictions = tee(forecast_it) 

# Calc metrics
evaluator = Evaluator(quantiles=[0.05, 0.5, 0.95])
agg_metrics, item_metrics = evaluator(
    ts_it, 
    forecast_it, 
    num_series=len(ds_current)
)

# Log metrics
print('\nagg_metrics:\n', json.dumps(agg_metrics, indent=4))
print('\nitem_metrics:\n', item_metrics)

# Log predictions
for item_index, item in enumerate(list(predictions)):
    df_target = list(targets)[item_index]
    prediction_actuals = df_target.iloc[-PREDICTION_LENGTH:, 0]
    prediction_averages = [sum(sub_list) / len(sub_list) for sub_list in zip(*item.samples)]
    print(
        f'\n{item.item_id} - Predictions for next {PREDICTION_LENGTH} periods:', 
        json.dumps({str(item.start_date + (item.freq * i)):f'{v:.1f} (should be {prediction_actuals[i]})' for i, v in enumerate(prediction_averages)}, indent=4)
    )

####################################################
print('\n', '*' * 80, '\n', '2b. Train / Predict', '\n')

# Using train/predict on current data
# Need future dynamic features for this

# Train model and make predictions
predictor = estimator.train(ds_current) # train on current data, no future dynamic_feat
predictions = predictor.predict(ds_future) # predict on current data + future dynamic_feat

# Log predictions
for item in list(predictions):
    prediction_averages = [sum(sub_list) / len(sub_list) for sub_list in zip(*item.samples)]
    print(
        f'\n{item.item_id} - Predictions for next {PREDICTION_LENGTH} periods:', 
        json.dumps({str(item.start_date + (item.freq * i)):f'{v:.1f} (should be {3 if FUTURE_DYNAMIC_REAL else 10})' for i, v in enumerate(prediction_averages)}, indent=4)
    )