""" RSI突破回调自动交易系统 - 完整版 EURUSD/GBPUSD/USDJPY/AUDUSD M15周期 G4最优参数配置 使用方法： 1. 安装依赖: pip install MetaTrader5 pandas numpy pytz 2. 确保MT5已登录账户 3. 运行: python rsi_breakout_system.py """ import MetaTrader5 as mt5 import pandas as pd import numpy as np from datetime import datetime, timedelta import pytz import time import threading from typing import Optional, Dict, List, Tuple from dataclasses import dataclass, field from enum import Enum from collections import deque import logging import sys import tkinter as tk from tkinter import ttk, scrolledtext # ==================== 配置参数 ==================== # 交易品种 SYMBOLS = ['EURUSD', 'GBPUSD', 'USDJPY', 'AUDUSD'] # MT5配置 MT5_LOGIN = 0 # 0表示使用已登录账户 MT5_PASSWORD = "" MT5_SERVER = "" # 策略参数 (G4最优) STRATEGY_CONFIG = { 'rsi_period': 14, 'rsi_long': 50, 'rsi_short': 65, 'atr_period': 14, 'sl_atr': 2.0, 'tp_atr': 2.5, 'ts_atr': 0.8, 'max_bars': 24, 'ema_period': 390, 'use_ema_filter': True } # 交易时间 (UTC+8) TRADE_HOURS = {'start': 7, 'end': 21, 'eod': 22} # 风控配置 RISK_CONFIG = { 'fixed_lot': 0.1, 'max_dir_daily': 1, 'max_daily_total': 3, 'magic_number': 20260413, 'deviation': 20, 'min_spread': 30 } # 数据配置 DATA_CONFIG = { 'timeframe': 'M15', 'bars_to_fetch': 100, 'update_interval': 1 } TIMEFRAME_MAP = { 'M1': mt5.TIMEFRAME_M1, 'M5': mt5.TIMEFRAME_M5, 'M15': mt5.TIMEFRAME_M15, 'M30': mt5.TIMEFRAME_M30, 'H1': mt5.TIMEFRAME_H1, 'H4': mt5.TIMEFRAME_H4, 'D1': mt5.TIMEFRAME_D1, } # ==================== 枚举和数据结构 ==================== class SignalType(Enum): NONE = 0 BUY = 1 SELL = -1 @dataclass class StrategyState: """策略状态""" symbol: str long_seen: bool = False short_seen: bool = False signal_bars: int = 0 pending_signal: SignalType = SignalType.NONE position_type: SignalType = SignalType.NONE entry_price: float = 0.0 entry_bar_time: datetime = None bars_held: int = 0 sl: float = 0.0 tp: float = 0.0 ts_trigger: float = 0.0 today_key: datetime = None buy_count: int = 0 sell_count: int = 0 total_count: int = 0 last_bar_time: datetime = None last_rsi: float = 0.0 last_atr: float = 0.0 last_close: float = 0.0 last_ema: float = 0.0 # ==================== 日志系统 ==================== class GUILogHandler(logging.Handler): def __init__(self, max_lines=1000): super().__init__() self.max_lines = max_lines self.logs = deque(maxlen=max_lines) self.callbacks = [] def emit(self, record): try: msg = self.format(record) timestamp = datetime.now().strftime('%H:%M:%S') formatted = f"[{timestamp}] [{record.levelname}] {msg}" self.logs.append(formatted) for callback in self.callbacks: callback(formatted) except Exception: self.handleError(record) def add_callback(self, callback): self.callbacks.append(callback) def get_logs(self): return list(self.logs) class StrategyLogger: def __init__(self, name="RSI_Strategy"): self.logger = logging.getLogger(name) self.logger.setLevel(logging.DEBUG) self.gui_handler = GUILogHandler() self.gui_handler.setLevel(logging.DEBUG) self.gui_handler.setFormatter(logging.Formatter('%(message)s')) self.logger.addHandler(self.gui_handler) console = logging.StreamHandler(sys.stdout) console.setLevel(logging.INFO) console.setFormatter(logging.Formatter('[%(asctime)s] [%(levelname)s] %(message)s', '%H:%M:%S')) self.logger.addHandler(console) file = logging.FileHandler(f"strategy_{datetime.now().strftime('%Y%m%d')}.log", encoding='utf-8') file.setLevel(logging.DEBUG) file.setFormatter(logging.Formatter('[%(asctime)s] [%(levelname)s] %(message)s', '%Y-%m-%d %H:%M:%S')) self.logger.addHandler(file) def debug(self, msg): self.logger.debug(msg) def info(self, msg): self.logger.info(msg) def warning(self, msg): self.logger.warning(msg) def error(self, msg): self.logger.error(msg) def get_logs(self): return self.gui_handler.get_logs() def add_gui_callback(self, callback): self.gui_handler.add_callback(callback) # ==================== MT5连接器 ==================== class MT5Connector: def __init__(self): self.connected = False self.timezone = pytz.timezone('Asia/Shanghai') def initialize(self) -> bool: if not mt5.initialize(): print(f"MT5初始化失败: {mt5.last_error()}") return False if MT5_LOGIN > 0: if not mt5.login(MT5_LOGIN, MT5_PASSWORD, MT5_SERVER): print(f"MT5登录失败: {mt5.last_error()}") mt5.shutdown() return False self.connected = True return True def shutdown(self): if self.connected: mt5.shutdown() self.connected = False def get_rates(self, symbol: str, timeframe: str, count: int) -> Optional[pd.DataFrame]: if not self.connected: return None tf = TIMEFRAME_MAP.get(timeframe, mt5.TIMEFRAME_M15) rates = mt5.copy_rates_from_pos(symbol, tf, 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') df.set_index('time', inplace=True) return df def get_tick(self, symbol: str) -> Optional[Dict]: if not self.connected: return None tick = mt5.symbol_info_tick(symbol) if tick is None: return None return { 'bid': tick.bid, 'ask': tick.ask, 'spread': (tick.ask - tick.bid) * 10**5, 'time': datetime.fromtimestamp(tick.time) } def get_symbol_info(self, symbol: str) -> Optional[Dict]: if not self.connected: return None info = mt5.symbol_info(symbol) if info is None: return None return { 'digits': info.digits, 'point': info.point, 'spread': info.spread, 'volume_min': info.volume_min } def calculate_rsi(self, closes: np.ndarray, period: int = 14) -> np.ndarray: if len(closes) < period + 1: return np.full(len(closes), np.nan) deltas = np.diff(closes) seed = deltas[:period] up = seed[seed >= 0].sum() / period down = -seed[seed < 0].sum() / period rsi = np.full(len(closes), np.nan) rsi[period] = 100 - 100 / (1 + up/down if down != 0 else 0) for i in range(period + 1, len(closes)): delta = deltas[i - 1] if delta > 0: up = (up * (period - 1) + delta) / period down = down * (period - 1) / period else: up = up * (period - 1) / period down = (down * (period - 1) - delta) / period rsi[i] = 100 - 100 / (1 + up/down if down != 0 else 0) return rsi def calculate_atr(self, high: np.ndarray, low: np.ndarray, close: np.ndarray, period: int = 14) -> np.ndarray: if len(high) < period + 1: return np.full(len(high), np.nan) tr = np.zeros(len(high)) tr[0] = high[0] - low[0] for i in range(1, len(high)): tr[i] = max(high[i] - low[i], abs(high[i] - close[i-1]), abs(low[i] - close[i-1])) atr = np.full(len(high), np.nan) atr[period] = np.mean(tr[1:period+1]) for i in range(period + 1, len(high)): atr[i] = (atr[i-1] * (period - 1) + tr[i]) / period return atr def calculate_ema(self, data: np.ndarray, period: int) -> np.ndarray: if len(data) < period: return np.full(len(data), np.nan) ema = np.full(len(data), np.nan) ema[period-1] = np.mean(data[:period]) mult = 2 / (period + 1) for i in range(period, len(data)): ema[i] = (data[i] - ema[i-1]) * mult + ema[i-1] return ema def place_order(self, symbol: str, order_type: str, volume: float, price: float, sl: float, tp: float, magic: int) -> Optional[int]: if not self.connected: return None request = { "action": mt5.TRADE_ACTION_DEAL, "symbol": symbol, "volume": volume, "type": mt5.ORDER_TYPE_BUY if order_type == 'buy' else mt5.ORDER_TYPE_SELL, "price": price, "sl": sl, "tp": tp, "deviation": RISK_CONFIG['deviation'], "magic": magic, "comment": f"RSI_{order_type.upper()}", "type_time": mt5.ORDER_TIME_GTC, "type_filling": mt5.ORDER_FILLING_IOC, } result = mt5.order_send(request) return result.order if result.retcode == mt5.TRADE_RETCODE_DONE else None def modify_position(self, ticket: int, sl: float, tp: float) -> bool: request = {"action": mt5.TRADE_ACTION_SLTP, "position": ticket, "sl": sl, "tp": tp} return mt5.order_send(request).retcode == mt5.TRADE_RETCODE_DONE def close_position(self, ticket: int) -> bool: position = mt5.positions_get(ticket=ticket) if not position: return False pos = position[0] tick = mt5.symbol_info_tick(pos.symbol) request = { "action": mt5.TRADE_ACTION_DEAL, "symbol": pos.symbol, "volume": pos.volume, "type": mt5.ORDER_TYPE_SELL if pos.type == mt5.POSITION_TYPE_BUY else mt5.ORDER_TYPE_BUY, "position": ticket, "price": tick.bid if pos.type == mt5.POSITION_TYPE_BUY else tick.ask, "deviation": RISK_CONFIG['deviation'], "magic": pos.magic, "comment": "Close", "type_time": mt5.ORDER_TIME_GTC, "type_filling": mt5.ORDER_FILLING_IOC, } return mt5.order_send(request).retcode == mt5.TRADE_RETCODE_DONE def get_positions(self, symbol: str = None, magic: int = None) -> List[Dict]: positions = mt5.positions_get(symbol=symbol) if symbol else mt5.positions_get() if not positions: return [] result = [] for pos in positions: if magic and pos.magic != magic: continue result.append({ 'ticket': pos.ticket, 'symbol': pos.symbol, 'type': 'buy' if pos.type == mt5.POSITION_TYPE_BUY else 'sell', 'volume': pos.volume, 'open_price': pos.price_open, 'sl': pos.sl, 'tp': pos.tp, 'profit': pos.profit, 'open_time': datetime.fromtimestamp(pos.time), 'magic': pos.magic }) return result def get_utc8_hour(self) -> int: return datetime.now(self.timezone).hour def in_trade_window(self) -> bool: hour = self.get_utc8_hour() return TRADE_HOURS['start'] <= hour <= TRADE_HOURS['end'] # ==================== 策略逻辑 ==================== class RSIBreakoutStrategy: def __init__(self, mt5_connector: MT5Connector, logger: StrategyLogger): self.mt5 = mt5_connector self.logger = logger self.states = {s: StrategyState(symbol=s) for s in SYMBOLS} def update_daily_stats(self, state: StrategyState): today = datetime.now(self.mt5.timezone).date() today_dt = datetime(today.year, today.month, today.day) if state.today_key != today_dt: state.today_key = today_dt state.buy_count = state.sell_count = state.total_count = 0 def can_trade(self, state: StrategyState, direction: SignalType) -> bool: self.update_daily_stats(state) if state.total_count >= RISK_CONFIG['max_daily_total']: return False if direction == SignalType.BUY and state.buy_count >= RISK_CONFIG['max_dir_daily']: return False if direction == SignalType.SELL and state.sell_count >= RISK_CONFIG['max_dir_daily']: return False return True def check_existing_position(self, symbol: str) -> bool: return len(self.mt5.get_positions(symbol=symbol, magic=RISK_CONFIG['magic_number'])) > 0 def analyze(self, symbol: str, df: pd.DataFrame) -> Tuple[Optional[Dict], Dict]: state = self.states[symbol] detailed = {} if df is None or len(df) < 50: return None, detailed closes = df['close'].values highs = df['high'].values lows = df['low'].values current_close = closes[-1] current_bar_time = df.index[-1] is_new_bar = state.last_bar_time != current_bar_time if is_new_bar: state.last_bar_time = current_bar_time rsi = self.mt5.calculate_rsi(closes, STRATEGY_CONFIG['rsi_period']) atr = self.mt5.calculate_atr(highs, lows, closes, STRATEGY_CONFIG['atr_period']) current_rsi = rsi[-1] current_atr = atr[-1] prev_rsi = rsi[-2] if len(rsi) > 1 else current_rsi prev_atr = atr[-2] if len(atr) > 1 else current_atr state.last_rsi = current_rsi state.last_atr = current_atr state.last_close = current_close trend_bull = trend_bear = True if STRATEGY_CONFIG['use_ema_filter']: daily_df = self.mt5.get_rates(symbol, 'D1', 50) if daily_df is not None and len(daily_df) > 0: ema = self.mt5.calculate_ema(daily_df['close'].values, STRATEGY_CONFIG['ema_period']) state.last_ema = ema[-1] trend_bull = current_close > state.last_ema trend_bear = current_close < state.last_ema detailed = { 'symbol': symbol, 'time': current_bar_time, 'close': current_close, 'rsi': current_rsi, 'prev_rsi': prev_rsi, 'atr': current_atr, 'ema': state.last_ema, 'trend_bull': trend_bull, 'trend_bear': trend_bear, 'long_seen': state.long_seen, 'short_seen': state.short_seen, 'is_new_bar': is_new_bar } # 已有持仓时检查平仓 if state.position_type != SignalType.NONE: self._check_exit(state, df, detailed) return None, detailed # 检查交易时间和现有持仓 if not self.mt5.in_trade_window() or self.check_existing_position(symbol): return None, detailed # 新K线时更新信号状态 if is_new_bar: state.signal_bars += 1 if state.signal_bars > 20: state.long_seen = state.short_seen = False state.signal_bars = 0 if prev_rsi < STRATEGY_CONFIG['rsi_long'] and trend_bull: state.long_seen = True self.logger.info(f"{symbol} RSI超跌: {prev_rsi:.2f} < {STRATEGY_CONFIG['rsi_long']}") if prev_rsi > STRATEGY_CONFIG['rsi_short'] and trend_bear: state.short_seen = True self.logger.info(f"{symbol} RSI超涨: {prev_rsi:.2f} > {STRATEGY_CONFIG['rsi_short']}") # 检查入场信号 signal = None tick = self.mt5.get_tick(symbol) spread_ok = tick and tick['spread'] <= RISK_CONFIG['min_spread'] if state.long_seen and self.can_trade(state, SignalType.BUY) and spread_ok: if prev_rsi >= STRATEGY_CONFIG['rsi_long']: entry = current_close if tick else current_close signal = { 'symbol': symbol, 'type': 'buy', 'price': entry, 'sl': entry - prev_atr * STRATEGY_CONFIG['sl_atr'], 'tp': entry + prev_atr * STRATEGY_CONFIG['tp_atr'], 'volume': RISK_CONFIG['fixed_lot'], 'atr': prev_atr, 'rsi': prev_rsi } state.long_seen = False state.signal_bars = 0 self.logger.info(f"{symbol} BUY信号: RSI={prev_rsi:.2f}, 入场={entry:.5f}") elif state.short_seen and self.can_trade(state, SignalType.SELL) and spread_ok: if prev_rsi <= STRATEGY_CONFIG['rsi_short']: entry = current_close if tick else current_close signal = { 'symbol': symbol, 'type': 'sell', 'price': entry, 'sl': entry + prev_atr * STRATEGY_CONFIG['sl_atr'], 'tp': entry - prev_atr * STRATEGY_CONFIG['tp_atr'], 'volume': RISK_CONFIG['fixed_lot'], 'atr': prev_atr, 'rsi': prev_rsi } state.short_seen = False state.signal_bars = 0 self.logger.info(f"{symbol} SELL信号: RSI={prev_rsi:.2f}, 入场={entry:.5f}") return signal, detailed def _check_exit(self, state: StrategyState, df: pd.DataFrame, data: Dict): state.bars_held += 1 close, high, low = df['close'].values[-1], df['high'].values[-1], df['low'].values[-1] atr = data['atr'] should_close = False reason = "" if state.position_type == SignalType.BUY: if low <= state.sl: should_close, reason = True, "止损" elif close >= state.tp: should_close, reason = True, "止盈" elif close > state.tp - atr * STRATEGY_CONFIG['ts_atr']: new_sl = close - atr * STRATEGY_CONFIG['ts_atr'] if new_sl > state.ts_trigger: state.ts_trigger = new_sl state.sl = new_sl data['ts_updated'] = new_sl else: if high >= state.sl: should_close, reason = True, "止损" elif close <= state.tp: should_close, reason = True, "止盈" elif close < state.tp + atr * STRATEGY_CONFIG['ts_atr']: new_sl = close + atr * STRATEGY_CONFIG['ts_atr'] if new_sl < state.ts_trigger or state.ts_trigger == 0: state.ts_trigger = new_sl state.sl = new_sl data['ts_updated'] = new_sl if not should_close and state.bars_held >= 2: if self.mt5.get_utc8_hour() >= TRADE_HOURS['eod']: should_close, reason = True, "尾盘强平" if not should_close and state.bars_held >= STRATEGY_CONFIG['max_bars']: should_close, reason = True, "超时强平" data['should_close'] = should_close data['close_reason'] = reason def on_order_filled(self, symbol: str, signal: Dict): state = self.states[symbol] state.position_type = SignalType.BUY if signal['type'] == 'buy' else SignalType.SELL state.entry_price = signal['price'] state.entry_bar_time = datetime.now(self.mt5.timezone) state.bars_held = 0 state.sl = signal['sl'] state.tp = signal['tp'] state.ts_trigger = signal['sl'] if state.position_type == SignalType.BUY: state.buy_count += 1 else: state.sell_count += 1 state.total_count += 1 def on_position_closed(self, symbol: str): state = self.states[symbol] state.position_type = SignalType.NONE state.entry_price = state.bars_held = 0 state.sl = state.tp = state.ts_trigger = 0 # ==================== 交易管理器 ==================== class TradeManager: def __init__(self, mt5_connector: MT5Connector, strategy: RSIBreakoutStrategy, logger: StrategyLogger): self.mt5 = mt5_connector self.strategy = strategy self.logger = logger self.running = False self.trade_history = [] def start(self): self.running = True self.logger.info("交易管理器已启动") def stop(self): self.running = False self.logger.info("交易管理器已停止") def execute_signal(self, signal: Dict) -> Optional[int]: if not self.running: return None symbol = signal['symbol'] # 检查同方向持仓 for pos in self.mt5.get_positions(symbol=symbol, magic=RISK_CONFIG['magic_number']): if (signal['type'] == 'buy' and pos['type'] == 'buy') or \ (signal['type'] == 'sell' and pos['type'] == 'sell'): self.logger.warning(f"{symbol} 已有同方向持仓，跳过") return None ticket = self.mt5.place_order( symbol, signal['type'], signal['volume'], signal['price'], signal['sl'], signal['tp'], RISK_CONFIG['magic_number'] ) if ticket: self.logger.info(f"{symbol} 订单成交, Ticket: {ticket}") self.strategy.on_order_filled(symbol, signal) self.trade_history.append({ 'time': datetime.now(), 'symbol': symbol, 'type': signal['type'], 'price': signal['price'], 'volume': signal['volume'], 'ticket': ticket, 'rsi': signal.get('rsi', 0), 'atr': signal.get('atr', 0) }) return ticket def manage_positions(self): if not self.running: return for symbol in SYMBOLS: state = self.strategy.states[symbol] for pos in self.mt5.get_positions(symbol=symbol, magic=RISK_CONFIG['magic_number']): if state.position_type != SignalType.NONE: tick = self.mt5.get_tick(symbol) if tick: price = tick['bid'] if pos['type'] == 'buy' else tick['ask'] atr = state.last_atr if pos['type'] == 'buy': if price > state.tp - atr * STRATEGY_CONFIG['ts_atr']: new_sl = price - atr * STRATEGY_CONFIG['ts_atr'] if new_sl > pos['sl']: self.mt5.modify_position(pos['ticket'], new_sl, pos['tp']) state.sl = state.ts_trigger = new_sl self.logger.info(f"{symbol} 移动止损: SL={new_sl:.5f}") else: if price < state.tp + atr * STRATEGY_CONFIG['ts_atr']: new_sl = price + atr * STRATEGY_CONFIG['ts_atr'] if new_sl < pos['sl'] or pos['sl'] == 0: self.mt5.modify_position(pos['ticket'], new_sl, pos['tp']) state.sl = state.ts_trigger = new_sl self.logger.info(f"{symbol} 移动止损: SL={new_sl:.5f}") # 检查平仓 df = self.mt5.get_rates(symbol, DATA_CONFIG['timeframe'], 50) if df is not None: data = {'atr': state.last_atr, 'should_close': False, 'close_reason': ''} self.strategy._check_exit(state, df, data) if data['should_close']: self.mt5.close_position(pos['ticket']) self.logger.info(f"{symbol} 平仓: {data['close_reason']}, Ticket={pos['ticket']}") self.strategy.on_position_closed(symbol) def get_positions_summary(self) -> List[Dict]: summary = [] for symbol in SYMBOLS: state = self.strategy.states[symbol] for pos in self.mt5.get_positions(symbol=symbol, magic=RISK_CONFIG['magic_number']): tick = self.mt5.get_tick(symbol) summary.append({ 'ticket': pos['ticket'], 'symbol': pos['symbol'], 'type': pos['type'], 'volume': pos['volume'], 'open_price': pos['open_price'], 'current_price': tick['bid'] if pos['type'] == 'buy' else tick['ask'] if tick else 0, 'sl': pos['sl'], 'tp': pos['tp'], 'profit': pos['profit'], 'bars_held': state.bars_held, 'open_time': pos['open_time'] }) return summary # ==================== GUI界面 ==================== class TradingGUI: def __init__(self, system): self.system = system self.root = tk.Tk() self.root.title("RSI突破回调交易系统 G4") self.root.geometry("1400x800") self.countdown_var = tk.StringVar(value="等待启动...") self.status_var = tk.StringVar(value="就绪") self._create_widgets() self.system.logger.add_gui_callback(self._on_log) def _create_widgets(self): main = ttk.Frame(self.root, padding="5") main.grid(row=0, column=0, sticky="nsew") self.root.columnconfigure(0, weight=1) self.root.rowconfigure(0, weight=1) main.columnconfigure(0, weight=3) main.columnconfigure(1, weight=2) main.rowconfigure(0, weight=1) main.rowconfigure(1, weight=2) # 信号面板 sig_frame = ttk.LabelFrame(main, text="信号状态", padding="5") sig_frame.grid(row=0, column=0, sticky="nsew", padx=(0,5)) cols = ('品种', 'RSI', 'ATR', 'EMA', '超跌', '超涨', '状态') self.signal_tree = ttk.Treeview(sig_frame, columns=cols, show='headings', height=6) for c in cols: self.signal_tree.heading(c, text=c) self.signal_tree.column(c, width=100) self.signal_tree.pack(fill="both", expand=True) # 持仓面板 pos_frame = ttk.LabelFrame(main, text="当前持仓", padding="5") pos_frame.grid(row=0, column=1, sticky="nsew") pcols = ('品种', '方向', '手数', '入场价', '当前价', '止损', '止盈', '盈亏', 'K线') self.pos_tree = ttk.Treeview(pos_frame, columns=pcols, show='headings', height=6) for c in pcols: self.pos_tree.heading(c, text=c) self.pos_tree.column(c, width=85) self.pos_tree.pack(side="left", fill="both", expand=True) ttk.Scrollbar(pos_frame, orient="vertical", command=self.pos_tree.yview).pack(side="right", fill="y") # 历史面板 hist_frame = ttk.LabelFrame(main, text="交易历史", padding="5") hist_frame.grid(row=1, column=0, sticky="nsew", padx=(0,5), pady=(5,0)) hcols = ('时间', '品种', '方向', '入场价', '手数', 'RSI', 'ATR') self.hist_tree = ttk.Treeview(hist_frame, columns=hcols, show='headings', height=10) for c in hcols: self.hist_tree.heading(c, text=c) self.hist_tree.column(c, width=100) self.hist_tree.pack(side="left", fill="both", expand=True) ttk.Scrollbar(hist_frame, orient="vertical", command=self.hist_tree.yview).pack(side="right", fill="y") # 日志面板 log_frame = ttk.LabelFrame(main, text="系统日志", padding="5") log_frame.grid(row=1, column=1, sticky="nsew", pady=(5,0)) self.log_text = scrolledtext.ScrolledText(log_frame, height=12, font=('Consolas', 9)) self.log_text.pack(fill="both", expand=True) # 控制栏 ctrl = ttk.Frame(main) ctrl.grid(row=2, column=0, columnspan=2, sticky="ew", pady=(5,0)) self.start_btn = ttk.Button(ctrl, text="启动交易", command=self._on_start) self.start_btn.pack(side="left", padx=5) self.stop_btn = ttk.Button(ctrl, text="停止交易", command=self._on_stop, state="disabled") self.stop_btn.pack(side="left", padx=5) ttk.Label(ctrl, text="下次检测:").pack(side="left", padx=(20,5)) ttk.Label(ctrl, textvariable=self.countdown_var, font=('Arial', 10, 'bold')).pack(side="left") ttk.Label(ctrl, text="状态:").pack(side="left", padx=(20,5)) ttk.Label(ctrl, textvariable=self.status_var).pack(side="left") ttk.Button(ctrl, text="清空日志", command=self._clear_log).pack(side="right", padx=5) ttk.Button(ctrl, text="全部平仓", command=self._close_all).pack(side="right", padx=5) def _on_log(self, msg): self.log_text.insert("end", msg + '\n') self.log_text.see("end") def _on_start(self): self.system.start() self.start_btn.config(state="disabled") self.stop_btn.config(state="normal") self.status_var.set("运行中") self._countdown() def _on_stop(self): self.system.stop() self.start_btn.config(state="normal") self.stop_btn.config(state="disabled") self.status_var.set("已停止") def _clear_log(self): self.log_text.delete(1.0, "end") def _close_all(self): self.system.close_all_positions() def _countdown(self): if self.system.running: next_update = self.system.last_update_time + timedelta(seconds=DATA_CONFIG['update_interval']) remaining = (next_update - datetime.now()).total_seconds() self.countdown_var.set(f"{remaining:.1f}秒" if remaining > 0 else "更新中...") self.root.after(100, self._countdown) def _get_status(self, state): if state.position_type.value == 1: return "持有多仓" elif state.position_type.value == -1: return "持有空仓" elif state.long_seen: return "等待BUY" elif state.short_seen: return "等待SELL" return "等待信号" def refresh(self): # 信号 for item in self.signal_tree.get_children(): self.signal_tree.delete(item) for s, state in self.system.strategy.states.items(): self.signal_tree.insert('', 'end', values=( s, f"{state.last_rsi:.2f}", f"{state.last_atr:.5f}", f"{state.last_ema:.5f}" if state.last_ema else "-", "是" if state.long_seen else "否", "是" if state.short_seen else "否", self._get_status(state) )) # 持仓 for item in self.pos_tree.get_children(): self.pos_tree.delete(item) for pos in self.system.trade_manager.get_positions_summary(): item = self.pos_tree.insert('', 'end', values=( pos['symbol'], '买入' if pos['type']=='buy' else '卖出', f"{pos['volume']:.2f}", f"{pos['open_price']:.5f}", f"{pos['current_price']:.5f}", f"{pos['sl']:.5f}" if pos['sl'] else "-", f"{pos['tp']:.5f}" if pos['tp'] else "-", f"{pos['profit']:.2f}", pos['bars_held'] )) tag = 'profit' if pos['profit'] >= 0 else 'loss' self.pos_tree.tag_configure('profit', foreground='green') self.pos_tree.tag_configure('loss', foreground='red') self.pos_tree.item(item, tags=(tag,)) # 历史 for item in self.hist_tree.get_children(): self.hist_tree.delete(item) for trade in self.system.trade_manager.trade_history[-20:]: self.hist_tree.insert('', 'end', values=( trade['time'].strftime('%H:%M:%S'), trade['symbol'], '买入' if trade['type']=='buy' else '卖出', f"{trade['price']:.5f}", f"{trade['volume']:.2f}", f"{trade.get('rsi',0):.2f}", f"{trade.get('atr',0):.5f}" )) def run(self): def periodic(): self.refresh() self.root.after(1000, periodic) self.root.after(1000, periodic) self.root.mainloop() # ==================== 主系统 ==================== class TradingSystem: def __init__(self): self.logger = StrategyLogger("RSI_System") self.mt5 = MT5Connector() self.strategy = RSIBreakoutStrategy(self.mt5, self.logger) self.trade_manager = TradeManager(self.mt5, self.strategy, self.logger) self.running = False self.thread = None self.last_update_time = datetime.now() self.gui = None def initialize(self) -> bool: self.logger.info("初始化交易系统...") if not self.mt5.initialize(): self.logger.error("MT5连接失败") return False acc = mt5.account_info() if acc: self.logger.info(f"账户: {acc.login}, 余额: {acc.balance:.2f}") for s in SYMBOLS: info = self.mt5.get_symbol_info(s) if info: self.logger.info(f"{s}: 小数位={info['digits']}, 点差={info['spread']}") self.logger.info("初始化完成") return True def start(self): if self.running: return self.running = True self.trade_manager.start() self.thread = threading.Thread(target=self._loop, daemon=True) self.thread.start() self.logger.info("交易系统已启动") def stop(self): self.running = False self.trade_manager.stop() if self.thread: self.thread.join(timeout=5) self.logger.info("交易系统已停止") def _loop(self): while self.running: try: self.last_update_time = datetime.now() for symbol in SYMBOLS: df = self.mt5.get_rates(symbol, DATA_CONFIG['timeframe'], DATA_CONFIG['bars_to_fetch']) if df is not None: signal, data = self.strategy.analyze(symbol, df) if data.get('is_new_bar'): self.logger.debug( f"{symbol} Close={data.get('close',0):.5f} " f"RSI={data.get('rsi',0):.2f} ATR={data.get('atr',0):.5f}" ) if signal: self.trade_manager.execute_signal(signal) self.trade_manager.manage_positions() except Exception as e: self.logger.error(f"主循环异常: {e}") time.sleep(DATA_CONFIG['update_interval']) def close_all_positions(self): for s in SYMBOLS: for p in self.mt5.get_positions(symbol=s, magic=RISK_CONFIG['magic_number']): self.mt5.close_position(p['ticket']) self.logger.info(f"手动平仓: {s} Ticket={p['ticket']}") def shutdown(self): self.stop() self.mt5.shutdown() self.logger.info("系统已关闭") def run_gui(self): self.gui = TradingGUI(self) self.gui.run() def main(): system = TradingSystem() if not system.initialize(): return try: system.run_gui() finally: system.shutdown() if __name__ == '__main__': main()