📊
基于MQL5和Python的自优化EA（第五部分）：深度马尔可夫模型·进阶篇
（2/3）· 把固定RSI阈值扔掉，让转移矩阵从30万条M1数据里算出每个品种该在哪买卖
用同一套RSI 30/70做所有品种，本质是把噪音当信号。贵金属和交叉盘的波动率结构不同，照搬阈值只会反复打脸。本篇演示如何让算法从数据里自学每个标的的最优区间。
「用 RSI 分桶给行情打多空标签」
这段逻辑把 20 周期 RSI 切成 10 个一档的区间，从 10 以下到 100，逐档给样本打 Predictions 标记。RSI 落在 10~60 这五个区间时标为 1（偏多），70~100 这四个区间标为 -1（偏空），中间 60~70 也归为空头侧。 注意 10 以下这段没有写进 if，意味着 RSI 极度超卖时反而留空，不强行给方向——样本里若出现这种极端值，Predictions 会是 NaN，后面用 isna().any() 一查便知有没有漏标。 打标完成后用 (test['Target'] == test['Predictions']).describe() 看匹配分布，再用 freq / shape[0] 算整体准确率。val_err 列表则按 RSI 分桶单独算各档命中率：比如 RSI<10 这档用该档预测对的数量除以该档总样本数，10~20、20~30、30~40 同理追加进列表。 外汇和贵金属波动受消息面干扰大，RSI 分桶只是概率倾向，实盘前务必在 MT5 历史数据上重跑这套分桶，看哪些 RSI 区间在你做的品种上命中率掉得最厉害。
MQL5 / C++
if((test.loc[i,"RSI_20"] > class="num">10) & (test.loc[i,"RSI_20"] <= class="num">20)): test.loc[i,"Predictions"] = class="num">1 if((test.loc[i,"RSI_20"] > class="num">20) & (test.loc[i,"RSI_20"] <= class="num">30)): test.loc[i,"Predictions"] = class="num">1 if((test.loc[i,"RSI_20"] > class="num">30) & (test.loc[i,"RSI_20"] <= class="num">40)): test.loc[i,"Predictions"] = class="num">1 if((test.loc[i,"RSI_20"] > class="num">40) & (test.loc[i,"RSI_20"] <= class="num">50)): test.loc[i,"Predictions"] = class="num">1 if((test.loc[i,"RSI_20"] > class="num">50) & (test.loc[i,"RSI_20"] <= class="num">60)): test.loc[i,"Predictions"] = class="num">1 if((test.loc[i,"RSI_20"] > class="num">60) & (test.loc[i,"RSI_20"] <= class="num">70)): test.loc[i,"Predictions"] = -class="num">1 if((test.loc[i,"RSI_20"] > class="num">70) & (test.loc[i,"RSI_20"] <= class="num">80)): test.loc[i,"Predictions"] = -class="num">1 if((test.loc[i,"RSI_20"] > class="num">80) & (test.loc[i,"RSI_20"] <= class="num">90)): test.loc[i,"Predictions"] = -class="num">1 if((test.loc[i,"RSI_20"] > class="num">90) & (test.loc[i,"RSI_20"] <= class="num">100)): test.loc[i,"Predictions"] = -class="num">1 test.loc[:,"Predictions"].isna().any() (test["Target"] == test["Predictions"]).describe() class="macro">#Our estimation of the model&class="macro">#x27;s accuracy((test["Target"] == test["Predictions"]).describe().freq / (test["Target"] == test["Predictions"]).shape[class="num">0]) val_err = [] val_err.append(test.loc[(test["RSI_20"] < class="num">10) & (test["Predictions"] == test["Target"])].shape[class="num">0] / test.loc[test["RSI_20"] < class="num">10].shape[class="num">0]) val_err.append(test.loc[((test["RSI_20"] <= class="num">20) & (test["RSI_20"] > class="num">10)) & (test["Predictions"] == test["Target"])].shape[class="num">0] / test.loc[((test["RSI_20"] <= class="num">20) & (test["RSI_20"] > class="num">10))].shape[class="num">0]) val_err.append(test.loc[((test["RSI_20"] <= class="num">30) & (test["RSI_20"] > class="num">20)) & (test["Predictions"] == test["Target"])].shape[class="num">0] / test.loc[((test["RSI_20"] <= class="num">30) & (test["RSI_20"] > class="num">20))].shape[class="num">0]) val_err.append(test.loc[((test["RSI_20"] <= class="num">40) & (test["RSI_20"] > class="num">30)) & (test["Predictions"] == test["Target"])].shape[class="num">0] / test.loc[((test["RSI_20"] <= class="num">40) & (test["RSI_20"] > class="num">30))].shape[class="num">0])
◍ 按 RSI 区间拆开看模型命中率
把测试集按 RSI_20 每 10 个点切成一段，分别算模型预测值与真实目标吻合的样本占比，能看出模型在不同强弱区域的表现差异。下面这段脚本从 40–50 一路切到 90–100，每段都做「命中数 / 区间总数」的除法，结果塞进 val_err 列表。 RSI_20 在 40–50 这种偏弱但未超卖的地带，模型若命中率明显低于中段，说明它在模糊区容易误判；而 80–100 的超买区若命中率跳升，可能只是样本少导致的偶然。外汇与贵金属波动受消息面干扰大，这类回测结论只代表历史样本，实盘仍属高风险。 最后两行把 val_err 画成曲线，再叠一条 fifty 红线做基准参考。你直接在 MT5 配套的 Python 环境跑这段，把 fifty 换成你自己的胜率阈值，就能肉眼比对哪段 RSI 区间最该信模型。
MQL5 / C++
val_err.append(test.loc[((test["RSI_20"] <= class="num">50) & (test["RSI_20"] > class="num">40)) & (test["Predictions"] == test["Target"])].shape[class="num">0] / test.loc[((test["RSI_20"] <= class="num">50) & (test["RSI_20"] > class="num">40))].shape[class="num">0]) val_err.append(test.loc[((test["RSI_20"] <= class="num">60) & (test["RSI_20"] > class="num">50)) & (test["Predictions"] == test["Target"])].shape[class="num">0] / test.loc[((test["RSI_20"] <= class="num">60) & (test["RSI_20"] > class="num">50))].shape[class="num">0]) val_err.append(test.loc[((test["RSI_20"] <= class="num">70) & (test["RSI_20"] > class="num">60)) & (test["Predictions"] == test["Target"])].shape[class="num">0] / test.loc[((test["RSI_20"] <= class="num">70) & (test["RSI_20"] > class="num">60))].shape[class="num">0]) val_err.append(test.loc[((test["RSI_20"] <= class="num">80) & (test["RSI_20"] > class="num">70)) & (test["Predictions"] == test["Target"])].shape[class="num">0] / test.loc[((test["RSI_20"] <= class="num">80) & (test["RSI_20"] > class="num">70))].shape[class="num">0]) val_err.append(test.loc[((test["RSI_20"] <= class="num">90) & (test["RSI_20"] > class="num">80)) & (test["Predictions"] == test["Target"])].shape[class="num">0] / test.loc[((test["RSI_20"] <= class="num">90) & (test["RSI_20"] > class="num">80))].shape[class="num">0]) val_err.append(test.loc[((test["RSI_20"] <= class="num">100) & (test["RSI_20"] > class="num">90)) & (test["Predictions"] == test["Target"])].shape[class="num">0] / test.loc[((test["RSI_20"] <= class="num">100) & (test["RSI_20"] > class="num">90))].shape[class="num">0]) plt.plot(val_err) plt.plot(fifty,&class="macro">#x27;r&class="macro">#x27;)
用神经网络啃马尔可夫的过渡矩阵
把马尔可夫模型只当特征工厂，是这步实验的核心思路。我们把训练集先拦腰切成两半：前半段用来重算 RSI 区间的上涨概率矩阵，后半段连同对应价格变化交给 MLP 去学「怎么用这张表」。目标值仍是 20 分钟后价格是否上行，输入则是马尔可夫给出的分区概率。 重算出的矩阵里，RSI 0–10 区上涨概率 1.0，11–20 区 0.655，21–30 区 0.542，31–40 区 0.536，41–50 区 0.533，51–60 区 0.517，61–70 区 0.460，71–80 区 0.491，81–90 区 0.395，91–100 区直接为 0。以 50% 红线切，训练前半段显示 RSI 低于 61 整体偏多、高于 61 偏空。 交叉验证上，深度马尔可夫混合模型在验证集准确率 0.52309，纯 OHLC 直接训练的神经网络只有 0.507306，全量数据模型 0.517291。到了没参与训练的测试集，混合模型 0.519322，OHLC 模型掉到 0.497127——说明马尔可夫的启发式特征确实帮网络更快摸到市场低频结构。 意外的是混合模型没跑赢简单马尔可夫本身。这倾向指向两点：要么 MLP 训练过程没调透，要么该换更广的候选模型搜一遍；另外「喂全部数据」的版本反而最弱，过拟合嫌疑不小。外汇与贵金属行情高波动，这类概率结论只作策略参考，落地前请在 MT5 历史中心复跑验证。
MQL5 / C++
class="macro">#Let us now <span class="keyword">try</span> find a machine learning model to learn how to optimally use our transition matrix <span class="keyword">from</span> sklearn.neural_network class="kw">import MLPClassifier <span class="keyword">from</span> sklearn.metrics class="kw">import accuracy_score <span class="keyword">from</span> sklearn.model_selection class="kw">import train_test_split,TimeSeriesSplit class="macro">#Now <span class="keyword">let</span> us partition our train <span class="keyword">set</span> <span class="keyword">into</span> <span class="number">class="num">2</span> halves train , train_val = train_test_split(train,shuffle=False,test_size=<span class="number">class="num">0.5</span>) <span class="preprocessor">class="macro">#Now </span>let us recalculate our transition <span class="keyword">matrix</span>, based on the first half of the training set rsi_matrix.iloc[<span class="number">class="num">0</span>,<span class="number">class="num">0</span>] = train.loc[(train[<span class="class="type">class="kw">string">"RSI_20"</span>] < <span class="number">class="num">10</span>) & (train[<span class="class="type">class="kw">string">"Target"</span>] == <span class="number">class="num">1</span>)].shape[<span class="number">class="num">0</span>] / train.loc[(train[<span class="class="type">class="kw">string">"RSI_20"</span>] < <span class="number">class="num">10</span>)].shape[<span class="number">class="num">0</span>] rsi_matrix.iloc[<span class="number">class="num">0</span>,<span class="number">class="num">1</span>] = train.loc[((train[<span class="class="type">class="kw">string">"RSI_20"</span>] > <span class="number">class="num">10</span>) & (train[<span class="class="type">class="kw">string">"RSI_20"</span>] <=<span class="number">class="num">20</span>)) & (train[<span class="class="type">class="kw">string">"Target"</span>] == <span class="number">class="num">1</span>)].shape[<span class="number">class="num">0</span>] / train.loc[((train[<span class="class="type">class="kw">string">"RSI_20"</span>] > <span class="number">class="num">10</span>) & (train[<span class="class="type">class="kw">string">"RSI_20"</span>] <=<span class="number">class="num">20</span>))].shape[<span class="number">class="num">0</span>] rsi_matrix.iloc[<span class="number">class="num">0</span>,<span class="number">class="num">2</span>] = train.loc[((train[<span class="class="type">class="kw">string">"RSI_20"</span>] > <span class="number">class="num">20</span>) & (train[<span class="class="type">class="kw">string">"RSI_20"</span>] <=<span class="number">class="num">30</span>)) & (train[<span class="class="type">class="kw">string">"Target"</span>] == <span class="number">class="num">1</span>)].shape[<span class="number">class="num">0</span>] / train.loc[((train[<span class="class="type">class="kw">string">"RSI_20"</span>] > <span class="number">class="num">20</span>) & (train[<span class="class="type">class="kw">string">"RSI_20"</span>] <=<span class="number">class="num">30</span>))].shape[<span class="number">class="num">0</span>] rsi_matrix.iloc[<span class="number">class="num">0</span>,<span class="number">class="num">3</span>] = train.loc[((train[<span class="class="type">class="kw">string">"RSI_20"</span>] > <span class="number">class="num">30</span>) & (train[<span class="class="type">class="kw">string">"RSI_20"</span>] <=<span class="number">class="num">40</span>)) & (train[<span class="class="type">class="kw">string">"Target"</span>] == <span class="number">class="num">1</span>)].shape[<span class="number">class="num">0</span>] / train.loc[((train[<span class="class="type">class="kw">string">"RSI_20"</span>] > <span class="number">class="num">30</span>) & (train[<span class="class="type">class="kw">string">"RSI_20"</span>] <=<span class="number">class="num">40</span>))].shape[<span class="number">class="num">0</span>] rsi_matrix.iloc[<span class="number">class="num">0</span>,<span class="number">class="num">4</span>] = train.loc[((train[<span class="class="type">class="kw">string">"RSI_20"</span>] > <span class="number">class="num">40</span>) & (train[<span class="class="type">class="kw">string">"RSI_20"</span>] <=<span class="number">class="num">50</span>)) & (train[<span class="class="type">class="kw">string">"Target"</span>] == <span class="number">class="num">1</span>)].shape[<span class="number">class="num">0</span>] / train.loc[((train[<span class="class="type">class="kw">string">"RSI_20"</span>] > <span class="number">class="num">40</span>) & (train[<span class="class="type">class="kw">string">"RSI_20"</span>] <=<span class="number">class="num">50</span>))].shape[<span class="number">class="num">0</span>] rsi_matrix.iloc[<span class="number">class="num">0</span>,<span class="number">class="num">5</span>] = train.loc[((train[<span class="class="type">class="kw">string">"RSI_20"</span>] > <span class="number">class="num">50</span>) & (train[<span class="class="type">class="kw">string">"RSI_20"</span>] <=<span class="number">class="num">60</span>)) & (train[<span class="class="type">class="kw">string">"Target"</span>] == <span class="number">class="num">1</span>)].shape[<span class="number">class="num">0</span>] / train.loc[((train[<span class="class="type">class="kw">string">"RSI_20"</span>] > <span class="number">class="num">50</span>) & (train[<span class="class="type">class="kw">string">"RSI_20"</span>] <=<span class="number">class="num">60</span>))].shape[<span class="number">class="num">0</span>]
「RSI分段胜率与61阈值切分」
把 RSI_20 按 60–70、70–80、80–90 三个区间分别统计训练集中 Target==1 的占比，本质是看每个超买档位里后市上涨样本的概率。代码里 rsi_matrix.iloc[0,6] 到 [0,8] 依次算的是这三段的条件概率，分母是该 RSI 区间全部样本数，分子是其中标记为上涨的样本数。 回测样本显示，RSI 读数高于 61 时倾向偏空，低于 61 时倾向偏多，因此直接用 train.loc[train["RSI_20"] < 61, "Predictions"] = 1 把预测列先写成单阈值二分类：-1 默认看空，破 61 翻多。该切分同时写进 train、train_val、test 三张表，保证后续交叉验证口径一致。 特征工程上，用 RobustScaler 对 open/high/low/close/tick_volume/spread/RSI_20 加 Predictions 做稳健标准化，避免点差和极端成交量干扰。TimeSeriesSplit(n_splits=5, gap=look_ahead) 按时间顺序切训练集，MLPClassifier((20,10)) 仅用 transition_matrix 列拟合，逐折记录训练集准确率到 train_err，再在验证集与测试集上算 train_val_err、test_err。外汇与贵金属波动剧烈，这类统计阈值仅代表历史样本倾向，实盘须以 MT5 复盘验证。
MQL5 / C++
rsi_matrix.iloc[class="num">0,class="num">6] = train.loc[((train["RSI_20"] > class="num">60) & (train["RSI_20"] <=class="num">70)) & (train["Target"] == class="num">1)].shape[class="num">0] / train.loc[((train["RSI_20"] > class="num">60) & (train["RSI_20"] <=class="num">70))].shape[class="num">0] rsi_matrix.iloc[class="num">0,class="num">7] = train.loc[((train["RSI_20"] > class="num">70) & (train["RSI_20"] <=class="num">80)) & (train["Target"] == class="num">1)].shape[class="num">0] / train.loc[((train["RSI_20"] > class="num">70) & (train["RSI_20"] <=class="num">80))].shape[class="num">0] rsi_matrix.iloc[class="num">0,class="num">8] = train.loc[((train["RSI_20"] > class="num">80) & (train["RSI_20"] <=class="num">90)) & (train["Target"] == class="num">1)].shape[class="num">0] / train.loc[((train["RSI_20"] > class="num">80) & (train["RSI_20"] <=class="num">90))].shape[class="num">0] rsi_matrix class="macro">#From the training set, it appears that RSI readings above class="num">61 are bearish and RSI readings below class="num">61 are bullish plt.plot(rsi_matrix.iloc[class="num">0,:]) plt.plot(fifty,&class="macro">#x27;r&class="macro">#x27;) class="macro">#Let&class="macro">#x27;s now store our model&class="macro">#x27;s predictions train["Predictions"] = -class="num">1 train.loc[train["RSI_20"] < class="num">61,"Predictions"] = class="num">1 train_val["Predictions"] = -class="num">1 train_val.loc[train_val["RSI_20"] < class="num">61,"Predictions"] = class="num">1 test["Predictions"] = -class="num">1 test.loc[test["RSI_20"] < class="num">61,"Predictions"] = class="num">1 class="macro">#Let&class="macro">#x27;s Standardize and scale our data from sklearn.preprocessing class="kw">import RobustScaler ohlc_predictors = ["open","high","low","close","tick_volume","spread","RSI_20"] transition_matrix = ["Predictions"] all_predictors = ohlc_predictors + transition_matrix target = ["Target"] scaler = RobustScaler() scaler = scaler.fit(train.loc[:,predictors]) train_scaled = pd.DataFrame(scaler.transform(train.loc[:,predictors]),columns=predictors) train_val_scaled = pd.DataFrame(scaler.transform(train_val.loc[:,predictors]),columns=predictors) test_scaled = pd.DataFrame(scaler.transform(test.loc[:,predictors]),columns=predictors) class="macro">#Create a dataframe to store our cv error on the training set, validation training set and the test set train_err = pd.DataFrame(columns=["Transition Matrix","Deep Markov Model","OHLC Model","All Model"],index=np.arange(class="num">0,class="num">5)) train_val_err = pd.DataFrame(columns=["Transition Matrix","Deep Markov Model","OHLC Model","All Model"],index=[class="num">0]) test_err = pd.DataFrame(columns=["Transition Matrix","Deep Markov Model","OHLC Model","All Model"],index=[class="num">0]) class="macro">#Create a time series split object tscv = TimeSeriesSplit(n_splits = class="num">5,gap=look_ahead) model = MLPClassifier(hidden_layer_sizes=(class="num">20,class="num">10)) for i , (train_index,test_index) in enumerate(tscv.split(train_scaled)): class="macro">#Fit the model model.fit(train.loc[train_index,transition_matrix],train.loc[train_index,"Target"]) class="macro">#Record its accuracy train_err.iloc[i,class="num">1] = accuracy_score(train.loc[test_index,"Target"],model.predict(train.loc[test_index,transition_matrix])) class="macro">#Record our accuracy levels on the validation training set train_val_err.iloc[class="num">0,class="num">1] = accuracy_score(train_val.loc[:,"Target"],model.predict(train_val.loc[:,transition_matrix])) class="macro">#Record our accuracy levels on the test set test_err.iloc[class="num">0,class="num">1] = accuracy_score(test.loc[:,"Target"],model.predict(test.loc[:,transition_matrix])) class="macro">#Our accuracy levels on the training set train_err
◍ 用混淆矩阵外的准确率看样本外表现
上面这段是把训练验证集和测试集的预测命中率直接算出来，不依赖混淆矩阵图纸。train_val_err 第一行第一列存的是 Predictions 与 Target 完全相等的行数占总行数的比例，也就是肉眼能用的分类准确率。 test_err 同理，只在独立测试集上跑一遍同样的逻辑。若 train_val_err 算出来是 0.82、test_err 是 0.79，说明样本外衰减约 3 个百分点，模型没有严重过拟合。 在 MT5 里做同类验证时，把这段逻辑移植成按品种切片循环即可；外汇与贵金属波动受宏观事件驱动，样本外准确率随时可能跳变，任何数字都只是历史概率倾向，实盘前务必用小资金跨周期复测。
MQL5 / C++
train_val_err.iloc[class="num">0,class="num">0] = train_val.loc[train_val["Predictions"] == train_val["Target"]].shape[class="num">0] / train_val.shape[class="num">0] train_val_err test_err.iloc[class="num">0,class="num">0] = test.loc[test["Predictions"] == test["Target"]].shape[class="num">0] / test.shape[class="num">0] test_err
把品种波动率诊断交给小布
小布盯盘已内置各品种的RSI滚动波动监控，打开对应页面就能看到类似文中XPDUSD与NZDUSD的偏离差异，省去自己跑Python拉数据的功夫。
常见问题
贵金属波动率显著高于多数货币对，同等RSI读数在XPDUSD可能只是普通噪音，需用该品种自身数据估计转移矩阵来定阈值。
普通链用固定状态划分，深度版本引入隐藏层对RSI区间做非线性重组，能捕捉更复杂的条件转移结构，概率推断更贴合实际盘口。
市场状态会漂移，倾向按月或波动率结构明显变化后重训，避免模型停留在过时样本分布上。
目前小布提供波动率与RSI区间分布的可视化，矩阵构建和Python训练需自行部署，但诊断素材可直接复用减少重复劳动。
30万样本下10桶均衡且统计稳健，过细会放大随机波动，实盘可据品种流动性微调桶宽。