# -*- coding: utf-8 -*- """ MACD Histogram Reversal 自动化交易系统 =============================================================== 策略：MACD 柱状图反转 监听：7 个主流货币对（M15） 功能：实时扫描 → 信号检测 → 自动下单 → GUI 可视化 =============================================================== """ import MetaTrader5 as mt5 import pandas as pd import numpy as np import tkinter as tk from tkinter import ttk, scrolledtext import threading import time import queue from datetime import datetime, timezone, timedelta import sys import os import logging # ================================================================ # CONFIG — 可在此处修改参数 # ================================================================ SYMBOLS = [ "EURUSD", "GBPUSD", "USDJPY", "USDCHF", "AUDUSD", "NZDUSD", "USDCAD" ] MAGIC_BASE = 88880000 LOT = 0.1 SLIPPAGE_PIPS = 1 # MACD 参数 MACD_FAST = 12 MACD_SLOW = 26 MACD_SIGNAL = 9 ATR_PERIOD = 14 # 止损止盈 ATR_SL_MULT = 1.5 ATR_TP_MULT = 2.0 # 交易时段（北京时间） TRADE_START = 8 TRADE_END = 20 EOD_HOUR = 21 # 其他限制 MAX_DAILY_TRADES = 2 MAX_BARS_HELD = 20 SCAN_INTERVAL = 5 # 秒 BARS_FOR_CALC = 100 # 计算 MACD 用的 K 线数量 # GUI 刷新间隔（毫秒） GUI_REFRESH_MS = 500 # ================================================================ # GLOBAL STATE # ================================================================ running = False account_info = {} last_bar_time = {} daily_trade_count = {} last_signal_time = {} signals_log = [] positions_log = [] log_lines = [] log_lock = threading.Lock() # ================================================================ # HELPERS - 时间相关 # ================================================================ def now_bj(): """当前北京时间（UTC+8）""" return datetime.now(timezone(timedelta(hours=8))) def now_bj_str(): """当前北京时间字符串 HH:MM:SS""" return now_bj().strftime("%H:%M:%S") def get_bj_hour(): """获取当前北京时间的小时数（0-23）""" return now_bj().hour def get_current_date_str(): """获取当前北京时间的日期字符串""" return now_bj().strftime("%Y-%m-%d") def is_trading_time(): """判断当前是否在交易时段""" hour = get_bj_hour() return TRADE_START <= hour < TRADE_END def is_eod_time(): """判断当前是否应该强制平仓""" hour = get_bj_hour() return hour >= EOD_HOUR def next_bar_close_seconds(): """距离下一个 M15 收盘还剩多少秒""" now = now_bj() minute = now.minute second = now.second quarter = (minute // 15) * 15 + 15 if quarter == 60: secs_left = (60 - minute) * 60 - second else: secs_left = (quarter - minute) * 60 - second return max(1, secs_left) # ================================================================ # HELPERS - 交易相关 # ================================================================ def get_pip_size(symbol): """根据货币对返回 pip 大小""" if "JPY" in symbol: return 0.01 return 0.0001 def get_pip_factor(symbol): """每个 pip 的钱（0.1 手）""" return 10.0 # ================================================================ # HELPERS - 日志和状态 # ================================================================ def add_log(msg, tag="INFO"): """添加日志""" ts = now_bj_str() line = f"[{ts}] [{tag}] {msg}" with log_lock: log_lines.append(line) if len(log_lines) > 500: log_lines.pop(0) def add_signal(sym, direction, price, macd_val, hist_val, atr_val): """记录信号""" ts = now_bj_str() signals_log.insert(0, { "time": ts, "symbol": sym, "direction": direction, "price": price, "macd": macd_val, "hist": hist_val, "atr": atr_val }) if len(signals_log) > 100: signals_log.pop() def add_position(ticket, sym, direction, entry, sl, tp): """添加持仓记录""" positions_log.append({ "ticket": ticket, "symbol": sym, "direction": direction, "entry": entry, "sl": sl, "tp": tp }) def remove_position(ticket): """移除持仓记录""" for i, p in enumerate(positions_log): if p["ticket"] == ticket: positions_log.pop(i) break def clear_positions(): """清空持仓记录""" positions_log.clear() # ================================================================ # MT5 CONNECTION # ================================================================ def mt5_connect(): """连接 MT5""" if not mt5.initialize(): add_log("MT5 初始化失败，请确认 MT5 已启动并允许 DLL 调用", "ERROR") return False acc = mt5.account_info() if acc is None: add_log("无法获取账户信息", "ERROR") mt5.shutdown() return False global account_info account_info = { "login": acc.login, "server": acc.server, "balance": acc.balance, "equity": acc.equity, "margin": acc.margin, "free_margin": acc.margin_free, } add_log(f"MT5 已连接 | Account: {acc.login} | Server: {acc.server}", "MT5") add_log(f"Balance: ${acc.balance:.2f} | Equity: ${acc.equity:.2f}", "MT5") return True def mt5_disconnect(): """断开 MT5""" mt5.shutdown() add_log("MT5 已断开", "MT5") def mt5_get_rates(symbol, count=100): """获取 K 线数据""" rates = mt5.copy_rates_from_pos(symbol, mt5.TIMEFRAME_M15, 0, count) if rates is None or len(rates) == 0: return None df = pd.DataFrame(rates) df["time"] = pd.to_datetime(df["time"], unit="s", utc=True) \ .dt.tz_convert("Asia/Shanghai").dt.tz_localize(None) return df def mt5_get_tick(symbol): """获取当前报价""" tick = mt5.symbol_info_tick(symbol) return tick def mt5_get_positions(): """获取所有持仓""" return mt5.positions_get() def mt5_has_position(symbol, magic): """检查是否存在同货币同 magic 的持仓""" positions = mt5.positions_get(symbol=symbol) if positions is None: return False for pos in positions: if pos.magic == magic: return True return False def mt5_place_order(symbol, direction, lot, price, sl, tp, magic, comment=""): """下单 - 市价单版本""" order_type = mt5.ORDER_TYPE_BUY if direction == 1 else mt5.ORDER_TYPE_SELL # 构建请求 request = { "action": mt5.TRADE_ACTION_DEAL, "symbol": symbol, "volume": lot, "type": order_type, "price": price, "sl": sl, "tp": tp, "deviation": 20, # 滑点容忍度 "magic": magic, "comment": comment, "type_time": mt5.ORDER_TIME_GTC, "type_filling": mt5.ORDER_FILLING_FOK, # 先尝试 FOK } # 发送订单 result = mt5.order_send(request) # 如果 FOK 失败，尝试 IOC if result is not None and result.retcode != mt5.TRADE_RETCODE_DONE: if "filling" in result.comment.lower() or result.retcode == 10030: add_log(f"{symbol}: FOK 填充模式失败，尝试 IOC 模式...", "DEBUG") request["type_filling"] = mt5.ORDER_FILLING_IOC result = mt5.order_send(request) # 如果还是失败，尝试 RETURN 模式（纯市价） if result is not None and result.retcode != mt5.TRADE_RETCODE_DONE: if "filling" in result.comment.lower() or result.retcode == 10030: add_log(f"{symbol}: IOC 填充模式失败，尝试 RETURN 模式...", "DEBUG") request["type_filling"] = mt5.ORDER_FILLING_RETURN # RETURN 模式不需要指定价格 if order_type == mt5.ORDER_TYPE_BUY: request["price"] = mt5.symbol_info_tick(symbol).ask else: request["price"] = mt5.symbol_info_tick(symbol).bid result = mt5.order_send(request) # 检查结果 if result is None: err = mt5.last_error() add_log(f"{symbol} 下单失败: error={err[0]} {err[1]}", "ERROR") return None if result.retcode != mt5.TRADE_RETCODE_DONE: add_log(f"{symbol} 下单失败: retcode={result.retcode} {result.comment}", "ERROR") return None add_log(f"{symbol} 下单成功: {'BUY' if direction > 0 else 'SELL'} {lot}@{price:.5f} " f"SL={sl:.5f} TP={tp:.5f} ticket={result.order}", "ORDER") return result # ================================================================ # STRATEGY ENGINE # ================================================================ def calc_indicators(df): """计算 MACD + ATR""" df = df.copy() ema_fast = df["close"].ewm(span=MACD_FAST, adjust=False).mean() ema_slow = df["close"].ewm(span=MACD_SLOW, adjust=False).mean() macd_line = ema_fast - ema_slow signal_line = macd_line.ewm(span=MACD_SIGNAL, adjust=False).mean() df["macd"] = macd_line.shift(1) df["macd_signal"] = signal_line.shift(1) df["macd_hist"] = (df["macd"] - df["macd_signal"]) tr = np.maximum(df["high"] - df["low"], np.maximum(np.abs(df["high"] - df["close"].shift(1)), np.abs(df["low"] - df["close"].shift(1)))) df["atr"] = tr.rolling(ATR_PERIOD).mean().shift(1) return df def detect_signal(df, symbol): """检测 MACD 柱状图反转信号""" if len(df) < 4: return None hist_prev = df["macd_hist"].iloc[-3] hist_curr = df["macd_hist"].iloc[-2] macd_curr = df["macd"].iloc[-2] macd_sig_curr = df["macd_signal"].iloc[-2] atr_curr = df["atr"].iloc[-2] bar_time = df["time"].iloc[-2] if pd.isna(hist_prev) or pd.isna(hist_curr) or pd.isna(atr_curr): return None sl_mult = ATR_SL_MULT * atr_curr tp_mult = ATR_TP_MULT * atr_curr # 金叉（柱状图从负转正） if hist_prev < 0 and hist_curr > 0: direction = 1 tick = mt5_get_tick(symbol) if tick is None: return None price = tick.ask sl = price - sl_mult tp = price + tp_mult return ("BUY", price, sl, tp, macd_curr, macd_sig_curr, hist_curr, atr_curr, bar_time) # 死叉（柱状图从正转负） if hist_prev > 0 and hist_curr < 0: direction = -1 tick = mt5_get_tick(symbol) if tick is None: return None price = tick.bid sl = price + sl_mult tp = price - tp_mult return ("SELL", price, sl, tp, macd_curr, macd_sig_curr, hist_curr, atr_curr, bar_time) return None # ================================================================ # SCANNER THREAD # ================================================================ def scanner_thread(gui_queue): """主扫描线程""" global running, last_bar_time, daily_trade_count, last_signal_time add_log("=" * 50, "SYSTEM") add_log("扫描线程启动", "SYSTEM") add_log(f"监听货币对: {', '.join(SYMBOLS)}", "SYSTEM") add_log(f"交易时段: {TRADE_START:02d}:00 - {TRADE_END:02d}:00 (北京时间)", "SYSTEM") add_log(f"当前时间: {now_bj().strftime('%Y-%m-%d %H:%M:%S')} (北京时间)", "SYSTEM") add_log(f"交易状态: {'🟢 交易中' if is_trading_time() else '🔴 非交易时段'}", "SYSTEM") for sym in SYMBOLS: last_bar_time[sym] = None daily_trade_count[sym] = 0 last_signal_time[sym] = None while running: try: current_hour = get_bj_hour() current_time_str = now_bj_str() # 检查交易时段 if not is_trading_time(): # 非交易时段，每 30 秒输出一次提示 if int(time.time()) % 30 == 0: add_log(f"非交易时段 ({current_time_str})，等待中...", "SCAN") time.sleep(SCAN_INTERVAL) continue for i, sym in enumerate(SYMBOLS): magic = MAGIC_BASE # 每日交易次数限制 if daily_trade_count.get(sym, 0) >= MAX_DAILY_TRADES: continue # 获取 K 线数据 df = mt5_get_rates(sym, BARS_FOR_CALC) if df is None or len(df) < 4: add_log(f"{sym}: 无法获取 K 线数据", "ERROR") continue df = calc_indicators(df) # 检查新 bar 是否已形成 last_bar_t = df["time"].iloc[-2] if last_bar_time.get(sym) is not None and last_bar_t == last_bar_time[sym]: continue last_bar_time[sym] = last_bar_t # 获取当前 bar 数据 bar_close = df["close"].iloc[-2] bar_macd = df["macd"].iloc[-2] bar_sig = df["macd_signal"].iloc[-2] bar_hist = df["macd_hist"].iloc[-2] bar_atr = df["atr"].iloc[-2] # 推送分析数据到 GUI gui_queue.put(("analysis_data", { "symbol": sym, "time": last_bar_t.strftime("%H:%M"), "close": bar_close, "macd": bar_macd, "signal": bar_sig, "hist": bar_hist, "atr": bar_atr })) add_log(f"{sym}: Bar {last_bar_t.strftime('%m-%d %H:%M')} | Close={bar_close:.5f} | " f"MACD={bar_macd:.6f} | Hist={bar_hist:.6f} | ATR={bar_atr:.5f}", "DATA") # 检测信号 signal = detect_signal(df, sym) if signal is None: continue direction_str, price, sl, tp, macd_v, sig_v, hist_v, atr_v, bar_t = signal direction_int = 1 if direction_str == "BUY" else -1 # 检查是否同 bar 重复信号 bar_key = f"{sym}_{bar_t}" if last_signal_time.get(sym) == bar_key: add_log(f"{sym}: 同 bar 重复信号，跳过", "SKIP") continue last_signal_time[sym] = bar_key # 检查是否已有持仓 if mt5_has_position(sym, magic): add_log(f"{sym}: 已有持仓，跳过", "SKIP") continue add_log(f"{sym}: ★ 信号触发！{direction_str} | MACD={macd_v:.6f} | " f"Hist={hist_v:.6f} | ATR={atr_v:.5f}", "SIGNAL") add_log(f"{sym}: 下单中... 价格={price:.5f} SL={sl:.5f} TP={tp:.5f}", "ORDER") # 下单 result = mt5_place_order(sym, direction_int, LOT, price, sl, tp, magic, f"MACD_Hist_{direction_str[0]}") if result is not None and result.order > 0: daily_trade_count[sym] = daily_trade_count.get(sym, 0) + 1 add_signal(sym, direction_str, price, macd_v, hist_v, atr_v) gui_queue.put(("new_signal", { "time": now_bj_str(), "symbol": sym, "direction": direction_str, "price": price, "macd": macd_v, "hist": hist_v, "atr": atr_v, "sl": sl, "tp": tp, "ticket": result.order })) add_position(result.order, sym, direction_str, price, sl, tp) # 刷新持仓 _refresh_positions() # 检查 EOD 平仓 _check_exits() # 每日重置 _reset_daily_counts() # 发送心跳 gui_queue.put(("heartbeat", None)) except Exception as e: add_log(f"扫描异常: {type(e).__name__}: {e}", "ERROR") import traceback add_log(traceback.format_exc(), "ERROR") time.sleep(SCAN_INTERVAL) def _refresh_positions(): """同步 MT5 持仓到本地状态""" try: mt5_pos = mt5_get_positions() if mt5_pos is None: return current_tickets = {pos.ticket for pos in mt5_pos} to_remove = [p["ticket"] for p in positions_log if p["ticket"] not in current_tickets] for t in to_remove: for p in positions_log: if p["ticket"] == t: add_log(f"平仓: ticket={t} {p['symbol']} {p['direction']}", "CLOSE") remove_position(t) break except Exception as e: add_log(f"持仓同步异常: {e}", "ERROR") def _check_exits(): """检查 EOD 强制平仓""" try: if not is_eod_time(): return mt5_pos = mt5_get_positions() if mt5_pos is None: return for pos in mt5_pos: magic = pos.magic if not (MAGIC_BASE <= magic < MAGIC_BASE + len(SYMBOLS)): continue tick = mt5_get_tick(pos.symbol) if tick is None: continue price = tick.bid if pos.type == mt5.ORDER_TYPE_BUY else tick.ask request = { "action": mt5.TRADE_ACTION_DEAL, "symbol": pos.symbol, "volume": pos.volume, "type": mt5.ORDER_TYPE_SELL if pos.type == mt5.ORDER_TYPE_BUY else mt5.ORDER_TYPE_BUY, "price": price, "deviation": 20, "magic": pos.magic, "comment": "EOD_Close", "type_time": mt5.ORDER_TIME_GTC, "type_filling": mt5.ORDER_FILLING_IOC, } result = mt5.order_send(request) if result and result.retcode == mt5.TRADE_RETCODE_DONE: add_log(f"EOD 平仓: {pos.symbol} ticket={pos.ticket}", "CLOSE") remove_position(pos.ticket) except Exception as e: add_log(f"EOD 平仓异常: {e}", "ERROR") _daily_reset_done = None def _reset_daily_counts(): """每日重置计数器""" global _daily_reset_done today = get_current_date_str() if _daily_reset_done != today: daily_trade_count.clear() last_signal_time.clear() _daily_reset_done = today add_log(f"每日计数已重置 ({today})", "SYSTEM") # ================================================================ # GUI - 简洁大方版 # ================================================================ class TradingGUI: def __init__(self, root, gui_queue): self.root = root self.gui_queue = gui_queue self.root.title("MACD 外汇交易系统") self.root.geometry("1300x750") self.root.configure(bg="#f0f0f0") # 存储各货币对的最新分析数据 self.symbol_data = {sym: {} for sym in SYMBOLS} self._setup_styles() self._build_ui() self._poll() def _setup_styles(self): style = ttk.Style() style.theme_use("clam") # 颜色方案 self.colors = { "bg": "#f5f5f5", "card": "#ffffff", "header": "#2c3e50", "text": "#333333", "text_light": "#666666", "border": "#e0e0e0", "success": "#27ae60", "danger": "#e74c3c", "warning": "#f39c12", "info": "#3498db", "buy": "#2ecc71", "sell": "#e74c3c" } style.configure("Card.TFrame", background=self.colors["card"], relief="flat", borderwidth=1) style.configure("Header.TLabel", background=self.colors["header"], foreground="white", font=("Microsoft YaHei UI", 11, "bold"), padding=(12, 8)) style.configure("Section.TLabel", background=self.colors["card"], foreground=self.colors["text"], font=("Microsoft YaHei UI", 11, "bold"), padding=(5, 5)) style.configure("Info.TLabel", background=self.colors["card"], foreground=self.colors["text_light"], font=("Consolas", 9)) # Treeview 样式 style.configure("Custom.Treeview", background=self.colors["card"], foreground=self.colors["text"], fieldbackground=self.colors["card"], rowheight=28, font=("Consolas", 9)) style.configure("Custom.Treeview.Heading", background="#ecf0f1", foreground=self.colors["text"], font=("Microsoft YaHei UI", 9, "bold"), relief="flat") style.map("Custom.Treeview", background=[("selected", "#d5dbdb")], foreground=[("selected", self.colors["text"])]) def _build_ui(self): # ==================== 顶部标题栏 ==================== header = tk.Frame(self.root, bg=self.colors["header"], height=55) header.pack(fill="x", padx=0, pady=0) header.pack_propagate(False) # 标题和状态 title_frame = tk.Frame(header, bg=self.colors["header"]) title_frame.pack(side="left", padx=20, pady=8) tk.Label(title_frame, text="📊 MACD 外汇交易系统", bg=self.colors["header"], fg="white", font=("Microsoft YaHei UI", 16, "bold")).pack(anchor="w") self.status_label = tk.Label(title_frame, text="⚪ 已停止", bg=self.colors["header"], fg="#ecf0f1", font=("Consolas", 9)) self.status_label.pack(anchor="w") # 账户信息 self.account_frame = tk.Frame(header, bg=self.colors["header"]) self.account_frame.pack(side="right", padx=20, pady=10) self.balance_label = tk.Label(self.account_frame, text="余额: --", bg=self.colors["header"], fg="white", font=("Consolas", 10, "bold")) self.balance_label.pack() self.equity_label = tk.Label(self.account_frame, text="净值: --", bg=self.colors["header"], fg="#bdc3c7", font=("Consolas", 9)) self.equity_label.pack() # 控制按钮 btn_frame = tk.Frame(header, bg=self.colors["header"]) btn_frame.pack(side="right", padx=10) self.start_btn = tk.Button(btn_frame, text="▶ 启动", bg=self.colors["success"], fg="white", font=("Microsoft YaHei UI", 10, "bold"), padx=20, pady=4, relief="flat", cursor="hand2", command=self._on_start) self.start_btn.pack(side="left", padx=5) self.stop_btn = tk.Button(btn_frame, text="■ 停止", bg=self.colors["danger"], fg="white", font=("Microsoft YaHei UI", 10, "bold"), padx=20, pady=4, relief="flat", cursor="hand2", state="disabled", command=self._on_stop) self.stop_btn.pack(side="left", padx=5) # ==================== 货币对状态条 ==================== status_bar = tk.Frame(self.root, bg=self.colors["card"], height=40) status_bar.pack(fill="x", padx=10, pady=(10, 0)) tk.Label(status_bar, text="货币对状态", bg=self.colors["card"], fg=self.colors["text_light"], font=("Microsoft YaHei UI", 9)).pack(side="left", padx=15, pady=8) self.symbol_frames = {} for sym in SYMBOLS: frame = tk.Frame(status_bar, bg=self.colors["card"]) frame.pack(side="left", padx=3, pady=5) lbl = tk.Label(frame, text=sym, bg=self.colors["border"], fg=self.colors["text_light"], font=("Consolas", 9, "bold"), padx=12, pady=2, relief="flat") lbl.pack() self.symbol_frames[sym] = lbl self.countdown_label = tk.Label(status_bar, text="⏱ 下一K线: --:--", bg=self.colors["card"], fg=self.colors["warning"], font=("Consolas", 9)) self.countdown_label.pack(side="right", padx=15, pady=8) # ==================== 主内容区域 ==================== main = tk.Frame(self.root, bg=self.colors["bg"]) main.pack(fill="both", expand=True, padx=10, pady=10) # 左侧：信号历史 left_card = ttk.Frame(main, style="Card.TFrame") left_card.pack(side="left", fill="both", expand=True, padx=(0, 5)) ttk.Label(left_card, text="📈 信号历史", style="Section.TLabel").pack(anchor="w", padx=12, pady=(10, 5)) tree_frame = tk.Frame(left_card, bg=self.colors["card"]) tree_frame.pack(fill="both", expand=True, padx=10, pady=(0, 10)) cols = ("time", "symbol", "dir", "price", "hist", "atr") self.signal_tree = ttk.Treeview(tree_frame, columns=cols, show="headings", height=12, style="Custom.Treeview") headers = {"time": "时间", "symbol": "货币", "dir": "方向", "price": "入场价", "hist": "Hist", "atr": "ATR"} widths = {"time": 90, "symbol": 80, "dir": 60, "price": 90, "hist": 90, "atr": 80} for col in cols: self.signal_tree.heading(col, text=headers[col]) self.signal_tree.column(col, width=widths[col], anchor="center") scrollbar = ttk.Scrollbar(tree_frame, orient="vertical", command=self.signal_tree.yview) self.signal_tree.configure(yscrollcommand=scrollbar.set) self.signal_tree.pack(side="left", fill="both", expand=True) scrollbar.pack(side="right", fill="y") self.signal_tree.tag_configure("BUY", foreground=self.colors["buy"]) self.signal_tree.tag_configure("SELL", foreground=self.colors["sell"]) # 右侧：日志区 right_card = ttk.Frame(main, style="Card.TFrame") right_card.pack(side="right", fill="both", expand=True, padx=(5, 0)) ttk.Label(right_card, text="📋 实时日志", style="Section.TLabel").pack(anchor="w", padx=12, pady=(10, 5)) log_frame = tk.Frame(right_card, bg=self.colors["card"]) log_frame.pack(fill="both", expand=True, padx=10, pady=(0, 10)) self.log_text = scrolledtext.ScrolledText( log_frame, bg="#1e1e1e", fg="#d4d4d4", font=("Consolas", 9), relief="flat", wrap="word", insertbackground="#d4d4d4" ) self.log_text.pack(fill="both", expand=True) # 日志颜色标签 self.log_text.tag_config("INFO", foreground="#569cd6") self.log_text.tag_config("DATA", foreground="#6a9955") self.log_text.tag_config("SCAN", foreground="#808080") self.log_text.tag_config("SIGNAL", foreground="#4ec9b0", font=("Consolas", 9, "bold")) self.log_text.tag_config("ORDER", foreground="#ce9178") self.log_text.tag_config("ERROR", foreground="#f44747") self.log_text.tag_config("SKIP", foreground="#dcdcaa") self.log_text.tag_config("MT5", foreground="#c586c0") self.log_text.tag_config("SYSTEM", foreground="#9cdcfe") self.log_text.tag_config("CLOSE", foreground="#f44747") self.log_text.tag_config("DEBUG", foreground="#808080") # ==================== 底部状态 ==================== footer = tk.Frame(self.root, bg=self.colors["bg"], height=25) footer.pack(fill="x", padx=15, pady=(0, 5)) self.trade_status_label = tk.Label(footer, text="交易时段: --", bg=self.colors["bg"], fg=self.colors["text_light"], font=("Consolas", 8)) self.trade_status_label.pack(side="left") self.update_time_label = tk.Label(footer, text="最后更新: --", bg=self.colors["bg"], fg=self.colors["text_light"], font=("Consolas", 8)) self.update_time_label.pack(side="right") def _poll(self): """轮询队列，更新 GUI""" try: while True: msg_type, data = self.gui_queue.get_nowait() if msg_type == "new_signal": self._add_signal_row(data) elif msg_type == "heartbeat": self._update_account_info() elif msg_type == "analysis_data": self._update_symbol_data(data) elif msg_type == "log": line, tag = data self._append_log(line, tag) except queue.Empty: pass # 更新倒计时 secs = next_bar_close_seconds() m, s = divmod(secs, 60) self.countdown_label.config(text=f"⏱ 下一K线: {int(m):02d}:{int(s):02d}") # 更新交易时段状态 if is_trading_time(): self.trade_status_label.config( text=f"🟢 交易中 ({TRADE_START:02d}:00 - {TRADE_END:02d}:00 北京时间)", fg=self.colors["success"] ) else: self.trade_status_label.config( text=f"🔴 非交易时段 ({TRADE_START:02d}:00 - {TRADE_END:02d}:00 北京时间)", fg=self.colors["danger"] ) # 更新时间戳 self.update_time_label.config(text=f"最后更新: {now_bj_str()}") self.root.after(GUI_REFRESH_MS, self._poll) def _update_symbol_data(self, data): """更新货币对状态标签""" sym = data["symbol"] self.symbol_data[sym] = data hist = data.get("hist", 0) if hist > 0: self.symbol_frames[sym].config(bg="#d5f5e3", fg=self.colors["buy"]) elif hist < 0: self.symbol_frames[sym].config(bg="#fadbd8", fg=self.colors["sell"]) else: self.symbol_frames[sym].config(bg=self.colors["border"], fg=self.colors["text_light"]) def _update_account_info(self): """更新账户信息""" acc = account_info if acc: self.balance_label.config(text=f"💰 余额: ${acc.get('balance', 0):,.2f}") self.equity_label.config(text=f"📊 净值: ${acc.get('equity', 0):,.2f}") def _add_signal_row(self, data): """添加信号行""" direction = data["direction"] tag = "BUY" if direction == "BUY" else "SELL" self.signal_tree.insert("", 0, values=( data["time"], data["symbol"], direction, f"{data['price']:.5f}", f"{data['hist']:.6f}", f"{data['atr']:.5f}" ), tags=(tag,)) # 限制显示行数 children = self.signal_tree.get_children() if len(children) > 50: self.signal_tree.delete(children[-1]) def _append_log(self, line, tag="INFO"): """添加日志""" self.log_text.configure(state="normal") self.log_text.insert("end", line + "\n", tag) self.log_text.see("end") self.log_text.configure(state="disabled") # 限制日志行数 lines = int(self.log_text.index('end-1c').split('.')[0]) if lines > 500: self.log_text.delete('1.0', '2.0') def _on_start(self): """启动交易""" global running if running: return add_log("=" * 50, "SYSTEM") add_log("启动交易系统...", "SYSTEM") if not mt5_connect(): self.status_label.config(text="🔴 MT5 连接失败", fg=self.colors["danger"]) return running = True self.start_btn.config(state="disabled", bg="#95a5a6") self.stop_btn.config(state="normal", bg=self.colors["danger"]) self.status_label.config(text="🟢 运行中", fg=self.colors["success"]) t = threading.Thread(target=scanner_thread, args=(self.gui_queue,), daemon=True) t.start() def _on_stop(self): """停止交易""" global running running = False self.start_btn.config(state="normal", bg=self.colors["success"]) self.stop_btn.config(state="disabled", bg="#95a5a6") self.status_label.config(text="⚪ 已停止", fg=self.colors["text_light"]) # 重置货币对状态标签 for sym in SYMBOLS: self.symbol_frames[sym].config(bg=self.colors["border"], fg=self.colors["text_light"]) mt5_disconnect() add_log("交易系统已停止", "SYSTEM") def run(self): """运行 GUI""" self.root.mainloop() # ================================================================ # MAIN # ================================================================ def main(): gui_queue = queue.Queue() # 重写 add_log 以推送到 GUI original_add_log = add_log def gui_add_log(msg, tag="INFO"): original_add_log(msg, tag) gui_queue.put(("log", (f"[{now_bj_str()}] [{tag}] {msg}", tag))) # 替换全局函数 this_module = sys.modules[__name__] setattr(this_module, 'add_log', gui_add_log) root = tk.Tk() gui = TradingGUI(root, gui_queue) gui.run() if __name__ == "__main__": main()