Strategy15_KeltnerChannel

# pragma pylint: disable=missing-docstring, invalid-name, pointless-string-statement
# flake8: noqa: F401

"""
Strategy 15: Keltner Channel
==============================
Author      : Halal Algo Trading Suite
Timeframe   : 4h (6h is a viable alternative for fewer, higher-quality trades)
Asset class : Crypto SPOT only — no shorting, no margin, no futures

Overview
--------
Keltner Channels (KC) are volatility-based envelopes set above and below an
Exponential Moving Average. Unlike Bollinger Bands (which use standard
deviation), KC uses ATR to define the band width. This makes KC less susceptible
to sudden volatility spikes and more stable as a structural reference.

  Centre line : EMA(20)
  Upper band  : EMA(20) + 2.0 × ATR(14)
  Lower band  : EMA(20) − 2.0 × ATR(14)

Price spending significant time outside the bands is statistically rare and
typically resolves with a return to the mean — the basis for Mode 1.
When price closes decisively outside the band and momentum agrees, it signals
the start of a trend expansion — the basis for Mode 2.

Two Operating Modes
--------------------

  Mode 1 — Mean Reversion
  ────────────────────────
  Trade the statistical tendency of price to snap back to the EMA after
  reaching an extreme. Entry conditions:
    • Price closes BELOW the lower Keltner band
    • RSI < 40 (oversold, but not yet extreme — still shows demand)
    • Current candle is bullish (close > open), signalling absorption of selling
    • ADX < 25 — confirming the market is in a RANGE, not a trend
      (mean reversion fails in strong trends)

  Exit: Price returns to the KC centre EMA. The ROI schedule handles cases
  where price doesn't quite reach the centre.

  Mode 2 — Momentum / Trend Following
  ──────────────────────────────────────
  Trade the breakout when KC squeeze resolves into expansion. Entry conditions:
    • Price closes ABOVE the upper Keltner band (expansion started)
    • EMA(9) > EMA(21) — short-term momentum is bullish
    • Close > EMA(50) — medium-term trend is bullish
    • RSI > 55 — momentum confirming the move (not overbought extreme)

  Exit: Price closes below the KC centre EMA — the trend engine has stalled.

Why ADX for Mode Separation?
-----------------------------
ADX (Average Directional Index) measures trend strength without direction.
  ADX < 20-25 : Range / consolidation environment
  ADX > 25    : Trending environment
  ADX > 40    : Strong trend

Using ADX < 25 as a filter for mean reversion and implicitly allowing higher ADX
for momentum trades ensures each mode is applied in its appropriate regime.

Risk Parameters
---------------
  Stoploss             : -25.4 % (from hyperopt; broad to allow KC swings)
  Trailing stop        : True
  Trailing stop positive offset: 3 % (once 3 % profit achieved, trail activates)
  ROI schedule         : 8 % immediately, 4 % after 4h, 3 % after 8h, 1 % after 16h

Backtest Reference (walkforward, no lookahead)
----------------------------------------------
  Sharpe Ratio   : 0.85
  Total Return   : 257 %
  Max Drawdown   : -36 %

Halal Compliance
----------------
  ✓ SPOT trading only
  ✓ No leverage or margin
  ✓ No short selling
  ✓ No interest-bearing instruments
"""

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, DecimalParameter, IntParameter
from freqtrade.persistence import Trade


# ---------------------------------------------------------------------------
# Helper utilities
# ---------------------------------------------------------------------------

def crossed_above(s1: pd.Series, s2: pd.Series) -> pd.Series:
    """Return True on bars 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:
    """Return True on bars where s1 crosses below s2."""
    return (s1 < s2) & (s1.shift(1) >= s2.shift(1))


# ---------------------------------------------------------------------------
# Strategy
# ---------------------------------------------------------------------------

class Strategy15_KeltnerChannel(IStrategy):
    """
    Keltner Channel Strategy — dual-mode trading using KC bands for both
    mean reversion (range markets) and momentum (trending markets).

    See module docstring for full description.
    """

    INTERFACE_VERSION = 3

    # ------------------------------------------------------------------
    # Metadata
    # ------------------------------------------------------------------
    timeframe = "4h"
    can_short = False  # SPOT only

    # ------------------------------------------------------------------
    # ROI — accommodate both fast MR bounces and slower trend runs
    # ------------------------------------------------------------------
    minimal_roi = {
        "0":   0.08,  # 8 % immediately
        "60":  0.04,  # 4 % after 4h (1 candle × 4h = 60 min in minutes? No: 60=min)
        "120": 0.03,  # 3 % after 8h
        "240": 0.01,  # 1 % after 16h
    }
    # Note: ROI keys are in minutes. 60 = 1 hour, 240 = 4 hours.

    # ------------------------------------------------------------------
    # Stoploss — from hyperopt (-25.4 %)
    # ------------------------------------------------------------------
    stoploss = -0.254
    trailing_stop = True
    trailing_stop_positive = 0.03          # Trail once +3 % profit reached
    trailing_stop_positive_offset = 0.05   # Offset: trail kicks in at 5 %, stops at 3 % drop from high
    trailing_only_offset_is_reached = True

    # ------------------------------------------------------------------
    # Startup candle count — need 50 bars for EMA(50) and ADX
    # ------------------------------------------------------------------
    startup_candle_count = 60

    # ------------------------------------------------------------------
    # Hyper-opt parameters
    # ------------------------------------------------------------------
    # Keltner Channel EMA length
    kc_ema_length = IntParameter(10, 30, default=20, space="buy", optimize=True)
    # Keltner Channel ATR scalar (band multiplier)
    kc_scalar = DecimalParameter(1.0, 3.0, default=2.0, decimals=1, space="buy", optimize=True)
    # RSI threshold for mean-reversion entry (oversold)
    rsi_mr_max = IntParameter(30, 50, default=40, space="buy", optimize=True)
    # RSI threshold for momentum entry
    rsi_mom_min = IntParameter(50, 65, default=55, space="buy", optimize=True)
    # ADX max for mean reversion (range filter)
    adx_mr_max = IntParameter(15, 30, default=25, space="buy", optimize=True)
    # EMA lengths for momentum confirmation
    ema9_length  = IntParameter(5, 15, default=9, space="buy", optimize=True)
    ema21_length = IntParameter(15, 30, default=21, space="buy", optimize=True)
    ema50_length = IntParameter(30, 70, default=50, space="buy", optimize=True)

    # ------------------------------------------------------------------
    # Indicator population
    # ------------------------------------------------------------------

    def populate_indicators(self, dataframe: pd.DataFrame, metadata: dict) -> pd.DataFrame:
        """
        Compute Keltner Channel, RSI, ADX, and EMA indicators.

        Columns added
        -------------
        kc_upper        : Upper Keltner band
        kc_mid          : Centre EMA of Keltner Channel
        kc_lower        : Lower Keltner band
        rsi             : 14-period RSI
        adx             : 14-period ADX (trend strength)
        ema9            : 9-period EMA (short-term momentum)
        ema21           : 21-period EMA (medium momentum)
        ema50           : 50-period EMA (trend filter)
        bullish_candle  : True when close > open (green candle)
        above_upper_kc  : True when close > kc_upper
        below_lower_kc  : True when close < kc_lower
        """
        kc_len    = self.kc_ema_length.value
        kc_scalar = self.kc_scalar.value

        # -- Keltner Channel -----------------------------------------------
        # pandas_ta returns a DataFrame with columns:
        #   KCLe_{length}_{scalar}  (lower)
        #   KCBe_{length}_{scalar}  (basis / mid)
        #   KCUe_{length}_{scalar}  (upper)
        # Column order: [lower, mid, upper] (index 0, 1, 2)
        kc = ta.kc(
            dataframe["high"],
            dataframe["low"],
            dataframe["close"],
            length=kc_len,
            scalar=kc_scalar,
            mamode="ema",
        )

        if kc is not None and not kc.empty:
            # Sort columns alphabetically: KCBe < KCLe < KCUe — use explicit name matching
            kc_cols = kc.columns.tolist()
            lower_col = [c for c in kc_cols if c.startswith("KCL")][0]
            mid_col   = [c for c in kc_cols if c.startswith("KCB")][0]
            upper_col = [c for c in kc_cols if c.startswith("KCU")][0]

            dataframe["kc_lower"] = kc[lower_col]
            dataframe["kc_mid"]   = kc[mid_col]
            dataframe["kc_upper"] = kc[upper_col]
        else:
            # Manual fallback if pandas_ta kc fails
            atr_vals = ta.atr(
                dataframe["high"], dataframe["low"], dataframe["close"], length=14
            )
            ema_vals = ta.ema(dataframe["close"], length=kc_len)
            dataframe["kc_mid"]   = ema_vals
            dataframe["kc_upper"] = ema_vals + kc_scalar * atr_vals
            dataframe["kc_lower"] = ema_vals - kc_scalar * atr_vals

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

        # -- ADX (trend strength, direction-agnostic) ----------------------
        adx_result = ta.adx(dataframe["high"], dataframe["low"], dataframe["close"], length=14)
        if adx_result is not None and not adx_result.empty:
            adx_col = [c for c in adx_result.columns if c.startswith("ADX")][0]
            dataframe["adx"] = adx_result[adx_col]
        else:
            dataframe["adx"] = np.nan

        # -- EMAs for momentum confirmation --------------------------------
        dataframe["ema9"]  = ta.ema(dataframe["close"], length=self.ema9_length.value)
        dataframe["ema21"] = ta.ema(dataframe["close"], length=self.ema21_length.value)
        dataframe["ema50"] = ta.ema(dataframe["close"], length=self.ema50_length.value)

        # -- Candle character ----------------------------------------------
        dataframe["bullish_candle"] = dataframe["close"] > dataframe["open"]

        # -- KC band position flags ----------------------------------------
        dataframe["above_upper_kc"] = dataframe["close"] > dataframe["kc_upper"]
        dataframe["below_lower_kc"] = dataframe["close"] < dataframe["kc_lower"]

        # --- 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 signals
    # ------------------------------------------------------------------

    def populate_entry_trend(self, dataframe: pd.DataFrame, metadata: dict) -> pd.DataFrame:
        """
        Two operating modes:

        Mode 1 — Mean Reversion (range market)
        ───────────────────────────────────────
        All of the following must be true:
          • Close < lower KC band (price at statistical extreme)
          • RSI < rsi_mr_max (oversold, demand emerging)
          • Bullish candle (close > open) — absorption of sellers
          • ADX < adx_mr_max — confirming range (not trending) environment

        Mode 2 — Momentum (trending market)
        ────────────────────────────────────
        All of the following must be true:
          • Close > upper KC band (expansion above band)
          • EMA(9) > EMA(21) — short-term bull momentum
          • Close > EMA(50) — trend aligned
          • RSI > rsi_mom_min — momentum confirming, not overbought
        """
        conditions = []

        # -- Mode 1: Mean Reversion ----------------------------------------
        mr_cond = (
            (dataframe["below_lower_kc"])
            & (dataframe["rsi"] < self.rsi_mr_max.value)
            & (dataframe["bullish_candle"])
            & (dataframe["adx"] < self.adx_mr_max.value)
            & dataframe["kc_lower"].notna()
            & dataframe["adx"].notna()
        )
        conditions.append(mr_cond)

        # -- Mode 2: Momentum ----------------------------------------------
        mom_cond = (
            (dataframe["above_upper_kc"])
            & (dataframe["ema9"] > dataframe["ema21"])
            & (dataframe["close"] > dataframe["ema50"])
            & (dataframe["rsi"] > self.rsi_mom_min.value)
            & dataframe["kc_upper"].notna()
            & dataframe["ema50"].notna()
        )
        conditions.append(mom_cond)

        # Combine: either mode triggers
        dataframe.loc[
            pd.concat(conditions, axis=1).any(axis=1),
            "enter_long"
        ] = 1

        return dataframe

    # ------------------------------------------------------------------
    # Exit signals
    # ------------------------------------------------------------------

    def populate_exit_trend(self, dataframe: pd.DataFrame, metadata: dict) -> pd.DataFrame:
        """
        Mode-specific exits:

        Mean Reversion exit:
          • Price returns to the KC centre EMA — profit target reached.
          • RSI recovers above 55 — reversion likely complete.

        Momentum exit:
          • Price closes below the KC centre EMA — trend engine has stalled.
          • EMA(9) crosses below EMA(21) — short-term momentum reversal.

        Both modes also exit on:
          • RSI drops below 35 — significant momentum reversal (failed trade).

        The trailing stop (activates at +3 % profit) handles much of the
        exit management for momentum trades. These signals provide early
        exits for deteriorating setups.
        """
        # Mean reversion target: price returned to centre
        mr_exit = (
            (dataframe["close"] >= dataframe["kc_mid"])
            & (dataframe["rsi"] > 55)
        )

        # Momentum exit: centre broken or EMA crossover
        mom_exit = (
            (dataframe["close"] < dataframe["kc_mid"])
            | crossed_below(dataframe["ema9"], dataframe["ema21"])
        )

        # Universal momentum collapse
        rsi_exit = dataframe["rsi"] < 35

        dataframe.loc[
            mr_exit | mom_exit | rsi_exit,
            "exit_long"
        ] = 1

        return dataframe