前言
系列专栏:【深度学习：算法项目实战】✨︎
涉及医疗健康、财经金融、商业零售、食品饮料、运动健身、交通运输、环境科学、社交媒体以及文本和图像处理等诸多领域，讨论了各种复杂的深度神经网络思想，如卷积神经网络、循环神经网络、生成对抗网络、门控循环单元、长短期记忆、自然语言处理、深度强化学习、大型语言模型和迁移学习。
在金融市场的分析中，股票价格预测一直是一个充满挑战且备受关注的领域。传统的时序预测方法，如ARIMA、LSTM等，虽然在一定程度上能够捕捉到时间序列数据的动态特性，但在处理复杂的非线性关系和长期依赖时往往力不从心。近年来，随着深度学习技术的快速发展，尤其是Transformer模型的出现，为时序预测问题提供了新的解决思路。
Transformer模型通过其独特的自注意力机制，能够有效地捕捉到时间序列数据中的长期依赖关系，这在股票价格预测等金融时序预测任务中显得尤为重要。然而，Transformer模型在处理局部依赖和时序信息方面可能不如LSTM等循环神经网络模型。因此，结合LSTM和Transformer的混合模型应运而生，旨在充分利用LSTM在处理时序信息和短期依赖方面的优势，以及Transformer在捕捉长期依赖关系方面的能力。
本文将介绍一种基于LSTM-Transformer混合模型的股票价格多变量时序预测方法，该方法结合了LSTM和Transformer的优点，旨在提高股票价格预测的准确性。我们将使用PyTorch框架来实现该模型，并通过实验验证其在股票价格预测任务中的有效性。希望通过本文的探讨，能够为金融市场的时序预测问题提供一些新的思路和解决方案。
1. 数据集介绍
AAPL股票数据集，是苹果公司（Apple Inc.）在股票市场上的交易数据集合。这些数据集包含了苹果公司的股票价格、交易量、市值等关键财务指标，是金融分析、量化交易、时间序列预测等领域的重要数据源。投资者可以通过分析AAPL股票数据集来评估苹果公司的基本面和市场表现，从而做出更为明智的投资决策。
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
import seaborn as sns
from sklearn.preprocessing import MinMaxScaler
from sklearn.model_selection import train_test_split
from sklearn.metrics import mean_absolute_error, \
mean_absolute_percentage_error, \
mean_squared_error, root_mean_squared_error, \
r2_score
import torch
import torch.nn as nn
from torch.utils.data import DataLoader, TensorDataset, Dataset
from torchinfo import summary
np.random.seed(0)
torch.manual_seed(0)
2. 数据预处理
使用 pandas.to_datetime 函数将标量、数组、Series 或 DataFrame/dict-like 转换为时间数据类型。
data = pd.read_csv('AAPL.csv')
print(type(data['Close'].iloc[0]),type(data['Date'].iloc[0]))
# Let's convert the data type of timestamp column to datatime format
data['Date'] = pd.to_datetime(data['Date'])
print(type(data['Close'].iloc[0]),type(data['Date'].iloc[0]))
# Selecting subset
cond_1 = data['Date'] >= '2021-04-23 00:00:00'
cond_2 = data['Date'] <= '2024-04-23 00:00:00'
data = data[cond_1 & cond_2].set_index('Date')
print(data.shape)
<class 'numpy.float64'> <class 'str'>
<class 'numpy.float64'> <class 'pandas._libs.tslibs.timestamps.Timestamp'>
(755, 6)
3. 数据可视化
通过 matplotlib 绘制收盘价格，收盘价是股票在正常交易日的最后交易价格，是投资者跟踪其长期表现的标准基准。通过图形可以快速识别股价的趋势走向。
# plt.style.available
plt.style.use('_mpl-gallery')
plt.figure(figsize=(18,6))
plt.title('Close Price History')
plt.plot(AAPL['Close'],label='AAPL')
plt.ylabel('Close Price USD ($)', fontsize=18)
plt.legend()
plt.show()
4. 特征工程
4.1 特征缩放（归一化）
MinMaxScaler() 函数是 scikit-learn 库中预处理模块的一个非常实用的工具，用于特征缩放，特别是将特征值缩放到一个指定的范围内，通常是[0, 1]。这种缩放方法对于许多机器学习算法来说是非常有用的，因为它可以帮助改善算法的收敛速度和性能，特别是在特征尺度差异较大的情况下。
# 使用选定的特征来训练模型
features = data.drop(['Adj Close', 'Volume'], axis=1)
target = data['Adj Close'].values.reshape(-1, 1)
# 创建 MinMaxScaler实例，对特征进行拟合和变换，生成NumPy数组
scaler = MinMaxScaler()
features_scaled = scaler.fit_transform(features)
target_scaled = scaler.fit_transform(target)
print(features_scaled.shape, target_scaled.shape)
4.2 构建时间序列数据
我们创建一个时间序列数据，时间步 time_steps 假设设置为30
time_steps = 30
X_list = []
y_list = []
for i in range(len(features_scaled) - time_steps):
X_list.append(features_scaled[i:i+time_steps])
y_list.append(target_scaled[i+time_steps])
X = np.array(X_list) # [samples, time_steps, num_features]
y = np.array(y_list) # [target]
上述代码的目的是进行时间序列数据的预处理，将原始的时间序列数据转换为适合机器学习模型输入的格式。具体来说，它通过滑动窗口的方式将时间序列数据分割成多个样本，每个样本包含一定数量的时间步 time_steps 的特征数据以及对应的一个目标值。time_steps：表示每个样本中包含的时间步数。它决定了模型在预测时考虑的历史数据长度。X_list：用于存储分割后的特征数据样本的列表。y_list：用于存储每个特征数据样本对应的目标值的列表。
X_list.append(features_scaled[i:i + time_steps])：将从当前位置 i 开始，长度为 time_steps 的特征数据切片添加到 X_list 中。这样就得到了一系列连续的时间步的特征数据样本。
y_list.append(target_scaled[i + time_steps])：将当前位置 i + time_steps 的目标值添加到 y_list 中。这个目标值对应于当前特征数据样本之后的一个时间步的目标值。
samples, time_steps, num_features = X.shape # 赋值
4.3 数据集划分
train_test_split 函数将数组或矩阵随机分成训练子集和测试子集。
X_train, X_valid,\
y_train, y_valid = train_test_split(X, y,
test_size=0.2,
random_state=45,
shuffle=False)
print(X_train.shape, X_valid.shape, y_train.shape, y_valid.shape)
以上代码中 random_state=45 设置了随机种子，以确保每次运行代码时分割结果的一致性。shuffle=False 表示在分割数据时不进行随机打乱。如果设置为True（默认值），则会在分割之前对数据进行随机打乱，这样可以增加数据的随机性，但时间序列数据具有连续性，所以设置为False。
4.4 数据加载器
# 将 NumPy数组转换为 tensor张量
X_train_tensor = torch.from_numpy(X_train).type(torch.Tensor)
X_valid_tensor = torch.from_numpy(X_valid).type(torch.Tensor)
y_train_tensor = torch.from_numpy(y_train).type(torch.Tensor).view(-1, 1)
y_valid_tensor = torch.from_numpy(y_valid).type(torch.Tensor).view(-1, 1)
print(X_train_tensor.shape, X_valid_tensor.shape, y_train_tensor.shape, y_valid_tensor.shape)
以上代码通过 train_test_split 划分得到的训练集和验证集中的特征数据 X_train、X_valid 以及标签数据 y_train、y_valid 从 numpy 数组转换为 PyTorch 的张量（tensor）类型。.type(torch.Tensor) 确保张量的数据类型为标准的 torch.Tensor 类型，.view(-1, 1) 对张量进行维度调整
class DataHandler(Dataset):
def __init__(self, X_train_tensor, y_train_tensor, X_valid_tensor, y_valid_tensor):
self.X_train_tensor = X_train_tensor
self.y_train_tensor = y_train_tensor
self.X_valid_tensor = X_valid_tensor
self.y_valid_tensor = y_valid_tensor
def __len__(self):
return len(self.X_train_tensor)
def __getitem__(self, idx):
sample = self.X_train_tensor[idx]
labels = self.y_train_tensor[idx]
return sample, labels
def train_loader(self):
train_dataset = TensorDataset(self.X_train_tensor, self.y_train_tensor)
return DataLoader(train_dataset, batch_size=32, shuffle=True)
def valid_loader(self):
valid_dataset = TensorDataset(self.X_valid_tensor, self.y_valid_tensor)
return DataLoader(valid_dataset, batch_size=32, shuffle=False)
在上述代码中，定义了一个名为 DataHandler 的类，它继承自 torch.utils.data.Dataset
__init__ 方法用于接收数据和标签。__len__ 方法返回数据集的长度。__getitem__ 方法根据给定的索引 idx 返回相应的数据样本和标签。
data_handler = DataHandler(X_train_tensor, y_train_tensor, X_valid_tensor, y_valid_tensor)
train_loader = data_handler.train_loader()
valid_loader = data_handler.valid_loader()
在上述代码中，创建了一个数据处理对象 data_handler，并通过该对象生成训练集数据加载器 train_loader 和验证集数据加载器valid_loader。通过这种方式，可以方便地管理和加载训练集和验证集数据，为深度学习模型的训练和评估提供了数据支持。
5. 构建时序模型（TSF）
5.1 构建LSTM-Transformer模型
class LSTM_Transformer(nn.Module):
def __init__(self, input_dim, hidden_dim, lstm_layers, transformer_heads, transformer_layers, output_dim, dropout=0.5):
super(LSTM_Transformer, self).__init__()
# LSTM 层
self.lstm = nn.LSTM(input_dim, hidden_dim, lstm_layers, batch_first=True)
# Transformer 编码器层
transformer_encoder_layer = nn.TransformerEncoderLayer(d_model=hidden_dim, nhead=transformer_heads, dim_feedforward=hidden_dim * 2, dropout=dropout, batch_first=True)
self.transformer_encoder = nn.TransformerEncoder(transformer_encoder_layer, num_layers=transformer_layers)
# 输出层
self.fc = nn.Linear(hidden_dim, output_dim)
def forward(self, x):
# LSTM 输出
lstm_out, _ = self.lstm(x)
# Transformer 输入
transformer_input = lstm_out
# Transformer 输出
transformer_out = self.transformer_encoder(transformer_input)
# 预测输出
output = self.fc(transformer_out[:, -1, :])
return output
5.2 实例化模型、损失函数和优化器
input_dim = num_features # 输入特征维度
hidden_dim = 64 # LSTM 和 Transformer 的隐藏维度
lstm_layers = 1 # LSTM 层数
transformer_heads = 8 # Transformer 头数
transformer_layers = 1 # Transformer 层数
output_dim = 1 # 输出维度
model = LSTM_Transformer(input_dim, hidden_dim, lstm_layers, transformer_heads, transformer_layers, output_dim)
criterion_mse = nn.MSELoss() # 定义均方误差损失函数
criterion_mae = nn.L1Loss() # 定义平均绝对误差损失
optimizer = torch.optim.Adam(model.parameters(), lr=0.0001) # 定义优化器
criterion_mse = nn.MSELoss()：
使用 PyTorch 中的nn.MSELoss定义了均方误差（Mean Squared Error）损失函数。均方误差是回归问题中常用的损失函数，它计算预测值与真实值之间的平方差的平均值。在时序预测中，MSE 损失可以衡量模型预测值与实际值之间的差异程度。criterion_mae = nn.L1Loss()：
定义了平均绝对误差（Mean Absolute Error）损失函数。平均绝对误差计算预测值与真实值之间的绝对差值的平均值。与 MSE 损失相比，MAE 损失对异常值不那么敏感。optimizer = torch.optim.Adam(model.parameters(), lr=0.0001)：- 使用 Adam 优化器来优化模型的参数。Adam 是一种自适应学习率的优化算法，结合了动量和 RMSProp 的优点，能够在训练过程中自动调整学习率，加快收敛速度。
model.parameters()表示要优化的模型参数，lr=0.0001是学习率，控制每次参数更新的步长。学习率的选择对于模型的训练效果很重要，过小的学习率可能导致收敛速度过慢，而过大的学习率可能导致模型无法收敛或振荡。
5.3 模型概要
# batch_size, seq_len(time_steps), input_size(in_channels)
summary(model, (32, time_steps, num_features))
===============================================================================================
Layer (type:depth-idx) Output Shape Param #
===============================================================================================
LSTM_Transformer [32, 1] --
├─LSTM: 1-1 [32, 30, 64] 17,920
├─TransformerEncoder: 1-2 [32, 30, 64] --
│ └─ModuleList: 2-1 -- --
│ │ └─TransformerEncoderLayer: 3-1 [32, 30, 64] 33,472
├─Linear: 1-3 [32, 1] 65
===============================================================================================
Total params: 51,457
Trainable params: 51,457
Non-trainable params: 0
Total mult-adds (Units.MEGABYTES): 17.21
===============================================================================================
Input size (MB): 0.02
Forward/backward pass size (MB): 0.49
Params size (MB): 0.07
Estimated Total Size (MB): 0.58
===============================================================================================
6. 模型训练与可视化
6.1 定义训练与评估函数
在模型训练之前，我们需先定义 train 函数来执行模型训练过程
def train(model, iterator, optimizer):
epoch_loss_mse = 0
epoch_loss_mae = 0
model.train() # 确保模型处于训练模式
for batch in iterator:
optimizer.zero_grad() # 清空梯度
inputs, targets = batch # 获取输入和目标值
outputs = model(inputs) # 前向传播
loss_mse = criterion_mse(outputs, targets) # 计算损失
loss_mae = criterion_mae(outputs, targets)
combined_loss = loss_mse + loss_mae # 可以根据需要调整两者的权重
combined_loss.backward()
optimizer.step()
epoch_loss_mse += loss_mse.item() # 累计损失
epoch_loss_mae += loss_mae.item()
average_loss_mse = epoch_loss_mse / len(iterator) # 计算平均损失
average_loss_mae = epoch_loss_mae / len(iterator)
return average_loss_mse, average_loss_mae
上述代码定义了一个名为 train 的函数，用于训练给定的模型。它接收模型、数据迭代器、优化器作为参数，并返回训练过程中的平均损失。
def evaluate(model, iterator):
epoch_loss_mse = 0
epoch_loss_mae = 0
model.eval() # 将模型设置为评估模式，例如关闭 Dropout 等
with torch.no_grad(): # 不需要计算梯度
for batch in iterator:
inputs, targets = batch
outputs = model(inputs) # 前向传播
loss_mse = criterion_mse(outputs, targets) # 计算损失
loss_mae = criterion_mae(outputs, targets)
epoch_loss_mse += loss_mse.item() # 累计损失
epoch_loss_mae += loss_mae.item()
return epoch_loss_mse / len(iterator), epoch_loss_mae / len(iterator)
上述代码定义了一个名为 evaluate 的函数，用于评估给定模型在给定数据迭代器上的性能。它接收模型、数据迭代器作为参数，并返回评估过程中的平均损失。这个函数通常在模型训练的过程中定期被调用，以监控模型在验证集或测试集上的性能。通过评估模型的性能，可以了解模型的泛化能力和训练的进展情况。
epoch = 1000
train_mselosses = []
valid_mselosses = []
train_maelosses = []
valid_maelosses = []
for epoch in range(epoch):
train_loss_mse, train_loss_mae = train(model, train_loader, optimizer)
valid_loss_mse, valid_loss_mae = evaluate(model, valid_loader)
train_mselosses.append(train_loss_mse)
valid_mselosses.append(valid_loss_mse)
train_maelosses.append(train_loss_mae)
valid_maelosses.append(valid_loss_mae)
print(f'Epoch: {epoch+1:02}, Train MSELoss: {train_loss_mse:.5f}, Train MAELoss: {train_loss_mae:.3f}, Val. MSELoss: {valid_loss_mse:.5f}, Val. MAELoss: {valid_loss_mae:.3f}')
上述代码主要进行了模型的训练和评估过程，并记录了每个 epoch 的训练和验证集上的均方误差损失(MSE Loss)和平均绝对误差损失(MAE Loss)。
Epoch: 1000, Train MSELoss: 0.00149, Train MAELoss: 0.029, Val. MSELoss: 0.00093, Val. MAELoss: 0.023
6.2 绘制训练与验证损失曲线
# 绘制 MSE损失图
plt.figure(figsize=(12, 5))
plt.subplot(1, 2, 1)
plt.plot(train_mselosses, label='Train MSELoss')
plt.plot(valid_mselosses, label='Validation MSELoss')
plt.xlabel('Epoch')
plt.ylabel('MSELoss')
plt.title('Train and Validation MSELoss')
plt.legend()
plt.grid(True)
# 绘制 MAE损失图
plt.subplot(1, 2, 2)
plt.plot(train_maelosses, label='Train MAELoss')
plt.plot(valid_maelosses, label='Validation MAELoss')
plt.xlabel('Epoch')
plt.ylabel('MAELoss')
plt.title('Train and Validation MAELoss')
plt.legend()
plt.grid(True)
plt.show()
7. 模型评估与可视化
7.1 构建预测函数
定义预测函数 prediction 方便调用
# 定义 prediction函数
def prediction(model, iterator):
all_targets = []
all_predictions = []
model.eval()
with torch.no_grad():
for batch in iterator:
inputs, targets = batch
predictions = model(inputs)
all_targets.extend(targets.numpy())
all_predictions.extend(predictions.numpy())
return all_targets, all_predictions
这段代码定义了一个名为 prediction 的函数，其主要目的是使用给定的模型对输入数据进行预测，并收集所有的目标值和预测值。
7.2 验证集预测
# 模型预测
targets, predictions = prediction(model, valid_loader)
# 反归一化
denormalized_targets = scaler.inverse_transform(targets)
denormalized_predictions = scaler.inverse_transform(predictions)
targets 是经过归一化处理后的目标值数组，predictions 是经过归一化处理后的预测值数组。scaler 是 MinMaxScaler() 归一化类的实例，inverse_transform 方法会将归一化后的数组还原为原始数据的尺度，即对预测值进行反归一化操作。
# Visualize the data
plt.figure(figsize=(12,6))
plt.style.use('_mpl-gallery')
plt.title('Comparison of validation set prediction results')
plt.plot(denormalized_targets, color='blue',label='Actual Value')
plt.plot(denormalized_predictions, color='orange', label='Valid Value')
plt.legend()
plt.show()
7.3 回归拟合图
使用 regplot() 函数绘制数据图并拟合线性回归模型。
plt.figure(figsize=(5, 5), dpi=100)
sns.regplot(x=denormalized_targets, y=denormalized_predictions, scatter=True, marker="*", color='orange',line_kws={'color': 'red'})
plt.show()
7.4 评估指标
以下代码使用了一些常见的评估指标：平均绝对误差（MAE）、平均绝对百分比误差（MAPE）、均方误差（MSE）、均方根误差（RMSE）和决定系数（R²）来衡量模型预测的性能。这里我们将通过调用 sklearn.metrics 模块中的 mean_absolute_error mean_absolute_percentage_error mean_squared_error root_mean_squared_error r2_score 函数来对模型的预测效果进行评估。
mae = mean_absolute_error(targets, predictions)
print(f"MAE: {mae:.4f}")
mape = mean_absolute_percentage_error(targets, predictions)
print(f"MAPE: {mape * 100:.4f}%")
mse = mean_squared_error(targets, predictions)
print(f"MSE: {mse:.4f}")
rmse = root_mean_squared_error(targets, predictions)
print(f"RMSE: {rmse:.4f}")
r2 = r2_score(targets, predictions)
print(f"R²: {r2:.4f}")
MAE: 0.0229
MAPE: 3.0055%
MSE: 0.0009
RMSE: 0.0302
R²: 0.9346
