from freqtrade.strategy import IStrategy, IntParameter, DecimalParameter, merge_informative_pair, informative from custom_indicators import divergences_rsi, cruce_medias_ema, cruce_medias_macd, confirmacion_volumen, confirmaciones, confirmaciones2, changes, trendlines_up from pandas import DataFrame from technical import trendline 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 import joblib import freqtrade.vendor.qtpylib.indicators as qtpylib from freqtrade.persistence.models import Trade, Order 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 = 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 DynamicTrend(IStrategy): timeframe = '1m' #inf_timeframes = ['5m', '15m', '30m', '1h', '4h', '1d'] inf_timeframes = ['5m', '15m'] optimized_params = {} trades_count = 0 # minimal_roi = { # "0": 0.049, # "8": 0.029, # "20": 0.015, # "35": 0 # } # minimal_roi = { # "0": 0.29 # } minimal_roi = { "0": 0.05 } roi_per_timeframe = { "1m": 0.005, # +1% "5m": 0.01, # +1.5% "15m": 0.02, # +2% "30m": 0.03, # +3% "1h": 0.04, # +4% } stoploss_per_timeframe = { "1m": -0.005, # -2% "5m": -0.01, # -1.5% "15m": -0.02, # -1% "30m": -0.03, # -0.8% "1h": -0.07, # -0.5% } roi_1m = DecimalParameter(0.001, 0.02, default=0.015, space='buy', optimize=True) roi_5m = DecimalParameter(0.005, 0.03, default=0.012, space='buy', optimize=True) roi_15m = DecimalParameter(0.01, 0.05, default=0.023, space='buy', optimize=True) roi_30m = DecimalParameter(0.01, 0.05, default=0.031, space='buy', optimize=True) roi_1h = DecimalParameter(0.02, 0.1, default=0.028, space='buy', optimize=True) stop_1m = DecimalParameter(-0.05, -0.001, default=-0.02, space='sell', optimize=True) stop_5m = DecimalParameter(-0.05, -0.001, default=-0.005, space='sell', optimize=True) stop_15m = DecimalParameter(-0.05, -0.001, default=-0.026, space='sell', optimize=True) stop_30m = DecimalParameter(-0.1, -0.005, default=-0.094, space='sell', optimize=True) stop_1h = DecimalParameter(-0.1, -0.005, default=-0.07, space='sell', optimize=True) use_custom_stoploss = True use_custom_exit = True trailing_stop = False can_short = True confirmation_window = IntParameter(5, 40, default=33, space="buy") min_confirmations = IntParameter(1, 4, default=3, space="buy") 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. """ print(f"load_optimized_parameters") 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' """@property def protections(self): return [ { "method": "CooldownPeriod", "stop_duration_candles": 15 }, ]""" @informative('5m') def populate_indicators_5m(self, dataframe: DataFrame, metadata: dict) -> DataFrame: from custom_indicators import divergences_rsi, cruce_medias_ema, cruce_medias_macd, confirmacion_volumen, confirmaciones, changes, acciones_precio print(f"buscamos los parámetros para el par {metadata['pair']}") params = self.get_pair_params(metadata['pair']) # BEGIN - Cargamos los valores por defecto confirmation_window = self.confirmation_window.value # END - Cargamos los valores por defecto # 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', {}) # Cargamos los valores optimizados del par confirmation_window = buy_params.get('confirmation_window', confirmation_window) dataframe['rsi'] = ta.RSI(dataframe['close'], timeperiod=14) dataframe['atr'] = ta.ATR(dataframe['high'], dataframe['low'], dataframe['close'], timeperiod=14) dataframe['adx'] = ta.ADX(dataframe['high'], dataframe['low'], dataframe['close'], timeperiod=14) dataframe = divergences_rsi(dataframe, window=5) dataframe = cruce_medias_ema(dataframe, fast=20, slow=50) dataframe = cruce_medias_macd(dataframe, fast=12, slow=26, period=9) dataframe = confirmacion_volumen(dataframe, fast=5, slow=10) #dataframe = confirmaciones(dataframe, window=confirmation_window) dataframe = confirmaciones2(dataframe, window=confirmation_window) dataframe = acciones_precio(dataframe) """print( dataframe[["date", "bullish_divergence", "bullish_divergence_hidden", "cruce_ema_alcista", "cruce_macd_alcista", "vol_confirmacion_alcista", "confirmaciones_alcistas", "confirmaciones_bajistas"]] .tail(120) # Muestra las últimas 30 filas (puedes ajustar) .to_string(index=False) )""" return dataframe @informative('15m') def populate_indicators_15m(self, dataframe: DataFrame, metadata: dict) -> DataFrame: from custom_indicators import divergences_rsi, cruce_medias_ema, cruce_medias_macd, confirmacion_volumen, confirmaciones, changes, acciones_precio print(f"buscamos los parámetros para el par {metadata['pair']}") params = self.get_pair_params(metadata['pair']) # BEGIN - Cargamos los valores por defecto confirmation_window = self.confirmation_window.value # END - Cargamos los valores por defecto # 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', {}) # Cargamos los valores optimizados del par confirmation_window = buy_params.get('confirmation_window', confirmation_window) dataframe['rsi'] = ta.RSI(dataframe['close'], timeperiod=14) dataframe['atr'] = ta.ATR(dataframe['high'], dataframe['low'], dataframe['close'], timeperiod=14) dataframe['adx'] = ta.ADX(dataframe['high'], dataframe['low'], dataframe['close'], timeperiod=14) dataframe = divergences_rsi(dataframe, window=5) dataframe = cruce_medias_ema(dataframe, fast=20, slow=50) dataframe = cruce_medias_macd(dataframe, fast=12, slow=26, period=9) dataframe = confirmacion_volumen(dataframe, fast=5, slow=10) #dataframe = confirmaciones(dataframe, window=confirmation_window) dataframe = confirmaciones2(dataframe, window=confirmation_window) dataframe = acciones_precio(dataframe) return dataframe """@informative('30m') def populate_indicators_30m(self, dataframe: DataFrame, metadata: dict) -> DataFrame: from custom_indicators import divergences_rsi, cruce_medias_ema, cruce_medias_macd, confirmacion_volumen, confirmaciones, changes print(f"buscamos los parámetros para el par {metadata['pair']}") params = self.get_pair_params(metadata['pair']) # BEGIN - Cargamos los valores por defecto confirmation_window = self.confirmation_window.value # END - Cargamos los valores por defecto # 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', {}) # Cargamos los valores optimizados del par confirmation_window = buy_params.get('confirmation_window', confirmation_window) dataframe['rsi'] = ta.RSI(dataframe['close'], timeperiod=14) dataframe['atr'] = ta.ATR(dataframe['high'], dataframe['low'], dataframe['close'], timeperiod=14) dataframe['adx'] = ta.ADX(dataframe['high'], dataframe['low'], dataframe['close'], timeperiod=14) dataframe = divergences_rsi(dataframe, window=5) dataframe = cruce_medias_ema(dataframe, fast=20, slow=50) dataframe = cruce_medias_macd(dataframe, fast=12, slow=26, period=9) dataframe = confirmacion_volumen(dataframe, fast=5, slow=10) #dataframe = confirmaciones(dataframe, window=confirmation_window) dataframe = confirmaciones2(dataframe, window=confirmation_window) return dataframe""" """@informative('1h') def populate_indicators_1h(self, dataframe: DataFrame, metadata: dict) -> DataFrame: from custom_indicators import divergences_rsi, cruce_medias_ema, cruce_medias_macd, confirmacion_volumen, confirmaciones, changes print(f"buscamos los parámetros para el par {metadata['pair']}") params = self.get_pair_params(metadata['pair']) # BEGIN - Cargamos los valores por defecto confirmation_window = self.confirmation_window.value # END - Cargamos los valores por defecto # 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', {}) # Cargamos los valores optimizados del par confirmation_window = buy_params.get('confirmation_window', confirmation_window) dataframe['rsi'] = ta.RSI(dataframe['close'], timeperiod=14) dataframe['adx'] = ta.ADX(dataframe['high'], dataframe['low'], dataframe['close'], timeperiod=14) dataframe['rsi_sobrecompra'] = 70 dataframe['rsi_sobrecompra_extrema'] = 80 dataframe['rsi_medio'] = 50 dataframe['rsi_sobreventa'] = 30 dataframe['rsi_sobreventa_extrema'] = 20 dataframe['atr'] = ta.ATR(dataframe['high'], dataframe['low'], dataframe['close'], timeperiod=14) dataframe['volume_positive'] = np.nan dataframe['volume_negative'] = np.nan # Vela verde: close > open dataframe.loc[dataframe['close'] > dataframe['open'], 'volume_positive'] = dataframe['volume'] # Vela roja: close < open dataframe.loc[dataframe['close'] < dataframe['open'], 'volume_negative'] = dataframe['volume'] dataframe['volume_ma5'] = ta.SMA(dataframe['volume'], timeperiod=5) dataframe = divergences_rsi(dataframe, window=5) dataframe = cruce_medias_ema(dataframe, fast=20, slow=50) dataframe = cruce_medias_macd(dataframe, fast=12, slow=26, period=9) dataframe = confirmacion_volumen(dataframe, fast=5, slow=10) #dataframe = confirmaciones(dataframe, window=confirmation_window) dataframe = confirmaciones2(dataframe, window=confirmation_window) return dataframe""" """@informative('4h') def populate_indicators_4h(self, dataframe: DataFrame, metadata: dict) -> DataFrame: from custom_indicators import divergences_rsi, cruce_medias_ema, cruce_medias_macd, confirmacion_volumen, confirmaciones, changes print(f"buscamos los parámetros para el par {metadata['pair']}") params = self.get_pair_params(metadata['pair']) # BEGIN - Cargamos los valores por defecto confirmation_window = self.confirmation_window.value # END - Cargamos los valores por defecto # 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', {}) # Cargamos los valores optimizados del par confirmation_window = buy_params.get('confirmation_window', confirmation_window) dataframe['rsi'] = ta.RSI(dataframe['close'], timeperiod=14) dataframe['atr'] = ta.ATR(dataframe['high'], dataframe['low'], dataframe['close'], timeperiod=14) dataframe['adx'] = ta.ADX(dataframe['high'], dataframe['low'], dataframe['close'], timeperiod=14) dataframe = divergences_rsi(dataframe, window=5) dataframe = cruce_medias_ema(dataframe, fast=20, slow=50) dataframe = cruce_medias_macd(dataframe, fast=12, slow=26, period=9) dataframe = confirmacion_volumen(dataframe, fast=5, slow=10) #dataframe = confirmaciones(dataframe, window=confirmation_window) dataframe = confirmaciones2(dataframe, window=confirmation_window) return dataframe""" """@informative('1d') def populate_indicators_1d(self, dataframe: DataFrame, metadata: dict) -> DataFrame: from custom_indicators import divergences_rsi, cruce_medias_ema, cruce_medias_macd, confirmacion_volumen, confirmaciones, changes print(f"buscamos los parámetros para el par {metadata['pair']}") params = self.get_pair_params(metadata['pair']) # BEGIN - Cargamos los valores por defecto confirmation_window = self.confirmation_window.value # END - Cargamos los valores por defecto # 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', {}) # Cargamos los valores optimizados del par confirmation_window = buy_params.get('confirmation_window', confirmation_window) dataframe['rsi'] = ta.RSI(dataframe['close'], timeperiod=14) dataframe['atr'] = ta.ATR(dataframe['high'], dataframe['low'], dataframe['close'], timeperiod=14) dataframe['adx'] = ta.ADX(dataframe['high'], dataframe['low'], dataframe['close'], timeperiod=14) dataframe = divergences_rsi(dataframe, window=5) dataframe = cruce_medias_ema(dataframe, fast=20, slow=50) dataframe = cruce_medias_macd(dataframe, fast=12, slow=26, period=9) dataframe = confirmacion_volumen(dataframe, fast=5, slow=10) #dataframe = confirmaciones(dataframe, window=confirmation_window) dataframe = confirmaciones2(dataframe, window=confirmation_window) return dataframe""" def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame: from custom_indicators import divergences_rsi, cruce_medias_ema, cruce_medias_macd, confirmacion_volumen, confirmaciones, changes, acciones_precio print(f"populate_indicators") print(f"buscamos los parámetros para el par {metadata['pair']}") params = self.get_pair_params(metadata['pair']) # BEGIN - Cargamos los valores por defecto confirmation_window = self.confirmation_window.value # END - Cargamos los valores por defecto # 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', {}) # Cargamos los valores optimizados del par confirmation_window = buy_params.get('confirmation_window', confirmation_window) dataframe['rsi_1m'] = ta.RSI(dataframe['close'], timeperiod=14) dataframe['rsi'] = ta.RSI(dataframe['close'], timeperiod=14) dataframe['atr'] = ta.ATR(dataframe['high'], dataframe['low'], dataframe['close'], timeperiod=14) dataframe['adx'] = ta.ADX(dataframe['high'], dataframe['low'], dataframe['close'], timeperiod=6) dataframe['plus_di'] = ta.PLUS_DI(dataframe['high'], dataframe['low'], dataframe['close'], timeperiod=6) dataframe['minus_di'] = ta.MINUS_DI(dataframe['high'], dataframe['low'], dataframe['close'], timeperiod=6) # Obtenemos las divergencias RSI dataframe = divergences_rsi(dataframe, window=5, trendlines_up_enable=True) dataframe = cruce_medias_ema(dataframe, fast=20, slow=50) dataframe = cruce_medias_macd(dataframe, fast=12, slow=26, period=9) dataframe = confirmacion_volumen(dataframe, fast=5, slow=10) #dataframe = confirmaciones(dataframe, window=confirmation_window) dataframe = confirmaciones2(dataframe, window=confirmation_window) #dataframe = changes(dataframe) dataframe = acciones_precio(dataframe) dataframe['inverted_hammer'] = np.nan # gentrend_df = trendline.gentrends(dataframe, field='close') # dataframe['gentrend_high'] = gentrend_df['Max Line'] # dataframe['gentrend_low'] = gentrend_df['Min Line'] # segtrend_df = trendline.segtrends(dataframe, field='close', segments=3) # dataframe['segtrend_high'] = segtrend_df['Max Line'] # dataframe['segtrend_low'] = segtrend_df['Min Line'] return dataframe """def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: pair = metadata.get("pair", "UNKNOWN_PAIR") runmode = self.config.get("runmode", "unknown") now = datetime.now(timezone.utc) params = self.get_pair_params(metadata['pair']) # BEGIN - Cargamos los valores por defecto min_confirmations = self.min_confirmations.value # END - Cargamos los valores por defecto # 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', {}) # Cargamos los valores optimizados del par min_confirmations = buy_params.get('min_confirmations', min_confirmations) # Inicializamos en 0 dataframe["enter_long"] = 0 dataframe["enter_short"] = 0 # Recorremos de timeframe mayor a menor timeframes_prioridad = ["1h", "30m", "15m", "5m", "1m"] # Para cada timeframe comprobamos las condiciones for tf in timeframes_prioridad: col_alcistas = f"confirmaciones_alcistas_{tf}" col_bajistas = f"confirmaciones_bajistas_{tf}" if tf == "1m": col_alcistas = "confirmaciones_alcistas" col_bajistas = "confirmaciones_bajistas" #if col_alcistas in dataframe.columns: #dataframe.loc[dataframe[col_alcistas] >= self.min_confirmations.value, "enter_long"] = 1 #dataframe.loc[dataframe[col_alcistas] >= self.min_confirmations.value, "tf_senal"] = tf #dataframe.loc[dataframe[col_alcistas] >= self.min_confirmations.value, 'enter_tag'] = tf #if col_bajistas in dataframe.columns: #dataframe.loc[dataframe[col_bajistas] >= self.min_confirmations.value, "enter_short"] = 1 #dataframe.loc[dataframe[col_bajistas] >= self.min_confirmations.value, "tf_senal"] = tf #dataframe.loc[dataframe[col_bajistas] >= self.min_confirmations.value, 'enter_tag'] = tf if col_alcistas in dataframe.columns and col_bajistas in dataframe.columns: # Entrada en largo: confirmaciones alcistas suficientes y mayores que las bajistas long_condition = ( (dataframe[col_alcistas] >= min_confirmations) & (dataframe[col_alcistas] > dataframe[col_bajistas]) & (dataframe['confirmacion_alcista_valida']) & (dataframe['adx'] < 40) & (dataframe['adx_5m'] < 40) & (dataframe['adx_15m'] < 40) ) dataframe.loc[long_condition, "enter_long"] = 1 dataframe.loc[long_condition, "tf_senal"] = tf dataframe.loc[long_condition, "enter_tag"] = tf # Entrada en corto: confirmaciones bajistas suficientes y mayores que las alcistas short_condition = ( (dataframe[col_bajistas] >= min_confirmations) & (dataframe[col_bajistas] > dataframe[col_alcistas]) & (dataframe['confirmacion_bajista_valida']) & (dataframe['adx'] < 40) & (dataframe['adx_5m'] < 40) & (dataframe['adx_15m'] < 40) ) dataframe.loc[short_condition, "enter_short"] = 1 dataframe.loc[short_condition, "tf_senal"] = tf dataframe.loc[short_condition, "enter_tag"] = tf # 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 = {} if "enter_long" not in dataframe.columns: dataframe["enter_long"] = 0 if "enter_short" not in dataframe.columns: dataframe["enter_short"] = 0 # Crear dos dataframes con la columna 'entry_type' añadida df_long = dataframe[dataframe['enter_long'] == 1].copy() df_long['entry_type'] = 'long' df_short = dataframe[dataframe['enter_short'] == 1].copy() df_short['entry_type'] = 'short' # Unirlos en uno solo df_trades = pd.concat([df_long, df_short]) for index, row in df_trades.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_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: pair = metadata.get("pair", "UNKNOWN_PAIR") runmode = self.config.get("runmode", "unknown") now = datetime.now(timezone.utc) params = self.get_pair_params(metadata['pair']) # BEGIN - Cargamos los valores por defecto min_confirmations = self.min_confirmations.value # END - Cargamos los valores por defecto # 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', {}) # Cargamos los valores optimizados del par min_confirmations = buy_params.get('min_confirmations', min_confirmations) # Inicializamos en 0 dataframe["enter_long"] = 0 dataframe["enter_short"] = 0 # Recorremos de timeframe mayor a menor timeframes_prioridad = ["5m", "1m"] # Para cada timeframe comprobamos las condiciones for tf in timeframes_prioridad: col_alcistas = f"confirmaciones_alcistas_{tf}" col_bajistas = f"confirmaciones_bajistas_{tf}" if tf == "1m": col_alcistas = "confirmaciones_alcistas" col_bajistas = "confirmaciones_bajistas" if col_alcistas in dataframe.columns and col_bajistas in dataframe.columns: # Entrada en largo: confirmaciones alcistas suficientes y mayores que las bajistas long_condition = ( ( (dataframe['bullish_divergence'].fillna(False).rolling(10, min_periods=10).sum() > 0) | (dataframe['bullish_divergence_hidden'].fillna(False).rolling(10, min_periods=10).sum() > 0) ) & #(dataframe['break_trend']) & (dataframe['adx'] < 40) & (dataframe['adx_5m'] < 40) & (dataframe['adx_15m'] < 40) ) dataframe.loc[long_condition, "enter_long"] = 1 dataframe.loc[long_condition, "tf_senal"] = tf dataframe.loc[long_condition, "enter_tag"] = tf # Entrada en corto: confirmaciones bajistas suficientes y mayores que las alcistas short_condition = ( ( (dataframe['bearish_divergence'].fillna(False).rolling(10, min_periods=10).sum() > 0) | (dataframe['bearish_divergence_hidden'].fillna(False).rolling(10, min_periods=10).sum() > 0) ) & (dataframe['adx'] < 40) & (dataframe['adx_5m'] < 40) & (dataframe['adx_15m'] < 40) ) dataframe.loc[short_condition, "enter_short"] = 1 dataframe.loc[short_condition, "tf_senal"] = tf dataframe.loc[short_condition, "enter_tag"] = tf # 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 = {} if "enter_long" not in dataframe.columns: dataframe["enter_long"] = 0 if "enter_short" not in dataframe.columns: dataframe["enter_short"] = 0 # Crear dos dataframes con la columna 'entry_type' añadida df_long = dataframe[dataframe['enter_long'] == 1].copy() df_long['entry_type'] = 'long' df_short = dataframe[dataframe['enter_short'] == 1].copy() df_short['entry_type'] = 'short' # Unirlos en uno solo df_trades = pd.concat([df_long, df_short]) for index, row in df_trades.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_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: params = self.get_pair_params(metadata['pair']) # BEGIN - Cargamos los valores por defecto min_confirmations = self.min_confirmations.value # END - Cargamos los valores por defecto # 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', {}) # Cargamos los valores optimizados del par min_confirmations = buy_params.get('min_confirmations', min_confirmations) # Inicializamos en 0 dataframe["exit_long"] = 0 dataframe["exit_short"] = 0 # Recorremos de timeframe mayor a menor timeframes_prioridad = ["1h", "30m", "15m", "5m", "1m"] # timeframes_prioridad = ["30m"] timeframes_prioridad = [] # Para cada timeframe comprobamos las condiciones for tf in timeframes_prioridad: if tf == "1m": col_alcistas = "confirmaciones_alcistas" col_bajistas = "confirmaciones_bajistas" else: col_alcistas = f"confirmaciones_alcistas_{tf}" col_bajistas = f"confirmaciones_bajistas_{tf}" if col_alcistas in dataframe.columns and col_bajistas in dataframe.columns: # Salir de largo si hay suficientes confirmaciones bajistas y son mayores que las alcistas exit_long_condition = ( (dataframe[col_bajistas] >= min_confirmations) & (dataframe[col_bajistas] > dataframe[col_alcistas]) ) dataframe.loc[exit_long_condition, "exit_long"] = 1 dataframe.loc[exit_long_condition, "tf_senal"] = tf dataframe.loc[exit_long_condition, "exit_tag"] = tf # Salir de corto si hay suficientes confirmaciones alcistas y son mayores que las bajistas exit_short_condition = ( (dataframe[col_alcistas] >= min_confirmations) & (dataframe[col_alcistas] > dataframe[col_bajistas]) ) dataframe.loc[exit_short_condition, "exit_short"] = 1 dataframe.loc[exit_short_condition, "tf_senal"] = tf dataframe.loc[exit_short_condition, "exit_tag"] = tf return dataframe def order_filled(self, pair: str, trade: Trade, order: Order, current_time: datetime, **kwargs) -> None: """ Called right after an order fills. Will be called for all order types (entry, exit, stoploss, position adjustment). :param pair: Pair for trade :param trade: trade object. :param order: Order object. :param current_time: datetime object, containing the current datetime :param **kwargs: Ensure to keep this here so updates to this won't break your strategy. """ # Obtain pair dataframe (just to show how to access it) dataframe, _ = self.dp.get_analyzed_dataframe(trade.pair, self.timeframe) last_candle = dataframe.iloc[-1].squeeze() if (trade.nr_of_successful_entries == 1) and (order.ft_order_side == trade.entry_side): trade.set_custom_data(key="entry_candle_high", value=last_candle["high"]) fill_price = float(order.average or order.price) type_trade = 'short' if trade.is_short else 'long' entry_time = trade.open_date_utc or current_time last_min_close = np.nan take_profit_ratio = 2 min_stop_pct = 0.002 # 0.20% if not trade.is_short: # LONG: stop = último mínimo antes de la entrada prev_mins = dataframe[(dataframe.index < entry_time) & dataframe['price_min'].notna()] if prev_mins.empty: return stop = float(prev_mins.iloc[-1]['close']) # Forzar mínimo 0.20% de distancia if (fill_price - stop) / fill_price < min_stop_pct: stop = fill_price * (1 - min_stop_pct) R = fill_price - stop if R <= 0: return change_pct = (fill_price / stop - 1) * 100 tp_price = fill_price + take_profit_ratio * R else: # SHORT: stop = último máximo antes de la entrada prev_maxs = dataframe[(dataframe.index < entry_time) & dataframe['price_max'].notna()] if prev_maxs.empty: return stop = float(prev_maxs.iloc[-1]['close']) # Forzar mínimo 0.20% de distancia if (stop - fill_price) / fill_price < min_stop_pct: stop = fill_price * (1 + min_stop_pct) R = stop - fill_price if R <= 0: return change_pct = (stop / fill_price - 1) * 100 tp_price = fill_price - take_profit_ratio * R # Guarda si quieres trade.set_custom_data("stop_price", stop) trade.set_custom_data("tp_price", tp_price) trade.set_custom_data("risk_R", R) trade.set_custom_data("change_pct", change_pct) print(f"{pair} {('SHORT' if trade.is_short else 'LONG')} " f"entry={fill_price:.8f} stop={stop:.8f} " f"Δ%={change_pct:.3f}% TP@{tp_price:.8f}") return None """def custom_stoploss(self, pair: str, trade, current_time: datetime, current_rate: float, current_profit: float, **kwargs) -> float: try: dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe) params = self.get_pair_params(pair) # BEGIN - Cargamos los valores por defecto stop_1m = self.stop_1m.value stop_5m = self.stop_5m.value stop_15m = self.stop_15m.value stop_30m = self.stop_30m.value stop_1h = self.stop_1h.value # END - Cargamos los valores por defecto # 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', {}) # Cargamos los valores optimizados del par stop_1m = sell_params.get('stop_1m', stop_1m) stop_5m = sell_params.get('stop_5m', stop_5m) stop_15m = sell_params.get('stop_15m', stop_15m) stop_30m = sell_params.get('stop_30m', stop_30m) stop_1h = sell_params.get('stop_1h', stop_1h) enter_tag = trade.enter_tag if enter_tag == "1m": return stop_1m elif enter_tag == "5m": return stop_5m elif enter_tag == "15m": return stop_15m elif enter_tag == "30m": return stop_30m elif enter_tag == "1h": return stop_1h else: return 0.03 except Exception as e: logger.error(f"[{pair}] Error en custom_stoploss: {e}") return 1""" def custom_stoploss(self, pair, trade, current_time, current_rate, current_profit, **kwargs) -> float: stop = trade.get_custom_data("stop_price") if stop is None: return self.stoploss # fallback global open_rate = float(trade.open_rate) loss_frac = abs(stop / open_rate - 1.0) # pérdida relativa hasta el stop return -loss_frac # Freqtrade espera un valor negativo """def custom_exit(self, pair: str, trade, current_time: datetime, current_rate: float, current_profit: float, **kwargs) -> bool: try: # # Calcular cuántos minutos lleva abierta la operación delta_min = int((current_time - trade.open_date_utc).total_seconds() // 60) params = self.get_pair_params(pair) # BEGIN - Cargamos los valores por defecto roi_1m = self.roi_1m.value roi_5m = self.roi_5m.value roi_15m = self.roi_15m.value roi_30m = self.roi_30m.value roi_1h = self.roi_1h.value # END - Cargamos los valores por defecto # 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', {}) # Cargamos los valores optimizados del par roi_1m = buy_params.get('roi_1m', roi_1m) roi_5m = buy_params.get('roi_5m', roi_5m) roi_15m = buy_params.get('roi_15m', roi_15m) roi_30m = buy_params.get('roi_30m', roi_30m) roi_1h = buy_params.get('roi_1h', roi_1h) # Cerrar si lleva más de 60 minutos # if delta_min >= 60: # return True # Recuperar enter_tag desde trade enter_tag = trade.enter_tag #target_roi = self.roi_per_timeframe.get(enter_tag, 0.01) if enter_tag == "1m": target_roi = roi_1m elif enter_tag == "5m": target_roi = roi_5m elif enter_tag == "15m": target_roi = roi_15m elif enter_tag == "30m": target_roi = roi_30m elif enter_tag == "1h": target_roi = roi_1h else: target_roi = 0.01 if current_profit >= target_roi: return True return False except Exception as e: logger.error(f"[{pair}] Error en custom_exit con ROI dinámico: {e}") return False""" def custom_exit(self, pair, trade, current_time, current_rate, current_profit, **kwargs): tp = trade.get_custom_data("tp_price") if tp is None: return None if trade.is_short: if current_rate <= tp: return "tp_hit" else: if current_rate >= tp: return "tp_hit" return None 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) if isinstance(rsi_val, str): try: rsi_val = float(rsi_val) except ValueError: print(f"Advertencia: valor no convertible a número: '{rsi_val}'") continue # 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, 7, 14, 2, 0, 0, tzinfo=timezone.utc) if not self.is_hyperopt() and current_time >= stats_time and not self.sections_printed: self.print_sections('adx', 0, 70, 10) 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) # volume_ratio_threshold_2 = self.volume_ratio_threshold_2.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', {}) # volume_ratio_threshold_2 = buy_params.get('volume_ratio_threshold_2', 0.2) posicion = "SHORT" if trade.is_short else " LONG" debug_string = ( f"[{trades_count_string}]{color}[{pair}]" f"[{open_date_spain_str} - {current_time_spain_str.split(' ')[1]}]" f"[{posicion}]" f"[{beneficio}]" f"[{trade_info.get('enter_tag')[:4].ljust(4)}][{exit_reason[:4].ljust(4)}] adx: {self.fmt(trade_info.get('adx'), decimals=0)} di_plus: {self.fmt(trade_info.get('plus_di'), decimals=0)} di_minus: {self.fmt(trade_info.get('minus_di'), decimals=0)} " f"5m: {self.fmt(trade_info.get('adx_5m'), decimals=0)} 15m: {self.fmt(trade_info.get('adx_15m'), decimals=0)} " f"30m: {self.fmt(trade_info.get('adx_30m'), decimals=0)} 1h: {self.fmt(trade_info.get('adx_1h'), decimals=0)} " f"4h: {self.fmt(trade_info.get('adx_4h'), decimals=0)} 1d: {self.fmt(trade_info.get('adx_1d'), decimals=0)} " f" {reset}" ) # logger.info(debug_string) print(debug_string) # 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)