"""
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
Henry Lin
Posted on
ADXTrendStrategy(ADX趋势强度策略)
For further actions, you may consider blocking this person and/or reporting abuse
Top comments (0)