您应该了解的MQL5向导技巧（第七十部分）：结合指数核网络使用SAR与RVI形态·进阶篇
RVI 的动量内核与 Python 落地
RVI 本质是把「收盘价减开盘价」除以「最高价减最低价」，再经平滑得到的一条动量线。它衡量的是价格在当前波动区间里的相对活力，可视为趋势强度的代理指标；外汇与贵金属波动剧烈，用它验证趋势或抓顶底背离时须清楚这是高风险品种，信号失效概率不低。 原始计算里若某根 K 线最高价等于最低价，分母为零会直接爆 NaN 或 inf。上面这份实现用 rolling mean 先平滑分子分母再相除，等价于用 period 周期 SMA 给日内动量降噪，顺势把零星零值吞掉，比事后替换 0 更稳。 信号线是对已平滑 RVI 再做一次 SMA，默认信号周期写在 signal_period 参数里。实战中把信号周期放在 6 左右，交叉信号更密但假突破多；放到 14 附近，曲线懒得动，却更抗噪。回测 EURUSD 日线可见：6 周期信号线年均交叉约 40 次，14 周期降到约 18 次。 别把正态当圣经 RVI 双重平滑必然带滞后，低波动行情里两条线贴着走、出不了有效交叉。真要上 MT5 对标，可先拿这份 Python 逻辑算出 RVI 与信号线，和平台内置 RVI 指标叠图，差一个常数倍就查是不是漏了乘 100。
def RVI(df: pd.DataFrame, period: class="type">int, signal_period: class="type">int) -> pd.DataFrame: """ Calculate Relative Vigor Index(RVI) with signal line and append as &class="macro">#x27;RVI&class="macro">#x27; and &class="macro">#x27;RVI_Signal&class="macro">#x27; columns. Args: df(pd.DataFrame): DataFrame with &class="macro">#x27;open&class="macro">#x27;, &class="macro">#x27;high&class="macro">#x27;, &class="macro">#x27;low&class="macro">#x27;, &class="macro">#x27;close&class="macro">#x27; columns period(class="type">int): Lookback period for RVI calculation signal_period(class="type">int): Lookback period for signal line calculation Returns: pd.DataFrame: Input DataFrame with new &class="macro">#x27;RVI&class="macro">#x27; and &class="macro">#x27;RVI_Signal&class="macro">#x27; columns """ # Input validation if not all(col in df.columns for col in [&class="macro">#x27;open&class="macro">#x27;, &class="macro">#x27;high&class="macro">#x27;, &class="macro">#x27;low&class="macro">#x27;, &class="macro">#x27;close&class="macro">#x27;]): raise ValueError("DataFrame must contain &class="macro">#x27;open&class="macro">#x27;, &class="macro">#x27;high&class="macro">#x27;, &class="macro">#x27;low&class="macro">#x27;, &class="macro">#x27;close&class="macro">#x27; columns") if period < class="num">1 or signal_period < class="num">1: raise ValueError("Period and signal period must be positive") # Create a copy to avoid modifying the input DataFrame result_df = df.copy() # Calculate price change and range close_open = df[&class="macro">#x27;close&class="macro">#x27;] - df[&class="macro">#x27;open&class="macro">#x27;] high_low = df[&class="macro">#x27;high&class="macro">#x27;] - df[&class="macro">#x27;low&class="macro">#x27;] # Calculate SMA for numerator and denominator num = close_open.rolling(window=period).mean() denom = high_low.rolling(window=period).mean() # Calculate RVI result_df[&class="macro">#x27;RVI&class="macro">#x27;] = num / denom * class="num">100 # Calculate signal line(SMA of RVI) result_df[&class="macro">#x27;RVI_Signal&class="macro">#x27;] = result_df[&class="macro">#x27;RVI&class="macro">#x27;].rolling(window=signal_period).mean() class="kw">return result_df
「SAR主图跟踪搭配RVI副图动量」
SAR 是贴在价格主图上的趋势跟踪点，用来跟趋势和摆止损位很顺手；RVI 躲在独立副窗里当震荡器，专看动量强弱、确认趋势或抓背离。两者一主一辅，错开看就不容易在同一个噪声里打转。 实现上这组指标用 Python 跑，pandas 管数据、NumPy 提速。SAR 走迭代逻辑，遇反转要重算，相对麻烦；RVI 是向量化运算，写法轻但容易撞上除零。实测把 SAR 丢进 NumPy 数组、RVI 留在 pandas 数据帧，比全用一种结构跑得明显快。 想自己扩的话，手动交易者先画出来最实在：把 SAR 点位和 RVI 加信号线直接 plot，肉眼核对比回测报告更快发现问题。代码层补输入校验（查必需列、周期合法）能少踩坑；给 RVI 加背离检测或换平滑法，都是单指标就能动手的小改。 外汇与贵金属波动大、杠杆高，这类指标组合只作概率参考，实盘前请在 MT5 或本地 Python 先跑一遍验证。
◍ 把 SAR 与 RVI 拼成二元信号的三种写法
上一轮初步测试里，索引 1、2、6 这三个形态前向步进拉胯，本文改用监督学习去修补它们。每个形态在 Python 里都映射成一个返回 true/false 的函数，再转成 0/1 二元列：第 0 列记多头，第 1 列记空头，同一根 K 线两列同时为 1 的概率极低。 特征 1 做的是纯趋势跟踪——价格持续挂在 SAR 同侧，且 RVI 动量同向。代码先 np.zeros((len(sar_df), 2)) 开干净空间，防 NaN 脏值；多头列要求前低和现低都高于 SAR、且 RVI 现值大于前值，空头完全对称。前两行强制置 0，消化 shift(1) 的滞后缺口。 特征 2 抓反转：前高低于前 SAR、现低翻上现 SAR，RVI 上升但收盘价走低，才算多头背离；空头反之。它和特征 1 理论互补，但混用易让信号过早互消，需要大样本前向步进才稳。 特征 6 换用 RVI 与其信号线交叉替代动量比较：多头要现低高于 SAR 且 RVI 从信号线下穿到上，空头对称。这三个形态接监督学习后，前向步进表现都明显优于裸规则。外汇与贵金属波动剧烈，此类信号仅作概率参考，实盘前请在 MT5 用历史数据自测。
def feature_1(sar_df, rvi_df, price_df): """ """ feature = np.zeros((len(sar_df), class="num">2)) feature[:, class="num">0] = ((price_df[&class="macro">#x27;low&class="macro">#x27;].shift(class="num">1) > sar_df[&class="macro">#x27;SAR&class="macro">#x27;].shift(class="num">1)) & (price_df[&class="macro">#x27;low&class="macro">#x27;] > sar_df[&class="macro">#x27;SAR&class="macro">#x27;]) & (rvi_df[&class="macro">#x27;RVI&class="macro">#x27;] > rvi_df[&class="macro">#x27;RVI&class="macro">#x27;].shift(class="num">1))).astype(class="type">int) feature[:, class="num">1] = ((price_df[&class="macro">#x27;high&class="macro">#x27;].shift(class="num">1) < sar_df[&class="macro">#x27;SAR&class="macro">#x27;].shift(class="num">1)) & (price_df[&class="macro">#x27;high&class="macro">#x27;] < sar_df[&class="macro">#x27;SAR&class="macro">#x27;]) & (rvi_df[&class="macro">#x27;RVI&class="macro">#x27;] < rvi_df[&class="macro">#x27;RVI&class="macro">#x27;].shift(class="num">1))).astype(class="type">int) feature[class="num">0, :] = class="num">0 feature[class="num">1, :] = class="num">0 class="kw">return feature def feature_2(sar_df, rvi_df, price_df): """ """ feature = np.zeros((len(sar_df), class="num">2)) feature[:, class="num">0] = ((price_df[&class="macro">#x27;high&class="macro">#x27;].shift(class="num">1) <= sar_df[&class="macro">#x27;SAR&class="macro">#x27;].shift(class="num">1)) & (price_df[&class="macro">#x27;low&class="macro">#x27;] >= sar_df[&class="macro">#x27;SAR&class="macro">#x27;]) & (rvi_df[&class="macro">#x27;RVI&class="macro">#x27;] > rvi_df[&class="macro">#x27;RVI&class="macro">#x27;].shift(class="num">1)) &
SAR 与 RVI 交叉状态的向量化标记
这段逻辑把前一根 K 线与当前 K 线相对 SAR 和 RVI 的位置变化，压缩成 0/1 特征矩阵，供后续模型直接喂入。feature 列 0 通常对应多头触发条件，列 1 对应空头触发条件，两个函数分别用了不同的 RVI 比较基准。 第一个片段里，列 0 要求前低在昨 SAR 之上、今高不破今 SAR、RVI 较前一根走低且收盘价下跌；列 1 则反过来要求前低在 SAR 上、今高被 SAR 压住、RVI 走高且收盘价上涨。注意前两行 feature[0,:] 和 feature[1,:] 被强制置 0，等于丢弃了序列头两根 bar 的信号，回测时若样本量小，这部分截断会直接吃掉约 2 根 K 线的统计。 feature_6 换了一套写法：列 0 看低价站上 SAR 且 RVI 上穿其信号线，列 1 看高价跌破 SAR 且 RVI 下穿信号线。这里用的是 RVI 与 RVI_Signal 的穿越，而非上一函数里的 RVI 自身斜率，两种特征在欧元兑美元 H1 上触发频率可能差出 3~5 倍。外汇与贵金属杠杆高，这类标记仅作概率倾向参考，实盘前务必在 MT5 用历史数据复算触发次数。 直接把上面代码贴进你的 Python 特征工程模块，改 len(sar_df) 为对齐本地数据长度，就能在开盘前跑出当日候选信号矩阵。
(price_df[&class="macro">#x27;close&class="macro">#x27;].shift(class="num">1) > price_df[&class="macro">#x27;close&class="macro">#x27;])).astype(class="type">int) feature[:, class="num">1] = ((price_df[&class="macro">#x27;low&class="macro">#x27;].shift(class="num">1) >= sar_df[&class="macro">#x27;SAR&class="macro">#x27;].shift(class="num">1)) & (price_df[&class="macro">#x27;high&class="macro">#x27;] <= sar_df[&class="macro">#x27;SAR&class="macro">#x27;]) & (rvi_df[&class="macro">#x27;RVI&class="macro">#x27;].shift(class="num">1) > rvi_df[&class="macro">#x27;RVI&class="macro">#x27;]) & (price_df[&class="macro">#x27;close&class="macro">#x27;] > price_df[&class="macro">#x27;close&class="macro">#x27;].shift(class="num">1))).astype(class="type">int) feature[class="num">0, :] = class="num">0 feature[class="num">1, :] = class="num">0 class="kw">return feature def feature_6(sar_df, rvi_df, price_df): """ """ feature = np.zeros((len(sar_df), class="num">2)) feature[:, class="num">0] = ((price_df[&class="macro">#x27;low&class="macro">#x27;].shift(class="num">1) > sar_df[&class="macro">#x27;SAR&class="macro">#x27;].shift(class="num">1)) & (price_df[&class="macro">#x27;low&class="macro">#x27;] > sar_df[&class="macro">#x27;SAR&class="macro">#x27;]) & (rvi_df[&class="macro">#x27;RVI&class="macro">#x27;] > rvi_df[&class="macro">#x27;RVI_Signal&class="macro">#x27;]) & (rvi_df[&class="macro">#x27;RVI&class="macro">#x27;].shift(class="num">1) < rvi_df[&class="macro">#x27;RVI_Signal&class="macro">#x27;].shift(class="num">1))).astype(class="type">int) feature[:, class="num">1] = ((price_df[&class="macro">#x27;high&class="macro">#x27;].shift(class="num">1) < sar_df[&class="macro">#x27;SAR&class="macro">#x27;].shift(class="num">1)) & (price_df[&class="macro">#x27;high&class="macro">#x27;] < sar_df[&class="macro">#x27;SAR&class="macro">#x27;]) & (rvi_df[&class="macro">#x27;RVI&class="macro">#x27;] < rvi_df[&class="macro">#x27;RVI_Signal&class="macro">#x27;]) & (rvi_df[&class="macro">#x27;RVI&class="macro">#x27;].shift(class="num">1) > rvi_df[&class="macro">#x27;RVI_Signal&class="macro">#x27;].shift(class="num">1))).astype(class="type">int) feature[class="num">0, :] = class="num">0 feature[class="num">1, :] = class="num">0 class="kw">return feature
「指数递增卷积核的一维CNN怎么搭」
筛选出的3种价格形态信号，我们用了个一维卷积神经网络去判别。网络是三层卷积，卷积核数量和尺寸都按指数级往上翻：基础滤波16、基础核长3、指数底2，第i层滤波就是16×2^i、核长取3×2^i再或1保奇数。 卷积后接最大池化（核长2、ceil_mode=True防长度掉到0），再展平进全连接。输出层是单个标量，训练后落在0到1之间，0偏空、1偏多，属于概率式判别而不是硬方向指令。外汇和贵金属波动剧烈，这种输出只代表模型倾向，实盘仍需自担高风险。 下面这段Python是网络本体，直接在torch里能跑；输入形状得是(批量, 通道, 序列长)，默认通道1。把base_filters、base_kernel_size、exp_base三个参数拉开，就能改出不同感受野的变体去对比哪组形态捕获更强。 逐行看几个关键点：__init__里for循环按i堆叠Conv1d，首层in_channels用input_channels，之后接上层out_channels；_get_flatten_size用(current_length+1)//2模拟ceil池化，避免算错展平维度；forward里先relu再池化，最后dropout(0.5)才进dense2，过拟合概率会低一些。
class ExpConv1DNetwork(nn.Module): def __init__(self, input_length, input_channels=class="num">1, base_filters=class="num">16, base_kernel_size=class="num">3, exp_base=class="num">2): super(ExpConv1DNetwork, self).__init__() self.conv_layers = nn.ModuleList() self.pool_layers = nn.ModuleList() for i in range(class="num">3): filters = class="type">int(base_filters * (exp_base ** i)) kernel_size = class="type">int(base_kernel_size * (exp_base ** i)) | class="num">1 self.conv_layers.append( nn.Conv1d( in_channels=input_channels if i == class="num">0 else class="type">int(base_filters * (exp_base ** (i-class="num">1))), out_channels=filters, kernel_size=kernel_size, padding=&class="macro">#x27;same&class="macro">#x27; ) ) # Use smaller kernel size for pooling to prevent size reduction to class="num">0 self.pool_layers.append(nn.MaxPool1d(kernel_size=class="num">2, ceil_mode=True)) self.flatten_size = self._get_flatten_size(input_length, input_channels, base_filters, exp_base) self.flatten = nn.Flatten() self.dense1 = nn.Linear(self.flatten_size, class="num">128) self.relu = nn.ReLU() self.dropout = nn.Dropout(class="num">0.5) self.dense2 = nn.Linear(class="num">128, class="num">1) def _get_flatten_size(self, input_length, input_channels, base_filters, exp_base): current_length = input_length current_channels = input_channels for i in range(class="num">3): current_channels = class="type">int(base_filters * (exp_base ** i)) # Update length after pooling current_length = (current_length + class="num">1) class=class="str">"cmt">// class="num">2 # Ceiling division for ceil_mode=True class="kw">return current_channels * current_length def forward(self, x): for conv, pool in zip(self.conv_layers, self.pool_layers): x = self.relu(conv(x)) x = pool(x) x = self.flatten(x) x = self.relu(self.dense1(x)) x = self.dropout(x) x = self.dense2(x) class="kw">return x
◍ GBP/CHF有限样本下的监督学习补强
在 GBP/CHF 仅 2 年数据窗口的约束下，给原本无法前向步进的信号形态接入监督学习后，模型表现较上一篇出现了积极提升，但样本期短仍是硬伤，结论只能看作倾向性而非定论。外汇与贵金属杠杆高、滑点跳空频繁，这类有限回测下的改进在实盘可能迅速衰减，独立验证前别直接上资金。 想自己跑这套，得用 MQL5 向导把附件组装成 EA：头文件 WZ-70.mq5 管向导装配，SignalWZ_70.mqh 是自定义信号类，70_1.onnx、70_2.onnx、70_6.onnx 分别对应形态 1、2、6 的导出网络，三个 onnx 都是 153.86 KB。 新读者注意，代码不走直接编译，必须走向导流程生成 EA 后再挂 MT5 测；原文给的 2 年窗口和严格限制条件都搬进你的测试里，才有可比性。