gavindiaz/mt5_python_ea_suite
1
0
Fork 0
mirror of https://github.com/silencesdg/mt5_python_ea_suite.git synced 2026-07-30 12:17:43 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: 远程MT5数据源 + 信号对冲/回撤锁仓模块

Browse Source 
新增:
- core/risk/hedge.py: 对冲管理器
  - 信号对冲: 加权信号反向超阈值→半仓反向
  - 回撤锁仓: 浮亏超-0.3%→全仓锁死
  - 自动解锁: 信号回正/对冲止盈0.5%

改动:
- run/backtest.py: 支持远程数据源回测
- run/realtime.py: 远程/本地双模式
- core/risk/position.py: 集成HedgeManager
- core/risk/controller.py: 传递weighted_signal
- execution/realtime_trader.py: 传入加权信号
- core/data/live.py, utils.py: MetaTrader5懒加载(ARM兼容)
- config.py: HEDGE_CONFIG, REMOTE配置, INITIAL_CAPITAL=1944

今日实盘: 9单, +7.4% (944→088)
This commit is contained in:
songkl
2026-05-11 23:46:49 +08:00
parent 4cb4f4a15e
commit 878e0f4a03
10 changed files with 478 additions and 89 deletions
Download Patch File Download Diff File Expand all files Collapse all files
config.py
+25 -2
View File
@@ -3,12 +3,15 @@ SERVER_HOST = "0.0.0.0"
SERVER_PORT = 5555
# 远端客户端配置（迁移到其他电脑时填写 MT5 机器的 IP）
REMOTE_SERVER_HOST = "127.0.0.1"
REMOTE_SERVER_HOST = "192.168.1.5"
REMOTE_SERVER_PORT = 5555
# 数据提供者模式: "remote" (远程HTTP API) / "local" (本机MT5)
DATA_PROVIDER_MODE = "remote"
# 交易配置
SYMBOL = "XAUUSDz"
INITIAL_CAPITAL = 1937 # 初始资金（2026-05-11 实盘余额）
INITIAL_CAPITAL = 1944 # 初始资金（2026-05-12 实盘余额 $1944.27）
# 时间配置
TIMEFRAME = 1# M1 (1分钟图) - MT5常量值
@@ -68,6 +71,26 @@ REALTIME_CONFIG = {
"trade_direction": "both", # 交易方向: "long"(只做多), "short"(只做空), "both"(多空都支持)
}
# 对冲配置（信号对冲 + 回撤锁仓）
HEDGE_CONFIG = {
# ── 信号对冲 ──
"signal_hedge_enabled": True, # 启用信号对冲
"signal_hedge_threshold": 2.0, # 加权信号绝对值超此值触发对冲
"signal_hedge_ratio": 0.5, # 对冲手数比例 (0.5=半仓对冲)
"signal_unhedge_threshold": 1.0, # 信号回到此值以下解锁
# ── 回撤锁仓 ──
"drawdown_hedge_enabled": True, # 启用回撤锁仓
"drawdown_hedge_pct": -0.003, # 浮亏超-0.3%触发锁仓
"drawdown_hedge_ratio": 1.0, # 锁仓比例 (1.0=全额锁仓)
# ── 对冲单止盈 ──
"hedge_take_profit_pct": 0.005, # 对冲单自身盈利0.5%止盈
# ── 风控限制 ──
"max_hedges_per_day": 5, # 每日最多对冲5次
}
# 数据获取配置
DATA_CONFIG = {
core/data/__init__.py
+19 -2
View File
@@ -1,6 +1,23 @@
from core.data.abc import DataProvider
from core.data.live import LiveDataProvider
from core.data.dryrun import DryRunDataProvider
from core.data.backtest import BacktestDataProvider
from core.data.multi_tf import MultiTimeframeDataStore
from core.data.remote import RemoteDataProvider
# LiveDataProvider 和 DryRunDataProvider 依赖 MetaTrader5，
# 仅在本地模式需要时延迟导入
_LiveDataProvider = None
_DryRunDataProvider = None
def LiveDataProvider(*args, **kwargs):
global _LiveDataProvider
if _LiveDataProvider is None:
from core.data.live import LiveDataProvider as LDP
_LiveDataProvider = LDP
return _LiveDataProvider(*args, **kwargs)
def DryRunDataProvider(*args, **kwargs):
global _DryRunDataProvider
if _DryRunDataProvider is None:
from core.data.dryrun import DryRunDataProvider as DDP
_DryRunDataProvider = DDP
return _DryRunDataProvider(*args, **kwargs)
core/data/live.py
+42 -34
View File
@@ -1,8 +1,16 @@
import MetaTrader5 as mt5
import pandas as pd
from logger import logger
from core.data.abc import DataProvider
# 惰性导入 — ARM 环境不装 MetaTrader5
_mt5 = None
def _get_mt5():
global _mt5
if _mt5 is None:
import MetaTrader5 as _mt5
return _mt5
class LiveDataProvider(DataProvider):
"""实盘数据提供者 — 封装真实MT5 API调用"""
@@ -12,18 +20,18 @@ class LiveDataProvider(DataProvider):
return True
def initialize(self):
if not mt5.initialize():
if not _get_mt5().initialize():
logger.error("MT5初始化失败")
return False
logger.info("MT5连接成功")
return True
def shutdown(self):
mt5.shutdown()
_get_mt5().shutdown()
logger.info("MT5连接已关闭")
def get_current_price(self, symbol):
tick = mt5.symbol_info_tick(symbol)
tick = _get_mt5().symbol_info_tick(symbol)
if tick:
last_price = tick.last if tick.last != 0 else (tick.bid + tick.ask) / 2
return {
@@ -34,16 +42,16 @@ class LiveDataProvider(DataProvider):
return None
def get_historical_data(self, symbol, timeframe, count, **kwargs):
return mt5.copy_rates_from_pos(symbol, timeframe, 0, count)
return _get_mt5().copy_rates_from_pos(symbol, timeframe, 0, count)
def get_account_info(self):
return mt5.account_info()
return _get_mt5().account_info()
def get_positions(self, symbol):
return mt5.positions_get(symbol=symbol)
return _get_mt5().positions_get(symbol=symbol)
def get_symbol_info(self, symbol):
return mt5.symbol_info(symbol)
return _get_mt5().symbol_info(symbol)
def send_order(self, symbol, order_type, volume):
price_data = self.get_current_price(symbol)
@@ -53,16 +61,16 @@ class LiveDataProvider(DataProvider):
price = price_data['ask'] if order_type == "buy" else price_data['bid']
if not mt5.terminal_info().trade_allowed:
if not _get_mt5().terminal_info().trade_allowed:
logger.error("MT5终端未启用自动交易")
return None
account_info = mt5.account_info()
account_info = _get_mt5().account_info()
if account_info and not account_info.trade_allowed:
logger.error("当前账户不允许自动交易")
return None
symbol_info = mt5.symbol_info(symbol)
symbol_info = _get_mt5().symbol_info(symbol)
if not symbol_info:
logger.error(f"无法获取 {symbol} 的品种信息")
return None
@@ -71,15 +79,15 @@ class LiveDataProvider(DataProvider):
fm = symbol_info.filling_mode
# MQL5 filling_mode 位图: FOK=1, IOC=2
if fm & 1:
filling_mode = mt5.ORDER_FILLING_FOK
filling_mode = _get_mt5().ORDER_FILLING_FOK
elif fm & 2:
filling_mode = mt5.ORDER_FILLING_IOC
filling_mode = _get_mt5().ORDER_FILLING_IOC
else:
filling_mode = mt5.ORDER_FILLING_RETURN
filling_mode = _get_mt5().ORDER_FILLING_RETURN
order_type_mt5 = mt5.ORDER_TYPE_BUY if order_type == "buy" else mt5.ORDER_TYPE_SELL
order_type_mt5 = _get_mt5().ORDER_TYPE_BUY if order_type == "buy" else _get_mt5().ORDER_TYPE_SELL
request = {
"action": mt5.TRADE_ACTION_DEAL,
"action": _get_mt5().TRADE_ACTION_DEAL,
"symbol": symbol,
"volume": volume,
"type": order_type_mt5,
@@ -89,18 +97,18 @@ class LiveDataProvider(DataProvider):
"comment": f"{order_type} order",
"type_filling": filling_mode,
}
result = mt5.order_send(request)
result = _get_mt5().order_send(request)
if result and hasattr(result, 'retcode'):
if result.retcode == 10027:
logger.error("自动交易被禁用")
elif result.retcode == 10030:
if filling_mode != mt5.ORDER_FILLING_IOC:
request["type_filling"] = mt5.ORDER_FILLING_IOC
result = mt5.order_send(request)
if result and result.retcode == 10030 and filling_mode != mt5.ORDER_FILLING_FOK:
request["type_filling"] = mt5.ORDER_FILLING_FOK
result = mt5.order_send(request)
if filling_mode != _get_mt5().ORDER_FILLING_IOC:
request["type_filling"] = _get_mt5().ORDER_FILLING_IOC
result = _get_mt5().order_send(request)
if result and result.retcode == 10030 and filling_mode != _get_mt5().ORDER_FILLING_FOK:
request["type_filling"] = _get_mt5().ORDER_FILLING_FOK
result = _get_mt5().order_send(request)
elif result.retcode != 10009:
logger.error(f"下单失败，错误代码: {result.retcode}")
@@ -122,33 +130,33 @@ class LiveDataProvider(DataProvider):
logger.error(f"未找到ticket为 {ticket} 的持仓")
return False
tick = mt5.symbol_info_tick(symbol)
tick = _get_mt5().symbol_info_tick(symbol)
if not tick:
logger.error(f"无法获取 {symbol} 的当前价格")
return False
if target_position.type == mt5.POSITION_TYPE_BUY:
if target_position.type == _get_mt5().POSITION_TYPE_BUY:
close_price = tick.bid
order_type = mt5.ORDER_TYPE_SELL
order_type = _get_mt5().ORDER_TYPE_SELL
else:
close_price = tick.ask
order_type = mt5.ORDER_TYPE_BUY
order_type = _get_mt5().ORDER_TYPE_BUY
symbol_info = mt5.symbol_info(symbol)
symbol_info = _get_mt5().symbol_info(symbol)
if not symbol_info:
return False
fm = symbol_info.filling_mode
# MQL5 filling_mode 位图: FOK=1, IOC=2
if fm & 1:
filling_mode = mt5.ORDER_FILLING_FOK
filling_mode = _get_mt5().ORDER_FILLING_FOK
elif fm & 2:
filling_mode = mt5.ORDER_FILLING_IOC
filling_mode = _get_mt5().ORDER_FILLING_IOC
else:
filling_mode = mt5.ORDER_FILLING_IOC
filling_mode = _get_mt5().ORDER_FILLING_IOC
request = {
"action": mt5.TRADE_ACTION_DEAL,
"action": _get_mt5().TRADE_ACTION_DEAL,
"position": target_position.ticket,
"symbol": symbol,
"volume": volume,
@@ -160,11 +168,11 @@ class LiveDataProvider(DataProvider):
"type_filling": filling_mode,
}
result = mt5.order_send(request)
result = _get_mt5().order_send(request)
if result is None:
logger.error(f"平仓请求返回None: Ticket={ticket}")
return False
if result.retcode == mt5.TRADE_RETCODE_DONE:
if result.retcode == _get_mt5().TRADE_RETCODE_DONE:
logger.info(f"平仓成功: Ticket {ticket}")
return True
else:
core/risk/controller.py
+6 -2
View File
@@ -1,5 +1,6 @@
from core.risk.market_state import MarketStateAnalyzer
from core.risk.position import PositionManager
from logger import logger
class RiskController:
@@ -16,8 +17,11 @@ class RiskController:
def process_trading_signal(self, direction, current_price, signal_strength=0.0):
return self.position_manager.open_position(direction, current_price, signal_strength)
def monitor_positions(self, current_price, dry_run=False):
self.position_manager.monitor_positions(current_price, dry_run)
def monitor_positions(self, current_price, dry_run=False, weighted_signal=0.0):
self.position_manager.monitor_positions(current_price, dry_run, weighted_signal)
# 对冲摘要日志
if self.position_manager.hedge_manager and self.position_manager.hedge_manager.active_hedges > 0:
logger.info(f"🔒 活跃对冲: {self.position_manager.hedge_manager.active_hedges} 个")
def sync_state(self):
self.position_manager.update_equity()
core/risk/hedge.py
+244
View File
@@ -0,0 +1,244 @@
"""
对冲管理器 — 信号对冲 + 回撤锁仓
两种对冲模式：
1. 信号对冲：持仓方向与当前加权信号相反 → 开反向单保护
2. 回撤锁仓：持仓亏损超过阈值 → 锁仓防进一步亏损
解锁条件：
- 信号回到中性/同向 → 平对冲单
- 对冲单自身止盈 → 平对冲单
- 原始仓位平仓 → 同时平对冲单
"""
from logger import logger
class HedgeManager:
"""对冲管理器 — 与 PositionManager 协作，不直接操作仓位"""
def __init__(self, position_manager, config: dict = None):
self._pm = position_manager # PositionManager 引用
cfg = config or {}
# ── 信号对冲 ──
self.signal_hedge_enabled = cfg.get("signal_hedge_enabled", True)
# 加权信号绝对值超过此阈值才触发对冲（避免噪音）
self.signal_hedge_threshold = cfg.get("signal_hedge_threshold", 2.0)
# 对冲比例：0.5=开一半手数，1.0=等量对冲
self.signal_hedge_ratio = cfg.get("signal_hedge_ratio", 0.5)
# 信号回到中性以下（绝对值<此值）时解锁
self.signal_unhedge_threshold = cfg.get("signal_unhedge_threshold", 1.0)
# ── 回撤锁仓 ──
self.drawdown_hedge_enabled = cfg.get("drawdown_hedge_enabled", True)
# 浮亏超过此比例触发锁仓
self.drawdown_hedge_pct = cfg.get("drawdown_hedge_pct", -0.003) # -0.3%
# 锁仓比例
self.drawdown_hedge_ratio = cfg.get("drawdown_hedge_ratio", 1.0) # 100%
# ── 对冲单元管理 ──
# 每个对冲单元: {original_ticket, hedge_ticket, hedge_type, hedge_volume}
self._hedge_units = []
# ── 对冲单止盈 ──
self.hedge_take_profit_pct = cfg.get("hedge_take_profit_pct", 0.005) # 0.5%
# ── 每日最大对冲次数 ──
self.max_hedges_per_day = cfg.get("max_hedges_per_day", 5)
self._daily_hedge_count = 0
# ═══════════════════════════════════════════════════════════
# 决策接口
# ═══════════════════════════════════════════════════════════
def evaluate(self, current_signal: float, current_price: dict) -> list:
"""
每轮主循环调用：评估所有持仓是否需要对冲/解锁。
返回需要执行的操作列表: [("hedge", position, reason), ("unhedge", unit, reason), ...]
"""
actions = []
# ① 先检查是否需要解锁已有对冲
for unit in list(self._hedge_units):
original = self._find_position(unit["original_ticket"])
if original is None:
# 原始仓位已平 → 解锁对冲
actions.append(("unhedge", unit, "原始仓位已平仓"))
continue
if self._should_unhedge(unit, original, current_signal, current_price):
actions.append(("unhedge", unit, unit.get("unhedge_reason", "解锁条件满足")))
# ② 再检查是否需要开新对冲
for pos in self._pm.positions:
ticket = pos["ticket"]
# 已有对冲的跳过
if any(u["original_ticket"] == ticket for u in self._hedge_units):
continue
# 信号对冲
if self.signal_hedge_enabled:
reason = self._check_signal_hedge(pos, current_signal)
if reason:
if self._can_hedge():
actions.append(("hedge", pos, reason))
# 回撤锁仓
if self.drawdown_hedge_enabled:
reason = self._check_drawdown_hedge(pos, current_price)
if reason:
if self._can_hedge():
actions.append(("hedge", pos, reason))
return actions
# ═══════════════════════════════════════════════════════════
# 对冲执行
# ═══════════════════════════════════════════════════════════
def execute_hedge(self, position, reason: str, current_price: dict) -> bool:
"""对指定持仓开反向对冲单"""
opposite = "sell" if position["position_type"] == "long" else "buy"
# 确定对冲手数
hedge_type = "signal" if "信号" in reason else "drawdown"
ratio = self.signal_hedge_ratio if hedge_type == "signal" else self.drawdown_hedge_ratio
hedge_volume = position["quantity"] * ratio
# 调取开仓
result = self._pm.data_provider.send_order(
self._pm.symbol, opposite, hedge_volume
)
if result is None:
logger.error(f"对冲开仓失败: {opposite} @ {hedge_volume}手")
return False
try:
order_id = result["order"] if isinstance(result, dict) else result.order
except Exception as e:
logger.error(f"解析对冲单号失败: {e}")
return False
if order_id > 0:
unit = {
"original_ticket": position["ticket"],
"hedge_ticket": order_id,
"hedge_type": hedge_type,
"hedge_volume": hedge_volume,
"hedge_direction": opposite,
"entry_price": current_price.get("last", 0),
"entry_time": current_price.get("time", None),
}
self._hedge_units.append(unit)
self._daily_hedge_count += 1
logger.info(
f"🔒 对冲开仓: Ticket {position['ticket']} → 反向 {opposite} "
f"@{hedge_volume}手 (Ticket {order_id}) | 原因: {reason}"
)
return True
logger.error(f"对冲开仓失败: order_id={order_id}")
return False
def execute_unhedge(self, unit: dict, reason: str) -> bool:
"""平掉对冲单，解锁"""
success = self._pm.data_provider.close_position(
unit["hedge_ticket"], self._pm.symbol, unit["hedge_volume"]
)
if success:
self._hedge_units.remove(unit)
logger.info(
f"🔓 对冲解锁: Ticket {unit['hedge_ticket']} 已平仓 | 原因: {reason}"
)
return True
logger.error(f"解锁平仓失败: Ticket {unit['hedge_ticket']}")
return False
# ═══════════════════════════════════════════════════════════
# 内部检查
# ═══════════════════════════════════════════════════════════
def _check_signal_hedge(self, position, current_signal: float) -> str | None:
"""检查信号是否触发对冲"""
is_long = position["position_type"] == "long"
# 持多 + 信号强烈看空 → 对冲
if is_long and current_signal < -self.signal_hedge_threshold:
return f"信号对冲(持多, 加权={current_signal:.2f} < -{self.signal_hedge_threshold})"
# 持空 + 信号强烈看多 → 对冲
if not is_long and current_signal > self.signal_hedge_threshold:
return f"信号对冲(持空, 加权={current_signal:.2f} > {self.signal_hedge_threshold})"
return None
def _check_drawdown_hedge(self, position, current_price: dict) -> str | None:
"""检查浮亏是否触发锁仓"""
price = current_price.get("last", 0)
if price <= 0:
return None
pnl_pct = self._pm._calculate_pnl_pct(position, price)
if pnl_pct <= self.drawdown_hedge_pct:
return f"回撤锁仓(浮亏={pnl_pct:.2%} <= {self.drawdown_hedge_pct:.2%})"
return None
def _should_unhedge(self, unit, original_pos, current_signal: float, current_price: dict) -> bool:
"""判断是否应该解锁对冲"""
# 条件1：信号回到中性
if abs(current_signal) < self.signal_unhedge_threshold:
unit["unhedge_reason"] = f"信号中性({current_signal:.2f})"
return True
# 条件2：信号与原始仓位同向
is_long = original_pos["position_type"] == "long"
if is_long and current_signal > 0:
unit["unhedge_reason"] = f"信号同向做多({current_signal:.2f})"
return True
if not is_long and current_signal < 0:
unit["unhedge_reason"] = f"信号同向做空({current_signal:.2f})"
return True
# 条件3：对冲单自身止盈
price = current_price.get("last", 0)
if price > 0 and unit.get("entry_price", 0) > 0:
if unit["hedge_direction"] == "sell":
hedge_pnl = (unit["entry_price"] - price) / unit["entry_price"]
else:
hedge_pnl = (price - unit["entry_price"]) / unit["entry_price"]
if hedge_pnl >= self.hedge_take_profit_pct:
unit["unhedge_reason"] = f"对冲单止盈({hedge_pnl:.2%})"
return True
return False
def _can_hedge(self) -> bool:
"""检查是否允许开新对冲（次数限制）"""
if self.max_hedges_per_day <= 0:
return True
return self._daily_hedge_count < self.max_hedges_per_day
def _find_position(self, ticket):
"""在持仓列表中查找指定 ticket 的仓位"""
return next((p for p in self._pm.positions if p["ticket"] == ticket), None)
# ═══════════════════════════════════════════════════════════
# 状态查询
# ═══════════════════════════════════════════════════════════
@property
def active_hedges(self) -> int:
return len(self._hedge_units)
def get_hedge_summary(self) -> dict:
return {
"active_hedges": len(self._hedge_units),
"daily_hedge_count": self._daily_hedge_count,
"units": [
{
"original": u["original_ticket"],
"hedge": u["hedge_ticket"],
"type": u["hedge_type"],
"volume": u["hedge_volume"],
}
for u in self._hedge_units
],
}
core/risk/position.py
+15 -1
View File
@@ -8,6 +8,7 @@ from config import (
RISK_CONFIG_CONST, SIMULATION_CONFIG
)
from core.risk.exit_rules import ExitRuleEngine, ExitContext
from core.risk.hedge import HedgeManager
class PositionManager:
@@ -70,6 +71,10 @@ class PositionManager:
if self._persist_peaks:
self._load_peak_data()
# 对冲管理器
from config import HEDGE_CONFIG
self.hedge_manager = HedgeManager(self, HEDGE_CONFIG)
# ── 仓位计算 ──
def _calculate_position_size(self, capital_to_allocate, current_price):
@@ -180,7 +185,7 @@ class PositionManager:
# ── 持仓监控 ──
def monitor_positions(self, current_price, dry_run=False):
def monitor_positions(self, current_price, dry_run=False, weighted_signal=0.0):
if not self.positions:
return
@@ -231,6 +236,15 @@ class PositionManager:
self.update_equity()
self.cleanup_peak_data()
# ── 对冲评估 ──
if self.positions and self.hedge_manager:
hedge_actions = self.hedge_manager.evaluate(weighted_signal, current_price)
for action, target, reason in hedge_actions:
if action == "hedge":
self.hedge_manager.execute_hedge(target, reason, current_price)
elif action == "unhedge":
self.hedge_manager.execute_unhedge(target, reason)
# ── 盈亏计算 ──
def _calculate_pnl_pct(self, position, current_price_value):
core/utils.py
+42 -34
View File
@@ -1,16 +1,24 @@
import MetaTrader5 as mt5
# 惰性导入 — ARM 环境不装 MetaTrader5
_mt5 = None
def _get_mt5():
global _mt5
if _mt5 is None:
import MetaTrader5 as _mt5_module
return _mt5
import pandas as pd
from logger import setup_logger
logger = setup_logger()
def initialize():
if not mt5.initialize():
logger.error("MT5初始化失败，错误代码：%d", mt5.last_error())
if not _get_mt5().initialize():
logger.error("MT5初始化失败，错误代码：%d", _get_mt5().last_error())
return False
return True
def shutdown():
mt5.shutdown()
_get_mt5().shutdown()
def get_rates(symbol, timeframe, count, start_date=None, end_date=None):
"""
@@ -23,21 +31,21 @@ def get_rates(symbol, timeframe, count, start_date=None, end_date=None):
- end_date: 结束日期 (格式: "YYYY-MM-DD" 或 datetime对象)
"""
# 检查MT5连接状态
if not mt5.terminal_info():
if not _get_mt5().terminal_info():
logger.warning("MT5终端未连接，尝试重新连接...")
if not initialize():
logger.error("MT5重新连接失败")
return None
# 检查交易品种是否可用
symbol_info = mt5.symbol_info(symbol)
symbol_info = _get_mt5().symbol_info(symbol)
if symbol_info is None:
logger.error(f"交易品种 {symbol} 不可用")
return None
if not symbol_info.visible:
logger.info(f"交易品种 {symbol} 不可见，尝试启用...")
if not mt5.symbol_select(symbol, True):
if not _get_mt5().symbol_select(symbol, True):
logger.error(f"无法启用交易品种 {symbol}")
return None
if start_date is not None and end_date is not None:
@@ -58,7 +66,7 @@ def get_rates(symbol, timeframe, count, start_date=None, end_date=None):
end_utc = end_date.replace(hour=23, minute=59, second=59, tzinfo=utc_timezone)
try:
rates = mt5.copy_rates_range(symbol, timeframe, start_utc, end_utc)
rates = _get_mt5().copy_rates_range(symbol, timeframe, start_utc, end_utc)
if rates is None:
logger.info(f"获取{symbol}从{start_date.date()}到{end_date.date()}的历史数据失败")
return None
@@ -66,14 +74,14 @@ def get_rates(symbol, timeframe, count, start_date=None, end_date=None):
except Exception as e:
logger.error(f"使用日期范围获取数据失败: {e}")
logger.info("回退到使用数据量获取数据")
rates = mt5.copy_rates_from_pos(symbol, timeframe, 0, count)
rates = _get_mt5().copy_rates_from_pos(symbol, timeframe, 0, count)
if rates is None:
logger.info(f"获取{symbol}历史数据失败")
return None
logger.debug(f"成功获取{symbol}历史数据，共{len(rates)}条")
else:
# 使用数据量获取数据
rates = mt5.copy_rates_from_pos(symbol, timeframe, 0, count)
rates = _get_mt5().copy_rates_from_pos(symbol, timeframe, 0, count)
if rates is None:
logger.info(f"获取{symbol}历史数据失败")
return None
@@ -82,39 +90,39 @@ def get_rates(symbol, timeframe, count, start_date=None, end_date=None):
return rates
def has_open_position(symbol):
positions = mt5.positions_get(symbol=symbol)
positions = _get_mt5().positions_get(symbol=symbol)
return positions is not None and len(positions) > 0
def close_all(symbol):
positions = mt5.positions_get(symbol=symbol)
positions = _get_mt5().positions_get(symbol=symbol)
if positions is None:
return
for pos in positions:
request = {
"action": mt5.TRADE_ACTION_DEAL,
"action": _get_mt5().TRADE_ACTION_DEAL,
"position": pos.ticket,
"symbol": symbol,
"volume": pos.volume,
"type": mt5.ORDER_TYPE_SELL if pos.type == 0 else mt5.ORDER_TYPE_BUY,
"price": mt5.symbol_info_tick(symbol).bid if pos.type == 0 else mt5.symbol_info_tick(symbol).ask,
"type": _get_mt5().ORDER_TYPE_SELL if pos.type == 0 else _get_mt5().ORDER_TYPE_BUY,
"price": _get_mt5().symbol_info_tick(symbol).bid if pos.type == 0 else _get_mt5().symbol_info_tick(symbol).ask,
"deviation": 20,
"magic": 234000,
"comment": "Close position",
"type_filling": mt5.ORDER_FILLING_RETURN,
"type_filling": _get_mt5().ORDER_FILLING_RETURN,
}
mt5.order_send(request)
_get_mt5().order_send(request)
def send_order(symbol, order_type, volume=0.01):
symbol_info_tick = mt5.symbol_info_tick(symbol)
symbol_info_tick = _get_mt5().symbol_info_tick(symbol)
if symbol_info_tick is None:
logger.error(f"无法获取{symbol}行情")
return None
price = symbol_info_tick.ask if order_type == "buy" else symbol_info_tick.bid
order_type_mt5 = mt5.ORDER_TYPE_BUY if order_type == "buy" else mt5.ORDER_TYPE_SELL
order_type_mt5 = _get_mt5().ORDER_TYPE_BUY if order_type == "buy" else _get_mt5().ORDER_TYPE_SELL
request = {
"action": mt5.TRADE_ACTION_DEAL,
"action": _get_mt5().TRADE_ACTION_DEAL,
"symbol": symbol,
"volume": volume,
"type": order_type_mt5,
@@ -122,11 +130,11 @@ def send_order(symbol, order_type, volume=0.01):
"deviation": 20,
"magic": 234000,
"comment": f"{order_type} order",
"type_filling": mt5.ORDER_FILLING_RETURN,
"type_filling": _get_mt5().ORDER_FILLING_RETURN,
}
result = mt5.order_send(request)
if result.retcode != mt5.TRADE_RETCODE_DONE:
result = _get_mt5().order_send(request)
if result.retcode != _get_mt5().TRADE_RETCODE_DONE:
logger.error(f"下单失败，retcode={result.retcode}")
return None
else:
@@ -137,7 +145,7 @@ def close_position(ticket, symbol, volume):
"""根据ticket平掉一个特定的仓位"""
# In MT5, you close a position by creating an opposite order.
# We need to get the position details first.
positions = mt5.positions_get(ticket=ticket)
positions = _get_mt5().positions_get(ticket=ticket)
if not positions:
logger.error(f"无法找到ticket为 {ticket} 的持仓")
return False
@@ -145,19 +153,19 @@ def close_position(ticket, symbol, volume):
pos = positions[0] # positions_get returns a tuple of objects
request = {
"action": mt5.TRADE_ACTION_DEAL,
"action": _get_mt5().TRADE_ACTION_DEAL,
"position": pos.ticket,
"symbol": symbol,
"volume": volume,
"type": mt5.ORDER_TYPE_SELL if pos.type == 0 else mt5.ORDER_TYPE_BUY, # pos.type == 0 is a BUY position
"price": mt5.symbol_info_tick(symbol).bid if pos.type == 0 else mt5.symbol_info_tick(symbol).ask,
"type": _get_mt5().ORDER_TYPE_SELL if pos.type == 0 else _get_mt5().ORDER_TYPE_BUY, # pos.type == 0 is a BUY position
"price": _get_mt5().symbol_info_tick(symbol).bid if pos.type == 0 else _get_mt5().symbol_info_tick(symbol).ask,
"deviation": 20,
"magic": 234000,
"comment": f"Close position {ticket}",
"type_filling": mt5.ORDER_FILLING_RETURN,
"type_filling": _get_mt5().ORDER_FILLING_RETURN,
}
result = mt5.order_send(request)
if result.retcode != mt5.TRADE_RETCODE_DONE:
result = _get_mt5().order_send(request)
if result.retcode != _get_mt5().TRADE_RETCODE_DONE:
logger.error(f"平仓失败 ticket {ticket}, retcode={result.retcode}")
return False
else:
@@ -168,25 +176,25 @@ def get_current_price(symbol):
"""获取当前价格"""
try:
# 检查MT5连接状态
if not mt5.terminal_info():
if not _get_mt5().terminal_info():
logger.warning("MT5终端未连接，尝试重新连接...")
if not initialize():
logger.error("MT5重新连接失败")
return None
# 检查交易品种是否可用
symbol_info = mt5.symbol_info(symbol)
symbol_info = _get_mt5().symbol_info(symbol)
if symbol_info is None:
logger.error(f"交易品种 {symbol} 不可用")
return None
if not symbol_info.visible:
logger.info(f"交易品种 {symbol} 不可见，尝试启用...")
if not mt5.symbol_select(symbol, True):
if not _get_mt5().symbol_select(symbol, True):
logger.error(f"无法启用交易品种 {symbol}")
return None
tick = mt5.symbol_info_tick(symbol)
tick = _get_mt5().symbol_info_tick(symbol)
if tick is None:
logger.error(f"无法获取 {symbol} 的价格信息")
return None
execution/realtime_trader.py
+1 -1
View File
@@ -83,7 +83,7 @@ class RealtimeTrader:
else:
logger.info(f"{direction}交易执行成功")
self.risk_controller.monitor_positions(current_price)
self.risk_controller.monitor_positions(current_price, weighted_signal=weighted_signal_sum)
# --- 状态汇总日志 ---
logger.info("--- 财务状况更新 ---")
run/backtest.py
+67 -9
View File
@@ -1,30 +1,88 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""回测入口 — python -m run.backtest"""
"""回测入口 — 支持远程/本地两种数据源"""
import sys
import os
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from core.utils import get_rates, initialize, shutdown
import numpy as np
from datetime import datetime
from core.data.remote import RemoteDataProvider
from execution.backtest import BacktestEngine
from config import SYMBOL, TIMEFRAME, BACKTEST_COUNT, BACKTEST_START_DATE, BACKTEST_END_DATE, USE_DATE_RANGE
from config import (
SYMBOL, TIMEFRAME, BACKTEST_COUNT,
BACKTEST_START_DATE, BACKTEST_END_DATE, USE_DATE_RANGE,
REMOTE_SERVER_HOST, REMOTE_SERVER_PORT, DATA_PROVIDER_MODE,
)
from logger import logger
# MT5 rates 结构化数组 dtype
MT5_RATES_DTYPE = np.dtype([
('time', 'i8'),
('open', 'f8'),
('high', 'f8'),
('low', 'f8'),
('close', 'f8'),
('tick_volume', 'i8'),
('spread', 'i4'),
('real_volume', 'i8'),
])
def fetch_remote_rates(symbol, timeframe, count):
"""通过远程 API 获取历史 K 线，转成 MT5 兼容的 numpy 数组"""
logger.info(f"使用远程数据源 {REMOTE_SERVER_HOST}:{REMOTE_SERVER_PORT}")
remote = RemoteDataProvider(host=REMOTE_SERVER_HOST, port=REMOTE_SERVER_PORT)
if not remote.initialize():
logger.error("远程API初始化失败")
return None
rates_data = remote.get_historical_data(symbol, timeframe, count)
remote.shutdown()
if not rates_data:
logger.error(f"远程API获取 {symbol} TF{timeframe} 历史数据失败")
return None
# 转成 MT5 的 numpy 结构化数组格式
records = []
for r in rates_data:
records.append((
r['time'],
r['open'],
r['high'],
r['low'],
r['close'],
r.get('tick_volume', 0),
r.get('spread', 0),
r.get('real_volume', 0),
))
rates = np.array(records, dtype=MT5_RATES_DTYPE)
logger.info(f"远程获取数据: {len(rates)} 条")
return rates
def main():
print("=" * 60)
print("MT5 智能交易系统 - 回测")
print(f"数据源: {DATA_PROVIDER_MODE.upper()}")
print(f"品种: {SYMBOL} | 周期: M{TIMEFRAME} | 数据量: {BACKTEST_COUNT}")
print("=" * 60)
try:
initialize()
if USE_DATE_RANGE:
rates = get_rates(SYMBOL, TIMEFRAME, BACKTEST_COUNT,
BACKTEST_START_DATE, BACKTEST_END_DATE)
if DATA_PROVIDER_MODE == "remote":
rates = fetch_remote_rates(SYMBOL, TIMEFRAME, BACKTEST_COUNT)
else:
rates = get_rates(SYMBOL, TIMEFRAME, BACKTEST_COUNT)
shutdown()
from core.utils import initialize, shutdown, get_rates
initialize()
if USE_DATE_RANGE:
rates = get_rates(SYMBOL, TIMEFRAME, BACKTEST_COUNT,
BACKTEST_START_DATE, BACKTEST_END_DATE)
else:
rates = get_rates(SYMBOL, TIMEFRAME, BACKTEST_COUNT)
shutdown()
if rates is None or len(rates) == 0:
logger.error("未能获取历史数据，回测终止。")
run/realtime.py
+17 -4
View File
@@ -1,6 +1,6 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""实时交易入口 — python -m run.realtime"""
"""实时交易入口 — 支持远程/本地两种数据源"""
import sys
import os
@@ -9,8 +9,11 @@ from datetime import datetime
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from execution.realtime_trader import RealtimeTrader
from core.data import LiveDataProvider, DryRunDataProvider
from config import REALTIME_CONFIG, INITIAL_CAPITAL
from core.data import LiveDataProvider, DryRunDataProvider, RemoteDataProvider
from config import (
REALTIME_CONFIG, INITIAL_CAPITAL,
REMOTE_SERVER_HOST, REMOTE_SERVER_PORT, DATA_PROVIDER_MODE,
)
from logger import setup_logger
@@ -23,10 +26,20 @@ def main():
print("=" * 60)
print("MT5 智能交易系统 - 实时交易")
print(f"启动时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
print(f"数据源: {DATA_PROVIDER_MODE.upper()}")
print(f"运行模式: {'模拟运行' if is_dry_run else '实盘交易'}")
print("=" * 60)
if is_dry_run:
if DATA_PROVIDER_MODE == "remote":
data_provider = RemoteDataProvider(host=REMOTE_SERVER_HOST, port=REMOTE_SERVER_PORT)
if not data_provider.initialize():
print("❌ 远程 MT5 API 连接失败，请检查 Windows 端服务是否运行")
return
print(f"✅ 已连接远程 MT5 {REMOTE_SERVER_HOST}:{REMOTE_SERVER_PORT}")
acct = data_provider.get_account_info()
if acct:
print(f" 账号: {acct.get('login','?')} 余额: ${acct.get('balance','?'):.2f}")
elif is_dry_run:
data_provider = DryRunDataProvider(initial_equity=INITIAL_CAPITAL)
else:
print("WARNING: 即将启动实盘交易模式！")