📘
使用PSAR、Heiken Ashi和深度学习进行交易（基础篇）
本章目录
◍ 把 PSAR 与 Heiken Ashi 塞进深度模型
Javier Santiago Gaston De Iriarte Cabrera 在 2025 年 5 月 12 日于 MetaTrader 5 社区发布了一个思路：用 PSAR（抛物线转向）与 Heiken Ashi 蜡烛作为特征输入，接一层轻量深度学习网络做方向判别。原帖基础互动数为 1061 次查看、0 条评论，说明这类「指标+神经网络」组合在零售圈仍属低热度试探。 PSAR 提供趋势反转的止损位移轨迹，Heiken Ashi 平滑了实体与影线噪声，两者叠加后喂给网络，理论上比裸价序列更容易让模型抓住「趋势延续 vs 反转」的概率边界。外汇与贵金属杠杆高、滑点随机，这类信号只倾向提高过滤效率，不保证胜率。 想验证的人可以直接开 MT5：先加载 iSAR 与 iCustom 的 Heiken Ashi，把每根 K 的 PSAR 值、HA 开盘收口差导出成 CSV，再用 Python 跑个两层 MLP 看混淆矩阵。这一步比空谈「AI 交易」实在。
把深度学习塞进MT5的实战闭环
在外汇与贵金属这类高杠杆市场里，海量tick数据靠人眼筛优势基本失效，失败往往就差那一点边际概率。把传统指标和深度学习揉在一起，本质是用模型替你跑遍历史样本，快速判断某套组合是否值得上实盘。 我们先用Python做离线实验，只验证策略有没有正期望，不碰下单逻辑；同一套核心判断再写成MQL5脚本，直接挂MT5实时跑。两者共用特征工程，意味着回测通过的信号结构，在实盘可能以极低延迟复现。 这种从离线脚本到MQL5实时嵌入的通路，降低了手动搬运逻辑的误差，但外汇/贵金属的高风险仍在，任何历史优势都只是概率倾向，不是必然。
「把EURUSD历史价压成ONNX模型」
做外汇深度学习预测，第一步是拿到干净且够长的历史序列。这套 Python 脚本直接调 MT5 终端拉 EURUSD 的 D1 数据，回看窗口设成 120 天——作者选这个数是为了兼顾模式识别量与训练负担，太长容易让模型钝化，太短又学不到日线级别的趋势结构。 原始价杂乱，脚本只抽每日收盘价，用 MinMaxScaler 压到 0~1 区间。这一步不是美化，是让梯度更新在同一量纲里跑，否则 LSTM 对量纲敏感，误差会漂。 数据切 80% 训练、20% 测试，避免模型死记。序列用 split_sequence 按 time_step=120 滑窗拆成监督样本：前 120 天收盘价进 X，第 121 天收盘价当 y。CNN 抓局部形态、LSTM 抓跨日依赖、Dense 出预测，再加 dropout 防过拟合；训练最多 300 epoch 配 early stopping，跑完用 RMSE 量预测偏离，最后 tf2onnx 导出成 .onnx 丢进 MQL5\Files 供 EA 调用。 外汇和贵金属都是高杠杆高风险品种，ONNX 模型只是工具箱里的其中一种信号源，实盘前务必在 MT5 用 2024 年以后的样本外数据重测 RMSE，别把回测吻合当概率优势。
MQL5 / C++
class="kw">import MetaTrader5 <span class="keyword">as</span> mt5 class="kw">import tensorflow <span class="keyword">as</span> tf class="kw">import numpy <span class="keyword">as</span> np class="kw">import pandas <span class="keyword">as</span> pd class="kw">import tf2onnx class="kw">import keras inp_history_size = <span class="number">class="num">120</span> sample_size = inp_history_size*<span class="number">class="num">3</span>*<span class="number">class="num">20</span> symbol = <span class="class="type">class="kw">string">"EURUSD"</span> optional = <span class="class="type">class="kw">string">"D1_2024"</span> inp_model_name = str(symbol)+<span class="class="type">class="kw">string">"_"</span>+str(optional)+<span class="class="type">class="kw">string">".onnx"</span> <span class="keyword">if</span> not mt5.initialize(): print(<span class="class="type">class="kw">string">"initialize() failed, error code ="</span>,mt5.last_error()) quit() <span class="preprocessor"># </span>we will save generated onnx-file near the our script to use <span class="keyword">as</span> resource from sys class="kw">import argv data_path=argv[<span class="number">class="num">0</span>] last_index=data_path.rfind(<span class="class="type">class="kw">string">"\\"</span>)+<span class="number">class="num">1</span> data_path=data_path[<span class="number">class="num">0</span>:last_index] print(<span class="class="type">class="kw">string">"data path to save onnx model"</span>,data_path) <span class="preprocessor"># </span>and save to MQL5\Files folder to use <span class="keyword">as</span> file terminal_info=mt5.terminal_info() file_path=terminal_info.data_path+<span class="class="type">class="kw">string">"\\MQL5\\Files\\"</span> print(<span class="class="type">class="kw">string">"file path to save onnx model"</span>,file_path) <span class="preprocessor"># </span>set start and end dates <span class="keyword">for</span> history data from <span class="keyword">class="type">class="kw">datetime</span> class="kw">import timedelta, <span class="keyword">class="type">class="kw">datetime</span> <span class="preprocessor">class="macro">#end_date </span>= <span class="keyword">class="type">class="kw">datetime</span>.now() end_date = <span class="keyword">class="type">class="kw">datetime</span>(<span class="number">class="num">2024</span>, <span class="number">class="num">1</span>, <span class="number">class="num">1</span>, <span class="number">class="num">0</span>) start_date = end_date - timedelta(days=inp_history_size*<span class="number">class="num">20</span>*<span class="number">class="num">3</span>) <span class="preprocessor"># </span>print start and end dates print(<span class="class="type">class="kw">string">"data start date ="</span>,start_date) print(<span class="class="type">class="kw">string">"data end date ="</span>,end_date) <span class="preprocessor"># </span>get rates eurusd_rates = mt5.copy_rates_from(symbol, mt5.TIMEFRAME_D1, end_date, sample_size) from sklearn.preprocessing class="kw">import MinMaxScaler scaler=MinMaxScaler(feature_range=(<span class="number">class="num">0</span>,<span class="number">class="num">1</span>)) scaled_data = scaler.fit_transform(data) <span class="preprocessor"># </span>scale data from sklearn.preprocessing class="kw">import MinMaxScaler scaler=MinMaxScaler(feature_range=(<span class="number">class="num">0</span>,<span class="number">class="num">1</span>)) scaled_data = scaler.fit_transform(data) <span class="preprocessor"># </span>training size is <span class="number">class="num">80</span>% of the data training_size = <span class="keyword">class="type">int</span>(len(scaled_data)*<span class="number">class="num">0.80</span>) print(<span class="class="type">class="kw">string">"Training_size:"</span>,training_size) train_data_initial = scaled_data[<span class="number">class="num">0</span>:training_size,:] test_data_initial = scaled_data[training_size:,:<span class="number">class="num">1</span>] <span class="preprocessor"># </span>split a univariate sequence into samples def split_sequence(sequence, n_steps): X, y = list(), list() <span class="keyword">for</span> i in range(len(sequence)): <span class="preprocessor"># </span>find the end of <span class="keyword">this</span> pattern end_ix = i + n_steps <span class="preprocessor"># </span>check <span class="keyword">if</span> we are beyond the sequence <span class="keyword">if</span> end_ix > len(sequence)-<span class="number">class="num">1</span>: <span class="keyword">break</span> <span class="preprocessor"># </span>gather <span class="keyword">input</span> and output parts of the pattern seq_x, seq_y = sequence[i:end_ix], sequence[end_ix] X.append(seq_x) y.append(seq_y) <span class="keyword">class="kw">return</span> np.array(X), np.array(y) <span class="preprocessor"># </span>split into samples time_step = inp_history_size x_train, y_train = split_sequence(train_data_initial, time_step) x_test, y_test = split_sequence(test_data_initial, time_step)
◍ 把LSTM预测模型压成ONNX给MT5喂数据
上面这段Python脚本干了两件事：先用Conv1D接双层LSTM搭一个时间序列模型，再把训好的权重转成ONNX文件。外汇和贵金属行情的高波动特性意味着任何模型预测都只是概率倾向，实盘前务必在小周期上验证。 输入维度必须压成[samples, time_steps, features]三维张量，否则Keras的LSTM层直接报错；代码里features恒为1，也就是单变量序列。卷积层用了256个滤波器、kernel_size=2，后面跟pool_size=2的下采样，再叠两个100单元的LSTM，Dropout都设0.3防过拟合。 训练时跑了最多300个epoch，但EarlyStopping的patience=20会在验证损失连续20轮不降时停手并回滚到最优权重。batch_size固定32，优化器adam，损失函数mse，顺带监控RMSE。 转ONNX用了tf2onnx，opset=13，input_signature写死[None, inp_history_size, 1]。MT5的Python环境若装了onnxruntime，就能直接load这个文件做推理，不必重跑Keras。 别把正态当圣经：脚本里最后三处save路径有重复调用，实际部署只留一个output_path即可，多余转换纯属浪费IO。
MQL5 / C++
# reshape input to be [samples, time steps, features] which is required for LSTM x_train =x_train.reshape(x_train.shape[class="num">0],x_train.shape[class="num">1],class="num">1) x_test = x_test.reshape(x_test.shape[class="num">0],x_test.shape[class="num">1],class="num">1) # define model from keras.models class="kw">import Sequential from keras.layers class="kw">import Dense, Activation, Conv1D, MaxPooling1D, Dropout, Flatten, LSTM from keras.metrics class="kw">import RootMeanSquaredError as rmse from tensorflow.keras class="kw">import callbacks model = Sequential() model.add(Conv1D(filters=class="num">256, kernel_size=class="num">2, strides=class="num">1, padding=&class="macro">#x27;same&class="macro">#x27;, activation=&class="macro">#x27;relu&class="macro">#x27;, input_shape=(inp_history_size,class="num">1))) model.add(MaxPooling1D(pool_size=class="num">2)) model.add(LSTM(class="num">100, return_sequences = True)) model.add(Dropout(class="num">0.3)) model.add(LSTM(class="num">100, return_sequences = False)) model.add(Dropout(class="num">0.3)) model.add(Dense(units=class="num">1, activation = &class="macro">#x27;sigmoid&class="macro">#x27;)) model.compile(optimizer=&class="macro">#x27;adam&class="macro">#x27;, loss= &class="macro">#x27;mse&class="macro">#x27; , metrics = [rmse()]) # Set up early stopping early_stopping = callbacks.EarlyStopping( monitor=&class="macro">#x27;val_loss&class="macro">#x27;, patience=class="num">20, restore_best_weights=True, ) # model training for class="num">300 epochs history = model.fit(x_train, y_train, epochs = class="num">300 , validation_data = (x_test,y_test), batch_size=class="num">32, callbacks=[early_stopping], verbose=class="num">2) # evaluate training data train_loss, train_rmse = model.evaluate(x_train,y_train, batch_size = class="num">32) print(f"train_loss={train_loss:.3f}") print(f"train_rmse={train_rmse:.3f}") # evaluate testing data test_loss, test_rmse = model.evaluate(x_test,y_test, batch_size = class="num">32) print(f"test_loss={test_loss:.3f}") print(f"test_rmse={test_rmse:.3f}") # Define a function that represents your model @tf.function(input_signature=[tf.TensorSpec([None, inp_history_size, class="num">1], tf.float32)]) def model_function(x): class="kw">return model(x) output_path = data_path+inp_model_name # Convert the model to ONNX onnx_model, _ = tf2onnx.convert.from_function( model_function, input_signature=[tf.TensorSpec([None, inp_history_size, class="num">1], tf.float32)], opset=class="num">13, output_path=output_path ) print(f"Saved ONNX model to {output_path}") # save model to ONNX output_path = data_path+inp_model_name onnx_model = tf2onnx.convert.from_keras(model, output_path=output_path) print(f"saved model to {output_path}") output_path = file_path+inp_model_name onnx_model = tf2onnx.convert.from_keras(model, output_path=output_path) print(f"saved model to {output_path}")
用Python脚本给策略做快速压力测试
这套 Python 脚本本质不是完整交易系统，而是给 EURUSD 小时级别策略做快速盈利潜力验证的探针。它把 MT5 历史行情、Heikin Ashi 平滑、PSAR/SMA/RSI/ATR 指标和预训练 ONNX 深度学习模型拼在一起，目的是在陷入复杂优化前先看清市场里有没有可捕捉的优势。外汇与贵金属属高风险品种，这类快速测试结果仅作进一步开发的参考，不代表实盘可行。 脚本先连 MT5 拉取欧元兑美元小时数据，用 Heikin Ashi 把噪音压下去，再叠知名指标做市场状态快检。核心是把缩放后的数据喂进 ONNX 模型出下一步预测，和「价在 SMA 上且 RSI 未超买」之类规则共振后给多空信号，并挂了简易自适应止损止盈。 回测概览里有个可盯的数字：测试期总回报 1.35%，1 万美元本金变成 10135.02 美元；夏普比率 0.39，说明那点收益伴着不小波动。对一个快速实验来说，这比率刚好用来判断「值不值得往下深挖」，而不是证明策略已成型。 可视化三张图把逻辑闭环画出来了：价格+模型预测+买卖点叠加图看信号落点，Heikin Ashi 图看趋势是否被策略咬住，权益曲线暴露增长段与回撤段。打开 MT5 终端跑通下面这段代码，你就能复现这套快速实验框架，再换品种或窗口长度试探自己的点子。
MQL5 / C++
class="kw">import MetaTrader5 as mt5 class="kw">import pandas as pd class="kw">import numpy as np class="kw">import onnxruntime as ort from sklearn.preprocessing class="kw">import MinMaxScaler from ta.trend class="kw">import PSARIndicator, SMAIndicator from ta.momentum class="kw">import RSIIndicator from ta.volatility class="kw">import AverageTrueRange class="kw">import matplotlib.pyplot as plt # Inicializar conexión con MetaTrader5 if not mt5.initialize(): print("Inicialización fallida") mt5.shutdown() def get_historical_data(symbol, timeframe, start_date, end_date): rates = mt5.copy_rates_range(symbol, timeframe, start_date, end_date) df = pd.DataFrame(rates) df[&class="macro">#x27;time&class="macro">#x27;] = pd.to_datetime(df[&class="macro">#x27;time&class="macro">#x27;], unit=&class="macro">#x27;s&class="macro">#x27;) df.set_index(&class="macro">#x27;time&class="macro">#x27;, inplace=True) class="kw">return df def calculate_heikin_ashi(df): ha_close = (df[&class="macro">#x27;open&class="macro">#x27;] + df[&class="macro">#x27;high&class="macro">#x27;] + df[&class="macro">#x27;low&class="macro">#x27;] + df[&class="macro">#x27;close&class="macro">#x27;]) / class="num">4 ha_open = (df[&class="macro">#x27;open&class="macro">#x27;].shift(class="num">1) + df[&class="macro">#x27;close&class="macro">#x27;].shift(class="num">1)) / class="num">2 ha_high = df[[&class="macro">#x27;high&class="macro">#x27;, &class="macro">#x27;open&class="macro">#x27;, &class="macro">#x27;close&class="macro">#x27;]].max(axis=class="num">1) ha_low = df[[&class="macro">#x27;low&class="macro">#x27;, &class="macro">#x27;open&class="macro">#x27;, &class="macro">#x27;close&class="macro">#x27;]].min(axis=class="num">1) df[&class="macro">#x27;ha_close&class="macro">#x27;] = ha_close df[&class="macro">#x27;ha_open&class="macro">#x27;] = ha_open df[&class="macro">#x27;ha_high&class="macro">#x27;] = ha_high df[&class="macro">#x27;ha_low&class="macro">#x27;] = ha_low class="kw">return df def add_indicators(df): # Calcular Heikin Ashi df = calculate_heikin_ashi(df) # PSAR con parámetros ajustados psar = PSARIndicator(df[&class="macro">#x27;high&class="macro">#x27;], df[&class="macro">#x27;low&class="macro">#x27;], df[&class="macro">#x27;close&class="macro">#x27;], step=class="num">0.02, max_step=class="num">0.2) df[&class="macro">#x27;psar&class="macro">#x27;] = psar.psar() # Añadir SMA sma = SMAIndicator(df[&class="macro">#x27;close&class="macro">#x27;], window=class="num">50) df[&class="macro">#x27;sma&class="macro">#x27;] = sma.sma_indicator() # Añadir RSI rsi = RSIIndicator(df[&class="macro">#x27;close&class="macro">#x27;], window=class="num">14) df[&class="macro">#x27;rsi&class="macro">#x27;] = rsi.rsi() # Añadir ATR para medir volatilidad atr = AverageTrueRange(df[&class="macro">#x27;high&class="macro">#x27;], df[&class="macro">#x27;low&class="macro">#x27;], df[&class="macro">#x27;close&class="macro">#x27;], window=class="num">14) df[&class="macro">#x27;atr&class="macro">#x27;] = atr.average_true_range() # Añadir filtro de tendencia simple df[&class="macro">#x27;trend&class="macro">#x27;] = np.where(df[&class="macro">#x27;close&class="macro">#x27;] > df[&class="macro">#x27;sma&class="macro">#x27;], class="num">1, -class="num">1) class="kw">return df def prepare_data(df, window_size=class="num">120): scaler = MinMaxScaler(feature_range=(class="num">0, class="num">1)) scaled_data = scaler.fit_transform(df[[&class="macro">#x27;close&class="macro">#x27;]]) X = [] for i in range(window_size, len(scaled_data)): X.append(scaled_data[i-window_size:i])
常见问题
先把 PSAR 的翻转点和 Heiken Ashi 的阴阳实体分别归一化，再作为双通道特征输入，避免量纲冲突导致模型学偏。
用 Python 把收盘价和 PSAR/HA 特征按时间窗切片，训练后通过 onnxruntime 导出，注意测试集要留出最近 3 个月防过拟合。
小布可对接你的 ONNX 预测结果，在对应品种页直接标注 LSTM 倾向与 HA 趋势共振区，省去手动切软件核对。
不靠谱，概率仅作趋势倾向参考；外汇贵金属高风险，必须叠加 PSAR 止损与资金风控再决策。
至少测跳空缺口、单日反转、以及连续窄幅震荡三种，看最大回撤是否超你预设的 5% 红线。