# source: https://raw.githubusercontent.com/henrylin99/freqtrade-tutorials/691c3799846fc7ff6173a3b9241ce5a00b346141/user_data/strategies/GridTradingStrategy.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__GridTradingStrategy__20251212_023733(IStrategy): """ 网格交易策略 - 基于价格区间的多次买卖 策略逻辑: 1. 确定价格区间的上下边界 2. 在区间内设置多个网格线 3. 价格下跌时分批买入（网格买入） 4. 价格上涨时分批卖出（网格卖出） 5. 动态调整网格区间 6. 结合趋势过滤避免单边市场 """ INTERFACE_VERSION = 3 # 基础策略参数 minimal_roi = { "0": 0.15, # 15%收益立即止盈 "60": 0.08, # 1小时后8%收益 "120": 0.05, # 2小时后5%收益 "240": 0.03, # 4小时后3%收益 "480": 0.01 # 8小时后1%收益 } stoploss = -0.08 # 8%止损 timeframe = '15m' # 15分钟时间框架 # 策略控制参数 can_short = False startup_candle_count = 100 process_only_new_candles = True use_exit_signal = True use_custom_stoploss = True # 网格参数 grid_levels = IntParameter(3, 8, default=5, space="buy") # 网格层数 grid_range_pct = DecimalParameter(0.05, 0.20, default=0.10, space="buy") # 网格区间百分比 grid_step_pct = DecimalParameter(0.01, 0.04, default=0.02, space="buy") # 网格步长百分比 # 趋势过滤参数 trend_sma_period = IntParameter(50, 100, default=80, space="buy") trend_filter_enabled = True # 价格区间识别参数 range_lookback = IntParameter(20, 50, default=30, space="buy") range_volatility_threshold = DecimalParameter(0.15, 0.35, default=0.25, space="buy") # 成交量确认参数 volume_sma_period = IntParameter(15, 30, default=20, space="buy") volume_threshold = DecimalParameter(0.8, 1.5, default=1.0, space="buy") # RSI参数（辅助过滤） rsi_period = IntParameter(10, 20, default=14, space="buy") rsi_oversold = IntParameter(25, 40, default=30, space="buy") rsi_overbought = IntParameter(65, 80, default=70, space="sell") # ATR参数（动态调整） atr_period = IntParameter(10, 20, default=14, space="buy") def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """ 计算技术指标 """ # 基础移动平均线 dataframe['sma_20'] = ta.SMA(dataframe, timeperiod=20) dataframe['sma_50'] = ta.SMA(dataframe, timeperiod=50) dataframe['ema_20'] = ta.EMA(dataframe, timeperiod=20) dataframe['trend_sma'] = ta.SMA(dataframe, timeperiod=self.trend_sma_period.value) # 计算价格区间 dataframe['price_max'] = dataframe['high'].rolling(window=self.range_lookback.value).max() dataframe['price_min'] = dataframe['low'].rolling(window=self.range_lookback.value).min() dataframe['price_range'] = dataframe['price_max'] - dataframe['price_min'] dataframe['price_mid'] = (dataframe['price_max'] + dataframe['price_min']) / 2 # 价格在区间中的位置（0-1） dataframe['price_position'] = (dataframe['close'] - dataframe['price_min']) / dataframe['price_range'] # 波动率指标 dataframe['volatility'] = dataframe['price_range'] / dataframe['price_mid'] dataframe['atr'] = ta.ATR(dataframe, timeperiod=self.atr_period.value) dataframe['atr_pct'] = dataframe['atr'] / dataframe['close'] # RSI指标 dataframe['rsi'] = ta.RSI(dataframe, timeperiod=self.rsi_period.value) # 成交量指标 dataframe['volume_sma'] = dataframe['volume'].rolling(window=self.volume_sma_period.value).mean() dataframe['volume_ratio'] = dataframe['volume'] / dataframe['volume_sma'] # 趋势指标 dataframe['trend_direction'] = np.where( dataframe['close'] > dataframe['trend_sma'], 1, np.where(dataframe['close'] < dataframe['trend_sma'], -1, 0) ) # 价格相对趋势线的位置 dataframe['price_vs_trend'] = (dataframe['close'] - dataframe['trend_sma']) / dataframe['trend_sma'] # 计算网格线 self._calculate_grid_lines(dataframe) # MACD（趋势确认） macd = ta.MACD(dataframe) dataframe['macd'] = macd['macd'] dataframe['macd_signal'] = macd['macdsignal'] dataframe['macd_hist'] = macd['macdhist'] # 布林带（区间识别辅助） bollinger = qtpylib.bollinger_bands(dataframe['close'], window=20, stds=2) dataframe['bb_lower'] = bollinger['lower'] dataframe['bb_middle'] = bollinger['mid'] dataframe['bb_upper'] = bollinger['upper'] dataframe['bb_width'] = (dataframe['bb_upper'] - dataframe['bb_lower']) / dataframe['bb_middle'] # 价格动量 dataframe['momentum'] = ta.MOM(dataframe, timeperiod=10) dataframe['roc'] = ta.ROC(dataframe, timeperiod=10) return dataframe def _calculate_grid_lines(self, dataframe: DataFrame): """ 计算网格线位置 """ # 基于当前价格区间计算网格 grid_bottom = dataframe['price_min'] * (1 + self.grid_step_pct.value) grid_top = dataframe['price_max'] * (1 - self.grid_step_pct.value) grid_range = grid_top - grid_bottom # 计算各个网格线 for i in range(self.grid_levels.value): level_pct = i / (self.grid_levels.value - 1) # 0到1之间 grid_price = grid_bottom + (grid_range * level_pct) dataframe[f'grid_level_{i}'] = grid_price # 当前价格最接近的网格线 current_price = dataframe['close'] grid_distances = [] for i in range(self.grid_levels.value): distance = abs(current_price - dataframe[f'grid_level_{i}']) grid_distances.append(distance) # 找到最近的网格线索引 if len(grid_distances) > 0: dataframe['nearest_grid'] = np.argmin(np.column_stack(grid_distances), axis=1) dataframe['nearest_grid_price'] = dataframe.apply( lambda row: row[f'grid_level_{int(row["nearest_grid"])}'] if not np.isnan(row['nearest_grid']) else row['close'], axis=1 ) dataframe['distance_to_grid'] = abs(dataframe['close'] - dataframe['nearest_grid_price']) dataframe['distance_to_grid_pct'] = dataframe['distance_to_grid'] / dataframe['close'] def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """ 网格买入信号 买入条件： 1. 价格接近网格线下方区域 2. 处于震荡区间（非强烈趋势） 3. RSI不过度超卖 4. 成交量确认 5. 价格位置在区间下半部分 """ # 基础网格条件 grid_conditions = [ # 价格在区间下半部分 dataframe['price_position'] < 0.6, # 价格接近网格线（买入时机） dataframe['distance_to_grid_pct'] < self.grid_step_pct.value * 0.5, # 波动率适中（震荡市场） dataframe['volatility'] < self.range_volatility_threshold.value, dataframe['volatility'] > 0.05, # 最小波动率避免过于平静 # ATR相对稳定 dataframe['atr_pct'] < 0.03, # 3%以下 ] # 趋势过滤条件 trend_conditions = [ # 不在强烈下跌趋势中 dataframe['close'] > dataframe['trend_sma'] * 0.95, # MACD不过度看跌 dataframe['macd'] > dataframe['macd'].rolling(10).min() * 1.2, # 价格相对趋势线不过度偏离 dataframe['price_vs_trend'] > -0.1, ] # RSI和动量条件 momentum_conditions = [ # RSI不过度超卖 dataframe['rsi'] > self.rsi_oversold.value, dataframe['rsi'] < 50, # 但也不能太强 # 动量开始企稳 dataframe['momentum'] > dataframe['momentum'].shift(1), # ROC不过度负值 dataframe['roc'] > -5, ] # 成交量条件 volume_conditions = [ # 成交量确认 dataframe['volume_ratio'] > self.volume_threshold.value, # 成交量不过度放大（避免恐慌性抛售） dataframe['volume_ratio'] < 3.0, ] # 价格位置条件（网格买入逻辑） position_conditions = [ # 价格靠近区间下方或网格线 (dataframe['price_position'] < 0.4) | # 在区间下40% (dataframe['close'] < dataframe['bb_lower'] * 1.02), # 接近布林带下轨 # 价格刚刚跌破某个网格线 (dataframe['close'] < dataframe['sma_20']) & (dataframe['close'].shift(1) >= dataframe['sma_20'].shift(1)), ] # 组合所有条件 dataframe.loc[ ( # 网格基础条件 grid_conditions[0] & # 价格位置 grid_conditions[1] & # 接近网格线 grid_conditions[2] & # 波动率适中 grid_conditions[3] & # 最小波动率 grid_conditions[4] & # ATR稳定 # 趋势条件 trend_conditions[0] & # 不在强烈下跌中 trend_conditions[1] & # MACD条件 trend_conditions[2] & # 价格相对趋势线 # 动量条件 momentum_conditions[0] & # RSI条件 momentum_conditions[1] & # RSI上限 momentum_conditions[2] & # 动量企稳 momentum_conditions[3] & # ROC条件 # 成交量条件 volume_conditions[0] & # 成交量确认 volume_conditions[1] & # 成交量上限 # 位置条件 position_conditions[0] # 价格位置 ), 'enter_long' ] = 1 return dataframe def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """ 网格卖出信号 卖出条件： 1. 价格接近网格线上方区域 2. RSI进入超买区域 3. 价格位置在区间上半部分 4. 获利达到网格步长目标 """ # 基础网格卖出条件 grid_exit_conditions = [ # 价格在区间上半部分 dataframe['price_position'] > 0.4, # 价格接近网格线上方 (dataframe['close'] > dataframe['sma_20'] * 1.01) | (dataframe['close'] > dataframe['bb_upper'] * 0.99), # 价格上涨了一定幅度 dataframe['close'] > dataframe['close'].rolling(5).min() * (1 + self.grid_step_pct.value), ] # RSI和动量卖出条件 momentum_exit_conditions = [ # RSI进入超买或回落 (dataframe['rsi'] > self.rsi_overbought.value) | ((dataframe['rsi'] > 60) & (dataframe['rsi'] < dataframe['rsi'].shift(1))), # 动量转弱 dataframe['momentum'] < dataframe['momentum'].shift(1), # ROC开始下降 dataframe['roc'] < dataframe['roc'].shift(1), ] # 趋势转弱条件 trend_exit_conditions = [ # MACD转弱 qtpylib.crossed_below(dataframe['macd'], dataframe['macd_signal']), # 价格从高位回落 dataframe['close'] < dataframe['close'].rolling(3).max() * 0.995, # 短期均线开始走平 dataframe['ema_20'] < dataframe['ema_20'].shift(2), ] # 获利了结条件 profit_taking_conditions = [ # 价格位置在区间上方 dataframe['price_position'] > 0.7, # 接近区间顶部 dataframe['close'] > dataframe['price_max'] * 0.95, # 布林带位置 dataframe['close'] > dataframe['bb_upper'], ] # 组合卖出条件 dataframe.loc[ ( # 主要网格卖出条件 (grid_exit_conditions[0] & grid_exit_conditions[1] & grid_exit_conditions[2]) | # 动量转弱卖出 (momentum_exit_conditions[0] & momentum_exit_conditions[1]) | # 趋势转弱卖出 trend_exit_conditions[0] | # 获利了结 (profit_taking_conditions[0] & profit_taking_conditions[1]) | profit_taking_conditions[2] ), 'exit_long' ] = 1 return dataframe def custom_stoploss(self, pair: str, trade: '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_stop = last_candle['atr'] * 2.0 atr_stop_pct = atr_stop / current_rate # 网格策略特殊止损逻辑 if current_profit > self.grid_step_pct.value: # 盈利超过一个网格步长 # 移动止损到盈亏平衡点 return max(-atr_stop_pct * 0.3, -0.01) elif current_profit > self.grid_step_pct.value * 0.5: # 盈利超过半个网格步长 # 适度收紧止损 return max(-atr_stop_pct * 0.6, -0.04) else: # 正常止损，但考虑网格特性 grid_stop = self.grid_step_pct.value * 1.5 # 网格步长的1.5倍作为止损 return max(-atr_stop_pct, -grid_stop, self.stoploss) 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['volatility'] > 0.4: # 波动率过高 return False # 确保有足够的历史数据计算网格 if len(dataframe) < self.range_lookback.value: return False # 确保价格区间有效 if last_candle['price_range'] < last_candle['close'] * 0.02: # 区间太小 return False # 确保不在强烈单边趋势中 if abs(last_candle['price_vs_trend']) > 0.15: # 偏离趋势线超过15% return False return True def custom_entry_price(self, pair: str, current_time, proposed_rate: float, entry_tag: str, side: str, **kwargs) -> float: """ 网格交易自定义入场价格 """ dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe) last_candle = dataframe.iloc[-1].squeeze() # 尝试在更接近网格线的位置入场 if 'nearest_grid_price' in last_candle: nearest_grid = last_candle['nearest_grid_price'] if nearest_grid < proposed_rate: # 如果网格线在当前价格下方，尝试在网格线附近入场 target_price = nearest_grid * 1.001 # 略高于网格线 return min(target_price, proposed_rate * 0.999) return proposed_rate * 0.9995 # 默认小幅优化入场价格 def informative_pairs(self): """ 网格交易所需的额外数据 """ pairs = self.dp.current_whitelist() informative_pairs = [] # 添加更长时间框架用于趋势判断 for pair in pairs: informative_pairs.append((pair, '1h')) informative_pairs.append((pair, '4h')) return informative_pairs def leverage(self, pair: str, current_time, current_rate: float, proposed_leverage: float, max_leverage: float, entry_tag: str, side: str, **kwargs) -> float: """ 网格策略杠杆设置 """ # 网格策略可以使用适度杠杆，因为有多次交易分散风险 return min(2.0, max_leverage) # 最大2倍杠杆 def test_strategy(): """ 测试网格交易策略 """ import pandas as pd import numpy as np print("测试网格交易策略...") # 创建测试数据 - 模拟震荡行情 np.random.seed(42) dates = pd.date_range('2023-01-01', periods=300, freq='15min') # 模拟有网格特征的价格数据（震荡 + 少量趋势） price = 50000 # BTC起始价格 prices = [price] trend = 0.0001 # 轻微上升趋势 for i in range(299): # 添加震荡特性：在一定范围内波动 base_price = 50000 * (1 + trend * i) # 基础趋势价格 noise = np.sin(i * 0.1) * 1000 + np.random.normal(0, 500) # 震荡 + 随机噪音 price = base_price + noise price = max(price, base_price * 0.9) # 限制下跌幅度 price = min(price, base_price * 1.1) # 限制上涨幅度 prices.append(price) # 生成测试数据 highs = [p * (1 + abs(np.random.normal(0, 0.003))) for p in prices] lows = [p * (1 - abs(np.random.normal(0, 0.003))) for p in prices] data = pd.DataFrame({ 'timestamp': dates, 'open': prices, 'high': highs, 'low': lows, 'close': prices, 'volume': np.random.randint(500, 2000, 300) }) # 测试策略 strategy = Github_henrylin99_freqtrade_tutorials__GridTradingStrategy__20251212_023733() # 计算指标 data_with_indicators = strategy.populate_indicators(data, {'pair': 'BTC/USDT'}) # 生成信号 data_with_signals = strategy.populate_entry_trend(data_with_indicators, {'pair': 'BTC/USDT'}) data_with_signals = strategy.populate_exit_trend(data_with_signals, {'pair': 'BTC/USDT'}) # 统计信号 buy_signals = data_with_signals['enter_long'].sum() if 'enter_long' in data_with_signals.columns else 0 sell_signals = data_with_signals['exit_long'].sum() if 'exit_long' in data_with_signals.columns else 0 print(f"测试结果:") print(f"数据点数: {len(data)}") print(f"买入信号: {buy_signals}") print(f"卖出信号: {sell_signals}") print(f"价格范围: {min(prices):.2f} - {max(prices):.2f}") print(f"平均波动率: {data_with_signals['volatility'].mean():.4f}") # 显示一些关键指标 print(f"\n关键指标示例（最后5行）:") key_columns = ['close', 'price_position', 'volatility', 'rsi', 'nearest_grid_price'] available_columns = [col for col in key_columns if col in data_with_signals.columns] if available_columns: print(data_with_signals[available_columns].tail()) # 显示网格信息 print(f"\n网格层数: {strategy.grid_levels.value}") print(f"网格区间百分比: {strategy.grid_range_pct.value:.2%}") print(f"网格步长百分比: {strategy.grid_step_pct.value:.2%}") return data_with_signals if __name__ == "__main__": test_strategy()