# source: https://raw.githubusercontent.com/ayhanarashtasin/Trading_Backtest/64a9ce0543d7663cf4a4b155a69d8715cc781d96/user_data/strategies/btc_momentum_v1_3_ma.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.

V1.3 adds exactly one trading-logic change to BTCMomentumV1: long entries
also require the completed 1m close to be above EMA(200).
"""

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_v1_3_ma__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), 1m ATR(14), and 1m EMA(200), 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
    TRAILING_ATR_MULTIPLIER = 1.50

    _ENTRY_ATR_KEY = "btc_momentum_v1_entry_atr"
    _HIGHEST_CLOSE_KEY = "btc_momentum_v1_highest_close"
    _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,
        )

        # V1.3 FILTER (the only trading-logic addition): 1m EMA(200).
        dataframe["ema_200_1m"] = ta.EMA(dataframe, timeperiod=200)
        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["close"] > dataframe["ema_200_1m"])
        )
        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,
        highest_completed_close: 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 (
            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 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)

        # A candle counts "since entry" only after the 1m candle whose open
        # timestamp is at or later than the fill timestamp has completed.
        post_entry = completed.loc[
            dataframe.loc[completed.index, "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,
                )

        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,
        )
        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