Place-Order-Trading-Bot/performance_analysis.py

#!/usr/bin/env python3
"""
📊 Trading Bot Performance Analysis
Umfassende Performance-Auswertung mit Session-, Confidence- und Zeitanalyse
"""

import sqlite3
import pandas as pd
from datetime import datetime, timedelta
import json

# ==========================================
# DATABASE CONNECTION
# ==========================================

def get_connection(db_path="trading_bot.db"):
    """Verbindung zur Datenbank"""
    conn = sqlite3.connect(db_path)
    conn.row_factory = sqlite3.Row
    return conn

# ==========================================
# DATA LOADING
# ==========================================

def load_closed_trades(conn, exclude_historical=True):
    """Lade geschlossene Trades"""
    query = """
        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 = 'closed'
    """

    if exclude_historical:
        query += " AND status != 'historical'"

    query += " ORDER BY exit_time DESC"

    df = pd.read_sql_query(query, conn)

    # Convert datetime columns
    if not df.empty:
        df['entry_time'] = pd.to_datetime(df['entry_time'], format='mixed')
        df['exit_time'] = pd.to_datetime(df['exit_time'], format='mixed')
        df['duration_hours'] = (df['exit_time'] - df['entry_time']).dt.total_seconds() / 3600
        df['win'] = df['net_profit'] > 0

    return df

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

def calculate_overall_metrics(df):
    """Berechne Overall Performance Metriken"""
    if df.empty:
        return None

    total_trades = len(df)
    winning_trades = len(df[df['win']])
    losing_trades = len(df[~df['win']])

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

    total_profit = df['net_profit'].sum()
    avg_profit = df['net_profit'].mean()

    avg_win = df[df['win']]['net_profit'].mean() if winning_trades > 0 else 0
    avg_loss = df[~df['win']]['net_profit'].mean() if losing_trades > 0 else 0

    profit_factor = abs(avg_win / avg_loss) if avg_loss != 0 else 0

    avg_duration = df['duration_hours'].mean()

    # Drawdown
    df_sorted = df.sort_values('exit_time')
    df_sorted['cumulative'] = df_sorted['net_profit'].cumsum()
    df_sorted['running_max'] = df_sorted['cumulative'].cummax()
    df_sorted['drawdown'] = df_sorted['cumulative'] - df_sorted['running_max']
    max_drawdown = df_sorted['drawdown'].min()

    return {
        'total_trades': total_trades,
        'winning_trades': winning_trades,
        'losing_trades': losing_trades,
        'win_rate': win_rate,
        'total_profit': total_profit,
        'avg_profit': avg_profit,
        'avg_win': avg_win,
        'avg_loss': avg_loss,
        'profit_factor': profit_factor,
        'avg_duration_hours': avg_duration,
        'max_drawdown': max_drawdown
    }

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

def analyze_by_session(df):
    """Performance pro Session"""
    if df.empty:
        return pd.DataFrame()

    session_stats = []

    for session in ['ny', 'london', 'asian', 'overlap']:
        session_df = df[df['session'] == session]

        if session_df.empty:
            continue

        total = len(session_df)
        wins = len(session_df[session_df['win']])
        losses = total - wins
        win_rate = (wins / total * 100) if total > 0 else 0

        total_profit = session_df['net_profit'].sum()
        avg_profit = session_df['net_profit'].mean()

        avg_win = session_df[session_df['win']]['net_profit'].mean() if wins > 0 else 0
        avg_loss = session_df[~session_df['win']]['net_profit'].mean() if losses > 0 else 0

        session_stats.append({
            'session': session.upper(),
            'trades': total,
            'wins': wins,
            'losses': losses,
            'win_rate': win_rate,
            'total_profit': total_profit,
            'avg_profit': avg_profit,
            'avg_win': avg_win,
            'avg_loss': avg_loss
        })

    return pd.DataFrame(session_stats).sort_values('total_profit', ascending=False)

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

def analyze_by_confidence(df, bins=[0, 60, 70, 75, 80, 100]):
    """Performance pro Confidence Level"""
    if df.empty:
        return pd.DataFrame()

    df['confidence_bin'] = pd.cut(df['confidence'], bins=bins,
                                   labels=[f"{bins[i]}-{bins[i+1]}" for i in range(len(bins)-1)])

    conf_stats = []

    for conf_range in df['confidence_bin'].unique():
        conf_df = df[df['confidence_bin'] == conf_range]

        total = len(conf_df)
        wins = len(conf_df[conf_df['win']])
        win_rate = (wins / total * 100) if total > 0 else 0

        total_profit = conf_df['net_profit'].sum()
        avg_profit = conf_df['net_profit'].mean()

        conf_stats.append({
            'confidence_range': str(conf_range),
            'trades': total,
            'wins': wins,
            'win_rate': win_rate,
            'total_profit': total_profit,
            'avg_profit': avg_profit
        })

    return pd.DataFrame(conf_stats).sort_values('confidence_range')

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

def analyze_by_exit_reason(df):
    """Performance pro Exit Reason"""
    if df.empty:
        return pd.DataFrame()

    exit_stats = []

    for reason in ['take_profit', 'stop_loss', 'manual_close']:
        reason_df = df[df['exit_reason'] == reason]

        if reason_df.empty:
            continue

        total = len(reason_df)
        wins = len(reason_df[reason_df['win']])
        win_rate = (wins / total * 100) if total > 0 else 0

        total_profit = reason_df['net_profit'].sum()
        avg_profit = reason_df['net_profit'].mean()

        exit_stats.append({
            'exit_reason': reason.upper().replace('_', ' '),
            'trades': total,
            'wins': wins,
            'win_rate': win_rate,
            'total_profit': total_profit,
            'avg_profit': avg_profit
        })

    return pd.DataFrame(exit_stats).sort_values('total_profit', ascending=False)

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

def analyze_by_hour(df):
    """Performance pro Stunde (UTC)"""
    if df.empty:
        return pd.DataFrame()

    df['hour'] = df['entry_time'].dt.hour

    hourly_stats = df.groupby('hour').agg({
        'ticket': 'count',
        'win': 'sum',
        'net_profit': ['sum', 'mean']
    }).round(2)

    hourly_stats.columns = ['trades', 'wins', 'total_profit', 'avg_profit']
    hourly_stats['win_rate'] = (hourly_stats['wins'] / hourly_stats['trades'] * 100).round(1)
    hourly_stats = hourly_stats.reset_index()

    return hourly_stats.sort_values('total_profit', ascending=False)

def analyze_by_weekday(df):
    """Performance pro Wochentag"""
    if df.empty:
        return pd.DataFrame()

    df['weekday'] = df['entry_time'].dt.day_name()

    weekday_order = ['Monday', 'Tuesday', 'Wednesday', 'Thursday', 'Friday', 'Saturday', 'Sunday']

    weekday_stats = []

    for day in weekday_order:
        day_df = df[df['weekday'] == day]

        if day_df.empty:
            continue

        total = len(day_df)
        wins = len(day_df[day_df['win']])
        win_rate = (wins / total * 100) if total > 0 else 0

        total_profit = day_df['net_profit'].sum()
        avg_profit = day_df['net_profit'].mean()

        weekday_stats.append({
            'weekday': day,
            'trades': total,
            'wins': wins,
            'win_rate': win_rate,
            'total_profit': total_profit,
            'avg_profit': avg_profit
        })

    return pd.DataFrame(weekday_stats)

# ==========================================
# REGIME ANALYSIS
# ==========================================

def analyze_by_regime(df):
    """Performance pro Market Regime"""
    if df.empty or 'regime' not in df.columns:
        return pd.DataFrame()

    regime_stats = []

    for regime in df['regime'].unique():
        if pd.isna(regime):
            continue

        regime_df = df[df['regime'] == regime]

        total = len(regime_df)
        wins = len(regime_df[regime_df['win']])
        win_rate = (wins / total * 100) if total > 0 else 0

        total_profit = regime_df['net_profit'].sum()
        avg_profit = regime_df['net_profit'].mean()

        regime_stats.append({
            'regime': regime,
            'trades': total,
            'wins': wins,
            'win_rate': win_rate,
            'total_profit': total_profit,
            'avg_profit': avg_profit
        })

    return pd.DataFrame(regime_stats).sort_values('total_profit', ascending=False)

# ==========================================
# PRINT REPORTS
# ==========================================

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

def print_overall_metrics(metrics):
    """Print overall performance"""
    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{'Average Duration:':<25} {metrics['avg_duration_hours']:.1f} hours")
    print(f"{'Max Drawdown:':<25} ${metrics['max_drawdown']:.2f}")

def print_dataframe_report(df, title):
    """Print DataFrame as formatted report"""
    print_section(title)

    if df.empty:
        print("\nNo data available")
        return

    print("\n" + df.to_string(index=False))

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

def run_performance_analysis(db_path="trading_bot.db", exclude_historical=True):
    """Führe komplette Performance-Analyse aus"""

    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
    conn = get_connection(db_path)
    df = load_closed_trades(conn, exclude_historical)

    if df.empty:
        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(df)} closed trades")

    # Overall Metrics
    overall = calculate_overall_metrics(df)
    print_overall_metrics(overall)

    # Session Analysis
    session_df = analyze_by_session(df)
    print_dataframe_report(session_df, "📍 PERFORMANCE BY SESSION")

    # Confidence Analysis
    conf_df = analyze_by_confidence(df)
    print_dataframe_report(conf_df, "🎯 PERFORMANCE BY CONFIDENCE LEVEL")

    # Exit Reason Analysis
    exit_df = analyze_by_exit_reason(df)
    print_dataframe_report(exit_df, "🚪 PERFORMANCE BY EXIT REASON")

    # Hourly Analysis (Top 10)
    hourly_df = analyze_by_hour(df)
    print_dataframe_report(hourly_df.head(10), "⏰ TOP 10 HOURS (UTC)")

    # Weekday Analysis
    weekday_df = analyze_by_weekday(df)
    print_dataframe_report(weekday_df, "📅 PERFORMANCE BY WEEKDAY")

    # Regime Analysis
    regime_df = analyze_by_regime(df)
    if not regime_df.empty:
        print_dataframe_report(regime_df, "📈 PERFORMANCE BY MARKET REGIME")

    # Recommendations
    print_section("💡 RECOMMENDATIONS")

    if not session_df.empty:
        best_session = session_df.iloc[0]
        worst_session = session_df.iloc[-1]

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

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

    if not conf_df.empty:
        best_conf = conf_df.loc[conf_df['total_profit'].idxmax()]
        print(f"\n🎯 Best Confidence Range: {best_conf['confidence_range']}")
        print(f"   Win Rate: {best_conf['win_rate']:.1f}%")
        print(f"   → Consider using confidence threshold >= {best_conf['confidence_range'].split('-')[0]}")

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

    conn.close()

    return {
        'overall': overall,
        'by_session': session_df,
        'by_confidence': conf_df,
        'by_exit_reason': exit_df,
        'by_hour': hourly_df,
        'by_weekday': weekday_df,
        'by_regime': regime_df
    }

# ==========================================
# EXPORT TO JSON
# ==========================================

def export_analysis_to_json(results, output_file="performance_analysis.json"):
    """Export analysis results to JSON"""

    output = {
        'timestamp': datetime.now().isoformat(),
        'overall_metrics': results['overall'],
        'by_session': results['by_session'].to_dict('records') if not results['by_session'].empty else [],
        'by_confidence': results['by_confidence'].to_dict('records') if not results['by_confidence'].empty else [],
        'by_exit_reason': results['by_exit_reason'].to_dict('records') if not results['by_exit_reason'].empty else [],
        'by_hour': results['by_hour'].to_dict('records') if not results['by_hour'].empty else [],
        'by_weekday': results['by_weekday'].to_dict('records') if not results['by_weekday'].empty else [],
        'by_regime': results['by_regime'].to_dict('records') if not results['by_regime'].empty else []
    }

    with open(output_file, 'w') as f:
        json.dump(output, f, indent=2)

    print(f"\n✅ Analysis exported to: {output_file}")

# ==========================================
# MAIN EXECUTION
# ==========================================

if __name__ == "__main__":
    # Run analysis
    results = run_performance_analysis(
        db_path="trading_bot.db",
        exclude_historical=True  # Set to False to include historical imports
    )

    # Export to JSON (optional)
    if results:
        export_analysis_to_json(results)