# source: https://raw.githubusercontent.com/ma-pony/crypto_trade_service/b9792ef228b3b7ab0305fca88e1d859e501d093b/backend/app/strategies/production/mean_reversion.py
"""
均值回归策略

基于布林带的均值回归策略。
在价格偏离均值过远时反向交易。

核心逻辑：
- 布林带判断价格偏离程度
- RSI 确认超买超卖
- 成交量萎缩确认反转时机
"""

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__mean_reversion__20260131_075609(IStrategy):
    """均值回归策略

    在价格触及布林带边界且出现反转信号时入场。
    """

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

    # 时间周期
    timeframe = "15m"

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

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

    # 数据配置
    startup_candle_count = 50
    can_short = True

    # ========== 可优化参数 ==========
    # 布林带参数
    bb_period = IntParameter(low=15, high=30, default=20, space="buy")
    bb_std = DecimalParameter(low=1.5, high=3.0, default=2.0, space="buy")

    # 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")

    def populate_indicators(
        self,
        dataframe: "DataFrame",
        metadata: dict[str, Any],
    ) -> "DataFrame":
        """计算技术指标"""
        # 布林带
        bb_upper, bb_middle, bb_lower = ta.bbands(
            dataframe["close"],
            length=self.bb_period.value,
            std=self.bb_std.value,
        )
        dataframe["bb_upper"] = bb_upper
        dataframe["bb_middle"] = bb_middle
        dataframe["bb_lower"] = bb_lower
        dataframe["bb_width"] = (dataframe["bb_upper"] - dataframe["bb_lower"]) / dataframe["bb_middle"]

        # RSI
        dataframe["rsi"] = ta.rsi(dataframe["close"], length=self.rsi_period.value)

        # 价格位置
        dataframe["bb_percent"] = (dataframe["close"] - dataframe["bb_lower"]) / (
            dataframe["bb_upper"] - dataframe["bb_lower"]
        )

        # 反转 K 线形态
        dataframe["bullish_pin"] = (
            (dataframe["close"] > dataframe["open"]) &
            ((dataframe["open"] - dataframe["low"]) > 2 * (dataframe["close"] - dataframe["open"]))
        )
        dataframe["bearish_pin"] = (
            (dataframe["close"] < dataframe["open"]) &
            ((dataframe["high"] - dataframe["open"]) > 2 * (dataframe["open"] - dataframe["close"]))
        )

        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 超卖 + 反转形态
        long_conditions = (
            (dataframe["close"] <= dataframe["bb_lower"]) &
            (dataframe["rsi"] < self.rsi_oversold.value) &
            (dataframe["bullish_pin"] | (dataframe["close"] > dataframe["open"])) &
            (dataframe["volume"] > 0)
        )
        dataframe.loc[long_conditions, "enter_long"] = 1
        dataframe.loc[long_conditions, "enter_tag"] = "mean_rev_long"

        # 做空：价格触及上轨 + RSI 超买 + 反转形态
        short_conditions = (
            (dataframe["close"] >= dataframe["bb_upper"]) &
            (dataframe["rsi"] > self.rsi_overbought.value) &
            (dataframe["bearish_pin"] | (dataframe["close"] < dataframe["open"])) &
            (dataframe["volume"] > 0)
        )
        dataframe.loc[short_conditions, "enter_short"] = 1
        dataframe.loc[short_conditions, "enter_tag"] = "mean_rev_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["bb_middle"]
        dataframe.loc[exit_long_conditions, "exit_long"] = 1
        dataframe.loc[exit_long_conditions, "exit_tag"] = "mean_rev_exit"

        # 空头出场：价格回归中轨
        exit_short_conditions = dataframe["close"] <= dataframe["bb_middle"]
        dataframe.loc[exit_short_conditions, "exit_short"] = 1
        dataframe.loc[exit_short_conditions, "exit_tag"] = "mean_rev_exit"

        return dataframe