RAPv1Strategy

# pragma pylint: disable=missing-docstring, invalid-name
from __future__ import annotations

from datetime import datetime, timedelta
from typing import Any

import pandas as pd
import talib.abstract as ta
from pandas import DataFrame
from technical import qtpylib

from freqtrade.persistence import Trade
from freqtrade.strategy import DecimalParameter, IStrategy, merge_informative_pair, stoploss_from_open


class RAPv1Strategy(IStrategy):
    """
    RAP-v1 (Regime-Adaptive Pullback), spot long-only.

    Core logic:
    - 1h regime filter: trade only when market is above rising EMA200.
    - 5m entry: pullback to EMA20 inside local trend + RSI recovery.
    - Risk: ATR-based custom stop with R-multiple profit protection.
    """

    INTERFACE_VERSION = 3
    can_short: bool = False

    timeframe = "5m"
    informative_timeframe = "1h"
    process_only_new_candles = True
    startup_candle_count: int = 2500

    use_exit_signal = True
    exit_profit_only = False
    ignore_roi_if_entry_signal = False
    use_custom_stoploss = True

    # Conservative defaults. Runtime config can override these.
    minimal_roi = {"120": 0.0, "60": 0.01, "0": 0.03}
    stoploss = -0.10
    trailing_stop = False

    order_types = {
        "entry": "limit",
        "exit": "limit",
        "emergency_exit": "market",
        "force_exit": "market",
        "force_entry": "market",
        "stoploss": "market",
        "stoploss_on_exchange": False,
    }

    order_time_in_force = {"entry": "GTC", "exit": "GTC"}

    # Entry control knobs (hyperopt-ready).
    pullback_tolerance = DecimalParameter(
        0.0, 0.010, decimals=4, default=0.0062, space="buy", optimize=True
    )
    min_volume_ratio = DecimalParameter(
        0.4, 1.0, decimals=2, default=0.62, space="buy", optimize=True
    )
    max_range_proxy = DecimalParameter(
        0.005, 0.025, decimals=3, default=0.018, space="buy", optimize=True
    )
    max_1h_atr_ratio = DecimalParameter(
        0.015, 0.060, decimals=3, default=0.050, space="buy", optimize=True
    )
    min_rsi_reclaim = DecimalParameter(40, 55, decimals=0, default=49, space="buy", optimize=True)
    min_trend_buffer = DecimalParameter(
        0.0, 0.010, decimals=4, default=0.0004, space="buy", optimize=True
    )
    min_ema50_slope = DecimalParameter(
        0.0, 0.005, decimals=4, default=0.0001, space="buy", optimize=True
    )
    max_5m_atr_pct = DecimalParameter(
        0.002, 0.020, decimals=3, default=0.010, space="buy", optimize=True
    )
    min_5m_atr_pct = DecimalParameter(
        0.0005, 0.006, decimals=4, default=0.0012, space="buy", optimize=True
    )
    test_mode_rsi_buffer = DecimalParameter(
        2.0, 10.0, decimals=1, default=4.0, space="buy", optimize=True
    )
    test_mode_ema20_buffer = DecimalParameter(
        0.0005, 0.0040, decimals=4, default=0.0010, space="buy", optimize=True
    )
    risk_atr_mult = DecimalParameter(
        1.2, 2.0, decimals=2, default=1.70, space="sell", optimize=True
    )
    trend_fail_risk_mult = DecimalParameter(
        0.12, 0.55, decimals=2, default=0.22, space="sell", optimize=True
    )
    stagnation_risk_mult = DecimalParameter(
        0.08, 0.35, decimals=2, default=0.14, space="sell", optimize=True
    )
    weak_momentum_rsi = DecimalParameter(
        44, 54, decimals=0, default=50, space="sell", optimize=True
    )
    # Test mode intentionally allows continuation entries in strong trend momentum.
    test_mode_relaxed_entry: bool = True

    @property
    def protections(self) -> list[dict[str, Any]]:
        return [
            {"method": "CooldownPeriod", "stop_duration_candles": 3},
            {
                "method": "StoplossGuard",
                "lookback_period_candles": 24,
                "trade_limit": 3,
                "stop_duration_candles": 5,
                "only_per_pair": True,
            },
            {
                "method": "MaxDrawdown",
                "lookback_period_candles": 48,
                "trade_limit": 20,
                "stop_duration_candles": 10,
                "max_allowed_drawdown": 0.08,
            },
            {
                "method": "LowProfitPairs",
                "lookback_period_candles": 72,
                "trade_limit": 4,
                "stop_duration_candles": 12,
                "required_profit": 0.003,
            },
        ]

    def informative_pairs(self) -> list[tuple[str, str]]:
        if not self.dp:
            return []
        return [(pair, self.informative_timeframe) for pair in self.dp.current_whitelist()]

    def populate_indicators(self, dataframe: DataFrame, metadata: dict[str, Any]) -> DataFrame:
        dataframe["ema20"] = ta.EMA(dataframe, timeperiod=20)
        dataframe["ema50"] = ta.EMA(dataframe, timeperiod=50)
        dataframe["rsi"] = ta.RSI(dataframe, timeperiod=14)
        dataframe["atr"] = ta.ATR(dataframe, timeperiod=14)
        dataframe["atr_pct"] = dataframe["atr"] / dataframe["close"]
        dataframe["ema50_slope"] = dataframe["ema50"].pct_change(periods=3)
        dataframe["vol_sma20"] = dataframe["volume"].rolling(20).mean()
        # Proxy for spread/instability in backtest-safe form.
        dataframe["range_proxy"] = (dataframe["high"] - dataframe["low"]) / dataframe["close"]

        if self.dp:
            inf = self.dp.get_pair_dataframe(pair=metadata["pair"], timeframe=self.informative_timeframe)
            if inf is not None and not inf.empty:
                inf["ema200"] = ta.EMA(inf, timeperiod=200)
                inf["atr_1h"] = ta.ATR(inf, timeperiod=14)
                inf["atr_ratio"] = inf["atr_1h"] / inf["close"]
                inf["regime_up"] = (inf["close"] > inf["ema200"]) & (
                    inf["ema200"] > inf["ema200"].shift(3)
                )
                dataframe = merge_informative_pair(
                    dataframe,
                    inf[["date", "close", "ema200", "atr_ratio", "regime_up"]],
                    self.timeframe,
                    self.informative_timeframe,
                    ffill=True,
                )

        if "regime_up_1h" not in dataframe.columns:
            dataframe["regime_up_1h"] = False
        if "atr_ratio_1h" not in dataframe.columns:
            dataframe["atr_ratio_1h"] = 999.0

        dataframe["volatility_ok_1h"] = dataframe["atr_ratio_1h"] <= float(self.max_1h_atr_ratio.value)
        dataframe["regime_up_1h"] = dataframe["regime_up_1h"].fillna(False)
        dataframe["ema50_slope"] = dataframe["ema50_slope"].fillna(0.0)
        dataframe["atr_pct"] = dataframe["atr_pct"].fillna(0.0)
        return dataframe

    def populate_entry_trend(self, dataframe: DataFrame, metadata: dict[str, Any]) -> DataFrame:
        rsi_reclaim = qtpylib.crossed_above(dataframe["rsi"], float(self.min_rsi_reclaim.value))
        ema20_reclaim = qtpylib.crossed_above(dataframe["close"], dataframe["ema20"])
        momentum_continuation = (
            dataframe["rsi"] >= (float(self.min_rsi_reclaim.value) + float(self.test_mode_rsi_buffer.value))
        ) & (
            dataframe["close"] >= (dataframe["ema20"] * (1.0 + float(self.test_mode_ema20_buffer.value)))
        )
        trigger_ok = (rsi_reclaim | ema20_reclaim)
        if self.test_mode_relaxed_entry:
            trigger_ok = trigger_ok | momentum_continuation

        entry_conditions = [
            dataframe["volume"] > 0,
            dataframe["regime_up_1h"],
            dataframe["volatility_ok_1h"],
            dataframe["close"] >= (dataframe["ema50"] * (1.0 + float(self.min_trend_buffer.value))),
            dataframe["ema50_slope"] >= float(self.min_ema50_slope.value),
            dataframe["low"] <= (dataframe["ema20"] * (1.0 + float(self.pullback_tolerance.value))),
            dataframe["close"] >= dataframe["ema20"],
            trigger_ok,
            dataframe["volume"] >= (dataframe["vol_sma20"] * float(self.min_volume_ratio.value)),
            dataframe["range_proxy"] <= float(self.max_range_proxy.value),
            dataframe["atr_pct"] >= float(self.min_5m_atr_pct.value),
            dataframe["atr_pct"] <= float(self.max_5m_atr_pct.value),
        ]

        if entry_conditions:
            dataframe.loc[
                pd.concat(entry_conditions, axis=1).all(axis=1),
                ["enter_long", "enter_tag"],
            ] = (1, "rap_pullback_reclaim")
        return dataframe

    def populate_exit_trend(self, dataframe: DataFrame, metadata: dict[str, Any]) -> DataFrame:
        dataframe["exit_long"] = 0
        dataframe["exit_tag"] = None

        regime_break = ~dataframe["regime_up_1h"]
        momentum_fade = (dataframe["rsi"] > 72) & qtpylib.crossed_below(dataframe["close"], dataframe["ema20"])

        dataframe.loc[regime_break, ["exit_long", "exit_tag"]] = (1, "regime_break")
        dataframe.loc[momentum_fade, ["exit_long", "exit_tag"]] = (1, "momentum_fade")
        return dataframe

    def _latest_pair_row(self, pair: str) -> pd.Series | None:
        if not self.dp:
            return None
        dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)
        if dataframe is None or dataframe.empty:
            return None
        return dataframe.iloc[-1]

    def _risk_pct(self, pair: str, trade: Trade | None = None) -> float:
        if trade is not None:
            stored = trade.get_custom_data("risk_pct")
            if stored is not None:
                return float(stored)

        latest = self._latest_pair_row(pair)
        atr_pct = float(latest["atr_pct"]) if latest is not None and "atr_pct" in latest else 0.01
        risk_pct = min(float(self.risk_atr_mult.value) * atr_pct, 0.016)
        risk_pct = max(risk_pct, 0.005)

        if trade is not None:
            trade.set_custom_data("risk_pct", risk_pct)
        return risk_pct

    def custom_stoploss(
        self,
        pair: str,
        trade: Trade,
        current_time: datetime,
        current_rate: float,
        current_profit: float,
        after_fill: bool,
        **kwargs: Any,
    ) -> float | None:
        risk_pct = self._risk_pct(pair, trade)

        if current_profit >= (risk_pct * 1.8):
            # Lock profit at roughly +0.8R once +1.8R is reached.
            open_relative_stop = max(0.001, current_profit - risk_pct)
            return stoploss_from_open(
                open_relative_stop,
                current_profit,
                is_short=trade.is_short,
                leverage=trade.leverage,
            )

        if current_profit >= risk_pct:
            # Move to tiny positive stop once trade has moved +1R.
            return stoploss_from_open(
                0.001,
                current_profit,
                is_short=trade.is_short,
                leverage=trade.leverage,
            )

        return stoploss_from_open(
            -risk_pct,
            current_profit,
            is_short=trade.is_short,
            leverage=trade.leverage,
        )

    def custom_exit(
        self,
        pair: str,
        trade: Trade,
        current_time: datetime,
        current_rate: float,
        current_profit: float,
        **kwargs: Any,
    ) -> str | bool | None:
        open_time = getattr(trade, "open_date_utc", None) or getattr(trade, "open_date", None)
        trade_age = (current_time - open_time) if open_time is not None else None
        risk_pct = self._risk_pct(pair, trade)

        if open_time is not None and (current_time - open_time) >= timedelta(hours=3) and current_profit < 0.003:
            return "time_stop_3h"

        latest = self._latest_pair_row(pair)
        if latest is not None:
            regime_ok = bool(latest.get("regime_up_1h", True))
            if not regime_ok:
                return "regime_break_exit"

            rsi_value = float(latest.get("rsi", 50.0))
            close_value = float(latest.get("close", 0.0))
            ema20_value = float(latest.get("ema20", 0.0))
            ema50_value = float(latest.get("ema50", 0.0))

            # Cut weak trades before full stop when momentum and trend fail together.
            if (
                trade_age is not None
                and trade_age >= timedelta(minutes=10)
                and current_profit < -(risk_pct * float(self.trend_fail_risk_mult.value))
                and rsi_value < (float(self.weak_momentum_rsi.value) + 1.0)
                and close_value < ema20_value
            ):
                return "trend_fail_early"

            if (
                trade_age is not None
                and trade_age >= timedelta(minutes=25)
                and current_profit < -(risk_pct * float(self.stagnation_risk_mult.value))
                and rsi_value < (float(self.weak_momentum_rsi.value) + 2.0)
                and close_value < ema20_value
            ):
                return "stagnation_cut"

            if (
                trade_age is not None
                and trade_age >= timedelta(minutes=50)
                and current_profit < 0.001
                and rsi_value < 52.0
                and close_value < ema20_value
            ):
                return "time_decay_exit"

        if current_profit >= (risk_pct * 1.8):
            return "tp_1_8r"

        return None