VolumeTrabar_v3

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

from freqtrade.strategy import (BooleanParameter, CategoricalParameter, DecimalParameter,
                                IStrategy, IntParameter, merge_informative_pair)

# --------------------------------
# Add your lib to import here
import talib.abstract as ta
from freqtrade.persistence import Trade
from functools import reduce


class VolumeTrabar_v3(IStrategy):
    """
    Strategy berdasarkan indikator Volume Trabar.id dari TradingView
    
    Konsep:
    - Menggunakan multiple Bollinger Bands dengan periode berbeda
    - Menghitung rank berdasarkan kondisi oversold/overbought
    - Rank negatif = oversold (buy signal)
    - Rank positif = overbought (sell signal)
    - Multi-timeframe analysis untuk konfirmasi signal
    """

    # Strategy interface version - allow new iterations of the strategy interface.
    # Check the documentation or the Sample strategy to get the latest version.
    INTERFACE_VERSION = 3

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

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

    # Minimal ROI designed for the strategy.
    minimal_roi = {
        "0": 0.06,
        "30": 0.05,
        "60": 0.03,
        "120": 0.01
    }

    # Optimal stoploss designed for the strategy.
    stoploss = -0.04

    # Trailing stoploss
    trailing_stop = True
    trailing_only_offset_is_reached = True
    trailing_stop_positive = 0.015
    trailing_stop_positive_offset = 0.025

    # 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 = False
    ignore_roi_if_entry_signal = False

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

    # Strategy parameters - Calculation Settings
    len1 = IntParameter(10, 20, default=14, space="buy", optimize=True)
    len2 = IntParameter(35, 50, default=42, space="buy", optimize=True)
    len3 = IntParameter(150, 200, default=168, space="buy", optimize=True)
    len4 = IntParameter(300, 400, default=336, space="buy", optimize=False)
    len5 = IntParameter(600, 750, default=672, space="buy", optimize=False)
    len6 = IntParameter(1200, 1400, default=1344, space="buy", optimize=False)
    len7 = IntParameter(1900, 2100, default=2016, space="buy", optimize=False)
    len8 = IntParameter(3900, 4100, default=4032, space="buy", optimize=False)
    
    # Standard deviation multiplier
    std_mult = DecimalParameter(1.5, 3.0, default=2.0, space="buy", optimize=True)
    
    # MA Type
    ma_type = CategoricalParameter(["SMA", "EMA"], default="SMA", space="buy", optimize=False)
    
    # Thresholds
    short_term_threshold = IntParameter(1, 3, default=1, space="buy", optimize=False)
    mid_term_threshold = IntParameter(3, 6, default=4, space="buy", optimize=True)
    long_term_threshold = IntParameter(5, 8, default=6, space="buy", optimize=True)
    
    # Entry/Exit thresholds
    buy_rank_threshold = IntParameter(-10, -3, default=-6, space="buy", optimize=True)
    sell_rank_threshold = IntParameter(3, 10, default=6, space="sell", optimize=True)
    
    # Multi-timeframe ready signal
    enable_mtf_ready = BooleanParameter(default=True, space="buy", optimize=False)
    ready_threshold = IntParameter(3, 7, default=5, space="buy", optimize=True)

    # Informative timeframes for MTF analysis (must be >= base timeframe)
    informative_timeframes = ['15m', '30m', '1h']

    def informative_pairs(self):
        """
        Define additional, informative pair/interval combinations to be cached from the exchange.
        """
        pairs = self.dp.current_whitelist()
        informative_pairs = []
        for tf in self.informative_timeframes:
            informative_pairs.extend([(pair, tf) for pair in pairs])
        return informative_pairs

    def calculate_ma(self, series: pd.Series, length: int, ma_type: str) -> pd.Series:
        """Calculate moving average based on type"""
        if ma_type == "SMA":
            return series.rolling(window=length).mean()
        elif ma_type == "EMA":
            return series.ewm(span=length, adjust=False).mean()
        else:  # RMA (same as EMA for simplicity)
            return series.ewm(span=length, adjust=False).mean()

    def check_bands(self, dataframe: DataFrame, length: int, mult: float, ma_type: str):
        """
        Check if price is outside Bollinger Bands
        Returns oversold and overbought conditions
        """
        ma = self.calculate_ma(dataframe['close'], length, ma_type)
        std = dataframe['close'].rolling(window=length).std()
        
        lower_band = ma - (std * mult)
        upper_band = ma + (std * mult)
        
        is_oversold = (dataframe['close'] < lower_band) & (~pd.isna(ma)) & (~pd.isna(std))
        is_overbought = (dataframe['close'] > upper_band) & (~pd.isna(ma)) & (~pd.isna(std))
        
        return is_oversold, is_overbought

    def calculate_rank(self, dataframe: DataFrame) -> pd.Series:
        """
        Calculate the volume rank based on multiple Bollinger Bands
        Negative rank = oversold (buy signal)
        Positive rank = overbought (sell signal)
        """
        ma_type_val = self.ma_type.value
        std_mult_val = self.std_mult.value
        
        # Check all timeframes
        oversold1, overbought1 = self.check_bands(dataframe, self.len1.value, std_mult_val, ma_type_val)
        oversold2, overbought2 = self.check_bands(dataframe, self.len2.value, std_mult_val, ma_type_val)
        oversold3, overbought3 = self.check_bands(dataframe, self.len3.value, std_mult_val, ma_type_val)
        oversold4, overbought4 = self.check_bands(dataframe, self.len4.value, std_mult_val, ma_type_val)
        oversold5, overbought5 = self.check_bands(dataframe, self.len5.value, std_mult_val, ma_type_val)
        oversold6, overbought6 = self.check_bands(dataframe, self.len6.value, std_mult_val, ma_type_val)
        oversold7, overbought7 = self.check_bands(dataframe, self.len7.value, std_mult_val, ma_type_val)
        oversold8, overbought8 = self.check_bands(dataframe, self.len8.value, std_mult_val, ma_type_val)
        
        # Calculate oversold rank (sequential check)
        oversold_rank = pd.Series(0, index=dataframe.index)
        condition = oversold1 & oversold2
        oversold_rank = np.where(condition, 2, oversold_rank)
        
        condition = condition & oversold3
        oversold_rank = np.where(condition, 3, oversold_rank)
        
        condition = condition & oversold4
        oversold_rank = np.where(condition, 4, oversold_rank)
        
        condition = condition & oversold5
        oversold_rank = np.where(condition, 5, oversold_rank)
        
        condition = condition & oversold6
        oversold_rank = np.where(condition, 6, oversold_rank)
        
        condition = condition & oversold7
        oversold_rank = np.where(condition, 7, oversold_rank)
        
        condition = condition & oversold8
        oversold_rank = np.where(condition, 8, oversold_rank)
        
        # Calculate overbought rank (sequential check)
        overbought_rank = pd.Series(0, index=dataframe.index)
        condition = overbought1 & overbought2
        overbought_rank = np.where(condition, 2, overbought_rank)
        
        condition = condition & overbought3
        overbought_rank = np.where(condition, 3, overbought_rank)
        
        condition = condition & overbought4
        overbought_rank = np.where(condition, 4, overbought_rank)
        
        condition = condition & overbought5
        overbought_rank = np.where(condition, 5, overbought_rank)
        
        condition = condition & overbought6
        overbought_rank = np.where(condition, 6, overbought_rank)
        
        condition = condition & overbought7
        overbought_rank = np.where(condition, 7, overbought_rank)
        
        condition = condition & overbought8
        overbought_rank = np.where(condition, 8, overbought_rank)
        
        # Final rank calculation
        rank = pd.Series(0, index=dataframe.index)
        
        # If only oversold, rank is negative
        rank = np.where((oversold_rank > 0) & (overbought_rank == 0), -oversold_rank, rank)
        
        # If only overbought, rank is positive
        rank = np.where((overbought_rank > 0) & (oversold_rank == 0), overbought_rank, rank)
        
        # If both, choose the stronger one
        rank = np.where((oversold_rank > overbought_rank) & (oversold_rank > 0) & (overbought_rank > 0), 
                       -oversold_rank, rank)
        rank = np.where((overbought_rank > oversold_rank) & (oversold_rank > 0) & (overbought_rank > 0), 
                       overbought_rank, rank)
        
        return pd.Series(rank, index=dataframe.index)

    def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        """
        Populate indicators that will be used in the Buy, Sell, and Exit strategy
        """
        # Calculate main rank for current timeframe
        dataframe['rank'] = self.calculate_rank(dataframe)
        
        # Add informative timeframes for MTF analysis
        for tf in self.informative_timeframes:
            informative = self.dp.get_pair_dataframe(pair=metadata['pair'], timeframe=tf)
            informative['rank'] = self.calculate_rank(informative)
            
            # Merge informative data
            dataframe = merge_informative_pair(dataframe, informative, self.timeframe, tf, ffill=True)
        
        # MTF Ready Signal
        if self.enable_mtf_ready.value and len(self.informative_timeframes) >= 3:
            tf1 = self.informative_timeframes[0]
            tf2 = self.informative_timeframes[1]
            tf3 = self.informative_timeframes[2]
            
            # Ready Buy: semua MTF rank <= -ready_threshold
            dataframe['mtf_ready_buy'] = (
                (dataframe[f'rank_{tf1}'] <= -self.ready_threshold.value) &
                (dataframe[f'rank_{tf2}'] <= -self.ready_threshold.value) &
                (dataframe[f'rank_{tf3}'] <= -self.ready_threshold.value)
            )
            
            # Ready Sell: semua MTF rank >= ready_threshold
            dataframe['mtf_ready_sell'] = (
                (dataframe[f'rank_{tf1}'] >= self.ready_threshold.value) &
                (dataframe[f'rank_{tf2}'] >= self.ready_threshold.value) &
                (dataframe[f'rank_{tf3}'] >= self.ready_threshold.value)
            )
        else:
            dataframe['mtf_ready_buy'] = False
            dataframe['mtf_ready_sell'] = False
        
        # Add volume for confirmation
        dataframe['volume_ma'] = dataframe['volume'].rolling(window=20).mean()
        
        return dataframe

    def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        """
        Based on TA indicators, populates the entry signal for the given dataframe
        """
        conditions = []
        
        # Basic condition: rank is below buy threshold (oversold)
        conditions.append(dataframe['rank'] <= self.buy_rank_threshold.value)
        
        # Volume confirmation
        conditions.append(dataframe['volume'] > 0)
        
        # Optional: MTF Ready confirmation
        if self.enable_mtf_ready.value:
            # Either MTF ready buy OR strong rank on current timeframe
            conditions.append(
                (dataframe['mtf_ready_buy'] == True) | 
                (dataframe['rank'] <= (self.buy_rank_threshold.value - 2))
            )
        
        if conditions:
            dataframe.loc[
                reduce(lambda x, y: x & y, conditions),
                'enter_long'] = 1

        return dataframe

    def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        """
        Based on TA indicators, populates the exit signal for the given dataframe
        """
        conditions = []
        
        # Basic condition: rank is above sell threshold (overbought)
        conditions.append(dataframe['rank'] >= self.sell_rank_threshold.value)
        
        # Optional: MTF Ready confirmation
        if self.enable_mtf_ready.value:
            # Either MTF ready sell OR strong rank on current timeframe
            conditions.append(
                (dataframe['mtf_ready_sell'] == True) | 
                (dataframe['rank'] >= (self.sell_rank_threshold.value + 2))
            )
        
        if conditions:
            dataframe.loc[
                reduce(lambda x, y: x & y, conditions),
                'exit_long'] = 1

        return dataframe

    def custom_stoploss(self, pair: str, trade: 'Trade', current_time: datetime,
                       current_rate: float, current_profit: float, **kwargs) -> float:
        """
        Custom stoploss logic, returning the new distance relative to current_rate
        """
        # After 20 minutes, allow stoploss to be closer
        if current_time - trade.open_date_utc > timedelta(minutes=20):
            return -0.05
        
        return self.stoploss