NeuralForecast

BiTCN (Bidirectional Temporal Convolutional Network) is a parameter-efficient architecture designed for probabilistic forecasting. It uses a 'forward' network to encode future covariates and a 'backward' network to encode past observations and covariates. It is a lightweight alternative to RNNs (LSTM, GRU) and Transformers, requiring significantly fewer parameters and fewer hyperparameters to tune.

Key characteristics:

• Low Space Complexity: Requires orders of magnitude fewer parameters than Transformer-based methods.
• Efficiency: Computationally more efficient than common RNN methods.
• Hyperparameters: Typically requires tuning only two main hyperparameters.

import pandas as pd
import matplotlib.pyplot as plt

from neuralforecast import NeuralForecast
from neuralforecast.losses.pytorch import GMM
from neuralforecast.models import BiTCN
from neuralforecast.utils import AirPassengersPanel, AirPassengersStatic

Y_train_df = AirPassengersPanel[AirPassengersPanel.ds<AirPassengersPanel['ds'].values[-12]] # 132 train
Y_test_df = AirPassengersPanel[AirPassengersPanel.ds>=AirPassengersPanel['ds'].values[-12]].reset_index(drop=True) # 12 test

fcst = NeuralForecast(
    models=[
            BiTCN(h=12,
                input_size=24,
                loss=GMM(n_components=7, level=[80,90]),
                max_steps=100,
                scaler_type='standard',
                futr_exog_list=['y_[lag12]'],
                hist_exog_list=None,
                stat_exog_list=['airline1'],
                windows_batch_size=2048,
                val_check_steps=10,
                early_stop_patience_steps=-1,
                ),
    ],
    freq='ME'
)
fcst.fit(df=Y_train_df, static_df=AirPassengersStatic)
forecasts = fcst.predict(futr_df=Y_test_df)

# Plot quantile predictions
Y_hat_df = forecasts.reset_index(drop=False).drop(columns=['unique_id','ds'])
plot_df = pd.concat([Y_test_df, Y_hat_df], axis=1)
plot_df = pd.concat([Y_train_df, plot_df])

plot_df = plot_df[plot_df.unique_id=='Airline1'].drop('unique_id', axis=1)
plt.plot(plot_df['ds'], plot_df['y'], c='black', label='True')
plt.plot(plot_df['ds'], plot_df['BiTCN-median'], c='blue', label='median')
plt.fill_between(x=plot_df['ds'][-12:], 
                 y1=plot_df['BiTCN-lo-90'][-12:].values,
                 y2=plot_df['BiTCN-hi-90'][-12:].values,
                 alpha=0.4, label='level 90')
plt.legend()
plt.grid()

The GRU (Gated Recurrent Unit) model is a sequential forecasting model that improves upon LSTM and Elman cells by using a simplified gating mechanism and an MLP decoder. It transforms hidden states into contexts that are decoded into time series predictions.

Key features include support for:

• Static exogenous inputs (stat_exog_list)
• Historic exogenous inputs (hist_exog_list)
• Future exogenous inputs (futr_exog_list)
• Probabilistic forecasting via DistributionLoss to provide prediction intervals (e.g., 80% or 90% levels).

import pandas as pd
from neuralforecast import NeuralForecast
from neuralforecast.models import GRU
from neuralforecast.losses.pytorch import DistributionLoss

# Example initialization
fcst = NeuralForecast(
    models=[GRU(h=12, input_size=24,
                loss=DistributionLoss(distribution='Normal', level=[80, 90]),
                scaler_type='robust',
                encoder_n_layers=2,
                encoder_hidden_size=128,
                decoder_hidden_size=128,
                decoder_layers=2,
                max_steps=200,
                futr_exog_list=None,
                hist_exog_list=['y_[lag12]'],
                stat_exog_list=['airline1'],
                )
    ],
    freq='ME'
)

The Hierarchical Mixture Networks (HINT) framework combines neural forecast architectures with mixture probability and hierarchical reconciliation. It uses a TemporalNorm module to improve training robustness and ensures forecast coherence via bootstrap sample reconciliation.

To use HINT, you must provide a base neural model (like NHITS), a summing matrix S representing aggregation constraints, and a reconciliation method.

Key requirements for data:

• The input dataframe Y_df should be sorted by unique_id (lexicographically) and ds to match the order of the summing matrix S_df.
• The summing matrix S is passed as S_df.values to the HINT constructor.

import matplotlib.pyplot as plt
from neuralforecast.losses.pytorch import GMM, sCRPS
from datasetsforecast.hierarchical import HierarchicalData
from neuralforecast import NeuralForecast
from neuralforecast.models import NHITS
from neuralforecast.models.hint import HINT

# 1. Prepare hierarchical data
horizon = 12
Y_df, S_df, tags = HierarchicalData.load('./data', 'TourismLarge')
Y_df['ds'] = pd.to_datetime(Y_df['ds'])

# Ensure Y_df matches S_df order
Y_df.unique_id = Y_df.unique_id.astype('category')
Y_df.unique_id = Y_df.unique_id.cat.set_categories(S_df.index)
Y_df = Y_df.sort_values(by=['unique_id', 'ds'])

level = [80, 90]

# 2. Define the base neural model with a mixture loss
nhits = NHITS(
    h=horizon,
    input_size=24,
    loss=GMM(n_components=10, level=level),
    max_steps=2000,
    early_stop_patience_steps=10,
    val_check_steps=50,
    scaler_type='robust',
    learning_rate=1e-3,
    valid_loss=sCRPS(level=level)
)

# 3. Instantiate HINT with reconciliation
model = HINT(h=horizon, S=S_df.values, model=nhits, reconciliation='BottomUp')

# 4. Fit and Predict using NeuralForecast
nf = NeuralForecast(models=[model], freq='MS')
Y_hat_df = nf.cross_validation(df=Y_df, val_size=12, n_windows=1)

The TFT model is a Temporal Fusion Transformer designed for interpretable multi-horizon forecasting. It combines gating layers, an LSTM recurrent encoder, and multi-head attention layers. It supports static exogenous variables, historic exogenous variables, future exogenous variables, and autoregressive features, all of which can be categorical or continuous. The model uses multi-quantile regression to model conditional probabilities.

import matplotlib.pyplot as plt
import numpy as np
import pandas as pd

from neuralforecast import NeuralForecast
from neuralforecast.models import TFT
from neuralforecast.losses.pytorch import DistributionLoss
from neuralforecast.utils import AirPassengersPanel, AirPassengersStatic

# Prepare data
AirPassengersPanel["month"] = AirPassengersPanel.ds.dt.month
Y_train_df = AirPassengersPanel[AirPassengersPanel.ds < AirPassengersPanel["ds"].values[-12]]
Y_test_df = AirPassengersPanel[AirPassengersPanel.ds >= AirPassengersPanel["ds"].values[-12]].reset_index(drop=True)

# Initialize TFT
nf = NeuralForecast(
    models=[
        TFT(
            h=12,
            input_size=48,
            hidden_size=20,
            grn_activation="ELU",
            rnn_type="lstm",
            n_rnn_layers=1,
            one_rnn_initial_state=False,
            loss=DistributionLoss(distribution="StudentT", level=[80, 90]),
            learning_rate=0.005,
            stat_exog_list=["airline1"],
            futr_exog_list=["y_[lag12]", "month"],
            hist_exog_list=["trend"],
            max_steps=300,
            val_check_steps=10,
            early_stop_patience_steps=10,
            scaler_type="robust",
            windows_batch_size=None,
            enable_progress_bar=True,
        ),
    ],
    freq="ME",
)

# Fit and Predict
nf.fit(df=Y_train_df, static_df=AirPassengersStatic, val_size=12)
Y_hat_df = nf.predict(futr_df=Y_test_df)

When using PyTorch Lightning for distributed training, implement a LightningDataModule to manage the data lifecycle. Follow these lifecycle method patterns:

• prepare_data(): Perform IO, downloads, or data loading. This is called on only one GPU/TPU in distributed settings (useful for shared filesystems).
• setup(stage): Perform data splitting (e.g., train/val/test) and assignments. This is called on every process in Distributed Data Parallel (DDP).
• train_dataloader(), val_dataloader(), test_dataloader(): Return the respective torch.utils.data.DataLoader instances.
• on_exception(exception): Clean up state if the trainer encounters an exception.
• teardown(): Clean up state or delete temporary files after the trainer stops. This is called on every process in DDP.

import lightning.pytorch as L
import torch.utils.data as data
import torch
from pytorch_lightning.demos.boring_classes import RandomDataset

class MyDataModule(L.LightningDataModule):
    def prepare_data(self):
        # download, IO, etc. Useful with shared filesystems
        # only called on 1 GPU/TPU in distributed
        pass

    def setup(self, stage):
        # make assignments here (val/train/test split)
        # called on every process in DDP
        dataset = RandomDataset(1, 100)
        self.train, self.val, self.test = data.random_split(
            dataset, [80, 10, 10], generator=torch.Generator().manual_seed(42)
        )

    def train_dataloader(self):
        return data.DataLoader(self.train)

    def val_dataloader(self):
        return data.DataLoader(self.val)

    def test_dataloader(self):
        return data.DataLoader(self.test)

    def on_exception(self, exception):
        # clean up state after the trainer faced an exception
        pass

    def teardown(self):
        # clean up state after the trainer stops, delete files...
        # called on every process in DDP
        pass