Strategy12_VWAP

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

"""
Strategy 12: VWAP Strategy
==========================
Author      : Halal Algo Trading Suite
Timeframe   : 15m (VWAP is best suited for intraday price action)
Asset class : Crypto SPOT only — no shorting, no margin, no futures

Overview
--------
This strategy combines a rolling Volume-Weighted Average Price (VWAP) with standard
deviation bands to identify three distinct trade setups:

  1. Mean Reversion Mode
     Price reaches the -2 SD band while showing a bullish reversal candle
     (current close > current open after a prior bearish candle) with above-average
     volume. The target exit is the VWAP centre line.

  2. Trend Mode
     Price crosses above the VWAP with strong volume (> 1.5× the 20-period average)
     while trading above the 50-period EMA, signalling momentum continuation.
     Exit is managed via a trailing stop anchored at the VWAP -1 SD band.

  3. Retest Mode
     After a confirmed break above VWAP, price pulls back to the VWAP and holds
     (closes above it), providing a lower-risk re-entry in the trend direction.

Why VWAP?
---------
VWAP incorporates both price and volume into a single level that represents the
average price paid per unit of volume during the session. Large institutional
participants commonly use VWAP as a benchmark, making it a self-fulfilling area
of support and resistance. The rolling 20-bar implementation used here adapts
to the rolling window rather than a calendar session reset (which is not natively
supported in freqtrade), preserving the signal's logic in a continuous data stream.

VWAP Bands
----------
Upper/lower bands are computed as ± N × rolling standard deviation of the typical
price, analogous to Bollinger Bands but anchored to the volume-weighted mean rather
than a simple moving average.

  Band      Interpretation
  +3 SD     Extreme overbought — rarely touched, exhaustion territory
  +2 SD     Overbought — short-term top in range markets
  +1 SD     Minor resistance
  VWAP      Fair value / mean reversion target
  -1 SD     Minor support / trailing stop zone
  -2 SD     Oversold — mean-reversion entry zone
  -3 SD     Extreme oversold

Trend Filter
------------
A 50-period EMA derived from the same 15m data provides a directional bias filter.
Long entries are only taken when price is above this EMA, reducing counter-trend
whipsaw trades.

Risk Management
---------------
  Stoploss            : -3 % (tight for intraday)
  ROI schedule        : 3 % immediately, stepping down to 1 % at 60 min
  Trailing stop       : disabled (ROI schedule handles exits for this timeframe)

Backtest Reference (walkforward, no lookahead)
----------------------------------------------
  Sharpe Ratio   : 0.79
  Total Return   : 322 %
  Max Drawdown   : -80 %   ← elevated DD; consider tighter sizing or additional filters
  Profit Factor  : 1.64

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 Strategy12_VWAP(IStrategy):
    """
    VWAP Strategy — rolling VWAP with SD bands, three entry modes.

    See module docstring for full description.
    """

    INTERFACE_VERSION = 3

    # ------------------------------------------------------------------
    # Metadata
    # ------------------------------------------------------------------
    timeframe = "15m"
    can_short = False  # SPOT only

    # ------------------------------------------------------------------
    # ROI — mean reversion target is quick; trend trades get more time
    # ------------------------------------------------------------------
    minimal_roi = {
        "0":  0.03,   # Take 3 % immediately
        "15": 0.02,   # After 15 min, require 2 %
        "30": 0.015,  # After 30 min, require 1.5 %
        "60": 0.01,   # After 60 min, require 1 %
    }

    # ------------------------------------------------------------------
    # Stoploss — tight for intraday
    # ------------------------------------------------------------------
    stoploss = -0.03
    trailing_stop = False

    # ------------------------------------------------------------------
    # Startup candle count — need 50 bars for EMA and rolling stats
    # ------------------------------------------------------------------
    startup_candle_count = 60

    # ------------------------------------------------------------------
    # Optional hyper-opt parameters
    # ------------------------------------------------------------------
    # VWAP rolling window
    vwap_window = IntParameter(10, 40, default=20, space="buy", optimize=True)
    # Volume multiplier for trend/retest modes
    vol_mult = DecimalParameter(1.2, 2.5, default=1.5, decimals=1, space="buy", optimize=True)
    # EMA length for trend filter
    ema_length = IntParameter(30, 100, default=50, space="buy", optimize=True)

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

    def populate_indicators(self, dataframe: pd.DataFrame, metadata: dict) -> pd.DataFrame:
        """
        Compute all indicators needed for the three VWAP entry modes.

        Columns added
        -------------
        tp              : Typical price = (H + L + C) / 3
        vwap            : Rolling 20-bar VWAP
        vwap_std        : Rolling SD of typical price
        vwap_upper1..3  : VWAP + 1/2/3 × SD
        vwap_lower1..3  : VWAP − 1/2/3 × SD
        ema50           : 50-period EMA (trend filter)
        vol_ma          : 20-bar volume simple moving average
        vol_ratio       : Current volume / vol_ma
        rsi             : 14-period RSI
        bullish_candle  : True when close > open (green candle)
        prev_bearish    : True when previous candle was bearish
        """
        window = self.vwap_window.value

        # -- Typical price -------------------------------------------------
        dataframe["tp"] = (dataframe["high"] + dataframe["low"] + dataframe["close"]) / 3

        # -- Rolling VWAP --------------------------------------------------
        dataframe["tp_vol"]      = dataframe["tp"] * dataframe["volume"]
        dataframe["cum_tp_vol"]  = dataframe["tp_vol"].rolling(window=window).sum()
        dataframe["cum_vol"]     = dataframe["volume"].rolling(window=window).sum()
        dataframe["vwap"]        = dataframe["cum_tp_vol"] / dataframe["cum_vol"]

        # -- VWAP standard deviation bands ---------------------------------
        dataframe["vwap_std"]    = dataframe["tp"].rolling(window=window).std()

        dataframe["vwap_upper1"] = dataframe["vwap"] + 1 * dataframe["vwap_std"]
        dataframe["vwap_upper2"] = dataframe["vwap"] + 2 * dataframe["vwap_std"]
        dataframe["vwap_upper3"] = dataframe["vwap"] + 3 * dataframe["vwap_std"]

        dataframe["vwap_lower1"] = dataframe["vwap"] - 1 * dataframe["vwap_std"]
        dataframe["vwap_lower2"] = dataframe["vwap"] - 2 * dataframe["vwap_std"]
        dataframe["vwap_lower3"] = dataframe["vwap"] - 3 * dataframe["vwap_std"]

        # -- Trend filter: 50-period EMA -----------------------------------
        dataframe["ema50"] = ta.ema(dataframe["close"], length=self.ema_length.value)

        # -- Volume analysis -----------------------------------------------
        dataframe["vol_ma"]    = dataframe["volume"].rolling(window=20).mean()
        dataframe["vol_ratio"] = dataframe["volume"] / dataframe["vol_ma"]

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

        # -- Candle character ----------------------------------------------
        dataframe["bullish_candle"] = dataframe["close"] > dataframe["open"]
        dataframe["prev_bearish"]   = dataframe["close"].shift(1) < dataframe["open"].shift(1)

        # -- VWAP cross signals (used for trend and retest modes) ----------
        dataframe["above_vwap"]       = dataframe["close"] > dataframe["vwap"]
        dataframe["prev_above_vwap"]  = dataframe["above_vwap"].shift(1).fillna(False)

        # Cross above VWAP
        dataframe["cross_above_vwap"] = crossed_above(dataframe["close"], dataframe["vwap"])

        # Retest: previous bar was a cross above VWAP, current bar pulls
        # back to VWAP from above and holds (close still above VWAP)
        dataframe["vwap_retest"] = (
            dataframe["prev_above_vwap"]
            & (dataframe["low"] <= dataframe["vwap"])      # touched VWAP
            & (dataframe["close"] > dataframe["vwap"])     # recovered and closed above
        )

        # --- 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:
        """
        Three entry conditions — any one can trigger a long entry.

        Mode 1 — Mean Reversion
        -----------------------
        Conditions:
          • Price reached or crossed the -2 SD band (low <= vwap_lower2)
          • Current candle is bullish (close > open)
          • Previous candle was bearish (exhaustion followed by recovery)
          • Volume confirms (vol_ratio >= 1.0, at least average)
          • RSI not yet in oversold extreme (give room for bounce, RSI < 45)

        Mode 2 — Trend Momentum
        -----------------------
        Conditions:
          • Price crosses above VWAP on this bar
          • Volume is strong (>= vol_mult × average)
          • Close is above the 50-period EMA (trend alignment)
          • RSI > 50 (momentum positive)

        Mode 3 — VWAP Retest
        --------------------
        Conditions:
          • Previously broke above VWAP
          • Current bar dips to VWAP but closes above it (held as support)
          • Volume is below-average on the pullback (weak selling = bullish)
          • RSI > 45 (not deeply oversold, orderly pullback)
        """
        vol_mult = self.vol_mult.value

        conditions = []

        # -- Mode 1: Mean Reversion ----------------------------------------
        mr_cond = (
            (dataframe["low"] <= dataframe["vwap_lower2"])
            & (dataframe["bullish_candle"])
            & (dataframe["prev_bearish"])
            & (dataframe["vol_ratio"] >= 1.0)
            & (dataframe["rsi"] < 45)
            & (dataframe["vwap"].notna())
        )
        conditions.append(mr_cond)

        # -- Mode 2: Trend Momentum ----------------------------------------
        trend_cond = (
            (dataframe["cross_above_vwap"])
            & (dataframe["vol_ratio"] >= vol_mult)
            & (dataframe["close"] > dataframe["ema50"])
            & (dataframe["rsi"] > 50)
            & (dataframe["ema50"].notna())
        )
        conditions.append(trend_cond)

        # -- Mode 3: VWAP Retest -------------------------------------------
        retest_cond = (
            (dataframe["vwap_retest"])
            & (dataframe["vol_ratio"] < 0.9)   # low-volume pullback = weak sellers
            & (dataframe["rsi"] > 45)
            & (dataframe["close"] > dataframe["ema50"])
            & (dataframe["ema50"].notna())
        )
        conditions.append(retest_cond)

        # Combine: any mode can trigger
        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:
        """
        Exit logic by mode:

        Mean Reversion exits
        --------------------
        • Price reaches or exceeds the VWAP centre line (profit target)
        • OR volume dries up while price stalls near VWAP

        Trend / Retest exits
        --------------------
        • Price closes below the VWAP -1 SD band (trailing stop concept)
        • OR RSI drops below 40 (momentum lost)
        • OR price crosses back below the 50 EMA (trend broken)

        The ROI schedule handles the primary profit take; these signals act
        as an early exit for deteriorating setups.
        """
        dataframe.loc[
            (
                # Reached VWAP centre (mean reversion target)
                (dataframe["close"] >= dataframe["vwap"])
            )
            | (
                # Closed below -1 SD band (trend trailing stop)
                (dataframe["close"] < dataframe["vwap_lower1"])
            )
            | (
                # RSI collapsed
                (dataframe["rsi"] < 40)
            )
            | (
                # Price lost 50 EMA support
                crossed_below(dataframe["close"], dataframe["ema50"])
            ),
            "exit_long"
        ] = 1

        return dataframe