MeanReversionStrategyV4

# pragma pylint: disable=missing-docstring, invalid-name, pointless-string-statement
# flake8: noqa: F401
# isort: skip_file
# --- Do not remove these imports ---
import numpy as np
import pandas as pd
from datetime import datetime, timedelta, timezone
from pandas import DataFrame
from typing import Dict, Optional, Union, Tuple

from freqtrade.strategy import (
    IStrategy,
    Trade,
    Order,
    PairLocks,
    informative,
    BooleanParameter,
    CategoricalParameter,
    DecimalParameter,
    IntParameter,
    RealParameter,
    timeframe_to_minutes,
    timeframe_to_next_date,
    timeframe_to_prev_date,
    merge_informative_pair,
    stoploss_from_absolute,
    stoploss_from_open,
    AnnotationType,
)

# --------------------------------
# Add your lib to import here
import talib.abstract as ta
from technical import qtpylib
from functools import reduce


class MeanReversionStrategyV4(IStrategy):
    """
    Mean Reversion Strategy with Short-Selling Capability
    
    Long entries: Price oversold (zscore < -threshold), expecting reversion to mean
    Short entries: Price overbought (zscore > +threshold), expecting reversion down to mean
    
    Both directions use:
    - Volume confirmation (spike required)
    - Trend filter (SMA slope)
    - RSI confirmation
    - Maximum hold time
    - Dynamic stop-loss based on ATR
    """
    
    # Strategy interface version
    INTERFACE_VERSION = 3

    # Can this strategy go short?
    can_short: bool = False

    # Minimal ROI designed for the strategy.
    minimal_roi = {
        "0": 0.11,   # 11% immediate (from hyperopt)
        "37": 0.045, # 4.5% after 37 candles
        "71": 0.013, # 1.3% after 71 candles
        "141": 0,    # 0% after 141 candles
    }

    # Optimal stoploss designed for the strategy.
    stoploss = -0.161  # -16.1% stop-loss (from hyperopt)

    # Trailing stoploss - DISABLED for mean reversion
    trailing_stop = False

    # Optimal timeframe for the strategy.
    timeframe = "5m"

    # Run "populate_indicators()" only for new candle.
    process_only_new_candles = True

    # These values can be overridden in the config.
    use_exit_signal = True
    exit_profit_only = True
    ignore_roi_if_entry_signal = True
    use_custom_stoploss = False  # Use fixed stop-loss

    # Number of candles the strategy requires before producing valid signals
    startup_candle_count: int = 200

    # ===== LONG PARAMETERS =====
    # Hyperoptable parameters for LONG entries
    buy_zscore = DecimalParameter(-3.0, -1.0, default=-2.0, space="buy", optimize=True, decimals=1)
    volume_threshold_long = DecimalParameter(1.2, 3.0, default=2.9, space="buy", optimize=True, decimals=1)
    rsi_buy = IntParameter(20, 35, default=24, space="buy", optimize=True)
    
    # ===== SHORT PARAMETERS =====
    # Hyperoptable parameters for SHORT entries
    short_zscore = DecimalParameter(1.0, 3.0, default=1.5, space="sell", optimize=True, decimals=1)
    volume_threshold_short = DecimalParameter(1.2, 3.0, default=2.9, space="sell", optimize=True, decimals=1)
    rsi_short = IntParameter(65, 85, default=76, space="sell", optimize=True)
    
    # ===== COMMON PARAMETERS =====
    # Moving average period for mean calculation
    ma_period = IntParameter(50, 150, default=92, space="buy", optimize=True)
    
    # Exit parameters
    sell_zscore_long = DecimalParameter(-0.5, 1.5, default=0.3, space="sell", optimize=True, decimals=1)
    rsi_sell_long = IntParameter(50, 80, default=80, space="sell", optimize=True)
    
    exit_zscore_short = DecimalParameter(-1.5, 0.5, default=-0.3, space="buy", optimize=True, decimals=1)
    rsi_exit_short = IntParameter(20, 50, default=24, space="buy", optimize=True)
    
    # Maximum hold time (candles)
    max_hold_candles = IntParameter(20, 100, default=86, space="sell", optimize=True)
    
    # SMA slope filter - different for long and short
    sma_slope_max_long = DecimalParameter(-1.0, 0.5, default=-0.09, space="buy", optimize=True, decimals=2)
    sma_slope_min_short = DecimalParameter(-0.5, 1.0, default=0.09, space="sell", optimize=True, decimals=2)
    
    # Volatility filter
    max_atr_pct = DecimalParameter(1.0, 5.0, default=3.0, space="buy", optimize=True, decimals=1)

    def populate_indicators(self, dataframe: DataFrame, metadata: Dict) -> DataFrame:
        """
        Adds several technical indicators to the dataframe.
        """
        # Calculate Simple Moving Average (mean)
        dataframe['sma'] = ta.SMA(dataframe, timeperiod=self.ma_period.value)
        
        # Calculate standard deviation for Z-score
        dataframe['std'] = dataframe['close'].rolling(window=self.ma_period.value).std()
        
        # Calculate Z-score: how many standard deviations from the mean
        dataframe['zscore'] = (dataframe['close'] - dataframe['sma']) / dataframe['std']
        
        # Calculate SMA slope (percentage change over last 12 candles)
        dataframe['sma_slope'] = (dataframe['sma'] - dataframe['sma'].shift(12)) / dataframe['sma'].shift(12) * 100
        
        # Bollinger Bands for additional confirmation
        bb = ta.BBANDS(dataframe, timeperiod=20, nbdevup=2.0, nbdevdn=2.0)
        dataframe['bb_lower'] = bb['lowerband']
        dataframe['bb_middle'] = bb['middleband']
        dataframe['bb_upper'] = bb['upperband']
        dataframe['bb_width'] = (dataframe['bb_upper'] - dataframe['bb_lower']) / dataframe['bb_middle']
        
        # Volume indicators
        dataframe['volume_sma'] = dataframe['volume'].rolling(window=20).mean()
        dataframe['volume_ratio'] = dataframe['volume'] / dataframe['volume_sma']
        
        # RSI
        dataframe['rsi'] = ta.RSI(dataframe, timeperiod=14)
        
        # Calculate distance from SMA as percentage
        dataframe['sma_distance_pct'] = (dataframe['close'] - dataframe['sma']) / dataframe['sma'] * 100
        
        # For visualization: flag oversold/overbought conditions
        dataframe['oversold'] = dataframe['zscore'] < self.buy_zscore.value
        dataframe['overbought'] = dataframe['zscore'] > self.short_zscore.value
        
        # Calculate average true range for volatility
        dataframe['atr'] = ta.ATR(dataframe, timeperiod=14)
        dataframe['atr_pct'] = dataframe['atr'] / dataframe['close'] * 100
        
        return dataframe

    def populate_entry_trend(self, dataframe: DataFrame, metadata: Dict) -> DataFrame:
        """
        Entry conditions for both LONG and SHORT trades.
        """
        # ===== LONG ENTRY CONDITIONS =====
        long_conditions = []
        
        # Core condition: Z-score indicates oversold
        long_conditions.append(dataframe['zscore'] < self.buy_zscore.value)
        
        # Safety: price should be below SMA (we're buying low)
        long_conditions.append(dataframe['close'] < dataframe['sma'])
        
        # Limit distance to avoid extreme deviations (max -5%)
        long_conditions.append(dataframe['sma_distance_pct'] > -5.0)
        
        # SMA slope filter: avoid strong downtrends for longs
        long_conditions.append(dataframe['sma_slope'] > self.sma_slope_max_long.value)
        
        # Volume confirmation: need volume spike
        long_conditions.append(dataframe['volume_ratio'] > self.volume_threshold_long.value)
        
        # RSI confirmation
        long_conditions.append(dataframe['rsi'] < self.rsi_buy.value)
        
        # Volatility filter: avoid extreme volatility
        long_conditions.append(dataframe['atr_pct'] < self.max_atr_pct.value)
        
        # Combine long conditions
        if long_conditions:
            dataframe.loc[
                reduce(lambda x, y: x & y, long_conditions),
                'enter_long'] = 1
        
        # ===== SHORT ENTRY CONDITIONS =====
        short_conditions = []
        
        # Core condition: Z-score indicates overbought
        short_conditions.append(dataframe['zscore'] > self.short_zscore.value)
        
        # Safety: price should be above SMA (we're shorting high)
        short_conditions.append(dataframe['close'] > dataframe['sma'])
        
        # Limit distance to avoid extreme deviations (max +5%)
        short_conditions.append(dataframe['sma_distance_pct'] < 5.0)
        
        # SMA slope filter: avoid strong uptrends for shorts
        short_conditions.append(dataframe['sma_slope'] < self.sma_slope_min_short.value)
        
        # Volume confirmation: need volume spike
        short_conditions.append(dataframe['volume_ratio'] > self.volume_threshold_short.value)
        
        # RSI confirmation (overbought)
        short_conditions.append(dataframe['rsi'] > self.rsi_short.value)
        
        # Volatility filter: avoid extreme volatility
        short_conditions.append(dataframe['atr_pct'] < self.max_atr_pct.value)
        
        # Combine short conditions
        if short_conditions:
            dataframe.loc[
                reduce(lambda x, y: x & y, short_conditions),
                'enter_short'] = 1
        
        return dataframe

    def populate_exit_trend(self, dataframe: DataFrame, metadata: Dict) -> DataFrame:
        """
        Exit conditions for both LONG and SHORT trades.
        """
        # ===== LONG EXIT CONDITIONS =====
        long_exit_conditions = []
        
        # Core condition: Z-score returned to mean
        long_exit_conditions.append(dataframe['zscore'] > self.sell_zscore_long.value)
        
        # Alternative: price crossed above SMA (returned to mean)
        long_exit_conditions.append(dataframe['close'] > dataframe['sma'])
        
        # RSI exit (overbought)
        long_exit_conditions.append(dataframe['rsi'] > self.rsi_sell_long.value)
        
        # Price reached Bollinger Band middle band
        long_exit_conditions.append(dataframe['close'] > dataframe['bb_middle'])
        
        # Combine long exit conditions with OR logic
        if long_exit_conditions:
            dataframe.loc[
                reduce(lambda x, y: x | y, long_exit_conditions),
                'exit_long'] = 1
        
        # ===== SHORT EXIT CONDITIONS =====
        short_exit_conditions = []
        
        # Core condition: Z-score returned to mean
        short_exit_conditions.append(dataframe['zscore'] < self.exit_zscore_short.value)
        
        # Alternative: price crossed below SMA (returned to mean)
        short_exit_conditions.append(dataframe['close'] < dataframe['sma'])
        
        # RSI exit (oversold)
        short_exit_conditions.append(dataframe['rsi'] < self.rsi_exit_short.value)
        
        # Price reached Bollinger Band middle band
        short_exit_conditions.append(dataframe['close'] < dataframe['bb_middle'])
        
        # Combine short exit conditions with OR logic
        if short_exit_conditions:
            dataframe.loc[
                reduce(lambda x, y: x | y, short_exit_conditions),
                'exit_short'] = 1
        
        return dataframe

    def custom_exit(self, pair: str, trade: Trade, current_time: datetime,
                    current_rate: float, current_profit: float, **kwargs) -> Optional[str]:
        """
        Custom exit logic for maximum hold time.
        """
        # Calculate how many candles we've held
        trade_duration = (current_time - trade.open_date_utc).total_seconds() / 60
        candle_minutes = timeframe_to_minutes(self.timeframe)
        candles_held = trade_duration / candle_minutes
        
        # Maximum hold period check
        if candles_held >= self.max_hold_candles.value:
            return 'max_hold_time_reached'
        
        # If we have small profit and have held enough candles, take it
        if candles_held >= 10 and current_profit > 0.005:  # 0.5% profit
            return 'quick_profit'
        
        return None

    def populate_buy(self, dataframe: DataFrame, metadata: Dict) -> DataFrame:
        # For compatibility with older freqtrade versions
        dataframe = self.populate_entry_trend(dataframe, metadata)
        dataframe.loc[dataframe['enter_long'] == 1, 'buy'] = 1
        return dataframe

    def populate_sell(self, dataframe: DataFrame, metadata: Dict) -> DataFrame:
        # For compatibility with older freqtrade versions
        dataframe = self.populate_exit_trend(dataframe, metadata)
        dataframe.loc[dataframe['exit_long'] == 1, 'sell'] = 1
        return dataframe