cbazzaandClaude Sonnet 4.6
dd2611eaef
Files were edited but not staged in earlier commits: - adaptive_rhythm_manager.py: mt→mt5, pytz→timezone, get_volatility_level, shutdown() - check_market_regime.py: ADX_THRESHOLD, Wilder EWM, try/finally, UTC timestamp, sys import - check_system_status.py: remove duplicate cursor.execute - drawdown_protection.py: float(inf), persist pause state, DB save_setting, Markdown fix - performance_analysis.py: KeyError export fix, profit factor, drawdown positive, SQL filter - performance_analysis_simple.py: fromisoformat, numeric bin sort, profit factor - trading_dashboard.py: st.rerun(), session_state auto-refresh, pathlib DB path, errors=coerce Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
482 lines
15 KiB
Python
482 lines
15 KiB
Python
#!/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)
|
|
return conn
|
|
|
|
# ==========================================
|
|
# DATA LOADING
|
|
# ==========================================
|
|
|
|
def load_closed_trades(conn, exclude_historical=True):
|
|
"""Lade geschlossene Trades"""
|
|
# 'historical' is a distinct status for imported legacy trades.
|
|
# exclude_historical=True → only live bot trades (status='closed')
|
|
# exclude_historical=False → live + historical imports
|
|
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
|
|
"""
|
|
|
|
df = pd.read_sql_query(query, conn)
|
|
|
|
if not df.empty:
|
|
df['entry_time'] = pd.to_datetime(df['entry_time'], errors='coerce')
|
|
df['exit_time'] = pd.to_datetime(df['exit_time'], errors='coerce')
|
|
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
|
|
|
|
gross_profit = df[df['win']]['net_profit'].sum()
|
|
gross_loss = abs(df[~df['win']]['net_profit'].sum())
|
|
profit_factor = round(gross_profit / gross_loss, 2) if gross_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 = abs(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()
|
|
|
|
results = {
|
|
'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
|
|
}
|
|
|
|
return results
|
|
|
|
# ==========================================
|
|
# 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', encoding='utf-8') 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)
|