# source: https://raw.githubusercontent.com/henrylin99/freqtrade-tutorials/05528803bacc805a08027b6cf006694a31620e20/user_data/strategies/MovingAverageCrossStrategy.py """ 移动平均线交叉趋势跟随策略 基于双移动平均线或多移动平均线的经典趋势跟随策略 """ 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 Github_henrylin99_freqtrade_tutorials__MovingAverageCrossStrategy__20251215_080324(IStrategy): """ 移动平均线交叉趋势跟随策略 策略逻辑: 1. 使用快速移动平均线和慢速移动平均线 2. 当快线上穿慢线时买入（金叉） 3. 当快线下穿慢线时卖出（死叉） 4. 结合成交量确认和趋势强度过滤 5. 使用ATR动态止损 """ INTERFACE_VERSION = 3 # 基础配置 minimal_roi = { "0": 0.15, # 15%收益立即止盈 "60": 0.08, # 1小时后8%收益 "120": 0.04, # 2小时后4%收益 "240": 0.02 # 4小时后2%收益 } stoploss = -0.08 # 8%止损 timeframe = '1h' # 1小时时间框架 # 策略控制 can_short = False startup_candle_count = 100 process_only_new_candles = True use_exit_signal = True use_custom_stoploss = False # 禁用 custom stoploss，避免 trailing stop 导致的大量亏损 # 可优化参数 # 移动平均线参数 fast_ma_period = IntParameter(5, 20, default=10, space="buy") slow_ma_period = IntParameter(20, 50, default=30, space="buy") # 移动平均线类型 ma_type_fast = IntParameter(0, 8, default=1, space="buy") # 0=SMA, 1=EMA, 2=WMA等 ma_type_slow = IntParameter(0, 8, default=1, space="buy") # 成交量确认参数 volume_factor = DecimalParameter(1.0, 3.0, default=1.5, space="buy") # 趋势强度过滤参数 adx_period = IntParameter(10, 20, default=14, space="buy") adx_threshold = IntParameter(15, 30, default=18, space="buy") # 止损参数 atr_period = IntParameter(10, 20, default=14, space="sell") atr_multiplier = DecimalParameter(1.5, 3.0, default=2.0, space="sell") # 趋势确认参数 trend_ema_period = IntParameter(50, 100, default=75, space="buy") def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """ 计算技术指标 """ # 移动平均线 dataframe['ma_fast'] = ta.MA(dataframe, timeperiod=self.fast_ma_period.value, matype=self.ma_type_fast.value) dataframe['ma_slow'] = ta.MA(dataframe, timeperiod=self.slow_ma_period.value, matype=self.ma_type_slow.value) # 移动平均线距离 dataframe['ma_distance'] = (dataframe['ma_fast'] - dataframe['ma_slow']) / dataframe['ma_slow'] dataframe['ma_distance_abs'] = abs(dataframe['ma_distance']) # 移动平均线斜率 dataframe['ma_fast_slope'] = (dataframe['ma_fast'] - dataframe['ma_fast'].shift(3)) / dataframe['ma_fast'].shift(3) dataframe['ma_slow_slope'] = (dataframe['ma_slow'] - dataframe['ma_slow'].shift(5)) / dataframe['ma_slow'].shift(5) # 趋势确认EMA dataframe['trend_ema'] = ta.EMA(dataframe, timeperiod=self.trend_ema_period.value) # 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) # ATR波动率 dataframe['atr'] = ta.ATR(dataframe, timeperiod=self.atr_period.value) dataframe['atr_percent'] = dataframe['atr'] / dataframe['close'] # 成交量指标 dataframe['volume_sma'] = dataframe['volume'].rolling(window=20).mean() dataframe['volume_ratio'] = dataframe['volume'] / dataframe['volume_sma'] # MACD确认 macd = ta.MACD(dataframe) dataframe['macd'] = macd['macd'] dataframe['macd_signal'] = macd['macdsignal'] dataframe['macd_hist'] = macd['macdhist'] # RSI过滤 dataframe['rsi'] = ta.RSI(dataframe, timeperiod=14) # 布林带 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_above_fast_ma'] = dataframe['close'] > dataframe['ma_fast'] dataframe['price_above_slow_ma'] = dataframe['close'] > dataframe['ma_slow'] dataframe['price_above_trend_ema'] = dataframe['close'] > dataframe['trend_ema'] # 移动平均线交叉信号 dataframe['ma_cross_up'] = qtpylib.crossed_above(dataframe['ma_fast'], dataframe['ma_slow']) dataframe['ma_cross_down'] = qtpylib.crossed_below(dataframe['ma_fast'], dataframe['ma_slow']) return dataframe def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """ 定义买入条件 - 趋势跟随策略 """ # 简化条件：金叉 + 趋势确认 + DI确认 dataframe.loc[ ( dataframe['ma_cross_up'] & # 金叉信号 (dataframe['close'] > dataframe['trend_ema']) & # 价格在长期趋势之上 (dataframe['di_plus'] > dataframe['di_minus']) & # DI确认上升趋势 (dataframe['rsi'] < 80) # RSI不能过度超买 ), 'enter_long' ] = 1 return dataframe def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """ 定义卖出条件 """ conditions = [ # 主要信号：快线下穿慢线（死叉） dataframe['ma_cross_down'], # 趋势转弱信号 dataframe['adx'] < 20, # 移动平均线开始走平或下降 dataframe['ma_fast_slope'] < -0.001, # DI转向 dataframe['di_minus'] > dataframe['di_plus'], # MACD转弱 dataframe['macd'] < dataframe['macd_signal'], # RSI超买 dataframe['rsi'] > 75, # 价格跌破趋势线 qtpylib.crossed_below(dataframe['close'], dataframe['trend_ema']), ] dataframe.loc[ ( # 主要退出：死叉 + ADX弱化（需要2个条件同时满足） (conditions[0] & conditions[1]) | # 或者：趋势明确转向（快线下降 + MACD转弱） (conditions[2] & conditions[4]) | # 或者：明确跌破趋势线 + DI转向 (conditions[6] & conditions[3]) ), 'exit_long' ] = 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) last_candle = dataframe.iloc[-1].squeeze() # ATR动态止损 atr_distance = last_candle['atr'] * self.atr_multiplier.value atr_stop_distance = atr_distance / current_rate # 趋势跟随止损策略 if current_profit > 0.08: # 盈利超过8% # 紧跟快速移动平均线 ma_fast_distance = abs(current_rate - last_candle['ma_fast']) / current_rate return max(-ma_fast_distance * 1.5, -0.02) elif current_profit > 0.04: # 盈利超过4% # 中等紧密止损 return max(-atr_stop_distance * 0.7, -0.03) elif current_profit > 0.02: # 盈利超过2% return max(-atr_stop_distance * 0.8, -0.04) else: # 正常ATR止损，但更宽松适合趋势跟随 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) last_candle = dataframe.iloc[-1].squeeze() # 趋势反转退出 if (last_candle['ma_fast'] < last_candle['ma_slow'] and last_candle['adx'] < 20): return "trend_reversal" # 动量衰减退出 if (last_candle['macd_hist'] < 0 and last_candle['rsi'] < 40 and current_profit > 0.02): return "momentum_weakness" # 时间止损（趋势策略允许更长持仓） trade_duration = (current_time - trade.open_date_utc).total_seconds() / 3600 if trade_duration > 72 and current_profit < 0.01: # 72小时无显著盈利 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) < 1: return False last_candle = dataframe.iloc[-1].squeeze() # 最终确认检查 # 确保趋势足够强劲 if last_candle['adx'] < self.adx_threshold.value: return False # 确保移动平均线排列正确 if not (last_candle['ma_fast'] > last_candle['ma_slow'] > last_candle['trend_ema']): return False # 确保不是假突破 if last_candle['ma_distance'] < 0.003: # 距离太小可能是假信号 return False # 波动率检查 if last_candle['atr_percent'] > 0.06: # 波动率过大 return False return True