# source: https://raw.githubusercontent.com/CrocoPruki/GalaxyWatch_BitGetApp_WearOs/a09a4c01078d3dd5ffc96388a812e64006284702/MACDL_LB.py
from __future__ import annotations

from datetime import datetime
from typing import Any

from pandas import DataFrame
from sklearn.cluster import KMeans

import talib.abstract as ta
from freqtrade.persistence import Trade
from freqtrade.strategy import DecimalParameter, IntParameter, IStrategy, stoploss_from_absolute


class Github_CrocoPruki_GalaxyWatch_BitGetApp_WearOs__MACDL_LB__20260719_235030(IStrategy):
    """
    Price-action + Github_CrocoPruki_GalaxyWatch_BitGetApp_WearOs__MACDL_LB__20260719_235030 strategy with lightweight KMeans S/R levels.
    Built for 5m timeframe with CPU-friendly behavior for live/dry-run.
    """

    timeframe = "5m"
    startup_candle_count: int = 500

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

    buy_macdl_level = IntParameter(350, 450, default=400, space="buy")
    sell_macdl_level = IntParameter(50, 150, default=100, space="sell")

    max_ema_dist = DecimalParameter(0.001, 0.01, default=0.005, space="buy")
    climax_multiplier = DecimalParameter(1.5, 3.5, default=2.5, space="sell")

    kmeans_lookback = 48
    kmeans_clusters = 3
    risk_per_trade = 0.03
    stop_lookback_candles = 3
    long_stop_buffer = 0.999
    short_stop_buffer = 1.001

    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:
            # Fast approximation for backtests: rolling quantiles as dynamic S/R.
            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:
            # Fail-safe: keep strategy running if clustering fails.
            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 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

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

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

        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 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

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

    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"] > 450)
            ),
            "exit_long",
        ] = 1

        dataframe.loc[
            (
                (dataframe["body_size"] > (dataframe["avg_body"] * self.climax_multiplier.value))
                & (dataframe["macdl_lb"] < 50)
            ),
            "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.5):
                return False

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

        return True