FreqaiBreakoutClassifierV1

# pragma pylint: disable=missing-docstring, invalid-name, pointless-string-statement
# flake8: noqa: F401
"""
FreqaiBreakoutClassifierV1

Long-only FreqAI classifier strategy for high-volatility breakout regimes.
Designed for research, backtesting, hyperopt, and dry-run in a Freqtrade lab.

Suggested model: LightGBMClassifier or XGBoostClassifier.
Suggested timeframe: 5m.

Hyperopt only entry/exit thresholds, ROI, stoploss, trailing, and protections.
Do not hyperopt feature_engineering_*() or set_freqai_targets().
"""

from __future__ import annotations

from functools import reduce

import numpy as np
import talib.abstract as ta
from pandas import DataFrame

from freqtrade.strategy import DecimalParameter, IStrategy, IntParameter
from freqtrade.vendor.qtpylib import indicators as qtpylib


class FreqaiBreakoutClassifierV1(IStrategy):
    """
    Classifies the forward move as down / flat / up. Enters only when the model
    predicts "up" and deterministic breakout/volume filters confirm expansion.
    """

    INTERFACE_VERSION = 3
    timeframe = "5m"
    can_short = False

    process_only_new_candles = True
    startup_candle_count: int = 400

    use_exit_signal = True
    exit_profit_only = False
    ignore_roi_if_entry_signal = False

    minimal_roi = {
        "0": 0.075,
        "60": 0.040,
        "180": 0.020,
        "600": 0.0,
    }

    stoploss = -0.10

    trailing_stop = True
    trailing_stop_positive = 0.020
    trailing_stop_positive_offset = 0.045
    trailing_only_offset_is_reached = True

    # Entry thresholds.
    buy_di_max = DecimalParameter(0.50, 3.00, decimals=2, default=1.60, space="buy", optimize=True)
    buy_adx_min = IntParameter(12, 45, default=22, space="buy", optimize=True)
    buy_volume_ratio_min = DecimalParameter(
        1.00, 4.00, decimals=2, default=1.40, space="buy", optimize=True
    )
    buy_atr_pct_min = DecimalParameter(0.002, 0.060, decimals=3, default=0.012, space="buy", optimize=True)
    buy_rsi_max = IntParameter(55, 88, default=78, space="buy", optimize=True)

    # Exit thresholds.
    sell_rsi_min = IntParameter(25, 58, default=42, space="sell", optimize=True)
    sell_volume_ratio_min = DecimalParameter(
        0.30, 2.00, decimals=2, default=0.80, space="sell", optimize=True
    )

    # Protections.
    protection_cooldown = IntParameter(1, 24, default=8, space="protection", optimize=True)
    protection_stop_duration = IntParameter(12, 120, default=48, space="protection", optimize=True)

    @property
    def protections(self):
        return [
            {
                "method": "CooldownPeriod",
                "stop_duration_candles": self.protection_cooldown.value,
            },
            {
                "method": "StoplossGuard",
                "lookback_period_candles": 96,
                "trade_limit": 2,
                "stop_duration_candles": self.protection_stop_duration.value,
                "only_per_pair": False,
            },
        ]

    def feature_engineering_expand_all(
        self, dataframe: DataFrame, period: int, metadata: dict, **kwargs
    ) -> DataFrame:
        dataframe["%-rsi-period"] = ta.RSI(dataframe, timeperiod=period)
        dataframe["%-mfi-period"] = ta.MFI(dataframe, timeperiod=period)
        dataframe["%-adx-period"] = ta.ADX(dataframe, timeperiod=period)
        dataframe["%-roc-period"] = ta.ROC(dataframe, timeperiod=period)
        dataframe["%-atr-period"] = ta.ATR(dataframe, timeperiod=period)
        dataframe["%-ema-period"] = ta.EMA(dataframe, timeperiod=period)

        rolling_high = dataframe["high"].rolling(period).max().shift(1)
        rolling_low = dataframe["low"].rolling(period).min().shift(1)
        channel_range = (rolling_high - rolling_low).replace(0, np.nan)
        dataframe["%-donchian_position-period"] = (dataframe["close"] - rolling_low) / channel_range
        dataframe["%-donchian_width-period"] = channel_range / dataframe["close"].replace(0, np.nan)

        dataframe["%-relative_volume-period"] = dataframe["volume"] / dataframe[
            "volume"
        ].rolling(period).mean().replace(0, np.nan)
        return dataframe

    def feature_engineering_expand_basic(
        self, dataframe: DataFrame, metadata: dict, **kwargs
    ) -> DataFrame:
        dataframe["%-pct_change"] = dataframe["close"].pct_change()
        dataframe["%-raw_volume"] = dataframe["volume"]
        dataframe["%-raw_price"] = dataframe["close"]
        dataframe["%-candle_range_pct"] = (dataframe["high"] - dataframe["low"]) / dataframe[
            "close"
        ].replace(0, np.nan)
        dataframe["%-candle_body_pct"] = (dataframe["close"] - dataframe["open"]) / dataframe[
            "open"
        ].replace(0, np.nan)
        return dataframe

    def feature_engineering_standard(
        self, dataframe: DataFrame, metadata: dict, **kwargs
    ) -> DataFrame:
        dataframe["%%-ema_21"] = ta.EMA(dataframe, timeperiod=21)
        dataframe["%%-ema_50"] = ta.EMA(dataframe, timeperiod=50)
        dataframe["%%-ema_200"] = ta.EMA(dataframe, timeperiod=200)
        dataframe["%%-rsi"] = ta.RSI(dataframe, timeperiod=14)
        dataframe["%%-adx"] = ta.ADX(dataframe, timeperiod=14)
        dataframe["%%-atr_pct"] = ta.ATR(dataframe, timeperiod=14) / dataframe["close"].replace(
            0, np.nan
        )
        dataframe["%%-volume_ratio_20"] = dataframe["volume"] / dataframe["volume"].rolling(
            20
        ).mean().replace(0, np.nan)
        dataframe["%%-donchian_high_20"] = dataframe["high"].rolling(20).max().shift(1)
        dataframe["%%-donchian_low_20"] = dataframe["low"].rolling(20).min().shift(1)
        dataframe["%-day_of_week"] = (dataframe["date"].dt.dayofweek + 1) / 7
        dataframe["%-hour_of_day"] = (dataframe["date"].dt.hour + 1) / 24
        return dataframe

    def set_freqai_targets(
        self, dataframe: DataFrame, metadata: dict, **kwargs
    ) -> DataFrame:
        """
        Classification target. Class threshold is deliberately static. If you change it,
        create a new freqai.identifier because you changed the target definition.
        """
        self.freqai.class_names = ["down", "flat", "up"]

        label_period = self.freqai_info["feature_parameters"]["label_period_candles"]
        class_threshold = 0.010
        future_return = dataframe["close"].shift(-label_period) / dataframe["close"] - 1.0

        dataframe["&-s_breakout"] = np.select(
            [future_return < -class_threshold, future_return > class_threshold],
            ["down", "up"],
            default="flat",
        )
        return dataframe

    def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        dataframe = self.freqai.start(dataframe, metadata, self)

        if "do_predict" not in dataframe:
            dataframe["do_predict"] = 0
        if "DI_values" not in dataframe:
            dataframe["DI_values"] = 0.0
        if "&-s_breakout" not in dataframe:
            dataframe["&-s_breakout"] = "flat"

        return dataframe

    def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        dataframe["enter_long"] = 0
        dataframe["enter_tag"] = ""

        breakout_trigger = qtpylib.crossed_above(
            dataframe["close"], dataframe["%%-donchian_high_20"]
        )

        conditions = [
            dataframe["do_predict"] == 1,
            dataframe["DI_values"] < self.buy_di_max.value,
            dataframe["&-s_breakout"] == "up",
            breakout_trigger,
            dataframe["%%-ema_50"] > dataframe["%%-ema_200"],
            dataframe["close"] > dataframe["%%-ema_21"],
            dataframe["%%-adx"] > self.buy_adx_min.value,
            dataframe["%%-atr_pct"] > self.buy_atr_pct_min.value,
            dataframe["%%-volume_ratio_20"] > self.buy_volume_ratio_min.value,
            dataframe["%%-rsi"] < self.buy_rsi_max.value,
            dataframe["volume"] > 0,
        ]

        dataframe.loc[reduce(lambda x, y: x & y, conditions), ["enter_long", "enter_tag"]] = (
            1,
            "freqai_breakout_classifier",
        )
        return dataframe

    def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        dataframe["exit_long"] = 0
        dataframe["exit_tag"] = ""

        model_no_longer_up = dataframe["&-s_breakout"].isin(["down", "flat"])
        failed_breakout = (
            (dataframe["close"] < dataframe["%%-ema_21"])
            | (dataframe["close"] < dataframe["%%-donchian_low_20"])
            | (dataframe["%%-rsi"] < self.sell_rsi_min.value)
        )
        volume_dried_up = dataframe["%%-volume_ratio_20"] < self.sell_volume_ratio_min.value

        conditions = [
            dataframe["volume"] > 0,
            model_no_longer_up | failed_breakout | volume_dried_up,
        ]

        dataframe.loc[reduce(lambda x, y: x & y, conditions), ["exit_long", "exit_tag"]] = (
            1,
            "freqai_breakout_exit",
        )
        return dataframe