#!/usr/bin/env python3 """ TradingBot V1.6 MT5 Profitability Analyzer Holt Trade-Historie aus MT5 und berechnet echte Performance """ import MetaTrader5 as mt import json import glob from datetime import datetime, timedelta from collections import defaultdict import keyring as kr class MT5ProfitabilityAnalyzer: """Analysiert echte Trading-Performance aus MT5""" def __init__(self, strategy_name="TradingBot_V1.6", symbol="XAUUSD"): self.strategy_name = strategy_name self.symbol = symbol self.json_trades = [] self.mt5_deals = [] self.mt5_positions = [] self.matched_trades = [] self.stats = {} def connect_mt5(self): """Verbindet zu MT5""" print("🔌 Verbinde zu MT5...") if not mt.initialize(): print(f"❌ MT5 Initialisierung fehlgeschlagen: {mt.last_error()}") return False # Login (wie im Bot) login = 10800246 server = 'VantageInternational-Demo' password = kr.get_password(server, str(login)) if not mt.login(login, password, server): print(f"❌ MT5 Login fehlgeschlagen: {mt.last_error()}") return False account_info = mt.account_info() if account_info: print(f"✅ MT5 verbunden") print(f" Account: {account_info.login}") print(f" Balance: ${account_info.balance:.2f}") print(f" Equity: ${account_info.equity:.2f}") print(f" Profit: ${account_info.profit:.2f}") return True def load_json_trades(self, json_pattern="trade_performance_v16_XAUUSD_*.json"): """Lädt JSON Trade-Daten""" print(f"\n📂 Lade JSON Trade-Daten...") json_files = glob.glob(json_pattern) if not json_files: print(f"❌ Keine JSON-Files gefunden") return False all_trades = [] for file in json_files: try: with open(file, 'r') as f: data = json.load(f) all_trades.extend(data) except Exception as e: print(f"⚠️ Fehler beim Laden von {file}: {e}") if not all_trades: return False self.json_trades = all_trades print(f"✅ {len(all_trades)} JSON Trades geladen") return True def fetch_mt5_history(self, days_back=30): """Holt Trade-Historie aus MT5""" print(f"\n📊 Hole MT5 Trade-Historie (letzte {days_back} Tage)...") # Zeitraum date_to = datetime.now() date_from = date_to - timedelta(days=days_back) # Hole ALLE Deals für das Symbol (nicht nur mit Comment) all_deals = mt.history_deals_get(date_from, date_to, symbol=self.symbol) if all_deals is None: print(f"❌ Keine Deals gefunden: {mt.last_error()}") return False print(f"📊 {len(all_deals)} Total Deals für {self.symbol} gefunden") # Filtere nach Strategy strategy_deals = [ deal for deal in all_deals if self.strategy_name in deal.comment ] # Zähle Entry vs Exit Deals entry_deals = [d for d in strategy_deals if d.entry == 0] # IN exit_deals = [d for d in strategy_deals if d.entry == 1] # OUT self.mt5_deals = strategy_deals print(f"✅ {len(strategy_deals)} Strategy Deals gefunden:") print(f" - {len(entry_deals)} Entry Deals (entry=0)") print(f" - {len(exit_deals)} Exit Deals (entry=1)") # Wenn keine Exit Deals, dann sind Positionen noch offen ODER # sie wurden per SL/TP geschlossen (anderer Comment?) if len(exit_deals) == 0: print(f"\n⚠️ KEINE Exit Deals gefunden!") print(f" Das bedeutet: Positionen wurden per SL/TP geschlossen,") print(f" aber Exit-Deals haben anderen Comment (nicht '{self.strategy_name}')") print(f"\n🔍 Prüfe alle Deals für Position-IDs...") # Sammle alle Position IDs aus Entry Deals entry_position_ids = set(d.position_id for d in entry_deals) # Suche ALLE Deals mit diesen Position IDs (auch ohne Strategy Comment) all_position_deals = [ d for d in all_deals if d.position_id in entry_position_ids ] print(f"✅ {len(all_position_deals)} Deals für diese Position-IDs gefunden") # Verwende ALLE Deals für diese Positionen self.mt5_deals = all_position_deals # Neu zählen entry_deals = [d for d in all_position_deals if d.entry == 0] exit_deals = [d for d in all_position_deals if d.entry == 1] print(f" - {len(entry_deals)} Entry Deals") print(f" - {len(exit_deals)} Exit Deals") # Hole auch geschlossene Positionen positions_history = mt.history_orders_get(date_from, date_to) if positions_history: strategy_positions = [ pos for pos in positions_history if pos.symbol == self.symbol and self.strategy_name in pos.comment ] self.mt5_positions = strategy_positions print(f"✅ {len(strategy_positions)} Order-Historie-Einträge gefunden") return True def match_trades(self): """Matched JSON Entry-Daten mit MT5 Exit-Daten""" print(f"\n🔗 Matche JSON Entries mit MT5 Exits...") matched = [] unmatched_json = [] # Gruppiere Deals nach Position ID deals_by_position = defaultdict(list) for deal in self.mt5_deals: deals_by_position[deal.position_id].append(deal) # Debug: Zeige erste Deals print(f"\n🔍 Debug: Erste 3 Deals:") for i, deal in enumerate(self.mt5_deals[:3]): print(f" Deal {i+1}:") print(f" ticket: {deal.ticket}") print(f" position_id: {deal.position_id}") print(f" entry: {deal.entry}") print(f" type: {deal.type}") print(f" price: {deal.price}") print(f" profit: {deal.profit}") # Extrahiere Deal-IDs aus JSON for json_trade in self.json_trades: order_str = json_trade.get('order_result', '') # Extrahiere Deal ID import re deal_match = re.search(r'deal=(\d+)', order_str) if not deal_match: unmatched_json.append(json_trade) continue entry_deal_id = int(deal_match.group(1)) # Finde Entry Deal in MT5 entry_deal = None for deal in self.mt5_deals: if deal.ticket == entry_deal_id: entry_deal = deal break if not entry_deal: unmatched_json.append(json_trade) continue # Finde zugehörigen Exit Deal # Ein Exit Deal hat: # - Gleiche position_id # - Andere ticket ID # - entry = 1 (OUT) statt 0 (IN) # - Späterer Zeitstempel position_id = entry_deal.position_id position_deals = deals_by_position[position_id] exit_deal = None if len(position_deals) >= 2: # Sortiere nach Zeit sorted_deals = sorted(position_deals, key=lambda d: d.time) # Entry sollte erster sein, Exit zweiter for deal in sorted_deals: if deal.ticket != entry_deal_id and deal.time > entry_deal.time: exit_deal = deal break # Erstelle Match-Entry match_entry = { 'json_trade': json_trade, 'entry_deal': entry_deal, 'exit_deal': exit_deal, 'is_closed': exit_deal is not None, 'entry_time': datetime.fromtimestamp(entry_deal.time), 'entry_price': entry_deal.price, 'entry_volume': entry_deal.volume, } if exit_deal: match_entry.update({ 'exit_time': datetime.fromtimestamp(exit_deal.time), 'exit_price': exit_deal.price, 'profit': exit_deal.profit, 'commission': exit_deal.commission, 'swap': exit_deal.swap, 'net_profit': exit_deal.profit + exit_deal.commission + exit_deal.swap, 'hold_time_hours': (datetime.fromtimestamp(exit_deal.time) - datetime.fromtimestamp(entry_deal.time)).total_seconds() / 3600, 'pips': abs(exit_deal.price - entry_deal.price), 'is_winner': exit_deal.profit > 0, }) matched.append(match_entry) self.matched_trades = matched print(f"\n✅ {len(matched)} Trades gematched") print(f" - {sum(1 for m in matched if m['is_closed'])} geschlossen") print(f" - {sum(1 for m in matched if not m['is_closed'])} noch offen") if unmatched_json: print(f"⚠️ {len(unmatched_json)} JSON Trades konnten nicht gematched werden") return True def calculate_performance_metrics(self): """Berechnet umfassende Performance-Metriken""" print(f"\n📈 Berechne Performance-Metriken...") closed_trades = [t for t in self.matched_trades if t['is_closed']] if not closed_trades: print("⚠️ Keine geschlossenen Trades gefunden!") return False # Basic Stats total_closed = len(closed_trades) winners = [t for t in closed_trades if t['is_winner']] losers = [t for t in closed_trades if not t['is_winner']] win_count = len(winners) loss_count = len(losers) win_rate = (win_count / total_closed * 100) if total_closed > 0 else 0 # P&L total_profit = sum(t['profit'] for t in closed_trades) total_commission = sum(t['commission'] for t in closed_trades) total_swap = sum(t['swap'] for t in closed_trades) net_profit = sum(t['net_profit'] for t in closed_trades) gross_profit = sum(t['profit'] for t in winners) if winners else 0 gross_loss = abs(sum(t['profit'] for t in losers)) if losers else 0 profit_factor = (gross_profit / gross_loss) if gross_loss > 0 else float('inf') # Average Trade avg_win = (sum(t['profit'] for t in winners) / win_count) if winners else 0 avg_loss = (sum(t['profit'] for t in losers) / loss_count) if losers else 0 avg_trade = net_profit / total_closed # Hold Time avg_hold_time = sum(t['hold_time_hours'] for t in closed_trades) / total_closed # Expectancy expectancy = (win_rate/100 * avg_win) + ((1 - win_rate/100) * avg_loss) # Drawdown Analyse cumulative_profits = [] running_profit = 0 for trade in sorted(closed_trades, key=lambda x: x['exit_time']): running_profit += trade['net_profit'] cumulative_profits.append(running_profit) peak = cumulative_profits[0] max_drawdown = 0 drawdown_pct = 0 for profit in cumulative_profits: if profit > peak: peak = profit drawdown = peak - profit if drawdown > max_drawdown: max_drawdown = drawdown drawdown_pct = (drawdown / peak * 100) if peak > 0 else 0 # Session Analysis session_performance = defaultdict(lambda: {'count': 0, 'profit': 0, 'wins': 0}) for trade in closed_trades: session = trade['json_trade'].get('session', 'unknown') session_performance[session]['count'] += 1 session_performance[session]['profit'] += trade['net_profit'] if trade['is_winner']: session_performance[session]['wins'] += 1 # Regime Analysis regime_performance = defaultdict(lambda: {'count': 0, 'profit': 0, 'wins': 0}) for trade in closed_trades: regime = trade['json_trade'].get('market_regime', 'unknown') regime_performance[regime]['count'] += 1 regime_performance[regime]['profit'] += trade['net_profit'] if trade['is_winner']: regime_performance[regime]['wins'] += 1 # Quality Analysis quality_performance = defaultdict(lambda: {'count': 0, 'profit': 0, 'wins': 0}) for trade in closed_trades: quality = trade['json_trade'].get('signal_quality', 'unknown') quality_performance[quality]['count'] += 1 quality_performance[quality]['profit'] += trade['net_profit'] if trade['is_winner']: quality_performance[quality]['wins'] += 1 # Best/Worst Trades best_trade = max(closed_trades, key=lambda x: x['profit']) worst_trade = min(closed_trades, key=lambda x: x['profit']) self.stats = { 'total_closed': total_closed, 'win_count': win_count, 'loss_count': loss_count, 'win_rate': win_rate, 'total_profit': total_profit, 'total_commission': total_commission, 'total_swap': total_swap, 'net_profit': net_profit, 'gross_profit': gross_profit, 'gross_loss': gross_loss, 'profit_factor': profit_factor, 'avg_win': avg_win, 'avg_loss': avg_loss, 'avg_trade': avg_trade, 'avg_hold_time': avg_hold_time, 'expectancy': expectancy, 'max_drawdown': max_drawdown, 'max_drawdown_pct': drawdown_pct, 'session_performance': dict(session_performance), 'regime_performance': dict(regime_performance), 'quality_performance': dict(quality_performance), 'best_trade': { 'profit': best_trade['profit'], 'entry_time': best_trade['entry_time'], 'session': best_trade['json_trade'].get('session'), }, 'worst_trade': { 'profit': worst_trade['profit'], 'entry_time': worst_trade['entry_time'], 'session': worst_trade['json_trade'].get('session'), }, 'cumulative_profits': cumulative_profits, } print("✅ Performance-Metriken berechnet") return True def print_profitability_report(self): """Druckt umfassenden Profitabilitäts-Report""" stats = self.stats print("\n" + "="*70) print("💰 TRADINGBOT V1.6 - PROFITABILITY REPORT") print("="*70) # Profitability Status is_profitable = stats['net_profit'] > 0 status_emoji = "✅" if is_profitable else "❌" status_text = "PROFITABEL" if is_profitable else "NICHT PROFITABEL" print(f"\n{status_emoji} STATUS: {status_text}") print(f" Net Profit: ${stats['net_profit']:.2f}") print(f"\n📊 TRADE STATISTICS:") print(f" Total Closed Trades: {stats['total_closed']}") print(f" Winners: {stats['win_count']} ({stats['win_rate']:.1f}%)") print(f" Losers: {stats['loss_count']} ({100-stats['win_rate']:.1f}%)") print(f"\n💵 PROFIT & LOSS:") print(f" Gross Profit: ${stats['gross_profit']:.2f}") print(f" Gross Loss: ${stats['gross_loss']:.2f}") print(f" Total Commission: ${stats['total_commission']:.2f}") print(f" Total Swap: ${stats['total_swap']:.2f}") print(f" Net Profit: ${stats['net_profit']:.2f}") print(f"\n📈 PERFORMANCE METRICS:") pf_display = f"{stats['profit_factor']:.2f}" if stats['profit_factor'] != float('inf') else "∞" print(f" Profit Factor: {pf_display}") print(f" Average Win: ${stats['avg_win']:.2f}") print(f" Average Loss: ${stats['avg_loss']:.2f}") print(f" Average Trade: ${stats['avg_trade']:.2f}") print(f" Expectancy: ${stats['expectancy']:.2f}") print(f"\n⏱️ TIMING:") print(f" Avg Hold Time: {stats['avg_hold_time']:.1f} hours") print(f"\n📉 RISK METRICS:") print(f" Max Drawdown: ${stats['max_drawdown']:.2f} ({stats['max_drawdown_pct']:.1f}%)") print(f"\n🏆 BEST TRADE:") best = stats['best_trade'] print(f" Profit: ${best['profit']:.2f}") print(f" Time: {best['entry_time'].strftime('%Y-%m-%d %H:%M')}") print(f" Session: {best['session']}") print(f"\n💔 WORST TRADE:") worst = stats['worst_trade'] print(f" Loss: ${worst['profit']:.2f}") print(f" Time: {worst['entry_time'].strftime('%Y-%m-%d %H:%M')}") print(f" Session: {worst['session']}") print(f"\n🌍 SESSION PERFORMANCE:") for session in ['asian', 'london', 'overlap', 'ny']: if session in stats['session_performance']: perf = stats['session_performance'][session] win_rate = (perf['wins'] / perf['count'] * 100) if perf['count'] > 0 else 0 profit_emoji = "✅" if perf['profit'] > 0 else "❌" print(f" {session.capitalize():8s}: {perf['count']:3d} trades | " f"${perf['profit']:7.2f} | Win Rate: {win_rate:5.1f}% {profit_emoji}") print(f"\n📈 REGIME PERFORMANCE:") for regime, perf in stats['regime_performance'].items(): win_rate = (perf['wins'] / perf['count'] * 100) if perf['count'] > 0 else 0 profit_emoji = "✅" if perf['profit'] > 0 else "❌" print(f" {regime.capitalize():10s}: {perf['count']:3d} trades | " f"${perf['profit']:7.2f} | Win Rate: {win_rate:5.1f}% {profit_emoji}") print(f"\n🎯 SIGNAL QUALITY PERFORMANCE:") for quality in ['excellent', 'good', 'fair']: if quality in stats['quality_performance']: perf = stats['quality_performance'][quality] win_rate = (perf['wins'] / perf['count'] * 100) if perf['count'] > 0 else 0 profit_emoji = "✅" if perf['profit'] > 0 else "❌" print(f" {quality.capitalize():10s}: {perf['count']:3d} trades | " f"${perf['profit']:7.2f} | Win Rate: {win_rate:5.1f}% {profit_emoji}") print("\n" + "="*70) # Interpretation print("\n💡 INTERPRETATION:") if is_profitable: print(" ✅ Die Strategie ist profitabel!") if stats['win_rate'] >= 50: print(" ✅ Gute Win-Rate") else: print(" ⚠️ Win-Rate unter 50% - Strategie profitiert von großen Wins") if stats['profit_factor'] >= 2.0: print(" ✅ Exzellenter Profit Factor (>=2.0)") elif stats['profit_factor'] >= 1.5: print(" ✅ Guter Profit Factor (>=1.5)") else: print(" ⚠️ Profit Factor könnte besser sein") else: print(" ❌ Die Strategie ist derzeit nicht profitabel") print(" ⚠️ Optimierung notwendig!") print("="*70) def save_results(self, output_file='profitability_analysis.json'): """Speichert Ergebnisse als JSON""" print(f"\n💾 Speichere Ergebnisse...") results = { 'analysis_date': datetime.now().isoformat(), 'strategy_name': self.strategy_name, 'symbol': self.symbol, 'statistics': { k: v for k, v in self.stats.items() if k not in ['best_trade', 'worst_trade', 'cumulative_profits'] }, 'best_trade': { 'profit': self.stats['best_trade']['profit'], 'entry_time': self.stats['best_trade']['entry_time'].isoformat(), 'session': self.stats['best_trade']['session'], }, 'worst_trade': { 'profit': self.stats['worst_trade']['profit'], 'entry_time': self.stats['worst_trade']['entry_time'].isoformat(), 'session': self.stats['worst_trade']['session'], }, } with open(output_file, 'w') as f: json.dump(results, f, indent=2) print(f"✅ Ergebnisse gespeichert: {output_file}") def disconnect_mt5(self): """Trennt MT5 Verbindung""" mt.shutdown() print("✅ MT5 Verbindung getrennt") def main(): """Haupt-Analyse""" print("="*70) print("💰 TradingBot V1.6 - MT5 Profitability Analyzer") print("="*70) analyzer = MT5ProfitabilityAnalyzer() # 1. Connect MT5 if not analyzer.connect_mt5(): return # 2. Load JSON Trades if not analyzer.load_json_trades(): analyzer.disconnect_mt5() return # 3. Fetch MT5 History if not analyzer.fetch_mt5_history(days_back=30): analyzer.disconnect_mt5() return # 4. Match Trades if not analyzer.match_trades(): analyzer.disconnect_mt5() return # 5. Calculate Performance if not analyzer.calculate_performance_metrics(): analyzer.disconnect_mt5() return # 6. Print Report analyzer.print_profitability_report() # 7. Save Results analyzer.save_results() # 8. Disconnect analyzer.disconnect_mt5() print("\n" + "="*70) print("✅ ANALYSE ABGESCHLOSSEN") print("="*70) print("\n📂 Generierte Files:") print(" - profitability_analysis.json") print("\n🎯 Nächste Schritte basierend auf Ergebnis:") print(" • Falls profitabel → SQLite + Telegram + Scaling") print(" • Falls nicht profitabel → Parameter-Optimierung + Backtesting") print("="*70) if __name__ == "__main__": main()