from __future__ import annotations

from datetime import datetime
from typing import Any

import talib.abstract as ta
from pandas import DataFrame
from sklearn.cluster import KMeans

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


class Agresywna(IStrategy):
    """
    More aggressive scalp-oriented variant of MACDL_LB.
    It accepts more entries, uses a shorter context window, and exits faster.
    """

    timeframe = "5m"
    startup_candle_count: int = 300

    stoploss = -0.12
    trailing_stop = False
    process_only_new_candles = True
    use_custom_stoploss = True
    can_short = True

    buy_macdl_level = IntParameter(380, 470, default=430, space="buy")
    sell_macdl_level = IntParameter(30, 120, default=70, space="sell")
    max_ema_dist = DecimalParameter(0.002, 0.015, default=0.008, space="buy")
    climax_multiplier = DecimalParameter(1.2, 2.5, default=1.8, space="sell")

    kmeans_lookback = 32
    kmeans_clusters = 4
    risk_per_trade = 0.03
    stop_lookback_candles = 2
    long_stop_buffer = 0.9995
    short_stop_buffer = 1.0005

    minimal_roi = {
        "0": 0.02,
        "15": 0.012,
        "45": 0.0,
    }

    def _is_live_like(self) -> bool:
        runmode: Any = self.config.get("runmode") if self.config else None
        if hasattr(runmode, "value"):
            runmode = runmode.value
        return runmode in ("live", "dry_run")

    def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        dataframe["ema20"] = ta.EMA(dataframe, timeperiod=20)

        macd = ta.MACD(dataframe)
        macd_val = macd["macd"]
        rolling_min = macd_val.rolling(window=self.startup_candle_count, min_periods=50).min()
        rolling_max = macd_val.rolling(window=self.startup_candle_count, min_periods=50).max()

        denom = (rolling_max - rolling_min).replace(0, float("nan"))
        dataframe["macdl_lb"] = (500 * (macd_val - rolling_min) / denom).clip(0, 500).fillna(250)

        dataframe["inside_bar"] = (dataframe["high"] < dataframe["high"].shift(1)) & (
            dataframe["low"] > dataframe["low"].shift(1)
        )
        dataframe["ii_pattern"] = dataframe["inside_bar"] & dataframe["inside_bar"].shift(1)

        dataframe["higher_high"] = dataframe["high"] > dataframe["high"].shift(1)
        dataframe["lower_high"] = dataframe["high"] < dataframe["high"].shift(1)
        dataframe["pullback"] = dataframe["lower_high"].rolling(3, min_periods=3).sum() >= 2
        dataframe["h2_signal"] = dataframe["higher_high"] & dataframe["pullback"].shift(1)

        if self._is_live_like():
            dataframe = self._calculate_kmeans_live(dataframe)
        else:
            lb = self.kmeans_lookback
            dataframe["cluster_support"] = dataframe["low"].rolling(lb, min_periods=10).quantile(0.2)
            dataframe["cluster_resistance"] = dataframe["high"].rolling(lb, min_periods=10).quantile(0.8)

        dataframe["cluster_support"] = dataframe["cluster_support"].ffill().bfill()
        dataframe["cluster_resistance"] = dataframe["cluster_resistance"].ffill().bfill()

        return dataframe

    def _calculate_kmeans_live(self, dataframe: DataFrame) -> DataFrame:
        if "cluster_support" not in dataframe.columns:
            dataframe["cluster_support"] = float("nan")
        if "cluster_resistance" not in dataframe.columns:
            dataframe["cluster_resistance"] = float("nan")

        if len(dataframe) < max(self.kmeans_lookback, 10):
            dataframe["cluster_support"] = dataframe["low"].rolling(10, min_periods=1).min()
            dataframe["cluster_resistance"] = dataframe["high"].rolling(10, min_periods=1).max()
            return dataframe

        try:
            price_points = (
                dataframe[["high", "low", "close"]]
                .tail(self.kmeans_lookback)
                .to_numpy()
                .reshape(-1, 1)
            )
            kmeans = KMeans(n_clusters=self.kmeans_clusters, n_init=5, random_state=42)
            kmeans.fit(price_points)
            centers = sorted(kmeans.cluster_centers_.flatten())

            dataframe.loc[dataframe.index[-1], "cluster_support"] = centers[0]
            dataframe.loc[dataframe.index[-1], "cluster_resistance"] = centers[-1]
        except Exception:
            dataframe.loc[dataframe.index[-1], "cluster_support"] = dataframe["low"].tail(10).min()
            dataframe.loc[dataframe.index[-1], "cluster_resistance"] = dataframe["high"].tail(10).max()

        dataframe["cluster_support"] = dataframe["cluster_support"].ffill()
        dataframe["cluster_resistance"] = dataframe["cluster_resistance"].ffill()
        return dataframe

    def _absolute_stop_from_dataframe(self, dataframe: DataFrame, is_short: bool) -> float | None:
        if dataframe is None or dataframe.empty:
            return None

        if is_short:
            recent_high = dataframe["high"].tail(self.stop_lookback_candles).max()
            return float(recent_high) * self.short_stop_buffer

        recent_low = dataframe["low"].tail(self.stop_lookback_candles).min()
        return float(recent_low) * self.long_stop_buffer

    def _resolve_total_stake_capital(self, fallback_value: float) -> float:
        wallets = getattr(self, "wallets", None)
        if wallets is None:
            return fallback_value

        for attr_name in ("get_total_stake_amount", "get_available_stake_amount"):
            getter = getattr(wallets, attr_name, None)
            if callable(getter):
                try:
                    resolved = float(getter())
                except Exception:
                    continue
                if resolved > 0:
                    return resolved

        return fallback_value

    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:
        dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)
        if dataframe is None or dataframe.empty or current_rate <= 0:
            return proposed_stake

        is_short = side == "short"
        stop_abs = self._absolute_stop_from_dataframe(dataframe, is_short)
        if stop_abs is None or stop_abs <= 0:
            return proposed_stake

        stop_distance_ratio = abs(current_rate - stop_abs) / current_rate
        effective_leverage = max(1.0, float(leverage or 1.0))
        leveraged_loss_ratio = stop_distance_ratio * effective_leverage
        if leveraged_loss_ratio <= 0:
            return proposed_stake

        capital_base = self._resolve_total_stake_capital(float(proposed_stake))
        risk_budget = capital_base * self.risk_per_trade
        stake = risk_budget / leveraged_loss_ratio

        if min_stake is not None:
            stake = max(stake, float(min_stake))
        stake = min(stake, float(max_stake))

        return max(0.0, float(stake))

    def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        macdl_growing = dataframe["macdl_lb"] > dataframe["macdl_lb"].shift(1)
        macdl_falling = dataframe["macdl_lb"] < dataframe["macdl_lb"].shift(1)

        dist_to_ema = (dataframe["close"] - dataframe["ema20"]).abs() / dataframe["close"]
        at_ema_magnet = dist_to_ema < self.max_ema_dist.value

        bullish_break = (dataframe["close"] > dataframe["high"].shift(1)) & (
            dataframe["close"] > dataframe["ema20"]
        )
        bearish_break = (dataframe["close"] < dataframe["low"].shift(1)) & (
            dataframe["close"] < dataframe["ema20"]
        )

        dataframe.loc[
            (
                (dataframe["macdl_lb"] < self.buy_macdl_level.value)
                & macdl_growing
                & at_ema_magnet
                & (
                    dataframe["h2_signal"]
                    | dataframe["ii_pattern"]
                    | (dataframe["low"] <= dataframe["cluster_support"])
                    | bullish_break
                )
                & (dataframe["close"] >= dataframe["open"])
            ),
            "enter_long",
        ] = 1

        dataframe.loc[
            (
                (dataframe["macdl_lb"] > self.sell_macdl_level.value)
                & macdl_falling
                & at_ema_magnet
                & (
                    (dataframe["high"] >= dataframe["cluster_resistance"])
                    | dataframe["ii_pattern"]
                    | bearish_break
                )
                & (dataframe["close"] <= dataframe["open"])
            ),
            "enter_short",
        ] = 1

        return dataframe

    def custom_stoploss(
        self,
        pair: str,
        trade: Trade,
        current_time: datetime,
        current_rate: float,
        current_profit: float,
        **kwargs,
    ) -> float:
        dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)
        if dataframe is None or dataframe.empty:
            return self.stoploss

        sl_abs = self._absolute_stop_from_dataframe(dataframe, trade.is_short)
        if sl_abs is None:
            return self.stoploss

        dyn_sl = stoploss_from_absolute(
            sl_abs,
            current_rate=current_rate,
            is_short=trade.is_short,
            leverage=max(1.0, float(getattr(trade, "leverage", 1.0))),
        )

        if dyn_sl is None:
            return self.stoploss

        dyn_sl = max(float(dyn_sl), self.stoploss)

        if current_profit > 0.03:
            return max(dyn_sl, -0.01)

        if current_profit > 0.015:
            return max(dyn_sl, -0.02)

        return dyn_sl

    def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        dataframe["body_size"] = (dataframe["close"] - dataframe["open"]).abs()
        dataframe["avg_body"] = dataframe["body_size"].rolling(5, min_periods=5).mean()

        dataframe.loc[
            (
                (dataframe["body_size"] > (dataframe["avg_body"] * self.climax_multiplier.value))
                & (dataframe["macdl_lb"] > 420)
            )
            | ((dataframe["macdl_lb"] > 470) & (dataframe["close"] < dataframe["close"].shift(1))),
            "exit_long",
        ] = 1

        dataframe.loc[
            (
                (dataframe["body_size"] > (dataframe["avg_body"] * self.climax_multiplier.value))
                & (dataframe["macdl_lb"] < 80)
            )
            | ((dataframe["macdl_lb"] < 30) & (dataframe["close"] > dataframe["close"].shift(1))),
            "exit_short",
        ] = 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: str | None,
        side: str,
        **kwargs,
    ) -> bool:
        dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)
        if dataframe is None or dataframe.empty:
            return False

        last_candle = dataframe.iloc[-1]
        resistance = float(last_candle.get("cluster_resistance", rate))
        support = float(last_candle.get("cluster_support", rate))

        if side == "long":
            risk = max(1e-9, rate - float(last_candle["low"]))
            reward = max(0.0, resistance - rate)
            if reward < (risk * 1.1):
                return False

        if side == "short":
            risk = max(1e-9, float(last_candle["high"]) - rate)
            reward = max(0.0, rate - support)
            if reward < (risk * 1.1):
                return False

        return True