# Filename: ICTSessionFakeoutStrategy.py from freqtrade.strategy import IStrategy from pandas import DataFrame import pandas as pd import pytz from typing import Optional from datetime import datetime class ICTSessionFakeoutStrategy(IStrategy): """ Estrategia inspirada en ICT - NY Session Fakeout con SL y TP dinámicos. Usa las 4 velas de 15m previas a la franja 20-22 NY. """ timeframe = "1m" minimal_roi = {"0": 0.02} stoploss = -0.99 trailing_stop = False position_adjustment_enable = False informative_timeframes = {"15m"} def informative_pairs(self): pairs = [(pair, "15m") for pair in self.dp.current_whitelist()] return pairs daily_levels = {} def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame: pair = metadata["pair"] ny_tz = pytz.timezone("America/New_York") # Procesar velas 1m primero dataframe["date"] = pd.to_datetime(dataframe["date"]) if dataframe["date"].dt.tz is None: dataframe["date_ny"] = dataframe["date"].dt.tz_localize("UTC").dt.tz_convert(ny_tz) else: dataframe["date_ny"] = dataframe["date"].dt.tz_convert(ny_tz) dataframe["date_ny_date"] = dataframe["date_ny"].dt.date dataframe["hour_ny"] = dataframe["date_ny"].dt.hour dataframe["minute_ny"] = dataframe["date_ny"].dt.minute # Cargar velas 15m df_15m = self.dp.get_pair_dataframe(pair=pair, timeframe="15m") df_15m["date"] = pd.to_datetime(df_15m["date"]) if df_15m["date"].dt.tz is None: df_15m["date_ny"] = df_15m["date"].dt.tz_localize("UTC").dt.tz_convert(ny_tz) else: df_15m["date_ny"] = df_15m["date"].dt.tz_convert(ny_tz) df_15m["date_ny_date"] = df_15m["date_ny"].dt.date df_15m["hour_ny"] = df_15m["date_ny"].dt.hour print("\n🔹 Fechas mín/max en 15m NY:") print("UTC:", df_15m["date"].min(), "->", df_15m["date"].max()) print("NY :", df_15m["date_ny"].min(), "->", df_15m["date_ny"].max()) self.daily_levels = {} for day, group in df_15m.groupby("date_ny_date"): # Extraer las 4 velas de 19:00 a 19:59 NY pre_session = group[group["hour_ny"] == 19] if len(pre_session) >= 1: high_row = pre_session.loc[pre_session["high"].idxmax()] low_row = pre_session.loc[pre_session["low"].idxmin()] self.daily_levels[day] = { "high": high_row["high"], "high_time": high_row["date_ny"], "low": low_row["low"], "low_time": low_row["date_ny"], } print( f"\n📅 {day}: " f"High={high_row['high']} ({high_row['date_ny']}), " f"Low={low_row['low']} ({low_row['date_ny']})" ) # Buscar rupturas en velas 1m de ese mismo día y franja 20-22 NY franja_1m = dataframe[ (dataframe["date_ny_date"] == day) & (dataframe["hour_ny"] >= 20) & (dataframe["hour_ny"] < 22) & ( (dataframe["high"] > high_row["high"]) | (dataframe["low"] < low_row["low"]) ) ] if not franja_1m.empty: print(" ➤ Rupturas en 1m dentro de 20-22 NY:") ruptura_detectada = None ruptura_idx = None stoploss_value = None # Aquí guardamos r["close"] # Detectar la primera ruptura y marcar for idx, r in franja_1m.iterrows(): if ruptura_detectada is None: if r["high"] > high_row["high"]: ruptura_detectada = "HIGH" nivel = high_row["high"] min_pre_franja = dataframe[ (dataframe["date_ny_date"] == day) & (dataframe["hour_ny"] < 20) ]["low"].min() detalle = f"Mínimo 1m previo={min_pre_franja}" else: ruptura_detectada = "LOW" nivel = low_row["low"] max_pre_franja = dataframe[ (dataframe["date_ny_date"] == day) & (dataframe["hour_ny"] < 20) ]["high"].max() detalle = f"Máximo 1m previo={max_pre_franja}" ruptura_idx = idx stoploss_value = r["close"] print( f" - {r['date_ny']} | PRIMERA ruptura {ruptura_detectada} del nivel {nivel} | " f"Cierre={r['close']} | {detalle}" ) break # Si hubo una ruptura inicial, buscar contraria desde esa vela en adelante if ruptura_detectada: posteriores = dataframe.loc[ruptura_idx + 1:] contraria = None if ruptura_detectada == "HIGH": contraria = posteriores[ (posteriores["date_ny_date"] == day) & (posteriores["hour_ny"] >= 20) & (posteriores["hour_ny"] < 22) & (posteriores["close"] < min_pre_franja) ] elif ruptura_detectada == "LOW": contraria = posteriores[ (posteriores["date_ny_date"] == day) & (posteriores["hour_ny"] >= 20) & (posteriores["hour_ny"] < 22) & (posteriores["close"] > max_pre_franja) ] if contraria is not None and not contraria.empty: print(" ➤ Ruptura CONTRARIA en 1m tras la primera ruptura:") for idx_c, rr in contraria.iterrows(): # Señal de entrada y stoploss de la PRIMERA ruptura if ruptura_detectada == "HIGH": dataframe.loc[idx_c, "open_short_signal"] = 1 dataframe.loc[idx_c, "short_stoploss"] = stoploss_value # Takeprofit: 2:1 ratio entry = rr["close"] sl = stoploss_value tp = entry - 2 * (sl - entry) dataframe.loc[idx_c, "short_takeprofit"] = tp elif ruptura_detectada == "LOW": dataframe.loc[idx_c, "open_long_signal"] = 1 dataframe.loc[idx_c, "long_stoploss"] = stoploss_value # Takeprofit: 2:1 ratio entry = rr["close"] sl = stoploss_value tp = entry + 2 * (entry - sl) dataframe.loc[idx_c, "long_takeprofit"] = tp print( f" - {rr['date_ny']} | Ruptura CONTRARIA | Cierre={rr['close']} sl: {stoploss_value} tp: {tp}" ) break # Solo la primera ruptura contraria else: print(" ➤ No hubo rupturas contrarias tras la primera ruptura.") else: print(" ➤ No hubo rupturas en 1m dentro de 20-22 NY.") else: print(f"\n📅 {day}: ❗ No hay velas de 19 NY") # Crear columnas auxiliares con el nivel de ese día dataframe["high_level"] = dataframe["date_ny_date"].map(lambda d: self.daily_levels.get(d, {}).get("high")) dataframe["low_level"] = dataframe["date_ny_date"].map(lambda d: self.daily_levels.get(d, {}).get("low")) return dataframe def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: dataframe["enter_long"] = dataframe.get("open_long_signal", 0) dataframe["enter_short"] = dataframe.get("open_short_signal", 0) # Filtrar solo las filas con entradas entradas = dataframe[(dataframe["enter_long"] == 1) | (dataframe["enter_short"] == 1)] print("\n🔹 Entradas con SL y TP configurados (últimas 5 filas):") print(entradas[[ "date_ny", "enter_long", "long_stoploss", "long_takeprofit", "enter_short", "short_stoploss", "short_takeprofit" ]].tail(5)) return dataframe def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: dataframe["exit_long"] = 0 dataframe["exit_short"] = 0 return dataframe def custom_exit( self, pair: str, trade, current_time: datetime, current_rate: float, current_profit: float, **kwargs ) -> Optional[str]: dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe) if dataframe is None or len(dataframe) < 1: return None row = dataframe.iloc[-1] if trade.is_long: sl = row["long_stoploss"] tp = row["long_takeprofit"] if sl and current_rate <= sl: print(f"\n🚨 Stoploss LONG alcanzado ({current_rate} <= {sl})") return "stoploss_reached" if tp and current_rate >= tp: print(f"\n✅ Takeprofit LONG alcanzado ({current_rate} >= {tp})") return "takeprofit_reached" if trade.is_short: sl = row["short_stoploss"] tp = row["short_takeprofit"] if sl and current_rate >= sl: print(f"\n🚨 Stoploss SHORT alcanzado ({current_rate} >= {sl})") return "stoploss_reached" if tp and current_rate <= tp: print(f"\n✅ Takeprofit SHORT alcanzado ({current_rate} <= {tp})") return "takeprofit_reached" return None