Strategy03_RSI_Momentum

# pragma pylint: disable=missing-docstring, invalid-name, pointless-string-statement
# flake8: noqa: F401
# isort:skip_file
# --- Do not remove these imports ---
from datetime import datetime
from typing import Optional

import numpy as np
import pandas as pd
from pandas import DataFrame

from freqtrade.strategy import IStrategy, IntParameter, DecimalParameter

import pandas_ta as ta

# ---------------------------------------------------------------------------
# Strategy03_RSI_Momentum
# ---------------------------------------------------------------------------


class Strategy03_RSI_Momentum(IStrategy):
    """
    RSI Momentum Strategy — SPOT ONLY (Halal Algo Trading, Batch 1)
    ================================================================
    Timeframe : 4h
    Author    : Halal Algo Trading
    Version   : 1.0

    Overview
    --------
    Uses RSI (14-period) as the primary momentum oscillator, filtered by a
    100-period EMA to ensure trades are only taken in an established uptrend.
    A volume confirmation gate prevents entries during low-participation moves.

    Two entry modes are implemented:

    Mode 1 — Momentum Cross (default)
        RSI crosses above 50 AND price is above 100 EMA AND volume is above average
        → Classic breakout into positive momentum

    Mode 2 — Dip-and-Recover (stronger entry)
        In an uptrend (price > 100 EMA), RSI dips below 40 on the previous bar
        then recovers back above 40 on the current bar
        → Buy the pullback within an established bull trend

    Both modes are active simultaneously; whichever fires first generates the signal.

    Market Regime Adjustment
    ------------------------
    The strategy stores OB/OS thresholds as class attributes, allowing easy
    tuning for market conditions without touching the core logic:

        Bull market : OB = 80, OS = 40
        Bear market : OB = 60, OS = 20

    Exit Conditions (ANY triggers exit)
    ------------------------------------
    A. RSI crosses below 50   → momentum lost
    B. RSI reaches 75         → take profit before overbought extreme
    C. Bearish RSI divergence → price higher high, RSI lower high (5-bar window)

    Risk Management
    ---------------
    • Hard stoploss         : -8 %
    • Trailing stop         : enabled
    • Trailing stop positive: 2 %
    • ROI ladder            : 5 % @ open, 3 % @ 60 min, 2 % @ 120 min,
                              1 % @ 240 min
    • SPOT only — no short, no margin, no futures

    Backtest Reference
    ------------------
    Sharpe 0.80 | Return 358 % | Max DD -71 % | Win Rate 74 % | PF 1.15
    """

    # -------------------------------------------------------------------------
    # Freqtrade metadata
    # -------------------------------------------------------------------------
    INTERFACE_VERSION = 3

    strategy_name = "Strategy03_RSI_Momentum"

    timeframe = "4h"

    process_only_new_candles = True

    # 100 EMA + RSI warmup
    startup_candle_count: int = 110

    # -------------------------------------------------------------------------
    # Market regime thresholds
    # Adjust these to tune the strategy for bull or bear markets:
    #   Bull market : OB_THRESHOLD = 80, OS_THRESHOLD = 40
    #   Bear market : OB_THRESHOLD = 60, OS_THRESHOLD = 20
    # -------------------------------------------------------------------------
    OB_THRESHOLD: float = 80.0   # Overbought — used for exit and partial TP
    OS_THRESHOLD: float = 40.0   # Oversold   — used for dip-and-recover entry

    # RSI level at which "take partial profit" exit fires
    RSI_TAKE_PROFIT: float = 75.0

    # -------------------------------------------------------------------------
    # ROI
    # -------------------------------------------------------------------------
    minimal_roi = {
        "0":   0.05,
        "60":  0.03,
        "120": 0.02,
        "240": 0.01,
    }

    # -------------------------------------------------------------------------
    # Stoploss & Trailing Stop
    # -------------------------------------------------------------------------
    stoploss = -0.08
    trailing_stop = True
    trailing_stop_positive = 0.02
    trailing_stop_positive_offset = 0.03
    trailing_only_offset_is_reached = True

    # -------------------------------------------------------------------------
    # Misc
    # -------------------------------------------------------------------------
    can_short = False

    use_exit_signal = True
    exit_profit_only = False
    ignore_roi_if_entry_signal = False

    # -------------------------------------------------------------------------
    # Helper functions
    # -------------------------------------------------------------------------

    @staticmethod
    def crossed_above(s1: pd.Series, s2: pd.Series) -> pd.Series:
        """Return True on the bar where s1 crosses from below to above s2."""
        return (s1 > s2) & (s1.shift(1) <= s2.shift(1))

    @staticmethod
    def crossed_below(s1: pd.Series, s2: pd.Series) -> pd.Series:
        """Return True on the bar where s1 crosses from above to below s2."""
        return (s1 < s2) & (s1.shift(1) >= s2.shift(1))

    @staticmethod
    def bearish_rsi_divergence(
        price: pd.Series,
        rsi: pd.Series,
        lookback: int = 5,
    ) -> pd.Series:
        """
        Detect bearish RSI divergence over a rolling window.

        Bearish divergence: price makes a higher high while RSI makes a lower
        high over the `lookback` period — signals momentum exhaustion.

        Parameters
        ----------
        price    : close price series
        rsi      : RSI series
        lookback : window size for comparing the rolling high

        Returns
        -------
        Boolean series — True where divergence is detected
        """
        price_hh = price > price.shift(lookback).rolling(lookback).max()
        rsi_lh   = rsi   < rsi.shift(lookback).rolling(lookback).max()
        return price_hh & rsi_lh

    # -------------------------------------------------------------------------
    # Indicators
    # -------------------------------------------------------------------------

    def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        """
        Compute all technical indicators.

        Indicators added to `dataframe`
        --------------------------------
        rsi          — 14-period RSI
        ema_trend    — 100-period EMA (trend filter)
        volume_sma   — 20-period SMA of volume
        rsi_prev     — RSI value from the prior bar (used for dip-and-recover)
        bearish_div  — True where bearish RSI divergence is detected
        """

        # --- RSI (14-period) ---
        dataframe["rsi"] = ta.rsi(dataframe["close"], length=14)

        # --- 100-period EMA trend filter ---
        dataframe["ema_trend"] = ta.ema(dataframe["close"], length=100)

        # --- Volume filter: 20-period SMA ---
        dataframe["volume_sma"] = ta.sma(dataframe["volume"], length=20)

        # --- Prior bar RSI — used for the dip-and-recover entry pattern ---
        dataframe["rsi_prev"] = dataframe["rsi"].shift(1)

        # --- Bearish RSI divergence ---
        dataframe["bearish_div"] = self.bearish_rsi_divergence(
            dataframe["close"], dataframe["rsi"], lookback=5
        )

        # --- NaN Safety: convert any None to NaN so pandas comparisons work ---
        for col in dataframe.columns:
            if col not in ['open', 'high', 'low', 'close', 'volume', 'date']:
                try:
                    dataframe[col] = pd.to_numeric(dataframe[col], errors='coerce')
                except (ValueError, TypeError):
                    pass

        return dataframe

    # -------------------------------------------------------------------------
    # Entry Signal
    # -------------------------------------------------------------------------

    def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        """
        Generate LONG entry signals.

        Mode 1 — RSI Momentum Cross (all three conditions):
            1a. RSI (14) crosses above 50
            1b. Price is above 100 EMA         → uptrend confirmed
            1c. Volume > 20-period SMA          → participation confirmed

        Mode 2 — Dip-and-Recover within Uptrend:
            2a. Previous bar RSI was below OS_THRESHOLD (40 in bull markets)
            2b. Current bar RSI is above OS_THRESHOLD
            2c. Price is above 100 EMA         → still in uptrend

        A signal fires if EITHER mode is active.
        """

        # Mode 1: RSI crosses above 50 in uptrend with volume
        mode_1 = (
            self.crossed_above(dataframe["rsi"], pd.Series(50, index=dataframe.index))
            & (dataframe["close"] > dataframe["ema_trend"])
            & (dataframe["volume"] > dataframe["volume_sma"])
        )

        # Mode 2: RSI dips below OS_THRESHOLD then recovers above it (in uptrend)
        mode_2 = (
            (dataframe["rsi_prev"] < self.OS_THRESHOLD)  # prior bar: RSI dipped
            & (dataframe["rsi"] >= self.OS_THRESHOLD)     # current bar: RSI recovered
            & (dataframe["close"] > dataframe["ema_trend"])  # still in uptrend
        )

        dataframe.loc[
            (mode_1 | mode_2)
            & dataframe["rsi"].notna()
            & dataframe["ema_trend"].notna()
            & dataframe["volume_sma"].notna(),
            "enter_long",
        ] = 1

        return dataframe

    # -------------------------------------------------------------------------
    # Exit Signal
    # -------------------------------------------------------------------------

    def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        """
        Generate LONG exit signals.

        Exit is triggered when ANY condition is met:
        A. RSI crosses below 50               → momentum lost, exit
        B. RSI reaches RSI_TAKE_PROFIT (75)   → approaching overbought, take profit
        C. Bearish RSI divergence detected    → momentum fading while price rises

        Note: OB_THRESHOLD (80) is the full overbought level and can be used
        as an additional exit by uncommenting the relevant line below.
        """

        dataframe.loc[
            (
                # Condition A — RSI crosses below 50 (momentum lost)
                self.crossed_below(dataframe["rsi"], pd.Series(50, index=dataframe.index))
                # Condition B — RSI reaches take-profit level (75)
                | (dataframe["rsi"] >= self.RSI_TAKE_PROFIT)
                # [Optional] Full overbought exit at OB_THRESHOLD (80):
                # | (dataframe["rsi"] >= self.OB_THRESHOLD)
                # Condition C — Bearish RSI divergence
                | dataframe["bearish_div"]
            )
            & dataframe["rsi"].notna(),
            "exit_long",
        ] = 1

        return dataframe