Torch 预测模型

本文档是为 darts 版本 0.15.0 及更高版本编写的。

我们假设您已经了解 Darts 中的协变量。如果您是这个主题的新手,我们建议您先阅读我们的 协变量指南

本指南的内容

  1. 介绍部分涵盖了关于Torch预测模型(TFMs)的最重要点:

  2. 输入数据使用部分提供了在使用TFMs进行训练和预测时,输入数据如何使用的深入指南:

  3. 高级功能部分提供了一些TFMs高级功能的示例:

  4. 性能优化部分 列出了在训练期间加速计算的技巧。

介绍

在 Darts 中,Torch 预测模型 (TFMs) 广义上来说是“基于机器学习”的模型,这些模型表示基于 PyTorch(深度学习)的模型。

TFMs 在固定长度的块(子样本)上训练和预测您的输入 target*_covariates 序列(如果支持)。Target 是我们想要预测未来的序列,*_covariates 是过去和/或未来的协变量。

每个块包含一个输入块——代表样本的过去——和一个输出块——样本的未来。样本的预测点位于输入块的末尾。这些块的长度必须在模型创建时通过参数 input_chunk_lengthoutput_chunk_length 指定(更多关于块的内容请参见 下一小节)。

# model that looks 7 time steps back (past) and 1 time step ahead (future)
model = SomeTorchForecastingModel(input_chunk_length=7,
                                  output_chunk_length=1,
                                  **model_kwargs)

所有TFM模型都可以在单个或多个``target``序列上进行训练,并且根据其协变量支持(在 此小节中介绍),可以使用``past_covariates``和/或``future_covariates``。当使用协变量时,你必须为每个目标序列提供一个专用的过去和/或未来协变量序列。

在训练过程中,您可以选择使用带有专用协变量的验证集。如果协变量具有所需的时间跨度,您可以将其用于训练、验证和预测。(详见 本小节

# fit the model on a single target series with optional past and / or future covariates
model.fit(target,
          past_covariates=past_covariates,
          future_covariates=future_covariates,
          val_series=target_val,  # optionally, use a validation set
          val_past_covariates=past_covariates_val,
          val_future_covariates=future_covariates_val)

# fit the model on multiple target series
model.fit([target, target2, ...],
          past_covariates=[past_covariates, past_covariates2, ...],
          ...
          )

您可以为任何输入 target 时间序列生成预测,或者为作为时间序列序列给出的多个目标生成预测。只要每条序列至少包含 input_chunk_length 个时间步,这同样适用于训练期间未见过的序列。

# predict the next n=3 time steps for any input series with `series`
prediction = model.predict(n=3,
                           series=target,
                           past_covariates=past_covariates,
                           future_covariates=future_covariates)

如果你想了解更多关于我们的 Torch 预测模型的训练和预测过程,以及它们如何与协变量一起工作,请继续阅读。

对使用块进行训练和预测的顶层概览

在图1中,您可以看到在调用 fit()predict() 时,您的数据是如何分布到每个样本的输入和输出块中的。在这个例子中,我们查看的是具有每日频率的数据。输入块从 target 中提取值,并可选地从 past_covariates 和/或 future_covariates 中提取值,这些值落在输入块的时间跨度内。这些 future_covariates 的“过去”值被称为“历史未来协变量”。

输出块只接受可选的 future_covariates 值,这些值落在输出块的时间跨度内。我们 past_covariates 的未来值——“未来过去协变量”——仅用于为即将到来的样本输入块提供新数据。

所有这些信息都被用来预测“未来目标”——在“过去目标”结束后的下一个 output_chunk_length 点。

figure0

图1:使用Torch预测模型对块进行训练/预测的顶层视图

当调用 predict() 并根据你的预测范围 n 时,模型可以一次性预测(如果 n <= output_chunk_length ),或者自回归地,通过在未来多个块上进行预测(如果 n > output_chunk_length )。这就是为什么在使用 past_covariates 进行预测时,你可能需要提供额外的“past_covariates 的未来值”。

Torch 预测模型协变量支持

在底层,Darts 实现了 5 种类型的 {X}CovariatesModel 类,以涵盖前面提到的不同协变量组合:

过去协变量

未来 过去 协变量

未来协变量

历史未来协变量

PastCovariatesModel

FutureCovariatesModel

DualCovariatesModel

SplitCovariatesModel

MixedCovariatesModel

表1:飞镖的”{X}CovariatesModels”协变量支持

每个 Torch 预测模型都继承自一个 {X}CovariatesModel (协变量类名由 X 部分缩写):

TFM

过去

Future

Dual

Split

Mixed

RNNModel

BlockRNNModel

NBEATSModel

TCNModel

TransformerModel

TFTModel

NLinearModel

DLinearModel

TiDEModel

TSMixerModel

表2:飞镖的Torch预测模型协变量支持

训练、验证和预测所需的时段

相关数据是根据序列的时间轴由模型自动提取的。如果它们满足以下要求,您可以对 fit()predict() 使用相同的协变量序列。

训练 只有在从传递给 fit() 的数据中至少提取出一个带有输入和输出块的样本时才能进行。这既适用于训练数据,也适用于验证数据。就所需的最短时间跨度而言,这意味着:

  • target 系列的最小长度为 input_chunk_length + output_chunk_length

  • *_covariatescovariates 指南第 2.3 节fit() 的时间跨度要求

对于 预测,您必须提供希望预测的 目标 序列。对于任何预测范围 n,最小时间跨度要求是:

  • target 系列的最小长度为 input_chunk_length

  • *_covariatespredict() 的时间跨度要求也来自 covariates 指南第2.3节

旁注:我们的 *RNNModels 在模型创建时接受 training_length 参数,而不是 output_chunk_length。这些模型的 output_chunk_length 在内部自动设置为 1。对于训练,过去的 target 必须具有至少 training_length + 1 的长度,而对于预测,长度必须为 input_chunk_length

深入了解在训练和预测时如何使用输入数据与TFMs

训练

让我们来看看这些模型是如何在底层工作的。

假设我们经营一家冰淇淋店,我们想要预测第二天的销售额。我们有一年的(365天)过去数据,包括每天结束时的冰淇淋销售额和平均每日环境温度。我们还注意到,我们的冰淇淋销售额取决于星期几,因此我们希望将这一点纳入我们的模型中。

  • 过去目标:实际过去的冰淇淋销售 ice_cream_sales

  • 未来目标:预测明天的冰淇淋销售量

  • 过去协变量:过去测量的平均日温度 temperature

  • 未来协变量:过去和未来的 weekday 的星期几

检查表1,能够适应这种协变量的模型将是一个 SplitCovariatesModel``(如果我们不使用未来协变量的历史值),或者 ``MixedCovariatesModel``(如果我们使用)。我们选择了一个 ``MixedCovariatesModel - TFTModel

想象一下,我们发现过去的冰淇淋销售模式每周重复一次。因此,我们设置 input_chunk_length = 7 天,让模型回顾过去整整一周。output_chunk_length 可以设置为 1 天,以预测第二天。

现在我们可以创建一个模型并训练它!图2展示了 TFTModel 将如何使用我们的数据。

from darts.models import TFTModel

model = TFTModel(input_chunk_length=7, output_chunk_length=1)
model.fit(series=ice_cream_sales,
          past_covariates=temperature,
          future_covariates=weekday)
图4

图2:冰淇淋销售示例中单个序列的概览;Mon1 - Sun1 代表我们训练数据集中的前7天(第1周)。Mon2 是第2周的星期一。

当调用 fit() 时,模型将构建一个适当的 darts.utils.data.TrainingDataset ,该数据集指定了如何切片数据以获取训练样本。如果你想自己控制这个切片过程,你可以实例化你自己的 TrainingDataset 并调用 model.fit_from_dataset() 而不是 fit()。默认情况下,大多数模型(尽管不是所有)将构建 顺序 数据集,这基本上意味着提供的序列中所有长度为 input_chunk_length + output_chunk_length 的子切片都将用于训练。

因此,在训练过程中,torch 模型将按顺序遍历训练数据(见图3)。利用 输入块输出块 的信息,模型预测输出块上的未来目标。训练损失是通过预测的未来目标与输出块上的实际目标值之间的差异来评估的。模型通过最小化所有序列的损失来进行自我训练。

图5

图3:单个序列中的预测和损失评估

在完成对第一个序列的计算后,模型移动到下一个序列并执行相同的训练步骤。每个序列的起始点是从顺序数据集中随机选择的。图4展示了如果纯粹出于偶然,第二个序列比第一个序列晚一个时间步(天)开始的情况。

这个序列到序列的过程会重复,直到覆盖所有365天。

旁注:拥有“长”的 target 序列可能会导致大量的训练序列/样本。你可以通过 fit() 参数 max_samples_per_ts 为每个 target 设置训练序列/样本的数量上限。这将选择每个 target 序列中最接近 target 结束的最新序列。

# fit only on the 10 "most recent" sequences
model.fit(target, max_samples_per_ts=10)
图6

图4:序列到序列:移动到下一个序列并重复训练步骤

使用验证数据集进行训练

你也可以使用验证数据集来训练你的模型:

# create train and validation sets
ice_cream_sales_train, ice_cream_sales_val = ice_cream_sales.split_after(training_cutoff)

# train with validation set
model.fit(series=ice_cream_sales_train,
          past_covariates=temperature,
          future_covariates=weekday,
          val_series=ice_cream_sales_val,
          val_past_covariates=temperature,
          val_future_covariates=weekday)

如果你分割数据,你必须定义一个 training_cutoff (一个日期或分割数据集的分数),以便训练和验证数据集都满足 这一小节 中的最小长度要求。

除了按时间分割,你还可以使用时间序列的另一个子集作为验证集。

该模型以与之前相同的方式进行训练,但另外在验证数据集上评估损失。如果你想跟踪验证集上表现最好的模型,你必须启用检查点保存,如下所示。

预测/预报

在训练模型之后,我们希望预测365天训练数据之后的任意天数的冰淇淋销售情况。

实际的预测工作与我们在序列上训练数据的方式非常相似。根据我们想要预测的天数 - 预测范围 n - 我们区分两种情况:

  • 如果 n <= output_chunk_length:我们可以一次性预测 ``n``(使用一次“内部模型调用”)

    • 在我们的示例中:预测第二天的冰淇淋销量(n = 1

  • 如果 n > output_chunk_length:我们需要通过多次调用内部模型来预测 n。每次调用输出 output_chunk_length 个预测点。我们以自回归的方式进行多次调用,直到获得最终的 n 个预测点。

    • 在我们的示例中:一次性预测未来3天的冰淇淋销量(n = 3

    为此,我们必须为训练数据结束后的接下来 n - output_chunk_length = 2 个时间步(天)提供额外的 past_covariates。不幸的是,我们没有未来测量的 temperature。但让我们假设我们可以获得接下来两天的温度预报。我们可以将它们附加到 temperature 上,预测就会起作用!

    temperature = temperature.concatenate(temperature_forecast, axis=0)
    
prediction = model.predict(n=n,
                           series=ice_cream_sales_train,
                           past_covariates=temperature,
                           future_covariates=weekday)
图7

图5:针对 ``n <= output_chunk_length`` 的单一序列预测

figure8

图6:对 ``n > output_chunk_length`` 的自回归预测

高级功能

保存和加载模型状态

❗ 警告 ❗ 在 Darts 开发的现阶段,我们尚未确保向后兼容性,因此可能无法始终加载由库的旧版本保存的模型。

对于在 GPU 上训练的 Darts 版本 <= 0.22.0 的模型,需要在 Darts 版本 >= 0.23.0 的 CPU 上加载,请查看此 issue 中提供的代码片段。

自动检查点

训练期间的自动检查点功能允许你:

  • 跟踪模型在最近5个epoch的状态,以及基于验证集损失表现最好的epoch。

  • 从检查点加载模型以在训练中断时恢复训练

  • 从检查点加载模型进行推理/预测

您可以在模型创建时激活检查点:

model = SomeTorchForecastingModel(..., model_name='my_model', save_checkpoints=True)

# checkpoints are saved automatically
model.fit(...)

# load the model state that performed best on validation set
best_model = model.load_from_checkpoint(model_name='my_model', best=True)

手动保存 / 加载

你也可以手动保存模型在其当前状态并加载它:

model.save("/your/path/to/save/model.pt")
loaded_model = model.load("/your/path/to/save/model.pt")

在GPU上训练/保存并在CPU上加载

你可以将一个在GPU上训练并保存的模型加载到CPU上(详见`文档 <https://unit8co.github.io/darts/userguide/gpu_and_tpu_usage.html>`_):

# define a model using gpu as accelerator
model = SomeTorchForecastingModel(...,
                                  model_name='my_model',
                                  save_checkpoints=True,
                                  pl_trainer_kwargs={
                                                     "accelerator":"gpu",
                                                     "devices": -1,
                                                     })

# train the model, automatic checkpoints will be created
model.fit(...)

# specify the device to which the model should be loaded
loaded_model = SomeTorchForecastingModel.load_from_checkpoint(model_name='my_model',
                                                              best=True,
                                                              map_location="cpu")
loaded_model.to_cpu()

# run inference
loaded_model.predict(...)

手动保存也可以加载到CPU:

model.save("/your/path/to/save/model.pt")
loaded_model = model.load("/your/path/to/save/model.pt", map_location="cpu")
loaded_model.to_cpu()

重新训练或微调预训练模型

要使用不同的优化器和/或学习率调度器重新训练或微调模型,您可以将自动检查点的权重加载到新模型中:

# model with identical architecture but different optimizer (default: torch.optim.Adam)
model_finetune = SomeTorchForecastingModel(...,  # use identical parameters & values as in original model
                                           optimizer_cls=torch.optim.SGD,
                                           optimizer_kwargs={"lr": 0.001})

# load the weights from a checkpoint
model_finetune.load_weights_from_checkpoint(model_name='my_model', best=True)

model_finetune.fit(...)

同样适用于手动保存和学习率调度器:

# model with identical architecture but different lr scheduler (default: None)
model_finetune = SomeTorchForecastingModel(...,  # use identical parameters & values as in original model
                                           lr_scheduler_cls=torch.optim.lr_scheduler.ExponentialLR,
                                           lr_scheduler_kwargs={"gamma": 0.09})

# load the weights from a manual save
model_finetune.load_weights("/your/path/to/save/model.pt")

回调

回调是在训练过程中监控或控制模型行为的强大方式。一些例子:

  • 性能监控:计算额外的指标(除了默认的损失之外)

  • 早停法:一旦模型收敛,就停止训练

通过回调,您可以在现有流程的预定义点/钩子上添加自定义代码。一旦流程执行到达相应的钩子,代码就会被触发。一些示例钩子:

  • 训练的开始 / 结束

  • 训练 / 验证步骤的开始 / 结束

一些有用的预定义 PyTorch Lightning 回调可以在 这里 找到。

带有早停的示例

早停是一种有效避免过拟合和减少训练时间的方法。一旦验证损失在某些周期内没有显著改善,它将退出训练过程。

你可以在任何 TorchForecastingModel 中使用早停法,利用 PyTorch Lightning 的 EarlyStopping 回调:

import pandas as pd
from pytorch_lightning.callbacks import EarlyStopping
from torchmetrics import MeanAbsolutePercentageError

from darts.dataprocessing.transformers import Scaler
from darts.datasets import AirPassengersDataset
from darts.models import NBEATSModel

# read data
series = AirPassengersDataset().load()

# create training and validation sets:
train, val = series.split_after(pd.Timestamp(year=1957, month=12, day=1))

# normalize the time series
transformer = Scaler()
train = transformer.fit_transform(train)
val = transformer.transform(val)

# any TorchMetric or val_loss can be used as the monitor
torch_metrics = MeanAbsolutePercentageError()

# early stop callback
my_stopper = EarlyStopping(
    monitor="val_MeanAbsolutePercentageError",  # "val_loss",
    patience=5,
    min_delta=0.05,
    mode='min',
)
pl_trainer_kwargs = {"callbacks": [my_stopper]}

# create the model
model = NBEATSModel(
    input_chunk_length=24,
    output_chunk_length=12,
    n_epochs=500,
    torch_metrics=torch_metrics,
    pl_trainer_kwargs=pl_trainer_kwargs)

# use validation set for early stopping
model.fit(
    series=train,
    val_series=val,
)

要在超参数优化的背景下使用早停和剪枝,请查看 这个指南

自定义回调存储损失的示例

训练和验证损失可以通过 tensorboard 自动记录。激活后,Darts 默认会将日志存储到当前工作目录中的 darts_logs 文件夹。您可以通过模型参数 work_dirmodel_name 更改此设置。

model = SomeTorchForecastingModel(..., log_tensorboad, save_checkpoints=True)
model.fit(...)

安装 tensorboard 库后,您可以从命令行可视化日志:

tensorboad --log_dir darts_logs

让我们看看如何实现一个 自定义回调 来在Python中访问模型损失。

from pytorch_lightning.callbacks import Callback

class LossLogger(Callback):
    def __init__(self):
        self.train_loss = []
        self.val_loss = []

    # will automatically be called at the end of each epoch
    def on_train_epoch_end(self, trainer: "pl.Trainer", pl_module: "pl.LightningModule") -> None:
        self.train_loss.append(float(trainer.callback_metrics["train_loss"]))

    def on_validation_epoch_end(self, trainer: "pl.Trainer", pl_module: "pl.LightningModule") -> None:
        self.val_loss.append(float(trainer.callback_metrics["val_loss"]))


loss_logger = LossLogger()

model = SomeTorchForecastingModel(
    ...,
    nr_epochs_val_period=1,  # perform validation after every epoch
    pl_trainer_kwargs={"callbacks": [loss_logger]}
)

# fit must include validation set for "val_loss"
model.fit(...)

注意 : 回调将在 loss_logger.val_loss 中多给出一个元素,因为在训练开始之前,模型训练器会执行一次验证检查。

性能建议

本节回顾了在训练和使用基于torch的模型时影响性能的主要因素。

使用32位数据构建您的 TimeSeries

Darts 中的模型会根据 TimeSeries 中的 dtype 动态地将自己转换为 64 位或 32 位。当所有内容(数据和模型)都为 float32 时,通常可以获得较大的性能和内存收益。要实现这一点,只需使用 dtype 为 np.float32 的数组(或 DataFrame 支持的数组)构建你的 TimeSeries,或者简单地调用 my_series32 = my_series.astype(np.float32)。调用 my_series.dtype 可以获取你的 TimeSeries 的 dtype。

使用 GPU

在许多情况下,使用GPU相比CPU会提供显著的速度提升。它也可能带来一些开销(例如数据传输到/从GPU),因此通常需要一些测试和调整。我们参考我们的 GPU/TPU指南 以获取更多关于如何通过PyTorch Lightning设置GPU(或TPU)的信息。

调整批次大小

较大的批量大小往往会加快训练速度,因为它减少了每个epoch的反向传播次数,并且有可能更好地并行化计算。然而,它也会改变训练动态(例如,您可能需要更多的epoch,并且收敛动态会受到影响)。此外,较大的批量大小会增加内存消耗。因此,这里也需要进行一些测试。

调整 num_loader_workers

Darts 中的所有深度学习模型在其 fit()predict() 函数中都有一个参数 num_loader_workers,该参数配置了 PyTorch DataLoaders 中的 num_workers 参数。默认情况下,它设置为 0,这意味着主进程也将负责加载数据。设置 num_workers > 0 将使用额外的工人来加载数据。这通常会产生一些开销(特别是增加内存消耗),但在某些情况下,它也可以显著提高性能。理想值取决于许多因素,如批量大小、是否使用 GPU、可用 CPU 核心的数量,以及加载数据是否涉及 I/O 操作(如果序列存储在磁盘上)。

从小模型开始

当然,影响性能的主要因素之一是模型大小(参数数量)以及前向/后向传递所需的操作数量。Darts 中的模型可以进行调整(例如,层数、注意力头、宽度等),这些超参数往往对性能有较大影响。开始时,最好先构建适度大小的模型。

内存中的数据和I/O瓶颈

如果可以的话,提前将所有 TimeSeries 加载到内存中是有帮助的。Darts 提供了在任何 Sequence[TimeSeries] 上训练模型的可能性,这意味着对于大型数据集,您可以编写自己的 Sequence 实现,并从磁盘上懒加载时间序列。不过,这通常会带来较高的 I/O 成本。因此,在训练多个序列时,首先尝试提前构建一个简单的 List[TimeSeries],看看它是否能保持在计算机内存中。

不要使用 所有 可能的子系列进行训练

默认情况下,当调用 fit() 时,Darts 中的模型会构建一个适合您正在使用的模型的 TrainingDataset 实例(例如,PastCovariatesTorchModelFutureCovariatesTorchModel 等)。默认情况下,这些训练数据集通常会包含每个 TimeSeries 中所有可能的连续(输入,输出)子序列。如果您的 TimeSeries 很长,这可能会导致大量的训练样本,这些样本直接(线性)影响训练模型一个 epoch 所需的时间。您有两种方法来限制这一点:

  • fit() 函数指定一些 max_samples_per_ts 参数。这将仅使用每个 TimeSeries 最近的 max_samples_per_ts 个样本进行训练。

  • 如果这个选项不能满足你的需求,你可以实现自己的 TrainingDataset 实例,并定义如何切片你的 TimeSeries 进行训练。我们建议查看 这个子模块 以查看如何操作的示例。


示例基准测试

作为一个示例,我们在这里展示在能量数据集(darts.datasets.EnergyDataset)的前80%上训练一个epoch所需的时间,该数据集由一个长度为28050个时间步且具有28个维度的多元序列组成。我们训练了两个模型:NBEATSModelTFTModel,使用默认参数,并且设置 input_chunk_length=48output_chunk_length=12``(这导致在默认的顺序训练数据集中有27991个训练样本)。对于TFT模型,我们还设置了参数 ``add_cyclic_encoder='hour'。测试在一台Intel CPU i9-10900K CPU @ 3.70GHz的计算机上进行,配备Nvidia RTX 2080s GPU和32 GB内存。所有 TimeSeries 都预先加载到内存中,并以列表形式提供给模型。

模型

数据集

dtype

CUDA

批量大小

num workers

每个epoch的时间

NBEATSModel

能量

64

32

0

283秒

NBEATSModel

能量

64

32

2

285秒

NBEATSModel

能量

64

32

4

282秒

NBEATSModel

能量

64

1024

0

58秒

NBEATSModel

能量

64

1024

2

57秒

NBEATSModel

能量

64

1024

4

58秒

NBEATSModel

能量

64

是的

32

0

63秒

NBEATSModel

能量

64

是的

32

2

62秒

NBEATSModel

能量

64

是的

1024

0

13.3秒

NBEATSModel

能量

64

是的

1024

2

12.1秒

NBEATSModel

能量

64

是的

1024

4

12.3秒

NBEATSModel

能量

32

32

0

117秒

NBEATSModel

能量

32

32

2

115秒

NBEATSModel

能量

32

32

4

117秒

NBEATSModel

能量

32

1024

0

28.4秒

NBEATSModel

能量

32

1024

2

27.4秒

NBEATSModel

能量

32

1024

4

27.5秒

NBEATSModel

能量

32

是的

32

0

41.5秒

NBEATSModel

能量

32

是的

32

2

40.6秒

NBEATSModel

能量

32

是的

1024

0

2.8秒

NBEATSModel

能量

32

是的

1024

2

1.65

NBEATSModel

能量

32

是的

1024

4

1.8秒

TFTModel

能量

64

32

0

78秒

TFTModel

能量

64

32

2

72秒

TFTModel

能量

64

32

4

72秒

TFTModel

能量

64

1024

0

46秒

TFTModel

能量

64

1024

2

38秒

TFTModel

能量

64

1024

4

39秒

TFTModel

能量

64

是的

32

0

125秒

TFTModel

能量

64

是的

32

2

115秒

TFTModel

能量

64

是的

1024

0

59秒

TFTModel

能量

64

是的

1024

2

50年代

TFTModel

能量

64

是的

1024

4

50年代

TFTModel

能量

32

32

0

70年代

TFTModel

能量

32

32

2

62.6秒

TFTModel

能量

32

32

4

63.6

TFTModel

能量

32

1024

0

31.9秒

TFTModel

能量

32

1024

2

45秒

TFTModel

能量

32

1024

4

44秒

TFTModel

能量

32

是的

32

0

73s

TFTModel

能量

32

是的

32

2

58秒

TFTModel

能量

32

是的

1024

0

41秒

TFTModel

能量

32

是的

1024

2

31秒

TFTModel

能量

32

是的

1024

4

31秒