class TelegramBot: def __init__(self, token, chat_id): self.token = token self.chat_id = chat_id self.base_url = f"https://api.telegram.org/bot{token}"

def send_message(self, message, parse_mode='HTML'): """Send text message""" url = f"{self.base_url}/sendMessage" payload = { 'chat_id': self.chat_id, 'text': message, 'parse_mode': parse_mode } response = requests.post(url, json=payload) return response.json()

def send_trade_alert(self, trade_type, symbol, price, lots, sl=None, tp=None): """Send formatted trade alert""" emoji = "🟢" if trade_type.upper() == "BUY" else "🔴"

message = f""" {emoji} NEW TRADE Type: {trade_type.upper()} Symbol: {symbol} Price: {price} Lots: {lots} SL: {sl if sl else 'None'} TP: {tp if tp else 'None'} Time: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')} """ return self.send_message(message)

def send_close_alert(self, symbol, profit, pips): """Send trade close alert""" emoji = "✅" if profit > 0 else "❌"

message = f""" {emoji} TRADE CLOSED Symbol: {symbol} Profit: ${profit:.2f} Pips: {pips:.1f} """ return self.send_message(message)

def send_error_alert(self, error_message): """Send error alert""" message = f""" ⚠️ ERROR ALERT {error_message} Time: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')} """ return self.send_message(message)

def send_daily_summary(self, stats): """Send daily performance summary""" message = f""" 📊 DAILY SUMMARY Date: {datetime.now().strftime('%Y-%m-%d')}

Trades: {stats['total_trades']} Wins: {stats['wins']} | Losses: {stats['losses']} Win Rate: {stats['win_rate']:.1%}

P&L: ${stats['pnl']:.2f} Pips: {stats['total_pips']:.1f}

Balance: ${stats['balance']:.2f} Equity: ${stats['equity']:.2f} Drawdown: {stats['drawdown']:.2%} """ return self.send_message(message)

# Setup bot = TelegramBot( token='YOUR_BOT_TOKEN', chat_id='YOUR_CHAT_ID' )

Creating Telegram Bot

Email Alerts

class EmailNotifier: def __init__(self, smtp_server, smtp_port, username, password): self.smtp_server = smtp_server self.smtp_port = smtp_port self.username = username self.password = password

def send_email(self, to_email, subject, body): """Send email notification""" msg = MIMEMultipart() msg['From'] = self.username msg['To'] = to_email msg['Subject'] = subject msg.attach(MIMEText(body, 'html'))

try: with smtplib.SMTP(self.smtp_server, self.smtp_port) as server: server.starttls() server.login(self.username, self.password) server.send_message(msg) return True except Exception as e: print(f"Email failed: {e}") return False

MQL5 Alert Integration

Sending Alerts from EA

void SendTelegramAlert(string message) { if(TelegramToken == "" || TelegramChatID == "") return;

string url = "https://api.telegram.org/bot" + TelegramToken + "/sendMessage";

string headers = "Content-Type: application/x-www-form-urlencoded\r\n"; string data = "chat_id=" + TelegramChatID + "&text=" + message + "&parse_mode=HTML";

char post[]; char result[]; string resultHeaders;

StringToCharArray(data, post); ArrayResize(post, ArraySize(post) - 1);

int res = WebRequest("POST", url, headers, 5000, post, result, resultHeaders);

if(res == -1) { Print("Telegram error: ", GetLastError()); } }

// Usage in OnTrade() void OnTrade() { static int prevPositions = 0; int currentPositions = PositionsTotal();

if(currentPositions > prevPositions) { // New position opened string msg = "New Trade Opened\n"; msg += "Check MT5 for details"; SendTelegramAlert(msg); } else if(currentPositions < prevPositions) { // Position closed string msg = "Trade Closed\n"; msg += "Check MT5 for details"; SendTelegramAlert(msg); }

System Health Monitoring

Python Health Monitor

class SystemMonitor: def __init__(self, telegram_bot, alert_thresholds=None): self.bot = telegram_bot self.thresholds = alert_thresholds or { 'cpu_percent': 90, 'memory_percent': 85, 'disk_percent': 90 } self.last_alert = {}

def check_system(self): """Check system resources""" cpu = psutil.cpu_percent(interval=1) memory = psutil.virtual_memory().percent disk = psutil.disk_usage('/').percent

alerts = []

if cpu > self.thresholds['cpu_percent']: alerts.append(f"CPU: {cpu}%")

if memory > self.thresholds['memory_percent']: alerts.append(f"Memory: {memory}%")

if disk > self.thresholds['disk_percent']: alerts.append(f"Disk: {disk}%")

return { 'cpu': cpu, 'memory': memory, 'disk': disk, 'alerts': alerts }

def check_process(self, process_name): """Check if process is running""" for proc in psutil.process_iter(['name']): if proc.info['name'] == process_name: return True return False

def run_monitoring(self, check_interval=60): """Continuous monitoring loop""" while True: try: # Check system status = self.check_system()

if status['alerts']: self.send_alert(f"System Alert: {', '.join(status['alerts'])}")

# Check MT5 process if not self.check_process('terminal64.exe'): self.send_alert("MT5 is not running!")

time.sleep(check_interval)

except Exception as e: self.send_alert(f"Monitor error: {str(e)}") time.sleep(check_interval)

def send_alert(self, message, cooldown=300): """Send alert with cooldown to prevent spam""" now = datetime.now() alert_key = hash(message)

if alert_key in self.last_alert: time_diff = (now - self.last_alert[alert_key]).seconds if time_diff < cooldown: return # Skip, too soon

Trade Performance Monitoring

Performance Tracker

def record_trade(self, trade): """Record completed trade""" self.daily_trades.append(trade) self.daily_pnl += trade['profit']

# Send individual trade alert if trade['profit'] >= 0: self.bot.send_close_alert( trade['symbol'], trade['profit'], trade['pips'] ) else: self.bot.send_close_alert( trade['symbol'], trade['profit'], trade['pips'] )

def check_drawdown(self, current_equity, max_drawdown_pct=10): """Check if drawdown exceeded""" if current_equity > self.peak_equity: self.peak_equity = current_equity

drawdown = (self.peak_equity - current_equity) / self.peak_equity * 100

if drawdown >= max_drawdown_pct: self.bot.send_error_alert( f"Drawdown alert: {drawdown:.2f}%\n" f"Peak: ${self.peak_equity:.2f}\n" f"Current: ${current_equity:.2f}" ) return True return False

def daily_report(self, balance, equity): """Generate and send daily report""" wins = sum(1 for t in self.daily_trades if t['profit'] > 0) losses = len(self.daily_trades) - wins

stats = { 'total_trades': len(self.daily_trades), 'wins': wins, 'losses': losses, 'win_rate': wins / len(self.daily_trades) if self.daily_trades else 0, 'pnl': self.daily_pnl, 'total_pips': sum(t['pips'] for t in self.daily_trades), 'balance': balance, 'equity': equity, 'drawdown': (self.peak_equity - equity) / self.peak_equity if self.peak_equity > 0 else 0 }

self.bot.send_daily_summary(stats)

Dashboard Monitoring

Simple Web Dashboard

app = Flask(__name__)

# Shared state trading_state = { 'status': 'running', 'balance': 10000, 'equity': 10000, 'open_positions': 0, 'daily_pnl': 0, 'last_update': None }

DASHBOARD_HTML = """

Trading Dashboard

Last Update: {{ last_update }}

@app.route('/') def dashboard(): return render_template_string( DASHBOARD_HTML, status=trading_state['status'], balance=trading_state['balance'], equity=trading_state['equity'], positions=trading_state['open_positions'], pnl=trading_state['daily_pnl'], last_update=trading_state['last_update'] )

@app.route('/api/status') def api_status(): return jsonify(trading_state)

def update_state(new_state): """Update trading state""" trading_state.update(new_state) trading_state['last_update'] = datetime.now().strftime('%Y-%m-%d %H:%M:%S')

Scheduled Reports

class ReportScheduler: def __init__(self, telegram_bot, performance_tracker): self.bot = telegram_bot self.tracker = performance_tracker

def setup_schedules(self): """Setup report schedules""" # Daily report at 5 PM schedule.every().day.at("17:00").do(self.send_daily_report)

# Weekly report on Sunday schedule.every().sunday.at("18:00").do(self.send_weekly_report)

# Heartbeat every hour schedule.every().hour.do(self.send_heartbeat)

def send_daily_report(self): """Send daily summary""" # Gather stats and send pass

def send_weekly_report(self): """Send weekly summary""" pass

def send_heartbeat(self): """Send hourly heartbeat""" self.bot.send_message("💓 System running normally")