Place-Order-Trading-Bot/performance_analysis_simple.py

#!/usr/bin/env python3
"""
📊 Trading Bot Performance Analysis (Simple Version - No Dependencies)
Umfassende Performance-Auswertung mit nur SQLite
"""

import sqlite3
from datetime import datetime
from collections import defaultdict
from typing import List, Dict

# ==========================================
# DATABASE QUERIES
# ==========================================

def get_closed_trades(db_path="trading_bot.db", exclude_historical=True):
    """Lade geschlossene Trades"""
    status_filter = "status = 'closed'" if exclude_historical else "status IN ('closed', 'historical')"

    query = f"""
        SELECT
            ticket, symbol, type, volume,
            entry_price, exit_price,
            sl_price, tp_price,
            entry_time, exit_time,
            session, regime, quality, confidence,
            timeframe_alignment,
            risk_amount, risk_pct,
            profit, commission, swap, net_profit,
            profit_pct, rr_ratio,
            exit_reason, status
        FROM trades
        WHERE {status_filter}
        ORDER BY exit_time DESC
    """

    with sqlite3.connect(db_path) as conn:
        conn.row_factory = sqlite3.Row
        cursor = conn.cursor()
        cursor.execute(query)
        return [dict(row) for row in cursor.fetchall()]

# ==========================================
# OVERALL PERFORMANCE
# ==========================================

def calculate_overall_metrics(trades):
    """Berechne Overall Performance"""
    if not trades:
        return None

    total_trades = len(trades)
    wins = [t for t in trades if t['net_profit'] > 0]
    losses = [t for t in trades if t['net_profit'] <= 0]

    win_rate = (len(wins) / total_trades * 100) if total_trades > 0 else 0

    total_profit = sum(t['net_profit'] for t in trades)
    avg_profit = total_profit / total_trades if total_trades > 0 else 0

    avg_win = sum(t['net_profit'] for t in wins) / len(wins) if wins else 0
    avg_loss = sum(t['net_profit'] for t in losses) / len(losses) if losses else 0

    gross_profit = sum(t['net_profit'] for t in wins)
    gross_loss = abs(sum(t['net_profit'] for t in losses))
    profit_factor = round(gross_profit / gross_loss, 2) if gross_loss > 0 else 0

    # Calculate drawdown
    cumulative = 0
    max_cumulative = 0
    max_drawdown = 0

    for trade in sorted(trades, key=lambda x: x['exit_time']):
        cumulative += trade['net_profit']
        if cumulative > max_cumulative:
            max_cumulative = cumulative
        drawdown = cumulative - max_cumulative
        if abs(drawdown) > max_drawdown:
            max_drawdown = abs(drawdown)

    return {
        'total_trades': total_trades,
        'winning_trades': len(wins),
        'losing_trades': len(losses),
        'win_rate': win_rate,
        'total_profit': total_profit,
        'avg_profit': avg_profit,
        'avg_win': avg_win,
        'avg_loss': avg_loss,
        'profit_factor': profit_factor,
        'max_drawdown': max_drawdown
    }

# ==========================================
# SESSION ANALYSIS
# ==========================================

def analyze_by_session(trades):
    """Performance pro Session"""
    sessions = defaultdict(lambda: {'trades': [], 'wins': 0, 'losses': 0, 'profit': 0})

    for trade in trades:
        session = trade['session']
        sessions[session]['trades'].append(trade)
        sessions[session]['profit'] += trade['net_profit']
        if trade['net_profit'] > 0:
            sessions[session]['wins'] += 1
        else:
            sessions[session]['losses'] += 1

    results = []
    for session, data in sessions.items():
        total = len(data['trades'])
        win_rate = (data['wins'] / total * 100) if total > 0 else 0
        avg_profit = data['profit'] / total if total > 0 else 0

        avg_win = sum(t['net_profit'] for t in data['trades'] if t['net_profit'] > 0)
        avg_win = avg_win / data['wins'] if data['wins'] > 0 else 0

        avg_loss = sum(t['net_profit'] for t in data['trades'] if t['net_profit'] <= 0)
        avg_loss = avg_loss / data['losses'] if data['losses'] > 0 else 0

        results.append({
            'session': session.upper(),
            'trades': total,
            'wins': data['wins'],
            'losses': data['losses'],
            'win_rate': win_rate,
            'total_profit': data['profit'],
            'avg_profit': avg_profit,
            'avg_win': avg_win,
            'avg_loss': avg_loss
        })

    return sorted(results, key=lambda x: x['total_profit'], reverse=True)

# ==========================================
# CONFIDENCE ANALYSIS
# ==========================================

def analyze_by_confidence(trades):
    """Performance pro Confidence Level"""
    bins = [(0, 60), (60, 70), (70, 75), (75, 80), (80, 100)]
    conf_groups = defaultdict(lambda: {'trades': [], 'wins': 0, 'profit': 0})

    for trade in trades:
        conf = trade['confidence']
        for bin_min, bin_max in bins:
            if bin_min <= conf < bin_max:
                key = f"{bin_min}-{bin_max}"
                conf_groups[key]['trades'].append(trade)
                conf_groups[key]['profit'] += trade['net_profit']
                if trade['net_profit'] > 0:
                    conf_groups[key]['wins'] += 1
                break

    results = []
    for conf_range, data in conf_groups.items():
        total = len(data['trades'])
        win_rate = (data['wins'] / total * 100) if total > 0 else 0
        avg_profit = data['profit'] / total if total > 0 else 0

        results.append({
            'confidence_range': conf_range,
            'trades': total,
            'wins': data['wins'],
            'win_rate': win_rate,
            'total_profit': data['profit'],
            'avg_profit': avg_profit
        })

    return sorted(results, key=lambda x: int(x['confidence_range'].split('-')[0]))

# ==========================================
# EXIT REASON ANALYSIS
# ==========================================

def analyze_by_exit_reason(trades):
    """Performance pro Exit Reason"""
    reasons = defaultdict(lambda: {'trades': [], 'wins': 0, 'profit': 0})

    for trade in trades:
        reason = trade['exit_reason']
        reasons[reason]['trades'].append(trade)
        reasons[reason]['profit'] += trade['net_profit']
        if trade['net_profit'] > 0:
            reasons[reason]['wins'] += 1

    results = []
    for reason, data in reasons.items():
        total = len(data['trades'])
        win_rate = (data['wins'] / total * 100) if total > 0 else 0
        avg_profit = data['profit'] / total if total > 0 else 0

        results.append({
            'exit_reason': reason.upper().replace('_', ' '),
            'trades': total,
            'wins': data['wins'],
            'win_rate': win_rate,
            'total_profit': data['profit'],
            'avg_profit': avg_profit
        })

    return sorted(results, key=lambda x: x['total_profit'], reverse=True)

# ==========================================
# TIME ANALYSIS
# ==========================================

def analyze_by_hour(trades):
    """Performance pro Stunde"""
    hours = defaultdict(lambda: {'trades': 0, 'wins': 0, 'profit': 0})

    for trade in trades:
        hour = datetime.fromisoformat(trade['entry_time']).hour
        hours[hour]['trades'] += 1
        hours[hour]['profit'] += trade['net_profit']
        if trade['net_profit'] > 0:
            hours[hour]['wins'] += 1

    results = []
    for hour, data in hours.items():
        win_rate = (data['wins'] / data['trades'] * 100) if data['trades'] > 0 else 0
        avg_profit = data['profit'] / data['trades'] if data['trades'] > 0 else 0

        results.append({
            'hour': hour,
            'trades': data['trades'],
            'wins': data['wins'],
            'win_rate': win_rate,
            'total_profit': data['profit'],
            'avg_profit': avg_profit
        })

    return sorted(results, key=lambda x: x['total_profit'], reverse=True)

# ==========================================
# PRINT FUNCTIONS
# ==========================================

def print_section(title):
    """Print section header"""
    print("\n" + "=" * 70)
    print(f"  {title}")
    print("=" * 70)

def print_overall(metrics):
    """Print overall metrics"""
    print_section("📊 OVERALL PERFORMANCE")

    print(f"\n{'Total Trades:':<25} {metrics['total_trades']}")
    print(f"{'Winning Trades:':<25} {metrics['winning_trades']} ({metrics['win_rate']:.1f}%)")
    print(f"{'Losing Trades:':<25} {metrics['losing_trades']}")
    print(f"\n{'Total Profit:':<25} ${metrics['total_profit']:.2f}")
    print(f"{'Average Profit/Trade:':<25} ${metrics['avg_profit']:.2f}")
    print(f"{'Average Win:':<25} ${metrics['avg_win']:.2f}")
    print(f"{'Average Loss:':<25} ${metrics['avg_loss']:.2f}")
    print(f"{'Profit Factor:':<25} {metrics['profit_factor']:.2f}")
    print(f"\n{'Max Drawdown:':<25} ${metrics['max_drawdown']:.2f}")

def print_table(data, title):
    """Print data as table"""
    print_section(title)

    if not data:
        print("\nNo data available")
        return

    # Print header
    headers = list(data[0].keys())
    print("\n" + " | ".join(f"{h:<15}" for h in headers))
    print("-" * (len(headers) * 18))

    # Print rows
    for row in data:
        values = []
        for key, val in row.items():
            if isinstance(val, float):
                values.append(f"{val:>15.2f}")
            else:
                values.append(f"{str(val):<15}")
        print(" | ".join(values))

# ==========================================
# MAIN ANALYSIS
# ==========================================

def run_analysis(db_path="trading_bot.db", exclude_historical=True):
    """Run complete performance analysis"""

    print("=" * 70)
    print("📊 TRADING BOT PERFORMANCE ANALYSIS")
    print("=" * 70)
    print(f"\nAnalysis Date: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
    print(f"Database: {db_path}")
    print(f"Exclude Historical: {exclude_historical}")

    # Load data
    trades = get_closed_trades(db_path, exclude_historical)

    if not trades:
        print("\n❌ No closed trades found!")
        print("\nPossible reasons:")
        print("  • Position Monitor not running")
        print("  • No trades have been closed yet")
        print("  • Database not synced from VPS")
        return

    print(f"\nLoaded {len(trades)} closed trades")

    # Overall Metrics
    overall = calculate_overall_metrics(trades)
    print_overall(overall)

    # Session Analysis
    session_data = analyze_by_session(trades)
    print_table(session_data, "📍 PERFORMANCE BY SESSION")

    # Confidence Analysis
    conf_data = analyze_by_confidence(trades)
    print_table(conf_data, "🎯 PERFORMANCE BY CONFIDENCE LEVEL")

    # Exit Reason Analysis
    exit_data = analyze_by_exit_reason(trades)
    print_table(exit_data, "🚪 PERFORMANCE BY EXIT REASON")

    # Hourly Analysis (Top 10)
    hourly_data = analyze_by_hour(trades)
    print_table(hourly_data[:10], "⏰ TOP 10 HOURS (UTC)")

    # Recommendations
    print_section("💡 RECOMMENDATIONS")

    if session_data:
        best = session_data[0]
        worst = session_data[-1]

        print(f"\n✅ Best Session: {best['session']}")
        print(f"   Win Rate: {best['win_rate']:.1f}%")
        print(f"   Total Profit: ${best['total_profit']:.2f}")

        if worst['total_profit'] < 0:
            print(f"\n❌ Worst Session: {worst['session']}")
            print(f"   Win Rate: {worst['win_rate']:.1f}%")
            print(f"   Total Loss: ${worst['total_profit']:.2f}")
            print(f"\n   → Consider disabling {worst['session']} session")

    if conf_data:
        best_conf = max(conf_data, key=lambda x: x['total_profit'])
        print(f"\n🎯 Best Confidence Range: {best_conf['confidence_range']}")
        print(f"   Win Rate: {best_conf['win_rate']:.1f}%")
        print(f"   Total Profit: ${best_conf['total_profit']:.2f}")

    print("\n" + "=" * 70)
    print("✅ Analysis Complete!")
    print("=" * 70)

if __name__ == "__main__":
    run_analysis(
        db_path="trading_bot.db",
        exclude_historical=True
    )