MetaTrader5 Python 库数据获取与交易接口详解 | 极客日志

Python算法

MetaTrader5 Python 库数据获取与交易接口详解

综述由AI生成如何使用 MetaTrader5 Python 库连接 MT5 终端并获取金融数据。内容包括环境安装、初始化连接、账户信息查询、交易品种规格获取、实时报价（Ticks）和历史 K 线数据（Bars）的获取方法。此外还涵盖了订单、持仓及历史成交记录的查询，以及错误处理和最佳实践。通过 Pandas 转换数据便于分析，适合量化交易初学者参考。

极光发布于 2026/3/30更新于 2026/5/2530 浏览

MetaTrader5 Python 库数据获取与交易接口详解

MetaTrader5 Python 库是 MetaQuotes 公司官方提供的接口，允许 Python 程序与 MetaTrader 5 交易终端进行交互。本教程将深入探讨如何使用该库连接到 MT5 终端并获取各种类型的金融数据。

1. 简介与安装

1.1 什么是 MetaTrader5 Python 库？

这是一个允许 Python 脚本直接与本地运行的 MetaTrader 5 交易终端进行通信的库。你可以用它来获取市场数据、账户信息、执行交易操作等。

1.2 环境准备与安装

Python: 确保你安装了 Python（建议 3.6 或更高版本）。
MetaTrader 5 终端: 你必须在你的电脑上安装 MetaTrader 5 交易终端。

安装库: 使用 pip 安装：

pip install MetaTrader5 --upgrade

1.3 MT5 终端设置

为了让 Python 库能够连接并执行操作，你需要在 MT5 终端进行设置：

打开 MT5 终端。
点击菜单栏'工具 (Tools)' -> '选项 (Options)'。
切换到'EA 交易 (Expert Advisors)'选项卡。
勾选'允许 Algo Trading' (Allow automated trading / Allow DLL imports - 不同版本名称可能略有差异)。
点击'确定'。
确保 MT5 终端已经登录到你的交易账户并且正在运行。

2. 连接与初始化

2.1 导入库

import MetaTrader5 as mt5
from datetime import datetime
import pandas as pd
import time

# 设置 Pandas 显示选项（可选）
pd.set_option('display.max_columns', 500)
pd.set_option('display.width', 1500)

2.2 初始化连接 (`initialize`)

这是连接到 MT5 终端的第一步。

# 尝试初始化连接到 MetaTrader 5 终端
# 如果 MT5 安装在非标准路径，可能需要指定 path 参数
# 如果需要登录特定账户，可以传入 login, password, server 参数，但通常 MT5 已登录即可
if not mt5.initialize():
    print(, mt5.last_error())
    quit()

()
()

相关免费在线工具

加密/解密文本
使用加密算法（如AES、TripleDES、Rabbit或RC4）加密和解密文本明文。在线工具，加密/解密文本在线工具，online
Gemini 图片去水印
基于开源反向 Alpha 混合算法去除 Gemini/Nano Banana 图片水印，支持批量处理与下载。在线工具，Gemini 图片去水印在线工具，online
curl 转代码
解析常见 curl 参数并生成 fetch、axios、PHP curl 或 Python requests 示例代码。在线工具，curl 转代码在线工具，online
Base64 字符串编码/解码
将字符串编码和解码为其 Base64 格式表示形式即可。在线工具，Base64 字符串编码/解码在线工具，online
Base64 文件转换器
将字符串、文件或图像转换为其 Base64 表示形式。在线工具，Base64 文件转换器在线工具，online
Markdown转HTML
将 Markdown（GFM）转为 HTML 片段，浏览器内 marked 解析；与 HTML转Markdown 互为补充。在线工具，Markdown转HTML在线工具，online

# 在脚本结束时或不再需要连接时调用
# mt5.shutdown()
# print("MetaTrader 5 connection closed.")

# 获取账户信息
account_info = mt5.account_info()
if account_info is not None:
    print("Account info retrieved successfully:")
    # account_info 是一个 named tuple (或类似结构)，可以通过 . 访问属性
    print(f" Login: {account_info.login}")
    print(f" Name: {account_info.name}")
    print(f" Server: {account_info.server}")
    print(f" Currency: {account_info.currency}")
    print(f" Balance: {account_info.balance}")
    print(f" Equity: {account_info.equity}")
    print(f" Profit: {account_info.profit}")
    print(f" Margin: {account_info.margin}")
    print(f" Margin Free: {account_info.margin_free}")
    print(f" Margin Level: {account_info.margin_level}%")
else:
    print("Failed to get account info, error code =", mt5.last_error())

# 也可以将账户信息转换为字典方便查看
account_info_dict = account_info._asdict()
print("\nAccount info as dictionary:")
print(account_info_dict)

symbol = "EURUSD"  # 示例品种
# 尝试将品种添加到 MarketWatch 并使其数据可用
selected = mt5.symbol_select(symbol, True)
# 第二个参数 True 表示选中
if not selected:
    print(f"Failed to select {symbol}, maybe it's not available on the server?")
    # 尝试取消选择（如果不需要）: mt5.symbol_select(symbol, False)
    # 如果无法选择，后续获取该品种数据的操作会失败
else:
    print(f"{symbol} selected and data available in MarketWatch.")

symbol_to_check = "GBPUSD"  # 确保品种被选中
if not mt5.symbol_select(symbol_to_check, True):
    print(f"Failed to select {symbol_to_check}")
else:
    # 获取品种信息
    info = mt5.symbol_info(symbol_to_check)
    if info is not None:
        print(f"\nInformation for {symbol_to_check}:")
        print(f" Digits (小数点位数): {info.digits}")
        print(f" Point (点值): {info.point}")
        print(f" Spread (当前点差): {info.spread}")
        print(f" Contract Size (合约大小): {info.trade_contract_size}")
        print(f" Tick Value (每点价值): {info.tick_value}")
        print(f" Tick Size (最小变动单位): {info.tick_size}")
        print(f" Trade Mode (交易模式): {info.trade_mode}")  # e.g., TRADE_MODE_FULL
        print(f" Description (描述): {info.description}")
        # 转换为字典
        # print("\nSymbol info as dictionary:")
        # print(info._asdict())
    else:
        print(f"Failed to get info for {symbol_to_check}, error code =", mt5.last_error())

# 获取 MarketWatch 中所有可见品种的信息
all_symbols_info = mt5.symbols_get()
print(f"\nTotal symbols in MarketWatch: {len(all_symbols_info)}")
if len(all_symbols_info) > 0:
    print("First few symbols:")
    for i, sym_info in enumerate(all_symbols_info[:5]):  # 只显示前 5 个
        print(f" {i+1}. {sym_info.name} - {sym_info.description}")

# 按组获取 (组名取决于你的 Broker 设置)
# forex_symbols_info = mt5.symbols_get("group_name")  # 例如 "Forex", "Indices"
# if forex_symbols_info:
#     print(f"\nForex symbols found: {len(forex_symbols_info)}")

symbol_tick = "USDJPY"  # 确保选中
if not mt5.symbol_select(symbol_tick, True):
    print(f"Failed to select {symbol_tick}")
else:
    # 获取最新 tick
    last_tick = mt5.symbol_info_tick(symbol_tick)
    if last_tick:
        print(f"\nLatest tick for {symbol_tick}:")
        # Tick 时间是 POSIX 时间戳 (自 1970-01-01 以来的秒数)
        tick_time_dt = datetime.fromtimestamp(last_tick.time)
        print(f" Time: {tick_time_dt.strftime('%Y-%m-%d %H:%M:%S')}")
        print(f" Bid: {last_tick.bid}")
        print(f" Ask: {last_tick.ask}")
        print(f" Last Price: {last_tick.last}")  # 如果有最后成交价 (股票/期货)
        print(f" Volume: {last_tick.volume}")  # 如果有成交量
        print(f" Flags: {last_tick.flags}")  # Tick 类型标志
    else:
        print(f"Failed to get tick for {symbol_tick}, error code =", mt5.last_error())

symbol_hist_ticks = "EURUSD"  # 确保选中
if not mt5.symbol_select(symbol_hist_ticks, True):
    print(f"Failed to select {symbol_hist_ticks}")
else:
    # 定义时间范围 (例如：过去 1 小时)
    time_to = datetime.now()
    time_from = time_to - pd.Timedelta(hours=1)  # 使用 Pandas Timedelta

    # 获取 Tick 数据
    # copy_ticks_range(symbol, dt_from, dt_to, flags)
    # flags 控制获取的数据类型：mt5.COPY_TICKS_ALL, mt5.COPY_TICKS_INFO (买卖价), mt5.COPY_TICKS_TRADE (最后价)
    try:
        ticks = mt5.copy_ticks_range(symbol_hist_ticks, time_from, time_to, mt5.COPY_TICKS_ALL)
        if ticks is not None:
            print(f"\nRetrieved {len(ticks)} ticks for {symbol_hist_ticks} in the last hour.")
            # 转换为 DataFrame
            ticks_df = pd.DataFrame(ticks)
            # 将时间戳转换为 datetime 对象
            ticks_df['time'] = pd.to_datetime(ticks_df['time'], unit='s')
            ticks_df = ticks_df.set_index('time')
            # 将时间设为索引
            print("Last 5 ticks:")
            print(ticks_df.tail())
        elif mt5.last_error() == mt5.RES_E_INTERNAL_ERROR:
            # 常见错误：可能是服务器限制或无数据
            print(f"Could not retrieve ticks for {symbol_hist_ticks}. No data or server restriction? Error: {mt5.last_error()}")
        else:
            print(f"Failed to retrieve ticks for {symbol_hist_ticks}, error code =", mt5.last_error())
    except Exception as e:
        print(f"An exception occurred while fetching ticks: {e}")

symbol_rates = "EURUSD"
timeframe = mt5.TIMEFRAME_H1  # 1 小时 K 线
count = 100  # 获取 100 根
start_date = datetime(2024, 1, 1)  # 指定开始日期 (注意是向前获取)

# 确保选中
if not mt5.symbol_select(symbol_rates, True):
    print(f"Failed to select {symbol_rates}")
else:
    rates_from = mt5.copy_rates_from(symbol_rates, timeframe, start_date, count)
    if rates_from is not None and len(rates_from) > 0:
        print(f"\nRetrieved {len(rates_from)} H1 bars for {symbol_rates} starting before {start_date}")
        # rates_from 是 NumPy array
    else:
        print(f"Failed to get rates using copy_rates_from, error code =", mt5.last_error())

start_pos = 0  # 从最新的 K 线开始 (第 0 根)
count_pos = 50  # 获取 50 根

# 确保选中
if not mt5.symbol_select(symbol_rates, True):
    print(f"Failed to select {symbol_rates}")
else:
    rates_pos = mt5.copy_rates_from_pos(symbol_rates, timeframe, start_pos, count_pos)
    if rates_pos is not None and len(rates_pos) > 0:
        print(f"\nRetrieved {len(rates_pos)} H1 bars for {symbol_rates} starting from position {start_pos}")
    else:
        print(f"Failed to get rates using copy_rates_from_pos, error code =", mt5.last_error())

symbol_rates = "XAUUSD"  # 黄金
timeframe_range = mt5.TIMEFRAME_D1  # 日线
date_from = datetime(2023, 1, 1)
date_to = datetime(2024, 1, 1)  # 获取 2023 全年的数据

# 确保选中
if not mt5.symbol_select(symbol_rates, True):
    print(f"Failed to select {symbol_rates}")
else:
    rates_range = mt5.copy_rates_range(symbol_rates, timeframe_range, date_from, date_to)
    if rates_range is not None and len(rates_range) > 0:
        print(f"\nRetrieved {len(rates_range)} D1 bars for {symbol_rates} from {date_from} to {date_to}")
    else:
        print(f"Failed to get rates using copy_rates_range, error code =", mt5.last_error())

if rates_range is not None and len(rates_range) > 0:
    # 将 NumPy 结构化数组转换为 Pandas DataFrame
    rates_df = pd.DataFrame(rates_range)
    # 转换时间戳列 (通常是第一列 'time') 为 datetime 对象
    # MT5 返回的是 POSIX 时间戳 (秒)
    rates_df['time'] = pd.to_datetime(rates_df['time'], unit='s')
    # 将 'time' 列设置为索引
    rates_df = rates_df.set_index('time')
    # 可以重命名列名，使其更符合习惯 (OHLCV)
    # rates_df.rename(columns={'open': 'Open', 'high': 'High', 'low': 'Low', 'close': 'Close', 'tick_volume': 'Volume'}, inplace=True)
    print("\nRates data as Pandas DataFrame (last 5 rows):")
    print(rates_df.tail())
    print("\nDataFrame Info:")
    rates_df.info()
else:
    print("No rates data to convert to DataFrame.")

# 获取挂单总数
orders_count = mt5.orders_total()
print(f"\nTotal pending orders: {orders_count}")
if orders_count > 0:
    # 获取所有挂单的详细信息
    # 可以按 symbol 或 ticket 过滤：orders_get(symbol="EURUSD")
    orders = mt5.orders_get()
    if orders is not None:
        print("Details of pending orders:")
        orders_list = []
        for order in orders:
            orders_list.append(order._asdict())  # 转为字典方便查看
            # print(f" Ticket: {order.ticket}, Symbol: {order.symbol}, Type: {order.type_str}, Price: {order.price_open}, Volume: {order.volume_current}")
        # 转换为 DataFrame (可选)
        orders_df = pd.DataFrame(orders_list)
        # 转换时间相关列
        time_cols_order = ['time_setup', 'time_expiration', 'time_done']
        for col in time_cols_order:
            if col in orders_df.columns:
                orders_df[col] = pd.to_datetime(orders_df[col], unit='s')
        print(orders_df[['ticket', 'symbol', 'type', 'volume_initial', 'price_open', 'sl', 'tp', 'time_setup']].head())
    else:
        print("Failed to get pending orders details, error code =", mt5.last_error())

# 获取持仓总数
positions_count = mt5.positions_total()
print(f"\nTotal open positions: {positions_count}")
if positions_count > 0:
    # 获取所有持仓的详细信息
    # 可以按 symbol 或 ticket 过滤：positions_get(symbol="EURUSD")
    positions = mt5.positions_get()
    if positions is not None:
        print("Details of open positions:")
        positions_list = []
        for position in positions:
            positions_list.append(position._asdict())
            # print(f" Ticket: {position.ticket}, Symbol: {position.symbol}, Type: {position.type}, Price: {position.price_open}, Profit: {position.profit}, Volume: {position.volume}")
        # 转换为 DataFrame (可选)
        positions_df = pd.DataFrame(positions_list)
        time_cols_pos = ['time', 'time_update']
        for col in time_cols_pos:
            if col in positions_df.columns:
                positions_df[col] = pd.to_datetime(positions_df[col], unit='s')
        print(positions_df[['ticket', 'symbol', 'type', 'volume', 'price_open', 'sl', 'tp', 'profit', 'time']].head())
    else:
        print("Failed to get open positions details, error code =", mt5.last_error())

# 定义要获取历史记录的时间范围
hist_date_from = datetime(2024, 1, 1)
hist_date_to = datetime.now()

# 获取指定时间范围内的所有成交记录
# 可以按 group (包含/排除的品种掩码) 或 position (特定头寸的 ticket) 过滤
history_deals = mt5.history_deals_get(hist_date_from, hist_date_to)  # 获取所有品种

if history_deals is not None:
    print(f"\nTotal deals found from {hist_date_from} to {hist_date_to}: {len(history_deals)}")
    if len(history_deals) > 0:
        deals_list = []
        for deal in history_deals:
            deals_list.append(deal._asdict())
            # print(f" Ticket: {deal.ticket}, Order: {deal.order}, Symbol: {deal.symbol}, Type: {deal.type}, Price: {deal.price}, Profit: {deal.profit}, Time: {datetime.fromtimestamp(deal.time)}")
        # 转换为 DataFrame
        deals_df = pd.DataFrame(deals_list)
        deals_df['time'] = pd.to_datetime(deals_df['time'], unit='s')
        deals_df = deals_df.sort_values(by='time')  # 按时间排序
        print("Last 5 history deals:")
        print(deals_df[['ticket', 'order', 'symbol', 'type', 'entry', 'volume', 'price', 'profit', 'commission', 'fee', 'time']].tail())
    else:
        print("No deals found in the specified period.")
else:
    print("Failed to get history deals, error code =", mt5.last_error())

import MetaTrader5 as mt5
from datetime import datetime
import pandas as pd
import time

# --- 1. 初始化连接 ---
if not mt5.initialize():
    print("initialize() failed, error code =", mt5.last_error())
    quit()
print("MT5 Initialized")

# --- 2. 获取账户信息 ---
account_info = mt5.account_info()
if account_info:
    print(f"Account login: {account_info.login}, Balance: {account_info.balance}{account_info.currency}")
else:
    print("Failed to get account info")

# --- 3. 获取 EURUSD 数据 ---
symbol = "EURUSD"
timeframe = mt5.TIMEFRAME_H1
bars_to_get = 10

# 确保选中
if not mt5.symbol_select(symbol, True):
    print(f"Failed to select {symbol}, error code =", mt5.last_error())
    mt5.shutdown()
    quit()
print(f"{symbol} selected.")
time.sleep(0.5)  # 短暂等待数据同步

# 获取最新 Tick
tick = mt5.symbol_info_tick(symbol)
if tick:
    tick_time = datetime.fromtimestamp(tick.time).strftime('%Y-%m-%d %H:%M:%S')
    print(f"\nLatest {symbol} Tick ({tick_time}): Bid={tick.bid}, Ask={tick.ask}")
else:
    print(f"Failed to get {symbol} tick, error code =", mt5.last_error())

# 获取最近的 K 线
rates = mt5.copy_rates_from_pos(symbol, timeframe, 0, bars_to_get)
if rates is not None and len(rates) > 0:
    rates_df = pd.DataFrame(rates)
    rates_df['time'] = pd.to_datetime(rates_df['time'], unit='s')
    rates_df = rates_df.set_index('time')
    print(f"\nLast {len(rates_df)} H1 bars for {symbol}:")
    print(rates_df[['open', 'high', 'low', 'close', 'tick_volume']])
else:
    print(f"Failed to get {symbol} rates, error code =", mt5.last_error())

# --- 4. 关闭连接 ---
mt5.shutdown()
print("\nMT5 Connection closed.")

MetaTrader5 Python 库数据获取与交易接口详解

MetaTrader5 Python 库数据获取与交易接口详解

1. 简介与安装

1.1 什么是 MetaTrader5 Python 库？

1.2 环境准备与安装

1.3 MT5 终端设置

2. 连接与初始化

2.1 导入库

2.2 初始化连接 (initialize)

微信扫一扫，关注极客日志

更多推荐文章

相关免费在线工具

2.3 检查连接状态

2.4 关闭连接 (shutdown)

3. 获取账户信息

3.1 account_info()

3.2 账户信息属性解析

4. 获取交易品种 (Symbol) 信息

4.1 什么是交易品种？

4.2 symbol_select() - 选择品种（重要）

4.3 symbol_info() - 获取单个品种的详细规格

4.4 symbols_get() - 获取多个品种的信息

4.5 品种信息属性解析

5. 获取实时报价 (Ticks)

5.1 symbol_info_tick() - 获取最新 Tick

5.2 Tick 数据属性解析

5.3 copy_ticks_from() / copy_ticks_range() - 获取历史 Tick 数据

6. 获取历史 K 线数据 (Bars/Candlesticks)

6.1 K 线数据的重要性

6.2 时间周期 (Timeframe) 常量

6.3 copy_rates_from(symbol, timeframe, start_date, count) - 从指定日期获取

6.4 copy_rates_from_pos(symbol, timeframe, start_pos, count) - 从指定位置获取

6.5 copy_rates_range(symbol, timeframe, date_from, date_to) - 按日期范围获取 (常用)

6.6 返回数据格式 (NumPy Array)

6.7 转换为 Pandas DataFrame (推荐)

6.8 K 线数据字段解析

7. 获取订单、持仓与交易历史

7.1 获取挂单 (Pending Orders)

7.2 获取当前持仓 (Open Positions)

7.3 获取历史成交记录 (History Deals)

7.4 数据结构解析与转换为 DataFrame

8. 错误处理与最佳实践

8.1 检查函数返回值

8.2 使用 mt5.last_error()

8.3 symbol_select 的重要性

8.4 使用 try...except 结构

8.5 注意数据获取的限制

9. 综合示例

10. 总结

微信扫一扫，关注极客日志

更多推荐文章

相关免费在线工具

2.2 初始化连接 (`initialize`)

2.4 关闭连接 (`shutdown`)

3.1 `account_info()`

4.2 `symbol_select()` - 选择品种（重要）

4.3 `symbol_info()` - 获取单个品种的详细规格

4.4 `symbols_get()` - 获取多个品种的信息

5.1 `symbol_info_tick()` - 获取最新 Tick

5.3 `copy_ticks_from()` / `copy_ticks_range()` - 获取历史 Tick 数据

6.3 `copy_rates_from(symbol, timeframe, start_date, count)` - 从指定日期获取

6.4 `copy_rates_from_pos(symbol, timeframe, start_pos, count)` - 从指定位置获取

6.5 `copy_rates_range(symbol, timeframe, date_from, date_to)` - 按日期范围获取 (常用)

8.2 使用 `mt5.last_error()`

8.3 `symbol_select` 的重要性

8.4 使用 `try...except` 结构