TrendRegimeBtcCycleV1

from datetime import datetime

from pandas import DataFrame
import talib.abstract as ta

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


class TrendRegimeBtcCycleV1(IStrategy):
    """BTC-first long/short research strategy.

    This variant intentionally ignores non-BTC pairs. It is meant to test the
    thesis that BTC carries the major market regime while SOL adds execution
    noise. Keep it in dry-run/backtest until it has enough evidence.
    """

    INTERFACE_VERSION = 3

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

    position_adjustment_enable = False
    max_entry_position_adjustment = 0

    minimal_roi = {
        "0": 0.055,
        "240": 0.028,
        "720": 0.0,
    }

    stoploss = -0.05
    trailing_stop = True
    trailing_stop_positive = 0.018
    trailing_stop_positive_offset = 0.04
    trailing_only_offset_is_reached = True

    use_exit_signal = True
    exit_profit_only = False
    ignore_roi_if_entry_signal = False

    rsi_long_min = IntParameter(42, 56, default=48, space="buy", optimize=False)
    rsi_long_max = IntParameter(58, 72, default=64, space="buy", optimize=False)
    rsi_short_min = IntParameter(30, 44, default=36, space="buy", optimize=False)
    rsi_short_max = IntParameter(44, 58, default=50, space="buy", optimize=False)
    adx_min = IntParameter(18, 35, default=24, space="buy", optimize=False)
    adx_1h_min = IntParameter(16, 35, default=20, space="buy", optimize=False)
    adx_4h_min = IntParameter(12, 30, default=16, space="buy", optimize=False)
    volume_min_factor = DecimalParameter(0.2, 1.2, default=0.45, decimals=2, space="buy", optimize=False)
    ema_spread_min = DecimalParameter(0.001, 0.02, default=0.003, decimals=3, space="buy", optimize=False)
    atr_pct_min = DecimalParameter(0.001, 0.02, default=0.002, decimals=3, space="buy", optimize=False)
    atr_pct_max = DecimalParameter(0.015, 0.08, default=0.05, decimals=3, space="buy", optimize=False)
    pullback_atr_max = DecimalParameter(0.6, 2.5, default=1.5, decimals=1, space="buy", optimize=False)
    cycle_rsi_long_min = IntParameter(48, 62, default=52, space="buy", optimize=False)
    cycle_rsi_short_max = IntParameter(38, 52, default=48, space="buy", optimize=False)
    rsi_long_exit = IntParameter(68, 84, default=74, space="sell", optimize=False)
    rsi_short_exit = IntParameter(16, 34, default=28, 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": 3,
                "stop_duration_candles": 12,
                "required_profit": 0.0,
                "only_per_pair": False,
                "only_per_side": False,
            },
            {
                "method": "MaxDrawdown",
                "calculation_mode": "equity",
                "lookback_period_candles": 96,
                "trade_limit": 10,
                "stop_duration_candles": 24,
                "max_allowed_drawdown": 0.10,
            },
        ]

    @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_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["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(3)
        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["adx"] = ta.ADX(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(3)
        return dataframe

    def populate_indicators(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["adx"] = ta.ADX(dataframe, timeperiod=14)
        dataframe["plus_di"] = ta.PLUS_DI(dataframe, timeperiod=14)
        dataframe["minus_di"] = ta.MINUS_DI(dataframe, timeperiod=14)
        dataframe["atr"] = ta.ATR(dataframe, timeperiod=14)
        dataframe["volume_mean_20"] = dataframe["volume"].rolling(20, min_periods=20).mean()
        dataframe["atr_pct"] = dataframe["atr"] / dataframe["close"]
        dataframe["ema_spread"] = (dataframe["ema_50"] - dataframe["ema_200"]).abs() / dataframe["close"]
        dataframe["ema_50_slope"] = dataframe["ema_50"] - dataframe["ema_50"].shift(3)
        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.volume_min_factor.value)
        )

        volatility_ok = (
            (dataframe["atr_pct"] >= self.atr_pct_min.value)
            & (dataframe["atr_pct"] <= self.atr_pct_max.value)
            & (dataframe["ema_spread"] >= self.ema_spread_min.value)
        )

        long_pullback = (
            (dataframe["close"] > dataframe["ema_50"])
            & (
                (dataframe["close"].shift(1) <= dataframe["ema_50"].shift(1))
                | ((dataframe["low"] <= dataframe["ema_50"]) & (dataframe["close"] > dataframe["open"]))
            )
            & (((dataframe["close"] - dataframe["ema_50"]) / dataframe["atr"]) <= self.pullback_atr_max.value)
        )

        short_pullback = (
            (dataframe["close"] < dataframe["ema_50"])
            & (
                (dataframe["close"].shift(1) >= dataframe["ema_50"].shift(1))
                | ((dataframe["high"] >= dataframe["ema_50"]) & (dataframe["close"] < dataframe["open"]))
            )
            & (((dataframe["ema_50"] - dataframe["close"]) / dataframe["atr"]) <= self.pullback_atr_max.value)
        )

        long_conditions = (
            volume_ok
            & volatility_ok
            & (dataframe["close"] > dataframe["ema_200"])
            & (dataframe["ema_50"] > dataframe["ema_200"])
            & (dataframe["ema_50_slope"] > 0)
            & (dataframe["plus_di"] > dataframe["minus_di"])
            & (dataframe["close_1h"] > dataframe["ema_200_1h"])
            & (dataframe["ema_50_1h"] > dataframe["ema_200_1h"])
            & (dataframe["ema_50_slope_1h"] > 0)
            & (dataframe["plus_di_1h"] > dataframe["minus_di_1h"])
            & (dataframe["rsi_1h"] > 50)
            & (dataframe["close_4h"] > dataframe["ema_200_4h"])
            & (dataframe["ema_50_4h"] > dataframe["ema_200_4h"])
            & (dataframe["ema_50_slope_4h"] > 0)
            & (dataframe["plus_di_4h"] > dataframe["minus_di_4h"])
            & (dataframe["rsi_4h"] > self.cycle_rsi_long_min.value)
            & (dataframe["adx"] > self.adx_min.value)
            & (dataframe["adx_1h"] > self.adx_1h_min.value)
            & (dataframe["adx_4h"] > self.adx_4h_min.value)
            & (dataframe["rsi"] > self.rsi_long_min.value)
            & (dataframe["rsi"] < self.rsi_long_max.value)
            & long_pullback
        )

        short_conditions = (
            volume_ok
            & volatility_ok
            & (dataframe["close"] < dataframe["ema_200"])
            & (dataframe["ema_50"] < dataframe["ema_200"])
            & (dataframe["ema_50_slope"] < 0)
            & (dataframe["minus_di"] > dataframe["plus_di"])
            & (dataframe["close_1h"] < dataframe["ema_200_1h"])
            & (dataframe["ema_50_1h"] < dataframe["ema_200_1h"])
            & (dataframe["ema_50_slope_1h"] < 0)
            & (dataframe["minus_di_1h"] > dataframe["plus_di_1h"])
            & (dataframe["rsi_1h"] < 50)
            & (dataframe["close_4h"] < dataframe["ema_200_4h"])
            & (dataframe["ema_50_4h"] < dataframe["ema_200_4h"])
            & (dataframe["ema_50_slope_4h"] < 0)
            & (dataframe["minus_di_4h"] > dataframe["plus_di_4h"])
            & (dataframe["rsi_4h"] < self.cycle_rsi_short_max.value)
            & (dataframe["adx"] > self.adx_min.value)
            & (dataframe["adx_1h"] > self.adx_1h_min.value)
            & (dataframe["adx_4h"] > self.adx_4h_min.value)
            & (dataframe["rsi"] > self.rsi_short_min.value)
            & (dataframe["rsi"] < self.rsi_short_max.value)
            & short_pullback
        )

        dataframe.loc[long_conditions, ["enter_long", "enter_tag"]] = (1, "btc_cycle_long_pullback")
        dataframe.loc[short_conditions, ["enter_short", "enter_tag"]] = (1, "btc_cycle_short_pullback")
        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 trade.is_short:
            if current_profit > 0.025 and candle["rsi"] < self.rsi_short_exit.value:
                return "btc_short_rsi_profit_take"
            if current_profit > 0.012 and current_rate > candle["ema_50"]:
                return "btc_short_profit_protect_ema50"
            if (
                current_profit < -0.008
                and candle["close_1h"] > candle["ema_50_1h"]
                and candle["plus_di_1h"] > candle["minus_di_1h"]
            ):
                return "btc_short_1h_reversal"
            if candle["ema_50_1h"] > candle["ema_200_1h"] and current_rate > candle["ema_200"]:
                return "btc_short_regime_lost"
            return None

        if current_profit > 0.025 and candle["rsi"] > self.rsi_long_exit.value:
            return "btc_long_rsi_profit_take"
        if current_profit > 0.012 and current_rate < candle["ema_50"]:
            return "btc_long_profit_protect_ema50"
        if current_rate < candle["ema_50"]:
            return "btc_long_ema50_break"
        if (
            current_profit < -0.008
            and candle["close_1h"] < candle["ema_50_1h"]
            and candle["minus_di_1h"] > candle["plus_di_1h"]
        ):
            return "btc_long_1h_reversal"
        if candle["ema_50_1h"] < candle["ema_200_1h"] and current_rate < candle["ema_200"]:
            return "btc_long_regime_lost"
        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