# source: https://raw.githubusercontent.com/Danson77/Freqtrade_AIO_UI/f0bf29851d0577f244c17c0aa7d593ba3a7f3551/user_data/backtest_reports/Registerd_with_1000USDT/ClucHAnix_5m1/ClucHAnix_5m1.py
# directory_url: https://github.com/Danson77/Freqtrade_AIO_UI/blob/main/user_data/backtest_reports/Registerd_with_1000USDT/ClucHAnix_5m1/
# User: Danson77
# Repository: Freqtrade_AIO_UI
# --------------------import logging
from datetime import datetime, timezone
from typing import Dict, List

import freqtrade.vendor.qtpylib.indicators as qtpylib
import numpy as np
import pandas as pd
import talib.abstract as ta

from freqtrade.persistence import Trade
from freqtrade.strategy import (
    DecimalParameter,
    RealParameter,
    merge_informative_pair,
    stoploss_from_open,
)
from freqtrade.strategy.interface import IStrategy
from pandas import DataFrame, Series


logger = logging.getLogger(__name__)


def bollinger_bands(stock_price, window_size, num_of_std):
    rolling_mean = stock_price.rolling(window=window_size).mean()
    rolling_std = stock_price.rolling(window=window_size).std()
    lower_band = rolling_mean - (rolling_std * num_of_std)
    return np.nan_to_num(rolling_mean), np.nan_to_num(lower_band)


def ha_typical_price(bars):
    res = (bars['ha_high'] + bars['ha_low'] + bars['ha_close']) / 3.0
    return Series(index=bars.index, data=res)


class Github_Danson77_Freqtrade_AIO_UI__ClucHAnix_5m1__20260520_191933(IStrategy):
    INTERFACE_VERSION = 3

    buy_params = {
        "bbdelta_close": 0.01889,
        "bbdelta_tail": 0.72235,
        "close_bblower": 0.0127,
        "closedelta_close": 0.00916,
        "rocr_1h": 0.79492,
    }

    sell_params = {
        "pHSL": -0.10,
        "pPF_1": 0.011,
        "pPF_2": 0.064,
        "pSL_1": 0.011,
        "pSL_2": 0.062,
        "sell_fisher": 0.39075,
        "sell_bbmiddle_close": 0.99754,
    }

    minimal_roi = {
        "0": 100
    }

    stoploss = -0.99
    trailing_stop = False
    trailing_stop_positive = 0.001
    trailing_stop_positive_offset = 0.012
    trailing_only_offset_is_reached = False

    timeframe = "5m"
    inf_tf = "1h"
    can_short = False

    use_exit_signal = True
    exit_profit_only = False
    ignore_roi_if_entry_signal = False

    use_custom_stoploss = True
    process_only_new_candles = True
    startup_candle_count = 168

    order_types = {
        "entry": "market",
        "exit": "market",
        "emergency_exit": "market",
        "force_entry": "market",
        "force_exit": "market",
        "stoploss": "market",
        "stoploss_on_exchange": False,
        "stoploss_on_exchange_interval": 60,
        "stoploss_on_exchange_limit_ratio": 0.99,
    }

    # buy params
    rocr_1h = RealParameter(0.5, 1.0, default=0.54904, space="buy", optimize=True)
    bbdelta_close = RealParameter(0.0005, 0.02, default=0.01965, space="buy", optimize=True)
    closedelta_close = RealParameter(0.0005, 0.02, default=0.00556, space="buy", optimize=True)
    bbdelta_tail = RealParameter(0.7, 1.0, default=0.95089, space="buy", optimize=True)
    close_bblower = RealParameter(0.0005, 0.02, default=0.00799, space="buy", optimize=True)

    # sell params
    sell_fisher = RealParameter(0.1, 0.5, default=0.38414, space="sell", optimize=True)
    sell_bbmiddle_close = RealParameter(0.97, 1.1, default=1.07634, space="sell", optimize=True)

    # stoploss params
    pHSL = DecimalParameter(-0.500, -0.040, default=-0.08, decimals=3, space="sell", load=True)
    pPF_1 = DecimalParameter(0.008, 0.020, default=0.016, decimals=3, space="sell", load=True)
    pSL_1 = DecimalParameter(0.008, 0.020, default=0.011, decimals=3, space="sell", load=True)
    pPF_2 = DecimalParameter(0.040, 0.100, default=0.080, decimals=3, space="sell", load=True)
    pSL_2 = DecimalParameter(0.020, 0.070, default=0.040, decimals=3, space="sell", load=True)

    def informative_pairs(self):
        pairs = self.dp.current_whitelist()
        return [(pair, self.inf_tf) for pair in pairs]

    @staticmethod
    def _ensure_datetime(dataframe: DataFrame) -> DataFrame:
        if dataframe is None or dataframe.empty:
            return dataframe
        if "date" in dataframe.columns:
            dataframe = dataframe.copy()
            dataframe["date"] = pd.to_datetime(dataframe["date"], utc=True, errors="coerce")
            dataframe = dataframe.dropna(subset=["date"])
        return dataframe

    def _add_empty_informative_cols(self, dataframe: DataFrame) -> DataFrame:
        for col in [f"ha_close_{self.inf_tf}", f"rocr_{self.inf_tf}"]:
            if col not in dataframe.columns:
                dataframe[col] = np.nan
        return dataframe

    def custom_stoploss(
        self,
        pair: str,
        trade: Trade,
        current_time: datetime,
        current_rate: float,
        current_profit: float,
        **kwargs
    ) -> float:
        HSL = self.pHSL.value
        PF_1 = self.pPF_1.value
        SL_1 = self.pSL_1.value
        PF_2 = self.pPF_2.value
        SL_2 = self.pSL_2.value

        if current_profit > PF_2:
            sl_profit = SL_2 + (current_profit - PF_2)
        elif current_profit > PF_1:
            sl_profit = SL_1 + ((current_profit - PF_1) * (SL_2 - SL_1) / (PF_2 - PF_1))
        else:
            sl_profit = HSL

        if sl_profit >= current_profit:
            return -0.99

        return stoploss_from_open(sl_profit, current_profit)

    def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        dataframe = dataframe.copy()
        dataframe = self._ensure_datetime(dataframe)

        # Heikin Ashi main timeframe
        if dataframe is not None and not dataframe.empty:
            heikinashi = qtpylib.heikinashi(dataframe)
            dataframe["ha_open"] = heikinashi["open"]
            dataframe["ha_close"] = heikinashi["close"]
            dataframe["ha_high"] = heikinashi["high"]
            dataframe["ha_low"] = heikinashi["low"]
        else:
            dataframe["ha_open"] = np.nan
            dataframe["ha_close"] = np.nan
            dataframe["ha_high"] = np.nan
            dataframe["ha_low"] = np.nan

        # Bollinger
        mid, lower = bollinger_bands(ha_typical_price(dataframe), window_size=40, num_of_std=2)
        dataframe["lower"] = lower
        dataframe["mid"] = mid

        dataframe["bbdelta"] = (dataframe["mid"] - dataframe["lower"]).abs()
        dataframe["closedelta"] = (dataframe["ha_close"] - dataframe["ha_close"].shift()).abs()
        dataframe["tail"] = (dataframe["ha_close"] - dataframe["ha_low"]).abs()

        dataframe["bb_lowerband"] = dataframe["lower"]
        dataframe["bb_middleband"] = dataframe["mid"]

        dataframe["ema_fast"] = ta.EMA(dataframe["ha_close"], timeperiod=3)
        dataframe["ema_slow"] = ta.EMA(dataframe["ha_close"], timeperiod=50)
        dataframe["volume_mean_slow"] = dataframe["volume"].rolling(window=30).mean()
        dataframe["rocr"] = ta.ROCR(dataframe["ha_close"], timeperiod=28)

        rsi = ta.RSI(dataframe)
        dataframe["rsi"] = rsi
        rsi_scaled = 0.1 * (rsi - 50)
        dataframe["fisher"] = (np.exp(2 * rsi_scaled) - 1) / (np.exp(2 * rsi_scaled) + 1)

        # informative 1h
        informative = self.dp.get_pair_dataframe(pair=metadata["pair"], timeframe=self.inf_tf)

        if informative is not None and not informative.empty:
            informative = informative.copy()
            informative = self._ensure_datetime(informative)

            if informative is not None and not informative.empty:
                inf_heikinashi = qtpylib.heikinashi(informative)
                informative["ha_close"] = inf_heikinashi["close"]
                informative["rocr"] = ta.ROCR(informative["ha_close"], timeperiod=168)

                dataframe = merge_informative_pair(
                    dataframe,
                    informative,
                    self.timeframe,
                    self.inf_tf,
                    ffill=True,
                )
            else:
                dataframe = self._add_empty_informative_cols(dataframe)
        else:
            dataframe = self._add_empty_informative_cols(dataframe)

        # dynamic offset
        dataframe["perc"] = ((dataframe["high"] - dataframe["low"]) / dataframe["low"] * 100)
        dataframe["avg3_perc"] = ta.EMA(dataframe["perc"], timeperiod=3)

        roll_min = dataframe["perc"].rolling(50).min()
        roll_max = dataframe["perc"].rolling(50).max()
        denom = (roll_max - roll_min).replace(0, np.nan)
        dataframe["norm_perc"] = (dataframe["perc"] - roll_min) / denom
        dataframe["norm_perc"] = dataframe["norm_perc"].fillna(0)

        return dataframe

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

        dataframe.loc[
            (
                dataframe[f"rocr_{self.inf_tf}"].gt(self.rocr_1h.value)
            ) &
            (
                (
                    (dataframe["lower"].shift().gt(0)) &
                    (dataframe["bbdelta"].gt(dataframe["ha_close"] * self.bbdelta_close.value)) &
                    (dataframe["closedelta"].gt(dataframe["ha_close"] * self.closedelta_close.value)) &
                    (dataframe["tail"].lt(dataframe["bbdelta"] * self.bbdelta_tail.value)) &
                    (dataframe["ha_close"].lt(dataframe["lower"].shift())) &
                    (dataframe["ha_close"].le(dataframe["ha_close"].shift()))
                ) |
                (
                    (dataframe["ha_close"] < dataframe["ema_slow"]) &
                    (dataframe["ha_close"] < self.close_bblower.value * dataframe["bb_lowerband"])
                )
            ),
            ["enter_long", "enter_tag"]
        ] = (1, "cluc_buy")

        return dataframe

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

        dataframe.loc[
            (dataframe["fisher"] > self.sell_fisher.value) &
            (dataframe["ha_high"].le(dataframe["ha_high"].shift(1))) &
            (dataframe["ha_high"].shift(1).le(dataframe["ha_high"].shift(2))) &
            (dataframe["ha_close"].le(dataframe["ha_close"].shift(1))) &
            (dataframe["ema_fast"] > dataframe["ha_close"]) &
            ((dataframe["ha_close"] * self.sell_bbmiddle_close.value) > dataframe["bb_middleband"]) &
            (dataframe["volume"] > 0),
            "exit_long"
        ] = 1

        return dataframe


class ClucHAnix_5mTB1(Github_Danson77_Freqtrade_AIO_UI__ClucHAnix_5m1__20260520_191933):
    process_only_new_candles = True

    custom_info_trail_buy = dict()

    trailing_buy_order_enabled = True
    trailing_expire_seconds = 300

    trailing_buy_uptrend_enabled = True
    trailing_expire_seconds_uptrend = 90
    min_uptrend_trailing_profit = 0.02

    debug_mode = True
    trailing_buy_max_stop = 0.01
    trailing_buy_max_buy = 0.002

    init_trailing_dict = {
        "trailing_buy_order_started": False,
        "trailing_buy_order_uplimit": 0,
        "start_trailing_price": 0,
        "buy_tag": None,
        "start_trailing_time": None,
        "offset": 0,
        "allow_trailing": False,
    }

    def trailing_buy(self, pair, reinit=False):
        if pair not in self.custom_info_trail_buy:
            self.custom_info_trail_buy[pair] = dict()
        if reinit or "trailing_buy" not in self.custom_info_trail_buy[pair]:
            self.custom_info_trail_buy[pair]["trailing_buy"] = self.init_trailing_dict.copy()
        return self.custom_info_trail_buy[pair]["trailing_buy"]

    def trailing_buy_info(self, pair: str, current_price: float):
        current_time = datetime.now(timezone.utc)
        if not self.debug_mode:
            return
        trailing_buy = self.trailing_buy(pair)

        try:
            duration = current_time - trailing_buy["start_trailing_time"]
        except TypeError:
            duration = 0

        logger.info(
            f"pair: {pair} : "
            f"start: {trailing_buy['start_trailing_price']:.4f}, "
            f"duration: {duration}, "
            f"current: {current_price:.4f}, "
            f"uplimit: {trailing_buy['trailing_buy_order_uplimit']:.4f}, "
            f"profit: {self.current_trailing_profit_ratio(pair, current_price) * 100:.2f}%, "
            f"offset: {trailing_buy['offset']}"
        )

    def current_trailing_profit_ratio(self, pair: str, current_price: float) -> float:
        trailing_buy = self.trailing_buy(pair)
        if trailing_buy["trailing_buy_order_started"]:
            return (trailing_buy["start_trailing_price"] - current_price) / trailing_buy["start_trailing_price"]
        return 0

    def trailing_buy_offset(self, dataframe, pair: str, current_price: float):
        current_trailing_profit_ratio = self.current_trailing_profit_ratio(pair, current_price)
        last_candle = dataframe.iloc[-1]

        adapt = abs(last_candle.get("norm_perc", 0))
        default_offset = 0.003 * (1 + adapt)

        trailing_buy = self.trailing_buy(pair)
        if not trailing_buy["trailing_buy_order_started"]:
            return default_offset

        current_time = datetime.now(timezone.utc)
        trailing_duration = current_time - trailing_buy["start_trailing_time"]

        if trailing_duration.total_seconds() > self.trailing_expire_seconds:
            if (current_trailing_profit_ratio > 0) and (last_candle.get("enter_long", 0) == 1):
                return "forcebuy"
            return None

        elif (
            self.trailing_buy_uptrend_enabled
            and (trailing_duration.total_seconds() < self.trailing_expire_seconds_uptrend)
            and (current_trailing_profit_ratio < (-1 * self.min_uptrend_trailing_profit))
        ):
            return "forcebuy"

        if current_trailing_profit_ratio < 0:
            return default_offset

        trailing_buy_offset_map = {
            0.06: 0.02,
            0.03: 0.01,
            0: default_offset,
        }

        for key in trailing_buy_offset_map:
            if current_trailing_profit_ratio > key:
                return trailing_buy_offset_map[key]

        return default_offset

    def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        dataframe = super().populate_indicators(dataframe, metadata)
        self.trailing_buy(metadata["pair"])
        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 = None,
        side: str = "long",
        **kwargs
    ) -> bool:
        val = super().confirm_trade_entry(
            pair, order_type, amount, rate, time_in_force, current_time, entry_tag, side, **kwargs
        )

        if val:
            if self.trailing_buy_order_enabled and self.config["runmode"].value in ("live", "dry_run"):
                val = False
                dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)

                if dataframe is not None and len(dataframe) >= 1:
                    last_candle = dataframe.iloc[-1].squeeze()
                    current_price = rate
                    trailing_buy = self.trailing_buy(pair)
                    trailing_buy_offset = self.trailing_buy_offset(dataframe, pair, current_price)

                    if trailing_buy["allow_trailing"]:
                        if (not trailing_buy["trailing_buy_order_started"]) and (last_candle.get("enter_long", 0) == 1):
                            trailing_buy["trailing_buy_order_started"] = True
                            trailing_buy["trailing_buy_order_uplimit"] = last_candle["close"]
                            trailing_buy["start_trailing_price"] = last_candle["close"]
                            trailing_buy["buy_tag"] = last_candle.get("enter_tag", "")
                            trailing_buy["start_trailing_time"] = datetime.now(timezone.utc)
                            trailing_buy["offset"] = 0

                            self.trailing_buy_info(pair, current_price)
                            logger.info(f"start trailing buy for {pair} at {last_candle['close']}")

                        elif trailing_buy["trailing_buy_order_started"]:
                            if trailing_buy_offset == "forcebuy":
                                val = True
                                ratio = "%.2f" % (self.current_trailing_profit_ratio(pair, current_price) * 100)
                                self.trailing_buy_info(pair, current_price)
                                logger.info(
                                    f"price OK for {pair} ({ratio} %, {current_price}), "
                                    f"order may not be triggered if all slots are full"
                                )

                            elif trailing_buy_offset is None:
                                self.trailing_buy(pair, reinit=True)
                                logger.info(f'STOP trailing buy for {pair} because "trailing buy offset" returned None')

                            elif current_price < trailing_buy["trailing_buy_order_uplimit"]:
                                old_uplimit = trailing_buy["trailing_buy_order_uplimit"]
                                self.custom_info_trail_buy[pair]["trailing_buy"]["trailing_buy_order_uplimit"] = min(
                                    current_price * (1 + trailing_buy_offset),
                                    self.custom_info_trail_buy[pair]["trailing_buy"]["trailing_buy_order_uplimit"]
                                )
                                self.custom_info_trail_buy[pair]["trailing_buy"]["offset"] = trailing_buy_offset
                                self.trailing_buy_info(pair, current_price)
                                logger.info(
                                    f"update trailing buy for {pair} at {old_uplimit} -> "
                                    f"{self.custom_info_trail_buy[pair]['trailing_buy']['trailing_buy_order_uplimit']}"
                                )

                            elif current_price < (trailing_buy["start_trailing_price"] * (1 + self.trailing_buy_max_buy)):
                                val = True
                                ratio = "%.2f" % (self.current_trailing_profit_ratio(pair, current_price) * 100)
                                self.trailing_buy_info(pair, current_price)
                                logger.info(
                                    f"current price ({current_price}) > uplimit ({trailing_buy['trailing_buy_order_uplimit']}) "
                                    f"and lower than starting price ({trailing_buy['start_trailing_price'] * (1 + self.trailing_buy_max_buy)}). "
                                    f"OK for {pair} ({ratio} %), order may not be triggered if all slots are full"
                                )

                            elif current_price > (trailing_buy["start_trailing_price"] * (1 + self.trailing_buy_max_stop)):
                                self.trailing_buy(pair, reinit=True)
                                self.trailing_buy_info(pair, current_price)
                                logger.info(
                                    f"STOP trailing buy for {pair} because the price is higher than "
                                    f"starting price * {1 + self.trailing_buy_max_stop}"
                                )
                            else:
                                self.trailing_buy_info(pair, current_price)
                                logger.info(f"price too high for {pair} !")
                    else:
                        logger.info(f"Wait for next buy signal for {pair}")

                if val is True:
                    self.trailing_buy_info(pair, rate)
                    self.trailing_buy(pair, reinit=True)
                    logger.info(f"STOP trailing buy for {pair} because I buy it")

        return val

    def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        dataframe = super().populate_entry_trend(dataframe, metadata)
    
        if dataframe.empty:
            return dataframe
    
        if self.trailing_buy_order_enabled and self.config["runmode"].value in ("live", "dry_run"):
            last_candle = dataframe.iloc[-1]
    
            trailing_buy = self.trailing_buy(metadata["pair"])
    
            enter_long = int(last_candle.get("enter_long", 0))
            enter_tag = str(last_candle.get("enter_tag", "buy signal"))
            close_price = float(last_candle.get("close", 0.0))
    
            if enter_long == 1:
                if not trailing_buy["trailing_buy_order_started"]:
                    open_trades = Trade.get_trades(
                        [
                            Trade.pair == metadata["pair"],
                            Trade.is_open.is_(True),
                        ]
                    ).all()
    
                    if not open_trades:
                        logger.info(
                            f"Set 'allow_trailing' to True for {metadata['pair']} to start trailing!!!"
                        )
                        trailing_buy["allow_trailing"] = True
                        dataframe.loc[:, "enter_tag"] = f"{enter_tag} (start trail price {close_price})"
            else:
                if trailing_buy["trailing_buy_order_started"] is True:
                    logger.info(
                        f"Continue trailing for {metadata['pair']}. Manually trigger buy signal!!"
                    )
                    dataframe.loc[:, "enter_long"] = 1
                    dataframe.loc[:, "enter_tag"] = trailing_buy["buy_tag"]
    
        return dataframe