# -*- coding: utf-8 -*- """黄金市价辅助下单窗：连接本机已登录的 MT5；可选激活 gold_monitor 上报复盘。""" from __future__ import annotations import math import sys import threading import time import webbrowser import tkinter as tk from datetime import datetime, timedelta, timezone from tkinter import messagebox, simpledialog, ttk from typing import Any, Optional, Tuple, Union BJT = timezone(timedelta(hours=8)) # 顶层导入，避免打包 exe 漏依赖 import numpy # noqa: F401 import MetaTrader5 as mt5 try: import hub_client from activation_dialog import ensure_activated from device_store import is_activated, load_device except ImportError: # 允许单独拷贝本文件调试 hub_client = None # type: ignore ensure_activated = None # type: ignore is_activated = lambda: False # type: ignore load_device = lambda: {} # type: ignore APP_TITLE = "黄金市价辅助" DEFAULT_SYMBOL = "XAUUSDm" DEFAULT_LOT = 0.10 DEFAULT_ATR_MULT = 1.0 DEFAULT_TP_ATR_MULT = 1.5 DEFAULT_TP_USD = 12.0 DEFAULT_PARTIAL = 0.10 ATR_PERIOD = 14 ATR_TIMEFRAME = None ATR_FLOOR_USD = 10.0 # ATR 低于该值时，止损/止盈按 10 美金计；高于则用实际 ATR MAGIC = 20261016 COMMENT = "GoldPanel" DEVIATION = 100 MUST_CONFIRM = False # 开仓/平仓默认不二次确认 REFRESH_MS = 400 # 报价/持仓轻量刷新 TODAY_STATS_MS = 2500 # 今日战绩（含 history）节流 ATR_CACHE_MS = 2000 DAY_OHLC_CACHE_MS = 1500 HUB_FLOAT_MS = 3000 # 浮盈上报间隔 HUB_RECONCILE_MS = 45000 # 历史回填/对账间隔（重） BALANCE_HIGH = 10000.0 BALANCE_LOW = 1000.0 LOT_HIGH = 0.10 LOT_MID = 0.05 LOT_LOW = 0.01 MAX_SIDE_LOTS = 0.50 MAX_OPEN_POSITIONS = 5 # 同时最多持仓笔数，满则买卖置灰 POS_ROW_HEIGHT = 46 # 持仓列表单行约高（含内边距） POS_LIST_HEIGHT = POS_ROW_HEIGHT * MAX_OPEN_POSITIONS # 默认可视 5 条持仓高度 DAY_LOSS_LOCK = 0.05 # 当日合计亏损达余额 5% 禁止开仓 DAY_PROFIT_LOCK = 0.10 # 当日合计盈利达余额 10% 禁止开仓 # 币圈习惯：绿涨红跌 COLOR_PROFIT = "#0ecb81" COLOR_LOSS = "#f6465d" COLOR_DANGER = "#f6465d" # 平仓按钮边框 COLOR_NEUTRAL = "#c0c0c0" COLOR_BE = "#3a5a80" COLOR_BE_OFF = "#555555" COLOR_BUY = "#108040" COLOR_SELL = "#a02828" COLOR_BTN_OFF = "#555555" COLOR_WARN = "#e6a23c" COLOR_BG = "#121218" COLOR_CARD = "#1a1a22" COLOR_CARD2 = "#22222e" COLOR_MUTED = "#8a8a9c" COLOR_TEXT = "#f0f0f5" COLOR_ACCENT = "#4cc2ff" BE_MIN_MOVE_USD = 3.0 # 价格向有利方向走出该美金距离才可保本（非账户浮盈） BE_OFFSET_USD = 1.0 # 保本止损 = 开仓价 ± 1$ SPREAD_WARN_USD = 0.50 # 点差超过该美金时，买卖按钮区提示（不禁开） def simpledialog_ask_server(parent: tk.Misc) -> str: """未激活或本地无中控地址时，让用户输入复盘服务器 URL。""" value = simpledialog.askstring( "打开复盘", "请输入中控地址（例如 http://192.168.1.10:5800）", parent=parent, initialvalue="http://127.0.0.1:5800", ) return (value or "").strip() def enable_dpi_awareness() -> None: """避免高分屏下 Tk 窗口被系统拉伸发虚。""" if sys.platform != "win32": return try: import ctypes try: ctypes.windll.shcore.SetProcessDpiAwareness(1) # SYSTEM_DPI_AWARE except Exception: ctypes.windll.user32.SetProcessDPIAware() except Exception: pass def _init_mt5_constants() -> None: global ATR_TIMEFRAME ATR_TIMEFRAME = mt5.TIMEFRAME_M15 def suggest_lot_by_balance(balance: float) -> float: """按余额档位手数：≥10000→0.10；1000~10000→0.05；<1000→0.01（点按钮即可）。""" if balance >= BALANCE_HIGH: return LOT_HIGH if balance < BALANCE_LOW: return LOT_LOW return LOT_MID def _as_bjt(now: Optional[datetime] = None) -> datetime: dt = now or datetime.now(BJT) if dt.tzinfo is None: return dt.replace(tzinfo=BJT) return dt.astimezone(BJT) def current_session_label(now: Optional[datetime] = None, market_open: bool = True) -> str: """北京时间盘段：亚盘/欧盘/美盘（可重叠显示）；休市单独标注。""" if not market_open: return "休市" dt = _as_bjt(now) h = dt.hour + dt.minute / 60.0 asia = 8.0 <= h < 16.0 euro = 15.0 <= h < 24.0 us = h >= 20.0 or h < 5.0 parts = [] if asia: parts.append("亚盘") if euro: parts.append("欧盘") if us: parts.append("美盘") if not parts: return "过渡时段" return "+".join(parts) # 黄金常见时段（一律按北京时间 UTC+8）：周一 07:00 开 → 周六 05:00 收；日内约 05:00–07:00 GOLD_OPEN_HOUR = 7 GOLD_OPEN_MINUTE = 0 GOLD_CLOSE_HOUR = 5 GOLD_CLOSE_MINUTE = 0 def fmt_bjt(dt: datetime, fmt: str = "%m-%d %H:%M") -> str: """格式化为北京时间字符串（不依赖系统时区）。""" return _as_bjt(dt).strftime(fmt) def next_market_open(now: Optional[datetime] = None) -> datetime: """估算下一次开盘时刻，返回带 BJT 时区的 datetime。""" dt = _as_bjt(now) open_hm = (GOLD_OPEN_HOUR, GOLD_OPEN_MINUTE, 0) close_hm = (GOLD_CLOSE_HOUR, GOLD_CLOSE_MINUTE, 0) t = (dt.hour, dt.minute, dt.second) wd = dt.weekday() # Mon=0 … Sun=6 # 周末休市：周六 05:00（北京）→ 下周一 07:00（北京） if wd == 5 and t >= close_hm: days = 2 elif wd == 6: days = 1 elif wd == 0 and t < open_hm: days = 0 # 工作日日内休市窗口 05:00–07:00（北京） elif wd <= 4 and close_hm <= t < open_hm: days = 0 else: # 非预期休市（节假日等）：推到下一个可交易日的北京时间 07:00 days = 1 while True: cand_date = (dt + timedelta(days=days)).date() cwd = datetime(cand_date.year, cand_date.month, cand_date.day, tzinfo=BJT).weekday() if cwd <= 4: break days += 1 if days > 10: break target = (dt + timedelta(days=days)).date() return datetime( target.year, target.month, target.day, GOLD_OPEN_HOUR, GOLD_OPEN_MINUTE, 0, tzinfo=BJT, ) def format_countdown(seconds: int) -> str: seconds = max(0, int(seconds)) days, rem = divmod(seconds, 86400) hours, rem = divmod(rem, 3600) mins, secs = divmod(rem, 60) if days > 0: return f"距开盘 {days}天{hours}小时{mins}分" if hours > 0: return f"距开盘 {hours}小时{mins}分" if mins > 0: return f"距开盘 {mins}分{secs}秒" return f"距开盘 {secs}秒" def format_open_elapsed(open_ts: float, market_open: bool, now: Optional[datetime] = None) -> str: """开盘后已过多久；休市时显示距下次开盘倒计时。""" if not market_open: nxt = next_market_open(now) left = max(0, int((nxt - _as_bjt(now)).total_seconds())) return format_countdown(left) elapsed = max(0, int(time.time() - open_ts)) days, rem = divmod(elapsed, 86400) hours, rem = divmod(rem, 3600) mins, _ = divmod(rem, 60) if days > 0: return f"已开盘 {days}天{hours}小时{mins}分" if hours > 0: return f"已开盘 {hours}小时{mins}分" return f"已开盘 {mins}分" class TradeError(Exception): pass class Mt5Bridge: def __init__(self) -> None: self.connected = False def connect(self) -> None: if self.connected: return if not mt5.initialize(): raise TradeError(f"连接 MT5 失败：{mt5.last_error()}（请先打开并登录 MT5）") self.connected = True def shutdown(self) -> None: if self.connected: mt5.shutdown() self.connected = False def account_balance(self) -> float: acc = mt5.account_info() if acc is None: raise TradeError("无法读取账户信息") return float(acc.balance) def ensure_symbol(self, symbol: str) -> None: info = mt5.symbol_info(symbol) if info is None: raise TradeError(f"找不到品种 {symbol}") if not info.visible and not mt5.symbol_select(symbol, True): raise TradeError(f"无法选择品种 {symbol}") def tick(self, symbol: str): self.ensure_symbol(symbol) t = mt5.symbol_info_tick(symbol) if t is None: raise TradeError("无法读取报价") return t def symbol_info(self, symbol: str): self.ensure_symbol(symbol) info = mt5.symbol_info(symbol) if info is None: raise TradeError(f"无法读取品种信息 {symbol}") return info def market_is_open(self, symbol: str) -> bool: """根据最新 tick 是否新鲜判断是否在交易（休市也仍可读日线）。""" self.ensure_symbol(symbol) info = mt5.symbol_info(symbol) if info is not None and hasattr(info, "trade_mode"): # 完全禁止交易 if int(info.trade_mode) == 0: return False tick = mt5.symbol_info_tick(symbol) if tick is None: return False age = time.time() - float(tick.time) return age < 180 # 3 分钟内有报价视为开市 def day_ohlc(self, symbol: str) -> dict[str, Any]: """当前/最近交易日日线开高低（休市仍显示）；开市时用 tick 修正最高/最低。""" self.ensure_symbol(symbol) rates = mt5.copy_rates_from_pos(symbol, mt5.TIMEFRAME_D1, 0, 3) if rates is None or len(rates) < 1: raise TradeError("无法读取日线，请确认品种有 D1 数据") # MT5 返回数组从旧到新：[-1] 才是当前日线 bar = rates[-1] open_px = float(bar["open"]) high = float(bar["high"]) low = float(bar["low"]) open_ts = int(bar["time"]) market_open = self.market_is_open(symbol) tick = mt5.symbol_info_tick(symbol) if market_open and tick is not None: bid = float(tick.bid) ask = float(tick.ask) high = max(high, bid, ask) low = min(low, bid, ask) open_dt = datetime.fromtimestamp(open_ts, BJT) bar_day = open_dt.strftime("%m-%d") nxt = None if market_open else next_market_open() return { "open": open_px, "high": high, "low": low, "open_time": open_dt, "open_ts": open_ts, "bar_day": bar_day, "market_open": market_open, "elapsed_text": format_open_elapsed(float(open_ts), market_open), "next_open": nxt, "next_open_text": fmt_bjt(nxt) if nxt else "", } def atr(self, symbol: str, period: int = ATR_PERIOD) -> float: self.ensure_symbol(symbol) rates = mt5.copy_rates_from_pos(symbol, ATR_TIMEFRAME, 1, period + 1) if rates is None or len(rates) < period: rates = mt5.copy_rates_from_pos(symbol, ATR_TIMEFRAME, 0, period + 1) if rates is None or len(rates) < period: raise TradeError("无法读取 M15 ATR，请确认已加载 15 分钟数据") trs = [] for i in range(1, len(rates)): high = float(rates[i]["high"]) low = float(rates[i]["low"]) prev_close = float(rates[i - 1]["close"]) trs.append(max(high - low, abs(high - prev_close), abs(low - prev_close))) if len(trs) < period: raise TradeError("ATR 数据不足") return sum(trs[-period:]) / period def normalize_volume(self, symbol: str, volume: float) -> float: info = self.symbol_info(symbol) step = float(info.volume_step) vmin = float(info.volume_min) vmax = float(info.volume_max) if step <= 0 or vmin <= 0: raise TradeError("无法读取手数规则") volume = math.floor(volume / step + 1e-8) * step volume = round(volume, 8) if volume < vmin - 1e-8: raise TradeError(f"手数过小，最小 {vmin:.2f}") if volume > vmax + 1e-8: raise TradeError(f"手数过大，最大 {vmax:.2f}") return volume def min_stop_distance(self, symbol: str) -> float: info = self.symbol_info(symbol) level = max(int(info.trade_stops_level), int(info.trade_freeze_level)) return level * float(info.point) def build_stops( self, symbol: str, order_type: int, price: float, sl_usd: float, tp_usd: float ) -> Tuple[float, float]: if sl_usd < -1e-8 or tp_usd < -1e-8: raise TradeError("SL/TP 不能为负数") info = self.symbol_info(symbol) digits = int(info.digits) min_dist = self.min_stop_distance(symbol) sl = 0.0 tp = 0.0 if sl_usd > 1e-8: sl = price - sl_usd if order_type == mt5.ORDER_TYPE_BUY else price + sl_usd sl = round(sl, digits) if abs(price - sl) + 1e-12 < min_dist: raise TradeError(f"止损过近，至少需要 {min_dist:.2f} 美元（含券商最小距离）") if tp_usd > 1e-8: tp = price + tp_usd if order_type == mt5.ORDER_TYPE_BUY else price - tp_usd tp = round(tp, digits) if abs(tp - price) + 1e-12 < min_dist: raise TradeError(f"止盈过近，至少需要 {min_dist:.2f} 美元（含券商最小距离）") return sl, tp def filling_candidates(self, symbol: str) -> list: info = self.symbol_info(symbol) mode = int(info.filling_mode) order = [] if mode & 2: order.append(mt5.ORDER_FILLING_IOC) if mode & 1: order.append(mt5.ORDER_FILLING_FOK) order.append(mt5.ORDER_FILLING_RETURN) # 去重保序 seen = set() out = [] for f in order: if f not in seen: seen.add(f) out.append(f) return out def _check_trade_allowed(self, symbol: str, order_type: int) -> None: term = mt5.terminal_info() if term is None or not term.connected: raise TradeError("终端未连接到交易服务器") if not term.trade_allowed: raise TradeError("终端未开启自动交易") acc = mt5.account_info() if acc is None or not acc.trade_allowed: raise TradeError("账户不允许交易") info = self.symbol_info(symbol) mode = int(info.trade_mode) if mode == mt5.SYMBOL_TRADE_MODE_DISABLED: raise TradeError("该品种禁止交易") if mode == mt5.SYMBOL_TRADE_MODE_CLOSEONLY: raise TradeError("该品种仅允许平仓") if order_type == mt5.ORDER_TYPE_BUY and mode == mt5.SYMBOL_TRADE_MODE_SHORTONLY: raise TradeError("该品种只允许做空") if order_type == mt5.ORDER_TYPE_SELL and mode == mt5.SYMBOL_TRADE_MODE_LONGONLY: raise TradeError("该品种只允许做多") def check_max_positions(self, symbol: str) -> None: n = len(self.positions(symbol)) if n >= MAX_OPEN_POSITIONS: raise TradeError(f"持仓已满 {MAX_OPEN_POSITIONS} 笔，请先平仓后再开") def check_max_addon(self, symbol: str, order_type: int, lots: float) -> None: buy, sell, _ = self.exposure(symbol) if order_type == mt5.ORDER_TYPE_BUY: total = buy + lots side = "多" cur = buy else: total = sell + lots side = "空" cur = sell if total > MAX_SIDE_LOTS + 1e-8: remain = max(0.0, MAX_SIDE_LOTS - cur) raise TradeError( f"同向加仓上限 {MAX_SIDE_LOTS:.2f} 手：当前{side}单 {cur:.2f}，" f"再开 {lots:.2f} 会超限（还可加 {remain:.2f}）" ) def day_lock_kind(self, symbol: str) -> Tuple[Optional[str], float]: """返回 (lock_kind, rate)。lock_kind: 'loss' | 'profit' | None。""" st = self.today_stats(symbol) rate = float(st["profit_rate"]) if rate <= -DAY_LOSS_LOCK + 1e-12: return "loss", rate if rate >= DAY_PROFIT_LOCK - 1e-12: return "profit", rate return None, rate def _send_deal(self, request: dict) -> object: last_detail = "" for filling in self.filling_candidates(request["symbol"]): req = dict(request) req["type_filling"] = filling result = mt5.order_send(req) if result is None: last_detail = str(mt5.last_error()) continue if result.retcode == mt5.TRADE_RETCODE_DONE: return result last_detail = f"{result.comment} ({result.retcode})" # 成交模式不对才继续试 if result.retcode not in ( mt5.TRADE_RETCODE_INVALID_FILL, 10030, # INVALID_FILL ): break raise TradeError(last_detail or "下单失败") def market_order( self, symbol: str, order_type: int, lots: float, sl_usd: float, tp_usd: float ) -> str: self._check_trade_allowed(symbol, order_type) self.check_day_trade_lock(symbol) self.check_max_positions(symbol) lots = self.normalize_volume(symbol, lots) self.check_max_addon(symbol, order_type, lots) t = self.tick(symbol) price = float(t.ask if order_type == mt5.ORDER_TYPE_BUY else t.bid) if price <= 0: raise TradeError("无法读取报价") sl, tp = self.build_stops(symbol, order_type, price, sl_usd, tp_usd) request = { "action": mt5.TRADE_ACTION_DEAL, "symbol": symbol, "volume": lots, "type": order_type, "price": price, "sl": sl, "tp": tp, "deviation": DEVIATION, "magic": MAGIC, "comment": COMMENT, "type_time": mt5.ORDER_TIME_GTC, } try: result = self._send_deal(request) except TradeError as exc: raise TradeError(f"下单失败：{exc}") from exc # 持仓单号可能与 order 不同，优先用持仓 ticket，避免平仓对不上仍显示「持仓中」 pos_ticket = int(result.order) open_px = float(result.price) open_ts = time.time() for _ in range(6): time.sleep(0.02) for p in self.positions(symbol): if int(p.ticket) == int(result.order) or abs(float(p.price_open) - open_px) < 0.05: pos_ticket = int(p.ticket) open_px = float(p.price_open) open_ts = float(p.time) sl = float(p.sl) if float(p.sl) > 0 else sl tp = float(p.tp) if float(p.tp) > 0 else tp break else: continue break side = "买入" if order_type == mt5.ORDER_TYPE_BUY else "卖出" msg = f"{side}成功 手数={lots:.2f} 成交价={open_px} 单号={pos_ticket}" report = { "ticket": str(pos_ticket), "symbol": symbol, "side": "buy" if order_type == mt5.ORDER_TYPE_BUY else "sell", "volume": lots, "open_price": open_px, "close_price": None, "open_time": open_ts, "close_time": None, "profit": 0.0, "sl": sl, "tp": tp, "snapshot": {"event": "open", "order": int(result.order)}, } return msg, report def positions(self, symbol: str, pos_type: Optional[int] = None): self.ensure_symbol(symbol) poss = mt5.positions_get(symbol=symbol) if poss is None: return [] items = list(poss) if pos_type is not None: items = [p for p in items if int(p.type) == pos_type] items.sort(key=lambda p: (int(p.time), int(p.ticket))) return items def get_position(self, ticket: int): poss = mt5.positions_get(ticket=ticket) if not poss: raise TradeError(f"找不到持仓 #{ticket}") return poss[0] def exposure(self, symbol: str) -> Tuple[float, float, float]: buy = sell = profit = 0.0 for p in self.positions(symbol): profit += float(p.profit) + float(p.swap) if int(p.type) == mt5.POSITION_TYPE_BUY: buy += float(p.volume) else: sell += float(p.volume) return buy, sell, profit def today_stats(self, symbol: str) -> dict: """今日开仓次数、已平盈亏、浮盈、盈亏比、盈利率、胜率等（仅当天）。""" self.ensure_symbol(symbol) now = datetime.now() day0 = now.replace(hour=0, minute=0, second=0, microsecond=0) deals = mt5.history_deals_get(day0 - timedelta(hours=12), now + timedelta(hours=1)) open_count = 0 closed_pnl = 0.0 win_sum = 0.0 loss_sum = 0.0 # 绝对值累加 # 按持仓单号汇总当日平仓盈亏，用于胜率（部分平仓合并为一笔） pos_pnl: dict = {} if deals: for d in deals: if d.symbol != symbol: continue deal_time = datetime.fromtimestamp(int(d.time)) if deal_time < day0: continue entry = int(d.entry) if entry in (mt5.DEAL_ENTRY_IN, mt5.DEAL_ENTRY_INOUT): open_count += 1 pnl = float(d.profit) + float(d.swap) + float(d.commission) if entry in (mt5.DEAL_ENTRY_OUT, mt5.DEAL_ENTRY_INOUT, mt5.DEAL_ENTRY_OUT_BY): closed_pnl += pnl pid = int(getattr(d, "position_id", 0) or 0) or int(d.ticket) pos_pnl[pid] = pos_pnl.get(pid, 0.0) + pnl if pnl > 1e-8: win_sum += pnl elif pnl < -1e-8: loss_sum += -pnl elif entry == mt5.DEAL_ENTRY_IN: closed_pnl += float(d.commission) if float(d.commission) < -1e-8: loss_sum += -float(d.commission) wins = sum(1 for p in pos_pnl.values() if p > 1e-8) losses = sum(1 for p in pos_pnl.values() if p < -1e-8) closed_n = len(pos_pnl) if closed_n > 0: win_rate = wins / closed_n win_rate_txt = f"{win_rate * 100:.0f}%（{wins}胜{losses}负）" else: win_rate_txt = "--" _, _, floating = self.exposure(symbol) total = closed_pnl + floating balance = self.account_balance() profit_rate = (total / balance) if balance > 1e-8 else 0.0 if loss_sum > 1e-8: pl_ratio = win_sum / loss_sum pl_ratio_txt = f"{pl_ratio:.2f}" elif win_sum > 1e-8: pl_ratio_txt = "∞" else: pl_ratio_txt = "--" return { "opens": open_count, "closed_pnl": closed_pnl, "floating": floating, "total": total, "balance": balance, "profit_rate": profit_rate, "pl_ratio_txt": pl_ratio_txt, "win_rate_txt": win_rate_txt, "wins": wins, "losses": losses, "closed_trades": closed_n, } def check_day_trade_lock(self, symbol: str) -> None: kind, rate = self.day_lock_kind(symbol) pct = rate * 100.0 if kind == "loss": raise TradeError(f"当日亏损已达 {pct:.2f}%（上限 -{DAY_LOSS_LOCK*100:.0f}%），禁止开仓") if kind == "profit": raise TradeError(f"当日盈利已达 {pct:.2f}%（上限 +{DAY_PROFIT_LOCK*100:.0f}%），禁止开仓") def recent_close_reports(self, symbol: str, hours: float = 72) -> list: """从 MT5 历史成交还原已平仓，用于补报/纠错仍显示「持仓中」的记录。""" self.ensure_symbol(symbol) now = datetime.now() deals = mt5.history_deals_get(now - timedelta(hours=hours), now + timedelta(hours=1)) if not deals: return [] out_entries = { mt5.DEAL_ENTRY_OUT, mt5.DEAL_ENTRY_OUT_BY, mt5.DEAL_ENTRY_INOUT, } by_pos: dict = {} for d in deals: if str(d.symbol) != symbol: continue if int(d.entry) not in out_entries: continue pid = int(d.position_id or 0) if pid <= 0: continue g = by_pos.get(pid) if g is None: # 平仓方向：卖出平仓 → 原多单 side = "buy" if int(d.type) == mt5.DEAL_TYPE_SELL else "sell" g = { "ticket": str(pid), "symbol": symbol, "side": side, "volume": 0.0, "open_price": 0.0, "close_price": float(d.price), "open_time": None, "close_time": float(d.time), "profit": 0.0, "sl": None, "tp": None, "snapshot": {"event": "history_close"}, } by_pos[pid] = g g["profit"] += float(d.profit) + float(d.swap) + float(d.commission) g["volume"] += float(d.volume) g["close_price"] = float(d.price) g["close_time"] = max(float(g["close_time"]), float(d.time)) for d in deals: if str(d.symbol) != symbol: continue pid = int(d.position_id or 0) if pid not in by_pos: continue if int(d.entry) in (mt5.DEAL_ENTRY_IN, mt5.DEAL_ENTRY_INOUT): by_pos[pid]["open_price"] = float(d.price) prev = by_pos[pid].get("open_time") ts = float(d.time) by_pos[pid]["open_time"] = ts if prev is None else min(float(prev), ts) return list(by_pos.values()) def close_ticket(self, ticket: int) -> str: p = self.get_position(ticket) symbol = str(p.symbol) pos_type = int(p.type) volume = float(p.volume) order_type = mt5.ORDER_TYPE_SELL if pos_type == mt5.POSITION_TYPE_BUY else mt5.ORDER_TYPE_BUY t = self.tick(symbol) price = float(t.bid if order_type == mt5.ORDER_TYPE_SELL else t.ask) request = { "action": mt5.TRADE_ACTION_DEAL, "symbol": symbol, "volume": volume, "type": order_type, "position": int(ticket), "price": price, "deviation": DEVIATION, "magic": MAGIC, "comment": COMMENT, "type_time": mt5.ORDER_TIME_GTC, } try: self._send_deal(request) except TradeError as exc: raise TradeError(f"平仓失败 #{ticket}：{exc}") from exc # 估算平仓盈亏（含 swap） profit = float(p.profit) + float(p.swap) report = { "ticket": str(ticket), "symbol": symbol, "side": "buy" if pos_type == mt5.POSITION_TYPE_BUY else "sell", "volume": volume, "open_price": float(p.price_open), "close_price": price, "open_time": float(p.time), "close_time": time.time(), "profit": profit, "sl": float(p.sl), "tp": float(p.tp), "snapshot": {"event": "close"}, } return f"已平 #{ticket}（{volume:.2f} 手）", report def _sltp_retcode_ok(self, retcode: int) -> bool: ok = { mt5.TRADE_RETCODE_DONE, getattr(mt5, "TRADE_RETCODE_DONE_PARTIAL", -1), getattr(mt5, "TRADE_RETCODE_PLACED", -1), } return int(retcode) in ok def modify_position_sltp(self, ticket: int, sl: float, tp: Optional[float] = None) -> float: """修改持仓止损/止盈；成功返回实际止损价，失败抛 TradeError。""" p = self.get_position(int(ticket)) symbol = str(p.symbol) self.ensure_symbol(symbol) term = mt5.terminal_info() if term is None or not term.connected: raise TradeError("终端未连接到交易服务器") if not getattr(term, "trade_allowed", True): raise TradeError("终端未开启自动交易（请勾选 Algo Trading）") acc = mt5.account_info() if acc is None or not getattr(acc, "trade_allowed", True): raise TradeError("账户不允许交易") digits = int(self.symbol_info(symbol).digits) sl = round(float(sl), digits) if tp is None: tp_val = float(p.tp) else: tp_val = float(tp) tp_val = round(tp_val, digits) if tp_val > 1e-12 else 0.0 old_sl = float(p.sl) # 多组请求：部分券商带 magic / 不带 magic 表现不同；TP=0 与保留原 TP 也分开试 variants: list[dict] = [] base = { "action": mt5.TRADE_ACTION_SLTP, "symbol": symbol, "position": int(ticket), "sl": sl, } variants.append({**base, "tp": tp_val}) variants.append({**base, "tp": tp_val, "magic": MAGIC}) if tp_val > 1e-12: variants.append({**base, "tp": 0.0}) # 仅改 SL # 去重 seen = set() uniq: list[dict] = [] for req in variants: key = (req.get("sl"), req.get("tp"), req.get("magic")) if key in seen: continue seen.add(key) uniq.append(req) last_detail = "" for request in uniq: check = mt5.order_check(request) if check is not None and int(getattr(check, "retcode", 0)) not in ( 0, mt5.TRADE_RETCODE_DONE, ): last_detail = f"校验失败 {check.comment} ({check.retcode})" # 校验失败仍尝试 send，部分终端 check 偏严 result = mt5.order_send(request) if result is None: last_detail = str(mt5.last_error()) continue if not self._sltp_retcode_ok(result.retcode): last_detail = f"{result.comment} ({result.retcode})" continue # 回读确认（券商可能延迟） for _ in range(8): time.sleep(0.2) try: p2 = self.get_position(int(ticket)) except TradeError: continue if abs(float(p2.sl) - sl) <= max(10 ** (-digits), 0.01): return float(p2.sl) last_detail = ( f"回报成功但止损未变（原 {old_sl:.{digits}f} → 目标 {sl:.{digits}f}，" f"当前仍为 {float(self.get_position(int(ticket)).sl):.{digits}f}）" ) raise TradeError(f"#{ticket} {last_detail or '改止损失败'}") def set_stop_loss(self, ticket: int, new_sl: float) -> Tuple[str, dict]: """手动改止损价（与保本无关）；保留原止盈。返回 (消息, 上报用 dict)。""" p = self.get_position(int(ticket)) symbol = str(p.symbol) pos_type = int(p.type) digits = int(self.symbol_info(symbol).digits) min_dist = self.min_stop_distance(symbol) t = self.tick(symbol) old_sl = float(p.sl) new_sl = round(float(new_sl), digits) if new_sl <= 1e-12: raise TradeError("止损价必须大于 0") if pos_type == mt5.POSITION_TYPE_BUY: ref = float(t.bid) if new_sl >= ref - 1e-12: raise TradeError( f"多单止损须低于现价：现价 {ref:.{digits}f}，止损 {new_sl:.{digits}f}" ) if ref - new_sl + 1e-12 < min_dist: raise TradeError( f"止损过近：距现价 {ref - new_sl:.{digits}f}，至少 {min_dist:.{digits}f}" ) else: ref = float(t.ask) if new_sl <= ref + 1e-12: raise TradeError( f"空单止损须高于现价：现价 {ref:.{digits}f}，止损 {new_sl:.{digits}f}" ) if new_sl - ref + 1e-12 < min_dist: raise TradeError( f"止损过近：距现价 {new_sl - ref:.{digits}f}，至少 {min_dist:.{digits}f}" ) applied = self.modify_position_sltp(int(ticket), new_sl, float(p.tp)) msg = ( f"#{ticket} 止损已改 " f"{(f'{old_sl:.{digits}f}' if old_sl > 0 else '无')}→{applied:.{digits}f}" ) report = { "ticket": str(ticket), "symbol": symbol, "side": "buy" if pos_type == mt5.POSITION_TYPE_BUY else "sell", "volume": float(p.volume), "open_price": float(p.price_open), "close_price": None, "open_time": float(p.time), "close_time": None, "profit": float(p.profit) + float(p.swap), "sl": float(applied), "tp": float(p.tp), "snapshot": {"event": "modify_sl"}, } return msg, report def break_even_tickets(self, tickets: list, offset_usd: float = 1.0) -> Tuple[str, list]: """保本：多单 SL=开仓价+offset，空单 SL=开仓价-offset。返回 (消息, 成功单号列表)。""" if not tickets: raise TradeError("没有可保本的持仓") ok_tickets: list = [] errors: list = [] details: list = [] for ticket in tickets: try: p = self.get_position(int(ticket)) symbol = str(p.symbol) pos_type = int(p.type) open_price = float(p.price_open) digits = int(self.symbol_info(symbol).digits) min_dist = self.min_stop_distance(symbol) t = self.tick(symbol) old_sl = float(p.sl) if pos_type == mt5.POSITION_TYPE_BUY: sl = round(open_price + offset_usd, digits) ref = float(t.bid) move = ref - open_price if move + 1e-12 < BE_MIN_MOVE_USD: raise TradeError( f"#{ticket} 有利方向仅 {move:.2f}$，需 ≥ {BE_MIN_MOVE_USD:.0f}$" ) if sl + max(min_dist, 0.0) > ref + 1e-12: raise TradeError( f"#{ticket} 现价 {ref:.{digits}f} 距保本价 {sl:.{digits}f} 过近" ) else: sl = round(open_price - offset_usd, digits) ref = float(t.ask) move = open_price - ref if move + 1e-12 < BE_MIN_MOVE_USD: raise TradeError( f"#{ticket} 有利方向仅 {move:.2f}$，需 ≥ {BE_MIN_MOVE_USD:.0f}$" ) if ref > sl - max(min_dist, 0.0) + 1e-12: raise TradeError( f"#{ticket} 现价 {ref:.{digits}f} 距保本价 {sl:.{digits}f} 过近" ) new_sl = self.modify_position_sltp(int(ticket), sl, float(p.tp)) ok_tickets.append(int(ticket)) details.append( f"#{ticket} SL {old_sl:.{digits}f}→{new_sl:.{digits}f}" ) except TradeError as exc: errors.append(str(exc)) if not ok_tickets: raise TradeError("保本失败：" + ("；".join(errors[:3]) if errors else "无可用持仓")) msg = f"已保本 {len(ok_tickets)} 笔：" + "，".join(details) if errors: msg += f"；失败 {len(errors)}：{errors[0]}" return msg, ok_tickets def close_positions(self, symbol: str, pos_type: Optional[int] = None): items = self.positions(symbol, pos_type) if not items: raise TradeError("没有可平的持仓") closed = 0 reports = [] for p in reversed(items): _msg, report = self.close_ticket(int(p.ticket)) reports.append(report) closed += 1 return f"已平 {closed} 笔", reports def partial_close(self, symbol: str, pos_type: int, volume: float) -> str: volume = self.normalize_volume(symbol, volume) items = self.positions(symbol, pos_type) if not items: raise TradeError("没有可平的持仓") total = sum(float(p.volume) for p in items) if volume > total + 1e-8: raise TradeError(f"部分平手数过大，当前可平 {total:.2f}") info = self.symbol_info(symbol) vmin = float(info.volume_min) step = float(info.volume_step) remaining = volume for p in items: if remaining <= 1e-8: break pos_vol = float(p.volume) close_vol = math.floor(min(remaining, pos_vol) / step + 1e-8) * step close_vol = round(close_vol, 8) if close_vol + 1e-8 < vmin: raise TradeError("剩余手数小于最小手数，无法继续部分平仓") order_type = ( mt5.ORDER_TYPE_SELL if int(p.type) == mt5.POSITION_TYPE_BUY else mt5.ORDER_TYPE_BUY ) t = self.tick(symbol) price = float(t.bid if order_type == mt5.ORDER_TYPE_SELL else t.ask) request = { "action": mt5.TRADE_ACTION_DEAL, "symbol": symbol, "volume": close_vol, "type": order_type, "position": int(p.ticket), "price": price, "deviation": DEVIATION, "magic": MAGIC, "comment": COMMENT, "type_time": mt5.ORDER_TIME_GTC, } try: self._send_deal(request) except TradeError as exc: raise TradeError(f"部分平仓失败 #{p.ticket}：{exc}") from exc remaining = round(remaining - close_vol, 8) return f"已部分平仓 {volume:.2f} 手" class OrderApp(tk.Tk): def __init__(self) -> None: super().__init__() _init_mt5_constants() self.title(APP_TITLE) self.geometry("1100x980") self.minsize(1000, 900) self.configure(bg=COLOR_BG) try: self.tk.call("tk", "scaling", 1.45) except tk.TclError: pass self.bridge = Mt5Bridge() self._busy = False self._lot_auto_done = False self._selected_tickets_set: set = set() self._be_done_tickets: set = set() # 已拉过保本的单号，不再判断 self._pos_sig = None self._pos_profit_labels: dict = {} self._metric_values: dict = {} self._last_spread: float = 0.0 self._force_today_stats = True self._last_today_stats_ts = 0.0 self._atr_cache: Tuple[float, float] = (0.0, 0.0) # (ts, value) self._day_ohlc_cache: Tuple[float, Any] = (0.0, None) # (ts, dict) self._last_float_sync = 0.0 self._last_reconcile_sync = 0.0 self._hub_sync_busy = False self._cached_today_stats: Optional[dict] = None self._build_ui() self.after(200, self._try_connect_and_refresh) self.after(3000, self._hub_heartbeat) def _card(self, parent: tk.Misc, pad: int = 12) -> tk.Frame: outer = tk.Frame(parent, bg=COLOR_BG) card = tk.Frame( outer, bg=COLOR_CARD, highlightthickness=1, highlightbackground="#2a2a38", highlightcolor="#2a2a38", ) card.pack(fill=tk.BOTH, expand=True, padx=0, pady=0) inner = tk.Frame(card, bg=COLOR_CARD) inner.pack(fill=tk.BOTH, expand=True, padx=pad, pady=pad) outer._inner = inner # type: ignore[attr-defined] return outer def _metric_box(self, parent: tk.Misc, key: str, title: str) -> None: box = tk.Frame(parent, bg=COLOR_CARD2, highlightthickness=0) box.pack(side=tk.LEFT, fill=tk.BOTH, expand=True, padx=4) tk.Label( box, text=title, bg=COLOR_CARD2, fg=COLOR_MUTED, font=("Microsoft YaHei UI", 10), ).pack(anchor="w", padx=10, pady=(8, 0)) val = tk.Label( box, text="--", bg=COLOR_CARD2, fg=COLOR_TEXT, font=("Microsoft YaHei UI", 13, "bold"), ) val.pack(anchor="w", padx=10, pady=(2, 8)) self._metric_values[key] = val def _build_ui(self) -> None: style = ttk.Style(self) try: style.theme_use("clam") except tk.TclError: pass style.configure("TLabel", background=COLOR_BG, foreground=COLOR_TEXT, font=("Microsoft YaHei UI", 12)) style.configure("Status.TLabel", background=COLOR_BG, foreground=COLOR_MUTED, font=("Microsoft YaHei UI", 11)) style.configure("TFrame", background=COLOR_BG) style.configure("Card.TFrame", background=COLOR_CARD) style.configure("TEntry", fieldbackground="#2a2a36", foreground="#ffffff", font=("Microsoft YaHei UI", 12)) style.configure("TCheckbutton", background=COLOR_CARD, foreground=COLOR_TEXT, font=("Microsoft YaHei UI", 11)) style.map("TCheckbutton", background=[("active", COLOR_CARD)]) root = tk.Frame(self, bg=COLOR_BG) root.pack(fill=tk.BOTH, expand=True, padx=16, pady=14) # —— 顶栏 —— head_outer = self._card(root, pad=12) head_outer.pack(fill=tk.X, pady=(0, 10)) head = head_outer._inner # type: ignore[attr-defined] left = tk.Frame(head, bg=COLOR_CARD) left.pack(side=tk.LEFT) tk.Label(left, text=APP_TITLE, bg=COLOR_CARD, fg=COLOR_ACCENT, font=("Microsoft YaHei UI", 14, "bold")).pack(side=tk.LEFT) tk.Label(left, text=" 品种", bg=COLOR_CARD, fg=COLOR_MUTED, font=("Microsoft YaHei UI", 11)).pack(side=tk.LEFT, padx=(16, 4)) self.var_symbol = tk.StringVar(value=DEFAULT_SYMBOL) ttk.Entry(left, textvariable=self.var_symbol, width=14).pack(side=tk.LEFT) self.lbl_account = tk.Label( left, text="余额 --", bg=COLOR_CARD, fg=COLOR_TEXT, font=("Microsoft YaHei UI", 12, "bold"), ) self.lbl_account.pack(side=tk.LEFT, padx=16) self.lbl_session = tk.Label( head, text="时段 --", bg="#243044", fg=COLOR_ACCENT, font=("Microsoft YaHei UI", 11, "bold"), padx=12, pady=4, ) self.lbl_session.pack(side=tk.RIGHT) # —— 报价 + 日线指标 —— quote_outer = self._card(root, pad=12) quote_outer.pack(fill=tk.X, pady=(0, 10)) q = quote_outer._inner # type: ignore[attr-defined] self.lbl_quote = tk.Label( q, text="Bid -- Ask --", bg=COLOR_CARD, fg=COLOR_TEXT, font=("Microsoft YaHei UI", 18, "bold"), ) self.lbl_quote.pack(anchor="w") self.lbl_info = tk.Label( q, text="点差 -- ATR -- SL$ -- TP$ -- 持仓 --", bg=COLOR_CARD, fg=COLOR_MUTED, font=("Microsoft YaHei UI", 11), ) self.lbl_info.pack(anchor="w", pady=(4, 10)) metrics = tk.Frame(q, bg=COLOR_CARD) metrics.pack(fill=tk.X) self._metric_box(metrics, "elapsed", "开盘时长") self._metric_box(metrics, "open", "开盘价") self._metric_box(metrics, "high", "最高价") self._metric_box(metrics, "low", "最低价") self._metric_box(metrics, "range", "振幅") self.lbl_day_ohlc = tk.Label( q, text="", bg=COLOR_CARD, fg=COLOR_MUTED, font=("Microsoft YaHei UI", 10), ) self.lbl_day_ohlc.pack(anchor="w", pady=(8, 0)) # —— 今日战绩 —— today_outer = self._card(root, pad=10) today_outer.pack(fill=tk.X, pady=(0, 10)) today_row = today_outer._inner # type: ignore[attr-defined] self.lbl_today_base = tk.Label( today_row, text="今日开单 -- 次", bg=COLOR_CARD, fg=COLOR_MUTED, font=("Microsoft YaHei UI", 11), ) self.lbl_today_base.pack(side=tk.LEFT) self.lbl_today_closed = tk.Label( today_row, text=" 已平 --", bg=COLOR_CARD, fg=COLOR_NEUTRAL, font=("Microsoft YaHei UI", 11, "bold"), ) self.lbl_today_closed.pack(side=tk.LEFT) self.lbl_today_mid = tk.Label( today_row, text=" 浮盈 -- 合计 ", bg=COLOR_CARD, fg=COLOR_MUTED, font=("Microsoft YaHei UI", 11), ) self.lbl_today_mid.pack(side=tk.LEFT) self.lbl_today_total = tk.Label( today_row, text="--", bg=COLOR_CARD, fg=COLOR_NEUTRAL, font=("Microsoft YaHei UI", 11, "bold"), ) self.lbl_today_total.pack(side=tk.LEFT) self.lbl_today_ratio = tk.Label( today_row, text=" 盈亏比 -- 盈利率 -- 胜率 --", bg=COLOR_CARD, fg=COLOR_MUTED, font=("Microsoft YaHei UI", 11, "bold"), ) self.lbl_today_ratio.pack(side=tk.LEFT) self.lbl_day_lock = tk.Label( today_row, text="", bg=COLOR_CARD, fg=COLOR_WARN, font=("Microsoft YaHei UI", 11, "bold"), padx=8, pady=2, ) self.lbl_day_lock.pack(side=tk.RIGHT) # —— 下单参数 —— form_outer = self._card(root, pad=12) form_outer.pack(fill=tk.X, pady=(0, 10)) form_wrap = form_outer._inner # type: ignore[attr-defined] form = ttk.Frame(form_wrap) form.pack(fill=tk.X) for i in range(3): form.columnconfigure(i, weight=1) style.configure("Form.TLabel", background=COLOR_CARD, foreground=COLOR_MUTED, font=("Microsoft YaHei UI", 11)) ttk.Label(form, text="手数", style="Form.TLabel").grid(row=0, column=0, sticky="w") self.lbl_sl_title = ttk.Label(form, text="SL倍", style="Form.TLabel") self.lbl_sl_title.grid(row=0, column=1, sticky="w", padx=6) self.lbl_tp_title = ttk.Label(form, text="TP倍", style="Form.TLabel") self.lbl_tp_title.grid(row=0, column=2, sticky="w") self.var_lot = tk.StringVar(value=f"{DEFAULT_LOT:.2f}") self.var_atr_mult = tk.StringVar(value=f"{DEFAULT_ATR_MULT:.2f}") self.var_tp = tk.StringVar(value=f"{DEFAULT_TP_ATR_MULT:.2f}") ttk.Entry(form, textvariable=self.var_lot, width=12).grid(row=1, column=0, sticky="ew", pady=6, ipady=5) ttk.Entry(form, textvariable=self.var_atr_mult, width=12).grid(row=1, column=1, sticky="ew", padx=8, pady=6, ipady=5) ttk.Entry(form, textvariable=self.var_tp, width=12).grid(row=1, column=2, sticky="ew", pady=6, ipady=5) opt = tk.Frame(form_wrap, bg=COLOR_CARD) opt.pack(fill=tk.X, pady=(6, 0)) self.var_use_atr = tk.BooleanVar(value=True) ttk.Checkbutton( opt, text="用 M15 ATR(14) 做止损/止盈（ATR<10 按 10）", variable=self.var_use_atr, command=self._sync_sl_label, ).pack(side=tk.LEFT) self._mk_btn(opt, "按资金手数", "#3a5a80", self.on_auto_lot, h=34, w=110).pack(side=tk.LEFT, padx=12) tk.Label( opt, text=f"小资金点一下→{LOT_LOW:.2f}手 · 同向≤{MAX_SIDE_LOTS:.2f} · 最多{MAX_OPEN_POSITIONS}笔", bg=COLOR_CARD, fg=COLOR_MUTED, font=("Microsoft YaHei UI", 10), ).pack(side=tk.LEFT) self.lbl_open_block = tk.Label( root, text="", bg=COLOR_BG, fg=COLOR_WARN, font=("Microsoft YaHei UI", 12, "bold"), anchor="center", ) self.lbl_open_block.pack(fill=tk.X, pady=(0, 4)) btns = tk.Frame(root, bg=COLOR_BG) btns.pack(fill=tk.X, pady=4) btns.columnconfigure(0, weight=1) btns.columnconfigure(1, weight=1) self.btn_buy = self._mk_btn(btns, "买入", COLOR_BUY, self.on_buy, h=48) self.btn_buy.grid(row=0, column=0, sticky="ew", padx=(0, 6), ipady=8) self.btn_sell = self._mk_btn(btns, "卖出", COLOR_SELL, self.on_sell, h=48) self.btn_sell.grid(row=0, column=1, sticky="ew", padx=(6, 0), ipady=8) close = tk.Frame(root, bg=COLOR_BG) close.pack(fill=tk.X, pady=8) for i in range(3): close.columnconfigure(i, weight=1) self._mk_btn(close, "平多", "#285032", self.on_close_buy, h=38).grid(row=0, column=0, sticky="ew", padx=(0, 4), ipady=4) self._mk_btn(close, "平空", "#5a3232", self.on_close_sell, h=38).grid(row=0, column=1, sticky="ew", padx=4, ipady=4) self._mk_btn(close, "全平", "#464628", self.on_close_all, h=38).grid(row=0, column=2, sticky="ew", padx=(4, 0), ipady=4) self.lbl_status = tk.Label( root, text="就绪", bg=COLOR_BG, fg=COLOR_MUTED, font=("Microsoft YaHei UI", 11), anchor="w", wraplength=960, ) self.lbl_status.pack(fill=tk.X, pady=(8, 4)) pos_bar = tk.Frame(root, bg=COLOR_BG) pos_bar.pack(fill=tk.X, pady=(8, 6)) tk.Label( pos_bar, text="持仓记录", bg=COLOR_BG, fg=COLOR_TEXT, font=("Microsoft YaHei UI", 13, "bold"), ).pack(side=tk.LEFT) self._be_ready = False self.btn_break_even = self._mk_btn(pos_bar, "保本", COLOR_BE, self.on_break_even, h=34, w=90) self.btn_break_even.pack(side=tk.RIGHT, padx=4) self._mk_btn(pos_bar, "打开复盘", "#3a5a80", self.on_open_review, h=34, w=100).pack( side=tk.RIGHT, padx=4 ) self._set_be_button_enabled(False) # 持仓列表：无竖线；列宽按权重拉满右侧；平仓红色边框按钮 self._pos_col_specs = [ ("单号", 12), ("方向", 5), ("手数", 6), ("开仓价", 12), ("止损幅度", 9), ("止损", 12), ("止盈", 12), ("浮盈", 9), ("操作", 16), ] # 持仓区固定为 5 行可视高度，超出可滚动 pos_wrap = tk.Frame( root, bg="#2b2b2b", highlightthickness=1, highlightbackground="#2a2a38", ) pos_wrap.pack(fill=tk.X, expand=False, pady=(0, 0)) self.pos_head = tk.Frame(pos_wrap, bg="#333333") self.pos_head.pack(fill=tk.X) self._config_pos_columns(self.pos_head) for i, (text, _) in enumerate(self._pos_col_specs): tk.Label( self.pos_head, text=text, bg="#333333", fg="#ffffff", font=("Microsoft YaHei UI", 11, "bold"), anchor="center", ).grid(row=0, column=i, sticky="nsew", padx=2, pady=6) body = tk.Frame(pos_wrap, bg="#2b2b2b", height=POS_LIST_HEIGHT) body.pack(fill=tk.X, expand=False) body.pack_propagate(False) self.pos_canvas = tk.Canvas( body, bg="#2b2b2b", highlightthickness=0, bd=0, height=POS_LIST_HEIGHT ) scroll = ttk.Scrollbar(body, orient=tk.VERTICAL, command=self.pos_canvas.yview) self.pos_inner = tk.Frame(self.pos_canvas, bg="#2b2b2b") self.pos_inner.bind( "", lambda e: self.pos_canvas.configure(scrollregion=self.pos_canvas.bbox("all")), ) self._pos_window = self.pos_canvas.create_window((0, 0), window=self.pos_inner, anchor="nw") self.pos_canvas.configure(yscrollcommand=scroll.set, height=POS_LIST_HEIGHT) self.pos_canvas.pack(side=tk.LEFT, fill=tk.BOTH, expand=True) scroll.pack(side=tk.RIGHT, fill=tk.Y) self.pos_canvas.bind("", self._on_pos_canvas_configure) self.pos_canvas.bind("", lambda e: self.pos_canvas.bind_all("", self._on_pos_mousewheel)) self.pos_canvas.bind("", lambda e: self.pos_canvas.unbind_all("")) tip = tk.Label( root, text=( f"最多同时 {MAX_OPEN_POSITIONS} 笔 · 同向≤{MAX_SIDE_LOTS:.2f} 手 · " f"点差>{SPREAD_WARN_USD:.2f} 买卖提示 · " "亏≥5%或盈≥10%禁止开仓 · 持仓「改止损」可手动设价 · 保本：有利≥3$ 后止损到开仓价±1$" ), bg=COLOR_BG, fg=COLOR_MUTED, font=("Microsoft YaHei UI", 10), wraplength=960, justify=tk.LEFT, anchor="w", ) tip.pack(anchor="w", pady=(8, 0)) def _mk_btn(self, parent, text, color, cmd, h=36, w=None): btn = tk.Button( parent, text=text, command=cmd, bg=color, fg="white", activebackground=color, activeforeground="white", relief=tk.FLAT, height=1, font=("Microsoft YaHei UI", 13, "bold"), padx=10, pady=4, ) if w: btn.configure(width=max(8, w // 10)) return btn def _sync_sl_label(self) -> None: if self.var_use_atr.get(): self.lbl_sl_title.configure(text="SL倍") self.lbl_tp_title.configure(text="TP倍") # 从美元模式切回来时，给合理默认倍 try: v = float(self.var_tp.get().replace(",", ".") or "0") if v >= 5: self.var_tp.set(f"{DEFAULT_TP_ATR_MULT:.2f}") except ValueError: self.var_tp.set(f"{DEFAULT_TP_ATR_MULT:.2f}") else: self.lbl_sl_title.configure(text="SL$") self.lbl_tp_title.configure(text="TP$") if self.var_atr_mult.get() in ("1", "1.0", "1.00"): self.var_atr_mult.set("8.0") try: v = float(self.var_tp.get().replace(",", ".") or "0") if v <= 3: self.var_tp.set(f"{DEFAULT_TP_USD:.1f}") except ValueError: self.var_tp.set(f"{DEFAULT_TP_USD:.1f}") def _symbol(self) -> str: s = self.var_symbol.get().strip() return s or DEFAULT_SYMBOL def set_status(self, text: str, ok: bool = True) -> None: self.lbl_status.configure(text=text, fg="#7CFC00" if ok else "#FF7F50") def _parse_float(self, value: str, name: str) -> float: s = value.strip().replace(",", ".") if s == "": raise TradeError(f"{name} 不能为空") try: return float(s) except ValueError as exc: raise TradeError(f"{name} 不是有效数字") from exc @staticmethod def _atr_for_stops(atr: float) -> float: """止损/止盈用的 ATR：小于 10 按 10，大于等于 10 用实际值。""" return max(float(atr), ATR_FLOOR_USD) def _resolve_stops(self, symbol: str) -> Tuple[float, float, float, float, float]: """返回 sl_usd, tp_usd, atr(原始), sl_mult, tp_mult""" if not self.var_use_atr.get(): sl = self._parse_float(self.var_atr_mult.get(), "SL$") tp = self._parse_float(self.var_tp.get(), "TP$") return sl, tp, 0.0, 0.0, 0.0 sl_mult = self._parse_float(self.var_atr_mult.get(), "SL倍") tp_mult = self._parse_float(self.var_tp.get(), "TP倍") if sl_mult < 0 or tp_mult < 0: raise TradeError("ATR 倍数不能为负") atr = self.bridge.atr(symbol) if (sl_mult > 1e-8 or tp_mult > 1e-8) else 0.0 base = self._atr_for_stops(atr) if atr > 1e-12 else 0.0 return base * sl_mult, base * tp_mult, atr, sl_mult, tp_mult def _confirm(self, title: str, body: str) -> bool: if not MUST_CONFIRM: return True return messagebox.askokcancel(title, body, parent=self) def _on_pos_mousewheel(self, event) -> None: if not hasattr(self, "pos_canvas"): return self.pos_canvas.yview_scroll(int(-1 * (event.delta / 120)), "units") def _on_pos_canvas_configure(self, event) -> None: self.pos_canvas.itemconfigure(self._pos_window, width=event.width) def _config_pos_columns(self, frame: tk.Frame) -> None: for i, (_, weight) in enumerate(self._pos_col_specs): frame.columnconfigure(i, weight=weight, uniform="poscol") def _set_trade_btn(self, btn: tk.Button, enabled: bool, on_color: str) -> None: if enabled: btn.configure( state=tk.NORMAL, bg=on_color, fg="white", activebackground=on_color, activeforeground="white", disabledforeground="white", cursor="hand2", ) else: btn.configure( state=tk.DISABLED, bg=COLOR_BTN_OFF, fg="#888888", activebackground=COLOR_BTN_OFF, activeforeground="#888888", disabledforeground="#888888", cursor="arrow", ) def _sync_open_buttons(self, symbol: str) -> None: """满仓 / 同向上限 / 日内达标 → 买卖置灰；点差过大时买卖区提示（不禁开）。""" if not hasattr(self, "btn_buy"): return try: positions = self.bridge.positions(symbol) buy_lots, sell_lots, _ = self.bridge.exposure(symbol) lock_kind, rate = self.bridge.day_lock_kind(symbol) except TradeError: self._set_trade_btn(self.btn_buy, False, COLOR_BUY) self._set_trade_btn(self.btn_sell, False, COLOR_SELL) return try: lot = float(self.var_lot.get().replace(",", ".") or "0") except ValueError: lot = 0.0 n = len(positions) full = n >= MAX_OPEN_POSITIONS day_locked = lock_kind is not None can_buy = (not day_locked) and (not full) and lot > 1e-8 and (buy_lots + lot <= MAX_SIDE_LOTS + 1e-8) can_sell = (not day_locked) and (not full) and lot > 1e-8 and (sell_lots + lot <= MAX_SIDE_LOTS + 1e-8) self._set_trade_btn(self.btn_buy, can_buy, COLOR_BUY) self._set_trade_btn(self.btn_sell, can_sell, COLOR_SELL) spread_wide = self._last_spread > SPREAD_WARN_USD + 1e-9 pct = rate * 100.0 if lock_kind == "loss": badge = f"亏损达标 {pct:.2f}% · 禁止开仓" banner = f"⚠ 当日亏损已达 {pct:.2f}%（上限 -{DAY_LOSS_LOCK*100:.0f}%）｜禁止开仓" color = COLOR_LOSS bg = "#3a1e1e" elif lock_kind == "profit": badge = f"盈利达标 {pct:.2f}% · 禁止开仓" banner = f"⚠ 当日盈利已达 {pct:.2f}%（上限 +{DAY_PROFIT_LOCK*100:.0f}%）｜禁止开仓" color = COLOR_PROFIT bg = "#1e3a2a" elif full: badge = f"持仓已满 {n}/{MAX_OPEN_POSITIONS} · 禁止开仓" banner = f"持仓已满 {MAX_OPEN_POSITIONS} 笔，平仓后再开" color = COLOR_WARN bg = "#3a3220" elif spread_wide: badge = f"点差偏大 {self._last_spread:.2f}$" banner = ( f"⚠ 点差过大 {self._last_spread:.2f}$（阈值 {SPREAD_WARN_USD:.2f}）｜" "建议暂缓开仓" ) color = COLOR_WARN bg = "#3a3220" else: badge = "" banner = "" color = COLOR_NEUTRAL bg = COLOR_BG # 买卖按钮文案：可开且点差大时直接写在按钮上 if can_buy: self.btn_buy.configure(text="买入 ⚠点差大" if spread_wide else "买入") else: self.btn_buy.configure(text="买入") if can_sell: self.btn_sell.configure(text="卖出 ⚠点差大" if spread_wide else "卖出") else: self.btn_sell.configure(text="卖出") self.lbl_day_lock.configure(text=badge, fg=color, bg=bg if badge else COLOR_CARD) self.lbl_open_block.configure(text=banner, fg=color, bg=bg if banner else COLOR_BG) def _toggle_ticket_select(self, ticket: int) -> None: if ticket in self._selected_tickets_set: self._selected_tickets_set.discard(ticket) else: self._selected_tickets_set.add(ticket) # 刷新选中底色 try: symbol = self._symbol() digits = int(self.bridge.symbol_info(symbol).digits) self._refresh_positions(symbol, digits) except Exception: pass def _mk_outline_btn( self, parent, text: str, color: str, command, padx: int = 8 ) -> tk.Frame: border = tk.Frame(parent, bg=color, padx=1, pady=1) btn = tk.Button( border, text=text, command=command, fg=color, bg="#2b2b2b", activeforeground="#ffffff", activebackground=color, relief=tk.FLAT, bd=0, font=("Microsoft YaHei UI", 10, "bold"), padx=padx, pady=2, cursor="hand2", ) btn.pack() return border def _mk_close_btn(self, parent, ticket: int) -> tk.Frame: """红色边框的平仓按钮。""" return self._mk_outline_btn( parent, "平仓", COLOR_DANGER, lambda t=ticket: self._close_ticket_now(t) ) def _mk_edit_sl_btn(self, parent, ticket: int) -> tk.Frame: """手动改止损（与保本独立）。""" return self._mk_outline_btn( parent, "改止损", COLOR_ACCENT, lambda t=ticket: self._edit_sl_now(t), padx=6, ) @staticmethod def _pos_profit(p) -> float: return float(p.profit) + float(p.swap) @staticmethod def _is_sl_at_breakeven(p, offset_usd: float = BE_OFFSET_USD, tol: float = 0.05) -> bool: """止损是否已在保本位（或更优）。""" sl = float(p.sl) if sl <= 0: return False open_price = float(p.price_open) if int(p.type) == mt5.POSITION_TYPE_BUY: return sl + 1e-12 >= open_price + offset_usd - tol return sl - 1e-12 <= open_price - offset_usd + tol @staticmethod def _favorable_move_usd(p, tick) -> float: """价格相对开仓价朝有利方向走出的美金距离。多=现价-开仓，空=开仓-现价。""" open_price = float(p.price_open) if int(p.type) == mt5.POSITION_TYPE_BUY: return float(tick.bid) - open_price return open_price - float(tick.ask) def _be_candidate_tickets(self, positions, tick=None) -> list: """价格朝有利方向≥3$ 且尚未保本的单号。""" tickets = [] if not positions: return tickets if tick is None: try: tick = self.bridge.tick(str(positions[0].symbol)) except Exception: return tickets for p in positions: ticket = int(p.ticket) if ticket in self._be_done_tickets: continue if self._is_sl_at_breakeven(p): self._be_done_tickets.add(ticket) continue if self._favorable_move_usd(p, tick) + 1e-12 >= BE_MIN_MOVE_USD: tickets.append(ticket) return tickets def _be_status_hint(self, positions, tick=None) -> str: """保本按钮不可用时的说明（含还差多少美金）。""" if not positions: return "当前无持仓，无法保本" symbol = str(positions[0].symbol) try: if tick is None: tick = self.bridge.tick(symbol) except Exception as exc: return f"无法读报价：{exc}" best = -1e9 for p in positions: ticket = int(p.ticket) if ticket in self._be_done_tickets or self._is_sl_at_breakeven(p): continue best = max(best, self._favorable_move_usd(p, tick)) if best < -1e8: return "持仓已保本或无需再拉" lack = BE_MIN_MOVE_USD - best if lack > 1e-8: return ( f"有利方向仅 {best:.2f}$，保本需 ≥ {BE_MIN_MOVE_USD:.0f}$" f"（之后止损拉到开仓价±{BE_OFFSET_USD:.0f}$），还差 {lack:.2f}$" ) return "可保本" def _set_be_button_enabled(self, enabled: bool) -> None: """外观区分可点/未就绪；始终 NORMAL，避免灰态点不到。""" if not hasattr(self, "btn_break_even"): return self._be_ready = bool(enabled) if enabled: self.btn_break_even.configure( state=tk.NORMAL, bg=COLOR_BE, fg="white", activebackground=COLOR_BE, activeforeground="white", cursor="hand2", ) else: self.btn_break_even.configure( state=tk.NORMAL, bg=COLOR_BE_OFF, fg="#aaaaaa", activebackground="#666666", activeforeground="#dddddd", cursor="hand2", ) def _sync_be_button(self, positions) -> None: try: ok = bool(self._be_candidate_tickets(positions)) except Exception: ok = False self._set_be_button_enabled(ok) def _hub_upload_bars(self, symbol: str) -> None: if hub_client is None or not is_activated(): return tf_map = { "M1": mt5.TIMEFRAME_M1, "M5": mt5.TIMEFRAME_M5, "M15": mt5.TIMEFRAME_M15, "H1": mt5.TIMEFRAME_H1, } for name, tf in tf_map.items(): rates = mt5.copy_rates_from_pos(symbol, tf, 0, 400) if rates is None or len(rates) == 0: continue bars = [ { "ts": float(r["time"]), "open": float(r["open"]), "high": float(r["high"]), "low": float(r["low"]), "close": float(r["close"]), "volume": float(r["tick_volume"]), } for r in rates ] hub_client.report_bars(symbol, name, bars) def _hub_report(self, payload: Union[dict, list, None], with_bars: bool = True) -> None: if hub_client is None or not is_activated() or not payload: return def run(): try: items = payload if isinstance(payload, list) else [payload] for item in items: if not isinstance(item, dict): continue hub_client.report_trade(item) if with_bars: sym = str(item.get("symbol") or self._symbol()) self._hub_upload_bars(sym) except Exception: pass threading.Thread(target=run, daemon=True).start() def _hub_sync_floating(self, symbol: str) -> None: """浮盈上报 +（低频）历史回填/对账。网络在后台，不堵界面。""" if hub_client is None or not is_activated() or self._hub_sync_busy: return now = time.time() if now - float(self._last_float_sync or 0) < HUB_FLOAT_MS / 1000.0: return self._last_float_sync = now do_reconcile = (now - float(self._last_reconcile_sync or 0)) >= HUB_RECONCILE_MS / 1000.0 if do_reconcile: self._last_reconcile_sync = now # 持仓快照在 UI 线程取，避免与刷新抢 MT5 try: poss = self.bridge.positions(symbol) except Exception: return open_tickets = [str(p.ticket) for p in poss] float_reports = [] for p in poss: float_reports.append({ "ticket": str(p.ticket), "symbol": str(p.symbol), "side": "buy" if int(p.type) == mt5.POSITION_TYPE_BUY else "sell", "volume": float(p.volume), "open_price": float(p.price_open), "close_price": None, "open_time": float(p.time), "close_time": None, "profit": float(p.profit) + float(p.swap), "sl": float(p.sl), "tp": float(p.tp), "snapshot": {"event": "float"}, }) self._hub_sync_busy = True def run() -> None: try: reports = list(float_reports) if do_reconcile: try: reports[0:0] = self.bridge.recent_close_reports(symbol, hours=48) except Exception: pass for item in reports: try: hub_client.report_trade(item) except Exception: pass if do_reconcile: try: hub_client.reconcile_trades(open_tickets, symbol) except Exception: pass except Exception: pass finally: self.after(0, lambda: setattr(self, "_hub_sync_busy", False)) threading.Thread(target=run, daemon=True).start() def _run_async(self, work) -> None: if self._busy: self.set_status("上一次操作还未完成", ok=False) return self._busy = True def runner(): try: result = work() report: Any = None if isinstance(result, tuple) and len(result) == 2: msg, second = result if isinstance(second, dict) or ( isinstance(second, list) and second and isinstance(second[0], dict) ): report = second else: msg = result self.after(0, lambda m=msg: self.set_status(m, ok=True)) if report is not None: # 成交后立刻刷新持仓；K 线上报延后，避免和 MT5 抢线程 self.after(0, self._kick_after_trade) self.after(0, lambda r=report: self._hub_report(r, with_bars=False)) self.after(800, lambda r=report: self._hub_upload_bars_for_report(r)) except TradeError as exc: self.after(0, lambda: self.set_status(str(exc), ok=False)) except Exception as exc: # noqa: BLE001 self.after(0, lambda: self.set_status(f"异常：{exc}", ok=False)) finally: self.after(0, self._done) threading.Thread(target=runner, daemon=True).start() def _kick_after_trade(self) -> None: """开平仓成功后立刻重绘持仓与今日战绩。""" self._force_today_stats = True self._pos_sig = None self._last_float_sync = 0.0 def _hub_upload_bars_for_report(self, payload: Union[dict, list, None]) -> None: if hub_client is None or not is_activated() or not payload: return def run() -> None: try: items = payload if isinstance(payload, list) else [payload] for item in items: if not isinstance(item, dict): continue sym = str(item.get("symbol") or self._symbol()) self._hub_upload_bars(sym) break # 一次成交只传一套 K 线即可 except Exception: pass threading.Thread(target=run, daemon=True).start() def _done(self) -> None: self._busy = False self._kick_after_trade() self.refresh_quotes() def on_auto_lot(self) -> None: """任意资金都可点：小资金（<1000）会设为 0.01 手。""" try: self.bridge.connect() bal = self.bridge.account_balance() lot = suggest_lot_by_balance(bal) self.var_lot.set(f"{lot:.2f}") if bal < BALANCE_LOW: tip = f"小资金档：余额 {bal:.2f} → 手数 {lot:.2f}" elif bal >= BALANCE_HIGH: tip = f"大资金档：余额 {bal:.2f} → 手数 {lot:.2f}" else: tip = f"中资金档：余额 {bal:.2f} → 手数 {lot:.2f}" self.set_status(tip, ok=True) except TradeError as exc: self.set_status(str(exc), ok=False) def _apply_auto_lot_once(self) -> None: if self._lot_auto_done: return try: bal = self.bridge.account_balance() lot = suggest_lot_by_balance(bal) self.var_lot.set(f"{lot:.2f}") self._lot_auto_done = True self.set_status(f"已连接 MT5；余额 {bal:.2f}，默认手数 {lot:.2f}", ok=True) except TradeError: pass def _try_connect_and_refresh(self) -> None: try: self.bridge.connect() self._apply_auto_lot_once() if not self._lot_auto_done: self.set_status("已连接 MT5", ok=True) except TradeError as exc: self.set_status(str(exc), ok=False) self.refresh_quotes() self.after(REFRESH_MS, self._try_connect_and_refresh) def _set_metric(self, key: str, text: str, fg: str = COLOR_TEXT) -> None: lbl = self._metric_values.get(key) if lbl is not None: lbl.configure(text=text, fg=fg) def refresh_quotes(self) -> None: symbol = self._symbol() try: if not self.bridge.connected: self.bridge.connect() info = self.bridge.symbol_info(symbol) digits = int(info.digits) market_open = False bid = ask = 0.0 try: t = self.bridge.tick(symbol) bid = float(t.bid) ask = float(t.ask) market_open = self.bridge.market_is_open(symbol) except TradeError: bid = float(getattr(info, "bid", 0) or getattr(info, "last", 0) or 0) ask = float(getattr(info, "ask", 0) or bid) market_open = False spread = ask - bid if ask and bid else 0.0 self._last_spread = spread try: buy, sell, profit = self.bridge.exposure(symbol) except TradeError: buy = sell = profit = 0.0 net = buy - sell pos = f"{net:.2f}" if buy > 0 and sell > 0: pos = f"多{buy:.2f} 空{sell:.2f}" atr_txt = sl_txt = tp_txt = "--" if self.var_use_atr.get(): try: atr = self._cached_atr(symbol) base = self._atr_for_stops(atr) atr_txt = f"{atr:.2f}" + (f"→{base:.0f}" if base > atr + 1e-9 else "") sl_m = float(self.var_atr_mult.get().replace(",", ".") or "0") tp_m = float(self.var_tp.get().replace(",", ".") or "0") sl_txt = f"{base * sl_m:.2f}" tp_txt = f"{base * tp_m:.2f}" except Exception: pass else: sl_txt = self.var_atr_mult.get() tp_txt = self.var_tp.get() try: bal = self.bridge.account_balance() self.lbl_account.configure(text=f"余额 {bal:.2f}") except TradeError: self.lbl_account.configure(text="余额 --") sess = current_session_label(market_open=market_open) if not market_open: sess_txt = f"休市 · {format_open_elapsed(0, False)}" else: sess_txt = f"时段 {sess}" self.lbl_session.configure( text=sess_txt, bg="#3a2820" if not market_open else "#243044", fg=COLOR_WARN if not market_open else COLOR_ACCENT, ) if bid > 0: self.lbl_quote.configure(text=f"Bid {bid:.{digits}f} Ask {ask:.{digits}f}") else: self.lbl_quote.configure(text="Bid -- Ask --") spread_wide = spread > SPREAD_WARN_USD + 1e-9 self.lbl_info.configure( text=( f"点差 {spread:.{digits}f}" f"{' ⚠偏大' if spread_wide else ''} ATR {atr_txt} " f"SL${sl_txt} TP${tp_txt} 持仓 {pos} 浮盈 {profit:.2f}" ), fg=COLOR_WARN if spread_wide else COLOR_MUTED, ) try: day = self._cached_day_ohlc(symbol, bid=bid, ask=ask, market_open=market_open) elapsed_fg = COLOR_WARN if not day["market_open"] else COLOR_ACCENT self._set_metric("elapsed", day["elapsed_text"], elapsed_fg) self._set_metric("open", f"{day['open']:.{digits}f}") self._set_metric("high", f"{day['high']:.{digits}f}", COLOR_PROFIT) self._set_metric("low", f"{day['low']:.{digits}f}", COLOR_LOSS) amp = float(day["high"]) - float(day["low"]) base = float(day["open"]) or float(day["low"]) or 1.0 amp_pct = (amp / base) * 100.0 if base else 0.0 self._set_metric( "range", f"{amp:.{digits}f} ({amp_pct:.2f}%)", COLOR_ACCENT, ) ot = fmt_bjt(day["open_time"]) if day["market_open"]: tag = "交易中" else: nxt_txt = day.get("next_open_text") or "" tag = f"休市 · {day['elapsed_text']}" if nxt_txt: tag += f"（预计开盘 {nxt_txt} 北京时间）" self.lbl_day_ohlc.configure(text=f"{tag} · 日线开盘 {ot} 北京时间") except Exception: left = format_open_elapsed(0, False) if not market_open else "--" self._set_metric("elapsed", left, COLOR_WARN) self._set_metric("open", "--") self._set_metric("high", "--") self._set_metric("low", "--") self._set_metric("range", "--") self.lbl_day_ohlc.configure(text="日线暂不可用（休市也会尽量显示最近交易日）") try: self._refresh_today_stats(symbol, floating=profit) except Exception: self.lbl_today_base.configure(text="今日开单 -- 次") self.lbl_today_closed.configure(text=" 已平 --", fg=COLOR_NEUTRAL) self.lbl_today_mid.configure(text=" 浮盈 -- 合计 ") self.lbl_today_total.configure(text="--", fg=COLOR_NEUTRAL) self.lbl_today_ratio.configure(text=" 盈亏比 -- 盈利率 -- 胜率 --", fg=COLOR_NEUTRAL) self._refresh_positions(symbol, digits) self._sync_open_buttons(symbol) self._hub_sync_floating(symbol) except TradeError: # 断连时仍尝试刷新日线卡片 try: day = self.bridge.day_ohlc(symbol) digits = 3 self._set_metric("elapsed", day["elapsed_text"], COLOR_WARN) self._set_metric("open", f"{day['open']:.{digits}f}") self._set_metric("high", f"{day['high']:.{digits}f}", COLOR_PROFIT) self._set_metric("low", f"{day['low']:.{digits}f}", COLOR_LOSS) amp = float(day["high"]) - float(day["low"]) base = float(day["open"]) or float(day["low"]) or 1.0 amp_pct = (amp / base) * 100.0 if base else 0.0 self._set_metric( "range", f"{amp:.{digits}f} ({amp_pct:.2f}%)", COLOR_ACCENT, ) self.lbl_session.configure( text=f"休市 · {day['elapsed_text']}", bg="#3a2820", fg=COLOR_WARN, ) nxt = day.get("next_open_text") or "" tip = f"休市 · {day['elapsed_text']}" if nxt: tip += f"（预计开盘 {nxt} 北京时间）" self.lbl_day_ohlc.configure(text=f"{tip} · 显示最近交易日日线") except Exception: pass except Exception: pass @staticmethod def _pnl_color(value: float) -> str: if value > 1e-8: return COLOR_PROFIT if value < -1e-8: return COLOR_LOSS return COLOR_NEUTRAL def _cached_atr(self, symbol: str) -> float: now = time.time() ts, val = self._atr_cache if val > 0 and (now - ts) * 1000.0 < ATR_CACHE_MS: return val atr = float(self.bridge.atr(symbol)) self._atr_cache = (now, atr) return atr def _cached_day_ohlc( self, symbol: str, bid: float = 0.0, ask: float = 0.0, market_open: bool = True ) -> dict: now = time.time() ts, cached = self._day_ohlc_cache if cached is not None and (now - ts) * 1000.0 < DAY_OHLC_CACHE_MS: day = dict(cached) # 缓存期内仍用最新报价刷新高低价与开盘时长 if market_open and bid > 0: day["high"] = max(float(day["high"]), bid, ask or bid) day["low"] = min(float(day["low"]), bid, ask or bid) day["market_open"] = market_open day["elapsed_text"] = format_open_elapsed(int(day.get("open_ts") or 0), market_open) return day day = self.bridge.day_ohlc(symbol) self._day_ohlc_cache = (now, dict(day)) return day def _refresh_today_stats(self, symbol: str, floating: Optional[float] = None) -> None: now = time.time() need = self._force_today_stats or ( (now - self._last_today_stats_ts) * 1000.0 >= TODAY_STATS_MS ) if need: st = self.bridge.today_stats(symbol) self._cached_today_stats = st self._last_today_stats_ts = now self._force_today_stats = False else: st = self._cached_today_stats if not st: st = self.bridge.today_stats(symbol) self._cached_today_stats = st self._last_today_stats_ts = now else: # 节流期内只刷新浮盈/合计，避免每拍扫成交历史 st = dict(st) if floating is not None: st["floating"] = float(floating) st["total"] = float(st.get("closed_pnl") or 0) + float(floating) bal = float(st.get("balance") or 0) st["profit_rate"] = (st["total"] / bal) if bal > 1e-8 else 0.0 closed = st["closed_pnl"] floating_v = st["floating"] if floating is None else float(floating) total = float(st.get("closed_pnl") or 0) + floating_v rate = (total / float(st.get("balance") or 1)) * 100.0 if float(st.get("balance") or 0) > 1e-8 else 0.0 self.lbl_today_base.configure(text=f"今日开单 {st['opens']} 次") self.lbl_today_closed.configure(text=f" 已平 {closed:+.2f}", fg=self._pnl_color(closed)) self.lbl_today_mid.configure(text=f" 浮盈 {floating_v:+.2f} 合计 ") self.lbl_today_total.configure(text=f"{total:+.2f}", fg=self._pnl_color(total)) self.lbl_today_ratio.configure( text=( f" | 盈亏比 {st['pl_ratio_txt']} | 盈利率 {rate:+.2f}%" f" | 胜率 {st['win_rate_txt']}" ), fg=self._pnl_color(total), ) def _refresh_positions(self, symbol: str, digits: int) -> None: positions = self.bridge.positions(symbol) live = {int(p.ticket) for p in positions} self._selected_tickets_set = {t for t in self._selected_tickets_set if t in live} self._be_done_tickets = {t for t in self._be_done_tickets if t in live} sig = tuple( ( int(p.ticket), int(p.type), round(float(p.volume), 4), round(float(p.price_open), 5), round(float(p.sl), 5), round(float(p.tp), 5), int(p.ticket) in self._selected_tickets_set, ) for p in positions ) # 结构未变时只更新浮盈，避免按钮闪烁点不中 if sig == self._pos_sig and self._pos_profit_labels: for p in positions: ticket = int(p.ticket) profit = self._pos_profit(p) lbl = self._pos_profit_labels.get(ticket) if lbl is not None and lbl.winfo_exists(): lbl.configure(text=f"{profit:.2f}", fg=self._pnl_color(profit)) self._sync_be_button(positions) return self._pos_sig = sig self._pos_profit_labels = {} for child in self.pos_inner.winfo_children(): child.destroy() for p in positions: ticket = int(p.ticket) is_buy = int(p.type) == mt5.POSITION_TYPE_BUY side = "多" if is_buy else "空" sl = float(p.sl) tp = float(p.tp) open_price = float(p.price_open) sl_dist = f"{abs(open_price - sl):.2f}u" if sl > 0 else "--" profit = self._pos_profit(p) selected = ticket in self._selected_tickets_set bg = "#3a4a5a" if selected else "#2b2b2b" row = tk.Frame(self.pos_inner, bg=bg) row.pack(fill=tk.X, pady=1) self._config_pos_columns(row) cells = [ str(ticket), side, f"{float(p.volume):.2f}", f"{open_price:.{digits}f}", sl_dist, ("--" if sl <= 0 else f"{sl:.{digits}f}"), ("--" if tp <= 0 else f"{tp:.{digits}f}"), f"{profit:.2f}", ] for i, text in enumerate(cells): title = self._pos_col_specs[i][0] if title == "浮盈": fg = self._pnl_color(profit) elif title == "方向": fg = COLOR_PROFIT if is_buy else COLOR_LOSS else: fg = "#f0f0f0" lbl = tk.Label( row, text=text, bg=bg, fg=fg, font=("Microsoft YaHei UI", 11), anchor="center", ) lbl.grid(row=0, column=i, sticky="nsew", padx=2, pady=5) lbl.bind("", lambda e, t=ticket: self._toggle_ticket_select(t)) if title == "浮盈": self._pos_profit_labels[ticket] = lbl btn_wrap = tk.Frame(row, bg=bg) btn_wrap.grid(row=0, column=len(cells), sticky="nsew", padx=4, pady=3) self._mk_edit_sl_btn(btn_wrap, ticket).pack(side=tk.LEFT, padx=(0, 4)) self._mk_close_btn(btn_wrap, ticket).pack(side=tk.LEFT) row.bind("", lambda e, t=ticket: self._toggle_ticket_select(t)) if not live: tk.Label( self.pos_inner, text="当前无持仓", bg="#2b2b2b", fg="#888888", font=("Microsoft YaHei UI", 11), ).pack(fill=tk.X, pady=12) self._sync_be_button(positions) def _selected_tickets(self) -> list: return sorted(self._selected_tickets_set) def _close_ticket_now(self, ticket: int) -> None: def work(): self.bridge.connect() return self.bridge.close_ticket(ticket) self._run_async(work) def _edit_sl_now(self, ticket: int) -> None: """弹窗输入实际止损价，与保本逻辑无关。""" try: self.bridge.connect() p = self.bridge.get_position(int(ticket)) digits = int(self.bridge.symbol_info(str(p.symbol)).digits) open_price = float(p.price_open) cur_sl = float(p.sl) side = "多" if int(p.type) == mt5.POSITION_TYPE_BUY else "空" cur_txt = f"{cur_sl:.{digits}f}" if cur_sl > 0 else "无" prompt = ( f"单号 #{ticket}（{side}）\n" f"开仓价 {open_price:.{digits}f}\n" f"当前止损 {cur_txt}\n\n" f"请输入新的止损价（与保本无关，按实际价格设置）：" ) initial = cur_sl if cur_sl > 0 else open_price new_sl = simpledialog.askfloat( "改止损", prompt, initialvalue=round(initial, digits), minvalue=0.01, parent=self, ) if new_sl is None: return except TradeError as exc: self.set_status(str(exc), ok=False) return except Exception as exc: # noqa: BLE001 self.set_status(f"改止损异常：{exc}", ok=False) return def work(): self.bridge.connect() return self.bridge.set_stop_loss(int(ticket), float(new_sl)) def after_ok(msg: str) -> None: self.set_status(msg, ok=True) self._pos_sig = None try: symbol = self._symbol() d = int(self.bridge.symbol_info(symbol).digits) self._refresh_positions(symbol, d) except Exception: pass def runner(): try: msg, report = work() self.after(0, lambda m=msg: after_ok(m)) self.after(0, lambda r=report: self._hub_report(r, with_bars=False)) except TradeError as exc: self.after(0, lambda: self.set_status(str(exc), ok=False)) except Exception as exc: # noqa: BLE001 self.after(0, lambda: self.set_status(f"改止损异常：{exc}", ok=False)) finally: self.after(0, self._done) if self._busy: self.set_status("上一次操作还未完成", ok=False) return self._busy = True threading.Thread(target=runner, daemon=True).start() def on_buy(self) -> None: self._order(mt5.ORDER_TYPE_BUY, "买入") def on_sell(self) -> None: self._order(mt5.ORDER_TYPE_SELL, "卖出") def _order(self, order_type: int, side: str) -> None: symbol = self._symbol() def work(): self.bridge.connect() lot = self._parse_float(self.var_lot.get(), "手数") if lot <= 0: raise TradeError("手数必须大于 0") self.bridge.check_max_positions(symbol) self.bridge.check_max_addon(symbol, order_type, lot) sl_usd, tp_usd, atr, sl_m, tp_m = self._resolve_stops(symbol) return self.bridge.market_order(symbol, order_type, lot, sl_usd, tp_usd) self._run_async(work) def on_close_buy(self) -> None: self._close(mt5.POSITION_TYPE_BUY, "平多") def on_close_sell(self) -> None: self._close(mt5.POSITION_TYPE_SELL, "平空") def on_close_all(self) -> None: self._close(None, "全平") def _close(self, pos_type: Optional[int], title: str) -> None: symbol = self._symbol() def work(): self.bridge.connect() return self.bridge.close_positions(symbol, pos_type) self._run_async(work) def on_break_even(self) -> None: """点击一次：对可保本持仓改止损到开仓价±1$。""" symbol = self._symbol() def work(): self.bridge.connect() positions = self.bridge.positions(symbol) tick = self.bridge.tick(symbol) tickets = self._be_candidate_tickets(positions, tick) selected = self._selected_tickets() if selected: picked = [t for t in tickets if t in selected] tickets = picked if picked else list(selected) if not tickets: raise TradeError(self._be_status_hint(positions, tick)) msg, done = self.bridge.break_even_tickets(tickets, BE_OFFSET_USD) self._be_done_tickets.update(done) return msg def after_ok(msg: str) -> None: self.set_status(msg, ok=True) # 强制重绘持仓行，避免止损列仍显示旧值 self._pos_sig = None try: digits = int(self.bridge.symbol_info(symbol).digits) except Exception: digits = 3 self._refresh_positions(symbol, digits) def runner(): try: msg = work() self.after(0, lambda m=msg: after_ok(m)) except TradeError as exc: self.after(0, lambda: self.set_status(str(exc), ok=False)) except Exception as exc: # noqa: BLE001 self.after(0, lambda: self.set_status(f"保本异常：{exc}", ok=False)) finally: self.after(0, self._done) if self._busy: self.set_status("上一次操作还未完成", ok=False) return self._busy = True threading.Thread(target=runner, daemon=True).start() def _hub_heartbeat(self) -> None: """约每 2 秒 ping 中控，供总览秒级在线状态。""" interval_ms = 2000 if hub_client is None or not is_activated(): self.after(interval_ms, self._hub_heartbeat) return def work(): try: return bool(hub_client.ping()) except Exception: return False def done(ok: bool): if not ok: self.set_status("中控离线：请检查服务器，客户端需保持在线", ok=False) self.after(interval_ms, self._hub_heartbeat) def run(): ok = work() self.after(0, lambda: done(ok)) threading.Thread(target=run, daemon=True).start() def on_open_review(self) -> None: """用本机激活 token 免密打开网页复盘。""" def work() -> None: try: if hub_client is None: raise RuntimeError("未打包中控客户端") url = hub_client.open_review_url("/records") self.after(0, lambda: self._open_review_done(url, None)) except Exception as exc: self.after(0, lambda: self._open_review_done(None, exc)) self.set_status("正在获取复盘免密链接…", ok=True) threading.Thread(target=work, daemon=True).start() def _open_review_done(self, url: Optional[str], exc: Optional[BaseException]) -> None: if exc is not None or not url: msg = str(exc) if exc else "无法打开复盘" self.set_status(msg, ok=False) return try: webbrowser.open(url) self.set_status("已免密打开复盘（本机 token）", ok=True) except Exception as open_exc: self.set_status(f"无法打开浏览器：{open_exc}", ok=False) def destroy(self) -> None: try: self.bridge.shutdown() finally: super().destroy() def main() -> None: enable_dpi_awareness() # 激活窗自管可见性（勿对隐藏根窗挂 Toplevel，Windows 上会双击无界面） if ensure_activated is not None: if not ensure_activated(): return app = OrderApp() if is_activated(): try: dev = load_device() app.set_status( f"已连接中控 {dev.get('server')} · 账户#{dev.get('account_id')} 机器#{dev.get('machine_id')}", ok=True, ) except Exception: pass app.mainloop() if __name__ == "__main__": try: main() except Exception as exc: try: import traceback from pathlib import Path log = Path.home() / ".gold_monitor" / "crash.log" log.parent.mkdir(parents=True, exist_ok=True) log.write_text(traceback.format_exc(), encoding="utf-8") except Exception: pass try: root = tk.Tk() root.withdraw() messagebox.showerror("黄金市价辅助", f"启动失败：{exc}\n详情见 %USERPROFILE%\\.gold_monitor\\crash.log") root.destroy() except Exception: pass raise