# source: https://raw.githubusercontent.com/ma-pony/crypto_trade_service/b9792ef228b3b7ab0305fca88e1d859e501d093b/backend/app/strategies/production/momentum.py
"""
动量策略

基于 RSI 和 MACD 的动量交易策略。
捕捉价格动量变化，在动量加速时入场。

核心逻辑：
- RSI 判断超买超卖区域
- MACD 确认动量方向和强度
- 成交量确认动量有效性
"""

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__momentum__20260131_075609(IStrategy):
    """动量策略

    使用 RSI + MACD 双重确认的动量交易系统。
    在动量启动时入场，动量衰竭时出场。
    """

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

    # 时间周期
    timeframe = "1h"

    # 风控配置
    minimal_roi = {
        "0": 0.12,
        "60": 0.08,
        "120": 0.04,
        "240": 0.02,
    }
    stoploss = -0.06

    # 追踪止损
    trailing_stop = True
    trailing_stop_positive = 0.015
    trailing_stop_positive_offset = 0.03
    trailing_only_offset_is_reached = True

    # 数据配置
    startup_candle_count = 50
    can_short = True

    # ========== 可优化参数 ==========
    # RSI 参数
    rsi_period = IntParameter(low=10, high=21, default=14, space="buy")
    rsi_oversold = DecimalParameter(low=20.0, high=35.0, default=30.0, space="buy")
    rsi_overbought = DecimalParameter(low=65.0, high=80.0, default=70.0, space="buy")

    # MACD 参数
    macd_fast = IntParameter(low=8, high=15, default=12, space="buy")
    macd_slow = IntParameter(low=20, high=30, default=26, space="buy")
    macd_signal = IntParameter(low=7, high=12, default=9, space="buy")

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

    def populate_indicators(
        self,
        dataframe: "DataFrame",
        metadata: dict[str, Any],
    ) -> "DataFrame":
        """计算技术指标"""
        # RSI
        dataframe["rsi"] = ta.rsi(dataframe["close"], length=self.rsi_period.value)

        # MACD
        macd_line, signal_line, histogram = ta.macd(
            dataframe["close"],
            fast=self.macd_fast.value,
            slow=self.macd_slow.value,
            signal=self.macd_signal.value,
        )
        dataframe["macd"] = macd_line
        dataframe["macd_signal"] = signal_line
        dataframe["macd_hist"] = histogram

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

        # MACD 交叉信号
        dataframe["macd_cross_up"] = (
            (dataframe["macd"] > dataframe["macd_signal"]) &
            (dataframe["macd"].shift(1) <= dataframe["macd_signal"].shift(1))
        )
        dataframe["macd_cross_down"] = (
            (dataframe["macd"] < dataframe["macd_signal"]) &
            (dataframe["macd"].shift(1) >= dataframe["macd_signal"].shift(1))
        )

        # RSI 动量方向
        dataframe["rsi_rising"] = dataframe["rsi"] > dataframe["rsi"].shift(1)
        dataframe["rsi_falling"] = dataframe["rsi"] < dataframe["rsi"].shift(1)

        # MACD 柱状图动量
        dataframe["macd_hist_rising"] = dataframe["macd_hist"] > dataframe["macd_hist"].shift(1)
        dataframe["macd_hist_falling"] = dataframe["macd_hist"] < dataframe["macd_hist"].shift(1)

        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"] = ""

        # 做多：RSI 从超卖区回升 + MACD 金叉 + 放量
        long_conditions = (
            (dataframe["rsi"] > self.rsi_oversold.value) &
            (dataframe["rsi"].shift(1) <= self.rsi_oversold.value) &
            (dataframe["macd"] > dataframe["macd_signal"]) &
            dataframe["macd_hist_rising"] &
            (dataframe["volume_ratio"] > self.volume_threshold.value)
        )
        dataframe.loc[long_conditions, "enter_long"] = 1
        dataframe.loc[long_conditions, "enter_tag"] = "momentum_long"

        # 做空：RSI 从超买区回落 + MACD 死叉 + 放量
        short_conditions = (
            (dataframe["rsi"] < self.rsi_overbought.value) &
            (dataframe["rsi"].shift(1) >= self.rsi_overbought.value) &
            (dataframe["macd"] < dataframe["macd_signal"]) &
            dataframe["macd_hist_falling"] &
            (dataframe["volume_ratio"] > self.volume_threshold.value)
        )
        dataframe.loc[short_conditions, "enter_short"] = 1
        dataframe.loc[short_conditions, "enter_tag"] = "momentum_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"] = ""

        # 多头出场：RSI 超买 或 MACD 死叉
        exit_long_conditions = (
            (dataframe["rsi"] > self.rsi_overbought.value) |
            dataframe["macd_cross_down"]
        )
        dataframe.loc[exit_long_conditions, "exit_long"] = 1
        dataframe.loc[exit_long_conditions, "exit_tag"] = "momentum_exit"

        # 空头出场：RSI 超卖 或 MACD 金叉
        exit_short_conditions = (
            (dataframe["rsi"] < self.rsi_oversold.value) |
            dataframe["macd_cross_up"]
        )
        dataframe.loc[exit_short_conditions, "exit_short"] = 1
        dataframe.loc[exit_short_conditions, "exit_tag"] = "momentum_exit"

        return dataframe