📘

使用 MetaTrader 5 的 Python 高频套利交易系统·进阶篇

第 2/3 篇

本章目录

拉 MT5 分时报价先把终端连起来
批量拉取多品种报价的轻量封装
2000 个合成价格计算：策略与实施
合成价怎么挤出行情盲区
把价差落盘成 CSV 再慢慢筛
用 open_test_limit_order 把信号送进 MT5

拉 MT5 分时报价先把终端连起来

做价格行为分析第一步是拿到真实分时流，而不是拿历史 K 线凑合。MetaTrader 5 终端必须显式初始化，且全程用 UTC 避免经纪商时区把你的时间戳搞乱——外汇和贵金属市场跨session报价一旦时区错配，后续所有微观结构统计都会偏。下面这段是取最近 1 天 tick 的最小可用骨架：先试连终端，连不上直接报路径并返回 None；再用 copy_ticks_from 拉 COPY_TICKS_ALL 全类型报价，转成 Pandas DataFrame 并把 time 列从秒戳还原成 datetime。

MQL5 / C++

if not mt5.initialize(path=terminal_path):
    print(f"Failed to connect to MetaTrader class="num">5 terminal at {terminal_path}")
    class="kw">return None
timezone = pytz.timezone("Etc/UTC")
utc_from = class="type">class="kw">datetime.now(timezone) - timedelta(days=class="num">1)
ticks = mt5.copy_ticks_from(symbol, utc_from, count, mt5.COPY_TICKS_ALL)
ticks_frame = pd.DataFrame(ticks)
ticks_frame[&class="macro">#x27;time&class="macro">#x27;] = pd.to_datetime(ticks_frame[&class="macro">#x27;time&class="macro">#x27;], unit=&class="macro">#x27;s&class="macro">#x27;)
if ticks is None:
    print(f"Failed to fetch data for {symbol}")
    class="kw">return None

逐行拆解：第1行试初始化指定路径的终端；第2–3行失败就打印终端路径并退出函数；第4行锁定 UTC 时区；第5行算「现在减 1 天」作为拉取起点；第6行按 symbol 和 count 拉全部 tick 类型；第7行塞进 DataFrame；第8行把整型秒时间转成可索引的 datetime；第9–11行若返回为空则报错返 None，调用方据此跳过该品种。实盘里把 terminal_path 换成你本机 MT5 的 terminal64.exe 路径，symbol 填 "XAUUSD" 这类，跑通后 ticks_frame 里就有 bid/ask/volume 可供算价差分布。外汇与贵金属杠杆高，tick 数据仅用于策略验证，不代表任何收益可能。

MQL5 / C++

if not mt5.initialize(path=terminal_path):
    print(f"Failed to connect to MetaTrader class="num">5 terminal at {terminal_path}")
    class="kw">return None
timezone = pytz.timezone("Etc/UTC")
utc_from = class="type">class="kw">datetime.now(timezone) - timedelta(days=class="num">1)
ticks = mt5.copy_ticks_from(symbol, utc_from, count, mt5.COPY_TICKS_ALL)
ticks_frame = pd.DataFrame(ticks)
ticks_frame[&class="macro">#x27;time&class="macro">#x27;] = pd.to_datetime(ticks_frame[&class="macro">#x27;time&class="macro">#x27;], unit=&class="macro">#x27;s&class="macro">#x27;)
if ticks is None:
    print(f"Failed to fetch data for {symbol}")
    class="kw">return None

「批量拉取多品种报价的轻量封装」

做跨品种相关性或套利扫描时，最忌在策略主循环里反复请求 MT5。把品种清单和取数逻辑收进一个函数，主程序只管拿字典，结构会干净很多。下面这段实现就干了这件事：先硬编码 25 个常见直盘与交叉盘（AUDUSD、EURJPY、GBPJPY 等），每品种取 1000 根 K 线的数据量。循环里对每一个 symbol 调 get_mt5_data，只有返回非 None 才落库，且只截 time / bid / ask 三列并以 time 做索引。返回的是一个以品种名为键的字典。实测中若某个品种在终端未开通行情，df 为 None 会被直接跳过，不会中断整个拉取——这点对同时扫几十个交叉盘的外汇、贵金属交易者很实用，但需注意部分平台交叉盘点差跳变大，高频取 bid/ask 仍属高风险操作。

MQL5 / C++

def get_currency_data():
    # Define currency pairs and the amount of data
    symbols = ["AUDUSD", "AUDJPY", "CADJPY", "AUDCHF", "AUDNZD", "USDCAD", "USDCHF", "USDJPY", "NZDUSD", "GBPUSD", "EURUSD", "CADCHF", "CHFJPY", "NZDCAD", "NZDCHF", "NZDJPY", "GBPCAD", "GBPCHF", "GBPJPY", "GBPNZD", "EURCAD", "EURCHF", "EURGBP", "EURJPY", "EURNZD"]
    count = class="num">1000  # number of data points for each currency pair
    data = {}
    for symbol in symbols:
        df = get_mt5_data(symbol, count, terminal_path)
        if df is not None:
            data[symbol] = df[[&class="macro">#x27;time&class="macro">#x27;, &class="macro">#x27;bid&class="macro">#x27;, &class="macro">#x27;ask&class="macro">#x27;]].set_index(&class="macro">#x27;time&class="macro">#x27;)
    class="kw">return data

◍ 2000 个合成价格计算：策略与实施

我们深入研究我们系统的基础 —— calculate_synthetic_prices 函数。它使我们能够获得合成数据。我们来看看代码： [CODE] def calculate_synthetic_prices(data):     synthetic_prices = {}     # Remove duplicate indices from all DataFrames in the data dictionary     for key in data:         data[key] = remove_duplicate_indices(data[key])     # Calculate synthetic prices for all pairs using multiple methods     pairs = [('AUDUSD', 'USDCHF'), ('AUDUSD', 'NZDUSD'), ('AUDUSD', 'USDJPY'),              ('USDCHF', 'USDCAD'), ('USDCHF', 'NZDCHF'), ('USDCHF', 'CHFJPY'),              ('USDJPY', 'USDCAD'), ('USDJPY', 'NZDJPY'), ('USDJPY', <s

合成价怎么挤出行情盲区

先把空字典 synthesize_prices 建好，后续所有合成报价都往里塞，同时顺手遍历原始 tick 数据把重复索引清掉，不然后面算叉盘容易因索引冲突直接报错。真正干活的是 pairs 列表，里面列好要做合成的基础货币对。对每一组配对跑两遍：用第一对的 bid 除以第二对的 ask，再用第一对的 ask 除以第二对的 bid，每算一次 method_count 加一。按这种双向除法铺开，全量配对循环下来实际能生成约 2000 个合成货币对。calculate_synthetic_prices 不只是算价，它把原本没有直接报价的路径硬造出来，外汇与贵金属市场流动性割裂处可能由此暴露出套利窗口，但这类机会转瞬即逝且点差滑点会吞噬利润，属高风险博弈。

「把价差落盘成 CSV 再慢慢筛」

analyze_arbitrage 这套逻辑的核心，是先建一个空的 spreads 字典，再拿真实货币对的 bid 去减合成价格序列，把每个差值塞进去。差值本身代表真实报价和交叉合成价之间的偏离，正数才意味着可能存在三角套利空间。代码里用 0.00008 作为门槛，等价于 8 个点。凡小于这个数的偏离直接被 DataFrame 的布尔比较过滤掉，剩下的是偏离更大、统计上获利概率可能更高的候选。这一步不复杂，但把噪声压下去了。落盘动作只有一行：arbitrage_opportunities.to_csv('arbitrage_opportunities.csv')。跑完你能在终端同目录拿到这张全表，拿 Excel 或 Python 接着画分布、看持续时间都行。外汇与贵金属价差套利属高风险，历史偏离不代表未来必然重现。别把 8 点门槛当圣旨阈值写死在脚本里，遇到点差扩大时段会漏掉大量机会，遇到流动性稀薄时段又可能塞满假信号。开 MT5 把同时段真实 tick 拉出来，对比 0.00008 和实际最小变动单位，再决定调不调参。

MQL5 / C++

def analyze_arbitrage(data, synthetic_prices, method_count):
    # Calculate spreads for each pair
    spreads = {}
    for pair in data.keys():
        for i in range(class="num">1, method_count + class="num">1):
            synthetic_pair = f&class="macro">#x27;{pair}_{i}&class="macro">#x27;
            if synthetic_pair in synthetic_prices.columns:
                print(f"Analyzing arbitrage opportunity for {synthetic_pair}")
                spreads[synthetic_pair] = data[pair][&class="macro">#x27;bid&class="macro">#x27;] - synthetic_prices[synthetic_pair]
    # Identify arbitrage opportunities
    arbitrage_opportunities = pd.DataFrame(spreads) > class="num">0.00008
    print("Arbitrage opportunities:")
    print(arbitrage_opportunities)
    # Save the full table of arbitrage opportunities to a CSV file
    arbitrage_opportunities.to_csv(&class="macro">#x27;arbitrage_opportunities.csv&class="macro">#x27;)
    class="kw">return arbitrage_opportunities
spreads[synthetic_pair] = data[pair][&class="macro">#x27;bid&class="macro">#x27;] - synthetic_prices[synthetic_pair]
arbitrage_opportunities = pd.DataFrame(spreads) > class="num">0.00008
arbitrage_opportunities.to_csv(&class="macro">#x27;arbitrage_opportunities.csv&class="macro">#x27;)

◍ 用 open_test_limit_order 把信号送进 MT5

下单函数先干两件事：连终端、验品种。mt5.initialize 用指定 terminal_path 拉起客户端，symbol_info 查不到就直接退出，避免在非法品种上挂单。仓位闸门靠 positions_total 与 MAX_OPEN_TRADES 比对实现。只要当前持仓数大于等于该上限，函数打印 MAX POSITIONS TOTAL 并 return None，从代码逻辑看这能防止同一终端累积过多敞口。外汇与贵金属杠杆高，这类硬性上限最好按账户净值反推，而非拍脑袋定。请求体里 deviation 写死 30 点、magic=123456、comment 标 'Stochastic Stupi Sustem'，tp/sl 用 symbol_info.point 按买卖方向加减。挂单成功回传 result.order，失败则打印 retcode。下面逐行拆关键段： if not mt5.initialize(path=terminal_path): 连终端失败就报错退出 if symbol_info is None: 品种不存在直接返回 if positions_total >= MAX_OPEN_TRADES: 超仓位数拒绝开单 request 中 'deviation': 30 表示允许 30 点成交偏差 'tp'/'sl' 用 point 乘系数，买单调上方止盈、下方止损，卖单反向 result = mt5.order_send(request) 真正发单，回码 DONE 才算成交把这套函数丢进 MT5 的 Python 环境跑一遍，改一下 magic 和 comment，看自己终端能不能正常回传 order id。

MQL5 / C++

def open_test_limit_order(symbol, order_type, price, volume, take_profit, stop_loss, terminal_path):
    if not mt5.initialize(path=terminal_path):
        print(f"Failed to connect to MetaTrader class="num">5 terminal at {terminal_path}")
        class="kw">return None
    symbol_info = mt5.symbol_info(symbol)
    positions_total = mt5.positions_total()
    if symbol_info is None:
        print(f"Instrument not found: {symbol}")
        class="kw">return None
    if positions_total >= MAX_OPEN_TRADES:
        print("MAX POSITIONS TOTAL!")
        class="kw">return None
    # Check if symbol_info is None before accessing its attributes
    if symbol_info is not None:
        request = {
            "action": mt5.TRADE_ACTION_DEAL,
            "symbol": symbol,
            "volume": volume,
            "type": order_type,
            "price": price,
            "deviation": class="num">30,
            "magic": class="num">123456,
            "comment": "Stochastic Stupi Sustem",
            "type_time": mt5.ORDER_TIME_GTC,
            "type_filling": mt5.ORDER_FILLING_IOC,
            "tp": price + take_profit * symbol_info.point if order_type == mt5.ORDER_TYPE_BUY else price - take_profit * symbol_info.point,
            "sl": price - stop_loss * symbol_info.point if order_type == mt5.ORDER_TYPE_BUY else price + stop_loss * symbol_info.point,
        }
        result = mt5.order_send(request)
        if result is not None and result.retcode == mt5.TRADE_RETCODE_DONE:
            print(f"Test limit order placed for {symbol}")
            class="kw">return result.order
        else:
            print(f"Error: Test limit order not placed for {symbol}, retcode={result.retcode if result is not None else &class="macro">#x27;None&class="macro">#x27;}")
            class="kw">return None
    else:
        print(f"Error: Symbol info not found for {symbol}")
        class="kw">return None
if positions_total >= MAX_OPEN_TRADES:
    print("MAX POSITIONS TOTAL!")
    class="kw">return None