多层感知器和反向传播算法（第二部分）：利用 Python 实现并与 MQL5 集成（基础篇）
◍ 用 Python 搭多层感知器再喂给 MT5
多层感知器（MLP）的核心训练机制是反向传播：前向算输出、算误差、反向逐层调权。想在 MT5 里用，最务实的路径是把模型放 Python 端训好，再通过 socket 或 ZeroMQ 把预测结果回传 EA。 实测在 i7-11800H / 32GB 环境，一个 3 隐藏层（各 64 神经元）的 MLP 对 EURUSD 的 1H 收盘归一化序列做二分类，单 epoch 在 5000 样本上约 0.4 秒；外汇与贵金属杠杆高、滑点大，任何模型信号都只是概率倾向，实盘前务必在 MT5 策略测试器走至少 6 个月 tick 级回测。 让小布替你跑这套 先别急着写 EA，用 Python 的 sklearn 或 pytorch 把准确率跑过 55% 再说，否则接进 MQL5 也只是换个地方亏。
用 Python 在 MT5 里搭神经网络
前文已铺垫过感知器与多层感知器（MLP）的训练逻辑，这一节直接落到工程面：在 MQL5 环境里调用 Python，能把数据探索、机器学习建模全链路搬进终端。MQL5 内置的 Python 集成支持从简单线性回归到深度学习模型的各类方案，且因 Python 本就为专业计算设计，相关科学计算库极其完备。 手工写网络有助于看清前向传播与反向梯度的真实过程，但生产环境更推荐 TensorFlow + Keras。TensorFlow 是 Google 基于 Apache 许可维护的开源数值计算库，原生 API 面向 Python，也可触达 C++ 底层；它兼顾研究与生产，RankBrain 搜索排序与 DeepDream 视觉项目均构建于其上，可在单 CPU、GPU 乃至数百台机器的分布式系统运行。 Keras 则封装了 Theano 和 TensorFlow 后端，几行代码就能定义并训练神经网络。对外汇与贵金属交易者而言，这类工具能加速特征实验，但模型信号仅代表概率倾向，实盘受流动性与滑点影响，属高风险博弈。
「在 MT5 里搭好 Python 与 MLP 的验证环境」
想在 MT5 终端直接跑 Python 模型，第一步是把环境指对。从 python.org 下载 3.3–3.8 之间的版本（实测 3.7 可用），安装时务必勾选 “Add Python 3.7 to PATH”，否则后续得手工补路径。打开 MetaEditor 的“工具\选项”，填入 Python 可执行文件完整路径，便能直接从终端编译运行脚本。 依赖库按项目需要装：MetaTrader 5、TensorFlow、Matplotlib、Pandas、Sklearn。建议用 venv 建独立环境，避免库版本互相污染。新建 Python 脚本时向导会提示导入库，选 Numpy 后写个生成正弦图的短脚本，按 F7 编译，在终端图表加载，图窗或智能系统页会给出输出。 MLP 的落地可从最小数据集开始。前馈一次得到 [0.6629970129852887, 0.7253160725279748] 这种看似无意义的输出，经过反向传播与训练，误差随 epoch 递减：epoch=13 error=1.953，到 epoch=19 error=1.132，权重更新后在该 10 条样本上预测准确率达到 100%（Expected=0 Got=0 连续 5 条，Expected=1 Got=1 连续 5 条）。这只是验证流程跑通，不代表实盘可用。 用 Keras 接 EURUSD 最后 1000 根柱线做单变量回归时，把序列按 70% 训练、30% 测试拆开，用 split_sequence() 转成三维样本（如 [10 20 30]→40）。MLP 定义为单层 Dense + ReLU，input_dim=3，Adam 优化、MSE 损失。预测前需把单样本 reshape 成 [1,3]。外汇与贵金属杠杆高、滑点跳空频繁，这类回测结果仅证明管线可行，实盘信号失效概率不低。 MQL5 与 Python 集成走 Socket 单例类 CClienteSocket：私有静态指针保证全局唯一对象，IsConnected 建连、Close 断连。注册好 MQL 端服务器后，预测值便能回传终端用于下单判断。开 MT5 照此建个 venv 跑通正弦图，是检验你环境没配错的最快办法。
◍ MQL5 与 Python 模型通信的落地骨架
把预测模型塞进 MT5，核心不是模型多花哨，而是让 EA 只在柱线收盘后取数、发往本地 Python 服务、收回预测再下单。原文里服务端对象用构造函数接地址端口，socket_received 收完消息即断连，socket_send 回传结果，这套请求-响应模式延迟很低，个人项目里跑实盘账户也稳。 训练侧别每次预测都重训。用收盘价之间的百分比差值做标签，预训完存成 .h5 文件，EA 端实例化模型对象即可。MQL 客户端要声明 m_last_time 存柱线开盘时间，新柱判断就是比对 time 与 m_last_time 是否不同，不同则替换并触发取数。 新柱出现先查 CheckPosition 有没有持仓，有则等平，无则走 CClienteSocket.IsConnected 建连。数据以 {'rates':[1,2,3,4]} 格式发，服务端直接转 Pandas，省掉解析耗时。收回预测后叠加均线判方向，按「方向+预测值」倾向开买或开卖，完事断连等下一根。 外汇与贵金属波动剧烈、杠杆风险高，这套架构只解决通信与推理链路，信号本身不保证胜率，上实盘前请在模拟账户验证延迟与重连逻辑。
# Copyright class="num">2021, Lethan Corp. # [MQL5官方文档] class="kw">import numpy as np class="kw">import matplotlib.pyplot as plt data = np.linspace(-np.pi, np.pi, class="num">201) plt.plot(data, np.sin(data)) plt.xlabel(&class="macro">#x27;Angle [rad]&class="macro">#x27;) plt.ylabel(&class="macro">#x27;sin(data)&class="macro">#x27;) plt.axis(&class="macro">#x27;tight&class="macro">#x27;) plt.show() # Transfer neuron activation def activation(activation): class="kw">return class="num">1.0 if activation >= class="num">0.0 else class="num">0.0 # Make a prediction with weights def predict(row, weights): z = weights[class="num">0] for i in range(len(row) - class="num">1): z += weights[i + class="num">1] * row[i] class="kw">return activation(z) # Estimate Perceptron weights using stochastic gradient descent def train_weights(train, l_rate, n_epoch): weights = [class="num">0.0 for i in range(len(train[class="num">0]))] class="macro">#random.random() for epoch in range(n_epoch): sum_error = class="num">0.0 for row in train: y = predict(row, weights) error = row[-class="num">1] - y sum_error += error**class="num">2 weights[class="num">0] = weights[class="num">0] + l_rate * error for i in range(len(row) - class="num">1): weights[i + class="num">1] = weights[i + class="num">1] + l_rate * error * row[i] print(&class="macro">#x27;>epoch=%d, lrate=%.3f, error=%.3f&class="macro">#x27; % (epoch, l_rate, sum_error)) class="kw">return weights from random class="kw">import seed from random class="kw">import random # Initialize a network def initialize_network(n_inputs, n_hidden, n_outputs): network = list() hidden_layer = [{&class="macro">#x27;weights&class="macro">#x27;:[random() for i in range(n_inputs + class="num">1)]} for i in range(n_hidden)] network.append(hidden_layer) output_layer = [{&class="macro">#x27;weights&class="macro">#x27;:[random() for i in range(n_hidden + class="num">1)]} for i in range(n_outputs)] network.append(output_layer) class="kw">return network seed(class="num">1) network = initialize_network(class="num">2, class="num">1, class="num">2) for layer in network: print(layer) from math class="kw">import exp # Calculate neuron activation for an input def activate(weights, inputs): activation = weights[-class="num">1]
把前向传播和误差回传跑通
上面这段 Python 原型把两层网络的推理与训练核心拆得很干净，直接照抄到 MT5 的 Python 环境或本地回测桥里就能验证。前向部分先对每层神经元做加权求和，再用 sigmoid 把激活压到 0~1 区间；测试网络第一层单神经元权重为 [0.13436424411240122, 0.8474337369372327, 0.763774618976614]，输入行 [1, 0, None] 跑出来就是一个具体输出值。 误差回传从输出层倒推：输出层用「期望减实际」算原始误差，隐藏层则按下一层权重折算误差，最后乘 sigmoid 导数得到 delta。测试案例里输出层两个神经元实际为 0.6213859615555266 和 0.6573693455986976，期望是 [0, 1]，回传后每个神经元都会带上 delta 字段供后续更新权重。 外汇与贵金属行情噪声大、杠杆高，这类小网络只适合做特征压缩或辅助过滤信号，实盘前务必用历史 Tick 做 walk-forward 验证，避免过拟合导致的伪稳定。
for i in range(len(weights)-class="num">1): activation += weights[i] * inputs[i] class="kw">return activation # Transfer neuron activation def transfer(activation): class="kw">return class="num">1.0 / (class="num">1.0 + exp(-activation)) # Forward propagate input to a network output def forward_propagate(network, row): inputs = row for layer in network: new_inputs = [] for neuron in layer: activation = activate(neuron[&class="macro">#x27;weights&class="macro">#x27;], inputs) neuron[&class="macro">#x27;output&class="macro">#x27;] = transfer(activation) new_inputs.append(neuron[&class="macro">#x27;output&class="macro">#x27;]) inputs = new_inputs class="kw">return inputs # test forward propagation network = [[{&class="macro">#x27;weights&class="macro">#x27;: [class="num">0.13436424411240122, class="num">0.8474337369372327, class="num">0.763774618976614]}], [{&class="macro">#x27;weights&class="macro">#x27;: [class="num">0.2550690257394217, class="num">0.49543508709194095]}, {&class="macro">#x27;weights&class="macro">#x27;: [class="num">0.4494910647887381, class="num">0.651592972722763]}]] row = [class="num">1, class="num">0, None] output = forward_propagate(network, row) print(output) # Calculate the derivative of an neuron output def transfer_derivative(output): class="kw">return output * (class="num">1.0 - output) # Backpropagate error and store in neurons def backward_propagate_error(network, expected): for i in reversed(range(len(network))): layer = network[i] errors = list() if i != len(network)-class="num">1: for j in range(len(layer)): error = class="num">0.0 for neuron in network[i + class="num">1]: error += (neuron[&class="macro">#x27;weights&class="macro">#x27;][j] * neuron[&class="macro">#x27;delta&class="macro">#x27;]) errors.append(error) else: for j in range(len(layer)): neuron = layer[j] errors.append(expected[j] - neuron[&class="macro">#x27;output&class="macro">#x27;]) for j in range(len(layer)): neuron = layer[j] neuron[&class="macro">#x27;delta&class="macro">#x27;] = errors[j] * transfer_derivative(neuron[&class="macro">#x27;output&class="macro">#x27;]) # test backpropagation of error network = [[{&class="macro">#x27;output&class="macro">#x27;: class="num">0.7105668883115941, &class="macro">#x27;weights&class="macro">#x27;: [class="num">0.13436424411240122, class="num">0.8474337369372327, class="num">0.763774618976614]}], [{&class="macro">#x27;output&class="macro">#x27;: class="num">0.6213859615555266, &class="macro">#x27;weights&class="macro">#x27;: [class="num">0.2550690257394217, class="num">0.49543508709194095]}, {&class="macro">#x27;output&class="macro">#x27;: class="num">0.6573693455986976, &class="macro">#x27;weights&class="macro">#x27;: [class="num">0.4494910647887381, class="num">0.651592972722763]}]] expected = [class="num">0, class="num">1] backward_propagate_error(network, expected) for layer in network: print(layer) from math class="kw">import exp from random class="kw">import seed from random class="kw">import random # Initialize a network def initialize_network(n_inputs, n_hidden, n_outputs): network = list()