from freqtrade.strategy import IStrategy, IntParameter, DecimalParameter, merge_informative_pair, informative from pandas import DataFrame import talib as ta import logging import numpy as np import pandas as pd from datetime import datetime import pytz from scipy.signal import argrelextrema import logging logger = logging.getLogger(__name__) class Hermes_v7(IStrategy): timeframe = '1m' inf_timeframes = ['5m'] def is_hyperopt(self) -> bool: return self.config.get('runmode') == 'hyperopt' @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 minimal_roi = { "0": 0.052 # 1% de ROI } # Parámetros configurables # Minutos sobre los que se buscará el RSI mínimo rsi_window = IntParameter(10, 150, default=30, space='buy', optimize=True) # Rango de margen para el RSI mínimo encontrado rsi_range = DecimalParameter(0.1, 10.0, default=0.8, decimals=2, 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) 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) use_custom_stoploss = True use_custom_exit = True trailing_stop = False 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 trade_open_time = trade.open_date_utc.replace(second=0, microsecond=0) if trade_open_time not in dataframe.index: trade_open_time = dataframe.index[dataframe.index.get_loc(trade_open_time, method='pad')] entry_price = trade.open_rate atr_at_entry = dataframe.loc[trade_open_time, 'atr'] atr_multiplier = self.atr_multiplier_stoploss.value # Calculamos el SL por ATR atr_stoploss_price = entry_price - (atr_at_entry * atr_multiplier) # Calculamos el SL máximo permitido (3%) max_stoploss_price = entry_price * 0.97 # 3% 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 populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame: # 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 logger.info(f"[{metadata['pair']}] Cálculo de indicadores iniciado.") # 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) # Calcula el RSI mínimo para la ventana configurada dataframe['rsi_min_custom'] = dataframe['rsi'].rolling(window=self.rsi_window.value).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=self.oscillation_window.value).mean() # Timeframe informativo (5m) #informative = self.dp.get_pair_dataframe(pair=metadata['pair'], timeframe='5m') # Verifica que el informativo también tenga datos y la columna 'close' #if informative is None or informative.empty or 'close' not in informative.columns: # return dataframe #informative['rsi_5m'] = ta.RSI(informative['close'], timeperiod=14) #informative['sma_rsi_5m'] = ta.SMA(informative['rsi_5m'], timeperiod=14) # Merge con el timeframe principal #dataframe = merge_informative_pair(dataframe, informative, self.timeframe, '5m', ffill=True) # BEGIN - Detectar divergencias (usamos una ventana móvil de N barras para identificar picos/valles) window = 5 # 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 # Solo si hay suficientes datos 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 # Detectar máximos locales rsi_maxs = argrelextrema(dataframe['rsi'].values, np.greater_equal, order=window)[0] price_maxs = argrelextrema(dataframe['close'].values, np.greater_equal, order=window)[0] dataframe.loc[dataframe.index[rsi_maxs], 'rsi_max_local'] = 1 dataframe.loc[dataframe.index[price_maxs], 'price_max_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: rsi1, rsi2 = dataframe.loc[dataframe.index[[i1, i2]], 'rsi'] price1, price2 = dataframe.loc[dataframe.index[[i1, i2]], 'close'] if rsi2 > rsi1 and price2 < price1: bullish_divs.append(i2) 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) 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") 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 ma_25_above_7 = dataframe['ma_7'] < dataframe['ma_25'] ma99_above_25 = (dataframe['ma_25'] < dataframe['ma_99']) ma_25_above_7 = dataframe['ma_25'] < dataframe['ma_7'] ma99_above_25 = (dataframe['ma_99'] < dataframe['ma_25']) # 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 dataframe['volume_threshold'] = dataframe['volume'].rolling(window=50).mean() * 1.5 for index, row in dataframe.iterrows(): c1 = row['rsi'] >= row['rsi_min_custom'] - self.rsi_range.value c2 = row['rsi'] <= row['rsi_min_custom'] + self.rsi_range.value c3 = row['rsi'] < 50 c4 = row['ma_25'] < row['ma_7'] c5 = row['ma_99'] < row['ma_25'] c6 = row['mid_price'] > row['ma_25'] c7 = row['volume'] > 0 c8 = row['volume_threshold'] > 0 and row['volume'] / row['volume_threshold'] > self.volume_ratio_threshold_1.value c9 = row['volume'] > row['volume_threshold'] c10 = row['avg_abs_oscillation_pct_30m'] >= self.oscillation_threshold.value c11 = row.get('rsi_5m_5m', 0) > self.rsi_5m_buy_threshold.value c12 = row['bullish_divergence'] == 1 c13 = row['volume'] > 0 c14 = row['volume_threshold'] > 0 and row['volume'] / row['volume_threshold'] > self.volume_ratio_threshold_2.value if all([c1, c2, c3, c4, c5, c6, c7, c8, c9, c10, c11]) or all([c12, c13, c14]): logger.info(f"[{metadata['pair']}] Señal de compra detectada en {index} -> " f"c1={c1} c2={c2} c3={c3} c4={c4} c5={c5} c6={c6} c7={c7} c8={c8} " f"c9={c9} c10={c10} c11={c11} | c12={c12} c13={c13} c14={c14}") logger.info(f"[{metadata['pair']}] Buy signal at {index} | " f"RSI={row['rsi']:.2f}, RSI_min_custom={row['rsi_min_custom']:.2f}, " f"MA7={row['ma_7']:.2f}, MA25={row['ma_25']:.2f}, MA99={row['ma_99']:.2f}, " f"MidPrice={row['mid_price']:.2f}, Volume={row['volume']:.2f}, " f"Threshold={row['volume_threshold']:.2f}, " f"OscilPct30m={row['avg_abs_oscillation_pct_30m']:.4f}, RSI_5m={row.get('rsi_5m_5m', 0):.2f}, " f"BullDiv={row['bullish_divergence']}") # Condiciones de compra dataframe.loc[ ( (dataframe['rsi'] >= dataframe['rsi_min_custom'] - self.rsi_range.value) & # RSI dentro del rango (dataframe['rsi'] <= dataframe['rsi_min_custom'] + self.rsi_range.value) & (dataframe['rsi'] < 50) & (ma_25_above_7) & (ma99_above_25) & (dataframe['mid_price'] > dataframe['ma_25']) & (dataframe['volume'] > 0) & (dataframe['volume'] / dataframe['volume_threshold'] > self.volume_ratio_threshold_1.value) & (dataframe['volume'] > dataframe['volume_threshold']) & (dataframe['avg_abs_oscillation_pct_30m'] >= self.oscillation_threshold.value) & (dataframe['rsi_5m_5m'] > self.rsi_5m_buy_threshold.value) ) | ( (dataframe['bullish_divergence'] == 1) & (dataframe['volume'] > 0) & (dataframe['volume'] / dataframe['volume_threshold'] > self.volume_ratio_threshold_2.value) ), 'buy' ] = 1 # 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 = {} for index, row in dataframe[dataframe['buy'] == 1].iterrows(): trade_time = index.to_pydatetime() for offset in [0, 1]: # 0 minutos y 1 minuto time_key = trade_time + pd.Timedelta(minutes=offset) self.custom_trade_info[time_key] = { 'avg_abs_oscillation_pct_30m': row.get('avg_abs_oscillation_pct_30m', 'No Data'), 'ma_7': row.get('ma_7', 'No Data'), 'ma_25': row.get('ma_25', 'No Data'), 'ma_99': row.get('ma_99', 'No Data'), 'rsi': row.get('rsi', 'No Data'), 'rsi_5m_5m': row.get('rsi_5m_5m', 'No Data'), 'sma_rsi_5m_5m': row.get('sma_rsi_5m_5m', 'No Data'), 'price': row.get('close', 'No Data'), 'bullish_divergence': row.get('bullish_divergence', 'No Data'), 'bearish_divergence': row.get('bearish_divergence', 'No Data'), 'volume': row.get('volume', 'No Data'), 'volume_threshold': row.get('volume_threshold', 'No Data') } #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_exit(self, pair: str, trade, current_time: datetime, current_rate: float, current_profit: float, **kwargs) -> bool: """ ROI dinámico basado en ATR: cuanto más alto el ATR, mayor el ROI requerido. """ 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 trade_open_time = trade.open_date_utc.replace(second=0, microsecond=0) if trade_open_time not in dataframe.index: trade_open_time = dataframe.index[dataframe.index.get_loc(trade_open_time, method='pad')] entry_price = trade.open_rate atr_at_entry = dataframe.loc[trade_open_time, 'atr'] # ROI basado en múltiplo del ATR atr_multiplier = self.atr_multiplier_roi.value roi_price = entry_price + (atr_at_entry * atr_multiplier) if current_rate >= roi_price: logger.info(f"[{pair}] ROI dinámico alcanzado | Entrada: {entry_price:.4f} | " f"ATR: {atr_at_entry:.4f} | ROI Target: {roi_price:.4f} | " f"Actual: {current_rate:.4f}") return True # Ejecutar venta return False except Exception as e: logger.error(f"[{pair}] Error en custom_exit: {e}") return False 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: trade_time = trade.open_date trade_info = self.custom_trade_info.get(trade_time, { 'avg_abs_oscillation_pct_30m' : 0, 'ma_7' : 0, 'ma_25' : 0, 'ma_99' : 0, 'rsi' : 0, 'rsi_5m_5m': 0, 'sma_rsi_5m_5m': 0, 'volume': 0, 'volume_threshold': 0 }) 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"{trade.calc_profit_ratio(rate) * 100:.2f}%".rjust(6) if not self.is_hyperopt(): # Imprime detalles de la operación al momento de salir print( f"[{pair}] " f"[{open_date_spain_str} | " f"{current_time_spain_str}] " f"Beneficio: {beneficio} | " f"[{exit_reason[:4].ljust(4)}] | " f"RSI: {trade_info['rsi']:.2f} | " f"RSI 5M: {trade_info['rsi_5m_5m']:.2f} | " f"RSI 5M SMA: {trade_info['sma_rsi_5m_5m']:.2f} | " f"avg: {trade_info['avg_abs_oscillation_pct_30m']:.4f} | " f"ma7: {trade_info['ma_7']:.4f} | " f"ma25: {trade_info['ma_25']:.4f} | " f"ma99: {trade_info['ma_99']:.4f} | " f"bull: {trade_info['bullish_divergence']} | " f"bear: {trade_info['bearish_divergence']} | " f"vol: {trade_info['volume']} | " f"thr: {trade_info['volume_threshold']} | " f"ratio: {(trade_info['volume'] / trade_info['volume_threshold']):.3f}" ) # 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)