# source: https://raw.githubusercontent.com/ayhanarashtasin/Trading_Backtest/ecbae6de6d58185a507baa49af5179d899bbba10/user_data/strategies/btc_momentum_v4_trendexpansion.py
# directory_url: https://github.com/ayhanarashtasin/Trading_Backtest/blob/main/user_data/strategies/
# User: ayhanarashtasin
# Repository: Trading_Backtest
# --------------------"""BTC/USDT multi-timeframe trend-expansion research strategy, V4.

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

V4 deliberately abandons the 1m V1-V3 architecture.  It waits for a 5m
volatility/volume breakout after a squeeze, permits entries only in a strong
macro trend, and lets trends run without a fixed candle-count regime cap.
"""

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_trendexpansion__20260826_110850(IStrategy):
    INTERFACE_VERSION = 3

    timeframe = "5m"
    informative_timeframes = ("1h", "4h")
    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,
    }

    # Freqtrade applies this count to each requested timeframe.  Two hundred
    # fifty native candles cover EMA(200), ADX(14), and every 5m reference.
    startup_candle_count: int = 250

    # 5m squeeze and breakout parameters.
    BB_PERIOD = 20
    BB_STDDEV = 2.0
    BANDWIDTH_QUANTILE_PERIOD = 100
    BANDWIDTH_QUANTILE = 0.25
    BREAKOUT_LOOKBACK = 20
    ATR_5M_PERIOD = 14
    ATR_REFERENCE_PERIOD = 50
    ATR_EXPANSION_MULTIPLIER = 1.25
    VOLUME_SMA_PERIOD = 20
    VOLUME_EXPANSION_MULTIPLIER = 1.20

    # Macro trend parameters.
    EMA_1H_PERIOD = 200
    EMA_4H_PERIOD = 50
    ADX_1H_PERIOD = 14
    ADX_ENTRY_THRESHOLD = 20.0
    ADX_ROLLOVER_THRESHOLD = 40.0

    # Risk and trend-exhaustion parameters.
    INITIAL_ATR_MULTIPLIER = 1.50
    TRAIL_ACTIVATION_PROFIT = 0.0100
    TRAILING_ATR_MULTIPLIER = 2.00
    EMA_5M_EXIT_PERIOD = 50
    EMA_EXIT_PROFIT_GATE = 0.0150

    _ENTRY_ATR_KEY = "btc_momentum_v4_entry_atr"
    _HIGHEST_CLOSE_KEY = "btc_momentum_v4_highest_close"
    _EFFECTIVE_STOP_KEY = "btc_momentum_v4_effective_stop"
    _TRAIL_ACTIVE_KEY = "btc_momentum_v4_trail_active"

    def informative_pairs(self):
        """Load native 1h and 4h 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,
        )
        bb_upper, bb_middle, bb_lower = ta.BBANDS(
            dataframe["close"],
            timeperiod=self.BB_PERIOD,
            nbdevup=self.BB_STDDEV,
            nbdevdn=self.BB_STDDEV,
            matype=0,
        )
        dataframe["bb_upper"] = bb_upper
        dataframe["bb_middle"] = bb_middle
        dataframe["bb_lower"] = bb_lower
        dataframe["bb_bandwidth"] = (
            (dataframe["bb_upper"] - dataframe["bb_lower"])
            / dataframe["bb_middle"]
        )

        # The breakout candle is excluded from every historical reference.
        prior_bandwidth = dataframe["bb_bandwidth"].shift(1)
        dataframe["prior_bandwidth_q25"] = prior_bandwidth.rolling(
            self.BANDWIDTH_QUANTILE_PERIOD,
            min_periods=self.BANDWIDTH_QUANTILE_PERIOD,
        ).quantile(self.BANDWIDTH_QUANTILE)
        dataframe["prior_squeeze"] = (
            prior_bandwidth <= dataframe["prior_bandwidth_q25"]
        ).fillna(False)
        dataframe["previous_breakout_level"] = (
            dataframe["high"]
            .shift(1)
            .rolling(self.BREAKOUT_LOOKBACK, min_periods=self.BREAKOUT_LOOKBACK)
            .max()
        )
        dataframe["previous_atr_reference"] = (
            dataframe["atr_14"]
            .shift(1)
            .rolling(
                self.ATR_REFERENCE_PERIOD,
                min_periods=self.ATR_REFERENCE_PERIOD,
            )
            .mean()
        )
        dataframe["previous_volume_sma"] = (
            dataframe["volume"]
            .shift(1)
            .rolling(
                self.VOLUME_SMA_PERIOD,
                min_periods=self.VOLUME_SMA_PERIOD,
            )
            .mean()
        )
        dataframe["ema_50"] = ta.EMA(
            dataframe,
            timeperiod=self.EMA_5M_EXIT_PERIOD,
        )

        # -------------------------- native 1h ---------------------------
        informative_1h = self.dp.get_pair_dataframe(
            pair=metadata["pair"],
            timeframe="1h",
        ).copy()
        informative_1h["ema_200"] = ta.EMA(
            informative_1h,
            timeperiod=self.EMA_1H_PERIOD,
        )
        informative_1h["adx_14"] = ta.ADX(
            informative_1h,
            timeperiod=self.ADX_1H_PERIOD,
        )
        informative_1h["adx_14_previous"] = informative_1h["adx_14"].shift(1)
        dataframe = merge_informative_pair(
            dataframe,
            informative_1h,
            self.timeframe,
            "1h",
            ffill=True,
        )

        # -------------------------- native 4h ---------------------------
        informative_4h = self.dp.get_pair_dataframe(
            pair=metadata["pair"],
            timeframe="4h",
        ).copy()
        informative_4h["ema_50"] = ta.EMA(
            informative_4h,
            timeperiod=self.EMA_4H_PERIOD,
        )
        dataframe = merge_informative_pair(
            dataframe,
            informative_4h,
            self.timeframe,
            "4h",
            ffill=True,
        )

        # merge_informative_pair exposes each higher-timeframe row only after
        # that native candle has closed; no unfinished 1h/4h value is visible.
        return dataframe

    def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        macro_trend = (
            (
                (dataframe["close_1h"] > dataframe["ema_200_1h"])
                | (dataframe["close_4h"] > dataframe["ema_50_4h"])
            )
            & (dataframe["adx_14_1h"] > self.ADX_ENTRY_THRESHOLD)
        )
        breakout = (
            dataframe["prior_squeeze"].fillna(False).astype(bool)
            & (dataframe["close"] > dataframe["previous_breakout_level"])
            & (
                dataframe["atr_14"]
                > self.ATR_EXPANSION_MULTIPLIER
                * dataframe["previous_atr_reference"]
            )
            & (
                dataframe["volume"]
                > self.VOLUME_EXPANSION_MULTIPLIER
                * dataframe["previous_volume_sma"]
            )
        )
        enter_long = macro_trend & breakout & (dataframe["volume"] > 0)
        dataframe.loc[enter_long, ["enter_long", "enter_tag"]] = (
            1,
            "mtf_squeeze_trend_expansion",
        )
        return dataframe

    def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        # Profit-dependent EMA exhaustion and the 1h ADX rollover are handled
        # in custom_exit, which has access to each trade's achieved 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,
        trail_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 (
            trail_active
            and 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 ATR trail."""
        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:
            # The first callback sees the signal candle's completed ATR, which
            # is the latest value known at the next-candle market entry.
            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,
                )

        trail_active = bool(trade.get_custom_data(self._TRAIL_ACTIVE_KEY, False))
        if (
            highest_completed_close is not None
            and highest_completed_close
            >= float(trade.open_rate) * (1.0 + self.TRAIL_ACTIVATION_PROFIT)
        ):
            trail_active = True
            trade.set_custom_data(self._TRAIL_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,
            trail_active=trail_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 on gated EMA exhaustion or a completed-1h ADX rollover."""
        completed = self._completed_candles(pair, current_time)
        if completed.empty:
            return None

        current = completed.iloc[-1]
        adx_now = current.get("adx_14_1h")
        adx_previous = current.get("adx_14_previous_1h")
        if (
            pd.notna(adx_now)
            and pd.notna(adx_previous)
            and float(adx_previous) > self.ADX_ROLLOVER_THRESHOLD
            and float(adx_now) < float(adx_previous)
        ):
            return "adx_1h_rollover_from_40"

        if len(completed) < 2:
            return None
        previous = completed.iloc[-2]
        ema_cross_below = (
            pd.notna(previous.get("ema_50"))
            and pd.notna(current.get("ema_50"))
            and float(previous["close"]) >= float(previous["ema_50"])
            and float(current["close"]) < float(current["ema_50"])
        )
        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 ema_cross_below:
            return "ema50_exhaustion_after_1_5pct"

        return None