AI_Strategy

# 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 pandas import DataFrame
from datetime import datetime
from typing import Optional, Union

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

# --------------------------------
# Import required indicators
import talib.abstract as ta
import freqtrade.vendor.qtpylib.indicators as qtpylib


class AI_Strategy(IStrategy):
    """
    AI-powered strategy that fetches dynamic configurations from core-app
    """
    INTERFACE_VERSION = 3

    # Strategy parameters
    can_short: bool = True
    
    # Define minimal_roi
    minimal_roi = {
        "0": 0.10,
        "30": 0.05,
        "60": 0.02,
        "120": 0.01
    }

    # Define stoploss
    stoploss = -0.10
    trailing_stop = True
    trailing_stop_positive = 0.01
    trailing_stop_positive_offset = 0.02
    trailing_only_offset_is_reached = True

    # Define timeframe
    timeframe = '15m'

    # 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 = 100

    # Define buy parameters
    buy_rsi_enabled = BooleanParameter(default=True, space="buy")
    buy_rsi = IntParameter(10, 40, default=30, space="buy")
    
    buy_ema_enabled = BooleanParameter(default=True, space="buy")
    buy_ema_short = IntParameter(5, 50, default=10, space="buy")
    buy_ema_long = IntParameter(50, 200, default=50, space="buy")
    
    buy_bb_enabled = BooleanParameter(default=True, space="buy")
    buy_bb_width = DecimalParameter(0.5, 3.0, default=2.0, space="buy")

    # Define sell parameters  
    sell_rsi_enabled = BooleanParameter(default=True, space="sell")
    sell_rsi = IntParameter(60, 90, default=70, space="sell")
    
    sell_ema_enabled = BooleanParameter(default=True, space="sell")
    
    sell_bb_enabled = BooleanParameter(default=True, space="sell")

    def informative_pairs(self):
        """
        Define additional, informative pair/interval combinations to be cached from the exchange.
        """
        return []

    def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        """
        Adds several different TA indicators to the given DataFrame
        """
        # RSI
        dataframe['rsi'] = ta.RSI(dataframe, timeperiod=14)
        
        # EMA - Exponential Moving Average
        dataframe['ema_10'] = ta.EMA(dataframe, timeperiod=self.buy_ema_short.value)
        dataframe['ema_50'] = ta.EMA(dataframe, timeperiod=self.buy_ema_long.value)
        dataframe['ema_200'] = ta.EMA(dataframe, timeperiod=200)
        
        # Bollinger Bands
        bollinger = qtpylib.bollinger_bands(qtpylib.typical_price(dataframe), window=20, stds=self.buy_bb_width.value)
        dataframe['bb_lowerband'] = bollinger['lower']
        dataframe['bb_middleband'] = bollinger['mid']
        dataframe['bb_upperband'] = bollinger['upper']
        dataframe["bb_percent"] = (
            (dataframe["close"] - dataframe["bb_lowerband"]) /
            (dataframe["bb_upperband"] - dataframe["bb_lowerband"])
        )
        dataframe["bb_width"] = (
            (dataframe["bb_upperband"] - dataframe["bb_lowerband"]) / dataframe["bb_middleband"]
        )
        
        # MACD
        macd = ta.MACD(dataframe)
        dataframe['macd'] = macd['macd']
        dataframe['macdsignal'] = macd['macdsignal']
        dataframe['macdhist'] = macd['macdhist']
        
        # Stochastic
        stoch = ta.STOCH(dataframe)
        dataframe['stoch_k'] = stoch['slowk']
        dataframe['stoch_d'] = stoch['slowd']
        
        # ADX
        dataframe['adx'] = ta.ADX(dataframe)
        
        # Volume
        dataframe['volume_mean'] = dataframe['volume'].rolling(window=30).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 = []
        
        # RSI condition
        if self.buy_rsi_enabled.value:
            conditions.append(dataframe['rsi'] < self.buy_rsi.value)
        
        # EMA condition
        if self.buy_ema_enabled.value:
            conditions.append(
                (dataframe['ema_10'] > dataframe['ema_50']) &
                (dataframe['close'] > dataframe['ema_10'])
            )
        
        # Bollinger Bands condition
        if self.buy_bb_enabled.value:
            conditions.append(
                (dataframe['close'] <= dataframe['bb_lowerband']) &
                (dataframe['bb_percent'] < 0.2)
            )

        # Volume condition
        conditions.append(dataframe['volume'] > 0)

        if conditions:
            dataframe.loc[
                reduce(lambda x, y: x & y, conditions),
                'enter_long'] = 1

        # Short entry conditions (opposite of long)
        short_conditions = []
        
        # RSI condition for short
        if self.buy_rsi_enabled.value:
            short_conditions.append(dataframe['rsi'] > (100 - self.buy_rsi.value))
        
        # EMA condition for short
        if self.buy_ema_enabled.value:
            short_conditions.append(
                (dataframe['ema_10'] < dataframe['ema_50']) &
                (dataframe['close'] < dataframe['ema_10'])
            )
        
        # Bollinger Bands condition for short
        if self.buy_bb_enabled.value:
            short_conditions.append(
                (dataframe['close'] >= dataframe['bb_upperband']) &
                (dataframe['bb_percent'] > 0.8)
            )

        # Volume condition
        short_conditions.append(dataframe['volume'] > 0)

        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:
        """
        Based on TA indicators, populates the exit signal for the given dataframe
        """
        conditions = []
        
        # RSI condition
        if self.sell_rsi_enabled.value:
            conditions.append(dataframe['rsi'] > self.sell_rsi.value)
        
        # EMA condition  
        if self.sell_ema_enabled.value:
            conditions.append(
                (dataframe['ema_10'] < dataframe['ema_50']) &
                (dataframe['close'] < dataframe['ema_10'])
            )
        
        # Bollinger Bands condition
        if self.sell_bb_enabled.value:
            conditions.append(
                (dataframe['close'] >= dataframe['bb_upperband']) &
                (dataframe['bb_percent'] > 0.8)
            )

        # Volume condition
        conditions.append(dataframe['volume'] > 0)

        if conditions:
            dataframe.loc[
                reduce(lambda x, y: x & y, conditions),
                'exit_long'] = 1

        # Short exit conditions (opposite of long exit)
        short_exit_conditions = []
        
        # RSI condition for short exit
        if self.sell_rsi_enabled.value:
            short_exit_conditions.append(dataframe['rsi'] < (100 - self.sell_rsi.value))
        
        # EMA condition for short exit
        if self.sell_ema_enabled.value:
            short_exit_conditions.append(
                (dataframe['ema_10'] > dataframe['ema_50']) &
                (dataframe['close'] > dataframe['ema_10'])
            )
        
        # Bollinger Bands condition for short exit
        if self.sell_bb_enabled.value:
            short_exit_conditions.append(
                (dataframe['close'] <= dataframe['bb_lowerband']) &
                (dataframe['bb_percent'] < 0.2)
            )

        # Volume condition
        short_exit_conditions.append(dataframe['volume'] > 0)

        if short_exit_conditions:
            dataframe.loc[
                reduce(lambda x, y: x & y, short_exit_conditions),
                'exit_short'] = 1

        return dataframe


# Import reduce function
from functools import reduce