from freqtrade.strategy import IStrategy, IntParameter, DecimalParameter from pandas import DataFrame import talib as ta import logging import numpy as np import pandas as pd from datetime import datetime import pytz class Hermes_v4(IStrategy): timeframe = '1m' minimal_roi = { "0": 0.01 # 1% de ROI } # Parámetros configurables # Minutos sobre los que se buscará el RSI mínimo rsi_window = IntParameter(10, 150, default=40, space='buy', optimize=True) # Rango de margen para el RSI mínimo encontrado rsi_range = DecimalParameter(0.1, 10.0, default=0.5, decimals=2, space='buy', optimize=True) use_custom_stoploss = False trailing_stop = False def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame: # 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 slope_window = 30 dataframe['ma_99_slope'] = dataframe['ma_99'].rolling(window=slope_window).apply(self.calculate_slope_in_degrees, raw=True) # 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=30).mean() return dataframe @property def protections(self): return [ { "method": "StoplossGuard", "lookback_period_candles": 1, # Considera solo el último trade para activar la protección "trade_limit": 1, # Solo se necesita un stoploss para activar la protección "stop_duration_candles": 120, # 120 minutos de bloqueo en timeframe de 1m "required_profit": 0.0, # No requiere beneficios para desactivar "only_per_pair": False, # Se aplica globalmente "only_per_side": False # Se aplica tanto a compras como a ventas } ] @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 def populate_buy_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: pair = metadata.get("pair", "UNKNOWN_PAIR") 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']) # 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) & (ma_25_above_7) & # MA(7) < MA(25) (ma99_above_25) & (dataframe['ma_99_slope'] > 0), # MA(99) < MA(25) '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() self.custom_trade_info[trade_time] = { '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'), 'price': row.get('close', 'No Data'), 'ma_99_slope': row.get('ma_99_slope', 'No Data') } print(f"Almacenado datos para {trade_time}: {self.custom_trade_info[trade_time]}") return dataframe def populate_sell_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: return dataframe 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_99_slope' : 0, 'ma_7' : 0, 'ma_25' : 0, 'ma_99' : 0, 'rsi' : 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) # if trade_info != "No Data": # print( # f"---[{pair}] | " # f"Fecha de entrada: {trade.open_date} | " # f"Fecha de salida: {current_time} | " # f"Beneficio: {trade.calc_profit_ratio(rate) * 100:.2f}% | " # f"Motivo: {exit_reason} | " # f"Detalles: {trade_info}" # ) # Imprime detalles de la operación al momento de salir print( f"[{pair}] | " f"[{open_date_spain} | " f"{current_time_spain}] " f"Beneficio: {trade.calc_profit_ratio(rate) * 100:.2f}% | " f"Motivo: {exit_reason} | " f"RSI: {trade_info['rsi']:.4f} | " 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"desv: {trade_info['ma_99_slope']:.8f}" ) # 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)