前言

系列专栏:【深度学习：算法项目实战】✨︎
涉及医疗健康、财经金融、商业零售、食品饮料、运动健身、交通运输、环境科学、社交媒体以及文本和图像处理等诸多领域，讨论了各种复杂的深度神经网络思想，如卷积神经网络、循环神经网络、生成对抗网络、门控循环单元、长短期记忆、自然语言处理、深度强化学习、大型语言模型和迁移学习。

近年来，深度学习技术的快速发展为股票价格预测提供了新的思路。其中，卷积神经网络（CNN）和长短期记忆网络（LSTM）作为深度学习中的两种重要模型，在图像识别、语音识别、自然语言处理等领域取得了显著成果。CNN擅长于捕捉数据的局部特征和空间依赖关系，而LSTM则擅长于处理时间序列数据中的长期依赖关系。因此，结合CNN和LSTM的模型在处理多变量时间序列预测问题方面具有巨大的潜力。

本文旨在探索基于CNN-LSTM的模型在股票多变量时间序列预测中的应用。通过构建CNN-LSTM模型，我们可以同时捕捉股票价格数据中的空间特征和时序依赖关系，从而实现对股票价格更准确的预测。本文的研究不仅有助于提升股票价格预测的精度，也为深度学习技术在金融领域的应用提供了新的思路和方法。

1. 股票时间序列数据

股票时间序列数据是指按照时间顺序排列的关于股票市场的历史数据。这些数据记录了股票市场在不同时间点上的各种信息，如股票的开盘价、最高价、最低价、收盘价、成交量等。时间序列数据是金融市场分析的重要基础，因为它们反映了市场参与者的行为、市场供求关系以及宏观经济和政策的影响等因素。

import numpy as np
import pandas as pd
import matplotlib.pyplot as plt

from sklearn.preprocessing import MinMaxScaler
from sklearn.model_selection import TimeSeriesSplit

import torch
import torch.nn as nn
from torch.utils.data import DataLoader, TensorDataset
from torchinfo import summary
from tqdm import tqdm

np.random.seed(0)
torch.manual_seed(0)

1.1 数据预处理

pandas.to_datetime 函数将标量、数组、Series 或 DataFrame/dict-like 转换为 pandas datetime 对象。

AAPL = pd.read_csv('AAPL.csv')
print(type(AAPL['Close'].iloc[0]),type(AAPL['Date'].iloc[0]))
# Let's convert the data type of timestamp column to datatime format
AAPL['Date'] = pd.to_datetime(AAPL['Date'])
print(type(AAPL['Close'].iloc[0]),type(AAPL['Date'].iloc[0]))

# Selecting subset
cond_1 = AAPL['Date'] >= '2021-04-23 00:00:00'
cond_2 = AAPL['Date'] <= '2024-04-23 00:00:00'
AAPL = AAPL[cond_1 & cond_2].set_index('Date')
print(AAPL.shape)

<class 'numpy.float64'> <class 'str'>
<class 'numpy.float64'> <class 'pandas._libs.tslibs.timestamps.Timestamp'>
(755, 6)

---

1.2 数据可视化

收盘价是股票在正常交易日的最后交易价格。股票的收盘价是投资者跟踪其长期表现的标准基准。

# 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()

2. 时间数据特征工程(APPL)

在时间序列分析中，时间窗口通常用于描述在训练模型时考虑的连续时间步 time steps 的数量。这个时间窗口的大小，即 window_size，对于模型预测的准确性至关重要。

具体来说，window_size 决定了模型在做出预测时所使用的历史数据的长度。例如，如果我们想要用前60天的股票数据来预测未来7天的收盘价，那么window_size 就是60。

# 设置时间窗口大小
window_size = 60

除此之外还需构建序列函数，该函数需要两个参数：dataset 和 lookback，前者是要转换成数据集的 NumPy 数组，后者是用作预测下一个时间段的输入变量的前一时间步数，默认设为 1。

# 构造序列数据函数，若target为 Close，即索引为 3
def create_dataset(dataset, lookback= 1):
    X, y = [], []
    for i in range(len(dataset)-lookback): 
        feature = dataset[i:(i+lookback), :]
        target = dataset[i + lookback, 3]
        X.append(feature)
        y.append(target)
    return np.array(X), np.array(y)

2.1 特征缩放（归一化）

MinMaxScaler() 函数主要用于将特征数据按比例缩放到指定的范围。默认情况下，它将数据缩放到[0, 1]区间内，但也可以通过参数设置将数据缩放到其他范围。在机器学习中，MinMaxScaler()函数常用于不同尺度特征数据的标准化，以提高模型的泛化能力。

# 选取 AAPL[['Open', 'High', 'Low', 'Close']]作为特征, 归一化数据
scaler = MinMaxScaler(feature_range=(0, 1))
scaled_data = scaler.fit_transform(AAPL[['Open', 'High', 'Low', 'Close']].values)

# 获取反归一化参数(即原始数据的最小值和最大值)
original_min = scaler.data_min_
original_max = scaler.data_max_

# scale_params是一个包含所有特征反归一化参数的列表或数组，
# 其中第四个元素是Close价格的反归一化参数
scale_params = original_max - original_min

2.2 数据集划分（TimeSeriesSplit）

TimeSeriesSplit() 函数与传统的交叉验证方法不同，TimeSeriesSplit 特别适用于需要考虑时间顺序的数据集，因为它确保测试集中的所有数据点都在训练集数据点之后，并且可以分割多个训练集和测试集。

# 创建数据集,数据形状为 [samples, time steps, features]
X, y = create_dataset(scaled_data, lookback = window_size)
print(X.shape, y.shape)

(695, 60, 4) (695,)

# 使用TimeSeriesSplit划分数据集，根据需要调整n_splits
tscv = TimeSeriesSplit(n_splits=3, test_size=90)
# 遍历所有划分进行交叉验证
for i, (train_index, test_index) in enumerate(tscv.split(X)):
    X_train, X_test = X[train_index], X[test_index]
    y_train, y_test = y[train_index], y[test_index]
    # print(f"Fold {i}:")
    # print(f"  Train: index={train_index}")
    # print(f"  Test:  index={test_index}")

# 查看最后一个 fold 数据帧的维度
print(X_train.shape, X_test.shape, y_train.shape, y_test.shape)

(605, 60, 4) (90, 60, 4) (605,) (90,)

2.3 数据集张量（TensorDataset）

张量是一个多维数组或矩阵的数学对象，可以看作是向量和矩阵的推广。在深度学习中，张量通常用于表示输入数据、模型参数以及输出数据

# 将 NumPy数组转换为 tensor张量
X_train_tensor = torch.from_numpy(X_train).type(torch.Tensor)
X_test_tensor = torch.from_numpy(X_test).type(torch.Tensor)
y_train_tensor = torch.from_numpy(y_train).type(torch.Tensor).view(-1, 1)
y_test_tensor = torch.from_numpy(y_test).type(torch.Tensor).view(-1, 1)

print(X_train_tensor.shape, X_test_tensor.shape, y_train_tensor.shape, y_test_tensor.shape)

view() 函数用于重塑张量对象，它等同于 NumPy 中的 reshape() 函数，允许我们重组数据，以匹配 BiLSTM 模型所需的输入形状。以这种方式重塑数据可确保 BiLSTM 模型以预期格式接收数据。

torch.Size([605, 60, 4]) torch.Size([90, 60, 4]) torch.Size([605, 1]) torch.Size([90, 1])

接下来，我们将使用 TensorDataset 和 DataLoader创建数据集和数据加载器

train_dataset = TensorDataset(X_train_tensor, y_train_tensor)
train_loader = DataLoader(train_dataset, batch_size=32, shuffle=True)
test_dataset = TensorDataset(X_test_tensor, y_test_tensor)
test_loader = DataLoader(test_dataset, batch_size=32, shuffle=False)

shuffle=True 表示在每个epoch开始时，数据集将被随机打乱，这有助于防止模型在训练时过拟合。与训练数据加载器类似，shuffle=False 表示在测试时不需要打乱数据集。因为测试集通常用于评估模型的性能，而不是用于训练，所以不需要打乱。

3. 构建时间序列模型（CNN+LSTM）

CNN-LSTM模型结合了CNN和LSTM的特点，首先通过CNN层对输入数据进行特征提取，然后将提取的特征序列输入到LSTM层进行进一步的序列建模。CNN层使用多个卷积核（每个具有不同的kernel_size和out_channels）来提取时间序列中的局部特征。卷积核的宽度（kernel_size）决定了在卷积过程中考虑多少个相邻的时间步。LSTM层接收CNN层的输出，并捕获时间序列中的长期依赖关系。全连接层在LSTM层之后，可以使用全连接层来进一步处理LSTM的输出，并产生最终的预测。

3.1 构建Conv2D+LSTM模型

# 参数
num_features = 4  # 每个时间步的特征数量 (如开盘价、最高价、最低价、收盘价)
out_channels = 64  # CNN卷积产生的通道数量用作 LSTM的输入
hidden_size = 128  # LSTM的隐藏层大小
num_layers = 2  # LSTM的层数
dropout = 0.2  # LSTM的dropout比率
num_classes = 1  # 预测类别数量

class CNN_LSTM(nn.Module):
    def __init__(self, num_features, out_channels, hidden_size, num_layers, dropout, num_classes):
        super(CNN_LSTM, self).__init__()        
        # CNN层使用二维卷积，其中输入通道数为 1(假设输入数据是灰度图)，输出通道数为out_channels，卷积核的大小为(3, num_features)
        # 这里假设在特征维度上有一个大小为3的窗口，并在时间维度上滑动整个num_features的长度
        self.conv = nn.Conv2d(in_channels = 1, out_channels=out_channels, kernel_size=(3, num_features), padding=(1, 0))
        self.relu = nn.ReLU()
        self.maxpool = nn.MaxPool2d((2, 1))  # 只在特征维度上进行池化，时间维度上保持不变
        
        # LSTM层使用了nn.LSTM，其输入特征大小为CNN的输出通道数out_channels，隐藏层大小为hidden_size，层数为num_layers
        self.lstm = nn.LSTM(out_channels, hidden_size, num_layers, batch_first=True, dropout=dropout if num_layers > 1 else 0)
        
        # 全连接层将LSTM的输出映射到num_classes个类别上
        self.fc = nn.Linear(hidden_size, num_classes)
    
    def forward(self, x):
        # (batch_size, time_steps, num_features) => (batch_size, 1, time_steps, num_features)
        x = x.unsqueeze(1)
        
        # CNN层
        x = self.conv(x)
        x = self.relu(x)
        x = self.maxpool(x)
        
        # LSTM层需要(batch_size, time_steps, input_size)的输入, input_size = out_channels
        # 因为CNN的maxpool只在特征维度上进行了池化，所以时间维度上保持不变
        # (batch_size, out_channels, time_steps, 1) => (batch_size, time_steps, out_channels)
        x = x.squeeze(3)
        x = x.permute(0, 2, 1)
        
        # LSTM层
        lstm_out, _ = self.lstm(x)
        
        # 全连接层，输出LSTM的最后一个时间步
        out = self.fc(lstm_out[:, -1, :])
        
        return out

二维卷积（nn.Conv2d）在应用于时间序列分析时，其in_channels参数通常不直接对应时间序列的特征数。在二维卷积的上下文中，in_channels指的是输入数据的通道数，这通常在处理图像数据时更有意义，其中每个通道可能代表图像的一个颜色分量（如RGB中的红、绿、蓝）。

然而，在时间序列分析中，我们通常使用一维卷积（nn.Conv1d），因为时间序列数据本质上是一维的。在nn.Conv1d中，in_channels参数会对应时间序列中的特征数（或称为变量数、维度数）。每个时间步长都可能有一个或多个特征值，这些特征值构成了卷积层的输入通道。

3.2 定义模型、损失函数与优化器

model = CNN_LSTM(num_features, out_channels, hidden_size, num_layers, dropout, num_classes)

criterion = torch.nn.MSELoss() # 定义均方误差损失函数
optimizer = torch.optim.Adam(model.parameters(), lr=0.01) # 定义优化器

summary(model, (32, window_size, num_features)) # batch_size, seq_len(time_steps), input_size(num_features)

==========================================================================================
Layer (type:depth-idx)                   Output Shape              Param #
==========================================================================================
CNN_LSTM                                 [32, 1]                   --
├─Conv2d: 1-1                            [32, 64, 60, 1]           832
├─ReLU: 1-2                              [32, 64, 60, 1]           --
├─MaxPool2d: 1-3                         [32, 64, 30, 1]           --
├─LSTM: 1-4                              [32, 30, 128]             231,424
├─Linear: 1-5                            [32, 1]                   129
==========================================================================================
Total params: 232,385
Trainable params: 232,385
Non-trainable params: 0
Total mult-adds (Units.MEGABYTES): 223.77
==========================================================================================
Input size (MB): 0.03
Forward/backward pass size (MB): 1.97
Params size (MB): 0.93
Estimated Total Size (MB): 2.93
==========================================================================================

4. 模型训练与可视化

4.1 模型训练与权重更新（保存最佳模型）

train_losses = [] # 初始化列表来存储损失值
test_losses = [] # 初始化列表来存储损失值
best_loss = float('100') # 初始化为一个很大的值
best_model = None # 初始化最佳模型为None

# 训练循环
num_epochs = 20  # 假设我们要训练 20个 epoch
for epoch in range(num_epochs):
    model.train()  # 初始化训练进程
    train_loss = 0.0  # 初始化每个epoch的损失总和

    pbar = tqdm(train_loader, desc=f"Epoch {epoch+1}/{num_epochs}",ncols=80)
    for batch_idx, (data, target) in enumerate(pbar):
        optimizer.zero_grad()  # 将优化器中所有参数的梯度归零
        output = model(data)  # 每个批次的预测值
        loss = criterion(output, target)  # 计算模型损失值(y_train_pred 与 y_train)
        # 累加每个批次的损失
        train_loss += loss.item() * data.size(0)
        # 反向传播和优化
        loss.backward()
        optimizer.step()

        # 更新进度条(显示当前批次的损失)
        pbar.update()

    # 计算当前epoch的平均损失
    average_loss = train_loss / len(train_loader.dataset)
    train_losses.append(average_loss)

    model.eval()  # 将模型设置为评估模式
    with torch.no_grad():  # 不计算梯度以节省内存和计算资源
        test_loss = 0.0
        for data, target in test_loader:
            output = model(data) # 每个批次的预测值
            loss = criterion(output, target)
            test_loss += loss.item() * data.size(0)

        # 计算测试集的平均损失
        test_loss = test_loss / len(test_loader.dataset)
        test_losses.append(test_loss)

    # 如果测试损失比之前的最小值小，则更新最佳模型权重
    if test_loss < best_loss:
        best_loss = test_loss
        best_model = model.state_dict() 

# 训练循环结束后保存最佳模型
if best_model is not None:
    torch.save(best_model, 'model_best.pth')
    print(f'Best model saved with test loss {best_loss:.4f}')
else:
    print('No model saved as no improvement was made.')

loss.item()：loss是一个PyTorch的Tensor对象，而.item()方法用于从只包含一个元素的Tensor中提取Python数值（通常是浮点数）。由于损失通常是一个标量（scalar），因此我们可以安全地使用.item()来获取其数值。
data.size(0)：这返回data这个Tensor在第一维度（通常是批次维度）的大小。在深度学习中，我们经常使用mini-batch梯度下降，这意味着我们将整个数据集分成较小的批次，并在每个批次上计算梯度并更新模型。data.size(0)就告诉我们当前批次中有多少个数据样本。
loss.item() * data.size(0)：这实际上是在对损失进行“加权”。我们乘以批次大小是为了稍后能够计算平均损失（average loss）。这是因为在累积损失时，我们不仅要考虑损失的大小，还要考虑该损失是基于多少数据样本计算出来的。
train_loss += ...：最后，我们将加权后的损失累加到train_loss变量中。这样做是为了在整个训练循环结束时计算平均损失。

Epoch 1/20: 100%|███████████████████████████████| 19/19 [00:00<00:00, 52.58it/s]
Epoch 2/20: 100%|███████████████████████████████| 19/19 [00:00<00:00, 55.02it/s]
Epoch 3/20: 100%|███████████████████████████████| 19/19 [00:00<00:00, 57.53it/s]
Epoch 4/20: 100%|███████████████████████████████| 19/19 [00:00<00:00, 50.46it/s]
Epoch 5/20: 100%|███████████████████████████████| 19/19 [00:00<00:00, 52.58it/s]
Epoch 6/20: 100%|███████████████████████████████| 19/19 [00:00<00:00, 50.44it/s]
Epoch 7/20: 100%|███████████████████████████████| 19/19 [00:00<00:00, 52.59it/s]
Epoch 8/20: 100%|███████████████████████████████| 19/19 [00:00<00:00, 50.46it/s]
Epoch 9/20: 100%|███████████████████████████████| 19/19 [00:00<00:00, 57.53it/s]
Epoch 10/20: 100%|██████████████████████████████| 19/19 [00:00<00:00, 50.12it/s]
Epoch 11/20: 100%|██████████████████████████████| 19/19 [00:00<00:00, 57.59it/s]
Epoch 12/20: 100%|██████████████████████████████| 19/19 [00:00<00:00, 55.01it/s]
Epoch 13/20: 100%|██████████████████████████████| 19/19 [00:00<00:00, 55.05it/s]
Epoch 14/20: 100%|██████████████████████████████| 19/19 [00:00<00:00, 50.35it/s]
Epoch 15/20: 100%|██████████████████████████████| 19/19 [00:00<00:00, 54.93it/s]
Epoch 16/20: 100%|██████████████████████████████| 19/19 [00:00<00:00, 55.00it/s]
Epoch 17/20: 100%|██████████████████████████████| 19/19 [00:00<00:00, 57.63it/s]
Epoch 18/20: 100%|██████████████████████████████| 19/19 [00:00<00:00, 55.11it/s]
Epoch 19/20: 100%|██████████████████████████████| 19/19 [00:00<00:00, 55.33it/s]
Epoch 20/20: 100%|██████████████████████████████| 19/19 [00:00<00:00, 57.54it/s]
Best model saved with test loss 0.0010

4.2 可视化训练过程（Loss损失）

# 绘制损失图
plt.figure(figsize=(18, 6))
plt.plot(train_losses, label='Train Loss')
plt.plot(test_losses, label='Test Loss')
plt.title('Training and Test Loss')
plt.xlabel('Epoch')
plt.ylabel('Loss')
plt.legend()
plt.show()

5. 模型评估与可视化

5.1 评估指标（MAE、RMSE、MAPE、R²）

RMSE：该指标计算预测值与实际值之间平均平方差的平方根。它对较大误差给予较高的惩罚。
MAPE：该指标计算实际值与预测值之间绝对百分比差异的平均值。它用百分比来表示平均绝对误差，这对了解相对预测误差非常有用。

# 加载最佳模型
model.load_state_dict(torch.load('model_best.pth'))

def mean_absolute_percentage_error(y_true, y_pred):
    y_true, y_pred = np.array(y_true), np.array(y_pred)
    mask = y_true != 0
    return np.mean(np.abs((y_true[mask] - y_pred[mask]) / y_true[mask])) * 100 if mask.any() else 0

# 将模型设置为评估模式
model.eval()
y_pred_all = []
y_true_all = []

with torch.no_grad():
    pbar = tqdm(test_loader, desc='Evaluating')
    for data, target in pbar:
        y_pred = model(data).detach().cpu().numpy()  # 确保张量在CPU上
        y_true = target.detach().cpu().numpy()  # 确保张量在CPU上
        y_pred_all.append(y_pred)
        y_true_all.append(y_true)
        pbar.update()

# 合并所有批次的预测和真实值
y_pred_all = np.concatenate(y_pred_all)
y_true_all = np.concatenate(y_true_all)

mae = np.mean(np.abs(y_pred_all - y_true_all))
print(f"MAE: {mae:.4f}")

rmse = np.sqrt(np.mean((y_pred_all - y_true_all) ** 2))
print(f"RMSE: {rmse:.4f}")

mape = mean_absolute_percentage_error(y_true_all, y_pred_all)
print(f"MAPE: {mape:.4f}%")

y_bar = np.mean(y_true_all)  # 真实值的平均值
SS_res = np.sum((y_true_all - y_pred_all) ** 2)  # 残差平方和
SS_tot = np.sum((y_true_all - y_bar) ** 2)  # 总平方和
R2 = 1 - (SS_res / SS_tot)
print(f"R²: {R2:.4f}")

Evaluating: 100%|██████████████████████| 3/3 [00:00<00:00, 95.95it/s]
MAE: 0.0242
RMSE: 0.0323
MAPE: 3.1446%
R²: 0.9351

5.2 反归一化

在进行数据特征缩放(归一化)时，我们已经提取了反归一化参数scale_params，接下来我们将进行反归一化

# 提取Close价格的反归一化参数
close_denorm_param = scale_params[3]  # 假设Close是第四个特征

# 如果反归一化包括一个偏移量shift，需要再加上 shift_param
# denormalized_predictions = (predictions[:, 0] * scale_param) + shift_param

# 将训练预测 Close的价格反归一化
y_train_pred = model(X_train_tensor).detach().numpy()
y_train_denormalized_predictions = (y_train_pred * scale_params[3]) + original_min[3]

# 将测试预测 Close的价格反归一化
y_test_pred = model(X_test_tensor).detach().numpy()
y_test_denormalized_predictions = (y_test_pred * scale_params[3]) + original_min[3]

5.3 结果可视化

计算训练预测与测试预测的绘图数据

# shift train predictions for plotting
trainPredict = AAPL[window_size:X_train.shape[0]+X_train.shape[1]]
trainPredictPlot = trainPredict.assign(TrainPrediction= y_train_denormalized_predictions)

testPredict = AAPL[X_train.shape[0]+X_train.shape[1]:]
testPredictPlot = testPredict.assign(TestPrediction= y_test_denormalized_predictions)

绘制模型收盘价格的原始数据与预测数据

# Visualize the data
plt.figure(figsize=(18,6))
plt.title('CNN-LSTM Close Price Validation')
plt.plot(AAPL['Close'], color='blue', label='original')
plt.plot(trainPredictPlot['TrainPrediction'], color='orange',label='Train Prediction')
plt.plot(testPredictPlot['TestPrediction'], color='red', label='Test Prediction')
plt.legend()
plt.show()

6. 模型预测

6.1 转换最新时间步收盘价的数组为张量

# 假设latest_closes是一个包含最新window_size个收盘价的列表或数组
latest_closes = AAPL[['Open', 'High', 'Low', 'Close']][-window_size:].values
scaled_latest_closes = scaler.transform(latest_closes)
tensor_latest_closes = torch.from_numpy(scaled_latest_closes).type(torch.Tensor).view(1, window_size, 4) 
print(tensor_latest_closes.shape)

torch.Size([1, 60, 4])

6.2 预测下一个时间点的收盘价格

# 使用模型预测下一个时间点的收盘价
next_close_pred = model(tensor_latest_closes).detach().numpy()
next_close_denormalized_pred = (next_close_pred * scale_params[3]) + original_min[3]
next_close_denormalized_pred

array([[166.56705]], dtype=float32)