🧠

多层感知器和反向传播算法（第二部分）：利用 Python 实现并与 MQL5 集成·进阶篇

（2/3）· 手写感知器只是理解训练过程的台阶，真正落地要靠 Keras+TensorFlow 与 MQL5 的管道打通

偏理论第 2/3 篇

很多交易者以为 MQL5 只能写指标和 EA，其实内置 Python 环境能把深度学习模型直接拉进行情终端。先理解网络怎么算，再谈集成，否则调参只是盲拧旋钮。

本章目录

用 Python 原型跑通三层网络前向传播
用反向传播跑通一个两层分类网络
把 EURUSD 日线拉进 Python 做序列切分
MT5 与 Python 的双向 socket 通路
用日线收益率跑通 LSTM 前的 Keras 基线

「用 Python 原型跑通三层网络前向传播」

这段 Python 原型先把网络结构搭出来：隐藏层每个神经元权重数为输入数加 1（偏置），输出层权重数为隐藏层数加 1。比如 n_inputs=5、n_hidden=8、n_outputs=1，隐藏层就生成 8 个含 6 个随机权的神经元，输出层 1 个含 9 个随机权的神经元。 activate 函数把偏置 weights[-1] 作为初始 activation，再逐项累加 weights[i]*inputs[i]，对应线性组合。transfer 用 1.0/(1.0+exp(-activation)) 做 sigmoid 压缩，输出恒在 0~1 之间，适合做概率类信号。 forward_propagate 逐层把上一层的输出当成下一层输入，new_inputs 收集本层各神经元 transfer 后的 output 并回写 neuron['output']，最终返回最后一层 inputs 作为网络预测。 backward_propagate_error 从输出层倒推：输出层误差直接用 expected[j]-output，隐藏层误差按下游权重乘 delta 求和，再乘 transfer_derivative（output*(1-output)）得到本层 delta。update_weights 用 l_rate * delta * input 修正每条权重，输入层用 row[:-1]，后续层用前层 output。在 MT5 外先用这套脚本验证梯度方向是否合理，再考虑把权重固化进 EA；外汇与贵金属杠杆高，信号仅作概率参考，实盘前务必小仓验证。

MQL5 / C++

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
# Calculate neuron activation for an input
def activate(weights, inputs):
	activation = weights[-class="num">1]
	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
# 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;])
# Update network weights with error
def update_weights(network, row, l_rate):
	for i in range(len(network)):
		inputs = row[:-class="num">1]
		if i != class="num">0:
			inputs = [neuron[&class="macro">#x27;output&class="macro">#x27;] for neuron in network[i - class="num">1]]
		for neuron in network[i]:
			for j in range(len(inputs)):
				neuron[&class="macro">#x27;weights&class="macro">#x27;][j] += l_rate * neuron[&class="macro">#x27;delta&class="macro">#x27;] * inputs[j]

◍ 用反向传播跑通一个两层分类网络

下面这段 Python 风格脚本把前面初始化的网络真正训了起来：固定 20 个 epoch、学习率 0.5，在 10 行二维样本（标签 0 与 1 各 5 行）上做分类。每轮打印累计平方误差，能看到误差随 epoch 下降的倾向，但具体收敛曲线取决于随机种子与初始权重。训练函数按样本逐行前向传播、算期望 one-hot、反向传播误差并更新权重；偏置项 weights[-1] 的更新规则是 l_rate * delta，和隐藏层权重共用同一学习率。跑完网络后，predict() 直接取输出层最大值下标作为类别。用 seed(1) 固定随机源后，同样的 dataset 在 2 隐神经元结构下，前几 epoch 的 error 通常从 2 点几降到 1 以下，说明小网络已能区分这两簇点。外汇与贵金属行情用类似结构做状态分类时，过拟合风险高，建议先在小样本验证再上实盘 tick。

MQL5 / C++

neuron[&class="macro">#x27;weights&class="macro">#x27;][-class="num">1] += l_rate * neuron[&class="macro">#x27;delta&class="macro">#x27;]
def train_network(network, train, l_rate, n_epoch, n_outputs):
	for epoch in range(n_epoch):
		sum_error = class="num">0
		for row in train:
			outputs = forward_propagate(network, row)
			expected = [class="num">0 for i in range(n_outputs)]
			expected[row[-class="num">1]] = class="num">1
			sum_error += sum([(expected[i]-outputs[i])**class="num">2 for i in range(len(expected))])
			backward_propagate_error(network, expected)
			update_weights(network, row, l_rate)
		print(&class="macro">#x27;>epoch=%d, lrate=%.3f, error=%.3f&class="macro">#x27; % (epoch, l_rate, sum_error))
seed(class="num">1)
dataset = [[class="num">2.7810836,class="num">2.550537003,class="num">0],
	[class="num">1.465489372,class="num">2.362125076,class="num">0],
	[class="num">3.396561688,class="num">4.400293529,class="num">0],
	[class="num">1.38807019,class="num">1.850220317,class="num">0],
	[class="num">3.06407232,class="num">3.005305973,class="num">0],
	[class="num">7.627531214,class="num">2.759262235,class="num">1],
	[class="num">5.332441248,class="num">2.088626775,class="num">1],
	[class="num">6.922596716,class="num">1.77106367,class="num">1],
	[class="num">8.675418651,-class="num">0.242068655,class="num">1],
	[class="num">7.673756466,class="num">3.508563011,class="num">1]]
n_inputs = len(dataset[class="num">0]) - class="num">1
n_outputs = len(set([row[-class="num">1] for row in dataset]))
network = initialize_network(n_inputs, class="num">2, n_outputs)
train_network(network, dataset, class="num">0.5, class="num">20, n_outputs)
for layer in network:
	print(layer)
from math class="kw">import exp
def activate(weights, inputs):
	activation = weights[-class="num">1]
	for i in range(len(weights)-class="num">1):
		activation += weights[i] * inputs[i]
	class="kw">return activation
def transfer(activation):
	class="kw">return class="num">1.0 / (class="num">1.0 + exp(-activation))
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
def predict(network, row):
	outputs = forward_propagate(network, row)
	class="kw">return outputs.index(max(outputs))
dataset = [[class="num">2.7810836,class="num">2.550537003,class="num">0],
	[class="num">1.465489372,class="num">2.362125076,class="num">0],

把 EURUSD 日线拉进 Python 做序列切分

上面那段网络权重只是示意，真正要在 MT5 上验证，得先拿到行情。用 mt5.copy_rates_from_pos 抓 EURUSD 的 D1 周期最近 1000 根 K 线，close 价就能直接喂给后面的序列模型。初始化失败时要显式 shutdown，否则终端会残留连接。rates 返回的是结构体数组，转成 DataFrame 后把 time 字段用 to_datetime(unit='s') 展开，再 set_index 成时间索引，画图就是一行 plt.plot(rates.close)。 split_sequence 是单变量序列的滑窗函数：n_steps=3 时，[10,20,30,40,50,60,70,80,90] 会被切成 X=[[10,20,30],[20,30,40]…]、y=[40,50…]。train_test_split 按因子切训练集和测试集，X_train 用 train 跑三步窗，X_test 用 test 跑同样步长。外汇和贵金属杠杆高、点差跳变频繁，D1 回测和实盘偏差可能很大，序列模型只给概率倾向，别当方向指令。

MQL5 / C++

class="kw">import MetaTrader5 as mt5
from pandas class="kw">import to_datetime, DataFrame
class="kw">import matplotlib.pyplot as plt
symbol = "EURUSD"
if not mt5.initialize():
    print("initialize() failed")
    mt5.shutdown()
rates = mt5.copy_rates_from_pos(symbol, mt5.TIMEFRAME_D1, class="num">0, class="num">1000)
mt5.shutdown()
rates = DataFrame(rates)
rates[&class="macro">#x27;time&class="macro">#x27;] = to_datetime(rates[&class="macro">#x27;time&class="macro">#x27;], unit=&class="macro">#x27;s&class="macro">#x27;)
rates = rates.set_index([&class="macro">#x27;time&class="macro">#x27;])
plt.figure(figsize = (class="num">15,class="num">10))
plt.plot(rates.close)
plt.show()

def train_test_split(values, fator):
    train_size = class="type">int(len(values) * fator)
    class="kw">return values[class="num">0:train_size], values[train_size:len(values)]

from numpy class="kw">import array
def split_sequence(sequence, n_steps):
    X, y = list(), list()
    for i in range(len(sequence)):
        end_ix = i + n_steps
        if end_ix > len(sequence)-class="num">1:
            class="kw">break
        seq_x, seq_y = sequence[i:end_ix], sequence[end_ix]
        X.append(seq_x)
        y.append(seq_y)
    class="kw">return array(X), array(y)

raw_seq = [class="num">10, class="num">20, class="num">30, class="num">40, class="num">50, class="num">60, class="num">70, class="num">80, class="num">90]
n_steps = class="num">3
X, y = split_sequence(raw_seq, n_steps)
for i in range(len(X)):
    print(X[i], y[i])

X_train, y_train = split_sequence(train, class="num">3)
X_test, y_test = split_sequence(test, class="num">3)
# define model
model = Sequential()
model.add(Dense(class="num">100, activation=&class="macro">#x27;relu&class="macro">#x27;, input_dim=n_steps))
model.add(Dense(class="num">1))
model.compile(optimizer=&class="macro">#x27;adam&class="macro">#x27;, loss=&class="macro">#x27;mse&class="macro">#x27;)

「MT5 与 Python 的双向 socket 通路」

把 LSTM 预测搬到 MT5 实盘，核心不是模型本身，而是让 EA 能跟 Python 端收发数据。下面这段 MQL5 的 CClientSocket 类用单例模式封装了连接、发送、接收和关闭，EA 侧只要调 Socket() 拿实例就能用。训练侧用 Keras 跑 100 个 epoch、verbose=2 看过程，预测时取 X_test 最后一组 n_steps 窗口 reshape 成 (1, n_steps) 再 model.predict，打印出预测值和 y_test[-1] 真实值做对照。这一步在本地 notebook 跑通后，才谈得上往 MT5 推。 Python 端 socketserver 类绑定 address:port，socket_receive 里 listen(1) 后阻塞等连接，recv(10000) 每次最多收 1 万字节。MT5 的 SocketConnect 超时由 m_time_out 控制，连不上 IsConnected 直接返 false，EA 就不会拿脏数据下单。外汇和贵金属杠杆高、滑点大，模型预测值仅作概率参考，实盘前务必在策略测试器用历史 tick 验证通路稳定性。

MQL5 / C++

model.fit(X_train, y_train, epochs=class="num">100, verbose=class="num">2)
# demonstrate prediction
x_input = X_test[-class="num">1]
x_input = x_input.reshape((class="num">1, n_steps))
yhat = model.predict(x_input, verbose=class="num">0)
print("Valor previsto: ", yhat)
print("Valor real: ", y_test[-class="num">1])
class CClientSocket
  {
class="kw">private:
   class="kw">static CClientSocket*   m_socket;
   class="type">int                     m_handler_socket;
   class="type">int                     m_port;
   class="type">class="kw">string                  m_host;
   class="type">int                     m_time_out;
                         CClientSocket(class="type">void);
                        ~CClientSocket(class="type">void);
class="kw">public:
   class="kw">static class="type">bool            DeleteSocket(class="type">void);
   class="type">bool                   SocketSend(class="type">class="kw">string payload);
   class="type">class="kw">string                 SocketReceive(class="type">void);
   class="type">bool                   IsConnected(class="type">void);
   class="kw">static CClientSocket *Socket(class="type">void);
   class="type">bool                   Config(class="type">class="kw">string host, class="type">int port);
   class="type">bool                   Close(class="type">void);
   };
class="kw">static CClientSocket *CClientSocket::Socket(class="type">void)
  {
   if(CheckPointer(m_socket)==POINTER_INVALID)
      m_socket=new CClientSocket();
   class="kw">return m_socket;
  }
class="type">bool CClientSocket::IsConnected(class="type">void)
  {
   ResetLastError();
   class="type">bool res=true;
   m_handler_socket=SocketCreate();
   if(m_handler_socket==INVALID_HANDLE)
      res=class="kw">false;
   if(!::SocketConnect(m_handler_socket,m_host,m_port,m_time_out))
      res=class="kw">false;
   class="kw">return res;
  }
class="type">bool CClientSocket::Close(class="type">void)
  {
   class="type">bool res=class="kw">false;
   if(SocketClose(m_handler_socket))
     {
      res=true;
      m_handler_socket=INVALID_HANDLE;
     }
   class="kw">return res;
  }
class="kw">import socket
class socketserver(object):
    def __init__(self, address, port):
        self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        self.address = address
        self.port = port
        self.sock.bind((self.address, self.port))
        
    def socket_receive(self):
        self.sock.listen(class="num">1)
        self.conn, self.addr = self.sock.accept()
        self.cummdata = &class="macro">#x27;&class="macro">#x27;
        while True:
            data = self.conn.recv(class="num">10000)

◍ 用日线收益率跑通 LSTM 前的 Keras 基线

这段脚本把 EURUSD 的日线收盘做成收益率序列，再切出训练与验证集喂给一个极简全连接网络。数据区间锁定在 2020-01-01 至 2021-07-01（UTC），用 mt5.copy_rates_range 取 TIMEFRAME_D1，close.pct_change(1) 算日收益，dropna 后剩约 400 个样本点。先按 0.70 把序列拆成 X/y，再对 X 内部按 0.7 分 train/test，n_steps 设为 60，意味着用过去 60 个日收益预测下一期。模型只是 Sequential 里塞一层 200 个 relu 单元加一个线性输出，optimizer 用 adam、loss 用 mse，跑 50 个 epoch。外汇与贵金属属高风险品种，这类基线网络仅用于验证数据管线是否通顺，预测倾向不可直接当作下单依据。开 MT5 把 date_ini/date_end 改成你关心的品种区间，就能复现这条管线。

MQL5 / C++

self.cummdata+=data.decode("utf-class="num">8")
if not data:
    self.conn.close()
    class="kw">break
class="kw">return self.cummdata

def socket_send(self, message):
    self.sock.listen(class="num">1)
    self.conn, self.addr = self.sock.accept()
    self.conn.send(bytes(message, "utf-class="num">8"))

    def __del__(self):
    self.conn.close()
class="kw">import MetaTrader5 as mt5
from numpy.lib.financial class="kw">import rate
from pandas class="kw">import to_datetime, DataFrame
from class="type">class="kw">datetime class="kw">import class="type">class="kw">datetime, timezone
from matplotlib class="kw">import pyplot
from sklearn.metrics class="kw">import mean_squared_error
from math class="kw">import sqrt
class="kw">import numpy as np
from tensorflow.keras class="kw">import Sequential
from tensorflow.keras.layers class="kw">import Dense
from tensorflow.keras.callbacks class="kw">import *
symbol = "EURUSD"
date_ini = class="type">class="kw">datetime(class="num">2020, class="num">1, class="num">1, tzinfo=timezone.utc)
date_end = class="type">class="kw">datetime(class="num">2021, class="num">7, class="num">1, tzinfo=timezone.utc)
period   = mt5.TIMEFRAME_D1
def train_test_split(values, fator):
    train_size = class="type">int(len(values) * fator)
    class="kw">return np.array(values[class="num">0:train_size]), np.array(values[train_size:len(values)])
# split a univariate sequence into samples
def split_sequence(sequence, n_steps):
    X, y = list(), list()
    for i in range(len(sequence)):
        end_ix = i + n_steps
        if end_ix > len(sequence)-class="num">1:
            class="kw">break
        seq_x, seq_y = sequence[i:end_ix], sequence[end_ix]
        X.append(seq_x)
        y.append(seq_y)
    class="kw">return np.array(X), np.array(y)
if not mt5.initialize():
    print("initialize() failed")
    mt5.shutdown()
    raise Exception("Error Getting Data")
rates = mt5.copy_rates_range(symbol, period, date_ini, date_end)
mt5.shutdown()
rates = DataFrame(rates)
if rates.empty:
    raise Exception("Error Getting Data")
rates[&class="macro">#x27;time&class="macro">#x27;] = to_datetime(rates[&class="macro">#x27;time&class="macro">#x27;], unit=&class="macro">#x27;s&class="macro">#x27;)
rates.set_index([&class="macro">#x27;time&class="macro">#x27;], inplace=True)
rates = rates.close.pct_change(class="num">1)
rates = rates.dropna()
X, y = train_test_split(rates, class="num">0.70)
X = X.reshape(X.shape[class="num">0])
y = y.reshape(y.shape[class="num">0])
train, test = train_test_split(X, class="num">0.7)
n_steps = class="num">60
verbose = class="num">1
epochs  = class="num">50
X_train, y_train = split_sequence(train, n_steps)
X_test, y_test   = split_sequence(test, n_steps)
X_val, y_val     = split_sequence(y, n_steps)
# define model
model = Sequential()
model.add(Dense(class="num">200, activation=&class="macro">#x27;relu&class="macro">#x27;, input_dim=n_steps))
model.add(Dense(class="num">1))
model.compile(optimizer=&class="macro">#x27;adam&class="macro">#x27;, loss=&class="macro">#x27;mse&class="macro">#x27;)
history = model.fit(X_train
                    ,y_train
                    ,epochs=epochs
                    ,verbose=verbose
                    ,validation_data=(X_test, y_test))

把模型诊断交给小布盯盘

这些诊断小布盯盘的 AIGC 已内置，打开对应品种页即可看到 Python 模型输出的置信区间和特征偏移，你只需判断信号是否值得跟单。

←

多层感知器和反向传播算法（第二部分）：利用 Python 实现并与 MQL5 集成（基础篇）

→

多层感知器和反向传播算法（第二部分）：利用 Python 实现并与 MQL5 集成·综合运用

常见问题

官方建议 3.3 到 3.8 之间的 Python，实践里 3.7 较稳，安装时勾选 Add to PATH 能省掉后续环境变量配置。

手写网络用来看清前向传播与反向传播的每一步梯度变化；Keras 封装了底层运算，适合快速训练和生产部署。

目前小布盯盘内置的是通用诊断与特征监控，你可将模型推理结果以数据形式接入观察，模型本体仍在你的 Python 侧运行。

本地 socket 或内置环境调用通常在毫秒到几十毫秒，高频场景要实测通道占用，外汇贵金属杠杆高、滑点风险需自担。

多层感知器和反向传播算法（第二部分）：利用 Python 实现并与 MQL5 集成·进阶篇

「用 Python 原型跑通三层网络前向传播」

◍ 用反向传播跑通一个两层分类网络

把 EURUSD 日线拉进 Python 做序列切分

「MT5 与 Python 的双向 socket 通路」

◍ 用日线收益率跑通 LSTM 前的 Keras 基线

常见问题

相关阅读