""" premium_signals.py - Sistema de Señales Premium con Scoring Avanzado Objetivo: >75% de tasa de éxito mediante confluencia de indicadores """ import pandas as pd import numpy as np from typing import Dict, List, Tuple, Optional from datetime import datetime class PremiumSignalScorer: """ Sistema de scoring profesional basado en confluencia de: - Tendencia (EMAs triples) - Momentum (RSI, MACD, Estocástico) - Volumen (OBV, Volume Ratio) - Volatilidad (Bollinger, ATR) - Soporte/Resistencia """ def __init__(self): self.MIN_SCORE_FOR_SIGNAL = 70 # Mínimo 70/100 para recomendar self.HIGH_CONFIDENCE_SCORE = 85 # Alta confianza >85 def calculate_premium_score(self, df: pd.DataFrame, symbol: str) -> Dict: """ Calcula score de 0-100 basado en múltiples factores Retorna: dict con score, señal, confianza y justificación """ if df is None or df.empty or len(df) < 100: return {'score': 0, 'signal': 'NEUTRAL', 'confidence': 0} scores = { 'trend': self._score_trend(df), # 0-25 puntos 'momentum': self._score_momentum(df), # 0-25 puntos 'volume': self._score_volume(df), # 0-20 puntos 'volatility': self._score_volatility(df), # 0-15 puntos 'structure': self._score_structure(df) # 0-15 puntos } total_score = sum(scores.values()) # Determinar señal if total_score >= self.MIN_SCORE_FOR_SIGNAL: if total_score >= 70: signal = 'COMPRA FUERTE' if scores['trend'] > 12 else 'VENTA FUERTE' else: signal = 'COMPRA' if scores['trend'] > 12 else 'VENTA' elif total_score >= 50: signal = 'NEUTRAL_ALCISTA' if scores['trend'] > 12 else 'NEUTRAL_BAJISTA' else: signal = 'NEUTRAL' # Calcular confianza confidence = min(total_score, 100) # Generar justificación justification = self._generate_justification(scores, df) return { 'score': total_score, 'signal': signal, 'confidence': confidence, 'scores_breakdown': scores, 'justification': justification, 'timestamp': datetime.now() } def _score_trend(self, df: pd.DataFrame) -> float: """ Score de tendencia (0-25 puntos) Basado en alineación de EMAs 9, 21, 50, 200 """ score = 0 # EMA 9 > EMA 21 (5 puntos) if df['ema_9'].iloc[-1] > df['ema_21'].iloc[-1]: score += 5 # EMA 21 > EMA 50 (10 puntos adicionales) if df['ema_21'].iloc[-1] > df['ema_50'].iloc[-1]: score += 10 # Tendencia fuerte confirmada if df['ema_9'].iloc[-1] > df['ema_21'].iloc[-1] > df['ema_50'].iloc[-1]: score += 5 # Bonus por alineación perfecta # EMA 50 > EMA 200 (si existe) if 'ema_200' in df.columns: if df['ema_50'].iloc[-1] > df['ema_200'].iloc[-1]: score += 5 else: # Tendencia bajista if df['ema_9'].iloc[-1] < df['ema_21'].iloc[-1]: score += 5 if df['ema_21'].iloc[-1] < df['ema_50'].iloc[-1]: score += 10 if df['ema_9'].iloc[-1] < df['ema_21'].iloc[-1] < df['ema_50'].iloc[-1]: score += 5 return score def _score_momentum(self, df: pd.DataFrame) -> float: """ Score de momentum (0-25 puntos) Basado en RSI, MACD, Estocástico """ score = 0 # RSI (8 puntos) rsi = df['rsi'].iloc[-1] if 40 <= rsi <= 60: score += 4 # Neutral elif 30 <= rsi < 40: score += 7 # Oversold (oportunidad compra) elif 60 < rsi <= 70: score += 7 # Fortaleza sin sobrecompra elif rsi < 30: score += 5 # Muy oversold (pero riesgoso) elif rsi > 70: score += 2 # Sobrecompra (riesgo) # MACD (10 puntos) macd_hist = df['macd_hist'].iloc[-1] macd_hist_prev = df['macd_hist'].iloc[-2] if len(df) > 1 else 0 if macd_hist > 0 and macd_hist > macd_hist_prev: score += 10 # MACD alcista y creciente elif macd_hist > 0: score += 7 # MACD positivo elif macd_hist < 0 and macd_hist < macd_hist_prev: score += 2 # MACD bajista elif macd_hist < 0: score += 5 # MACD negativo pero mejorando # Divergencia MACD (5 puntos bonus) if self._detect_bullish_divergence(df): score += 5 elif self._detect_bearish_divergence(df): score += 0 # No suma para venta (ya está en MACD) return min(score, 25) def _score_volume(self, df: pd.DataFrame) -> float: """ Score de volumen (0-20 puntos) El volumen confirma la tendencia """ score = 0 # Volume Ratio (10 puntos) vol_ratio = df['volume_ratio'].iloc[-1] if vol_ratio > 2.0: score += 10 # Volumen muy alto (confirmación fuerte) elif vol_ratio > 1.5: score += 8 # Volumen alto elif vol_ratio > 1.2: score += 5 # Volumen moderado elif vol_ratio > 0.8: score += 3 # Volumen normal else: score += 1 # Volumen bajo (poca confirmación) # Tendencia de volumen (10 puntos) avg_vol_last_5 = df['volume'].iloc[-5:].mean() avg_vol_prev_10 = df['volume'].iloc[-15:-5].mean() if avg_vol_last_5 > avg_vol_prev_10 * 1.3: score += 10 # Volumen creciente elif avg_vol_last_5 > avg_vol_prev_10: score += 6 # Volumen ligeramente creciente return min(score, 20) def _score_volatility(self, df: pd.DataFrame) -> float: """ Score de volatilidad (0-15 puntos) Volatilidad óptima = oportunidad sin riesgo excesivo """ score = 0 # Posición en Bandas de Bollinger (8 puntos) bb_upper = df['bb_upper'].iloc[-1] bb_lower = df['bb_lower'].iloc[-1] close = df['close'].iloc[-1] bb_width = bb_upper - bb_lower if close < bb_lower * 1.02: score += 8 # Precio cerca de banda inferior (oportunidad compra) elif close > bb_upper * 0.98: score += 2 # Precio cerca de banda superior (riesgo) elif bb_lower < close < bb_upper: # Posición central position = (close - bb_lower) / bb_width if 0.3 <= position <= 0.7: score += 5 # Zona neutral elif position < 0.3: score += 7 # Zona baja (oportunidad) else: score += 3 # Zona alta # ATR (volatilidad razonable) (7 puntos) atr = df['atr'].iloc[-1] atr_percent = (atr / close) * 100 if 1.0 <= atr_percent <= 5.0: score += 7 # Volatilidad óptima elif 0.5 <= atr_percent < 1.0: score += 4 # Volatilidad baja elif 5.0 < atr_percent <= 8.0: score += 5 # Volatilidad alta (oportunidad pero riesgo) else: score += 2 # Volatilidad extrema return min(score, 15) def _score_structure(self, df: pd.DataFrame) -> float: """ Score de estructura de mercado (0-15 puntos) Soportes, resistencias, máximos/mínimos """ score = 0 # Máximos y mínimos crecientes/decrecientes (10 puntos) highs = df['high'].values lows = df['low'].values # Últimos 5 períodos recent_highs = highs[-5:] recent_lows = lows[-5:] # Tendencia alcista: máximos y mínimos crecientes if all(recent_highs[i] > recent_highs[i-1] for i in range(1, len(recent_highs))): score += 10 # Máximos crecientes elif all(recent_lows[i] > recent_lows[i-1] for i in range(1, len(recent_lows))): score += 8 # Mínimos crecientes # Tendencia bajista if all(recent_highs[i] < recent_highs[i-1] for i in range(1, len(recent_highs))): score += 2 # Máximos decrecientes (no suma mucho para compra) elif all(recent_lows[i] < recent_lows[i-1] for i in range(1, len(recent_lows))): score += 0 # Mínimos decrecientes # Patrón de consolidación rota (5 puntos bonus) if self._detect_breakout(df): score += 5 return min(score, 15) def _detect_bullish_divergence(self, df: pd.DataFrame) -> bool: """Detecta divergencia alcista entre precio y MACD/RSI""" if len(df) < 20: return False # Precio haciendo mínimos más bajos price_min1 = df['close'].iloc[-10:-5].min() price_min2 = df['close'].iloc[-5:].min() # MACD haciendo mínimos más altos macd_min1 = df['macd'].iloc[-10:-5].min() macd_min2 = df['macd'].iloc[-5:].min() return price_min2 < price_min1 and macd_min2 > macd_min1 def _detect_bearish_divergence(self, df: pd.DataFrame) -> bool: """Detecta divergencia bajista""" if len(df) < 20: return False price_max1 = df['close'].iloc[-10:-5].max() price_max2 = df['close'].iloc[-5:].max() macd_max1 = df['macd'].iloc[-10:-5].max() macd_max2 = df['macd'].iloc[-5:].max() return price_max2 > price_max1 and macd_max2 < macd_max1 def _detect_breakout(self, df: pd.DataFrame) -> bool: """Detecta ruptura de consolidación""" if len(df) < 50: return False # Rango de consolidación (últimas 20 velas) recent_high = df['high'].iloc[-20:-5].max() recent_low = df['low'].iloc[-20:-5].min() range_size = recent_high - recent_low # Ruptura con volumen current_close = df['close'].iloc[-1] current_vol = df['volume_ratio'].iloc[-1] breakout_up = current_close > recent_high * 1.01 and current_vol > 1.5 breakout_down = current_close < recent_low * 0.99 and current_vol > 1.5 return breakout_up or breakout_down def _generate_justification(self, scores: Dict, df: pd.DataFrame) -> str: """Genera explicación de la señal""" justifications = [] # Tendencia if scores['trend'] >= 20: justifications.append("✅ Tendencia fuerte confirmada (EMAs alineadas)") elif scores['trend'] >= 15: justifications.append("✓ Tendencia alcista") elif scores['trend'] <= 5: justifications.append("⚠️ Tendencia débil o lateral") # Momentum if scores['momentum'] >= 20: justifications.append("✅ Momentum muy fuerte") elif scores['momentum'] >= 15: justifications.append("✓ Buen momentum") # Volumen if scores['volume'] >= 15: justifications.append("✅ Volumen de confirmación alto") elif scores['volume'] < 10: justifications.append("️ Volumen bajo (poca confirmación)") # Volatilidad if scores['volatility'] >= 10: justifications.append("✓ Volatilidad óptima") return "\n".join(justifications) if justifications else "⚠️ Sin señales claras" def get_premium_recommendations(market_data: Dict[str, pd.DataFrame], market_type: str = 'spot', min_score: int = 70, top_n: int = 3) -> List[Dict]: """ Obtiene las mejores recomendaciones premium Filtra solo las señales con score >= min_score """ scorer = PremiumSignalScorer() recommendations = [] for symbol, df in market_data.items(): if df is None or df.empty: continue # Calcular indicadores si no existen if 'ema_9' not in df.columns: from indicators import calculate_all_indicators df = calculate_all_indicators(df) # Obtener score premium result = scorer.calculate_premium_score(df, symbol) # Solo incluir si supera el mínimo if result['score'] >= min_score: recommendations.append({ 'symbol': symbol, **result, 'price': df['close'].iloc[-1], 'change_24h': ((df['close'].iloc[-1] / df['close'].iloc[-24]) - 1) * 100 if len(df) > 24 else 0 }) # Ordenar por score recommendations.sort(key=lambda x: x['score'], reverse=True) return recommendations[:top_n]