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

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

pip install MetaTrader5 --upgrade

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)

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

print("MetaTrader 5 initialized successfully!")
print(f"MetaTrader 5 version: {mt5.version()}")

# 在脚本结束时或不再需要连接时调用
# 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` 结构