#!/usr/bin/env python3 """ đŸ–Ĩī¸ Trading Bot GUI - Tkinter Application Professional Desktop App fÃŧr Trading Bot V1.9 """ import tkinter as tk from tkinter import ttk, scrolledtext, messagebox import MetaTrader5 as mt5 import threading import queue from datetime import datetime import logging from apscheduler.schedulers.background import BackgroundScheduler # MT5 connection config — edit here instead of in code MT5_LOGIN = 10800246 MT5_SERVER = 'VantageInternational-Demo' # Trading Bot Imports from infrastructure_patch import TradingInfrastructure, create_scheduled_reports from position_monitor import PositionMonitor from session_filter_patch import ( create_session_filtered_check, SESSION_WHITELIST_CONFIG, is_session_allowed ) from drawdown_protection import create_protected_trading_check, print_protection_status from adaptive_rhythm_manager import AdaptiveRhythmManager from execute_trade import execute_trade_v2_adaptive # Setup Logging logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s' ) logger = logging.getLogger(__name__) class TradingBotGUI: """ Main Trading Bot GUI Application """ def __init__(self, root): self.root = root self.root.title("🤖 Trading Bot V1.9 - Control Center") self.root.geometry("1200x800") # Bot State self.bot_running = False self.scheduler = None self.infra = None self.position_monitor = None self.rhythm_manager = None self.drawdown_protection = None self.trading_check = None # MT5 Config self.symbol = "XAUUSD" self.strategy_name = "TradingBot_V1.9" self.max_positions = 1 # GUI Queue for thread-safe updates self.gui_queue = queue.Queue() # Build GUI self.create_widgets() # Start GUI update loop self.process_queue() # Clean shutdown when window is closed self.root.protocol("WM_DELETE_WINDOW", self.on_closing) def create_widgets(self): """Create all GUI widgets""" # ========================================== # TITLE BAR # ========================================== title_frame = tk.Frame(self.root, bg="#2c3e50", height=60) title_frame.pack(fill=tk.X) title_label = tk.Label( title_frame, text="🤖 Trading Bot V1.9 - Control Center", font=("Arial", 20, "bold"), bg="#2c3e50", fg="white" ) title_label.pack(pady=15) # ========================================== # MAIN CONTAINER # ========================================== main_frame = tk.Frame(self.root) main_frame.pack(fill=tk.BOTH, expand=True, padx=10, pady=10) # Left Panel (Controls) left_frame = tk.Frame(main_frame, width=400) left_frame.pack(side=tk.LEFT, fill=tk.BOTH, padx=(0, 5)) # Right Panel (Logs & Status) right_frame = tk.Frame(main_frame) right_frame.pack(side=tk.RIGHT, fill=tk.BOTH, expand=True, padx=(5, 0)) # ========================================== # LEFT PANEL: CONTROLS # ========================================== # Connection Status self.create_connection_panel(left_frame) # Bot Controls self.create_bot_controls(left_frame) # Session Filter Config self.create_session_config(left_frame) # Drawdown Protection Status self.create_drawdown_status(left_frame) # Manual Controls self.create_manual_controls(left_frame) # ========================================== # RIGHT PANEL: STATUS & LOGS # ========================================== # Trading Status self.create_trading_status(right_frame) # Logs self.create_log_panel(right_frame) def create_connection_panel(self, parent): """MT5 Connection Status Panel""" frame = tk.LabelFrame(parent, text="📡 MT5 Connection", font=("Arial", 10, "bold")) frame.pack(fill=tk.X, pady=5) # Status Label self.connection_status = tk.Label( frame, text="❌ Not Connected", font=("Arial", 12), fg="red" ) self.connection_status.pack(pady=10) # Connect Button self.connect_btn = tk.Button( frame, text="Connect to MT5", command=self.connect_mt5, bg="#3498db", fg="white", font=("Arial", 10, "bold"), cursor="hand2" ) self.connect_btn.pack(pady=5) # Account Info self.account_info = tk.Label(frame, text="", font=("Arial", 9)) self.account_info.pack(pady=5) def create_bot_controls(self, parent): """Bot Start/Stop Controls""" frame = tk.LabelFrame(parent, text="🤖 Bot Controls", font=("Arial", 10, "bold")) frame.pack(fill=tk.X, pady=5) # Bot Status self.bot_status_label = tk.Label( frame, text="⏸ī¸ Bot Stopped", font=("Arial", 12, "bold"), fg="orange" ) self.bot_status_label.pack(pady=10) # Control Buttons btn_frame = tk.Frame(frame) btn_frame.pack(pady=5) self.start_btn = tk.Button( btn_frame, text="â–ļī¸ Start Bot", command=self.start_bot, bg="#27ae60", fg="white", font=("Arial", 10, "bold"), width=12, cursor="hand2" ) self.start_btn.pack(side=tk.LEFT, padx=5) self.stop_btn = tk.Button( btn_frame, text="⏚ī¸ Stop Bot", command=self.stop_bot, bg="#e74c3c", fg="white", font=("Arial", 10, "bold"), width=12, cursor="hand2", state=tk.DISABLED ) self.stop_btn.pack(side=tk.LEFT, padx=5) def create_session_config(self, parent): """Session Filter Configuration""" frame = tk.LabelFrame(parent, text="📊 Session Filter", font=("Arial", 10, "bold")) frame.pack(fill=tk.X, pady=5) # Session Checkboxes self.session_vars = {} sessions = ['asian', 'london', 'overlap', 'ny'] for session in sessions: var = tk.BooleanVar(value=SESSION_WHITELIST_CONFIG['enabled_sessions'][session]) self.session_vars[session] = var cb = tk.Checkbutton( frame, text=f"{session.upper()}", variable=var, font=("Arial", 9), command=lambda s=session: self.update_session_config(s) ) cb.pack(anchor=tk.W, padx=20, pady=2) # Confidence Threshold conf_frame = tk.Frame(frame) conf_frame.pack(fill=tk.X, padx=20, pady=5) tk.Label(conf_frame, text="Confidence:", font=("Arial", 9)).pack(side=tk.LEFT) self.confidence_var = tk.IntVar(value=SESSION_WHITELIST_CONFIG['base_confidence']) self.confidence_slider = tk.Scale( conf_frame, from_=50, to=90, orient=tk.HORIZONTAL, variable=self.confidence_var, command=self.update_confidence ) self.confidence_slider.pack(side=tk.LEFT, fill=tk.X, expand=True, padx=5) self.confidence_label = tk.Label( conf_frame, text=f"{self.confidence_var.get()}%", font=("Arial", 9, "bold") ) self.confidence_label.pack(side=tk.LEFT) def create_drawdown_status(self, parent): """Drawdown Protection Status""" frame = tk.LabelFrame(parent, text="🛡ī¸ Drawdown Protection", font=("Arial", 10, "bold")) frame.pack(fill=tk.X, pady=5) self.drawdown_status = tk.Label( frame, text="⏸ī¸ Not Active", font=("Arial", 9), justify=tk.LEFT ) self.drawdown_status.pack(pady=10, padx=10) def create_manual_controls(self, parent): """Manual Trading Controls""" frame = tk.LabelFrame(parent, text="🎮 Manual Controls", font=("Arial", 10, "bold")) frame.pack(fill=tk.X, pady=5) btn_frame = tk.Frame(frame) btn_frame.pack(pady=10) tk.Button( btn_frame, text="📊 Check Status", command=self.check_status, width=15, cursor="hand2" ).pack(pady=2) tk.Button( btn_frame, text="🔄 Check Positions", command=self.check_positions, width=15, cursor="hand2" ).pack(pady=2) tk.Button( btn_frame, text="❌ Close All Positions", command=self.close_all_positions, width=15, cursor="hand2", bg="#e74c3c", fg="white" ).pack(pady=2) def create_trading_status(self, parent): """Trading Status Display""" frame = tk.LabelFrame(parent, text="📈 Trading Status", font=("Arial", 10, "bold")) frame.pack(fill=tk.X, pady=5) # Status Grid status_frame = tk.Frame(frame) status_frame.pack(fill=tk.X, padx=10, pady=10) # Current Session tk.Label(status_frame, text="Session:", font=("Arial", 9, "bold")).grid(row=0, column=0, sticky=tk.W) self.current_session = tk.Label(status_frame, text="-", font=("Arial", 9)) self.current_session.grid(row=0, column=1, sticky=tk.W, padx=10) # Interval tk.Label(status_frame, text="Interval:", font=("Arial", 9, "bold")).grid(row=1, column=0, sticky=tk.W) self.current_interval = tk.Label(status_frame, text="-", font=("Arial", 9)) self.current_interval.grid(row=1, column=1, sticky=tk.W, padx=10) # Positions tk.Label(status_frame, text="Positions:", font=("Arial", 9, "bold")).grid(row=2, column=0, sticky=tk.W) self.positions_count = tk.Label(status_frame, text="-", font=("Arial", 9)) self.positions_count.grid(row=2, column=1, sticky=tk.W, padx=10) # Daily P/L tk.Label(status_frame, text="Daily P/L:", font=("Arial", 9, "bold")).grid(row=3, column=0, sticky=tk.W) self.daily_pl = tk.Label(status_frame, text="-", font=("Arial", 9)) self.daily_pl.grid(row=3, column=1, sticky=tk.W, padx=10) def create_log_panel(self, parent): """Log Output Panel""" frame = tk.LabelFrame(parent, text="📋 Activity Log", font=("Arial", 10, "bold")) frame.pack(fill=tk.BOTH, expand=True, pady=5) # Scrolled Text self.log_text = scrolledtext.ScrolledText( frame, wrap=tk.WORD, font=("Courier", 9), bg="#1e1e1e", fg="#00ff00", insertbackground="white" ) self.log_text.pack(fill=tk.BOTH, expand=True, padx=5, pady=5) # Clear Button tk.Button( frame, text="🗑ī¸ Clear Log", command=lambda: self.log_text.delete(1.0, tk.END), cursor="hand2" ).pack(pady=5) # ========================================== # FUNCTIONALITY # ========================================== def log(self, message, level="INFO"): """Thread-safe logging to GUI""" timestamp = datetime.now().strftime("%H:%M:%S") log_message = f"[{timestamp}] {level}: {message}\n" self.gui_queue.put(("log", log_message)) logger.info(message) def process_queue(self): """Process GUI updates from queue (runs in main thread — thread-safe)""" try: while True: action, data = self.gui_queue.get_nowait() if action == "log": self.log_text.insert(tk.END, data) self.log_text.see(tk.END) elif action == "widget": # data = (widget, config_dict) e.g. (self.connection_status, {"text": "...", "fg": "green"}) widget, kwargs = data widget.config(**kwargs) elif action == "status": self.update_status_display(data) except queue.Empty: pass self.root.after(100, self.process_queue) def _update_widget(self, widget, **kwargs): """Queue a thread-safe widget update""" self.gui_queue.put(("widget", (widget, kwargs))) def connect_mt5(self): """Connect to MT5""" def connect_thread(): try: self.log("Connecting to MT5...") if not mt5.initialize(): self.log("MT5 initialization failed!", "ERROR") return import keyring as kr password = kr.get_password(MT5_SERVER, str(MT5_LOGIN)) if not mt5.login(MT5_LOGIN, password, MT5_SERVER): self.log(f"MT5 login failed: {mt5.last_error()}", "ERROR") return account_info = mt5.account_info() if account_info: self.log(f"✅ Connected — Account: {account_info.login} | Balance: ${account_info.balance:.2f}") self._update_widget(self.connection_status, text="✅ Connected", fg="green") self._update_widget(self.account_info, text=f"Account: {account_info.login} | Balance: ${account_info.balance:.2f}") self._update_widget(self.connect_btn, state=tk.DISABLED) except Exception as e: self.log(f"Connection error: {e}", "ERROR") threading.Thread(target=connect_thread, daemon=True).start() def start_bot(self): """Start Trading Bot""" if self.bot_running: self.log("Bot is already running", "WARNING") return def start_thread(): try: self.log("🚀 Starting Trading Bot...") # Initialize Infrastructure self.log("Initializing infrastructure...") self.infra = TradingInfrastructure( db_path="trading_bot.db", enable_telegram=True, enable_database=True ) # Initialize Position Monitor self.log("Initializing Position Monitor...") self.position_monitor = PositionMonitor(self.infra.db, self.infra.telegram) # Initialize Rhythm Manager self.log("Initializing Adaptive Rhythm Manager...") self.rhythm_manager = AdaptiveRhythmManager(self.symbol) # Create Session-Filtered Trading Check self.log("Setting up Session Filter...") base_trading_check = create_session_filtered_check( rhythm_manager=self.rhythm_manager, execute_func=execute_trade_v2_adaptive, symbol=self.symbol, strategy_name=self.strategy_name, max_positions=self.max_positions, logger=logger, datetime=datetime ) # Add Drawdown Protection self.log("Setting up Drawdown Protection...") self.trading_check = create_protected_trading_check(self.infra, base_trading_check) self.drawdown_protection = self.trading_check.protection # Start Scheduler self.log("Starting Scheduler...") self.scheduler = BackgroundScheduler() # Add Jobs self.scheduler.add_job( func=self.trading_check, trigger='cron', minute='*', id='adaptive_trading_check' ) self.scheduler.add_job( func=self.position_monitor.check_open_positions, trigger='interval', minutes=1, id='position_monitor' ) create_scheduled_reports(self.infra, self.scheduler) self.scheduler.start() self.bot_running = True self.log("✅ Trading Bot Started Successfully!") self._update_widget(self.bot_status_label, text="✅ Bot Running", fg="green") self._update_widget(self.start_btn, state=tk.DISABLED) self._update_widget(self.stop_btn, state=tk.NORMAL) bot_config = { 'version': 'V1.9', 'enabled_sessions': SESSION_WHITELIST_CONFIG['enabled_sessions'], 'base_confidence': SESSION_WHITELIST_CONFIG['base_confidence'] } # Log bot status to DB so bot_status table is populated self.infra.log_bot_status('running', bot_config) # Send Telegram notification if self.infra.telegram: self.infra.send_bot_started(bot_config) except Exception as e: self.log(f"Error starting bot: {e}", "ERROR") threading.Thread(target=start_thread, daemon=True).start() def stop_bot(self): """Stop Trading Bot""" if self.scheduler: self.scheduler.shutdown() self.bot_running = False self.log("🛑 Trading Bot Stopped") if self.infra: self.infra.log_bot_status('stopped') # Update GUI self.bot_status_label.config(text="⏸ī¸ Bot Stopped", fg="orange") self.start_btn.config(state=tk.NORMAL) self.stop_btn.config(state=tk.DISABLED) def update_session_config(self, session): """Update session filter config""" enabled = self.session_vars[session].get() SESSION_WHITELIST_CONFIG['enabled_sessions'][session] = enabled self.log(f"Session {session.upper()}: {'ENABLED' if enabled else 'DISABLED'}") def update_confidence(self, value): """Update confidence threshold""" confidence = int(float(value)) SESSION_WHITELIST_CONFIG['base_confidence'] = confidence self.confidence_label.config(text=f"{confidence}%") self.log(f"Confidence threshold updated: {confidence}%") def check_status(self): """Check current bot status""" def status_thread(): try: if not self.bot_running: self.log("Bot is not running", "WARNING") return self.log("📊 Checking Status...") session = self.rhythm_manager.get_current_session() self._update_widget(self.current_session, text=session.upper()) interval = self.rhythm_manager.calculate_optimal_interval() self._update_widget(self.current_interval, text=f"{interval} min") positions = mt5.positions_get(symbol=self.symbol) count = len(positions) if positions else 0 self._update_widget(self.positions_count, text=f"{count}/{self.max_positions}") # Drawdown Status if self.drawdown_protection: status = self.drawdown_protection.get_status() status_text = f"Trading: {'✅ Allowed' if status['trading_allowed'] else '🛑 Paused'}\n" status_text += f"Daily: ${status['daily_loss']:.2f}/${status['daily_limit']:.2f}\n" status_text += f"Consecutive: {status['consecutive_losses']}/{status['consecutive_limit']}" self._update_widget(self.drawdown_status, text=status_text) self.log("✅ Status updated") except Exception as e: self.log(f"Error checking status: {e}", "ERROR") threading.Thread(target=status_thread, daemon=True).start() def check_positions(self): """Check open positions""" def positions_thread(): try: positions = mt5.positions_get(symbol=self.symbol) if not positions: self.log("No open positions") return self.log(f"📊 {len(positions)} Open Position(s):") for pos in positions: self.log(f" Ticket: {pos.ticket} | {pos.type} | P/L: ${pos.profit:.2f}") except Exception as e: self.log(f"Error checking positions: {e}", "ERROR") threading.Thread(target=positions_thread, daemon=True).start() def close_all_positions(self): """Close all open positions""" confirm = messagebox.askyesno("Confirm Close", "Are you sure you want to close ALL positions?") if not confirm: return def close_thread(): try: positions = mt5.positions_get(symbol=self.symbol) if not positions: self.log("No positions to close") return self.log(f"Closing {len(positions)} position(s)...") # Fetch tick once outside loop tick = mt5.symbol_info_tick(self.symbol) if not tick: self.log("Could not get current tick price", "ERROR") return for pos in positions: close_price = tick.bid if pos.type == 0 else tick.ask close_request = { "action": mt5.TRADE_ACTION_DEAL, "symbol": self.symbol, "volume": pos.volume, "type": mt5.ORDER_TYPE_SELL if pos.type == 0 else mt5.ORDER_TYPE_BUY, "position": pos.ticket, "price": close_price, "deviation": 20, "magic": 234000, "comment": "Manual close from GUI", "type_time": mt5.ORDER_TIME_GTC, "type_filling": mt5.ORDER_FILLING_IOC, } result = mt5.order_send(close_request) if result.retcode == mt5.TRADE_RETCODE_DONE: self.log(f"✅ Closed position {pos.ticket}") else: self.log(f"❌ Failed to close {pos.ticket}: {result.comment}", "ERROR") except Exception as e: self.log(f"Error closing positions: {e}", "ERROR") threading.Thread(target=close_thread, daemon=True).start() def update_status_display(self, data): """Update status display from data""" pass def on_closing(self): """Shutdown scheduler and MT5 before closing""" if self.scheduler and self.scheduler.running: self.scheduler.shutdown(wait=False) mt5.shutdown() self.root.destroy() def main(): """Main Entry Point""" root = tk.Tk() app = TradingBotGUI(root) root.mainloop() if __name__ == "__main__": main()