Python与时间序列预测：如何使用`PyTorch Forecasting`进行深度学习预测。

Python与时间序列预测：如何使用PyTorch Forecasting进行深度学习预测

大家好！今天我们将深入探讨如何使用 PyTorch Forecasting 库进行深度学习时间序列预测。时间序列预测在各个领域都至关重要，例如金融、销售预测、能源需求预测等。传统的统计方法，如ARIMA模型，在某些情况下表现出色，但当数据复杂且非线性时，深度学习模型往往能取得更好的效果。PyTorch Forecasting 提供了一个高级接口，简化了使用 PyTorch 构建和训练各种时间序列模型的流程。

1. 为什么选择 PyTorch Forecasting?

PyTorch Forecasting 具有以下几个关键优势：

• 易用性： 提供高级 API，简化了模型构建、训练和评估过程。
• 灵活性： 基于 PyTorch 构建，可以自定义模型结构和训练流程。
• 可解释性： 提供工具来分析模型预测的原因和影响因素。
• 多种模型： 支持多种深度学习模型，例如 Temporal Fusion Transformer (TFT)，N-BEATS， DeepAR等。
• 内置数据处理： 提供了处理时间序列数据的常用功能，例如缺失值处理，数据缩放等。

2. 环境搭建与数据准备

首先，我们需要安装 PyTorch Forecasting 和其他必要的库：

pip install pytorch-forecasting pytorch-lightning pandas matplotlib scikit-learn

接下来，我们将使用一个示例数据集，比如来自 Kaggle 的“Store Sales – Time Series Forecasting”数据集（为了简化，我们将使用一个模拟数据集）。这个数据集包含不同商店的销售数据。
这里我们生成一个模拟数据集，模拟不同商店的销售额数据：

import pandas as pd
import numpy as np

def generate_sales_data(n_stores=3, n_days=365*2):
    """Generates simulated sales data for multiple stores."""
    store_ids = [f"store_{i+1}" for i in range(n_stores)]
    dates = pd.date_range(start="2022-01-01", periods=n_days)
    data = []

    for store_id in store_ids:
        # Base sales level (different for each store)
        base_sales = np.random.randint(50, 200)

        # Seasonal pattern (e.g., higher sales in summer/winter)
        seasonal_amplitude = np.random.uniform(0.1, 0.3)  # Random amplitude for each store
        seasonal_pattern = np.sin(2 * np.pi * np.arange(n_days) / 365) * seasonal_amplitude * base_sales

        # Trend (gradual increase or decrease over time)
        trend = np.linspace(0, np.random.uniform(-0.05, 0.05) * base_sales * n_days, n_days) # Random upward or downward trend

        # Random noise
        noise = np.random.normal(0, 0.1 * base_sales, n_days)

        # Combine all components
        sales = base_sales + seasonal_pattern + trend + noise
        sales = np.maximum(sales, 0)  # Ensure sales are non-negative

        store_data = pd.DataFrame({
            "date": dates,
            "store": store_id,
            "sales": sales
        })
        data.append(store_data)

    df = pd.concat(data)
    df['date'] = pd.to_datetime(df['date'])
    df["time_idx"] = (df["date"] - df["date"].min()).dt.days
    df["month"] = df.date.dt.month.astype(str)
    return df

# Generate the data
data = generate_sales_data(n_stores=5)
print(data.head())

这段代码生成了一个包含 date（日期）、store（商店 ID）和 sales（销售额）列的 Pandas DataFrame。 time_idx 是一个从0开始的索引，表示时间步长。 month 表示月份。

3. 数据预处理

在使用数据之前，我们需要进行一些预处理步骤：

• 缺失值处理： 检查并处理缺失值。
• 数据缩放： 将数据缩放到一个合适的范围，例如 [0, 1] 或使用标准化。
• 创建时间序列特征： 创建有用的时间序列特征，例如月份、星期几、年份等。
• 定义训练集和验证集： 将数据分为训练集和验证集。

from sklearn.preprocessing import MinMaxScaler

# 1. 缺失值处理 (简单起见，我们假设没有缺失值)
# data.isna().sum()  # 检查缺失值
# data = data.dropna()  # 如果有缺失值，删除包含缺失值的行

# 2. 数据缩放
scaler = MinMaxScaler()
data['sales'] = scaler.fit_transform(data[['sales']])

# 3. 创建时间序列特征 (已在数据生成中完成一部分)
# data['month'] = data['date'].dt.month

# 4. 定义训练集和验证集
max_prediction_length = 30 # 预测30天
max_encoder_length = 365 # 使用过去365天的数据进行编码
training_cutoff = data["time_idx"].max() - max_prediction_length

train_data = data[lambda x: x.time_idx <= training_cutoff]
validation_data = data[lambda x: (x.time_idx > training_cutoff) & (x.time_idx <= data["time_idx"].max())]

# 打印数据集大小
print(f"Train size: {len(train_data)}")
print(f"Validation size: {len(validation_data)}")

我们使用 MinMaxScaler 将销售额缩放到 0 到 1 之间。 我们定义了 max_prediction_length 和 max_encoder_length，分别表示预测的未来天数和用于编码历史数据的天数。 training_cutoff 用于将数据分为训练集和验证集。

4. 定义数据加载器

PyTorch Forecasting 使用 TimeSeriesDataSet 类来定义数据集。我们需要指定时间 ID、组 ID、目标变量和已知输入的列。

from pytorch_forecasting import TimeSeriesDataSet

# 定义训练数据集
training = TimeSeriesDataSet(
    train_data,
    group_ids=["store"],
    target="sales",
    time_idx="time_idx",
    min_encoder_length=max_encoder_length // 2,  # keep encoder length long (at least half of max_encoder_length)
    max_encoder_length=max_encoder_length,
    min_prediction_length=1,
    max_prediction_length=max_prediction_length,
    static_categoricals=["store"],
    time_varying_known_reals=["time_idx"],
    time_varying_unknown_reals=['sales'],
    categorical_encoders={"store": None}, # use LabelEncoder to encode categorical variables
    add_relative_time_idx=True,
    add_target_scales=True,
    add_encoder_length=True,
)

# 定义验证数据集
validation = TimeSeriesDataSet(
    validation_data,
    group_ids=["store"],
    target="sales",
    time_idx="time_idx",
    min_encoder_length=max_encoder_length // 2,
    max_encoder_length=max_encoder_length,
    min_prediction_length=1,
    max_prediction_length=max_prediction_length,
    static_categoricals=["store"],
    time_varying_known_reals=["time_idx"],
    time_varying_unknown_reals=['sales'],
    categorical_encoders={"store": None},
    add_relative_time_idx=True,
    add_target_scales=True,
    add_encoder_length=True,
)

from torch.utils.data import DataLoader

# 创建数据加载器
batch_size = 64  # 调整批次大小
train_dataloader = training.to_dataloader(train=True, batch_size=batch_size, num_workers=0)
val_dataloader = validation.to_dataloader(train=False, batch_size=batch_size, num_workers=0)

我们创建了 TimeSeriesDataSet 对象，并指定了 group_ids（商店 ID），target（销售额），time_idx（时间索引），known_reals（已知的实数输入），unknown_reals（未知的实数输入）。我们还使用 DataLoader 创建了数据加载器，用于在训练期间批量加载数据。

5. 模型选择与训练

PyTorch Forecasting 提供了多种深度学习模型。这里我们选择 Temporal Fusion Transformer (TFT) 模型，因为它在各种时间序列预测任务中表现出色。

from pytorch_forecasting import TemporalFusionTransformer, QuantileLoss
import pytorch_lightning as pl
from pytorch_lightning.callbacks import EarlyStopping, LearningRateMonitor
import torch

# 配置训练参数
early_stop_callback = EarlyStopping(monitor="val_loss", min_delta=1e-4, patience=10, verbose=False, mode="min")
lr_logger = LearningRateMonitor()  # log the learning rate
trainer = pl.Trainer(
    max_epochs=30,
    callbacks=[lr_logger, early_stop_callback],
    gradient_clip_val=0.1,
    accelerator="auto", # 使用gpu or cpu
    # limit_train_batches=30,  # comment in for training, running val step every 20 batches
)

# 定义模型
tft = TemporalFusionTransformer.from_dataset(
    training,
    learning_rate=0.03,
    hidden_size=16,
    attention_head_size=1,
    dropout=0.1,
    hidden_continuous_size=8,
    output_size=7,  # 定义分位数
    loss=QuantileLoss(),
    optimizer="RAdam",
)
print(tft)

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

# 保存模型
best_model_path = trainer.checkpoint_callback.best_model_path
best_tft = TemporalFusionTransformer.load_from_checkpoint(best_model_path)

这段代码首先定义了 EarlyStopping 回调函数，用于在验证损失停止改善时提前停止训练。然后，我们创建了一个 TemporalFusionTransformer 模型，并指定了各种超参数，例如 hidden_size（隐藏层大小），attention_head_size（注意力头大小），dropout（dropout 率）和 loss（损失函数）。 我们使用 QuantileLoss 作为损失函数，这允许我们预测不同的分位数。 最后，我们使用 trainer.fit() 方法开始训练模型。

6. 模型评估与预测

训练完成后，我们可以使用训练好的模型进行预测并评估其性能。

from pytorch_forecasting import TimeSeriesDataSet, GroupNormalizer
from pytorch_forecasting.metrics import MAE, SMAPE, PoissonLoss, QuantileLoss
from pytorch_forecasting.models.baseline import Baseline

# 评估基线模型
actuals = torch.cat([y[0] for x, y in iter(val_dataloader)])
baseline_predictions = Baseline().predict(val_dataloader)
MAE()(baseline_predictions, actuals)

# 预测验证集
predictions = best_tft.predict(val_dataloader)
MAE()(predictions, actuals)

# Raw predictions are a dictionary from which all quantiles can be extracted
raw_predictions, x = best_tft.predict(val_dataloader, mode="raw", return_x=True)

# 选择一个batch进行可视化
for i in range(10):
    best_tft.plot_prediction(x, raw_predictions, idx=i, add_loss_to_title=True);

我们首先使用 best_tft.predict() 方法对验证集进行预测。然后，我们使用 MAE（平均绝对误差）作为评估指标来评估模型的性能。 我们还展示了如何使用 plot_prediction() 方法可视化预测结果。

7. 自定义预测

除了对整个验证集进行预测外，我们还可以对特定时间段进行自定义预测。

# 选择一个商店进行预测
new_prediction_data = data.copy()[lambda x: x.time_idx > x["time_idx"].max() - max_encoder_length]
new_prediction_data["time_idx"] = list(range(new_prediction_data["time_idx"].max() + 1, new_prediction_data["time_idx"].max() + 1 + max_prediction_length))
new_prediction_data["sales"] = np.nan
new_prediction_data = pd.concat([data, new_prediction_data], ignore_index=True)

# 创建数据集和数据加载器
new_prediction_data_set = TimeSeriesDataSet.from_dataset(
    training,
    new_prediction_data,
    predict=True,
    stop_randomization=True,
)
new_prediction_data_loader = new_prediction_data_set.to_dataloader(train=False, batch_size=batch_size * 2, num_workers=0)

# 进行预测
predictions = best_tft.predict(new_prediction_data_loader)

# 将预测结果添加到数据中
actuals = torch.cat([y[0] for x, y in iter(new_prediction_data_loader)])
raw_predictions, x = best_tft.predict(new_prediction_data_loader, mode="raw", return_x=True)

# 可视化预测结果
for i in range(10):
    best_tft.plot_prediction(x, raw_predictions, idx=i, add_loss_to_title=True);

我们首先创建一个新的 DataFrame，其中包含要预测的未来时间步长的数据。 然后，我们使用 TimeSeriesDataSet.from_dataset() 方法从现有数据集创建新的数据集，并指定 predict=True。 最后，我们使用 best_tft.predict() 方法进行预测，并将预测结果添加到数据中。

8. 一些注意事项

• 超参数调整： TemporalFusionTransformer 模型有许多超参数，需要根据具体的数据集进行调整。
• 特征工程： 特征工程是时间序列预测的关键步骤。创建有用的时间序列特征可以显著提高模型的性能。
• 模型选择： PyTorch Forecasting 提供了多种深度学习模型。选择合适的模型取决于具体的数据集和预测任务。
• 数据质量： 数据质量对模型的性能至关重要。确保数据准确、完整和一致。
• 可解释性： 使用 PyTorch Forecasting 提供的工具来分析模型预测的原因和影响因素。

9. 总结和展望

PyTorch Forecasting 是一个功能强大的库，可以简化使用 PyTorch 构建和训练时间序列模型的流程。它提供了高级 API、多种模型和内置数据处理功能，使其成为时间序列预测的理想选择。 通过本讲座，您已经了解了如何使用 PyTorch Forecasting 进行深度学习时间序列预测。希望这些知识能帮助您在实际项目中取得成功！

10. 使用PyTorch Forecasting的简单步骤

总而言之，使用PyTorch Forecasting进行时间序列预测的主要步骤包括：数据准备和预处理，定义TimeSeriesDataSet，选择合适的模型（例如TemporalFusionTransformer），训练模型，以及使用训练好的模型进行预测和评估。

11. 深度学习进行时间序列预测的优势

相较于传统的时间序列方法，深度学习模型能够捕捉数据中的复杂非线性关系，并可以方便地集成各种特征，从而在许多情况下获得更好的预测结果。

12. 未来时间序列预测的发展趋势

时间序列预测的未来发展方向包括：更强大的模型架构，更好的可解释性方法，以及更自动化和易于使用的工具。