cbazzaandClaude Sonnet 4.6
f37e7adcf3
enhanced_trailing_stop.py: - mt -> mt5 (all occurrences) - Add pandas + timezone imports at file top - Fix UTC bug: datetime.fromtimestamp(..., tz=timezone.utc).replace(tzinfo=None) - Fetch symbol_info once per call, reuse for point (was called twice) - cleanup_closed_positions: handle None from positions_get() - Remove pandas import from inside function body dynamic_threshold_optimizer.py: - Fix SQL injection: replace f-string session filter with parameterized query (?) - Add logging module, replace all print() with logger calls - Use context manager (with sqlite3.connect()) to prevent connection leak on exception - save_thresholds_to_config: add try/except with logger.error signal_cache.py: - Fix bare except -> except Exception in _cleanup_old_entries Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
440 lines
14 KiB
Python
440 lines
14 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
🎯 Dynamic Confidence Threshold Optimizer
|
|
Passt Confidence Threshold automatisch basierend auf Performance an
|
|
|
|
FEATURES:
|
|
1. Win Rate Tracking (letzte N Trades)
|
|
2. Automatische Threshold-Anpassung
|
|
3. Session-spezifische Optimization
|
|
4. Performance-basiertes Learning
|
|
"""
|
|
|
|
import sqlite3
|
|
import pandas as pd
|
|
import logging
|
|
from datetime import datetime, timedelta
|
|
from typing import Dict, Optional, Tuple
|
|
import json
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class DynamicThresholdOptimizer:
|
|
"""
|
|
Optimiert Confidence Threshold basierend auf tatsächlicher Performance
|
|
|
|
Logik:
|
|
- Hohe Win Rate → Senke Threshold (mehr Trades)
|
|
- Niedrige Win Rate → Erhöhe Threshold (nur beste Setups)
|
|
- Adaptiert sich automatisch an Marktbedingungen
|
|
"""
|
|
|
|
def __init__(self,
|
|
db_path: str = "trading_bot.db",
|
|
lookback_trades: int = 20,
|
|
target_win_rate: float = 0.60,
|
|
min_threshold: int = 60,
|
|
max_threshold: int = 95,
|
|
adjustment_step: int = 5):
|
|
"""
|
|
Args:
|
|
db_path: Pfad zur Trading Database
|
|
lookback_trades: Wie viele Trades für Berechnung (default: 20)
|
|
target_win_rate: Ziel Win Rate (default: 60%)
|
|
min_threshold: Minimum Confidence Threshold (default: 60%)
|
|
max_threshold: Maximum Confidence Threshold (default: 95%)
|
|
adjustment_step: Schritte für Anpassung (default: 5%)
|
|
"""
|
|
self.db_path = db_path
|
|
self.lookback_trades = lookback_trades
|
|
self.target_win_rate = target_win_rate
|
|
self.min_threshold = min_threshold
|
|
self.max_threshold = max_threshold
|
|
self.adjustment_step = adjustment_step
|
|
|
|
# Cache für Session-spezifische Thresholds
|
|
self.session_thresholds = {
|
|
'asian': 70,
|
|
'ny': 70,
|
|
'london': 70,
|
|
'overlap': 70
|
|
}
|
|
|
|
logger.info(f"Dynamic Threshold Optimizer initialized — "
|
|
f"lookback={lookback_trades}, target_wr={target_win_rate*100:.0f}%, "
|
|
f"range={min_threshold}%-{max_threshold}%")
|
|
|
|
def get_recent_performance(self, session: Optional[str] = None) -> Dict:
|
|
"""
|
|
Holt Performance der letzten N Trades
|
|
|
|
Args:
|
|
session: Optional - nur für diese Session (asian/ny/london/overlap)
|
|
|
|
Returns:
|
|
Dict mit Performance-Metriken
|
|
"""
|
|
try:
|
|
base_query = """
|
|
SELECT
|
|
confidence,
|
|
session,
|
|
net_profit,
|
|
CASE WHEN net_profit > 0 THEN 1 ELSE 0 END as win
|
|
FROM trades
|
|
WHERE status = 'closed'
|
|
"""
|
|
params: list = []
|
|
if session:
|
|
base_query += " AND session = ?"
|
|
params.append(session)
|
|
base_query += " ORDER BY exit_time DESC LIMIT ?"
|
|
params.append(self.lookback_trades)
|
|
|
|
with sqlite3.connect(self.db_path) as conn:
|
|
df = pd.read_sql_query(base_query, conn, params=params)
|
|
|
|
if df.empty:
|
|
return {
|
|
'trades': 0,
|
|
'win_rate': 0.0,
|
|
'avg_confidence': 0.0,
|
|
'total_profit': 0.0,
|
|
'recommendation': 'insufficient_data'
|
|
}
|
|
|
|
trades = len(df)
|
|
wins = int(df['win'].sum())
|
|
win_rate = wins / trades if trades > 0 else 0.0
|
|
avg_confidence = float(df['confidence'].mean())
|
|
total_profit = float(df['net_profit'].sum())
|
|
|
|
return {
|
|
'trades': trades,
|
|
'wins': wins,
|
|
'losses': trades - wins,
|
|
'win_rate': win_rate,
|
|
'avg_confidence': avg_confidence,
|
|
'total_profit': total_profit,
|
|
'recommendation': self._get_recommendation(win_rate)
|
|
}
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error getting performance: {e}")
|
|
return {
|
|
'trades': 0,
|
|
'win_rate': 0.0,
|
|
'avg_confidence': 0.0,
|
|
'total_profit': 0.0,
|
|
'recommendation': 'error'
|
|
}
|
|
|
|
def _get_recommendation(self, win_rate: float) -> str:
|
|
"""Gibt Empfehlung basierend auf Win Rate"""
|
|
if win_rate >= 0.70:
|
|
return "excellent" # Sehr gut - kann aggressiver werden
|
|
elif win_rate >= 0.60:
|
|
return "good" # Gut - am Target
|
|
elif win_rate >= 0.50:
|
|
return "moderate" # OK - leicht konservativer
|
|
elif win_rate >= 0.40:
|
|
return "poor" # Schlecht - deutlich konservativer
|
|
else:
|
|
return "critical" # Kritisch - sehr konservativ
|
|
|
|
def calculate_optimal_threshold(self, session: Optional[str] = None) -> Tuple[int, str]:
|
|
"""
|
|
Berechnet optimalen Confidence Threshold
|
|
|
|
Args:
|
|
session: Optional - für diese Session
|
|
|
|
Returns:
|
|
(optimal_threshold, reason)
|
|
"""
|
|
perf = self.get_recent_performance(session)
|
|
|
|
if perf['trades'] < 10:
|
|
return (70, f"Insufficient data ({perf['trades']} trades), using default 70%")
|
|
|
|
current_threshold = self.session_thresholds.get(session, 70) if session else 70
|
|
win_rate = perf['win_rate']
|
|
|
|
# Berechne Anpassung basierend auf Win Rate
|
|
if win_rate >= 0.70:
|
|
# Exzellent - senke Threshold für mehr Trades
|
|
adjustment = -self.adjustment_step * 2 # -10%
|
|
reason = f"Excellent WR {win_rate*100:.1f}% → Lower threshold for more trades"
|
|
|
|
elif win_rate >= 0.65:
|
|
# Sehr gut - leicht senken
|
|
adjustment = -self.adjustment_step # -5%
|
|
reason = f"Very good WR {win_rate*100:.1f}% → Slightly lower threshold"
|
|
|
|
elif win_rate >= 0.55:
|
|
# Gut - bleibe oder leicht senken
|
|
adjustment = 0
|
|
reason = f"Good WR {win_rate*100:.1f}% → Maintain threshold"
|
|
|
|
elif win_rate >= 0.50:
|
|
# OK - leicht erhöhen
|
|
adjustment = self.adjustment_step # +5%
|
|
reason = f"Moderate WR {win_rate*100:.1f}% → Slightly raise threshold"
|
|
|
|
elif win_rate >= 0.40:
|
|
# Schlecht - deutlich erhöhen
|
|
adjustment = self.adjustment_step * 2 # +10%
|
|
reason = f"Poor WR {win_rate*100:.1f}% → Raise threshold significantly"
|
|
|
|
else:
|
|
# Kritisch - stark erhöhen
|
|
adjustment = self.adjustment_step * 3 # +15%
|
|
reason = f"Critical WR {win_rate*100:.1f}% → Raise threshold aggressively"
|
|
|
|
# Neuer Threshold
|
|
new_threshold = current_threshold + adjustment
|
|
|
|
# Clamp zu min/max
|
|
new_threshold = max(self.min_threshold, min(self.max_threshold, new_threshold))
|
|
|
|
return (new_threshold, reason)
|
|
|
|
def update_session_threshold(self, session: str) -> Dict:
|
|
"""
|
|
Updated Threshold für eine Session
|
|
|
|
Args:
|
|
session: Session Name (asian/ny/london/overlap)
|
|
|
|
Returns:
|
|
Dict mit Update-Info
|
|
"""
|
|
old_threshold = self.session_thresholds.get(session, 70)
|
|
new_threshold, reason = self.calculate_optimal_threshold(session)
|
|
|
|
self.session_thresholds[session] = new_threshold
|
|
|
|
perf = self.get_recent_performance(session)
|
|
|
|
return {
|
|
'session': session,
|
|
'old_threshold': old_threshold,
|
|
'new_threshold': new_threshold,
|
|
'change': new_threshold - old_threshold,
|
|
'reason': reason,
|
|
'recent_trades': perf['trades'],
|
|
'win_rate': perf['win_rate'],
|
|
'avg_confidence': perf['avg_confidence'],
|
|
'total_profit': perf['total_profit']
|
|
}
|
|
|
|
def optimize_all_sessions(self) -> Dict[str, Dict]:
|
|
"""
|
|
Optimiert Thresholds für alle Sessions
|
|
|
|
Returns:
|
|
Dict mit Updates für jede Session
|
|
"""
|
|
results = {}
|
|
|
|
for session in ['asian', 'ny', 'london', 'overlap']:
|
|
results[session] = self.update_session_threshold(session)
|
|
|
|
return results
|
|
|
|
def get_threshold_for_session(self, session: str) -> int:
|
|
"""
|
|
Holt aktuellen Threshold für Session
|
|
|
|
Args:
|
|
session: Session Name
|
|
|
|
Returns:
|
|
Confidence Threshold (%)
|
|
"""
|
|
return self.session_thresholds.get(session, 70)
|
|
|
|
def generate_report(self) -> str:
|
|
"""
|
|
Erstellt Optimization Report
|
|
|
|
Returns:
|
|
Formatted Report String
|
|
"""
|
|
report = []
|
|
report.append("=" * 70)
|
|
report.append("🎯 DYNAMIC THRESHOLD OPTIMIZATION REPORT")
|
|
report.append("=" * 70)
|
|
report.append(f"\nGenerated: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
|
|
report.append(f"Lookback: {self.lookback_trades} trades")
|
|
report.append(f"Target Win Rate: {self.target_win_rate*100:.1f}%")
|
|
report.append("")
|
|
|
|
for session in ['asian', 'ny', 'london', 'overlap']:
|
|
perf = self.get_recent_performance(session)
|
|
threshold = self.session_thresholds[session]
|
|
|
|
if perf['trades'] < 5:
|
|
continue
|
|
|
|
report.append(f"\n{'='*70}")
|
|
report.append(f"📊 {session.upper()} SESSION")
|
|
report.append(f"{'='*70}")
|
|
report.append(f"Recent Trades: {perf['trades']}")
|
|
report.append(f"Win Rate: {perf['win_rate']*100:.1f}% ({perf['wins']}W / {perf['losses']}L)")
|
|
report.append(f"Avg Confidence: {perf['avg_confidence']:.1f}%")
|
|
report.append(f"Total Profit: ${perf['total_profit']:.2f}")
|
|
report.append(f"Performance: {perf['recommendation'].upper()}")
|
|
report.append(f"\nCurrent Threshold: {threshold}%")
|
|
|
|
# Recommendation
|
|
new_threshold, reason = self.calculate_optimal_threshold(session)
|
|
if new_threshold != threshold:
|
|
change = new_threshold - threshold
|
|
emoji = "🔽" if change < 0 else "🔼"
|
|
report.append(f"Recommended: {new_threshold}% ({emoji} {abs(change):+d}%)")
|
|
report.append(f"Reason: {reason}")
|
|
else:
|
|
report.append(f"Recommended: Keep at {threshold}% ✅")
|
|
|
|
report.append("\n" + "=" * 70)
|
|
report.append("✅ Optimization Complete")
|
|
report.append("=" * 70)
|
|
|
|
return "\n".join(report)
|
|
|
|
def save_thresholds_to_config(self, config_file: str = "dynamic_thresholds.json"):
|
|
"""
|
|
Speichert optimierte Thresholds in JSON-Datei
|
|
|
|
Args:
|
|
config_file: Output Datei
|
|
"""
|
|
config = {
|
|
'timestamp': datetime.now().isoformat(),
|
|
'session_thresholds': self.session_thresholds,
|
|
'settings': {
|
|
'lookback_trades': self.lookback_trades,
|
|
'target_win_rate': self.target_win_rate,
|
|
'min_threshold': self.min_threshold,
|
|
'max_threshold': self.max_threshold
|
|
}
|
|
}
|
|
|
|
try:
|
|
with open(config_file, 'w') as f:
|
|
json.dump(config, f, indent=2)
|
|
logger.info(f"Thresholds saved to: {config_file}")
|
|
except Exception as e:
|
|
logger.error(f"Failed to save thresholds: {e}")
|
|
|
|
|
|
# ==========================================
|
|
# AUTO-OPTIMIZATION SCHEDULER
|
|
# ==========================================
|
|
|
|
def auto_optimize_thresholds(optimizer: DynamicThresholdOptimizer,
|
|
apply_changes: bool = False) -> Dict:
|
|
"""
|
|
Automatische Optimization (für Scheduler)
|
|
|
|
Args:
|
|
optimizer: DynamicThresholdOptimizer Instanz
|
|
apply_changes: Wenn True, werden Änderungen angewendet
|
|
|
|
Returns:
|
|
Optimization Results
|
|
"""
|
|
logger.info(f"\n{'='*70}")
|
|
logger.info(f"🔄 AUTO-OPTIMIZATION STARTED - {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
|
|
logger.info(f"{'='*70}\n")
|
|
|
|
results = optimizer.optimize_all_sessions()
|
|
|
|
# Print Summary
|
|
for session, info in results.items():
|
|
if info['recent_trades'] < 5:
|
|
continue
|
|
|
|
change_emoji = "🔽" if info['change'] < 0 else ("🔼" if info['change'] > 0 else "➡️")
|
|
|
|
logger.info(f"{session.upper():8s}: {info['old_threshold']}% → {info['new_threshold']}% "
|
|
f"{change_emoji} | WR: {info['win_rate']*100:.1f}% ({info['recent_trades']} trades)")
|
|
|
|
if apply_changes:
|
|
optimizer.save_thresholds_to_config()
|
|
logger.info("\n✅ Changes applied and saved!")
|
|
else:
|
|
logger.info("\n⚠️ Dry-run mode - changes NOT applied")
|
|
|
|
logger.info(f"\n{'='*70}\n")
|
|
|
|
return results
|
|
|
|
|
|
# ==========================================
|
|
# USAGE EXAMPLE
|
|
# ==========================================
|
|
|
|
"""
|
|
INTEGRATION IN NOTEBOOK:
|
|
|
|
# Cell: Setup Dynamic Optimizer
|
|
|
|
from dynamic_threshold_optimizer import DynamicThresholdOptimizer, auto_optimize_thresholds
|
|
|
|
# Initialize Optimizer
|
|
threshold_optimizer = DynamicThresholdOptimizer(
|
|
db_path="trading_bot.db",
|
|
lookback_trades=20, # Letzte 20 Trades
|
|
target_win_rate=0.60, # 60% Target
|
|
min_threshold=60, # Minimum 60%
|
|
max_threshold=95 # Maximum 95%
|
|
)
|
|
|
|
print("✅ Dynamic Threshold Optimizer activated!")
|
|
|
|
|
|
# Cell: Manual Optimization (run when you want)
|
|
|
|
# Generate Report
|
|
print(threshold_optimizer.generate_report())
|
|
|
|
# Apply Optimization
|
|
results = auto_optimize_thresholds(threshold_optimizer, apply_changes=True)
|
|
|
|
|
|
# Cell: Use in Trading Logic
|
|
|
|
# Get optimized threshold for current session
|
|
session = rhythm_manager.get_current_session()
|
|
optimal_threshold = threshold_optimizer.get_threshold_for_session(session)
|
|
|
|
print(f"Using threshold: {optimal_threshold}% for {session.upper()} session")
|
|
|
|
# Use in execute_trade_v2_adaptive
|
|
execute_trade_v2_adaptive(
|
|
symbol="XAUUSD",
|
|
base_confidence=optimal_threshold, # ← Dynamic!
|
|
...
|
|
)
|
|
|
|
|
|
# Cell: Add to Scheduler (auto-optimize daily)
|
|
|
|
scheduler.add_job(
|
|
func=lambda: auto_optimize_thresholds(threshold_optimizer, apply_changes=True),
|
|
trigger='cron',
|
|
hour=0, # Run at midnight
|
|
id='threshold_optimization'
|
|
)
|
|
|
|
print("✅ Auto-optimization scheduled (daily at midnight)")
|
|
"""
|
|
|
|
if __name__ == "__main__":
|
|
# Test
|
|
optimizer = DynamicThresholdOptimizer()
|
|
logger.info(optimizer.generate_report())
|