BtcBearMarketShortV3

from datetime import datetime

from pandas import DataFrame
import talib.abstract as ta

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


class BtcBearMarketShortV3(IStrategy):
    """BTC short-only bear-market research strategy.

    Two setups share the same priority: avoid trading unless higher timeframes
    are already bearish. This is a research strategy, not the active dry-run bot.
    """

    INTERFACE_VERSION = 3

    can_short = True
    timeframe = "15m"
    startup_candle_count = 420
    process_only_new_candles = True

    position_adjustment_enable = False
    max_entry_position_adjustment = 0

    minimal_roi = {
        "0": 0.050,
        "180": 0.026,
        "540": 0.0,
    }

    stoploss = -0.060
    trailing_stop = True
    trailing_stop_positive = 0.014
    trailing_stop_positive_offset = 0.036
    trailing_only_offset_is_reached = True

    use_exit_signal = True
    exit_profit_only = False
    ignore_roi_if_entry_signal = False

    volume_breakdown_factor = DecimalParameter(0.8, 1.8, default=1.20, decimals=2, space="buy", optimize=False)
    max_candle_range_atr = DecimalParameter(1.5, 3.5, default=2.5, decimals=1, space="buy", optimize=False)
    atr_pct_min = DecimalParameter(0.001, 0.010, default=0.0025, decimals=4, space="buy", optimize=False)
    rally_rsi_1h_min = IntParameter(40, 52, default=45, space="buy", optimize=False)
    rally_rsi_1h_max = IntParameter(55, 70, default=65, space="buy", optimize=False)
    min_1h_rsi_not_oversold = IntParameter(24, 38, default=30, space="buy", optimize=False)
    min_4h_rsi_not_oversold = IntParameter(22, 38, default=28, space="buy", optimize=False)
    wick_body_factor = DecimalParameter(0.8, 2.0, default=1.2, decimals=1, space="buy", optimize=False)
    atr_stop_mult = DecimalParameter(1.0, 2.5, default=1.5, decimals=1, space="sell", optimize=False)
    profit_take_rsi = IntParameter(18, 32, default=25, space="sell", optimize=False)

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

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

    @informative("1h")
    def populate_indicators_1h(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        dataframe["ema_20"] = ta.EMA(dataframe, timeperiod=20)
        dataframe["ema_50"] = ta.EMA(dataframe, timeperiod=50)
        dataframe["ema_100"] = ta.EMA(dataframe, timeperiod=100)
        dataframe["ema_200"] = ta.EMA(dataframe, timeperiod=200)
        dataframe["rsi"] = ta.RSI(dataframe, timeperiod=14)
        dataframe["atr"] = ta.ATR(dataframe, timeperiod=14)
        dataframe["plus_di"] = ta.PLUS_DI(dataframe, timeperiod=14)
        dataframe["minus_di"] = ta.MINUS_DI(dataframe, timeperiod=14)
        return dataframe

    @informative("4h")
    def populate_indicators_4h(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        dataframe["ema_50"] = ta.EMA(dataframe, timeperiod=50)
        dataframe["ema_200"] = ta.EMA(dataframe, timeperiod=200)
        dataframe["rsi"] = ta.RSI(dataframe, timeperiod=14)
        dataframe["atr"] = ta.ATR(dataframe, timeperiod=14)
        dataframe["plus_di"] = ta.PLUS_DI(dataframe, timeperiod=14)
        dataframe["minus_di"] = ta.MINUS_DI(dataframe, timeperiod=14)
        dataframe["ema_50_slope"] = dataframe["ema_50"] - dataframe["ema_50"].shift(4)
        return dataframe

    @informative("1d")
    def populate_indicators_1d(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        dataframe["ema_50"] = ta.EMA(dataframe, timeperiod=50)
        dataframe["ema_200"] = ta.EMA(dataframe, timeperiod=200)
        dataframe["rsi"] = ta.RSI(dataframe, timeperiod=14)
        dataframe["atr"] = ta.ATR(dataframe, timeperiod=14)
        dataframe["plus_di"] = ta.PLUS_DI(dataframe, timeperiod=14)
        dataframe["minus_di"] = ta.MINUS_DI(dataframe, timeperiod=14)
        dataframe["ema_50_slope"] = dataframe["ema_50"] - dataframe["ema_50"].shift(5)
        return dataframe

    def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        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["atr"] = ta.ATR(dataframe, timeperiod=14)
        dataframe["plus_di"] = ta.PLUS_DI(dataframe, timeperiod=14)
        dataframe["minus_di"] = ta.MINUS_DI(dataframe, timeperiod=14)
        dataframe["volume_mean_20"] = dataframe["volume"].rolling(20, min_periods=20).mean()
        dataframe["donchian_low_20"] = dataframe["low"].rolling(20, min_periods=20).min().shift(1)

        bollinger = ta.BBANDS(dataframe, timeperiod=20, nbdevup=2.0, nbdevdn=2.0)
        dataframe["bb_upper"] = bollinger["upperband"]
        dataframe["bb_middle"] = bollinger["middleband"]
        dataframe["bb_lower"] = bollinger["lowerband"]

        dataframe["atr_pct"] = dataframe["atr"] / dataframe["close"]
        dataframe["range_atr"] = (dataframe["high"] - dataframe["low"]) / dataframe["atr"]
        dataframe["body"] = (dataframe["close"] - dataframe["open"]).abs()
        dataframe["upper_wick"] = dataframe["high"] - dataframe[["open", "close"]].max(axis=1)
        dataframe["bearish_engulfing"] = (
            (dataframe["close"] < dataframe["open"])
            & (dataframe["close"].shift(1) > dataframe["open"].shift(1))
            & (dataframe["open"] >= dataframe["close"].shift(1))
            & (dataframe["close"] <= dataframe["open"].shift(1))
        )
        return dataframe

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

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

        daily_bear = (
            (dataframe["close_1d"] < dataframe["ema_200_1d"])
            | (dataframe["ema_50_1d"] < dataframe["ema_200_1d"])
        )

        four_h_bear = (
            (dataframe["ema_50_4h"] < dataframe["ema_200_4h"])
            & (dataframe["close_4h"] < dataframe["ema_50_4h"])
            & (dataframe["rsi_4h"] > self.min_4h_rsi_not_oversold.value)
        )

        one_h_breakdown_ready = (
            (dataframe["ema_20_1h"] < dataframe["ema_50_1h"])
            & (dataframe["rsi_1h"] < 50)
            & (dataframe["rsi_1h"] > self.min_1h_rsi_not_oversold.value)
        )

        not_panic_candle = (
            (dataframe["atr"] > 0)
            & (dataframe["atr_pct"] >= self.atr_pct_min.value)
            & (dataframe["range_atr"] <= self.max_candle_range_atr.value)
        )

        volume_breakdown_ok = (
            (dataframe["volume"] > 0)
            & (dataframe["volume_mean_20"] > 0)
            & (dataframe["volume"] > dataframe["volume_mean_20"] * self.volume_breakdown_factor.value)
        )

        breakdown = (
            daily_bear
            & four_h_bear
            & one_h_breakdown_ready
            & not_panic_candle
            & volume_breakdown_ok
            & (dataframe["close"] < dataframe["donchian_low_20"])
            & (dataframe["close"] < dataframe["open"])
        )

        rally_regime = (
            (dataframe["close_4h"] < dataframe["ema_200_4h"])
            & (dataframe["ema_50_4h"] < dataframe["ema_200_4h"])
            & (dataframe["close_1h"] < dataframe["ema_100_1h"])
            & (dataframe["rsi_1h"] >= self.rally_rsi_1h_min.value)
            & (dataframe["rsi_1h"] <= self.rally_rsi_1h_max.value)
        )

        bb_touch_recent = (
            (dataframe["high"] >= dataframe["bb_upper"])
            | (dataframe["high"].shift(1) >= dataframe["bb_upper"].shift(1))
        )
        ema20_reclaim_failed = (
            (dataframe["close"] < dataframe["ema_20"])
            & (
                (dataframe["close"].shift(1) >= dataframe["ema_20"].shift(1))
                | bb_touch_recent
            )
        )
        rejection_candle = (
            (dataframe["upper_wick"] > dataframe["body"].clip(lower=dataframe["atr"] * 0.05) * self.wick_body_factor.value)
            | dataframe["bearish_engulfing"]
        )

        bear_rally_rejection = (
            rally_regime
            & not_panic_candle
            & bb_touch_recent
            & ema20_reclaim_failed
            & rejection_candle
        )

        dataframe.loc[breakdown, ["enter_short", "enter_tag"]] = (1, "v3_breakdown_continuation")
        dataframe.loc[bear_rally_rejection, ["enter_short", "enter_tag"]] = (1, "v3_bear_rally_rejection")
        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,
    ) -> str | bool | None:
        if not self.dp:
            return None

        dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)
        if len(dataframe) < 2 or not trade.is_short:
            return None

        candle = dataframe.iloc[-1]
        previous = dataframe.iloc[-2]

        atr_stop_rate = trade.open_rate + candle["atr"] * self.atr_stop_mult.value
        if current_rate > atr_stop_rate:
            return "v3_atr_invalidation"

        if current_profit > 0 and current_rate > candle["ema_20"]:
            return "v3_profit_ema20_reclaim"

        if candle["close_1h"] > candle["ema_50_1h"]:
            return "v3_1h_ema50_reclaim"

        momentum_slowing = (candle["rsi"] > previous["rsi"]) or (candle["close"] > previous["close"])
        if current_profit > 0.012 and candle["rsi"] < self.profit_take_rsi.value and momentum_slowing:
            return "v3_oversold_momentum_slow"

        return None

    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 1.0


class BtcBearMarketShortV3RallyOnly(BtcBearMarketShortV3):
    """Research variant: keep only the bear-rally rejection setup.

    The full V3 backtest showed that breakdown continuation was destructive.
    This variant isolates the more promising setup instead of tuning both at
    once.
    """

    def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        dataframe = super().populate_entry_trend(dataframe, metadata)
        dataframe.loc[dataframe["enter_tag"] != "v3_bear_rally_rejection", "enter_short"] = 0
        return dataframe