1、简介

时序数据无处不在，如我们观看的视频直播、chatGPT生成的对话内容、天气预报中的气温数据、在股票交易中的股价数据，均存在时间与顺序上的相关性。而这种相关性，仅凭肉眼观察，难以获取可信结果。为此，本文以LSTM为基础，以股票数据为例，建立时序关联，辅助股票趋势预测。

2、任务描述

我们以百度美股（BIDU.US）为例，采用2022-01-31至2023-01-31日，共计252天交易数据，作为本次实测数据。从markert watch网站下载该时间段交易数据。

图1. BIDU交易数据

目标：利用部分数据作为训练，预测下1天、下4天、下8天、下16天股票趋势。

3、模型搭建

在模型搭建前，先介绍数据划分方式。如图2，在任务数据中，有五类数据，分别为开盘价、收盘价、高点、低点以及交易量。我们以Open数据为例进行时序分析。

图2. BIDU交易数据

怎样对数据进行划分？类比在前述文章中所介绍的情感分类任务。举例，我们想通过如下语句预测情感类别是正向或负向。

'i am good and happy': True

在上述例子中，训练数据为一段语句，所对应的标签为True。该段语句所包含的单词数量为15。因此，seq_len=15，每个单词写成one-hot方式，特征维度为1000。因此，输入数据可表示为：input：shape=【seq_len，batch_size，input_size】=【15,1,1000】。

类比情感分类任务，我们对股票数据做如下划分：

图3. 数据划分

采用10个观测值，预测第11个数据值。依次循环，获取训练输入数据与标签数据。若训练样本数量为192，观测值长度为10，观测值中每个特征维度为1。则输入数据可表示为：

input：shape=【seq_len，batch_size，input_size】=【10,192,1】。

图4. many to one结构图

在这里采用many to one方式lstm，将最后一个时间步的输出，外接全连接层，加深特征表示，输出值作为预测结果。模型构建代码如下：

class LSTM(nn.Module):
    def __init__(self, num_classes, input_size, hidden_size, num_layers):
        super(LSTM, self).__init__()
        self.num_classes = num_classes
        self.num_layers = num_layers
        self.input_size = input_size
        self.hidden_size = hidden_size
        self.seq_length = seq_length
        self.lstm = nn.LSTM(input_size=input_size, hidden_size=hidden_size,
                            num_layers=num_layers, batch_first=True)
        self.fc = nn.Linear(hidden_size, num_classes)

    def forward(self, x):
        h_0 = Variable(torch.zeros(
            self.num_layers, x.size(0), self.hidden_size))
        c_0 = Variable(torch.zeros(
            self.num_layers, x.size(0), self.hidden_size))
        # Propagate input through LSTM
        output, (h_out, c_out) = self.lstm(x, (h_0, c_0))
        h_out_ = h_out[-1, :, :].view(-1, self.hidden_size)
        out = self.fc(h_out_)
        return out

上述代码中，输入层，隐藏层以及输出层参数如下：

• input_size：1，因为输入数据的特征维度为1。
• hidden_size：50，设置的隐藏层节点个数为50。
• num_layers：3，设置的隐藏层层数为3，即3层堆叠lstm。
• num_classes：1，因为输出数据维度为1。
• batch_first=True：表示数据数据shape=【batch_size，seq_len，input_size】

4、实测验证

我们采用192天数据作为训练，预测下1天、4天、8天、16天的股票开盘价。

图5. 1天预测结果

图5，橙色曲线为预测值，蓝色曲线为真实值。单步预测较好逼近真实值。

图6. 4天预测结果

图6，预测基于183-192天的数据预测第193天股票值，采用183-192天以及预测的第193天数据预测第194天数据。可发现趋势基本符合。

图7. 8天预测结果

图8. 16天股票预测

如图8，由于预测值存在误差，而基于预测值推测下一步，误差会累积增加，因此导致与真实值偏离逐渐增加，即图8中从第202天开始，预测值与真实值偏差建增。因此，以上实验发现，在8步内预测与真实值趋势相符。

完整代码如下：

import torch
import torch.nn as nn
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
from torch.autograd import Variable
from sklearn.preprocessing import MinMaxScaler

from matplotlib import rcParams

rcParams['font.family'] = 'SimHei'


class LSTM(nn.Module):
    def __init__(self, num_classes, input_size, hidden_size, num_layers):
        super(LSTM, self).__init__()
        self.num_classes = num_classes
        self.num_layers = num_layers
        self.input_size = input_size
        self.hidden_size = hidden_size
        self.seq_length = seq_length
        self.lstm = nn.LSTM(input_size=input_size, hidden_size=hidden_size,
                            num_layers=num_layers, batch_first=True)
        self.fc = nn.Linear(hidden_size, num_classes)

    def forward(self, x):
        h_0 = Variable(torch.zeros(
            self.num_layers, x.size(0), self.hidden_size))
        c_0 = Variable(torch.zeros(
            self.num_layers, x.size(0), self.hidden_size))
        # Propagate input through LSTM
        output, (h_out, c_out) = self.lstm(x, (h_0, c_0))
        h_out_ = h_out[-1, :, :].view(-1, self.hidden_size)
        out = self.fc(h_out_)
        return out


training = pd.read_csv('Data_STOCK_US_XNAS_BIDU.csv')
training_set = training.iloc[:, 1:2].values


def sliding_windows(data, seq_length):
    x = []
    y = []
    for i in range(len(data) - seq_length - 1):
        _x = data[i:(i + seq_length)]
        _y = data[i + seq_length]
        x.append(_x)
        y.append(_y)
    return np.array(x), np.array(y)


sc = MinMaxScaler()
training_data = sc.fit_transform(training_set)

seq_length = 10
x, y = sliding_windows(training_data, seq_length)

train_size = int(len(y) * 0.8)
test_size = len(y) - train_size

dataX = Variable(torch.Tensor(np.array(x)))
dataY = Variable(torch.Tensor(np.array(y)))

trainX = Variable(torch.Tensor(np.array(x[0:train_size])))
trainY = Variable(torch.Tensor(np.array(y[0:train_size])))

testX = Variable(torch.Tensor(np.array(x[train_size:len(x)])))
testY = Variable(torch.Tensor(np.array(y[train_size:len(y)])))

num_epochs = 3000
learning_rate = 0.01

input_size = 1
hidden_size = 50
num_layers = 3

num_classes = 1

lstm = LSTM(num_classes, input_size, hidden_size, num_layers)

criterion = torch.nn.MSELoss()
optimizer = torch.optim.Adam(lstm.parameters(), lr=learning_rate)

for epoch in range(num_epochs):
    outputs = lstm(trainX)
    optimizer.zero_grad()
    # obtain the loss function
    loss = criterion(outputs, trainY)
    loss.backward()
    optimizer.step()
    if epoch % 100 == 0:
        print("Epoch: %d, loss: %1.5f" % (epoch, loss.item()))

lstm.eval()
train_predict = lstm(dataX)

data_predict = train_predict.data.numpy()
dataY_plot = dataY.data.numpy()

data_predict = sc.inverse_transform(data_predict)
dataY_plot = sc.inverse_transform(dataY_plot)

start = 190
step = 32
time_step = 10
multistep_result = np.zeros((training_data.shape[0], 1))
multistep_result[:start] = training_data[:start]

for i in range(step):
    input_X = multistep_result[start - time_step + i:start + i]
    input_X = np.reshape(input_X, (1, input_X.shape[0], input_X.shape[1]))
    input_X = torch.tensor(input_X, dtype=torch.float32)
    output_X = lstm(input_X)
    output_X = output_X.detach().numpy()
    multistep_result[start + i] = output_X

multistep_result = sc.inverse_transform(multistep_result)

x1 = np.arange(step)
y1 = multistep_result[start:start + step]
plt.figure()
plt.plot(training_set[start:start + step], color='red', label='真实股价')
plt.plot(x1, y1, color='blue', label='预测股价')
plt.title('百度股票趋势预测')
plt.xlabel('时间')
plt.ylabel('开盘股价')
plt.legend()

plt.figure()
plt.axvline(x=train_size, c='r', linestyle='--')
plt.plot(dataY_plot)
plt.plot(data_predict)
plt.xticks(range(0, training.shape[0], 30), training['Date'].loc[::-30], rotation=30)
plt.title('百度股票趋势预测')
# plt.xlabel('时间')
plt.ylabel('开盘股价')
plt.show()