# source: https://raw.githubusercontent.com/henrylin99/freqtrade-tutorials/691c3799846fc7ff6173a3b9241ce5a00b346141/user_data/strategies/CryptoBreakoutStrategy.py """ 加密货币突破型趋势跟随策略 基于伪代码实现的专业突破交易策略，专为加密货币市场特性优化 策略特点： - 24小时交易适应性 - 高波动性过滤机制 - 情绪驱动行情识别 - 多重确认减少假突破 """ 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 Github_henrylin99_freqtrade_tutorials__CryptoBreakoutStrategy__20251212_023733(IStrategy): """ 加密货币突破型趋势跟随策略 核心逻辑： 1. 动态支撑阻力识别（30日高低点） 2. 成交量确认突破有效性（2倍成交量） 3. RSI过滤避免极端位置入场 4. 突破幅度要求过滤假突破 5. 连续确认机制减少噪音 """ INTERFACE_VERSION = 3 # 基础配置 minimal_roi = { "0": 0.30, # 30%收益立即止盈 "30": 0.20, # 30分钟后20%收益 "60": 0.15, # 1小时后15%收益 "120": 0.10, # 2小时后10%收益 "240": 0.05, # 4小时后5%收益 "480": 0.03 # 8小时后3%收益 } stoploss = -0.08 # 8%止损，适应高波动 timeframe = '30m' # 30分钟时间框架，平衡响应速度和稳定性 # 策略控制 can_short = False # 暂时禁用做空，专注做多机会 startup_candle_count = 150 # 需要更多K线计算30日高低点 process_only_new_candles = True use_exit_signal = True use_custom_stoploss = True # 可优化参数 # 支撑阻力参数 support_resistance_period = IntParameter(20, 40, default=30, space="buy") # 30日高低点 # 突破确认参数 breakout_threshold = DecimalParameter(0.5, 2.5, default=1.0, space="buy") # 1%突破幅度 volume_multiplier = DecimalParameter(1.5, 3.0, default=2.0, space="buy") # 2倍成交量确认 volume_short_multiplier = DecimalParameter(1.2, 2.5, default=1.8, space="buy") # 做空1.8倍 # RSI过滤参数 rsi_period = IntParameter(10, 20, default=14, space="buy") rsi_buy_upper = IntParameter(65, 80, default=70, space="buy") # 做多RSI上限 rsi_buy_lower = IntParameter(35, 50, default=40, space="buy") # 做多RSI下限 rsi_sell_upper = IntParameter(60, 75, default=70, space="sell") # 做空RSI上限 rsi_sell_lower = IntParameter(20, 40, default=30, space="sell") # 做空RSI下限 # 成交量参数 volume_ma_period = IntParameter(20, 40, default=30, space="buy") # 动态止损参数 atr_period = IntParameter(10, 20, default=14, space="sell") atr_multiplier = DecimalParameter(2.0, 4.0, default=2.5, space="sell") # 趋势确认参数 ema_trend_period = IntParameter(20, 50, default=30, space="buy") ema_fast_period = IntParameter(10, 20, default=15, space="buy") def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """ 计算技术指标 - 基于伪代码实现 """ # 动态支撑阻力（30日高低点） dataframe['resistance'] = dataframe['high'].rolling(window=self.support_resistance_period.value).max() dataframe['support'] = dataframe['low'].rolling(window=self.support_resistance_period.value).min() # 成交量指标 dataframe['volume_ma'] = dataframe['volume'].rolling(window=self.volume_ma_period.value).mean() dataframe['volume_ratio'] = dataframe['volume'] / dataframe['volume_ma'] # RSI指标 dataframe['rsi'] = ta.RSI(dataframe, timeperiod=self.rsi_period.value) # 突破幅度计算 dataframe['breakout_distance'] = (dataframe['close'] - dataframe['resistance'].shift(1)) / dataframe['close'] dataframe['breakdown_distance'] = (dataframe['support'].shift(1) - dataframe['close']) / dataframe['close'] # 趋势确认EMA dataframe['ema_trend'] = ta.EMA(dataframe, timeperiod=self.ema_trend_period.value) dataframe['ema_fast'] = ta.EMA(dataframe, timeperiod=self.ema_fast_period.value) dataframe['price_above_ema'] = dataframe['close'] > dataframe['ema_trend'] dataframe['ema_bullish'] = dataframe['ema_fast'] > dataframe['ema_trend'] # ATR动态止损 dataframe['atr'] = ta.ATR(dataframe, timeperiod=self.atr_period.value) dataframe['atr_percent'] = dataframe['atr'] / dataframe['close'] # 价格位置指标 dataframe['price_range'] = dataframe['high'] - dataframe['low'] dataframe['price_position'] = (dataframe['close'] - dataframe['support']) / (dataframe['resistance'] - dataframe['support']) # 成交量异常检测 dataframe['volume_spike'] = dataframe['volume_ratio'] > self.volume_multiplier.value dataframe['volume_spike_short'] = dataframe['volume_ratio'] > self.volume_short_multiplier.value # 突破强度评估 dataframe['breakout_strength'] = np.where( dataframe['close'] > dataframe['resistance'].shift(1), dataframe['breakout_distance'] * 100, 0 ) dataframe['breakdown_strength'] = np.where( dataframe['close'] < dataframe['support'].shift(1), dataframe['breakdown_distance'] * 100, 0 ) # 连续确认机制（减少假突破） dataframe['prev_above_resistance'] = dataframe['close'].shift(1) > dataframe['resistance'].shift(2) dataframe['prev_below_support'] = dataframe['close'].shift(1) < dataframe['support'].shift(2) # 假突破检测 dataframe['false_breakout_up'] = ( (dataframe['close'].shift(1) > dataframe['resistance'].shift(2)) & (dataframe['close'] <= dataframe['resistance'].shift(1)) ) dataframe['false_breakdown'] = ( (dataframe['close'].shift(1) < dataframe['support'].shift(2)) & (dataframe['close'] >= dataframe['support'].shift(1)) ) # 布林带（辅助判断） 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'] # MACD（趋势确认） macd = ta.MACD(dataframe) dataframe['macd'] = macd['macd'] dataframe['macd_signal'] = macd['macdsignal'] dataframe['macd_hist'] = macd['macdhist'] # 市场结构分析 dataframe['higher_highs'] = dataframe['high'] > dataframe['high'].shift(1) dataframe['higher_lows'] = dataframe['low'] > dataframe['low'].shift(1) dataframe['uptrend_structure'] = dataframe['higher_highs'] & dataframe['higher_lows'] return dataframe def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """ 定义买入条件 - 基于伪代码的突破条件 """ # 做多突破条件（完全基于伪代码） long_breakout = ( # 突破关键阻力 (dataframe['close'] > dataframe['resistance'].shift(1)) & # 成交量确认（2倍成交量） (dataframe['volume'] > dataframe['volume_ma'] * self.volume_multiplier.value) & # RSI不过热（RSI < 70） (dataframe['rsi'] < self.rsi_buy_upper.value) & # RSI不过低（RSI > 40） (dataframe['rsi'] > self.rsi_buy_lower.value) & # 突破幅度要求（1%以上） (dataframe['breakout_distance'] > self.breakout_threshold.value / 100) & # 连续确认（前一根K线也在阻力之上） (dataframe['close'].shift(1) > dataframe['resistance'].shift(2)) & # 趋势确认 (dataframe['price_above_ema']) & # 避免假突破 ~(dataframe['false_breakout_up']) & # 突破强度足够 (dataframe['breakout_strength'] > 0.5) ) dataframe.loc[long_breakout, 'enter_long'] = 1 # 做空突破条件（如果启用） if self.can_short: short_breakout = ( # 跌破关键支撑 (dataframe['close'] < dataframe['support'].shift(1)) & # 成交量确认（1.8倍成交量） (dataframe['volume'] > dataframe['volume_ma'] * self.volume_short_multiplier.value) & # RSI不超卖（RSI > 30） (dataframe['rsi'] > self.rsi_sell_lower.value) & # RSI不过高（RSI < 70） (dataframe['rsi'] < self.rsi_sell_upper.value) & # 跌破幅度要求（1%以上） (dataframe['breakdown_distance'] > self.breakout_threshold.value / 100) & # 连续确认（前一根K线也在支撑之下） (dataframe['close'].shift(1) < dataframe['support'].shift(2)) & # 趋势确认（下降趋势） ~(dataframe['price_above_ema']) & # 避免假突破 ~(dataframe['false_breakdown']) & # 跌破强度足够 (dataframe['breakdown_strength'] > 0.5) ) dataframe.loc[short_breakout, 'enter_short'] = 1 return dataframe def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """ 定义卖出条件 - 突破策略的退出逻辑 """ # 多头退出条件 exit_long_condition = ( # 条件1：跌破关键支撑 (dataframe['close'] < dataframe['support'].shift(1)) | # 条件2：跌破EMA趋势线 (dataframe['close'] < dataframe['ema_trend']) | # 条件3：RSI极度超买 (dataframe['rsi'] > 85) | # 条件4：MACD死叉 (dataframe['macd'] < dataframe['macd_signal']) | # 条件5：假突破确认 (dataframe['false_breakout_up'] & (dataframe['close'] < dataframe['resistance'] * 0.98)) | # 条件6：结构破坏 ~(dataframe['uptrend_structure']) | # 条件7：成交量萎缩 (dataframe['volume_ratio'] < 0.5) ) dataframe.loc[exit_long_condition, 'exit_long'] = 1 # 空头退出条件（如果启用） if self.can_short: exit_short_condition = ( # 条件1：涨破关键阻力 (dataframe['close'] > dataframe['resistance'].shift(1)) | # 条件2：涨破EMA趋势线 (dataframe['close'] > dataframe['ema_trend']) | # 条件3：RSI极度超卖 (dataframe['rsi'] < 15) | # 条件4：MACD金叉 (dataframe['macd'] > dataframe['macd_signal']) | # 条件5：假突破确认 (dataframe['false_breakdown'] & (dataframe['close'] > dataframe['support'] * 1.02)) ) dataframe.loc[exit_short_condition, '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.value) / current_rate # 支撑位止损 support_stop_distance = abs(current_rate - last_candle['support']) / current_rate # 突破策略的特殊止损逻辑 if current_profit > 0.20: # 盈利超过20%，非常严格 return max(min(-support_stop_distance * 0.6, -atr_stop_distance * 0.3), -0.015) elif current_profit > 0.15: # 盈利超过15%，严格止损 return max(min(-support_stop_distance * 0.7, -atr_stop_distance * 0.4), -0.02) elif current_profit > 0.10: # 盈利超过10% return max(min(-support_stop_distance * 0.8, -atr_stop_distance * 0.5), -0.025) elif current_profit > 0.05: # 盈利超过5% return max(-atr_stop_distance * 0.7, -0.035) elif current_profit > 0.02: # 盈利超过2% return max(-atr_stop_distance * 0.85, -0.05) 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: """ 自定义退出逻辑 - 突破失败检测 """ dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe) if len(dataframe) < 1: return None last_candle = dataframe.iloc[-1].squeeze() # 突破失败退出 if (current_profit < -0.03 and # 亏损超过3% last_candle['close'] < last_candle['resistance'] * 0.99): # 跌破阻力位99% return "breakout_failure" # 假突破退出 if (current_profit < -0.02 and last_candle['false_breakout_up']): return "false_breakout" # 动能衰减退出 if (current_profit > 0.05 and last_candle['rsi'] < 45 and # RSI回落 last_candle['macd_hist'] < 0 and # MACD柱状图转负 last_candle['volume_ratio'] < 1.0): # 成交量萎缩 return "momentum_exhaustion" # 结构破坏退出 if (not last_candle['uptrend_structure'] and current_profit < 0.02): return "structure_breakdown" # 时间止损 - 突破策略需要时间验证 trade_duration = (current_time - trade.open_date_utc).total_seconds() / 3600 if trade_duration > 24 and current_profit < 0.01: # 24小时无显著收益 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: """ 交易确认 - 突破策略的严格验证 """ 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 side == 'long': # 确保是真实突破 if last_candle['breakout_strength'] < 0.5: return False # 确保成交量足够 if last_candle['volume_ratio'] < self.volume_multiplier.value * 0.8: return False # 确保RSI在合理范围 if last_candle['rsi'] > self.rsi_buy_upper.value: return False # 确保突破幅度足够 if last_candle['breakout_distance'] < self.breakout_threshold.value / 100: return False # 确保不是极端波动 if last_candle['atr_percent'] > 0.10: # 单根K线波动超过10% return False # 连续确认：确保前一根K线也在突破状态 if not (prev_candle['close'] > prev_candle['resistance'].shift(1)): return False # 做空交易确认（如果启用） elif side == 'short': if last_candle['breakdown_strength'] < 0.5: return False if last_candle['volume_ratio'] < self.volume_short_multiplier.value * 0.8: return False if last_candle['rsi'] < self.rsi_sell_lower.value: return False if last_candle['breakdown_distance'] < self.breakout_threshold.value / 100: return False return True 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() # 根据突破强度调整仓位 if side == 'long': breakout_strength = last_candle['breakout_strength'] volume_confirmation = last_candle['volume_ratio'] # 突破强度因子 strength_factor = min(breakout_strength / 2.0, 1.5) # 最高1.5倍仓位 # 成交量确认因子 volume_factor = min(volume_confirmation / self.volume_multiplier.value, 1.2) # 综合仓位因子 position_factor = min(strength_factor * volume_factor, 1.3) adjusted_stake = proposed_stake * position_factor else: adjusted_stake = proposed_stake return min(max(adjusted_stake, min_stake), max_stake)