BearMarketShortV2

from datetime import datetime

from pandas import DataFrame
import talib.abstract as ta

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


class BearMarketShortV2(IStrategy):
    """BTC-only research strategy for bear-market short setups.

    The strategy is deliberately not an always-on short bot. It looks for a
    bearish BTC regime, then waits for either a retest rejection or a controlled
    breakdown. It is meant for backtesting and dry-run research only.
    """

    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.055,
        "180": 0.028,
        "480": 0.0,
    }

    stoploss = -0.055
    trailing_stop = True
    trailing_stop_positive = 0.014
    trailing_stop_positive_offset = 0.034
    trailing_only_offset_is_reached = True

    use_exit_signal = True
    exit_profit_only = False
    ignore_roi_if_entry_signal = False

    adx_min = IntParameter(18, 35, default=23, space="buy", optimize=False)
    rsi_min = IntParameter(24, 40, default=31, space="buy", optimize=False)
    rsi_max = IntParameter(45, 62, default=55, space="buy", optimize=False)
    daily_rsi_min = IntParameter(24, 40, default=31, space="buy", optimize=False)
    daily_rsi_max = IntParameter(42, 58, default=52, space="buy", optimize=False)
    max_daily_ema50_atr_extension = DecimalParameter(1.0, 4.0, default=2.4, decimals=1, space="buy", optimize=False)
    max_15m_ema50_atr_extension = DecimalParameter(0.8, 3.0, default=1.7, decimals=1, space="buy", optimize=False)
    min_volume_factor = DecimalParameter(0.3, 1.5, default=0.65, decimals=2, space="buy", optimize=False)
    retest_buffer = DecimalParameter(0.000, 0.010, default=0.003, decimals=3, space="buy", optimize=False)
    ema50_exit_buffer = DecimalParameter(0.000, 0.012, default=0.004, decimals=3, space="sell", optimize=False)
    profit_take_rsi = IntParameter(18, 35, default=26, space="sell", optimize=False)

    @property
    def protections(self) -> list[dict]:
        return [
            {
                "method": "CooldownPeriod",
                "stop_duration_candles": 3,
            },
            {
                "method": "StoplossGuard",
                "lookback_period_candles": 72,
                "trade_limit": 2,
                "stop_duration_candles": 16,
                "required_profit": 0.0,
                "only_per_pair": False,
                "only_per_side": True,
            },
            {
                "method": "MaxDrawdown",
                "calculation_mode": "equity",
                "lookback_period_candles": 144,
                "trade_limit": 8,
                "stop_duration_candles": 24,
                "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_200"] = ta.EMA(dataframe, timeperiod=200)
        dataframe["rsi"] = ta.RSI(dataframe, timeperiod=14)
        dataframe["adx"] = ta.ADX(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(6)
        return dataframe

    @informative("4h")
    def populate_indicators_4h(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["adx"] = ta.ADX(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_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["adx"] = ta.ADX(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["atr_pct"] = dataframe["atr"] / dataframe["close"]
        dataframe["dist_ema200_pct"] = (dataframe["ema_200"] - dataframe["close"]) / dataframe["ema_200"]
        dataframe["ema_20_slope_atr"] = (dataframe["ema_20"] - dataframe["ema_20"].shift(5)) / dataframe["atr"]
        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["adx"] = ta.ADX(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_40"] = dataframe["low"].rolling(40, min_periods=40).min().shift(1)
        dataframe["range_atr"] = (dataframe["high"] - dataframe["low"]) / dataframe["atr"]
        dataframe["ema50_extension_atr"] = (dataframe["ema_50"] - dataframe["close"]) / dataframe["atr"]
        dataframe["ema_50_slope"] = dataframe["ema_50"] - dataframe["ema_50"].shift(6)
        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

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

        daily_extension = (dataframe["ema_50_1d"] - dataframe["close_1d"]) / dataframe["atr_1d"]
        daily_risk_off = (
            (
                (dataframe["close_1d"] < dataframe["ema_50_1d"])
                & (dataframe["ema_50_slope_1d"] < 0)
            )
            | (
                (dataframe["close_1d"] < dataframe["ema_200_1d"])
                & (dataframe["minus_di_1d"] > dataframe["plus_di_1d"])
            )
        )

        higher_tf_bear = (
            daily_risk_off
            & (daily_extension < self.max_daily_ema50_atr_extension.value)
            & (dataframe["rsi_1d"] > self.daily_rsi_min.value)
            & (dataframe["rsi_1d"] < self.daily_rsi_max.value)
            & (dataframe["close_4h"] < dataframe["ema_50_4h"])
            & (dataframe["ema_50_slope_4h"] < 0)
            & (dataframe["minus_di_4h"] > dataframe["plus_di_4h"])
            & (dataframe["close_1h"] < dataframe["ema_200_1h"])
            & (dataframe["ema_50_slope_1h"] < 0)
            & (dataframe["minus_di_1h"] > dataframe["plus_di_1h"])
        )

        not_late = (
            (dataframe["ema50_extension_atr"] < self.max_15m_ema50_atr_extension.value)
            & (dataframe["rsi"] > self.rsi_min.value)
            & (dataframe["rsi"] < self.rsi_max.value)
            & (dataframe["range_atr"] < 3.5)
        )

        base_short = (
            volume_ok
            & higher_tf_bear
            & not_late
            & (dataframe["close"] < dataframe["ema_200"])
            & (dataframe["ema_50"] < dataframe["ema_200"])
            & (dataframe["ema_50_slope"] < 0)
            & (dataframe["minus_di"] > dataframe["plus_di"])
            & (dataframe["adx"] > self.adx_min.value)
        )

        retest_rejection = (
            base_short
            & (dataframe["high"] >= dataframe["ema_50"] * (1 - self.retest_buffer.value))
            & (dataframe["close"] < dataframe["ema_20"])
            & (dataframe["close"] < dataframe["open"])
        )

        breakdown = (
            base_short
            & (dataframe["close"] < dataframe["donchian_low_40"])
            & (dataframe["close"] < dataframe["open"])
            & (dataframe["adx_1h"] > self.adx_min.value)
            & (dataframe["rsi_1h"] > self.rsi_min.value)
        )

        dataframe.loc[retest_rejection, ["enter_short", "enter_tag"]] = (1, "bear_v2_retest_rejection")
        dataframe.loc[breakdown, ["enter_short", "enter_tag"]] = (1, "bear_v2_breakdown")
        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 dataframe.empty:
            return None

        candle = dataframe.iloc[-1]
        if not trade.is_short:
            return None

        if current_profit > 0.025 and candle["rsi"] < self.profit_take_rsi.value:
            return "bear_v2_rsi_profit_take"

        if current_profit > 0.015 and current_rate > candle["ema_20"]:
            return "bear_v2_profit_ema20_reclaim"

        if current_rate > candle["ema_50"] * (1 + self.ema50_exit_buffer.value):
            return "bear_v2_ema50_reclaim"

        if candle["close_1h"] > candle["ema_50_1h"] and current_profit < 0.01:
            return "bear_v2_1h_reclaim"

        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