一款专业级执行机器人，连接Python机器学习模型与MetaTrader 5终端，实现实时算法交易。
📖 概述
该机器人根据预训练的XGBoost/ML模型（格式）预测自动执行交易。它采用严格的事件驱动循环，与蜡烛闭合同步以防止滑动并确保数据完整性。joblib
它包含一个强大的动态风险管理引擎，基于账户股权和波动性计算手型，并自动实现盈亏平衡跟踪逻辑。
⚙️ 核心架构
1. 同步引擎（心跳）
- 逻辑：计算距离下一次M5蜡烛收盘还有多少秒。
- 优点：确保在闭合蜡烛上计算特征（RSI、BBands、VWAP），避免重新涂装问题。
2. 多时间框架特征工程
M5（战术）和H4（战略）时间框架的实时技术指标：
- RSI（相对强度指数）带有滞后特征。
- 布林带（波动率）。
- VWAP（成交量加权平均价格）用于机构资金流跟踪。
3. 动态风险管理师
- 资本保护：风险固定为每笔交易账户股本的1%。
- 波动率评估：手数根据到止损的距离计算（信号烛的高/低）。
- 收支平衡的逻辑：当交易达到1R利润时，将止损调整为入场价。
🔄 执行逻辑
🚀 主要特征
| 特色 | 描述 |
|---|---|
| 🤖 机器学习集成 | 加载 joblib 模型以进行实时推理。 |
| 🛡️ 风险控制 | 硬性设定的风险限额（最大批次，风险百分比），以避免过度杠杆。 |
| ⏱️ 零延迟 | 新蜡烛开启后毫秒内执行命令。 |
| 📊 多TF数据 | 结合 M5 和 H4 数据流，实现上下文感知决策。 |
📂 项目结构
python-mt5-live-trading-bot/ ├── master.py # Main Event Loop & Execution Logic ├── output/ # Model Directory │ └── eurusd_model.joblib # Pre-trained ML Model ├── requirements.txt # Dependencies └── README.md # Documentation
💻 用途
1. 先决条件
- MetaTrader 5终端：已安装并登录对冲账户。
- 启用算法交易：在MT5工具栏点击“算法交易”。
2. 配置编辑以设置风险参数：master.py
RISK_PER_TRADE_PERCENT = 1.0 # Risk 1% per trade SYMBOL = "EURUSD" MAGIC_NUMBER = 777 # Unique ID for this bot's trades
3. 启动
python master.py
控制台会显示“等待下一根蜡烛……”并且与市场同步。
⚠️ 免责声明
高风险投资警示
该软件在一个活体金融账户上执行交易。算法交易涉及显著的资本风险。
- 使用请自行承担风险：作者不对经济损失负责。
- 先测试演示：在正式上线前，先在演示账号上运行机器人至少4周。
- 市场状况：过去的ML表现不保证未来结果。
由雷杜安·邦德拉（Redouane Boundra）担任工程师。
该项目基于MIT协议
原文
Copyright (C) <year> <copyright holders>
Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the “Software”), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED “AS IS”, WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. [1]
说明
- 被授权人权利
被授权人有权利使用、复制、修改、合并、出版发行、散布、再授权及贩售软件及软件的副本。
被授权人可根据程序的需要修改授权条款为适当的内容。
- 被授权人义务
在软件和软件的所有副本中都必须包含版权声明和许可声明。
- 其他重要特性
此授权条款并非属copyleft的自由软件授权条款，允许在自由/开放源码软件或非自由软件（proprietary software）所使用。
MIT的内容可依照程序著作权者的需求更改内容。此亦为MIT与BSD（The BSD license, 3-clause BSD license）本质上不同处。
MIT条款可与其他授权条款并存。另外，MIT条款也是自由软件基金会（FSF）所认可的自由软件授权条款，与GPL兼容。
代码全文
import pandas as pd
import joblib
import os
import numpy as np
import MetaTrader5 as mt5
from datetime import datetime, timedelta
import pytz
import time
print("--- Live Trading Bot Initializing ---")
RISK_PER_TRADE_PERCENT = 1.0
CONTRACT_SIZE = 100000
MAX_LOT_SIZE = 50.0
MIN_RISK_PIPS = 2.0
RR_RATIO = 2.0
MAGIC_NUMBER = 777
OUTPUT_DIRECTORY = "output/"
SYMBOL = "EURUSD"
TIMEFRAME = mt5.TIMEFRAME_M5
MODEL_FILENAME = f"{SYMBOL.lower()}_model.joblib"
def calculate_rsi(data, window=14):
delta = data.diff()
gain = delta.where(delta > 0, 0).rolling(window=window, min_periods=1).mean()
loss = -delta.where(delta < 0, 0).rolling(window=window, min_periods=1).mean()
rs = gain / loss
return 100 - (100 / (1 + rs))
def calculate_bollinger_bands(data, window=20):
rolling_mean = data.rolling(window=window, min_periods=1).mean()
rolling_std = data.rolling(window=window, min_periods=1).std()
upper = rolling_mean + (rolling_std * 2)
lower = rolling_mean - (rolling_std * 2)
return upper, lower
def calculate_daily_vwap(df):
df['typical_price_volume'] = ((df['high'] + df['low'] + df['close']) / 3) * df['tick_volume']
df['cumulative_volume'] = df.groupby(df.index.date)['tick_volume'].cumsum()
df['cumulative_tpv'] = df.groupby(df.index.date)['typical_price_volume'].cumsum()
return df['cumulative_tpv'] / df['cumulative_volume']
def run_live_bot():
print("--- Connecting to MetaTrader 5 terminal... ---")
if not mt5.initialize():
print("initialize() failed, error code =", mt5.last_error())
return
print("--- Loading the predictive model... ---")
model_path = os.path.join(OUTPUT_DIRECTORY, MODEL_FILENAME)
if not os.path.exists(model_path):
print(f"Error: Model file not found at {model_path}")
mt5.shutdown()
return
model = joblib.load(model_path)
print("--- Model loaded successfully. Bot is now running. ---")
symbol_info = mt5.symbol_info(SYMBOL)
if symbol_info is None:
print(f"{SYMBOL} not found on broker.")
mt5.shutdown()
return
point = symbol_info.point
try:
while True:
now_utc = datetime.now(pytz.utc)
next_candle_time = now_utc.replace(second=0, microsecond=0) + timedelta(minutes=5 - (now_utc.minute % 5))
sleep_seconds = (next_candle_time - now_utc).total_seconds()
print(f"Current time: {now_utc.strftime('%H:%M:%S')}. Waiting for {sleep_seconds:.1f}s until next candle at {next_candle_time.strftime('%H:%M:%S')}...")
time.sleep(sleep_seconds)
positions = mt5.positions_get(symbol=SYMBOL)
my_position = None
if positions:
for pos in positions:
if pos.magic == MAGIC_NUMBER:
my_position = pos
break
if my_position:
if my_position.sl == my_position.price_open:
print(f"Trade #{my_position.ticket} is active and already at breakeven. Monitoring...")
continue
breakeven_price = my_position.price_open + (my_position.price_open - my_position.sl)
tick = mt5.symbol_info_tick(SYMBOL)
if my_position.type == mt5.ORDER_TYPE_BUY and tick.bid >= breakeven_price:
print(f"Trade #{my_position.ticket} hit 1R profit. Moving SL to breakeven.")
request = {
"action": mt5.TRADE_ACTION_SLTP,
"position": my_position.ticket,
"sl": my_position.price_open,
"tp": my_position.tp,
}
mt5.order_send(request)
continue
rates_m5_df = pd.DataFrame(mt5.copy_rates_from_pos(SYMBOL, TIMEFRAME, 0, 50))
rates_m5_df['time'] = pd.to_datetime(rates_m5_df['time'], unit='s', utc=True)
rates_m5_df.set_index('time', inplace=True)
rates_h4_df = pd.DataFrame(mt5.copy_rates_from_pos(SYMBOL, mt5.TIMEFRAME_H4, 0, 50))
rates_h4_df['time'] = pd.to_datetime(rates_h4_df['time'], unit='s', utc=True)
rates_h4_df.set_index('time', inplace=True)
rates_m5_df['m5_rsi'] = calculate_rsi(rates_m5_df['close'])
rates_m5_df['m5_bb_upper'], rates_m5_df['m5_bb_lower'] = calculate_bollinger_bands(rates_m5_df['close'])
rates_m5_df['m5_vwap'] = calculate_daily_vwap(rates_m5_df)
rates_h4_df['h4_rsi'] = calculate_rsi(rates_h4_df['close'])
rates_h4_df['h4_bb_upper'], rates_h4_df['h4_bb_lower'] = calculate_bollinger_bands(rates_h4_df['close'])
rates_h4_df['h4_vwap'] = calculate_daily_vwap(rates_h4_df)
last_candle = rates_m5_df.iloc[-2]
prev_candle = rates_m5_df.iloc[-3]
is_bullish_reversal = last_candle['close'] > last_candle['open'] and prev_candle['close'] < prev_candle['open']
is_bearish_reversal = last_candle['close'] < last_candle['open'] and prev_candle['close'] > prev_candle['open']
if is_bullish_reversal or is_bearish_reversal:
print(f"Signal detected at {last_candle.name}. Preparing features...")
features = {}
for i in range(1, 11):
features[f'm5_rsi_lag_{i}'] = rates_m5_df['m5_rsi'].iloc[-(i+1)]
h4_candle = rates_h4_df[rates_h4_df.index <= last_candle.name].iloc[-1]
features['h4_rsi'] = h4_candle['h4_rsi']
feature_df = pd.DataFrame([features], columns=model.feature_names_in_)
prediction = model.predict(feature_df)[0]
trade_direction = 1 if prediction == 1 else -1
print(f"Model predicts: {'BUY' if trade_direction == 1 else 'SELL'}. Validating trade...")
if trade_direction == 1:
sl_price = last_candle['low']
risk_points = last_candle['close'] - sl_price
else:
sl_price = last_candle['high']
risk_points = sl_price - last_candle['close']
if risk_points >= (MIN_RISK_PIPS * point):
account_info = mt5.account_info()
equity = account_info.equity
risk_amount_dollars = equity * (RISK_PER_TRADE_PERCENT / 100)
risk_per_lot_dollars = risk_points * CONTRACT_SIZE
lot_size = min(risk_amount_dollars / risk_per_lot_dollars, MAX_LOT_SIZE)
if lot_size >= 0.01:
lot_size = round(lot_size, 2)
tp_price = last_candle['close'] + (risk_points * RR_RATIO * trade_direction)
tick = mt5.symbol_info_tick(SYMBOL)
price = tick.ask if trade_direction == 1 else tick.bid
print(f"Executing {'BUY' if trade_direction == 1 else 'SELL'} trade. Size: {lot_size}, SL: {sl_price}, TP: {tp_price}")
request = {
"action": mt5.TRADE_ACTION_DEAL,
"symbol": SYMBOL,
"volume": lot_size,
"type": mt5.ORDER_TYPE_BUY if trade_direction == 1 else mt5.ORDER_TYPE_SELL,
"price": price,
"sl": sl_price,
"tp": tp_price,
"magic": MAGIC_NUMBER,
"comment": "XGBoost Bot",
"type_time": mt5.ORDER_TIME_GTC,
"type_filling": mt5.ORDER_FILLING_IOC,
}
result = mt5.order_send(request)
if result.retcode != mt5.TRADE_RETCODE_DONE:
print(f"Order send failed, retcode={result.retcode}")
except KeyboardInterrupt:
print("\n--- Bot shutdown requested. ---")
finally:
mt5.shutdown()
print("--- MetaTrader 5 connection closed. Bot has stopped. ---")
if __name__ == '__main__':
run_live_bot()
# WARNING: This bot will execute LIVE trades.
# Run on a DEMO account first. I'm not responsible for anything