从0开始基于transformer进行股价预测（pytorch版本）

数据阶段

！！！注意！！！，本文不会讲原理，因为之前两篇文章已经讲过了，只会解释一些结构性问题，和思路问题。

所谓工欲善其事，必先利其器做量化分析的股价预测，完美必须要先把数据处理好。
那么本人的数据下载是在聚宽平台股票代码为601398的数据2014-3 到 2024-3年的默认数据。如何下载可以按照我的方式

在这里插入图片描述
进入研究环境后随便创建一个ipynb文件进行数据下载，运行以下代码就行

# 1.获取数据
data = get_price('601398.XSHG', start_date='2014-01-01', end_date='2024-01-01', frequency='daily', fields=None, skip_paused=False, fq='pre', panel=True)
# 2.保存数据
data.to_csv('data_沪深300/601398.XSHG(工商银行14-24).csv')

两个问题

1.为什么我们只需要用encoder部分去预测就行而不需要decoder部分？
答：编码器用于将输入序列编码成一个上下文表示（contextual representation），然后解码器根据该上下文表示生成目标序列在时间序列预测任务中，我们不需要生成一个序列，而是预测单个或少量几个未来数据点。因此，编码器的上下文表示已经包含了足够的信息来进行预测，无需使用解码器。还有我觉得使用解码器的意思是，你用上一天的数据去预测下一天的数据，我感觉这样就没意思了，这和我们个人看有什么区别。而且对最后的结果也会造成不精准的效果。为什么这么说呢，你看解码器的mask编码部分应该可以理解了。
2.我们的维度为什么不是[batch, len, feature]? 因为这是pytorch要求，自己能实现的话，自己改吧。

开始利用我们的代码进行切分

我的思路用的是用五天的数据去预测下一天，数据集和测试及8/2分
但是我们要记住一点，就是我们必须要理解我们这么做的思路，就比如我们的特征有6列分别是,open,close,high,low,volume,money,我们可以通过训练得到我们想预测的某一特征。OK，我们这就开始。

说起数据分割里面的代码不难，最难的是
for i in range(len(X_CONVERT) - seq_length):
X_data.append(X_CONVERT[i:i+seq_length, :])
y_data.append(X_CONVERT[i+seq_length, 1])
你要知道我在干什么，就是用8成的数据集去预测得到我们所需要的train数据集和我们对应train数据集的label，举个例子就是，我们要炒菜，我们拿上原料后我们要知道炒的什么菜，那么菜单必须要知道。是吧，不然你炒完菜后说是红烧肉，但是没有菜单图片对比你怎么知道这是红烧肉？这也就是这一步的意义。

def split_data(batch_size,seq_length, pred_length, train_ratio):data_all = pd.read_csv(data_path)data_ha = []length = len(data_all)# 将数据转换为numpy数组，并添加到列表中for element in elements:data_element = data_all[element].values.astype(np.float32)data_element = data_element.reshape(length, 1)data_ha.append(data_element)X_hat = np.concatenate(data_ha, axis=1)X_CONVERT = torch.from_numpy(X_hat).float()X_CONVERT = X_CONVERT.flip(dims=[0])# 进行归一化min_val = np.min(X_hat, axis=0)max_val = np.max(X_hat, axis=0)X_normalized = (X_hat - min_val) / (max_val - min_val)X_CONVERT = torch.from_numpy(X_normalized).float()X_CONVERT = X_CONVERT.flip(dims=[0])#数据翻转# 划分训练集和验证集X_data = []y_data = []for i in range(len(X_CONVERT) - seq_length):#划分的时候是用8成的训练集去训练然后label是某##一列X_data.append(X_CONVERT[i:i+seq_length, :])y_data.append(X_CONVERT[i+seq_length, 1])X_data = torch.stack(X_data)y_data = torch.stack(y_data).squeeze(-1)print(X_data.shape, y_data.shape)dataset = TensorDataset(X_data, y_data)train_size = int(len(dataset) * train_ratio)val_size = len(dataset) - train_sizetrain_dataset, val_dataset = random_split(dataset, [train_size, val_size])train_loader = DataLoader(train_dataset, batch_size, shuffle=False)val_loader = DataLoader(val_dataset, batch_size, shuffle=False)return train_loader, val_loader,min_val, max_val

backbone网络

如其名，我们都知道这是这是transformer当然是用的transformer的结构。但是我们用，但是只用一部分，具体用什么部分开头说了，只用encoder

**但是具体操作起来的时候encoder里面的embadding部分我们需要修改，因为我们不是机器翻译，所以我们不需要把他变成词向量，我们时间序列数据，输入通常是连续的数值特征，使用线性层更直接地将这些数值特征映射到高维空间。并且我们的embadding嵌入层，适用于离散的输入，输出是固定维度的嵌入向量。而线性层，适用于连续的输入，可以灵活处理不同维度的输入特征，将其映射到高维表示。**具体看下面代码

class Encoder(nn.Module):def __init__(self):super(Encoder, self).__init__()self.src_emb = nn.Linear(feature, d_model)#这里替换了self.pos_emb = PositionalEncoding(d_model)self.layers = nn.ModuleList([EncoderLayer() for _ in range(n_layers)])def forward(self, enc_inputs):enc_outputs = self.src_emb(enc_inputs)  # [batch_size, src_len, d_model]enc_outputs = self.pos_emb(enc_outputs)  # [batch_size, src_len, d_model]enc_self_attn_mask = get_attn_pad_mask(enc_inputs, enc_inputs)  # [batch_size, src_len, src_len]enc_self_attns = []for layer in self.layers:enc_outputs, enc_self_attn = layer(enc_outputs, enc_self_attn_mask)enc_self_attns.append(enc_self_attn)return enc_outputs, enc_self_attns

当然其他的部分和我上一篇的一样，但是就是decode不要了，当然也可以换成其他结果，或者加个注意力机制

讲下各个参数


d_model = 512   # linnerer的输入维度 也就是字embedding的维度
d_ff = 2048     # 前向传播隐藏层维度
d_k = d_v = 64  # K(=Q), V的维度
n_layers = 6    # 有多少个encoder和decoder
n_heads = 8     # Multi-Head Attention设置为8
feature=6       # 输入特征维度

当然主体还是要看一下的最重要的是通过encoder后的维度转换比较繁琐，要和我们之前split的数据集得到的y_train一致这样才能计算损失


class Transformer(nn.Module):def __init__(self):super(Transformer, self).__init__()self.Encoder = Encoder()self.projection = nn.Linear(d_model, 1, bias=False)def forward(self, enc_inputs):  # enc_inputs: [batch_size, src_len, feature]enc_outputs, enc_self_attns = self.Encoder(enc_inputs)  # enc_outputs: [batch_size, src_len, d_model]dec_logits = self.projection(enc_outputs)  # dec_logits: [batch_size, src_len, 1]dec_logits = dec_logits.mean(dim=1)  # 将每个时间步的预测结果取平均，得到 [batch_size, 1]return dec_logits.squeeze(-1), enc_self_attns  # 输出 [batch_size]

训练

先解释参数

batch_size=64#批处理大小
seq_length=7#时间序列长度 也就是通过seq_length天预测后面pred_length天
pred_length=1#预测长度
train_ratio=0.8#训练集比例
epochs = 50 # 训练轮数
lr= 0.001 # 学习率
png_save_path="diytransformers/12.24transformer/picture"#所有的图片保存的地方
loss_history = []# 存储每个 epoch 的损失

训练代码很长，挺简单的


# 训练模型
for epoch in range(epochs):epoch_loss = 0y_pre = []y_true = []# 训练阶段for X, y in train_loader:X = X.float()  # 确保输入数据类型为float32y = y.float()  # 确保目标数据类型为float32outputs, enc_self_attns = model(X)# 计算损失，确保形状一致loss = criterion(outputs, y)epoch_loss += loss.item()optimizer.zero_grad()loss.backward()torch.nn.utils.clip_grad_norm_(model.parameters(), 0.5)optimizer.step()#转换我们的label和训练后得到的训练集的预测值 y_pre.append(outputs.detach())y_true.append(y.detach())avg_loss = epoch_loss / len(train_loader)loss_history.append(avg_loss)#获得最好的lossif avg_loss < best_loss:best_loss = avg_lossbest_epoch = epochbest_model_wts = copy.deepcopy(model.state_dict())torch.save(best_model_wts, path_train)y_pre_concat = torch.cat(y_pre, dim=0)y_true_concat = torch.cat(y_true, dim=0)# 计算并打印评估指标metrics = evaluate(y_pre_concat, y_true_concat, min_val, max_val)print(f'Epoch {epoch + 1}, Loss: {avg_loss:.6f}')# 可视化结果ht(y_true_concat.detach().cpu().numpy(), y_pre_concat.detach().cpu().numpy(), min_val, max_val,png_save_path)

最后是看我们的一些指标效果如何比如这里我计算的mae，rmse，pcc等

# 加载最佳模型权重
model.load_state_dict(torch.load(train_over_path))# 测试模型并计算评估指标
test_metrics = test_model(model, val_loader, min_val, max_val)print(f'Test Metrics: {test_metrics}')