""" MT5 + DeepSeek 日内交易分析系统 - GUI版本（带MT5图表标记和自动交易） 修复MT5对象操作问题 """ import MetaTrader5 as mt5 import pandas as pd import numpy as np from datetime import datetime, timedelta import json import time from typing import Dict, List, Tuple, Optional from openai import OpenAI import os from dotenv import load_dotenv from threading import Thread, Lock import tkinter as tk from tkinter import ttk, scrolledtext, messagebox from queue import Queue # 加载环境变量 load_dotenv() # ============================================ # 第一部分：MT5 数据获取模块 # ============================================ class MT5DataFetcher: """MT5 数据获取器""" TIMEFRAME_MAP = { 'M15': mt5.TIMEFRAME_M15, 'H1': mt5.TIMEFRAME_H1, 'D1': mt5.TIMEFRAME_D1, } def __init__(self, symbol: str = "EURUSD"): self.symbol = symbol self.connected = False def connect(self) -> bool: """连接 MT5 终端""" if not mt5.initialize(): print(f"❌ MT5 初始化失败: {mt5.last_error()}") return False self.connected = True return True def disconnect(self): """断开 MT5 连接""" if self.connected: mt5.shutdown() self.connected = False def get_ohlc_data(self, timeframe: str, bars: int) -> pd.DataFrame: """获取指定周期的 K 线数据""" if not self.connected: raise ConnectionError("MT5 未连接") tf = self.TIMEFRAME_MAP.get(timeframe) if tf is None: raise ValueError(f"不支持的周期: {timeframe}") rates = mt5.copy_rates_from_pos(self.symbol, tf, 0, bars) if rates is None or len(rates) == 0: raise ValueError(f"获取 {timeframe} 数据失败: {mt5.last_error()}") df = pd.DataFrame(rates) df['time'] = pd.to_datetime(df['time'], unit='s') df.set_index('time', inplace=True) return df def get_current_price(self) -> Dict[str, float]: """获取当前买卖价格""" tick = mt5.symbol_info_tick(self.symbol) if tick is None: return {'bid': 0, 'ask': 0, 'spread': 0} return { 'bid': tick.bid, 'ask': tick.ask, 'spread': tick.ask - tick.bid } def place_arrow_on_chart(self, symbol: str, direction: str, price: float): """在MT5图表上标记箭头 - 使用正确的MT5 API""" try: # 获取当前M15 K线的时间（对齐到15分钟） now = datetime.now() current_minute = now.minute aligned_minute = (current_minute // 15) * 15 arrow_time = now.replace(minute=aligned_minute, second=0, microsecond=0) arrow_timestamp = int(arrow_time.timestamp()) # 获取该K线的OHLC数据，确定箭头位置 rates = mt5.copy_rates_from(self.symbol, mt5.TIMEFRAME_M15, arrow_timestamp, 1) if rates is not None and len(rates) > 0: candle = rates[0] high = candle['high'] low = candle['low'] # 根据方向设置箭头位置和类型 if direction == "看涨": arrow_price = low - (high - low) * 0.1 arrow_char = "▲" else: arrow_price = high + (high - low) * 0.1 arrow_char = "▼" else: arrow_price = price arrow_char = "▲" if direction == "看涨" else "▼" obj_name = f"Signal_{symbol}_{direction}_{int(arrow_timestamp)}" chart_id = mt5.chart_open(symbol, mt5.TIMEFRAME_M15) if chart_id is None: print(f"⚠️ 无法打开 {symbol} 图表") return False try: mt5.chart_object_delete(chart_id, obj_name) except: pass point = { "type": mt5.OBJ_TEXT, "time": arrow_timestamp, "price": arrow_price, "text": arrow_char, "font_size": 14, "color": "LimeGreen" if direction == "看涨" else "Red", "anchor": mt5.ANCHOR_CENTER, "symbol": symbol, "timeframe": mt5.TIMEFRAME_M15 } result = mt5.obj_add(chart_id, obj_name, point) if result: print(f"✅ 已标记箭头: {symbol} {direction} @ {arrow_price:.5f}") return True else: print(f"⚠️ 标记箭头失败: {symbol} 错误码: {mt5.last_error()}") return False except Exception as e: print(f"❌ 标记箭头异常 {symbol}: {e}") return False # ============================================ # 交易管理模块 # ============================================ class TradeManager: """交易管理器 - 处理自动下单、持仓管理""" MAGIC_NUMBER = 11223344 LOT_SIZE = 1.0 def __init__(self): self.order_lock = Lock() def get_positions(self, symbol: str = None) -> List: """获取当前持仓""" if symbol: return mt5.positions_get(symbol=symbol) else: return mt5.positions_get() def get_positions_by_direction(self, symbol: str, direction: str) -> List: """获取指定方向和品种的持仓""" positions = self.get_positions(symbol) if not positions: return [] result = [] direction_map = {"看涨": mt5.POSITION_TYPE_BUY, "看跌": mt5.POSITION_TYPE_SELL} target_type = direction_map.get(direction) for pos in positions: if pos.type == target_type: result.append(pos) return result def has_position(self, symbol: str, direction: str) -> bool: """检查是否存在指定方向和品种的持仓""" return len(self.get_positions_by_direction(symbol, direction)) > 0 def close_position(self, position) -> bool: """平仓单个订单""" try: symbol = position.symbol order_type = mt5.ORDER_TYPE_BUY if position.type == mt5.POSITION_TYPE_SELL else mt5.ORDER_TYPE_SELL price = mt5.symbol_info_tick(symbol).bid if order_type == mt5.ORDER_TYPE_SELL else mt5.symbol_info_tick(symbol).ask request = { "action": mt5.TRADE_ACTION_DEAL, "symbol": symbol, "volume": position.volume, "type": order_type, "position": position.ticket, "price": price, "deviation": 20, "magic": self.MAGIC_NUMBER, "comment": "Auto close by DeepSeek signal", "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"❌ 平仓失败 {symbol} ticket={position.ticket}: {result.comment}") return False print(f"✅ 平仓成功 {symbol} ticket={position.ticket}") return True except Exception as e: print(f"❌ 平仓异常: {e}") return False def close_opposite_positions(self, symbol: str, direction: str) -> bool: """平仓所有反方向持仓""" opposite_direction = "看跌" if direction == "看涨" else "看涨" opposite_positions = self.get_positions_by_direction(symbol, opposite_direction) if not opposite_positions: return True success = True for pos in opposite_positions: if not self.close_position(pos): success = False return success def open_position(self, symbol: str, direction: str, price: float, stop_loss: float = None, take_profit: float = None) -> bool: """开仓""" try: # 获取交易类型和价格 if direction == "看涨": order_type = mt5.ORDER_TYPE_BUY order_price = mt5.symbol_info_tick(symbol).ask else: order_type = mt5.ORDER_TYPE_SELL order_price = mt5.symbol_info_tick(symbol).bid # 构建请求 request = { "action": mt5.TRADE_ACTION_DEAL, "symbol": symbol, "volume": self.LOT_SIZE, "type": order_type, "price": order_price, "deviation": 20, "magic": self.MAGIC_NUMBER, "comment": f"DeepSeek {direction} signal", "type_time": mt5.ORDER_TIME_GTC, "type_filling": mt5.ORDER_FILLING_IOC, } # 添加止损止盈 if stop_loss: request["sl"] = stop_loss if take_profit: request["tp"] = take_profit result = mt5.order_send(request) if result.retcode != mt5.TRADE_RETCODE_DONE: print(f"❌ 开仓失败 {symbol} {direction}: {result.comment}") return False print(f"✅ 开仓成功 {symbol} {direction} {self.LOT_SIZE}手 @ {order_price:.5f}") if stop_loss: print(f" 止损: {stop_loss:.5f}") if take_profit: print(f" 止盈: {take_profit:.5f}") return True except Exception as e: print(f"❌ 开仓异常: {e}") return False def execute_signal(self, symbol: str, direction: str, stop_loss: float = None, take_profit: float = None) -> Dict: """执行交易信号""" with self.order_lock: result = { 'executed': False, 'action': None, 'message': '' } # 1. 检查是否已有同方向订单 if self.has_position(symbol, direction): result['message'] = f"{symbol} 已有{direction}方向持仓，跳过" print(f"⏭️ {result['message']}") return result # 2. 平仓反方向订单 if self.close_opposite_positions(symbol, direction): # 3. 开新仓 if self.open_position(symbol, direction, None, stop_loss, take_profit): result['executed'] = True result['action'] = f"开仓{direction}" result['message'] = f"{symbol} {direction} 开仓成功" else: result['message'] = f"{symbol} 开仓失败" else: result['message'] = f"{symbol} 平仓反方向订单失败" return result # ============================================ # 第二部分：技术指标计算与特征提取 # ============================================ class MarketAnalyzer: """市场特征分析器""" @staticmethod def calculate_atr(df: pd.DataFrame, period: int = 14) -> pd.Series: """计算 ATR""" high = df['high'] low = df['low'] close = df['close'] tr1 = high - low tr2 = abs(high - close.shift()) tr3 = abs(low - close.shift()) tr = pd.concat([tr1, tr2, tr3], axis=1).max(axis=1) atr = tr.rolling(window=period).mean() return atr @staticmethod def calculate_rsi(df: pd.DataFrame, period: int = 14) -> pd.Series: """计算 RSI""" delta = df['close'].diff() gain = (delta.where(delta > 0, 0)).rolling(window=period).mean() loss = (-delta.where(delta < 0, 0)).rolling(window=period).mean() rs = gain / loss rsi = 100 - (100 / (1 + rs)) return rsi @staticmethod def calculate_macd(df: pd.DataFrame, fast: int = 12, slow: int = 26, signal: int = 9) -> Dict: """计算 MACD""" ema_fast = df['close'].ewm(span=fast, adjust=False).mean() ema_slow = df['close'].ewm(span=slow, adjust=False).mean() macd_line = ema_fast - ema_slow signal_line = macd_line.ewm(span=signal, adjust=False).mean() histogram = macd_line - signal_line return { 'macd': macd_line, 'signal': signal_line, 'histogram': histogram } @staticmethod def analyze_candlestick_patterns(df: pd.DataFrame) -> Dict: """分析 K 线形态特征""" last_bars = df.tail(10) upper_shadows = (last_bars['high'] - last_bars[['close', 'open']].max(axis=1)) lower_shadows = (last_bars[['close', 'open']].min(axis=1) - last_bars['low']) bodies = abs(last_bars['close'] - last_bars['open']) long_upper = (upper_shadows > bodies * 2).sum() long_lower = (lower_shadows > bodies * 2).sum() bullish = (last_bars['close'] > last_bars['open']).sum() bearish = (last_bars['close'] < last_bars['open']).sum() return { 'long_upper_count': int(long_upper), 'long_lower_count': int(long_lower), 'bullish_count': int(bullish), 'bearish_count': int(bearish), 'last_candle_body_ratio': float(bodies.iloc[-1] / (upper_shadows.iloc[-1] + lower_shadows.iloc[-1] + 0.00001)) } @staticmethod def find_support_resistance(df: pd.DataFrame, lookback: int = 50) -> Dict: """识别支撑阻力位""" recent_df = df.tail(lookback) window = 5 highs = recent_df['high'].rolling(window=window, center=True).max() lows = recent_df['low'].rolling(window=window, center=True).min() resistance_levels = recent_df[recent_df['high'] == highs]['high'].tail(3).tolist() support_levels = recent_df[recent_df['low'] == lows]['low'].tail(3).tolist() current_price = recent_df['close'].iloc[-1] nearest_resistance = min([r for r in resistance_levels if r > current_price], default=None) nearest_support = max([s for s in support_levels if s < current_price], default=None) return { 'nearest_resistance': nearest_resistance, 'nearest_support': nearest_support, 'distance_to_resistance_pips': (nearest_resistance - current_price) * 10000 if nearest_resistance else None, 'distance_to_support_pips': (current_price - nearest_support) * 10000 if nearest_support else None } # ============================================ # 第三部分：语义描述生成器 # ============================================ class SemanticDescriptionGenerator: """将技术数据转换为自然语言描述""" def __init__(self, symbol: str): self.symbol = symbol self.analyzer = MarketAnalyzer() def generate_d1_description(self, df: pd.DataFrame) -> str: """生成日线级别描述""" last_5 = df.tail(5) price_change = (last_5['close'].iloc[-1] - last_5['close'].iloc[0]) / last_5['close'].iloc[0] * 100 consecutive_bullish = 0 consecutive_bearish = 0 for i in range(len(last_5)-1, -1, -1): if last_5['close'].iloc[i] > last_5['open'].iloc[i]: consecutive_bullish += 1 else: break for i in range(len(last_5)-1, -1, -1): if last_5['close'].iloc[i] < last_5['open'].iloc[i]: consecutive_bearish += 1 else: break desc_parts = [] if consecutive_bullish >= 3: desc_parts.append(f"D1连续{consecutive_bullish}日收阳") elif consecutive_bearish >= 3: desc_parts.append(f"D1连续{consecutive_bearish}日收阴") else: desc_parts.append(f"D1近5日涨跌幅为{price_change:.2f}%") last_candle = df.iloc[-1] upper_shadow = last_candle['high'] - max(last_candle['close'], last_candle['open']) lower_shadow = min(last_candle['close'], last_candle['open']) - last_candle['low'] body = abs(last_candle['close'] - last_candle['open']) if upper_shadow > body * 1.5: desc_parts.append("昨日收出长上影线，上方抛压明显") elif lower_shadow > body * 1.5: desc_parts.append("昨日收出长下影线，下方买盘支撑强劲") recent_high = df['high'].tail(20).max() recent_low = df['low'].tail(20).min() current = last_candle['close'] position_pct = (current - recent_low) / (recent_high - recent_low) * 100 if position_pct > 80: desc_parts.append(f"价格处于近20日高位区间({position_pct:.0f}%分位)") elif position_pct < 20: desc_parts.append(f"价格处于近20日低位区间({position_pct:.0f}%分位)") return "；".join(desc_parts) def generate_h1_description(self, df: pd.DataFrame) -> str: """生成小时线描述""" rsi = self.analyzer.calculate_rsi(df) macd_data = self.analyzer.calculate_macd(df) current_rsi = rsi.iloc[-1] rsi_prev = rsi.iloc[-4] hist_last_3 = macd_data['histogram'].tail(3) hist_trend = "伸长" if abs(hist_last_3.iloc[-1]) > abs(hist_last_3.iloc[-2]) else "缩短" desc_parts = [] if current_rsi > 70: desc_parts.append(f"H1的RSI读数为{current_rsi:.1f}，处于超买区域") elif current_rsi < 30: desc_parts.append(f"H1的RSI读数为{current_rsi:.1f}，处于超卖区域") else: desc_parts.append(f"H1的RSI读数为{current_rsi:.1f}，处于中性区间") if current_rsi > rsi_prev + 3: desc_parts.append("RSI呈上升趋势") elif current_rsi < rsi_prev - 3: desc_parts.append("RSI呈下降趋势") macd_val = macd_data['macd'].iloc[-1] signal_val = macd_data['signal'].iloc[-1] if macd_val > signal_val: desc_parts.append(f"MACD金叉状态，动能柱{hist_trend}") else: desc_parts.append(f"MACD死叉状态，动能柱{hist_trend}") return "；".join(desc_parts) def generate_m15_description(self, df: pd.DataFrame, current_price: float) -> str: """生成15分钟线描述""" atr = self.analyzer.calculate_atr(df) patterns = self.analyzer.analyze_candlestick_patterns(df) sr_levels = self.analyzer.find_support_resistance(df) desc_parts = [] current_atr = atr.iloc[-1] atr_avg = atr.tail(20).mean() vol_ratio = current_atr / atr_avg if vol_ratio > 1.3: desc_parts.append("M15波动率显著放大") elif vol_ratio < 0.7: desc_parts.append("M15处于低波动收缩状态") recent_high = df['high'].tail(20).max() recent_low = df['low'].tail(20).min() range_pips = (recent_high - recent_low) * 10000 desc_parts.append(f"过去3小时形成{range_pips:.0f}点的波动区间") if patterns['long_upper_count'] >= 3: desc_parts.append("近期频繁出现长上影线，上方抛压沉重") elif patterns['long_lower_count'] >= 3: desc_parts.append("近期频繁出现长下影线，下方买盘积极") if sr_levels['distance_to_resistance_pips'] and sr_levels['distance_to_resistance_pips'] < 10: desc_parts.append(f"现价紧贴阻力位{sr_levels['nearest_resistance']:.5f}") elif sr_levels['distance_to_support_pips'] and sr_levels['distance_to_support_pips'] < 10: desc_parts.append(f"现价紧贴支撑位{sr_levels['nearest_support']:.5f}") if patterns['bullish_count'] >= 7: desc_parts.append("多头力量占据主导") elif patterns['bearish_count'] >= 7: desc_parts.append("空头力量占据主导") return "；".join(desc_parts) # ============================================ # 第四部分：DeepSeek API 调用 # ============================================ class DeepSeekAnalyzer: """DeepSeek API 分析器""" def __init__(self, api_key: str = None): self.api_key = api_key or os.getenv('DEEPSEEK_API_KEY') if not self.api_key: raise ValueError("请设置 DEEPSEEK_API_KEY 环境变量") self.client = OpenAI( api_key=self.api_key, base_url="https://api.deepseek.com" ) def analyze_direction(self, symbol: str, d1_desc: str, h1_desc: str, m15_desc: str, current_price: float) -> Dict: """调用 DeepSeek 分析方向置信度""" prompt = f"""角色设定：你是一位专注于 M15 日内动量突破的交易架构师。请忽略基本面消息，仅基于提供的价格行为结构数据，输出方向研判与置信度。 输入数据快照（{symbol}）： 1. 大周期背景 (D1 & H1)：{d1_desc}；{h1_desc} 2. M15 关键形态：{m15_desc} 3. 当前价格：{current_price:.5f} 输出要求： 请严格按以下JSON格式回复，不要发散： {{ "direction": "看涨" 或 "看跌" 或 "震荡无方向", "confidence": "高" 或 "中" 或 "低", "reasoning": "用2-3句话解释判断依据，重点说明当前是真突破动能还是流动性扫损陷阱" }} 注意：如果 confidence 为 "高"，请在JSON中额外包含以下两个字段： "stop_loss": "止损点位（基于ATR或关键支撑/阻力位计算，给出具体价格，保留5位小数）", "take_profit": "止盈点位（基于风险回报比1:1.5或2:1计算，给出具体价格，保留5位小数）" 如果 confidence 不是 "高"，则不需要包含 stop_loss 和 take_profit 字段。""" try: response = self.client.chat.completions.create( model="deepseek-chat", messages=[ {"role": "system", "content": "你是一个专业的交易分析师，输出必须是严格的JSON格式。"}, {"role": "user", "content": prompt} ], temperature=0.3, max_tokens=500 ) result_text = response.choices[0].message.content # 提取 JSON if "```json" in result_text: result_text = result_text.split("```json")[1].split("```")[0] elif "```" in result_text: result_text = result_text.split("```")[1].split("```")[0] result = json.loads(result_text.strip()) analysis_result = { 'direction': result.get('direction', '未知'), 'confidence': result.get('confidence', '未知'), 'reasoning': result.get('reasoning', '无') } if analysis_result['confidence'] == '高': analysis_result['stop_loss'] = result.get('stop_loss', '未提供') analysis_result['take_profit'] = result.get('take_profit', '未提供') return analysis_result except Exception as e: print(f"❌ {symbol} DeepSeek API 调用失败: {e}") return { 'direction': '错误', 'confidence': '错误', 'reasoning': f'API调用异常: {str(e)}' } # ============================================ # 第五部分：GUI 主程序 # ============================================ class MT5DeepSeekGUI: """GUI主程序""" SUPPORTED_SYMBOLS = ["EURUSD", "GBPUSD", "USDJPY", "USDCHF", "AUDUSD", "USDCAD", "NZDUSD"] def __init__(self, root): self.root = root self.root.title("MT5 + DeepSeek 日内交易分析系统 (带自动交易)") self.root.geometry("1400x900") # MT5 连接 self.mt5_fetcher = None self.mt5_connected = False # DeepSeek 分析器 self.deepseek = None # 交易管理器 self.trade_manager = TradeManager() # 信号存储 self.signals = {} # 线程控制 self.running = False self.auto_analysis = False # 日志队列 self.log_queue = Queue() # 初始化 GUI self.setup_gui() # 启动日志处理 self.process_log_queue() # 启动持仓更新 self.update_positions_display() # 设置窗口关闭事件 self.root.protocol("WM_DELETE_WINDOW", self.on_closing) def setup_gui(self): """设置GUI界面""" # 创建主框架 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(0, weight=1) main_frame.rowconfigure(3, weight=1) # ============================================ # 功能区 # ============================================ control_frame = ttk.LabelFrame(main_frame, text="功能控制", padding="10") control_frame.grid(row=0, column=0, sticky=(tk.W, tk.E), pady=(0, 10)) # 第一行：连接按钮 btn_frame1 = ttk.Frame(control_frame) btn_frame1.pack(fill=tk.X, pady=(0, 5)) self.connect_btn = ttk.Button(btn_frame1, text="连接 MT5", command=self.connect_mt5, width=15) self.connect_btn.pack(side=tk.LEFT, padx=(0, 10)) self.disconnect_btn = ttk.Button(btn_frame1, text="断开 MT5", command=self.disconnect_mt5, width=15, state=tk.DISABLED) self.disconnect_btn.pack(side=tk.LEFT, padx=(0, 10)) self.start_btn = ttk.Button(btn_frame1, text="开始分析", command=self.start_analysis, width=15, state=tk.DISABLED) self.start_btn.pack(side=tk.LEFT, padx=(0, 10)) self.stop_btn = ttk.Button(btn_frame1, text="停止分析", command=self.stop_analysis, width=15, state=tk.DISABLED) self.stop_btn.pack(side=tk.LEFT, padx=(0, 10)) self.manual_close_btn = ttk.Button(btn_frame1, text="全部平仓", command=self.close_all_positions, width=15) self.manual_close_btn.pack(side=tk.LEFT) # 第二行：状态显示 status_frame = ttk.Frame(control_frame) status_frame.pack(fill=tk.X, pady=(5, 0)) ttk.Label(status_frame, text="MT5状态:").pack(side=tk.LEFT, padx=(0, 5)) self.mt5_status_label = ttk.Label(status_frame, text="未连接", foreground="red") self.mt5_status_label.pack(side=tk.LEFT, padx=(0, 20)) ttk.Label(status_frame, text="DeepSeek:").pack(side=tk.LEFT, padx=(0, 5)) self.api_status_label = ttk.Label(status_frame, text="未初始化", foreground="red") self.api_status_label.pack(side=tk.LEFT, padx=(0, 20)) ttk.Label(status_frame, text="分析状态:").pack(side=tk.LEFT, padx=(0, 5)) self.analysis_status_label = ttk.Label(status_frame, text="空闲", foreground="gray") self.analysis_status_label.pack(side=tk.LEFT) # 第三行：倒计时和设置 countdown_frame = ttk.Frame(control_frame) countdown_frame.pack(fill=tk.X, pady=(5, 0)) ttk.Label(countdown_frame, text="下次分析倒计时:").pack(side=tk.LEFT, padx=(0, 10)) self.countdown_label = ttk.Label(countdown_frame, text="--:--:--", font=("Arial", 12, "bold")) self.countdown_label.pack(side=tk.LEFT, padx=(0, 20)) self.auto_mark_var = tk.BooleanVar(value=True) ttk.Checkbutton(countdown_frame, text="自动标记MT5图表", variable=self.auto_mark_var).pack(side=tk.LEFT, padx=(0, 20)) self.auto_trade_var = tk.BooleanVar(value=True) ttk.Checkbutton(countdown_frame, text="自动交易(高置信度信号)", variable=self.auto_trade_var).pack(side=tk.LEFT) # ============================================ # 信号区 # ============================================ signal_frame = ttk.LabelFrame(main_frame, text="实时信号（各品种最新分析结果）", padding="10") signal_frame.grid(row=1, column=0, sticky=(tk.W, tk.E), pady=(0, 10)) # 创建表格 columns = ("品种", "方向", "置信度", "止损", "止盈", "分析逻辑", "交易状态", "时间") self.signal_tree = ttk.Treeview(signal_frame, columns=columns, show="headings", height=6) # 设置列 column_widths = {"品种": 80, "方向": 60, "置信度": 60, "止损": 100, "止盈": 100, "分析逻辑": 350, "交易状态": 100, "时间": 120} for col in columns: self.signal_tree.heading(col, text=col) self.signal_tree.column(col, width=column_widths.get(col, 100)) # 滚动条 signal_scrollbar = ttk.Scrollbar(signal_frame, orient=tk.VERTICAL, command=self.signal_tree.yview) self.signal_tree.configure(yscrollcommand=signal_scrollbar.set) self.signal_tree.pack(side=tk.LEFT, fill=tk.BOTH, expand=True) signal_scrollbar.pack(side=tk.RIGHT, fill=tk.Y) # 配置颜色标签 self.signal_tree.tag_configure('bullish', foreground='green') self.signal_tree.tag_configure('bearish', foreground='red') self.signal_tree.tag_configure('neutral', foreground='orange') # ============================================ # 持仓区 # ============================================ positions_frame = ttk.LabelFrame(main_frame, text="当前持仓", padding="10") positions_frame.grid(row=2, column=0, sticky=(tk.W, tk.E), pady=(0, 10)) # 持仓表格 position_columns = ("订单号", "品种", "方向", "手数", "开仓价", "当前价", "盈亏点数", "盈亏金额", "止损", "止盈") self.position_tree = ttk.Treeview(positions_frame, columns=position_columns, show="headings", height=4) position_widths = {"订单号": 80, "品种": 80, "方向": 60, "手数": 60, "开仓价": 100, "当前价": 100, "盈亏点数": 80, "盈亏金额": 100, "止损": 100, "止盈": 100} for col in position_columns: self.position_tree.heading(col, text=col) self.position_tree.column(col, width=position_widths.get(col, 80)) pos_scrollbar = ttk.Scrollbar(positions_frame, orient=tk.VERTICAL, command=self.position_tree.yview) self.position_tree.configure(yscrollcommand=pos_scrollbar.set) self.position_tree.pack(side=tk.LEFT, fill=tk.BOTH, expand=True) pos_scrollbar.pack(side=tk.RIGHT, fill=tk.Y) # 持仓统计 pos_stats_frame = ttk.Frame(positions_frame) pos_stats_frame.pack(fill=tk.X, pady=(5, 0)) self.position_stats_label = ttk.Label(pos_stats_frame, text="持仓统计: 总手数 0.00 | 浮动盈亏 0.00 USD", font=("Arial", 10)) self.position_stats_label.pack(side=tk.LEFT) # ============================================ # 历史交易记录区 # ============================================ history_frame = ttk.LabelFrame(main_frame, text="历史交易记录", padding="10") history_frame.grid(row=3, column=0, sticky=(tk.W, tk.E, tk.N, tk.S)) # 历史交易表格 history_columns = ("订单号", "品种", "方向", "手数", "开仓价", "平仓价", "盈亏点数", "盈亏金额", "开仓时间", "平仓时间") self.history_tree = ttk.Treeview(history_frame, columns=history_columns, show="headings", height=6) history_widths = {"订单号": 80, "品种": 80, "方向": 60, "手数": 60, "开仓价": 100, "平仓价": 100, "盈亏点数": 80, "盈亏金额": 100, "开仓时间": 130, "平仓时间": 130} for col in history_columns: self.history_tree.heading(col, text=col) self.history_tree.column(col, width=history_widths.get(col, 80)) history_scrollbar = ttk.Scrollbar(history_frame, orient=tk.VERTICAL, command=self.history_tree.yview) self.history_tree.configure(yscrollcommand=history_scrollbar.set) self.history_tree.pack(side=tk.LEFT, fill=tk.BOTH, expand=True) history_scrollbar.pack(side=tk.RIGHT, fill=tk.Y) # 历史统计 history_stats_frame = ttk.Frame(history_frame) history_stats_frame.pack(fill=tk.X, pady=(5, 0)) self.history_stats_label = ttk.Label(history_stats_frame, text="历史统计: 总交易 0 笔 | 胜率 0% | 总盈亏 0.00 USD", font=("Arial", 10)) self.history_stats_label.pack(side=tk.LEFT) ttk.Button(history_stats_frame, text="刷新历史", command=self.refresh_history, width=10).pack(side=tk.RIGHT) # ============================================ # 日志区 # ============================================ log_frame = ttk.LabelFrame(main_frame, text="运行日志", padding="10") log_frame.grid(row=4, column=0, sticky=(tk.W, tk.E, tk.N, tk.S)) self.log_text = scrolledtext.ScrolledText(log_frame, height=10, wrap=tk.WORD) self.log_text.pack(fill=tk.BOTH, expand=True) # 配置日志颜色 self.log_text.tag_config("INFO", 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.log_text.tag_config("SIGNAL", foreground="purple", font=("Arial", 10, "bold")) self.log_text.tag_config("TRADE", foreground="darkgreen", font=("Arial", 10, "bold")) def add_log(self, message: str, level: str = "INFO"): """添加日志""" timestamp = datetime.now().strftime("%H:%M:%S") log_message = f"[{timestamp}] {message}\n" self.log_queue.put((log_message, level)) def process_log_queue(self): """处理日志队列（在主线程中）""" try: while not self.log_queue.empty(): log_message, level = self.log_queue.get_nowait() self.log_text.insert(tk.END, log_message, level) self.log_text.see(tk.END) except: pass finally: self.root.after(100, self.process_log_queue) def connect_mt5(self): """连接MT5""" try: self.mt5_fetcher = MT5DataFetcher("EURUSD") if self.mt5_fetcher.connect(): self.mt5_connected = True self.mt5_status_label.config(text="已连接", foreground="green") self.connect_btn.config(state=tk.DISABLED) self.disconnect_btn.config(state=tk.NORMAL) self.start_btn.config(state=tk.NORMAL) self.add_log("✅ MT5 连接成功", "SUCCESS") # 初始化DeepSeek self.init_deepseek() # 加载历史交易 self.refresh_history() else: self.add_log("❌ MT5 连接失败", "ERROR") except Exception as e: self.add_log(f"❌ MT5 连接异常: {e}", "ERROR") def disconnect_mt5(self): """断开MT5""" if self.mt5_fetcher: self.mt5_fetcher.disconnect() self.mt5_connected = False self.mt5_status_label.config(text="未连接", foreground="red") self.connect_btn.config(state=tk.NORMAL) self.disconnect_btn.config(state=tk.DISABLED) self.start_btn.config(state=tk.DISABLED) self.stop_btn.config(state=tk.DISABLED) self.add_log("🔌 MT5 已断开", "WARNING") def init_deepseek(self): """初始化DeepSeek""" try: api_key = os.getenv('DEEPSEEK_API_KEY') if not api_key: self.add_log("⚠️ 未找到DEEPSEEK_API_KEY环境变量", "WARNING") api_key = "sk-7845bf14aa97444e9d8f6c271e4c9a72" self.deepseek = DeepSeekAnalyzer(api_key) self.api_status_label.config(text="已就绪", foreground="green") self.add_log("✅ DeepSeek API 初始化成功", "SUCCESS") except Exception as e: self.add_log(f"❌ DeepSeek API 初始化失败: {e}", "ERROR") def start_analysis(self): """开始分析""" if not self.mt5_connected or not self.deepseek: self.add_log("❌ 请先连接MT5并确保DeepSeek已初始化", "ERROR") return self.auto_analysis = True self.start_btn.config(state=tk.DISABLED) self.stop_btn.config(state=tk.NORMAL) self.add_log("🚀 开始自动分析模式", "SUCCESS") # 启动分析线程 self.analysis_thread = Thread(target=self.analysis_loop, daemon=True) self.analysis_thread.start() # 启动倒计时更新 self.update_countdown() def stop_analysis(self): """停止分析""" self.auto_analysis = False self.running = False self.start_btn.config(state=tk.NORMAL) self.stop_btn.config(state=tk.DISABLED) self.analysis_status_label.config(text="已停止", foreground="red") self.add_log("⏹️ 停止自动分析模式", "WARNING") def close_all_positions(self): """全部平仓""" if not self.mt5_connected: self.add_log("❌ MT5未连接", "ERROR") return if messagebox.askyesno("确认", "确定要平掉所有持仓吗？"): positions = self.trade_manager.get_positions() if not positions: self.add_log("当前无持仓", "INFO") return success_count = 0 for pos in positions: if self.trade_manager.close_position(pos): success_count += 1 self.add_log(f"✅ 平仓 {pos.symbol} ticket={pos.ticket}", "TRADE") self.add_log(f"📊 平仓完成: {success_count}/{len(positions)} 成功", "SUCCESS") def analysis_loop(self): """分析循环""" self.running = True while self.running and self.auto_analysis: try: # 计算到下一个M15整点的时间 now = datetime.now() minutes = now.minute next_minute = ((minutes // 15) + 1) * 15 if next_minute == 60: next_minute = 0 next_hour = now.hour + 1 else: next_hour = now.hour next_time = now.replace(hour=next_hour, minute=next_minute, second=0, microsecond=0) sleep_seconds = (next_time - now).total_seconds() if sleep_seconds > 0: self.root.after(0, lambda: self.analysis_status_label.config(text=f"等待下次分析", foreground="orange")) time.sleep(sleep_seconds) # 执行分析 if self.running and self.auto_analysis: self.root.after(0, lambda: self.analysis_status_label.config(text="分析中...", foreground="blue")) self.add_log("📊 开始分析所有货币对", "INFO") # 并行分析所有品种 threads = [] for symbol in self.SUPPORTED_SYMBOLS: thread = Thread(target=self.analyze_symbol, args=(symbol,)) thread.daemon = True thread.start() threads.append(thread) # 等待所有线程完成 for thread in threads: thread.join() self.add_log("✅ 本轮分析完成", "SUCCESS") self.root.after(0, lambda: self.analysis_status_label.config(text="空闲", foreground="green")) except Exception as e: self.add_log(f"❌ 分析循环异常: {e}", "ERROR") time.sleep(1) def analyze_symbol(self, symbol: str): """分析单个品种""" if not self.mt5_connected or not self.deepseek: return try: self.add_log(f"🔍 开始分析 {symbol}", "INFO") # 更新MT5品种 self.mt5_fetcher.symbol = symbol # 获取数据 df_m15 = self.mt5_fetcher.get_ohlc_data('M15', 100) df_h1 = self.mt5_fetcher.get_ohlc_data('H1', 50) df_d1 = self.mt5_fetcher.get_ohlc_data('D1', 50) current_price_data = self.mt5_fetcher.get_current_price() current_price = current_price_data['bid'] # 生成描述 generator = SemanticDescriptionGenerator(symbol) d1_desc = generator.generate_d1_description(df_d1) h1_desc = generator.generate_h1_description(df_h1) m15_desc = generator.generate_m15_description(df_m15, current_price) # 调用 DeepSeek analysis = self.deepseek.analyze_direction( symbol, d1_desc, h1_desc, m15_desc, current_price ) trade_status = "未交易" trade_result = None # 如果是高置信度信号且自动交易开启 if analysis.get('confidence') == '高' and self.auto_trade_var.get(): self.add_log(f"🎯 {symbol} 检测到高置信度{analysis.get('direction')}信号！开始执行自动交易...", "SIGNAL") # 解析止损止盈 stop_loss = None take_profit = None if 'stop_loss' in analysis and analysis['stop_loss'] != '未提供': try: stop_loss = float(analysis['stop_loss']) except: pass if 'take_profit' in analysis and analysis['take_profit'] != '未提供': try: take_profit = float(analysis['take_profit']) except: pass # 执行交易 trade_result = self.trade_manager.execute_signal( symbol, analysis.get('direction'), stop_loss, take_profit ) if trade_result['executed']: trade_status = f"✅ {trade_result['action']}" self.add_log(f"💰 {trade_result['message']}", "TRADE") else: trade_status = f"❌ {trade_result['message'][:30]}" self.add_log(f"⚠️ {trade_result['message']}", "WARNING") # 标记到MT5图表 if self.auto_mark_var.get(): mark_thread = Thread(target=self.mt5_fetcher.place_arrow_on_chart, args=(symbol, analysis.get('direction'), current_price)) mark_thread.daemon = True mark_thread.start() # 更新信号表 self.root.after(0, self.update_signal_table, symbol, analysis, trade_status) self.add_log(f"✅ {symbol} 分析完成 - {analysis.get('direction')} ({analysis.get('confidence')})", "SUCCESS") except Exception as e: self.add_log(f"❌ 分析 {symbol} 失败: {e}", "ERROR") def update_signal_table(self, symbol: str, analysis: Dict, trade_status: str = ""): """更新信号表格""" existing_items = self.signal_tree.get_children() item_to_update = None for item in existing_items: values = self.signal_tree.item(item)['values'] if values and len(values) > 0 and values[0] == symbol: item_to_update = item break values = ( symbol, analysis.get('direction', 'N/A'), analysis.get('confidence', 'N/A'), analysis.get('stop_loss', '-'), analysis.get('take_profit', '-'), (analysis.get('reasoning', 'N/A')[:50] + "...") if len(analysis.get('reasoning', 'N/A')) > 50 else analysis.get('reasoning', 'N/A'), trade_status if trade_status else "-", datetime.now().strftime("%H:%M:%S") ) direction = analysis.get('direction', '') if direction == '看涨': tag = 'bullish' elif direction == '看跌': tag = 'bearish' else: tag = 'neutral' if item_to_update: self.signal_tree.item(item_to_update, values=values, tags=(tag,)) else: self.signal_tree.insert("", 0, values=values, tags=(tag,)) def update_positions_display(self): """更新持仓显示""" if not self.mt5_connected: self.root.after(2000, self.update_positions_display) return try: # 清除现有显示 for item in self.position_tree.get_children(): self.position_tree.delete(item) positions = self.trade_manager.get_positions() total_volume = 0 total_profit = 0 if positions: for pos in positions: # 获取当前价格 tick = mt5.symbol_info_tick(pos.symbol) if tick: current_price = tick.bid if pos.type == mt5.POSITION_TYPE_SELL else tick.ask price_diff = current_price - pos.price_open if pos.type == mt5.POSITION_TYPE_SELL: price_diff = -price_diff point_value = self.get_point_value(pos.symbol) profit_points = price_diff / point_value if point_value else 0 profit_usd = pos.profit else: current_price = pos.price_open profit_points = 0 profit_usd = 0 direction_text = "买入" if pos.type == mt5.POSITION_TYPE_BUY else "卖出" self.position_tree.insert("", tk.END, values=( pos.ticket, pos.symbol, direction_text, f"{pos.volume:.2f}", f"{pos.price_open:.5f}", f"{current_price:.5f}", f"{profit_points:.1f}", f"{profit_usd:.2f}", f"{pos.sl:.5f}" if pos.sl else "-", f"{pos.tp:.5f}" if pos.tp else "-" )) total_volume += pos.volume total_profit += profit_usd profit_color = "green" if total_profit >= 0 else "red" self.position_stats_label.config( text=f"持仓统计: 总手数 {total_volume:.2f} | 浮动盈亏 {total_profit:.2f} USD", foreground=profit_color ) else: self.position_stats_label.config(text="持仓统计: 总手数 0.00 | 浮动盈亏 0.00 USD", foreground="black") except Exception as e: print(f"更新持仓显示异常: {e}") self.root.after(2000, self.update_positions_display) def get_point_value(self, symbol: str) -> float: """获取品种的点值""" try: symbol_info = mt5.symbol_info(symbol) if symbol_info: return symbol_info.point return 0.00001 except: return 0.00001 def refresh_history(self): """刷新历史交易记录""" if not self.mt5_connected: return try: # 清除现有显示 for item in self.history_tree.get_children(): self.history_tree.delete(item) # 获取历史订单（最近30天） from_date = datetime.now() - timedelta(days=30) history_orders = mt5.history_deals_get(from_date, datetime.now()) if history_orders: # 筛选出平仓订单（关联position_id的deal） deals_by_position = {} for deal in history_orders: if deal.position_id != 0: if deal.position_id not in deals_by_position: deals_by_position[deal.position_id] = [] deals_by_position[deal.position_id].append(deal) # 处理每个持仓 for position_id, deals in deals_by_position.items(): # 找出开仓和平仓的deal entry_deal = None exit_deal = None for deal in deals: if deal.entry == mt5.ORDER_ENTRY_IN: entry_deal = deal elif deal.entry == mt5.ORDER_ENTRY_OUT: exit_deal = deal if entry_deal and exit_deal: profit = exit_deal.profit + exit_deal.swap + exit_deal.commission direction_text = "买入" if entry_deal.type == mt5.ORDER_TYPE_BUY else "卖出" point_value = self.get_point_value(entry_deal.symbol) profit_points = profit / point_value if point_value else 0 self.history_tree.insert("", tk.END, values=( entry_deal.position_id, entry_deal.symbol, direction_text, f"{entry_deal.volume:.2f}", f"{entry_deal.price:.5f}", f"{exit_deal.price:.5f}", f"{profit_points:.1f}", f"{profit:.2f}", datetime.fromtimestamp(entry_deal.time).strftime("%Y-%m-%d %H:%M:%S"), datetime.fromtimestamp(exit_deal.time).strftime("%Y-%m-%d %H:%M:%S") )) # 计算统计数据 total_trades = len(self.history_tree.get_children()) total_profit = 0 winning_trades = 0 for item in self.history_tree.get_children(): values = self.history_tree.item(item)['values'] if len(values) >= 8: profit = float(values[7]) total_profit += profit if profit > 0: winning_trades += 1 win_rate = (winning_trades / total_trades * 100) if total_trades > 0 else 0 profit_color = "green" if total_profit >= 0 else "red" self.history_stats_label.config( text=f"历史统计: 总交易 {total_trades} 笔 | 胜率 {win_rate:.1f}% | 总盈亏 {total_profit:.2f} USD", foreground=profit_color ) else: self.history_stats_label.config(text="历史统计: 总交易 0 笔 | 胜率 0% | 总盈亏 0.00 USD", foreground="black") except Exception as e: print(f"刷新历史记录异常: {e}") def update_countdown(self): """更新倒计时""" if not self.auto_analysis: self.countdown_label.config(text="--:--:--") return now = datetime.now() minutes = now.minute next_minute = ((minutes // 15) + 1) * 15 if next_minute == 60: next_minute = 0 next_hour = now.hour + 1 else: next_hour = now.hour next_time = now.replace(hour=next_hour, minute=next_minute, second=0, microsecond=0) remaining = (next_time - now).total_seconds() if remaining > 0 and remaining <= 900: minutes = int(remaining // 60) seconds = int(remaining % 60) self.countdown_label.config(text=f"{minutes:02d}:{seconds:02d}") else: self.countdown_label.config(text="00:00") self.root.after(1000, self.update_countdown) def on_closing(self): """关闭窗口""" self.running = False self.auto_analysis = False if self.mt5_fetcher: self.mt5_fetcher.disconnect() self.root.destroy() # ============================================ # 主程序入口 # ============================================ def main(): """主函数""" root = tk.Tk() app = MT5DeepSeekGUI(root) root.mainloop() if __name__ == "__main__": main()