# source: https://raw.githubusercontent.com/ayhanarashtasin/Trading_Backtest/ecbae6de6d58185a507baa49af5179d899bbba10/user_data/strategies/btc_momentum_v5_base.py
# directory_url: https://github.com/ayhanarashtasin/Trading_Backtest/blob/main/user_data/strategies/
# User: ayhanarashtasin
# Repository: Trading_Backtest
# --------------------"""BTC/USDT multi-timeframe swing momentum strategy, V5 Base.

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

V5 Base implements a multi-timeframe swing momentum architecture:
1. Long-only trend alignment across 1d (daily EMA20 > EMA50, Close > EMA50) and 4h (EMA20 > EMA50, Close > EMA20, ADX > 18).
2. 1h breakout of the 20-period highest high accompanied by volume expansion and volatility confirmation.
3. Asymmetric risk-reward trailing stoploss with ATR-based initial buffer, breakeven lock at +1.0%, and dynamic ATR trailing ratchet.
"""

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

    timeframe = "1h"
    informative_timeframes = ("4h", "1d")
    can_short = False

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

    # Fixed safety floor in case analyzed candle data is temporarily 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,
    }

    startup_candle_count: int = 250

    # 1h Breakout and volatility parameters
    BREAKOUT_LOOKBACK = 20
    ATR_1H_PERIOD = 14
    ATR_REFERENCE_PERIOD = 50
    ATR_EXPANSION_MULTIPLIER = 1.05
    VOLUME_SMA_PERIOD = 20
    VOLUME_EXPANSION_MULTIPLIER = 1.05
    BB_PERIOD = 20
    BB_STDDEV = 2.0
    BANDWIDTH_QUANTILE_PERIOD = 100
    BANDWIDTH_QUANTILE = 0.50

    # 1h Momentum indicators
    RSI_PERIOD = 14
    RSI_MIN = 50.0
    RSI_MAX = 75.0

    # Macro trend parameters
    EMA_4H_FAST = 20
    EMA_4H_SLOW = 50
    ADX_4H_PERIOD = 14
    ADX_4H_THRESHOLD = 18.0

    EMA_1D_FAST = 20
    EMA_1D_SLOW = 50
    EMA_1D_LONG = 200

    # Risk and trailing ratchet parameters
    INITIAL_ATR_MULTIPLIER = 3.00
    BE_ACTIVATION_PROFIT = 0.0100
    BE_LOCK_PROFIT = 0.0020
    TRAIL_ACTIVATION_PROFIT = 0.0150
    TRAILING_ATR_MULTIPLIER = 2.50

    _ENTRY_ATR_KEY = "btc_momentum_v5_entry_atr"
    _HIGHEST_CLOSE_KEY = "btc_momentum_v5_highest_close"
    _EFFECTIVE_STOP_KEY = "btc_momentum_v5_effective_stop"
    _BE_LOCKED_KEY = "btc_momentum_v5_be_locked"
    _TRAIL_ACTIVE_KEY = "btc_momentum_v5_trail_active"

    def informative_pairs(self):
        """Load native 4h and 1d 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 1h ---------------------------
        dataframe["atr_14"] = ta.ATR(
            dataframe,
            timeperiod=self.ATR_1H_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"]
        )

        prior_bandwidth = dataframe["bb_bandwidth"].shift(1)
        dataframe["prior_bandwidth_q50"] = 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_q50"]
        ).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_20"] = ta.EMA(dataframe, timeperiod=20)
        dataframe["ema_50"] = ta.EMA(dataframe, timeperiod=50)
        dataframe["rsi_14"] = ta.RSI(dataframe, timeperiod=self.RSI_PERIOD)
        dataframe["adx_14"] = ta.ADX(dataframe, timeperiod=14)

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

        # -------------------------- native 1d ---------------------------
        informative_1d = self.dp.get_pair_dataframe(
            pair=metadata["pair"],
            timeframe="1d",
        ).copy()
        informative_1d["ema_20"] = ta.EMA(
            informative_1d,
            timeperiod=self.EMA_1D_FAST,
        )
        informative_1d["ema_50"] = ta.EMA(
            informative_1d,
            timeperiod=self.EMA_1D_SLOW,
        )
        informative_1d["ema_200"] = ta.EMA(
            informative_1d,
            timeperiod=self.EMA_1D_LONG,
        )
        dataframe = merge_informative_pair(
            dataframe,
            informative_1d,
            self.timeframe,
            "1d",
            ffill=True,
        )

        return dataframe

    def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        # Macro Daily Bullish Alignment
        daily_trend = (
            (dataframe["close_1d"] > dataframe["ema_50_1d"])
            & (dataframe["ema_20_1d"] > dataframe["ema_50_1d"])
        )

        # Macro 4h Trend & Directional Expansion
        four_hour_trend = (
            (dataframe["close_4h"] > dataframe["ema_20_4h"])
            & (dataframe["ema_20_4h"] > dataframe["ema_50_4h"])
            & (dataframe["adx_14_4h"] > self.ADX_4H_THRESHOLD)
        )

        # 1h Breakout with volume and RSI momentum sanity
        breakout = (
            (dataframe["close"] > dataframe["previous_breakout_level"])
            & (
                dataframe["volume"]
                > self.VOLUME_EXPANSION_MULTIPLIER * dataframe["previous_volume_sma"]
            )
            & (dataframe["rsi_14"] > self.RSI_MIN)
            & (dataframe["rsi_14"] < self.RSI_MAX)
        )

        enter_long = daily_trend & four_hour_trend & breakout & (dataframe["volume"] > 0)
        dataframe.loc[enter_long, ["enter_long", "enter_tag"]] = (
            1,
            "mtf_swing_breakout",
        )
        return dataframe

    def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        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,
        be_locked: bool,
        trail_active: bool,
    ) -> float:
        """Return monotonic absolute stop price with initial ATR, BE lock, and ATR trail."""
        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 be_locked:
            candidates.append(entry_price * (1.0 + cls.BE_LOCK_PROFIT))

        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 1h 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(hours=1)
        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, BE ratchet, and trailing ATR stop."""
        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:
            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,
                )

        be_locked = bool(trade.get_custom_data(self._BE_LOCKED_KEY, False))
        if (
            highest_completed_close is not None
            and highest_completed_close
            >= float(trade.open_rate) * (1.0 + self.BE_ACTIVATION_PROFIT)
        ):
            be_locked = True
            trade.set_custom_data(self._BE_LOCKED_KEY, True)

        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,
            be_locked=be_locked,
            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