# source: https://raw.githubusercontent.com/Cailiang/an-autumn-leaf/ddf8cbc3840cc7dfb9e2a370a4a5a465560fa534/freqtrade_strategies.py """ Freqtrade策略文件 - 包含两个策略 1. Github_Cailiang_an_autumn_leaf__freqtrade_strategies__20251127_081901 - 保守型做多策略 2. Github_Cailiang_an_autumn_leaf__freqtrade_strategies__20251127_081901 - RSI超卖反弹策略 使用方法： freqtrade backtesting --strategy Github_Cailiang_an_autumn_leaf__freqtrade_strategies__20251127_081901 ... freqtrade backtesting --strategy Github_Cailiang_an_autumn_leaf__freqtrade_strategies__20251127_081901 ... """ import talib.abstract as ta import pandas as pd from pandas import DataFrame from freqtrade.strategy import IStrategy, DecimalParameter, IntParameter from datetime import datetime, timedelta # ============================================================ # 策略1: Github_Cailiang_an_autumn_leaf__freqtrade_strategies__20251127_081901 - 保守型做多策略 # ============================================================ class Github_Cailiang_an_autumn_leaf__freqtrade_strategies__20251127_081901(IStrategy): """ 保守型做多策略，适合大市值币种（BTC/ETH等） 时间周期：4h 买入信号： - 成交量过滤：平均成交量 >= 阈值 - MACD金叉：DIF > DEA - KDJ金叉：K > D，且满足上限过滤 - 市场状态过滤：ATR比率、ADX、价格波动范围、RSI、StochRSI等 卖出信号： - 固定止损：5% - ATR移动止损：盈利达到1×ATR后，止损线为最高价-3×ATR """ # 策略参数 timeframe = '4h' # ROI 配置（禁用，使用移动止损） minimal_roi = { "0": 100 # 不使用ROI，完全依靠止损 } # 固定止损 stoploss = -0.05 # -5% # 关闭百分比移动止损（使用ATR移动止损） trailing_stop = False # 使用自定义止损 use_custom_stoploss = True # 参数配置 min_volume_threshold = DecimalParameter(1000, 2000, default=1500, space='buy', optimize=True) long_kdj_max_k = DecimalParameter(80, 90, default=88, space='buy', optimize=True) long_kdj_max_d = DecimalParameter(75, 85, default=83, space='buy', optimize=True) long_kdj_max_j = DecimalParameter(100, 120, default=110, space='buy', optimize=True) long_min_atr_ratio = DecimalParameter(0.5, 0.8, default=0.6, space='buy', optimize=True) long_min_adx = DecimalParameter(10, 20, default=16, space='buy', optimize=True) long_min_price_range_30 = DecimalParameter(3, 6, default=4.8, space='buy', optimize=True) long_max_price_range_30 = DecimalParameter(14, 18, default=16, space='buy', optimize=True) long_max_rsi = DecimalParameter(70, 80, default=76, space='buy', optimize=True) long_min_stochrsi = DecimalParameter(40, 60, default=52, space='buy', optimize=True) long_min_macd_dif = DecimalParameter(-300, -100, default=-200, space='buy', optimize=True) long_min_macd_dea = DecimalParameter(-500, -200, default=-400, space='buy', optimize=True) # ATR移动止损参数 atr_trailing_period = IntParameter(10, 20, default=14, space='buy', optimize=False) atr_trailing_multiplier = DecimalParameter(2.0, 4.0, default=3.0, space='sell', optimize=True) atr_trailing_trigger = DecimalParameter(0.5, 2.0, default=1.0, space='sell', optimize=True) # 盈利达到1×ATR时激活 def __init__(self, config: dict) -> None: super().__init__(config) self.entry_atr = {} # pair -> entry_atr (入场时的ATR值) def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """计算技术指标""" # MACD (12, 26, 9) macd = ta.MACD(dataframe, fastperiod=12, slowperiod=26, signalperiod=9) dataframe['macd'] = macd['macd'] dataframe['macd_signal'] = macd['macdsignal'] dataframe['macd_hist'] = macd['macdhist'] # RSI (14) dataframe['rsi'] = ta.RSI(dataframe, timeperiod=14) # Stochastic RSI stoch_rsi = ta.STOCHRSI(dataframe, timeperiod=14, fastk_period=14, fastd_period=3) dataframe['stochrsi_k'] = stoch_rsi['fastk'] dataframe['stochrsi_d'] = stoch_rsi['fastd'] # KDJ (基于Stochastic) stoch = ta.STOCH(dataframe, fastk_period=9, slowk_period=3, slowd_period=3) dataframe['kdj_k'] = stoch['slowk'] dataframe['kdj_d'] = stoch['slowd'] dataframe['kdj_j'] = 3 * dataframe['kdj_k'] - 2 * dataframe['kdj_d'] # ATR (14) dataframe['atr'] = ta.ATR(dataframe, timeperiod=14) # ATR比率 (ATR(14) / SMA(ATR, 60)) dataframe['atr_sma60'] = ta.SMA(dataframe['atr'], timeperiod=60) dataframe['atr_ratio'] = dataframe['atr'] / dataframe['atr_sma60'] # ADX (14) dataframe['adx'] = ta.ADX(dataframe, timeperiod=14) # 30根K线价格波动范围百分比 dataframe['price_high_30'] = dataframe['high'].rolling(window=30).max() dataframe['price_low_30'] = dataframe['low'].rolling(window=30).min() dataframe['price_range_30'] = ( (dataframe['price_high_30'] - dataframe['price_low_30']) / dataframe['price_low_30'] * 100 ) # EMA (40) - 用于可视化 dataframe['ema40'] = ta.EMA(dataframe, timeperiod=40) return dataframe def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """买入信号""" dataframe.loc[ ( # 成交量过滤 (dataframe['volume'] >= self.min_volume_threshold.value) & # MACD金叉 (dataframe['macd'] > dataframe['macd_signal']) & # KDJ金叉及上限 (dataframe['kdj_k'] > dataframe['kdj_d']) & (dataframe['kdj_j'] < self.long_kdj_max_j.value) & (dataframe['kdj_k'] < self.long_kdj_max_k.value) & (dataframe['kdj_d'] < self.long_kdj_max_d.value) & # ATR比率（避免震荡期） (dataframe['atr_ratio'] >= self.long_min_atr_ratio.value) & # ADX（趋势强度） (dataframe['adx'] >= self.long_min_adx.value) & # 价格波动范围 (dataframe['price_range_30'] >= self.long_min_price_range_30.value) & (dataframe['price_range_30'] <= self.long_max_price_range_30.value) & # RSI过滤 (dataframe['rsi'] <= self.long_max_rsi.value) & # StochRSI过滤 (dataframe['stochrsi_k'] >= self.long_min_stochrsi.value) & # MACD DIF/DEA过滤 (dataframe['macd'] >= self.long_min_macd_dif.value) & (dataframe['macd_signal'] >= self.long_min_macd_dea.value) & # 有效数据 (dataframe['volume'] > 0) ), 'enter_long'] = 1 return dataframe def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """卖出信号（主要依靠止损，这里可以添加其他退出条件）""" # 不设置主动退出信号，完全依靠止损 dataframe['exit_long'] = 0 return dataframe def confirm_trade_entry(self, pair: str, order_type: str, amount: float, rate: float, time_in_force: str, current_time: datetime, entry_tag: str, **kwargs) -> bool: """ 确认交易入场时，存储入场时的ATR值 注意：使用前一根已完成K线的ATR（因为入场信号基于前一根K线计算） """ dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe) if len(dataframe) >= 2: # 使用前一根已完成K线的ATR（-2位置），而不是最新K线（-1位置） # 因为入场信号是基于前一根K线的指标计算的 prev_candle = dataframe.iloc[-2].squeeze() if 'atr' in prev_candle: self.entry_atr[pair] = float(prev_candle['atr']) return True def custom_stoploss(self, pair: str, trade: 'Trade', current_time: datetime, current_rate: float, current_profit: float, **kwargs) -> float: """ 自定义止损逻辑 - ATR移动止损 逻辑： 1. 触发条件：盈利金额 >= EntryATR × atr_trailing_trigger 2. 止损价 = 最高价 - EntryATR × atr_trailing_multiplier 3. 止损只能向上调整（保护利润） Returns: float: 止损百分比（负数，相对于入场价） 例如：-0.05 = -5% 止损 """ # 如果没有存储入场ATR，使用固定止损 if pair not in self.entry_atr: return self.stoploss # -0.05 entry_atr = self.entry_atr[pair] entry_price = trade.open_rate highest_price = trade.max_rate if hasattr(trade, 'max_rate') else current_rate # 计算当前盈利金额（基于当前价格） current_profit_amount = current_rate - entry_price # ATR触发金额：EntryATR × atr_trailing_trigger atr_trigger_amount = entry_atr * self.atr_trailing_trigger.value # 检查是否达到触发条件 if current_profit_amount >= atr_trigger_amount: # 计算ATR止损价：最高价 - (EntryATR × atr_trailing_multiplier) atr_stop_price = highest_price - (entry_atr * self.atr_trailing_multiplier.value) # 计算止损百分比（相对于入场价） stop_loss_pct = (atr_stop_price - entry_price) / entry_price # 确保止损不会低于固定止损 if stop_loss_pct > self.stoploss: return stop_loss_pct # 未达到触发条件或计算出的止损低于固定止损，使用固定止损 return self.stoploss def confirm_trade_exit(self, pair: str, trade: 'Trade', order_type: str, amount: float, rate: float, time_in_force: str, exit_reason: str, **kwargs) -> bool: """确认退出时清理entry_atr记录""" if pair in self.entry_atr: del self.entry_atr[pair] return True # ============================================================ # 策略2: Github_Cailiang_an_autumn_leaf__freqtrade_strategies__20251127_081901 - RSI超卖反弹策略 # ============================================================ class Github_Cailiang_an_autumn_leaf__freqtrade_strategies__20251127_081901(IStrategy): """ RSI超卖反弹策略 时间周期：4h 买入信号： - RSI < 30 (超卖) - 额外过滤：StochRSI <= 10, KDJ K <= 35, KDJ D <= 35, -3 <= KDJ J <= 1 卖出信号： - 固定持仓3根K线（12小时） - 止损：2% """ # 策略参数 timeframe = '4h' # ROI 配置（禁用） minimal_roi = { "0": 100 } # 固定止损 stoploss = -0.02 # -2% # 禁用移动止损 trailing_stop = False # 参数配置 rsi_period = IntParameter(10, 20, default=14, space='buy', optimize=True) rsi_oversold_threshold = DecimalParameter(20, 35, default=30, space='buy', optimize=True) holding_bars = IntParameter(2, 5, default=3, space='sell', optimize=True) # 额外过滤条件 enable_filters = True stochrsi_threshold = DecimalParameter(5, 15, default=10, space='buy', optimize=True) kdj_k_threshold = DecimalParameter(25, 40, default=35, space='buy', optimize=True) kdj_d_threshold = DecimalParameter(25, 40, default=35, space='buy', optimize=True) kdj_j_max_threshold = DecimalParameter(0, 5, default=1, space='buy', optimize=True) kdj_j_min_threshold = DecimalParameter(-5, 0, default=-3, space='buy', optimize=True) # 用于追踪持仓时间 def __init__(self, config: dict) -> None: super().__init__(config) self.entry_candle_time = {} # pair -> entry candle timestamp def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """计算技术指标""" # RSI dataframe['rsi'] = ta.RSI(dataframe, timeperiod=self.rsi_period.value) # Stochastic RSI stoch_rsi = ta.STOCHRSI(dataframe, timeperiod=self.rsi_period.value, fastk_period=14, fastd_period=3) dataframe['stochrsi_k'] = stoch_rsi['fastk'] dataframe['stochrsi_d'] = stoch_rsi['fastd'] # KDJ stoch = ta.STOCH(dataframe, fastk_period=9, slowk_period=3, slowd_period=3) dataframe['kdj_k'] = stoch['slowk'] dataframe['kdj_d'] = stoch['slowd'] dataframe['kdj_j'] = 3 * dataframe['kdj_k'] - 2 * dataframe['kdj_d'] # MACD (用于记录) macd = ta.MACD(dataframe, fastperiod=12, slowperiod=26, signalperiod=9) dataframe['macd'] = macd['macd'] dataframe['macd_signal'] = macd['macdsignal'] # 标记买入信号出现的位置（前一根K线RSI超卖） dataframe['rsi_oversold_prev'] = (dataframe['rsi'].shift(1) < self.rsi_oversold_threshold.value) return dataframe def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """买入信号 - 前一根K线RSI超卖，当前K线开盘买入""" conditions = [ # 前一根K线RSI超卖 (dataframe['rsi'].shift(1) < self.rsi_oversold_threshold.value), (dataframe['rsi'].shift(1) > 0), # 确保RSI有效 ] # 额外过滤条件 if self.enable_filters: conditions.extend([ # StochRSI超卖 (dataframe['stochrsi_k'].shift(1) <= self.stochrsi_threshold.value), # KDJ超卖 (dataframe['kdj_k'].shift(1) <= self.kdj_k_threshold.value), (dataframe['kdj_d'].shift(1) <= self.kdj_d_threshold.value), (dataframe['kdj_j'].shift(1) >= self.kdj_j_min_threshold.value), (dataframe['kdj_j'].shift(1) <= self.kdj_j_max_threshold.value), ]) # 合并所有条件 dataframe.loc[ ( conditions[0] & (conditions[1] if len(conditions) > 1 else True) & (conditions[2] if len(conditions) > 2 else True) & (conditions[3] if len(conditions) > 3 else True) & (conditions[4] if len(conditions) > 4 else True) & (conditions[5] if len(conditions) > 5 else True) & (conditions[6] if len(conditions) > 6 else True) & (dataframe['volume'] > 0) ), 'enter_long'] = 1 return dataframe def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """卖出信号 - 持仓固定K线数后卖出（通过custom_exit处理）""" # 不在这里设置退出信号，使用 custom_exit_long dataframe['exit_long'] = 0 return dataframe def custom_exit(self, pair: str, trade: 'Trade', current_time: datetime, current_rate: float, current_profit: float, **kwargs) -> bool: """ 自定义退出逻辑 - 持仓固定K线数后卖出 Returns: bool: True表示退出 """ # 记录入场时的K线时间 if pair not in self.entry_candle_time: self.entry_candle_time[pair] = trade.open_date_utc # 计算持仓时间（按4h K线数量） time_diff = current_time - self.entry_candle_time[pair] hours_held = time_diff.total_seconds() / 3600 bars_held = int(hours_held / 4) # 4h周期 # 持仓达到目标K线数后退出 if bars_held >= self.holding_bars.value: return True return False def confirm_trade_exit(self, pair: str, trade: 'Trade', order_type: str, amount: float, rate: float, time_in_force: str, exit_reason: str, **kwargs) -> bool: """确认退出时清理记录""" if pair in self.entry_candle_time: del self.entry_candle_time[pair] return True # ============================================================ # 策略3: Github_Cailiang_an_autumn_leaf__freqtrade_strategies__20251127_081901 - 组合策略（可选） # ============================================================ class Github_Cailiang_an_autumn_leaf__freqtrade_strategies__20251127_081901(IStrategy): """ 组合策略 - 同时使用UnifiedLong和RSIOversold的信号 买入信号：满足任一策略的买入条件 卖出信号：根据入场策略选择对应的退出条件 注意：此策略仅作为示例，实际使用时需要仔细测试和调优 """ timeframe = '4h' minimal_roi = { "0": 100 } stoploss = -0.05 trailing_stop = False def __init__(self, config: dict) -> None: super().__init__(config) self.unified = Github_Cailiang_an_autumn_leaf__freqtrade_strategies__20251127_081901(config) self.rsi_oversold = Github_Cailiang_an_autumn_leaf__freqtrade_strategies__20251127_081901(config) self.entry_strategy = {} # pair -> strategy_name def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """计算两个策略的所有指标""" dataframe = self.unified.populate_indicators(dataframe, metadata) dataframe = self.rsi_oversold.populate_indicators(dataframe, metadata) return dataframe def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """组合买入信号""" # 计算两个策略的信号 df_unified = self.unified.populate_entry_trend(dataframe.copy(), metadata) df_rsi = self.rsi_oversold.populate_entry_trend(dataframe.copy(), metadata) # 优先级：RSI超卖策略 > Unified策略 dataframe.loc[ (df_rsi['enter_long'] == 1), 'enter_long'] = 1 dataframe.loc[ ((df_unified['enter_long'] == 1) & (dataframe['enter_long'] != 1)), 'enter_long'] = 1 return dataframe def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """组合卖出信号""" dataframe['exit_long'] = 0 return dataframe def custom_exit(self, pair: str, trade: 'Trade', current_time: datetime, current_rate: float, current_profit: float, **kwargs) -> bool: """根据入场策略选择退出逻辑""" # 这里可以根据需要实现组合退出逻辑 # 简化处理：使用RSI策略的退出条件 return self.rsi_oversold.custom_exit(pair, trade, current_time, current_rate, current_profit, **kwargs)