# source: https://raw.githubusercontent.com/raulcarreno/trader-bot/1e811ea1c84acdabce4bafe74c14ce1920da7549/user_data/strategies/FreqaiExampleStrategy.py
import logging
from datetime import datetime
from functools import reduce

import talib.abstract as ta
from pandas import DataFrame
from technical import qtpylib

from freqtrade.strategy import IStrategy, Trade, timeframe_to_minutes


logger = logging.getLogger(__name__)


class Github_raulcarreno_trader_bot__FreqaiExampleStrategy__20260610_164544(IStrategy):
    """
    FreqAI futures strategy (10x): high-conviction entries, prediction-scaled stake,
    profit-oriented exits. Tuned for Bitget USDT perpetuals.
    """

    timeframe = "5m"
    TARGET_LEVERAGE = 10.0
    PREDICTION_THRESHOLD = 0.012
    PREDICTION_STRONG_MULT = 1.25
    PREDICTION_WEAK_MULT = 0.25
    MAX_STAKE_PRED_MULT = 2.0
    DI_MAX_DEFAULT = 0.95

    stoploss = -0.15
    use_custom_stoploss = True
    trailing_stop = False

    _tf_min = timeframe_to_minutes(timeframe)
    minimal_roi = {
        "0": 0.10,
        str(_tf_min * 12): 0.06,
        str(_tf_min * 18): 0.04,
        str(_tf_min * 24): 0.02,
        str(_tf_min * 30): 0,
    }

    MIN_ADX = 22
    MAX_ATR_PCT = 0.012

    plot_config = {
        "main_plot": {},
        "subplots": {
            "&-s_close": {"&-s_close": {"color": "blue"}},
            "do_predict": {"do_predict": {"color": "brown"}},
        },
    }

    process_only_new_candles = True
    use_exit_signal = True
    startup_candle_count: int = 310
    can_short = True

    @property
    def protections(self):
        return [
            {
                "method": "CooldownPeriod",
                "stop_duration_candles": 2,
            },
            {
                "method": "StoplossGuard",
                "lookback_period_candles": 36,
                "trade_limit": 4,
                "stop_duration_candles": 12,
                "only_per_pair": False,
            },
            {
                "method": "MaxDrawdown",
                "calculation_mode": "equity",
                "lookback_period_candles": 288,
                "trade_limit": 3,
                "stop_duration_candles": 72,
                "max_allowed_drawdown": 0.08,
            },
        ]

    def _strategy_params(self) -> dict:
        defaults = {
            "prediction_threshold": self.PREDICTION_THRESHOLD,
            "prediction_strong_mult": self.PREDICTION_STRONG_MULT,
            "prediction_weak_mult": self.PREDICTION_WEAK_MULT,
            "min_adx": self.MIN_ADX,
            "max_atr_pct": self.MAX_ATR_PCT,
            "max_stake_pred_mult": self.MAX_STAKE_PRED_MULT,
            "target_leverage": self.TARGET_LEVERAGE,
            "exit_requires_profit": False,
        }
        overrides = self.config.get("strategy_params") or {}
        merged = {**defaults, **overrides}
        return merged

    def _prediction_threshold(self) -> float:
        return float(self._strategy_params()["prediction_threshold"])

    def _strong_prediction_threshold(self) -> float:
        p = self._strategy_params()
        return self._prediction_threshold() * float(p["prediction_strong_mult"])

    def _weak_prediction_threshold(self) -> float:
        p = self._strategy_params()
        return self._prediction_threshold() * float(p["prediction_weak_mult"])

    def _min_adx(self) -> float:
        return float(self._strategy_params()["min_adx"])

    def _max_atr_pct(self) -> float:
        return float(self._strategy_params()["max_atr_pct"])

    def _target_leverage(self) -> float:
        return float(self._strategy_params()["target_leverage"])

    def _exit_requires_profit(self) -> bool:
        return bool(self._strategy_params().get("exit_requires_profit", False))

    def _di_threshold(self) -> float:
        return float(
            self.config.get("freqai", {})
            .get("feature_parameters", {})
            .get("DI_threshold", self.DI_MAX_DEFAULT)
        )

    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["%-sma-period"] = ta.SMA(dataframe, timeperiod=period)
        dataframe["%-ema-period"] = ta.EMA(dataframe, timeperiod=period)

        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"]
        dataframe["%-roc-period"] = ta.ROC(dataframe, timeperiod=period)
        dataframe["%-relative_volume-period"] = (
            dataframe["volume"] / dataframe["volume"].rolling(period).mean()
        )
        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"]
        return dataframe

    def feature_engineering_standard(
        self, dataframe: DataFrame, metadata: dict, **kwargs
    ) -> DataFrame:
        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:
        dataframe["&-s_close"] = (
            dataframe["close"]
            .shift(-self.freqai_info["feature_parameters"]["label_period_candles"])
            .rolling(self.freqai_info["feature_parameters"]["label_period_candles"])
            .mean()
            / dataframe["close"]
            - 1
        )
        return dataframe

    def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        dataframe = self.freqai.start(dataframe, metadata, self)
        dataframe["atr_pct"] = ta.ATR(dataframe, timeperiod=14) / dataframe["close"]
        dataframe["adx"] = ta.ADX(dataframe, timeperiod=14)
        return dataframe

    def populate_entry_trend(self, df: DataFrame, metadata: dict) -> DataFrame:
        strong_thr = self._strong_prediction_threshold()
        vol_ok = df["atr_pct"] < self._max_atr_pct()
        trend_ok = df["adx"] > self._min_adx()

        di_ok = df["do_predict"] == 1
        if "DI_values" in df.columns:
            di_ok = di_ok & (df["DI_values"] < self._di_threshold())

        enter_long_conditions = [
            di_ok,
            df["&-s_close"] > strong_thr,
            vol_ok,
            trend_ok,
        ]
        if enter_long_conditions:
            df.loc[
                reduce(lambda x, y: x & y, enter_long_conditions), ["enter_long", "enter_tag"]
            ] = (1, "long")

        enter_short_conditions = [
            di_ok,
            df["&-s_close"] < -strong_thr,
            vol_ok,
            trend_ok,
        ]
        if enter_short_conditions:
            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:
        if self._exit_requires_profit():
            return df

        weak = self._weak_prediction_threshold()
        exit_long_conditions = [
            df["do_predict"] == 1,
            df["&-s_close"] < -weak,
        ]
        if exit_long_conditions:
            df.loc[reduce(lambda x, y: x & y, exit_long_conditions), "exit_long"] = 1

        exit_short_conditions = [
            df["do_predict"] == 1,
            df["&-s_close"] > weak,
        ]
        if exit_short_conditions:
            df.loc[reduce(lambda x, y: x & y, exit_short_conditions), "exit_short"] = 1

        return df

    def custom_exit(
        self,
        pair: str,
        trade: Trade,
        current_time: datetime,
        current_rate: float,
        current_profit: float,
        **kwargs,
    ) -> str | bool | None:
        df, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)
        if df.empty or current_profit <= 0:
            return None

        pred = float(df.iloc[-1].get("&-s_close", 0) or 0)
        weak = self._weak_prediction_threshold()

        if trade.is_short:
            if pred > -weak:
                return "pred_weak_profit"
            if current_profit > 0.04 and pred > 0:
                return "profit_lock"
        else:
            if pred < weak:
                return "pred_weak_profit"
            if current_profit > 0.04 and pred < 0:
                return "profit_lock"

        return None

    def leverage(
        self,
        pair: str,
        current_time,
        current_rate: float,
        proposed_leverage: float,
        max_leverage: float,
        entry_tag: str | None,
        side: str,
        **kwargs,
    ) -> float:
        return min(self._target_leverage(), max_leverage)

    def custom_stake_amount(
        self,
        pair: str,
        current_time: datetime,
        current_rate: float,
        proposed_stake: float,
        min_stake: float | None,
        max_stake: float,
        leverage: float,
        entry_tag: str | None,
        side: str,
        **kwargs,
    ) -> float:
        df, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)
        if df.empty:
            return proposed_stake

        pred = abs(float(df.iloc[-1].get("&-s_close", 0) or 0))
        thr = self._prediction_threshold()
        if pred <= thr:
            return proposed_stake

        max_mult = float(self._strategy_params()["max_stake_pred_mult"])
        multiplier = min(pred / thr, max_mult)
        stake = proposed_stake * multiplier
        floor = min_stake if min_stake is not None else 0.0
        return max(floor, min(stake, max_stake))

    def custom_stoploss(
        self,
        pair: str,
        trade: Trade,
        current_time: datetime,
        current_rate: float,
        current_profit: float,
        after_fill: bool,
        **kwargs,
    ) -> float | None:
        df, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)
        if df.empty:
            return None

        last = df.iloc[-1].squeeze()
        atr_pct = float(last.get("atr_pct", 0.006) or 0.006)
        atr_stop = -min(0.018, max(0.008, atr_pct * 2.0))

        if current_profit > 0.12:
            return -0.02
        if current_profit > 0.08:
            return -0.025

        return max(atr_stop, self.stoploss)

    def confirm_trade_entry(
        self,
        pair: str,
        order_type: str,
        amount: float,
        rate: float,
        time_in_force: str,
        current_time,
        entry_tag,
        side: str,
        **kwargs,
    ) -> bool:
        df, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)
        if df.empty:
            return False

        last_candle = df.iloc[-1].squeeze()
        atr_pct = float(last_candle.get("atr_pct", 0) or 0)

        max_atr = self._max_atr_pct()
        if atr_pct > max_atr:
            logger.info(
                "Entry rejected %s: ATR%% %.4f above max %.4f", pair, atr_pct, max_atr
            )
            return False

        if "DI_values" in df.columns:
            di_val = float(last_candle.get("DI_values", 0) or 0)
            if di_val >= self._di_threshold():
                logger.info(
                    "Entry rejected %s: DI_values %.4f >= %.4f", pair, di_val, self._di_threshold()
                )
                return False

        pred = float(last_candle.get("&-s_close", 0) or 0)
        strong_thr = self._strong_prediction_threshold()
        if side == "long" and pred <= strong_thr:
            return False
        if side == "short" and pred >= -strong_thr:
            return False

        if side == "long":
            if rate > (last_candle["close"] * (1 + 0.0025)):
                return False
        else:
            if rate < (last_candle["close"] * (1 - 0.0025)):
                return False

        return True