""" 缠论同级别分解策略 (Chan Same-Level Decomposition Strategy) 核心思想：按同级别分解操作，实现a+A结构的机械化操作 以5分钟级别为例： 1. a+A结构：a是5分钟走势类型（定义为A0），A分解为m段5分钟走势类型：A=A1+A2+...+Am 2. 如果a+A向上，则Ai当i为奇数时向下，i为偶数时向上 3. 中枢形成： - A1不能跌破a的低点 - 如果A2升破a的高点而A3不跌回a的高点，可以把a+A1+A2+A3当成一个新的a'（还是5分钟级别） - 如果A3跌破a的高点，则A1、A2、A3必然构成30分钟中枢 操作程式（机械化操作）： 1. 盘整背驰情况： - Ai与Ai+2之间比较力度（盘整背驰） - i+2为偶数时卖出 - i+2为奇数时买入 2. 非背驰情况： - 当i为偶数，若Ai+3不跌破Ai高点，则继续持有到Ai+k+3跌破Ai+k高点后在不创新高或盘整顶背驰的Ai+k+4卖出，其中k为偶数 - 当i为奇数，若Ai+3不升破Ai低点，则继续保持不回补直到Ai+k+3升破Ai+k低点后在不创新低或盘整底背驰的Ai+k+4回补 使用命令： freqtrade backtesting -c ./user_data/Chan/config/Local_Test.json \ --strategy ChanSameLevelStrategy --strategy-path ./user_data/Chan/strategies \ --timerange=20250301- """ import logging from datetime import datetime from typing import Optional import numpy as np import pandas as pd import talib.abstract as ta from pandas import DataFrame from technical.util import resample_to_interval, resampled_merge from freqtrade.strategy import IStrategy logger = logging.getLogger(__name__) class ChanSameLevelStrategy(IStrategy): INTERFACE_VERSION: int = 3 # === 基础配置 === # 底层使用 1m K线，resample 到 30m 进行同级别分解 can_short = True startup_candle_count: int = 2000 # 需要足够的数据来识别走势段 # 止损和止盈（优化：改善风险回报比） stoploss = -0.015 # 1.5% 硬止损（更紧，减少单笔亏损） use_custom_stoploss = False # Trailing stop（优化：更激进的保护利润） trailing_stop = True trailing_stop_positive = 0.006 # 回撤 0.6% 触发退出（更紧） trailing_stop_positive_offset = 0.012 # 盈利 1.2% 后才开始追踪（降低门槛） trailing_only_offset_is_reached = True # ROI（优化：更合理的止盈目标，改善风险回报比） minimal_roi = { "0": 0.03, # 3% 立即止盈（降低目标，提高胜率） "60": 0.02, # 60分钟后 2% "120": 0.015, # 120分钟后 1.5% "240": 0.01, # 240分钟后 1% "480": 0.005, # 480分钟后 0.5% "720": 0, # 720分钟后不设止盈 } order_types = { "entry": "market", "exit": "market", "stoploss": "market", "stoploss_on_exchange": False, } # 同级别分解的级别（5分钟） same_level_timeframe = 5 def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """计算指标并识别同级别走势段""" ticker = self.get_ticker_indicator() # Resample 到 30m 进行同级别分解 dataframe_30m = resample_to_interval(dataframe, ticker * self.same_level_timeframe) # 在 30m 上计算指标 dataframe_30m = self.add_indicators_30m(dataframe_30m) # 识别同级别走势段和背驰 dataframe_30m = self.identify_same_level_segments(dataframe_30m) # 合并回 1m dataframe dataframe = resampled_merge(dataframe, dataframe_30m) # 在 1m 上也计算基础指标 dataframe = self.add_indicators_1m(dataframe) return dataframe def add_indicators_30m(self, dataframe: DataFrame) -> DataFrame: """在30m级别计算指标""" # MACD 用于识别背驰 macd = ta.MACD(dataframe, fastperiod=12, slowperiod=26, signalperiod=9) dataframe['macd'] = macd['macd'] dataframe['macdsignal'] = macd['macdsignal'] dataframe['macdhist'] = macd['macdhist'] # EMA 用于识别趋势（增加更多EMA用于趋势确认） dataframe['ema12'] = ta.EMA(dataframe, timeperiod=12) dataframe['ema26'] = ta.EMA(dataframe, timeperiod=26) dataframe['ema50'] = ta.EMA(dataframe, timeperiod=50) dataframe['ema200'] = ta.EMA(dataframe, timeperiod=200) # 长期趋势 # EMA趋势方向 dataframe['ema_trend_up'] = (dataframe['ema12'] > dataframe['ema26']) & (dataframe['ema26'] > dataframe['ema50']) dataframe['ema_trend_dn'] = (dataframe['ema12'] < dataframe['ema26']) & (dataframe['ema26'] < dataframe['ema50']) # 市场整体趋势（基于价格和EMA200） dataframe['price_above_ema200'] = dataframe['close'] > dataframe['ema200'] dataframe['price_below_ema200'] = dataframe['close'] < dataframe['ema200'] # 趋势强度（EMA斜率） dataframe['ema12_slope'] = dataframe['ema12'].diff(5) / dataframe['ema12'].shift(5) dataframe['ema26_slope'] = dataframe['ema26'].diff(5) / dataframe['ema26'].shift(5) dataframe['strong_uptrend'] = (dataframe['ema12_slope'] > 0) & (dataframe['ema26_slope'] > 0) & (dataframe['price_above_ema200']) dataframe['strong_downtrend'] = (dataframe['ema12_slope'] < 0) & (dataframe['ema26_slope'] < 0) & (dataframe['price_below_ema200']) # RSI 用于确认 dataframe['rsi'] = ta.RSI(dataframe, timeperiod=14) # ATR 用于波动率过滤 dataframe['atr'] = ta.ATR(dataframe, timeperiod=14) dataframe['atr_mean'] = dataframe['atr'].rolling(window=20).mean() # 波动率过滤：只在波动率足够时交易 dataframe['volatility_ok'] = dataframe['atr'] > dataframe['atr_mean'] * 0.8 return dataframe def add_indicators_1m(self, dataframe: DataFrame) -> DataFrame: """在1m级别计算基础指标""" dataframe['rsi_1m'] = ta.RSI(dataframe, timeperiod=14) dataframe['volume_mean'] = dataframe['volume'].rolling(window=20).mean() # MACD 用于1m级别确认 macd_1m = ta.MACD(dataframe, fastperiod=12, slowperiod=26, signalperiod=9) dataframe['macd_1m'] = macd_1m['macd'] dataframe['macdsignal_1m'] = macd_1m['macdsignal'] dataframe['macdhist_1m'] = macd_1m['macdhist'] # MACD交叉确认 dataframe['macd_cross_up_1m'] = ( (dataframe['macd_1m'] > dataframe['macdsignal_1m']) & (dataframe['macd_1m'].shift(1) <= dataframe['macdsignal_1m'].shift(1)) ) dataframe['macd_cross_dn_1m'] = ( (dataframe['macd_1m'] < dataframe['macdsignal_1m']) & (dataframe['macd_1m'].shift(1) >= dataframe['macdsignal_1m'].shift(1)) ) return dataframe def identify_same_level_segments(self, dataframe: DataFrame) -> DataFrame: """ 识别同级别走势段（a+A结构） 实现5分钟级别的同级别分解： 1. 识别A0（即a）、A1、A2、A3等走势段 2. 判断每个段的类型（上涨/下跌）：如果a+A向上，则Ai当i为奇数时向下，i为偶数时向上 3. 识别中枢形成条件 4. 计算盘整背驰（Ai与Ai+2比较力度） 5. 标记买卖点 """ df = dataframe.copy() # 初始化列 df['ai_index'] = -1 # Ai的索引（A0, A1, A2, ...） df['ai_type'] = 0 # 1: 上涨, -1: 下跌 df['ai_high'] = np.nan # Ai的高点 df['ai_low'] = np.nan # Ai的低点 df['ai_macd_max'] = np.nan # Ai的MACD最大值 df['ai_macd_min'] = np.nan # Ai的MACD最小值 df['zs_formed'] = False # 是否形成中枢 df['panzheng_beichi'] = False # 盘整背驰信号 df['buy_signal'] = False # 买入信号 df['sell_signal'] = False # 卖出信号 # 识别关键转折点（局部高点和低点） window = 3 # 确认窗口 lookback = window + 1 # 高点识别（延迟确认） df['temp_high'] = df['high'].shift(window) df['is_pivot_high'] = ( (df['temp_high'] == df['temp_high'].rolling(window=lookback).max()) & (df['temp_high'].notna()) ) # 低点识别（延迟确认） df['temp_low'] = df['low'].shift(window) df['is_pivot_low'] = ( (df['temp_low'] == df['temp_low'].rolling(window=lookback).min()) & (df['temp_low'].notna()) ) # 逐行处理，识别走势段 ai_list = [] # 存储Ai段的信息：[(start_idx, end_idx, type, high, low, macd_max, macd_min), ...] current_ai_start = None current_ai_type = None # 1: 上涨, -1: 下跌 last_pivot_idx = None last_pivot_type = None # 'high' or 'low' for i in range(window, len(df)): # 检查是否有新的转折点 is_new_pivot = False pivot_type = None if df.iloc[i]['is_pivot_high']: is_new_pivot = True pivot_type = 'high' elif df.iloc[i]['is_pivot_low']: is_new_pivot = True pivot_type = 'low' if is_new_pivot and last_pivot_idx is not None: # 完成一个走势段 if current_ai_start is not None: seg_df = df.iloc[current_ai_start:i] if len(seg_df) >= 3: # 至少3根K线 # 使用已确认的数据计算（不包括当前转折点） # 为了安全，只使用到 last_pivot_idx 之前的数据 confirmed_seg_df = df.iloc[current_ai_start:last_pivot_idx] if last_pivot_idx > current_ai_start else seg_df if len(confirmed_seg_df) > 0: high_val = confirmed_seg_df['high'].max() low_val = confirmed_seg_df['low'].min() macd_max = confirmed_seg_df['macd'].max() macd_min = confirmed_seg_df['macd'].min() else: high_val = seg_df['high'].max() low_val = seg_df['low'].min() macd_max = seg_df['macd'].max() macd_min = seg_df['macd'].min() # 判断走势类型 if current_ai_type is None: # 第一个段（A0），根据价格变化判断 if high_val > df.iloc[current_ai_start]['close']: current_ai_type = 1 # 上涨 else: current_ai_type = -1 # 下跌 else: # 后续段：如果a+A向上，则Ai当i为奇数时向下，i为偶数时向上 # 简化处理：根据转折点类型判断 if pivot_type == 'high' and last_pivot_type == 'low': current_ai_type = 1 # 上涨段 elif pivot_type == 'low' and last_pivot_type == 'high': current_ai_type = -1 # 下跌段 ai_list.append({ 'start': current_ai_start, 'end': i, 'type': current_ai_type, 'high': high_val, 'low': low_val, 'macd_max': macd_max, 'macd_min': macd_min }) # 标记到dataframe（只在段结束时标记，避免未来数据） # 使用滚动窗口：只在确认转折点后才标记前一段的信息 # 为了安全，只在段的最后几根K线标记（确认段已结束） confirm_window = min(3, i - current_ai_start) # 确认窗口，最多3根K线 mark_start = max(current_ai_start, i - confirm_window) df.iloc[mark_start:i, df.columns.get_loc('ai_index')] = len(ai_list) - 1 df.iloc[mark_start:i, df.columns.get_loc('ai_type')] = current_ai_type # 高点和低点使用已确认的数据 df.iloc[mark_start:i, df.columns.get_loc('ai_high')] = high_val df.iloc[mark_start:i, df.columns.get_loc('ai_low')] = low_val df.iloc[mark_start:i, df.columns.get_loc('ai_macd_max')] = macd_max df.iloc[mark_start:i, df.columns.get_loc('ai_macd_min')] = macd_min # 开始新的走势段 current_ai_start = last_pivot_idx last_pivot_idx = i last_pivot_type = pivot_type elif is_new_pivot: # 第一个转折点 last_pivot_idx = i last_pivot_type = pivot_type if current_ai_start is None: current_ai_start = 0 # 处理最后一个段 if current_ai_start is not None: # 标记当前未完成的段 if len(df) - current_ai_start >= 3: seg_df = df.iloc[current_ai_start:] high_val = seg_df['high'].max() low_val = seg_df['low'].min() macd_max = seg_df['macd'].max() macd_min = seg_df['macd'].min() # 使用最后一个段的类型 if len(ai_list) > 0: last_type = ai_list[-1]['type'] # 如果上一个段是上涨，当前应该是下跌（或相反） current_ai_type = -last_type else: current_ai_type = 1 if high_val > df.iloc[current_ai_start]['close'] else -1 ai_list.append({ 'start': current_ai_start, 'end': len(df), 'type': current_ai_type, 'high': high_val, 'low': low_val, 'macd_max': macd_max, 'macd_min': macd_min }) df.iloc[current_ai_start:, df.columns.get_loc('ai_index')] = len(ai_list) - 1 df.iloc[current_ai_start:, df.columns.get_loc('ai_type')] = current_ai_type df.iloc[current_ai_start:, df.columns.get_loc('ai_high')] = high_val df.iloc[current_ai_start:, df.columns.get_loc('ai_low')] = low_val df.iloc[current_ai_start:, df.columns.get_loc('ai_macd_max')] = macd_max df.iloc[current_ai_start:, df.columns.get_loc('ai_macd_min')] = macd_min # 识别中枢和盘整背驰 df = self.identify_zs_and_beichi(df, ai_list) # 清理临时列 df = df.drop(columns=['temp_high', 'temp_low', 'is_pivot_high', 'is_pivot_low']) return df def identify_zs_and_beichi(self, dataframe: DataFrame, ai_list: list) -> DataFrame: """ 识别中枢和盘整背驰 1. 中枢形成：如果A3跌破a的高点，则A1、A2、A3必然构成30分钟中枢 2. 盘整背驰：Ai与Ai+2之间比较力度（MACD面积或幅度） 3. 标记买卖点： - 盘整背驰：i+2为偶数时卖出，i+2为奇数时买入 - 非背驰情况：根据Ai+3是否跌破/升破Ai的高低点决定 注意：为了避免未来数据，只在段确认结束后才标记信号 """ df = dataframe.copy() if len(ai_list) < 3: return df # 逐行处理，只在当前行可以确认历史段的信息时才标记 # 这样可以避免使用未来数据 for row_idx in range(len(df)): # 找到当前行属于哪个段 current_ai_idx = -1 for ai_idx, ai in enumerate(ai_list): if ai['start'] <= row_idx < ai['end']: current_ai_idx = ai_idx break if current_ai_idx < 0: continue # 只在段的最后几根K线才处理，确保段已确认结束 current_ai = ai_list[current_ai_idx] if row_idx < current_ai['end'] - 3: # 只在段的最后3根K线处理 continue # 识别中枢（A1、A2、A3构成中枢） # 只在A3段结束时才标记中枢，避免使用未来数据 if current_ai_idx >= 2: # 至少需要A0, A1, A2 a0 = ai_list[0] a1 = ai_list[current_ai_idx - 2] if current_ai_idx >= 2 else None a2 = ai_list[current_ai_idx - 1] if current_ai_idx >= 1 else None a3 = ai_list[current_ai_idx] if a1 and a2 and a3: # 如果A3跌破a（A0）的高点，则A1、A2、A3构成中枢 if a3['low'] < a0['high']: # 只在A3段的最后几根K线标记中枢 df.iloc[row_idx, df.columns.get_loc('zs_formed')] = True # 盘整背驰判断：Ai与Ai+2比较力度 # 只在ai_plus_2段结束时才判断，避免使用未来数据 if current_ai_idx >= 2: ai = ai_list[current_ai_idx - 2] ai_plus_2 = ai_list[current_ai_idx] # 计算力度（使用MACD面积或价格幅度） if ai['type'] == ai_plus_2['type']: # 同方向才能比较 # 上涨段：比较MACD最大值和价格涨幅 if ai['type'] == 1: # 上涨 price_strength_ai = (ai['high'] - ai['low']) / ai['low'] if ai['low'] > 0 else 0 price_strength_ai2 = (ai_plus_2['high'] - ai_plus_2['low']) / ai_plus_2['low'] if ai_plus_2['low'] > 0 else 0 macd_strength_ai = ai['macd_max'] macd_strength_ai2 = ai_plus_2['macd_max'] # 盘整顶背驰：价格创新高或接近，但MACD力度减弱 beichi = ( (price_strength_ai2 <= price_strength_ai * 1.1) & # 价格涨幅相近或更小 (macd_strength_ai2 < macd_strength_ai * 0.9) # MACD力度明显减弱 ) else: # 下跌 price_strength_ai = (ai['high'] - ai['low']) / ai['low'] if ai['low'] > 0 else 0 price_strength_ai2 = (ai_plus_2['high'] - ai_plus_2['low']) / ai_plus_2['low'] if ai_plus_2['low'] > 0 else 0 macd_strength_ai = abs(ai['macd_min']) macd_strength_ai2 = abs(ai_plus_2['macd_min']) # 盘整底背驰：价格创新低或接近，但MACD力度减弱 beichi = ( (abs(price_strength_ai2) <= abs(price_strength_ai) * 1.1) & # 价格跌幅相近或更小 (macd_strength_ai2 < macd_strength_ai * 0.9) # MACD力度明显减弱 ) if beichi: # 只在ai_plus_2段的最后几根K线标记信号 # i+2为偶数时卖出，i+2为奇数时买入 if (current_ai_idx) % 2 == 0: # 偶数，卖出 df.iloc[row_idx, df.columns.get_loc('sell_signal')] = True df.iloc[row_idx, df.columns.get_loc('panzheng_beichi')] = True else: # 奇数，买入 df.iloc[row_idx, df.columns.get_loc('buy_signal')] = True df.iloc[row_idx, df.columns.get_loc('panzheng_beichi')] = True # 非背驰情况的处理（简化版） # 只在Ai+4段结束时才标记，避免使用未来数据 if current_ai_idx >= 4: ai = ai_list[current_ai_idx - 4] ai_plus_3 = ai_list[current_ai_idx - 1] ai_plus_4 = ai_list[current_ai_idx] if (current_ai_idx - 4) % 2 == 0: # i为偶数 # 若Ai+3不跌破Ai高点，继续持有（不标记卖出） if ai_plus_3['low'] < ai['high']: # Ai+3跌破Ai高点，在不创新高或盘整顶背驰的Ai+k+4卖出 if ai_plus_4['high'] <= ai_plus_3['high']: # 不创新高 df.iloc[row_idx, df.columns.get_loc('sell_signal')] = True else: # i为奇数 # 若Ai+3不升破Ai低点，继续保持不回补 if ai_plus_3['high'] > ai['low']: # Ai+3升破Ai低点，在不创新低或盘整底背驰的Ai+k+4回补 if ai_plus_4['low'] >= ai_plus_3['low']: # 不创新低 df.iloc[row_idx, df.columns.get_loc('buy_signal')] = True return df def detect_divergence(self, dataframe: DataFrame) -> DataFrame: """ 检测背驰（使用滚动窗口，避免未来函数） 顶背驰：价格创新高，但MACD不创新高 底背驰：价格创新低，但MACD不创新低 """ df = dataframe.copy() # 使用滚动窗口检测背驰（只使用历史数据） lookback = 20 # 向前看20根K线 # 顶背驰检测：当前价格是近期最高，但MACD不是近期最高 df['recent_high'] = df['high'].rolling(window=lookback).max() df['recent_macd_max'] = df['macd'].rolling(window=lookback).max() df['prev_recent_high'] = df['high'].rolling(window=lookback).max().shift(1) df['prev_recent_macd_max'] = df['macd'].rolling(window=lookback).max().shift(1) # 当前价格创新高，但MACD没有创新高（或降低） df['divergence_top'] = ( (df['high'] >= df['recent_high']) & # 当前是近期最高 (df['high'] > df['prev_recent_high']) & # 比之前的最高更高 (df['macd'] < df['prev_recent_macd_max']) & # MACD没有创新高 (df['macd'] < 0) # MACD在零轴下方（下跌趋势中的顶背驰） ) # 底背驰检测：当前价格是近期最低，但MACD不是近期最低 df['recent_low'] = df['low'].rolling(window=lookback).min() df['recent_macd_min'] = df['macd'].rolling(window=lookback).min() df['prev_recent_low'] = df['low'].rolling(window=lookback).min().shift(1) df['prev_recent_macd_min'] = df['macd'].rolling(window=lookback).min().shift(1) # 当前价格创新低，但MACD没有创新低（或升高） df['divergence_bottom'] = ( (df['low'] <= df['recent_low']) & # 当前是近期最低 (df['low'] < df['prev_recent_low']) & # 比之前的最低更低 (df['macd'] > df['prev_recent_macd_min']) & # MACD没有创新低 (df['macd'] > 0) # MACD在零轴上方（上涨趋势中的底背驰） ) return df def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """ 入场逻辑：基于同级别分解的a+A结构 1. 盘整背驰买入：i+2为奇数时的盘整背驰信号 2. 非背驰情况的买入：Ai+3升破Ai低点后的回补信号 """ ticker = self.get_ticker_indicator() resample_col = f"resample_{ticker * self.same_level_timeframe}_" # 获取5m级别的指标 buy_signal_col = f"{resample_col}buy_signal" panzheng_beichi_col = f"{resample_col}panzheng_beichi" ai_type_col = f"{resample_col}ai_type" rsi_5m_col = f"{resample_col}rsi" # 获取30m级别的趋势指标 ema_trend_up_col = f"{resample_col}ema_trend_up" ema_trend_dn_col = f"{resample_col}ema_trend_dn" volatility_ok_col = f"{resample_col}volatility_ok" strong_uptrend_col = f"{resample_col}strong_uptrend" strong_downtrend_col = f"{resample_col}strong_downtrend" price_above_ema200_col = f"{resample_col}price_above_ema200" price_below_ema200_col = f"{resample_col}price_below_ema200" # 做多条件（激进优化：在下跌趋势中禁止做多，只在强上涨趋势中做多） # 1. 盘整背驰买入信号（i+2为奇数） # 2. 非背驰情况的回补信号 # 3. 确认是上涨段或即将上涨 # 4. 强上涨趋势确认（必须价格在EMA200上方且EMA斜率向上） # 5. 波动率确认 # 6. MACD确认 # 7. 禁止在下跌趋势中做多 dataframe.loc[ ( (dataframe[buy_signal_col] == True) & # 买入信号 ( (dataframe[panzheng_beichi_col] == True) | # 盘整背驰 (dataframe[ai_type_col] == 1) # 或当前是上涨段 ) & (dataframe[strong_uptrend_col] == True) & # 强上涨趋势（新增：必须强趋势） (dataframe[price_above_ema200_col] == True) & # 价格在EMA200上方（新增） (dataframe[volatility_ok_col] == True) & # 波动率足够 (dataframe[rsi_5m_col] < 60) & # RSI不过度超买（收紧） (dataframe[rsi_5m_col] > 40) & # RSI在合理区间（收紧） (dataframe['rsi_1m'] > 40) & # 1m RSI确认（收紧） (dataframe['rsi_1m'] < 65) & # 1m RSI不过度超买（收紧） (dataframe['macd_1m'] > dataframe['macdsignal_1m']) & # MACD向上 (dataframe['macd_1m'] > 0) & # MACD在零轴上方（新增） (dataframe['volume'] > dataframe['volume_mean'] * 1.5) & # 成交量确认（提高阈值） ~(dataframe[strong_downtrend_col] == True) # 禁止在强下跌趋势中做多（新增） ), ["enter_long", "enter_tag"], ] = (1, "same_level_long") # 做空条件（优化：收紧条件，提高质量） # 1. 盘整背驰卖出信号（i+2为偶数，但这里作为做空入场） # 2. 非背驰情况的卖出信号 # 3. 确认是下跌段或即将下跌 # 4. 强下跌趋势确认（必须价格在EMA200下方且EMA斜率向下） # 5. 波动率确认 # 6. MACD确认 sell_signal_col = f"{resample_col}sell_signal" dataframe.loc[ ( (dataframe[sell_signal_col] == True) & # 卖出信号 ( (dataframe[panzheng_beichi_col] == True) | # 盘整背驰（但i+2为偶数） (dataframe[ai_type_col] == -1) # 或当前是下跌段 ) & ( (dataframe[strong_downtrend_col] == True) | # 强下跌趋势（优先） ( (dataframe[ema_trend_dn_col] == True) & # 30m趋势向下 (dataframe[price_below_ema200_col] == True) # 且价格在EMA200下方 ) ) & (dataframe[volatility_ok_col] == True) & # 波动率足够 (dataframe[rsi_5m_col] > 40) & # RSI不过度超卖（收紧） (dataframe[rsi_5m_col] < 65) & # RSI不过度超买（收紧） (dataframe['rsi_1m'] < 65) & # 1m RSI确认（收紧） (dataframe['rsi_1m'] > 35) & # 1m RSI不过度超卖（收紧） (dataframe['macd_1m'] < dataframe['macdsignal_1m']) & # MACD向下 (dataframe['macd_1m'] < 0) & # MACD在零轴下方（新增） (dataframe['volume'] > dataframe['volume_mean'] * 1.3) # 成交量确认（提高阈值） ), ["enter_short", "enter_tag"], ] = (1, "same_level_short") return dataframe def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """ 出场逻辑：基于同级别分解的a+A结构 1. 盘整背驰卖出：i+2为偶数时的盘整背驰信号 2. 非背驰情况的卖出：Ai+3跌破Ai高点后的卖出信号 """ ticker = self.get_ticker_indicator() resample_col = f"resample_{ticker * self.same_level_timeframe}_" # 获取5m级别的指标 sell_signal_col = f"{resample_col}sell_signal" buy_signal_col = f"{resample_col}buy_signal" panzheng_beichi_col = f"{resample_col}panzheng_beichi" ai_type_col = f"{resample_col}ai_type" rsi_5m_col = f"{resample_col}rsi" ema_trend_up_col = f"{resample_col}ema_trend_up" ema_trend_dn_col = f"{resample_col}ema_trend_dn" strong_uptrend_col = f"{resample_col}strong_uptrend" strong_downtrend_col = f"{resample_col}strong_downtrend" # 做多出场（优化：更早退出，保护利润） # 在趋势转弱或明确反转时退出 dataframe.loc[ ( ( (dataframe[sell_signal_col] == True) & # 明确的卖出信号 (dataframe[panzheng_beichi_col] == True) # 且是背驰信号 ) | ( (dataframe[ai_type_col] == -1) & # 转为下跌段 (dataframe[rsi_5m_col] > 55) & # RSI确认（降低阈值，更早退出） (dataframe[ema_trend_dn_col] == True) # 且趋势确实向下 ) | ( (dataframe[strong_downtrend_col] == True) & # 强下跌趋势（新增） (dataframe[rsi_5m_col] > 50) # RSI确认 ) ), ["exit_long", "exit_tag"], ] = (1, "same_level_exit_long") # 做空出场（优化：更早退出，保护利润） # 在趋势转弱或明确反转时退出 dataframe.loc[ ( ( (dataframe[buy_signal_col] == True) & # 明确的买入信号 (dataframe[panzheng_beichi_col] == True) # 且是背驰信号 ) | ( (dataframe[ai_type_col] == 1) & # 转为上涨段 (dataframe[rsi_5m_col] < 45) & # RSI确认（提高阈值，更早退出） (dataframe[ema_trend_up_col] == True) # 且趋势确实向上 ) | ( (dataframe[strong_uptrend_col] == True) & # 强上涨趋势（新增） (dataframe[rsi_5m_col] < 50) # RSI确认 ) ), ["exit_short", "exit_tag"], ] = (1, "same_level_exit_short") return dataframe 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: return 1.0 def get_ticker_indicator(self) -> int: """获取 timeframe 的分钟数""" return int(self.timeframe[:-1])