AlexStrategyFinalV9

import numpy as np
import pandas_ta as pta
from typing import Optional
import math
import freqtrade.vendor.qtpylib.indicators as qtpylib
import logging
from functools import reduce
from typing import Dict
from datetime import timedelta, datetime, timezone
from freqtrade.persistence import Trade
from pandas import DataFrame, Series
import talib.abstract as ta
from pandas import DataFrame

from freqtrade.strategy import DecimalParameter, IntParameter, IStrategy, BooleanParameter
from technical.pivots_points import pivots_points

from scipy.signal import argrelextrema

import logging
from functools import reduce

import numpy as np
import pandas as pd
import talib.abstract as ta
from pandas import DataFrame
from technical import qtpylib

from freqtrade.exchange.exchange_utils import *
from freqtrade.strategy import IStrategy, RealParameter
from technical.pivots_points import pivots_points

logger = logging.getLogger(__name__)

"""
AlexStrategyFinalV9 - FreqAI Strategy with Advanced Technical Analysis

This strategy combines multiple technical indicators with FreqAI predictions
and includes dynamic features like:
- Multi-indicator normalized scoring
- Market regime filtering
- Volatility-based adjustments
- Dynamic trailing stops and leverage
"""

class AlexStrategyFinalV9(IStrategy):
    """
    This is an example strategy that uses the LSTMRegressor model to predict the target score.
    Use at your own risk.
    This is a simple example strategy and should be used for educational purposes only.
    """

    # Hyperspace parameters:
    buy_params = {
        "threshold_buy": 0.59453,
        "w0": 0.54347,
        "w1": 0.82226,
        "w2": 0.56675,
        "w3": 0.77918,
        "w4": 0.98488,
        "w5": 0.31368,
        "w6": 0.75916,
        "w7": 0.09226,
        "w8": 0.85667,
    }

    sell_params = {
        "threshold_sell": 0.80573,
    }

    # ROI table:
    minimal_roi = {
        "0": 0.239,
        "79": 0.058,
        "231": 0.029,
        "543": 0
    }

    # Stoploss:
    stoploss = -0.305  # Were letting the model decide when to sell

    # Trailing stop:
    #trailing_stop = False
    #trailing_only_offset_is_reached = False

    timeframe = "1h"
    can_short = True
    use_exit_signal = True
    process_only_new_candles = True
    exit_profit_only = False
    exit_profit_offset = 0.01
    ignore_roi_if_entry_signal = False

    startup_candle_count = 100
    leverage_value = 10.0

    threshold_buy = RealParameter(-1, 1, default=0, space='buy')
    threshold_sell = RealParameter(-1, 1, default=0, space='sell')

    # Weights for calculating the aggregate score - normalized to sum to 1
    w0 = RealParameter(0, 1, default=0.10, space='buy')  # moving average (normalized_ma)
    w1 = RealParameter(0, 1, default=0.15, space='buy')  # MACD (normalized_macd)
    w2 = RealParameter(0, 1, default=0.10, space='buy')  # Rate of Change (ROC)
    w3 = RealParameter(0, 1, default=0.15, space='buy')  # RSI (normalized_rsi)
    w4 = RealParameter(0, 1, default=0.10, space='buy')  # Bollinger Band width
    w5 = RealParameter(0, 1, default=0.10, space='buy')  # CCI (normalized_cci)
    w6 = RealParameter(0, 1, default=0.10, space='buy')  # OBV (normalized_obv)
    w7 = RealParameter(0, 1, default=0.05, space='buy')  # ATR (normalized_atr)
    w8 = RealParameter(0, 1, default=0.10, space='buy')  # Stochastic Oscillator (normalized_stoch)

    plot_config = {
        'main_plot': {
            'ma_100': {'color': 'blue'},
            'bb_upperband': {'color': 'orange'},
            'bb_middleband': {'color': 'yellow'},
            'bb_lowerband': {'color': 'orange'},
        },
        'subplots': {
            "Target": {
                'T': {'color': 'blue'},
                '&-target': {'color': 'green'},
            },
            "Scores": {
                'S': {'color': 'purple'},
            },
            "Filters": {
                'R': {'color': 'red'},
                'R2': {'color': 'brown'},
                'V': {'color': 'green'},
                'V2': {'color': 'black'},
            },
            "Normalized Indicators": {
                'normalized_ma': {'color': 'blue'},
                'normalized_macd': {'color': 'red'},
                'normalized_rsi': {'color': 'green'},
            }
        }
    }

    def feature_engineering_expand_all(self, dataframe: DataFrame, period: int,
                                       metadata: Dict, **kwargs):
        """
        Feature engineering for the AI model using rolling windows
        """
        # Basic technical indicators with rolling periods
        dataframe["%-cci-period"] = ta.CCI(dataframe, timeperiod=20)
        dataframe["%-rsi-period"] = ta.RSI(dataframe, timeperiod=10)
        dataframe["%-momentum-period"] = ta.MOM(dataframe, timeperiod=4)
        dataframe['%-ma-period'] = ta.SMA(dataframe, timeperiod=10)
        dataframe['%-macd-period'], dataframe['%-macdsignal-period'], dataframe['%-macdhist-period'] = ta.MACD(
            dataframe['close'], slowperiod=12, fastperiod=26)
        dataframe['%-roc-period'] = ta.ROC(dataframe, timeperiod=2)

        # Bollinger Bands
        bollinger = qtpylib.bollinger_bands(
            qtpylib.typical_price(dataframe), window=period, stds=2.2
        )
        dataframe["bb_lowerband-period"] = bollinger["lower"]
        dataframe["bb_middleband-period"] = bollinger["mid"]
        dataframe["bb_upperband-period"] = bollinger["upper"]
        dataframe["%-bb_width-period"] = (
            dataframe["bb_upperband-period"]
            - dataframe["bb_lowerband-period"]
        ) / dataframe["bb_middleband-period"]
        dataframe["%-close-bb_lower-period"] = (
            dataframe["close"] / dataframe["bb_lowerband-period"]
        )

        return dataframe

    def feature_engineering_expand_basic(self, dataframe: DataFrame, metadata: Dict, **kwargs):
        """
        Basic feature engineering
        """
        dataframe["%-pct-change"] = dataframe["close"].pct_change()
        dataframe["%-raw_volume"] = dataframe["volume"]
        dataframe["%-raw_price"] = dataframe["close"]

        return dataframe

    def feature_engineering_standard(self, dataframe: DataFrame, metadata: Dict, **kwargs):
        """
        Standard feature engineering with time-based features
        """
        dataframe['date'] = pd.to_datetime(dataframe['date'])
        dataframe["%-day_of_week"] = dataframe["date"].dt.dayofweek
        dataframe["%-hour_of_day"] = dataframe["date"].dt.hour
        return dataframe

    def set_freqai_targets(self, dataframe: DataFrame, metadata: Dict, **kwargs) -> DataFrame:
        """
        Calculate the target variable for FreqAI prediction
        This is the core of the strategy's logic
        """
        
        # Calculate basic technical indicators
        dataframe['ma'] = ta.SMA(dataframe, timeperiod=10)
        dataframe['roc'] = ta.ROC(dataframe, timeperiod=2)
        dataframe['macd'], dataframe['macdsignal'], dataframe['macdhist'] = ta.MACD(dataframe['close'], slowperiod=12,
                                                                                    fastperiod=26)
        dataframe['momentum'] = ta.MOM(dataframe, timeperiod=4)
        dataframe['rsi'] = ta.RSI(dataframe, timeperiod=10)
        bollinger = ta.BBANDS(dataframe, timeperiod=20)
        dataframe['bb_upperband'] = bollinger['upperband']
        dataframe['bb_middleband'] = bollinger['middleband']
        dataframe['bb_lowerband'] = bollinger['lowerband']
        dataframe['cci'] = ta.CCI(dataframe, timeperiod=20)
        dataframe['stoch'] = ta.STOCH(dataframe)['slowk']
        dataframe['atr'] = ta.ATR(dataframe, timeperiod=14)
        dataframe['obv'] = ta.OBV(dataframe)
        dataframe['ma_100'] = ta.SMA(dataframe, timeperiod=100)

        # Step 1: Normalize Indicators:
        # Why? Normalizing the indicators will make them comparable and allow us to assign weights to them.
        # How? We will calculate the z-score of each indicator by subtracting the rolling mean and dividing by the
        # rolling standard deviation. This will give us a normalized value that is centered around 0 with a standard
        # deviation of 1.
        dataframe['normalized_stoch'] = (dataframe['stoch'] - dataframe['stoch'].rolling(window=14).mean()) / dataframe['stoch'].rolling(window=14).std()
        dataframe['normalized_atr'] = (dataframe['atr'] - dataframe['atr'].rolling(window=14).mean()) / dataframe['atr'].rolling(window=14).std()
        dataframe['normalized_obv'] = (dataframe['obv'] - dataframe['obv'].rolling(window=14).mean()) / dataframe['obv'].rolling(window=14).std()
        dataframe['normalized_ma'] = (dataframe['close'] - dataframe['close'].rolling(window=10).mean()) / dataframe['close'].rolling(window=10).std()
        dataframe['normalized_macd'] = (dataframe['macd'] - dataframe['macd'].rolling(window=26).mean()) / dataframe['macd'].rolling(window=26).std()
        dataframe['normalized_roc'] = (dataframe['roc'] - dataframe['roc'].rolling(window=2).mean()) / dataframe['roc'].rolling(window=2).std()
        dataframe['normalized_momentum'] = (dataframe['momentum'] - dataframe['momentum'].rolling(window=4).mean()) / \
                                             dataframe['momentum'].rolling(window=4).std()
        dataframe['normalized_rsi'] = (dataframe['rsi'] - dataframe['rsi'].rolling(window=10).mean()) / dataframe['rsi'].rolling(window=10).std()
        dataframe['normalized_bb_width'] = (dataframe['bb_upperband'] - dataframe['bb_lowerband']).rolling(
            window=20).mean() / (dataframe['bb_upperband'] - dataframe['bb_lowerband']).rolling(window=20).std()
        dataframe['normalized_cci'] = (dataframe['cci'] - dataframe['cci'].rolling(window=20).mean()) / dataframe['cci'].rolling(window=20).std()

        # Dynamic Weights Adjustment
        # Calculate trend strength as the absolute difference between MA and close price
        trend_strength = abs(dataframe['ma'] - dataframe['close'])
        # Calculate rolling mean and stddev once to avoid redundancy
        rolling_mean = trend_strength.rolling(window=14).mean()
        rolling_stddev = trend_strength.rolling(window=14).std()
        # Calculate a more dynamic strong trend threshold
        strong_trend_threshold = rolling_mean + 1.5 * rolling_stddev
        # Determine strong trend condition
        is_strong_trend = trend_strength > strong_trend_threshold
        # Apply dynamic weight adjustment, could also consider a more gradual adjustment
        dataframe['w_momentum'] = self.w3.value * (1 + 0.5 * (trend_strength / strong_trend_threshold))
        # Optional: Clip the w_momentum values to prevent extreme cases
        dataframe['w_momentum'] = dataframe['w_momentum'].clip(lower=self.w3.value, upper=self.w3.value * 2)

        # Step 2: Calculate aggregate score S
        w = [self.w0.value, self.w1.value, self.w2.value, self.w3.value, self.w4.value, self.w5.value, self.w6.value, self.w7.value, self.w8.value]
      
        dataframe['S'] = w[0] * dataframe['normalized_ma'] + \
                         w[1] * dataframe['normalized_macd'] + \
                         w[2] * dataframe['normalized_roc'] + \
                         w[3] * dataframe['normalized_rsi'] + \
                         w[4] * dataframe['normalized_bb_width'] + \
                         w[5] * dataframe['normalized_cci'] + \
                         dataframe['w_momentum'] * dataframe['normalized_momentum'] + \
                         self.w8.value * dataframe['normalized_stoch'] + \
                         self.w7.value * dataframe['normalized_atr'] + \
                         self.w6.value * dataframe['normalized_obv']

        # Step 3: Market Regime Filter R
        dataframe['R'] = 0
        dataframe.loc[dataframe['close'] > dataframe['bb_upperband'], 'R'] = 1
        dataframe.loc[dataframe['close'] < dataframe['bb_lowerband'], 'R'] = -1
        buffer_pct = 0.01  # 1% buffer
        
        dataframe['R2'] = np.where(dataframe['close'] > dataframe['ma_100'] * (1 + buffer_pct), 1, 
                                    np.where(dataframe['close'] < dataframe['ma_100'] * (1 - buffer_pct), -1, np.nan))

        # Step 4: Volatility Adjustment V
        # EXPLANATION: Calculate the Bollinger Band width and assign it to V. The Bollinger Band width is the
        # difference between the upper and lower Bollinger Bands divided by the middle Bollinger Band. The idea is
        # that when the Bollinger Bands are wide, the market is volatile, and when the Bollinger Bands are narrow,
        # the market is less volatile. So we are using the Bollinger Band width as a measure of volatility. You can
        # use other indicators to measure volatility as well. For example, you can use the ATR (Average True Range)

        bb_width = (dataframe['bb_upperband'] - dataframe['bb_lowerband']) / dataframe['bb_middleband']
        dataframe['V_mean'] = 1 / (bb_width + 1e-8)  # Avoid division by zero

        # ATR-based Volatility Measure
        dataframe['V2_mean'] = 1 / (dataframe['atr'] + 1e-8)  # Avoid division by zero

        # Rolling window size for adaptive normalization
        rolling_window = 50

        # Normalize V_mean using a rolling window
        mean_v = dataframe['V_mean'].rolling(window=rolling_window).mean()
        std_v = dataframe['V_mean'].rolling(window=rolling_window).std()
        dataframe['V_norm'] = (dataframe['V_mean'] - mean_v) / std_v
        dataframe['V_norm'] = dataframe['V_norm'].fillna(0) 

        # Normalize V2_mean using a rolling window
        mean_v2 = dataframe['V2_mean'].rolling(window=rolling_window).mean()
        std_v2 = dataframe['V2_mean'].rolling(window=rolling_window).std()
        dataframe['V2_norm'] = (dataframe['V2_mean'] - mean_v2) / std_v2
        dataframe['V2_norm'] = dataframe['V2_norm'].fillna(0)

        # Signal assignment using hysteresis
        upper_threshold = 1.0
        lower_threshold = -1.0

        dataframe['V'] = np.where(dataframe['V_norm'] > upper_threshold, 1,
                          np.where(dataframe['V_norm'] < lower_threshold, -1, np.nan))
        dataframe['V2'] = np.where(dataframe['V2_norm'] > upper_threshold, 1,
                           np.where(dataframe['V2_norm'] < lower_threshold, -1, np.nan))

        # Forward-fill to maintain the last state of the signal
        dataframe['V'] = dataframe['V'].ffill()  # Correct ffill usage
        dataframe['V2'] = dataframe['V2'].ffill()  # Correct ffill usage

        # Get Final Target Score to incorporate new calculations
        dataframe['T'] = dataframe['S'] * dataframe['R'] * dataframe['R2'] * dataframe['V'] * dataframe['V2']

        # Assign the target score T to the AI target column
        target_horizon = 1  # Define your prediction horizon here
        dataframe['&-target'] = dataframe['T'].shift(-target_horizon)
        return dataframe
        
    def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        """
        This is where FreqAI is called to make predictions
        """
        self.freqai_info = self.config["freqai"]
        dataframe = self.freqai.start(dataframe, metadata, self)
        
        # One can define indicators here if needed and add logic to populate_entry_trend and populate_exit_trend
        # dataframe["target_roi"] = dataframe["&-target_mean"] + dataframe["&-target_std"] * 1.25
        # dataframe["sell_roi"] = dataframe["&-target_mean"] - dataframe["&-target_std"] * 1.25

        return dataframe

    def populate_entry_trend(self, df: DataFrame, metadata: dict) -> DataFrame:
        """
        Entry signal logic based on FreqAI predictions
        """
        enter_long_conditions = [
            df["do_predict"] == 1,
            df['&-target'] > self.threshold_buy.value,
            df['volume'] > 0
        ]

        enter_short_conditions = [
            df["do_predict"] == 1,
            df['&-target'] < self.threshold_sell.value,
            df["volume"] > 0
        ]

        df.loc[
            reduce(lambda x, y: x & y, enter_long_conditions), ["enter_long", "enter_tag"]
        ] = (1, "long")

        df.loc[
            reduce(lambda x, y: x & y, enter_short_conditions), ["enter_short", "enter_tag"]
        ] = (1, "short")

        return df

    def populate_exit_trend(self, df: DataFrame, metadata: dict) -> DataFrame:
        """
        Exit signal logic based on FreqAI predictions
        """
        exit_long_conditions = [
            df['&-target'] < self.threshold_sell.value
        ]

        exit_short_conditions = [
            df['&-target'] > self.threshold_buy.value
        ]

        if exit_long_conditions:
            df.loc[
                reduce(lambda x, y: x & y, exit_long_conditions), ["exit_long", "exit_tag"]
            ] = (1, "exit_long")

        if exit_short_conditions:
            df.loc[
                reduce(lambda x, y: x & y, exit_short_conditions), ["exit_short", "exit_tag"]
            ] = (1, "exit_short")

        return df
    
    def custom_stoploss(self, pair: str, trade: 'Trade', current_time: datetime, current_rate: float,
                        current_profit: float, **kwargs) -> float:
        """
        Calculate custom stoploss based on dynamic trailing stop logic.
        """
        dataframe, row = self.dp.get_analyzed_dataframe(pair, self.timeframe)

        # If no valid dataframe is available (initial trades)
        if dataframe is None or row is None:
            return self.stoploss

        # Simple ATR-based dynamic stoploss (you can enhance this)
        try:
            atr = dataframe['atr'].iat[-1] if 'atr' in dataframe.columns else None
            if atr and not pd.isna(atr):
                # Dynamic stoploss based on ATR
                dynamic_stop = min(-0.02, -(atr * 2) / current_rate)  # 2x ATR or 2% minimum
                return max(self.stoploss, dynamic_stop)
        except (IndexError, KeyError):
            pass

        return self.stoploss

    def leverage(
        self,
        pair: str,
        current_time: "datetime",
        current_rate: float,
        proposed_leverage: float,
        **kwargs,
    ) -> float:
        """
        Dynamically adjust leverage based on volatility (ATR).
        """
        # Get the analyzed dataframe for the pair and timeframe
        dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)
        
        if dataframe is None or dataframe.empty:
            return self.leverage_value  # Fallback to default leverage

        # Get the latest ATR value from the last candle
        try:
            atr = dataframe['atr'].iat[-1]  # Get the ATR of the most recent candle
        except (IndexError, KeyError):
            return self.leverage_value  # Fallback if ATR is not available

        if pd.isna(atr):
            return self.leverage_value  # Fallback if ATR is NaN

        # Example: Lower leverage when volatility (ATR) is high
        try:
            rolling_mean = dataframe['atr'].rolling(window=14).mean().iat[-1]
            rolling_stddev = dataframe['atr'].rolling(window=14).std().iat[-1]
            
            if atr > rolling_mean + 1.5 * rolling_stddev:
                return self.leverage_value * 0.5  # Reduce leverage in volatile markets
        except (IndexError, KeyError):
            pass

        return self.leverage_value


def chaikin_mf(df, periods=20):
    """
    Chaikin Money Flow indicator
    """
    close = df["close"]
    low = df["low"]
    high = df["high"]
    volume = df["volume"]
    mfv = ((close - low) - (high - close)) / (high - low)
    mfv = mfv.fillna(0.0)
    mfv *= volume
    cmf = mfv.rolling(periods).sum() / volume.rolling(periods).sum()
    return Series(cmf, name="cmf")

def top_percent_change(dataframe: DataFrame, length: int) -> float:
    """
    Calculate top percent change over a given length
    """
    if length == 0:
        return (dataframe["open"] - dataframe["close"]) / dataframe["close"]
    else:
        return (dataframe["open"].rolling(length).max() - dataframe["close"]) / dataframe["close"]