from freqtrade.strategy import IStrategy, IntParameter, DecimalParameter, merge_informative_pair, informative from pandas import DataFrame import talib as ta import numpy as np import pandas as pd from datetime import datetime, timezone, timedelta import pytz from scipy.signal import argrelextrema import json import os import arrow import logging logger = None def setup_logger(runmode: str): global logger print(f"[setup_logger] Inicializando logger para runmode: {runmode}") date_str = datetime.now().strftime('%Y-%m-%d') #log_file_path = f"/root/freqtrade/logfile_{__name__}_{runmode}_{date_str}.log" log_file_path = os.path.expanduser(f"~/freqtrade/logfile_{__name__}_{runmode}_{date_str}.log") logger = logging.getLogger(f"strategy_logger_{__name__}_{runmode}") logger.setLevel(logging.INFO) # 🔐 Asegura que se agrega el handler si no existe if not any(isinstance(h, logging.FileHandler) and h.baseFilename == log_file_path for h in logger.handlers): handler = logging.FileHandler(log_file_path, mode="a", encoding="utf-8") formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s') handler.setFormatter(formatter) logger.addHandler(handler) # ✔ Prueba de que el logger está configurado print(f"[setup_logger] Logger inicializado en ruta: {log_file_path}") logger.info("Logger configurado correctamente") class Hermes_v10(IStrategy): timeframe = '1m' inf_timeframes = ['5m', '1d'] optimized_params = {} trades_count = 0 minimal_roi = { "0": 0.11 } use_custom_stoploss = True use_custom_exit = True trailing_stop = False def __init__(self, config: dict) -> None: super().__init__(config) self.load_optimized_parameters() runmode = self.config.get("runmode", "unknown") setup_logger(runmode) logger.info(f"Logger de estrategia {__name__} inicializado.") if logger and logger.hasHandlers(): print(f"Logger inicializado en ruta: {logger.handlers[0].baseFilename}") else: print("⚠️ Logger no tiene handlers. No se ha inicializado correctamente.") def load_optimized_parameters(self): """ Cargar todos los archivos .json de parámetros optimizados en una carpeta. """ if self.is_hyperopt(): return {} directory = 'user_data/strategies/' for filename in os.listdir(directory): if filename.endswith('.json') and filename.startswith(self.__class__.__name__): pair_name = filename.replace('.json', '').replace(f"{self.__class__.__name__}_", '').replace('_', '/') print(f"Buscamos el archivo {filename} con el pair_name {pair_name}") with open(os.path.join(directory, filename)) as f: print(f"Cargamos el json de {filename} con el pair_name {pair_name}") self.optimized_params[pair_name] = json.load(f) def get_pair_params(self, pair: str) -> dict: """ Devuelve los parámetros para el par dado. """ result = self.optimized_params.get(pair, {}) return result def is_hyperopt(self) -> bool: return self.config.get('runmode') == 'hyperopt' def shift_if_needed(self, series): #if self.config.get('runmode') == 'backtest' or self.config.get('runmode') == 'hyperopt': #return series.shift(1) return series @informative('5m') def populate_indicators_5m(self, dataframe: DataFrame, metadata: dict) -> DataFrame: dataframe['rsi_5m'] = ta.RSI(dataframe['close'], timeperiod=14) dataframe['sma_rsi_5m'] = ta.SMA(dataframe['rsi_5m'], timeperiod=14) return dataframe @informative('1d') def populate_indicators_1d(self, dataframe: DataFrame, metadata: dict) -> DataFrame: dataframe['rsi_1d'] = ta.RSI(dataframe['close'], timeperiod=14) dataframe['sma_rsi_1d'] = ta.SMA(dataframe['rsi_1d'], timeperiod=14) dataframe['pct'] = ((dataframe['close'] - dataframe['open']) / dataframe['open']) * 100 dataframe['pct_prev'] = dataframe['pct'].shift(1) dataframe['ma_7'] = ta.SMA(dataframe['close'], timeperiod=7) dataframe['ma_25'] = ta.SMA(dataframe['close'], timeperiod=25) dataframe['ma_99'] = ta.SMA(dataframe['close'], timeperiod=99) return dataframe # Parámetros configurables rsi_window = IntParameter(10, 150, default=30, space='buy', optimize=True) rsi_range = DecimalParameter(0.1, 10.0, default=0.8, decimals=2, space='buy', optimize=True) oscillation_threshold = DecimalParameter(0.01, 1.0, default=0.05, decimals=3, space='buy', optimize=True) rsi_5m_buy_threshold = IntParameter(50, 80, default=65, space='buy', optimize=True) volume_ratio_threshold_1 = DecimalParameter(0.05, 1.0, default=0.1, decimals=2, space='buy', optimize=True) volume_ratio_threshold_2 = DecimalParameter(0.05, 1.0, default=0.2, decimals=2, space='buy', optimize=True) oscillation_window = IntParameter(10, 60, default=30, space='buy', optimize=True) atr_multiplier_stoploss = DecimalParameter(0.1, 3.0, default=0.8, decimals=2, space='sell', optimize=True) atr_multiplier_roi = DecimalParameter(1.0, 5.0, default=2.0, decimals=2, space='sell', optimize=True) divergence_window = IntParameter(3, 10, default=6, space='buy', optimize=False) def is_causal_minimum(self, series: pd.Series, idx: int, window: int) -> bool: """ Verifica si el índice `idx` es un mínimo local mirando solo hacia atrás. """ if idx < window: return False prev_window = series[idx - window:idx + 1] # incluye el actual return series[idx] == min(prev_window) def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame: print(f"populate_indicators") # Verifica que el dataframe no esté vacío y contenga la columna 'close' if dataframe is None or dataframe.empty or 'close' not in dataframe.columns: return dataframe print(f"buscamos los parámetros para el par {metadata['pair']}") params = self.get_pair_params(metadata['pair']) rsi_window = self.rsi_window.value rsi_range = self.rsi_range.value oscillation_threshold = self.oscillation_threshold.value rsi_5m_buy_threshold = self.rsi_5m_buy_threshold.value volume_ratio_threshold_1 = self.volume_ratio_threshold_1.value volume_ratio_threshold_2 = self.volume_ratio_threshold_2.value oscillation_window = self.oscillation_window.value atr_multiplier_stoploss = self.atr_multiplier_stoploss.value atr_multiplier_roi = self.atr_multiplier_roi.value divergence_window = self.divergence_window.value # Si existen valores optimizados para este par, los usamos if params and not self.is_hyperopt(): print(f"parámetros encontrados para el par {metadata['pair']}") buy_params = params.get('params', {}).get('buy', {}) sell_params = params.get('params', {}).get('sell', {}) rsi_window = buy_params.get('rsi_window', 30) rsi_range = buy_params.get('rsi_range', 0.8) oscillation_threshold = buy_params.get('oscillation_threshold', 0.05) rsi_5m_buy_threshold = buy_params.get('rsi_5m_buy_threshold', 65) volume_ratio_threshold_1 = buy_params.get('volume_ratio_threshold_1', 0.1) volume_ratio_threshold_2 = buy_params.get('volume_ratio_threshold_2', 0.2) oscillation_window = buy_params.get('oscillation_window', 30) atr_multiplier_stoploss = sell_params.get('atr_multiplier_stoploss', 0.8) atr_multiplier_roi = sell_params.get('atr_multiplier_roi', 2.0) divergence_window = buy_params.get('divergence_window', 6) #logger.info(f"[{metadata['pair']}] Cálculo de indicadores iniciado.") velas = 60 dataframe['change_pct'] = (dataframe['close'] - dataframe['close'].shift(velas)) / dataframe['close'].shift(velas) * 100 dataframe['acceleration'] = dataframe['change_pct'] - dataframe['change_pct'].shift(1) # Calcula el RSI dataframe['rsi'] = ta.RSI(dataframe['close'], timeperiod=14) # Calcula las medias móviles dataframe['ma_7'] = ta.SMA(dataframe['close'], timeperiod=7) dataframe['ma_25'] = ta.SMA(dataframe['close'], timeperiod=25) dataframe['ma_99'] = ta.SMA(dataframe['close'], timeperiod=99) dataframe['ma_7_above_25'] = dataframe['ma_7'] > dataframe['ma_25'] dataframe['ma_7_above_99'] = dataframe['ma_7'] > dataframe['ma_99'] dataframe['ma_25_above_99'] = dataframe['ma_25'] > dataframe['ma_99'] # Detecta si en las últimas 10 velas la ma_25 ha cruzado hacia abajo la ma_99 velas_cruce = 15 #dataframe['crossover_ma_25_99'] = ( # (dataframe['ma_25'] < dataframe['ma_99']) & # (dataframe['ma_25'].shift(1) > dataframe['ma_99'].shift(1)) #).rolling(window=velas_cruce).max().fillna(0).astype(bool) #dataframe['crossover_ma_7_25'] = ( # (dataframe['ma_7'] < dataframe['ma_25']) & # (dataframe['ma_7'].shift(1) > dataframe['ma_25'].shift(1)) #).rolling(window=velas_cruce).max().fillna(0).astype(bool) dataframe['ma_7_above_25_1d'] = dataframe['ma_7_1d'] > dataframe['ma_25_1d'] dataframe['ma_7_above_99_1d'] = dataframe['ma_7_1d'] > dataframe['ma_99_1d'] dataframe['ma_25_above_99_1d'] = dataframe['ma_25_1d'] > dataframe['ma_99_1d'] # Calcula el RSI mínimo para la ventana configurada dataframe['rsi_min_custom'] = dataframe['rsi'].rolling(window=rsi_window).min() # Calcula la pendiente de la MA(99) en los últimos 30 minutos # Calcula la oscilación absoluta del precio por minuto en los últimos 30 minutos dataframe['abs_oscillation_pct'] = ((dataframe['high'] - dataframe['low']) / dataframe['close']) * 100 dataframe['avg_abs_oscillation_pct_30m'] = dataframe['abs_oscillation_pct'].rolling(window=oscillation_window).mean() # BEGIN - Detectar divergencias (usamos una ventana móvil de N barras para identificar picos/valles) window = divergence_window # Detectar mínimos locales en RSI y precio dataframe['rsi_min_local'] = 0 dataframe['price_min_local'] = 0 dataframe['rsi_max_local'] = 0 dataframe['price_max_local'] = 0 dataframe['bullish_divergence'] = 0 dataframe['bearish_divergence'] = 0 # Detectar si la última vela está abierta if 'date' in dataframe.columns: last_candle_time = dataframe['date'].iloc[-1] elif 'timestamp' in dataframe.columns: last_candle_time = dataframe['timestamp'].iloc[-1] else: raise ValueError("No hay columna de fecha para comprobar la hora de la última vela.") timeframe_minutes = int(self.timeframe.replace('m', '')) now = arrow.utcnow().floor('minute') expected_close_time = now.shift(minutes=-now.minute % timeframe_minutes) logger.info(f"[{metadata['pair']}][{now}] Cálculo de indicadores. expected_close_time: {expected_close_time} last_candle_time: {last_candle_time} dw: {divergence_window}") #print(f"now: {now} expected_close_time: {expected_close_time} last_candle_time: {last_candle_time}") #logger.info(f"now: {now} expected_close_time: {expected_close_time} last_candle_time: {last_candle_time} runmode: {self.config.get('runmode')}") # if self.config.get('runmode') not in ['backtest', 'hyperopt'] and last_candle_time >= now.datetime: # print(f"cogemos la vela anterior") # logger.info(f"[{metadata['pair']}] Cogemos la vela anterior.") # extrema_df = dataframe.iloc[:-1].copy() # else: # extrema_df = dataframe.copy() # extrema_df = dataframe.copy() runmode = self.config.get("runmode", "") # Si hay suficientes datos, calcular los extremos if len(dataframe) > window: # Detectar mínimos locales rsi_mins = argrelextrema(dataframe['rsi'].values, np.less_equal, order=window)[0] price_mins = argrelextrema(dataframe['close'].values, np.less_equal, order=window)[0] dataframe.loc[dataframe.index[rsi_mins], 'rsi_min_local'] = 1 dataframe.loc[dataframe.index[price_mins], 'price_min_local'] = 1 # ---------------------------- # 🟢 Divergencias Alcistas bullish_divs = [] for i in range(1, len(rsi_mins)): i1 = rsi_mins[i - 1] i2 = rsi_mins[i] if i1 in price_mins and i2 in price_mins: # Evita usar divergencias que terminen fuera de la penúltima vela (solo en dry_run y live) if runmode in ("dry_run", "live") and dataframe.index[i2] != dataframe.index[-2]: continue rsi1, rsi2 = dataframe.loc[dataframe.index[[i1, i2]], 'rsi'] price1, price2 = dataframe.loc[dataframe.index[[i1, i2]], 'close'] date1, date2 = dataframe.loc[dataframe.index[[i1, i2]], 'date'] #print(f"[{metadata['pair']}][{date1} - {date2}] i1: {i1} i2: {i2}") date1_datetime = date1 date2_datetime = date2 if isinstance(date1_datetime, str): date1_datetime = pd.to_datetime(date1_datetime) if isinstance(date2_datetime, str): date2_datetime = pd.to_datetime(date2_datetime) diff_minutes = (date2_datetime - date1_datetime).total_seconds() / 60 if rsi2 > rsi1 and price2 < price1: # Guardar info solo si hay divergencia dataframe.loc[dataframe.index[i2], 'rsi_div1'] = rsi1 dataframe.loc[dataframe.index[i2], 'rsi_div2'] = rsi2 dataframe.loc[dataframe.index[i2], 'price_div1'] = price1 dataframe.loc[dataframe.index[i2], 'price_div2'] = price2 dataframe.loc[dataframe.index[i2], 'date_div1'] = date1 dataframe.loc[dataframe.index[i2], 'date_div2'] = date2 dataframe.loc[dataframe.index[i2], 'diff_div'] = diff_minutes dataframe.loc[dataframe.index[i2], 'divergence_strength'] = (rsi2 - rsi1) / max(rsi1, 1) dataframe.loc[dataframe.index[i2], 'divergence_window'] = window bullish_divs.append(i2) now = datetime.now(timezone.utc) if isinstance(date2, str): date2 = pd.to_datetime(date2) if (now - date2 < timedelta(minutes=30)) and runmode in ("dry_run", "live"): logger.info(f"--→[{metadata['pair']}][{now}][{i2}] Divergencia alcista detectada - " f"[{date1} → {date2}] diff: {diff_minutes} " f"RSI: {rsi1:.3f} → {rsi2:.3f}, " f"Precio: {price1:.3f} → {price2:.3f}, " f"Window: {window} diff: {diff_minutes}") dataframe.loc[dataframe.index[bullish_divs], 'bullish_divergence'] = 1 # 🔴 Divergencias Bajistas bearish_divs = [] # for i in range(1, len(rsi_maxs)): # i1 = rsi_maxs[i - 1] # i2 = rsi_maxs[i] # if i1 in price_maxs and i2 in price_maxs: # rsi1, rsi2 = dataframe.loc[dataframe.index[[i1, i2]], 'rsi'] # price1, price2 = dataframe.loc[dataframe.index[[i1, i2]], 'close'] # if rsi2 < rsi1 and price2 > price1: # bearish_divs.append(i2) # dataframe.loc[dataframe.index[bearish_divs], 'bearish_divergence'] = 1 if not self.is_hyperopt(): print(f"Divergencias alcistas detectadas: {dataframe['bullish_divergence'].sum()}") print(f"Divergencias bajistas detectadas: {dataframe['bearish_divergence'].sum()}") # END - Detectar divergencias (usamos una ventana móvil de N barras para identificar picos/valles) # Cálculo del ATR para custom_stoploss dataframe['atr'] = ta.ATR(dataframe['high'], dataframe['low'], dataframe['close'], timeperiod=14) dataframe['atr_pct'] = dataframe['atr'] / dataframe['close'] * 100 dataframe['volume_threshold'] = dataframe['volume'].rolling(window=50).mean() * 1.5 dataframe['volume_ratio'] = dataframe['volume'] / dataframe['volume_threshold'] return dataframe @staticmethod def calculate_slope_in_degrees(data): # Si no hay suficientes puntos, no calcular if len(data) < 2: return 0 # Ajusta una línea recta a los datos x = np.arange(len(data)) y = data slope, _ = np.polyfit(x, y, 1) # Convierte la pendiente a grados angle_deg = np.degrees(np.arctan(slope)) return angle_deg @staticmethod def calculate_slope(data): data = np.array(data) if len(data) < 2 or np.all(np.isnan(data)): return 0.0 x = np.arange(len(data)) y = data slope, _ = np.polyfit(x, y, 1) return slope def populate_buy_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: pair = metadata.get("pair", "UNKNOWN_PAIR") #logger.info(f"populate_buy_trend[{pair}]") required_columns = [ 'rsi', 'rsi_min_custom', 'ma_7', 'ma_25', 'ma_99', 'close', 'open', 'avg_abs_oscillation_pct_30m', 'volume' ] for col in required_columns: if col not in dataframe.columns: # O puedes usar logger.warning si prefieres no imprimir if not self.is_hyperopt(): print(f"[{metadata.get('pair')}] Ignorado por falta de columna: {col}") logger.info(f"[{metadata.get('pair')}] Ignorado por falta de columna: {col}") return dataframe # Devuelve sin hacer nada params = self.get_pair_params(pair) if not self.is_hyperopt() and not params: print(f"populate_buy_trend({pair}) params optimized not found") return dataframe rsi_window = self.rsi_window.value rsi_range = self.rsi_range.value oscillation_threshold = self.oscillation_threshold.value rsi_5m_buy_threshold = self.rsi_5m_buy_threshold.value volume_ratio_threshold_1 = self.volume_ratio_threshold_1.value volume_ratio_threshold_2 = self.volume_ratio_threshold_2.value oscillation_window = self.oscillation_window.value atr_multiplier_stoploss = self.atr_multiplier_stoploss.value atr_multiplier_roi = self.atr_multiplier_roi.value # Si existen valores optimizados para este par, los usamos if params and not self.is_hyperopt(): buy_params = params.get('params', {}).get('buy', {}) sell_params = params.get('params', {}).get('sell', {}) rsi_window = buy_params.get('rsi_window', 30) rsi_range = buy_params.get('rsi_range', 0.8) oscillation_threshold = buy_params.get('oscillation_threshold', 0.05) rsi_5m_buy_threshold = buy_params.get('rsi_5m_buy_threshold', 65) volume_ratio_threshold_1 = buy_params.get('volume_ratio_threshold_1', 0.1) volume_ratio_threshold_2 = buy_params.get('volume_ratio_threshold_2', 0.2) oscillation_window = buy_params.get('oscillation_window', 30) atr_multiplier_stoploss = sell_params.get('atr_multiplier_stoploss', 0.8) atr_multiplier_roi = sell_params.get('atr_multiplier_roi', 2.0) # Calcula las pendientes (velocidad de cambio) de las medias móviles dataframe['ma_7_slope'] = dataframe['ma_7'].diff() dataframe['ma_25_slope'] = dataframe['ma_25'].diff() # Calcula el tiempo estimado para que se crucen dataframe['ma_diff'] = dataframe['ma_7'] - dataframe['ma_25'] dataframe['ma_diff_next'] = dataframe['ma_diff'] + (dataframe['ma_7_slope'] - dataframe['ma_25_slope']) # Verifica si se tocarán en la próxima iteración dataframe['will_touch'] = (dataframe['ma_diff'] > 0) & (dataframe['ma_diff_next'] <= 0) dataframe['mid_price'] = (dataframe['close'] + dataframe['open']) / 2 bullish_cond = ( (dataframe['bullish_divergence'] == 1) & (dataframe['volume'] > 0) & (dataframe['volume_ratio'] > volume_ratio_threshold_2) & ((dataframe['change_pct'] < -0.3) | (dataframe['change_pct'] > 0.25)) ) not_bullish_cond = ( #(dataframe['rsi'] >= dataframe['rsi_min_custom'] - rsi_range) & (dataframe['rsi'] <= dataframe['rsi_min_custom'] + rsi_range) & (dataframe['rsi'] < 50) & ((dataframe['change_pct'] < -0.3) | (dataframe['change_pct'] > 0.25)) & (dataframe['mid_price'] > dataframe['ma_25']) & (dataframe['volume'] > 0) & (dataframe['volume_ratio'] > volume_ratio_threshold_1) & (dataframe['volume'] > dataframe['volume_threshold']) & (dataframe['avg_abs_oscillation_pct_30m'] >= oscillation_threshold) & (dataframe['rsi_5m_5m'] > rsi_5m_buy_threshold) & (dataframe['rsi_1d_1d'] < 60) ) # Condiciones de compra dataframe.loc[ not_bullish_cond | bullish_cond, 'buy' ] = 1 runmode = self.config.get("runmode", "unknown") now = datetime.now(timezone.utc) # Filtrar las filas que cumplen la condición bullish_rows = dataframe[bullish_cond] # Loguear cada fila que cumple la condición for idx, row in bullish_rows.iterrows(): date_bullish = row.get('date', 'N/A') if isinstance(date_bullish, str): date_bullish = pd.to_datetime(date_bullish) if isinstance(now, str): now = pd.to_datetime(now) #logger.info(f"now: {now} date_bullish: {date_bullish} runmode: {runmode} timedelta: {timedelta(minutes=500)} cond: {(now - date_bullish)}") #if ((now - date_bullish) < timedelta(minutes=400)) and runmode in ("dry_run", "live"): logger.info( f"-- [{pair}] BULLISH ENTRY [{idx}] - " f"rsi: {row['rsi']:.2f}, " f"change_pct: {row['change_pct']:.2f}, " f"volume: {row['volume']:.2f}, " f"volume_ratio: {row['volume_ratio']:.2f}, " f"bullish_divergence: {row['bullish_divergence']}, " f"price: {row.get('close', 'N/A')}, " f"time: {row.get('date', 'N/A')} " f"diff: {row.get('diff_div', 'N/A')}" ) # Filtrar las filas que cumplen la condición not_bullish not_bullish_rows = dataframe[not_bullish_cond] # Loguear cada fila que cumple la condición not_bullish for idx, row in not_bullish_rows.iterrows(): logger.info( f"-- [{pair}] NOT BULLISH ENTRY [{idx}] - " f"rsi: {row['rsi']:.2f}, " f"rsi_min_custom: {row['rsi_min_custom']:.2f}, " f"change_pct: {row['change_pct']:.2f}, " f"mid_price: {row['mid_price']:.2f}, " f"ma_25: {row['ma_25']:.2f}, " f"volume: {row['volume']:.2f}, " f"volume_ratio: {row['volume_ratio']:.2f}, " f"volume_threshold: {row['volume_threshold']:.2f}, " f"oscillation_30m: {row['avg_abs_oscillation_pct_30m']:.2f}, " f"rsi_5m_5m: {row['rsi_5m_5m']:.2f}, " f"rsi_1d_1d: {row['rsi_1d_1d']:.2f}, " f"price: {row.get('close', 'N/A')}, " f"time: {row.get('date', 'N/A')}" ) # Verifica que el índice sea de tipo `DatetimeIndex` if not isinstance(dataframe.index, pd.DatetimeIndex): dataframe.set_index(pd.to_datetime(dataframe['date']), inplace=True) # Almacena más detalles de la operación en custom_trade_info if not hasattr(self, 'custom_trade_info'): self.custom_trade_info = {} #extra_columns = ['crossover_ma_7_25', 'crossover_ma_25_99'] #for col in extra_columns: # if col not in dataframe.columns: # dataframe[col] = False for index, row in dataframe[dataframe['buy'] == 1].iterrows(): trade_time = index.to_pydatetime() # row.to_dict() te devuelve {columna: valor, ...} row_dict = row.to_dict() # Si quieres sustituir los NaN por 'No Data': row_dict = {col: (val if pd.notna(val) else 'No Data') for col, val in row_dict.items()} for offset in [0, 1]: # 0 minutos y 1 minuto time_key = trade_time + pd.Timedelta(minutes=offset) self.custom_trade_info[time_key] = row_dict.copy() #print(f"Almacenado datos para {time_key}: {self.custom_trade_info[time_key]}") return dataframe def populate_sell_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: return dataframe def custom_stoploss(self, pair: str, trade, current_time: datetime, current_rate: float, current_profit: float, **kwargs) -> float: """ Stoploss dinámico basado en ATR, limitado a un máximo del 3% de pérdida. """ try: dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe) if dataframe is None or dataframe.empty or 'atr' not in dataframe.columns: logger.warning(f"[{pair}] No se pudo aplicar custom_stoploss: sin ATR.") return 1 params = self.get_pair_params(pair) atr_multiplier_stoploss = self.atr_multiplier_stoploss.value # Si existen valores optimizados para este par, los usamos if params and not self.is_hyperopt(): sell_params = params.get('params', {}).get('sell', {}) atr_multiplier_stoploss = sell_params.get('atr_multiplier_stoploss', 0.8) trade_open_time = trade.open_date_utc.replace(second=0, microsecond=0) # Asegura que el índice esté ordenado dataframe = dataframe.sort_index() # Usa asof() para encontrar el índice más cercano anterior o igual trade_open_time = dataframe.index.asof(trade_open_time) if pd.isna(trade_open_time): logger.warning(f"[{pair}] No se encontró un índice válido para {trade_open_time}.") return 1 entry_price = trade.open_rate atr_at_entry = dataframe.loc[trade_open_time, 'atr'] atr_multiplier = atr_multiplier_stoploss # Calculamos el SL por ATR atr_stoploss_price = entry_price - (atr_at_entry * atr_multiplier) # Calculamos el SL máximo permitido (1.5%) max_stoploss_price = entry_price * 0.988 # 1.5% por debajo del entry # Tomamos el más conservador (más alto) stoploss_price = max(atr_stoploss_price, max_stoploss_price) if current_rate <= stoploss_price: # logger.info(f"[{pair}] Stoploss activado | Entrada: {entry_price:.4f} | " # f"ATR: {atr_at_entry:.4f} | SL ATR: {atr_stoploss_price:.4f} | " # f"SL Máximo (3%): {max_stoploss_price:.4f} | Usado: {stoploss_price:.4f} | " # f"Actual: {current_rate:.4f}") return 0.01 # Ejecutar venta inmediata return 1 # Mantener except Exception as e: logger.error(f"[{pair}] Error en custom_stoploss: {e}") return 1 def custom_exit(self, pair: str, trade, current_time: datetime, current_rate: float, current_profit: float, **kwargs) -> bool: """ ROI dinámico personalizado por par, basado en ATR y archivo optimizado. """ try: dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe) if dataframe is None or dataframe.empty or 'atr' not in dataframe.columns: return False params = self.get_pair_params(pair) roi_dict = {"0": 0.11} # ROI por defecto if params and not self.is_hyperopt(): roi = params.get('params', {}).get('roi', {}) if isinstance(roi, dict): roi_dict = roi # Calcular cuántos minutos lleva abierta la operación delta_min = int((current_time - trade.open_date_utc).total_seconds() // 60) # Encontrar el ROI más apropiado según el tiempo transcurrido selected_roi = 0.0 for t in sorted(map(int, roi_dict.keys()), reverse=True): if delta_min >= t: selected_roi = roi_dict[str(t)] break # Obtener precio de entrada entry_price = trade.open_rate # Obtener comisión total por la operación completa (entrada + salida) fee = self.config.get('fee', 0) # valor por defecto: 0.1% total_fee = fee * 2 # ROI objetivo ajustado al fee roi_price = entry_price * (1 + selected_roi + total_fee) #print(f"delta_min: {delta_min} fee: {fee} selected_roi: {selected_roi} entry_price: {entry_price} roi_price: {roi_price} current_rate: {current_rate}") # Si el precio actual ha alcanzado o superado el ROI objetivo, salir if current_rate >= roi_price: return True return False except Exception as e: logger.error(f"[{pair}] Error en custom_exit con ROI dinámico: {e}") return False # @property # def protections(self): # return [ # { # "method": "CooldownPeriod", # "stop_duration_candles": 15, # "only_per_pair": True # } # ] sections_printed = False def print_sections(self, indicador, sections_min, sections_max, sections_step) : """ Divide los tramos de RSI de 100→90, 90→80, ..., 10→0 y calcula porcentaje de operaciones ganadoras/perdedoras. """ total_rows = len(self.custom_trade_info) print(f"Total rows en custom_trade_info: {total_rows}") sorted_items = sorted(self.custom_trade_info.items(), key=lambda x: x[0]) custom_trade_info_aux = {} ultima_fecha_agregada = None for fecha, info in sorted_items: # print(f"****** trade_info date: {fecha} close: {info.get('close')}") if info and info.get("profit") is not None: if (ultima_fecha_agregada is None or (fecha - ultima_fecha_agregada).total_seconds() > 60): custom_trade_info_aux[fecha] = info ultima_fecha_agregada = fecha total_rows_aux = len(custom_trade_info_aux) print(f"Total rows en custom_trade_info_aux: {total_rows_aux}") if isinstance(sections_min, bool) and isinstance(sections_max, bool): total = total_rows_aux wins_true = 0 wins_false = 0 loss_true = 0 loss_false = 0 for trade_time, info in custom_trade_info_aux.items(): val = info.get(indicador) is_win = info.get('is_win', False) if val: if is_win: wins_true += 1 else: loss_true += 1 else: if is_win: wins_false += 1 else: loss_false += 1 total_true = wins_true + loss_true total_false = wins_false + loss_false if total > 0: pct_true = wins_true / total_true * 100 pct_false = wins_false / total_false * 100 else: pct_true = pct_false = 0.0 print(f"--- Estadísticas booleanas para '{indicador}' ---") print(f"True : {wins_true}/{total_true} = {pct_true:.1f}%") print(f"False: {wins_false}/{total_false} = {pct_false:.1f}%") self.sections_printed = True return # ¡terminas aquí! sections = [] current = sections_max while current > sections_min: lower = max(current - sections_step, sections_min) sections.append((current, lower)) current -= sections_step data = [] for trade_time, info in custom_trade_info_aux.items(): rsi_val = info.get(indicador, 0) # Asumimos que 'is_win' se guardó en custom_trade_info is_win = info.get('is_win', False) data.append((rsi_val, is_win)) print(f"--- Estadísticas por tramo ({indicador}) ---") for hi, lo in sections: if isinstance(rsi_val, str): continue bin_data = [win for (rsi_val, win) in data if lo <= rsi_val < hi] total = len(bin_data) wins = sum(bin_data) if total > 0: win_pct = wins / total * 100 loss_pct = 100 - win_pct else: win_pct = loss_pct = 0.0 print(f"{indicador} {hi}-{lo}: {win_pct:.1f}% wins ({wins}/{total}), {loss_pct:.1f}% losses") # Marcar como impreso para no repetir self.sections_printed = True def fmt(self, value, decimals=2, default='n/a'): try: return f"{value:.{decimals}f}" except (TypeError, ValueError): return default def confirm_trade_exit(self, pair: str, trade, order_type: str, amount: float, rate: float, time_in_force: str, exit_reason: str, current_time: datetime, **kwargs) -> bool: logger.info(f"confirm_trade_exit[{pair}]") self.trades_count += 1 trade_time = trade.open_date trade_info = self.custom_trade_info.get(trade_time, {}) beneficio_float = trade.calc_profit_ratio(rate) * 100 is_win = beneficio_float > 0 if trade_time in self.custom_trade_info: self.custom_trade_info[trade_time]['is_win'] = is_win self.custom_trade_info[trade_time]['profit'] = beneficio_float self.custom_trade_info[trade_time]['rate'] = rate stats_time = datetime(2025, 6, 3, 2, 0, 0, tzinfo=timezone.utc) # if not self.is_hyperopt() and current_time >= stats_time and not self.sections_printed: # self.print_sections('divergence_strength', 0, 1, 0.05) utc = pytz.utc madrid = pytz.timezone('Europe/Madrid') open_date_spain = trade.open_date.replace(tzinfo=utc).astimezone(madrid) current_time_spain = current_time.replace(tzinfo=utc).astimezone(madrid) open_date_spain_str = str(open_date_spain).replace(':00+01:00', '') current_time_spain_str = str(current_time_spain).replace(':00+01:00', '') open_date_spain_str = str(open_date_spain_str).replace(':00+02:00', '') current_time_spain_str = str(current_time_spain_str).replace(':00+02:00', '') beneficio = f"{beneficio_float:.2f}%".rjust(6) if not self.is_hyperopt(): params = self.get_pair_params(pair) rsi_window = self.rsi_window.value rsi_range = self.rsi_range.value oscillation_threshold = self.oscillation_threshold.value rsi_5m_buy_threshold = self.rsi_5m_buy_threshold.value volume_ratio_threshold_1 = self.volume_ratio_threshold_1.value volume_ratio_threshold_2 = self.volume_ratio_threshold_2.value oscillation_window = self.oscillation_window.value atr_multiplier_stoploss = self.atr_multiplier_stoploss.value atr_multiplier_roi = self.atr_multiplier_roi.value trades_count_string = f"{self.trades_count}".rjust(4) RED = '\033[91m' RESET = '\033[0m' color = RED if beneficio_float < 0 else '' reset = RESET if beneficio_float < 0 else '' # Si existen valores optimizados para este par, los usamos if params: buy_params = params.get('params', {}).get('buy', {}) sell_params = params.get('params', {}).get('sell', {}) rsi_window = buy_params.get('rsi_window', 30) rsi_range = buy_params.get('rsi_range', 0.8) oscillation_threshold = buy_params.get('oscillation_threshold', 0.05) rsi_5m_buy_threshold = buy_params.get('rsi_5m_buy_threshold', 65) volume_ratio_threshold_1 = buy_params.get('volume_ratio_threshold_1', 0.1) volume_ratio_threshold_2 = buy_params.get('volume_ratio_threshold_2', 0.2) oscillation_window = buy_params.get('oscillation_window', 30) atr_multiplier_stoploss = sell_params.get('atr_multiplier_stoploss', 0.8) atr_multiplier_roi = sell_params.get('atr_multiplier_roi', 2.0) logger.info( f"[{trades_count_string}]{color}[{pair}]" f"[{open_date_spain_str} - {current_time_spain_str.split(' ')[1]}]" f"[{beneficio}]" f"[{exit_reason[:4].ljust(4)}]{reset}" f"[RSI: {self.fmt(trade_info.get('rsi', 'n/a'), decimals=2)}]" f"[1d: {self.fmt(trade_info.get('pct_1d'), decimals=2)} prev: {self.fmt(trade_info.get('pct_prev_1d'), decimals=2)}]" f"[change: {self.fmt(trade_info.get('change_pct'), decimals=2)}]" f"[acc: {self.fmt(trade_info.get('acceleration'), decimals=2)}]" f"avg: {self.fmt(trade_info.get('avg_abs_oscillation_pct_30m'), decimals=4)} " f"bd: {trade_info.get('bullish_divergence')} " f"vl: {trade_info.get('volume')} " f"th: {self.fmt(trade_info.get('volume_threshold'), decimals=0)} " f"rt: {self.fmt(trade_info.get('volume_ratio'), decimals=3)} " f"divRsi: {self.fmt(trade_info.get('rsi_div1'), decimals=2)}->{self.fmt(trade_info.get('rsi_div2'), decimals=2)} " f"divPrice: {self.fmt(trade_info.get('price_div1'), decimals=2)}->{self.fmt(trade_info.get('price_div2'), decimals=2)} " f"divDate: {trade_info.get('date_div1')}->{trade_info.get('date_div2')} diff: {trade_info.get('diff_div')} " f"str: {self.fmt(trade_info.get('divergence_strength'), decimals=2)} " f"w: {trade_info.get('divergence_window')}{reset}" ) #logger.info(f"confirm_trade_exit[{pair}] trades_count: {trades_count_string}") #logger.info(f"RSI: {self.fmt(trade_info.get('rsi'), decimals=2)}") #print(f"RSI: {self.fmt(trade_info.get('rsi'), decimals=2)}") print( f"[{trades_count_string}]{color}[{pair}]" f"[{open_date_spain_str} - {current_time_spain_str.split(' ')[1]}]" f"[{beneficio}]" f"[{exit_reason[:4].ljust(4)}] " f"[RSI: {self.fmt(trade_info['rsi'], decimals=2)}]" f"[1d: {self.fmt(trade_info['pct_1d'], decimals=2)} prev: {self.fmt(trade_info['pct_prev_1d'], decimals=2)}]" f"[change: {self.fmt(trade_info['change_pct'], decimals=2)}]" f"[acc: {self.fmt(trade_info['acceleration'], decimals=2)}]" f"avg: {self.fmt(trade_info['avg_abs_oscillation_pct_30m'], decimals=4)} " f"bd: {trade_info['bullish_divergence']} " f"vl: {trade_info['volume']} " f"th: {self.fmt(trade_info['volume_threshold'], decimals=0)} " f"rt: {self.fmt((trade_info['volume'] / trade_info['volume_threshold']), decimals=3)} " f"divRsi: {self.fmt(trade_info['rsi_div1'], decimals=2)}->{self.fmt(trade_info['rsi_div2'], decimals=2)} " f"divPrice: {self.fmt(trade_info['price_div1'], decimals=2)}->{self.fmt(trade_info['price_div2'], decimals=2)} " f"divDate: {trade_info['date_div1']}->{trade_info['date_div2']} diff: {trade_info['diff_div']} " f"str: {self.fmt(trade_info['divergence_strength'], decimals=2)} " f"w: {trade_info['divergence_window']}{reset}" ) # Llama al método original para no romper la lógica base return super().confirm_trade_exit(pair, trade, order_type, amount, rate, time_in_force, exit_reason, current_time, **kwargs)