LunoFreqAIStrategy

# pragma pylint: disable=missing-docstring, invalid-name, pointless-string-statement
# flake8: noqa: F401
# isort: skip_file
# --- Do not remove these libs ---

import logging
import numpy as np
import pandas as pd
from datetime import datetime, timedelta
from typing import Optional, Union

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

import freqtrade.vendor.qtpylib.indicators as qtpylib

logger = logging.getLogger(__name__)


class LunoFreqAIStrategy(IStrategy):
    """
    Real FreqAI Strategy for Luno-style trading
    This strategy uses the actual FreqAI framework for machine learning predictions
    """

    INTERFACE_VERSION = 3

    # Minimal ROI designed for the strategy.
    minimal_roi = {
        "0": 0.10,      # 10% profit
        "40": 0.06,     # 6% profit after 40 minutes
        "100": 0.04,    # 4% profit after 100 minutes
        "180": 0.02,    # 2% profit after 180 minutes
        "360": 0.01     # 1% profit after 360 minutes
    }

    # Optimal stoploss designed for the strategy.
    stoploss = -0.05  # 5% stop loss (4% risk management + buffer)

    # Trailing stoploss
    trailing_stop = True
    trailing_stop_positive = 0.02
    trailing_stop_positive_offset = 0.03
    trailing_only_offset_is_reached = True

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

    # FreqAI required parameters
    can_short = False  # We only do long trades
    startup_candle_count: int = 40
    process_only_new_candles = False

    # Strategy parameters
    buy_params = {
        "ai_signal_threshold": 0.65,  # Confidence threshold for AI buy signals
        "rsi_buy": 30,
        "volume_factor": 1.2,
    }

    sell_params = {
        "ai_signal_threshold": -0.3,  # Confidence threshold for AI sell signals
        "rsi_sell": 70,
    }

    # FreqAI specific parameters
    ai_signal_threshold = DecimalParameter(0.5, 0.9, default=0.65, space="buy", optimize=True)
    rsi_buy = IntParameter(20, 40, default=30, space="buy", optimize=True)
    rsi_sell = IntParameter(60, 80, default=70, space="sell", optimize=True)
    volume_factor = DecimalParameter(1.0, 2.0, default=1.2, space="buy", optimize=True)

    def bot_loop_start(self, current_time: datetime, **kwargs) -> None:
        """
        Called at the start of the bot iteration (one loop).
        """
        if self.config.get("freqai", {}).get("enabled", False):
            logger.info("FreqAI enabled - using real ML predictions for trading decisions")

    def populate_indicators(self, dataframe: pd.DataFrame, metadata: dict) -> pd.DataFrame:
        """
        Adds several different TA indicators to the given DataFrame
        """

        # Basic indicators for feature engineering and fallback signals
        dataframe['rsi'] = qtpylib.rsi(dataframe, timeperiod=14)
        
        # MACD
        macd = qtpylib.macd(dataframe)
        dataframe['macd'] = macd['macd']
        dataframe['macd_signal'] = macd['macdsignal']
        dataframe['macd_hist'] = macd['macdhist']
        
        # Bollinger Bands
        bollinger = qtpylib.bollinger_bands(dataframe, window=20, stds=2)
        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']
        )

        # EMA
        dataframe['ema12'] = qtpylib.ema(dataframe, timeperiod=12)
        dataframe['ema26'] = qtpylib.ema(dataframe, timeperiod=26)
        dataframe['ema50'] = qtpylib.ema(dataframe, timeperiod=50)

        # Volume indicators
        dataframe['volume_sma'] = qtpylib.sma(dataframe['volume'], timeperiod=20)
        dataframe['volume_ratio'] = dataframe['volume'] / dataframe['volume_sma']

        # Price action indicators
        dataframe['price_change'] = dataframe['close'].pct_change()
        dataframe['momentum'] = qtpylib.momentum(dataframe, timeperiod=14)

        return dataframe

    def populate_any_indicators(
        self, pair, df, tf, informative=None, set_generalized_indicators=False
    ):
        """
        Function which takes a dataframe, pair and timeframe and populates
        indicators for FreqAI feature engineering.
        """

        coin = pair.split('/')[0]

        if informative is None:
            informative = self.dp.get_pair_dataframe(pair, tf)

        # FreqAI feature engineering - the "%" prefix indicates features for ML model
        informative[f"%-{coin}rsi-period-10"] = qtpylib.rsi(informative, timeperiod=10)
        informative[f"%-{coin}rsi-period-14"] = qtpylib.rsi(informative, timeperiod=14)
        informative[f"%-{coin}rsi-period-20"] = qtpylib.rsi(informative, timeperiod=20)

        # MACD features
        macd = qtpylib.macd(informative)
        informative[f"%-{coin}macd"] = macd['macd']
        informative[f"%-{coin}macd_signal"] = macd['macdsignal']
        informative[f"%-{coin}macd_hist"] = macd['macdhist']

        # Bollinger Band features
        bollinger = qtpylib.bollinger_bands(informative, window=20, stds=2)
        informative[f"%-{coin}bb_lowerband"] = bollinger['lower']
        informative[f"%-{coin}bb_middleband"] = bollinger['mid']
        informative[f"%-{coin}bb_upperband"] = bollinger['upper']
        informative[f"%-{coin}bb_percent"] = (
            (informative['close'] - bollinger['lower']) /
            (bollinger['upper'] - bollinger['lower'])
        )
        informative[f"%-{coin}bb_width"] = (
            (bollinger['upper'] - bollinger['lower']) / bollinger['mid']
        )

        # EMA features
        for period in [10, 21, 50]:
            informative[f"%-{coin}ema-period-{period}"] = qtpylib.ema(informative, timeperiod=period)

        # SMA features
        for period in [10, 20, 50]:
            informative[f"%-{coin}sma-period-{period}"] = qtpylib.sma(informative, timeperiod=period)

        # Volume features
        informative[f"%-{coin}volume_sma"] = qtpylib.sma(informative['volume'], timeperiod=20)
        informative[f"%-{coin}volume_ratio"] = informative['volume'] / informative[f"%-{coin}volume_sma"]

        # Price action features
        informative[f"%-{coin}price_change"] = informative['close'].pct_change()
        informative[f"%-{coin}high_low_ratio"] = informative['high'] / informative['low']

        # Momentum features
        for period in [10, 14, 20]:
            informative[f"%-{coin}momentum-period-{period}"] = qtpylib.momentum(informative, timeperiod=period)

        # Volatility features (ATR)
        for period in [14, 20]:
            informative[f"%-{coin}atr-period-{period}"] = qtpylib.atr(informative, timeperiod=period)

        # Add targets for FreqAI to predict - the "&" prefix indicates targets
        # These are what FreqAI will try to predict
        informative[f"&-{coin}close_price_2"] = informative['close'].shift(-2)
        informative[f"&-{coin}close_price_5"] = informative['close'].shift(-5)
        informative[f"&-{coin}close_price_10"] = informative['close'].shift(-10)

        # Classification targets (up/down/sideways)
        future_return_2 = (informative['close'].shift(-2) / informative['close'] - 1)
        future_return_5 = (informative['close'].shift(-5) / informative['close'] - 1)

        # Binary classification: 1 for up, 0 for down/sideways
        informative[f"&-{coin}up_or_down_2"] = (future_return_2 > 0.01).astype(int)
        informative[f"&-{coin}up_or_down_5"] = (future_return_5 > 0.01).astype(int)

        # Multi-class classification: 2=strong_up, 1=up, 0=down/sideways
        def classify_movement(returns):
            if returns > 0.02:
                return 2  # Strong up
            elif returns > 0.005:
                return 1  # Up
            else:
                return 0  # Down/sideways

        informative[f"&-{coin}price_class_2"] = future_return_2.apply(classify_movement)
        informative[f"&-{coin}price_class_5"] = future_return_5.apply(classify_movement)

        return informative

    def populate_entry_trend(self, dataframe: pd.DataFrame, metadata: dict) -> pd.DataFrame:
        """
        Based on TA indicators and FreqAI predictions, populates the entry signal
        """

        # Check if FreqAI predictions are available
        freqai_predictions = [col for col in dataframe.columns if col.startswith('do_predict')]
        
        if freqai_predictions:
            # Use FreqAI predictions as primary signal
            logger.debug("Using FreqAI predictions for entry signals")
            
            # Look for prediction columns that indicate buy signals
            # FreqAI typically creates columns like 'do_predict_close_price_5'
            prediction_cols = [col for col in dataframe.columns if 'do_predict' in col and 'close_price' in col]
            
            if prediction_cols:
                # Use the first available prediction (e.g., 5-period ahead)
                pred_col = prediction_cols[0]
                
                # Calculate expected return from prediction
                current_price = dataframe['close']
                predicted_price = dataframe[pred_col]
                expected_return = (predicted_price / current_price - 1)
                
                # Entry condition: AI predicts significant upward movement
                dataframe.loc[
                    (
                        (expected_return > self.ai_signal_threshold.value / 100) &  # AI signal
                        (dataframe['rsi'] < 70) &  # Not overbought
                        (dataframe['volume_ratio'] > 1.0) &  # Volume confirmation
                        (dataframe['volume'] > 0)  # Basic sanity check
                    ),
                    'enter_long'] = 1
            else:
                # Fallback to classification predictions
                class_cols = [col for col in dataframe.columns if 'do_predict' in col and ('up_or_down' in col or 'price_class' in col)]
                
                if class_cols:
                    class_col = class_cols[0]
                    
                    # Entry condition: AI predicts upward movement
                    dataframe.loc[
                        (
                            (dataframe[class_col] >= 1) &  # AI predicts up or strong up
                            (dataframe['rsi'] < 70) &  # Not overbought
                            (dataframe['volume_ratio'] > 1.0) &  # Volume confirmation
                            (dataframe['volume'] > 0)  # Basic sanity check
                        ),
                        'enter_long'] = 1

        else:
            # Fallback to traditional technical analysis
            logger.debug("No FreqAI predictions available, using technical analysis fallback")
            
            dataframe.loc[
                (
                    (dataframe['rsi'] < self.rsi_buy.value) &
                    (dataframe['volume_ratio'] > self.volume_factor.value) &
                    (dataframe['close'] > dataframe['ema12']) &
                    (dataframe['ema12'] > dataframe['ema26']) &
                    (dataframe['macd'] > dataframe['macd_signal']) &
                    (dataframe['bb_percent'] < 0.2) &  # Near lower Bollinger Band
                    (dataframe['volume'] > 0)
                ),
                'enter_long'] = 1

        return dataframe

    def populate_exit_trend(self, dataframe: pd.DataFrame, metadata: dict) -> pd.DataFrame:
        """
        Based on TA indicators and FreqAI predictions, populates the exit signal
        """

        # Check if FreqAI predictions are available
        freqai_predictions = [col for col in dataframe.columns if col.startswith('do_predict')]
        
        if freqai_predictions:
            # Use FreqAI predictions for exit signals
            prediction_cols = [col for col in dataframe.columns if 'do_predict' in col and 'close_price' in col]
            
            if prediction_cols:
                pred_col = prediction_cols[0]
                
                # Calculate expected return from prediction
                current_price = dataframe['close']
                predicted_price = dataframe[pred_col]
                expected_return = (predicted_price / current_price - 1)
                
                # Exit condition: AI predicts downward movement or low confidence
                dataframe.loc[
                    (
                        (expected_return < -0.01) |  # AI predicts decline
                        (dataframe['rsi'] > self.rsi_sell.value) |  # Overbought
                        (dataframe['bb_percent'] > 0.8)  # Near upper Bollinger Band
                    ),
                    'exit_long'] = 1
            else:
                # Use classification predictions
                class_cols = [col for col in dataframe.columns if 'do_predict' in col and ('up_or_down' in col or 'price_class' in col)]
                
                if class_cols:
                    class_col = class_cols[0]
                    
                    # Exit condition: AI predicts down/sideways movement
                    dataframe.loc[
                        (
                            (dataframe[class_col] == 0) |  # AI predicts down/sideways
                            (dataframe['rsi'] > self.rsi_sell.value) |  # Overbought
                            (dataframe['bb_percent'] > 0.8)  # Near upper Bollinger Band
                        ),
                        'exit_long'] = 1

        else:
            # Fallback to traditional technical analysis
            dataframe.loc[
                (
                    (dataframe['rsi'] > self.rsi_sell.value) |
                    (dataframe['bb_percent'] > 0.8) |
                    (dataframe['macd'] < dataframe['macd_signal'])
                ),
                'exit_long'] = 1

        return dataframe

    def confirm_trade_entry(self, pair: str, order_type: str, amount: float, rate: float,
                           time_in_force: str, current_time: datetime,
                           entry_tag: Optional[str], side: str, **kwargs) -> bool:
        """
        Called right before placing an entry order.
        Implement 4% risk management as per user requirements
        """
        try:
            # Get current portfolio value
            portfolio_value = 154273.71  # User's capital (in production this would be dynamic)
            
            # Calculate 4% risk
            max_risk_amount = portfolio_value * 0.04  # R6170.95
            
            # Calculate position size based on 4% risk
            trade_value = amount * rate
            
            if trade_value > max_risk_amount:
                logger.warning(f"Trade value {trade_value} exceeds 4% risk limit {max_risk_amount}")
                return False
            
            # Special handling for XRP (user wants to keep 1000 XRP long-term)
            if 'XRP' in pair:
                # Conservative limit for XRP trades to protect reserves
                if amount > 100:
                    logger.info(f"XRP trade rejected - protecting 1000 XRP reserve")
                    return False
            
            logger.info(f"FreqAI Trade confirmed: {pair} {side} {amount} at {rate}")
            return True
            
        except Exception as e:
            logger.error(f"Error in confirm_trade_entry: {e}")
            return False

    def custom_exit(self, pair: str, trade, current_time: datetime, current_rate: float,
                   current_profit: float, **kwargs):
        """
        Custom exit logic for advanced profit taking with FreqAI insights
        """
        try:
            # Progressive profit taking based on user's R8000/month target
            # Target: R8000/month means ~R267/day profit target
            
            if current_profit > 0.15:  # 15% profit
                return "freqai_take_profit_15"
            elif current_profit > 0.10:  # 10% profit
                return "freqai_take_profit_10"
            elif current_profit > 0.05:  # 5% profit
                return "freqai_take_profit_5"
            elif current_profit < -0.04:  # 4% loss (strict risk management)
                return "freqai_stop_loss_4"
                
            return None
            
        except Exception as e:
            logger.error(f"Error in custom_exit: {e}")
            return None