""" ADX趋势强度策略 基于ADX指标识别和跟随强趋势的策略 """ from freqtrade.strategy import IStrategy, IntParameter, DecimalParameter from pandas import DataFrame import talib.abstract as ta from technical import qtpylib import numpy as np import logging logger = logging.getLogger(__name__) class ADXTrendStrategy(IStrategy): """ ADX趋势强度策略 策略逻辑: 1. 使用ADX识别趋势强度 2. 使用DI+和DI-确定趋势方向 3. 当ADX上升且DI+>DI-时买入 4. 当ADX下降或DI->DI+时卖出 5. 结合其他指标过滤假信号 """ INTERFACE_VERSION = 3 # 基础配置 minimal_roi = { "0": 0.20, # 20%收益立即止盈 "60": 0.12, # 1小时后12%收益 "120": 0.08, # 2小时后8%收益 "240": 0.04, # 4小时后4%收益 "480": 0.02 # 8小时后2%收益 } stoploss = -0.06 # 6%止损 timeframe = '1h' # 1小时时间框架 # 策略控制 can_short = False startup_candle_count = 100 process_only_new_candles = True use_exit_signal = True use_custom_stoploss = True # 可优化参数 # ADX参数 adx_period = IntParameter(10, 20, default=14, space="buy") adx_threshold_strong = IntParameter(25, 40, default=30, space="buy") adx_threshold_weak = IntParameter(15, 25, default=20, space="sell") # DI差值参数 di_diff_threshold = DecimalParameter(2.0, 8.0, default=5.0, space="buy") # ADX趋势参数 adx_slope_periods = IntParameter(3, 7, default=5, space="buy") adx_min_slope = DecimalParameter(0.5, 3.0, default=1.5, space="buy") # 确认指标参数 # EMA趋势确认 ema_fast = IntParameter(8, 20, default=12, space="buy") ema_slow = IntParameter(24, 40, default=30, space="buy") # RSI过滤 rsi_period = IntParameter(10, 20, default=14, space="buy") rsi_buy_threshold = IntParameter(45, 60, default=50, space="buy") rsi_sell_threshold = IntParameter(65, 85, default=75, space="sell") # MACD确认 macd_fast = IntParameter(8, 16, default=12, space="buy") macd_slow = IntParameter(21, 35, default=26, space="buy") macd_signal = IntParameter(7, 12, default=9, space="buy") # 成交量参数 volume_factor = DecimalParameter(1.2, 2.5, default=1.6, space="buy") # ATR止损参数 atr_period = IntParameter(10, 20, default=14, space="sell") atr_multiplier = DecimalParameter(2.0, 4.0, default=2.5, space="sell") # 抛物线SAR参数 sar_acceleration = DecimalParameter(0.01, 0.05, default=0.02, space="buy") sar_maximum = DecimalParameter(0.1, 0.3, default=0.2, space="buy") def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """ 计算技术指标 """ # ADX指标组 dataframe['adx'] = ta.ADX(dataframe, timeperiod=self.adx_period.value) dataframe['di_plus'] = ta.PLUS_DI(dataframe, timeperiod=self.adx_period.value) dataframe['di_minus'] = ta.MINUS_DI(dataframe, timeperiod=self.adx_period.value) # DI差值和比率 dataframe['di_diff'] = dataframe['di_plus'] - dataframe['di_minus'] dataframe['di_ratio'] = dataframe['di_plus'] / (dataframe['di_minus'] + 0.001) # 避免除零 # ADX趋势和斜率 dataframe['adx_slope'] = (dataframe['adx'] - dataframe['adx'].shift(self.adx_slope_periods.value)) / self.adx_slope_periods.value dataframe['adx_rising'] = dataframe['adx_slope'] > self.adx_min_slope.value dataframe['adx_falling'] = dataframe['adx_slope'] < -self.adx_min_slope.value # ADX强度分类 dataframe['adx_very_strong'] = dataframe['adx'] > 50 dataframe['adx_strong'] = (dataframe['adx'] > self.adx_threshold_strong.value) & (dataframe['adx'] <= 50) dataframe['adx_moderate'] = (dataframe['adx'] > 25) & (dataframe['adx'] <= self.adx_threshold_strong.value) dataframe['adx_weak'] = dataframe['adx'] <= self.adx_threshold_weak.value # 趋势方向 dataframe['bullish_trend'] = (dataframe['di_plus'] > dataframe['di_minus']) & (dataframe['di_diff'] > self.di_diff_threshold.value) dataframe['bearish_trend'] = (dataframe['di_minus'] > dataframe['di_plus']) & (dataframe['di_diff'] < -self.di_diff_threshold.value) # EMA趋势确认 dataframe['ema_fast'] = ta.EMA(dataframe, timeperiod=self.ema_fast.value) dataframe['ema_slow'] = ta.EMA(dataframe, timeperiod=self.ema_slow.value) dataframe['ema_trend_up'] = dataframe['ema_fast'] > dataframe['ema_slow'] dataframe['price_above_ema_fast'] = dataframe['close'] > dataframe['ema_fast'] # RSI dataframe['rsi'] = ta.RSI(dataframe, timeperiod=self.rsi_period.value) # MACD macd = ta.MACD(dataframe, fastperiod=self.macd_fast.value, slowperiod=self.macd_slow.value, signalperiod=self.macd_signal.value) dataframe['macd'] = macd['macd'] dataframe['macd_signal'] = macd['macdsignal'] dataframe['macd_hist'] = macd['macdhist'] dataframe['macd_bullish'] = dataframe['macd'] > dataframe['macd_signal'] # ATR dataframe['atr'] = ta.ATR(dataframe, timeperiod=self.atr_period.value) dataframe['atr_percent'] = dataframe['atr'] / dataframe['close'] # 抛物线SAR dataframe['sar'] = ta.SAR(dataframe, acceleration=self.sar_acceleration.value, maximum=self.sar_maximum.value) dataframe['sar_bullish'] = dataframe['close'] > dataframe['sar'] # 成交量指标 dataframe['volume_sma'] = dataframe['volume'].rolling(window=20).mean() dataframe['volume_ratio'] = dataframe['volume'] / dataframe['volume_sma'] # 布林带 bollinger = qtpylib.bollinger_bands(dataframe['close'], window=20, stds=2) dataframe['bb_upper'] = bollinger['upper'] dataframe['bb_middle'] = bollinger['mid'] dataframe['bb_lower'] = bollinger['lower'] dataframe['bb_percent'] = (dataframe['close'] - dataframe['bb_lower']) / (dataframe['bb_upper'] - dataframe['bb_lower']) # 价格动量 dataframe['price_momentum'] = (dataframe['close'] - dataframe['close'].shift(5)) / dataframe['close'].shift(5) # Aroon指标（趋势确认） aroon = ta.AROON(dataframe, timeperiod=14) dataframe['aroon_up'] = aroon['aroonup'] dataframe['aroon_down'] = aroon['aroondown'] dataframe['aroon_osc'] = dataframe['aroon_up'] - dataframe['aroon_down'] # 趋势综合评分 # 基于多个指标的趋势强度评分 trend_score = 0 trend_score += np.where(dataframe['bullish_trend'], 2, 0) # ADX方向 trend_score += np.where(dataframe['adx_strong'] | dataframe['adx_very_strong'], 2, 0) # ADX强度 trend_score += np.where(dataframe['adx_rising'], 1, 0) # ADX上升 trend_score += np.where(dataframe['ema_trend_up'], 1, 0) # EMA趋势 trend_score += np.where(dataframe['price_above_ema_fast'], 1, 0) # 价格位置 trend_score += np.where(dataframe['macd_bullish'], 1, 0) # MACD trend_score += np.where(dataframe['sar_bullish'], 1, 0) # SAR trend_score += np.where(dataframe['aroon_osc'] > 20, 1, 0) # Aroon dataframe['trend_score'] = trend_score return dataframe def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """ 定义买入条件 - ADX趋势策略 """ conditions = [ # 主要ADX信号 dataframe['bullish_trend'], # DI+>DI-且差值足够大 dataframe['adx'] > self.adx_threshold_strong.value, # ADX强度足够 dataframe['adx_rising'], # ADX上升趋势 # 趋势确认 dataframe['ema_trend_up'], # EMA趋势向上 dataframe['price_above_ema_fast'], # 价格在快EMA之上 # RSI确认 dataframe['rsi'] > self.rsi_buy_threshold.value, dataframe['rsi'] < 80, # 避免极度超买 # MACD确认 dataframe['macd_bullish'], dataframe['macd_hist'] > 0, dataframe['macd_hist'] > dataframe['macd_hist'].shift(1), # MACD柱状图增强 # SAR确认 dataframe['sar_bullish'], # 成交量确认 dataframe['volume_ratio'] > self.volume_factor.value, # 价格动量确认 dataframe['price_momentum'] > 0.01, # 至少1%的价格动量 # Aroon确认 dataframe['aroon_osc'] > 20, # Aroon振荡器显示上升趋势 # 布林带位置 dataframe['bb_percent'] > 0.3, # 不在布林带下轨附近 dataframe['bb_percent'] < 0.9, # 不在布林带上轨附近 # 综合趋势评分 dataframe['trend_score'] >= 6, # 趋势评分足够高 # ADX强度分类 dataframe['adx_strong'] | dataframe['adx_very_strong'], ] # 组合条件 dataframe.loc[ ( conditions[0] & # 牛市趋势 conditions[1] & # ADX强度 conditions[2] & # ADX上升 conditions[3] & # EMA趋势 conditions[4] & # 价格位置 conditions[5] & # RSI下限 conditions[6] & # RSI上限 conditions[7] & # MACD信号 conditions[8] & # MACD柱状图 conditions[9] & # MACD增强 conditions[10] & # SAR信号 conditions[11] & # 成交量 conditions[12] & # 价格动量 conditions[13] & # Aroon信号 conditions[14] & # 布林带下限 conditions[15] & # 布林带上限 conditions[16] & # 趋势评分 conditions[17] # ADX强度分类 ), 'enter_long' ] = 1 return dataframe def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """ 定义卖出条件 """ conditions = [ # 主要ADX信号转向 dataframe['bearish_trend'], # DI->DI+ dataframe['adx'] < self.adx_threshold_weak.value, # ADX弱化 dataframe['adx_falling'], # ADX下降 # 趋势转向 ~dataframe['ema_trend_up'], # EMA趋势向下 dataframe['close'] < dataframe['ema_fast'], # 价格跌破快EMA # RSI转向 dataframe['rsi'] > self.rsi_sell_threshold.value, # RSI超买 dataframe['rsi'] < 40, # RSI转弱 # MACD转向 ~dataframe['macd_bullish'], # MACD死叉 dataframe['macd_hist'] < 0, # MACD柱状图转负 # SAR转向 ~dataframe['sar_bullish'], # SAR信号转向 # Aroon转向 dataframe['aroon_osc'] < -20, # Aroon显示下降趋势 # 综合趋势评分下降 dataframe['trend_score'] <= 3, # DI差值收窄 abs(dataframe['di_diff']) < 2, # DI差值过小，趋势不明确 ] dataframe.loc[ ( conditions[0] | # 熊市趋势 conditions[1] | # ADX弱化 conditions[2] | # ADX下降 conditions[3] | # EMA趋势转向 conditions[4] | # 价格跌破EMA conditions[5] | # RSI超买 conditions[6] | # RSI转弱 conditions[7] | # MACD死叉 conditions[8] | # MACD柱状图 conditions[9] | # SAR转向 conditions[10] | # Aroon转向 conditions[11] | # 趋势评分下降 conditions[12] # DI差值收窄 ), 'exit_long' ] = 1 return dataframe def custom_stoploss(self, pair: str, trade, current_time, current_rate: float, current_profit: float, **kwargs) -> float: """ 基于ADX的动态止损 """ dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe) last_candle = dataframe.iloc[-1].squeeze() # ATR动态止损 atr_distance = last_candle['atr'] * self.atr_multiplier.value atr_stop_distance = atr_distance / current_rate # 基于ADX强度的止损调整 adx_value = last_candle['adx'] if current_profit > 0.10: # 盈利超过10% if adx_value > 40: # 极强趋势 return max(-atr_stop_distance * 0.6, -0.015) # 紧跟趋势 elif adx_value > 30: # 强趋势 return max(-atr_stop_distance * 0.7, -0.02) else: # 趋势减弱 return max(-atr_stop_distance * 0.8, -0.025) elif current_profit > 0.05: # 盈利超过5% if adx_value > 35: return max(-atr_stop_distance * 0.7, -0.025) else: return max(-atr_stop_distance * 0.9, -0.03) elif current_profit > 0.02: # 盈利超过2% return max(-atr_stop_distance * 0.8, -0.035) else: # ADX策略需要给趋势发展更多空间 if adx_value > 30: return max(-atr_stop_distance * 1.0, self.stoploss) else: return max(-atr_stop_distance * 1.2, self.stoploss) def custom_exit(self, pair: str, trade, current_time, current_rate: float, current_profit: float, **kwargs) -> str: """ 基于ADX的自定义退出 """ dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe) last_candle = dataframe.iloc[-1].squeeze() # ADX急剧下降退出 if (last_candle['adx'] < 20 and last_candle['adx_slope'] < -2 and current_profit > 0.02): return "adx_collapse" # DI交叉退出 if (last_candle['di_minus'] > last_candle['di_plus'] and abs(last_candle['di_diff']) > 3 and current_profit > 0.01): return "di_crossover" # 趋势评分急剧下降 if (last_candle['trend_score'] <= 2 and current_profit > 0.02): return "trend_score_collapse" # SAR反转 if (not last_candle['sar_bullish'] and current_profit > 0.03): return "sar_reversal" # 时间止损 trade_duration = (current_time - trade.open_date_utc).total_seconds() / 3600 if trade_duration > 48 and current_profit < 0.01: # 48小时无显著盈利 return "time_exit" return None def confirm_trade_entry(self, pair: str, order_type: str, amount: float, rate: float, time_in_force: str, current_time, entry_tag: str, side: str, **kwargs) -> bool: """ ADX策略特有的交易确认 """ dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe) if len(dataframe) < 1: return False last_candle = dataframe.iloc[-1].squeeze() # 确保ADX足够强 if last_candle['adx'] < self.adx_threshold_strong.value: return False # 确保DI差值足够大 if last_candle['di_diff'] < self.di_diff_threshold.value: return False # 确保ADX正在上升 if not last_candle['adx_rising']: return False # 确保趋势评分足够高 if last_candle['trend_score'] < 6: return False # 确保不是极端市况 if last_candle['atr_percent'] > 0.08: # 波动率过大 return False # 确保价格不在极端位置 if last_candle['rsi'] > 85 or last_candle['rsi'] < 20: return False return True