📘

数据科学与机器学习（第22部分）：利用自编码器神经网络实现更智能的交易从噪声中提炼信号·进阶篇

第 2/3 篇

本章目录

先跑通一个编码器看噪声滤得干不干净
矩阵切片里的 OHLC 数据堆叠
用潜在空间给 PCA 降维加控制阀
自编码器特征经PCA压到二维后的形态
把自编码器落盘成 MT5 能读的 ONNX
用自动编码器给K线降噪

「先跑通一个编码器看噪声滤得干不干净」

编码器堆了 [256, 128, 64] 三层，解码器反向 [64, 128, 256]，中间潜空间压到 32 个神经元。这种结构在 EURUSD 小时级收盘价（4 维特征）上过拟合风险偏高，只是拿来演示，实盘建议从更浅的层数和更窄的潜层起步。切分用了 70% 训练、30% 验证，random_state=42 固定随机种子，MinMaxScaler 把输入压到 [0,1]。Adam 学习率设成 1e-5 很保守，EarlyStopping 盯 val_loss、patience=5，50 个 epoch 内若连续 5 轮没改善就停。实测训练集 7000 行、测试集 3000 行，首轮 loss 0.0669、val_loss 0.0636，到第 50 轮 val_loss 降到 8.7e-05。把原始归一化数据丢进模型再反归一化，肉眼比对收盘价序列，新数据明显削掉了一些毛刺，异常值更容易被拎出来。外汇和贵金属波动受杠杆与消息面驱动，这类重构滤波只降低噪声、不预示方向，信号失效时亏损可能快速放大，务必用小仓位验证。

MQL5 / C++

class="kw">import sklearn
from sklearn.model_selection class="kw">import train_test_split
from keras class="kw">import optimizers
from keras.callbacks class="kw">import EarlyStopping
x_train, x_test = train_test_split(dataset, test_size=class="num">0.3, random_state=class="num">42) class="macro">#train test the data
# Normalizing the class="kw">input data
scaler = sklearn.preprocessing.MinMaxScaler()
x_train = scaler.fit_transform(x_train)
x_test = scaler.transform(x_test)
print(f"x_train {x_train.shape}.dtype({x_train.dtype}) x_test {x_test.shape}.dtype({x_test.dtype})")
# compile the autoencoder
input_dim = dataset.shape[class="num">1]
latent_dim = class="num">32  # Dimension of latent space
hidden_dims = [class="num">256, class="num">128, class="num">64]
autoencoder = Autoencoder(input_dim, latent_dim, hidden_dims)
optimizer = optimizers.Adam(learning_rate=class="num">1e-5)
autoencoder.compile(optimizer=optimizer, loss=losses.MeanSquaredError())
early_stopping = EarlyStopping(monitor=&class="macro">#x27;val_loss&class="macro">#x27;, patience = class="num">5, restore_best_weights=True) class=class="str">"cmt">//stop the training process if class="num">5 epochs have no change in loss
history = autoencoder.fit(x_train, x_train, epochs=class="num">50, shuffle=True, callbacks=[early_stopping], validation_data=(x_test, x_test), batch_size=class="num">64, verbose=class="num">1)
x_train(class="num">7000, class="num">4).dtype(float64) x_test(class="num">3000, class="num">4).dtype(float64)
Epoch class="num">1/class="num">50
class="num">110/class="num">110 ━━━━━━━━━━━━━━━━━━━━ 3s 5ms/step - loss: class="num">0.0669 - val_loss: class="num">0.0636
Epoch class="num">2/class="num">50
class="num">110/class="num">110 ━━━━━━━━━━━━━━━━━━━━ 0s 4ms/step - loss: class="num">0.0648 - val_loss: class="num">0.0608
Epoch class="num">3/class="num">50
class="num">110/class="num">110 ━━━━━━━━━━━━━━━━━━━━ 0s 4ms/step - loss: class="num">0.0624 - val_loss: class="num">0.0550
....
....
....
Epoch class="num">46/class="num">50
class="num">110/class="num">110 ━━━━━━━━━━━━━━━━━━━━ 0s 4ms/step - loss: class="num">1.2096e-04 - val_loss: class="num">1.0195e-04
Epoch class="num">47/class="num">50
class="num">110/class="num">110 ━━━━━━━━━━━━━━━━━━━━ 0s 4ms/step - loss: class="num">1.0758e-04 - val_loss: class="num">9.7759e-05
Epoch class="num">48/class="num">50
class="num">110/class="num">110 ━━━━━━━━━━━━━━━━━━━━ 0s 4ms/step - loss: class="num">1.0923e-04 - val_loss: class="num">9.4798e-05
Epoch class="num">49/class="num">50
class="num">110/class="num">110 ━━━━━━━━━━━━━━━━━━━━ 1s 5ms/step - loss: class="num">1.0243e-04 - val_loss: class="num">9.0442e-05
Epoch class="num">50/class="num">50
class="num">110/class="num">110 ━━━━━━━━━━━━━━━━━━━━ 1s 4ms/step - loss: class="num">1.0222e-04 - val_loss: class="num">8.7384e-05
original_norm_data = scaler.transform(dataset)
new_data = autoencoder.call(original_norm_data)
new_data = scaler.inverse_transform(new_data) class="macro">#class="kw">return data to the original form
print("original data
" ,dataset,"\nnew data\n" ,new_data)
原始数据
[[class="num">1.06507 class="num">1.06633 class="num">1.06497 class="num">1.06538]
[class="num">1.06628 class="num">1.06685 class="num">1.06463 class="num">1.06508]
[class="num">1.06771 class="num">1.06797 class="num">1.06599 class="num">1.06627]
...
[class="num">0.99941 class="num">0.99996 class="num">0.9991  class="num">0.99916]

矩阵切片里的 OHLC 数据堆叠

上面这段原始输出把两组行情压成了二维数组，每组四列对应 O、H、L、C。前一组是 0.99 附近的报价，例如首行 [0.99687, 0.99999, 0.99646, 0.99941]，高点 0.99999 距低点仅 5.3 点，波动极窄。随后 new data 标记之后，数组跳到 1.06 上方，如 [1.06612682, 1.06676685, 1.06537819, 1.06605109]，单根振幅约 13.8 点；尾部又落回 0.99 区间，末行 [0.99581326, 0.99789913, 0.99494114, 0.99651365] 振幅约 19.6 点。这种跨品种、跨时段的拼接，在 MT5 里常见于多符号批量回采。直接把这段贴进脚本打印，能确认你的数据接口是否按 O H L C 顺序吐数；外汇与贵金属杠杆高，窄幅数组也可能是流动性真空段，验证时务必核对服务器时间。

MQL5 / C++

[[class="num">0.99687 class="num">0.99999 class="num">0.99646 class="num">0.99941]
[class="num">0.99536 class="num">0.99724 class="num">0.99444 class="num">0.99687]]
new data
[[class="num">1.06612682 class="num">1.06676685 class="num">1.06537819 class="num">1.06605109]
[class="num">1.06617137 class="num">1.06679912 class="num">1.06541834 class="num">1.06609218]
[class="num">1.06742607 class="num">1.06804771 class="num">1.06668032 class="num">1.06736937]
...
[class="num">0.99906356 class="num">1.00121275 class="num">0.9980908  class="num">0.99980352]
[class="num">0.998204   class="num">1.00034005 class="num">0.9972261  class="num">0.99893805]
[class="num">0.99581326 class="num">0.99789913 class="num">0.99494114 class="num">0.99651365]]

◍ 用潜在空间给 PCA 降维加控制阀

自动编码器把高维输入压进低维潜在空间后，只取编码器部分就能做特征压缩。相比直接对原始数据跑 PCA，先过一遍编码器再降维，能顺手滤掉一部分噪声，对后续机器学习任务的计算效率有帮助。关键在类的构建顺序：实例化 Autoencoder 之后必须先调 build()，再 compile 和训练，否则层形状未知会直接报错。训练无误后，用 encoder_model.predict() 拿到编码矩阵，其第二维就是潜在层组件数（本例 32）。把 encoded_data.shape[1] 喂给 PCA 组件数，画累积解释方差碎石图。实测 10000 条样本、32 维潜在输出下，约 2 个组件就解释了 0.9989 的总方差，拐点明显；单组件轴向上各点差异也不大，说明大幅压缩几乎不丢信息。下次调 PCA 直接用 n_components=2。把 32 个潜在组件画在同一轴上，仅 1 个特征明显游离、其余几乎重合，进一步印证少数组件够用。但注意：若要把降维数据送预测模型，只对自变量做 PCA 更合理，因变量别跟着压。在拿这套编码器去清 MetaTrader 5 行情噪声前，得先存成 ONNX——这是跨到 MT5 调用的必经一步。

MQL5 / C++

class Autoencoder(Model):
  def __init__(self, input_dim, latent_dim, hidden_dims=[]):
    super(Autoencoder, self).__init__()
    self.hidden_dims = hidden_dims
    self.input_dim = input_dim
    
    # Encoder
    self.encoder = tf.keras.Sequential(name=&class="macro">#x27;encoder&class="macro">#x27;) class="macro">#give the encoder Sequential layer name=encoder
    # Decoder( mirrored structure )
    self.decoder = tf.keras.Sequential(name=&class="macro">#x27;decoder&class="macro">#x27;) class="macro">#give the decoder Sequential layer name=decoder
    
  def build(self):
    # Add hidden layers to the encoder(if any)
    for dim in hidden_dims:
      self.encoder.add(layers.Dense(dim, activation=&class="macro">#x27;relu&class="macro">#x27;))
      self.encoder.add(layers.Dropout(class="num">0.5))
    # Define the latent layer
    self.encoder.add(layers.Dense(latent_dim, activation=&class="macro">#x27;relu&class="macro">#x27;))
      
    # Add hidden layers to the decoder(in reverse order)
    for dim in hidden_dims[::-class="num">1]:
      self.decoder.add(layers.Dense(dim, activation=&class="macro">#x27;relu&class="macro">#x27;))
      self.decoder.add(layers.Dropout(class="num">0.5))
    # Define the output layer
    self.decoder.add(layers.Dense(self.input_dim, activation=&class="macro">#x27;sigmoid&class="macro">#x27;))  class="macro">#the output layer with dimensions matching the original class="kw">input data
  def call(self, x):
    encoded = self.encoder(x)
    decoded = self.decoder(encoded)
    class="kw">return decoded
# Instantiate the autoencoder and build the model
autoencoder = Autoencoder(input_dim, latent_dim, hidden_dims)
autoencoder.build()
optimizer = optimizers.Adam(learning_rate=class="num">1e-5)
autoencoder.compile(optimizer=optimizer, loss=losses.MeanSquaredError())
# Extract Encoder
encoder_input = autoencoder.encoder.layers[class="num">0].class="kw">input
encoder_output = autoencoder.encoder.get_layer(index=-class="num">1).output # the layer at index -class="num">1 is the last layer
# Define the encoder model
encoder_model = tf.keras.Model(inputs=encoder_input, outputs=encoder_output)
# Extract Decoder
decoder_input = autoencoder.decoder.layers[class="num">0].class="kw">input
decoder_output = autoencoder.decoder.get_layer(index=-class="num">1).output # the layer at index -class="num">1 is the last layer
# Define the decoder model
decoder_model = tf.keras.Model(inputs=decoder_input, outputs=decoder_output)
from sklearn.decomposition class="kw">import PCA
# Fit & transform the encoded data
encoded_data = encoder_model.predict(original_norm_data)
print("decoded data.shape: ",encoded_data.shape)
# Create PCA object
pca = PCA(n_components=encoded_data.shape[class="num">1])
reduced_data = pca.fit_transform(encoded_data)
print("pca reduced data.shape: ",reduced_data.shape)
print("explained var:\n",np.cumsum(pca.explained_variance_ratio_))
# Plotting the scree plot
plt.figure(figsize=(class="num">10, class="num">6))
plt.plot(np.cumsum(pca.explained_variance_ratio_))
plt.xlabel(&class="macro">#x27;Number of Components&class="macro">#x27;)
plt.ylabel(&class="macro">#x27;Cumulative Explained Variance&class="macro">#x27;)
plt.title(&class="macro">#x27;Scree Plot&class="macro">#x27;)
plt.grid(True)
plt.show()

「自编码器特征经PCA压到二维后的形态」

把上一节训出的 10000 条编码向量丢进 PCA，设 n_components=2，得到 reduced_data.shape = (10000, 2)。这一步不是为预测，而是肉眼看高维特征是不是真被压出了结构。从打印的降维数据头部能看出，第一主成分的值从 0.996 一路爬到 0.9999998，越往后越贴着 1；而第二主成分在前 25 个样本里全是 1.0，没有任何分散。换句话说，二维平面上多数样本会挤在 (≈1, 1) 附近。下面这段代码就是做降维和画图的完整过程，直接在 python 环境里跑就能复现上面的形状打印和特征折线图。外汇与贵金属行情做这类特征压缩时波动结构可能完全不同，高杠杆下误读特征有爆仓风险。

MQL5 / C++

# Create PCA object
pca = PCA(n_components=class="num">2)
reduced_data = pca.fit_transform(encoded_data)
print("pca reduced data.shape: ",reduced_data.shape)
pca降维后的数据形状为：(class="num">10000, class="num">2)
bar = [count+class="num">1 for count in range(reduced_data.shape[class="num">0])]
plt.figure(figsize = (class="num">7,class="num">10))
for col in range(reduced_data.shape[class="num">1]):
    plt.plot(bar,  reduced_data[:, col],label=f&class="macro">#x27;feature {col}&class="macro">#x27;)

plt.xlabel("index")
plt.ylabel("feature")
plt.title("PCA encoded features")
plt.legend()
plt.savefig("pca-encoded features")

把自编码器落盘成 MT5 能读的 ONNX

编码器和解码器之前已经拆开，转 ONNX 只是顺手的事。关键点在于给 Keras 模型显式声明 input_signature，明确输入张量名和 float16 类型，否则 MT5 加载时容易撞上 TensorFlow 与 ONNX 版本错配的坑。保存时 opset 选 13，编码器输出文件如 encoder.eurusd.h1.onnx，解码器同理。缩放器必须 Python 训练端与 MQL5 推理端完全一致，把 MinMax 的 data_min_ 和 data_max_ 写成二进制，塞进 MQL5\Files 才能两边对得上。别以为缩放器差不多就行同一组 minmax_min.bin / minmax_max.bin 若在训练侧和指标侧各算各的，重构误差会直接失真，异常检测信号基本废掉。这是 EURUSD H1 自编码器里最容易被忽略、却最致命的一环。下面这段把 Python 导出与 MQL5 内嵌资源写在一起，开 MT5 把三个文件丢进 Files 目录即可编译验证。

MQL5 / C++

class="kw">import tf2onnx
class="kw">import onnx
class="kw">import os
output_path = os.path.<span class="keyword">join</span>(<span class="class="type">class="kw">string">&class="macro">#x27;/kaggle/working/&class="macro">#x27;</span>,<span class="class="type">class="kw">string">"encoder.eurusd.h1.onnx"</span>)
# saving the encoder <span class="keyword">for</span> MetaTrader <span class="number">class="num">5</span>
<b>input_signature</b> = [tf.TensorSpec(encoder_input.shape, tf.float16, name=<span class="class="type">class="kw">string">&class="macro">#x27;x_inputs&class="macro">#x27;</span>)] class="macro">#onnx class="kw">input signature
# Use from_function <span class="keyword">for</span> tf functions
onnx_model, _ = tf2onnx.convert.from_keras(encoder_model, input_signature, opset=<span class="number">class="num">13</span>)
onnx.save(onnx_model, output_path)
# saving the decoder
output_path = os.path.join(&class="macro">#x27;/kaggle/working/&class="macro">#x27;,"decoder.eurusd.h1.onnx")
input_signature = [tf.TensorSpec(decoder_input.shape, tf.float16, name=&class="macro">#x27;decoder_inputs&class="macro">#x27;)] class="macro">#onnx class="kw">input signature
onnx_model, _ = tf2onnx.convert.from_keras(decoder_model, input_signature, opset=class="num">13) class="macro">#conver keras model to onnx
onnx.save(onnx_model, output_path)
scaler.data_min_.tofile(<span class="class="type">class="kw">string">"minmax_min.bin"</span>)
scaler.data_max_.tofile(<span class="class="type">class="kw">string">"minmax_max.bin"</span>)
<span class="comment">class=class="str">"cmt">//Load both the encoder_model and the decoder_model</span>
<span class="preprocessor">class="macro">#resource </span><span class="class="type">class="kw">string">"\Files\encoder.eurusd.h1.onnx"</span> <span class="keyword">as</span> <span class="keyword">class="type">uchar</span> encoder_onnx[];
<span class="preprocessor">class="macro">#resource </span><span class="class="type">class="kw">string">"\Files\decoder.eurusd.h1.onnx"</span> <span class="keyword">as</span> <span class="keyword">class="type">uchar</span> decoder_onnx[];
<span class="comment">class=class="str">"cmt">// Load the MinMax scaler also</span>
<span class="preprocessor">class="macro">#resource </span><span class="class="type">class="kw">string">"\Files\minmax_min.bin"</span> <span class="keyword">as</span> <span class="keyword">class="type">class="kw">double</span> min_values[];
<span class="preprocessor">class="macro">#resource </span><span class="class="type">class="kw">string">"\Files\minmax_max.bin"</span> <span class="keyword">as</span> <span class="keyword">class="type">class="kw">double</span> max_values[];

◍ 用自动编码器给K线降噪

自动编码器在图像去噪上已被验证有效，但金融时序里还没人坐实。把原始收盘价和编码器重建的收盘价叠图看，重建后的曲线明显更平滑，毛刺少了一截。要在 MT5 里跑这套，得先包一个 CAutoEncoderONNX 类，把 ONNX 模型当 Python 里那样调。编码器和解码器各实例化一个对象，原始数据先送编码器压成隐向量，再把隐向量喂给解码器出重建 OHLC。实际加载指标后有个现象：重建 K 线的实体大小几乎一致，高低点价差偏大且整齐划一；颜色上绝大多数被标成红色空头，灰色多头只占极少数。外汇和贵金属波动受事件驱动，这类重建数据仅作降噪参考，实盘信号仍需结合价格行为判断，杠杆品种高风险。指标给了两个开关：show_bars 控制是否画重建 K 线，show_bullish_bearish 决定是否按真实开收价区分多空着色。还可以隐藏原生 K 线，只留自编码那套，图表干净不少。

MQL5 / C++

class="macro">#class="kw">property indicator_chart_window
class="macro">#class="kw">property indicator_plots class="num">1
class="macro">#class="kw">property indicator_buffers class="num">5
class="kw">input class="type">bool show_bars = true;
class="kw">input class="type">bool show_bullish_bearish = false;
class=class="str">"cmt">//--- plot Candle
class="macro">#class="kw">property indicator_label1  "autoencoded open; high; low; close"
class="macro">#class="kw">property indicator_type1   DRAW_COLOR_CANDLES
class="macro">#class="kw">property indicator_color1  clrRed, clrGray
class="macro">#class="kw">property indicator_style1  STYLE_SOLID
class="macro">#class="kw">property indicator_width1  class="num">1
class CAutoEncoderONNX
  {
class="kw">protected:
   class="type">bool initialized;
   class="type">long onnx_handle;
   class="type">void PrintTypeInfo(const class="type">long num,const class="type">class="kw">string layer,const OnnxTypeInfo& type_info);
   class="type">long inputs[], outputs[];

   class="type">void replace(class="type">long &arr[]) { for (class="type">uint i=class="num">0; i<arr.Size(); i++) if (arr[i] <= -class="num">1) arr[i] = UNDEFINED_REPLACE; }

class="kw">public:
                     CAutoEncoderONNX(class="type">void);
                    ~CAutoEncoderONNX(class="type">void);

                     class="type">bool Init(const class="type">uchar &onnx_buff[], class="type">ulong flags=ONNX_DEFAULT); class=class="str">"cmt">//load the onnx model from a resource class="type">uchar array
                     class="type">bool Init(class="type">class="kw">string onnx_filename, class="type">uint flags=ONNX_DEFAULT); class=class="str">"cmt">//load the onnx model from a .onnx file 

                     matrix predict(const matrix &x); class=class="str">"cmt">//passing inputs for either the encoder or the decoder to the outputs in matrix form
                     vector predict(const vector &x); class=class="str">"cmt">//passing inputs for either the encoder or the decoder to the outputs in matrix form
  };
class="macro">#include <Autoencoder-onnx.mqh>
class="macro">#include <MALE5\preprocessing.mqh>
CAutoEncoderONNX encoder_model; class=class="str">"cmt">//for the encoder model
CAutoEncoderONNX decoder_model; class=class="str">"cmt">//for the decoder model
MinMaxScaler *scaler; class=class="str">"cmt">//Python-like MinMax scaler
class=class="str">"cmt">//+------------------------------------------------------------------+
class=class="str">"cmt">//| Custom indicator initialization function                         |