ChronoDipV3

# pragma pylint: disable=missing-docstring, invalid-name, pointless-string-statement
# flake8: noqa
# isort: skip_file

from datetime import datetime
from typing import Optional, Union

import pandas as pd
from pandas import DataFrame

import talib.abstract as ta

from freqtrade.strategy import (
    IStrategy,
    BooleanParameter,
    CategoricalParameter,
    DecimalParameter,
    IntParameter,
    timeframe_to_minutes,
    timeframe_to_prev_date,
    stoploss_from_absolute,
)


class ChronoDipV3(IStrategy):
    """
    ChronoDipV3

    这是比上一版更宽松的 4h 版本，主要解决“没有任何交易记录”的问题。

    入场逻辑默认 entry_mode = "both"：
    1. below_ma:
       连续强势后，当前 close 跌破 LSMA，类似你原始策略。
    2. reclaim_ma:
       连续强势后，当前 low 跌破 LSMA，但 close 收回 LSMA。

    默认不启用 EMA200 和 LSMA 斜率过滤，避免信号被过滤光。
    等确认有交易后，再逐步打开过滤条件。
    """

    INTERFACE_VERSION = 3

    # =========================
    # 基础设置
    # =========================
    timeframe = "1h"
    can_short: bool = True

    # 禁用默认 ROI，主要交给突破出场、时间出场、ATR 止损
    minimal_roi = {
        "0": 100.0
    }

    # 兜底止损。ATR 止损会在 custom_stoploss 中接管。
    stoploss = -0.35

    # 必须开启，否则 custom_stoploss 不会执行
    use_custom_stoploss = True

    trailing_stop = False
    use_exit_signal = True
    process_only_new_candles = True

    # EMA200 可选，所以 startup 需要稍微大一点
    startup_candle_count: int = 250

    # =========================
    # 参数
    # =========================

    # LSMA 周期。默认 14，比上一版 50 宽松很多。
    ma_len = IntParameter(8, 40, default=14, space="buy", optimize=True)

    # 连续站上均线的根数。默认 6，接近你原始版本。
    # 4h 下 6 根约等于 1 天。
    pianli = IntParameter(3, 24, default=6, space="buy", optimize=True)

    # 入场模式：
    # below_ma = 原始逻辑，跌破均线就买，信号最多；
    # reclaim_ma = 回踩收回均线再买，信号更少；
    # both = 两种都允许，默认，避免 0 交易。
    entry_mode = CategoricalParameter(
        ["below_ma", "reclaim_ma", "both"],
        default="both",
        space="buy",
        optimize=True,
    )

    # 是否使用 EMA200 大趋势过滤。默认 False，避免 0 交易。
    use_ema_filter = BooleanParameter(default=False, space="buy", optimize=True)

    # 是否使用 LSMA 斜率过滤。默认 False，避免 0 交易。
    use_slope_filter = BooleanParameter(default=False, space="buy", optimize=True)

    # 突破前高平仓窗口
    tupo_len = IntParameter(10, 50, default=20, space="sell", optimize=True)

    # 时间出场。默认 12 根 4h，约 2 天。
    pingcang_time = IntParameter(6, 48, default=12, space="sell", optimize=True)

    # ATR 动态止损
    atr_len = IntParameter(10, 40, default=20, space="sell", optimize=True)
    atr_mult = DecimalParameter(1.5, 4.0, default=2.5, decimals=2, space="sell", optimize=True)

    # EMA 大趋势过滤周期
    trend_ema_len: int = 200

    # =========================
    # 画图配置
    # =========================
    plot_config = {
        "main_plot": {
            "ma": {},
            "ema200": {},
            "hh": {},
        },
        "subplots": {
            "ATR": {
                "atr": {},
                "atr_pct": {},
            },
            "Debug": {
                "consec_above": {},
                "debug_run_ready": {},
                "debug_below_ma": {},
                "debug_reclaim_ma": {},
                "debug_entry_raw": {},
            },
        },
    }

    def version(self) -> str:
        return "3.0.0"

    # =========================
    # 指标
    # =========================
    def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        ma_period = int(self.ma_len.value)
        atr_period = int(self.atr_len.value)
        breakout_period = int(self.tupo_len.value)
        run_period = int(self.pianli.value)

        # LSMA：ta-lib 的 LINEARREG 可近似作为最小二乘移动平均
        dataframe["ma"] = ta.LINEARREG(dataframe["close"], timeperiod=ma_period)

        # 均线斜率
        dataframe["ma_slope"] = dataframe["ma"] - dataframe["ma"].shift(1)

        # EMA200，可选过滤
        dataframe["ema200"] = ta.EMA(dataframe["close"], timeperiod=self.trend_ema_len)

        # ATR
        dataframe["atr"] = ta.ATR(dataframe, timeperiod=atr_period)
        dataframe["atr_pct"] = dataframe["atr"] / dataframe["close"]

        # 前高线。出场时使用 shift(1)，避免当前 K 线参与自己的突破判断。
        dataframe["hh"] = (
            dataframe["high"]
            .rolling(window=breakout_period, min_periods=breakout_period)
            .max()
        )

        # 连续站上均线
        dataframe["is_above_ma"] = (dataframe["close"] > dataframe["ma"]).astype(int)
        dataframe["consec_above"] = (
            dataframe["is_above_ma"]
            .rolling(window=run_period, min_periods=run_period)
            .sum()
        )

        # 调试列：方便你后面 plot 或导出分析
        dataframe["debug_run_ready"] = (
            dataframe["consec_above"].shift(1) >= run_period
        ).astype(int)

        dataframe["debug_below_ma"] = (
            dataframe["close"] < dataframe["ma"]
        ).astype(int)

        dataframe["debug_reclaim_ma"] = (
            (dataframe["low"] < dataframe["ma"])
            &
            (dataframe["close"] > dataframe["ma"])
        ).astype(int)

        dataframe["debug_entry_raw"] = 0

        return dataframe

    # =========================
    # 入场
    # =========================
    def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        run_period = int(self.pianli.value)

        run_ready = (
            dataframe["consec_above"].shift(1) >= run_period
        )

        # 模式 1：原始逻辑，连续强势后，当前 close 跌破 LSMA。
        below_ma_signal = (
            run_ready
            &
            (dataframe["close"] < dataframe["ma"])
        )

        # 模式 2：回踩 LSMA 后收回。
        reclaim_ma_signal = (
            run_ready
            &
            (dataframe["low"] < dataframe["ma"])
            &
            (dataframe["close"] > dataframe["ma"])
        )

        mode = self.entry_mode.value

        if mode == "below_ma":
            entry_signal = below_ma_signal
            entry_tag = "below_ma"
        elif mode == "reclaim_ma":
            entry_signal = reclaim_ma_signal
            entry_tag = "reclaim_ma"
        else:
            entry_signal = below_ma_signal | reclaim_ma_signal
            entry_tag = "both_below_or_reclaim"

        # 可选 EMA200 过滤。默认关闭。
        if self.use_ema_filter.value:
            entry_signal = entry_signal & (dataframe["close"] > dataframe["ema200"])

        # 可选 LSMA 斜率过滤。默认关闭。
        if self.use_slope_filter.value:
            entry_signal = entry_signal & (dataframe["ma_slope"] > 0)

        # 避免同一根 K 线既入场又出场。
        # Freqtrade 文档说明，如果同一根 K 线同时有 entry 和 exit 信号，会被视为信号冲突，不会下单。
        exit_same_candle = (
            dataframe["close"] >= dataframe["hh"].shift(1)
        )

        entry_signal = (
            entry_signal
            &
            (~exit_same_candle)
            &
            (dataframe["volume"] > 0)
        )

        dataframe.loc[entry_signal, ["enter_short", "enter_tag"]] = (1, entry_tag)
        dataframe.loc[entry_signal, "debug_entry_raw"] = 1

        return dataframe

    # =========================
    # 出场：突破前高
    # =========================
    def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        breakout_exit = (
            (dataframe["close"] >= dataframe["hh"].shift(1))
            &
            (dataframe["volume"] > 0)
        )

        dataframe.loc[
            breakout_exit,
            ["exit_short", "exit_tag"]
        ] = (1, "breakout_prev_high")

        return dataframe

    # =========================
    # 自定义出场：时间淘汰
    # =========================
    def custom_exit(
        self,
        pair: str,
        trade,
        current_time: datetime,
        current_rate: float,
        current_profit: float,
        **kwargs
    ) -> Optional[Union[str, bool]]:
        """
        时间出场。

        这版比上一版更宽松：
        到时间后，不再要求必须跌破均线才出；
        只要达到持仓时间，就出场。
        这样可以先保证有完整交易闭环。
        后面如果交易太多、质量差，再改成“只平失败单”。
        """

        tf_minutes = timeframe_to_minutes(self.timeframe)
        minutes_passed = (current_time - trade.open_date_utc).total_seconds() / 60
        candles_passed = int(minutes_passed / tf_minutes)

        if candles_passed >= int(self.pingcang_time.value):
            return "time_limit_exit"

        return None

    # =========================
    # 自定义止损：ATR
    # =========================
    def custom_stoploss(
        self,
        pair: str,
        trade,
        current_time: datetime,
        current_rate: float,
        current_profit: float,
        after_fill: bool = False,
        **kwargs
    ) -> Optional[float]:
        """
        ATR 动态止损。

        多单止损价：
            入场价 - 入场附近 ATR * atr_mult
        """

        dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)

        if dataframe is None or dataframe.empty:
            return None

        trade_date = timeframe_to_prev_date(self.timeframe, trade.open_date_utc)
        candle_df = dataframe.loc[dataframe["date"] <= trade_date]

        if candle_df.empty:
            return None

        entry_candle = candle_df.iloc[-1].squeeze()
        entry_atr = entry_candle.get("atr")

        if pd.isna(entry_atr) or entry_atr <= 0:
            return None

        sl_price = trade.open_rate - (float(entry_atr) * float(self.atr_mult.value))

        # 多单止损价必须低于当前价
        if sl_price < current_rate:
            return stoploss_from_absolute(
                sl_price,
                current_rate=current_rate,
                is_short=getattr(trade, "is_short", False),
                leverage=getattr(trade, "leverage", 1.0),
            )

        return None