FreqaiRangeReversionV1

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

Long-only FreqAI strategy for low-volatility range / mean-reversion regimes.
Designed for research, backtesting, hyperopt, and dry-run in a Freqtrade lab.

Suggested model: LightGBMRegressor or XGBoostRegressor.
Suggested timeframe: 5m or 15m.

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 FreqaiRangeReversionV1(IStrategy):
    """
    Mean-reversion strategy. It only enters stretched, relatively quiet ranges when
    FreqAI predicts a positive forward average return.
    """

    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.035,
        "60": 0.020,
        "180": 0.010,
        "480": 0.0,
    }

    stoploss = -0.06

    trailing_stop = True
    trailing_stop_positive = 0.010
    trailing_stop_positive_offset = 0.025
    trailing_only_offset_is_reached = True

    # Entry thresholds.
    buy_z = DecimalParameter(0.10, 1.75, decimals=2, default=0.65, space="buy", optimize=True)
    buy_pred_floor = DecimalParameter(0.001, 0.020, decimals=3, default=0.004, space="buy", optimize=True)
    buy_di_max = DecimalParameter(0.50, 3.00, decimals=2, default=1.60, space="buy", optimize=True)
    buy_rsi_max = IntParameter(15, 45, default=32, space="buy", optimize=True)
    buy_adx_max = IntParameter(10, 32, default=24, space="buy", optimize=True)
    buy_bb_percent_max = DecimalParameter(0.00, 0.35, decimals=2, default=0.12, space="buy", optimize=True)
    buy_bb_width_max = DecimalParameter(0.010, 0.200, decimals=3, default=0.080, space="buy", optimize=True)
    buy_atr_pct_max = DecimalParameter(0.002, 0.080, decimals=3, default=0.030, space="buy", optimize=True)

    # Exit thresholds.
    sell_z = DecimalParameter(0.00, 1.25, decimals=2, default=0.25, space="sell", optimize=True)
    sell_pred_floor = DecimalParameter(-0.020, 0.010, decimals=3, default=-0.002, space="sell", optimize=True)
    sell_bb_percent_min = DecimalParameter(0.45, 1.00, decimals=2, default=0.72, space="sell", optimize=True)
    sell_rsi_min = IntParameter(50, 82, default=64, space="sell", optimize=True)

    # Protections.
    protection_cooldown = IntParameter(1, 24, default=4, space="protection", optimize=True)
    protection_stop_duration = IntParameter(12, 96, default=30, 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["%-cci-period"] = ta.CCI(dataframe, timeperiod=period)
        dataframe["%-atr-period"] = ta.ATR(dataframe, timeperiod=period)

        bollinger = qtpylib.bollinger_bands(
            qtpylib.typical_price(dataframe), window=period, stds=2.0
        )
        bb_range = (bollinger["upper"] - bollinger["lower"]).replace(0, np.nan)
        dataframe["%-bb_width-period"] = bb_range / bollinger["mid"].replace(0, np.nan)
        dataframe["%-bb_percent-period"] = (dataframe["close"] - bollinger["lower"]) / bb_range

        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_body_pct"] = (dataframe["close"] - dataframe["open"]) / dataframe[
            "open"
        ].replace(0, np.nan)
        dataframe["%-upper_wick_pct"] = (
            dataframe["high"] - np.maximum(dataframe["open"], dataframe["close"])
        ) / dataframe["close"].replace(0, np.nan)
        dataframe["%-lower_wick_pct"] = (
            np.minimum(dataframe["open"], dataframe["close"]) - dataframe["low"]
        ) / dataframe["close"].replace(0, np.nan)
        return dataframe

    def feature_engineering_standard(
        self, dataframe: DataFrame, metadata: dict, **kwargs
    ) -> DataFrame:
        dataframe["%%-rsi"] = ta.RSI(dataframe, timeperiod=14)
        dataframe["%%-adx"] = ta.ADX(dataframe, timeperiod=14)
        dataframe["%%-ema_50"] = ta.EMA(dataframe, timeperiod=50)
        dataframe["%%-ema_200"] = ta.EMA(dataframe, timeperiod=200)
        dataframe["%%-atr_pct"] = ta.ATR(dataframe, timeperiod=14) / dataframe["close"].replace(
            0, np.nan
        )

        bollinger = qtpylib.bollinger_bands(qtpylib.typical_price(dataframe), window=20, stds=2.0)
        bb_range = (bollinger["upper"] - bollinger["lower"]).replace(0, np.nan)
        dataframe["%%-bb_lower"] = bollinger["lower"]
        dataframe["%%-bb_mid"] = bollinger["mid"]
        dataframe["%%-bb_upper"] = bollinger["upper"]
        dataframe["%%-bb_width"] = bb_range / bollinger["mid"].replace(0, np.nan)
        dataframe["%%-bb_percent"] = (dataframe["close"] - bollinger["lower"]) / bb_range

        dataframe["%%-volume_ratio_20"] = dataframe["volume"] / dataframe["volume"].rolling(
            20
        ).mean().replace(0, np.nan)
        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:
        label_period = self.freqai_info["feature_parameters"]["label_period_candles"]
        dataframe["&-s_close"] = (
            dataframe["close"].shift(-label_period).rolling(label_period).mean()
            / dataframe["close"]
            - 1.0
        )
        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_close_mean" not in dataframe:
            dataframe["&-s_close_mean"] = 0.0
        if "&-s_close_std" not in dataframe:
            dataframe["&-s_close_std"] = 0.0

        return dataframe

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

        prediction = dataframe["&-s_close"]
        dynamic_target = dataframe["&-s_close_mean"] + dataframe["&-s_close_std"] * self.buy_z.value

        conditions = [
            dataframe["do_predict"] == 1,
            dataframe["DI_values"] < self.buy_di_max.value,
            prediction > dynamic_target,
            prediction > self.buy_pred_floor.value,
            dataframe["%%-bb_percent"] < self.buy_bb_percent_max.value,
            dataframe["%%-bb_width"] < self.buy_bb_width_max.value,
            dataframe["%%-atr_pct"] < self.buy_atr_pct_max.value,
            dataframe["%%-rsi"] < self.buy_rsi_max.value,
            dataframe["%%-adx"] < self.buy_adx_max.value,
            dataframe["close"] < dataframe["%%-bb_lower"],
            dataframe["volume"] > 0,
        ]

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

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

        dynamic_exit = dataframe["&-s_close_mean"] - dataframe["&-s_close_std"] * self.sell_z.value
        model_turn = (dataframe["&-s_close"] < dynamic_exit) | (
            dataframe["&-s_close"] < self.sell_pred_floor.value
        )
        mean_reversion_complete = (
            (dataframe["%%-bb_percent"] > self.sell_bb_percent_min.value)
            | (dataframe["close"] > dataframe["%%-bb_mid"])
            | (dataframe["%%-rsi"] > self.sell_rsi_min.value)
        )

        conditions = [
            dataframe["volume"] > 0,
            model_turn | mean_reversion_complete,
        ]

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