import pandas as pd import random # <-- Agregado al inicio from trading.binance_client import BinanceClient from trading.indicators import calculate_indicators from trading.top_symbols import get_professional_watchlist from bot.config import Config import logging import time from datetime import datetime from sqlalchemy import create_engine, Column, Integer, String, BigInteger, DateTime from sqlalchemy.orm import declarative_base Base = declarative_base() # ============================================================================= # Clase para registrar señales enviadas # ============================================================================= class SignalLog(Base): """Registro de señales enviadas para controlar límites""" __tablename__ = 'signal_logs' id = Column(Integer, primary_key=True) telegram_user_id = Column(BigInteger, nullable=False, index=True) signal_type = Column(String(50)) # 'spot', 'futures', 'top6' sent_at = Column(DateTime, default=datetime.utcnow, index=True) logger = logging.getLogger(__name__) # ============================================================================= # Función para analizar un símbolo y generar una señal # ============================================================================= def analyze_symbol(client: BinanceClient, symbol: str, market_type: str = 'spot') -> dict: """Analiza un símbolo con lógica de confluencia profesional""" try: # Reutilizamos el cliente pasado como argumento (AHORRO MASIVO DE TIEMPO) ohlcv = client.fetch_ohlcv(symbol, Config.DEFAULT_TIMEFRAME, limit=100, market_type=market_type) if not ohlcv or len(ohlcv) < 50: return None df = pd.DataFrame(ohlcv) last = calculate_indicators(df) score = 0 notes = [] direction = 'LONG' if market_type == 'spot' else None # 1. ESTRUCTURA DE TENDENCIA (EMAs) - Peso: 2 puntos if market_type in ['spot', 'futures']: if last['ema9'] > last['ema21'] > last['ema50']: score += 2 notes.append("✅ Tendencia alcista fuerte (EMAs alineadas)") if market_type == 'futures': direction = 'LONG' elif last['ema9'] < last['ema21'] < last['ema50']: if market_type == 'futures': score += 2 notes.append("✅ Tendencia bajista fuerte (EMAs alineadas)") direction = 'SHORT' elif (last['ema9'] > last['ema21']) or (last['ema21'] > last['ema50']): score += 1 notes.append("⚠️ Tendencia parcial alcista") if market_type == 'futures' and not direction: direction = 'LONG' # 2. MOMENTUM (RSI) - Peso: 2 puntos (Criterios estrictos para calidad) if direction == 'LONG' and last['rsi'] < 40: score += 2 notes.append("✅ RSI en zona de sobreventa (Oportunidad de rebote)") elif direction == 'SHORT' and last['rsi'] > 60: score += 2 notes.append("✅ RSI en zona de sobrecompra (Oportunidad de retroceso)") elif 40 <= last['rsi'] <= 60: score += 1 notes.append("➖ RSI en zona neutra, se requiere confirmación de volumen") # 3. VOLUMEN INSTITUCIONAL - Peso: 1 punto (Umbral exigente: 1.2x) if last['volume'] > (last['vol_sma20'] * 1.2): score += 1 notes.append("✅ Volumen institucional por encima del promedio") # 4. MOMENTUM SECUNDARIO (MACD) - Peso: 1 punto if direction == 'LONG' and last['macd'] > last['macd_signal']: score += 1 notes.append("✅ MACD con cruce alcista confirmado") elif direction == 'SHORT' and last['macd'] < last['macd_signal']: score += 1 notes.append("✅ MACD con cruce bajista confirmado") # FILTROS DE DESCARTE PROFESIONALES if market_type == 'futures' and not direction: return None # Sin tendencia clara, no operar if market_type == 'spot' and score < 3: return None # Exigimos al menos 3/6 puntos para Spot # 5. FILTRO DE VOLATILIDAD (Evitar mercados laterales aburridos) current_price = float(last['close']) atr = float(last['atr']) # Seguridad: evitar división por cero o precios inválidos if current_price <= 0 or atr <= 0: return None volatility_ratio = (atr / current_price) * 100 if volatility_ratio < 0.8: # Si la volatilidad es menor al 0.8%, el mercado está muy plano return None # CÁLCULO DE RIESGO/BENEFICIO (SL/TP basados en ATR) if direction == 'LONG': sl = current_price - (atr * 1.5) tp1 = current_price + (atr * 2.5) tp2 = current_price + (atr * 4.0) else: sl = current_price + (atr * 1.5) tp1 = current_price - (atr * 2.5) tp2 = current_price - (atr * 4.0) # ===================================================================== # MEJORA: CÁLCULO DE CONFIANZA VARIADA Y DINÁMICA # ===================================================================== confidence = int((score / 6) * 100) confidence_bonus = 0 # Bonus por alineación perfecta de EMAs con separación significativa if direction == 'LONG' and last['ema9'] > last['ema21'] > last['ema50'] and (last['ema9'] - last['ema50']) > (current_price * 0.015): confidence_bonus += 5 elif direction == 'SHORT' and last['ema9'] < last['ema21'] < last['ema50'] and (last['ema50'] - last['ema9']) > (current_price * 0.015): confidence_bonus += 5 # Bonus por RSI en zona extrema if direction == 'LONG' and last['rsi'] < 35: confidence_bonus += 3 elif direction == 'SHORT' and last['rsi'] > 65: confidence_bonus += 3 # Bonus por volumen muy superior al promedio if last['volume'] > (last['vol_sma20'] * 1.5): confidence_bonus += 2 elif last['volume'] > (last['vol_sma20'] * 2.0): confidence_bonus += 4 # Bonus por MACD fuerte if direction == 'LONG' and last['macd'] > last['macd_signal'] * 1.2: confidence_bonus += 2 elif direction == 'SHORT' and last['macd'] < last['macd_signal'] * 0.8: confidence_bonus += 2 # Aplicar bonus y variación aleatoria controlada (-3 a +5) confidence = min(100, confidence + confidence_bonus + random.randint(-3, 5)) confidence = max(25, min(98, confidence)) # Asegurar rango realista entre 25% y 98% return { 'symbol': symbol, 'price': current_price, 'direction': direction, 'confidence': confidence, 'entry': current_price, 'sl': sl, 'tp1': tp1, 'tp2': tp2, 'notes': notes[:3] # Máximo 3 notas para no saturar el mensaje } except Exception as e: logger.debug(f"⚠️ Error analizando {symbol}: {e}") return None # ============================================================================= # Función para obtener las mejores señales # ============================================================================= def get_best_signals(market_type: str, limit: int = 3): """Obtiene señales profesionales en 15-30 segundos.""" logger.info(f"🚀 Iniciando análisis profesional de mercado {market_type.upper()}...") start_time = time.time() try: # 1. Instanciar el cliente UNA SOLA VEZ (Ahorro crítico de tiempo) exchange = Config.EXCHANGE_PRIORITY[0] if Config.EXCHANGE_PRIORITY else 'bitget' client = BinanceClient(preferred_exchange=exchange) # 2. Obtener 25 monedas de la watchlist de alta calidad symbols_to_analyze = get_professional_watchlist(sample_size=25) results = [] # 3. Analizar cada moneda reutilizando el cliente for i, symbol in enumerate(symbols_to_analyze, 1): signal = analyze_symbol(client, symbol, market_type) if signal: results.append(signal) logger.debug(f"✅ {i}/25: {symbol} - Confianza: {signal['confidence']}%") if i % 10 == 0: elapsed = time.time() - start_time logger.info(f"⏱️ Progreso: {i}/25 activos analizados en {elapsed:.1f}s | Señales encontradas: {len(results)}") # 4. Ordenar por confianza (de mayor a menor) results.sort(key=lambda x: x['confidence'], reverse=True) final_signals = results[:limit] total_time = time.time() - start_time logger.info(f"✅ Análisis completado en {total_time:.1f}s. Se retornan {len(final_signals)} señales de alta calidad.") return final_signals except Exception as e: logger.error(f"❌ Error crítico en get_best_signals: {e}", exc_info=True) return []