# source: https://raw.githubusercontent.com/ayhanarashtasin/Trading_Backtest/64a9ce0543d7663cf4a4b155a69d8715cc781d96/user_data/strategies/btc_momentum_v4_1_trendpullback.py
# directory_url: https://github.com/ayhanarashtasin/Trading_Backtest/blob/main/user_data/strategies/
# User: ayhanarashtasin
# Repository: Trading_Backtest
# --------------------"""BTC/USDT multi-timeframe trend-pullback research strategy, V4.1.

Market: Binance BTC/USDT spot, long only
Execution timeframe: completed 5-minute candles
Macro context: completed native 1-hour candles

V4.1 removes V4's squeeze/breakout entry and instead enters a confirmed
bounce from 5m EMA support during a healthy bullish 1h trend.  Its initial
ATR stop is wider, and its activated stop ratchets from breakeven behind the
highest completed 5m close without ever loosening.
"""

from datetime import datetime
from math import isfinite

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

from freqtrade.persistence import Trade
from freqtrade.strategy import IStrategy, merge_informative_pair, stoploss_from_absolute


class Github_ayhanarashtasin_Trading_Backtest__btc_momentum_v4_1_trendpullback__20260826_055757(IStrategy):
    INTERFACE_VERSION = 3

    timeframe = "5m"
    informative_timeframes = ("1h",)
    can_short = False

    minimal_roi = {}
    use_exit_signal = True
    exit_profit_only = False
    ignore_roi_if_entry_signal = False

    # The custom ATR stop is the normal risk control.  This fixed value is a
    # wide emergency floor in case analyzed candle data is unavailable.
    stoploss = -0.10
    use_custom_stoploss = True
    trailing_stop = False

    process_only_new_candles = True

    order_types = {
        "entry": "market",
        "exit": "market",
        "stoploss": "market",
        "stoploss_on_exchange": False,
    }

    # Applied to each requested timeframe; sufficient for the 1h EMA(200)
    # and all 5m pullback, volume, and exit references.
    startup_candle_count: int = 250

    # Native 5m pullback and bounce parameters.
    EMA_5M_FAST_PERIOD = 20
    EMA_5M_TREND_PERIOD = 50
    PULLBACK_LOOKBACK = 3
    MIN_BODY_RATIO = 0.50
    VOLUME_SMA_PERIOD = 20
    ATR_5M_PERIOD = 14

    # Completed native 1h macro-trend parameters.
    EMA_1H_FAST_PERIOD = 50
    EMA_1H_SLOW_PERIOD = 200
    ADX_1H_PERIOD = 14
    ADX_ENTRY_THRESHOLD = 18.0
    RSI_1H_PERIOD = 14
    RSI_ENTRY_MIN = 48.0
    RSI_ENTRY_MAX = 72.0

    # Initial risk, activated breakeven step, and trailing ratchet.
    INITIAL_ATR_MULTIPLIER = 2.25
    RATCHET_ACTIVATION_PROFIT = 0.0075
    BREAKEVEN_LOCK_PROFIT = 0.0005
    TRAILING_ATR_MULTIPLIER = 1.75
    EMA_EXIT_PROFIT_GATE = 0.0100

    _ENTRY_ATR_KEY = "btc_momentum_v4_1_entry_atr"
    _HIGHEST_CLOSE_KEY = "btc_momentum_v4_1_highest_close"
    _EFFECTIVE_STOP_KEY = "btc_momentum_v4_1_effective_stop"
    _RATCHET_ACTIVE_KEY = "btc_momentum_v4_1_ratchet_active"

    def informative_pairs(self):
        """Load native 1h candles for every configured spot pair."""
        return [
            (pair, timeframe)
            for pair in self.dp.current_whitelist()
            for timeframe in self.informative_timeframes
        ]

    def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        # -------------------------- native 5m ---------------------------
        dataframe["atr_14"] = ta.ATR(
            dataframe,
            timeperiod=self.ATR_5M_PERIOD,
        )
        dataframe["ema_20"] = ta.EMA(
            dataframe,
            timeperiod=self.EMA_5M_FAST_PERIOD,
        )
        dataframe["ema_50"] = ta.EMA(
            dataframe,
            timeperiod=self.EMA_5M_TREND_PERIOD,
        )
        dataframe["volume_sma_20"] = dataframe["volume"].rolling(
            self.VOLUME_SMA_PERIOD,
            min_periods=self.VOLUME_SMA_PERIOD,
        ).mean()

        # Includes the current completed candle and the prior two candles.
        dataframe["pullback_touched_ema20"] = (
            (dataframe["low"] <= dataframe["ema_20"])
            .rolling(
                self.PULLBACK_LOOKBACK,
                min_periods=self.PULLBACK_LOOKBACK,
            )
            .max()
            .eq(1.0)
        )
        dataframe["previous_high"] = dataframe["high"].shift(1)
        candle_range = dataframe["high"] - dataframe["low"]
        dataframe["bullish_body_ratio"] = (
            (dataframe["close"] - dataframe["open"])
            / candle_range.where(candle_range > 0)
        )

        # -------------------------- native 1h ---------------------------
        informative_1h = self.dp.get_pair_dataframe(
            pair=metadata["pair"],
            timeframe="1h",
        ).copy()
        informative_1h["ema_50"] = ta.EMA(
            informative_1h,
            timeperiod=self.EMA_1H_FAST_PERIOD,
        )
        informative_1h["ema_200"] = ta.EMA(
            informative_1h,
            timeperiod=self.EMA_1H_SLOW_PERIOD,
        )
        informative_1h["adx_14"] = ta.ADX(
            informative_1h,
            timeperiod=self.ADX_1H_PERIOD,
        )
        informative_1h["rsi_14"] = ta.RSI(
            informative_1h,
            timeperiod=self.RSI_1H_PERIOD,
        )
        dataframe = merge_informative_pair(
            dataframe,
            informative_1h,
            self.timeframe,
            "1h",
            ffill=True,
        )

        # merge_informative_pair delays each 1h row until that native candle
        # has closed, so unfinished macro values are never visible to 5m.
        return dataframe

    def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        macro_trend = (
            (dataframe["close_1h"] > dataframe["ema_50_1h"])
            & (dataframe["ema_50_1h"] > dataframe["ema_200_1h"])
            & (dataframe["adx_14_1h"] > self.ADX_ENTRY_THRESHOLD)
            & (dataframe["rsi_14_1h"] > self.RSI_ENTRY_MIN)
            & (dataframe["rsi_14_1h"] < self.RSI_ENTRY_MAX)
        )
        local_trend = dataframe["close"] > dataframe["ema_50"]
        bullish_confirmation = (
            (dataframe["close"] > dataframe["open"])
            & (dataframe["close"] > dataframe["previous_high"])
            & (dataframe["bullish_body_ratio"] >= self.MIN_BODY_RATIO)
            & (dataframe["volume"] > dataframe["volume_sma_20"])
        )
        enter_long = (
            macro_trend
            & local_trend
            & dataframe["pullback_touched_ema20"]
            & bullish_confirmation
            & (dataframe["volume"] > 0)
        )
        dataframe.loc[enter_long, ["enter_long", "enter_tag"]] = (
            1,
            "mtf_trend_pullback_bounce",
        )
        return dataframe

    def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        # The profit-gated EMA exhaustion is evaluated in custom_exit, which
        # can inspect each trade's achieved gross profit.
        dataframe["exit_long"] = 0
        return dataframe

    @classmethod
    def _next_stop_price(
        cls,
        entry_price: float,
        entry_atr: float,
        previous_effective_stop: float | None,
        highest_completed_close: float | None,
        current_atr: float,
        ratchet_active: bool,
    ) -> float:
        """Return the monotonic absolute stop price for a long trade."""
        initial_stop = entry_price - cls.INITIAL_ATR_MULTIPLIER * entry_atr
        candidates = [initial_stop]

        if previous_effective_stop is not None and isfinite(previous_effective_stop):
            candidates.append(previous_effective_stop)

        if ratchet_active:
            candidates.append(
                entry_price * (1.0 + cls.BREAKEVEN_LOCK_PROFIT)
            )
            if (
                highest_completed_close is not None
                and isfinite(highest_completed_close)
                and isfinite(current_atr)
                and current_atr > 0
            ):
                candidates.append(
                    highest_completed_close
                    - cls.TRAILING_ATR_MULTIPLIER * current_atr
                )

        return max(candidates)

    def _completed_candles(
        self,
        pair: str,
        current_time: datetime,
    ) -> DataFrame:
        """Return only 5m candles completed by the callback timestamp."""
        dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)
        if dataframe.empty:
            return dataframe
        candle_cutoff = pd.Timestamp(current_time) - pd.Timedelta(minutes=5)
        return dataframe.loc[dataframe["date"] <= candle_cutoff]

    def custom_stoploss(
        self,
        pair: str,
        trade: Trade,
        current_time: datetime,
        current_rate: float,
        current_profit: float,
        after_fill: bool,
        **kwargs,
    ) -> float | None:
        """Maintain the initial ATR stop and activated monotonic ratchet."""
        completed = self._completed_candles(pair, current_time)
        if completed.empty or "atr_14" not in completed.columns:
            return None

        current_atr = float(completed.iloc[-1]["atr_14"])
        if not isfinite(current_atr) or current_atr <= 0:
            return None

        entry_atr = trade.get_custom_data(self._ENTRY_ATR_KEY)
        if entry_atr is None:
            # At next-candle market entry this is the signal candle's ATR,
            # which is the latest completed 5m ATR available at entry time.
            entry_atr = current_atr
            trade.set_custom_data(self._ENTRY_ATR_KEY, float(entry_atr))
        entry_atr = float(entry_atr)

        post_entry = completed.loc[
            completed["date"] >= pd.Timestamp(trade.open_date_utc)
        ]
        stored_highest = trade.get_custom_data(self._HIGHEST_CLOSE_KEY)
        highest_completed_close = (
            float(stored_highest) if stored_highest is not None else None
        )
        if not post_entry.empty:
            observed_highest = float(post_entry["close"].max())
            if isfinite(observed_highest):
                highest_completed_close = (
                    observed_highest
                    if highest_completed_close is None
                    else max(highest_completed_close, observed_highest)
                )
                trade.set_custom_data(
                    self._HIGHEST_CLOSE_KEY,
                    highest_completed_close,
                )

        ratchet_active = bool(
            trade.get_custom_data(self._RATCHET_ACTIVE_KEY, False)
        )
        observed_peak_rate = max(
            float(trade.max_rate or trade.open_rate),
            float(current_rate),
        )
        if observed_peak_rate >= float(trade.open_rate) * (
            1.0 + self.RATCHET_ACTIVATION_PROFIT
        ):
            ratchet_active = True
            trade.set_custom_data(self._RATCHET_ACTIVE_KEY, True)

        stored_stop = trade.get_custom_data(self._EFFECTIVE_STOP_KEY)
        previous_effective_stop = (
            float(stored_stop) if stored_stop is not None else None
        )
        effective_stop = self._next_stop_price(
            entry_price=float(trade.open_rate),
            entry_atr=entry_atr,
            previous_effective_stop=previous_effective_stop,
            highest_completed_close=highest_completed_close,
            current_atr=current_atr,
            ratchet_active=ratchet_active,
        )
        trade.set_custom_data(self._EFFECTIVE_STOP_KEY, float(effective_stop))

        distance = stoploss_from_absolute(
            stop_rate=effective_stop,
            current_rate=current_rate,
            is_short=trade.is_short,
            leverage=trade.leverage or 1.0,
        )
        return distance if distance > 0 else None

    def custom_exit(
        self,
        pair: str,
        trade: Trade,
        current_time: datetime,
        current_rate: float,
        current_profit: float,
        **kwargs,
    ) -> str | None:
        """Exit after a +1% peak when a completed 5m close is below EMA(50)."""
        completed = self._completed_candles(pair, current_time)
        if completed.empty:
            return None

        current = completed.iloc[-1]
        peak_rate = max(
            float(trade.max_rate or trade.open_rate),
            float(current_rate),
        )
        achieved_profit = peak_rate / float(trade.open_rate) - 1.0
        if (
            achieved_profit >= self.EMA_EXIT_PROFIT_GATE
            and pd.notna(current.get("ema_50"))
            and float(current["close"]) < float(current["ema_50"])
        ):
            return "ema50_exhaustion_after_1pct"

        return None