基于MQL5和Python的自优化EA(第五部分):深度马尔可夫模型(基础篇)
📘

基于MQL5和Python的自优化EA(第五部分):深度马尔可夫模型(基础篇)

第 1/3 篇

用RSI重写马尔可夫状态定义

前一篇用移动平均线划分市场状态来搭马尔可夫链,转移矩阵能同时给出择时信号和趋势/均值回归倾向。这次把状态变量从均线换成RSI,思路不变但更贴近振荡类品种的惯性。 很多人学RSI只记30买、70卖,但不同外汇或贵金属品种的高波动时段里这套阈值经常失效。用算法从目标品种历史数据里自己学状态边界,比拍脑袋设固定超买超卖线更扛折腾。 具体做法是把RSI离散成若干区间当作隐状态,再用转移概率判断下一根K线更可能延续还是反转。整个过程随你导入的行情样本动态重算,黄金15分钟和欧美1小时跑出来的链不会一样。 外汇与贵金属杠杆高、跳空频繁,任何概率模型都只是倾向而非保证,实盘前务必在MT5用历史数据回测确认转移矩阵稳定性。

「RSI极端值不能跨市场套用」

RSI常被用来抓极端价位反手,逻辑是价格倾向回归均值。但原文图1把XPDUSD(蓝)和NZDUSD(橙)的RSI滚动标准差叠在一起,能直接看出两个市场的RSI波动结构差很离谱:贵金属XPDUSD的RSI标准差均值明显比NZDUSD高出一个量级。 对一个货币对算“高RSI”的阈值,放到XPDUSD里可能只是普通噪音。外汇和贵金属都是高杠杆高风险品种,拿统一RSI参数硬切所有品种,逆势单容易被扫。 所以每个标的都有自己的RSI兴趣区。要让EA自己学“这个品种该在哪个RSI水平买卖”,可以用转移矩阵(transition matrix)去统计任意标的的历史状态迁移,而不是写死30/70。开MT5把RSI标准差按品种分开算一遍,比背教科书阈值有用。

◍ 用 RSI 分桶给 M1 数据找概率优势

我们用 Python 的 MetaTrader 5 库拉取了 30 万条 M1 行情,按 RSI 读数切成 0–10、11–20 直到 91–100 共 10 个区间,标记价格穿过每个区间后的未来表现。训练集里,价格穿过 41–50 区间时后续上涨倾向最强,穿过 61–70 时下跌倾向最强。 直接拿训练出的转移矩阵跑贪婪模型(每次取先验里概率最高的结果),测试集整体准确率 52%。这类概率模型的卖点不在黑箱,而在可解释——你能看清每一笔决策依据哪个区间,外汇和贵金属这种高杠杆品种上,合规风险也低。 但我们没盯整体准确率,而是逐个看 10 个区间的细分表现。训练集分布最高的两个区间在验证集并不可靠;验证集上 11–20 区间做多、71–80 区间做空最稳,各自准确率 51.4% 和 75.8%。最终选这两区间作为 NZDJPY 上开多和开空的触发带。 落地到 MT5,我们写了个 EA 把上述结论跑起来,并给了两种平仓路径:RSI 穿过可能削减持仓风险的区间时平,或价格穿越均线时平,交易者按自己的盯盘习惯二选一。

用 RSI 区间给分钟线打方向标签

做这类统计模型,第一步是把 MT5 的裸数据拉进来并清洗干净。下面这段代码以 NZDJPY 的 M1 周期为例,一次性取 30 万根 K 线,用 pandas_ta 直接算 length=20 的 RSI,并把时间字段从 Unix 秒转成可读 datetime。 标签定义很直接:往后看 20 根(look_ahead=20),若 20 根后的收盘价高于当前,标 -1(空方占优);低于当前则标 1(多方占优)。dropna 后训练集约 10 万行,验证集同样规模。 核心动作是把 RSI 切成 10 个区间,遍历训练集每条记录:RSI 落在哪一档,就把该档计数加上对应的 Target。涨加 1、跌减 1,最终矩阵里正值的档位倾向于引发行情上行,负值则相反。 训练计数暴露出一个坑:91-100 这档在全 10 万行里一次都没出现。邻近档位都是负向,于是人为塞了一个任意负值进去,这点后面验证时会反噬。 验证集上整体准确率 0.5208,刚过随机线。分档看更有意思:11-20 档准确率 0.75,是看涨区间里最高的;71-80 档约 0.51,在看跌档里相对突出;而 91-100 档显示 1.0 的满分,只是因为验证集里它也没几条样本,对实盘基本没参考意义。外汇与贵金属杠杆高,这类统计倾向只能当过滤条件,不能直接当作入场信号。

MQL5 / C++
class="macro">#Let&class="macro">#x27;s get started
class="kw">import MetaTrader5        as mt5
class="kw">import pandas             as pd
class="kw">import numpy              as np
class="kw">import seaborn            as sns
class="kw">import matplotlib.pyplot as plt
class="kw">import pandas_ta          as ta
初始化
class="macro">#Fetch market data
SYMBOL    = "NZDJPY"
TIMEFRAME = mt5.TIMEFRAME_M1
data = pd.DataFrame(mt5.copy_rates_from_pos(SYMBOL,TIMEFRAME,class="num">0,class="num">300000))
data["time"]  = pd.to_datetime(data["time"],unit=&class="macro">#x27;s&class="macro">#x27;)
data.ta.rsi(length=class="num">20,append=True)
class="macro">#Define the look ahead
look_ahead = class="num">20
class="macro">#Label the data
data["Target"] = np.nan
data.loc[data["close"] > data["close"].shift(-class="num">20),"Target"] = -class="num">1
data.loc[data["close"] < data["close"].shift(-class="num">20),"Target"] = class="num">1
data.dropna(inplace=True)
data.reset_index(inplace=True,drop=True)
class="macro">#Create a dataframe
rsi_matrix = pd.DataFrame(columns=["class="num">0-class="num">10","class="num">11-class="num">20","class="num">21-class="num">30","class="num">31-class="num">40","class="num">41-class="num">50","class="num">51-class="num">60","class="num">61-class="num">70","class="num">71-class="num">80","class="num">81-class="num">90","class="num">91-class="num">100"],index=[class="num">0])
数据
class="macro">#Initialize the rsi matrix to class="num">0
for i in np.arange(class="num">0,class="num">9):
    rsi_matrix.iloc[class="num">0,i] = class="num">0
class="macro">#Split the data into train and test sets
train = data.loc[:(data.shape[class="num">0]class=class="str">"cmt">//class="num">2),:]
test  = data.loc[(data.shape[class="num">0]class=class="str">"cmt">//class="num">2):,:]
for i in np.arange(class="num">0,train.shape[class="num">0]):
    class="macro">#Fill in the rsi matrix, what happened in the future when we saw RSI readings below class="num">10?
    if((train.loc[i,"RSI_20"] <= class="num">10)):
        rsi_matrix.iloc[class="num">0,class="num">0] = rsi_matrix.iloc[class="num">0,class="num">0] + train.loc[i,"Target"]
    class="macro">#What tends to happen in the future, after seeing RSI readings between class="num">11 and class="num">20?
    if((train.loc[i,"RSI_20"] > class="num">10) & (train.loc[i,"RSI_20"] <= class="num">20)):
        rsi_matrix.iloc[class="num">0,class="num">1] = rsi_matrix.iloc[class="num">0,class="num">1] + train.loc[i,"Target"]
    class="macro">#What tends to happen in the future, after seeing RSI readings between class="num">21 and class="num">30?

「把 RSI 分桶结果压到预测列里」

上面那段循环把 20 周期 RSI 从 20 到 100 切成 10 个桶,每个桶往 rsi_matrix 第 0 行对应列累加 train 里的 Target 值。跑完直接 print(rsi_matrix) 再用 sns.barplot(rsi_matrix) 画出来,哪一桶累计收益高,后面就倾向按那一桶方向挂单——外汇和贵金属杠杆高,这种统计优势随时被滑点和跳空吃掉,只能当概率参考。 test 集先 reset_index 丢到 drop=True 清掉旧索引,开一列全 nan 的 Predictions。循环里只写了 RSI_20<=10 就给 1,意思是超卖极端桶直接看多;其余桶的映射你得照着前面 rsi_matrix 的柱高自己补 if 分支,别照抄就以为模型完了。 这种写法在 MT5 里没法直接跑,得用 Python 算完把信号 csv 丢给 EA 读。开 MT5 拿 EURUSD 的 H1 跑一遍 RSI(20),看 20~30 桶和 70~80 桶的历史胜率差,大概率能复现非线性衰减,而不是教科书说的对称反转。

MQL5 / C++
if((train.loc[i,"RSI_20"] > class="num">20) & (train.loc[i,"RSI_20"] <= class="num">30)):
    rsi_matrix.iloc[class="num">0,class="num">2] = rsi_matrix.iloc[class="num">0,class="num">2] + train.loc[i,"Target"]
if((train.loc[i,"RSI_20"] > class="num">30) & (train.loc[i,"RSI_20"] <= class="num">40)):
    rsi_matrix.iloc[class="num">0,class="num">3] = rsi_matrix.iloc[class="num">0,class="num">3] + train.loc[i,"Target"]
if((train.loc[i,"RSI_20"] > class="num">40) & (train.loc[i,"RSI_20"] <= class="num">50)):
    rsi_matrix.iloc[class="num">0,class="num">4] = rsi_matrix.iloc[class="num">0,class="num">4] + train.loc[i,"Target"]
if((train.loc[i,"RSI_20"] > class="num">50) & (train.loc[i,"RSI_20"] <= class="num">60)):
    rsi_matrix.iloc[class="num">0,class="num">5] = rsi_matrix.iloc[class="num">0,class="num">5] + train.loc[i,"Target"]
if((train.loc[i,"RSI_20"] > class="num">60) & (train.loc[i,"RSI_20"] <= class="num">70)):
    rsi_matrix.iloc[class="num">0,class="num">6] = rsi_matrix.iloc[class="num">0,class="num">6] + train.loc[i,"Target"]
if((train.loc[i,"RSI_20"] > class="num">70) & (train.loc[i,"RSI_20"] <= class="num">80)):
    rsi_matrix.iloc[class="num">0,class="num">7] = rsi_matrix.iloc[class="num">0,class="num">7] + train.loc[i,"Target"]
if((train.loc[i,"RSI_20"] > class="num">80) & (train.loc[i,"RSI_20"] <= class="num">90)):
    rsi_matrix.iloc[class="num">0,class="num">8] = rsi_matrix.iloc[class="num">0,class="num">8] + train.loc[i,"Target"]
if((train.loc[i,"RSI_20"] > class="num">90) & (train.loc[i,"RSI_20"] <= class="num">100)):
    rsi_matrix.iloc[class="num">0,class="num">9] = rsi_matrix.iloc[class="num">0,class="num">9] + train.loc[i,"Target"]
rsi_matrix
sns.barplot(rsi_matrix)
test.reset_index(inplace=True,drop=True)
test["Predictions"]  = np.nan
for i in np.arange(class="num">0,test.shape[class="num">0]):
    if((test.loc[i,"RSI_20"] <= class="num">10)):
        test.loc[i,"Predictions"] = class="num">1

常见问题

不能。RSI极端值具有市场特异性,跨市场套用会失真,需在本品种历史数据上重算分桶边界。
将RSI按区间切桶,统计各桶内下一根K线涨跌频率,桶内胜率显著高于随机即为可用优势。
可以。小布能对接品种数据自动算RSI区间、打方向标签并输出分桶概率表,你只需复核阈值。
会。震荡市RSI反复穿区间,标签噪声大,建议叠加波动率过滤或仅取极端桶参与。
它把连续RSI转成离散状态特征,降低过拟合,让模型直接学状态间转移而非裸数值。