#!/usr/bin/env python3 """ 🔍 Enhanced Multi-Timeframe Signal Scoring Verbesserte Signal-Bewertung mit zusätzlichen Faktoren NEUE FEATURES: 1. Volume Analysis (Trending Volume = stärkerer Move) 2. Momentum Indicators (RSI, MACD) 3. Support/Resistance Levels 4. Fibonacci Retracements 5. Weighted Scoring System """ import MetaTrader5 as mt5 import pandas as pd import numpy as np from typing import Dict, List, Tuple, Optional from dataclasses import dataclass import logging logger = logging.getLogger(__name__) @dataclass class EnhancedSignal: """Enhanced Signal mit allen Scoring-Komponenten""" symbol: str direction: int # 1=Long, -1=Short, 0=No Signal total_score: float # 0-100 confidence: float # Original confidence # Sub-Scores trend_score: float volume_score: float momentum_score: float support_resistance_score: float fibonacci_score: float # Metadata timeframe_alignment: str signal_quality: str reason: str class EnhancedSignalScorer: """ Erweiterte Signal-Bewertung mit Multi-Faktor-Analyse Weighted Scoring: - Trend Alignment: 30% - Volume Confirmation: 20% - Momentum Strength: 20% - Support/Resistance: 15% - Fibonacci Levels: 15% """ def __init__(self, weights: Optional[Dict[str, float]] = None): """ Args: weights: Custom weights für Scoring (default: siehe oben) """ self.weights = weights or { 'trend': 0.30, 'volume': 0.20, 'momentum': 0.20, 'support_resistance': 0.15, 'fibonacci': 0.15 } # Verify weights sum to 1.0 total = sum(self.weights.values()) if abs(total - 1.0) > 0.01: raise ValueError(f"Weights must sum to 1.0 (got {total})") # ========================================== # 1. VOLUME ANALYSIS # ========================================== def calculate_volume_score(self, symbol: str, timeframe: str = "H1", lookback: int = 50) -> float: """ Analysiert Volume für Trend-Bestätigung Logic: - Steigendes Volume in Trend-Richtung = stark (Score: 80-100) - Fallendes Volume in Trend-Richtung = schwach (Score: 20-50) - Kein klares Volume-Pattern = neutral (Score: 50) Args: symbol: Trading Symbol timeframe: Timeframe lookback: Anzahl Bars Returns: Volume Score (0-100) """ try: # Get OHLCV data rates = mt5.copy_rates_from_pos(symbol, self._tf_to_mt5(timeframe), 0, lookback) if rates is None or len(rates) < 20: return 50.0 # Neutral if no data df = pd.DataFrame(rates) # Calculate Volume MA df['volume_ma_20'] = df['tick_volume'].rolling(20).mean() # Recent vs Average Volume recent_volume = df['tick_volume'].iloc[-5:].mean() avg_volume = df['volume_ma_20'].iloc[-1] volume_ratio = recent_volume / avg_volume if avg_volume > 0 else 1.0 # Score berechnen if volume_ratio >= 1.5: score = 90.0 # Sehr hohes Volume elif volume_ratio >= 1.2: score = 75.0 # Hohes Volume elif volume_ratio >= 0.8: score = 60.0 # Normales Volume else: score = 40.0 # Niedriges Volume return score except Exception as e: logger.warning(f"Volume calculation error: {e}") return 50.0 # ========================================== # 2. MOMENTUM INDICATORS # ========================================== def calculate_momentum_score(self, symbol: str, timeframe: str = "H1", lookback: int = 50) -> Tuple[float, Dict]: """ Berechnet Momentum-Score mit RSI und MACD Args: symbol: Trading Symbol timeframe: Timeframe lookback: Anzahl Bars Returns: (momentum_score, details_dict) """ try: rates = mt5.copy_rates_from_pos(symbol, self._tf_to_mt5(timeframe), 0, lookback) if rates is None or len(rates) < 30: return 50.0, {} df = pd.DataFrame(rates) df['close'] = df['close'] # 1. RSI Calculation rsi = self._calculate_rsi(df['close'], period=14) current_rsi = rsi.iloc[-1] # 2. MACD Calculation macd, signal, hist = self._calculate_macd(df['close']) current_macd = macd.iloc[-1] current_signal = signal.iloc[-1] current_hist = hist.iloc[-1] # RSI Score if 40 <= current_rsi <= 60: rsi_score = 80.0 # Neutral = gut für Entry elif 30 <= current_rsi <= 70: rsi_score = 60.0 # OK elif current_rsi < 30 or current_rsi > 70: rsi_score = 40.0 # Overbought/Oversold = vorsichtig else: rsi_score = 50.0 # MACD Score if current_macd > current_signal and current_hist > 0: macd_score = 80.0 # Bullish elif current_macd < current_signal and current_hist < 0: macd_score = 80.0 # Bearish (consistent) else: macd_score = 50.0 # Mixed # Combined Momentum Score momentum_score = (rsi_score * 0.5 + macd_score * 0.5) details = { 'rsi': current_rsi, 'rsi_score': rsi_score, 'macd': current_macd, 'macd_signal': current_signal, 'macd_hist': current_hist, 'macd_score': macd_score } return momentum_score, details except Exception as e: logger.warning(f"Momentum calculation error: {e}") return 50.0, {} def _calculate_rsi(self, prices: pd.Series, period: int = 14) -> pd.Series: """RSI Calculation""" delta = prices.diff() gain = (delta.where(delta > 0, 0)).rolling(window=period).mean() loss = (-delta.where(delta < 0, 0)).rolling(window=period).mean() # Avoid division by zero: when loss == 0 the RSI is 100 rs = gain / loss.replace(0, np.nan) rsi = 100 - (100 / (1 + rs)) return rsi.fillna(100.0) def _calculate_macd(self, prices: pd.Series, fast: int = 12, slow: int = 26, signal: int = 9) -> Tuple[pd.Series, pd.Series, pd.Series]: """MACD Calculation""" ema_fast = prices.ewm(span=fast).mean() ema_slow = prices.ewm(span=slow).mean() macd = ema_fast - ema_slow signal_line = macd.ewm(span=signal).mean() histogram = macd - signal_line return macd, signal_line, histogram # ========================================== # 3. SUPPORT/RESISTANCE LEVELS # ========================================== def calculate_support_resistance_score(self, symbol: str, current_price: float, timeframe: str = "H4", lookback: int = 200) -> Tuple[float, Dict]: """ Findet Support/Resistance und bewertet Distanz Logic: - Nahe an Support (Long) oder Resistance (Short) = gut (Score: 80-100) - Weit entfernt = schlecht (Score: 30-50) Args: symbol: Trading Symbol current_price: Aktueller Preis timeframe: Timeframe lookback: Anzahl Bars Returns: (sr_score, details_dict) """ try: rates = mt5.copy_rates_from_pos(symbol, self._tf_to_mt5(timeframe), 0, lookback) if rates is None or len(rates) < 50: return 50.0, {} df = pd.DataFrame(rates) # Find Swing Highs/Lows highs = df['high'].values lows = df['low'].values # Simple peak/trough detection resistance_levels = self._find_peaks(highs, distance=10) support_levels = self._find_peaks(-lows, distance=10) # Invert for troughs support_levels = [-s for s in support_levels] # Negate each element back # Closest Support/Resistance closest_support = max([s for s in support_levels if s < current_price], default=None) closest_resistance = min([r for r in resistance_levels if r > current_price], default=None) # Calculate distances if closest_support: support_distance = (current_price - closest_support) / current_price else: support_distance = float('inf') if closest_resistance: resistance_distance = (closest_resistance - current_price) / current_price else: resistance_distance = float('inf') # Score based on proximity (näher = besser) if support_distance < 0.01: # Within 1% score = 85.0 elif support_distance < 0.02: # Within 2% score = 70.0 elif resistance_distance < 0.01: score = 85.0 elif resistance_distance < 0.02: score = 70.0 else: score = 50.0 details = { 'closest_support': closest_support, 'closest_resistance': closest_resistance, 'support_distance_pct': support_distance * 100 if support_distance != float('inf') else None, 'resistance_distance_pct': resistance_distance * 100 if resistance_distance != float('inf') else None } return score, details except Exception as e: logger.warning(f"Support/Resistance calculation error: {e}") return 50.0, {} def _find_peaks(self, data: np.ndarray, distance: int = 10) -> List[float]: """Simple peak detection""" peaks = [] for i in range(distance, len(data) - distance): if data[i] == max(data[i-distance:i+distance+1]): peaks.append(data[i]) return peaks # ========================================== # 4. FIBONACCI RETRACEMENTS # ========================================== def calculate_fibonacci_score(self, symbol: str, current_price: float, timeframe: str = "D1", lookback: int = 100) -> Tuple[float, Dict]: """ Bewertet Fibonacci Level Proximity Args: symbol: Trading Symbol current_price: Aktueller Preis timeframe: Timeframe lookback: Anzahl Bars Returns: (fib_score, details_dict) """ try: rates = mt5.copy_rates_from_pos(symbol, self._tf_to_mt5(timeframe), 0, lookback) if rates is None or len(rates) < 50: return 50.0, {} df = pd.DataFrame(rates) # Find swing high/low for Fibonacci swing_high = df['high'].max() swing_low = df['low'].min() diff = swing_high - swing_low # Fibonacci Levels fib_levels = { '0.0': swing_low, '0.236': swing_low + 0.236 * diff, '0.382': swing_low + 0.382 * diff, '0.5': swing_low + 0.5 * diff, '0.618': swing_low + 0.618 * diff, '0.786': swing_low + 0.786 * diff, '1.0': swing_high } # Find closest Fib level distances = {level: abs(current_price - price) / current_price for level, price in fib_levels.items()} closest_level = min(distances, key=distances.get) closest_distance = distances[closest_level] # Score based on proximity to key Fib levels key_levels = ['0.382', '0.5', '0.618'] if closest_level in key_levels and closest_distance < 0.005: # Within 0.5% score = 90.0 # Perfect bounce area elif closest_level in key_levels and closest_distance < 0.01: # Within 1% score = 75.0 # Good area elif closest_distance < 0.02: # Within 2% score = 60.0 # OK else: score = 50.0 # No special Fib level details = { 'swing_high': swing_high, 'swing_low': swing_low, 'fib_levels': fib_levels, 'closest_level': closest_level, 'closest_price': fib_levels[closest_level], 'distance_pct': closest_distance * 100 } return score, details except Exception as e: logger.warning(f"Fibonacci calculation error: {e}") return 50.0, {} # ========================================== # 5. COMBINED SCORING # ========================================== def calculate_enhanced_score(self, symbol: str, base_confidence: float, trend_direction: int, current_price: float) -> EnhancedSignal: """ Berechnet Enhanced Score mit allen Faktoren Args: symbol: Trading Symbol base_confidence: Original Confidence vom Trend-System trend_direction: 1=Long, -1=Short, 0=No Signal current_price: Aktueller Preis Returns: EnhancedSignal Object """ # Trend Score (basierend auf original confidence) trend_score = base_confidence # Volume Score volume_score = self.calculate_volume_score(symbol) # Momentum Score momentum_score, momentum_details = self.calculate_momentum_score(symbol) # Support/Resistance Score sr_score, sr_details = self.calculate_support_resistance_score(symbol, current_price) # Fibonacci Score fib_score, fib_details = self.calculate_fibonacci_score(symbol, current_price) # Weighted Total Score total_score = ( trend_score * self.weights['trend'] + volume_score * self.weights['volume'] + momentum_score * self.weights['momentum'] + sr_score * self.weights['support_resistance'] + fib_score * self.weights['fibonacci'] ) # Signal Quality if total_score >= 85: signal_quality = "excellent" elif total_score >= 75: signal_quality = "very_good" elif total_score >= 65: signal_quality = "good" elif total_score >= 55: signal_quality = "fair" else: signal_quality = "poor" # Reason reasons = [] if trend_score >= 80: reasons.append(f"Strong trend ({trend_score:.0f}%)") if volume_score >= 75: reasons.append("High volume") if momentum_score >= 75: reasons.append("Strong momentum") if sr_score >= 70: reasons.append("Near S/R level") if fib_score >= 75: reasons.append("Key Fib level") reason = ", ".join(reasons) if reasons else "Standard setup" return EnhancedSignal( symbol=symbol, direction=trend_direction, total_score=total_score, confidence=base_confidence, trend_score=trend_score, volume_score=volume_score, momentum_score=momentum_score, support_resistance_score=sr_score, fibonacci_score=fib_score, timeframe_alignment="multi", signal_quality=signal_quality, reason=reason ) # ========================================== # HELPER # ========================================== def _tf_to_mt5(self, timeframe: str): """Convert string timeframe to MT5 constant""" tf_map = { 'M1': mt5.TIMEFRAME_M1, 'M5': mt5.TIMEFRAME_M5, 'M15': mt5.TIMEFRAME_M15, 'M30': mt5.TIMEFRAME_M30, 'H1': mt5.TIMEFRAME_H1, 'H4': mt5.TIMEFRAME_H4, 'D1': mt5.TIMEFRAME_D1, 'W1': mt5.TIMEFRAME_W1 } return tf_map.get(timeframe.upper(), mt5.TIMEFRAME_H1) # ========================================== # USAGE EXAMPLE # ========================================== """ INTEGRATION IN NOTEBOOK: # Cell: Setup Enhanced Signal Scorer from enhanced_signal_scoring import EnhancedSignalScorer # Initialize Scorer signal_scorer = EnhancedSignalScorer( weights={ 'trend': 0.30, 'volume': 0.20, 'momentum': 0.20, 'support_resistance': 0.15, 'fibonacci': 0.15 } ) print("✅ Enhanced Signal Scorer activated!") # Cell: Use in Trading Logic # Get base signal from existing system signal_info = extended_top_down_v2_adaptive(symbol) # Get current price price = signal_info['trend_info']['M5']['price'] # Calculate enhanced score enhanced_signal = signal_scorer.calculate_enhanced_score( symbol=symbol, base_confidence=signal_info['confidence'], trend_direction=signal_info['entry_signal'], current_price=price ) # Print details print(f"🎯 ENHANCED SIGNAL SCORING:") print(f" Total Score: {enhanced_signal.total_score:.1f}/100") print(f" Quality: {enhanced_signal.signal_quality.upper()}") print(f" Reason: {enhanced_signal.reason}") print(f"") print(f" 📊 Component Scores:") print(f" Trend: {enhanced_signal.trend_score:.1f}/100") print(f" Volume: {enhanced_signal.volume_score:.1f}/100") print(f" Momentum: {enhanced_signal.momentum_score:.1f}/100") print(f" S/R: {enhanced_signal.support_resistance_score:.1f}/100") print(f" Fibonacci: {enhanced_signal.fibonacci_score:.1f}/100") # Use enhanced score instead of base confidence if enhanced_signal.total_score >= 70: execute_trade_v2_adaptive( symbol=symbol, base_confidence=enhanced_signal.total_score, # ← Enhanced! ... ) """