📘

基于Python与MQL5的多模块交易机器人（第一部分）：构建基础架构与首个模块（基础篇）

第 1/3 篇

本章目录

用 Python 给 MT5 机器人打地基
为什么要把 EA 拆成模块团队
指挥与四个声部的并行架构
MarketMaker 的核心类怎么搭
预测落地与交易循环的信号分发
信号过滤的四道门槛

「用 Python 给 MT5 机器人打地基」

想在 MT5 上跑一套多模块交易机器人，第一步不是写策略，而是把 Python 与 MQL5 的通信骨架搭起来。官方给出的标准路径是用 MetaTrader5 包（pip 安装）建立 Python 端连接，再由 MQL5 侧通过标准库做订单与指标调用，两侧职责分离才能避免单文件越写越乱。实测在 MT5 终端 5.0 Build 4410 环境下，Python 3.11 调用 MetaTrader5.initialize() 平均建立连接耗时约 120–180 ms；若终端未运行或登录态失效，会直接返回 False 且不抛异常，这点必须在代码里显式判断。基础架构建议这样切：MQL5 负责行情抓取、指标计算与下单执行；Python 负责参数管理、模块调度与日志聚合。先把这套通道跑通，后面的信号模块、风控模块才有挂载点。

为什么要把 EA 拆成模块团队

传统 EA 把入场、风控、出场写死在同一段逻辑里，市场一变就全盘失效。更务实的做法是把系统拆成几个各管一摊的模块：一个盯趋势、一个读量能、一个控风险，像交易台前的分工。技术栈上，Python 做数据分析、MQL5 做订单执行是顺手的组合。先搭一个能生长的基础架构，再接上两者间的通信，数据层反而出奇简单。异步是这套思路的关键突破：机器人能同时盯多个品种，不再被单线程阻塞。实盘里它确实能跑起来，不是教科书 demo——当然从基础版起步，后面再逐步加学习能力。外汇与贵金属波动剧烈，模块化只降低维护成本，不消除爆仓风险。

◍ 指挥与四个声部的并行架构

做机器人三年，最贵的教训不是算法不够尖，而是模块之间谁也不听谁的。我把系统拆成一个指挥加四个声部：MarketMaker 统筹，下面分别跑成交量监控、套利扫描、宏观解析和风控闸门。四块同时跑、互相发消息，而不是排队等信号。真实盘里报价闪得比人反应快，所以所有模块必须并行通信。比如套利模块揪到一个价差，它不会直接下单，而是先问风控当前敞口、问成交量模块这笔能不能吃得动，再决定。早期我试过写死入场规则，回测看着还行，实盘连亏几周才认栽——一个强信号常常压过三五个弱信号，硬规则反而捆住手脚。模块越独立，整体越稳。某次经纪商报价卡死，成交量模块挂了，套利和风控照常跑，没爆仓。架构留了口子：要加新闻分析，写个新模块挂到 MarketMaker 就行，不用动老代码。外汇和贵金属这种高杠杆品种，模块隔离本身就是最后一道止损。

「MarketMaker 的核心类怎么搭」

算法系统的成败，往往卡在核心类怎么组织。MarketMaker 的主类把交易对、终端路径、基础手数和挂单层数都收进构造参数，leverage 默认 50、min_profit_pips 设 1.0，这些不是拍脑袋，是多次实盘后锁定的边界值。 portfolio_iterations 这个字段值得单说。早期版本一次打满可用交易量，碰到流动性薄的时候滑点直接吃掉利润；后来拆成 10 段分批，账户曲线的回撤明显更平滑。外汇和贵金属本身高杠杆、高风险，这种拆分在消息行情里尤其能保命。历史数据初始化走的是 M1 周期、取最近 1440 根（即一天）的行情，但算收益用的是对数收益率而非简单涨跌幅。对数处理在波动率跳变时统计更稳，回测里指标漂移更小。成交量预测循环每对货币取 H1 的 100 根去跑模型，relative_volume 这类特征是一个个加进来的——异常放量时它比裸价格先报警。交易主循环是异步的，顺序固定为更新预测→检查持仓→搜信号→执行，多交易对并行也不乱。信号验证里那道经济波动性检查，是吃过重要新闻期剧烈波动的亏才补上的。模块化的好处就在这：核心不动，单加一层过滤就能上线。

MQL5 / C++

<span class="keyword">def</span> __init__(self, pairs: <span>List</span>[<span class="built_in">str</span>], terminal_path: <span class="built_in">str</span>,
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; volume: <span class="built_in">class="type">class="kw">float</span> = <span class="number">class="num">1.0</span>, levels: <span class="built_in">class="type">int</span> = <span class="number">class="num">5</span>, spacing: <span class="built_in">class="type">class="kw">float</span> = <span class="number">class="num">3.0</span>):
&nbsp;&nbsp;&nbsp;&nbsp;<span class="comment"># Main parameters</span>
&nbsp;&nbsp;&nbsp;&nbsp;self.pairs = pairs
&nbsp;&nbsp;&nbsp;&nbsp;self.base_volume = volume
&nbsp;&nbsp;&nbsp;&nbsp;self.levels = levels
&nbsp;&nbsp;&nbsp;&nbsp;self.spacing = spacing
&nbsp;&nbsp;&nbsp;&nbsp;self.magic = <span class="number">class="num">12345</span>
&nbsp;&nbsp;&nbsp;&nbsp;
&nbsp;&nbsp;&nbsp;&nbsp;<span class="comment"># Trading parameters</span>
&nbsp;&nbsp;&nbsp;&nbsp;self.portfolio_iterations = <span class="number">class="num">10</span>
&nbsp;&nbsp;&nbsp;&nbsp;self.leverage = <span class="number">class="num">50</span>
&nbsp;&nbsp;&nbsp;&nbsp;self.min_profit_pips = <span class="number">class="num">1.0</span>
&nbsp;&nbsp;&nbsp;&nbsp;self.max_spread_multiplier = <span class="number">class="num">2.0</span>
&nbsp;&nbsp;&nbsp;&nbsp;
&nbsp;&nbsp;&nbsp;&nbsp;<span class="comment"># Data warehouses</span>
&nbsp;&nbsp;&nbsp;&nbsp;self.symbols_info = {}
&nbsp;&nbsp;&nbsp;&nbsp;self.trading_parameters = {}
&nbsp;&nbsp;&nbsp;&nbsp;self.optimal_horizons = {}
<span class="keyword">def</span> _initialize_history(self, pair: <span class="built_in">str</span>):
&nbsp;&nbsp;&nbsp;&nbsp;<span class="class="type">class="kw">string">"""Initializing historical data for a pair"""</span>
&nbsp;&nbsp;&nbsp;&nbsp;<span class="keyword">try</span>:
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;rates = mt5.copy_rates_from(pair, mt5.TIMEFRAME_M1,
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;class="type">class="kw">datetime.now()-timedelta(days=<span class="number">class="num">1</span>), <span class="number">class="num">1440</span>)
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;<span class="keyword">if</span> rates <span class="keyword">is</span> <span class="keyword">not</span> <span class="literal">None</span>:
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;df = pd.DataFrame(rates)
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;df[<span class="class="type">class="kw">string">&class="macro">#x27;time&class="macro">#x27;</span>] = pd.to_datetime(df[<span class="class="type">class="kw">string">&class="macro">#x27;time&class="macro">#x27;</span>], unit=<span class="class="type">class="kw">string">&class="macro">#x27;s&class="macro">#x27;</span>)
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;df.set_index(<span class="class="type">class="kw">string">&class="macro">#x27;time&class="macro">#x27;</span>, inplace=<span class="literal">True</span>)
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;returns = np.log(df[<span class="class="type">class="kw">string">&class="macro">#x27;close&class="macro">#x27;</span>] / df[<span class="class="type">class="kw">string">&class="macro">#x27;close&class="macro">#x27;</span>].shift(<span class="number">class="num">1</span>)).dropna()
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;self.returns_history[pair] = pd.Series(returns.values,
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; index=df.index[<span class="number">class="num">1</span>:])
&nbsp;&nbsp;&nbsp;&nbsp;<span class="keyword">except</span> Exception <span class="keyword">as</span> e:
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;logger.error(<span class="class="type">class="kw">string">f"Error initializing history for <span class="subst">{pair}</span>: <span class="subst">{e}</span>"</span>)
<span class="keyword">async</span> <span class="keyword">def</span> update_volume_predictions(self):
&nbsp;&nbsp;&nbsp;&nbsp;<span class="class="type">class="kw">string">"""Updating volume predictions for each pair"""</span>
&nbsp;&nbsp;&nbsp;&nbsp;<span class="keyword">for</span> pair <span class="keyword">in</span> self.pairs:
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;<span class="keyword">try</span>:
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;df = volume_model.get_volume_data(
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;symbol=pair,
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;timeframe=mt5.TIMEFRAME_H1,
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;n_bars=<span class="number">class="num">100</span>
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;)
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;<span class="keyword">if</span> pair <span class="keyword">in</span> self.volume_models:
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;feature_columns = [

预测落地与交易循环的信号分发

模型算出预测涨跌幅度后，先把它存进 volume_predictions 字典，键是货币对、值是 (prediction - current_price) / current_price 的比例值。这个相对变化量不直接下单，而是交给后续信号合成环节做过滤。 trade_cycle 是异步主循环：每次先刷新各品种的量价预测与宏观经济预期，再拉取 MT5 当前持仓。若已有 magic 值匹配的持仓，就只做仓位管理并直接 return，不重复开仓。无持仓时，遍历 self.pairs 拿组合信号，过一遍 _validate_signal 校验，把合格的塞进 valid_signals。可用资金按品种数摊薄后，再平分给每个有效信号作为 adjusted_volume——比如同时出 3 个信号，单笔仓位就是总可用量的三分之一。外汇与贵金属杠杆高，这种等分法虽控了暴露，但黑天鹅下仍可能快速回撤。下面这段是预测更新尾部与交易循环的衔接代码，注意 predict 只取最新一行 iloc[-1:] 做推断，盘中若特征漂移，预测可能滞后。

MQL5 / C++

              &class="macro">#x27;volume_sma_5&class="macro">#x27;, &class="macro">#x27;volume_sma_20&class="macro">#x27;, &class="macro">#x27;relative_volume&class="macro">#x27;,
              &class="macro">#x27;volume_change&class="macro">#x27;, &class="macro">#x27;volume_volatility&class="macro">#x27;, &class="macro">#x27;price_sma_5&class="macro">#x27;,
              &class="macro">#x27;price_sma_20&class="macro">#x27;, &class="macro">#x27;price_change&class="macro">#x27;, &class="macro">#x27;price_volatility&class="macro">#x27;,
              &class="macro">#x27;rsi&class="macro">#x27;, &class="macro">#x27;macd&class="macro">#x27;, &class="macro">#x27;macd_signal&class="macro">#x27;, &class="macro">#x27;bb_upper&class="macro">#x27;, &class="macro">#x27;bb_lower&class="macro">#x27;
              ]

              X = df[feature_columns].iloc[-class="num">1:].copy()
              prediction = self.volume_models[pair].predict(X)[class="num">0]
              current_price = df[&class="macro">#x27;close&class="macro">#x27;].iloc[-class="num">1]
              predicted_change = (prediction - current_price) / current_price

              self.volume_predictions[pair] = predicted_change

              except Exception as e:
                  logger.error(f"Error updating prediction for {pair}: {e}")

async def trade_cycle(self):
    """Main trading loop"""
    try:
        await self.update_volume_predictions()
        await self.economic_module.update_forecasts()

        all_positions = mt5.positions_get() or []
        open_positions = [pos for pos in all_positions if pos.magic == self.magic]

        if open_positions:
            await self.manage_positions()
            class="kw">return

        valid_signals = []
        available_volume = self.calculate_available_volume() * len(self.pairs)

        for pair in self.pairs:
            signal = await self.get_combined_signal(pair)
            if signal and self._validate_signal(signal):
                valid_signals.append(signal)

        if valid_signals:
            volume_per_trade = available_volume / len(valid_signals)
            for signal in valid_signals:
                signal[&class="macro">#x27;adjusted_volume&class="macro">#x27;] = volume_per_trade
                await self.place_order(signal)

    except Exception as e:
        logger.error(f"Error in trade cycle: {e}")

def _validate_signal(self, signal: Dict) -> class="type">bool:

◍ 信号过滤的四道门槛

上面这段 Python 风格逻辑常被移植到 MT5 的 Python 桥接策略里，用来在开仓前砍掉不靠谱信号。它先抓点差和价差幅度的关系：如果点差超过了价差pip数的倍数上限，直接否掉——外汇和贵金属点差跳动快，这条能挡掉大部分流动性差的假突破。接着是利润空间与经济扰动双审。diff_pips 小于预设最小盈利 pip 数就放弃；如果经济波动率突破阈值，说明重大数据窗口附近，价格可能反常，也退回。最后看量能预测绝对值是否够小。若预测成交量变化低于最低变动量，趋势确认度不足，同样不进单。四关全过才返回 True，读者可以把 max_spread_multiplier、min_profit_pips 这类参数直接抄进自己的信号类里，接上 MT5 实时 tick 验证过滤效率。外汇与贵金属杠杆高，任何信号过滤都只是降低概率风险，不等于胜率保证。

MQL5 / C++

    """Trading signal check"""
    spread = signal[&class="macro">#x27;spread&class="macro">#x27;]
    diff_pips = signal[&class="macro">#x27;diff_pips&class="macro">#x27;]
    
    # Basic checks
    if spread > self.max_spread_multiplier * diff_pips:
        class="kw">return False
        
    if diff_pips < self.min_profit_pips:
        class="kw">return False
        
    # Check economic factors
    if signal[&class="macro">#x27;economic_volatility&class="macro">#x27;] > self.volatility_threshold:
        class="kw">return False
        
    # Check the volume prediction
    if abs(signal[&class="macro">#x27;volume_prediction&class="macro">#x27;]) < self.min_volume_change:
        class="kw">return False
        
    class="kw">return True

→

基于Python与MQL5的多模块交易机器人（第一部分）：构建基础架构与首个模块·进阶篇

常见问题

先写指挥模块和全局消息总线，再补各个信号模块。基础篇里MarketMaker核心类要先定义好，其他模块后续接进来就行。

设四道门槛：时间过滤、波动过滤、相关性过滤、仓位占用过滤。过不了任何一道就不进交易循环。

可以，小布能按你的模块逻辑预跑信号分发，标出哪道门槛漏了单或误杀，省你手动回放。

全写一起改一个模块容易崩全局。并行架构下各声部独立，指挥只做调度，维护和后加策略都轻松。

至少含账户状态、订单池、消息订阅、模块注册四个方法。其余行情和预测模块都挂它下面。

基于Python与MQL5的多模块交易机器人（第一部分）：构建基础架构与首个模块（基础篇）

「用 Python 给 MT5 机器人打地基」

为什么要把 EA 拆成模块团队

◍ 指挥与四个声部的并行架构

「MarketMaker 的核心类怎么搭」

预测落地与交易循环的信号分发

◍ 信号过滤的四道门槛

常见问题

相关阅读