SolLargeRedRebound1hV1

from __future__ import annotations

from datetime import datetime

from pandas import DataFrame
import talib.abstract as ta

from freqtrade.strategy import DecimalParameter, IStrategy, IntParameter, stoploss_from_absolute


class SolLargeRedRebound1hV1(IStrategy):
    """Research-only SOL rebound after large red 1h candle.

    Hypothesis from edge lab:
    - On recent SOL data, large red 1h candles often mean-reverted over the
      next few hours.
    - This is a tactical long-only module, not a cycle strategy.

    Do not use live without a separate dry-run and kill-switch review.
    """

    INTERFACE_VERSION = 3

    can_short = False
    timeframe = "1h"
    startup_candle_count = 240
    process_only_new_candles = True

    position_adjustment_enable = False
    max_entry_position_adjustment = 0

    minimal_roi = {"0": 0.99}
    stoploss = -0.06
    trailing_stop = False
    use_custom_stoploss = False

    use_exit_signal = True
    exit_profit_only = False
    ignore_roi_if_entry_signal = False

    take_profit_pct: float | None = None
    body_recovery_target: float | None = None
    range_recovery_target: float | None = None
    dynamic_tp_min_pct = 0.015
    dynamic_tp_max_pct = 0.08
    break_even_after_pct: float | None = None
    break_even_stop_pct = 0.001
    trail_after_pct: float | None = None
    trail_distance_pct: float | None = None
    leverage_value = 1.0

    min_range_atr = DecimalParameter(1.60, 2.60, default=2.00, decimals=2, space="buy", optimize=False)
    max_range_atr = DecimalParameter(2.60, 4.00, default=3.80, decimals=2, space="buy", optimize=False)
    min_volume_ratio = DecimalParameter(0.50, 1.50, default=0.80, decimals=2, space="buy", optimize=False)
    max_btc_drop_6h = DecimalParameter(-0.12, -0.03, default=-0.07, decimals=3, space="buy", optimize=False)
    time_exit_candles = IntParameter(4, 24, default=8, space="sell", optimize=False)
    panic_exit_btc_drop_4h = DecimalParameter(-0.12, -0.03, default=-0.08, decimals=3, space="sell", optimize=False)

    @property
    def protections(self) -> list[dict]:
        return [
            {"method": "CooldownPeriod", "stop_duration_candles": 2},
            {
                "method": "StoplossGuard",
                "lookback_period_candles": 48,
                "trade_limit": 2,
                "stop_duration_candles": 24,
                "required_profit": 0.0,
                "only_per_pair": True,
                "only_per_side": False,
            },
            {
                "method": "MaxDrawdown",
                "calculation_mode": "equity",
                "lookback_period_candles": 168,
                "trade_limit": 10,
                "stop_duration_candles": 24,
                "max_allowed_drawdown": 0.08,
            },
        ]

    @staticmethod
    def _is_sol_pair(pair: str) -> bool:
        return pair.upper().startswith("SOL/")

    def _btc_reference_pair(self) -> str:
        stake_currency = str(self.config.get("stake_currency", "USDT")).upper()
        exchange_name = str(self.config.get("exchange", {}).get("name", "")).lower()
        if stake_currency == "USDC" or exchange_name == "hyperliquid":
            return "BTC/USDC:USDC"
        return "BTC/USDT:USDT"

    def informative_pairs(self) -> list[tuple[str, str]]:
        return [(self._btc_reference_pair(), self.timeframe)]

    def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        dataframe["atr"] = ta.ATR(dataframe, timeperiod=14)
        dataframe["ema_20"] = ta.EMA(dataframe, timeperiod=20)
        dataframe["ema_50"] = ta.EMA(dataframe, timeperiod=50)
        dataframe["ema_200"] = ta.EMA(dataframe, timeperiod=200)
        dataframe["rsi"] = ta.RSI(dataframe, timeperiod=14)
        dataframe["volume_mean_24"] = dataframe["volume"].rolling(24, min_periods=24).mean()
        dataframe["volume_ratio"] = dataframe["volume"] / dataframe["volume_mean_24"]
        dataframe["range_atr"] = (dataframe["high"] - dataframe["low"]) / dataframe["atr"]
        dataframe["red_candle"] = dataframe["close"] < dataframe["open"]
        dataframe["ret_6"] = dataframe["close"].pct_change(6)
        dataframe["ret_24"] = dataframe["close"].pct_change(24)

        if self.dp:
            btc = self.dp.get_pair_dataframe(pair=self._btc_reference_pair(), timeframe=self.timeframe)
            if not btc.empty:
                btc = btc[["date", "close"]].copy()
                btc["btc_ret_6"] = btc["close"].pct_change(6)
                btc = btc[["date", "btc_ret_6"]]
                dataframe = dataframe.merge(btc, on="date", how="left")
        if "btc_ret_6" not in dataframe.columns:
            dataframe["btc_ret_6"] = 0.0

        return dataframe

    def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        dataframe["enter_long"] = 0

        if not self._is_sol_pair(metadata["pair"]):
            return dataframe

        valid = (
            (dataframe["volume"] > 0)
            & dataframe["atr"].notna()
            & (dataframe["volume_mean_24"] > 0)
            & dataframe["btc_ret_6"].notna()
        )

        large_red_rebound = (
            valid
            & dataframe["red_candle"]
            & (dataframe["range_atr"] >= self.min_range_atr.value)
            & (dataframe["range_atr"] <= self.max_range_atr.value)
            & (dataframe["volume_ratio"] >= self.min_volume_ratio.value)
            & (dataframe["btc_ret_6"] > self.max_btc_drop_6h.value)
        )

        dataframe.loc[large_red_rebound, ["enter_long", "enter_tag"]] = (1, "large_red_rebound")
        return dataframe

    def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        dataframe["exit_long"] = 0
        return dataframe

    @staticmethod
    def _signal_candle(dataframe: DataFrame, trade):
        frame = dataframe.loc[dataframe["date"] < trade.open_date_utc]
        if frame.empty:
            frame = dataframe.loc[dataframe["date"] <= trade.open_date_utc]
        if frame.empty:
            return None
        return frame.iloc[-1]

    def _body_recovery_profit_target(self, dataframe: DataFrame, trade) -> float | None:
        if self.body_recovery_target is None:
            return None

        signal = self._signal_candle(dataframe, trade)
        if signal is None:
            return None

        signal_open = float(signal.get("open", 0.0))
        signal_close = float(signal.get("close", 0.0))
        if signal_open <= signal_close or signal_close <= 0:
            return None

        recovery_price = signal_close + ((signal_open - signal_close) * self.body_recovery_target)
        raw_profit_target = (recovery_price / float(trade.open_rate)) - 1.0
        return min(max(raw_profit_target, self.dynamic_tp_min_pct), self.dynamic_tp_max_pct)

    def _range_recovery_profit_target(self, dataframe: DataFrame, trade) -> float | None:
        if self.range_recovery_target is None:
            return None

        signal = self._signal_candle(dataframe, trade)
        if signal is None:
            return None

        signal_high = float(signal.get("high", 0.0))
        signal_low = float(signal.get("low", 0.0))
        signal_close = float(signal.get("close", 0.0))
        if signal_high <= signal_low or signal_close <= 0:
            return None

        candle_range_pct = (signal_high - signal_low) / signal_close
        raw_profit_target = candle_range_pct * self.range_recovery_target
        return min(max(raw_profit_target, self.dynamic_tp_min_pct), self.dynamic_tp_max_pct)

    def custom_exit(
        self,
        pair: str,
        trade,
        current_time: datetime,
        current_rate: float,
        current_profit: float,
        **kwargs,
    ) -> str | bool | None:
        elapsed_candles = int((current_time - trade.open_date_utc).total_seconds() // 3600)

        if self.take_profit_pct is not None and current_profit >= self.take_profit_pct:
            return f"take_profit_{self.take_profit_pct:.0%}"

        if self.dp:
            dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)
            if not dataframe.empty:
                dynamic_target = self._range_recovery_profit_target(dataframe, trade)
                if dynamic_target is not None and current_profit >= dynamic_target:
                    return f"range_recovery_{self.range_recovery_target:.0%}"

                dynamic_target = self._body_recovery_profit_target(dataframe, trade)
                if dynamic_target is not None and current_profit >= dynamic_target:
                    return f"body_recovery_{self.body_recovery_target:.0%}"

                candle = dataframe.iloc[-1]
                if elapsed_candles >= 2 and float(candle.get("btc_ret_6", 0.0)) <= self.panic_exit_btc_drop_4h.value:
                    return "btc_panic_exit"

        if elapsed_candles >= self.time_exit_candles.value:
            return f"time_exit_{self.time_exit_candles.value}h"

        return None

    def custom_stoploss(
        self,
        pair: str,
        trade,
        current_time: datetime,
        current_rate: float,
        current_profit: float,
        after_fill: bool = False,
        **kwargs,
    ) -> float | None:
        if trade.is_short:
            return None

        stop_level: float | None = None

        if self.break_even_after_pct is not None and current_profit >= self.break_even_after_pct:
            stop_level = float(trade.open_rate) * (1 + self.break_even_stop_pct)

        if (
            self.trail_after_pct is not None
            and self.trail_distance_pct is not None
            and current_profit >= self.trail_after_pct
        ):
            trailing_level = current_rate * (1 - self.trail_distance_pct)
            stop_level = max(stop_level or 0.0, trailing_level)

        if stop_level is None:
            return None

        if stop_level >= current_rate:
            return 0.001

        return stoploss_from_absolute(
            stop_level,
            current_rate,
            is_short=False,
            leverage=getattr(trade, "leverage", 1.0) or 1.0,
        )

    def leverage(
        self,
        pair: str,
        current_time: datetime,
        current_rate: float,
        proposed_leverage: float,
        max_leverage: float,
        entry_tag: str | None,
        side: str,
        **kwargs,
    ) -> float:
        return min(self.leverage_value, max_leverage)


class SolLargeRedRebound1hV1Exit8h(SolLargeRedRebound1hV1):
    time_exit_candles = IntParameter(4, 24, default=8, space="sell", optimize=False)


class SolLargeRedRebound1hV1Exit24h(SolLargeRedRebound1hV1):
    time_exit_candles = IntParameter(4, 24, default=24, space="sell", optimize=False)


class SolLargeRedRebound1hV1Exit24hLev2(SolLargeRedRebound1hV1Exit24h):
    leverage_value = 2.0


class SolLargeRedRebound1hV1Exit24hLev3(SolLargeRedRebound1hV1Exit24h):
    leverage_value = 3.0


class SolLargeRedRebound1hV1Exit24hTp2(SolLargeRedRebound1hV1Exit24h):
    take_profit_pct = 0.02


class SolLargeRedRebound1hV1Exit24hTp3(SolLargeRedRebound1hV1Exit24h):
    take_profit_pct = 0.03


class SolLargeRedRebound1hV1Exit24hTp4(SolLargeRedRebound1hV1Exit24h):
    take_profit_pct = 0.04


class SolLargeRedRebound1hV1Exit24hBody35(SolLargeRedRebound1hV1Exit24h):
    body_recovery_target = 0.35


class SolLargeRedRebound1hV1Exit24hBody50(SolLargeRedRebound1hV1Exit24h):
    body_recovery_target = 0.50


class SolLargeRedRebound1hV1Exit24hBody65(SolLargeRedRebound1hV1Exit24h):
    body_recovery_target = 0.65


class SolLargeRedRebound1hV1Exit24hRange35(SolLargeRedRebound1hV1Exit24h):
    range_recovery_target = 0.35


class SolLargeRedRebound1hV1Exit24hRange50(SolLargeRedRebound1hV1Exit24h):
    range_recovery_target = 0.50


class SolLargeRedRebound1hV1Exit24hRange65(SolLargeRedRebound1hV1Exit24h):
    range_recovery_target = 0.65


class SolLargeRedRebound1hV1Exit24hBeOnly(SolLargeRedRebound1hV1Exit24h):
    use_custom_stoploss = True
    break_even_after_pct = 0.015
    break_even_stop_pct = 0.001


class SolLargeRedRebound1hV1Exit24hTp4Be(SolLargeRedRebound1hV1Exit24h):
    take_profit_pct = 0.04
    use_custom_stoploss = True
    break_even_after_pct = 0.015
    break_even_stop_pct = 0.001


class SolLargeRedRebound1hV1Exit24hBeTrail(SolLargeRedRebound1hV1Exit24h):
    use_custom_stoploss = True
    break_even_after_pct = 0.015
    break_even_stop_pct = 0.001
    trail_after_pct = 0.035
    trail_distance_pct = 0.025