# source: https://raw.githubusercontent.com/MaicolO-O/trading/13e0c1e92daa6a64416f5d56f27bf585680e0583/IntelligentMomentumStrategyULTRATE.py """ 🚀 INTELLIGENT MOMENTUM STRATEGY ULTRA-TE 🚀 Estrategia híbrida ultra-avanzada que combina lo mejor de: - IntelligentMomentumStrategyPro (conservador) - SmartTradingStrategy (agresivo) Características principales: - Análisis multi-timeframe inteligente - Sistema de scoring avanzado (0-100) - Gestión de riesgo adaptativa - Trading bidireccional (LONG/SHORT) - Stop loss dinámico basado en ATR - Machine Learning integration ready - Detección de condiciones de mercado """ import numpy as np import pandas as pd from pandas import DataFrame from datetime import datetime, timedelta from typing import Optional, Union from functools import reduce import logging from freqtrade.strategy import IStrategy, informative, merge_informative_pair from freqtrade.strategy import (BooleanParameter, CategoricalParameter, DecimalParameter, IntParameter, RealParameter, timeframe_to_minutes) import talib.abstract as ta from freqtrade.persistence import Trade # Configurar logging logger = logging.getLogger(__name__) class Github_MaicolO_O_trading__IntelligentMomentumStrategyULTRATE__20250607_171022(IStrategy): """ 🎯 ESTRATEGIA HÍBRIDA ULTRA-AVANZADA Combina análisis conservador con oportunidades agresivas """ # === CONFIGURACIÓN BÁSICA === INTERFACE_VERSION = 3 # Configuración de timeframes timeframe = '5m' # Timeframe principal optimizado can_short = False # Habilitar trading bidireccional # ROI dinámico - Será manejado por trailing stop inteligente minimal_roi = { "0": 0.12, # 12% ROI máximo teórico "30": 0.08, # 8% después de 30 min "60": 0.05, # 5% después de 1 hora "120": 0.03, # 3% después de 2 horas "240": 0.02, # 2% después de 4 horas "480": 0.01 # 1% después de 8 horas } # Stop loss base - Será dinámico stoploss = -0.08 # -8% como máximo de seguridad # Trailing stop ultra-inteligente trailing_stop = True trailing_stop_positive = 0.015 # 1.5% ganancia mínima trailing_stop_positive_offset = 0.025 # 2.5% offset trailing_only_offset_is_reached = True # Configuración avanzada use_exit_signal = True exit_profit_only = False ignore_roi_if_entry_signal = False # Número de velas para análisis startup_candle_count: int = 300 # Configuración de órdenes order_types = { 'entry': 'limit', 'exit': 'limit', 'stoploss': 'market', 'stoploss_on_exchange': True, 'stoploss_on_exchange_interval': 60, 'stoploss_on_exchange_limit_ratio': 0.99, } # === PARÁMETROS OPTIMIZABLES ULTRA === # Modo de operación trading_mode = CategoricalParameter( ['conservative', 'aggressive', 'adaptive'], default='adaptive', space="buy" ) # Timeframes informativos use_multi_timeframe = BooleanParameter(default=True, space="buy") # Parámetros RSI rsi_period = IntParameter(12, 16, default=14, space="buy") rsi_oversold = IntParameter(25, 35, default=30, space="buy") rsi_overbought = IntParameter(65, 75, default=70, space="sell") # Parámetros MACD macd_fast = IntParameter(10, 14, default=12, space="buy") macd_slow = IntParameter(24, 28, default=26, space="buy") macd_signal = IntParameter(8, 10, default=9, space="buy") # Parámetros EMA ema_fast = IntParameter(8, 12, default=10, space="buy") ema_medium = IntParameter(18, 24, default=21, space="buy") ema_slow = IntParameter(45, 55, default=50, space="buy") # Bollinger Bands bb_period = IntParameter(18, 22, default=20, space="buy") bb_std = DecimalParameter(1.8, 2.2, default=2.0, space="buy") # ADX - Fuerza de tendencia adx_period = IntParameter(12, 16, default=14, space="buy") adx_threshold = IntParameter(20, 30, default=25, space="buy") # Volumen volume_check = BooleanParameter(default=True, space="buy") volume_multiplier = DecimalParameter(1.2, 2.5, default=1.5, space="buy") # Sistema de scoring min_score_entry_long = IntParameter(65, 80, default=75, space="buy") min_score_entry_short = IntParameter(65, 80, default=75, space="sell") min_score_exit = IntParameter(25, 35, default=30, space="sell") # Gestión de riesgo max_volatility = DecimalParameter(0.02, 0.05, default=0.03, space="buy") atr_multiplier = DecimalParameter(1.5, 3.0, default=2.0, space="buy") # Filtros de mercado market_condition_filter = BooleanParameter(default=True, space="buy") trend_strength_min = IntParameter(20, 40, default=30, space="buy") def informative_pairs(self): """ Pares informativos para análisis multi-timeframe """ pairs = self.dp.current_whitelist() informative_pairs = [] if self.use_multi_timeframe.value: for pair in pairs: informative_pairs.append((pair, '1m')) # Micro-tendencias informative_pairs.append((pair, '15m')) # Tendencias cortas informative_pairs.append((pair, '1h')) # Tendencias medias informative_pairs.append((pair, '4h')) # Tendencias largas return informative_pairs def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """ 🔧 Poblado de indicadores ultra-avanzados """ # === INDICADORES BÁSICOS MEJORADOS === # RSI multinivel dataframe['rsi'] = ta.RSI(dataframe, timeperiod=self.rsi_period.value) dataframe['rsi_fast'] = ta.RSI(dataframe, timeperiod=7) dataframe['rsi_slow'] = ta.RSI(dataframe, timeperiod=21) # MACD optimizado macd = ta.MACD(dataframe, fastperiod=self.macd_fast.value, slowperiod=self.macd_slow.value, signalperiod=self.macd_signal.value) dataframe['macd'] = macd['macd'] dataframe['macd_signal'] = macd['macdsignal'] dataframe['macd_histogram'] = macd['macdhist'] # MACD adicional para confirmación macd_fast = ta.MACD(dataframe, fastperiod=8, slowperiod=21, signalperiod=5) dataframe['macd_fast'] = macd_fast['macd'] dataframe['macd_signal_fast'] = macd_fast['macdsignal'] # === MEDIAS MÓVILES MÚLTIPLES === dataframe['ema_fast'] = ta.EMA(dataframe, timeperiod=self.ema_fast.value) dataframe['ema_medium'] = ta.EMA(dataframe, timeperiod=self.ema_medium.value) dataframe['ema_slow'] = ta.EMA(dataframe, timeperiod=self.ema_slow.value) dataframe['ema_200'] = ta.EMA(dataframe, timeperiod=200) dataframe['sma_10'] = ta.SMA(dataframe, timeperiod=10) dataframe['sma_20'] = ta.SMA(dataframe, timeperiod=20) dataframe['sma_50'] = ta.SMA(dataframe, timeperiod=50) dataframe['sma_100'] = ta.SMA(dataframe, timeperiod=100) # === BOLLINGER BANDS AVANZADAS === bollinger = ta.BBANDS(dataframe, timeperiod=self.bb_period.value, nbdevup=self.bb_std.value, nbdevdn=self.bb_std.value) dataframe['bb_upper'] = bollinger['upperband'] dataframe['bb_middle'] = bollinger['middleband'] dataframe['bb_lower'] = bollinger['lowerband'] dataframe['bb_percent'] = (dataframe['close'] - dataframe['bb_lower']) / (dataframe['bb_upper'] - dataframe['bb_lower']) dataframe['bb_width'] = (dataframe['bb_upper'] - dataframe['bb_lower']) / dataframe['bb_middle'] dataframe['bb_width_sma'] = ta.SMA(dataframe['bb_width'], timeperiod=20) dataframe['bb_expansion'] = dataframe['bb_width'] > dataframe['bb_width_sma'] * 1.2 dataframe['bb_squeeze'] = dataframe['bb_width'] < dataframe['bb_width_sma'] * 0.8 # === INDICADORES DE MOMENTUM === # Stochastic stoch = ta.STOCH(dataframe) dataframe['stoch_k'] = stoch['slowk'] dataframe['stoch_d'] = stoch['slowd'] # Williams %R dataframe['williams_r'] = ta.WILLR(dataframe) # CCI dataframe['cci'] = ta.CCI(dataframe) # === INDICADORES DE TENDENCIA === # ADX - Fuerza de tendencia dataframe['adx'] = ta.ADX(dataframe, timeperiod=self.adx_period.value) dataframe['plus_di'] = ta.PLUS_DI(dataframe, timeperiod=self.adx_period.value) dataframe['minus_di'] = ta.MINUS_DI(dataframe, timeperiod=self.adx_period.value) # Parabolic SAR dataframe['sar'] = ta.SAR(dataframe) # === INDICADORES DE VOLUMEN === dataframe['volume_sma'] = ta.SMA(dataframe['volume'], timeperiod=20) dataframe['volume_ratio'] = dataframe['volume'] / dataframe['volume_sma'] # OBV (On Balance Volume) dataframe['obv'] = ta.OBV(dataframe) dataframe['obv_ema'] = ta.EMA(dataframe['obv'], timeperiod=10) # A/D Line dataframe['ad'] = ta.AD(dataframe) # === INDICADORES DE VOLATILIDAD === # ATR (Average True Range) dataframe['atr'] = ta.ATR(dataframe, timeperiod=14) dataframe['atr_ratio'] = dataframe['atr'] / dataframe['close'] dataframe['natr'] = ta.NATR(dataframe) # === PATRONES DE VELAS === dataframe['hammer'] = ta.CDLHAMMER(dataframe) dataframe['doji'] = ta.CDLDOJI(dataframe) dataframe['engulfing'] = ta.CDLENGULFING(dataframe) dataframe['shooting_star'] = ta.CDLSHOOTINGSTAR(dataframe) dataframe['hanging_man'] = ta.CDLHANGINGMAN(dataframe) # === ANÁLISIS MULTI-TIMEFRAME === if self.use_multi_timeframe.value: dataframe = self.analyze_multi_timeframe(dataframe, metadata) # === ANÁLISIS DE CONTEXTO DE MERCADO === dataframe = self.detect_market_condition(dataframe) # === SISTEMA DE SCORING ULTRA === dataframe = self.calculate_ultra_score(dataframe) # === SEÑALES PERSONALIZADAS === dataframe = self.calculate_custom_signals(dataframe) return dataframe def analyze_multi_timeframe(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """ 🔍 Análisis multi-timeframe inteligente """ if not self.dp: return dataframe pair = metadata['pair'] # Análisis 1m - Micro-tendencias try: inf_1m = self.dp.get_pair_dataframe(pair=pair, timeframe='1m') if not inf_1m.empty: inf_1m['ema_21_1m'] = ta.EMA(inf_1m, timeperiod=21) inf_1m['rsi_1m'] = ta.RSI(inf_1m, timeperiod=14) inf_1m['trend_1m'] = inf_1m['close'] > inf_1m['ema_21_1m'] # Merge con resampling dataframe = merge_informative_pair(dataframe, inf_1m, self.timeframe, '1m', ffill=True, append_timeframe=True) except Exception as e: logger.warning(f"Error obteniendo datos 1m: {e}") # Análisis 15m - Tendencias cortas try: inf_15m = self.dp.get_pair_dataframe(pair=pair, timeframe='15m') if not inf_15m.empty: inf_15m['ema_50_15m'] = ta.EMA(inf_15m, timeperiod=50) inf_15m['rsi_15m'] = ta.RSI(inf_15m, timeperiod=14) inf_15m['adx_15m'] = ta.ADX(inf_15m, timeperiod=14) inf_15m['trend_15m'] = inf_15m['close'] > inf_15m['ema_50_15m'] dataframe = merge_informative_pair(dataframe, inf_15m, self.timeframe, '15m', ffill=True, append_timeframe=True) except Exception as e: logger.warning(f"Error obteniendo datos 15m: {e}") # Análisis 1h - Tendencias medias try: inf_1h = self.dp.get_pair_dataframe(pair=pair, timeframe='1h') if not inf_1h.empty: inf_1h['ema_50_1h'] = ta.EMA(inf_1h, timeperiod=50) inf_1h['ema_200_1h'] = ta.EMA(inf_1h, timeperiod=200) inf_1h['rsi_1h'] = ta.RSI(inf_1h, timeperiod=14) inf_1h['trend_1h'] = inf_1h['close'] > inf_1h['ema_200_1h'] dataframe = merge_informative_pair(dataframe, inf_1h, self.timeframe, '1h', ffill=True, append_timeframe=True) except Exception as e: logger.warning(f"Error obteniendo datos 1h: {e}") # Análisis 4h - Tendencias largas try: inf_4h = self.dp.get_pair_dataframe(pair=pair, timeframe='4h') if not inf_4h.empty: inf_4h['ema_200_4h'] = ta.EMA(inf_4h, timeperiod=200) inf_4h['rsi_4h'] = ta.RSI(inf_4h, timeperiod=14) inf_4h['trend_4h'] = inf_4h['close'] > inf_4h['ema_200_4h'] dataframe = merge_informative_pair(dataframe, inf_4h, self.timeframe, '4h', ffill=True, append_timeframe=True) except Exception as e: logger.warning(f"Error obteniendo datos 4h: {e}") return dataframe def detect_market_condition(self, dataframe: DataFrame) -> DataFrame: """ 🎯 Detección inteligente de condiciones de mercado """ # Alineación de EMAs dataframe['ema_bullish_alignment'] = ( (dataframe['ema_fast'] > dataframe['ema_medium']) & (dataframe['ema_medium'] > dataframe['ema_slow']) ) dataframe['ema_bearish_alignment'] = ( (dataframe['ema_fast'] < dataframe['ema_medium']) & (dataframe['ema_medium'] < dataframe['ema_slow']) ) # Fuerza de tendencia dataframe['trend_strength'] = dataframe['adx'] dataframe['is_trending'] = dataframe['trend_strength'] > self.trend_strength_min.value # Análisis de volatilidad dataframe['is_high_volatility'] = dataframe['atr_ratio'] > self.max_volatility.value dataframe['is_low_volatility'] = dataframe['atr_ratio'] < (self.max_volatility.value * 0.5) # Detección de mercado lateral dataframe['is_sideways'] = ( (~dataframe['is_trending']) | (dataframe['bb_width'] < dataframe['bb_width_sma'] * 0.8) ) # Condiciones de mercado favorables dataframe['market_favorable_long'] = ( dataframe['ema_bullish_alignment'] & dataframe['is_trending'] & ~dataframe['is_high_volatility'] & (dataframe['close'] > dataframe['ema_fast']) ) dataframe['market_favorable_short'] = ( dataframe['ema_bearish_alignment'] & dataframe['is_trending'] & ~dataframe['is_high_volatility'] & (dataframe['close'] < dataframe['ema_fast']) ) # Detección de breakout dataframe['breakout_up'] = ( (dataframe['close'] > dataframe['bb_upper']) & (dataframe['volume_ratio'] > 1.5) & (dataframe['rsi'] > 50) ) dataframe['breakout_down'] = ( (dataframe['close'] < dataframe['bb_lower']) & (dataframe['volume_ratio'] > 1.5) & (dataframe['rsi'] < 50) ) return dataframe def calculate_ultra_score(self, dataframe: DataFrame) -> DataFrame: """ 🏆 Sistema de scoring ultra-avanzado (0-100) """ # Inicializar scores dataframe['long_score'] = 0 dataframe['short_score'] = 0 # === TENDENCIA MULTI-TIMEFRAME (25 puntos) === # Tendencia principal (10 puntos) dataframe.loc[dataframe['ema_bullish_alignment'], 'long_score'] += 10 dataframe.loc[dataframe['ema_bearish_alignment'], 'short_score'] += 10 # Tendencias superiores (15 puntos) - si disponible if 'trend_1h_1h' in dataframe.columns: dataframe.loc[dataframe['trend_1h_1h'] == True, 'long_score'] += 5 dataframe.loc[dataframe['trend_1h_1h'] == False, 'short_score'] += 5 if 'trend_4h_4h' in dataframe.columns: dataframe.loc[dataframe['trend_4h_4h'] == True, 'long_score'] += 10 dataframe.loc[dataframe['trend_4h_4h'] == False, 'short_score'] += 10 # === INDICADORES TÉCNICOS (25 puntos) === # RSI (8 puntos) dataframe.loc[ (dataframe['rsi'] > 40) & (dataframe['rsi'] < 60), 'long_score' ] += 4 dataframe.loc[ (dataframe['rsi'] > 40) & (dataframe['rsi'] < 60), 'short_score' ] += 4 dataframe.loc[dataframe['rsi'] > dataframe['rsi'].shift(1), 'long_score'] += 4 dataframe.loc[dataframe['rsi'] < dataframe['rsi'].shift(1), 'short_score'] += 4 # MACD (8 puntos) dataframe.loc[ (dataframe['macd'] > dataframe['macd_signal']) & (dataframe['macd_histogram'] > 0), 'long_score' ] += 8 dataframe.loc[ (dataframe['macd'] < dataframe['macd_signal']) & (dataframe['macd_histogram'] < 0), 'short_score' ] += 8 # Stochastic (5 puntos) dataframe.loc[ (dataframe['stoch_k'] > dataframe['stoch_d']) & (dataframe['stoch_k'] < 80), 'long_score' ] += 5 dataframe.loc[ (dataframe['stoch_k'] < dataframe['stoch_d']) & (dataframe['stoch_k'] > 20), 'short_score' ] += 5 # ADX (4 puntos) dataframe.loc[ (dataframe['adx'] > self.adx_threshold.value) & (dataframe['plus_di'] > dataframe['minus_di']), 'long_score' ] += 4 dataframe.loc[ (dataframe['adx'] > self.adx_threshold.value) & (dataframe['plus_di'] < dataframe['minus_di']), 'short_score' ] += 4 # === VOLUMEN Y MOMENTUM (20 puntos) === # Volumen confirmativo (10 puntos) dataframe.loc[ dataframe['volume_ratio'] > self.volume_multiplier.value, 'long_score' ] += 10 dataframe.loc[ dataframe['volume_ratio'] > self.volume_multiplier.value, 'short_score' ] += 10 # OBV (5 puntos) dataframe.loc[dataframe['obv'] > dataframe['obv_ema'], 'long_score'] += 5 dataframe.loc[dataframe['obv'] < dataframe['obv_ema'], 'short_score'] += 5 # Williams %R (5 puntos) dataframe.loc[ (dataframe['williams_r'] > -80) & (dataframe['williams_r'] < -20), 'long_score' ] += 5 dataframe.loc[ (dataframe['williams_r'] > -80) & (dataframe['williams_r'] < -20), 'short_score' ] += 5 # === PATRONES DE VELAS (15 puntos) === # Patrones bullish bullish_patterns = ( (dataframe['hammer'] > 0) | (dataframe['engulfing'] > 0) | (dataframe['close'] > dataframe['open']) ) dataframe.loc[bullish_patterns, 'long_score'] += 15 # Patrones bearish bearish_patterns = ( (dataframe['shooting_star'] < 0) | (dataframe['hanging_man'] < 0) | (dataframe['engulfing'] < 0) | (dataframe['close'] < dataframe['open']) ) dataframe.loc[bearish_patterns, 'short_score'] += 15 # === CONTEXTO DE MERCADO (15 puntos) === # Mercado favorable (10 puntos) dataframe.loc[dataframe['market_favorable_long'], 'long_score'] += 10 dataframe.loc[dataframe['market_favorable_short'], 'short_score'] += 10 # Breakouts (5 puntos) dataframe.loc[dataframe['breakout_up'], 'long_score'] += 5 dataframe.loc[dataframe['breakout_down'], 'short_score'] += 5 # === PENALIZACIONES === # Mercado lateral (-20 puntos) dataframe.loc[dataframe['is_sideways'], 'long_score'] -= 20 dataframe.loc[dataframe['is_sideways'], 'short_score'] -= 20 # Alta volatilidad (-10 puntos) dataframe.loc[dataframe['is_high_volatility'], 'long_score'] -= 10 dataframe.loc[dataframe['is_high_volatility'], 'short_score'] -= 10 # Límites del score (0-100) dataframe['long_score'] = dataframe['long_score'].clip(0, 100) dataframe['short_score'] = dataframe['short_score'].clip(0, 100) return dataframe def calculate_custom_signals(self, dataframe: DataFrame) -> DataFrame: """ 🔄 Señales personalizadas avanzadas """ # Divergencia RSI bullish dataframe['rsi_divergence_bullish'] = ( (dataframe['rsi'] > dataframe['rsi'].shift(1)) & (dataframe['close'] < dataframe['close'].shift(1)) & (dataframe['rsi'].shift(1) < 35) ) # Divergencia RSI bearish dataframe['rsi_divergence_bearish'] = ( (dataframe['rsi'] < dataframe['rsi'].shift(1)) & (dataframe['close'] > dataframe['close'].shift(1)) & (dataframe['rsi'].shift(1) > 65) ) # Squeeze Bollinger Bands dataframe['bb_squeeze'] = dataframe['bb_width'] < dataframe['bb_width_sma'] * 0.8 dataframe['bb_expansion'] = dataframe['bb_width'] > dataframe['bb_width_sma'] * 1.2 # Confluencia de señales dataframe['bullish_confluence'] = ( (dataframe['rsi'] > 50) & (dataframe['macd'] > dataframe['macdsignal']) & (dataframe['ema_fast'] > dataframe['ema_medium']) & (dataframe['adx'] > self.adx_threshold.value) ) dataframe['bearish_confluence'] = ( (dataframe['rsi'] < 50) & (dataframe['macd'] < dataframe['macd_signal']) & (dataframe['ema_fast'] < dataframe['ema_medium']) & (dataframe['adx'] > self.adx_threshold.value) ) return dataframe def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """ 🚀 Lógica de entrada ultra-inteligente """ # === CONDICIONES DE ENTRADA LONG === long_conditions = [ # Score mínimo (dataframe['long_score'] >= self.min_score_entry_long.value), # Filtros básicos (dataframe['volume'] > 0), (~dataframe['is_sideways']) if self.market_condition_filter.value else True, # Confirmaciones técnicas (dataframe['rsi'] > self.rsi_oversold.value), (dataframe['rsi'] < 75), # No extremadamente sobrecomprado (dataframe['close'] > dataframe['ema_fast']), # Contexto de mercado (dataframe['market_favorable_long'] | ~self.market_condition_filter.value), # Volumen confirmativo (dataframe['volume_ratio'] > 1.0) if self.volume_check.value else True, ] # Condiciones adicionales según modo if self.trading_mode.value == 'conservative': long_conditions.extend([ (dataframe['long_score'] >= 80), (dataframe['trend_strength'] > 30), (dataframe['bullish_confluence']), ]) elif self.trading_mode.value == 'aggressive': long_conditions.extend([ (dataframe['long_score'] >= 65), (dataframe['breakout_up'] | dataframe['rsi_divergence_bullish']), ]) # Modo adaptive usa condiciones base # === CONDICIONES DE ENTRADA SHORT === # === CONDICIONES DE ENTRADA SHORT === short_conditions = [ # Score mínimo (dataframe['short_score'] >= self.min_score_entry_short.value), # Filtros básicos (dataframe['volume'] > 0), (~dataframe['is_sideways']) if self.market_condition_filter.value else True, # Confirmaciones técnicas (dataframe['rsi'] < self.rsi_overbought.value), (dataframe['rsi'] > 25), # No extremadamente sobrevendido (dataframe['close'] < dataframe['ema_fast']), # Contexto de mercado (dataframe['market_favorable_short'] | ~self.market_condition_filter.value), # Volumen confirmativo (dataframe['volume_ratio'] > 1.0) if self.volume_check.value else True, ] # Condiciones adicionales según modo if self.trading_mode.value == 'conservative': short_conditions.extend([ (dataframe['short_score'] >= 80), (dataframe['trend_strength'] > 30), (dataframe['bearish_confluence']), ]) elif self.trading_mode.value == 'aggressive': short_conditions.extend([ (dataframe['short_score'] >= 65), (dataframe['breakout_down'] | dataframe['rsi_divergence_bearish']), ]) # Modo adaptive usa condiciones base # Aplicar condiciones if long_conditions: dataframe.loc[ reduce(lambda x, y: x & y, long_conditions), 'enter_long'] = 1 if short_conditions and self.can_short: dataframe.loc[ reduce(lambda x, y: x & y, short_conditions), 'enter_short'] = 1 return dataframe def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """ 🎯 Señales de salida ultra-optimizadas """ # === CONDICIONES DE SALIDA LONG === exit_long_conditions = [ # Score bajo (dataframe['long_score'] <= self.min_score_exit.value), # RSI extremadamente sobrecomprado (dataframe['rsi'] > 80), # MACD bearish (dataframe['macd'] < dataframe['macd_signal']), # Tendencia revertida en múltiples timeframes (dataframe['trend_reversal_count'] >= 2), # Volumen decreciente + momentum perdido ((dataframe['volume_ratio'] < 0.8) & (dataframe['momentum_strength'] < 20)), ] # === CONDICIONES DE SALIDA SHORT === exit_short_conditions = [ # Score bajo (dataframe['short_score'] <= self.min_score_exit.value), # RSI extremadamente sobrevendido (dataframe['rsi'] < 20), # MACD bullish (dataframe['macd'] > dataframe['macd_signal']), # Tendencia revertida en múltiples timeframes (dataframe['trend_reversal_count'] >= 2), # Volumen decreciente + momentum perdido ((dataframe['volume_ratio'] < 0.8) & (dataframe['momentum_strength'] < 20)), ] # Aplicar condiciones de salida if self.use_exit_signal.value: if exit_long_conditions: dataframe.loc[ reduce(lambda x, y: x | y, exit_long_conditions), 'exit_long'] = 1 if exit_short_conditions and self.can_short: dataframe.loc[ reduce(lambda x, y: x | y, exit_short_conditions), 'exit_short'] = 1 return dataframe def analyze_multi_timeframe(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """ 🔄 Análisis cruzado de múltiples timeframes Analiza 1m, 15m, 1h para confirmación de señales """ if not self.dp: return dataframe pair = metadata['pair'] timeframes = ['1m', '15m', '1h'] # Contadores de tendencias dataframe['bullish_timeframes'] = 0 dataframe['bearish_timeframes'] = 0 dataframe['trend_reversal_count'] = 0 for tf in timeframes: try: # Obtener datos del timeframe tf_data = self.dp.get_pair_dataframe(pair=pair, timeframe=tf) if tf_data.empty: continue # Calcular indicadores básicos para el timeframe tf_data[f'ema_fast_{tf}'] = ta.EMA(tf_data, timeperiod=10) tf_data[f'ema_slow_{tf}'] = ta.EMA(tf_data, timeperiod=21) tf_data[f'rsi_{tf}'] = ta.RSI(tf_data, timeperiod=14) tf_data[f'macd_{tf}'] = ta.MACD(tf_data)['macd'] tf_data[f'macd_signal_{tf}'] = ta.MACD(tf_data)['macdsignal'] # Determinar tendencia del timeframe tf_data[f'trend_{tf}'] = np.where( (tf_data[f'ema_fast_{tf}'] > tf_data[f'ema_slow_{tf}']) & (tf_data[f'macd_{tf}'] > tf_data[f'macd_signal_{tf}']), 1, # Bullish np.where( (tf_data[f'ema_fast_{tf}'] < tf_data[f'ema_slow_{tf}']) & (tf_data[f'macd_{tf}'] < tf_data[f'macd_signal_{tf}']), -1, # Bearish 0 # Neutral ) ) # Detectar reversiones de tendencia tf_data[f'trend_reversal_{tf}'] = ( tf_data[f'trend_{tf}'] != tf_data[f'trend_{tf}'].shift(1) ) # Merge con dataframe principal merge_columns = [ 'date', f'trend_{tf}', f'rsi_{tf}', f'trend_reversal_{tf}' ] tf_data_merge = tf_data[merge_columns].copy() dataframe = pd.merge(dataframe, tf_data_merge, on='date', how='left') # Rellenar valores faltantes dataframe[f'trend_{tf}'].fillna(method='ffill', inplace=True) dataframe[f'rsi_{tf}'].fillna(method='ffill', inplace=True) dataframe[f'trend_reversal_{tf}'].fillna(False, inplace=True) # Actualizar contadores dataframe.loc[dataframe[f'trend_{tf}'] == 1, 'bullish_timeframes'] += 1 dataframe.loc[dataframe[f'trend_{tf}'] == -1, 'bearish_timeframes'] += 1 dataframe.loc[dataframe[f'trend_reversal_{tf}'], 'trend_reversal_count'] += 1 except Exception as e: logger.warning(f"Error analizando timeframe {tf}: {e}") continue # Fuerza de tendencia multi-timeframe dataframe['multi_tf_trend_strength'] = ( dataframe['bullish_timeframes'] - dataframe['bearish_timeframes'] ) * 100 / len(timeframes) # Consenso de timeframes dataframe['tf_consensus_bullish'] = dataframe['bullish_timeframes'] >= 2 dataframe['tf_consensus_bearish'] = dataframe['bearish_timeframes'] >= 2 return dataframe def calculate_ultra_score(self, dataframe: DataFrame) -> DataFrame: """ 🎯 Sistema de puntuación híbrido ultra-preciso (0-100 puntos) Distribución: - Tendencia Multi-timeframe: 25 puntos - Indicadores Técnicos: 25 puntos - Volumen y Momentum: 20 puntos - Patrones de Velas: 15 puntos - Contexto de Mercado: 15 puntos """ # Inicializar scores dataframe['long_score'] = 0.0 dataframe['short_score'] = 0.0 # === 1. TENDENCIA MULTI-TIMEFRAME (25 puntos) === # Long score - Tendencia dataframe.loc[dataframe['tf_consensus_bullish'], 'long_score'] += 25 dataframe.loc[dataframe['bullish_timeframes'] == 2, 'long_score'] += 20 dataframe.loc[dataframe['bullish_timeframes'] == 1, 'long_score'] += 10 # Short score - Tendencia dataframe.loc[dataframe['tf_consensus_bearish'], 'short_score'] += 25 dataframe.loc[dataframe['bearish_timeframes'] == 2, 'short_score'] += 20 dataframe.loc[dataframe['bearish_timeframes'] == 1, 'short_score'] += 10 # === 2. INDICADORES TÉCNICOS (25 puntos) === # RSI (8 puntos) dataframe.loc[(dataframe['rsi'] > 40) & (dataframe['rsi'] < 60), 'long_score'] += 8 dataframe.loc[(dataframe['rsi'] > 40) & (dataframe['rsi'] < 60), 'short_score'] += 8 dataframe.loc[dataframe['rsi'] < 40, 'long_score'] += 5 dataframe.loc[dataframe['rsi'] > 60, 'short_score'] += 5 # MACD (8 puntos) dataframe.loc[dataframe['macd'] > dataframe['macd_signal'], 'long_score'] += 8 dataframe.loc[dataframe['macd'] < dataframe['macd_signal'], 'short_score'] += 8 # Bollinger Bands (9 puntos) dataframe.loc[dataframe['bb_squeeze'] & dataframe['bb_expansion'], 'long_score'] += 9 dataframe.loc[dataframe['bb_squeeze'] & dataframe['bb_expansion'], 'short_score'] += 9 dataframe.loc[dataframe['close'] > dataframe['bb_middle'], 'long_score'] += 5 dataframe.loc[dataframe['close'] < dataframe['bb_middle'], 'short_score'] += 5 # === 3. VOLUMEN Y MOMENTUM (20 puntos) === # Volumen (10 puntos) dataframe.loc[dataframe['volume_ratio'] > 2.0, 'long_score'] += 10 dataframe.loc[dataframe['volume_ratio'] > 2.0, 'short_score'] += 10 dataframe.loc[dataframe['volume_ratio'] > 1.5, 'long_score'] += 7 dataframe.loc[dataframe['volume_ratio'] > 1.5, 'short_score'] += 7 dataframe.loc[dataframe['volume_ratio'] > 1.2, 'long_score'] += 4 dataframe.loc[dataframe['volume_ratio'] > 1.2, 'short_score'] += 4 # Momentum (10 puntos) dataframe.loc[dataframe['adx'] > 30, 'long_score'] += 10 dataframe.loc[dataframe['adx'] > 30, 'short_score'] += 10 dataframe.loc[dataframe['adx'] > 25, 'long_score'] += 7 dataframe.loc[dataframe['adx'] > 25, 'short_score'] += 7 # === 4. PATRONES DE VELAS (15 puntos) === # Patrones bullish bullish_candle_conditions = [ (dataframe['close'] > dataframe['open']), # Vela verde (dataframe['close'] > dataframe['high'].shift(1)), # Nuevo máximo ((dataframe['high'] - dataframe['close']) < (dataframe['close'] - dataframe['open']) * 0.3) # Poca sombra superior ] # Patrones bearish bearish_candle_conditions = [ (dataframe['close'] < dataframe['open']), # Vela roja (dataframe['close'] < dataframe['low'].shift(1)), # Nuevo mínimo ((dataframe['close'] - dataframe['low']) < (dataframe['open'] - dataframe['close']) * 0.3) # Poca sombra inferior ] # Aplicar puntos por patrones for condition in bullish_candle_conditions: dataframe.loc[condition, 'long_score'] += 5 for condition in bearish_candle_conditions: dataframe.loc[condition, 'short_score'] += 5 # === 5. CONTEXTO DE MERCADO (15 puntos) === # Confluencia de señales dataframe.loc[dataframe['bullish_confluence'], 'long_score'] += 15 dataframe.loc[dataframe['bearish_confluence'], 'short_score'] += 15 # Divergencias dataframe.loc[dataframe['rsi_divergence_bullish'], 'long_score'] += 10 dataframe.loc[dataframe['rsi_divergence_bearish'], 'short_score'] += 10 # Condición de mercado favorable dataframe.loc[dataframe['market_favorable_long'], 'long_score'] += 8 dataframe.loc[dataframe['market_favorable_short'], 'short_score'] += 8 # === PENALIZACIONES === # Mercado lateral dataframe.loc[dataframe['is_sideways'], 'long_score'] -= 20 dataframe.loc[dataframe['is_sideways'], 'short_score'] -= 20 # Volatilidad extrema dataframe.loc[dataframe['atr_ratio'] > 0.08, 'long_score'] -= 15 dataframe.loc[dataframe['atr_ratio'] > 0.08, 'short_score'] -= 15 # Limitar scores a rango 0-100 dataframe['long_score'] = dataframe['long_score'].clip(0, 100) dataframe['short_score'] = dataframe['short_score'].clip(0, 100) return dataframe def detect_market_condition(self, dataframe: DataFrame) -> DataFrame: """ 🎯 Detección automática de contexto de mercado """ # Análisis de volatilidad dataframe['volatility_percentile'] = dataframe['atr_ratio'].rolling(50).rank(pct=True) # Condiciones de mercado dataframe['market_condition'] = 'normal' # Mercado volátil dataframe.loc[dataframe['volatility_percentile'] > 0.8, 'market_condition'] = 'volatile' # Mercado lateral dataframe.loc[ (dataframe['adx'] < 20) & (dataframe['bb_width'] < dataframe['bb_width'].rolling(20).mean() * 0.8), 'market_condition' ] = 'sideways' # Mercado trending dataframe.loc[ (dataframe['adx'] > 30) & (dataframe['tf_consensus_bullish'] | dataframe['tf_consensus_bearish']), 'market_condition' ] = 'trending' # Mercado de ruptura dataframe.loc[ (dataframe['bb_expansion']) & (dataframe['volume_ratio'] > 2.0), 'market_condition' ] = 'breakout' return dataframe def adaptive_risk_management(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """ 🛡️ Gestión de riesgo contextual y adaptativa """ # Calcular tamaño de posición basado en confianza (score) dataframe['position_size_multiplier'] = 1.0 # Aumentar tamaño con score alto dataframe.loc[dataframe['long_score'] >= 85, 'position_size_multiplier'] = 1.5 dataframe.loc[dataframe['short_score'] >= 85, 'position_size_multiplier'] = 1.5 # Reducir tamaño con score medio dataframe.loc[ (dataframe['long_score'] >= 70) & (dataframe['long_score'] < 85), 'position_size_multiplier' ] = 1.2 dataframe.loc[ (dataframe['short_score'] >= 70) & (dataframe['short_score'] < 85), 'position_size_multiplier' ] = 1.2 # Reducir mucho con score bajo dataframe.loc[dataframe['long_score'] < 70, 'position_size_multiplier'] = 0.7 dataframe.loc[dataframe['short_score'] < 70, 'position_size_multiplier'] = 0.7 # Ajustes por condición de mercado dataframe.loc[dataframe['market_condition'] == 'volatile', 'position_size_multiplier'] *= 0.7 dataframe.loc[dataframe['market_condition'] == 'sideways', 'position_size_multiplier'] *= 0.5 dataframe.loc[dataframe['market_condition'] == 'trending', 'position_size_multiplier'] *= 1.2 dataframe.loc[dataframe['market_condition'] == 'breakout', 'position_size_multiplier'] *= 1.3 # Stop loss dinámico basado en ATR y score dataframe['dynamic_stoploss'] = np.where( dataframe['long_score'] >= 80, -0.02, # Stop más ajustado para alta confianza np.where( dataframe['long_score'] >= 70, -0.03, # Stop medio -0.05 # Stop más amplio para baja confianza ) ) return dataframe def custom_stoploss(self, pair: str, trade: Trade, current_time: datetime, current_rate: float, current_profit: float, **kwargs) -> float: """ 🛡️ Stop loss dinámico inteligente """ try: # Obtener datos actuales dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe) if dataframe.empty: return self.stoploss current_candle = dataframe.iloc[-1] # Stop loss basado en score de confianza if hasattr(current_candle, 'long_score'): score = current_candle['long_score'] if not trade.is_short else current_candle['short_score'] if score >= 85: dynamic_stop = -0.02 # 2% para alta confianza elif score >= 75: dynamic_stop = -0.03 # 3% para confianza media else: dynamic_stop = -0.05 # 5% para baja confianza else: dynamic_stop = self.stoploss # Ajuste por ATR if hasattr(current_candle, 'atr') and current_candle['atr'] > 0: atr_stop = -(current_candle['atr'] * 2.5) / current_rate dynamic_stop = max(dynamic_stop, atr_stop) # Trailing stop para ganancias if current_profit > 0.02: # Más del 2% de ganancia trailing_stop = -0.01 # Trailing de 1% return max(dynamic_stop, trailing_stop) return max(dynamic_stop, self.stoploss) except Exception as e: logger.error(f"Error en custom_stoploss: {e}") return self.stoploss def confirm_trade_entry(self, pair: str, order_type: str, amount: float, rate: float, time_in_force: str, current_time: datetime, entry_tag: Optional[str], side: str, **kwargs) -> bool: """ 🔍 Confirmación ultra-inteligente antes de entrada """ try: # Obtener datos actuales dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe) if dataframe.empty: return False current_candle = dataframe.iloc[-1] # Verificar score mínimo if side == "long": score = current_candle.get('long_score', 0) min_score = self.min_score_entry_long.value else: score = current_candle.get('short_score', 0) min_score = self.min_score_entry_short.value if score < min_score: logger.info(f"Trade rechazado para {pair}: Score {score} < {min_score}") return False # Verificar condición de mercado market_condition = current_candle.get('market_condition', 'normal') if market_condition == 'sideways' and self.market_condition_filter.value: logger.info(f"Trade rechazado para {pair}: Mercado lateral") return False # Verificar ratio riesgo/recompensa atr = current_candle.get('atr', 0) if atr > 0: if side == "long": potential_stop = rate - (atr * 2.5) potential_tp = rate + (atr * 4.0) rr_ratio = (potential_tp - rate) / (rate - potential_stop) else: potential_stop = rate + (atr * 2.5) potential_tp = rate - (atr * 4.0) rr_ratio = (rate - potential_tp) / (potential_stop - rate) if rr_ratio < 1.2: # Mínimo 1.2:1 logger.info(f"Trade rechazado para {pair}: RR ratio {rr_ratio:.2f} < 1.2") return False logger.info(f"✅ Trade confirmado para {pair}: Score {score}, Market: {market_condition}") return True except Exception as e: logger.error(f"Error en confirm_trade_entry: {e}") return False def custom_exit(self, pair: str, trade: Trade, current_time: datetime, current_rate: float, current_profit: float, **kwargs): """ 🎯 Lógica de salida ultra-inteligente """ try: # Obtener datos actuales dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe) if dataframe.empty: return None current_candle = dataframe.iloc[-1] # Salida por score bajo if trade.is_short: score = current_candle.get('short_score', 50) if score <= self.min_score_exit.value: return f"low_score_{score}" else: score = current_candle.get('long_score', 50) if score <= self.min_score_exit.value: return f"low_score_{score}" # Salida por cambio de condición de mercado market_condition = current_candle.get('market_condition', 'normal') if market_condition == 'sideways': return "market_sideways" # Salida por reversión de tendencia multi-timeframe trend_reversal_count = current_candle.get('trend_reversal_count', 0) if trend_reversal_count >= 2: return "trend_reversal" # Salida por momentum contrario fuerte if trade.is_short and current_candle.get('long_score', 0) >= 85: return "strong_counter_momentum" elif not trade.is_short and current_candle.get('short_score', 0) >= 85: return "strong_counter_momentum" # Take profit basado en score y ratio RR if current_profit > 0.03: # Más de 3% ganancia if score >= 90 and current_profit > 0.08: # Score muy alto, TP alto return "high_confidence_tp" elif score < 70 and current_profit > 0.04: # Score bajo, TP conservador return "low_confidence_tp" return None except Exception as e: logger.error(f"Error en custom_exit: {e}") return None # === FUNCIONES AUXILIARES === from functools import reduce import logging # Configurar logging para debugging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__)