您应该了解的MQL5向导技巧（第七十六部分）：利用AO形态与包络通道进行监督学习建模
从形态检验到神经网络过滤
前两篇文章里，动量震荡叠加包络通道的组合策略在两年前向测试窗口中，10类形态有7至8类通过了步进检验。这个通过率说明形态本身有边际有效性，但仍有近三成形态在前向样本上失效。 既然指标组合已经筛过一轮，下一个自然问题是：机器学习能不能替我们把剩下的噪声信号再压一压。本文挑了第4、8、9类形态做对象，用监督学习网络当信号过滤器，看它能不能改善这三类的交易表现。 具体用的是CNN模型，靠跨时间注意力点积核去自适应调卷积核和特征通道的维度参数。外汇和贵金属市场高风险，这类优化只代表历史样本上的概率倾向，实盘仍需自行验证。
◍ 用跨时间点积给CNN卷积核瘦身扩容
跨时间注意力点积核干的事很直接：拿两组时序特征 X=[x1…xT] 和 Y=[y1…yS]，对每一对 (xi, yj) 算 Scorei,j = xi·yj，再对 i 或 j 做 Softmax 拿到权重，拿去重新加权融合跨时间特征。除投影参数外不训额外参数，能啃变长序列、非均匀采样，也能让模型盯着关键时序节点而不是卷积累积的局部邻域。 把它塞进 CNN 不是为了炫技。CNN 天生盯局部，点积核能按数据自身的时序节律自适应放大或收缩有效感受野；在通道维还能筛出谁被持续忽略——一直没权重的通道就是冗余，可直接剪掉压缩规模，若注意力散得一塌糊涂则说明维度不够，反向指导加通道。思路和我们先前动态 CNN 的路线一致：别拿一套固定核/通道硬套非平稳的贵金属、外汇行情。 代价也得摊开说。复杂度随序列长走 O(N²)，传统静态 CNN 才 O(N)，长时序下显卡吃不消；注意力分数叠上动态核选择，解释性反而更糊。实测里纯局部任务（如图像边缘）传统 CNN 仍胜出，且这机制训起来比传统 CNN 更饿数据，动态调控过头容易把行情里的随机扰动当规律拟合，过拟合风险偏高——外汇/贵金属杠杆品种本就高波动，这么搞更得留神。 真要换路子，空洞卷积不涨核就能扩感受野，动态卷积按输入调权重，深度可分离卷积只动通道换效率，SE 模块、TCN 也都是现成改良。开 MT5 接历史 tick 跑一遍 O(N²) 开销对比，比纸上谈参数实在。
「动态卷积加交叉时间注意力的混合骨架」
这套 1D 时序网络把 CNN 局部提取和两种注意力机制揉在一起：常规多头注意力负责全局跨步依赖，自写的点积核负责显式计算时间步两两相似度。初始化时不再写死卷积尺寸，而是用 _design_architecture 按规则生成——代码里循环 10 层，核宽按 3,5,7,9 周期轮转，通道数从 32 倍增到 320，相当于每层学更抽象表征。 层堆叠用 nn.ModuleList 实现可变深度，填充取 kernel_size//2，所以绝大多数层输入输出长度对齐，时序任务里这能保住每个时间步不错位。每隔一层（i%2==0）插一个 4 头 MultiheadAttention，交替而不是逐层插，明显省算力，也让局部和全局推理能切换。 _dot_product_kernel 里先把形状 permute 成 (B,L,C)，用 torch.bmm 算 (B,L,L) 的点积分数，再除以嵌入维平方根做 softmax 温度缩放，最后用注意力权重回混特征并恢复原形。这个跨时间感知是单独卷积拿不到的。外汇和贵金属行情用这类模型预测趋势属于高风险，历史适配不代表后续胜率。 前向里先 unsqueeze(1) 补通道维满足 1D 卷积要求，每层卷积后若有注意力层就 permute 成 (L,B,C) 喂给 PyTorch 多头接口，出来再 permute 回去并叠一次点积核。头部用 AdaptiveAvgPool1d(1) 把任意长度压成定长，展平后三层全连接加 dropout，Sigmoid 吐单值：<0.5 倾向跌，>0.5 倾向涨。想验证就直接把下面代码丢进 MT5 的 Python 环境跑 forward，看不同 input_length 下输出是否稳定。
class="kw">import torch class="kw">import torch.nn as nn class="kw">import torch.nn.functional as F class DotProductAttentionConv1D(nn.Module): def __init__(self, input_length=class="num">100): super(DotProductAttentionConv1D, self).__init__() self.input_length = input_length # Deeper and wider design with attention self.kernel_sizes, self.channels = self._design_architecture() self.conv_layers = nn.ModuleList() self.attention_layers = nn.ModuleList() # For cross-time attention in_channels = class="num">1 for i, (out_channels, kernel_size) in enumerate(zip(self.channels, self.kernel_sizes)): # Convolutional block conv_layer = nn.Sequential( nn.Conv1d(in_channels, out_channels, kernel_size=kernel_size, padding=kernel_size class=class="str">"cmt">// class="num">2), nn.BatchNorm1d(out_channels), nn.ReLU(), nn.Dropout(class="num">0.2)) self.conv_layers.append(conv_layer) # Attention block(cross-time attention) if i % class="num">2 == class="num">0: # Apply attention to every other layer attn_layer = nn.MultiheadAttention(embed_dim=out_channels, num_heads=class="num">4) self.attention_layers.append(attn_layer) else: self.attention_layers.append(None) in_channels = out_channels # Fully connected head(same as original) self.head = nn.Sequential( nn.AdaptiveAvgPool1d(class="num">1), nn.Flatten(), nn.Linear(in_channels, class="num">128), nn.ReLU(), nn.Dropout(class="num">0.2), nn.Linear(class="num">128, class="num">64), nn.ReLU(), nn.Linear(class="num">64, class="num">1), nn.Sigmoid() ) def _dot_product_kernel(self, x): """Dot product similarity kernel with positional encoding""" # x shape: (B, C, L) x = x.permute(class="num">0, class="num">2, class="num">1) # (B, L, C) # Compute dot product attention attention_scores = torch.bmm(x, x.transpose(class="num">1, class="num">2)) # (B, L, L) attention_weights = F.softmax(attention_scores / (x.size(-class="num">1) ** class="num">0.5), dim=-class="num">1) class="kw">return torch.bmm(attention_weights, x).permute(class="num">0, class="num">2, class="num">1) # (B, C, L) def _design_architecture(self): # Simulate attention response pattern num_layers = class="num">10 kernel_sizes = [class="num">3 + (i % class="num">4) * class="num">2 for i in range(num_layers)] # cyclic class="num">3, class="num">5, class="num">7, class="num">9 channels = [class="num">32 * (i + class="num">1) for i in range(num_layers)] # class="num">32, class="num">64, ..., class="num">320 class="kw">return kernel_sizes, channels def forward(self, x): x = x.unsqueeze(class="num">1) # (B, class="num">1, L) for conv_layer, attn_layer in zip(self.conv_layers, self.attention_layers): x = conv_layer(x) if attn_layer is not None: # Reshape for attention(MultiheadAttention expects seq_len first) attn_input = x.permute(class="num">2, class="num">0, class="num">1) # (L, B, C) attn_output, _ = attn_layer(attn_input, attn_input, attn_input) x = attn_output.permute(class="num">1, class="num">2, class="num">0) # (B, C, L) # Also apply dot product kernel x = self._dot_product_kernel(x) class="kw">return self.head(x)
在 Python 里手写 AO 动量震荡函数
MT5 的 Python 接口不会自带 MQL5 那套指标函数，想用比尔·威廉姆斯的动量震荡指标（AO）只能自己造轮子。上面这段函数用 pandas 做时间序列滚动窗口，NumPy 在底层被隐式调用撑住数值效率，函数签名里 short_period 默认 5、long_period 默认 34，和经典设置一致。 进函数先卡两道防御：必须有 high、low 列，周期必须是正整数，否则直接抛带信息的 ValueError 早失败。外部数据永远别信，漏掉这步后面 NaN 会咬人。 接着 df.copy() 非破坏性复制，保住原始 K 线。中位数价取 (high+low)/2 而非收盘价，避开开收震荡噪音；再用 rolling().mean() 算长短两条 SMA，差值写进新列 AO。AO 是零线驱动——短期 SMA 减长期 SMA 为正偏多、为负偏空，交叉动作比绝对值更关键。 外汇与贵金属波动剧烈，AO 仅作概率倾向参考，实盘前请在 MT5 Python 环境用历史数据回测验证。
def Awesome_Oscillator(df: pd.DataFrame, short_period: class="type">int = class="num">5, long_period: class="type">int = class="num">34) -> pd.DataFrame: """ Calculate the Bill Williams Awesome Oscillator(AO) and append it to the class="kw">input DataFrame. AO = SMA(Median Price, short_period) - SMA(Median Price, long_period) Args: df(pd.DataFrame): DataFrame with &class="macro">#x27;high&class="macro">#x27; and &class="macro">#x27;low&class="macro">#x27; columns. short_period(class="type">int): Short period for SMA(class="kw">default class="num">5). long_period(class="type">int): Long period for SMA(class="kw">default class="num">34). Returns: pd.DataFrame: Input DataFrame with &class="macro">#x27;AO&class="macro">#x27; column added. """ required_cols = {&class="macro">#x27;high&class="macro">#x27;, &class="macro">#x27;low&class="macro">#x27;} if not required_cols.issubset(df.columns): raise ValueError("DataFrame must contain &class="macro">#x27;high&class="macro">#x27; and &class="macro">#x27;low&class="macro">#x27; columns") if not all(p > class="num">0 for p in [short_period, long_period]): raise ValueError("Period values must be positive integers") result_df = df.copy() median_price = (result_df[&class="macro">#x27;high&class="macro">#x27;] + result_df[&class="macro">#x27;low&class="macro">#x27;]) / class="num">2 short_sma = median_price.rolling(window=short_period).mean() long_sma = median_price.rolling(window=long_period).mean() result_df[&class="macro">#x27;AO&class="macro">#x27;] = short_sma - long_sma class="kw">return result_df