"""
ClawQuantAdaptiveV2

Spot-only adaptive ensemble strategy for Freqtrade V3.

Design notes:
- This strategy is strictly spot-only (`can_short = False`).
- Exit signal is only one exit layer. `minimal_roi`, `stoploss`, and `trailing_stop*`
  remain additional safety nets.
- Runtime config overlays may override strategy defaults.
- `startup_candle_count = 250` is a safe baseline and MUST be validated using
  `recursive-analysis`; increase it when recursive variance is not yet 0%.
"""

from __future__ import annotations

from datetime import datetime
from typing import Any

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

from freqtrade.persistence import Trade
from freqtrade.strategy import DecimalParameter, IntParameter, IStrategy, merge_informative_pair


class ClawQuantAdaptiveV2(IStrategy):
    INTERFACE_VERSION = 3
    can_short = False
    timeframe = "15m"
    startup_candle_count = 250

    minimal_roi = {"0": 0.03, "60": 0.015, "180": 0}
    stoploss = -0.08
    trailing_stop = True
    trailing_stop_positive = 0.01
    trailing_stop_positive_offset = 0.02
    process_only_new_candles = True
    use_exit_signal = True
    exit_profit_only = False

    order_types = {
        "entry": "limit",
        "exit": "limit",
        "stoploss": "market",
        "stoploss_on_exchange": True,
        "emergency_exit": "market",
        "force_entry": "market",
        "force_exit": "market",
    }

    order_time_in_force = {
        "entry": "GTC",
        "exit": "GTC",
    }

    mom_score_threshold = IntParameter(3, 5, default=4, space="buy")
    mr_score_threshold = IntParameter(3, 5, default=4, space="buy")
    mr_bb_period = IntParameter(18, 24, default=20, space="buy")
    mr_bb_std = DecimalParameter(2.0, 3.0, decimals=1, default=2.5, space="buy")
    entry_cooldown_candles = IntParameter(2, 8, default=5, space="buy")

    # Exit guardrails tuned from 2-day pre-tuning dataset (2026-04-12..2026-04-13):
    # `exit_signal_mean_return` had the weakest expectancy and higher loss ratio than ROI exits.
    # We require stronger recovery before signal-based exits to reduce premature churn.
    meanrev_exit_mid_buffer = 1.001
    meanrev_exit_rsi_floor = 50
    meanrev_exit_rsi_recovery = 62

    @staticmethod
    def _bb_key(period: int, std: float) -> str:
        return f"p{int(period)}_s{int(round(float(std) * 10))}"

    def _active_bb_columns(self) -> tuple[str, str]:
        key = self._bb_key(int(self.mr_bb_period.value), float(self.mr_bb_std.value))
        return f"bb_lower_{key}", f"bb_middle_{key}"

    @staticmethod
    def _safe_float(value: Any) -> float | None:
        if value is None:
            return None
        try:
            out = float(value)
        except Exception:
            return None
        if np.isnan(out):
            return None
        return out

    def informative_pairs(self):
        return [("BTC/USDT", "1d")]

    def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        dataframe["sma_50"] = ta.SMA(dataframe, timeperiod=50)
        dataframe["rsi"] = ta.RSI(dataframe, timeperiod=14)
        dataframe["adx"] = ta.ADX(dataframe, timeperiod=14)
        dataframe["mfi"] = ta.MFI(dataframe, timeperiod=14)
        dataframe["willr"] = ta.WILLR(dataframe, timeperiod=14)
        dataframe["volume_sma_24"] = dataframe["volume"].rolling(24, min_periods=1).mean()

        macd = ta.MACD(dataframe)
        dataframe["macd"] = macd["macd"]
        dataframe["macdsignal"] = macd["macdsignal"]
        dataframe["macdhist"] = macd["macdhist"]

        # Hyperopt-safe Bollinger precompute across full parameter ranges.
        for period in self.mr_bb_period.range:
            p_int = int(period)
            for std in self.mr_bb_std.range:
                s_float = float(std)
                bands = ta.BBANDS(
                    dataframe,
                    timeperiod=p_int,
                    nbdevup=s_float,
                    nbdevdn=s_float,
                    matype=0,
                )
                key = self._bb_key(p_int, s_float)
                dataframe[f"bb_lower_{key}"] = bands["lowerband"]
                dataframe[f"bb_middle_{key}"] = bands["middleband"]
                dataframe[f"bb_upper_{key}"] = bands["upperband"]

        dataframe["btc_bull_1d"] = False
        dataframe["btc_relief_1d"] = False
        dataframe["btc_short_up_1d"] = False
        dataframe["btc_regime_ok_1d"] = False

        if self.dp:
            btc_df = self.dp.get_pair_dataframe(pair="BTC/USDT", timeframe="1d")
            if btc_df is not None and not btc_df.empty:
                btc_df = btc_df.copy()
                btc_df["btc_sma_200"] = ta.SMA(btc_df, timeperiod=200)
                btc_df["btc_ema_20"] = ta.EMA(btc_df, timeperiod=20)
                btc_df["btc_roc_5"] = ta.ROC(btc_df, timeperiod=5)
                btc_df["btc_bull"] = btc_df["close"] > btc_df["btc_sma_200"]
                btc_df["btc_relief"] = btc_df["close"] > btc_df["btc_ema_20"]
                btc_df["btc_short_up"] = btc_df["btc_roc_5"] > 0
                btc_df["btc_regime_ok"] = btc_df["btc_bull"] | (
                    btc_df["btc_relief"] & btc_df["btc_short_up"]
                )

                dataframe = merge_informative_pair(
                    dataframe,
                    btc_df[
                        [
                            "date",
                            "btc_sma_200",
                            "btc_ema_20",
                            "btc_roc_5",
                            "btc_bull",
                            "btc_relief",
                            "btc_short_up",
                            "btc_regime_ok",
                        ]
                    ],
                    self.timeframe,
                    "1d",
                    ffill=True,
                )

                for col in ["btc_bull", "btc_relief", "btc_short_up", "btc_regime_ok"]:
                    merged_col = f"{col}_1d"
                    if merged_col in dataframe.columns:
                        dataframe[merged_col] = dataframe[merged_col].fillna(False).astype(bool)

        for required_col in ["btc_bull_1d", "btc_relief_1d", "btc_short_up_1d", "btc_regime_ok_1d"]:
            if required_col not in dataframe.columns:
                dataframe[required_col] = False
            dataframe[required_col] = dataframe[required_col].fillna(False).astype(bool)

        active_lower_col, active_middle_col = self._active_bb_columns()
        dataframe["bb_lower_active"] = np.nan
        dataframe["bb_middle_active"] = np.nan
        if active_lower_col in dataframe.columns:
            dataframe["bb_lower_active"] = dataframe[active_lower_col]
        if active_middle_col in dataframe.columns:
            dataframe["bb_middle_active"] = dataframe[active_middle_col]

        return dataframe

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

        score_m1 = dataframe["close"] > dataframe["sma_50"]
        score_m2 = dataframe["rsi"] > 55
        score_m3 = dataframe["macd"] > dataframe["macdsignal"]
        score_m4 = dataframe["adx"] > 25
        score_m5 = dataframe["volume"] > (dataframe["volume_sma_24"] * 1.1)

        dataframe["mom_score"] = (
            score_m1.astype(int)
            + score_m2.astype(int)
            + score_m3.astype(int)
            + score_m4.astype(int)
            + score_m5.astype(int)
        )

        score_mr1 = dataframe["close"] < dataframe["bb_lower_active"]
        score_mr2 = dataframe["rsi"] < 30
        score_mr3 = dataframe["willr"] < -80
        score_mr4 = dataframe["mfi"] < 20
        score_mr5 = dataframe["volume"] > (dataframe["volume_sma_24"] * 1.5)

        dataframe["mr_score"] = (
            score_mr1.astype(int)
            + score_mr2.astype(int)
            + score_mr3.astype(int)
            + score_mr4.astype(int)
            + score_mr5.astype(int)
        )

        regime_ok = dataframe.get("btc_regime_ok_1d", pd.Series(False, index=dataframe.index))
        regime_ok = regime_ok.fillna(False).astype(bool)

        raw_entry_momentum = (
            regime_ok
            & (dataframe["mom_score"] >= int(self.mom_score_threshold.value))
            & (dataframe["volume"] > 0)
        )

        raw_entry_meanrev = (
            (~regime_ok)
            & (dataframe["mr_score"] >= int(self.mr_score_threshold.value))
            & (dataframe["volume"] > 0)
        )

        raw_entry_signal = raw_entry_momentum | raw_entry_meanrev

        recent_raw_entries = (
            raw_entry_signal.shift(1)
            .rolling(window=int(self.entry_cooldown_candles.value), min_periods=1)
            .sum()
            .fillna(0)
        )
        cooldown_ok = recent_raw_entries == 0

        final_momentum_entry = raw_entry_momentum & cooldown_ok
        final_meanrev_entry = raw_entry_meanrev & cooldown_ok
        final_entry = final_momentum_entry | final_meanrev_entry

        # Recorder-friendly observability columns (for skip-reason telemetry).
        dataframe["raw_entry_signal"] = raw_entry_signal.astype(int)
        dataframe["cooldown_ok"] = cooldown_ok.astype(int)
        dataframe["entry_ready_before_system_gate"] = final_entry.astype(int)
        dataframe["skip_reason_strategy"] = "no_raw_signal"
        dataframe.loc[raw_entry_signal & ~cooldown_ok, "skip_reason_strategy"] = "cooldown_blocked"
        dataframe.loc[final_entry, "skip_reason_strategy"] = "entry_ready"

        dataframe.loc[final_entry, "enter_long"] = 1

        dataframe.loc[final_momentum_entry, "enter_tag"] = (
            "entry_momentum_score_"
            + dataframe.loc[final_momentum_entry, "mom_score"].astype(int).astype(str)
        )
        dataframe.loc[final_meanrev_entry, "enter_tag"] = (
            "entry_meanrev_score_"
            + dataframe.loc[final_meanrev_entry, "mr_score"].astype(int).astype(str)
        )

        return dataframe

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

        valid_volume = dataframe["volume"] > 0

        exit_signal_mom_weak = (
            (dataframe["rsi"] > 72)
            & (dataframe["macd"] < dataframe["macdsignal"])
            & (dataframe["adx"] < 25)
        )

        exit_signal_mean_return = (
            (dataframe["close"] > (dataframe["bb_middle_active"] * self.meanrev_exit_mid_buffer))
            & (dataframe["rsi"] > self.meanrev_exit_rsi_floor)
        )

        exit_signal_rsi_recovery = (
            (dataframe["rsi"] > self.meanrev_exit_rsi_recovery)
            & (dataframe["willr"] > -30)
        )

        dataframe.loc[exit_signal_mom_weak & valid_volume, "exit_tag"] = "exit_signal_mom_weak"
        dataframe.loc[
            exit_signal_mean_return & valid_volume & dataframe["exit_tag"].isna(),
            "exit_tag",
        ] = "exit_signal_mean_return"
        dataframe.loc[
            exit_signal_rsi_recovery & valid_volume & dataframe["exit_tag"].isna(),
            "exit_tag",
        ] = "exit_signal_rsi_recovery"

        exit_any = (
            (exit_signal_mom_weak | exit_signal_mean_return | exit_signal_rsi_recovery)
            & valid_volume
        )
        dataframe.loc[exit_any, "exit_long"] = 1

        return dataframe

    def custom_exit(
        self,
        pair: str,
        trade: Trade,
        current_time: datetime,
        current_rate: float,
        current_profit: float,
        **kwargs,
    ) -> str | None:
        if not self.dp:
            return None

        dataframe, _ = self.dp.get_analyzed_dataframe(pair=pair, timeframe=self.timeframe)
        if dataframe is None or dataframe.empty:
            return None

        last_candle = dataframe.iloc[-1]

        rsi = self._safe_float(last_candle.get("rsi"))
        adx = self._safe_float(last_candle.get("adx"))
        macd = self._safe_float(last_candle.get("macd"))
        macdsignal = self._safe_float(last_candle.get("macdsignal"))
        close = self._safe_float(last_candle.get("close"))

        _, active_middle_col = self._active_bb_columns()
        bb_middle = self._safe_float(last_candle.get(active_middle_col))
        if bb_middle is None:
            bb_middle = self._safe_float(last_candle.get("bb_middle_active"))

        enter_tag = str(trade.enter_tag or "")

        if "entry_momentum_" in enter_tag:
            if rsi is not None and rsi > 75:
                return "exit_mom_overbought"

            if (
                macd is not None
                and macdsignal is not None
                and adx is not None
                and macd < macdsignal
                and adx < 20
            ):
                return "exit_mom_macd_weak"

        if "entry_meanrev_" in enter_tag:
            if (
                close is not None
                and bb_middle is not None
                and close > (bb_middle * self.meanrev_exit_mid_buffer)
                and rsi is not None
                and rsi > self.meanrev_exit_rsi_floor
            ):
                return "exit_mr_mean_return"

            if rsi is not None and rsi > self.meanrev_exit_rsi_recovery:
                return "exit_mr_mean_return"

        return None