Place-Order-Trading-Bot/demo_test_tracker.py

#!/usr/bin/env python3
"""
📊 Demo Test Tracker
Sammelt Statistiken während der Demo-Phase und gibt Go-Live Empfehlungen

FEATURES:
1. Trade Logging (Entry, Exit, Profit/Loss)
2. Performance Metriken (Win Rate, Profit Factor, etc.)
3. Drawdown Tracking
4. Session-basierte Analyse
5. Error/Bug Tracking
6. Go-Live Readiness Check
7. Automatische Reports

VERWENDUNG:
    from demo_test_tracker import DemoTestTracker

    tracker = DemoTestTracker()

    # Nach jedem Trade:
    tracker.log_trade(trade_result)

    # Report anzeigen:
    tracker.print_report()

    # Go-Live Check:
    tracker.check_go_live_readiness()
"""

import json
import os
from datetime import datetime, timedelta
from typing import Dict, List, Optional, Tuple
from dataclasses import dataclass, asdict
from collections import defaultdict
import logging

logger = logging.getLogger(__name__)


@dataclass
class TradeRecord:
    """Einzelner Trade-Eintrag"""
    ticket: int
    symbol: str
    direction: str  # "LONG" or "SHORT"
    entry_price: float
    exit_price: float
    volume: float
    profit: float
    profit_pips: float
    entry_time: str
    exit_time: str
    duration_minutes: float
    session: str

    # Signal Info
    base_confidence: float
    enhanced_score: float
    hybrid_score: float
    signal_quality: str

    # Filters applied
    equity_curve_multiplier: float

    # Result
    is_win: bool

    # Optional metadata
    stop_loss: float = 0.0
    take_profit: float = 0.0
    close_reason: str = ""  # "TP", "SL", "Manual", "Trailing"


class DemoTestTracker:
    """
    Demo Test Tracker für Go-Live Vorbereitung

    Sammelt alle relevanten Statistiken und gibt
    Empfehlungen wann der Bot produktionsreif ist.
    """

    # Go-Live Kriterien
    GO_LIVE_CRITERIA = {
        'min_trades': 50,
        'min_win_rate': 0.55,  # 55%
        'min_profit_factor': 1.3,
        'max_drawdown': 0.15,  # 15%
        'min_days': 14,
        'max_errors': 5,
        'min_sessions_tested': 2,  # Mindestens 2 verschiedene Sessions
    }

    def __init__(self,
                 data_file: str = "demo_test_stats.json",
                 criteria: Optional[Dict] = None):
        """
        Args:
            data_file: Datei für persistente Speicherung
            criteria: Custom Go-Live Kriterien (optional)
        """
        self.data_file = data_file
        self.criteria = criteria or self.GO_LIVE_CRITERIA.copy()

        # Daten laden oder initialisieren
        self.data = self._load_data()

        logger.info("=" * 60)
        logger.info("📊 DEMO TEST TRACKER INITIALIZED")
        logger.info("=" * 60)
        logger.info(f"   Data File:     {data_file}")
        logger.info(f"   Total Trades:  {len(self.data['trades'])}")
        logger.info(f"   Start Date:    {self.data['start_date']}")
        logger.info(f"   Days Running:  {self._days_running()}")
        logger.info("=" * 60)

    # ==========================================
    # TRADE LOGGING
    # ==========================================

    def log_trade(self,
                  ticket: int,
                  symbol: str,
                  direction: str,
                  entry_price: float,
                  exit_price: float,
                  volume: float,
                  profit: float,
                  entry_time: datetime,
                  exit_time: datetime,
                  session: str = "unknown",
                  base_confidence: float = 0.0,
                  enhanced_score: float = 0.0,
                  hybrid_score: float = 0.0,
                  signal_quality: str = "unknown",
                  equity_curve_multiplier: float = 1.0,
                  stop_loss: float = 0.0,
                  take_profit: float = 0.0,
                  close_reason: str = "") -> TradeRecord:
        """
        Loggt einen abgeschlossenen Trade

        Returns:
            TradeRecord object
        """
        # Calculate derived values
        if direction == "LONG":
            profit_pips = (exit_price - entry_price) * 10  # For Gold
        else:
            profit_pips = (entry_price - exit_price) * 10

        duration = (exit_time - entry_time).total_seconds() / 60
        is_win = profit > 0

        record = TradeRecord(
            ticket=ticket,
            symbol=symbol,
            direction=direction,
            entry_price=entry_price,
            exit_price=exit_price,
            volume=volume,
            profit=profit,
            profit_pips=profit_pips,
            entry_time=entry_time.isoformat(),
            exit_time=exit_time.isoformat(),
            duration_minutes=duration,
            session=session,
            base_confidence=base_confidence,
            enhanced_score=enhanced_score,
            hybrid_score=hybrid_score,
            signal_quality=signal_quality,
            equity_curve_multiplier=equity_curve_multiplier,
            is_win=is_win,
            stop_loss=stop_loss,
            take_profit=take_profit,
            close_reason=close_reason
        )

        # Add to data
        self.data['trades'].append(asdict(record))
        self.data['last_updated'] = datetime.now().isoformat()

        # Update running stats
        self._update_stats()

        # Save
        self._save_data()

        # Log
        emoji = "🟢" if is_win else "🔴"
        logger.info(f"📊 Trade logged: {emoji} #{ticket} {direction} {symbol} | Profit: ${profit:.2f}")

        return record

    def log_trade_from_mt5(self, position, close_reason: str = "",
                          session: str = "unknown",
                          base_confidence: float = 0.0,
                          enhanced_score: float = 0.0,
                          hybrid_score: float = 0.0,
                          signal_quality: str = "unknown",
                          equity_curve_multiplier: float = 1.0) -> TradeRecord:
        """
        Loggt Trade direkt von MT5 Position Object
        """
        direction = "LONG" if position.type == 0 else "SHORT"

        return self.log_trade(
            ticket=position.ticket,
            symbol=position.symbol,
            direction=direction,
            entry_price=position.price_open,
            exit_price=position.price_current,
            volume=position.volume,
            profit=position.profit,
            entry_time=datetime.fromtimestamp(position.time),
            exit_time=datetime.now(),
            session=session,
            base_confidence=base_confidence,
            enhanced_score=enhanced_score,
            hybrid_score=hybrid_score,
            signal_quality=signal_quality,
            equity_curve_multiplier=equity_curve_multiplier,
            stop_loss=position.sl,
            take_profit=position.tp,
            close_reason=close_reason
        )

    def log_error(self, error_type: str, message: str, details: Optional[Dict] = None):
        """Loggt einen Fehler/Bug"""
        error_entry = {
            'timestamp': datetime.now().isoformat(),
            'type': error_type,
            'message': message,
            'details': details or {}
        }
        self.data['errors'].append(error_entry)
        self._save_data()

        logger.warning(f"📊 Error logged: {error_type} - {message}")

    def log_filtered_signal(self, reason: str, base_confidence: float,
                           enhanced_score: float, hybrid_score: float):
        """Loggt ein Signal das gefiltert wurde"""
        entry = {
            'timestamp': datetime.now().isoformat(),
            'reason': reason,
            'base_confidence': base_confidence,
            'enhanced_score': enhanced_score,
            'hybrid_score': hybrid_score
        }
        self.data['filtered_signals'].append(entry)
        self._save_data()

    # ==========================================
    # STATISTICS
    # ==========================================

    def _update_stats(self):
        """Aktualisiert alle Statistiken"""
        trades = self.data['trades']

        if not trades:
            return

        # Basic stats
        wins = [t for t in trades if t['is_win']]
        losses = [t for t in trades if not t['is_win']]

        total_profit = sum(t['profit'] for t in trades)
        total_wins = sum(t['profit'] for t in wins) if wins else 0
        total_losses = abs(sum(t['profit'] for t in losses)) if losses else 0

        # Win Rate
        win_rate = len(wins) / len(trades) if trades else 0

        # Profit Factor
        profit_factor = total_wins / total_losses if total_losses > 0 else float('inf')

        # Average Win/Loss
        avg_win = total_wins / len(wins) if wins else 0
        avg_loss = total_losses / len(losses) if losses else 0

        # Drawdown calculation
        equity_curve = []
        running_equity = 0
        peak_equity = 0
        max_drawdown = 0

        for t in trades:
            running_equity += t['profit']
            equity_curve.append(running_equity)

            if running_equity > peak_equity:
                peak_equity = running_equity

            drawdown = (peak_equity - running_equity) / peak_equity if peak_equity > 0 else 0
            if drawdown > max_drawdown:
                max_drawdown = drawdown

        # Session stats
        session_stats = defaultdict(lambda: {'trades': 0, 'wins': 0, 'profit': 0})
        for t in trades:
            session = t.get('session', 'unknown')
            session_stats[session]['trades'] += 1
            session_stats[session]['wins'] += 1 if t['is_win'] else 0
            session_stats[session]['profit'] += t['profit']

        # Signal quality stats
        quality_stats = defaultdict(lambda: {'trades': 0, 'wins': 0, 'profit': 0})
        for t in trades:
            quality = t.get('signal_quality', 'unknown')
            quality_stats[quality]['trades'] += 1
            quality_stats[quality]['wins'] += 1 if t['is_win'] else 0
            quality_stats[quality]['profit'] += t['profit']

        # Store stats
        self.data['stats'] = {
            'total_trades': len(trades),
            'wins': len(wins),
            'losses': len(losses),
            'win_rate': win_rate,
            'profit_factor': profit_factor,
            'total_profit': total_profit,
            'total_wins': total_wins,
            'total_losses': total_losses,
            'avg_win': avg_win,
            'avg_loss': avg_loss,
            'avg_trade': total_profit / len(trades) if trades else 0,
            'max_drawdown': max_drawdown,
            'current_equity': running_equity,
            'peak_equity': peak_equity,
            'sessions_tested': list(session_stats.keys()),
            'session_stats': dict(session_stats),
            'quality_stats': dict(quality_stats),
            'best_trade': max(trades, key=lambda t: t['profit'])['profit'] if trades else 0,
            'worst_trade': min(trades, key=lambda t: t['profit'])['profit'] if trades else 0,
            'avg_duration_minutes': sum(t['duration_minutes'] for t in trades) / len(trades) if trades else 0,
        }

    def get_stats(self) -> Dict:
        """Gibt aktuelle Statistiken zurück"""
        self._update_stats()
        return self.data['stats']

    # ==========================================
    # GO-LIVE READINESS CHECK
    # ==========================================

    def check_go_live_readiness(self) -> Tuple[bool, Dict]:
        """
        Prüft ob alle Kriterien für Go-Live erfüllt sind

        Returns:
            (is_ready, detailed_check_results)
        """
        self._update_stats()
        stats = self.data['stats']

        checks = {}

        # 1. Minimum Trades
        checks['min_trades'] = {
            'passed': stats['total_trades'] >= self.criteria['min_trades'],
            'current': stats['total_trades'],
            'required': self.criteria['min_trades'],
            'label': f"Trades: {stats['total_trades']}/{self.criteria['min_trades']}"
        }

        # 2. Win Rate
        checks['win_rate'] = {
            'passed': stats['win_rate'] >= self.criteria['min_win_rate'],
            'current': stats['win_rate'],
            'required': self.criteria['min_win_rate'],
            'label': f"Win Rate: {stats['win_rate']*100:.1f}% (min {self.criteria['min_win_rate']*100:.0f}%)"
        }

        # 3. Profit Factor
        pf = stats['profit_factor'] if stats['profit_factor'] != float('inf') else 999
        checks['profit_factor'] = {
            'passed': pf >= self.criteria['min_profit_factor'],
            'current': pf,
            'required': self.criteria['min_profit_factor'],
            'label': f"Profit Factor: {pf:.2f} (min {self.criteria['min_profit_factor']:.1f})"
        }

        # 4. Max Drawdown
        checks['max_drawdown'] = {
            'passed': stats['max_drawdown'] <= self.criteria['max_drawdown'],
            'current': stats['max_drawdown'],
            'required': self.criteria['max_drawdown'],
            'label': f"Max Drawdown: {stats['max_drawdown']*100:.1f}% (max {self.criteria['max_drawdown']*100:.0f}%)"
        }

        # 5. Days Running
        days = self._days_running()
        checks['min_days'] = {
            'passed': days >= self.criteria['min_days'],
            'current': days,
            'required': self.criteria['min_days'],
            'label': f"Days Running: {days}/{self.criteria['min_days']}"
        }

        # 6. Errors
        error_count = len(self.data['errors'])
        checks['max_errors'] = {
            'passed': error_count <= self.criteria['max_errors'],
            'current': error_count,
            'required': self.criteria['max_errors'],
            'label': f"Errors: {error_count} (max {self.criteria['max_errors']})"
        }

        # 7. Sessions Tested
        sessions = len(stats.get('sessions_tested', []))
        checks['sessions_tested'] = {
            'passed': sessions >= self.criteria['min_sessions_tested'],
            'current': sessions,
            'required': self.criteria['min_sessions_tested'],
            'label': f"Sessions Tested: {sessions}/{self.criteria['min_sessions_tested']}"
        }

        # Overall
        all_passed = all(c['passed'] for c in checks.values())
        passed_count = sum(1 for c in checks.values() if c['passed'])

        return all_passed, {
            'checks': checks,
            'passed_count': passed_count,
            'total_checks': len(checks),
            'is_ready': all_passed
        }

    # ==========================================
    # REPORTS
    # ==========================================

    def print_report(self):
        """Druckt einen vollständigen Report"""
        self._update_stats()
        stats = self.data['stats']

        print("\n" + "=" * 70)
        print("📊 DEMO TEST TRACKER - PERFORMANCE REPORT")
        print("=" * 70)

        print(f"\n📅 Demo Period:")
        print(f"   Start:         {self.data['start_date'][:10]}")
        print(f"   Days Running:  {self._days_running()}")
        print(f"   Last Trade:    {self.data.get('last_updated', 'N/A')[:10] if self.data.get('last_updated') else 'N/A'}")

        print(f"\n📈 PERFORMANCE METRICS:")
        print(f"   Total Trades:  {stats.get('total_trades', 0)}")
        print(f"   Wins/Losses:   {stats.get('wins', 0)}/{stats.get('losses', 0)}")
        print(f"   Win Rate:      {stats.get('win_rate', 0)*100:.1f}%")
        print(f"   Profit Factor: {stats.get('profit_factor', 0):.2f}")

        print(f"\n💰 PROFIT/LOSS:")
        print(f"   Total Profit:  ${stats.get('total_profit', 0):,.2f}")
        print(f"   Avg Win:       ${stats.get('avg_win', 0):,.2f}")
        print(f"   Avg Loss:      ${stats.get('avg_loss', 0):,.2f}")
        print(f"   Avg Trade:     ${stats.get('avg_trade', 0):,.2f}")
        print(f"   Best Trade:    ${stats.get('best_trade', 0):,.2f}")
        print(f"   Worst Trade:   ${stats.get('worst_trade', 0):,.2f}")

        print(f"\n📉 RISK METRICS:")
        print(f"   Max Drawdown:  {stats.get('max_drawdown', 0)*100:.1f}%")
        print(f"   Current Equity: ${stats.get('current_equity', 0):,.2f}")
        print(f"   Peak Equity:   ${stats.get('peak_equity', 0):,.2f}")

        print(f"\n⏱️ TIMING:")
        print(f"   Avg Duration:  {stats.get('avg_duration_minutes', 0):.0f} min")

        # Session breakdown
        session_stats = stats.get('session_stats', {})
        if session_stats:
            print(f"\n🌍 SESSION BREAKDOWN:")
            for session, data in session_stats.items():
                wr = data['wins'] / data['trades'] * 100 if data['trades'] > 0 else 0
                print(f"   {session.upper():10} | Trades: {data['trades']:3} | Win Rate: {wr:5.1f}% | Profit: ${data['profit']:,.2f}")

        # Signal quality breakdown
        quality_stats = stats.get('quality_stats', {})
        if quality_stats:
            print(f"\n🎯 SIGNAL QUALITY BREAKDOWN:")
            for quality, data in quality_stats.items():
                wr = data['wins'] / data['trades'] * 100 if data['trades'] > 0 else 0
                print(f"   {quality.upper():10} | Trades: {data['trades']:3} | Win Rate: {wr:5.1f}% | Profit: ${data['profit']:,.2f}")

        # Errors
        if self.data['errors']:
            print(f"\n⚠️ ERRORS ({len(self.data['errors'])} total):")
            for err in self.data['errors'][-5:]:  # Last 5
                print(f"   {err['timestamp'][:10]} | {err['type']}: {err['message'][:50]}")

        # Filtered signals
        filtered = len(self.data.get('filtered_signals', []))
        if filtered > 0:
            print(f"\n🚫 Filtered Signals: {filtered}")

        print("\n" + "=" * 70)

    def print_go_live_check(self):
        """Druckt den Go-Live Readiness Check"""
        is_ready, results = self.check_go_live_readiness()

        print("\n" + "=" * 70)
        print("🚦 GO-LIVE READINESS CHECK")
        print("=" * 70)

        for name, check in results['checks'].items():
            emoji = "✅" if check['passed'] else "⬜"
            print(f"   {emoji} {check['label']}")

        print("-" * 70)
        print(f"   Passed: {results['passed_count']}/{results['total_checks']}")

        if is_ready:
            print("\n   🎉 STATUS: READY FOR GO-LIVE!")
            print("   ✅ All criteria met. You can start with small real money.")
        else:
            remaining = [c['label'] for c in results['checks'].values() if not c['passed']]
            print(f"\n   ⏳ STATUS: NOT READY YET")
            print(f"   Still needed:")
            for r in remaining:
                print(f"      • {r}")

        print("=" * 70)

        return is_ready

    def get_daily_summary(self) -> str:
        """Generiert eine tägliche Zusammenfassung"""
        self._update_stats()
        stats = self.data['stats']

        today = datetime.now().date().isoformat()
        today_trades = [t for t in self.data['trades']
                       if t['exit_time'].startswith(today)]

        today_profit = sum(t['profit'] for t in today_trades)
        today_wins = sum(1 for t in today_trades if t['is_win'])
        today_wr = today_wins / len(today_trades) * 100 if today_trades else 0

        summary = f"""
📊 Daily Summary - {today}
━━━━━━━━━━━━━━━━━━━━━━━━━━━
Today:    {len(today_trades)} trades | {today_wins} wins | WR: {today_wr:.0f}% | P/L: ${today_profit:+.2f}
Overall:  {stats['total_trades']} trades | WR: {stats['win_rate']*100:.1f}% | Total: ${stats['total_profit']:+.2f}
━━━━━━━━━━━━━━━━━━━━━━━━━━━
"""
        return summary

    # ==========================================
    # HELPER METHODS
    # ==========================================

    def _days_running(self) -> int:
        """Berechnet Tage seit Start"""
        start = datetime.fromisoformat(self.data['start_date'])
        return (datetime.now() - start).days

    def _load_data(self) -> Dict:
        """Lädt Daten aus Datei"""
        default_data = {
            'start_date': datetime.now().isoformat(),
            'last_updated': None,
            'trades': [],
            'errors': [],
            'filtered_signals': [],
            'stats': {}
        }

        try:
            if os.path.exists(self.data_file):
                with open(self.data_file, 'r') as f:
                    data = json.load(f)
                    # Merge with defaults for any missing keys
                    for key in default_data:
                        if key not in data:
                            data[key] = default_data[key]
                    return data
        except Exception as e:
            logger.warning(f"Could not load demo tracker data: {e}")

        return default_data

    def _save_data(self):
        """Speichert Daten in Datei"""
        try:
            with open(self.data_file, 'w') as f:
                json.dump(self.data, f, indent=2, default=str)
        except Exception as e:
            logger.error(f"Could not save demo tracker data: {e}")

    def reset(self):
        """Setzt alle Daten zurück (Vorsicht!)"""
        self.data = {
            'start_date': datetime.now().isoformat(),
            'last_updated': None,
            'trades': [],
            'errors': [],
            'filtered_signals': [],
            'stats': {}
        }
        self._save_data()
        logger.warning("⚠️ Demo tracker data has been reset!")


# ==========================================
# STANDALONE USAGE
# ==========================================

if __name__ == "__main__":
    print("📊 Demo Test Tracker - Demo Mode")
    print("=" * 50)

    tracker = DemoTestTracker(data_file="demo_test_example.json")

    # Simulate some trades
    from datetime import timedelta

    base_time = datetime.now() - timedelta(days=10)

    test_trades = [
        {"profit": 15.50, "direction": "LONG", "session": "ny"},
        {"profit": -8.20, "direction": "SHORT", "session": "ny"},
        {"profit": 22.30, "direction": "LONG", "session": "asian"},
        {"profit": 12.10, "direction": "LONG", "session": "ny"},
        {"profit": -5.50, "direction": "SHORT", "session": "asian"},
        {"profit": 18.90, "direction": "LONG", "session": "london"},
        {"profit": -12.30, "direction": "LONG", "session": "ny"},
        {"profit": 25.60, "direction": "SHORT", "session": "asian"},
        {"profit": 8.40, "direction": "LONG", "session": "ny"},
        {"profit": -3.20, "direction": "SHORT", "session": "london"},
    ]

    for i, t in enumerate(test_trades):
        entry_time = base_time + timedelta(hours=i*8)
        exit_time = entry_time + timedelta(minutes=45)

        tracker.log_trade(
            ticket=1000 + i,
            symbol="XAUUSD",
            direction=t["direction"],
            entry_price=2850.00,
            exit_price=2850.00 + (t["profit"] / 0.10),  # Reverse calculate
            volume=0.10,
            profit=t["profit"],
            entry_time=entry_time,
            exit_time=exit_time,
            session=t["session"],
            base_confidence=75.0,
            enhanced_score=68.0,
            hybrid_score=72.2,
            signal_quality="good"
        )

    # Print reports
    tracker.print_report()
    tracker.print_go_live_check()

    # Cleanup
    os.remove("demo_test_example.json")