#!/usr/bin/env python3 """ ML SIGNAL QUALITY PREDICTOR XGBoost-basierter Predictor für Trade-Erfolgswahrscheinlichkeit Lernt aus historischen Trades und sagt voraus, ob ein Signal wahrscheinlich zu einem erfolgreichen Trade führt. VERWENDUNG: from ml_signal_predictor import MLSignalPredictor predictor = MLSignalPredictor() predictor.train() # Trainiert auf historischen Daten # Vorhersage für neues Signal result = predictor.predict(signal_features) print(f"Win Probability: {result['win_probability']:.1%}") """ import json import logging import os import pickle from datetime import datetime from typing import Dict, List, Optional, Tuple import warnings import numpy as np import pandas as pd # Suppress warnings for cleaner output # Scoped suppression only during XGBoost training, not globally logger = logging.getLogger(__name__) # ========================================== # CONFIGURATION # ========================================== ML_CONFIG = { 'enabled': False, # Standardmäßig AUS bis Model besser trainiert 'model_path': 'ml_model_xgboost.pkl', 'min_trades_for_training': 50, # Mindestens 50 Trades zum Trainieren 'retrain_interval_trades': 50, # Alle 50 neuen Trades retrainieren 'min_win_probability': 0.45, # Unter 45% = Trade blockieren (konservativ) 'reduce_threshold': 0.50, # Unter 50% = Lot reduzieren 'confidence_threshold': 0.55, # Ab 55% = Volle Lot-Size # XGBoost Parameters (konservativ für kleine Datasets) 'xgb_params': { 'max_depth': 3, 'n_estimators': 100, 'learning_rate': 0.1, 'reg_alpha': 1.0, 'reg_lambda': 1.0, 'min_child_weight': 10, 'subsample': 0.8, 'colsample_bytree': 0.8, 'objective': 'binary:logistic', 'eval_metric': 'auc', 'random_state': 42, 'use_label_encoder': False, } } # ========================================== # FEATURE DEFINITIONS # ========================================== FEATURE_COLUMNS = [ # Signal Features 'confidence', 'adaptive_threshold', 'regime_strength', 'risk_adjusted_strength_norm', # Session (One-Hot) 'session_asian', 'session_london', 'session_ny', 'session_overlap', # Direction 'direction_long', # Market Regime (One-Hot) 'regime_trending', 'regime_ranging', # Signal Quality (Ordinal) 'quality_score', # Time Features 'hour_sin', 'hour_cos', 'day_of_week', # Derived Features 'confidence_above_threshold', 'confidence_margin', ] # ========================================== # ML SIGNAL PREDICTOR CLASS # ========================================== class MLSignalPredictor: """XGBoost-basierter Signal Quality Predictor""" def __init__(self, demo_stats_file: str = 'demo_test_stats.json', performance_file: str = 'trade_performance_v16_XAUUSD_202601.json'): # NOTE: Both default files are excluded from git (runtime data). # Export them from the SQLite DB or provide a custom path before training. """ Initialisiert den ML Predictor Args: demo_stats_file: Pfad zur Demo-Stats Datei (Trade-Ergebnisse) performance_file: Pfad zur Performance Datei (Signal-Features) """ self.demo_stats_file = demo_stats_file self.performance_file = performance_file self.model = None self.feature_columns = FEATURE_COLUMNS self.is_trained = False self.training_stats = {} self.trades_since_training = 0 # Try to load existing model self._load_model() logger.info("=" * 60) logger.info("ML SIGNAL PREDICTOR INITIALIZED") logger.info("=" * 60) logger.info(f" Model Loaded: {self.is_trained}") logger.info(f" Min Trades for Training: {ML_CONFIG['min_trades_for_training']}") logger.info(f" Min Win Probability: {ML_CONFIG['min_win_probability']:.0%}") logger.info("=" * 60) # ========================================== # DATA LOADING & PREPROCESSING # ========================================== def _load_data(self) -> Tuple[pd.DataFrame, pd.DataFrame]: """Lädt Trade-Daten aus beiden Quellen""" # Load demo stats (trade results) try: with open(self.demo_stats_file, 'r') as f: demo_data = json.load(f) trades_df = pd.DataFrame(demo_data.get('trades', [])) logger.info(f"Loaded {len(trades_df)} trades from demo_stats") except FileNotFoundError: logger.warning(f"Demo stats file not found: {self.demo_stats_file}") trades_df = pd.DataFrame() # Load performance data (signal features) try: with open(self.performance_file, 'r') as f: perf_data = json.load(f) perf_df = pd.DataFrame(perf_data) logger.info(f"Loaded {len(perf_df)} performance records") except FileNotFoundError: logger.warning(f"Performance file not found: {self.performance_file}") perf_df = pd.DataFrame() return trades_df, perf_df def _merge_data(self, trades_df: pd.DataFrame, perf_df: pd.DataFrame) -> pd.DataFrame: """Merged Trade-Ergebnisse mit Signal-Features""" if trades_df.empty or perf_df.empty: return pd.DataFrame() # Extract order ticket from performance data def extract_ticket(order_result): if pd.isna(order_result): return None try: # Parse "order=641628738" from the string if 'order=' in str(order_result): parts = str(order_result).split('order=') if len(parts) > 1: ticket = int(parts[1].split(',')[0].split(')')[0]) return ticket except: pass return None perf_df['ticket'] = perf_df['order_result'].apply(extract_ticket) # Remove rows without ticket perf_df = perf_df.dropna(subset=['ticket']) perf_df['ticket'] = perf_df['ticket'].astype(int) # Merge on ticket merged = pd.merge( trades_df, perf_df, on='ticket', how='inner', suffixes=('_trade', '_perf') ) logger.info(f"Merged {len(merged)} records") return merged def _extract_features(self, df: pd.DataFrame) -> pd.DataFrame: """Extrahiert Features aus den Rohdaten""" features = pd.DataFrame() # Signal Features features['confidence'] = df['confidence'].fillna(0) features['adaptive_threshold'] = df['adaptive_threshold'].fillna(70) features['regime_strength'] = df['regime_strength'].fillna(50) # Normalize risk_adjusted_strength (can be very large) ras = df['risk_adjusted_strength'].fillna(0) features['risk_adjusted_strength_norm'] = np.log1p(ras) / 15 # Normalize to ~0-1 # Session One-Hot session = df.get('session_trade', df.get('session', 'unknown')).str.lower() features['session_asian'] = (session == 'asian').astype(int) features['session_london'] = (session == 'london').astype(int) features['session_ny'] = (session == 'ny').astype(int) features['session_overlap'] = (session == 'overlap').astype(int) # Direction direction = df.get('direction', df.get('entry_signal', 0)) if direction.dtype == 'object': features['direction_long'] = (direction.str.upper() == 'LONG').astype(int) else: features['direction_long'] = (direction == 1).astype(int) # Market Regime One-Hot regime = df.get('market_regime', 'unknown').str.lower() features['regime_trending'] = (regime == 'trending').astype(int) features['regime_ranging'] = (regime == 'ranging').astype(int) # Signal Quality (Ordinal: unknown=0, poor=1, average=2, good=3, excellent=4) quality_map = {'unknown': 0, 'poor': 1, 'average': 2, 'good': 3, 'excellent': 4} quality = df.get('signal_quality', 'unknown') if hasattr(quality, 'str'): features['quality_score'] = quality.str.lower().map(quality_map).fillna(0) else: features['quality_score'] = 0 # Time Features try: entry_time = pd.to_datetime(df.get('entry_time', df.get('timestamp'))) hour = entry_time.dt.hour features['hour_sin'] = np.sin(2 * np.pi * hour / 24) features['hour_cos'] = np.cos(2 * np.pi * hour / 24) features['day_of_week'] = entry_time.dt.dayofweek except Exception as e: logger.warning(f"Could not extract time features: {e}") features['hour_sin'] = 0 features['hour_cos'] = 1 features['day_of_week'] = 0 # Derived Features features['confidence_above_threshold'] = ( features['confidence'] > features['adaptive_threshold'] ).astype(int) features['confidence_margin'] = features['confidence'] - features['adaptive_threshold'] # Target Variable if 'is_win' in df.columns: features['is_win'] = df['is_win'].astype(int) return features def _extract_single_signal_features(self, signal_info: Dict) -> pd.DataFrame: """Extrahiert Features aus einem einzelnen Signal für Prediction""" features = {} # Signal Features features['confidence'] = signal_info.get('confidence', 0) features['adaptive_threshold'] = signal_info.get('adaptive_threshold', 70) features['regime_strength'] = signal_info.get('regime_strength', 50) # Normalize risk_adjusted_strength ras = signal_info.get('risk_adjusted_strength', 0) features['risk_adjusted_strength_norm'] = np.log1p(ras) / 15 # Session One-Hot session = signal_info.get('session', 'unknown').lower() features['session_asian'] = 1 if session == 'asian' else 0 features['session_london'] = 1 if session == 'london' else 0 features['session_ny'] = 1 if session == 'ny' else 0 features['session_overlap'] = 1 if session == 'overlap' else 0 # Direction entry_signal = signal_info.get('entry_signal', 0) features['direction_long'] = 1 if entry_signal == 1 else 0 # Market Regime regime = signal_info.get('market_regime', 'unknown').lower() features['regime_trending'] = 1 if regime == 'trending' else 0 features['regime_ranging'] = 1 if regime == 'ranging' else 0 # Signal Quality quality_map = {'unknown': 0, 'poor': 1, 'average': 2, 'good': 3, 'excellent': 4} quality = signal_info.get('signal_quality', 'unknown').lower() features['quality_score'] = quality_map.get(quality, 0) # Time Features try: now = datetime.now() hour = now.hour features['hour_sin'] = np.sin(2 * np.pi * hour / 24) features['hour_cos'] = np.cos(2 * np.pi * hour / 24) features['day_of_week'] = now.weekday() except Exception as e: logger.warning(f"Could not extract time features: {e}") features['hour_sin'] = 0 features['hour_cos'] = 1 features['day_of_week'] = 0 # Derived Features features['confidence_above_threshold'] = 1 if features['confidence'] > features['adaptive_threshold'] else 0 features['confidence_margin'] = features['confidence'] - features['adaptive_threshold'] return pd.DataFrame([features]) # ========================================== # TRAINING # ========================================== def train(self, force: bool = False) -> Dict: """ Trainiert das XGBoost Modell Args: force: Erzwingt Retraining auch wenn Model existiert Returns: Dict mit Training-Statistiken """ try: # Import XGBoost import xgboost as xgb from sklearn.model_selection import cross_val_score, StratifiedKFold except ImportError: logger.error("XGBoost not installed. Run: pip install xgboost scikit-learn") return {'error': 'XGBoost not installed'} print("\n" + "=" * 60) print("ML SIGNAL PREDICTOR - TRAINING") print("=" * 60) # Load and merge data trades_df, perf_df = self._load_data() merged_df = self._merge_data(trades_df, perf_df) if len(merged_df) < ML_CONFIG['min_trades_for_training']: msg = f"Not enough data: {len(merged_df)}/{ML_CONFIG['min_trades_for_training']} trades" print(f" {msg}") logger.warning(msg) return {'error': msg, 'trades_available': len(merged_df)} # Extract features features_df = self._extract_features(merged_df) # Prepare training data X = features_df[FEATURE_COLUMNS] y = features_df['is_win'] print(f"\n Training Data:") print(f" - Total Trades: {len(X)}") print(f" - Wins: {y.sum()} ({y.mean()*100:.1f}%)") print(f" - Losses: {len(y) - y.sum()} ({(1-y.mean())*100:.1f}%)") print(f" - Features: {len(FEATURE_COLUMNS)}") # Initialize model self.model = xgb.XGBClassifier(**ML_CONFIG['xgb_params']) # Cross-validation print(f"\n Cross-Validation (5-fold)...") cv = StratifiedKFold(n_splits=5, shuffle=True, random_state=42) cv_scores = cross_val_score(self.model, X, y, cv=cv, scoring='roc_auc') print(f" - AUC Scores: {cv_scores}") print(f" - Mean AUC: {cv_scores.mean():.3f} (+/- {cv_scores.std()*2:.3f})") # Train final model on all data print(f"\n Training final model...") self.model.fit(X, y) # Feature importance importance = dict(zip(FEATURE_COLUMNS, self.model.feature_importances_)) importance_sorted = sorted(importance.items(), key=lambda x: x[1], reverse=True) print(f"\n Feature Importance (Top 5):") for feat, imp in importance_sorted[:5]: print(f" - {feat}: {imp:.3f}") # Save model self._save_model() # Update stats self.is_trained = True self.trades_since_training = 0 self.training_stats = { 'trained_at': datetime.now().isoformat(), 'total_trades': len(X), 'win_rate': float(y.mean()), 'cv_auc_mean': float(cv_scores.mean()), 'cv_auc_std': float(cv_scores.std()), 'feature_importance': importance } print(f"\n Model saved to: {ML_CONFIG['model_path']}") print("=" * 60) return self.training_stats # ========================================== # PREDICTION # ========================================== def predict(self, signal_info: Dict) -> Dict: """ Vorhersage für ein Signal Args: signal_info: Dict mit Signal-Informationen Returns: Dict mit: - win_probability: float (0-1) - should_trade: bool - lot_multiplier: float - action: str (ALLOW, REDUCE, BLOCK) - reason: str """ if not ML_CONFIG['enabled']: return { 'win_probability': 0.5, 'should_trade': True, 'lot_multiplier': 1.0, 'action': 'ALLOW', 'reason': 'ML Predictor disabled' } if not self.is_trained or self.model is None: return { 'win_probability': 0.5, 'should_trade': True, 'lot_multiplier': 1.0, 'action': 'ALLOW', 'reason': 'ML Model not trained yet' } try: # Extract features X = self._extract_single_signal_features(signal_info) X = X[FEATURE_COLUMNS] # Predict probability win_prob = self.model.predict_proba(X)[0][1] # Determine action if win_prob < ML_CONFIG['min_win_probability']: action = 'BLOCK' lot_multiplier = 0.0 should_trade = False reason = f"Low win probability ({win_prob:.1%} < {ML_CONFIG['min_win_probability']:.0%})" elif win_prob < ML_CONFIG['reduce_threshold']: action = 'REDUCE' lot_multiplier = 0.5 should_trade = True reason = f"Moderate win probability ({win_prob:.1%}), lot reduced to 50%" elif win_prob < ML_CONFIG['confidence_threshold']: action = 'CAUTION' lot_multiplier = 0.75 should_trade = True reason = f"Acceptable win probability ({win_prob:.1%}), lot at 75%" else: action = 'ALLOW' lot_multiplier = 1.0 should_trade = True reason = f"Good win probability ({win_prob:.1%})" return { 'win_probability': float(win_prob), 'should_trade': should_trade, 'lot_multiplier': lot_multiplier, 'action': action, 'reason': reason } except Exception as e: logger.error(f"Prediction error: {e}") return { 'win_probability': 0.5, 'should_trade': True, 'lot_multiplier': 1.0, 'action': 'ALLOW', 'reason': f'Prediction error: {e}' } # ========================================== # MODEL PERSISTENCE # ========================================== def _save_model(self): """Speichert das Model""" try: with open(ML_CONFIG['model_path'], 'wb') as f: pickle.dump({ 'model': self.model, 'feature_columns': self.feature_columns, 'training_stats': self.training_stats }, f) logger.info(f"Model saved to {ML_CONFIG['model_path']}") except Exception as e: logger.error(f"Failed to save model: {e}") def _load_model(self): """Lädt ein existierendes Model""" try: if os.path.exists(ML_CONFIG['model_path']): # pickle.load executes arbitrary code — only load models you generated yourself with open(ML_CONFIG['model_path'], 'rb') as f: data = pickle.load(f) self.model = data['model'] self.feature_columns = data.get('feature_columns', FEATURE_COLUMNS) self.training_stats = data.get('training_stats', {}) self.is_trained = True logger.info(f"Model loaded from {ML_CONFIG['model_path']}") except Exception as e: logger.warning(f"Failed to load model: {e}") self.is_trained = False # ========================================== # UTILITY METHODS # ========================================== def get_status(self) -> str: """Gibt den aktuellen Status zurück""" lines = [] lines.append("=" * 60) lines.append("ML SIGNAL PREDICTOR STATUS") lines.append("=" * 60) lines.append(f" Enabled: {ML_CONFIG['enabled']}") lines.append(f" Model Trained: {self.is_trained}") if self.is_trained and self.training_stats: lines.append(f" Trained On: {self.training_stats.get('total_trades', 'N/A')} trades") lines.append(f" Training Win Rate: {self.training_stats.get('win_rate', 0)*100:.1f}%") lines.append(f" CV AUC Score: {self.training_stats.get('cv_auc_mean', 0):.3f}") lines.append(f" Trained At: {self.training_stats.get('trained_at', 'N/A')}") lines.append(f"\n Thresholds:") lines.append(f" - Block if < {ML_CONFIG['min_win_probability']:.0%} win probability") lines.append(f" - Reduce if < {ML_CONFIG['reduce_threshold']:.0%} win probability") lines.append(f" - Full lot if >= {ML_CONFIG['confidence_threshold']:.0%} win probability") lines.append("=" * 60) return "\n".join(lines) def should_retrain(self) -> bool: """Prüft ob Retraining nötig ist""" return self.trades_since_training >= ML_CONFIG['retrain_interval_trades'] def record_trade_result(self, was_win: bool): """Zeichnet Trade-Ergebnis auf für Retraining-Trigger""" self.trades_since_training += 1 if self.should_retrain(): logger.info("Retraining threshold reached, consider calling train()") # ========================================== # INTEGRATION FUNCTIONS # ========================================== # Global instance ml_predictor = None def get_ml_predictor() -> MLSignalPredictor: """Gibt die globale ML Predictor Instanz zurück""" global ml_predictor if ml_predictor is None: ml_predictor = MLSignalPredictor() return ml_predictor def check_ml_signal(signal_info: Dict) -> Dict: """ Wrapper-Funktion für einfache Integration Args: signal_info: Signal-Dict von extended_top_down_v2_adaptive() Returns: Dict mit win_probability, should_trade, lot_multiplier, action, reason """ predictor = get_ml_predictor() return predictor.predict(signal_info) def train_ml_model(force: bool = False) -> Dict: """Trainiert das ML Model""" predictor = get_ml_predictor() return predictor.train(force=force) def get_ml_status() -> str: """Gibt ML Predictor Status zurück""" predictor = get_ml_predictor() return predictor.get_status() def enable_ml_predictor(): """Aktiviert den ML Predictor""" ML_CONFIG['enabled'] = True print("ML Signal Predictor ENABLED") def disable_ml_predictor(): """Deaktiviert den ML Predictor""" ML_CONFIG['enabled'] = False print("ML Signal Predictor DISABLED") # ========================================== # STANDALONE TEST # ========================================== if __name__ == "__main__": logging.basicConfig(level=logging.INFO) # Initialize predictor predictor = MLSignalPredictor() print(predictor.get_status()) # Train model print("\nTraining model...") stats = predictor.train() if 'error' not in stats: # Test prediction test_signal = { 'entry_signal': 1, 'confidence': 75.0, 'adaptive_threshold': 70.0, 'signal_quality': 'good', 'market_regime': 'ranging', 'regime_strength': 45.0, 'risk_adjusted_strength': 50000, 'session': 'asian' } print("\nTest Prediction:") result = predictor.predict(test_signal) print(f" Win Probability: {result['win_probability']:.1%}") print(f" Action: {result['action']}") print(f" Lot Multiplier: {result['lot_multiplier']}") print(f" Reason: {result['reason']}")