cbazza/Place-Order-Trading-Bot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
main
Place-Order-Trading-Bot/ml_signal_predictor.py
T
cbazzaandClaude Sonnet 4.6 4e45db967b fix: core module review fixes (signal scoring, ML, session filter, telegram, MT filter) 
enhanced_signal_scoring.py:
- mt -> mt5, add logging module, replace print() with logger
- Remove no-op df['tick_volume'] = df['tick_volume'] line
- Fix RSI division-by-zero: loss.replace(0, nan) + fillna(100)

ml_signal_predictor.py:
- Remove global warnings.filterwarnings('ignore') suppression
- Fix bare except -> except Exception with logger.warning
- Add note: default data files excluded from git, need manual export
- Add pickle security warning comment

session_confidence_filter.py:
- Move imports to file top, add logging + functools.wraps
- Remove repeated AdaptiveRhythmManager() per-call instantiation
- Add functools.wraps to preserve wrapped function metadata
- Document that extended_top_down_v2_adaptive is notebook-only
- Replace print() with logger, pass **kwargs through wrapper

telegram_notifier.py:
- Remove network call from __init__ -> explicit test_connection() method
- Add logging module, replace all print() with logger calls
- Narrow exception type: Exception -> requests.RequestException
- Remove unused imports (timedelta)
- notify_trade_entry/exit now return bool from send_message

multi_timeframe_regime_filter.py:
- Change debug default from True to False in wrapper to avoid verbose production output

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-05-12 10:12:25 +02:00

661 lines
23 KiB
Python
Raw Permalink Blame History

#!/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']}")
Reference in New Issue View Git Blame Copy Permalink