SupertrendFuturesStrategyV5

# SupertrendFuturesStrategyV5 - 30m 优化版 + Phase 1 改进
# 更新时间: 2026-02-22 10:15
# 优化数据: 90 天 30m 数据
# 改进: 分批止盈、RSI过滤加强、动态仓位

import numpy as np
import pandas as pd
from pandas import DataFrame
from datetime import datetime
from typing import Optional
from functools import reduce
import logging

from freqtrade.strategy import IStrategy, DecimalParameter, IntParameter
from freqtrade.persistence import Trade
import talib.abstract as ta

logger = logging.getLogger(__name__)


class SupertrendFuturesStrategyV5(IStrategy):
    """
    Supertrend + EMA 趋势跟踪策略 (合约版 V5)
    
    改进点 (2026-02-22):
    1. ✅ 时间周期优化: 15m -> 30m
    2. ✅ 参数优化: ATR period 11, multiplier 2.884
    3. 📋 分批止盈: 3%/5%/10% 三级止盈
    4. 📋 RSI 过滤加强: 避免极端超买超卖
    5. 📋 动态仓位: 基于 ATR 调整
    """
    
    INTERFACE_VERSION = 3

    # 参数 - 30m 优化后（90天数据，2026-02-22）
    atr_period = IntParameter(5, 30, default=11, space="buy")
    atr_multiplier = DecimalParameter(2.0, 5.0, default=2.884, space="buy")
    ema_fast = IntParameter(5, 50, default=48, space="buy")
    ema_slow = IntParameter(20, 200, default=151, space="buy")

    # ADX 参数 - 30m 优化后
    adx_threshold_long = IntParameter(20, 35, default=33, space="buy")
    adx_threshold_short = IntParameter(15, 30, default=23, space="buy")
    
    # RSI 参数 - 新增
    rsi_upper_limit = IntParameter(65, 80, default=70, space="buy")  # 做多时RSI上限
    rsi_lower_limit = IntParameter(20, 35, default=30, space="buy")  # 做空时RSI下限
    
    # 止盈参数 - 新增
    tp_level_1 = DecimalParameter(0.02, 0.05, default=0.03, space="sell")  # 一级止盈 3%
    tp_level_2 = DecimalParameter(0.04, 0.08, default=0.05, space="sell")  # 二级止盈 5%
    tp_level_3 = DecimalParameter(0.08, 0.15, default=0.10, space="sell")  # 三级止盈 10%
    
    # 动态仓位参数 - 新增
    atr_low_threshold = DecimalParameter(0.02, 0.04, default=0.03, space="buy")   # 低波动阈值
    atr_high_threshold = DecimalParameter(0.04, 0.06, default=0.05, space="buy")  # 高波动阈值

    minimal_roi = {"0": 0.06}
    stoploss = -0.03  # 3% (最优止损)

    timeframe = '30m'  # 2026-02-22: 15m -> 30m (回测显示30m表现更优)

    trailing_stop = True
    trailing_stop_positive = 0.02
    trailing_stop_positive_offset = 0.03
    trailing_only_offset_is_reached = True

    startup_candle_count = 200

    order_types = {
        'entry': 'limit',
        'exit': 'limit',
        'stoploss': 'market',
        'stoploss_on_exchange': False
    }

    can_short: bool = True
    leverage_default = 2
    
    # 记录止盈状态
    custom_info_trail = {}

    def supertrend(self, dataframe, period=14, multiplier=3):
        """计算 Supertrend 指标"""
        df = dataframe.copy()
        hl2 = (df['high'] + df['low']) / 2
        atr = ta.ATR(df, timeperiod=period)
        upperband = hl2 + (multiplier * atr)
        lowerband = hl2 - (multiplier * atr)
        supertrend = [0] * len(df)
        direction = [1] * len(df)
        for i in range(1, len(df)):
            if df['close'].iloc[i] > upperband.iloc[i-1]:
                direction[i] = 1
            elif df['close'].iloc[i] < lowerband.iloc[i-1]:
                direction[i] = -1
            else:
                direction[i] = direction[i-1]
            supertrend[i] = lowerband.iloc[i] if direction[i] == 1 else upperband.iloc[i]
        return pd.Series(supertrend, index=df.index), pd.Series(direction, index=df.index)

    def leverage(self, pair, current_time, current_rate, proposed_leverage, max_leverage, entry_tag, side, **kwargs):
        return min(self.leverage_default, max_leverage)

    def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        """计算技术指标"""
        dataframe['ema_fast'] = ta.EMA(dataframe, timeperiod=self.ema_fast.value)
        dataframe['ema_slow'] = ta.EMA(dataframe, timeperiod=self.ema_slow.value)
        dataframe['supertrend'], dataframe['st_dir'] = self.supertrend(
            dataframe, period=self.atr_period.value, multiplier=self.atr_multiplier.value
        )
        dataframe['rsi'] = ta.RSI(dataframe, timeperiod=14)
        dataframe['adx'] = ta.ADX(dataframe, timeperiod=14)
        dataframe['adx_pos'] = ta.PLUS_DI(dataframe, timeperiod=14)
        dataframe['adx_neg'] = ta.MINUS_DI(dataframe, timeperiod=14)
        dataframe['atr'] = ta.ATR(dataframe, timeperiod=14)
        dataframe['volume_ma'] = dataframe['volume'].rolling(20).mean()

        # 趋势判断：EMA 200 判断大趋势
        dataframe['ema_200'] = ta.EMA(dataframe, timeperiod=200)
        dataframe['is_uptrend'] = dataframe['close'] > dataframe['ema_200']
        dataframe['is_downtrend'] = dataframe['close'] < dataframe['ema_200']
        
        # ATR 占比 (用于动态仓位)
        dataframe['atr_ratio'] = dataframe['atr'] / dataframe['close']

        return dataframe

    def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        """
        做多 - 优化版
        
        改进:
        1. ✅ ADX > 33 趋势强度确认
        2. ✅ RSI < 70 避免超买
        3. ✅ 成交量确认
        4. ✅ EMA 趋势确认
        5. ✅ 大趋势确认 (EMA 200)
        """
        dataframe.loc[:, 'enter_long'] = 0

        conditions = [
            # Supertrend 信号
            dataframe['st_dir'] == 1,
            # EMA 趋势确认
            dataframe['ema_fast'] > dataframe['ema_slow'],
            # ADX 趋势强度
            dataframe['adx'] > self.adx_threshold_long.value,
            dataframe['adx_pos'] > dataframe['adx_neg'],
            # RSI 避免极端超买
            dataframe['rsi'] < self.rsi_upper_limit.value,
            dataframe['rsi'] > 20,  # 避免极端超卖
            # 成交量确认
            dataframe['volume'] > dataframe['volume_ma'],
            # 价格在 Supertrend 上方
            dataframe['close'] > dataframe['supertrend'],
            # 大趋势确认
            dataframe['is_uptrend'],
        ]

        if conditions:
            dataframe.loc[reduce(lambda x, y: x & y, conditions), 'enter_long'] = 1

        return dataframe

    def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        """做多退出"""
        dataframe.loc[:, 'exit_long'] = 0
        conditions = [dataframe['st_dir'] == -1]
        if conditions:
            dataframe.loc[reduce(lambda x, y: x & y, conditions), 'exit_long'] = 1
        return dataframe

    def populate_entry_trend_short(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        """
        做空 - 优化版
        
        改进:
        1. ✅ ADX > 23 趋势强度确认
        2. ✅ RSI > 30 避免超卖
        3. ✅ 成交量确认
        4. ✅ EMA 趋势确认
        5. ✅ 大趋势确认 (EMA 200)
        """
        dataframe.loc[:, 'enter_short'] = 0

        conditions = [
            # Supertrend 信号
            dataframe['st_dir'] == -1,
            # EMA 趋势确认
            dataframe['ema_fast'] < dataframe['ema_slow'],
            # ADX 趋势强度
            dataframe['adx'] > self.adx_threshold_short.value,
            dataframe['adx_neg'] > dataframe['adx_pos'],
            # RSI 避免极端超卖
            dataframe['rsi'] > self.rsi_lower_limit.value,
            dataframe['rsi'] < 80,  # 避免极端超买
            # 价格在 Supertrend 下方
            dataframe['close'] < dataframe['supertrend'],
            # 大趋势确认
            dataframe['is_downtrend'],
        ]

        if conditions:
            dataframe.loc[reduce(lambda x, y: x & y, conditions), 'enter_short'] = 1

        return dataframe

    def populate_exit_trend_short(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        """做空退出"""
        dataframe.loc[:, 'exit_short'] = 0
        conditions = [dataframe['st_dir'] == 1]
        if conditions:
            dataframe.loc[reduce(lambda x, y: x & y, conditions), 'exit_short'] = 1
        return dataframe
    
    def custom_stake_amount(self, pair: str, current_time: datetime, current_rate: float,
                           proposed_stake: float, min_stake: Optional[float], max_stake: float,
                           entry_tag: Optional[str], side: str, **kwargs) -> float:
        """
        动态仓位管理 - 基于 ATR
        
        原理:
        - 高波动 (ATR > 5%) → 小仓位 (50%)
        - 中波动 (ATR 3-5%) → 中仓位 (75%)
        - 低波动 (ATR < 3%) → 正常仓位 (100%)
        """
        dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)
        
        if len(dataframe) < 1:
            return proposed_stake
        
        atr_ratio = dataframe['atr_ratio'].iloc[-1]
        
        # 根据波动率调整仓位
        if atr_ratio > self.atr_high_threshold.value:
            # 高波动 - 减半仓位
            stake = proposed_stake * 0.5
            logger.info(f"{pair} 高波动 (ATR ratio: {atr_ratio:.4f}), 仓位减半")
        elif atr_ratio > self.atr_low_threshold.value:
            # 中波动 - 75% 仓位
            stake = proposed_stake * 0.75
            logger.info(f"{pair} 中波动 (ATR ratio: {atr_ratio:.4f}), 仓位 75%")
        else:
            # 低波动 - 正常仓位
            stake = proposed_stake
            logger.info(f"{pair} 低波动 (ATR ratio: {atr_ratio:.4f}), 正常仓位")
        
        # 确保不低于最小值
        if min_stake is not None and stake < min_stake:
            return min_stake
        
        return stake
    
    def custom_exit(self, pair: str, trade: Trade, current_time: datetime, 
                   current_rate: float, current_profit: float, **kwargs) -> Optional[str]:
        """
        分批止盈逻辑
        
        改进:
        - 3% 利润 → 部分止盈
        - 5% 利润 → 加速止盈
        - 10% 利润 → 全部平仓
        """
        dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)
        
        if len(dataframe) < 1:
            return None
        
        # 获取当前利润百分比
        profit_pct = current_profit
        
        # 三级止盈
        if profit_pct >= self.tp_level_3.value:
            logger.info(f"{pair} 达到三级止盈 {profit_pct:.2%} (目标: {self.tp_level_3.value:.2%})")
            return f'profit_{int(self.tp_level_3.value*100)}pct'
        
        elif profit_pct >= self.tp_level_2.value:
            logger.info(f"{pair} 达到二级止盈 {profit_pct:.2%} (目标: {self.tp_level_2.value:.2%})")
            return f'profit_{int(self.tp_level_2.value*100)}pct'
        
        elif profit_pct >= self.tp_level_1.value:
            logger.info(f"{pair} 达到一级止盈 {profit_pct:.2%} (目标: {self.tp_level_1.value:.2%})")
            return f'profit_{int(self.tp_level_1.value*100)}pct'
        
        # RSI 反转信号
        last_candle = dataframe.iloc[-1]
        if trade.is_short:
            # 做空时 RSI 超卖
            if last_candle['rsi'] < 30:
                logger.info(f"{pair} RSI 超卖反转 (RSI: {last_candle['rsi']:.2f})")
                return 'rsi_oversold_exit'
        else:
            # 做多时 RSI 超买
            if last_candle['rsi'] > 70:
                logger.info(f"{pair} RSI 超买反转 (RSI: {last_candle['rsi']:.2f})")
                return 'rsi_overbought_exit'
        
        return None
    
    def confirm_trade_entry(self, pair: str, order_type: str, amount: float, 
                           rate: float, time_in_force: str, current_time: datetime,
                           entry_tag: Optional[str], side: str, **kwargs) -> bool:
        """
        入场确认 - 记录交易信息
        """
        logger.info(f"确认入场: {pair} | 方向: {side} | 金额: {amount:.2f} | 价格: {rate:.2f}")
        return True