CTAAggressiveBreakout

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

import numpy as np
import pandas as pd
from datetime import datetime
from pandas import DataFrame
from typing import Optional

from freqtrade.strategy import (
    IStrategy,
    Trade,
    # Hyperopt
    BooleanParameter,
    CategoricalParameter,
    DecimalParameter,
    IntParameter,
    RealParameter,
    # helpers
    merge_informative_pair,
    stoploss_from_absolute,
)

import talib.abstract as ta
from technical import qtpylib


class CTAAggressiveBreakout(IStrategy):
    """
    CTA风格 - 激进通道突破（多头为主）
    - 5m 基础周期
    - 入场：Donchian 短通道向上突破 + 量能过滤（激进）
    - 退出：Chandelier/ATR 跟踪止损（不敏感，延长持仓） + 慢速 Donchian 退场
            另加极端拉升的获利了结（触发频率低）
    """

    INTERFACE_VERSION = 3
    can_short: bool = False

    # 关闭 ROI（实质上让退出依赖 custom_stoploss / exit 信号）
    minimal_roi = {"0": 10.0}

    # 初始基础止损（作为兜底；真正的退出逻辑走 custom_stoploss）
    stoploss = -0.12

    trailing_stop = False  # 使用自定义止损，不启用内置 TSL

    timeframe = "5m"
    process_only_new_candles = True

    use_exit_signal = True
    exit_profit_only = False
    ignore_roi_if_entry_signal = True

    # 启动需要的K线数量（需覆盖最长滚动窗口）
    startup_candle_count: int = 300

    # ========== Hyperoptable params ==========
    breakout_len = IntParameter(low=15, high=40, default=20, space="buy", optimize=True, load=True)
    exit_len = IntParameter(low=25, high=80, default=35, space="sell", optimize=True, load=True)

    atr_period = IntParameter(low=7, high=35, default=14, space="sell", optimize=True, load=True)
    atr_mult = RealParameter(low=2.5, high=5.0, default=3.5, space="sell", optimize=True, load=True)

    vol_sma_period = IntParameter(low=10, high=60, default=20, space="buy", optimize=True, load=True)
    vol_mult = RealParameter(low=0.8, high=1.5, default=1.0, space="buy", optimize=True, load=True)

    rsi_exit = IntParameter(low=80, high=95, default=90, space="sell", optimize=True, load=True)
    bb_window = IntParameter(low=10, high=40, default=20, space="sell", optimize=True, load=True)

    order_types = {
        "entry": "limit",
        "exit": "limit",
        "stoploss": "market",
        "stoploss_on_exchange": False,
    }
    order_time_in_force = {"entry": "GTC", "exit": "GTC"}

    plot_config = {
        "main_plot": {
            "donchian_high": {"color": "orange"},
            "donchian_low": {"color": "orange"},
            "donchian_exit": {"color": "yellow"},
            "chandelier_stop": {"color": "white"},
        },
        "subplots": {
            "ATR": {"atr": {"color": "blue"}},
            "RSI": {"rsi": {"color": "red"}},
        },
    }

    def informative_pairs(self):
        """
        如果你后续想做多周期（如1h过滤），可在此返回 [(pair, '1h'), ...]
        本策略基础版仅用 5m，避免依赖 DataProvider 的额外缓存。
        """
        return []

    # -------------- 指标计算 --------------
    def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        n = int(self.breakout_len.value)
        n_exit = int(self.exit_len.value)

        # Donchian 通道
        dataframe["donchian_high"] = dataframe["high"].rolling(n).max().shift(1)
        dataframe["donchian_low"] = dataframe["low"].rolling(n).min().shift(1)
        # 退出通道（更慢）
        dataframe["donchian_exit"] = dataframe["low"].rolling(n_exit).min().shift(1)

        # ATR（用于 Chandelier 止损）
        dataframe["atr"] = ta.ATR(dataframe, timeperiod=int(self.atr_period.value))

        # RSI / Bollinger（用于爆顶退出）
        dataframe["rsi"] = ta.RSI(dataframe, timeperiod=14)
        bb = qtpylib.bollinger_bands(
            qtpylib.typical_price(dataframe), window=int(self.bb_window.value), stds=2
        )
        dataframe["bb_upper"] = bb["upper"]
        dataframe["bb_lower"] = bb["lower"]

        # 量能过滤
        dataframe["vol_sma"] = dataframe["volume"].rolling(int(self.vol_sma_period.value)).mean()

        # 画图辅助：计算当前 Chandelier 止损（基于最近 n_exit 或 n 的最高价，你也可改为更长）
        # 仅用于可视化，不参与信号（真正止损在 custom_stoploss）
        basis_h = dataframe["high"].rolling(max(n, n_exit)).max()
        dataframe["chandelier_stop"] = basis_h - self.atr_mult.value * dataframe["atr"]

        return dataframe

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

        # 新突破（close 刚刚突破上轨）
        new_breakout = (
            (dataframe["close"] > dataframe["donchian_high"]) &
            (dataframe["close"].shift(1) <= dataframe["donchian_high"].shift(1))
        )

        # 量能过滤（更激进可把 self.vol_mult 默认调到 0.9 或 0.8）
        vol_ok = dataframe["volume"] > (dataframe["vol_sma"] * float(self.vol_mult.value))

        dataframe.loc[(new_breakout & vol_ok), "enter_long"] = 1
        dataframe.loc[(new_breakout & vol_ok), "enter_tag"] = "DonchBreakout(aggr)"

        # 不做空
        if "enter_short" in dataframe.columns:
            dataframe["enter_short"] = 0

        return dataframe

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

        # 1) 慢速 Donchian 退场：趋势破位才出，避免过敏
        slow_exit = dataframe["close"] < dataframe["donchian_exit"]

        # 2) 爆顶获利了结：仅在极端拉升时触发（锁定大幅盈利，非频繁）
        blowoff_exit = (dataframe["rsi"] > int(self.rsi_exit.value)) & (dataframe["close"] > dataframe["bb_upper"])

        dataframe.loc[(slow_exit | blowoff_exit), "exit_long"] = 1
        dataframe.loc[slow_exit, "exit_tag"] = "DonchExit(slow)"
        dataframe.loc[blowoff_exit, "exit_tag"] = "BlowoffTP"

        if "exit_short" in dataframe.columns:
            dataframe["exit_short"] = 0

        return dataframe

    # -------------- 自定义止损（核心：Chandelier/ATR） --------------
    def custom_stoploss(
        self,
        pair: str,
        trade: Trade,
        current_time: datetime,
        current_rate: float,
        current_profit: float,
        **kwargs,
    ) -> float:
        """
        以 5m 的 ATR 与近 N 根最高价计算 Chandelier 止损，N 取 exit_len 或更大。
        通过 stoploss_from_absolute 把“绝对价位”转换为 freqtrade 需要的比例返回。
        """
        # 若 DataProvider 不可用，使用基础 stoploss 兜底
        if not hasattr(self, "dp") or self.dp is None:
            return self.stoploss

        df = self.dp.get_pair_dataframe(pair=pair, timeframe=self.timeframe)
        if df is None or df.empty:
            return self.stoploss

        # 选取当前时刻可用的最新数据
        n_exit = int(self.exit_len.value)
        atr_period = int(self.atr_period.value)

        atr = ta.ATR(df, timeperiod=atr_period)
        hh = df["high"].rolling(max(n_exit, atr_period)).max()

        # 最新值
        atr_v = float(atr.iloc[-1]) if atr.notnull().any() else None
        hh_v = float(hh.iloc[-1]) if hh.notnull().any() else None

        if atr_v is None or hh_v is None or atr_v == 0.0:
            return self.stoploss

        chandelier = hh_v - float(self.atr_mult.value) * atr_v

        # 将绝对价位转换为相对开仓价的止损比例（负数）
        desired_sl = stoploss_from_absolute(trade.open_rate, chandelier)

        # 为安全起见，若计算异常导致止损过低/过高，这里回退到基础 stoploss
        if desired_sl is None or desired_sl >= 0:
            return self.stoploss

        return desired_sl