在 Hugging Face 中使用 Patch Time Series Transformer - 入门指南

在本博客中，我们提供了一些如何开始使用 PatchTST 的示例。我们首先在 Electricity 数据上演示 PatchTST 的预测能力。然后，我们将通过使用先前训练好的模型在电力变压器 (ETTh1) 数据集上进行零样本预测，来展示 PatchTST 的迁移学习能力。零样本预测性能将表示模型在 目标 域中的 测试 性能，而无需在目标域上进行任何训练。随后，我们将在目标数据的 训练 部分对预训练模型进行线性探测和（然后）微调，并将在目标数据的 测试 部分验证预测性能。

PatchTST 模型由 Yuqi Nie、Nam H. Nguyen、Phanwadee Sinthong 和 Jayant Kalagnanam 在论文《A Time Series is Worth 64 Words: Long-term Forecasting with Transformers》中提出，并于 ICLR 2023 发表。

PatchTST 快速概览

从高层次来看，该模型将批次中的单个时间序列向量化为给定大小的补丁 (patch)，并通过一个 Transformer 编码器对生成的向量序列进行编码，然后通过一个合适的头部 (head) 输出预测长度的预测值。

该模型基于两个关键组成部分

1. 将时间序列分割成子序列级别的补丁，这些补丁作为 Transformer 的输入词元 (token)；
2. 通道独立性 (channel-independence)，其中每个通道包含一个单变量时间序列，所有序列共享相同的嵌入和 Transformer 权重，即一个全局单变量模型。

补丁设计天然具有三重好处

• 局部语义信息在嵌入中得以保留；
• 在给定相同回溯窗口的情况下，通过补丁之间的步幅，注意力图的计算和内存使用量呈二次方级减少；以及
• 模型可以通过权衡补丁长度（输入向量大小）和上下文长度（序列数量）来关注更长的历史记录。

此外，PatchTST 采用模块化设计，无缝支持掩码时间序列预训练以及直接时间序列预测。

安装

此演示需要 Hugging Face Transformers 来获取模型，以及 IBM tsfm 包用于辅助数据预处理。我们可以通过克隆 tsfm 仓库并按照以下步骤来安装两者。

1. 克隆公开的 IBM 时间序列基础模型仓库 tsfm。

   pip install git+https://github.com/IBM/tsfm.git
   

2. 安装 Hugging Face Transformers

   pip install transformers
   

3. 在 python 终端中使用以下命令进行测试。

   from transformers import PatchTSTConfig
   from tsfm_public.toolkit.dataset import ForecastDFDataset
   

第 1 部分：在 Electricity 数据集上进行预测

在这里，我们直接在 Electricity 数据集（可从 https://github.com/zhouhaoyi/Informer2020 获取）上训练一个 PatchTST 模型，并评估其性能。

# Standard
import os

# Third Party
from transformers import (
    EarlyStoppingCallback,
    PatchTSTConfig,
    PatchTSTForPrediction,
    Trainer,
    TrainingArguments,
)
import numpy as np
import pandas as pd

# First Party
from tsfm_public.toolkit.dataset import ForecastDFDataset
from tsfm_public.toolkit.time_series_preprocessor import TimeSeriesPreprocessor
from tsfm_public.toolkit.util import select_by_index

设置随机种子

from transformers import set_seed

set_seed(2023)

加载并准备数据集

在下一个单元格中，请根据您的应用调整以下参数

• dataset_path：本地 .csv 文件的路径，或目标数据的 csv 文件的网址。数据使用 pandas 加载，因此 pd.read_csv 支持的任何格式都受支持：(https://pandas.ac.cn/pandas-docs/stable/reference/api/pandas.read_csv.html)。
• timestamp_column：包含时间戳信息的列名，如果没有此列，请使用 None。
• id_columns：指定不同时间序列 ID 的列名列表。如果不存在 ID 列，请使用 []。
• forecast_columns：需要建模的列的列表
• context_length：用作模型输入的历史数据量。将从输入数据框中提取长度等于 context_length 的输入时间序列数据窗口。对于多时间序列数据集，将创建上下文窗口，使其包含在单个时间序列（即单个 ID）内。
• forecast_horizon：未来要预测的时间戳数量。
• train_start_index, train_end_index：加载数据中用于划分训练数据的起始和结束索引。
• valid_start_index, eval_end_index：加载数据中用于划分验证数据的起始和结束索引。
• test_start_index, eval_end_index：加载数据中用于划分测试数据的起始和结束索引。
• patch_length：PatchTST 模型的补丁长度。建议选择一个能被 context_length 整除的值。
• num_workers：PyTorch 数据加载器中的 CPU 工作进程数。
• batch_size：批次大小。

数据首先被加载到 Pandas 数据框中，并被分割为训练、验证和测试部分。然后，Pandas 数据框被转换为训练所需的适当 PyTorch 数据集。

# The ECL data is available from https://github.com/zhouhaoyi/Informer2020?tab=readme-ov-file#data
dataset_path = "~/data/ECL.csv"
timestamp_column = "date"
id_columns = []

context_length = 512
forecast_horizon = 96
patch_length = 16
num_workers = 16  # Reduce this if you have low number of CPU cores
batch_size = 64  # Adjust according to GPU memory

data = pd.read_csv(
    dataset_path,
    parse_dates=[timestamp_column],
)
forecast_columns = list(data.columns[1:])

# get split
num_train = int(len(data) * 0.7)
num_test = int(len(data) * 0.2)
num_valid = len(data) - num_train - num_test
border1s = [
    0,
    num_train - context_length,
    len(data) - num_test - context_length,
]
border2s = [num_train, num_train + num_valid, len(data)]

train_start_index = border1s[0]  # None indicates beginning of dataset
train_end_index = border2s[0]

# we shift the start of the evaluation period back by context length so that
# the first evaluation timestamp is immediately following the training data
valid_start_index = border1s[1]
valid_end_index = border2s[1]

test_start_index = border1s[2]
test_end_index = border2s[2]

train_data = select_by_index(
    data,
    id_columns=id_columns,
    start_index=train_start_index,
    end_index=train_end_index,
)
valid_data = select_by_index(
    data,
    id_columns=id_columns,
    start_index=valid_start_index,
    end_index=valid_end_index,
)
test_data = select_by_index(
    data,
    id_columns=id_columns,
    start_index=test_start_index,
    end_index=test_end_index,
)

time_series_preprocessor = TimeSeriesPreprocessor(
    timestamp_column=timestamp_column,
    id_columns=id_columns,
    input_columns=forecast_columns,
    output_columns=forecast_columns,
    scaling=True,
)
time_series_preprocessor = time_series_preprocessor.train(train_data)

train_dataset = ForecastDFDataset(
    time_series_preprocessor.preprocess(train_data),
    id_columns=id_columns,
    timestamp_column="date",
    input_columns=forecast_columns,
    output_columns=forecast_columns,
    context_length=context_length,
    prediction_length=forecast_horizon,
)
valid_dataset = ForecastDFDataset(
    time_series_preprocessor.preprocess(valid_data),
    id_columns=id_columns,
    timestamp_column="date",
    input_columns=forecast_columns,
    output_columns=forecast_columns,
    context_length=context_length,
    prediction_length=forecast_horizon,
)
test_dataset = ForecastDFDataset(
    time_series_preprocessor.preprocess(test_data),
    id_columns=id_columns,
    timestamp_column="date",
    input_columns=forecast_columns,
    output_columns=forecast_columns,
    context_length=context_length,
    prediction_length=forecast_horizon,
)

配置 PatchTST 模型

接下来，我们使用一个配置实例化一个随机初始化的 PatchTST 模型。以下设置控制了与架构相关的不同超参数。

• num_input_channels：时间序列数据中的输入通道（或维度）数量。这个值会自动设置为预测列的数量。
• context_length：如上所述，用作模型输入的历史数据量。
• patch_length：从上下文窗口（长度为 context_length）中提取的补丁的长度。
• patch_stride：从上下文窗口提取补丁时使用的步幅。
• random_mask_ratio：为预训练模型而完全掩码的输入补丁的比例。
• d_model：Transformer 层的维度。
• num_attention_heads：Transformer 编码器中每个注意力层的注意力头数量。
• num_hidden_layers：编码器层的数量。
• ffn_dim：编码器中中间层（通常称为前馈层）的维度。
• dropout：编码器中所有全连接层的丢弃概率。
• head_dropout：模型头部中使用的丢弃概率。
• pooling_type：嵌入的池化方式。支持 "mean"、"max" 和 None。
• channel_attention：激活 Transformer 中的通道注意力模块，以允许通道之间相互关注。
• scaling：是否通过 "mean" 缩放器、"std" 缩放器对输入目标进行缩放，如果为 None 则不进行缩放。如果为 True，缩放器设置为 "mean"。
• loss：对应于 distribution_output 头的模型损失函数。对于参数分布，它是负对数似然 ("nll")，对于点估计，它是均方误差 "mse"。
• pre_norm：如果 pre_norm 设置为 True，则在自注意力之前应用归一化。否则，在残差块之后应用归一化。
• norm_type：每个 Transformer 层的归一化类型。可以是 "BatchNorm" 或 "LayerNorm"。

有关参数的完整详细信息，请参阅文档。

config = PatchTSTConfig(
    num_input_channels=len(forecast_columns),
    context_length=context_length,
    patch_length=patch_length,
    patch_stride=patch_length,
    prediction_length=forecast_horizon,
    random_mask_ratio=0.4,
    d_model=128,
    num_attention_heads=16,
    num_hidden_layers=3,
    ffn_dim=256,
    dropout=0.2,
    head_dropout=0.2,
    pooling_type=None,
    channel_attention=False,
    scaling="std",
    loss="mse",
    pre_norm=True,
    norm_type="batchnorm",
)
model = PatchTSTForPrediction(config)

训练模型

接下来，我们可以利用 Hugging Face 的 Trainer 类，基于直接预测策略来训练模型。我们首先定义 TrainingArguments，其中列出了用于训练的各种超参数，例如训练周期数、学习率等。

training_args = TrainingArguments(
    output_dir="./checkpoint/patchtst/electricity/pretrain/output/",
    overwrite_output_dir=True,
    # learning_rate=0.001,
    num_train_epochs=100,
    do_eval=True,
    evaluation_strategy="epoch",
    per_device_train_batch_size=batch_size,
    per_device_eval_batch_size=batch_size,
    dataloader_num_workers=num_workers,
    save_strategy="epoch",
    logging_strategy="epoch",
    save_total_limit=3,
    logging_dir="./checkpoint/patchtst/electricity/pretrain/logs/",  # Make sure to specify a logging directory
    load_best_model_at_end=True,  # Load the best model when training ends
    metric_for_best_model="eval_loss",  # Metric to monitor for early stopping
    greater_is_better=False,  # For loss
    label_names=["future_values"],
)

# Create the early stopping callback
early_stopping_callback = EarlyStoppingCallback(
    early_stopping_patience=10,  # Number of epochs with no improvement after which to stop
    early_stopping_threshold=0.0001,  # Minimum improvement required to consider as improvement
)

# define trainer
trainer = Trainer(
    model=model,
    args=training_args,
    train_dataset=train_dataset,
    eval_dataset=valid_dataset,
    callbacks=[early_stopping_callback],
    # compute_metrics=compute_metrics,
)

# pretrain
trainer.train()

轮次	训练损失	验证损失
1	0.455400	0.215057
2	0.241000	0.179336
3	0.209000	0.158522
...	...	...
83	0.128000	0.111213

在源域的测试集上评估模型

接下来，我们可以利用 trainer.evaluate() 来计算测试指标。虽然这不是此任务中要判断的目标指标，但它提供了一个合理的检查，以确保预训练模型已正确训练。请注意，PatchTST 的训练和评估损失是均方误差 (MSE) 损失。因此，在以下任何评估实验中，我们都不再单独计算 MSE 指标。

results = trainer.evaluate(test_dataset)
print("Test result:")
print(results)

>>> Test result:
    {'eval_loss': 0.1316315233707428, 'eval_runtime': 5.8077, 'eval_samples_per_second': 889.332, 'eval_steps_per_second': 3.616, 'epoch': 83.0}

0.131 的 MSE 值与原始 PatchTST 论文中报告的 Electricity 数据集的值非常接近。

保存模型

save_dir = "patchtst/electricity/model/pretrain/"
os.makedirs(save_dir, exist_ok=True)
trainer.save_model(save_dir)

第 2 部分：从 Electricity 到 ETTh1 的迁移学习

在本节中，我们将展示 PatchTST 模型的迁移学习能力。我们使用在 Electricity 数据集上预训练的模型，在 ETTh1 数据集上进行零样本预测。

所谓迁移学习，是指我们首先在一个 源 数据集（如我们上面在 Electricity 数据集上所做的）上为一个预测任务预训练模型。然后，我们将使用预训练的模型在一个 目标 数据集上进行零样本预测。所谓零样本，是指我们在 目标 域中测试性能，而无需任何额外的训练。我们希望模型从预训练中获得了足够的知识，可以迁移到另一个不同的数据集。随后，我们将在目标数据的 训练 部分对预训练模型进行线性探测和（然后）微调，并将在目标数据的 测试 部分验证预测性能。在本例中，源数据集是 Electricity 数据集，目标数据集是 ETTh1。

在 ETTh1 数据上进行迁移学习。

所有评估都在 ETTh1 数据的 测试 部分进行。

步骤 1：直接评估在 electricity 上预训练的模型。这是零样本性能。

步骤 2：在进行线性探测后进行评估。

步骤 3：在进行全量微调后进行评估。

加载 ETTh 数据集

下面，我们将 ETTh1 数据集加载为 Pandas 数据框。接下来，我们创建 3 个分割：训练、验证和测试。然后，我们利用 TimeSeriesPreprocessor 类为模型准备每个分割。

dataset = "ETTh1"

print(f"Loading target dataset: {dataset}")
dataset_path = f"https://raw.githubusercontent.com/zhouhaoyi/ETDataset/main/ETT-small/{dataset}.csv"
timestamp_column = "date"
id_columns = []
forecast_columns = ["HUFL", "HULL", "MUFL", "MULL", "LUFL", "LULL", "OT"]
train_start_index = None  # None indicates beginning of dataset
train_end_index = 12 * 30 * 24

# we shift the start of the evaluation period back by context length so that
# the first evaluation timestamp is immediately following the training data
valid_start_index = 12 * 30 * 24 - context_length
valid_end_index = 12 * 30 * 24 + 4 * 30 * 24

test_start_index = 12 * 30 * 24 + 4 * 30 * 24 - context_length
test_end_index = 12 * 30 * 24 + 8 * 30 * 24

>>> Loading target dataset: ETTh1

data = pd.read_csv(
    dataset_path,
    parse_dates=[timestamp_column],
)

train_data = select_by_index(
    data,
    id_columns=id_columns,
    start_index=train_start_index,
    end_index=train_end_index,
)
valid_data = select_by_index(
    data,
    id_columns=id_columns,
    start_index=valid_start_index,
    end_index=valid_end_index,
)
test_data = select_by_index(
    data,
    id_columns=id_columns,
    start_index=test_start_index,
    end_index=test_end_index,
)

time_series_preprocessor = TimeSeriesPreprocessor(
    timestamp_column=timestamp_column,
    id_columns=id_columns,
    input_columns=forecast_columns,
    output_columns=forecast_columns,
    scaling=True,
)
time_series_preprocessor = time_series_preprocessor.train(train_data)

train_dataset = ForecastDFDataset(
    time_series_preprocessor.preprocess(train_data),
    id_columns=id_columns,
    input_columns=forecast_columns,
    output_columns=forecast_columns,
    context_length=context_length,
    prediction_length=forecast_horizon,
)
valid_dataset = ForecastDFDataset(
    time_series_preprocessor.preprocess(valid_data),
    id_columns=id_columns,
    input_columns=forecast_columns,
    output_columns=forecast_columns,
    context_length=context_length,
    prediction_length=forecast_horizon,
)
test_dataset = ForecastDFDataset(
    time_series_preprocessor.preprocess(test_data),
    id_columns=id_columns,
    input_columns=forecast_columns,
    output_columns=forecast_columns,
    context_length=context_length,
    prediction_length=forecast_horizon,
)

在 ETTh 上的零样本预测

由于我们将测试开箱即用的预测性能，我们加载了上面预训练的模型。

finetune_forecast_model = PatchTSTForPrediction.from_pretrained(
    "patchtst/electricity/model/pretrain/",
    num_input_channels=len(forecast_columns),
    head_dropout=0.7,
)

finetune_forecast_args = TrainingArguments(
    output_dir="./checkpoint/patchtst/transfer/finetune/output/",
    overwrite_output_dir=True,
    learning_rate=0.0001,
    num_train_epochs=100,
    do_eval=True,
    evaluation_strategy="epoch",
    per_device_train_batch_size=batch_size,
    per_device_eval_batch_size=batch_size,
    dataloader_num_workers=num_workers,
    report_to="tensorboard",
    save_strategy="epoch",
    logging_strategy="epoch",
    save_total_limit=3,
    logging_dir="./checkpoint/patchtst/transfer/finetune/logs/",  # Make sure to specify a logging directory
    load_best_model_at_end=True,  # Load the best model when training ends
    metric_for_best_model="eval_loss",  # Metric to monitor for early stopping
    greater_is_better=False,  # For loss
    label_names=["future_values"],
)

# Create a new early stopping callback with faster convergence properties
early_stopping_callback = EarlyStoppingCallback(
    early_stopping_patience=10,  # Number of epochs with no improvement after which to stop
    early_stopping_threshold=0.001,  # Minimum improvement required to consider as improvement
)

finetune_forecast_trainer = Trainer(
    model=finetune_forecast_model,
    args=finetune_forecast_args,
    train_dataset=train_dataset,
    eval_dataset=valid_dataset,
    callbacks=[early_stopping_callback],
)

print("\n\nDoing zero-shot forecasting on target data")
result = finetune_forecast_trainer.evaluate(test_dataset)
print("Target data zero-shot forecasting result:")
print(result)

>>> Doing zero-shot forecasting on target data

    Target data zero-shot forecasting result:
    {'eval_loss': 0.3728715181350708, 'eval_runtime': 0.95, 'eval_samples_per_second': 2931.527, 'eval_steps_per_second': 11.579}

可以看到，通过零样本预测方法，我们获得了 0.370 的 MSE，这接近于原始 PatchTST 论文中的最先进结果。

接下来，让我们看看通过执行线性探测能达到什么效果，这涉及到在冻结的预训练模型之上训练一个线性层。线性探测通常用于测试预训练模型特征的性能。

在 ETTh1 上进行线性探测

我们可以在目标数据的 训练 部分进行快速线性探测，看看是否有任何可能的 测试 性能提升。

# Freeze the backbone of the model
for param in finetune_forecast_trainer.model.model.parameters():
    param.requires_grad = False

print("\n\nLinear probing on the target data")
finetune_forecast_trainer.train()
print("Evaluating")
result = finetune_forecast_trainer.evaluate(test_dataset)
print("Target data head/linear probing result:")
print(result)

>>> Linear probing on the target data

轮次	训练损失	验证损失
1	0.384600	0.688319
2	0.374200	0.678159
3	0.368400	0.667633
...	...	...

>>> Evaluating

    Target data head/linear probing result:
    {'eval_loss': 0.35652095079421997, 'eval_runtime': 1.1537, 'eval_samples_per_second': 2413.986, 'eval_steps_per_second': 9.535, 'epoch': 18.0}

可以看到，仅在冻结的主干网络上训练一个简单的线性层，MSE 就从 0.370 降至 0.357，超过了最初报告的结果！

save_dir = f"patchtst/electricity/model/transfer/{dataset}/model/linear_probe/"
os.makedirs(save_dir, exist_ok=True)
finetune_forecast_trainer.save_model(save_dir)

save_dir = f"patchtst/electricity/model/transfer/{dataset}/preprocessor/"
os.makedirs(save_dir, exist_ok=True)
time_series_preprocessor = time_series_preprocessor.save_pretrained(save_dir)

最后，让我们看看通过对模型进行全量微调是否可以获得额外的改进。

在 ETTh1 上进行全量微调

我们可以在目标数据的 训练 部分进行全模型微调（而不是像上面那样探测最后一个线性层），以查看可能的 测试 性能提升。代码看起来与上面的线性探测任务类似，只是我们没有冻结任何参数。

# Reload the model
finetune_forecast_model = PatchTSTForPrediction.from_pretrained(
    "patchtst/electricity/model/pretrain/",
    num_input_channels=len(forecast_columns),
    dropout=0.7,
    head_dropout=0.7,
)
finetune_forecast_trainer = Trainer(
    model=finetune_forecast_model,
    args=finetune_forecast_args,
    train_dataset=train_dataset,
    eval_dataset=valid_dataset,
    callbacks=[early_stopping_callback],
)
print("\n\nFinetuning on the target data")
finetune_forecast_trainer.train()
print("Evaluating")
result = finetune_forecast_trainer.evaluate(test_dataset)
print("Target data full finetune result:")
print(result)

>>> Finetuning on the target data

轮次	训练损失	验证损失
1	0.348600	0.709915
2	0.328800	0.706537
3	0.319700	0.741892
...	...	...

>>> Evaluating

    Target data full finetune result:
    {'eval_loss': 0.354232519865036, 'eval_runtime': 1.0715, 'eval_samples_per_second': 2599.18, 'eval_steps_per_second': 10.266, 'epoch': 12.0}

在这种情况下，在 ETTh1 数据集上，全量微调只带来了微小的改进。对于其他数据集，可能会有更显著的改进。无论如何，我们还是保存模型。

save_dir = f"patchtst/electricity/model/transfer/{dataset}/model/fine_tuning/"
os.makedirs(save_dir, exist_ok=True)
finetune_forecast_trainer.save_model(save_dir)

总结

在本博客中，我们提供了一个关于训练 PatchTST 以完成预测和迁移学习任务的逐步指南，并演示了各种微调方法。我们旨在促进 PatchTST HF 模型轻松集成到您的预测用例中，并希望这些内容能成为加速采用 PatchTST 的有用资源。感谢您关注我们的博客，希望您觉得这些信息对您的项目有益。