### 使用 Informer 实现多金融时间序列预测 #### 1. 数据准备 为了使用 Informer 进行多金融时间序列预测，首先需要准备好合适的数据集。通常情况下，金融时间序列数据会包含多个特征（如开盘价、收盘价、最高价、最低价等），并且这些数据通常是按时间顺序排列的。 ```python import pandas as pd from sklearn.preprocessing import MinMaxScaler # 假设有一个 CSV 文件 'financial_data.csv' 包含了历史金融数据 data = pd.read_csv('financial_data.csv', parse_dates=['date'], index_col='date') # 对数据进行标准化处理 scaler = MinMaxScaler() scaled_data = scaler.fit_transform(data) # 将数据转换为适合模型输入的形式 def create_sequences(data, seq_length): xs, ys = [], [] for i in range(len(data)-seq_length): x = data[i:i+seq_length] y = data[i+seq_length] xs.append(x) ys.append(y) return np.array(xs), np.array(ys) sequence_length = 96 # 可以根据实际情况调整 X_train, y_train = create_sequences(scaled_data[:int(len(scaled_data)*0.7)], sequence_length) X_val, y_val = create_sequences(scaled_data[int(len(scaled_data)*0.7):], sequence_length) ``` #### 2. 构建 Informer 模型 接下来构建 Informer 模型。Informer 是一种改进版的 Transformer 结构，在长时记忆方面表现优异[^1]。 ```python import torch import torch.nn as nn from informer_model import Informer # 需要安装对应的库或自定义实现 class TimeSeriesPredictor(nn.Module): def __init__(self, enc_in=9, dec_in=9, c_out=9, seq_len=96, label_len=48, out_len=24, factor=5, d_model=512, n_heads=8, e_layers=2, d_ff=2048, dropout=0.0, activation='gelu'): super(TimeSeriesPredictor, self).__init__() self.informer = Informer(enc_in, dec_in, c_out, seq_len, label_len, out_len, factor=factor, d_model=d_model, n_heads=n_heads, e_layers=e_layers, d_ff=d_ff, dropout=dropout, activation=activation) def forward(self, x_enc, x_mark_enc, x_dec, x_mark_dec): output = self.informer(x_enc, x_mark_enc, x_dec, x_mark_dec) return output model = TimeSeriesPredictor().cuda() if torch.cuda.is_available() else TimeSeriesPredictor() criterion = nn.MSELoss() optimizer = torch.optim.Adam(model.parameters(), lr=1e-4) ``` #### 3. 训练过程 训练过程中需要注意的是，对于长时间序列来说，合理的超参数设置非常重要。比如学习率的选择、批次大小等都会影响最终的效果。 ```python batch_size = 32 num_epochs = 10 for epoch in range(num_epochs): model.train() total_loss = 0. for batch_idx in range((len(X_train) - 1) // batch_size + 1): start_idx = batch_idx * batch_size end_idx = min(start_idx + batch_size, len(X_train)) X_batch = torch.tensor(X_train[start_idx:end_idx]).float() y_batch = torch.tensor(y_train[start_idx:end_idx]).float() optimizer.zero_grad() outputs = model(X_batch[:, :-1].unsqueeze(-1).repeat(1, 1, 9), None, X_batch[:, -label_len:].unsqueeze(-1).repeat(1, 1, 9), None) loss = criterion(outputs.squeeze(), y_batch) loss.backward() optimizer.step() total_loss += loss.item() avg_loss = total_loss / ((len(X_train) - 1) // batch_size + 1) print(f"Epoch {epoch}, Loss: {avg_loss}") ``` #### 4. 测试与评估 完成训练之后可以对验证集上的性能进行测试，并通过可视化等方式来直观展示预测效果。 ```python with torch.no_grad(): predictions = [] actuals = [] for idx in range(len(X_val)): input_seq = torch.tensor([X_val[idx]]).float() pred = model(input_seq[:, :-1].unsqueeze(-1).repeat(1, 1, 9), None, input_seq[:, -label_len:].unsqueeze(-1).repeat(1, 1, 9), None) predictions.extend(pred.cpu().numpy()) actuals.extend(y_val[idx]) predictions = scaler.inverse_transform(predictions) actuals = scaler.inverse_transform(actuals) plt.figure(figsize=(15, 5)) plt.plot(range(len(predictions)), predictions, color="blue", label="Prediction") plt.plot(range(len(actuals)), actuals, color="red", linestyle="--", label="Actual Value") plt.legend(loc="best") plt.show() ```