LiquiditySweepStrategy

"""
LiquiditySweepStrategy - Vectorized Liquidity Sweep detection strategy (Improved v2).

Concept: Identifies "sweeps" of key swing high/low points, confirmed by a candle
close back within the range, filtered by volume, trend, and market structure.

Improvements (v2):
- Trend Filter: EMA alignment + ADX to avoid trading against strong trends
- Market Structure Confirmation: Strong reclaim candle (closes in top 30% of range)
- Pivot Freshness: Min age requirement to avoid sweeping very recent pivots
- Enhanced exit logic with trailing and bearish sweeps

Entry (Long): Price sweeps below a recent pivot low, reclaims strongly, with trend confirmation.
Exit (Long): Opposing sweep, RSI overbought, or trailing stop.
"""

from freqtrade.strategy import IStrategy, IntParameter, DecimalParameter, BooleanParameter
from pandas import DataFrame
import pandas_ta as ta
import numpy as np


class LiquiditySweepStrategy(IStrategy):
    """
    A liquidity sweep strategy adapted from Pine Script to Freqtrade (v2 Improved).

    Detects sweeps of key swing highs/lows using vectorized pandas operations.
    Includes trend filtering, market structure confirmation, and smart exits.

    Key Improvements:
    - EMA + ADX trend filter to avoid catching falling knives
    - Strong reclaim confirmation (candle closes in upper portion of range)
    - Pivot age requirement to ensure we're sweeping significant levels
    """

    INTERFACE_VERSION = 3

    # ==========================================================================
    # Strategy Settings
    # ==========================================================================
    minimal_roi = {
        "0": 0.15,    # 15% at any time
        "60": 0.08,   # 8% after 60 minutes
        "120": 0.04,  # 4% after 2 hours
        "240": 0.02,  # 2% after 4 hours
    }

    # Stoploss - fallback fixed stoploss (custom_stoploss overrides this)
    stoploss = -0.08  # -8%

    # Enable custom stoploss
    use_custom_stoploss = True

    # Timeframe
    timeframe = '1h'

    # Startup candle count - need enough data for EMA-200 and pivot detection
    startup_candle_count = 250

    # ==========================================================================
    # HYPEROPT PARAMETERS - Original
    # ==========================================================================

    # Pivot Lookback: How many candles to look before/after finding a pivot
    # Pivot requires N candles on each side, so window = 2 * pivot_len + 1
    pivot_len = IntParameter(2, 10, default=5, space='buy', optimize=True)

    # Volume Filter - whether to require above-average volume on sweep
    use_vol_check = BooleanParameter(default=True, space='buy', optimize=True)

    # Volume MA Length for calculating average volume
    vol_len = IntParameter(10, 50, default=20, space='buy', optimize=True)

    # Volume Multiplier - sweep candle volume must exceed (vol_ma * vol_mult)
    vol_mult = DecimalParameter(1.0, 3.0, default=1.5, decimals=1, space='buy', optimize=True)

    # How far back to search for the "swept" liquidity level (for ffill limit)
    sweep_lookback = IntParameter(10, 60, default=20, space='buy', optimize=True)

    # Noise Filter: Minimum percentage distance for a sweep to be valid (0.1% to 1%)
    min_sweep_dist = DecimalParameter(0.001, 0.01, default=0.002, decimals=3, space='buy', optimize=True)

    # RSI exit threshold - exit when RSI exceeds this (overbought)
    rsi_exit_threshold = IntParameter(60, 90, default=75, space='sell', optimize=True)

    # ==========================================================================
    # HYPEROPT PARAMETERS - New (v2 Improvements)
    # ==========================================================================

    # ---------- Trend Filter ----------
    # Short EMA period for trend detection
    trend_ema_short = IntParameter(20, 100, default=50, space='buy', optimize=True)

    # Long EMA period for trend detection
    trend_ema_long = IntParameter(100, 300, default=200, space='buy', optimize=True)

    # ADX Filter - whether to use ADX to detect strong trends
    use_adx_filter = BooleanParameter(default=True, space='buy', optimize=True)

    # ADX threshold - above this value indicates a strong trend
    adx_threshold = IntParameter(20, 40, default=25, space='buy', optimize=True)

    # ---------- Confirmation ----------
    # Require strong reclaim candle (closes in upper portion of range)
    require_strong_reclaim = BooleanParameter(default=True, space='buy', optimize=True)

    # Reclaim strength threshold (0.7 = candle closes in top 30% of its range)
    reclaim_strength_threshold = DecimalParameter(0.5, 0.9, default=0.7, decimals=1, space='buy', optimize=True)

    # ---------- Pivot Freshness ----------
    # Minimum candles between pivot formation and sweep (avoid sweeping very recent lows)
    min_pivot_age = IntParameter(5, 30, default=10, space='buy', optimize=True)

    # ==========================================================================
    # INDICATOR CALCULATION
    # ==========================================================================

    def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        """
        Calculate all technical indicators needed for the strategy.
        """
        pivot_window = 2 * self.pivot_len.value + 1

        # ---------------------------------------------------------------------
        # Trend Indicators (NEW - v2)
        # ---------------------------------------------------------------------
        dataframe['ema_short'] = ta.ema(dataframe['close'], length=self.trend_ema_short.value)
        dataframe['ema_long'] = ta.ema(dataframe['close'], length=self.trend_ema_long.value)

        # ADX for trend strength
        adx_result = ta.adx(dataframe['high'], dataframe['low'], dataframe['close'], length=14)
        dataframe['adx'] = adx_result['ADX_14']

        # ---------------------------------------------------------------------
        # Pivot Low Detection (Vectorized)
        # A pivot low occurs when the low is the minimum over a centered window
        # ---------------------------------------------------------------------
        dataframe['rolling_min'] = dataframe['low'].rolling(
            window=pivot_window,
            center=True
        ).min()
        dataframe['is_pivot_low'] = dataframe['low'] == dataframe['rolling_min']

        # Extract pivot low values, NaN elsewhere
        dataframe['pivot_low_value'] = np.where(
            dataframe['is_pivot_low'],
            dataframe['low'],
            np.nan
        )

        # Calculate pivot age (bars since last pivot)
        # Create a cumulative count that resets at each pivot
        dataframe['pivot_low_idx'] = np.where(
            dataframe['is_pivot_low'],
            dataframe.index,
            np.nan
        )
        dataframe['last_pivot_low_idx'] = dataframe['pivot_low_idx'].ffill()
        dataframe['pivot_age'] = dataframe.index - dataframe['last_pivot_low_idx']

        # Forward fill to propagate the last known pivot low
        # Limit ffill to sweep_lookback to avoid stale levels
        dataframe['last_pivot_low'] = dataframe['pivot_low_value'].ffill(
            limit=self.sweep_lookback.value
        )

        # CRITICAL: A centered rolling window confirms a pivot only after pivot_len
        # future candles have been seen. Shift by (pivot_len + 1) to ensure we only
        # compare against pivots that are fully confirmed and avoid look-ahead bias.
        pivot_shift = self.pivot_len.value + 1
        dataframe['last_pivot_low'] = dataframe['last_pivot_low'].shift(pivot_shift)
        dataframe['pivot_age'] = dataframe['pivot_age'].shift(pivot_shift)

        # ---------------------------------------------------------------------
        # Pivot High Detection (Vectorized) - for opposing sweep exits
        # ---------------------------------------------------------------------
        dataframe['rolling_max'] = dataframe['high'].rolling(
            window=pivot_window,
            center=True
        ).max()
        dataframe['is_pivot_high'] = dataframe['high'] == dataframe['rolling_max']

        # Extract pivot high values
        dataframe['pivot_high_value'] = np.where(
            dataframe['is_pivot_high'],
            dataframe['high'],
            np.nan
        )

        # Forward fill for last known pivot high
        dataframe['last_pivot_high'] = dataframe['pivot_high_value'].ffill(
            limit=self.sweep_lookback.value
        )

        # Shift by (pivot_len + 1) to match pivot low shift and avoid look-ahead
        dataframe['last_pivot_high'] = dataframe['last_pivot_high'].shift(pivot_shift)

        # ---------------------------------------------------------------------
        # Volume Indicators
        # ---------------------------------------------------------------------
        dataframe['vol_ma'] = dataframe['volume'].rolling(
            window=self.vol_len.value
        ).mean()

        dataframe['high_volume'] = dataframe['volume'] > (
            dataframe['vol_ma'] * self.vol_mult.value
        )

        # ---------------------------------------------------------------------
        # RSI for Exit
        # ---------------------------------------------------------------------
        dataframe['rsi'] = ta.rsi(dataframe['close'], length=14)

        # ---------------------------------------------------------------------
        # Sweep Depth Calculation (for noise filter)
        # How much the wick went below the pivot as a percentage
        # ---------------------------------------------------------------------
        dataframe['sweep_depth'] = (
            (dataframe['last_pivot_low'] - dataframe['low']) / dataframe['low']
        )

        # ---------------------------------------------------------------------
        # Candle Reclaim Strength (NEW - v2)
        # How strongly the candle closed back above the sweep
        # 1.0 = closed at high, 0.0 = closed at low
        # ---------------------------------------------------------------------
        candle_range = dataframe['high'] - dataframe['low']
        # Avoid division by zero for doji candles
        candle_range = candle_range.replace(0, np.nan)
        dataframe['reclaim_strength'] = (dataframe['close'] - dataframe['low']) / candle_range
        dataframe['reclaim_strength'] = dataframe['reclaim_strength'].fillna(0.5)

        # Store the sweep candle low for custom stoploss
        dataframe['sweep_candle_low'] = dataframe['low']

        return dataframe

    # ==========================================================================
    # ENTRY LOGIC (Improved v2)
    # ==========================================================================

    def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        """
        Define entry signals for long positions (v2 with trend + confirmation).

        Improved Entry Conditions:
        1. TREND FILTER: EMA alignment + ADX (don't buy in strong downtrends)
        2. SWEEP: Wick sweeps below the last pivot low (low < last_pivot_low)
        3. PIVOT AGE: Pivot must be old enough (not a very recent low)
        4. RECLAIM: Body reclaims above the pivot (close > last_pivot_low)
        5. CONFIRMATION: Strong reclaim candle (closes in top 30% of range)
        6. VOLUME: Above-average volume on sweep candle
        7. NOISE FILTER: Sweep depth exceeds minimum threshold
        """

        # =====================================================================
        # 1. TREND FILTER (NEW - v2)
        # =====================================================================
        # Basic uptrend: short EMA > long EMA AND price above long EMA
        is_uptrend = (
            (dataframe['ema_short'] > dataframe['ema_long']) &
            (dataframe['close'] > dataframe['ema_long'])
        )

        # ADX filter: block entries in STRONG downtrends
        if self.use_adx_filter.value:
            is_strong_downtrend = (
                (dataframe['adx'] > self.adx_threshold.value) &
                (dataframe['ema_short'] < dataframe['ema_long'])
            )
            trend_ok = is_uptrend & (~is_strong_downtrend)
        else:
            trend_ok = is_uptrend

        # =====================================================================
        # 2. SWEEP CONDITION
        # =====================================================================
        sweep_condition = (
            # Wick went below pivot
            (dataframe['low'] < dataframe['last_pivot_low']) &
            # Close is back above pivot (reclaim)
            (dataframe['close'] > dataframe['last_pivot_low']) &
            # Must have valid pivot data
            (dataframe['last_pivot_low'].notna())
        )

        # =====================================================================
        # 3. PIVOT AGE FILTER (NEW - v2)
        # =====================================================================
        pivot_fresh_enough = dataframe['pivot_age'] > self.min_pivot_age.value

        # =====================================================================
        # 4. CONFIRMATION - Strong Reclaim Candle (NEW - v2)
        # =====================================================================
        if self.require_strong_reclaim.value:
            strong_reclaim = dataframe['reclaim_strength'] > self.reclaim_strength_threshold.value
        else:
            strong_reclaim = True  # Always true if disabled

        # =====================================================================
        # 5. NOISE FILTER
        # =====================================================================
        noise_filter = dataframe['sweep_depth'] > self.min_sweep_dist.value

        # =====================================================================
        # 6. VOLUME FILTER
        # =====================================================================
        if self.use_vol_check.value:
            volume_ok = dataframe['high_volume']
        else:
            volume_ok = True

        # =====================================================================
        # COMBINE ALL CONDITIONS
        # =====================================================================
        conditions = (
            trend_ok &
            sweep_condition &
            pivot_fresh_enough &
            strong_reclaim &
            noise_filter &
            volume_ok &
            (dataframe['volume'] > 0)  # Safety check
        )

        dataframe.loc[conditions, 'enter_long'] = 1

        return dataframe

    # ==========================================================================
    # EXIT LOGIC
    # ==========================================================================

    def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        """
        Define exit signals for long positions.

        Exit Conditions (OR logic):
        A. Opposing Sweep: high sweeps above pivot high and closes below (bearish sweep)
        B. RSI Overbought: RSI exceeds threshold
        C. Trend Reversal: Short EMA crosses below Long EMA (optional, implicit via next entry filter)
        """
        # Condition A: Opposing (Bearish) Sweep
        bearish_sweep = (
            # Wick swept above pivot high
            (dataframe['high'] > dataframe['last_pivot_high']) &
            # Body closed back below pivot (failed breakout)
            (dataframe['close'] < dataframe['last_pivot_high']) &
            # Must have valid pivot data
            (dataframe['last_pivot_high'].notna())
        )

        # Condition B: RSI Overbought Exit
        rsi_exit = dataframe['rsi'] > self.rsi_exit_threshold.value

        # Combine exits with OR logic
        dataframe.loc[
            (bearish_sweep | rsi_exit) &
            (dataframe['volume'] > 0),  # Safety check
            'exit_long'
        ] = 1

        return dataframe

    # ==========================================================================
    # CUSTOM STOPLOSS (Dynamic)
    # ==========================================================================

    def custom_stoploss(
        self,
        pair: str,
        trade: 'Trade',
        current_time: 'datetime',
        current_rate: float,
        current_profit: float,
        after_fill: bool,
        **kwargs
    ) -> float:
        """
        Dynamic stoploss based on the sweep candle low.

        Instead of a fixed percentage, we place the stop just below
        the low of the entry candle (the sweep candle).

        Returns: stoploss as a negative ratio from current_rate, or -1 to disable.
        """
        # Access the dataframe to get the sweep candle low at entry
        dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)

        if dataframe.empty:
            return -1  # Use default stoploss

        # Find the candle at trade entry time
        entry_candle = dataframe.loc[dataframe['date'] <= trade.open_date_utc]

        if entry_candle.empty:
            return -1  # Use default stoploss

        # Get the low of the entry (sweep) candle
        sweep_low = entry_candle.iloc[-1]['low']

        # Place stop slightly below (0.5% buffer)
        stop_price = sweep_low * 0.995

        # Calculate stoploss ratio from entry price
        stoploss_ratio = (stop_price - trade.open_rate) / trade.open_rate

        # Return the stoploss (must be negative for a long position stop below entry)
        # If sweep low is above entry (shouldn't happen), use default
        if stoploss_ratio >= 0:
            return -1  # Use fallback fixed stoploss

        return stoploss_ratio