在 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.pydata.org/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 的有用資源。感謝您關注我們的部落格，希望您覺得這些資訊對您的專案有益。