# source: https://raw.githubusercontent.com/ayhanarashtasin/Trading_Backtest/64a9ce0543d7663cf4a4b155a69d8715cc781d96/user_data/strategies/btc_momentum_v2_2_structuretrail.py
# directory_url: https://github.com/ayhanarashtasin/Trading_Backtest/blob/main/user_data/strategies/
# User: ayhanarashtasin
# Repository: Trading_Backtest
# --------------------"""
Plan-compliant BTC/USDT momentum baseline, version 1.

Market: Binance BTC/USDT spot
Signal timeframe: 1 minute
Informative timeframe: 5 minutes (completed candles only)
Direction: long only

The informative indicators are calculated on the native 5m dataframe before
``merge_informative_pair`` makes them available to the 1m strategy.  The
helper delays each informative row until its 5m candle has closed, preventing
unfinished higher-timeframe data from leaking into a 1m decision.
"""

from datetime import datetime
from math import isfinite

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

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


class Github_ayhanarashtasin_Trading_Backtest__btc_momentum_v2_2_structuretrail__20260826_055757(IStrategy):
    INTERFACE_VERSION = 3

    timeframe = "1m"
    informative_timeframe = "5m"
    can_short = False

    # An empty ROI table disables fixed take-profit exits.  The strategy exits
    # only when the confirmed 5m regime ends or its custom ATR stop is reached.
    minimal_roi = {}
    use_exit_signal = True
    exit_profit_only = False
    ignore_roi_if_entry_signal = False

    # Wide emergency floor.  Normal risk control is entirely handled by the
    # ATR-based custom stop below.
    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,
    }

    # 1,000 1m candles provide 200 5m candles.  This covers BB(20), the
    # 100-candle bandwidth quantile, the six-candle consolidation lookback,
    # ATR(14), the previous-50 ATR mean, the previous-20 breakout level,
    # EMA(20), and 1m ATR(14), with additional warmup margin.
    startup_candle_count: int = 1000

    # Frozen 5m consolidation/breakout parameters.
    ATR_5M_PERIOD = 14
    BB_PERIOD = 20
    BB_STDDEV = 2.0
    BANDWIDTH_QUANTILE_PERIOD = 100
    BANDWIDTH_QUANTILE = 0.25
    CONSOLIDATION_LOOKBACK = 6
    CONSOLIDATION_MIN_COUNT = 4
    BREAKOUT_LOOKBACK = 20
    ATR_REFERENCE_PERIOD = 50
    ATR_EXPANSION_MULTIPLIER = 1.20
    REGIME_MAX_CANDLES = 6
    EMA_5M_PERIOD = 20

    # Frozen 1m risk parameters.
    ATR_1M_PERIOD = 14
    INITIAL_ATR_MULTIPLIER = 1.25
    STRUCTURE_LOOKBACK = 5
    STRUCTURE_ATR_BUFFER_MULTIPLIER = 0.25

    _ENTRY_ATR_KEY = "btc_momentum_v1_entry_atr"
    _EFFECTIVE_STOP_KEY = "btc_momentum_v1_effective_stop"

    def informative_pairs(self):
        """Load 5m candles for every configured (BTC/USDT) spot pair."""
        return [
            (pair, self.informative_timeframe)
            for pair in self.dp.current_whitelist()
        ]

    @classmethod
    def _calculate_bullish_regime(
        cls,
        breakout: Series,
        close: Series,
        ema: Series,
    ) -> Series:
        """
        Build a causal regime state.

        A breakout starts (or refreshes) a six-completed-candle window,
        including the breakout candle.  An EMA breach ends that regime
        immediately and it cannot reactivate without another breakout.
        """
        remaining = 0
        active: list[bool] = []

        for breakout_now, close_now, ema_now in zip(breakout, close, ema):
            if bool(breakout_now):
                remaining = cls.REGIME_MAX_CANDLES

            above_ema = (
                pd.notna(close_now)
                and pd.notna(ema_now)
                and float(close_now) > float(ema_now)
            )
            regime_now = remaining > 0 and above_ema
            active.append(regime_now)

            if regime_now:
                remaining -= 1
            else:
                # Once price loses EMA(20), the old breakout cannot reactivate
                # the regime even if price recovers inside the original window.
                remaining = 0

        return Series(active, index=breakout.index, dtype=bool)

    def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        informative = self.dp.get_pair_dataframe(
            pair=metadata["pair"],
            timeframe=self.informative_timeframe,
        ).copy()

        # ------------------------- 5m indicators -------------------------
        informative["atr"] = ta.ATR(
            informative,
            timeperiod=self.ATR_5M_PERIOD,
        )
        bb_upper, bb_middle, bb_lower = ta.BBANDS(
            informative["close"],
            timeperiod=self.BB_PERIOD,
            nbdevup=self.BB_STDDEV,
            nbdevdn=self.BB_STDDEV,
            matype=0,
        )
        informative["bb_upper"] = bb_upper
        informative["bb_middle"] = bb_middle
        informative["bb_lower"] = bb_lower
        informative["bb_bandwidth"] = (
            (informative["bb_upper"] - informative["bb_lower"])
            / informative["bb_middle"]
        )
        informative["bandwidth_q25"] = informative["bb_bandwidth"].rolling(
            self.BANDWIDTH_QUANTILE_PERIOD,
            min_periods=self.BANDWIDTH_QUANTILE_PERIOD,
        ).quantile(self.BANDWIDTH_QUANTILE)
        informative["consolidation"] = (
            informative["bb_bandwidth"] < informative["bandwidth_q25"]
        ).fillna(False)

        # The breakout candle itself is deliberately excluded from all three
        # "previous" references below.
        previous_consolidations = (
            informative["consolidation"]
            .astype(int)
            .shift(1)
            .rolling(
                self.CONSOLIDATION_LOOKBACK,
                min_periods=self.CONSOLIDATION_LOOKBACK,
            )
            .sum()
        )
        informative["recent_consolidation"] = (
            previous_consolidations >= self.CONSOLIDATION_MIN_COUNT
        ).fillna(False)
        informative["previous_breakout_level"] = (
            informative["high"]
            .shift(1)
            .rolling(self.BREAKOUT_LOOKBACK, min_periods=self.BREAKOUT_LOOKBACK)
            .max()
        )
        informative["previous_atr_reference"] = (
            informative["atr"]
            .shift(1)
            .rolling(
                self.ATR_REFERENCE_PERIOD,
                min_periods=self.ATR_REFERENCE_PERIOD,
            )
            .mean()
        )
        informative["breakout"] = (
            informative["recent_consolidation"]
            & (
                informative["close"]
                > informative["previous_breakout_level"]
            )
            & (
                informative["atr"]
                > self.ATR_EXPANSION_MULTIPLIER
                * informative["previous_atr_reference"]
            )
        ).fillna(False)

        informative["ema_20"] = ta.EMA(
            informative,
            timeperiod=self.EMA_5M_PERIOD,
        )
        informative["bullish_regime"] = self._calculate_bullish_regime(
            informative["breakout"],
            informative["close"],
            informative["ema_20"],
        )

        # merge_informative_pair shifts each 5m row to the first compatible
        # 1m row whose close is at or after that 5m candle's close.
        dataframe = merge_informative_pair(
            dataframe,
            informative,
            self.timeframe,
            self.informative_timeframe,
            ffill=True,
        )

        # -------------------------- 1m indicators -------------------------
        dataframe["atr_1m"] = ta.ATR(
            dataframe,
            timeperiod=self.ATR_1M_PERIOD,
        )
        return dataframe

    def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        raw_entry = (
            dataframe["bullish_regime_5m"].fillna(False).astype(bool)
            & (dataframe["close"] > dataframe["high"].shift(1))
            & (dataframe["volume"] > 0)
        )
        dataframe["raw_entry"] = raw_entry.astype(int)
        dataframe.loc[raw_entry, ["enter_long", "enter_tag"]] = (
            1,
            "bullish_breakout_regime",
        )
        return dataframe

    def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        regime = dataframe["bullish_regime_5m"].fillna(False).astype(bool)
        regime_ended = regime.shift(1).fillna(False) & ~regime
        dataframe.loc[regime_ended, ["exit_long", "exit_tag"]] = (
            1,
            "bullish_regime_ended",
        )
        return dataframe

    @classmethod
    def _next_stop_price(
        cls,
        entry_price: float,
        entry_atr: float,
        previous_effective_stop: float | None,
        structure_level: float | None,
        current_atr: float,
    ) -> 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 (
            structure_level is not None
            and isfinite(structure_level)
            and isfinite(current_atr)
            and current_atr > 0
        ):
            candidates.append(
                structure_level
                - cls.STRUCTURE_ATR_BUFFER_MULTIPLIER * current_atr
            )

        return max(candidates)

    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 ATR stop using only candles completed by ``current_time``.

        Freqtrade also enforces monotonic stop movement internally.  Persisting
        the absolute effective stop here makes the same invariant explicit and
        keeps it intact across bot loops and restarts.
        """
        dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)
        if dataframe.empty or "atr_1m" not in dataframe.columns:
            return None

        candle_cutoff = pd.Timestamp(current_time) - pd.Timedelta(minutes=1)
        completed = dataframe.loc[dataframe["date"] <= candle_cutoff]
        if completed.empty:
            return None

        current_atr = float(completed.iloc[-1]["atr_1m"])
        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 entry time, this is the newest ATR known without peeking into
            # the just-opened candle.
            entry_atr = current_atr
            trade.set_custom_data(self._ENTRY_ATR_KEY, float(entry_atr))
        entry_atr = float(entry_atr)

        structure_window = completed.tail(self.STRUCTURE_LOOKBACK)
        structure_level = None
        if len(structure_window) == self.STRUCTURE_LOOKBACK:
            observed_structure = float(structure_window["low"].min())
            if isfinite(observed_structure):
                structure_level = observed_structure

        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,
            structure_level=structure_level,
            current_atr=current_atr,
        )
        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,
        )
        # A zero distance means price is already beyond the requested stop.
        # Keeping the engine's previous stop is safer than asking it to refresh
        # the stop after a gap.
        return distance if distance > 0 else None