### 使用 Transformers 进行长短期记忆的时间序列预测 为了实现长时间跨度下的时间序列预测，基于 Transformer 的架构因其并行处理能力和捕捉长距离依赖的能力而成为理想的选择。下面是一个简化版的 Python 实现案例，该例子展示了如何构建一个用于多步前向预测的时间序列模型。 #### 数据预处理阶段 考虑到时间序列数据的特点，在将其送入 Transformer 模型之前，通常会先转换成矩阵形式[^1]： ```python import numpy as np from sklearn.preprocessing import StandardScaler def create_dataset(dataset, look_back=1): dataX, dataY = [], [] for i in range(len(dataset)-look_back-1): a = dataset[i:(i+look_back), 0] dataX.append(a) dataY.append(dataset[i + look_back, 0]) return np.array(dataX), np.array(dataY) # 加载您的时间序列数据集 data = ... # 替换成您自己的时间序列数据加载方式 scaler = StandardScaler() scaled_data = scaler.fit_transform(data.reshape(-1, 1)) train_size = int(len(scaled_data) * 0.8) test_size = len(scaled_data) - train_size train, test = scaled_data[0:train_size,:], scaled_data[train_size:len(scaled_data),:] look_back = 100 trainX, trainY = create_dataset(train, look_back) testX, testY = create_dataset(test, look_back) ``` #### 构建 Transformer 模型结构 接下来定义一个简单的 Transformer 编码器层来处理上述准备好的训练样本。这里特别注意要加入位置编码以保留原始时间戳的信息[^2]: ```python import torch import torch.nn as nn import math class PositionalEncoding(nn.Module): def __init__(self, d_model, max_len=5000): super(PositionalEncoding, self).__init__() pe = torch.zeros(max_len, d_model) position = torch.arange(0, max_len, dtype=torch.float).unsqueeze(1) div_term = torch.exp(torch.arange(0, d_model, 2).float() * (-math.log(10000.0) / d_model)) pe[:, 0::2] = torch.sin(position * div_term) pe[:, 1::2] = torch.cos(position * div_term) pe = pe.unsqueeze(0).transpose(0, 1) self.register_buffer('pe', pe) def forward(self, x): return x + self.pe[:x.size(0), :] class TransformerModel(nn.Module): def __init__(self, input_dim, nhead, nhid, nlayers, dropout=0.5): super(TransformerModel, self).__init__() from torch.nn import TransformerEncoder, TransformerEncoderLayer self.model_type = 'Transformer' self.src_mask = None self.pos_encoder = PositionalEncoding(input_dim) encoder_layers = TransformerEncoderLayer(input_dim, nhead, nhid, dropout) self.transformer_encoder = TransformerEncoder(encoder_layers, nlayers) self.encoder = nn.Linear(look_back, input_dim) self.decoder = nn.Linear(input_dim, 1) self.init_weights() def init_weights(self): initrange = 0.1 self.encoder.weight.data.uniform_(-initrange, initrange) self.decoder.bias.data.zero_() self.decoder.weight.data.uniform_(-initrange, initrange) def forward(self, src): src = self.encoder(src) * math.sqrt(nhid) src = self.pos_encoder(src) output = self.transformer_encoder(src) output = self.decoder(output[-1]) return output ``` #### 训练过程配置与执行 最后一步是设置优化算法以及损失函数，并启动迭代更新循环完成整个学习流程: ```python device = torch.device("cuda" if torch.cuda.is_available() else "cpu") input_dim = 200 nhead = 2 nhid = 200 nlayers = 2 dropout = 0.2 model = TransformerModel(input_dim, nhead, nhid, nlayers, dropout).to(device) criterion = nn.MSELoss() lr = 0.001 optimizer = torch.optim.Adam(model.parameters(), lr=lr) batch_size = 64 for epoch in range(epochs): model.train() total_loss = 0. start_time = time.time() for batch, i in enumerate(range(0, len(trainX) - 1, batch_size)): data = torch.tensor(trainX[i:i+batch_size]).float().permute(1,0,2).to(device) target = torch.tensor(trainY[i:i+batch_size]).float().view(-1,1).to(device) optimizer.zero_grad() output = model(data) loss = criterion(output, target) loss.backward() torch.nn.utils.clip_grad_norm_(model.parameters(), 0.5) optimizer.step() total_loss += loss.item() ```