# ══════════════════════════════════════════════════════════════ # anis solidscale - Elite Spot Trading Suite # STRATÉGIE : LiquiditySweepOB # CATÉGORIE : Smart Money — Liquidity Sweep + Order Block # ══════════════════════════════════════════════════════════════ # # LOGIQUE : # 1. Détection de Liquidity Sweep : le prix fait un faux breakout # sous un swing low récent (déclenche les stops) puis remonte # 2. Order Block bullish : dernière bougie rouge avant un mouvement # haussier de >1% dans les 3 bougies suivantes # 3. RSI < seuil d'entrée (pas de surachat) # 4. Sortie : prix atteint le swing high OU RSI > seuil exit # ══════════════════════════════════════════════════════════════ import sys from pathlib import Path import numpy as np from pandas import DataFrame from freqtrade.strategy import IStrategy, IntParameter, DecimalParameter sys.path.insert(0, str(Path(__file__).resolve().parent.parent.parent)) from utils.indicators import CommonIndicators from utils.logging_utils import TradeLogger from utils.telegram_notifier import TelegramNotifier class LiquiditySweepOB(IStrategy): INTERFACE_VERSION = 3 can_short = False timeframe = "1h" startup_candle_count = 100 minimal_roi = {"0": 0.10, "360": 0.05, "720": 0.02} stoploss = -0.05 trailing_stop = True trailing_stop_positive = 0.02 trailing_stop_positive_offset = 0.03 trailing_only_offset_is_reached = True # ── Buy params ── swing_lookback = IntParameter(3, 10, default=5, space="buy") rsi_period = IntParameter(7, 21, default=14, space="buy") rsi_entry = IntParameter(40, 65, default=55, space="buy") ob_threshold = DecimalParameter(0.005, 0.03, default=0.01, space="buy") volume_period = IntParameter(10, 50, default=20, space="buy") # ── Sell params ── rsi_exit = IntParameter(65, 85, default=75, space="sell") _logger = None _notifier = None def _init_utils(self) -> None: if self._logger is None: self._logger = TradeLogger(strategy_name="LiquiditySweepOB") self._notifier = TelegramNotifier() def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame: self._init_utils() # Pre-calculer RSI pour TOUTES les valeurs possibles (hyperopt-safe) for rsi_p in range(self.rsi_period.low, self.rsi_period.high + 1): dataframe = CommonIndicators.add_rsi(dataframe, period=rsi_p) # Pre-calculer volume SMA pour TOUTES les valeurs possibles for vol_p in range(self.volume_period.low, self.volume_period.high + 1): dataframe = CommonIndicators.add_volume_sma(dataframe, period=vol_p) # Swing highs/lows pour TOUTES les valeurs de lookback for lookback in range(self.swing_lookback.low, self.swing_lookback.high + 1): sh_col = f"swing_high_{lookback}" sl_col = f"swing_low_{lookback}" dataframe[sh_col] = np.nan dataframe[sl_col] = np.nan for i in range(lookback, len(dataframe) - lookback): high_val = dataframe["high"].iloc[i] is_swing = True for j in range(1, lookback + 1): if high_val <= dataframe["high"].iloc[i - j] or high_val <= dataframe["high"].iloc[i + j]: is_swing = False break if is_swing: dataframe.iloc[i, dataframe.columns.get_loc(sh_col)] = high_val low_val = dataframe["low"].iloc[i] is_swing = True for j in range(1, lookback + 1): if low_val >= dataframe["low"].iloc[i - j] or low_val >= dataframe["low"].iloc[i + j]: is_swing = False break if is_swing: dataframe.iloc[i, dataframe.columns.get_loc(sl_col)] = low_val dataframe[f"recent_swing_high_{lookback}"] = dataframe[sh_col].ffill() dataframe[f"recent_swing_low_{lookback}"] = dataframe[sl_col].ffill() # Order Block bullish : bougie rouge (close < open) suivie d'une hausse > threshold # dans les 3 bougies suivantes # Pre-calculer pour toutes les valeurs de ob_threshold # On utilise la valeur max du threshold pour le calcul, le filtrage se fait dans entry dataframe["is_red_candle"] = (dataframe["close"] < dataframe["open"]).astype(int) # Hausse max dans les 3 bougies suivantes par rapport au close de la bougie courante dataframe["future_max_close"] = dataframe["close"].shift(-1).rolling(3).max() dataframe["future_return"] = ( (dataframe["future_max_close"] - dataframe["close"]) / dataframe["close"] ) # Marquer les Order Blocks (bougie rouge avec hausse future > threshold) # On forward-fill le prix de l'OB pour savoir si le prix actuel est dans/au-dessus dataframe["ob_price_high"] = np.nan dataframe["ob_price_low"] = np.nan # On pre-calcule pour le seuil minimum (toutes les valeurs >= min seront couvertes) min_threshold = 0.005 ob_mask = (dataframe["is_red_candle"] == 1) & (dataframe["future_return"] > min_threshold) dataframe.loc[ob_mask, "ob_price_high"] = dataframe.loc[ob_mask, "open"] dataframe.loc[ob_mask, "ob_price_low"] = dataframe.loc[ob_mask, "close"] dataframe["ob_price_high"] = dataframe["ob_price_high"].ffill() dataframe["ob_price_low"] = dataframe["ob_price_low"].ffill() dataframe["ob_future_return"] = np.nan dataframe.loc[ob_mask, "ob_future_return"] = dataframe.loc[ob_mask, "future_return"] dataframe["ob_future_return"] = dataframe["ob_future_return"].ffill() return dataframe def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: rsi_col = f"rsi_{self.rsi_period.value}" sl_col = f"recent_swing_low_{self.swing_lookback.value}" sh_col = f"recent_swing_high_{self.swing_lookback.value}" # Liquidity sweep : low passe sous le swing low mais close reste au-dessus (faux breakout) sweep = ( (dataframe["low"] < dataframe[sl_col]) & (dataframe["close"] > dataframe[sl_col]) ) # Prix dans/au-dessus de l'Order Block + OB valide pour le threshold courant in_ob = ( (dataframe["close"] >= dataframe["ob_price_low"]) & (dataframe["ob_future_return"] >= self.ob_threshold.value) ) conditions = ( sweep & in_ob & (dataframe[rsi_col] < self.rsi_entry.value) & (dataframe["volume"] > 0) ) dataframe.loc[conditions, "enter_long"] = 1 return dataframe def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: rsi_col = f"rsi_{self.rsi_period.value}" sh_col = f"recent_swing_high_{self.swing_lookback.value}" conditions = ( (dataframe["close"] >= dataframe[sh_col]) | (dataframe[rsi_col] > self.rsi_exit.value) ) dataframe.loc[conditions, "exit_long"] = 1 return dataframe