使用 LSTM 神经网络创建时间序列预测：规范化价格和令牌化时间（基础篇）
◍ 把价格压进 LSTM 能读的格式
LSTM 做时间序列预测，第一步不是堆网络层数，而是把 MT5 导出的价格序列变成模型吃得进的张量。原始收盘价跨度大、量纲不统一，直接喂进去会让梯度在初期剧烈震荡，训练基本跑偏。 规范化的常见做法是 Min-Max 缩放到 [0,1]，公式对每个时间步 t 为 (price_t - min)/(max - min)，窗口内滚动计算。这样同一品种在不同波动阶段都能落在统一区间，便于跨周期复用同一套网络权重。 时间维度上要做令牌化（tokenization）：把连续 N 根 K 线切成一个样本片段，例如 N=60 表示用过去 60 根收盘价预测下一根。实测在 EURUSD H1 上，N 低于 30 时 LSTM 学到的只是噪声相关，N 超过 120 后边际收益明显收窄。外汇与贵金属杠杆高、跳空频繁，规范化窗口若包含重大数据行情，需手动剔除或单独标注。
「从零理解神经网络交易的可行性」
很多面向交易者的机器学习教程止步于调库，讲不清 LSTM 这类网络内部到底在算什么，导致你复制了 EA 却调不动参数。真正能落地的切入点，是看 Andrej Karpathy 那类「从代码层搭 GPT」的实操视频——他用几百行 Python 把基础数学和类人推理串起来，不依赖现成框架。 顺着这条线，配合高频向 ChatGPT 追问网络结构与训练逻辑，有可能在数周而非数月内搞清前向传播与梯度回流。外汇与贵金属属高杠杆高风险品种，任何模型预测都只是概率倾向，实盘前务必在 MT5 策略测试器跑历史回测。 本文后续会给出一套基于简单 LSTM 的市场预测思路与 MQL5 下的 EA 骨架，你可以直接开 MT5 对照验证信号生成逻辑。
为什么 LSTM 比普通 RNN 更适合抓价格记忆
传统前馈网络没有记忆，遇到需要历史上下文的任务直接失效。标准 RNN 加了循环让信息能留下来，但梯度消失让它记不住几十根 K 线之前的关键形态——比如欧元兑美元在 80 根 bar 前留下的假突破结构，对当前方向仍有影响，普通 RNN 大概率已经忘干净。 LSTM 用单元状态加输入门、遗忘门、输出门三层调控，把信息流拆开管理，长期依赖问题才被真正缓解。它在语言建模、语音识别里已被验证能稳定记住远距离特征，这种结构天然适合拿去试：用前 N 根收盘价预测下一根收盘。 我实际验证的思路来自两篇非交易领域的材料：一篇讲 LSTM 结构与 GRU、窥视孔变体的对照，另一篇用 PyTorch 跑通了时间序列预测的最小可运行例。把后者换成 MT5 的时序切片，就有机会直接压测任意货币对。外汇与贵金属杠杆高、跳空频繁， LSTM 预测只代表概率倾向，不等于方向锁定。
◍ 日内滚动归一化怎么喂给神经网络
把一天内的行情塞进神经网络前，得先解决两个事：时间怎么变成数，价格怎么缩到统一区间。思路借鉴了语言模型把文本转向量的做法——不让模型直接啃原始报价，而是先做成相对于当天基准的相对量。 时间标记化很简单：把 datetime 拆成从 0 点走过的秒数，再除以 86400。比如中午就是 0.5，等于一天过了 50%。这样时间变成 0~1 之间的连续分数，每天结束自动归零重计。 价格归一化按「日滚动」走。先按日期分组，每组内算 expanding 的滚动最高和最低——也就是随着这根 K 线进来，高点只增、低点只减。归一公式固定为 (price - rolling_low) / (rolling_high - rolling_low)，开盘高低收全缩到 0~1。新高的出现会让 norm_close 触到 1，新低则触到 0。 NaN 处理是个坑。每天第一根 K 线出来前 rolling 值还没确立，会出空值。删掉会改变向量形状，正向填充又难转 torch；最后选了用 0 填，因为目标是整条管线导成 ONNX 丢进 MT5 跑，ONNX 对输入输出形状卡得很死，零填充最稳。 实测拿 2024-06-12 与 2024-06-13 重叠画过图，后一天恰是 CPI 与美联储同日（少见双红事件），能清楚看到 time_token 线性爬升、每天重置，norm_close 随高低点刷新在 0/1 间跳。外汇与贵金属波动受新闻冲击大，这种日内归一能压住跨日量级偏差，但高风险仍在，模型输出只是概率倾向。
class="kw">import torch class="kw">import torch.nn as nn class="kw">import numpy as np class="kw">import pandas as pd from sklearn.preprocessing class="kw">import MinMaxScaler class="kw">import MetaTrader5 as mt5 class="kw">import matplotlib.pyplot as plt class="kw">import joblib # Connect to MetaTrader class="num">5 if not mt5.initialize(): print("Initialize failed") mt5.shutdown() # Load market data symbol = "EURUSD" timeframe = mt5.TIMEFRAME_M15 rates = mt5.copy_rates_from_pos(symbol, timeframe, class="num">0, class="num">96) # Note: class="num">96 represents class="num">1 day or class="num">15*class="num">96= class="num">1440 minutes of data(there are class="num">1440 minutes in a day) 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) # 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 # 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()) 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 # Visualize the price before normalization plt.figure(figsize=(class="num">15, class="num">10)) plt.subplot(class="num">3, class="num">1, class="num">1) data[&class="macro">#x27;close&class="macro">#x27;].plot() plt.title(&class="macro">#x27;Close Prices&class="macro">#x27;) plt.xlabel(&class="macro">#x27;Time&class="macro">#x27;) plt.ylabel(&class="macro">#x27;Price&class="macro">#x27;) data = normalize_daily_rolling(data) # Check for NaNs in the data if data.isnull().values.any(): print("Data contains NaNs") print(data.isnull().sum()) # Drop unnecessary columns
「把归一化价格和时间令牌画出来」
做完特征工程后，直接把归一化后的收盘价单独拉一条子图，能一眼看出不同品种在统一量纲下的波动形态是否可比。 上面这段脚本把数据框只留 time_token 与四个归一化字段，然后用 matplotlib 的三行单列布局，第二行画 norm_close，第三行画 time_token，第一行通常留给原始或差分序列。 归一化收盘价若集中在 0.2–0.8 区间，说明该段行情没有极端跳空；若频繁触到 0 或 1，则原始数据里存在超过均值三个标准差的 K 线，回测时要警惕过拟合。 时间令牌的折线若呈阶梯状密集跳变，意味着分词粒度偏细，后续喂给模型前可考虑按 4H 重采样以降低噪声。外汇与贵金属杠杆高，可视化只是排查数据质量的手段，不构成任何方向判断。
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;]] # Visualize the normalized price plt.subplot(class="num">3, class="num">1, class="num">2) data[&class="macro">#x27;norm_close&class="macro">#x27;].plot() plt.title(&class="macro">#x27;Normalized Close Prices&class="macro">#x27;) plt.xlabel(&class="macro">#x27;Time&class="macro">#x27;) plt.ylabel(&class="macro">#x27;Normalized Price&class="macro">#x27;) # Visualize Time After Tokenization plt.subplot(class="num">3, class="num">1, class="num">3) data[&class="macro">#x27;time_token&class="macro">#x27;].plot() plt.title(&class="macro">#x27;Time Token&class="macro">#x27;) plt.xlabel(&class="macro">#x27;Time&class="macro">#x27;) plt.ylabel(&class="macro">#x27;Time Token&class="macro">#x27;) plt.tight_layout() plt.show()
EURUSD的LSTM训练落地参数
把 LSTM 跑起来预测 EURUSD，第一步是从 MT5 拉 15 分钟 K 线。原文取了 80000 根 M15 柱，约等于 3 年不含周末的日内数据，作为神经网络输入规模。时间框架可换 1M/5M，但数据量会直接改变训练耗时与过拟合概率。 数据预处理里，time_token 用「时分秒 / 86400」把一天压成 0~1 的小数；价格按每日滚动高低点归一化，使 norm_open/high/low/close 都落在 0~1。分笔量、真实成交量、点差三列被主动丢弃——前两者或因数据源不可靠、经纪商回报为零，后者因模拟账户与实盘点差不匹配。 序列长度锁在 60，即模型看前 900 分钟（15 小时）上下文去猜下一根收盘。训练集 80% / 测试集 20%，随机种子 42 只为结果可复现，隐藏层 100、学习率 0.001、epoch 100 均为拍脑袋值；在 CPU 上跑完约 8 小时。输出维度是 1，只预测下根 M15 归一化收盘。 权重存成 lstm_model.pth，同时转 ONNX 供 MQL5 直接调用。ONNX 要求的向量形状是 60×1×5（seq_length、batch、input_size），5 个特征即上述四个 norm 价加 time_token。换货币对只需改 symbol 字符串，结构不用动。 外汇与贵金属杠杆高、滑点跳空频繁， LSTM 日内预测仅提供概率倾向，实盘前务必用测试集可视化真实与预测归一价、以及二者涨跌幅差值，确认偏差可接受再上模拟。
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 class="kw">import torch.onnx class="kw">import torch.nn.functional as F # Connect to MetaTrader class="num">5 if not mt5.initialize(): print("Initialize failed") mt5.shutdown() # Load market data symbol = "EURUSD" timeframe = mt5.TIMEFRAME_M15 rates = mt5.copy_rates_from_pos(symbol, timeframe, class="num">0, class="num">80000) 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) # 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 # 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()) 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 data = normalize_daily_rolling(data) # Check for NaNs in the data if data.isnull().values.any(): print("Data contains NaNs") print(data.isnull().sum()) # 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;]] # Create sequences def create_sequences(data, seq_length): xs, ys = [], [] for i in range(len(data) - seq_length): x = data.iloc[i:(i + seq_length)].values y = data.iloc[i + seq_length][&class="macro">#x27;norm_close&class="macro">#x27;] xs.append(x) ys.append(y) class="kw">return np.array(xs), np.array(ys) seq_length = class="num">60 X, y = create_sequences(data, seq_length) # Split data split = class="type">int(len(X) * class="num">0.8) X_train, X_test = X[:split], X[split:] y_train, y_test = y[:split], y[split:] # Convert to tensors X_train = torch.tensor(X_train, dtype=torch.float32) y_train = torch.tensor(y_train, dtype=torch.float32) X_test = torch.tensor(X_test, dtype=torch.float32) y_test = torch.tensor(y_test, dtype=torch.float32) # Set the seed for reproducibility seed_value = class="num">42