LSTM外汇价格预测 - 精简版

本Notebook包含LSTM模型训练的核心代码，用于预测EURUSD价格。

主要步骤

导入库和设置设备
从MT5获取历史数据
数据归一化
构造训练/验证数据集
定义LSTM模型
训练模型（早停机制）
可视化训练过程
导出ONNX模型

1. 导入库和设置设备

In [ ]:

import torch
import torch.nn as nn
import numpy as np
import pandas as pd
import MetaTrader5 as mt5
from sklearn.preprocessing import MinMaxScaler
from datetime import datetime, timedelta
import matplotlib.pyplot as plt
import os

# 设置中文字体
plt.rcParams['font.sans-serif'] = ['SimHei']
plt.rcParams['axes.unicode_minus'] = False

# 选择计算设备（GPU或CPU）
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
print(f"使用设备: {device}")

2. 从MT5获取EURUSD历史数据

In [ ]:

# 连接MT5
mt5.initialize()

# 设置时间范围（最近5年）
end_date = datetime.now()
start_date = end_date - timedelta(days=5*365)

# 获取EURUSD H1数据
rates = mt5.copy_rates_range("EURUSD", mt5.TIMEFRAME_H1, start_date, end_date)
mt5.shutdown()

# 转换为DataFrame
df = pd.DataFrame(rates)
print(f"获取到 {len(df):,} 条数据")
df.head()

3. 数据归一化

In [ ]:

# 创建归一化器，将数据缩放到[0,1]范围
scaler = MinMaxScaler()

# 对开、高、低、收、成交量进行归一化
df[['open', 'high', 'low', 'close', 'tick_volume']] = scaler.fit_transform(
    df[['open', 'high', 'low', 'close', 'tick_volume']]
)

print("归一化完成")
print(f"最小值: {scaler.data_min_}")
print(f"最大值: {scaler.data_max_}")

4. 构造时间序列数据集

In [ ]:

def create_dataset(data, lookback=10):
    """将时间序列转换为监督学习数据集"""
    X, y = [], []
    for i in range(len(data) - lookback - 1):
        X.append(data[i:i+lookback])  # 输入：前10根K线
        y.append(data[i+lookback, 3])  # 输出：第11根K线的收盘价
    return np.array(X, dtype=np.float32), np.array(y, dtype=np.float32).reshape(-1, 1)

# 提取特征
features = df[['open', 'high', 'low', 'close', 'tick_volume']].values
X, y = create_dataset(features)

# 划分训练集（80%）和验证集（20%）
train_size = int(len(X) * 0.8)
X_train = torch.from_numpy(X[:train_size]).to(device)
y_train = torch.from_numpy(y[:train_size]).to(device)
X_val = torch.from_numpy(X[train_size:]).to(device)
y_val = torch.from_numpy(y[train_size:]).to(device)

print(f"训练集: {X_train.shape}")
print(f"验证集: {X_val.shape}")

5. 定义LSTM模型

In [ ]:

class SimpleLSTM(nn.Module):
    """简单的LSTM神经网络"""
    def __init__(self):
        super().__init__()
        # LSTM层：输入5个特征，隐藏层20个神经元
        self.lstm = nn.LSTM(input_size=5, hidden_size=20, num_layers=1, batch_first=True)
        # 全连接层：将20维输出映射到1维（预测值）
        self.fc = nn.Linear(20, 1)

    def forward(self, x):
        lstm_out, _ = self.lstm(x)
        return self.fc(lstm_out[:, -1, :])  # 取最后一个时间步的输出

# 创建模型
model = SimpleLSTM().to(device)
optimizer = torch.optim.Adam(model.parameters(), lr=0.001)
criterion = nn.MSELoss()

print("模型创建完成")

6. 训练模型（带早停机制）

In [ ]:

# 训练参数
max_epochs = 500
patience = 20  # 验证Loss不降20轮则停止

# 记录训练过程
train_losses = []
val_losses = []
best_val_loss = float('inf')
best_epoch = 0
patience_counter = 0
best_model_state = None

print("开始训练...")
for epoch in range(max_epochs):
    # 训练阶段
    model.train()
    optimizer.zero_grad()
    train_output = model(X_train)
    train_loss = criterion(train_output, y_train)
    train_loss.backward()
    optimizer.step()
    
    # 验证阶段
    model.eval()
    with torch.no_grad():
        val_output = model(X_val)
        val_loss = criterion(val_output, y_val)
    
    # 记录Loss
    train_losses.append(train_loss.item())
    val_losses.append(val_loss.item())
    
    # 每10轮打印一次
    if (epoch + 1) % 10 == 0:
        print(f"Epoch {epoch+1:3d} | 训练Loss: {train_loss.item():.6f} | 验证Loss: {val_loss.item():.6f}")
    
    # 早停检查
    if val_loss.item() < best_val_loss:
        best_val_loss = val_loss.item()
        best_epoch = epoch + 1
        patience_counter = 0
        best_model_state = {k: v.cpu().clone() for k, v in model.state_dict().items()}
    else:
        patience_counter += 1
    
    if patience_counter >= patience:
        print(f"\n早停触发！最佳模型在Epoch {best_epoch}")
        break

# 恢复最佳模型
if best_model_state:
    model.load_state_dict(best_model_state)
    model = model.to(device)
    print(f"已恢复最佳模型 (Epoch {best_epoch}, 验证Loss: {best_val_loss:.6f})")

7. 可视化训练过程

In [ ]:

# 绘制训练和验证Loss曲线
plt.figure(figsize=(12, 5))

# 完整曲线
plt.subplot(1, 2, 1)
plt.plot(train_losses, label='训练Loss', color='blue', linewidth=2)
plt.plot(val_losses, label='验证Loss', color='orange', linewidth=2)
plt.axvline(x=best_epoch-1, color='red', linestyle='--', label=f'最佳模型 (Epoch {best_epoch})')
plt.xlabel('Epoch')
plt.ylabel('Loss')
plt.title('训练和验证损失曲线')
plt.legend()
plt.grid(True, alpha=0.3)

# 后80%放大曲线
plt.subplot(1, 2, 2)
start_idx = int(len(train_losses) * 0.2)
plt.plot(range(start_idx, len(train_losses)), train_losses[start_idx:], label='训练Loss', color='blue', linewidth=2)
plt.plot(range(start_idx, len(val_losses)), val_losses[start_idx:], label='验证Loss', color='orange', linewidth=2)
plt.axvline(x=best_epoch-1, color='red', linestyle='--', label=f'最佳模型')
plt.xlabel('Epoch')
plt.ylabel('Loss')
plt.title('后期收敛细节')
plt.legend()
plt.grid(True, alpha=0.3)

plt.tight_layout()
plt.savefig('training_loss.png', dpi=150)
plt.show()

print("训练曲线已保存为 training_loss.png")

8. 导出ONNX模型和归一化参数

In [ ]:

# 导出ONNX模型
model.eval()
model_cpu = model.cpu()
dummy_input = torch.randn(1, 10, 5)

torch.onnx.export(
    model_cpu,
    dummy_input,
    "lstm_model.onnx",
    export_params=True,
    opset_version=11,
    input_names=['input'],
    output_names=['output']
)

print("✓ ONNX模型已导出: lstm_model.onnx")

# 保存归一化参数
np.save('scaler_params.npy', {'min': scaler.data_min_, 'max': scaler.data_max_})
print("✓ 归一化参数已保存: scaler_params.npy")

# 输出MQL5格式的归一化参数
print("\n=" * 60)
print("归一化参数 (复制到LSTM_EA.mq5):")
print("=" * 60)
print(f"double data_min[5] = {{{', '.join([f'{x:.5f}' for x in scaler.data_min_])}}};")
print(f"double data_max[5] = {{{', '.join([f'{x:.5f}' for x in scaler.data_max_])}}};")
print("=" * 60)

9. 训练总结

In [ ]:

print("=" * 60)
print("训练总结")
print("=" * 60)
print(f"总训练轮数: {len(train_losses)} epochs")
print(f"最佳模型: Epoch {best_epoch}")
print(f"最佳验证Loss: {best_val_loss:.6f}")
print(f"最终训练Loss: {train_losses[-1]:.6f}")
print(f"最终验证Loss: {val_losses[-1]:.6f}")

loss_diff = abs(train_losses[-1] - val_losses[-1])
print(f"\nLoss差异: {loss_diff:.6f}")
if loss_diff < 0.0001:
    print("✓ 模型状态: 良好")
elif loss_diff < 0.001:
    print("⚠ 模型状态: 尚可，有轻微过拟合")
else:
    print("✗ 模型状态: 可能过拟合")
print("=" * 60)

12 KiB Raw Blame History Unescape Escape

LSTM外汇价格预测 - 精简版

主要步骤

1. 导入库和设置设备

2. 从MT5获取EURUSD历史数据

3. 数据归一化

4. 构造时间序列数据集

5. 定义LSTM模型

6. 训练模型（带早停机制）

7. 可视化训练过程

8. 导出ONNX模型和归一化参数

9. 训练总结

12 KiB

Raw Blame History