📘

使用 LSTM 神经网络创建时间序列预测：规范化价格和令牌化时间·进阶篇

第 2/3 篇

本章目录

把LSTM预测模型跑通并导出ONNX
用百分比变化校验预测曲线的可信度
四年老机跑完百代训练的损失曲线
训练曲线不单调时该怎么处理
用 Python 先跑一遍 LSTM 滚动预测再进 MT5
把 LSTM 模型接上 MT5 实盘数据跑一遍

「把LSTM预测模型跑通并导出ONNX」

用 PyTorch 搭一个单层 LSTM 做收盘价归一化预测，输入维度设为 5（时间令牌 + 开高低收的归一值），隐藏层 100 单元，输出 1 个值。训练前先 torch.manual_seed 固定随机种子，保证你本地复跑时权重初始化和 dropout 走向一致，方便对照 loss 曲线。训练循环跑 100 个 epoch，batch 级别用 Adam（lr=0.001）更新，每 10 轮打印一次 single_loss.item()。代码里专门埋了 NaN 检测：一旦 loss 变 nan，直接把当前 seq、y_pred、labels 全 print 出来，省得你盲调。外汇与贵金属价格序列波动剧烈，输入未做充分归一时极容易出现梯度爆炸，这块检测建议保留。训完先 torch.save 存 state_dict 为 lstm_model.pth，再切 model.eval() 用 torch.onnx.export 导出 lstm_model.onnx，dummy_input 形状为 (seq_length, 1, 5)，opset_version=11，dynamic_axes 把第 0 维标成 sequence。这样导出的模型能直接丢进 MT5 的 ONNX 推理环境跑。测试阶段用 X_test 逐条 unsqueeze(1) 推理，把预测值和 y_test 画到一起看拟合。最后算真实价格百分比变化时，分母加了 1e-10 防零除——这是处理贵金属跳空缺口时的实用写法，复制时别手滑删掉。

MQL5 / C++

torch.manual_seed(seed_value)
# Define LSTM model class
class LSTMModel(nn.Module):
    def __init__(self, input_size, hidden_layer_size, output_size):
        super(LSTMModel, self).__init__()
        self.hidden_layer_size = hidden_layer_size
        self.lstm = nn.LSTM(input_size, hidden_layer_size)
        self.linear = nn.Linear(hidden_layer_size, output_size)
    def forward(self, input_seq):
        h0 = torch.zeros(class="num">1, input_seq.size(class="num">1), self.hidden_layer_size).to(input_seq.device)
        c0 = torch.zeros(class="num">1, input_seq.size(class="num">1), self.hidden_layer_size).to(input_seq.device)
        lstm_out, _ = self.lstm(input_seq, (h0, c0))
        predictions = self.linear(lstm_out.view(input_seq.size(class="num">0), -class="num">1))
        class="kw">return predictions[-class="num">1]
print(f"Seed value used: {seed_value}")
input_size = class="num">5   # time_token, norm_open, norm_high, norm_low, norm_close
hidden_layer_size = class="num">100
output_size = class="num">1
model = LSTMModel(input_size, hidden_layer_size, output_size)
class="macro">#model = torch.compile(model)
loss_function = nn.MSELoss()
optimizer = torch.optim.Adam(model.parameters(), lr=class="num">0.001)
# Training
epochs = class="num">100
for epoch in range(epochs + class="num">1):
    for seq, labels in zip(X_train, y_train):
        optimizer.zero_grad()
        y_pred = model(seq.unsqueeze(class="num">1))
        # Ensure both are tensors of shape [class="num">1]
        y_pred = y_pred.view(-class="num">1)
        labels = labels.view(-class="num">1)
        single_loss = loss_function(y_pred, labels)
        
        # Print intermediate values to debug NaN loss
        if torch.isnan(single_loss):
            print(f&class="macro">#x27;Epoch {epoch} NaN loss detected&class="macro">#x27;)
            print(&class="macro">#x27;Sequence:&class="macro">#x27;, seq)
            print(&class="macro">#x27;Prediction:&class="macro">#x27;, y_pred)
            print(&class="macro">#x27;Label:&class="macro">#x27;, labels)
        single_loss.backward()
        optimizer.step()
    if epoch % class="num">10 == class="num">0 or epoch == epochs:  # Include the final epoch
        print(f&class="macro">#x27;Epoch {epoch} loss: {single_loss.item()}&class="macro">#x27;)
# Save the model&class="macro">#x27;s state dictionary
torch.save(model.state_dict(), &class="macro">#x27;lstm_model.pth&class="macro">#x27;)
# Convert the model to ONNX format
model.eval()
dummy_input = torch.randn(seq_length, class="num">1, input_size, dtype=torch.float32)
onnx_model_path = "lstm_model.onnx"
torch.onnx.class="kw">export(model,
                dummy_input,
                onnx_model_path,
                input_names=[&class="macro">#x27;input&class="macro">#x27;],
                output_names=[&class="macro">#x27;output&class="macro">#x27;],
                dynamic_axes={&class="macro">#x27;input&class="macro">#x27;: {class="num">0: &class="macro">#x27;sequence&class="macro">#x27;}, &class="macro">#x27;output&class="macro">#x27;: {class="num">0: &class="macro">#x27;sequence&class="macro">#x27;}},
                opset_version=class="num">11)
print(f"Model has been converted to ONNX format and saved to {onnx_model_path}")
# Predictions
model.eval()
predictions = []
for seq in X_test:
    with torch.no_grad():
        predictions.append(model(seq.unsqueeze(class="num">1)).item())
# Evaluate the model
plt.plot(y_test.numpy(), label=&class="macro">#x27;True Prices(Normalized)&class="macro">#x27;)
plt.plot(predictions, label=&class="macro">#x27;Predicted Prices(Normalized)&class="macro">#x27;)
plt.legend()
plt.show()
# Calculate percent changes with a small value added to the denominator to prevent divide by zero error
true_prices = y_test.numpy()
predicted_prices = np.array(predictions)
true_pct_change = np.diff(true_prices) / (true_prices[:-class="num">1] + class="num">1e-10)

◍ 用百分比变化校验预测曲线的可信度

直接比绝对价格意义不大，归一化后两张线贴得再近，也可能只是趋势同向而幅度失真。把预测序列做一阶差分再除以上一期价格，得到 predicted_pct_change，这一步能暴露模型在拐点处的放大或滞后偏差。上面那段 Python 把真实与预测价格画在上半子图，百分比变化画在下半子图，12x6 的画布、2行1列布局。当下半图里 Predicted Percent Change 频繁越过 True Percent Change 的波峰波谷，说明模型在波动率上做了过拟合，MT5 里接同类 LSTM 输出时就要降学习率或砍序列长度。外汇与贵金属杠杆高、跳空多，这类离线回看只能作为特征筛选参考，不能直接当进场信号；真要上实盘，先把同一段历史在 MT5 用 OnTester 跑一遍样本外误差。

MQL5 / C++

predicted_pct_change = np.diff(predicted_prices) / (predicted_prices[:-class="num">1] + class="num">1e-10)
# Plot the true and predicted prices
plt.figure(figsize=(class="num">12, class="num">6))
plt.subplot(class="num">2, class="num">1, class="num">1)
plt.plot(true_prices, label=&class="macro">#x27;True Prices(Normalized)&class="macro">#x27;)
plt.plot(predicted_prices, label=&class="macro">#x27;Predicted Prices(Normalized)&class="macro">#x27;)
plt.legend()
plt.title(&class="macro">#x27;True vs Predicted Prices(Normalized)&class="macro">#x27;)
# Plot the percent change
plt.subplot(class="num">2, class="num">1, class="num">2)
plt.plot(true_pct_change, label=&class="macro">#x27;True Percent Change&class="macro">#x27;)
plt.plot(predicted_pct_change, label=&class="macro">#x27;Predicted Percent Change&class="macro">#x27;)
plt.legend()
plt.title(&class="macro">#x27;True vs Predicted Percent Change&class="macro">#x27;)
plt.tight_layout()
plt.show()

四年老机跑完百代训练的损失曲线

这台用于训练的机器是四年前的游戏本，AMD Ryzen 5 4600H 配 Radeon Graphics 3.00 GHz、64 GB 内存，没走 GPU，纯 CPU 硬扛。整轮训练约 8 小时跑满 100 个世代，种子固定为 42，每 10 代打印一次均方差损失。从控制台记录看，损失并非单调下降：世代 0 为 0.01436，世代 50 骤降到 0.00439 的全程最低，但到世代 100 又弹到 0.03328，反而高于起点。这种中后段发散倾向，说明模型可能在小样本序列上出现过拟合征兆。收尾时框架抛出一个警告，提示应以不同方式指定模型结构。我试过排查，但因单轮重训成本就是 8 小时，而批次内序列长度一致、不会变长，便选择忽略。外汇与贵金属数据噪声大，此类损失回弹不代表策略失效，但上实盘前务必用 MT5 历史数据复算一遍。

「训练曲线不单调时该怎么处理」

种子值设为 42 的世代损失呈现非单调下降，说明模型可能还没训透。遇到这种抖动，先别急着加轮次——换一组随机种子往往更省事，比如用 Python 的 torch.seed() 把自动生成的种子打印出来再跑一遍，看损失走向是否更平稳。数据量上去性能可能改善，但 80000 根 15 分钟柱的训练会直接拉长耗时并吃满显存，硬件账单要先算清。外汇与贵金属模型训练属高风险实验，过拟合和算力浪费都可能发生。我之前用同一套图去塞 16000+ 根 15 分钟柱，结果大多数点挤成一团没法读。这种“全局”图对小数据集还有参考意义，对大样本基本失效。下一段打算切到“本地”视图：用训好的模型做连续预测，序列长度锁 60，再外推 100 根柱（合计 160 根 15 分钟量），从第 100 根倒推回第 0 根画出来，比全局拥挤图更可能看出价格行为的脉络。

◍ 用 Python 先跑一遍 LSTM 滚动预测再进 MT5

做 EURUSD 的 15 分钟 LSTM 预测，别急着往 MQL5 里搬。先在 Python 里拉最近 160 根 15 分钟 K 线（MT5 实时取数），用训好的 lstm_model.pth 跑滚动预测，前 60 根作为输入、后 100 根做对照，能快速看出模型当天靠不靠谱。脚本结构不复杂：LSTMModel 类跟训练时一致，一层 LSTM 接一层线性层；normalize_daily_rolling 按交易日做滚动归一化，每天重新算 expanding 的 high/low，避免跨日量纲污染。时间特征也做了数字标记化，和训练管道保持同构。实测 2024-06-14 约 00:45（UTC+3）跑出来的下一根归一化预测价是 0.9003，按当天已形成的日区间 1.07402–1.07336（约 8 点）折算，下一价大概率贴近日高上沿，但绝对点数空间很薄。同次输出的「归一化价格百分比变化 73.64%」是相对前一根归一化值的增幅，不是真实点数盈利，直接拿去开仓会误判。那张 matplotlib 图把最近 100 个预测和真实收盘画在一起（红点是下一预测），这次明显滞后于实际走势。外汇、贵金属属高风险品种，这种模型跑偏的交易日，倾向观望而非硬做；不代表模型全数据集失效，不必马上重训。验证建议：在 Python 里补一行，把 (当日最高-最低) 乘归一化预测收盘价，直接得出可预期的点数；这行我们后面转 MQL5 时会写进 EA，顺便给出准确价格而非分数。

MQL5 / C++

class="kw">import torch
class="kw">import torch.nn as nn
class="kw">import numpy as np
class="kw">import pandas as pd
class="kw">import MetaTrader5 as mt5
class="kw">import matplotlib.pyplot as plt
# Define LSTM model class (same as during training)
class LSTMModel(nn.Module):
    def __init__(self, input_size, hidden_layer_size, output_size):
        super(LSTMModel, self).__init__()
        self.hidden_layer_size = hidden_layer_size
        self.lstm = nn.LSTM(input_size, hidden_layer_size)
        self.linear = nn.Linear(hidden_layer_size, output_size)
        self.hidden_cell = (torch.zeros(class="num">1, class="num">1, self.hidden_layer_size),
                            torch.zeros(class="num">1, class="num">1, self.hidden_layer_size))
    def forward(self, input_seq):
        lstm_out, self.hidden_cell = self.lstm(input_seq.view(len(input_seq), class="num">1, -class="num">1), self.hidden_cell)
        predictions = self.linear(lstm_out.view(len(input_seq), -class="num">1))
        class="kw">return predictions[-class="num">1]
# Normalize prices on a rolling basis resetting at the start of each day
def normalize_daily_rolling(data):
    data[&class="macro">#x27;date&class="macro">#x27;] = data.index.date
    data[&class="macro">#x27;rolling_high&class="macro">#x27;] = data.groupby(&class="macro">#x27;date&class="macro">#x27;)[&class="macro">#x27;high&class="macro">#x27;].transform(lambda x: x.expanding(min_periods=class="num">1).max())
    data[&class="macro">#x27;rolling_low&class="macro">#x27;] = data.groupby(&class="macro">#x27;date&class="macro">#x27;)[&class="macro">#x27;low&class="macro">#x27;].transform(lambda x: x.expanding(min_periods=class="num">1).min())

把 LSTM 模型接上 MT5 实盘数据跑一遍

模型训完不等于能用，得拉 MT5 的真实行情喂进去验证。这段脚本把 EURUSD 的 M15 周期最近 160 根 K 线取回来，前 60 根做序列长度，后 100 根做滚动评估步数，逻辑很直接。归一化用滚动高低点做分母，把 open/high/low/close 压到 0~1 区间；时间也单独编码成 time_token（当日秒数除 86400）。这样输入维度是 5：time_token 加四个归一化价格。隐藏层设 100、输出层 1，和训练时 LSTMModel(input_size=5, hidden_layer_size=100, output_size=1) 必须对齐，否则 load_state_dict 直接报错。评估循环从 step=100 倒退到 1，每次截一段 60 长序列送进模型，预测值存 all_predicted_prices，真实值存 all_true_prices。最后算两者逐阶百分比变化做对比，再对最新 60 根做 next_prediction，打印出「下一根归一化收盘价」和基于预测值的涨跌幅百分比。外汇和贵金属属高风险品种，LSTM 在 EURUSD M15 上的预测仅反映历史统计关系，下一步价格偏向哪边不代表实际成交结果，开 MT5 跑这段代码时请先用策略测试器或旁路上线观察，别直接挂真实订单。

MQL5 / C++

data[&class="macro">#x27;norm_open&class="macro">#x27;] = (data[&class="macro">#x27;open&class="macro">#x27;] - data[&class="macro">#x27;rolling_low&class="macro">#x27;]) / (data[&class="macro">#x27;rolling_high&class="macro">#x27;] - data[&class="macro">#x27;rolling_low&class="macro">#x27;])
data[&class="macro">#x27;norm_high&class="macro">#x27;] = (data[&class="macro">#x27;high&class="macro">#x27;] - data[&class="macro">#x27;rolling_low&class="macro">#x27;]) / (data[&class="macro">#x27;rolling_high&class="macro">#x27;] - data[&class="macro">#x27;rolling_low&class="macro">#x27;])
data[&class="macro">#x27;norm_low&class="macro">#x27;] = (data[&class="macro">#x27;low&class="macro">#x27;] - data[&class="macro">#x27;rolling_low&class="macro">#x27;]) / (data[&class="macro">#x27;rolling_high&class="macro">#x27;] - data[&class="macro">#x27;rolling_low&class="macro">#x27;])
data[&class="macro">#x27;norm_close&class="macro">#x27;] = (data[&class="macro">#x27;close&class="macro">#x27;] - data[&class="macro">#x27;rolling_low&class="macro">#x27;]) / (data[&class="macro">#x27;rolling_high&class="macro">#x27;] - data[&class="macro">#x27;rolling_low&class="macro">#x27;])
# Replace NaNs with zeros
data.fillna(class="num">0, inplace=True)
class="kw">return data[[&class="macro">#x27;norm_open&class="macro">#x27;, &class="macro">#x27;norm_high&class="macro">#x27;, &class="macro">#x27;norm_low&class="macro">#x27;, &class="macro">#x27;norm_close&class="macro">#x27;]]
# Load the saved model
input_size = class="num">5  # time_token, norm_open, norm_high, norm_low, norm_close
hidden_layer_size = class="num">100
output_size = class="num">1
model = LSTMModel(input_size, hidden_layer_size, output_size)
model.load_state_dict(torch.load(&class="macro">#x27;lstm_model.pth&class="macro">#x27;))
model.eval()
# Connect to MetaTrader class="num">5
if not mt5.initialize():
    print("Initialize failed")
    mt5.shutdown()
# Load the latest class="num">160 bars of market data
symbol = "EURUSD"
timeframe = mt5.TIMEFRAME_M15
bars = class="num">160  # class="num">60 for sequence length + class="num">100 for evaluation steps
rates = mt5.copy_rates_from_pos(symbol, timeframe, class="num">0, bars)
mt5.shutdown()
# Convert to DataFrame
data = pd.DataFrame(rates)
data[&class="macro">#x27;time&class="macro">#x27;] = pd.to_datetime(data[&class="macro">#x27;time&class="macro">#x27;], unit=&class="macro">#x27;s&class="macro">#x27;)
data.set_index(&class="macro">#x27;time&class="macro">#x27;, inplace=True)
# Normalize the new data
data[[&class="macro">#x27;norm_open&class="macro">#x27;, &class="macro">#x27;norm_high&class="macro">#x27;, &class="macro">#x27;norm_low&class="macro">#x27;, &class="macro">#x27;norm_close&class="macro">#x27;]] = normalize_daily_rolling(data)
# Tokenize time
data[&class="macro">#x27;time_token&class="macro">#x27;] = (data.index.hour * class="num">3600 + data.index.minute * class="num">60 + data.index.second) / class="num">86400
# Drop unnecessary columns
data = data[[&class="macro">#x27;time_token&class="macro">#x27;, &class="macro">#x27;norm_open&class="macro">#x27;, &class="macro">#x27;norm_high&class="macro">#x27;, &class="macro">#x27;norm_low&class="macro">#x27;, &class="macro">#x27;norm_close&class="macro">#x27;]]
# Fetch the last class="num">100 sequences for evaluation
seq_length = class="num">60
evaluation_steps = class="num">100
# Initialize lists for storing evaluation results
all_true_prices = []
all_predicted_prices = []
model.eval()
for step in range(evaluation_steps, class="num">0, -class="num">1):
    # Get the sequence ending at &class="macro">#x27;step&class="macro">#x27;
    seq = data.values[-step-seq_length:-step]
    seq = torch.tensor(seq, dtype=torch.float32)
    # Make prediction
    with torch.no_grad():
        model.hidden_cell = (torch.zeros(class="num">1, class="num">1, model.hidden_layer_size),
                             torch.zeros(class="num">1, class="num">1, model.hidden_layer_size))
        prediction = model(seq).item()
    
    all_true_prices.append(data[&class="macro">#x27;norm_close&class="macro">#x27;].values[-step])
    all_predicted_prices.append(prediction)
# Calculate percent changes and convert to percentages
true_pct_change = (np.diff(all_true_prices) / np.array(all_true_prices[:-class="num">1])) * class="num">100
predicted_pct_change = (np.diff(all_predicted_prices) / np.array(all_predicted_prices[:-class="num">1])) * class="num">100
# Make next prediction
next_seq = data.values[-seq_length:]
next_seq = torch.tensor(next_seq, dtype=torch.float32)
with torch.no_grad():
    model.hidden_cell = (torch.zeros(class="num">1, class="num">1, model.hidden_layer_size),
                        torch.zeros(class="num">1, class="num">1, model.hidden_layer_size))
    next_prediction = model(next_seq).item()
# Calculate percent change for the next prediction
next_true_price = data[&class="macro">#x27;norm_close&class="macro">#x27;].values[-class="num">1]
next_price_pct_change = ((next_prediction - all_predicted_prices[-class="num">1]) / all_predicted_prices[-class="num">1]) * class="num">100
print(f"Next predicted close price(normalized): {next_prediction}")
print(f"Percent change for the next prediction based on normalized price: {next_price_pct_change:.5f}%")
print("All Predicted Prices: ", all_predicted_prices)
# Plot the evaluation results with capped y-axis
plt.figure(figsize=(class="num">12, class="num">8))
plt.subplot(class="num">2, class="num">1, class="num">1)
plt.plot(all_true_prices, label=&class="macro">#x27;True Prices(Normalized)&class="macro">#x27;)
plt.plot(all_predicted_prices, label=&class="macro">#x27;Predicted Prices(Normalized)&class="macro">#x27;)
plt.scatter(len(all_true_prices), next_prediction, class="type">color=&class="macro">#x27;red&class="macro">#x27;, label=&class="macro">#x27;Next Prediction&class="macro">#x27;)
plt.legend()
plt.title(&class="macro">#x27;True vs Predicted Prices(Normalized, Last class="num">100 Steps)&class="macro">#x27;)