前言

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

近来，机器学习得到了长足的发展，并引起了广泛的关注，其中语音和图像识别领域的成果最为显著。本文分析了深度学习模型——堆叠门控循环单元 Stacked GRU 在股市的表现。论文显示，虽然这种技术在自然语言处理、语音识别等其他领域取得了不错的成绩，但在金融时间序列预测上却表现不佳。事实上，金融数据的特点是噪声信号比高，这使得机器学习模型难以找到模式并预测未来价格。

本文通过对 GRU 时间序列模型的介绍，探讨Stacked GRU在股市科技股中的表现。本研究文章的结构如下。第一节介绍金融时间序列数据。第二节对金融时间数进行特征工程。第三节是构建模型、定义参数空间、损失函数与优化器。第四节是训练模型。第五节是评估模型与结果可视化。第六部分是预测下一个时间点的收盘价。

1. 金融时间序列数据

金融时间序列数据是指按照时间顺序记录的各种金融指标的数值序列，这些指标包括但不限于股票价格、汇率、利率等。这些数据具有以下几个显著特点：

1. 时间连续性：数据按照时间的先后顺序排列，反映了金融市场的动态变化过程。
2. 噪声和不确定性：金融市场受到多种复杂因素的影响，因此数据中存在大量噪声和不确定性。
3. 非线性和非平稳性：金融时间序列数据通常呈现出明显的非线性和非平稳性特征。

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 TimeSeriesSplit

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

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 探索性分析（可视化）

探索性数据分析 $EDA$ 是一种使用视觉技术分析数据的方法。它用于发现趋势和模式，或借助统计摘要和图形表示来检查假设。

1.2.1 股票的日收盘价

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

# plt.style.available
plt.style.use('seaborn-v0_8')

# 绘制收盘价
plt.figure(figsize=(18, 6))
plt.plot(AAPL['Adj Close'], label='AAPL')

# 设置图表标题和轴标签
plt.title('Close Price with Moving Averages')
plt.xlabel('')
plt.ylabel('Price $', fontsize=18)

# 显示图例
plt.legend()
plt.show()

1.2.2 股票的日收益率

股票的日收益率是反映投资者在一天内从股票投资中获得的回报比例。它通常用百分比来表示，计算公式为：日收益率 = (今日收盘价 - 前一日收盘价) / 前一日收盘价 × 100%，这里我们可是使用 .pct_change() 函数来实现。

plt.figure(figsize=(18,6))
plt.title('Daily Return History')
plt.plot(AAPL['Adj Close'].pct_change(),linestyle='--',marker='*',label='AAPL')
plt.ylabel('Daily Return', fontsize=18)
plt.legend()
plt.show()

1.2.3 股票收益率自相关性

股票收益率自相关性是描述一个股票在不同时间点的收益率如何相互关联的一个概念。具体来说，它指的是一个股票过去的收益率与其未来收益率之间的相关性。这种相关性可以是正相关（即过去的收益率上升预示着未来的收益率也可能上升），也可以是负相关（即过去的收益率上升预示着未来的收益率可能下降），或者两者之间没有显著的相关性。

AAPL['Returns'] = AAPL['Adj Close'].pct_change()

# 使用pandas的autocorr函数计算自相关系数
# 注意：autocorr默认计算的是滞后1的自相关系数，要计算其他滞后的，需要循环或使用其他方法
autocorr_values = [AAPL['Returns'].autocorr(lag=i) for i in range(1, 301)]  # 假设我们查看滞后1到300的自相关

# 使用matplotlib绘制自相关系数
plt.figure(figsize=(18, 6))
plt.plot(range(1, 301), autocorr_values, linestyle='-.', marker='*')
plt.title('Autocorrelation of Stock Returns')
plt.xlabel('Lag')
plt.ylabel('Autocorrelation')
plt.grid(True)
plt.show()

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

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

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

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

2.1 构造序列数据

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

# 构造序列数据函数
def create_dataset(dataset, lookback=1):
    """Transform a time series into a prediction dataset
    
    Args:
        dataset: A numpy array of time series, first dimension is the time steps
        lookback: Size of window for prediction
    """
    X, y = [], []
    for i in range(len(dataset)-lookback): 
        feature = dataset[i:(i+lookback), 0]
        target = dataset[i + lookback, 0]
        X.append(feature)
        y.append(target)
    return np.array(X), np.array(y)

2.2 特征缩放（归一化）

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

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

# 创建数据集
X, y = create_dataset(scaled_data, lookback=window_size)

# 重塑输入数据为[samples, time steps, features]
X = np.reshape(X, (X.shape[0], X.shape[1], 1))

2.3 数据集划分（TimeSeriesSplit）

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

# 使用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, 1) (90, 60, 1) (605,) (90,)

2.4 数据集张量（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() 函数，允许我们重组数据，以匹配 LSTM 模型所需的输入形状。以这种方式重塑数据可确保 LSTM 模型以预期格式接收数据。

torch.Size([605, 60, 1]) torch.Size([90, 60, 1]) 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. 构建时间序列模型（Stacked GRU）

GRU (Gated Recurrent Unit)是一种循环神经网络$RNN$的变体，用于处理和预测序列数据。与标准RNN相比，GRU能够更有效地捕捉长期依赖关系，并且在训练时更不容易出现梯度消失或梯度爆炸的问题。

🔗 PyTorch所提供的数学公式及解释如下：

Apply a multi-layer gated recurrent unit (GRU) RNN to an input sequence. For each element in the input sequence, each layer computes the following function:
$$\begin{array}{l}{r}_t=\sigma (W_{ir}{x}_t+b_{ir}+W_{hr}{h}_{(t-1)}+b_{hr})\\ z_t=\sigma (W_{iz}{x}_t+b_{iz}+W_{hz}{h}_{(t-1)}+b_{hz})\\ n_t=\tanh (W_{in}{x}_t+b_{in}+r_t\odot (W_{hn}{h}_{(t-1)}+b_{hn}))\\ h_t=(1-z_t)\odot n_t+z_t\odot h_{(t-1)}\end{array}$$
where $h_t$ is the hidden state at time $t$, $x_t$ is the input at time $t$, $h_{(t-1)}$ is the hidden state of the layer at time $t-1$ or the initial hidden state at time $0$, and $r_t$, $z_t$, $n_t$ are the reset, update, and new gates, respectively. $\sigma$ is the sigmoid function, and $\odot$ is the Hadamard product.

In a multilayer GRU, the input $x_t^{(l)}$ of the $l$ -th layer ($l\ge 2$) is the hidden state $h_t^{(l-1)}$ of the previous layer multiplied by dropout ${\delta }_t^{(l-1)}$ where each ${\delta }_t^{(l-1)}$ is a Bernoulli random variable which is $0$ with probability $dropout$.

3.1 构建 GRU 模型

class GRUNet(nn.Module):
    def __init__(self, input_dim, hidden_dim, output_dim=1, num_layers=2):
        # input_dim 是输入特征的维度，hidden_dim 是隐藏层神经单元维度或称为隐藏状态的大小，output_dim 是输出维度，
        # num_layers 是网络层数，设置 num_layers=2 表示将两个 GRU 堆叠在一起形成一个堆叠 GRU，第二个 GRU 接收第一个 GRU 的输出并计算最终结果
        super(GRUNet, self).__init__()
        # 通过调用 super(GRUNet, self).__init__() 初始化父类 nn.Module
        self.hidden_dim = hidden_dim
        self.num_layers = num_layers
        self.gru = nn.GRU(input_dim, hidden_dim, num_layers, batch_first=True)
        # 定义 GRU 层，使用 batch_first=True 表示输入数据的形状是 [batch_size, seq_len(time_steps), input_dim]
        self.fc = nn.Linear(hidden_dim, output_dim)
        # 定义全连接层，将 GRU 的最后一个隐藏状态映射到输出维度 output_dim
    def forward(self, x):
        h0 = torch.zeros(self.num_layers, x.size(0), self.hidden_dim).to(x.device)
        # 初始化h0为全零张量，h0代表隐藏状态(hidden state)的初始值，形状为 [num_layers * num_directions, batch, hidden_dim]
        # 如果没有指定双向参数 bidirectional为 True，num_directions 默认为 1
        out, _ = self.gru(x, h0)
        # 将输入数据 x 和初始隐藏状态 h0 传递给 GRU层，得到输出 out(即所有时间步的输出)和最后一个时间步的隐藏状态 hn(这里用 _忽略)
        out = self.fc(out[:, -1, :])
        # GRU 的输出是一个三维张量，其形状是 [batch_size, seq_len(time_steps), hidden_dim]，
        # 这里我们只取最后一个时间步的隐藏状态 out[:, -1, :] 并传递给全连接层。
        return out

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

要在 PyTorch 中构建堆叠 GRU，我们需要调用 GRUNet 类，通过输入 num_layers 的参数来实现

model = GRUNet(input_dim=1, # 输入数据的特征数量 X_train.shape[2]
               hidden_dim=64,
               output_dim=1,
               num_layers=2) # 表示将两个 GRU 堆叠在一起形成一个堆叠 GRU
criterion = torch.nn.MSELoss() # 定义均方误差损失函数
optimizer = torch.optim.Adam(model.parameters(), lr=0.01) # 定义优化器

summary(model, (32, 60, 1)) # batch_size, seq_len(time_steps), input_dim

==========================================================================================
Layer (type:depth-idx)                   Output Shape              Param #
==========================================================================================
GRUNet                                   [32, 1]                   --
├─GRU: 1-1                               [32, 60, 64]              37,824
├─Linear: 1-2                            [32, 1]                   65
==========================================================================================
Total params: 37,889
Trainable params: 37,889
Non-trainable params: 0
Total mult-adds (Units.MEGABYTES): 72.62
==========================================================================================
Input size (MB): 0.01
Forward/backward pass size (MB): 0.98
Params size (MB): 0.15
Estimated Total Size (MB): 1.14
==========================================================================================

4. 模型训练与可视化

train_loss = []
num_epochs = 20

for epoch in range(num_epochs):
    model.train()  # 初始化训练进程
    pbar = tqdm(train_loader, desc=f"Epoch {epoch+1}/{num_epochs}")
    for batch_idx, (data, target) in enumerate(pbar):
        # 前向传播
        outputs = model(data)  # 每个批次的预测值
        loss = criterion(outputs, target)

        # 反向传播和优化
        optimizer.zero_grad()
        loss.backward()
        optimizer.step()
        
        # 记录损失值
        train_loss.append(loss.item())
        # 更新进度条
        pbar.update()
        # 这里只用于显示当前批次的损失，不是平均损失
        pbar.set_postfix({'Train loss': f'{loss.item():.4f}'})

这里我们使用 tqdm模块来展示进度条

Epoch 1/20: 100%|███████████████████████████████████████████████████| 19/19 [00:01<00:00, 13.25it/s, Train loss=0.0211]
Epoch 2/20: 100%|███████████████████████████████████████████████████| 19/19 [00:01<00:00, 13.07it/s, Train loss=0.0052]
Epoch 3/20: 100%|███████████████████████████████████████████████████| 19/19 [00:01<00:00, 13.24it/s, Train loss=0.0030]
Epoch 4/20: 100%|███████████████████████████████████████████████████| 19/19 [00:01<00:00, 13.06it/s, Train loss=0.0014]
Epoch 5/20: 100%|███████████████████████████████████████████████████| 19/19 [00:01<00:00, 13.19it/s, Train loss=0.0011]
Epoch 6/20: 100%|███████████████████████████████████████████████████| 19/19 [00:01<00:00, 13.21it/s, Train loss=0.0007]
Epoch 7/20: 100%|███████████████████████████████████████████████████| 19/19 [00:01<00:00, 13.21it/s, Train loss=0.0015]
Epoch 8/20: 100%|███████████████████████████████████████████████████| 19/19 [00:01<00:00, 13.22it/s, Train loss=0.0015]
Epoch 9/20: 100%|███████████████████████████████████████████████████| 19/19 [00:01<00:00, 13.21it/s, Train loss=0.0012]
Epoch 10/20: 100%|██████████████████████████████████████████████████| 19/19 [00:01<00:00, 13.07it/s, Train loss=0.0010]
Epoch 11/20: 100%|██████████████████████████████████████████████████| 19/19 [00:01<00:00, 13.07it/s, Train loss=0.0014]
Epoch 12/20: 100%|██████████████████████████████████████████████████| 19/19 [00:01<00:00, 13.19it/s, Train loss=0.0011]
Epoch 13/20: 100%|██████████████████████████████████████████████████| 19/19 [00:01<00:00, 13.21it/s, Train loss=0.0013]
Epoch 14/20: 100%|██████████████████████████████████████████████████| 19/19 [00:01<00:00, 13.07it/s, Train loss=0.0013]
Epoch 15/20: 100%|██████████████████████████████████████████████████| 19/19 [00:01<00:00, 13.03it/s, Train loss=0.0020]
Epoch 16/20: 100%|██████████████████████████████████████████████████| 19/19 [00:01<00:00, 13.08it/s, Train loss=0.0012]
Epoch 17/20: 100%|██████████████████████████████████████████████████| 19/19 [00:01<00:00, 13.07it/s, Train loss=0.0009]
Epoch 18/20: 100%|██████████████████████████████████████████████████| 19/19 [00:01<00:00, 13.22it/s, Train loss=0.0014]
Epoch 19/20: 100%|██████████████████████████████████████████████████| 19/19 [00:01<00:00, 13.07it/s, Train loss=0.0019]
Epoch 20/20: 100%|██████████████████████████████████████████████████| 19/19 [00:01<00:00, 13.11it/s, Train loss=0.0013]

plt.plot(train_loss)

5. 模型评估与可视化

5.1 均方误差

model.eval()  # 将模型设置为评估模式
test_loss = []  # 初始化损失
pbar = tqdm(test_loader, desc="Evaluating")
with torch.no_grad():
    for data, target in pbar:
        test_pred = model(data)
        loss = criterion(test_pred, target)
        test_loss.append(loss.item())
        # 计算当前批次的平均损失
        batch_avg_loss = sum(test_loss)/len(test_loss)
        pbar.set_postfix({'Test Loss': f'{batch_avg_loss:.4f}'})
        pbar.update()  # 更新进度条

pbar.close()  # 关闭进度条

Evaluating: 100%|██████████████████████████████████████████████████████| 3/3 [00:00<00:00, 51.30it/s, Test Loss=0.0011]

5.2 反归一化

.inverse_transform 将经过转换或缩放的数据转换回其原始形式或接近原始形式

# 反归一化预测结果
train_pred = scaler.inverse_transform(model(X_train_tensor).detach().numpy())
y_train = scaler.inverse_transform(y_train_tensor.detach().numpy())
test_pred = scaler.inverse_transform(model(X_test_tensor).detach().numpy())
y_test = scaler.inverse_transform(y_test_tensor.detach().numpy())

print(train_pred.shape, y_train.shape, test_pred.shape, y_test.shape)

(605, 1) (605, 1) (90, 1) (90, 1)

5.3 结果可视化

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

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

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

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

# Visualize the data
plt.figure(figsize=(18,6))
plt.title('GRU 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['Close'][-window_size:].values
latest_closes = latest_closes.reshape(-1, 1)
scaled_latest_closes = scaler.fit_transform(latest_closes)
tensor_latest_closes = torch.from_numpy(scaled_latest_closes).type(torch.Tensor).view(1, window_size, 1)
print(tensor_latest_closes.shape)

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

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

# 使用模型预测下一个时间点的收盘价
next_close_pred = model(tensor_latest_closes)
next_close_pred = scaler.inverse_transform(next_close_pred.detach().numpy())
next_close_pred

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