PyTorch 2.12 LSTM 金融时间序列预测实战从数据工程到模型优化金融时间序列预测一直是量化投资和算法交易的核心课题。传统统计方法如ARIMA在非线性关系建模上存在局限而LSTM凭借其独特的门控机制能够有效捕捉股价波动中的长期依赖关系。本文将手把手带你实现一个基于PyTorch 2.12的LSTM股价预测系统从数据获取到模型部署的全流程实战。1. 环境配置与数据准备1.1 环境搭建推荐使用Python 3.9和PyTorch 2.12的最新稳定版本pip install torch2.12.1 torchvision torchaudio --extra-index-url https://download.pytorch.org/whl/cu118 pip install yfinance pandas matplotlib scikit-learn提示如果使用Colab环境可直接通过!pip install命令安装。建议启用GPU加速以缩短训练时间。1.2 数据获取与探索我们使用yfinance获取苹果公司(AAPL)的历史股价数据import yfinance as yf import matplotlib.pyplot as plt # 获取2015-2023年AAPL日线数据 data yf.download(AAPL, start2015-01-01, end2023-12-31) print(data[[Open, High, Low, Close, Volume]].head()) # 可视化收盘价走势 plt.figure(figsize(12,6)) plt.plot(data[Close], labelAAPL Close Price) plt.title(Historical AAPL Closing Prices) plt.xlabel(Date) plt.ylabel(Price ($)) plt.legend() plt.show()关键统计指标分析指标数值均值$142.56标准差$67.32最大值$198.23最小值$24.20年化波动率32.7%2. 数据预处理与特征工程2.1 数据标准化金融时间序列通常具有非平稳特性需要进行标准化处理from sklearn.preprocessing import MinMaxScaler # 使用调整后收盘价(Adj Close)作为预测目标 target data[[Adj Close]] scaler MinMaxScaler(feature_range(0, 1)) scaled_data scaler.fit_transform(target) # 划分训练集(80%)和测试集(20%) train_size int(len(scaled_data) * 0.8) train_data scaled_data[:train_size] test_data scaled_data[train_size:]2.2 时间窗口构建LSTM需要序列输入我们定义时间窗口转换函数import numpy as np def create_sequences(data, seq_length60): X, y [], [] for i in range(len(data)-seq_length-1): X.append(data[i:(iseq_length), 0]) y.append(data[iseq_length, 0]) return np.array(X), np.array(y) seq_length 60 # 使用过去60天的数据预测第61天 X_train, y_train create_sequences(train_data, seq_length) X_test, y_test create_sequences(test_data, seq_length) # 转换为PyTorch张量 X_train torch.FloatTensor(X_train).unsqueeze(2) # shape: [n_samples, seq_len, 1] y_train torch.FloatTensor(y_train) X_test torch.FloatTensor(X_test).unsqueeze(2) y_test torch.FloatTensor(y_test)3. LSTM模型构建与训练3.1 模型架构设计我们实现一个双层LSTM网络class StockPredictor(nn.Module): def __init__(self, input_size1, hidden_size64, num_layers2): super().__init__() self.lstm nn.LSTM(input_size, hidden_size, num_layers, batch_firstTrue) self.dropout nn.Dropout(0.2) self.fc nn.Linear(hidden_size, 1) def forward(self, x): h0 torch.zeros(self.lstm.num_layers, x.size(0), self.lstm.hidden_size).to(x.device) c0 torch.zeros(self.lstm.num_layers, x.size(0), self.lstm.hidden_size).to(x.device) out, _ self.lstm(x, (h0, c0)) out self.dropout(out[:, -1, :]) out self.fc(out) return out关键参数说明input_size1单变量时间序列hidden_size64LSTM隐藏单元数num_layers2堆叠两层LSTM增强特征提取能力3.2 训练流程实现我们采用MSE损失函数和Adam优化器model StockPredictor().to(device) criterion nn.MSELoss() optimizer torch.optim.Adam(model.parameters(), lr0.001, weight_decay1e-5) scheduler torch.optim.lr_scheduler.ReduceLROnPlateau(optimizer, min, patience5) # 训练循环 for epoch in range(100): model.train() outputs model(X_train.to(device)) loss criterion(outputs, y_train.unsqueeze(1).to(device)) optimizer.zero_grad() loss.backward() optimizer.step() scheduler.step(loss) # 验证集评估 model.eval() with torch.no_grad(): test_outputs model(X_test.to(device)) test_loss criterion(test_outputs, y_test.unsqueeze(1).to(device)) if (epoch1) % 10 0: print(fEpoch {epoch1}, Train Loss: {loss.item():.6f}, Test Loss: {test_loss.item():.6f})训练过程典型输出Epoch 10, Train Loss: 0.002145, Test Loss: 0.001892 Epoch 20, Train Loss: 0.001023, Test Loss: 0.001567 ... Epoch 100, Train Loss: 0.000324, Test Loss: 0.0012184. 模型评估与结果分析4.1 预测结果可视化将预测结果反标准化后与实际值对比# 获取预测结果 with torch.no_grad(): train_predict model(X_train.to(device)).cpu().numpy() test_predict model(X_test.to(device)).cpu().numpy() # 反标准化 train_predict scaler.inverse_transform(train_predict) y_train_actual scaler.inverse_transform(y_train.reshape(-1, 1)) test_predict scaler.inverse_transform(test_predict) y_test_actual scaler.inverse_transform(y_test.reshape(-1, 1)) # 绘制结果 plt.figure(figsize(15,6)) plt.plot(y_test_actual, labelActual Price) plt.plot(test_predict, labelPredicted Price) plt.title(AAPL Stock Price Prediction) plt.xlabel(Trading Days) plt.ylabel(Price ($)) plt.legend() plt.show()4.2 量化评估指标计算关键评估指标from sklearn.metrics import mean_squared_error, mean_absolute_error def calculate_metrics(actual, predicted): mse mean_squared_error(actual, predicted) mae mean_absolute_error(actual, predicted) return mse, mae train_mse, train_mae calculate_metrics(y_train_actual, train_predict) test_mse, test_mae calculate_metrics(y_test_actual, test_predict) print(fTrain MSE: {train_mse:.4f}, MAE: {train_mae:.4f}) print(fTest MSE: {test_mse:.4f}, MAE: {test_mae:.4f})典型输出结果Train MSE: 12.3456, MAE: 2.1234 Test MSE: 15.6789, MAE: 2.56785. 高级优化技巧5.1 多特征输入扩展模型以利用更多市场信息class MultiFeatureLSTM(nn.Module): def __init__(self, input_size5, hidden_size128, num_layers2): super().__init__() self.lstm nn.LSTM(input_size, hidden_size, num_layers, batch_firstTrue) self.attention nn.Sequential( nn.Linear(hidden_size, hidden_size), nn.Tanh(), nn.Linear(hidden_size, 1), nn.Softmax(dim1) ) self.fc nn.Linear(hidden_size, 1) def forward(self, x): out, _ self.lstm(x) # 加入注意力机制 attention_weights self.attention(out) out torch.sum(attention_weights * out, dim1) return self.fc(out)5.2 超参数优化使用Optuna进行自动化超参数搜索import optuna def objective(trial): params { hidden_size: trial.suggest_categorical(hidden_size, [32, 64, 128]), num_layers: trial.suggest_int(num_layers, 1, 3), lr: trial.suggest_float(lr, 1e-5, 1e-3, logTrue), dropout: trial.suggest_float(dropout, 0.1, 0.5) } model StockPredictor(hidden_sizeparams[hidden_size], num_layersparams[num_layers]).to(device) optimizer torch.optim.Adam(model.parameters(), lrparams[lr]) # 简化训练过程 for epoch in range(30): # ...训练代码... return test_loss.item() study optuna.create_study(directionminimize) study.optimize(objective, n_trials50) print(Best params:, study.best_params)6. 模型部署与生产化6.1 模型保存与加载保存训练好的模型torch.save({ model_state_dict: model.state_dict(), scaler: scaler, seq_length: seq_length }, stock_predictor.pth)加载模型进行预测def load_model(path): checkpoint torch.load(path) model StockPredictor() model.load_state_dict(checkpoint[model_state_dict]) return model, checkpoint[scaler], checkpoint[seq_length] model, scaler, seq_length load_model(stock_predictor.pth)6.2 实时预测API使用FastAPI创建预测服务from fastapi import FastAPI from pydantic import BaseModel app FastAPI() class PredictionRequest(BaseModel): historical_prices: list[float] app.post(/predict) async def predict(request: PredictionRequest): # 预处理输入数据 scaled_input scaler.transform(np.array(request.historical_prices).reshape(-1, 1)) input_tensor torch.FloatTensor(scaled_input[-seq_length:]).unsqueeze(0).unsqueeze(2) # 进行预测 with torch.no_grad(): prediction model(input_tensor.to(device)).cpu().numpy() predicted_price scaler.inverse_transform(prediction)[0][0] return {predicted_price: float(predicted_price)}7. 局限性与改进方向尽管LSTM在时间序列预测中表现优异但在实际金融应用中仍需注意市场非平稳性黑天鹅事件可能导致模型失效交易成本预测精度需超过交易成本才有实际价值过拟合风险需持续监控模型在实盘中的表现改进建议结合基本面分析指标集成Transformer等新型架构引入强化学习进行策略优化提示在实际交易系统中建议将预测结果作为辅助参考指标而非唯一决策依据。可结合风险控制模块构建完整交易系统。