ChronoDrop

# 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 ChronoDrop(IStrategy):
    """
    ChronoDrop (做空死猫跳版)
    
    核心逻辑：
    寻找处于长期阴跌（连续多根K线在均线下方）的山寨币，
    当它突然反弹突破均线（诱多）时，果断开空。
    目标是跌破前期低点平仓赚取暴利，或者时间耗尽平仓。
    """

    INTERFACE_VERSION = 3

    # =========================
    # 基础设置
    # =========================
    timeframe = "4h"
    
    # 【核心改动】关闭做多，开启做空
    can_long: bool = False
    can_short: bool = True

    minimal_roi = {
        "0": 100.0  # 禁用默认ROI
    }

    # 兜底止损
    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)

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

    # 时间出场
    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)

    def version(self) -> str:
        return "1.0.0_ShortOnly"

    # =========================
    # 指标
    # =========================
    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
        dataframe["ma"] = ta.LINEARREG(dataframe["close"], timeperiod=ma_period)

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

        # 【核心改动】前低线 (做空看跌破前低)
        dataframe["ll"] = (
            dataframe["low"]
            .rolling(window=breakout_period, min_periods=breakout_period)
            .min()
        )

        # 【核心改动】统计连续在均线【下方】的根数
        dataframe["is_below_ma"] = (dataframe["close"] < dataframe["ma"]).astype(int)
        dataframe["consec_below"] = (
            dataframe["is_below_ma"]
            .rolling(window=run_period, min_periods=run_period)
            .sum()
        )

        return dataframe

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

        # 准备条件：之前连续 N 根 K 线都在均线下方 (处于弱势阴跌)
        run_ready = (
            dataframe["consec_below"].shift(1) >= run_period
        )

        # 触发条件：当前 K 线突然抽风，收盘价站上均线 (死猫跳反弹)
        bounce_signal = (
            run_ready
            &
            (dataframe["close"] > dataframe["ma"])
        )

        # 避免同一根 K 线既入场又出场
        exit_same_candle = (
            dataframe["close"] <= dataframe["ll"].shift(1)
        )

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

        # 【核心改动】发送 enter_short 信号
        dataframe.loc[entry_signal, ["enter_short", "enter_tag"]] = (1, "bounce_short")

        return dataframe

    # =========================
    # 出场 (平空)
    # =========================
    def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        # 【核心改动】当价格向下跌破前低时，获利平空
        breakdown_exit = (
            (dataframe["close"] <= dataframe["ll"].shift(1))
            &
            (dataframe["volume"] > 0)
        )

        dataframe.loc[
            breakdown_exit,
            ["exit_short", "exit_tag"]
        ] = (1, "breakdown_prev_low")

        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 * 倍数)
        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", True), # 明确告诉系统这是空单
                leverage=getattr(trade, "leverage", 1.0),
            )

        return None