UnifiedLong4hBTC

"""
Freqtrade策略文件 - 包含两个策略
1. UnifiedLong4hBTC - 保守型做多策略
2. RSIOversold4hBTC - RSI超卖反弹策略

使用方法：
freqtrade backtesting --strategy UnifiedLong4hBTC ...
freqtrade backtesting --strategy RSIOversold4hBTC ...
"""

import talib.abstract as ta
import pandas as pd
from pandas import DataFrame
from freqtrade.strategy import IStrategy, DecimalParameter, IntParameter
from datetime import datetime, timedelta


# ============================================================
# 策略1: UnifiedLong4hBTC - 保守型做多策略
# ============================================================
class UnifiedLong4hBTC(IStrategy):
    """
    保守型做多策略，适合大市值币种（BTC/ETH等）
    时间周期：4h

    买入信号：
    - 成交量过滤：平均成交量 >= 阈值
    - MACD金叉：DIF > DEA
    - KDJ金叉：K > D，且满足上限过滤
    - 市场状态过滤：ATR比率、ADX、价格波动范围、RSI、StochRSI等

    卖出信号：
    - 固定止损：5%
    - ATR移动止损：盈利达到1×ATR后，止损线为最高价-3×ATR
    """

    # 策略参数
    timeframe = '4h'

    # ROI 配置（禁用，使用移动止损）
    minimal_roi = {
        "0": 100  # 不使用ROI，完全依靠止损
    }

    # 固定止损
    stoploss = -0.05  # -5%

    # 关闭百分比移动止损（使用ATR移动止损）
    trailing_stop = False

    # 使用自定义止损
    use_custom_stoploss = True

    # 参数配置
    min_volume_threshold = DecimalParameter(1000, 2000, default=1500, space='buy', optimize=True)
    long_kdj_max_k = DecimalParameter(80, 90, default=88, space='buy', optimize=True)
    long_kdj_max_d = DecimalParameter(75, 85, default=83, space='buy', optimize=True)
    long_kdj_max_j = DecimalParameter(100, 120, default=110, space='buy', optimize=True)
    long_min_atr_ratio = DecimalParameter(0.5, 0.8, default=0.6, space='buy', optimize=True)
    long_min_adx = DecimalParameter(10, 20, default=16, space='buy', optimize=True)
    long_min_price_range_30 = DecimalParameter(3, 6, default=4.8, space='buy', optimize=True)
    long_max_price_range_30 = DecimalParameter(14, 18, default=16, space='buy', optimize=True)
    long_max_rsi = DecimalParameter(70, 80, default=76, space='buy', optimize=True)
    long_min_stochrsi = DecimalParameter(40, 60, default=52, space='buy', optimize=True)
    long_min_macd_dif = DecimalParameter(-300, -100, default=-200, space='buy', optimize=True)
    long_min_macd_dea = DecimalParameter(-500, -200, default=-400, space='buy', optimize=True)

    # ATR移动止损参数
    atr_trailing_period = IntParameter(10, 20, default=14, space='buy', optimize=False)
    atr_trailing_multiplier = DecimalParameter(2.0, 4.0, default=3.0, space='sell', optimize=True)
    atr_trailing_trigger = DecimalParameter(0.5, 2.0, default=1.0, space='sell', optimize=True)  # 盈利达到1×ATR时激活

    def __init__(self, config: dict) -> None:
        super().__init__(config)
        self.entry_atr = {}  # pair -> entry_atr (入场时的ATR值)

    def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        """计算技术指标"""

        # MACD (12, 26, 9)
        macd = ta.MACD(dataframe, fastperiod=12, slowperiod=26, signalperiod=9)
        dataframe['macd'] = macd['macd']
        dataframe['macd_signal'] = macd['macdsignal']
        dataframe['macd_hist'] = macd['macdhist']

        # RSI (14)
        dataframe['rsi'] = ta.RSI(dataframe, timeperiod=14)

        # Stochastic RSI
        stoch_rsi = ta.STOCHRSI(dataframe, timeperiod=14, fastk_period=14, fastd_period=3)
        dataframe['stochrsi_k'] = stoch_rsi['fastk']
        dataframe['stochrsi_d'] = stoch_rsi['fastd']

        # KDJ (基于Stochastic)
        stoch = ta.STOCH(dataframe, fastk_period=9, slowk_period=3, slowd_period=3)
        dataframe['kdj_k'] = stoch['slowk']
        dataframe['kdj_d'] = stoch['slowd']
        dataframe['kdj_j'] = 3 * dataframe['kdj_k'] - 2 * dataframe['kdj_d']

        # ATR (14)
        dataframe['atr'] = ta.ATR(dataframe, timeperiod=14)

        # ATR比率 (ATR(14) / SMA(ATR, 60))
        dataframe['atr_sma60'] = ta.SMA(dataframe['atr'], timeperiod=60)
        dataframe['atr_ratio'] = dataframe['atr'] / dataframe['atr_sma60']

        # ADX (14)
        dataframe['adx'] = ta.ADX(dataframe, timeperiod=14)

        # 30根K线价格波动范围百分比
        dataframe['price_high_30'] = dataframe['high'].rolling(window=30).max()
        dataframe['price_low_30'] = dataframe['low'].rolling(window=30).min()
        dataframe['price_range_30'] = (
            (dataframe['price_high_30'] - dataframe['price_low_30']) / dataframe['price_low_30'] * 100
        )

        # EMA (40) - 用于可视化
        dataframe['ema40'] = ta.EMA(dataframe, timeperiod=40)

        return dataframe

    def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        """买入信号"""

        dataframe.loc[
            (
                # 成交量过滤
                (dataframe['volume'] >= self.min_volume_threshold.value) &

                # MACD金叉
                (dataframe['macd'] > dataframe['macd_signal']) &

                # KDJ金叉及上限
                (dataframe['kdj_k'] > dataframe['kdj_d']) &
                (dataframe['kdj_j'] < self.long_kdj_max_j.value) &
                (dataframe['kdj_k'] < self.long_kdj_max_k.value) &
                (dataframe['kdj_d'] < self.long_kdj_max_d.value) &

                # ATR比率（避免震荡期）
                (dataframe['atr_ratio'] >= self.long_min_atr_ratio.value) &

                # ADX（趋势强度）
                (dataframe['adx'] >= self.long_min_adx.value) &

                # 价格波动范围
                (dataframe['price_range_30'] >= self.long_min_price_range_30.value) &
                (dataframe['price_range_30'] <= self.long_max_price_range_30.value) &

                # RSI过滤
                (dataframe['rsi'] <= self.long_max_rsi.value) &

                # StochRSI过滤
                (dataframe['stochrsi_k'] >= self.long_min_stochrsi.value) &

                # MACD DIF/DEA过滤
                (dataframe['macd'] >= self.long_min_macd_dif.value) &
                (dataframe['macd_signal'] >= self.long_min_macd_dea.value) &

                # 有效数据
                (dataframe['volume'] > 0)
            ),
            'enter_long'] = 1

        return dataframe

    def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        """卖出信号（主要依靠止损，这里可以添加其他退出条件）"""

        # 不设置主动退出信号，完全依靠止损
        dataframe['exit_long'] = 0

        return dataframe

    def confirm_trade_entry(self, pair: str, order_type: str, amount: float, rate: float,
                           time_in_force: str, current_time: datetime, entry_tag: str, **kwargs) -> bool:
        """
        确认交易入场时，存储入场时的ATR值

        注意：使用前一根已完成K线的ATR（因为入场信号基于前一根K线计算）
        """
        dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)

        if len(dataframe) >= 2:
            # 使用前一根已完成K线的ATR（-2位置），而不是最新K线（-1位置）
            # 因为入场信号是基于前一根K线的指标计算的
            prev_candle = dataframe.iloc[-2].squeeze()
            if 'atr' in prev_candle:
                self.entry_atr[pair] = float(prev_candle['atr'])

        return True

    def custom_stoploss(self, pair: str, trade: 'Trade', current_time: datetime,
                       current_rate: float, current_profit: float, **kwargs) -> float:
        """
        自定义止损逻辑 - ATR移动止损

        逻辑：
        1. 触发条件：盈利金额 >= EntryATR × atr_trailing_trigger
        2. 止损价 = 最高价 - EntryATR × atr_trailing_multiplier
        3. 止损只能向上调整（保护利润）

        Returns:
            float: 止损百分比（负数，相对于入场价）
                  例如：-0.05 = -5% 止损
        """

        # 如果没有存储入场ATR，使用固定止损
        if pair not in self.entry_atr:
            return self.stoploss  # -0.05

        entry_atr = self.entry_atr[pair]
        entry_price = trade.open_rate
        highest_price = trade.max_rate if hasattr(trade, 'max_rate') else current_rate

        # 计算当前盈利金额（基于当前价格）
        current_profit_amount = current_rate - entry_price

        # ATR触发金额：EntryATR × atr_trailing_trigger
        atr_trigger_amount = entry_atr * self.atr_trailing_trigger.value

        # 检查是否达到触发条件
        if current_profit_amount >= atr_trigger_amount:
            # 计算ATR止损价：最高价 - (EntryATR × atr_trailing_multiplier)
            atr_stop_price = highest_price - (entry_atr * self.atr_trailing_multiplier.value)

            # 计算止损百分比（相对于入场价）
            stop_loss_pct = (atr_stop_price - entry_price) / entry_price

            # 确保止损不会低于固定止损
            if stop_loss_pct > self.stoploss:
                return stop_loss_pct

        # 未达到触发条件或计算出的止损低于固定止损，使用固定止损
        return self.stoploss

    def confirm_trade_exit(self, pair: str, trade: 'Trade', order_type: str, amount: float,
                          rate: float, time_in_force: str, exit_reason: str, **kwargs) -> bool:
        """确认退出时清理entry_atr记录"""
        if pair in self.entry_atr:
            del self.entry_atr[pair]
        return True


# ============================================================
# 策略2: RSIOversold4hBTC - RSI超卖反弹策略
# ============================================================
class RSIOversold4hBTC(IStrategy):
    """
    RSI超卖反弹策略
    时间周期：4h

    买入信号：
    - RSI < 30 (超卖)
    - 额外过滤：StochRSI <= 10, KDJ K <= 35, KDJ D <= 35, -3 <= KDJ J <= 1

    卖出信号：
    - 固定持仓3根K线（12小时）
    - 止损：2%
    """

    # 策略参数
    timeframe = '4h'

    # ROI 配置（禁用）
    minimal_roi = {
        "0": 100
    }

    # 固定止损
    stoploss = -0.02  # -2%

    # 禁用移动止损
    trailing_stop = False

    # 参数配置
    rsi_period = IntParameter(10, 20, default=14, space='buy', optimize=True)
    rsi_oversold_threshold = DecimalParameter(20, 35, default=30, space='buy', optimize=True)
    holding_bars = IntParameter(2, 5, default=3, space='sell', optimize=True)

    # 额外过滤条件
    enable_filters = True
    stochrsi_threshold = DecimalParameter(5, 15, default=10, space='buy', optimize=True)
    kdj_k_threshold = DecimalParameter(25, 40, default=35, space='buy', optimize=True)
    kdj_d_threshold = DecimalParameter(25, 40, default=35, space='buy', optimize=True)
    kdj_j_max_threshold = DecimalParameter(0, 5, default=1, space='buy', optimize=True)
    kdj_j_min_threshold = DecimalParameter(-5, 0, default=-3, space='buy', optimize=True)

    # 用于追踪持仓时间
    def __init__(self, config: dict) -> None:
        super().__init__(config)
        self.entry_candle_time = {}  # pair -> entry candle timestamp

    def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        """计算技术指标"""

        # RSI
        dataframe['rsi'] = ta.RSI(dataframe, timeperiod=self.rsi_period.value)

        # Stochastic RSI
        stoch_rsi = ta.STOCHRSI(dataframe, timeperiod=self.rsi_period.value,
                               fastk_period=14, fastd_period=3)
        dataframe['stochrsi_k'] = stoch_rsi['fastk']
        dataframe['stochrsi_d'] = stoch_rsi['fastd']

        # KDJ
        stoch = ta.STOCH(dataframe, fastk_period=9, slowk_period=3, slowd_period=3)
        dataframe['kdj_k'] = stoch['slowk']
        dataframe['kdj_d'] = stoch['slowd']
        dataframe['kdj_j'] = 3 * dataframe['kdj_k'] - 2 * dataframe['kdj_d']

        # MACD (用于记录)
        macd = ta.MACD(dataframe, fastperiod=12, slowperiod=26, signalperiod=9)
        dataframe['macd'] = macd['macd']
        dataframe['macd_signal'] = macd['macdsignal']

        # 标记买入信号出现的位置（前一根K线RSI超卖）
        dataframe['rsi_oversold_prev'] = (dataframe['rsi'].shift(1) < self.rsi_oversold_threshold.value)

        return dataframe

    def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        """买入信号 - 前一根K线RSI超卖，当前K线开盘买入"""

        conditions = [
            # 前一根K线RSI超卖
            (dataframe['rsi'].shift(1) < self.rsi_oversold_threshold.value),
            (dataframe['rsi'].shift(1) > 0),  # 确保RSI有效
        ]

        # 额外过滤条件
        if self.enable_filters:
            conditions.extend([
                # StochRSI超卖
                (dataframe['stochrsi_k'].shift(1) <= self.stochrsi_threshold.value),

                # KDJ超卖
                (dataframe['kdj_k'].shift(1) <= self.kdj_k_threshold.value),
                (dataframe['kdj_d'].shift(1) <= self.kdj_d_threshold.value),
                (dataframe['kdj_j'].shift(1) >= self.kdj_j_min_threshold.value),
                (dataframe['kdj_j'].shift(1) <= self.kdj_j_max_threshold.value),
            ])

        # 合并所有条件
        dataframe.loc[
            (
                conditions[0] &
                (conditions[1] if len(conditions) > 1 else True) &
                (conditions[2] if len(conditions) > 2 else True) &
                (conditions[3] if len(conditions) > 3 else True) &
                (conditions[4] if len(conditions) > 4 else True) &
                (conditions[5] if len(conditions) > 5 else True) &
                (conditions[6] if len(conditions) > 6 else True) &
                (dataframe['volume'] > 0)
            ),
            'enter_long'] = 1

        return dataframe

    def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        """卖出信号 - 持仓固定K线数后卖出（通过custom_exit处理）"""

        # 不在这里设置退出信号，使用 custom_exit_long
        dataframe['exit_long'] = 0

        return dataframe

    def custom_exit(self, pair: str, trade: 'Trade', current_time: datetime,
                   current_rate: float, current_profit: float, **kwargs) -> bool:
        """
        自定义退出逻辑 - 持仓固定K线数后卖出

        Returns:
            bool: True表示退出
        """

        # 记录入场时的K线时间
        if pair not in self.entry_candle_time:
            self.entry_candle_time[pair] = trade.open_date_utc

        # 计算持仓时间（按4h K线数量）
        time_diff = current_time - self.entry_candle_time[pair]
        hours_held = time_diff.total_seconds() / 3600
        bars_held = int(hours_held / 4)  # 4h周期

        # 持仓达到目标K线数后退出
        if bars_held >= self.holding_bars.value:
            return True

        return False

    def confirm_trade_exit(self, pair: str, trade: 'Trade', order_type: str, amount: float,
                          rate: float, time_in_force: str, exit_reason: str, **kwargs) -> bool:
        """确认退出时清理记录"""

        if pair in self.entry_candle_time:
            del self.entry_candle_time[pair]

        return True


# ============================================================
# 策略3: CombinedStrategy - 组合策略（可选）
# ============================================================
class CombinedStrategy(IStrategy):
    """
    组合策略 - 同时使用UnifiedLong和RSIOversold的信号

    买入信号：满足任一策略的买入条件
    卖出信号：根据入场策略选择对应的退出条件

    注意：此策略仅作为示例，实际使用时需要仔细测试和调优
    """

    timeframe = '4h'

    minimal_roi = {
        "0": 100
    }

    stoploss = -0.05

    trailing_stop = False

    def __init__(self, config: dict) -> None:
        super().__init__(config)
        self.unified = UnifiedLong4hBTC(config)
        self.rsi_oversold = RSIOversold4hBTC(config)
        self.entry_strategy = {}  # pair -> strategy_name

    def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        """计算两个策略的所有指标"""
        dataframe = self.unified.populate_indicators(dataframe, metadata)
        dataframe = self.rsi_oversold.populate_indicators(dataframe, metadata)
        return dataframe

    def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        """组合买入信号"""

        # 计算两个策略的信号
        df_unified = self.unified.populate_entry_trend(dataframe.copy(), metadata)
        df_rsi = self.rsi_oversold.populate_entry_trend(dataframe.copy(), metadata)

        # 优先级：RSI超卖策略 > Unified策略
        dataframe.loc[
            (df_rsi['enter_long'] == 1),
            'enter_long'] = 1

        dataframe.loc[
            ((df_unified['enter_long'] == 1) & (dataframe['enter_long'] != 1)),
            'enter_long'] = 1

        return dataframe

    def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        """组合卖出信号"""
        dataframe['exit_long'] = 0
        return dataframe

    def custom_exit(self, pair: str, trade: 'Trade', current_time: datetime,
                   current_rate: float, current_profit: float, **kwargs) -> bool:
        """根据入场策略选择退出逻辑"""

        # 这里可以根据需要实现组合退出逻辑
        # 简化处理：使用RSI策略的退出条件
        return self.rsi_oversold.custom_exit(pair, trade, current_time,
                                            current_rate, current_profit, **kwargs)