深度学习做股票预测靠谱吗？

背景

趁着完成了本周assignments来佛系更新一下。其实本文模型的原型是ALSTM(Attentive LSTM), 是机器学习选股常见的，也是微软qlib自带的一个baseline，只不过本文作者为了降低模型的过拟合并提高泛化能力，在每一次Iteration里面增添了对抗样本 e_{adv} 以及对应的损失函数部分。我对网络的完整架构和单次BP的过程做了手动复现。由于未参考源代码，或存在部分不妥之处欢迎交流指正。

原论文：《 Enhancing Stock Movement Prediction with Adversarial Training》

作者将该问题抽象成一个经典的低频机器学习预测涨跌的分类模型（可参考我往期的若干篇文章），对于股票 s ,给定时间戳 t 的特征矩阵 X_s \in \mathbb{R}^{T\times D} ,希望训练模型预测该股票在下个持仓期的涨跌标签：

\hat{y_s} = f(X_s,\Theta) \in \{-1,1\}

由于股票市场高随机波动特性，传统的机器学习模型通常会表现出较差的泛化能力，尽管会添加如L2正则化这样的方法，还是会难以避免出现模型过拟合的情况。比如，现在很多模型所喂入的都是短期价量数据（可见此前我的文章AlphaNet因子挖掘网络），学习短期价量特征和收益率的关系，但是这种price-based features是随时间演化的高度随机变量，可以认为是从时变联合分布中不停采样的一组样本，为了适应这种随机性，作者在ALSTM的基础上引入了对抗学习，提升了模型在不同股票数据集上的表现。因此我的复现也基于两大部分：ALSTM的架构和对抗学习部分。

ALSTM模型的整体架构如下：

最后的损失函数架构：

首先是定义一些参数：

import torch 
import torch.nn as nn
from torch.utils.data import Dataset
from torch.utils.data import DataLoader
from torch import optim
import torch.nn.functional as F
batch_size = 100
D = 40 #原始的单日的特征维度数
T = 30 #单个样本用于预测所使用的数据的天数
E = 30 #映射后的特征维度数
U = 10 #LSTM网络中的隐藏维度数
Q = 10
num_lstmlayer = 1#LSTM的层数
beta = 1#对抗学习损失函数部分的权重
#模拟一个batch的数据输入
origin_input = torch.randn(batch_size,T,D)
y_label = torch.ones(batch_size)

Attentive LSTM

整个模型被拆分成四个主要部分：

• Feature Mapping Layer : 对于一个Batch的原始输入，通过一个fc层映映射成laten space里的特征表示：

m_t^s = tanh(W_mx_t^s+b_m) \in \mathbb{R}^{T \times E}

mapping_layer = nn.Linear(D,E,bias = True)
origin_input = origin_input.flatten(start_dim = 0,end_dim = 1)
tanh = nn.Tanh()
mapping_feature = tanh(mapping_layer(origin_input)
mapping_feature = mapping_feature.reshape(batch_size,T,-1)
mapping_feature = mapping_feature.transpose(0,1)

• LSTM Layer: 捕捉时序上的依赖关系，这可以表示成：

h_t^s = LSTM(m_t^s,h_{t-1}^s),\quad [h_1^s,...h_T^s] \in \mathbb{R}^{T \times U}

LSTM_layer = nn.LSTM(E,U,num_lstmlayer)
#初始化参数
h_0 = torch.randn(num_lstmlayer,batch_size,U)
c_0 = torch.randn(num_lstmlayer,batch_size,U)
output, (hn, cn) = LSTM_layer(mapping_feature, (h_0, c_0))
#output: T*batch*U
output = output.transpose(0,1)
hn = hn.squeeze(0) #now turn to batchsize * U
hn.shape
output.shape #100*30*10

注意PyTorch里的LSTM层的输入参数，batch_size是在中间的，RNN也是这样。

• Temporal Attention Layer: 老生常谈的话题了，通过一个注意力层聚合同一个股票不同timesteps的特征信息得到该只股票的最终向量表示。这里直接放出公式：

\mathbf{a}^s = \sum_{i =1}^T a_t^s \mathbf{h}_t^s, \\ a_t^s = \frac{e^{\hat{a}_t^s}}{\sum_{i=1}^T e^{\hat{a}_t^s}},\\ \hat{a}_t^s = u_a^Ttanh(W_a \mathbf{h}_t^s+b_a)

对应的代码：

#Temporal attention layer
h_s = output.flatten(start_dim = 0,end_dim = 1)
layer = nn.Linear(U,Q,bias = True)
x = layer(h_s)
tanh = nn.Tanh()
x = tanh(x)
layer_2 = nn.Linear(Q,1,bias = False)
x = layer_2(x).squeeze(1)
x = x.reshape(batch_size,T)
weighted_vector = torch.exp(x)/torch.sum(torch.exp(x),dim = 1).unsqueeze(1)#weight of each time stamp
out = output.transpose(1,2)
weighted_vector = weighted_vector.unsqueeze(2)
a_s = torch.bmm(out,weighted_vector).squeeze(2)

• Prediction Layer ：注意，个股最终的向量表示是 a^s 和LSTM层最后一个timestamp输出cat的结果：

\mathbf{e}^s = [\mathbf{a}^s,\mathbf{h}_T^s]^T \in \mathbb{R}^{2U}

最终的涨跌标签预测：

\hat{y}^s = sign(W_P \mathbf{e}^s+b_P) \in\{-1,1\}

h_T = output[:,-1,:]
e_s = torch.cat([a_s,h_T],dim=1)
final_map = nn.Linear(e_s.shape[-1],1,bias = True)
y_s = torch.sign(final_map(e_s))#classification result

完整的模型实现（代码有模糊处理）

class ALSTM(nn.Module):
    def __init__(self,T,D,E,U,Q,num_lstmlayer):
        super(ALSTM,self).__init__()
        self.D = D
        self.T = T 
        self.E = E 
        self.U = U
        self.Q = Q
        self.num_lstmlayer = num_lstmlayer
        self.mapping_layer = nn.Linear(D,E,bias = True)
        self.LSTM_layer = nn.LSTM(E,U,num_lstmlayer)
        self.layer = nn.Linear(U,Q,bias = True)
        self.layer_2 = nn.Linear(Q,1,bias = False) 
        self.tanh = nn.tanh()
    def forward(self,origin_input):
        batch_size = origin_input.shape[0]
        origin_input = origin_input.flatten(start_dim = 0,end_dim = 1)
        mapping_feature = self.tanh(mapping_layer(origin_input))
        mapping_feature = mapping_feature.reshape(batch_size,T,-1)
        mapping_feature = mapping_feature.transpose(0,1)
        #LSTM Embedding
        h_0 = torch.rand(self.num_lstmlayer,batch_size,self.U)
        c_0 = torch.randn(self.num_lstmlayer,batch_size,self.U)
        output, (hn, cn) = self.LSTM_layer(mapping_feature, (h_0, c_0))
        output = output.transpose(0,1)
        hn = hn.squeeze(0) #now turn to batchsize * U
        #Temporal attention layer
        h_s = output.flatten(start_dim = 0,end_dim = 1)
        x = self.layer(h_s)
        x = self.tanh(x) 
        x = self.layer_2(x).squeeze(1)
        x = x.reshape(batch_size,self.T) 
        weighted_vector = torch.exp(x)/torch.sum(torch.exp(x),dim = 1).unsqueeze(1)#weight of each time stamp
        #feature aggreagation and final output
        out = output.transpose(1,2)
        weighted_vector = weight_vector.unsqueeze(2)
        a_s = torch.bmm(out,weighted_vector).squeeze(2)
        h_T = output[:,-1,:]#last output of the hidden layers of the LSTM block
        es = torch.cat([a_s,h_T],dim=1)
        return es

Adversarial Training

针对分类问题，本文的损失函数主要基于Hinge Loss：

l(y^s,\hat{y}^s) = max(0,1-y^s*\hat{y}^s)

因此我先实现了一个类：

class HingeLoss(nn.Module):
    def __init__(self):
        super(HingeLoss,self).__init__()
    def forward(self,y_label,y_predict):
        #y_label,y_predict都是长度为batch_size的一维tensor
        if y_label.shape == y_predict.shape:
            loss = torch.sum(torch.max(torch.zeros_like(y_label),torch.ones_like(y_label)-y_label * y_predict))
            return loss
        else:
            raise Exception("The size of label and predicted value is not equal !")

所谓对抗学习，即在每次迭代训练的时候，给一个batch内的数据添加一个自适应的perturbation，相应的仍然计算这些攻击样本的输出并纳入损失函数，从而提高模型的泛化能力。基于前文给出的最终向量表示 e^s , 先计算一次hinge loss，并计算loss对于 e^s 的梯度，从而构造对抗样本：

e_{adv}^s = e^s + \epsilon \frac{g^s}{||g^s||}, \quad g^s = \frac{\partial l(y^s,\hat{y}^s)}{\partial e^s}

#构造对抗样本函数
def get_adv(origin_input,y_label,final_map,model,epsilon,criterion):
    origin_input: batch_size * T * feature_dim
    y_label: batch_size
    final_map: final mapping layer
    model: Attentive lSTM
    epsilon: learning rate to control the adv examples
    criterion: loss function
    e_s = model(origin_input)
    e_s.retain_grad()
    y_s = torch.sign(final_map(e_s)).squeeze(1)
    loss_1 = criterion(y_s,y_label)
    g_s = torch.autograd.grad(outputs = loss_1,inputs=e_s,grad_outputs=None)[0]
    g_snorm = torch.sqrt(torch.norm(g_s,p = 2))
    if g_snorm == 0:
        return 0
    else:
        r_adv = epsilon(g_s/g_snorm)
        return r_adv.detach()

从而得到对抗样本的输出 \hat{y_{adv}}^s 。最终的损失函数可以表示为：

\Gamma_{adv} = \sum_{s = 1}^S l(y^s,\hat{y}^s) + \beta\sum_{s = 1}^S l(y^s,\hat{y}_{adv}^s) + \frac{\alpha}{2} ||\Theta||_F^2

最后一项是模型复杂度的L2正则化，单次BP的过程如下(代码经过模糊处理）：

#input batch data
origin_input = torch.randn(batch_size,T,D)
y_label = torch.ones(batch_size)
#hyper parameters
batch_size = 100
D = 40 #原始的单日的特征维度数
T = 30 #单个样本用于预测所使用的数据的天数
E = 30 #映射后的特征维度数
U = 10 #LSTM网络中的隐藏维度数
Q = 10
num_lstmlayer = 1
beta = 1 
#building model
model = ALSTM(T,D,E,U,Q,num_lstmlayer)
final_map = nn.Linear(2*U,1,bias = True)
criterion = HingeLoss()
#L2 normalization
weight_list,bias_list = [],[]
for name,p in model.named_parameters():
    if 'bias' in name:
        bias_list += [p]
    else:
        weight_list += [p]
optimizer = optim.SGD([{'params': weight_list, 'weight_decay':1e-5},
                       {'params': bias_list, 'weight_decay':0}],
                      lr = 1e-3,
                      momentum = 0.9)
#初始化：Xavier Initialization
for n in model.modules():  
    if isinstance(n,nn.Linear): #线性全连接层初始化 
        n.weight = nn.init.xavier_normal_(n.weight, gain=1.)   
#forward propogation
#得到e_s
e_s = model(origin_input)
y_s = torch.sign(final_map(e_s)).unsqueeze(1)
r_adv = get_adv(origin_input,y_label,final_map,model,1e-2,criterion)
e_adv = e_s + r_adv
y_adv = torch.sign(final_map(e_adv)).squeeze(1)
#计算loss