# source: https://raw.githubusercontent.com/ayhanarashtasin/Trading_Backtest/64a9ce0543d7663cf4a4b155a69d8715cc781d96/user_data/strategies/BTCMomentumATRExpansionV1.py
# directory_url: https://github.com/ayhanarashtasin/Trading_Backtest/blob/main/user_data/strategies/
# User: ayhanarashtasin
# Repository: Trading_Backtest
# --------------------"""
Github_ayhanarashtasin_Trading_Backtest__BTCMomentumATRExpansionV1__20260826_055757  --  PHASE 5 VARIANT D (ATR VOLATILITY EXPANSION)
===========================================================================
COPY of the frozen BTCMomentumBaselineV1 with EXACTLY ONE change:
one additional entry filter, marked by a PHASE 5 FILTER block.
Stop, exits, bucket rules, HTF boundary, sizing and warmup are
identical to the baseline and must not be touched.

Added filter: 1m ATR(14) must be expanded relative to its own
SMA(50) baseline:
    ATR(14)  >=  1.10 * SMA(ATR(14), 50)
Both ATR and SMA are standard TA-Lib implementations.
Computed on the 1m execution frame, consistent with the other
1m-based filters.

Frozen migration of the TradingView BTC/USDT long-only momentum baseline.

Market      : Binance BTC/USDT SPOT
Signal TF   : 1m
Context TF  : 5m (completed candles only)
Direction   : LONG ONLY

--------------------------------------------------------------------------
EXECUTION MODEL: TradingView vs Freqtrade  (ACCEPTED, DOCUMENTED DIFFERENCE)
--------------------------------------------------------------------------
The TradingView baseline ran with process_orders_on_close = true:

    condition true on candle T  ->  order filled at close[T]

Freqtrade's backtester cannot fill on the signal candle. Its engine shifts
the enter/exit columns by one row (see backtesting.py: "To avoid using data
from future, we use entry/exit signals shifted from the previous candle"):

    condition true on candle T  ->  order filled at open[T+1]

Therefore EVERY Freqtrade fill timestamp is expected to be ~1 minute later
than the corresponding TradingView timestamp. This is CORRECT and is NOT
compensated for anywhere in this file:

  * we do NOT shift the trading signal backwards,
  * we do NOT fake a fill at close[T],
  * we do NOT peek at candle T+1 to decide candle T.

For Phase 3 validation compare TWO SEPARATE things:
  * SIGNAL time : Freqtrade signal candle T   vs TradingView bar T  -> should match
  * FILL time   : Freqtrade fill candle T+1   vs TradingView bar T  -> +1 minute

Because BTC trades continuously, close[T] is essentially equal to open[T+1],
so the PRICE impact of this difference is near zero even though the
TIMESTAMP differs by one minute.

--------------------------------------------------------------------------
5-MINUTE LOOKAHEAD SAFETY
--------------------------------------------------------------------------
A 1m candle stamped T may only see a 5m candle that has FULLY CLOSED.
merge_informative_pair() sets:

    date_merge = inf_date + timeframe_inf - timeframe
               = inf_date + 5min - 1min

so the 5m candle 12:00 (spanning 12:00 -> 12:05) is attached from the 1m
candle 12:04 onward -- the 1m candle that closes at 12:05:00, i.e. the exact
instant the 5m candle closes. That is genuinely lookahead-free, but it is
ONE 1m BAR EARLIER than TradingView's lookahead_off behaviour, which only
publishes a completed HTF bar on the first LTF bar that OPENS after the HTF
bar closed (i.e. the 1m candle 12:05).

HTF_TV_BOUNDARY (below) applies that single extra 1m delay so the mapping
matches TradingView and the timing table in the project specification:

    1m 12:00..12:04 -> sees completed 5m 11:55
    1m 12:05..12:09 -> sees completed 5m 12:00
    1m 12:10..12:14 -> sees completed 5m 12:05

This is an EXECUTION-SEMANTICS flag, not a tunable strategy parameter. It
must stay fixed once the baseline is validated. Setting it False is still
lookahead-free (it reverts to stock Freqtrade behaviour) and it exists only
so that Phase 3 can isolate boundary effects. NEVER optimise it.

The direction of the shift is conservative: it can only make information
arrive LATER, never earlier, so it cannot introduce lookahead.
"""

from datetime import datetime, timedelta

import talib.abstract as ta
from pandas import DataFrame

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


class Github_ayhanarashtasin_Trading_Backtest__BTCMomentumATRExpansionV1__20260826_055757(IStrategy):
    INTERFACE_VERSION = 3

    # ---------------- timeframes ----------------
    timeframe = "1m"
    informative_timeframe = "5m"

    # ---------------- direction ----------------
    # Spot, long only. There is no short logic anywhere in this file.
    can_short = False

    # ---------------- exits ----------------
    # ROI must NEVER close a trade: the baseline has no take-profit.
    # 100 means +10000% required, which is unreachable on BTC 1m. A single
    # "0" key also means the ROI table has no time decay.
    minimal_roi = {"0": 100}

    # Fixed 0.25% stop below the ACTUAL Freqtrade fill price (open[T+1]),
    # not below the signal candle close. Freqtrade computes
    #     stop_price = open_rate * (1 + stoploss) = open_rate * 0.9975
    stoploss = -0.0025

    # No trailing stop of any kind.
    trailing_stop = False
    use_custom_stoploss = False

    # The 5m-context exit must be able to fire.
    use_exit_signal = True
    exit_profit_only = False           # exit on context loss even at a loss
    ignore_roi_if_entry_signal = False

    process_only_new_candles = True

    # Market orders so the backtester fills deterministically at open[T+1]
    # with no unfilled-limit-order edge cases.
    order_types = {
        "entry": "market",
        "exit": "market",
        "stoploss": "market",
        "stoploss_on_exchange": False,
    }

    # ---------------- warmup ----------------
    # Expressed in 1m candles (the strategy timeframe).
    #   5m EMA(50) : TA-Lib seeds with SMA(50) then recurses; roughly 5x the
    #                period is needed for convergence -> 250 completed 5m
    #                candles -> 250 * 5 = 1250 1m candles
    #   5m ROC(12) : 12 * 5 = 60 1m candles
    #   1m ROC(3)  : 3 1m candles
    #   merge/shift: +5 1m candles
    # 1500 yields 300 5m candles (6x the EMA50 period) with margin. Data
    # begins 2025-08-20, i.e. 7200 1m candles before the 2025-08-25 start.
    startup_candle_count: int = 1500

    # ---------------- frozen constants (DO NOT OPTIMISE) ----------------
    # TA-Lib ROC = 100 * (close - close[n]) / close[n]  -> PERCENT units.
    # TradingView ta.roc() uses the identical formula and identical units.
    # So 0.10 here means +0.10%, NOT +10%. Verified numerically by
    # user_data/tools/inspect_baseline.py --check-roc
    ROC_1M_PERIOD = 3
    ROC_1M_THRESHOLD = 0.10    # percent -> +0.10%
    EMA_5M_PERIOD = 50
    ROC_5M_PERIOD = 12

    # ---- PHASE 5 FILTER D constants (frozen, DO NOT OPTIMISE) ----
    ATR_PERIOD = 14
    ATR_SMA_PERIOD = 50
    ATR_EXPANSION_MULT = 1.10

    # Match TradingView's lookahead_off HTF publication boundary (see the
    # module docstring). EXECUTION SEMANTICS, NOT A STRATEGY PARAMETER.
    HTF_TV_BOUNDARY = True

    def __init__(self, config: dict) -> None:
        super().__init__(config)
        # Bucket in which this pair most recently CLOSED a trade, used to
        # enforce "no re-entry in the same 5m bucket as an exit". Written
        # only from confirm_trade_exit (a past event) and read only in
        # confirm_trade_entry, so it cannot look ahead.
        self._last_exit_bucket: dict[str, datetime] = {}

    # ------------------------------------------------------------------
    # helpers
    # ------------------------------------------------------------------
    @staticmethod
    def _bucket(ts: datetime) -> datetime:
        """Floor a timestamp to its 5-minute bucket (12:07 -> 12:05)."""
        return ts.replace(minute=(ts.minute // 5) * 5, second=0, microsecond=0)

    def informative_pairs(self):
        """Tell Freqtrade to load 5m data for every whitelisted pair."""
        pairs = self.dp.current_whitelist()
        return [(pair, self.informative_timeframe) for pair in pairs]

    # ------------------------------------------------------------------
    # indicators
    # ------------------------------------------------------------------
    def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        itf = self.informative_timeframe

        # ---------- 5m context, computed on the 5m dataframe itself ----------
        inf = self.dp.get_pair_dataframe(pair=metadata["pair"], timeframe=itf)

        # Standard TA-Lib implementations, no hand-rolled indicator maths.
        inf["ema_50"] = ta.EMA(inf, timeperiod=self.EMA_5M_PERIOD)
        inf["roc_12"] = ta.ROC(inf, timeperiod=self.ROC_5M_PERIOD)

        # Lookahead-safe merge.
        # Produces date_5m / open_5m / ... / ema_50_5m / roc_12_5m.
        dataframe = merge_informative_pair(
            dataframe, inf, self.timeframe, itf, ffill=True
        )

        htf_cols = [c for c in dataframe.columns if c.endswith(f"_{itf}")]

        if self.HTF_TV_BOUNDARY:
            # One extra 1m bar of delay so that a completed 5m candle first
            # becomes visible on the 1m candle that OPENS after it closed
            # (TradingView lookahead_off convention) instead of the 1m candle
            # that closes simultaneously with it (stock Freqtrade convention).
            # Shifting forward can only DELAY information -> never lookahead.
            dataframe[htf_cols] = dataframe[htf_cols].shift(1)

        # ---------- confirmed 5m bullish context ----------
        # Every value below comes from a 5m candle that closed strictly
        # before the current 1m candle closed, so no unfinished 5m data is
        # ever used.
        dataframe["ctx_valid_5m"] = (
            dataframe[f"ema_50_{itf}"].notna() & dataframe[f"roc_12_{itf}"].notna()
        )
        dataframe["bullish_ctx_5m"] = (
            dataframe["ctx_valid_5m"]
            # rule 1: confirmed 5m close > confirmed 5m EMA(50)
            & (dataframe[f"close_{itf}"] > dataframe[f"ema_50_{itf}"])
            # rule 2: confirmed 5m ROC(12) > 0
            & (dataframe[f"roc_12_{itf}"] > 0)
        )

        # ---------- 1m momentum ----------
        # rule 3: current COMPLETED 1m ROC(3) >= +0.10%
        dataframe["roc_3"] = ta.ROC(dataframe, timeperiod=self.ROC_1M_PERIOD)

        # 5-minute bucket of each 1m candle (12:00..12:04 -> 12:00).
        dataframe["bucket_5m"] = dataframe["date"].dt.floor("5min")

        # ---- PHASE 5 FILTER D: ATR expansion vs its SMA baseline ----
        # Standard TA-Lib ATR and SMA, no hand-rolled maths.
        dataframe["atr_14"] = ta.ATR(dataframe, timeperiod=self.ATR_PERIOD)
        dataframe["atr_base"] = ta.SMA(
            dataframe, timeperiod=self.ATR_SMA_PERIOD, price="atr_14"
        )

        return dataframe

    # ------------------------------------------------------------------
    # entries
    # ------------------------------------------------------------------
    def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        # Rules 1+2 (confirmed 5m context) and rule 3 (1m momentum).
        # UNCHANGED from the frozen baseline.
        baseline = dataframe["bullish_ctx_5m"] & (dataframe["roc_3"] >= self.ROC_1M_THRESHOLD)
        # Diagnostics only (never used for trading): lets the Phase 5 signal
        # funnel measure the filter pass rate on identical candles.
        dataframe["baseline_raw_entry"] = baseline.astype(int)

        # ============ PHASE 5 FILTER D (ATR EXPANSION) -- THE ONLY LOGIC CHANGE ============
        dataframe["filter_ok"] = (
            dataframe["atr_14"] >= self.ATR_EXPANSION_MULT * dataframe["atr_base"]
        )
        # ======================================================================

        raw = baseline & dataframe["filter_ok"]
        dataframe["raw_entry"] = raw.astype(int)

        # Rule 5: at most ONE entry opportunity per 5m bucket.
        # Keep the FIRST qualifying signal in each bucket and suppress the
        # rest. The cumsum runs within a bucket and is backward-looking only,
        # so it introduces no lookahead.
        rank_in_bucket = raw.astype(int).groupby(dataframe["bucket_5m"]).cumsum()
        first_of_bucket = raw & (rank_in_bucket == 1)

        # Rule 4 ("no existing long position") is enforced by the engine via
        # max_open_trades = 1, not here.
        dataframe.loc[first_of_bucket, ["enter_long", "enter_tag"]] = (
            1,
            "1m_momentum_long",
        )
        return dataframe

    # ------------------------------------------------------------------
    # exits
    # ------------------------------------------------------------------
    def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        # Exit when the CONFIRMED 5m bullish context disappears:
        #   confirmed 5m close <= confirmed 5m EMA50  OR  confirmed 5m ROC12 <= 0
        # Requires a valid context so that startup NaNs cannot fire a
        # spurious exit.
        context_lost = dataframe["ctx_valid_5m"] & (~dataframe["bullish_ctx_5m"])
        dataframe.loc[context_lost, ["exit_long", "exit_tag"]] = (1, "5m_context_exit")
        return dataframe

    # ------------------------------------------------------------------
    # "no re-entry in the same 5m bucket as an exit"
    # ------------------------------------------------------------------
    def confirm_trade_exit(
        self,
        pair: str,
        trade: Trade,
        order_type: str,
        amount: float,
        rate: float,
        time_in_force: str,
        exit_reason: str,
        current_time: datetime,
        **kwargs,
    ) -> bool:
        # current_time is the candle on which the exit FILLS.
        #   stoploss        : triggered intrabar on candle T, fills on T  -> bucket(T)
        #   5m_context_exit : signalled on T-1, fills on T (engine shift) -> bucket(T-1)
        # We record the bucket of the EVENT as TradingView would have seen
        # it, so that entries and exits are compared on the same
        # (signal-time) basis.
        event_time = current_time
        if exit_reason not in ("stop_loss", "stoploss", "liquidation"):
            event_time = current_time - timedelta(minutes=1)
        self._last_exit_bucket[pair] = self._bucket(event_time)
        return True    # never veto the exit itself

    def confirm_trade_entry(
        self,
        pair: str,
        order_type: str,
        amount: float,
        rate: float,
        time_in_force: str,
        current_time: datetime,
        entry_tag: str | None = None,
        side: str = "long",
        **kwargs,
    ) -> bool:
        # current_time is the FILL candle T+1; the signal was on candle T.
        signal_bucket = self._bucket(current_time - timedelta(minutes=1))
        if self._last_exit_bucket.get(pair) == signal_bucket:
            # A position already closed inside this 5m bucket, so the bucket
            # is spent: do not re-enter until the next bucket begins.
            return False
        return True
