xau-ai-trading-bot/scripts/monitor_bot.py

#!/usr/bin/env python3
"""
Bot Health Monitor & Trade Analyzer
Runs every 1 hour to check bot status and analyze trades
"""
import sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).parent.parent))

import MetaTrader5 as mt5
from datetime import datetime, timedelta
from zoneinfo import ZoneInfo
import polars as pl
from dotenv import load_dotenv
import os

load_dotenv()

def check_bot_health():
    """Check if bot is running and healthy"""
    print("\n" + "="*60)
    print(f"BOT HEALTH CHECK - {datetime.now(ZoneInfo('Asia/Jakarta')).strftime('%Y-%m-%d %H:%M:%S WIB')}")
    print("="*60)

    # Check lock file
    lock_file = Path("data/bot.lock")
    if lock_file.exists():
        with open(lock_file) as f:
            content = f.read().strip()
            print(f"[OK] Bot lock exists: {content}")
    else:
        print("[ERROR] WARNING: No bot lock file found - bot may not be running!")
        return False

    # Check bot_status.json
    status_file = Path("data/bot_status.json")
    if status_file.exists():
        import json
        with open(status_file) as f:
            status = json.load(f)
        print(f"[OK] Bot connected: {status['connected']}")
        print(f"  Price: ${status['price']:.2f}")
        print(f"  Balance: ${status['balance']:.2f}")
        print(f"  Equity: ${status['equity']:.2f}")
        print(f"  Open positions: {len(status['positions'])}")
        print(f"  Daily P/L: ${status['dailyProfit']:.2f} / -${status['dailyLoss']:.2f}")

        # Check last update time
        try:
            last_update = datetime.strptime(status['timestamp'], "%H:%M:%S").replace(
                year=datetime.now().year,
                month=datetime.now().month,
                day=datetime.now().day
            )
            time_since_update = (datetime.now() - last_update).total_seconds()
            if time_since_update > 300:  # 5 minutes
                print(f"[WARN] WARNING: Status last updated {time_since_update/60:.1f} minutes ago!")
            else:
                print(f"[OK] Status fresh ({time_since_update:.0f}s ago)")
        except:
            pass
    else:
        print("[ERROR] WARNING: No bot status file found!")
        return False

    return True

def analyze_todays_trades():
    """Analyze all trades from today"""
    print("\n" + "="*60)
    print("TODAY'S TRADE ANALYSIS")
    print("="*60)

    # Connect to MT5
    if not mt5.initialize():
        print("[X] Failed to connect to MT5")
        return

    try:
        # Get today's trades
        today_start = datetime.now(ZoneInfo("Asia/Jakarta")).replace(hour=0, minute=0, second=0, microsecond=0)
        today_start_utc = today_start.astimezone(ZoneInfo("UTC"))

        deals = mt5.history_deals_get(today_start_utc, datetime.now())
        if not deals:
            print("No trades today")
            return

        # Convert to DataFrame
        deals_dict = [deal._asdict() for deal in deals]
        df = pl.DataFrame(deals_dict)

        # Filter closed positions (deals with profit/loss)
        closed_trades = df.filter(
            (pl.col("entry") == 1) & (pl.col("profit") != 0)
        )

        if len(closed_trades) == 0:
            print("No closed trades today")
            return

        # Calculate statistics
        total_trades = len(closed_trades)
        wins = closed_trades.filter(pl.col("profit") > 0)
        losses = closed_trades.filter(pl.col("profit") < 0)

        total_profit = closed_trades["profit"].sum()
        win_count = len(wins)
        loss_count = len(losses)
        win_rate = (win_count / total_trades * 100) if total_trades > 0 else 0

        avg_win = wins["profit"].mean() if len(wins) > 0 else 0
        avg_loss = abs(losses["profit"].mean()) if len(losses) > 0 else 0

        print(f"Total Trades: {total_trades}")
        print(f"Wins: {win_count} | Losses: {loss_count}")
        print(f"Win Rate: {win_rate:.1f}%")
        print(f"Total P/L: ${total_profit:.2f}")
        print(f"Avg Win: ${avg_win:.2f} | Avg Loss: ${avg_loss:.2f}")

        # Identify issues
        print("\n" + "-"*60)
        print("ISSUE DETECTION")
        print("-"*60)

        issues = []

        # Check win rate
        if win_rate < 45:
            issues.append(f"[WARN]  CRITICAL: Win rate too low ({win_rate:.1f}% < 45%)")
        elif win_rate < 50:
            issues.append(f"[WARN]  WARNING: Win rate below target ({win_rate:.1f}% < 50%)")

        # Check average loss vs win
        if avg_loss > avg_win * 1.5:
            issues.append(f"[WARN]  WARNING: Average loss (${avg_loss:.2f}) > 1.5x average win (${avg_win:.2f})")

        # Check for large losses
        if len(losses) > 0:
            max_loss = abs(losses["profit"].min())
            if max_loss > 20:
                issues.append(f"[WARN]  CRITICAL: Large loss detected: ${max_loss:.2f}")

        # Check consecutive losses
        closed_sorted = closed_trades.sort("time")
        consecutive_losses = 0
        max_consecutive_losses = 0
        for profit in closed_sorted["profit"].to_list():
            if profit < 0:
                consecutive_losses += 1
                max_consecutive_losses = max(max_consecutive_losses, consecutive_losses)
            else:
                consecutive_losses = 0

        if max_consecutive_losses >= 3:
            issues.append(f"[WARN]  WARNING: {max_consecutive_losses} consecutive losses detected")

        # Check BUY vs SELL performance
        buy_trades = closed_trades.filter(pl.col("type") == 0)
        sell_trades = closed_trades.filter(pl.col("type") == 1)

        if len(buy_trades) > 0:
            buy_winrate = len(buy_trades.filter(pl.col("profit") > 0)) / len(buy_trades) * 100
            print(f"\nBUY Trades: {len(buy_trades)} | Win Rate: {buy_winrate:.1f}%")

        if len(sell_trades) > 0:
            sell_winrate = len(sell_trades.filter(pl.col("profit") > 0)) / len(sell_trades) * 100
            print(f"SELL Trades: {len(sell_trades)} | Win Rate: {sell_winrate:.1f}%")

            if sell_winrate < 45 and len(sell_trades) >= 5:
                issues.append(f"[WARN]  CRITICAL: SELL win rate very low ({sell_winrate:.1f}%)")

        # Print issues
        if issues:
            print("\n" + "="*60)
            print("DETECTED ISSUES:")
            for issue in issues:
                print(issue)
            print("="*60)
        else:
            print("\n[OK] No critical issues detected")

        # Recent trades detail
        print("\n" + "-"*60)
        print("LAST 5 TRADES")
        print("-"*60)

        recent = closed_sorted.tail(5)
        for row in recent.iter_rows(named=True):
            trade_time = datetime.fromtimestamp(row['time'], ZoneInfo("Asia/Jakarta"))
            trade_type = "BUY" if row['type'] == 0 else "SELL"
            profit = row['profit']
            emoji = "[OK]" if profit > 0 else "[X]"
            print(f"{emoji} #{row['ticket']} {trade_type} ${profit:+.2f} @ {trade_time.strftime('%H:%M:%S')}")

    finally:
        mt5.shutdown()

def check_open_positions():
    """Check current open positions"""
    print("\n" + "="*60)
    print("OPEN POSITIONS")
    print("="*60)

    if not mt5.initialize():
        print("[X] Failed to connect to MT5")
        return

    try:
        positions = mt5.positions_get(symbol="XAUUSD")
        if not positions or len(positions) == 0:
            print("No open positions")
            return

        print(f"Open Positions: {len(positions)}\n")

        total_profit = 0
        for pos in positions:
            pos_type = "BUY" if pos.type == 0 else "SELL"
            duration = (datetime.now().timestamp() - pos.time) / 60  # minutes
            emoji = "[+]" if pos.profit > 0 else "[-]"
            print(f"{emoji} #{pos.ticket} {pos_type} {pos.volume} lots")
            print(f"   Entry: ${pos.price_open:.2f} | Current: ${pos.price_current:.2f}")
            print(f"   Profit: ${pos.profit:.2f} | Duration: {duration:.1f}m")
            print(f"   SL: ${pos.sl:.2f} | TP: ${pos.tp:.2f}")
            print()
            total_profit += pos.profit

        print(f"Total Floating P/L: ${total_profit:.2f}")

    finally:
        mt5.shutdown()

if __name__ == "__main__":
    try:
        # Run all checks
        bot_healthy = check_bot_health()
        analyze_todays_trades()
        check_open_positions()

        print("\n" + "="*60)
        print(f"Monitor completed at {datetime.now(ZoneInfo('Asia/Jakarta')).strftime('%H:%M:%S WIB')}")
        print("="*60 + "\n")

    except Exception as e:
        print(f"\n[X] ERROR during monitoring: {e}")
        import traceback
        traceback.print_exc()