数据科学和机器学习(第 27 部分):MetaTrader 5 中训练卷积神经网络(CNN)交易机器人 值得吗?·进阶篇
(2/3)·从卷积层到Python建模,拆解CNN在外汇贵金属行情识别中的真实收益与坑点
不少交易者把卷积神经网络当成行情预测的万能钥匙,直接套用图像识别里的池化与卷积参数,结果在MT5回测里反复过拟合。金融时间序列不是静态图片,网格状特征的空间层次若生搬硬套,信号衰减比想象中更快。本篇接上篇基础概念,继续深挖CNN组件与实盘建模的断层。
「全连接层」
全连接层中的神经元、与前一层中的所有激活都有连接。这些层典型情况下偏向用在网络末尾,以便基于卷积层和池化层提取的特征执行分类或回归。 偏平层 将一维池化特征映射转换为一维向量,如此就能将其投喂到完全连接(密集)层之中。 密集层( Dense )是完全连接的层,用于基于卷积层和池化层提取的特征制定最终决策。 密集层本质上是传统人工神经网络(ANN)的核心组件。 [CODE] from tensorflow.keras.layers import Conv1D, MaxPooling1D, Flatten, Dense, Dropout model = Sequential() model.add(Conv1D(filters=64, kernel_size=3, activation='relu', input_shape=(window_size, X_train.shape[2]))) model.add(MaxPooling1D(pool_size=2)) model.add(Flatten()) <span style="background-color:rgb(232, 183, 215);">model.add(Dense(100, activation='relu')) model.add(Dropout(0.5)) model.add(Dense(units=len(np.unique(y)), activation='sigmoid')) # For binary classification (e.g., buy/sell signal)</span> model.summary() [/CODE]
◍ 用 Dropout 切断过拟合通路
舍弃层(Dropout)本质是一张随机掩码:训练时按概率把部分神经元输出强制置零,使这些神经元不参与当前批次的前向与反向传播,其余神经元照常计算。放在输入向量上,等于随机删掉若干特征;插在隐藏层之间,则是随机废掉一些隐藏节点,逼网络不依赖特定通路。
在 CNN 类行情模型里,没有 Dropout 时先到的样本会过度主导权重,只在后面批次才出现的形态(比如某类贵金属跳空后的回归)可能根本学不到,实盘泛化就会塌。
上面这段 Keras 代码里 Dropout(0.5) 直接挂在全连接层后,意味着每次更新有半数节点被随机屏蔽。对 MT5 导出的窗口序列做二分类(买/卖)时,这个 0.5 是高频调试点:调高更抗过拟合但收敛慢,调低训练快却易记噪声。外汇与贵金属杠杆高,模型过拟合会放大回撤,调参前先用小样本跑通再说。
from tensorflow.keras.layers class="kw">import Conv1D, MaxPooling1D, Flatten, Dense, Dropout model = Sequential() model.add(Conv1D(filters=class="num">64, kernel_size=class="num">3, activation=&class="macro">#x27;relu&class="macro">#x27;, input_shape=(window_size, X_train.shape[class="num">2]))) model.add(MaxPooling1D(pool_size=class="num">2)) model.add(Flatten()) model.add(Dense(class="num">100, activation=&class="macro">#x27;relu&class="macro">#x27;)) model.add(Dropout(class="num">0.5)) model.add(Dense(units=len(np.unique(y)), activation=&class="macro">#x27;sigmoid&class="macro">#x27;)) # For binary classification(e.g., buy/sell signal) model.summary()
CNN 进金融盘面的七个硬理由
把 CNN 拿来做外汇或贵金属的表格型时序分析,乍看违背它的图像出身——它本是为像素卷积设计的。但和 FFNN、RNN、LSTM、GRU 比,它在财经数据上反而有几处实打实的优势。 CNN 的局部卷积核能自动抠出趋势、季节轮动这类局部时态形态,传统统计常漏掉特征间的交互。多层堆叠还能学层次特征:浅层抓简单波形,深层把波形拼成抽象结构,正好吃下盘面里的复杂耦合。 金融序列满是高噪和冗余,池化层做降采样,把微小波动的影响压下去,模型对杂讯更稳。面对多标的联立(比如金价与美元指数、成交量),CNN 直接捕跨序列依赖, multivariate 预测不掉链。 权重共享加局部连接,让它在高维特征下比全连接网络省算量,MT5 里跑高维因子不至于卡死。端到端训练免手工特征工程,原生 OHLCV 喂进去就能出信号。 卷积运算本身会强化突变量:一段骤升骤降、或特定 K 线组合,核响应直接拉高。外汇/贵金属波动剧烈、黑天鹅频出,这种突发检测在高杠杆高风险环境下有实用价值,但信号仅作概率参考,不构成方向保证。 理由列完,下一步就是真建一个 CNN 训练,再看怎么塞进 MT5 的 EA 里跑。
「用 CNN 在日线窗口里捞形态信号」
CNN 对高维数据里的局部形态敏感,做外汇或贵金属日线判别时,不必堆复杂结构,挑 OPEN/HIGH/LOW/CLOSE 几个基础量就够它抓特征。我们按前一篇的时间序列思路,把 TARGET_OPEN 与 TARGET_CLOSE 比大小做成涨跌标签,只留 4 个自变量加 1 个因变量,造出轻量数据集 new_df。 交易数据天然有序,形态常跨多个时间步而非单点。用 10 根日线做滑动窗口,输入形状压成 (样本数, 10, 4),CNN 才能吃到时间上下文。下面这段代码就是造标签和滑窗预处理的骨架。 [CODE 见 code 字段] 逐行看:先取目标开盘收盘,循环里收大于开则标 1(偏多),否则 0(偏空);new_df 只装四个行情字段加标签。preprocess_data 按 window_size=10 切重叠窗,X 是 990 个三维样本,y 是对应标签。 切分后训练集 792、测试集 198,标准化只作用在特征维;标签再独热成 (n,2)。一维卷积层放 64 个核、步长 2、causal 填充,接池化、Dropout 0.5 防过拟合。训练 EarlyStopping 在第 34 轮停,样本外准确率约 57%——对外汇贵金属这种噪声市,仅略高于随机,实盘前务必在 MT5 用历史数据复跑验证,杠杆品种高风险。 模型够给 EA 用之前,先存成 .onnx 并把 scaler 参数落二进制,后面接 MQL5 推理才不脱节。
open_price = df[&class="macro">#x27;TARGET_OPEN&class="macro">#x27;] close_price = df[&class="macro">#x27;TARGET_CLOSE&class="macro">#x27;] # making the target variable target_var = [] for i in range(len(open_price)): if close_price[i] > open_price[i]: # if the price closed above where it opened target_var.append(class="num">1) # bullish signal else: target_var.append(class="num">0) # bearish signal new_df = pd.DataFrame({ &class="macro">#x27;OPEN&class="macro">#x27;: df[&class="macro">#x27;OPEN&class="macro">#x27;], &class="macro">#x27;HIGH&class="macro">#x27;: df[&class="macro">#x27;HIGH&class="macro">#x27;], &class="macro">#x27;LOW&class="macro">#x27;: df[&class="macro">#x27;LOW&class="macro">#x27;], &class="macro">#x27;CLOSE&class="macro">#x27;: df[&class="macro">#x27;CLOSE&class="macro">#x27;], &class="macro">#x27;TARGET_VAR&class="macro">#x27;: target_var }) print(new_df.shape) # Example data preprocessing function def preprocess_data(df, window_size): X, y = [], [] for i in range(len(df) - window_size): X.append(df.iloc[i:i+window_size, :-class="num">1].values) y.append(df.iloc[i+window_size, -class="num">1]) class="kw">return np.array(X), np.array(y) window_size = class="num">10 X, y = preprocess_data(new_df, window_size) print(f"x_shape = {X.shape}\ny_shape = {y.shape}") x_shape = (class="num">990, class="num">10, class="num">4) y_shape = (class="num">990,) # Split data into training and testing sets X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=class="num">0.2, shuffle=False) # Standardize the data scaler = StandardScaler() X_train = scaler.fit_transform(X_train.reshape(-class="num">1, X_train.shape[-class="num">1])).reshape(X_train.shape) X_test = scaler.transform(X_test.reshape(-class="num">1, X_test.shape[-class="num">1])).reshape(X_test.shape) print(f"x_train\n{X_train.shape}\nx_test\n{X_test.shape}\n\ny_train {y_train.shape} y_test {y_test.shape}") x_train(class="num">792, class="num">10, class="num">4) x_test(class="num">198, class="num">10, class="num">4) y_train(class="num">792,) y_test(class="num">198,) from tensorflow.keras.utils class="kw">import to_categorical y_train_encoded = to_categorical(y_train) y_test_encoded = to_categorical(y_test) print(f"One hot encoded\n\ny_train {y_train_encoded.shape}\ny_test {y_test_encoded.shape}") One hot encoded y_train(class="num">792, class="num">2) y_test(class="num">198, class="num">2) # Defining the CNN model model = Sequential() model.add(Conv1D(filters=class="num">64, kernel_size=class="num">3, activation=&class="macro">#x27;relu&class="macro">#x27;, strides=class="num">2, padding=&class="macro">#x27;causal&class="macro">#x27;, input_shape=(window_size, X_train.shape[class="num">2]) ) ) model.add(MaxPooling1D(pool_size=class="num">2)) model.add(Flatten()) model.add(Dense(class="num">100, activation=&class="macro">#x27;relu&class="macro">#x27;)) model.add(Dropout(class="num">0.5)) model.add(Dense(units=len(np.unique(y)), activation=&class="macro">#x27;softmax&class="macro">#x27;)) # For binary classification(buy/sell signal) model.summary() # Compiling the model optimizer = Adam(learning_rate=class="num">0.001) model.compile(optimizer=optimizer, loss=&class="macro">#x27;categorical_crossentropy&class="macro">#x27;, metrics=[&class="macro">#x27;accuracy&class="macro">#x27;]) # Training the model early_stopping = EarlyStopping(monitor=&class="macro">#x27;val_loss&class="macro">#x27;, patience=class="num">10, restore_best_weights=True) history = model.fit(X_train, y_train_encoded, epochs=class="num">100, batch_size=class="num">16, validation_split=class="num">0.2, callbacks=[early_stopping]) plt.figure(figsize=(class="num">7.5, class="num">6))
◍ CNN 分类器的训练曲线与混淆矩阵落地
把 Keras 里这段 CNN 二分类脚本直接跑起来,先看训练与验证损失怎么分叉。下面这段 matplotlib 代码会把每个 epoch 的 loss 画成两条线,存成 png 再弹窗,肉眼能立刻判断模型有没有过拟合倾向。 [CODE]plt.plot(history.history['loss'], label='Training Loss') plt.plot(history.history['val_loss'], label='Validation Loss') plt.xlabel('Epochs') plt.ylabel('Loss') plt.title('Training Loss Curve') plt.legend() plt.savefig("training loss cuver-cnn-clf.png") plt.show()[/CODE] 上面逐行拆解:第1行取训练集损失序列并打标签;第2行取验证集损失;第3、4行设横纵轴名;第5行写图标题;第6行显示图例;第7行存盘(注意文件名拼错成 cuver);第8行弹窗渲染。 模型结构 sequential_2 里 Conv1D 输出 (None,5,64) 占 832 参数,Flatten 后接 12900 参数的 Dense(100),最后 Dense(2) 做二分类。总参数量不大,但回测片段显示第 32–34 epoch 验证准确率在 0.4843–0.5346 间晃,val_loss 卡在 0.69 附近,基本贴近随机猜。 [CODE]y_pred = model.predict(X_test) classes_in_y = np.unique(y) y_pred_binary = classes_in_y[np.argmax(y_pred, axis=1)] cm = confusion_matrix(y_test, y_pred_binary) sns.heatmap(cm, annot=True, fmt='d', cmap='Blues') plt.xlabel("Predicted Label") plt.ylabel("True Label") plt.title("Confusion Matrix") plt.savefig("confusion-matrix CNN") print("Classification Report\n", classification_report(y_test, y_pred_binary))[/CODE] 这段先对测试集推理,再用 np.unique 映射回类别标签,confusion_matrix 配 seaborn 热力图直接看错分格子。分类报告里类别 0 的 precision 0.53、recall 0.24、f1 0.33、support 88——召回极低说明大量真实 0 被漏判,外汇或贵金属信号若按此模型跟单,错漏概率偏高,属高风险用法。 开盘前拿自己的 X_test 复跑这份脚本,把混淆矩阵截图和 val_loss 曲线对照,若 val_loss 不降反平,就别把该 CNN 当日内过滤层。
plt.plot(history.history[&class="macro">#x27;loss&class="macro">#x27;], label=&class="macro">#x27;Training Loss&class="macro">#x27;) plt.plot(history.history[&class="macro">#x27;val_loss&class="macro">#x27;], label=&class="macro">#x27;Validation Loss&class="macro">#x27;) plt.xlabel(&class="macro">#x27;Epochs&class="macro">#x27;) plt.ylabel(&class="macro">#x27;Loss&class="macro">#x27;) plt.title(&class="macro">#x27;Training Loss Curve&class="macro">#x27;) plt.legend() plt.savefig("training loss cuver-cnn-clf.png") plt.show() y_pred = model.predict(X_test) classes_in_y = np.unique(y) y_pred_binary = classes_in_y[np.argmax(y_pred, axis=class="num">1)] cm = confusion_matrix(y_test, y_pred_binary) sns.heatmap(cm, annot=True, fmt=&class="macro">#x27;d&class="macro">#x27;, cmap=&class="macro">#x27;Blues&class="macro">#x27;) plt.xlabel("Predicted Label") plt.ylabel("True Label") plt.title("Confusion Matrix") plt.savefig("confusion-matrix CNN") print("Classification Report\n", classification_report(y_test, y_pred_binary))