import pandas as pd import ta as ta_lib from utils import setup_logger, retry_api logger = setup_logger() class Strategy: """ Aplica filtros tecnicos a los candidatos para decidir si entramos. Incluye validacion de tendencia macro (BTC) para evitar operar en contra del mercado. Usa la libreria 'ta' (Technical Analysis) para indicadores. """ def __init__(self, exchange): self.exchange = exchange self._btc_bullish = None # Cache para no consultar BTC por cada par @retry_api() def fetch_ohlcv_df(self, symbol, timeframe, limit=200): """Obtiene las velas japonesas y devuelve un DataFrame de Pandas.""" ohlcv = self.exchange.fetch_ohlcv(symbol, timeframe, limit=limit) df = pd.DataFrame(ohlcv, columns=["timestamp", "open", "high", "low", "close", "volume"]) df["timestamp"] = pd.to_datetime(df["timestamp"], unit="ms") return df def is_btc_bullish(self): """ Filtro macro: verifica que BTC este en tendencia alcista en 1h. Si BTC esta cayendo, es peligroso comprar altcoins. Se cachea por ciclo de ejecucion para no repetir la consulta. """ if self._btc_bullish is not None: return self._btc_bullish try: df_btc = self.fetch_ohlcv_df("BTC/USDT", "1h", limit=50) if len(df_btc) < 50: self._btc_bullish = False return False df_btc["ema_21"] = ta_lib.trend.EMAIndicator(df_btc["close"], window=21).ema_indicator() df_btc["ema_50"] = ta_lib.trend.EMAIndicator(df_btc["close"], window=50).ema_indicator() last = df_btc.iloc[-1] btc_price = float(last["close"]) btc_ema21 = float(last["ema_21"]) btc_ema50 = float(last["ema_50"]) is_bull = btc_price > btc_ema21 and btc_ema21 > btc_ema50 #if not is_bull: # logger.info( # f"[MACRO] BTC no alcista " # f"(Precio={btc_price:.0f}, EMA21={btc_ema21:.0f}, EMA50={btc_ema50:.0f}). Precaucion." # ) #else: # logger.info( # f"[MACRO] BTC alcista " # f"(Precio={btc_price:.0f} > EMA21={btc_ema21:.0f} > EMA50={btc_ema50:.0f})" # ) self._btc_bullish = is_bull return is_bull except Exception as e: logger.error(f"Error evaluando tendencia de BTC: {e}") self._btc_bullish = False return False def is_valid_entry(self, symbol): """ Filtro de entrada: 0. BTC debe estar en tendencia alcista (filtro macro) 1. RSI(14) en 5m entre 30 y 55 (zona de oportunidad, no sobrecompra) 2. EMA12 > EMA50 en 5m (tendencia alcista del par) 3. Precio actual < 92% del maximo de 7 dias (margen de subida) 4. Volumen reciente > 1.2x el promedio (confirmacion de interes) """ try: # 0. Filtro macro #if not self.is_btc_bullish(): # logger.info(f"[{symbol}] Rechazado: BTC no esta en tendencia alcista.") # return False if not self.is_btc_bullish(): logger.info(f"[{symbol}] Advertencia: BTC no alcista. Continuando (modo no estricto).") # 1. Maximo de los ultimos 7 dias df_1d = self.fetch_ohlcv_df(symbol, "1d", limit=7) if len(df_1d) < 7: return False max_7d = float(df_1d["high"].max()) current_price = float(df_1d.iloc[-1]["close"]) if current_price >= (max_7d * 0.97): logger.debug( f"[{symbol}] Rechazado: Precio demasiado cerca del maximo de 7d " f"({current_price:.4f} vs {max_7d:.4f})." ) return False # 2. Analisis tecnico en 5 minutos df_5m = self.fetch_ohlcv_df(symbol, "5m", limit=100) if len(df_5m) < 50: return False df_5m["rsi"] = ta_lib.momentum.RSIIndicator(df_5m["close"], window=14).rsi() df_5m["ema_12"] = ta_lib.trend.EMAIndicator(df_5m["close"], window=12).ema_indicator() df_5m["ema_50"] = ta_lib.trend.EMAIndicator(df_5m["close"], window=50).ema_indicator() df_5m["vol_sma"] = ta_lib.trend.SMAIndicator(df_5m["volume"], window=20).sma_indicator() last_row = df_5m.iloc[-1] rsi_val = float(last_row["rsi"]) ema12_val = float(last_row["ema_12"]) ema50_val = float(last_row["ema_50"]) vol_val = float(last_row["volume"]) vol_sma_val = float(last_row["vol_sma"]) rsi_cond = 35 <= rsi_val <= 60 ema_cond = ema12_val > ema50_val vol_cond = vol_val > (vol_sma_val * 1.05) # Filtro de tendencia local: precio por encima de EMA20 en 1h try: df_1h = self.fetch_ohlcv_df(symbol, "1h", limit=25) ema20_1h = float( ta_lib.trend.EMAIndicator(df_1h["close"], window=20) .ema_indicator() .iloc[-1] ) price_1h = float(df_1h.iloc[-1]["close"]) trend_cond = price_1h > ema20_1h except Exception: trend_cond = True # Si falla, no bloqueamos la entrada if rsi_cond and ema_cond and vol_cond and trend_cond: logger.info( f"[{symbol}] ENTRADA VALIDA | " f"RSI: {rsi_val:.2f}, EMA12: {ema12_val:.4f} > EMA50: {ema50_val:.4f}, " f"Vol: {vol_val:.0f} > Avg: {vol_sma_val:.0f}" ) return True else: reasons = [] if not rsi_cond: reasons.append(f"RSI={rsi_val:.1f} fuera de [30-55]") if not ema_cond: reasons.append("EMA12 < EMA50") if not vol_cond: reasons.append("Volumen insuficiente") logger.debug(f"[{symbol}] Rechazado: {', '.join(reasons)}") except Exception as e: logger.error(f"Error evaluando estrategia en {symbol}: {e}") return False def calculate_dynamic_tp_sl(self, symbol): """ Calcula TP y SL dinamicos basados en ATR (Average True Range). - Moneda muy volatil (ej: DOGE) -> TP y SL mas amplios - Moneda estable (ej: BTC) -> TP y SL mas ajustados TP = 2.0 x ATR, SL = 1.0 x ATR (ratio 2:1) Retorna (tp_pct, sl_pct) como decimales o None si falla. """ try: df_5m = self.fetch_ohlcv_df(symbol, "5m", limit=100) if len(df_5m) < 50: return None df_5m["atr"] = ta_lib.volatility.AverageTrueRange( df_5m["high"], df_5m["low"], df_5m["close"], window=14 ).average_true_range() atr_val = float(df_5m.iloc[-1]["atr"]) current_price = float(df_5m.iloc[-1]["close"]) if current_price <= 0 or atr_val <= 0: return None atr_pct = atr_val / current_price tp_pct = max(0.008, min(atr_pct * 2.0, 0.05)) # Entre 0.8% y 5% sl_pct = max(0.004, min(atr_pct * 1.0, 0.025)) # Entre 0.4% y 2.5% logger.info( f"[{symbol}] ATR dinamico: ATR={atr_val:.6f} ({atr_pct * 100:.2f}%) -> " f"TP={tp_pct * 100:.2f}%, SL={sl_pct * 100:.2f}%" ) return tp_pct, sl_pct except Exception as e: logger.error(f"Error calculando ATR para {symbol}: {e}") return None def is_valid_hyperscalping(self, symbol): """ (Modo Opcional) Entrada rapida usando cruce de EMAs y volumen en 1 minuto. """ try: df_1m = self.fetch_ohlcv_df(symbol, "1m", limit=60) if len(df_1m) < 20: return False df_1m["ema_5"] = ta_lib.trend.EMAIndicator(df_1m["close"], window=5).ema_indicator() df_1m["ema_15"] = ta_lib.trend.EMAIndicator(df_1m["close"], window=15).ema_indicator() last_ema5 = float(df_1m.iloc[-1]["ema_5"]) last_ema15 = float(df_1m.iloc[-1]["ema_15"]) prev_ema5 = float(df_1m.iloc[-2]["ema_5"]) prev_ema15 = float(df_1m.iloc[-2]["ema_15"]) cross_up = prev_ema5 <= prev_ema15 and last_ema5 > last_ema15 if cross_up: logger.info(f"[{symbol}] Hyperscalping: Cruce alcista EMA5/EMA15 detectado.") return True except Exception as e: logger.error(f"Error evaluando hyperscalping en {symbol}: {e}") return False