import MetaTrader5 as mt5 import pandas as pd import numpy as np from datetime import datetime, timedelta import pytz import time import threading import queue import tkinter as tk from tkinter import ttk, scrolledtext # ==================== 全局配置 ==================== SYMBOLS = ["EURUSD", "GBPUSD", "USDJPY", "USDCHF", "AUDUSD", "USDCAD", "NZDUSD"] MAGIC_NUMBER = 99990000 LOT_SIZE = 0.1 PIP_MULTIPLIERS = { "EURUSD": 0.0001, "GBPUSD": 0.0001, "USDJPY": 0.01, "USDCHF": 0.0001, "AUDUSD": 0.0001, "USDCAD": 0.0001, "NZDUSD": 0.0001 } # 策略参数 LOOKBACK = 20 SL_PIPS = 15 TP_PIPS = 45 MIN_BREAKOUT_PIPS = 5 # 交易时间（北京时间） TRADE_START_HOUR = 8 # 15:00 TRADE_END_HOUR = 1 # 次日01:00 # ==================== 交易引擎 ==================== class TradingEngine: def __init__(self, log_callback=None, signal_callback=None, position_callback=None, countdown_callback=None): self.log_callback = log_callback self.signal_callback = signal_callback self.position_callback = position_callback self.countdown_callback = countdown_callback self.running = False self.signals_history = deque(maxlen=100) self.last_check_hour = {} # 记录每个货币对最后检查的小时 self.data_queue = queue.Queue() def log(self, msg, level="INFO"): timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S") log_msg = f"[{timestamp}] [{level}] {msg}" if self.log_callback: self.log_callback(log_msg) print(log_msg) def get_ema(self, series, period): return series.ewm(span=period, adjust=False).mean() def get_pip_multiplier(self, symbol): return PIP_MULTIPLIERS.get(symbol, 0.0001) def is_trading_time(self): """检查是否在北京时间交易时段内""" now = datetime.now() hour = now.hour if TRADE_START_HOUR <= hour <= 23: return True elif 0 <= hour < TRADE_END_HOUR: return True return False def get_next_hour_close_time(self): """计算下一个整点时间（H1收盘时间）""" now = datetime.now() # 下一个整点 next_hour = now.replace(minute=0, second=0, microsecond=0) + timedelta(hours=1) return next_hour def get_seconds_to_next_hour(self): """计算距离下一个整点还有多少秒""" now = datetime.now() next_hour = self.get_next_hour_close_time() seconds = (next_hour - now).total_seconds() return max(1, int(seconds)) def get_mt5_timeframe(self, timeframe_str): """获取MT5时间框架""" timeframes = { 'D1': mt5.TIMEFRAME_D1, 'H1': mt5.TIMEFRAME_H1 } return timeframes.get(timeframe_str) def fetch_data(self, symbol, timeframe, count=200): """获取K线数据""" tf = self.get_mt5_timeframe(timeframe) rates = mt5.copy_rates_from_pos(symbol, tf, 0, count) if rates is None or len(rates) == 0: self.log(f"{symbol} {timeframe} 数据获取失败", "ERROR") return None df = pd.DataFrame(rates) df['time'] = pd.to_datetime(df['time'], unit='s') df.set_index('time', inplace=True) return df def should_check_signal(self, symbol): """判断是否应该检查信号（新H1 K线形成时）""" current_hour = datetime.now().replace(minute=0, second=0, microsecond=0) # 如果是第一次检查这个货币对 if symbol not in self.last_check_hour: self.last_check_hour[symbol] = current_hour return True # 如果当前小时和上次检查的小时不同，说明新K线已形成 if current_hour > self.last_check_hour[symbol]: self.last_check_hour[symbol] = current_hour return True return False def check_signal(self, symbol): """检查单个货币对的交易信号""" try: self.log(f"{symbol} H1 K线收盘，开始分析信号...", "INFO") # 获取日线数据 d1_df = self.fetch_data(symbol, 'D1', 100) if d1_df is None: return None # 计算日线EMA60和斜率 d1_df['ema60'] = self.get_ema(d1_df['close'], 60) d1_df['slope'] = d1_df['ema60'].diff(3) # 获取小时线数据 h1_df = self.fetch_data(symbol, 'H1', LOOKBACK + 50) if h1_df is None: return None # 映射日线数据到小时线 h1_df['date'] = h1_df.index.date d1_df['date'] = d1_df.index.date ema_map = d1_df.groupby('date')['ema60'].last() slope_map = d1_df.groupby('date')['slope'].last() h1_df['d1_ema60'] = h1_df['date'].map(ema_map) h1_df['d1_slope'] = h1_df['date'].map(slope_map) h1_df = h1_df.dropna(subset=['d1_ema60']) if len(h1_df) < LOOKBACK + 2: return None # 获取最新数据（刚收盘的H1 K线） curr_bar = h1_df.iloc[-1] prev_bar = h1_df.iloc[-2] # 日志记录关键数据 pip_mult = self.get_pip_multiplier(symbol) self.log(f"{symbol} 收盘价: {curr_bar['close']:.5f}, " f"EMA60: {curr_bar['d1_ema60']:.5f}, " f"斜率: {curr_bar['d1_slope']:.6f}, " f"箱体高: {h1_df['high'].iloc[-LOOKBACK-1:-1].max():.5f}, " f"箱体低: {h1_df['low'].iloc[-LOOKBACK-1:-1].min():.5f}", "DEBUG") # D1趋势判定 d1_trend = 0 if curr_bar['d1_slope'] > 0.0001 and curr_bar['close'] > curr_bar['d1_ema60']: d1_trend = 1 self.log(f"{symbol} 日线趋势: 多头", "DEBUG") elif curr_bar['d1_slope'] < -0.0001 and curr_bar['close'] < curr_bar['d1_ema60']: d1_trend = -1 self.log(f"{symbol} 日线趋势: 空头", "DEBUG") else: self.log(f"{symbol} 日线趋势: 无方向", "DEBUG") return None # 计算前N根的高低点（不包括当前K线） lookback_high = h1_df['high'].iloc[-LOOKBACK-1:-1].max() lookback_low = h1_df['low'].iloc[-LOOKBACK-1:-1].min() signal = None # 多头突破 if d1_trend == 1: breakout_pips = (curr_bar['close'] - lookback_high) / pip_mult self.log(f"{symbol} 多头突破检测: 收盘价{curr_bar['close']:.5f} vs 阻力{lookback_high:.5f}, " f"突破{breakout_pips:.1f}点", "DEBUG") if breakout_pips >= MIN_BREAKOUT_PIPS: if prev_bar['close'] <= lookback_high * 1.0005: signal = { 'symbol': symbol, 'direction': 'buy', 'entry_price': mt5.symbol_info_tick(symbol).ask, 'stop_loss': None, 'take_profit': None, 'breakout_pips': breakout_pips, 'lookback_high': lookback_high } self.log(f"{symbol} 多头信号确认! 前根收盘{prev_bar['close']:.5f}在阻力位附近", "INFO") # 空头突破 elif d1_trend == -1: breakout_pips = (lookback_low - curr_bar['close']) / pip_mult self.log(f"{symbol} 空头突破检测: 收盘价{curr_bar['close']:.5f} vs 支撑{lookback_low:.5f}, " f"突破{breakout_pips:.1f}点", "DEBUG") if breakout_pips >= MIN_BREAKOUT_PIPS: if prev_bar['close'] >= lookback_low * 0.9995: signal = { 'symbol': symbol, 'direction': 'sell', 'entry_price': mt5.symbol_info_tick(symbol).bid, 'stop_loss': None, 'take_profit': None, 'breakout_pips': breakout_pips, 'lookback_low': lookback_low } self.log(f"{symbol} 空头信号确认! 前根收盘{prev_bar['close']:.5f}在支撑位附近", "INFO") if signal: # 计算止损止盈 pip_mult = self.get_pip_multiplier(symbol) if signal['direction'] == 'buy': signal['stop_loss'] = signal['entry_price'] - SL_PIPS * pip_mult signal['take_profit'] = signal['entry_price'] + TP_PIPS * pip_mult else: signal['stop_loss'] = signal['entry_price'] + SL_PIPS * pip_mult signal['take_profit'] = signal['entry_price'] - TP_PIPS * pip_mult self.log(f"{symbol} 产生{signal['direction']}信号! " f"入场: {signal['entry_price']:.5f}, " f"止损: {signal['stop_loss']:.5f}, " f"止盈: {signal['take_profit']:.5f}", "SIGNAL") # 添加到信号历史 signal_record = { 'time': datetime.now().strftime("%Y-%m-%d %H:%M:%S"), 'symbol': symbol, 'direction': signal['direction'], 'entry_price': signal['entry_price'], 'breakout_pips': breakout_pips } self.signals_history.append(signal_record) if self.signal_callback: self.signal_callback(signal_record) return signal else: self.log(f"{symbol} 无突破信号", "DEBUG") return None except Exception as e: self.log(f"{symbol} 信号检测异常: {str(e)}", "ERROR") import traceback self.log(traceback.format_exc(), "ERROR") return None def check_existing_position(self, symbol, direction): """检查是否已存在同方向同货币的持仓""" positions = mt5.positions_get(symbol=symbol) if positions: for pos in positions: if pos.magic == MAGIC_NUMBER: if (direction == 'buy' and pos.type == mt5.POSITION_TYPE_BUY) or \ (direction == 'sell' and pos.type == mt5.POSITION_TYPE_SELL): return True return False def execute_trade(self, signal): """执行交易""" try: symbol = signal['symbol'] direction = signal['direction'] # 检查现有持仓 if self.check_existing_position(symbol, direction): self.log(f"{symbol} 已存在同方向持仓，跳过", "INFO") return False # 准备订单请求 request = { "action": mt5.TRADE_ACTION_DEAL, "symbol": symbol, "volume": LOT_SIZE, "type": mt5.ORDER_TYPE_BUY if direction == 'buy' else mt5.ORDER_TYPE_SELL, "price": signal['entry_price'], "sl": signal['stop_loss'], "tp": signal['take_profit'], "deviation": 10, "magic": MAGIC_NUMBER, "comment": "Breakout Strategy", "type_time": mt5.ORDER_TIME_GTC, "type_filling": mt5.ORDER_FILLING_IOC, } # 发送订单 result = mt5.order_send(request) if result.retcode != mt5.TRADE_RETCODE_DONE: self.log(f"{symbol} 订单失败: {result.comment}", "ERROR") return False self.log(f"{symbol} {direction.upper()} 订单成功! " f"单号: {result.order}, " f"入场: {signal['entry_price']:.5f}, " f"止损: {signal['stop_loss']:.5f}, " f"止盈: {signal['take_profit']:.5f}", "SUCCESS") return True except Exception as e: self.log(f"{symbol} 执行交易异常: {str(e)}", "ERROR") return False def update_positions(self): """更新持仓信息""" positions_data = [] positions = mt5.positions_get() if positions: for pos in positions: if pos.magic == MAGIC_NUMBER: profit = pos.profit positions_data.append({ 'symbol': pos.symbol, 'type': 'BUY' if pos.type == mt5.POSITION_TYPE_BUY else 'SELL', 'volume': pos.volume, 'open_price': pos.price_open, 'current_price': pos.price_current, 'sl': pos.sl, 'tp': pos.tp, 'profit': profit, 'time': datetime.fromtimestamp(pos.time).strftime("%m-%d %H:%M") }) if self.position_callback: self.position_callback(positions_data) return positions_data def scan_all_symbols(self): """扫描所有货币对（仅在H1收盘时）""" if not self.is_trading_time(): self.log("当前非交易时段，跳过扫描", "DEBUG") return check_count = 0 for symbol in SYMBOLS: # 检查是否需要分析（新H1 K线形成） if self.should_check_signal(symbol): check_count += 1 self.log(f"{symbol} 新H1 K线形成，开始分析", "INFO") # 确保该品种可用 mt5.symbol_select(symbol, True) signal = self.check_signal(symbol) if signal: self.execute_trade(signal) time.sleep(0.5) # 避免请求过于频繁 if check_count > 0: self.log(f"本轮共分析了 {check_count} 个货币对", "INFO") self.update_positions() def run(self): """主运行循环""" self.running = True self.log("交易引擎启动 - H1 K线收盘时自动分析", "INFO") self.log(f"交易时间: 北京时间 {TRADE_START_HOUR}:00 - 次日{TRADE_END_HOUR}:00", "INFO") while self.running: try: # 检查MT5连接 if not mt5.terminal_info(): self.log("MT5连接断开，尝试重连...", "WARNING") mt5.shutdown() time.sleep(5) if not mt5.initialize(): self.log("MT5重新初始化失败", "ERROR") time.sleep(30) continue # 计算距离下一个整点的秒数 seconds_to_next_hour = self.get_seconds_to_next_hour() # 如果距离整点小于5秒，执行扫描 if seconds_to_next_hour <= 5: self.log(f"距离H1收盘还有 {seconds_to_next_hour} 秒，准备扫描", "INFO") time.sleep(seconds_to_next_hour + 1) # 等待K线完全收盘 self.scan_all_symbols() else: # 倒计时更新 for i in range(min(seconds_to_next_hour, 60), 0, -1): if not self.running: break # 每60秒更新一次倒计时显示 if i % 60 == 0 or i <= 10: minutes = i // 60 secs = i % 60 countdown_str = f"距下次分析: {minutes:02d}:{secs:02d}" self.data_queue.put({'type': 'countdown', 'value': countdown_str}) if self.countdown_callback: self.countdown_callback(countdown_str) time.sleep(1) except Exception as e: self.log(f"运行异常: {str(e)}", "ERROR") time.sleep(10) def stop(self): """停止引擎""" self.running = False self.log("交易引擎停止", "INFO") # ==================== GUI界面 ==================== class TradingGUI: def __init__(self): self.root = tk.Tk() self.root.title("D1/H1箱体突破策略v1") self.root.geometry("1200x800") # 设置样式 self.setup_styles() # 创建引擎 self.engine = TradingEngine( log_callback=self.add_log, signal_callback=self.add_signal, position_callback=self.update_positions_display, countdown_callback=self.update_countdown ) # 创建界面 self.create_widgets() # 启动引擎线程 self.engine_thread = None self.start_engine() # 定时更新UI self.update_ui() def setup_styles(self): """设置样式""" style = ttk.Style() style.configure("Header.TLabel", font=("Arial", 12, "bold")) style.configure("Signal.TLabel", font=("Arial", 10)) style.configure("Profit.TLabel", font=("Arial", 10, "bold")) def create_widgets(self): """创建界面组件""" # 主框架 main_frame = ttk.Frame(self.root, padding="10") main_frame.grid(row=0, column=0, sticky=(tk.W, tk.E, tk.N, tk.S)) # 配置网格权重 self.root.columnconfigure(0, weight=1) self.root.rowconfigure(0, weight=1) main_frame.columnconfigure(1, weight=1) main_frame.rowconfigure(1, weight=1) # 标题栏 title_frame = ttk.Frame(main_frame) title_frame.grid(row=0, column=0, columnspan=2, sticky=(tk.W, tk.E), pady=(0, 10)) ttk.Label(title_frame, text="H1突破策略自动交易系统", style="Header.TLabel").pack(side=tk.LEFT) # 状态指示器 self.status_label = ttk.Label(title_frame, text="● 运行中", foreground="green", font=("Arial", 10, "bold")) self.status_label.pack(side=tk.LEFT, padx=20) # 倒计时 self.countdown_label = ttk.Label(title_frame, text="距下次分析: --:--", font=("Arial", 10, "bold")) self.countdown_label.pack(side=tk.LEFT, padx=10) # 交易时间指示 self.time_status = ttk.Label(title_frame, text="", font=("Arial", 10)) self.time_status.pack(side=tk.LEFT, padx=10) # 下次分析时间 self.next_check_label = ttk.Label(title_frame, text="", font=("Arial", 9)) self.next_check_label.pack(side=tk.LEFT, padx=10) # 控制按钮 ttk.Button(title_frame, text="停止", command=self.stop_engine).pack(side=tk.RIGHT, padx=5) ttk.Button(title_frame, text="启动", command=self.start_engine).pack(side=tk.RIGHT, padx=5) # 左侧面板 - 信号和持仓 left_frame = ttk.Frame(main_frame) left_frame.grid(row=1, column=0, sticky=(tk.W, tk.E, tk.N, tk.S), padx=(0, 5)) # 信号历史 signal_frame = ttk.LabelFrame(left_frame, text="信号历史", padding="5") signal_frame.grid(row=0, column=0, sticky=(tk.W, tk.E, tk.N, tk.S), pady=(0, 5)) self.signal_tree = ttk.Treeview(signal_frame, columns=('time', 'symbol', 'direction', 'price', 'breakout'), show='headings', height=10) self.signal_tree.heading('time', text='时间') self.signal_tree.heading('symbol', text='货币对') self.signal_tree.heading('direction', text='方向') self.signal_tree.heading('price', text='价格') self.signal_tree.heading('breakout', text='突破点数') self.signal_tree.column('time', width=150) self.signal_tree.column('symbol', width=80) self.signal_tree.column('direction', width=60) self.signal_tree.column('price', width=80) self.signal_tree.column('breakout', width=80) scrollbar = ttk.Scrollbar(signal_frame, orient=tk.VERTICAL, command=self.signal_tree.yview) self.signal_tree.configure(yscrollcommand=scrollbar.set) self.signal_tree.grid(row=0, column=0, sticky=(tk.W, tk.E, tk.N, tk.S)) scrollbar.grid(row=0, column=1, sticky=(tk.N, tk.S)) signal_frame.columnconfigure(0, weight=1) signal_frame.rowconfigure(0, weight=1) # 持仓信息 position_frame = ttk.LabelFrame(left_frame, text="当前持仓", padding="5") position_frame.grid(row=1, column=0, sticky=(tk.W, tk.E, tk.N, tk.S)) self.position_tree = ttk.Treeview(position_frame, columns=('symbol', 'type', 'volume', 'open', 'current', 'sl', 'tp', 'profit', 'time'), show='headings', height=8) self.position_tree.heading('symbol', text='货币对') self.position_tree.heading('type', text='类型') self.position_tree.heading('volume', text='手数') self.position_tree.heading('open', text='开仓价') self.position_tree.heading('current', text='当前价') self.position_tree.heading('sl', text='止损') self.position_tree.heading('tp', text='止盈') self.position_tree.heading('profit', text='盈亏') self.position_tree.heading('time', text='时间') columns_width = {'symbol': 70, 'type': 50, 'volume': 50, 'open': 80, 'current': 80, 'sl': 70, 'tp': 70, 'profit': 80, 'time': 90} for col, width in columns_width.items(): self.position_tree.column(col, width=width) pos_scrollbar = ttk.Scrollbar(position_frame, orient=tk.VERTICAL, command=self.position_tree.yview) self.position_tree.configure(yscrollcommand=pos_scrollbar.set) self.position_tree.grid(row=0, column=0, sticky=(tk.W, tk.E, tk.N, tk.S)) pos_scrollbar.grid(row=0, column=1, sticky=(tk.N, tk.S)) position_frame.columnconfigure(0, weight=1) position_frame.rowconfigure(0, weight=1) left_frame.columnconfigure(0, weight=1) left_frame.rowconfigure(1, weight=1) # 右侧面板 - 日志 right_frame = ttk.Frame(main_frame) right_frame.grid(row=1, column=1, sticky=(tk.W, tk.E, tk.N, tk.S)) log_frame = ttk.LabelFrame(right_frame, text="运行日志", padding="5") log_frame.grid(row=0, column=0, sticky=(tk.W, tk.E, tk.N, tk.S)) self.log_text = scrolledtext.ScrolledText(log_frame, width=70, height=35, font=("Consolas", 9)) self.log_text.grid(row=0, column=0, sticky=(tk.W, tk.E, tk.N, tk.S)) # 日志控制 log_control = ttk.Frame(log_frame) log_control.grid(row=1, column=0, sticky=(tk.W, tk.E), pady=5) ttk.Button(log_control, text="清空日志", command=self.clear_log).pack(side=tk.RIGHT) self.show_debug_var = tk.BooleanVar(value=True) ttk.Checkbutton(log_control, text="显示调试信息", variable=self.show_debug_var).pack(side=tk.RIGHT, padx=10) log_frame.columnconfigure(0, weight=1) log_frame.rowconfigure(0, weight=1) right_frame.columnconfigure(0, weight=1) right_frame.rowconfigure(0, weight=1) main_frame.columnconfigure(0, weight=1) main_frame.columnconfigure(1, weight=2) main_frame.rowconfigure(1, weight=1) def add_log(self, msg): """添加日志""" if "[DEBUG]" in msg and not self.show_debug_var.get(): return self.log_text.insert(tk.END, msg + "\n") self.log_text.see(tk.END) def clear_log(self): """清空日志""" self.log_text.delete(1.0, tk.END) def add_signal(self, signal): """添加信号到显示""" self.signal_tree.insert('', 0, values=( signal['time'], signal['symbol'], signal['direction'].upper(), f"{signal['entry_price']:.5f}", f"{signal['breakout_pips']:.1f}" )) def update_positions_display(self, positions): """更新持仓显示""" # 清空现有显示 for item in self.position_tree.get_children(): self.position_tree.delete(item) # 添加新数据 total_profit = 0 for pos in positions: profit_color = "green" if pos['profit'] >= 0 else "red" profit_tag = f"profit_{pos['symbol']}_{pos['profit']}" self.position_tree.insert('', 'end', values=( pos['symbol'], pos['type'], f"{pos['volume']:.2f}", f"{pos['open_price']:.5f}", f"{pos['current_price']:.5f}", f"{pos['sl']:.5f}" if pos['sl'] > 0 else "0.00000", f"{pos['tp']:.5f}" if pos['tp'] > 0 else "0.00000", f"{pos['profit']:.2f}", pos['time'] ), tags=(profit_tag,)) self.position_tree.tag_configure(profit_tag, foreground=profit_color) total_profit += pos['profit'] # 更新状态栏 if total_profit != 0: profit_text = f"总盈亏: ${total_profit:.2f}" profit_color = "green" if total_profit > 0 else "red" self.status_label.config(text=profit_text, foreground=profit_color) else: self.status_label.config(text="● 运行中", foreground="green") def update_countdown(self, countdown_str): """更新倒计时""" self.countdown_label.config(text=countdown_str) def update_ui(self): """定期更新UI""" try: # 处理队列消息 while True: try: data = self.engine.data_queue.get_nowait() if data['type'] == 'countdown': self.countdown_label.config(text=data['value']) except queue.Empty: break # 更新交易时间状态 if self.engine.is_trading_time(): self.time_status.config(text="交易时段 ●", foreground="green") else: self.time_status.config(text="非交易时段 ○", foreground="gray") # 更新下次分析时间 next_check = self.engine.get_next_hour_close_time() self.next_check_label.config(text=f"下次分析: {next_check.strftime('%H:%M:%S')}") # 定期更新持仓 if self.engine.running: self.engine.update_positions() except Exception as e: print(f"UI更新异常: {e}") # 每秒更新一次 self.root.after(1000, self.update_ui) def start_engine(self): """启动交易引擎""" if not self.engine.running: if not mt5.initialize(): self.add_log("[ERROR] MT5初始化失败，请检查MetaTrader 5是否运行") return self.add_log("[INFO] MT5初始化成功") self.add_log("[INFO] 策略将在每个H1 K线收盘时自动分析") self.engine_thread = threading.Thread(target=self.engine.run, daemon=True) self.engine_thread.start() def stop_engine(self): """停止交易引擎""" self.engine.stop() mt5.shutdown() self.add_log("[INFO] 系统已停止") def run(self): """运行GUI""" self.root.protocol("WM_DELETE_WINDOW", self.on_closing) self.root.mainloop() def on_closing(self): """关闭窗口""" self.stop_engine() self.root.destroy() # ==================== 主程序 ==================== if __name__ == "__main__": # 需要添加deque导入 from collections import deque app = TradingGUI() app.run()