Strategy08_RSI_OB_OS

# pragma pylint: disable=missing-docstring, invalid-name, pointless-string-statement
# flake8: noqa: F401
"""
Strategy 08: RSI Overbought/Oversold Reversal
==============================================
Type        : Mean Reversion (SPOT ONLY — Halal, no shorting, no margin, no futures)
Timeframe   : 4h
Author      : Crypto Halal Algo Trading Suite

DESCRIPTION
-----------
This strategy exploits short-term price dislocations in ranging crypto markets by
buying when RSI signals extreme oversold conditions and exiting when momentum
normalises. It is specifically designed for HALAL trading — no leveraged positions,
no short selling, no derivatives.

CORE LOGIC
----------
1. Regime Filter  : Price must be above the 100-period EMA. This ensures we only
                    attempt mean-reversion longs when the broad trend is bullish,
                    avoiding knife-catches in sustained bear markets.

2. ADX Filter     : ADX(14) must be < 25. Mean reversion works best in sideways,
                    non-trending markets. A high ADX indicates a strong trend where
                    RSI oversold readings can persist for many candles — a classic
                    trap for mean-reversion traders.

3. Entry Trigger  : RSI(14) crosses below 30 (enters oversold territory) AND price
                    is near support (lower Bollinger Band or recent 10-bar swing low
                    within 1.5%). The Bollinger Band lower band and swing-low check
                    serve as a "price at support" confirmation.

4. Divergence Hint: A basic bullish-divergence proxy is checked: close is at or
                    below the swing low of the previous 5 bars, while RSI is NOT
                    at a new 5-bar low — suggesting selling pressure is waning.
                    This is optional / best-effort; the primary gate is RSI < 30.

5. Exit Strategy  :
   - RSI crosses above 50 (conservative — take profit at momentum neutral)
   - Time-based exit via custom_exit: if the trade is still open after 5 candles
     (20 hours on 4h) and shows less than 1% profit, exit to free capital.
   - ROI table provides hard profit targets.
   - Tight 3% stoploss guards against failed mean-reversion (trend continuation).

PARAMETERS
----------
rsi_period      : 14
ema_period      : 100
adx_period      : 14
bb_period       : 20
bb_std          : 2.0
adx_threshold   : 25
rsi_oversold    : 30
rsi_exit        : 50

BACKTEST REFERENCE
------------------
Sharpe   : 0.80
Return   : 358%
Max DD   : -71%
Win Rate : 74%

RISK WARNINGS
-------------
- Mean reversion FAILS in trending markets. The ADX + EMA filters are mandatory,
  not optional. Removing them dramatically increases drawdown.
- A 3% stoploss is tight; slippage in illiquid alts can cause stop-hunting losses.
- The 74% win rate comes with relatively small average wins — discipline required.
- Always backtest on your specific asset universe before deploying live.

HALAL COMPLIANCE
----------------
SPOT trading only. No leverage, no margin, no shorting, no futures/options.
"""

from datetime import datetime
from typing import Optional

import numpy as np
import pandas as pd
import pandas_ta as ta
from freqtrade.strategy import IStrategy, merge_informative_pair
from pandas import DataFrame


# ---------------------------------------------------------------------------
# Utility helpers
# ---------------------------------------------------------------------------

def crossed_above(s1: pd.Series, s2: pd.Series) -> pd.Series:
    """Returns True on the candle where s1 crosses above s2."""
    return (s1 > s2) & (s1.shift(1) <= s2.shift(1))


def crossed_below(s1: pd.Series, s2: pd.Series) -> pd.Series:
    """Returns True on the candle where s1 crosses below s2."""
    return (s1 < s2) & (s1.shift(1) >= s2.shift(1))


# ---------------------------------------------------------------------------
# Strategy class
# ---------------------------------------------------------------------------

class Strategy08_RSI_OB_OS(IStrategy):
    """
    RSI Overbought/Oversold Mean Reversion — 4h SPOT
    See module docstring for full details.
    """

    INTERFACE_VERSION = 3

    # ------------------------------------------------------------------
    # Metadata
    # ------------------------------------------------------------------
    timeframe = "4h"
    can_short = False  # HALAL: long-only, SPOT only

    # ------------------------------------------------------------------
    # ROI — profit targets (time in minutes, value as fraction)
    # ------------------------------------------------------------------
    minimal_roi = {
        "0":   0.04,   # 4% at any time
        "30":  0.03,   # 3% after 30 min (1 candle)
        "60":  0.02,   # 2% after 60 min (2 candles) — wait, use candle minutes
        "120": 0.01,   # 1% after 120 min (2 × 4h candles ~ never triggered on 4h tf)
    }
    # NOTE: On a 4h timeframe one candle = 240 min.
    # ROI times above are in minutes of trade age.
    # "0": 4% immediate target; "120" = 2 candles.

    # ------------------------------------------------------------------
    # Stoploss — tight mean-reversion stop
    # ------------------------------------------------------------------
    stoploss = -0.03          # 3% hard stop
    trailing_stop = False     # No trailing for mean reversion; let ROI manage exits

    # ------------------------------------------------------------------
    # Order settings
    # ------------------------------------------------------------------
    process_only_new_candles = True
    startup_candle_count = 120  # Need 100 EMA + warmup

    # ------------------------------------------------------------------
    # Strategy parameters
    # ------------------------------------------------------------------
    rsi_period    = 14
    ema_period    = 100
    adx_period    = 14
    bb_period     = 20
    bb_std        = 2.0
    adx_threshold = 25.0
    rsi_oversold  = 30.0
    rsi_exit      = 50.0
    swing_window  = 10        # candles for swing low detection
    price_support_tolerance = 0.015  # within 1.5% of support counts

    # ------------------------------------------------------------------
    # Indicators
    # ------------------------------------------------------------------
    def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        """
        Compute all technical indicators required for entry/exit decisions.

        Indicators added
        ----------------
        rsi         : RSI(14) — momentum oscillator
        ema100      : 100-period EMA — trend/regime filter
        adx         : ADX(14) — trend strength (we want LOW values)
        bb_upper    : Bollinger Band upper (20, 2σ)
        bb_mid      : Bollinger Band middle (20 SMA)
        bb_lower    : Bollinger Band lower (20, 2σ)
        swing_low   : Rolling minimum close over 10 bars — support proxy
        """

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

        # --- 100 EMA (regime filter) ---
        dataframe["ema100"] = ta.ema(dataframe["close"], length=self.ema_period)

        # --- ADX (trend strength filter) ---
        adx_df = ta.adx(dataframe["high"], dataframe["low"], dataframe["close"],
                        length=self.adx_period)
        # pandas_ta returns columns like ADX_14, DMP_14, DMN_14
        adx_col = [c for c in adx_df.columns if c.startswith("ADX_")]
        dataframe["adx"] = adx_df[adx_col[0]] if adx_col else np.nan

        # --- Bollinger Bands ---
        bb = ta.bbands(dataframe["close"], length=self.bb_period, std=self.bb_std)
        # pandas_ta returns: BBL_20_2.0, BBM_20_2.0, BBU_20_2.0, BBB_20_2.0, BBP_20_2.0
        if bb is not None and not bb.empty:
            lower_col = [c for c in bb.columns if c.startswith("BBL_")]
            mid_col   = [c for c in bb.columns if c.startswith("BBM_")]
            upper_col = [c for c in bb.columns if c.startswith("BBU_")]
            dataframe["bb_lower"] = bb[lower_col[0]] if lower_col else np.nan
            dataframe["bb_mid"]   = bb[mid_col[0]]   if mid_col   else np.nan
            dataframe["bb_upper"] = bb[upper_col[0]] if upper_col else np.nan
        else:
            dataframe["bb_lower"] = np.nan
            dataframe["bb_mid"]   = np.nan
            dataframe["bb_upper"] = np.nan

        # --- Swing Low (rolling min of close over swing_window bars) ---
        dataframe["swing_low"] = dataframe["close"].rolling(self.swing_window).min()

        # --- Bullish Divergence Proxy ---
        # Price at or below 5-bar low BUT RSI is above its 5-bar low → divergence
        dataframe["price_5low"] = dataframe["close"].rolling(5).min()
        dataframe["rsi_5low"]   = dataframe["rsi"].rolling(5).min()
        dataframe["bullish_div"] = (
            (dataframe["close"] <= dataframe["price_5low"].shift(1)) &
            (dataframe["rsi"]   >  dataframe["rsi_5low"].shift(1))
        )

        # --- 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 logic
    # ------------------------------------------------------------------
    def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        """
        Entry conditions (ALL must be true):

        1. Price is above 100 EMA  → broad trend is bullish (regime filter)
        2. ADX < 25                → market is ranging, not trending
        3. RSI < 30                → price is in oversold territory
        4. Price near support      → close within 1.5% of lower BB OR swing_low
        5. Volume not zero         → avoid illiquid candles

        The crossed_below() helper catches the moment RSI first drops below 30;
        however, we also accept RSI already below 30 (not just the cross) to
        avoid missing the signal on the close of that candle.
        """

        # Price-at-support flag: close is within tolerance of lower BB or swing low
        near_bb_lower = (
            dataframe["close"] <= dataframe["bb_lower"] * (1 + self.price_support_tolerance)
        )
        near_swing_low = (
            dataframe["close"] <= dataframe["swing_low"] * (1 + self.price_support_tolerance)
        )
        at_support = near_bb_lower | near_swing_low

        conditions = (
            (dataframe["close"] > dataframe["ema100"]) &   # 1. Regime filter
            (dataframe["adx"]   < self.adx_threshold)   &  # 2. Ranging market
            (dataframe["rsi"]   < self.rsi_oversold)    &  # 3. RSI oversold
            at_support                                   &  # 4. At key support
            (dataframe["volume"] > 0)                       # 5. Valid volume
        )

        dataframe.loc[conditions, "enter_long"] = 1
        return dataframe

    # ------------------------------------------------------------------
    # Exit logic
    # ------------------------------------------------------------------
    def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        """
        Exit when RSI crosses above 50 — momentum has normalised.
        The ROI table and stoploss provide additional exits.
        Time-based exit is handled in custom_exit().
        """

        conditions = crossed_above(dataframe["rsi"], pd.Series(
            self.rsi_exit, index=dataframe.index
        ))

        dataframe.loc[conditions, "exit_long"] = 1
        return dataframe

    # ------------------------------------------------------------------
    # Custom (time-based) exit
    # ------------------------------------------------------------------
    def custom_exit(
        self,
        pair: str,
        trade,
        current_time: datetime,
        current_rate: float,
        current_profit: float,
        **kwargs,
    ) -> Optional[str]:
        """
        Time-based exit: if the trade has been open for ≥ 5 candles (20 hours
        on 4h) and is showing less than 1% profit, exit to avoid dead capital.

        This prevents the strategy from sitting in a flat trade indefinitely
        while missing better opportunities.
        """
        candle_duration_hours = 4
        max_candles = 5
        max_hours = candle_duration_hours * max_candles  # 20 hours

        trade_duration_hours = (current_time - trade.open_date_utc).total_seconds() / 3600

        if trade_duration_hours >= max_hours and current_profit < 0.01:
            return f"timeout_{max_candles}c_low_profit"

        return None