FreqtradeFuturesStrategy

from __future__ import annotations

from datetime import datetime
from typing import Any

import pandas as pd

try:
    from freqtrade.persistence import Trade  # pyright: ignore[reportMissingImports]
    from freqtrade.strategy import (  # pyright: ignore[reportMissingImports]
        DecimalParameter,
        IntParameter,
        IStrategy,
        stoploss_from_absolute,
    )
except ModuleNotFoundError:
    # Fallback for IDE/lint environments without freqtrade installed.
    class _Param:  # type: ignore[override]
        def __init__(self, *args: Any, **kwargs: Any) -> None:
            self.value = kwargs.get("default", args[2] if len(args) > 2 else 0)
        def __repr__(self) -> str:
            return str(self.value)

    IntParameter = _Param       # type: ignore[misc, assignment]
    DecimalParameter = _Param   # type: ignore[misc, assignment]

    class IStrategy:  # type: ignore[override]
        pass

    class Trade:  # type: ignore[override]
        open_rate: float = 0.0
        is_short: bool = False


class FreqtradeFuturesStrategy(IStrategy):
    """
    MA + RSI + ADX + MACD + ATR futures strategy (long/short).

    Hyperopt spaces:
      buy  : ma_short, ma_long, rsi_overbought, rsi_oversold, adx_threshold
      sell : atr_sl_mult, rr_ratio

    Anti-overfitting protocol:
      - Train  : --timerange 20240101-20250601  (17 months)
      - Validate: --timerange 20250601-20260401  (10 months, out-of-sample)
      - Loss   : SharpeHyperOptLossDaily
      - Min trades: 80
    """

    INTERFACE_VERSION = 3

    timeframe = "1h"
    can_short = True
    process_only_new_candles = True
    # startup_candle_count 须覆盖最大 MA 周期（ma_long 上限 = 60）
    startup_candle_count = 100

    minimal_roi = {"0": 10}
    stoploss = -0.99
    use_custom_stoploss = True   # T1后启用ATR跟踪止损
    use_custom_exit = True

    # 加仓 & 部分平仓开关
    position_adjustment_enable = True

    # 加仓参数（与 bot.py 对齐）
    SCALE_TRIGGER_MULT  = 0.5    # 价格顺方向移动 N×ATR 后触发加仓
    SCALE_INIT_RATIO    = 0.60   # 首次建仓比例（60%）
    SCALE_ADD_RATIO     = 0.40   # 加仓补足比例（40%）
    PARTIAL_CLOSE_RATIO = 0.50   # 1R 处部分平仓比例（50%）
    PARTIAL_TARGET_MULT = 1.0    # 第一目标：1×止损距离

    # 日线趋势过滤
    DAILY_MA_PERIOD = 20         # 日线 MA20：close > MA20 → 牛市，< MA20 → 熊市

    # ----------------------------------------------------------------
    # 固定参数（不参与 hyperopt）
    # ----------------------------------------------------------------
    RSI_PERIOD   = 14
    ADX_PERIOD   = 14
    ATR_PERIOD   = 14
    MACD_FAST    = 12
    MACD_SLOW    = 26
    MACD_SIGNAL  = 9
    MAX_LEVERAGE     = 10.0
    TARGET_LEVERAGE  = 10.0

    # ----------------------------------------------------------------
    # Hyperopt 参数空间
    # ----------------------------------------------------------------

    # 均线周期 —— buy 空间
    ma_short = IntParameter(5,  20, default=12, space="buy", optimize=True)
    ma_long  = IntParameter(21, 60, default=51, space="buy", optimize=True)

    # RSI 阈值 —— buy 空间
    rsi_overbought = IntParameter(60, 85, default=66, space="buy", optimize=True)
    rsi_oversold   = IntParameter(15, 45, default=40, space="buy", optimize=True)

    # ADX 趋势过滤阈值 —— buy 空间（多空独立阈值）
    adx_threshold       = IntParameter(15, 45, default=27, space="buy", optimize=True)
    adx_threshold_short = IntParameter(20, 50, default=33, space="buy", optimize=True)

    # ATR 止损倍数 & 跟踪止损倍数 —— sell 空间
    # 关键权衡：止损太宽 → 单次亏损大；止损太窄 → 被震出频繁
    # 回测数据驱动：3.5×ATR @ 10x 杠杆 ≈ 账户 -32%/笔，盈亏比被稀释到 1:1
    atr_sl_mult    = DecimalParameter(1.0, 3.5, default=2.5, decimals=1, space="sell", optimize=True)
    atr_trail_mult = DecimalParameter(1.0, 3.0, default=2.0, decimals=1, space="sell", optimize=True)

    # 波动率过滤：ATR/Close 超过此阈值不开仓（极端行情易扫损）
    MAX_ATR_PCT = 0.035

    # 最大持仓时长（小时），超过强平（防长期横盘消耗）
    MAX_HOLD_HOURS = 72

    # ----------------------------------------------------------------
    # 工具函数
    # ----------------------------------------------------------------

    def leverage(
        self,
        pair: str,
        current_time: datetime,
        current_rate: float,
        proposed_leverage: float,
        max_leverage: float,
        entry_tag: str | None,
        side: str,
        **kwargs: Any,
    ) -> float:
        return min(self.TARGET_LEVERAGE, self.MAX_LEVERAGE, max_leverage)

    @staticmethod
    def _true_range(dataframe: pd.DataFrame) -> pd.Series:
        high       = dataframe["high"]
        low        = dataframe["low"]
        prev_close = dataframe["close"].shift(1)
        return pd.concat(
            [(high - low), (high - prev_close).abs(), (low - prev_close).abs()],
            axis=1,
        ).max(axis=1)

    def _compute_rsi(self, series: pd.Series, period: int) -> pd.Series:
        delta    = series.diff()
        gain     = delta.clip(lower=0)
        loss     = (-delta).clip(lower=0)
        avg_gain = gain.ewm(com=period - 1, min_periods=period).mean()
        avg_loss = loss.ewm(com=period - 1, min_periods=period).mean()
        return 100 - (100 / (1 + avg_gain / avg_loss))

    def _compute_macd(self, series: pd.Series) -> tuple[pd.Series, pd.Series]:
        ema_fast    = series.ewm(span=self.MACD_FAST,   min_periods=self.MACD_FAST).mean()
        ema_slow    = series.ewm(span=self.MACD_SLOW,   min_periods=self.MACD_SLOW).mean()
        macd_line   = ema_fast - ema_slow
        signal_line = macd_line.ewm(span=self.MACD_SIGNAL, min_periods=self.MACD_SIGNAL).mean()
        return macd_line, signal_line

    def _compute_atr(self, dataframe: pd.DataFrame) -> pd.Series:
        return self._true_range(dataframe).ewm(
            span=self.ATR_PERIOD, min_periods=self.ATR_PERIOD
        ).mean()

    def _compute_adx(self, dataframe: pd.DataFrame) -> pd.Series:
        up       = dataframe["high"].diff()
        down     = -dataframe["low"].diff()
        plus_dm  = up.where((up > down) & (up > 0), 0.0)
        minus_dm = down.where((down > up) & (down > 0), 0.0)
        atr      = self._true_range(dataframe).ewm(
            span=self.ADX_PERIOD, min_periods=self.ADX_PERIOD
        ).mean()
        plus_di  = 100 * plus_dm.ewm(span=self.ADX_PERIOD, min_periods=self.ADX_PERIOD).mean() / atr
        minus_di = 100 * minus_dm.ewm(span=self.ADX_PERIOD, min_periods=self.ADX_PERIOD).mean() / atr
        dx       = ((plus_di - minus_di).abs() / (plus_di + minus_di)) * 100
        return dx.ewm(span=self.ADX_PERIOD, min_periods=self.ADX_PERIOD).mean()

    # ----------------------------------------------------------------
    # 指标计算 —— 只跑一次（不受 hyperopt 参数影响）
    #
    # MA 预计算技巧：一次性算出所有候选周期，存为独立列
    # populate_entry_trend 根据当前 hyperopt 参数值选列，
    # 避免每轮 hyperopt 重跑 populate_indicators（性能提升 ~10x）
    # ----------------------------------------------------------------

    # ma_short 候选范围 [5, 20]，ma_long 候选范围 [21, 60]
    _MA_SHORT_RANGE = range(5,  21)    # 16 个周期
    _MA_LONG_RANGE  = range(21, 61)    # 40 个周期

    def populate_indicators(self, dataframe: pd.DataFrame, metadata: dict) -> pd.DataFrame:
        # 预计算所有候选 MA
        for p in self._MA_SHORT_RANGE:
            dataframe[f"ma_{p}"] = dataframe["close"].rolling(p).mean()
        for p in self._MA_LONG_RANGE:
            dataframe[f"ma_{p}"] = dataframe["close"].rolling(p).mean()

        # 固定周期指标（只算一次）
        dataframe["rsi"]                         = self._compute_rsi(dataframe["close"], self.RSI_PERIOD)
        dataframe["atr"]                         = self._compute_atr(dataframe)
        dataframe["adx"]                         = self._compute_adx(dataframe)
        dataframe["macd"], dataframe["macd_sig"] = self._compute_macd(dataframe["close"])

        # 日线趋势（merge 1d MA20）
        informative_1d = self.dp.get_pair_dataframe(metadata["pair"], "1d")
        if not informative_1d.empty and "date" in dataframe.columns:
            informative_1d = informative_1d.copy()
            informative_1d["daily_ma20"] = informative_1d["close"].rolling(self.DAILY_MA_PERIOD).mean()
            informative_1d["daily_bull"] = (informative_1d["close"] > informative_1d["daily_ma20"]).astype(int)
            informative_1d["date"] = pd.to_datetime(informative_1d["date"])

            # 以 1h K 线的 date 对应的"当日"去匹配日线：用 merge_asof 向前对齐
            df_sorted = dataframe.sort_values("date").reset_index(drop=True)
            df_sorted["date"] = pd.to_datetime(df_sorted["date"])
            info_sorted = informative_1d[["date", "daily_bull"]].sort_values("date").reset_index(drop=True)
            merged = pd.merge_asof(
                df_sorted, info_sorted, on="date", direction="backward"
            )
            dataframe = merged
            dataframe["daily_bull"] = dataframe["daily_bull"].ffill().fillna(-1).astype(int)
        else:
            dataframe["daily_bull"] = -1   # 获取失败时不过滤

        return dataframe

    # ----------------------------------------------------------------
    # 入场信号 —— 六因子过滤（多空独立 ADX + 日线趋势）
    # ----------------------------------------------------------------

    def populate_entry_trend(self, dataframe: pd.DataFrame, metadata: dict) -> pd.DataFrame:
        ms = self.ma_short.value
        ml = self.ma_long.value

        if ms >= ml or f"ma_{ms}" not in dataframe or f"ma_{ml}" not in dataframe:
            dataframe["enter_long"]  = 0
            dataframe["enter_short"] = 0
            return dataframe

        cur_short  = dataframe[f"ma_{ms}"]
        cur_long   = dataframe[f"ma_{ml}"]
        prev_short = cur_short.shift(1)
        prev_long  = cur_long.shift(1)

        # 多空使用独立 ADX 阈值
        long_adx_ok  = dataframe["adx"] >= self.adx_threshold.value
        short_adx_ok = dataframe["adx"] >= self.adx_threshold_short.value

        # 日线趋势：1=牛，0=熊，-1=未知
        daily_bull = dataframe.get("daily_bull", pd.Series(-1, index=dataframe.index))
        bull_ok  = (daily_bull != 0)   # 牛市或未知 → 可做多
        bear_ok  = (daily_bull != 1)   # 熊市或未知 → 可做空

        # 波动率过滤：ATR/Close 过高 → 极端行情，不入场
        atr_pct = dataframe["atr"] / dataframe["close"]
        vol_ok  = atr_pct <= self.MAX_ATR_PCT

        long_cond = (
            (prev_short < prev_long)
            & (cur_short  > cur_long)
            & (dataframe["rsi"]  < self.rsi_overbought.value)
            & (dataframe["macd"] > dataframe["macd_sig"])
            & long_adx_ok
            & bull_ok
            & vol_ok
        )

        short_cond = (
            (prev_short > prev_long)
            & (cur_short  < cur_long)
            & (dataframe["rsi"]  > self.rsi_oversold.value)
            & (dataframe["macd"] < dataframe["macd_sig"])
            & short_adx_ok
            & bear_ok
            & vol_ok
        )

        dataframe.loc[long_cond,  "enter_long"]  = 1
        dataframe.loc[short_cond, "enter_short"] = 1
        return dataframe

    def populate_exit_trend(self, dataframe: pd.DataFrame, metadata: dict) -> pd.DataFrame:
        dataframe["exit_long"]  = 0
        dataframe["exit_short"] = 0
        return dataframe

    # ----------------------------------------------------------------
    # ATR 动态止损止盈
    # ----------------------------------------------------------------

    def custom_exit(
        self,
        pair: str,
        trade: Trade,
        current_time: datetime,
        current_rate: float,
        current_profit: float,
        **kwargs: Any,
    ) -> str | None:
        dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)
        if dataframe.empty:
            return None

        last = dataframe.iloc[-1]
        atr  = float(last["atr"]) if not pd.isna(last["atr"]) else 0.0
        if atr <= 0:
            return None

        stop_dist = self.atr_sl_mult.value * atr
        entry     = trade.open_rate

        # 最大持仓时长：超过 MAX_HOLD_HOURS 强平（防横盘消耗）
        hold_hours = (current_time - trade.open_date_utc).total_seconds() / 3600
        if hold_hours >= self.MAX_HOLD_HOURS:
            return "max_hold_timeout"

        # T1 触发前：固定 ATR 止损（T1后止损由 custom_stoploss 跟踪接管）
        if trade.nr_of_successful_exits == 0:
            if trade.is_short:
                if current_rate >= entry + stop_dist:
                    return "atr_stop_short"
            else:
                if current_rate <= entry - stop_dist:
                    return "atr_stop_long"

        return None

    def adjust_trade_position(
        self,
        trade: Trade,
        current_time: datetime,
        current_rate: float,
        current_profit: float,
        min_stake: float | None,
        max_stake: float,
        current_entry_rate: float,
        current_exit_rate: float,
        current_entry_profit: float,
        current_exit_profit: float,
        **kwargs: Any,
    ) -> float | None:
        """
        加仓 & 部分平仓逻辑（与 bot.py 对齐）

        加仓规则：
          - 首次建仓 60%（通过 stake_amount × SCALE_INIT_RATIO 控制初始仓位）
          - 触发条件：价格顺方向移动 SCALE_TRIGGER_MULT × ATR
          - 仅加仓一次（nr_of_successful_entries == 1 时检查）

        部分平仓规则：
          - 1R 处平掉 50%（返回负数 stake，Freqtrade 卖出对应仓位）
          - 仅触发一次（nr_of_successful_exits == 0 时检查）
        """
        dataframe, _ = self.dp.get_analyzed_dataframe(trade.pair, self.timeframe)
        if dataframe.empty:
            return None

        last = dataframe.iloc[-1]
        atr  = float(last["atr"]) if not pd.isna(last["atr"]) else 0.0
        if atr <= 0:
            return None

        stop_dist  = self.atr_sl_mult.value * atr
        entry      = trade.open_rate
        nr_entries = trade.nr_of_successful_entries
        nr_exits   = trade.nr_of_successful_exits

        # ---- 加仓逻辑（仅执行一次）----
        if nr_entries == 1:
            scale_trigger = (
                entry + atr * self.SCALE_TRIGGER_MULT
                if not trade.is_short
                else entry - atr * self.SCALE_TRIGGER_MULT
            )
            triggered = (
                current_rate >= scale_trigger
                if not trade.is_short
                else current_rate <= scale_trigger
            )
            if triggered:
                add_stake = trade.stake_amount * (self.SCALE_ADD_RATIO / self.SCALE_INIT_RATIO)
                add_stake = min(add_stake, max_stake)
                if min_stake and add_stake >= min_stake:
                    return add_stake

        # ---- 部分平仓逻辑（1R 处，仅执行一次）----
        # T1 触发后止损由 custom_stoploss 跟踪接管（利润自由奔跑）
        if nr_exits == 0:
            target1 = (
                entry + stop_dist * self.PARTIAL_TARGET_MULT
                if not trade.is_short
                else entry - stop_dist * self.PARTIAL_TARGET_MULT
            )
            hit = (
                current_rate >= target1
                if not trade.is_short
                else current_rate <= target1
            )
            if hit:
                reduce_stake = -(trade.stake_amount * self.PARTIAL_CLOSE_RATIO)
                return reduce_stake

        return None

    # ----------------------------------------------------------------
    # 跟踪止损 —— T1 触发后启用，T1前直接用固定止损
    # ----------------------------------------------------------------

    def custom_stoploss(
        self,
        pair: str,
        trade: Trade,
        current_time: datetime,
        current_rate: float,
        current_profit: float,
        after_fill: bool,
        **kwargs: Any,
    ) -> float:
        """
        T1（部分平仓）触发前：使用固定 ATR 止损（由 custom_exit 处理，此处返回默认）
        T1 触发后：切换为 ATR 跟踪止损，以 max_rate/min_rate 为基准
        """
        # T1 触发前，返回 stoploss 默认值（custom_exit 处理固定止损）
        if trade.nr_of_successful_exits == 0:
            return self.stoploss

        dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)
        if dataframe.empty:
            return self.stoploss

        last = dataframe.iloc[-1]
        atr  = float(last["atr"]) if not pd.isna(last["atr"]) else 0.0
        if atr <= 0:
            return self.stoploss

        trail_dist = self.atr_trail_mult.value * atr

        if trade.is_short:
            # 跟踪最低价，止损在 min_rate + trail_dist
            stop_price = (trade.min_rate or current_rate) + trail_dist
            if stop_price >= current_rate:
                return -0.001   # 立即平仓
            # 返回相对于 current_rate 的负数百分比
            return (stop_price / current_rate) - 1.0
        else:
            # 跟踪最高价，止损在 max_rate - trail_dist
            stop_price = (trade.max_rate or current_rate) - trail_dist
            if stop_price <= 0 or stop_price >= current_rate:
                return -0.001
            # stoploss_from_absolute 将绝对价格转为相对值
            try:
                return stoploss_from_absolute(stop_price, current_rate, is_short=False)
            except Exception:
                return (stop_price / current_rate) - 1.0

    # ----------------------------------------------------------------
    # 日线 informative pairs（Freqtrade 内部数据接口）
    # ----------------------------------------------------------------

    def informative_pairs(self):
        pairs = self.dp.current_whitelist()
        return [(pair, "1d") for pair in pairs]