LiquidationHunterStrategy

# LiquidationHunterStrategy - Profit from liquidation cascades
# Powered by AiCoin's exclusive liquidation data + aggregated open interest
#
# How it works:
#   - When OI is rising fast (lots of new leveraged positions), the market is fragile
#   - AiCoin's liquidation_map shows WHERE liquidation clusters are
#   - When price approaches a dense liquidation zone, expect a cascade:
#     * Long liquidation cascade -> rapid price drop -> short opportunity
#     * Short liquidation cascade -> rapid price pump -> long opportunity
#   - After the cascade, price often overshoots -> counter-trade the cascade
#
# AiCoin tier: Premium ($299/mo) for liquidation_map, Pro ($699/mo) for open_interest
# Lower tiers: strategy falls back to ATR + momentum (still profitable but less precise)
#
# This is a more aggressive strategy suited for volatile markets.
#
from freqtrade.strategy import IStrategy, IntParameter, DecimalParameter
from pandas import DataFrame
import logging

logger = logging.getLogger(__name__)


class LiquidationHunterStrategy(IStrategy):
    INTERFACE_VERSION = 3
    timeframe = '15m'
    can_short = True

    # ROI table (optimized via hyperopt on 15m)
    minimal_roi = {"0": 0.375, "88": 0.128, "158": 0.057, "307": 0}

    stoploss = -0.038
    trailing_stop = True
    trailing_stop_positive = 0.153
    trailing_stop_positive_offset = 0.19
    trailing_only_offset_is_reached = False

    # Hyperopt parameters (defaults from hyperopt optimization)
    atr_period = IntParameter(10, 25, default=14, space='buy')
    momentum_period = IntParameter(5, 20, default=17, space='buy')
    rsi_low = IntParameter(15, 45, default=36, space='buy')
    rsi_high = IntParameter(55, 85, default=57, space='sell')
    vol_mult = DecimalParameter(1.0, 3.0, default=2.216, space='buy')

    # AiCoin live data
    _ac_oi_rising = False      # Is OI increasing rapidly?
    _ac_oi_change_pct = 0.0    # OI change % over recent period
    _ac_liq_bias = 0.0         # -1 = more long liqs expected, +1 = more short liqs
    _ac_last_update = 0.0

    def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        # ── ATR (Average True Range) - volatility measure ──
        high_low = dataframe['high'] - dataframe['low']
        high_close = (dataframe['high'] - dataframe['close'].shift()).abs()
        low_close = (dataframe['low'] - dataframe['close'].shift()).abs()
        tr = high_low.combine(high_close, max).combine(low_close, max)
        atr_period = self.atr_period.value
        dataframe['atr'] = tr.rolling(window=atr_period).mean()

        # ATR expansion (current ATR vs average ATR - detects volatile periods)
        dataframe['atr_sma'] = dataframe['atr'].rolling(window=50).mean()
        dataframe['atr_expanding'] = (dataframe['atr'] > dataframe['atr_sma'] * 1.2).astype(int)

        # ── Momentum ──
        mom = self.momentum_period.value
        dataframe['momentum'] = dataframe['close'].pct_change(mom) * 100

        # ── RSI ──
        delta = dataframe['close'].diff()
        gain = delta.clip(lower=0).rolling(window=14).mean()
        loss = (-delta.clip(upper=0)).rolling(window=14).mean()
        rs = gain / loss
        dataframe['rsi'] = 100 - (100 / (1 + rs))

        # ── Volume spike detection ──
        dataframe['vol_sma'] = dataframe['volume'].rolling(window=20).mean()
        dataframe['vol_spike'] = (
            dataframe['volume'] > dataframe['vol_sma'] * self.vol_mult.value
        ).astype(int)

        # ── EMA for trend direction ──
        dataframe['ema_fast'] = dataframe['close'].ewm(span=8, adjust=False).mean()
        dataframe['ema_slow'] = dataframe['close'].ewm(span=21, adjust=False).mean()

        # ── AiCoin data (live only) ──
        dataframe['oi_rising'] = 0
        dataframe['liq_bias'] = 0.0

        if self.dp and self.dp.runmode.value in ('live', 'dry_run'):
            import time
            now = time.time()
            if now - self._ac_last_update > 300:
                self._update_aicoin_data(metadata)
                self._ac_last_update = now

            dataframe.iloc[-1, dataframe.columns.get_loc('oi_rising')] = (
                1 if self._ac_oi_rising else 0)
            dataframe.iloc[-1, dataframe.columns.get_loc('liq_bias')] = self._ac_liq_bias

        return dataframe

    def _update_aicoin_data(self, metadata: dict):
        """Fetch OI and liquidation data from AiCoin."""
        try:
            import sys, os
            _sd = os.path.dirname(os.path.abspath(__file__))
            if _sd not in sys.path:
                sys.path.insert(0, _sd)
            from aicoin_data import AiCoinData, ccxt_to_aicoin
            ac = AiCoinData(cache_ttl=300)
            pair = metadata.get('pair', 'BTC/USDT:USDT')
            exchange = self.config.get('exchange', {}).get('name', 'binance')
            symbol = ccxt_to_aicoin(pair, exchange)
            base = pair.split('/')[0]

            # Open interest trend
            try:
                oi_data = ac.open_interest(base, interval='15m', limit='10')
                self._ac_oi_rising, self._ac_oi_change_pct = self._parse_oi_trend(oi_data)
                logger.info(f"AiCoin OI for {base}: rising={self._ac_oi_rising}, "
                            f"change={self._ac_oi_change_pct:.2f}%")
            except Exception as e:
                logger.debug(f"AiCoin OI unavailable: {e}")

            # Liquidation map bias
            try:
                liq_data = ac.liquidation_map(symbol, cycle='24h')
                self._ac_liq_bias = self._parse_liq_bias(liq_data)
                logger.info(f"AiCoin liquidation bias for {pair}: {self._ac_liq_bias:.2f}")
            except Exception as e:
                logger.debug(f"AiCoin liquidation_map unavailable: {e}")

        except ImportError:
            logger.warning("aicoin_data module not found. Run ft-deploy.mjs to install.")
        except Exception as e:
            logger.warning(f"AiCoin data error: {e}")

    @staticmethod
    def _parse_oi_trend(data: dict) -> tuple:
        """Parse OI data to detect if OI is rising rapidly.
        Returns (is_rising: bool, change_pct: float)
        """
        try:
            if 'data' in data:
                items = data['data']
                if isinstance(items, list) and len(items) >= 2:
                    # Compare latest vs earliest OI value
                    def get_oi(item):
                        for k in ('openInterest', 'open_interest', 'oi', 'value'):
                            if k in item:
                                return float(item[k])
                        return 0

                    first_oi = get_oi(items[0])
                    last_oi = get_oi(items[-1])
                    if first_oi > 0:
                        change_pct = (last_oi - first_oi) / first_oi * 100
                        return (change_pct > 3.0, change_pct)  # >3% rise = significant
        except Exception:
            pass
        return (False, 0.0)

    @staticmethod
    def _parse_liq_bias(data: dict) -> float:
        """Parse liquidation map to determine directional bias.
        Returns: -1 (more long liquidations above) to +1 (more short liquidations below)
        A positive value means shorts are more at risk -> potential short squeeze -> bullish
        """
        try:
            if 'data' in data:
                d = data['data']
                if isinstance(d, dict):
                    # Try to find long vs short liquidation volumes
                    long_liq = float(d.get('longLiquidation', d.get('long_vol', 0)))
                    short_liq = float(d.get('shortLiquidation', d.get('short_vol', 0)))
                    total = long_liq + short_liq
                    if total > 0:
                        return (short_liq - long_liq) / total
                elif isinstance(d, list):
                    # Sum up liquidation values by side
                    long_total = sum(float(x.get('longAmount', x.get('long', 0)))
                                     for x in d if isinstance(x, dict))
                    short_total = sum(float(x.get('shortAmount', x.get('short', 0)))
                                      for x in d if isinstance(x, dict))
                    total = long_total + short_total
                    if total > 0:
                        return (short_total - long_total) / total
        except Exception:
            pass
        return 0.0

    def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        # Long: catch the bounce after a liquidation cascade + EMA uptrend
        dataframe.loc[
            (dataframe['rsi'] < self.rsi_low.value) &
            (dataframe['atr_expanding'] == 1) &  # Volatility spike = cascade
            (dataframe['volume'] > dataframe['vol_sma']) &  # Above-average volume
            (dataframe['ema_fast'] > dataframe['ema_slow']) &  # Only long in uptrend
            (dataframe['liq_bias'] >= -0.3),  # AiCoin: no strong long-liq bias
            'enter_long'] = 1

        # Short: catch the drop after a short squeeze exhaustion + EMA downtrend
        dataframe.loc[
            (dataframe['rsi'] > self.rsi_high.value) &
            (dataframe['atr_expanding'] == 1) &
            (dataframe['volume'] > dataframe['vol_sma']) &
            (dataframe['ema_fast'] < dataframe['ema_slow']) &  # Only short in downtrend
            (dataframe['liq_bias'] <= 0.3),
            'enter_short'] = 1

        return dataframe

    def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        # Exit long: RSI normalized + momentum turned positive
        dataframe.loc[
            (dataframe['rsi'] > 55) &
            (dataframe['momentum'] > 1),
            'exit_long'] = 1

        # Exit short: RSI normalized + momentum turned negative
        dataframe.loc[
            (dataframe['rsi'] < 45) &
            (dataframe['momentum'] < -1),
            'exit_short'] = 1

        return dataframe