PyTorch Forecasting 是一个基于 PyTorch 的预测包，采用最先进的深度学习架构进行时间序列预测。它提供了高级 API，并利用 PyTorch Lightning 在 GPU 或 CPU 上进行扩展训练，同时支持自动日志记录。

	文档 · 教程 · 发布说明
开源
社区
CI/CD
代码
下载量

我们在 Towards Data Science 上的文章介绍了该包并提供了背景信息。

PyTorch Forecasting 旨在简化使用神经网络进行时间序列预测的流程，适用于实际场景和研究。其目标是提供高级 API，为专业人士提供最大灵活性，为初学者提供合理的默认设置。具体来说，该包提供：

• 一个时间序列数据集类，抽象处理变量变换、缺失值、随机子采样、多历史长度等
• 一个基础模型类，提供时间序列模型的基本训练功能，支持 TensorBoard 日志记录以及通用可视化（如实际值 vs 预测值和依赖图）
• 多种时间序列预测神经网络架构，经过增强可用于实际部署，并内置解释能力
• 多步预测时间序列指标
• 使用 optuna 进行超参数调优

该包基于 pytorch-lightning 构建，支持开箱即用地在 CPU、单 GPU 和多 GPU 上进行训练。

安装

如果在 Windows 上工作，需要先使用以下命令安装 PyTorch：

pip install torch -f https://download.pytorch.org/whl/torch_stable.html

否则，可以直接进行安装：

pip install pytorch-forecasting

或者，可以通过 conda 安装该包：

conda install pytorch-forecasting pytorch -c pytorch>=1.7 -c conda-forge

PyTorch Forecasting 现在从 conda-forge 渠道安装，而 PyTorch 从 pytorch 渠道安装。

要使用 MQF2 损失（多变量分位数损失），还需要安装：

pip install pytorch-forecasting[mqf2]

文档

访问 https://pytorch-forecasting.readthedocs.io 阅读包含详细教程的文档。

可用模型

文档提供了可用模型的比较。

• 时间融合 Transformer，用于可解释的多步时间序列预测，在基准测试中比 Amazon 的 DeepAR 表现提升 36-69%
• N-BEATS：用于可解释时间序列预测的神经基扩展分析，在 M4 竞赛中以集成形式使用，表现优于包括传统统计方法集成的所有其他方法。M4 竞赛可以说是单变量时间序列预测中最重要的基准。
• N-HiTS：用于时间序列预测的神经层级插值，支持协变量，持续优于 N-BEATS，特别适合长步长预测。
• DeepAR：基于自回归循环网络的概率预测，是最流行的预测算法之一，通常用作基准方法
• 用于基准测试的简单标准网络：LSTM 和 GRU 网络，以及解码器上的 MLP
• 始终预测最新已知值的基准模型

要实现新模型或其他自定义组件，请参阅如何实现新模型教程，涵盖基础和高级架构。

使用示例

网络可以使用 PyTorch Lightning Trainer 在 pandas Dataframes 上进行训练，这些数据首先转换为 TimeSeriesDataSet。

# 导入训练所需模块
import lightning.pytorch as pl
from lightning.pytorch.loggers import TensorBoardLogger
from lightning.pytorch.callbacks import EarlyStopping, LearningRateMonitor
# 导入数据集、要训练的网络和要优化的指标
from pytorch_forecasting import TimeSeriesDataSet, TemporalFusionTransformer, QuantileLoss
from lightning.pytorch.tuner import Tuner

# 加载数据：这是一个 pandas 数据框，至少包含以下列：
# * 目标变量（要预测的内容）
# * 时间序列 ID（应该是唯一字符串，用于标识每个时间序列）
# * 观察时间（应该是单调递增的整数）
data = ...

# 定义数据集，即为 pandas 数据框添加元数据，以便模型理解
max_encoder_length = 36
max_prediction_length = 6
training_cutoff = "YYYY-MM-DD"  # 截止日期

training = TimeSeriesDataSet(
    data[lambda x: x.date <= training_cutoff],
    time_idx= ...,  # 观察时间列名
    target= ...,  # 要预测的目标列名
    group_ids=[ ... ],  # 时间序列 ID 列名
    max_encoder_length=max_encoder_length,  # 使用多少历史数据
    max_prediction_length=max_prediction_length,  # 预测未来的长度
    # 时间序列 ID 的静态协变量
    static_categoricals=[ ... ],
    static_reals=[ ... ],
    # 未来已知和未知的协变量，用于预测
    time_varying_known_categoricals=[ ... ],
    time_varying_known_reals=[ ... ],
    time_varying_unknown_categoricals=[ ... ],
    time_varying_unknown_reals=[ ... ],
)

# 使用与训练数据集相同的归一化技术创建验证数据集
validation = TimeSeriesDataSet.from_dataset(training, data, min_prediction_idx=training.index.time.max() + 1, stop_randomization=True)

# 将数据集转换为数据加载器以用于训练
batch_size = 128
train_dataloader = training.to_dataloader(train=True, batch_size=batch_size, num_workers=2)
val_dataloader = validation.to_dataloader(train=False, batch_size=batch_size, num_workers=2)

# 创建带有早停功能的 PyTorch Lightning Trainer
early_stop_callback = EarlyStopping(monitor="val_loss", min_delta=1e-4, patience=1, verbose=False, mode="min")
lr_logger = LearningRateMonitor()
trainer = pl.Trainer(
    max_epochs=100,
    accelerator="auto",  # 在 CPU 上运行；如果使用多个 GPU，请使用 strategy="ddp"
    gradient_clip_val=0.1,
    limit_train_batches=30,  # 每个 epoch 使用 30 个批次
    callbacks=[lr_logger, early_stop_callback],
    logger=TensorBoardLogger("lightning_logs")
)

# 定义要训练的网络 - 架构主要由数据集推断得出，因此用户只需设置少数几个超参数
tft = TemporalFusionTransformer.from_dataset(
    # 数据集
    training,
    # 架构超参数
    hidden_size=32,
    attention_head_size=1,
    dropout=0.1,
    hidden_continuous_size=16,
    # 要优化的损失指标
    loss=QuantileLoss(),
    # 日志记录频率
    log_interval=2,
    # 优化器参数
    learning_rate=0.03,
    reduce_on_plateau_patience=4
)
print(f"网络中的参数数量: {tft.size()/1e3:.1f}k")

# 找到最优学习率
res = Tuner(trainer).lr_find(
    tft, train_dataloaders=train_dataloader, val_dataloaders=val_dataloader, early_stop_threshold=1000.0, max_lr=0.3,
)
# 并绘制结果 - 始终目视确认建议的学习率是否合理
print(f"建议的学习率: {res.suggestion()}")
fig = res.plot(show=True, suggest=True)
fig.show()

# 在数据上拟合模型 - 如有必要，使用正确学习率重新定义模型
trainer.fit(
    tft, train_dataloaders=train_dataloader, val_dataloaders=val_dataloader,
)