香草 Transformer,遵循 Informer 论文的实现,用作基准模型。

该架构具有三个显著特点: - 时间和内存复杂度为 O(L^2) 的全注意力机制。 - 遵循 Vaswani 等人 (2017) 提出的带有多头注意力机制的经典编码器-解码器。 - 一个 MLP 多步解码器,它通过单次前向操作预测长时序序列,而不是逐步预测。

Vanilla Transformer 模型采用三组件方法来定义其嵌入: - 它使用从卷积网络获得的编码自回归特征。 - 它使用从谐波函数导出的窗口相对位置嵌入。 - 它使用从日历特征获得的绝对位置嵌入。

参考文献
- Haoyi Zhou, Shanghang Zhang, Jieqi Peng, Shuai Zhang, Jianxin Li, Hui Xiong, Wancai Zhang. “Informer:超越高效 Transformer 的长序列时序预测”

1. VanillaTransformer


源代码

VanillaTransformer

 VanillaTransformer (h:int, input_size:int, stat_exog_list=None,
                     hist_exog_list=None, futr_exog_list=None,
                     exclude_insample_y=False,
                     decoder_input_size_multiplier:float=0.5,
                     hidden_size:int=128, dropout:float=0.05,
                     n_head:int=4, conv_hidden_size:int=32,
                     activation:str='gelu', encoder_layers:int=2,
                     decoder_layers:int=1, loss=MAE(), valid_loss=None,
                     max_steps:int=5000, learning_rate:float=0.0001,
                     num_lr_decays:int=-1,
                     early_stop_patience_steps:int=-1,
                     val_check_steps:int=100, batch_size:int=32,
                     valid_batch_size:Optional[int]=None,
                     windows_batch_size=1024,
                     inference_windows_batch_size:int=1024,
                     start_padding_enabled=False, step_size:int=1,
                     scaler_type:str='identity', random_seed:int=1,
                     drop_last_loader:bool=False,
                     alias:Optional[str]=None, optimizer=None,
                     optimizer_kwargs=None, lr_scheduler=None,
                     lr_scheduler_kwargs=None, dataloader_kwargs=None,
                     **trainer_kwargs)

*VanillaTransformer

香草 Transformer,遵循 Informer 论文的实现,用作基准模型。

该架构具有三个显著特点: - 时间和内存复杂度为 O(L^2) 的全注意力机制。 - 一个 MLP 多步解码器,它通过单次前向操作预测长时序序列,而不是逐步预测。

Vanilla Transformer 模型采用三组件方法来定义其嵌入: - 它使用从卷积网络获得的编码自回归特征。 - 它使用从谐波函数导出的窗口相对位置嵌入。 - 它使用从日历特征获得的绝对位置嵌入。

参数
h: int,预测范围。
input_size: int,截断训练反向传播的最大序列长度。
stat_exog_list: str list,静态外部变量列。
hist_exog_list: str list,历史外部变量列。
futr_exog_list: str list,未来外部变量列。
exclude_insample_y: bool=False,是否从输入中排除目标变量。
decoder_input_size_multiplier: float = 0.5,。
hidden_size: int=128,嵌入和编码器的单元数。
dropout: float (0, 1),Informer 架构中的 dropout 率。
n_head: int=4,控制多头注意力的头数。
conv_hidden_size: int=32,卷积编码器的通道数。
activation: str=GELU,激活函数,可选值包括 [‘ReLU’, ‘Softplus’, ‘Tanh’, ‘SELU’, ‘LeakyReLU’, ‘PReLU’, ‘Sigmoid’, ‘GELU’]。
encoder_layers: int=2,TCN 编码器的层数。
decoder_layers: int=1,MLP 解码器的层数。
loss: PyTorch 模块,来自损失集合的训练损失类实例。
valid_loss: PyTorch 模块=loss,来自损失集合的验证损失类实例。

max_steps: int=1000,最大训练步数。
learning_rate: float=1e-3,学习率,范围为 (0, 1)。
num_lr_decays: int=-1,学习率衰减次数,均匀分布在 max_steps 中。
early_stop_patience_steps: int=-1,早停前所需的验证迭代次数。
val_check_steps: int=100,每次验证损失检查之间的训练步数。
batch_size: int=32,每个批次中不同序列的数量。
valid_batch_size: int=None,每个验证和测试批次中不同序列的数量,如果为 None 则使用 batch_size。
windows_batch_size: int=1024,每个训练批次中采样的窗口数量,默认为所有窗口。
inference_windows_batch_size: int=1024,每个推理批次中采样的窗口数量。
start_padding_enabled: bool=False,如果为 True,模型将在时间序列开头用零填充,填充长度等于 input size。
step_size: int=1,每个时间数据窗口之间的步长。
scaler_type: str=‘robust’,用于时间输入归一化的缩放器类型,参见时间缩放器
random_seed: int=1,用于 pytorch 初始化器和 numpy 生成器的随机种子。
drop_last_loader: bool=False,如果为 True,则 TimeSeriesDataLoader 将丢弃最后一个非完整批次。
alias: str,可选,模型的自定义名称。
optimizer: ‘torch.optim.Optimizer’ 的子类,可选,用户指定的优化器而非默认选择 (Adam)。
optimizer_kwargs: dict,可选,用户指定的 optimizer 使用的参数列表。
lr_scheduler: ‘torch.optim.lr_scheduler.LRScheduler’ 的子类,可选,用户指定的 lr_scheduler 而非默认选择 (StepLR)。
lr_scheduler_kwargs: dict,可选,用户指定的 lr_scheduler 使用的参数列表。
dataloader_kwargs: dict,可选,由 TimeSeriesDataLoader 传递给 PyTorch Lightning dataloader 的参数列表。
**trainer_kwargs: int,继承自PyTorch Lightning 的 trainer 的关键字 trainer 参数。

*References*<br/>
- [Haoyi Zhou, Shanghang Zhang, Jieqi Peng, Shuai Zhang, Jianxin Li, Hui Xiong, Wancai Zhang. "Informer: Beyond Efficient Transformer for Long Sequence Time-Series Forecasting"](https://arxiv.org/abs/2012.07436)<br/>*

VanillaTransformer.fit

 VanillaTransformer.fit (dataset, val_size=0, test_size=0,
                         random_seed=None, distributed_config=None)

*训练。

fit 方法使用初始化参数 (learning_rate, windows_batch_size, ...) 和初始化时定义的 loss 函数来优化神经网络的权重。在 fit 方法内部,我们使用 PyTorch Lightning 的 Trainer,它继承了初始化时的 self.trainer_kwargs 参数以定制输入,参见 PL 的 trainer 参数

该方法旨在与 SKLearn 风格的类兼容,特别是与 StatsForecast 库兼容。

默认情况下,model 不保存训练检查点以保护磁盘空间,如需保存,请在 __init__ 中将 enable_checkpointing 设置为 True

参数
dataset: NeuralForecast 的 TimeSeriesDataset,参见文档
val_size: int,时间交叉验证的验证集大小。
random_seed: int=None,用于 pytorch 初始化器和 numpy 生成器的随机种子,会覆盖 model.__init__ 中的设置。
test_size: int,时间交叉验证的测试集大小。
*


VanillaTransformer.predict

 VanillaTransformer.predict (dataset, test_size=None, step_size=1,
                             random_seed=None, quantiles=None,
                             **data_module_kwargs)

*预测。

使用 PL 的 Trainer 执行 predict_step 进行神经网络预测。

参数
dataset: NeuralForecast 的 TimeSeriesDataset,参见文档
test_size: int=None,时间交叉验证的测试集大小。
step_size: int=1,每个窗口之间的步长。
random_seed: int=None,用于 pytorch 初始化器和 numpy 生成器的随机种子,会覆盖 model.__init__ 中的设置。
quantiles: float 列表,可选 (默认值=None),要预测的目标分位数。
**data_module_kwargs: PL 的 TimeSeriesDataModule 参数,参见文档。*

使用示例

import pandas as pd
import matplotlib.pyplot as plt

from neuralforecast import NeuralForecast
from neuralforecast.models import VanillaTransformer
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

model = VanillaTransformer(h=12,
                 input_size=24,
                 hidden_size=16,
                 conv_hidden_size=32,
                 n_head=2,
                 loss=MAE(),
                 scaler_type='robust',
                 learning_rate=1e-3,
                 max_steps=500,
                 val_check_steps=50,
                 early_stop_patience_steps=2)

nf = NeuralForecast(
    models=[model],
    freq='ME'
)
nf.fit(df=Y_train_df, static_df=AirPassengersStatic, val_size=12)
forecasts = nf.predict(futr_df=Y_test_df)

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

if model.loss.is_distribution_output:
    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['VanillaTransformer-median'], c='blue', label='median')
    plt.fill_between(x=plot_df['ds'][-12:], 
                    y1=plot_df['VanillaTransformer-lo-90'][-12:].values, 
                    y2=plot_df['VanillaTransformer-hi-90'][-12:].values,
                    alpha=0.4, label='level 90')
    plt.grid()
    plt.legend()
    plt.plot()
else:
    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['VanillaTransformer'], c='blue', label='Forecast')
    plt.legend()
    plt.grid()