ChronoDip

# 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 ChronoDip(IStrategy):
    """
    ChronoDipV3 - SHORT-ONLY SIMPLE FLIP VERSION
    
    WARNING: This version does NOT reverse the logic.
    It simply takes every original LONG entry signal and opens a SHORT instead.
    This is likely to be an anti-trend strategy and may perform poorly in non-bear markets.
    """

    INTERFACE_VERSION = 3

    # =========================
    # 基础设置：只允许做空
    # =========================
    timeframe = "4h"
    can_long: bool = False   # 禁用做多
    can_short: bool = True   # 启用做空

    minimal_roi = {
        "0": 100.0
    }

    stoploss = -0.35
    use_custom_stoploss = True

    trailing_stop = False
    use_exit_signal = True
    process_only_new_candles = True

    startup_candle_count: int = 250

    # =========================
    # 参数（保持不变）
    # =========================
    ma_len = IntParameter(8, 40, default=14, space="buy", optimize=True)
    pianli = IntParameter(3, 24, default=6, space="buy", optimize=True)
    entry_mode = CategoricalParameter(
        ["below_ma", "reclaim_ma", "both"],
        default="both",
        space="buy",
        optimize=True,
    )
    use_ema_filter = BooleanParameter(default=False, space="buy", optimize=False)
    use_slope_filter = BooleanParameter(default=False, space="buy", optimize=True)
    tupo_len = IntParameter(10, 50, default=20, space="sell", optimize=True)
    pingcang_time = IntParameter(6, 48, default=12, space="sell", optimize=True)
    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)
    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-simple-flip-short"

    # =========================
    # 指标（完全不变）
    # =========================
    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)

        dataframe["ma"] = ta.LINEARREG(dataframe["close"], timeperiod=ma_period)
        dataframe["ma_slope"] = dataframe["ma"] - dataframe["ma"].shift(1)
        dataframe["ema200"] = ta.EMA(dataframe["close"], timeperiod=self.trend_ema_len)
        dataframe["atr"] = ta.ATR(dataframe, timeperiod=atr_period)
        dataframe["atr_pct"] = dataframe["atr"] / dataframe["close"]
        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()
        )

        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

    # =========================
    # 入场：原逻辑生成 buy 信号，现在改为开 short
    # =========================
    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
        )

        below_ma_signal = (
            run_ready
            &
            (dataframe["close"] < dataframe["ma"])
        )

        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_short_flip"
        elif mode == "reclaim_ma":
            entry_signal = reclaim_ma_signal
            entry_tag = "reclaim_ma_short_flip"
        else:
            entry_signal = below_ma_signal | reclaim_ma_signal
            entry_tag = "both_short_flip"

        # 过滤器保持原样（注意：这可能导致在多头趋势中做空！）
        if self.use_ema_filter.value:
            entry_signal = entry_signal & (dataframe["close"] > dataframe["ema200"])

        if self.use_slope_filter.value:
            entry_signal = entry_signal & (dataframe["ma_slope"] > 0)

        exit_same_candle = (
            dataframe["close"] >= dataframe["hh"].shift(1)
        )

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

        # 关键改动：原 enter_long → enter_short
        dataframe.loc[entry_signal, ["enter_short", "enter_tag"]] = (1, entry_tag)
        dataframe.loc[entry_signal, "debug_entry_raw"] = 1

        return dataframe

    # =========================
    # 出场：原 exit_long → exit_short（但逻辑仍是“突破前高”）
    # =========================
    def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        breakout_exit = (
            (dataframe["close"] >= dataframe["hh"].shift(1))
            &
            (dataframe["volume"] > 0)
        )

        # 关键改动：原 exit_long → exit_short
        dataframe.loc[
            breakout_exit,
            ["exit_short", "exit_tag"]
        ] = (1, "breakout_prev_high_short_flip")

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

        # 空单止损价 = 入场价 + ATR * mult
        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=True,
                leverage=getattr(trade, "leverage", 1.0),
            )

        return None