Chan/strategies/ElliottWaveBTCStrategyV2.py

"""
Elliott Wave Strategy for BTC Perpetual Futures V2
基于真正的艾略特波浪理论 + 缠论分型确认

核心逻辑:
- 艾略特波浪识别: 自动识别1-5浪上涨和A-C浪下跌
- 多时间框架确认: 5m入场，1h确认趋势方向，1d确认大周期浪型
- 双向交易: 根据波浪位置决定做多或做空
- 动态风险管理: 根据波动率调整仓位和止损

改进点:
1. 实现真正的波浪计数器 (Wave Counter)
2. 斐波那契回撤/扩展用于止盈止损
3. 波浪完成度评估
4. 多周期共振确认
5. 市场情绪过滤

作者: AI Assistant (Optimized)
"""

import sys
import os
sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from chan.pipeline.ChanLun import ChanLun

from freqtrade.strategy import IStrategy
from pandas import DataFrame
import pandas as pd
import numpy as np
import talib.abstract as ta
import logging
from datetime import datetime
from typing import Optional, Tuple, List
from dataclasses import dataclass
from enum import Enum

logger = logging.getLogger(__name__)


class WaveType(Enum):
    """波浪类型"""
    IMPULSE = "impulse"      # 推动浪 (1,2,3,4,5)
    CORRECTIVE = "corrective"  # 调整浪 (A,B,C)
    UNKNOWN = "unknown"


class WavePosition(Enum):
    """当前在波浪中的位置"""
    WAVE_1 = 1
    WAVE_2 = 2
    WAVE_3 = 3
    WAVE_4 = 4
    WAVE_5 = 5
    WAVE_A = 6
    WAVE_B = 7
    WAVE_C = 8
    UNKNOWN = 0


@dataclass
class Wave:
    """波浪数据结构"""
    start_idx: int
    end_idx: int
    start_price: float
    end_price: float
    wave_num: int  # 1-5 or 6-8 (A-C)
    wave_type: WaveType
    is_complete: bool = False


class ElliottWaveBTCStrategyV2(IStrategy):
    """
    艾略特波浪理论策略 V2
    结合缠论分型进行波浪确认
    """

    INTERFACE_VERSION = 3
    can_short = True

    # 基础止损止盈 (会根据波动率动态调整)
    stoploss = -0.015
    minimal_roi = {
        "0": 0.08,
        "60": 0.05,
        "120": 0.03,
        "240": 0.015
    }

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

    startup_candle_count = 1000
    position_adjustment_enable = True

    pair = 'BTC/USDT:USDT'
    timeframe = '5m'
    chan = ChanLun()

    # ========== 策略参数 (可优化) ==========
    # 波浪检测参数
    wave_pivot_lookback = 5          # 波浪枢轴点回看周期
    min_wave_bars = 8                # 最小波浪K线数
    max_wave_bars = 200              # 最大波浪K线数

    # 斐波那契参数
    fib_entry_threshold = 0.618      # 入场回撤位
    fib_target_1 = 1.272             # 第一目标位
    fib_target_2 = 1.618             # 第二目标位
    fib_stop_loss = 0.5              # 止损位 (低于/高于0.5)

    # RSI参数
    rsi_oversold = 35
    rsi_overbought = 65
    rsi_period = 14

    # 波动率参数
    atr_period = 14
    atr_multiplier_entry = 1.5       # 入场ATR倍数
    atr_multiplier_stop = 2.0        # 止损ATR倍数

    # 趋势过滤参数
    ema_trend_period = 200
    trend_filter_strict = True       # 严格趋势过滤

    # 波浪完成度阈值
    wave_completion_threshold = 0.8

    def informative_pairs(self):
        return [
            (self.pair, '5m'),
            (self.pair, '1h'),
            (self.pair, '4h'),
            (self.pair, '1d'),
        ]

    def _add_indicators(self, df: DataFrame) -> DataFrame:
        """添加技术指标"""
        # 基础EMA
        df['ema20'] = ta.EMA(df, timeperiod=20)
        df['ema50'] = ta.EMA(df, timeperiod=50)
        df['ema200'] = ta.EMA(df, timeperiod=self.ema_trend_period)

        # RSI
        df['rsi'] = ta.RSI(df, timeperiod=self.rsi_period)
        df['rsi_ma'] = df['rsi'].rolling(window=9).mean()

        # ATR
        df['atr'] = ta.ATR(df, timeperiod=self.atr_period)
        df['atr_percent'] = df['atr'] / df['close'] * 100

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

        # 布林带
        bb = ta.BBANDS(df, timeperiod=20, nbdevup=2.0, nbdevdn=2.0)
        df['bb_upper'] = bb['upper']
        df['bb_lower'] = bb['lower']
        df['bb_middle'] = bb['middle']
        df['bb_percent'] = (df['close'] - df['bb_lower']) / (df['bb_upper'] - df['bb_lower'])

        # 成交量
        df['volume_sma'] = ta.SMA(df, timeperiod=20)
        df['volume_ratio'] = df['volume'] / df['volume_sma']

        # 波动率
        df['volatility'] = df['close'].pct_change().rolling(20).std() * np.sqrt(365 * 24 * 12)

        return df

    def _detect_pivots(self, df: DataFrame, left_bars: int = 5) -> Tuple[List[int], List[int]]:
        """
        检测价格枢轴点 (用于波浪识别) - 无未来数据版本
        只使用左侧已确认的数据，避免lookahead bias
        返回: (高点索引列表, 低点索引列表)
        """
        highs = []
        lows = []

        # 只使用左侧数据确认枢轴点，不使用right_bars避免未来数据
        for i in range(left_bars, len(df)):
            # 检测高点: 当前点比之前left_bars个bar都高
            is_high = True
            for j in range(1, left_bars + 1):
                if df['high'].iloc[i] <= df['high'].iloc[i - j]:
                    is_high = False
                    break
            if is_high:
                highs.append(i)

            # 检测低点: 当前点比之前left_bars个bar都低
            is_low = True
            for j in range(1, left_bars + 1):
                if df['low'].iloc[i] >= df['low'].iloc[i - j]:
                    is_low = False
                    break
            if is_low:
                lows.append(i)

        return highs, lows

    def _calculate_wave(self, pivots: List[int], df: DataFrame, is_up: bool) -> Optional[Wave]:
        """
        计算单个波浪的属性
        """
        if len(pivots) < 2:
            return None

        start_idx = pivots[0]
        end_idx = pivots[-1]
        start_price = df['low'].iloc[start_idx] if is_up else df['high'].iloc[start_idx]
        end_price = df['high'].iloc[end_idx] if is_up else df['low'].iloc[end_idx]

        wave_height = abs(end_price - start_price)
        wave_bars = end_idx - start_idx

        if wave_bars < self.min_wave_bars or wave_bars > self.max_wave_bars:
            return None

        return Wave(
            start_idx=start_idx,
            end_idx=end_idx,
            start_price=start_price,
            end_price=end_price,
            wave_num=0,  # 稍后分配
            wave_type=WaveType.UNKNOWN
        )

    def _identify_elliott_waves(self, df: DataFrame) -> List[Wave]:
        """
        识别艾略特波浪结构
        简化版：基于枢轴点识别5浪上涨或3浪下跌
        """
        highs, lows = self._detect_pivots(df, self.wave_pivot_lookback)

        waves = []
        all_pivots = sorted(highs + lows)

        if len(all_pivots) < 4:
            return waves

        # 简化波浪识别：基于价格走势判断当前处于哪个浪
        recent_pivots = all_pivots[-8:]  # 取最近8个枢轴点

        for i in range(0, len(recent_pivots) - 1, 2):
            if i + 1 >= len(recent_pivots):
                break

            start_idx = recent_pivots[i]
            end_idx = recent_pivots[i + 1]

            # 确定是上涨还是下跌浪
            price_change = df['close'].iloc[end_idx] - df['close'].iloc[start_idx]
            is_up = price_change > 0

            wave = Wave(
                start_idx=start_idx,
                end_idx=end_idx,
                start_price=df['close'].iloc[start_idx],
                end_price=df['close'].iloc[end_idx],
                wave_num=(i // 2) + 1,
                wave_type=WaveType.IMPULSE if is_up else WaveType.CORRECTIVE,
                is_complete=True
            )
            waves.append(wave)

        return waves

    def _get_current_wave_position(self, df: DataFrame, waves: List[Wave]) -> WavePosition:
        """
        判断当前处于波浪的哪个位置
        """
        if not waves:
            return WavePosition.UNKNOWN

        last_wave = waves[-1]
        current_price = df['close'].iloc[-1]

        # 基于最后一浪的特征判断位置
        if last_wave.wave_num == 1:
            return WavePosition.WAVE_2 if current_price < last_wave.end_price else WavePosition.WAVE_1
        elif last_wave.wave_num == 2:
            return WavePosition.WAVE_3 if current_price > last_wave.end_price else WavePosition.WAVE_2
        elif last_wave.wave_num == 3:
            return WavePosition.WAVE_4 if current_price < last_wave.end_price else WavePosition.WAVE_3
        elif last_wave.wave_num == 4:
            return WavePosition.WAVE_5 if current_price > last_wave.end_price else WavePosition.WAVE_4
        elif last_wave.wave_num >= 5:
            return WavePosition.WAVE_A

        return WavePosition.UNKNOWN

    def _calculate_fibonacci_levels(self, wave: Wave) -> dict:
        """
        计算斐波那契回撤和扩展位
        """
        if wave is None:
            return {}

        price_range = abs(wave.end_price - wave.start_price)
        is_up = wave.end_price > wave.start_price

        if is_up:
            levels = {
                '0.0': wave.end_price,
                '0.236': wave.end_price - price_range * 0.236,
                '0.382': wave.end_price - price_range * 0.382,
                '0.5': wave.end_price - price_range * 0.5,
                '0.618': wave.end_price - price_range * 0.618,
                '0.786': wave.end_price - price_range * 0.786,
                '1.0': wave.start_price,
                '1.272': wave.end_price + price_range * 0.272,
                '1.618': wave.end_price + price_range * 0.618,
            }
        else:
            levels = {
                '0.0': wave.end_price,
                '0.236': wave.end_price + price_range * 0.236,
                '0.382': wave.end_price + price_range * 0.382,
                '0.5': wave.end_price + price_range * 0.5,
                '0.618': wave.end_price + price_range * 0.618,
                '0.786': wave.end_price + price_range * 0.786,
                '1.0': wave.start_price,
                '1.272': wave.end_price - price_range * 0.272,
                '1.618': wave.end_price - price_range * 0.618,
            }

        return levels

    def _get_dataframe(self, timeframe: str) -> DataFrame:
        return self.dp.get_pair_dataframe(pair=self.pair, timeframe=timeframe)

    def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        """计算所有指标"""
        dataframe = self._add_indicators(dataframe)

        # 获取多时间框架数据
        df_1h = self._get_dataframe('1h')
        df_4h = self._get_dataframe('4h')
        df_1d = self._get_dataframe('1d')

        # 1小时指标
        if len(df_1h) > 50:
            df_1h = self._add_indicators(df_1h)
            dataframe['1h_ema200'] = df_1h['ema200'].reindex(dataframe.index, method='ffill')
            dataframe['1h_trend'] = np.where(dataframe['close'] > dataframe['1h_ema200'], 1, -1)
            dataframe['1h_rsi'] = df_1h['rsi'].reindex(dataframe.index, method='ffill')
            dataframe['1h_atr'] = df_1h['atr'].reindex(dataframe.index, method='ffill')

            # 1h波浪识别
            waves_1h = self._identify_elliott_waves(df_1h)
            dataframe['1h_wave_position'] = self._get_current_wave_position(df_1h, waves_1h).value
        else:
            dataframe['1h_trend'] = 0
            dataframe['1h_rsi'] = 50
            dataframe['1h_wave_position'] = 0

        # 4小时指标
        if len(df_4h) > 50:
            df_4h = self._add_indicators(df_4h)
            dataframe['4h_ema200'] = df_4h['ema200'].reindex(dataframe.index, method='ffill')
            dataframe['4h_trend'] = np.where(dataframe['close'] > dataframe['4h_ema200'], 1, -1)
        else:
            dataframe['4h_trend'] = 0

        # 日线指标
        if len(df_1d) > 50:
            df_1d = self._add_indicators(df_1d)
            dataframe['1d_ema200'] = df_1d['ema200'].reindex(dataframe.index, method='ffill')
            dataframe['1d_trend'] = np.where(dataframe['close'] > dataframe['1d_ema200'], 1, -1)
            dataframe['1d_rsi'] = df_1d['rsi'].reindex(dataframe.index, method='ffill')

            # 日线波浪 (大趋势)
            waves_1d = self._identify_elliott_waves(df_1d)
            dataframe['1d_wave_position'] = self._get_current_wave_position(df_1d, waves_1d).value
        else:
            dataframe['1d_trend'] = 0
            dataframe['1d_rsi'] = 50
            dataframe['1d_wave_position'] = 0

        # 当前时间框架波浪识别
        waves = self._identify_elliott_waves(dataframe)
        dataframe['wave_position'] = self._get_current_wave_position(dataframe, waves).value

        # 缠论分型
        dataframe['chan_state'] = self.chan.get_klu_state(dataframe)

        # 生成交易信号
        dataframe = self._generate_signals(dataframe, waves)

        return dataframe

    def _generate_signals(self, df: DataFrame, waves: List[Wave]) -> DataFrame:
        """
        基于艾略特波浪理论生成交易信号
        """
        n = len(df)
        if n < 50:
            return df

        # 获取当前波浪位置
        current_wave = self._get_current_wave_position(df, waves)

        # 延迟分型 (避免未来数据)
        df['_fx'] = df['chan_state'].shift(1).fillna('00')

        # 趋势方向
        trend_up = df['1h_trend'] > 0
        trend_down = df['1h_trend'] < 0
        trend_aligned_daily = df['1d_trend'] == df['1h_trend']

        # RSI条件
        rsi_oversold = df['rsi'] < self.rsi_oversold
        rsi_overbought = df['rsi'] > self.rsi_overbought
        rsi_divergence_long = (df['rsi'] > df['rsi'].shift(5)) & (df['close'] < df['close'].shift(5))
        rsi_divergence_short = (df['rsi'] < df['rsi'].shift(5)) & (df['close'] > df['close'].shift(5))

        # 波动率过滤
        low_volatility = df['atr_percent'] < df['atr_percent'].rolling(50).mean() * 0.8
        high_volatility = df['atr_percent'] > df['atr_percent'].rolling(50).mean() * 1.5

        # ========== 多头信号 ==========
        long_conditions = []

        # 浪2回调做多 (最佳入场点)
        # 条件: 浪2位置 + 底分型 + RSI超卖 + 趋势向上
        long_cond_wave2 = (
            (df['wave_position'] == WavePosition.WAVE_2.value) |
            (df['1h_wave_position'] == WavePosition.WAVE_2.value)
        ) & (
            (df['_fx'] == '-10') |
            ((df['close'] > df['ema20']) & (df['ema20'] > df['ema50']))
        ) & rsi_oversold & trend_up

        long_conditions.append(('wave2', long_cond_wave2))

        # 浪4回调做多 (谨慎入场)
        long_cond_wave4 = (
            (df['wave_position'] == WavePosition.WAVE_4.value) |
            (df['1h_wave_position'] == WavePosition.WAVE_4.value)
        ) & (df['_fx'] == '-10') & rsi_oversold & trend_up & (
            df['rsi_divergence_long'] if 'rsi_divergence_long' in df.columns else True
        )

        long_conditions.append(('wave4', long_cond_wave4))

        # C浪结束做多 (趋势反转)
        long_cond_wave_c = (
            (df['wave_position'] == WavePosition.WAVE_C.value) |
            (df['1h_wave_position'] == WavePosition.WAVE_C.value)
        ) & (df['_fx'] == '-10') & rsi_oversold & (
            df['volume_ratio'] > 1.5  # 放量确认
        )

        long_conditions.append(('wave_c', long_cond_wave_c))

        # 强势突破做多
        long_cond_breakout = (
            (df['close'] > df['bb_upper']) &
            (df['volume_ratio'] > 2.0) &
            trend_up &
            (df['macdhist'] > 0) &
            (df['1h_wave_position'].isin([WavePosition.WAVE_3.value, WavePosition.WAVE_5.value]))
        )

        long_conditions.append(('breakout', long_cond_breakout))

        # 合并多头信号
        df['elliott_long'] = False
        for name, cond in long_conditions:
            df[f'long_{name}'] = cond & ~low_volatility  # 避免低波动时入场
            df['elliott_long'] |= df[f'long_{name}']

        # ========== 空头信号 ==========
        short_conditions = []

        # 浪2回调做空 (下跌趋势)
        short_cond_wave2 = (
            (df['wave_position'] == WavePosition.WAVE_2.value) |
            (df['1h_wave_position'] == WavePosition.WAVE_2.value)
        ) & (
            (df['_fx'] == '10') |
            ((df['close'] < df['ema20']) & (df['ema20'] < df['ema50']))
        ) & rsi_overbought & trend_down

        short_conditions.append(('wave2', short_cond_wave2))

        # 浪4回调做空 (谨慎)
        short_cond_wave4 = (
            (df['wave_position'] == WavePosition.WAVE_4.value) |
            (df['1h_wave_position'] == WavePosition.WAVE_4.value)
        ) & (df['_fx'] == '10') & rsi_overbought & trend_down

        short_conditions.append(('wave4', short_cond_wave4))

        # 浪5结束做空 (趋势反转)
        short_cond_wave5 = (
            (df['wave_position'] == WavePosition.WAVE_5.value) |
            (df['1h_wave_position'] == WavePosition.WAVE_5.value)
        ) & (df['_fx'] == '10') & rsi_overbought & (
            df['volume_ratio'] > 1.5
        )

        short_conditions.append(('wave5', short_cond_wave5))

        # B浪反弹做空 (继续下跌)
        short_cond_wave_b = (
            (df['wave_position'] == WavePosition.WAVE_B.value) |
            (df['1h_wave_position'] == WavePosition.WAVE_B.value)
        ) & (df['_fx'] == '10') & rsi_overbought & trend_down

        short_conditions.append(('wave_b', short_cond_wave_b))

        # 强势跌破做空
        short_cond_breakdown = (
            (df['close'] < df['bb_lower']) &
            (df['volume_ratio'] > 2.0) &
            trend_down &
            (df['macdhist'] < 0) &
            (df['1h_wave_position'].isin([WavePosition.WAVE_3.value, WavePosition.WAVE_C.value]))
        )

        short_conditions.append(('breakdown', short_cond_breakdown))

        # 合并空头信号
        df['elliott_short'] = False
        for name, cond in short_conditions:
            df[f'short_{name}'] = cond & ~low_volatility
            df['elliott_short'] |= df[f'short_{name}']

        # 强趋势过滤
        if self.trend_filter_strict:
            df['elliott_long'] &= trend_up | (df['1d_trend'] > 0)
            df['elliott_short'] &= trend_down | (df['1d_trend'] < 0)

        # 避免高波动时期入场
        df['elliott_long'] &= ~high_volatility
        df['elliott_short'] &= ~high_volatility

        return df

    def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        """入场信号"""
        dataframe['enter_long'] = 0
        dataframe['enter_short'] = 0
        dataframe['enter_tag'] = ''

        if 'elliott_long' not in dataframe.columns:
            return dataframe

        # 多头入场
        long_mask = dataframe['elliott_long'].fillna(False)
        dataframe.loc[long_mask, 'enter_long'] = 1

        # 标记入场类型
        for col in dataframe.columns:
            if col.startswith('long_') and col != 'elliott_long':
                mask = dataframe[col].fillna(False) & (dataframe['enter_long'] == 1)
                dataframe.loc[mask, 'enter_tag'] = col.replace('long_', 'elliott_')

        # 空头入场
        short_mask = dataframe['elliott_short'].fillna(False)
        dataframe.loc[short_mask, 'enter_short'] = 1

        for col in dataframe.columns:
            if col.startswith('short_') and col != 'elliott_short':
                mask = dataframe[col].fillna(False) & (dataframe['enter_short'] == 1)
                dataframe.loc[mask, 'enter_tag'] = col.replace('short_', 'elliott_')

        return dataframe

    def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        """出场信号"""
        dataframe['exit_long'] = 0
        dataframe['exit_short'] = 0

        if len(dataframe) < 2:
            return dataframe

        df = dataframe.copy()
        df['_fx'] = df['chan_state'].shift(1).fillna('00')

        # 多头出场条件
        exit_long_cond = (
            # 顶分型出场
            (df['_fx'] == '10') |
            # RSI超买
            (df['rsi'] > 75) |
            # 跌破EMA20
            (df['close'] < df['ema20']) & (df['close'].shift(1) > df['ema20'].shift(1)) |
            # MACD死叉
            (df['macd'] < df['macdsignal']) & (df['macd'].shift(1) > df['macdsignal'].shift(1))
        )

        # 波浪位置出场
        exit_long_wave = df['wave_position'].isin([
            WavePosition.WAVE_5.value,
            WavePosition.WAVE_C.value
        ])

        dataframe['exit_long'] = (exit_long_cond | exit_long_wave).astype(int)

        # 空头出场条件
        exit_short_cond = (
            # 底分型出场
            (df['_fx'] == '-10') |
            # RSI超卖
            (df['rsi'] < 25) |
            # 突破EMA20
            (df['close'] > df['ema20']) & (df['close'].shift(1) < df['ema20'].shift(1)) |
            # MACD金叉
            (df['macd'] > df['macdsignal']) & (df['macd'].shift(1) < df['macdsignal'].shift(1))
        )

        # 波浪位置出场
        exit_short_wave = df['wave_position'].isin([
            WavePosition.WAVE_C.value,
            WavePosition.WAVE_5.value
        ])

        dataframe['exit_short'] = (exit_short_cond | exit_short_wave).astype(int)

        return dataframe

    def custom_stoploss(self, pair: str, trade: 'Trade', current_time: datetime,
                        current_rate: float, current_profit: float, **kwargs) -> float:
        """
        动态止损：基于ATR和波浪位置调整
        """
        dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)
        if len(dataframe) < 2:
            return self.stoploss

        current_candle = dataframe.iloc[-1]
        atr = current_candle['atr_percent']

        # 基于ATR的动态止损
        dynamic_stop = -atr * self.atr_multiplier_stop / 100

        # 根据盈利情况收紧止损
        if current_profit > 0.03:  # 盈利3%后收紧止损
            return max(dynamic_stop, -0.01)
        elif current_profit > 0.05:  # 盈利5%后更紧
            return max(dynamic_stop, -0.005)

        return max(dynamic_stop, self.stoploss)

    def leverage(self, pair: str, current_time: datetime, current_rate: float,
                 proposed_leverage: float, max_leverage: float, entry_tag: Optional[str], side: str,
                 **kwargs) -> float:
        """
        动态杠杆：根据波动率调整
        """
        dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)
        if len(dataframe) < 2:
            return 2.0

        current_candle = dataframe.iloc[-1]
        volatility = current_candle['atr_percent']

        # 低波动时提高杠杆，高波动时降低杠杆
        if volatility < 0.5:
            return 3.0
        elif volatility < 1.0:
            return 2.0
        elif volatility < 2.0:
            return 1.5
        else:
            return 1.0

    def adjust_trade_position(self, trade: 'Trade', current_time: datetime,
                              current_rate: float, current_profit: float,
                              min_stake: Optional[float], max_stake: float,
                              current_entry_rate: float, current_exit_rate: float,
                              current_entry_profit: float, current_exit_profit: float,
                              **kwargs) -> Optional[float]:
        """
        仓位调整：金字塔加仓
        """
        if current_profit < -0.01:  # 亏损时不加仓
            return None

        if current_profit > 0.02 and current_profit < 0.03:  # 盈利2-3%时加仓
            return min_stake * 0.5 if min_stake else None

        return None