paper：A Deep Learning based Stock Trading Model with 2-D CNN Trend Detection 2D CNN模型

Posted 2023-01-12 David_7788

2D CNN进行时序预测

paper：A Deep Learning based Stock Trading Model with 2-D CNN Trend Detection

paper使用了2-D的cnn对时序序列进行预测

步骤：

① 首先先对时序数据进行第一差分的处理方法，即前一天减去后一天，得到△值，然后使用tanh把值压缩到-1,1之间。

②由于2D CNN输入是一个正方形，这一篇paper的CNN输入是（28,28），第一个28是时间序列的长度，第二个是通过维度扩充把时序数据的特征维度扩充为28，这一篇paper扩维的方法是：由于这是一个股票预测的任务，因此他使用了股票方面的专家知识，通过计算下图的计算公式，把原数据的维度进行了扩充。

下表的Parameters就是每一个特征计算中的一些参数取值，Amount就是这一个参数计算了多少次，计算次数是与参数的不同取值相对应的

③处理好数据后就把数据送入模型的进行预测任务。下图是模型的流程图

④最后输出的是一个值，如何通过这个值AI判断是指买入，持有，还是卖出。

本paper中使用的是一个聚类算法，即在把原数据进行第一差分，并且使用tanh激活函数后，进行聚类算法得到阈值，即可知道模型输出值来判断买入，持有，还是卖出。

代码实现部分 （CNN_1D)

由于任务是进行变压器的时序预测，因此不适用使用股票方面的知识，无法通过计算股票相关的指标进行维度扩充，因此使用的是CNN_1D

输入数据维度为 (batch_size,seq_len.d_feature)，其中batch_size为批量大小

seq_len为已知时间序列的长度，d_feature为特征维度

初始化模型

构建CNN的模型结构：

模型初始化部分的代码，和流程图不同的就是我没有使用dropout层，因为该模型对于本任务来说不会导致过拟合，因此没有使用dropout层，其他的就是和流程图的相同

class CNN_1D(nn.Module):
    def __init__(self,args,in_channels=7,in_channels_other=96,conv_kernel_size=33,conv_stride=1,out_channels1=32,out_channels2=64,out_channels3=128,pool_kernel_size=4,padding=16):
        '''
        Args:
        in_channels=7代表输入数据的第一个维度（特征维度）；in_channels_other=96代表输入数据的第二个维度（时间维度：已知多长的时间序列）
        conv_kernel_size=33是卷积核的大小，这里使用的三个卷积核大小都是一样的，conv_stride=1代表卷积核一次滑动的步长
        out_channels1=32是第一个卷积后的输出维度，out_channels2=64是第二个卷积后的输出维度，out_channels3=128是第三个卷积后的输出维度
        padding是卷积时候周围补零的数目，为了使的卷积出来的数据第二个维度即时间维度不改变，
        padding=(conv_kernel_size-1)/2，（当string为1时）
        pool_kernel_size=4是maxPool1d的核大小，并且maxPooling的步长默认和pooling核大小一样，
        '''
        self.seq_len = args.seq_len # seq_len 已知的时间序列的长度
        self.pred_len = args.pred_len # pred_len 是预测的时间序列的长度

        super(CNN_1D, self).__init__()
        # 输入是（batch_size,in_channels，seq_len），输出是（batch_size,out_channels1,seq_len）,扩充了特征维度
        self.conv1 = nn.Conv1d(in_channels=in_channels, out_channels=out_channels1, kernel_size=conv_kernel_size,padding=padding, stride=conv_stride)
        # 输入是（batch_size,out_channels1,seq_len） 输出是（batch_size,out_channels2,seq_len）
        self.conv2 = nn.Conv1d(in_channels=out_channels1, out_channels=out_channels2, kernel_size=conv_kernel_size,padding=padding, stride=conv_stride)
        # 输入是（batch_size,out_channels2,seq_len），输出是（batch_size,out_channels3，seq_len）
        self.conv3=nn.Conv1d(in_channels=out_channels2,out_channels=out_channels3,kernel_size=conv_kernel_size,padding=padding,stride=conv_stride)
        # 输入是（batch_size,out_channels3，seq_len）,输出是（batch_size,out_channels3,seq_len//pool_kernel_size） 压缩时间维度
        self.pool = nn.MaxPool1d(kernel_size=pool_kernel_size, stride=pool_kernel_size)
        # 输入是（batch_size,out_channels3,seq_len//pool_kernel_size），输出是(batch_size,out_channels3, 1） 对时间维度进行压缩
        self.fc1 = nn.Linear(in_channels_other//pool_kernel_size, 1)
        # 输入 (batch_size,1,out_channels3) 输出是（batch_size,1,out_channels2） 在特征维度上进行降维
        self.fc2 = nn.Linear(out_channels3, out_channels2)
        # 输入（batch_size,1, out_channels2）-->输出（batch_size,1，in_channels） 得到一天的预测结果
        self.fc3 = nn.Linear(out_channels2, in_channels) # 在特征维度上进行降维

前向传播

由于一次只预测一天，因此以下的代码就是根据输入数据预测下一天的时间序列，并且把得到的结果拼接回已知的时间序列中，然后输入窗口在时间序列上滑动，每一次滑动的步长为1，窗口的尺寸为(batch_size,seq_len,d_feature)，也就是代表每一次的输入的序列时间长度都为seq_len。

以下的代码是一次只预测一天，即就是预测给定时间窗口的下一天

其输入input_x就是时间窗口

def pred_onestep(self, input_x): # 每一次只预测一天
    #input_x [batch,seq_len,dim]
    # 因为conv1d 转为 input_x[batch,dim,seq_len]
    x=self.conv1(input_x.permute(0,2,1))# 对时间维度卷积,变为（batch_size,out_channels1,seq_len）
    x=torch.relu(x)
    x=self.conv2(x)# 对时间维度卷积,变为（batch_size,out_channels2,seq_len）
    x=torch.relu(x)
    x=torch.relu(self.conv3(x))# 对时间维度卷积，变为（batch_size,out_channels3，seq_len）
    x=self.pool(x) #输出变为（batch_size,out_channels3,seq_len//pool_kernel_size）
    x=self.fc1(x)#把预测一天的结果seq=1
    x=torch.relu(x)
    x=x.permute(0,2,1)#开始处理dim维度
    x=torch.relu(self.fc2(x)) # 输出是（batch_size,1,out_channels2）
    x=self.fc3(x) # 输出（batch_size,1，in_channels）

    return x

forward部分

本code只使用了enc_x，表示的是已知的时间序列的数据，不包括时间数据，其他的变量本code未使用，可以忽略

初始化预测结果pred_zero，初始化x_cat_pred用来装已知的时间序列和预测的时间序列

def forward(self, enc_x, enc_mark, y, y_mark): # 每一次预测一天，然后把这一次预测的结果拼到已知的时间序列后，然后在移动窗口，预测下一天的
    '''
    :param enc_x: 已知的时间序列 （batch_size,seq_len,dim）
    以下的param本model未使用，不做过多介绍
    :param enc_mark: 已知的时序序列的时间对应的时间矩阵，
    :param y:
    :param y_mark:
    :return:  x_cat_pred[:,-self.pred_len:,:] 将预测的时间序列的部分返回回去 (batch_size,pred)len,dim)
    '''

全零初始化预测结果，叫pred_zero

将已知的时间序列和全零初始化的预测结果拼起来 ，叫x_cat_pred，后面输入数据在该处取

pred_zero = torch.zeros_like(enc_x[:, -self.pred_len:, :]).float() # 初始化预测的结果,shape(batch_size,pred_len,dim)
# 将已知的时间序列和pred拼接在一起，
x_cat_pred = torch.cat([enc_x[:, :self.seq_len, :], pred_zero], dim=1).float().to(enc_x.device) # shape(batch_size , seq_len+pred_len , dim)

以下是循环进行单步预测，把单步预测的结果拼接起来然后组合成结果

input_x就是滑动窗口，滑动的步长为1，滑动次数取决于预测的时间序列的长度,这里类似于CNN_1D的滑动方式

得到输入数据，就送入pred_onestep函数中，可以改输入时间序列下一天的结果，然后把结果再拼接回之前已经初始化过的x_cat_pred中。然后继续滑动，即下一步的预测是建立在上一步预测的结果之上。

for i in range(self.pred_len): # 循环预测的时间序列的长度，因为一次只预测一天的
    input_x = x_cat_pred[:, i:i + self.seq_len, :].clone() # 得到每一次已知的时间序列，shape(batch_size,seq_len,dim)
    pred = self.pred_onestep(input_x) # 得到下一天的预测结果,shape(batch_size,1,dim)
    x_cat_pred[:, self.seq_len + i, :] = x_cat_pred[:, self.seq_len + i, :].clone() + pred.squeeze() # 将这一次的预测结果放入x_cat_pred中

当滑动次数和pred_len长度相同的时候，那么就停止滑动，把预测的部分的结果返回，与真实值进行loss计算，进行反向传播和更新梯度等操作

CNN_1D完整代码

import torch
import torch.nn as nn

#-------------------------------------------------------------------
#   一次取seq_len列，每一次滑动一列
#   输入模型的是（batch_size,seq_len,dim）,如果想要预测后pred_len天
#   可以现在后pred_len天的位置全部初始化为0，那么滚动一次，就填上后面的值
#   直至滑动到最后，那么就把后面的pred_len切分出来，然后在和目标值计算loss
# -------------------------------------------------------------------

class CNN_1D(nn.Module):
    def __init__(self,args,in_channels=7,in_channels_other=96,conv_kernel_size=33,conv_stride=1,out_channels1=32,out_channels2=64,out_channels3=128,pool_kernel_size=4,padding=16):
        '''
        Args:
        in_channels=7代表输入数据的第一个维度（特征维度）；in_channels_other=96代表输入数据的第二个维度（时间维度：已知多长的时间序列）
        conv_kernel_size=33是卷积核的大小，这里使用的三个卷积核大小都是一样的，conv_stride=1代表卷积核一次滑动的步长
        out_channels1=32是第一个卷积后的输出维度，out_channels2=64是第二个卷积后的输出维度，out_channels3=128是第三个卷积后的输出维度
        padding是卷积时候周围补零的数目，为了使的卷积出来的数据第二个维度即时间维度不改变，
        padding=(conv_kernel_size-1)/2，（当string为1时）
        pool_kernel_size=4是maxPool1d的核大小，并且maxPooling的步长默认和pooling核大小一样，
        '''
        self.seq_len = args.seq_len # seq_len 已知的时间序列的长度
        self.pred_len = args.pred_len # pred_len 是预测的时间序列的长度

        super(CNN_1D, self).__init__()
        # 输入是（batch_size,in_channels，seq_len），输出是（batch_size,out_channels1,seq_len）,扩充了特征维度
        self.conv1 = nn.Conv1d(in_channels=in_channels, out_channels=out_channels1, kernel_size=conv_kernel_size,padding=padding, stride=conv_stride)
        # 输入是（batch_size,out_channels1,seq_len） 输出是（batch_size,out_channels2,seq_len）
        self.conv2 = nn.Conv1d(in_channels=out_channels1, out_channels=out_channels2, kernel_size=conv_kernel_size,padding=padding, stride=conv_stride)
        # 输入是（batch_size,out_channels2,seq_len），输出是（batch_size,out_channels3，seq_len）
        self.conv3=nn.Conv1d(in_channels=out_channels2,out_channels=out_channels3,kernel_size=conv_kernel_size,padding=padding,stride=conv_stride)
        # 输入是（batch_size,out_channels3，seq_len）,输出是（batch_size,out_channels3,seq_len//pool_kernel_size） 压缩时间维度
        self.pool = nn.MaxPool1d(kernel_size=pool_kernel_size, stride=pool_kernel_size)
        # 输入是（batch_size,out_channels3,seq_len//pool_kernel_size），输出是(batch_size,out_channels3, 1） 对时间维度进行压缩
        self.fc1 = nn.Linear(in_channels_other//pool_kernel_size, 1)
        # 输入 (batch_size,1,out_channels3) 输出是（batch_size,1,out_channels2） 在特征维度上进行降维
        self.fc2 = nn.Linear(out_channels3, out_channels2)
        # 输入（batch_size,1, out_channels2）-->输出（batch_size,1，in_channels） 得到一天的预测结果
        self.fc3 = nn.Linear(out_channels2, in_channels) # 在特征维度上进行降维


    def pred_onestep(self, input_x): # 每一次只预测一天
        #input_x [batch,seq_len,dim]
        # 因为conv1d 转为 input_x[batch,dim,seq_len]
        x=self.conv1(input_x.permute(0,2,1))# 对时间维度卷积,变为（batch_size,out_channels1,seq_len）
        x=torch.relu(x)
        x=self.conv2(x)# 对时间维度卷积,变为（batch_size,out_channels2,seq_len）
        x=torch.relu(x)
        x=torch.relu(self.conv3(x))# 对时间维度卷积，变为（batch_size,out_channels3，seq_len）
        x=self.pool(x) #输出变为（batch_size,out_channels3,seq_len//pool_kernel_size）
        x=self.fc1(x)#把预测一天的结果seq=1
        x=torch.relu(x)
        x=x.permute(0,2,1)#开始处理dim维度
        x=torch.relu(self.fc2(x)) # 输出是（batch_size,1,out_channels2）
        x=self.fc3(x) # 输出（batch_size,1，in_channels）

        return x

    def forward(self, enc_x, enc_mark, y, y_mark): # 每一次预测一天，然后把这一次预测的结果拼到已知的时间序列后，然后在移动窗口，预测下一天的
        '''
        :param enc_x: 已知的时间序列 （batch_size,seq_len,dim）
        以下的param本model未使用，不做过多介绍
        :param enc_mark: 已知的时序序列的时间对应的时间矩阵，
        :param y:
        :param y_mark:
        :return:  x_cat_pred[:,-self.pred_len:,:] 将预测的时间序列的部分返回回去 (batch_size,pred)len,dim)
        '''
        pred_zero = torch.zeros_like(enc_x[:, -self.pred_len:, :]).float() # 初始化预测的结果,shape(batch_size,pred_len,dim)
        # 将已知的时间序列和pred拼接在一起，
        x_cat_pred = torch.cat([enc_x[:, :self.seq_len, :], pred_zero], dim=1).float().to(enc_x.device) # shape(batch_size , seq_len+pred_len , dim)

        for i in range(self.pred_len): # 循环预测的时间序列的长度，因为一次只预测一天的
            input_x = x_cat_pred[:, i:i + self.seq_len, :].clone() # 得到每一次已知的时间序列，shape(batch_size,seq_len,dim)
            pred = self.pred_onestep(input_x) # 得到下一天的预测结果,shape(batch_size,1,dim)
            x_cat_pred[:, self.seq_len + i, :] = x_cat_pred[:, self.seq_len + i, :].clone() + pred.squeeze() # 将这一次的预测结果放入x_cat_pred中

        return x_cat_pred[:,-self.pred_len:,:] # 返回总的预测的时间序列的结果,shape(batch_size,pred_len,dim)