import MetaTrader5 as mt5 import pandas as pd import numpy as np from datetime import datetime, timedelta import time import threading import queue from typing import Dict, List, Optional import json import os # GUI相关 import tkinter as tk from tkinter import ttk, scrolledtext, messagebox import tkinter.font as tkFont class ForexAutoTrader: """外汇自动化交易系统 - 直接市价成交版""" def __init__(self): # MT5连接状态 self.mt5_connected = False # 交易参数 self.lot_size = 0.1 self.magic_number = 77770000 self.pip_multiplier = 10 self.max_spread_pips = 3 # 最大允许点差 # 货币对配置 self.symbols_config = { 'EURUSD': {'tp': 15, 'sl': 25, 'name': '欧元/美元', 'enabled': True}, 'GBPUSD': {'tp': 20, 'sl': 35, 'name': '英镑/美元', 'enabled': True}, 'USDJPY': {'tp': 15, 'sl': 20, 'name': '美元/日元', 'enabled': True}, 'AUDUSD': {'tp': 12, 'sl': 20, 'name': '澳元/美元', 'enabled': True}, 'USDCAD': {'tp': 15, 'sl': 25, 'name': '美元/加元', 'enabled': True}, } # 风控参数 self.max_daily_trades = 3 # 每日最大交易次数 self.max_daily_loss_pips = 50 # 每日最大亏损点数 self.cooldown_minutes = 30 # 亏损后冷却时间（分钟） # 当日统计 self.daily_trades = 0 self.daily_pips = 0 self.current_day = None self.last_loss_time = None # MACD参数 self.macd_fast = 12 self.macd_slow = 26 self.macd_signal = 9 self.ema_fast = 21 self.ema_slow = 55 # 数据存储 self.data_buffers = {} self.positions = {} self.signals_history = [] self.trades_history = [] # 运行状态 self.is_running = False self.monitor_thread = None self.update_interval = 10 self.last_bar_times = {} self.scan_mode = 'bar_close' # 日志队列 self.log_queue = queue.Queue() # GUI组件 self.root = None self.log_text = None self.positions_tree = None self.signals_tree = None self.status_label = None self.countdown_label = None self.symbol_checkboxes = {} self.risk_label = None # 下次检查时间 self.next_check_time = None self.countdown_seconds = 0 # 信号记录标记 self._current_signal_recorded = False def connect_mt5(self) -> bool: """连接MT5""" self.log("正在连接MT5...", "INFO") if not mt5.initialize(): error = mt5.last_error() self.log(f"MT5初始化失败: {error}", "ERROR") return False self.mt5_connected = True version = mt5.version() self.log(f"MT5连接成功，版本: {version[0]}.{version[1]}", "SUCCESS") # 获取账户信息 account_info = mt5.account_info() if account_info: self.log(f"账户: {account_info.login}, 余额: ${account_info.balance:.2f}, 杠杆: 1:{account_info.leverage}", "INFO") # 初始化数据缓存 for symbol in self.symbols_config: if mt5.symbol_select(symbol, True): self.data_buffers[symbol] = [] symbol_info = mt5.symbol_info(symbol) self.log(f"加载品种: {symbol}, 点差: {symbol_info.spread} points", "INFO") else: self.log(f"加载品种失败: {symbol}", "WARNING") return True def log(self, message: str, level: str = "INFO"): """添加日志""" timestamp = datetime.now().strftime("%H:%M:%S") log_entry = f"[{timestamp}] [{level}] {message}" self.log_queue.put(log_entry) print(log_entry) def reset_daily_stats(self): """重置每日统计""" today = datetime.now().date() if self.current_day != today: self.current_day = today self.daily_trades = 0 self.daily_pips = 0 self.last_loss_time = None self.log(f"新的一天开始，重置每日统计", "INFO") def check_daily_limits(self) -> tuple: """检查每日限制""" self.reset_daily_stats() if self.daily_trades >= self.max_daily_trades: return False, f"已达每日交易上限({self.max_daily_trades}笔)" if self.daily_pips <= -self.max_daily_loss_pips: return False, f"已达每日亏损上限({self.max_daily_loss_pips}点)" # 检查冷却时间 if self.last_loss_time: cooldown_end = self.last_loss_time + timedelta(minutes=self.cooldown_minutes) if datetime.now() < cooldown_end: remaining = (cooldown_end - datetime.now()).seconds // 60 return False, f"冷却中，剩余{remaining}分钟" return True, "OK" def get_current_spread(self, symbol: str) -> float: """获取当前点差（点数）""" tick = mt5.symbol_info_tick(symbol) if tick: symbol_info = mt5.symbol_info(symbol) if symbol_info and symbol_info.point: return (tick.ask - tick.bid) / (symbol_info.point * self.pip_multiplier) return 999 def fetch_latest_data(self, symbol: str, bars: int = 200) -> pd.DataFrame: """获取最新数据""" rates = mt5.copy_rates_from_pos(symbol, mt5.TIMEFRAME_M15, 0, bars) if rates is None or len(rates) == 0: return pd.DataFrame() df = pd.DataFrame(rates) df['time'] = pd.to_datetime(df['time'], unit='s') df.set_index('time', inplace=True) return df def calculate_indicators(self, df: pd.DataFrame, symbol: str) -> pd.DataFrame: """计算技术指标""" df = df.copy() # EMA df['ema_fast'] = df['close'].ewm(span=self.ema_fast, adjust=False).mean() df['ema_slow'] = df['close'].ewm(span=self.ema_slow, adjust=False).mean() # EMA斜率 df['ema_fast_slope'] = df['ema_fast'] - df['ema_fast'].shift(3) # MACD ema_12 = df['close'].ewm(span=self.macd_fast, adjust=False).mean() ema_26 = df['close'].ewm(span=self.macd_slow, adjust=False).mean() df['macd_line'] = ema_12 - ema_26 df['macd_signal_line'] = df['macd_line'].ewm(span=self.macd_signal, adjust=False).mean() df['macd_histogram'] = df['macd_line'] - df['macd_signal_line'] # 柱状图变化 df['hist_prev'] = df['macd_histogram'].shift(1) df['hist_turn_positive'] = (df['macd_histogram'] > 0) & (df['hist_prev'] <= 0) df['hist_turn_negative'] = (df['macd_histogram'] < 0) & (df['hist_prev'] >= 0) # 连续同色柱 df['hist_color'] = np.sign(df['macd_histogram']) df['color_change'] = df['hist_color'].diff().fillna(0) != 0 df['color_group'] = df['color_change'].cumsum() df['consecutive_bars'] = df.groupby('color_group').cumcount() + 1 # 价格与均线 df['price_above_slow'] = df['close'] > df['ema_slow'] df['price_below_slow'] = df['close'] < df['ema_slow'] df['ema_bullish'] = df['ema_fast'] > df['ema_slow'] df['ema_bearish'] = df['ema_fast'] < df['ema_slow'] # K线形态 df['is_bullish_bar'] = df['close'] > df['open'] df['is_bearish_bar'] = df['close'] < df['open'] # 前一根数据 df['high_prev'] = df['high'].shift(1) df['low_prev'] = df['low'].shift(1) df['close_prev'] = df['close'].shift(1) df['open_prev'] = df['open'].shift(1) df['ema_fast_prev'] = df['ema_fast'].shift(1) df['is_bullish_prev'] = df['is_bullish_bar'].shift(1) df['is_bearish_prev'] = df['is_bearish_bar'].shift(1) return df def check_existing_positions(self, symbol: str, direction: str) -> bool: """检查是否已有同方向同货币同magic的持仓""" positions = mt5.positions_get(symbol=symbol) if positions is None: return False for pos in positions: if pos.magic == self.magic_number: if (direction == 'LONG' and pos.type == mt5.POSITION_TYPE_BUY) or \ (direction == 'SHORT' and pos.type == mt5.POSITION_TYPE_SELL): return True return False def check_signal(self, df: pd.DataFrame, symbol: str) -> Dict: """检查交易信号""" if len(df) < 60: return {'signal': 0} row = df.iloc[-1] prev = df.iloc[-2] config = self.symbols_config[symbol] if pd.isna(row['ema_fast']) or pd.isna(row['ema_slow']): return {'signal': 0} symbol_info = mt5.symbol_info(symbol) if symbol_info is None: return {'signal': 0} point = symbol_info.point pip_value = point * self.pip_multiplier # 记录关键变量 self.log(f"[{symbol}] 价格:{row['close']:.5f}, EMA21:{row['ema_fast']:.5f}, EMA55:{row['ema_slow']:.5f}, " f"MACD柱:{row['macd_histogram']*10000:.2f}, 连续:{row['consecutive_bars']}, " f"前K线:{'阳' if prev['is_bullish_bar'] else '阴'}", "DEBUG") # ============ 做多信号（需要当前K线收阳确认）============ long_condition = ( row['price_above_slow'] and row['ema_bullish'] and row['ema_fast_slope'] > 0 and # EMA21向上 prev['low'] <= prev['ema_fast'] * 1.0003 and prev['is_bullish_bar'] and # 前一根必须是阳线 row['is_bullish_bar'] and # 当前根也是阳线 row['macd_histogram'] > 0 and (row['hist_turn_positive'] or row['consecutive_bars'] <= 3) and row['consecutive_bars'] < 8 ) if long_condition: # 直接使用当前市价 tick = mt5.symbol_info_tick(symbol) entry_price = tick.ask if tick else row['close'] return { 'signal': 1, 'type': 'MARKET_BUY', 'entry': round(entry_price, 5), 'sl': round(entry_price - config['sl'] * pip_value, 5), 'tp': round(entry_price + config['tp'] * pip_value, 5), 'reason': f"EMA多头+MACD金叉+阳线确认", 'details': f"价格>{row['ema_slow']:.5f}, EMA21>EMA55, 阳线确认, MACD柱{row['macd_histogram']*10000:.2f}" } # ============ 做空信号（需要当前K线收阴确认）============ short_condition = ( row['price_below_slow'] and row['ema_bearish'] and row['ema_fast_slope'] < 0 and # EMA21向下 prev['high'] >= prev['ema_fast'] * 0.9997 and prev['is_bearish_bar'] and # 前一根必须是阴线 row['is_bearish_bar'] and # 当前根也是阴线 row['macd_histogram'] < 0 and (row['hist_turn_negative'] or row['consecutive_bars'] <= 3) and row['consecutive_bars'] < 8 ) if short_condition: tick = mt5.symbol_info_tick(symbol) entry_price = tick.bid if tick else row['close'] return { 'signal': -1, 'type': 'MARKET_SELL', 'entry': round(entry_price, 5), 'sl': round(entry_price + config['sl'] * pip_value, 5), 'tp': round(entry_price - config['tp'] * pip_value, 5), 'reason': f"EMA空头+MACD死叉+阴线确认", 'details': f"价格<{row['ema_slow']:.5f}, EMA21 bool: """市价单成交""" config = self.symbols_config[symbol] # 检查点差 spread = self.get_current_spread(symbol) if spread > self.max_spread_pips: self.log(f"[{symbol}] 点差过大({spread:.1f}点 > {self.max_spread_pips}点)，放弃下单", "WARNING") # 更新信号记录 if self.signals_history: self.signals_history[-1]['result'] = f"点差过大({spread:.1f}点)" return False # 获取品种信息 symbol_info = mt5.symbol_info(symbol) if symbol_info is None: self.log(f"[{symbol}] 无法获取品种信息", "ERROR") if self.signals_history: self.signals_history[-1]['result'] = "获取品种信息失败" return False # 构建市价单请求 - 使用 TRADE_ACTION_DEAL 进行市价成交 order_type = mt5.ORDER_TYPE_BUY if signal['signal'] == 1 else mt5.ORDER_TYPE_SELL price = signal['entry'] # 获取当前最新价格确保市价单准确 tick = mt5.symbol_info_tick(symbol) if tick: if signal['signal'] == 1: price = tick.ask else: price = tick.bid # 更新入场价 signal['entry'] = price if self.signals_history: self.signals_history[-1]['entry'] = price request = { "action": mt5.TRADE_ACTION_DEAL, # 市价单 "symbol": symbol, "volume": self.lot_size, "type": order_type, "price": price, "sl": signal['sl'], "tp": signal['tp'], "deviation": 20, # 允许滑点20点 "magic": self.magic_number, "comment": f"Auto_{signal['reason'][:20]}", "type_time": mt5.ORDER_TIME_GTC, "type_filling": mt5.ORDER_FILLING_IOC, # 立即成交或取消 } # 发送订单 self.log(f"[{symbol}] 发送市价单: {'买入' if signal['signal']==1 else '卖出'} {self.lot_size}手 @ {price}", "INFO") result = mt5.order_send(request) if result.retcode != mt5.TRADE_RETCODE_DONE: error_msg = f"{result.comment} (retcode={result.retcode})" self.log(f"[{symbol}] 市价单失败: {error_msg}", "ERROR") # 尝试备选填充模式 self.log(f"[{symbol}] 尝试备选填充模式...", "INFO") # 备选模式列表 alt_modes = [mt5.ORDER_FILLING_FOK, mt5.ORDER_FILLING_RETURN] for alt_mode in alt_modes: if alt_mode == request["type_filling"]: continue try: request["type_filling"] = alt_mode result = mt5.order_send(request) if result.retcode == mt5.TRADE_RETCODE_DONE: self.log(f"[{symbol}] 备选模式成功! 使用填充模式: {alt_mode}", "SUCCESS") break except Exception as e: self.log(f"[{symbol}] 填充模式 {alt_mode} 失败: {e}", "DEBUG") if result.retcode != mt5.TRADE_RETCODE_DONE: self.log(f"[{symbol}] 所有尝试均失败: {error_msg}", "ERROR") # 更新信号记录 if self.signals_history: self.signals_history[-1]['result'] = f"下单失败: {error_msg[:30]}" return False # 获取成交价格 if result.price > 0: actual_price = result.price else: actual_price = price self.log(f"[{symbol}] ✅ 市价单成交! 方向:{'买入' if signal['signal']==1 else '卖出'}, " f"价格:{actual_price}, 止损:{signal['sl']}, 止盈:{signal['tp']}, " f"单号:{result.order}, 点差:{spread:.1f}", "SUCCESS") # 更新每日统计 self.daily_trades += 1 # 更新信号记录中的结果 if self.signals_history: self.signals_history[-1]['entry'] = actual_price self.signals_history[-1]['result'] = f"成交成功, 单号:{result.order}" self.signals_history[-1]['order_id'] = result.order return True def update_positions(self): """更新持仓信息并计算每日盈亏""" positions = mt5.positions_get() self.positions = {} if positions: for pos in positions: if pos.magic == self.magic_number: # 计算点数盈亏 symbol_info = mt5.symbol_info(pos.symbol) if symbol_info and symbol_info.point: point = symbol_info.point if pos.type == 0: # BUY pips = (pos.price_current - pos.price_open) / (point * self.pip_multiplier) else: # SELL pips = (pos.price_open - pos.price_current) / (point * self.pip_multiplier) else: pips = 0 self.positions[pos.ticket] = { 'symbol': pos.symbol, 'type': 'BUY' if pos.type == 0 else 'SELL', 'volume': pos.volume, 'open_price': pos.price_open, 'current_price': pos.price_current, 'sl': pos.sl, 'tp': pos.tp, 'profit': pos.profit, 'profit_pips': round(pips, 1), 'swap': pos.swap, 'time': datetime.fromtimestamp(pos.time).strftime("%m-%d %H:%M") } # 计算当日累计点数盈亏 total_pips = sum(pos['profit_pips'] for pos in self.positions.values()) self.daily_pips = total_pips def get_next_bar_time(self, timeframe_minutes: int = 15) -> datetime: """计算下一个K线收盘时间""" now = datetime.now() minutes = now.minute next_minute = ((minutes // timeframe_minutes) + 1) * timeframe_minutes if next_minute >= 60: next_time = now.replace(hour=now.hour + 1, minute=next_minute - 60, second=0, microsecond=0) else: next_time = now.replace(minute=next_minute, second=0, microsecond=0) return next_time def is_new_bar(self, symbol: str, df: pd.DataFrame) -> bool: """检查是否有新K线产生""" if df.empty: return False current_bar_time = df.index[-1] if symbol not in self.last_bar_times: self.last_bar_times[symbol] = current_bar_time return True if current_bar_time > self.last_bar_times[symbol]: self.last_bar_times[symbol] = current_bar_time return True return False def smart_scan_symbol(self, symbol: str): """智能扫描单个货币对""" if not self.symbols_config[symbol]['enabled']: return config = self.symbols_config[symbol] # 检查每日限制 can_trade, reason = self.check_daily_limits() if not can_trade: self.log(f"[{symbol}] {reason}，跳过扫描", "INFO") return # 获取数据 df = self.fetch_latest_data(symbol) if df.empty: return # 检查是否有新K线 if not self.is_new_bar(symbol, df): return current_time = df.index[-1] self.log(f"[{symbol}] 新K线产生 {current_time.strftime('%H:%M')}，开始扫描...", "DEBUG") # 计算指标 df = self.calculate_indicators(df, symbol) # 检查信号 signal = self.check_signal(df, symbol) if signal['signal'] != 0: direction = 'LONG' if signal['signal'] == 1 else 'SHORT' # 检查是否已有持仓 if self.check_existing_positions(symbol, direction): self.log(f"[{symbol}] 已有同方向持仓，跳过", "INFO") return # 输出信号详情 self.log(f"[{symbol}] 🎯 发现信号! {signal['reason']}", "SIGNAL") self.log(f"[{symbol}] {signal['details']}", "INFO") self.log(f"[{symbol}] 入场:{signal['entry']}, SL:{signal['sl']}(-{config['sl']}点), " f"TP:{signal['tp']}(+{config['tp']}点)", "INFO") # 先记录信号到历史（无论是否成交都记录） spread = self.get_current_spread(symbol) self.record_signal_to_history(symbol, signal, spread, "待成交") self._current_signal_recorded = True # 直接市价成交 if self.place_market_order(symbol, signal): self.update_positions() def monitor_loop(self): """监控主循环""" self.log("🚀 监控循环启动，等待K线收盘...", "SUCCESS") while self.is_running: try: # 计算下一个K线时间 next_bar = self.get_next_bar_time(15) wait_seconds = (next_bar - datetime.now()).total_seconds() if wait_seconds > 0: # 提前3秒开始准备 sleep_time = max(1, wait_seconds - 3) self.countdown_seconds = int(sleep_time) # 分段等待，便于快速响应停止信号 for _ in range(int(sleep_time)): if not self.is_running: break time.sleep(1) self.countdown_seconds -= 1 if not self.is_running: break # 最后几秒精确等待 while datetime.now() < next_bar - timedelta(milliseconds=500): if not self.is_running: break time.sleep(0.1) if not self.is_running: break # K线收盘时扫描 self.log("="*60, "INFO") self.log(f"M15 K线收盘 {datetime.now().strftime('%H:%M:%S')}，开始扫描...", "INFO") for symbol in self.symbols_config: self.smart_scan_symbol(symbol) self.update_positions() self.countdown_seconds = 0 except Exception as e: self.log(f"监控循环错误: {e}", "ERROR") import traceback self.log(traceback.format_exc(), "ERROR") time.sleep(5) def start_monitoring(self): """开始监控""" if self.is_running: return if not self.mt5_connected: if not self.connect_mt5(): messagebox.showerror("错误", "无法连接MT5") return self.is_running = True self.monitor_thread = threading.Thread(target=self.monitor_loop, daemon=True) self.monitor_thread.start() self.log("🚀 自动交易系统已启动 (市价成交模式)", "SUCCESS") self.log(f"风控: 每日最多{self.max_daily_trades}笔, 最大亏损{self.max_daily_loss_pips}点", "INFO") self.status_label.config(text="运行中", foreground="green") self.start_btn.config(state=tk.DISABLED) self.stop_btn.config(state=tk.NORMAL) def stop_monitoring(self): """停止监控""" self.is_running = False self.log("⏹️ 自动交易系统已停止", "WARNING") self.status_label.config(text="已停止", foreground="red") self.start_btn.config(state=tk.NORMAL) self.stop_btn.config(state=tk.DISABLED) # ==================== GUI ==================== def create_gui(self): """创建GUI界面""" self.root = tk.Tk() self.root.title("EMA均线趋势+MACD动量确认+K线实体验证的三重过滤系统V334") self.root.geometry("1300x850") # 设置字体 default_font = tkFont.nametofont("TkDefaultFont") default_font.configure(size=10) # 设置样式 style = ttk.Style() style.configure("Treeview", font=('Arial', 9)) style.configure("Treeview.Heading", font=('Arial', 9, 'bold')) # ========== 顶部控制栏 ========== control_frame = ttk.Frame(self.root) control_frame.pack(fill=tk.X, padx=10, pady=5) # 连接状态 ttk.Label(control_frame, text="状态:").pack(side=tk.LEFT, padx=5) self.status_label = ttk.Label(control_frame, text="未连接", foreground="red") self.status_label.pack(side=tk.LEFT, padx=5) # 倒计时 ttk.Label(control_frame, text="下次K线:").pack(side=tk.LEFT, padx=(20, 5)) self.countdown_label = ttk.Label(control_frame, text="--", font=("Arial", 12, "bold")) self.countdown_label.pack(side=tk.LEFT, padx=5) # 风控状态 self.risk_label = ttk.Label(control_frame, text="今日: 0/3笔, 0点", foreground="blue") self.risk_label.pack(side=tk.LEFT, padx=(20, 5)) # 按钮 self.start_btn = ttk.Button(control_frame, text="▶ 开始", command=self.start_monitoring) self.start_btn.pack(side=tk.LEFT, padx=5) self.stop_btn = ttk.Button(control_frame, text="⏹ 停止", command=self.stop_monitoring, state=tk.DISABLED) self.stop_btn.pack(side=tk.LEFT, padx=5) ttk.Button(control_frame, text="🔄 刷新持仓", command=self.update_positions_display).pack(side=tk.LEFT, padx=5) # 货币对选择 ttk.Label(control_frame, text="监控品种:").pack(side=tk.LEFT, padx=(20, 5)) for symbol, config in self.symbols_config.items(): var = tk.BooleanVar(value=config['enabled']) self.symbol_checkboxes[symbol] = var cb = ttk.Checkbutton(control_frame, text=symbol, variable=var, command=lambda s=symbol, v=var: self.toggle_symbol(s, v)) cb.pack(side=tk.LEFT, padx=2) # ========== 主内容区 ========== main_frame = ttk.Frame(self.root) main_frame.pack(fill=tk.BOTH, expand=True, padx=10, pady=5) # 左侧面板 - 持仓 left_frame = ttk.LabelFrame(main_frame, text="当前持仓") left_frame.pack(side=tk.LEFT, fill=tk.BOTH, expand=True, padx=(0, 5)) columns = ('时间', '品种', '方向', '手数', '开仓价', '现价', '止损', '止盈', '盈亏', '点数') self.positions_tree = ttk.Treeview(left_frame, columns=columns, show='headings', height=12) col_widths = [90, 70, 50, 50, 80, 80, 80, 80, 80, 60] for col, width in zip(columns, col_widths): self.positions_tree.heading(col, text=col) self.positions_tree.column(col, width=width, anchor='center') scrollbar = ttk.Scrollbar(left_frame, orient=tk.VERTICAL, command=self.positions_tree.yview) self.positions_tree.configure(yscrollcommand=scrollbar.set) self.positions_tree.pack(side=tk.LEFT, fill=tk.BOTH, expand=True) scrollbar.pack(side=tk.RIGHT, fill=tk.Y) # 配置Treeview标签样式 self.positions_tree.tag_configure('profit', foreground='green') self.positions_tree.tag_configure('loss', foreground='red') # 右侧面板 - 信号历史 right_frame = ttk.LabelFrame(main_frame, text="信号历史") right_frame.pack(side=tk.RIGHT, fill=tk.BOTH, expand=True, padx=(5, 0)) columns = ('时间', '品种', '方向', '入场价', '止损', '止盈', '点差', '结果', '原因') self.signals_tree = ttk.Treeview(right_frame, columns=columns, show='headings', height=12) col_widths = [140, 70, 50, 80, 80, 80, 50, 120, 150] for col, width in zip(columns, col_widths): self.signals_tree.heading(col, text=col) self.signals_tree.column(col, width=width, anchor='center') scrollbar2 = ttk.Scrollbar(right_frame, orient=tk.VERTICAL, command=self.signals_tree.yview) self.signals_tree.configure(yscrollcommand=scrollbar2.set) self.signals_tree.pack(side=tk.LEFT, fill=tk.BOTH, expand=True) scrollbar2.pack(side=tk.RIGHT, fill=tk.Y) # 配置信号历史标签样式 self.signals_tree.tag_configure('success', foreground='green') self.signals_tree.tag_configure('failed', foreground='red') self.signals_tree.tag_configure('pending', foreground='orange') # ========== 底部日志区 ========== log_frame = ttk.LabelFrame(self.root, text="运行日志") log_frame.pack(fill=tk.BOTH, expand=True, padx=10, pady=5) self.log_text = scrolledtext.ScrolledText(log_frame, height=15, font=("Consolas", 9)) self.log_text.pack(fill=tk.BOTH, expand=True, padx=5, pady=5) # 配置日志颜色标签 self.log_text.tag_config("INFO", foreground="black") self.log_text.tag_config("DEBUG", foreground="gray") self.log_text.tag_config("SIGNAL", foreground="blue") self.log_text.tag_config("SUCCESS", foreground="green") self.log_text.tag_config("WARNING", foreground="orange") self.log_text.tag_config("ERROR", foreground="red") # 启动定时更新 self.update_log_display() self.update_countdown_display() self.update_trees() self.update_risk_display() # 自动连接MT5 self.root.after(1000, self.auto_connect) self.root.protocol("WM_DELETE_WINDOW", self.on_closing) self.root.mainloop() def auto_connect(self): """自动连接MT5""" if self.connect_mt5(): self.status_label.config(text="已连接", foreground="green") self.start_btn.config(state=tk.NORMAL) else: self.status_label.config(text="连接失败", foreground="red") def toggle_symbol(self, symbol: str, var: tk.BooleanVar): """切换货币对监控状态""" self.symbols_config[symbol]['enabled'] = var.get() self.log(f"{symbol} 监控状态: {'启用' if var.get() else '禁用'}", "INFO") def update_positions_display(self): """更新持仓显示""" self.update_positions() # 清空现有显示 for item in self.positions_tree.get_children(): self.positions_tree.delete(item) # 添加持仓 for ticket, pos in self.positions.items(): tag = 'profit' if pos['profit'] > 0 else 'loss' self.positions_tree.insert('', 'end', values=(pos['time'], pos['symbol'], pos['type'], pos['volume'], f"{pos['open_price']:.5f}", f"{pos['current_price']:.5f}", f"{pos['sl']:.5f}" if pos['sl'] else "-", f"{pos['tp']:.5f}" if pos['tp'] else "-", f"${pos['profit']:.2f}", f"{pos['profit_pips']}"), tags=(tag,)) def update_signals_display(self): """更新信号历史显示""" # 清空现有显示 for item in self.signals_tree.get_children(): self.signals_tree.delete(item) # 添加最近30条信号 for sig in self.signals_history[-30:]: result = sig.get('result', '-') if '成功' in result or '成交' in result: tag = 'success' elif '失败' in result or '过大' in result: tag = 'failed' else: tag = 'pending' self.signals_tree.insert('', 'end', values=(sig['time'], sig['symbol'], sig['direction'], f"{sig['entry']:.5f}", f"{sig['sl']:.5f}", f"{sig['tp']:.5f}", f"{sig.get('spread', '-')}", result, sig['reason']), tags=(tag,)) def update_risk_display(self): """更新风控显示""" self.reset_daily_stats() self.risk_label.config(text=f"今日: {self.daily_trades}/{self.max_daily_trades}笔, {self.daily_pips:.0f}点") self.root.after(5000, self.update_risk_display) def update_log_display(self): """更新日志显示""" try: while True: log_entry = self.log_queue.get_nowait() self.log_text.insert(tk.END, log_entry + "\n") for tag in ["INFO", "DEBUG", "SIGNAL", "SUCCESS", "WARNING", "ERROR"]: if f"[{tag}]" in log_entry: start = self.log_text.index("end-2l") end = self.log_text.index("end-1l") self.log_text.tag_add(tag, start, end) break self.log_text.see(tk.END) except queue.Empty: pass self.root.after(500, self.update_log_display) def update_countdown_display(self): """更新倒计时显示""" if self.is_running: next_bar = self.get_next_bar_time(15) remaining = (next_bar - datetime.now()).total_seconds() if remaining > 0: minutes = int(remaining // 60) seconds = int(remaining % 60) self.countdown_label.config(text=f"{minutes:02d}:{seconds:02d}") else: self.countdown_label.config(text="扫描中...") else: self.countdown_label.config(text="--") self.root.after(1000, self.update_countdown_display) def update_trees(self): """定期更新树形视图""" try: self.update_positions_display() self.update_signals_display() except Exception as e: self.log(f"更新界面错误: {e}", "ERROR") self.root.after(5000, self.update_trees) def on_closing(self): """关闭窗口""" self.is_running = False if self.mt5_connected: mt5.shutdown() self.root.destroy() def run(self): """运行GUI""" self.create_gui() if __name__ == "__main__": trader = ForexAutoTrader() trader.run()