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

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

---

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

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

• 一个时间序列数据集类，抽象了变量转换、缺失值处理、随机子采样、多种历史长度等操作。
• 一个基础模型类，提供时间序列模型的基本训练功能，以及 TensorBoard 日志记录和通用可视化功能，例如实际值与预测值对比图和依赖性图。
• 多种针对时间序列预测优化的神经网络架构，这些架构经过改进以适应实际部署，并内置了解释能力。
• 多步长时间序列指标。
• 使用 optuna 进行超参数调优。

该包基于 pytorch-lightning 构建，因此可以开箱即用地在 CPU、单个或多个 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，阅读包含详细教程的文档。

可用模型

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

• 用于可解释多步长时间序列预测的 Temporal Fusion Transformers，在基准测试中比亚马逊的 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",  # 如果有多块 GPU，则使用 strategy="ddp"；否则在 CPU 上运行
    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,
)

PyTorch Forecasting 快速上手指南

PyTorch Forecasting 是一个基于 PyTorch 的时间序列预测包，集成了 Temporal Fusion Transformers (TFT)、N-BEATS、DeepAR 等最先进的深度学习架构。它利用 PyTorch Lightning 实现高效的 GPU/CPU 训练，并提供高级 API 以简化数据预处理、模型训练及超参数调优流程。

环境准备

在开始之前，请确保您的开发环境满足以下要求：

• 操作系统: Linux, macOS 或 Windows
• Python 版本: 3.8 - 3.11 (推荐最新稳定版)
• 核心依赖:
  • PyTorch (>= 1.7)
  • PyTorch Lightning
  • pandas, numpy, scikit-learn
• 硬件建议: 支持 CUDA 的 NVIDIA GPU 可显著加速训练（可选，CPU 亦可运行）

安装步骤

1. 安装 PyTorch (Windows 用户必读)

如果您使用的是 Windows 系统，请先单独安装 PyTorch：

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

Linux/macOS 用户通常可直接通过下一步安装，或访问 PyTorch 官网获取对应命令。

2. 安装 PyTorch Forecasting

推荐使用 pip 进行安装。国内用户可使用清华源或阿里源加速下载：

标准安装：

pip install pytorch-forecasting -i https://pypi.tuna.tsinghua.edu.cn/simple

Conda 安装方式：

conda install pytorch-forecasting pytorch -c pytorch -c conda-forge

可选组件： 如果您需要使用多变量分位数损失 (MQF2 loss)，请额外安装：

pip install pytorch-forecasting[mqf2]

基本使用

以下是使用 Temporal Fusion Transformer (TFT) 模型进行训练的最小化工作流示例。主要步骤包括：数据加载 -> 构建数据集 -> 创建 DataLoader -> 初始化模型 -> 训练。

# 导入必要的库
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

# 1. 加载数据
# 数据应为 pandas DataFrame，必须包含：
# * target: 预测目标列
# * time_idx: 时间索引列 (单调递增整数)
# * group_ids: 时间序列 ID 列 (用于区分不同序列)
data = ... 

# 2. 定义数据集
max_encoder_length = 36       # 历史输入长度
max_prediction_length = 6     # 未来预测长度
training_cutoff = "YYYY-MM-DD" # 训练集截止日期

training = TimeSeriesDataSet(
    data[lambda x: x.date <= training_cutoff],
    time_idx="time_idx",           # 替换为实际列名
    target="target",               # 替换为实际列名
    group_ids=["series_id"],       # 替换为实际列名
    max_encoder_length=max_encoder_length,
    max_prediction_length=max_prediction_length,
    # 定义协变量 (根据实际数据填写)
    static_categoricals=["static_cat"],
    static_reals=["static_real"],
    time_varying_known_categoricals=["known_cat"],
    time_varying_known_reals=["known_real"],
    time_varying_unknown_categoricals=["unknown_cat"],
    time_varying_unknown_reals=["unknown_real"],
)

# 创建验证集 (复用训练集的归一化统计量)
validation = TimeSeriesDataSet.from_dataset(
    training, 
    data, 
    min_prediction_idx=training.index.time.max() + 1, 
    stop_randomization=True
)

# 3. 转换为 DataLoader
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)

# 4. 配置 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",  # 自动检测 GPU/CPU
    gradient_clip_val=0.1,
    limit_train_batches=30,
    callbacks=[lr_logger, early_stop_callback],
    logger=TensorBoardLogger("lightning_logs")
)

# 5. 初始化模型
# 大部分架构参数会自动从数据集中推断
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")

# 6. (可选) 寻找最佳学习率
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()}")
# res.plot(show=True, suggest=True) # 取消注释以查看图表

# 7. 开始训练
trainer.fit(
    tft, 
    train_dataloaders=train_dataloader, 
    val_dataloaders=val_dataloader,
)

训练完成后，您可以使用该模型进行预测、可视化结果或导出模型。更多高级功能（如自定义模型、详细调参）请参考官方文档。