# source: https://raw.githubusercontent.com/kuker24/NewEra/f8cef4e1d0889a99408c80ecdc84f6de680ca59e/analysis_exports/claw_simons_20260421/tuning_round2/SimonsSniperV1.py
"""Github_kuker24_NewEra__SimonsSniperV1__20260423_015443 strategy for anomaly-based long entries on spot markets."""

from __future__ import annotations

import logging
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, Series

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

logger = logging.getLogger(__name__)

ENTRY_TAG_SIMONS_ANOMALY = "simons_anomaly"


class Github_kuker24_NewEra__SimonsSniperV1__20260423_015443(IStrategy):
    """An anomaly-driven long-only strategy with ATR-based dynamic stoploss."""

    INTERFACE_VERSION = 3
    timeframe = "15m"
    can_short = False

    stoploss = -0.99
    use_custom_stoploss = True
    trailing_stop = False

    minimal_roi = {
        "0": 0.04,
        "60": 0.02,
        "180": 0.01,
        "360": 0,
    }

    startup_candle_count = 100

    zscore_window = IntParameter(60, 95, default=81, space="buy")
    anomaly_threshold = DecimalParameter(1.7, 2.4, default=2.0, space="buy")
    roc_period = IntParameter(10, 22, default=15, space="buy")

    atr_period = IntParameter(12, 20, default=14, space="sell")
    atr_stop_multiplier = DecimalParameter(1.5, 2.3, default=1.65, space="sell")

    def populate_indicators(self, dataframe: DataFrame, metadata: dict[str, Any]) -> DataFrame:
        """Compute vectorized indicators used by entry and custom stoploss logic."""
        _ = metadata

        window = int(self.zscore_window.value)

        close_mean = dataframe["close"].rolling(window=window, min_periods=window).mean()
        close_std = (
            dataframe["close"]
            .rolling(window=window, min_periods=window)
            .std(ddof=0)
            .replace(0.0, np.nan)
        )
        close_zscore = (dataframe["close"] - close_mean) / close_std

        volume_mean = dataframe["volume"].rolling(window=window, min_periods=window).mean()
        volume_std = (
            dataframe["volume"]
            .rolling(window=window, min_periods=window)
            .std(ddof=0)
            .replace(0.0, np.nan)
        )
        volume_zscore = (dataframe["volume"] - volume_mean) / volume_std

        dataframe["close_zscore"] = close_zscore.replace([np.inf, -np.inf], np.nan)
        dataframe["volume_zscore"] = volume_zscore.replace([np.inf, -np.inf], np.nan)
        dataframe["atr"] = ta.ATR(dataframe, timeperiod=int(self.atr_period.value))
        dataframe["roc"] = ta.ROC(dataframe, timeperiod=int(self.roc_period.value))

        return dataframe

    def _calculate_anomaly_score(self, dataframe: DataFrame) -> Series:
        """Calculate anomaly score from separate mean-reversion and momentum components."""
        mean_rev_signal = (-dataframe["close_zscore"]).clip(lower=0.0) * dataframe[
            "volume_zscore"
        ].clip(lower=0.0)
        momentum_signal = dataframe["close_zscore"].clip(lower=0.0) * (
            dataframe["roc"] / 100.0
        ).clip(lower=0.0)

        anomaly_score = pd.concat([mean_rev_signal, momentum_signal], axis=1).max(axis=1)
        return anomaly_score

    def populate_entry_trend(self, dataframe: DataFrame, metadata: dict[str, Any]) -> DataFrame:
        """Create long entries when anomaly score exceeds configured threshold."""
        _ = metadata

        dataframe["anomaly_score"] = self._calculate_anomaly_score(dataframe)
        dataframe["enter_long"] = 0
        dataframe.loc[
            dataframe["anomaly_score"] > self.anomaly_threshold.value,
            ["enter_long", "enter_tag"],
        ] = (1, ENTRY_TAG_SIMONS_ANOMALY)

        return dataframe

    def populate_exit_trend(self, dataframe: DataFrame, metadata: dict[str, Any]) -> DataFrame:
        """Disable indicator-driven exits and rely on ROI plus custom stoploss."""
        _ = metadata

        dataframe["exit_long"] = 0
        return dataframe

    def custom_stoploss(
        self,
        pair: str,
        trade: Trade,
        current_time: datetime,
        current_rate: float,
        current_profit: float,
        after_fill: bool,
        **kwargs: Any,
    ) -> float | None:
        """Return ATR-based stop distance, clamped by the hard strategy stoploss."""
        _ = (current_time, current_profit, after_fill, kwargs)

        if self.dp is None:
            logger.debug("DataProvider unavailable. Keeping existing stoploss for %s.", pair)
            return None

        dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)
        if dataframe is None or dataframe.empty:
            logger.debug("No analyzed dataframe for %s on timeframe %s.", pair, self.timeframe)
            return None

        last_candle = dataframe.iloc[-1]
        last_atr = float(last_candle.get("atr", np.nan))

        if not np.isfinite(last_atr) or last_atr <= 0.0:
            logger.debug("ATR is invalid for %s. last_atr=%s", pair, last_atr)
            return None

        if current_rate <= 0.0 or trade.open_rate <= 0.0:
            return None

        absolute_stop = trade.open_rate - (last_atr * float(self.atr_stop_multiplier.value))
        if not np.isfinite(absolute_stop) or absolute_stop <= 0.0:
            logger.debug("Calculated absolute stop is invalid for %s. stop=%s", pair, absolute_stop)
            return None

        stop_distance = stoploss_from_absolute(
            stop_rate=absolute_stop,
            current_rate=current_rate,
            is_short=trade.is_short,
            leverage=trade.leverage or 1.0,
        )

        if not np.isfinite(stop_distance):
            return None

        clamped_stop = min(max(float(stop_distance), 0.0), abs(self.stoploss))
        if clamped_stop <= 0.0:
            return None

        return clamped_stop