Hyperliquid动量突破策略
Timeframe
15m
Direction
Long & Short
Stoploss
-10.0%
Trailing Stop
No
ROI
0m: 40.0%, 20m: 25.0%, 40m: 20.0%, 60m: 15.0%
Interface Version
3
Startup Candles
200
Indicators
5
freqtrade/freqtrade-strategies
Strategy 003 author@: Gerald Lonlas github@: https://github.com/freqtrade/freqtrade-strategies
"""
Hyperliquid动量突破策略
基于专业交易代理提示词实现的freqtrade策略
策略特点:
- 多重技术指标确认(EMA趋势、MACD动量、RSI强弱、ATR波动)
- 严格的风险管理(动态止损、仓位控制)
- 24/7加密货币市场适应性
- 基于置信度的仓位调整
- 突破确认机制减少假信号
"""
from freqtrade.strategy import IStrategy, IntParameter, DecimalParameter
from pandas import DataFrame
from typing import Optional
import talib.abstract as ta
import numpy as np
import logging
logger = logging.getLogger(__name__)
class HyperliquidMomentumStrategy(IStrategy):
"""
Hyperliquid动量突破策略
核心逻辑:
1. EMA趋势过滤 - 确保顺势交易
2. MACD动量确认 - 捕捉趋势转折点
3. RSI超买超卖过滤 - 避免极端位置入场
4. ATR波动性管理 - 动态调整止损和仓位
5. 多重确认机制 - 减少假突破
6. 基于置信度的仓位管理 - 高置信度加大仓位
"""
INTERFACE_VERSION = 3
# 基础配置
minimal_roi = {
"0": 0.40, # 40%收益立即止盈
"20": 0.25, # 20分钟后25%收益
"40": 0.20, # 40分钟后20%收益
"60": 0.15, # 1小时后15%收益
"120": 0.10, # 2小时后10%收益
"240": 0.06, # 4小时后6%收益
"480": 0.04 # 8小时后4%收益
}
stoploss = -0.10 # 10%基础止损
timeframe = '15m' # 15分钟时间框架,适应提示词的2-3分钟决策频率
# 策略控制
can_short = True # 启用做空
startup_candle_count = 200 # 需要足够K线计算指标
process_only_new_candles = True
use_exit_signal = True
use_custom_stoploss = True
# 订单类型配置
order_types = {
'entry': 'market',
'exit': 'market',
'stoploss': 'market',
'stoploss_on_exchange': False,
'stoploss_on_exchange_interval': 60,
}
# 可优化参数
# EMA趋势参数
ema_fast = IntParameter(10, 25, default=20, space="buy")
ema_slow = IntParameter(40, 80, default=50, space="buy")
ema_trend = IntParameter(100, 200, default=150, space="buy")
# MACD参数
macd_fast = IntParameter(8, 15, default=12, space="buy")
macd_slow = IntParameter(20, 30, default=26, space="buy")
macd_signal = IntParameter(6, 12, default=9, space="buy")
# RSI参数
rsi_period = IntParameter(10, 20, default=14, space="buy")
rsi_overbought = IntParameter(70, 80, default=75, space="sell")
rsi_oversold = IntParameter(20, 30, default=25, space="sell")
rsi_neutral_high = IntParameter(60, 70, default=65, space="buy")
rsi_neutral_low = IntParameter(35, 45, default=40, space="buy")
# ATR参数
atr_period = IntParameter(10, 20, default=14, space="sell")
atr_multiplier_stoploss = DecimalParameter(2.0, 4.0, default=3.0, space="sell")
atr_multiplier_entry = DecimalParameter(0.5, 2.0, default=1.0, space="buy")
# 成交量确认参数
volume_ma_period = IntParameter(15, 35, default=20, space="buy")
volume_multiplier_min = DecimalParameter(1.2, 2.0, default=1.5, space="buy")
volume_multiplier_strong = DecimalParameter(2.0, 3.5, default=2.5, space="buy")
# 价格突破参数
breakout_threshold = DecimalParameter(0.3, 1.5, default=0.8, space="buy")
pullback_threshold = DecimalParameter(0.2, 1.0, default=0.5, space="sell")
# 仓位管理参数
position_size_base = DecimalParameter(0.01, 0.05, default=0.02, space="buy") # 基础仓位2%
position_size_multiplier = DecimalParameter(1.5, 3.0, default=2.0, space="buy") # 高置信度2倍仓位
max_position_size = DecimalParameter(0.08, 0.20, default=0.10, space="buy") # 最大仓位10%
def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
"""
计算技术指标
"""
# 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'] = ta.EMA(dataframe, timeperiod=self.ema_trend.value)
# EMA趋势状态
dataframe['price_above_ema_fast'] = dataframe['close'] > dataframe['ema_fast']
dataframe['price_above_ema_slow'] = dataframe['close'] > dataframe['ema_slow']
dataframe['price_above_ema_trend'] = dataframe['close'] > dataframe['ema_trend']
dataframe['ema_bullish_alignment'] = (dataframe['ema_fast'] > dataframe['ema_slow']) & (dataframe['ema_slow'] > dataframe['ema_trend'])
dataframe['ema_bearish_alignment'] = (dataframe['ema_fast'] < dataframe['ema_slow']) & (dataframe['ema_slow'] < dataframe['ema_trend'])
# 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']
# MACD信号
dataframe['macd_bullish'] = dataframe['macd'] > dataframe['macd_signal']
dataframe['macd_bearish'] = dataframe['macd'] < dataframe['macd_signal']
dataframe['macd_cross_above'] = (dataframe['macd'] > dataframe['macd_signal']) & (dataframe['macd'].shift(1) <= dataframe['macd_signal'].shift(1))
dataframe['macd_cross_below'] = (dataframe['macd'] < dataframe['macd_signal']) & (dataframe['macd'].shift(1) >= dataframe['macd_signal'].shift(1))
dataframe['macd_hist_positive'] = dataframe['macd_hist'] > 0
dataframe['macd_hist_growing'] = dataframe['macd_hist'] > dataframe['macd_hist'].shift(1)
# RSI强弱指标
dataframe['rsi'] = ta.RSI(dataframe, timeperiod=self.rsi_period.value)
dataframe['rsi_overbought'] = dataframe['rsi'] > self.rsi_overbought.value
dataframe['rsi_oversold'] = dataframe['rsi'] < self.rsi_oversold.value
dataframe['rsi_neutral'] = (dataframe['rsi'] >= self.rsi_neutral_low.value) & (dataframe['rsi'] <= self.rsi_neutral_high.value)
dataframe['rsi_rising'] = dataframe['rsi'] > dataframe['rsi'].shift(1)
# ATR波动性指标
dataframe['atr'] = ta.ATR(dataframe, timeperiod=self.atr_period.value)
dataframe['atr_percent'] = dataframe['atr'] / dataframe['close']
dataframe['volatility_high'] = dataframe['atr_percent'] > dataframe['atr_percent'].rolling(20).mean() * 1.5
# 支撑阻力位
dataframe['resistance_20'] = dataframe['high'].rolling(20).max()
dataframe['support_20'] = dataframe['low'].rolling(20).min()
dataframe['resistance_50'] = dataframe['high'].rolling(50).max()
dataframe['support_50'] = dataframe['low'].rolling(50).min()
# 价格位置
dataframe['price_position_20'] = (dataframe['close'] - dataframe['support_20']) / (dataframe['resistance_20'] - dataframe['support_20'])
dataframe['near_resistance_20'] = dataframe['close'] > dataframe['resistance_20'].shift(1) * 0.98
dataframe['near_support_20'] = dataframe['close'] < dataframe['support_20'].shift(1) * 1.02
# 成交量指标
dataframe['volume_ma'] = dataframe['volume'].rolling(self.volume_ma_period.value).mean()
dataframe['volume_ratio'] = dataframe['volume'] / dataframe['volume_ma']
dataframe['volume_high'] = dataframe['volume_ratio'] > self.volume_multiplier_strong.value
dataframe['volume_normal'] = dataframe['volume_ratio'] > self.volume_multiplier_min.value
# 价格突破检测
dataframe['breakout_up'] = dataframe['close'] > dataframe['resistance_20'].shift(1)
dataframe['breakout_down'] = dataframe['close'] < dataframe['support_20'].shift(1)
dataframe['breakout_strength'] = np.where(
dataframe['breakout_up'],
(dataframe['close'] - dataframe['resistance_20'].shift(1)) / dataframe['close'] * 100,
np.where(
dataframe['breakout_down'],
(dataframe['support_20'].shift(1) - dataframe['close']) / dataframe['close'] * 100,
0
)
)
# 布林带辅助指标
bb = ta.BBANDS(dataframe, timeperiod=20, nbdevup=2.0, nbdevdn=2.0)
dataframe['bb_upper'] = bb['upperband']
dataframe['bb_lower'] = bb['lowerband']
dataframe['bb_middle'] = bb['middleband']
dataframe['bb_width'] = (dataframe['bb_upper'] - dataframe['bb_lower']) / dataframe['bb_middle']
dataframe['price_above_bb_upper'] = dataframe['close'] > dataframe['bb_upper']
dataframe['price_below_bb_lower'] = dataframe['close'] < dataframe['bb_lower']
# 置信度计算
dataframe['confidence_score'] = self._calculate_confidence_score(dataframe)
return dataframe
def _calculate_confidence_score(self, dataframe: DataFrame) -> DataFrame:
"""
计算交易置信度分数 (0-1)
"""
score = np.zeros(len(dataframe))
# 趋势一致性得分 (0-0.3)
trend_score = np.where(
dataframe['ema_bullish_alignment'],
0.3,
np.where(
dataframe['ema_bearish_alignment'],
-0.3,
0
)
)
# 动量确认得分 (0-0.25)
momentum_score = np.where(
dataframe['macd_bullish'] & dataframe['macd_hist_positive'],
0.25,
np.where(
dataframe['macd_bearish'] & ~dataframe['macd_hist_positive'],
-0.25,
0
)
)
# RSI位置得分 (0-0.2)
rsi_score = np.where(
dataframe['rsi_neutral'] & dataframe['rsi_rising'],
0.2,
np.where(
dataframe['rsi_oversold'],
0.15,
np.where(
dataframe['rsi_overbought'],
-0.15,
0
)
)
)
# 成交量确认得分 (0-0.15)
volume_score = np.where(
dataframe['volume_high'],
0.15,
np.where(
dataframe['volume_normal'],
0.1,
0
)
)
# 突破强度得分 (0-0.1)
breakout_score = np.clip(dataframe['breakout_strength'] / 2.0, -0.1, 0.1)
# 综合得分
total_score = trend_score + momentum_score + rsi_score + volume_score + breakout_score
# 归一化到0-1范围
normalized_score = (total_score + 1) / 2
return np.clip(normalized_score, 0, 1)
def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
"""
定义入场条件
"""
# 做多入场条件
long_conditions = (
# 趋势确认
dataframe['price_above_ema_trend'] &
dataframe['ema_bullish_alignment'] &
# 动量确认
dataframe['macd_cross_above'] &
dataframe['macd_hist_positive'] &
# RSI确认
(dataframe['rsi'] > self.rsi_neutral_low.value) &
(dataframe['rsi'] < self.rsi_neutral_high.value) &
dataframe['rsi_rising'] &
# 成交量确认
dataframe['volume_high'] &
# 突破确认
(dataframe['breakout_strength'] > self.breakout_threshold.value) &
# 波动性过滤
~dataframe['volatility_high'] &
# 置信度要求
(dataframe['confidence_score'] > 0.65)
)
dataframe.loc[long_conditions, 'enter_long'] = 1
# 做空入场条件
short_conditions = (
# 趋势确认
~dataframe['price_above_ema_trend'] &
dataframe['ema_bearish_alignment'] &
# 动量确认
dataframe['macd_cross_below'] &
~dataframe['macd_hist_positive'] &
# RSI确认
(dataframe['rsi'] > self.rsi_neutral_low.value) &
(dataframe['rsi'] < self.rsi_neutral_high.value) &
~dataframe['rsi_rising'] &
# 成交量确认
dataframe['volume_high'] &
# 突破确认
(dataframe['breakout_strength'] > self.breakout_threshold.value) &
# 波动性过滤
~dataframe['volatility_high'] &
# 置信度要求
(dataframe['confidence_score'] > 0.65)
)
dataframe.loc[short_conditions, 'enter_short'] = 1
return dataframe
def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
"""
定义出场条件
"""
# 多头出场条件
exit_long_conditions = (
# 趋势反转
(~dataframe['price_above_ema_slow']) |
(dataframe['ema_fast'] < dataframe['ema_slow']) |
# 动量衰竭
(dataframe['macd_cross_below']) |
# RSI过热
(dataframe['rsi'] > self.rsi_overbought.value) |
# 破位下跌
(dataframe['close'] < dataframe['support_20'].shift(1)) |
# 置信度过低
(dataframe['confidence_score'] < 0.35)
)
dataframe.loc[exit_long_conditions, 'exit_long'] = 1
# 空头出场条件
exit_short_conditions = (
# 趋势反转
(dataframe['price_above_ema_slow']) |
(dataframe['ema_fast'] > dataframe['ema_slow']) |
# 动量衰竭
(dataframe['macd_cross_above']) |
# RSI超卖
(dataframe['rsi'] < self.rsi_oversold.value) |
# 破位上涨
(dataframe['close'] > dataframe['resistance_20'].shift(1)) |
# 置信度过低
(dataframe['confidence_score'] < 0.35)
)
dataframe.loc[exit_short_conditions, 'exit_short'] = 1
return dataframe
def custom_stoploss(self, pair: str, trade, current_time, current_rate: float,
current_profit: float, **kwargs) -> float:
"""
动态止损策略
"""
dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)
if len(dataframe) < 1:
return self.stoploss
last_candle = dataframe.iloc[-1].squeeze()
# ATR动态止损
atr_stop_distance = (last_candle['atr'] * self.atr_multiplier_stoploss.value) / current_rate
# 支撑/阻力位止损
if trade.is_short:
support_stop_distance = abs(last_candle['support_20'] - current_rate) / current_rate
stop_distance = max(atr_stop_distance, support_stop_distance)
else:
resistance_stop_distance = abs(last_candle['resistance_20'] - current_rate) / current_rate
stop_distance = max(atr_stop_distance, resistance_stop_distance)
# 根据盈利情况调整止损
if current_profit > 0.30: # 盈利30%以上,收紧止损
return max(-stop_distance * 0.5, -0.03)
elif current_profit > 0.20: # 盈利20%以上
return max(-stop_distance * 0.6, -0.04)
elif current_profit > 0.10: # 盈利10%以上
return max(-stop_distance * 0.7, -0.05)
elif current_profit > 0.05: # 盈利5%以上
return max(-stop_distance * 0.8, -0.06)
elif current_profit > -0.02: # 小幅亏损或盈利
return max(-stop_distance * 1.0, self.stoploss)
else: # 较大亏损
return max(-stop_distance * 1.2, self.stoploss)
def custom_stake_amount(self, pair: str, current_time: str, current_rate: float,
proposed_stake: float, min_stake: Optional[float], max_stake: float,
leverage: float, entry_tag: Optional[str], side: str,
**kwargs) -> float:
"""
自定义仓位管理
"""
dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)
if len(dataframe) < 1:
return proposed_stake
last_candle = dataframe.iloc[-1].squeeze()
# 基础仓位计算
available_capital = self.wallets.get_total_stake_amount()
base_stake = available_capital * self.position_size_base.value
# 根据置信度调整仓位
confidence = last_candle['confidence_score']
if confidence > 0.85: # 超高置信度
position_multiplier = self.position_size_multiplier.value
elif confidence > 0.75: # 高置信度
position_multiplier = 1.5
elif confidence > 0.65: # 中等置信度
position_multiplier = 1.2
else: # 低置信度
position_multiplier = 0.8
# 根据波动性调整仓位
if last_candle['volatility_high']:
position_multiplier *= 0.7 # 高波动时减仓
# 根据突破强度调整仓位
if last_candle['breakout_strength'] > 1.5:
position_multiplier *= 1.2 # 强突破时加仓
adjusted_stake = base_stake * position_multiplier
# 限制最大仓位
max_allowed = available_capital * self.max_position_size.value
adjusted_stake = min(adjusted_stake, max_allowed)
return min(max(adjusted_stake, min_stake or 0), max_stake)
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:
"""
交易确认
"""
dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)
if len(dataframe) < 2:
return False
last_candle = dataframe.iloc[-1].squeeze()
prev_candle = dataframe.iloc[-2].squeeze()
# 置信度检查
if last_candle['confidence_score'] < 0.65:
return False
# 做多交易确认
if side == 'long':
# 确保趋势向上
if not (last_candle['price_above_ema_trend'] and last_candle['ema_bullish_alignment']):
return False
# 确保动量向上
if not (last_candle['macd_hist_positive'] and last_candle['macd_hist_growing']):
return False
# 确保RSI合理
if not (self.rsi_neutral_low.value <= last_candle['rsi'] <= self.rsi_neutral_high.value):
return False
# 确保成交量充足
if last_candle['volume_ratio'] < self.volume_multiplier_min.value:
return False
# 做空交易确认
elif side == 'short':
# 确保趋势向下
if not (not last_candle['price_above_ema_trend'] and last_candle['ema_bearish_alignment']):
return False
# 确保动量向下
if not (not last_candle['macd_hist_positive'] and not last_candle['macd_hist_growing']):
return False
# 确保RSI合理
if not (self.rsi_neutral_low.value <= last_candle['rsi'] <= self.rsi_neutral_high.value):
return False
# 确保成交量充足
if last_candle['volume_ratio'] < self.volume_multiplier_min.value:
return False
return True
def custom_exit(self, pair: str, trade, current_time, current_rate: float,
current_profit: float, **kwargs) -> str:
"""
自定义退出逻辑
"""
dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)
if len(dataframe) < 1:
return None
last_candle = dataframe.iloc[-1].squeeze()
# 置信度急剧下降退出
if last_candle['confidence_score'] < 0.3:
return "confidence_drop"
# 动量衰竭退出
if trade.is_short:
if (last_candle['macd_cross_above'] and current_profit < 0.02):
return "momentum_reversal_long"
else:
if (last_candle['macd_cross_below'] and current_profit < 0.02):
return "momentum_reversal_short"
# 趋势破坏退出
if trade.is_short:
if (last_candle['price_above_ema_slow'] and current_profit < 0.01):
return "trend_breakdown_long"
else:
if (not last_candle['price_above_ema_slow'] and current_profit < 0.01):
return "trend_breakdown_short"
# 时间止损
trade_duration = (current_time - trade.open_date_utc).total_seconds() / 3600
if trade_duration > 48 and abs(current_profit) < 0.02: # 48小时无显著收益
return "time_exit"
return None
def leverage(self, pair: str, current_time: str, current_rate: float,
proposed_leverage: float, max_leverage: int, entry_tag: Optional[str],
side: str, **kwargs) -> float:
"""
动态杠杆管理
"""
dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)
if len(dataframe) < 1:
return 1.0
last_candle = dataframe.iloc[-1].squeeze()
# 基于置信度调整杠杆
confidence = last_candle['confidence_score']
if confidence > 0.85:
return min(3.0, max_leverage) # 最高3倍杠杆
elif confidence > 0.75:
return min(2.0, max_leverage) # 2倍杠杆
elif confidence > 0.65:
return 1.5 # 1.5倍杠杆
else:
return 1.0 # 1倍杠杆