# source: https://raw.githubusercontent.com/ma-pony/crypto_trade_service/a92aa6dd67142e2882b343b4553ea535854bee4b/backend/app/strategies/production/breakout.py
"""
突破策略

基于支撑阻力位和成交量的突破交易策略。
在价格突破关键位置时顺势入场。

核心逻辑：
- Donchian 通道识别支撑阻力
- 成交量确认突破有效性
- ATR 过滤假突破
"""

from __future__ import annotations

from typing import TYPE_CHECKING, Any

from freqtrade.strategy import IStrategy, DecimalParameter, IntParameter
from app.strategies.production import indicators as ta

if TYPE_CHECKING:
    from pandas import DataFrame


class Github_ma_pony_crypto_trade_service__breakout__20260201_050828(IStrategy):
    """突破策略

    在价格突破 Donchian 通道且成交量放大时入场。
    """

    # 策略元数据
    strategy_id = "breakout_v1"
    strategy_version = "1.0.0"

    # 时间周期
    timeframe = "1h"

    # 风控配置
    minimal_roi = {
        "0": 0.20,
        "120": 0.10,
        "240": 0.05,
    }
    stoploss = -0.05

    # 追踪止损
    trailing_stop = True
    trailing_stop_positive = 0.02
    trailing_stop_positive_offset = 0.05
    trailing_only_offset_is_reached = True

    # 数据配置
    startup_candle_count = 50
    can_short = True

    # ========== 可优化参数 ==========
    # Donchian 通道参数
    donchian_period = IntParameter(low=15, high=30, default=20, space="buy")

    # ATR 参数
    atr_period = IntParameter(low=10, high=20, default=14, space="buy")
    atr_threshold = DecimalParameter(low=0.5, high=2.0, default=1.0, space="buy")

    # 成交量参数
    volume_ma_period = IntParameter(low=10, high=30, default=20, space="buy")
    volume_threshold = DecimalParameter(low=1.2, high=2.5, default=1.5, space="buy")

    def populate_indicators(
        self,
        dataframe: "DataFrame",
        metadata: dict[str, Any],
    ) -> "DataFrame":
        """计算技术指标"""
        # Donchian 通道
        dc_upper, dc_mid, dc_lower = ta.donchian(
            dataframe["high"],
            dataframe["low"],
            length=self.donchian_period.value,
        )
        dataframe["dc_upper"] = dc_upper
        dataframe["dc_lower"] = dc_lower
        dataframe["dc_mid"] = dc_mid

        # ATR
        dataframe["atr"] = ta.atr(
            dataframe["high"],
            dataframe["low"],
            dataframe["close"],
            length=self.atr_period.value,
        )
        dataframe["atr_pct"] = dataframe["atr"] / dataframe["close"] * 100

        # 成交量
        dataframe["volume_ma"] = ta.sma(dataframe["volume"], length=self.volume_ma_period.value)
        dataframe["volume_ratio"] = dataframe["volume"] / dataframe["volume_ma"]

        # 突破信号
        dataframe["break_up"] = (
            (dataframe["close"] > dataframe["dc_upper"].shift(1)) &
            (dataframe["close"].shift(1) <= dataframe["dc_upper"].shift(2))
        )
        dataframe["break_down"] = (
            (dataframe["close"] < dataframe["dc_lower"].shift(1)) &
            (dataframe["close"].shift(1) >= dataframe["dc_lower"].shift(2))
        )

        return dataframe

    def populate_entry_trend(
        self,
        dataframe: "DataFrame",
        metadata: dict[str, Any],
    ) -> "DataFrame":
        """生成入场信号"""
        dataframe.loc[:, "enter_long"] = 0
        dataframe.loc[:, "enter_short"] = 0
        dataframe.loc[:, "enter_tag"] = ""

        # 做多：向上突破 + 放量 + ATR 确认波动
        long_conditions = (
            dataframe["break_up"] &
            (dataframe["volume_ratio"] > self.volume_threshold.value) &
            (dataframe["atr_pct"] > self.atr_threshold.value)
        )
        dataframe.loc[long_conditions, "enter_long"] = 1
        dataframe.loc[long_conditions, "enter_tag"] = "breakout_long"

        # 做空：向下突破 + 放量 + ATR 确认波动
        short_conditions = (
            dataframe["break_down"] &
            (dataframe["volume_ratio"] > self.volume_threshold.value) &
            (dataframe["atr_pct"] > self.atr_threshold.value)
        )
        dataframe.loc[short_conditions, "enter_short"] = 1
        dataframe.loc[short_conditions, "enter_tag"] = "breakout_short"

        return dataframe

    def populate_exit_trend(
        self,
        dataframe: "DataFrame",
        metadata: dict[str, Any],
    ) -> "DataFrame":
        """生成出场信号"""
        dataframe.loc[:, "exit_long"] = 0
        dataframe.loc[:, "exit_short"] = 0
        dataframe.loc[:, "exit_tag"] = ""

        # 多头出场：价格跌破中轨
        exit_long_conditions = dataframe["close"] < dataframe["dc_mid"]
        dataframe.loc[exit_long_conditions, "exit_long"] = 1
        dataframe.loc[exit_long_conditions, "exit_tag"] = "breakout_exit"

        # 空头出场：价格突破中轨
        exit_short_conditions = dataframe["close"] > dataframe["dc_mid"]
        dataframe.loc[exit_short_conditions, "exit_short"] = 1
        dataframe.loc[exit_short_conditions, "exit_tag"] = "breakout_exit"

        return dataframe