#交易 #量化交易 #策略研究 #freqtrade #DeepSeek ## 复杂形态识别在 Freqtrade 中的实现 是的，Freqtrade 完全支持自定义复杂形态识别，即使是分析 10-50 根K线的形态组合。下面是一个完整的实现方案，展示如何识别"上升通道"、"头肩顶"等复杂形态： ### 核心实现方案 ``` import numpy as np import pandas as pd from scipy.stats import linregress from freqtrade.strategy import IStrategy, DecimalParameter, IntParameter class ComplexPatternStrategy(IStrategy): timeframe = '4h' # 推荐4小时以上时间框架 minimal_roi = {"0": 0.15, "240": 0.05, "480": 0} stoploss = -0.1 # 形态检测参数 pattern_window = IntParameter(10, 50, default=30, space='buy', optimize=True) channel_deviation = DecimalParameter(0.01, 0.1, default=0.03, space='buy', optimize=True) def detect_market_background(self, dataframe: pd.DataFrame, window: int) -> dict: """ 分析市场背景：识别趋势、通道、反转形态等 返回包含市场状态和形态特征的字典 """ df = dataframe.copy().tail(window) results = { 'market_state': 'neutral', 'pattern': None, 'trend_strength': 0, 'volatility': df['close'].pct_change().std(), 'key_levels': [] } # 1. 趋势强度分析 slope, intercept, _, _, _ = linregress(np.arange(len(df)), df['close']) results['trend_strength'] = slope * len(df) / df['close'].mean() # 2. 通道识别 upper_channel = self.detect_price_channel(df, 'high', deviation=self.channel_deviation.value) lower_channel = self.detect_price_channel(df, 'low', deviation=self.channel_deviation.value) if upper_channel and lower_channel: results['market_state'] = 'channel' results['pattern'] = 'price_channel' results['key_levels'] = [upper_channel[-1], lower_channel[-1]] # 3. 头肩形态识别 elif head_shoulder := self.detect_head_shoulder(df): results['market_state'] = 'reversal' results['pattern'] = 'head_shoulder' if head_shoulder['type'] == 'top' else 'inverse_head_shoulder' results['key_levels'] = [head_shoulder['neckline']] # 4. 三角形形态识别 elif triangle := self.detect_triangle(df): results['market_state'] = 'consolidation' results['pattern'] = triangle['type'] results['key_levels'] = [triangle['upper'], triangle['lower']] # 5. 趋势状态判断 elif abs(results['trend_strength']) > 0.05: results['market_state'] = 'uptrend' if slope > 0 else 'downtrend' return results def detect_price_channel(self, df: pd.DataFrame, price_col: str, deviation=0.03) -> list: """ 检测价格通道 deviation: 允许的偏离度 """ prices = df[price_col].values # 线性回归通道 x = np.arange(len(prices)) slope, intercept, _, _, _ = linregress(x, prices) regression_line = intercept + slope * x # 检查价格是否在通道内 deviations = np.abs(prices - regression_line) if np.max(deviations) / np.mean(prices) < deviation: return regression_line.tolist() return None def detect_head_shoulder(self, df: pd.DataFrame) -> dict: """ 检测头肩顶/底形态 返回形态特征字典 """ # 寻找关键转折点 (使用最高价/最低价) highs = df['high'].values lows = df['low'].values # 寻找局部极值点 max_idx = np.argsort(highs)[-5:] # 前5个高点 min_idx = np.argsort(lows)[:5] # 前5个低点 # 排序索引以确保时间顺序 max_idx.sort() min_idx.sort() # 头肩顶模式检查 if len(max_idx) >= 5: # 典型头肩顶: 左肩-头-右肩 形态 if (max_idx[1] > max_idx[0] and max_idx[2] > max_idx[1] and max_idx[3] < max_idx[2] and max_idx[4] < max_idx[3]): neckline = np.mean([lows[max_idx[1]], lows[max_idx[3]]]) return { 'type': 'top', 'left_shoulder': max_idx[0], 'head': max_idx[2], 'right_shoulder': max_idx[4], 'neckline': neckline } # 头肩底模式检查 if len(min_idx) >= 5: # 典型头肩底: 左肩-头-右肩 形态 if (min_idx[1] < min_idx[0] and min_idx[2] < min_idx[1] and min_idx[3] > min_idx[2] and min_idx[4] > min_idx[3]): neckline = np.mean([highs[min_idx[1]], highs[min_idx[3]]]) return { 'type': 'bottom', 'left_shoulder': min_idx[0], 'head': min_idx[2], 'right_shoulder': min_idx[4], 'neckline': neckline } return None def detect_triangle(self, df: pd.DataFrame) -> dict: """ 检测三角形整理形态 """ # 计算收敛趋势线 highs = df['high'].values lows = df['low'].values # 高点下降趋势线 high_slope, high_intercept, _, _, _ = linregress(np.arange(len(highs)), highs) high_line = high_intercept + high_slope * np.arange(len(highs)) # 低点上升趋势线 low_slope, low_intercept, _, _, _ = linregress(np.arange(len(lows)), lows) low_line = low_intercept + low_slope * np.arange(len(lows)) # 确定三角形类型 triangle_type = None if high_slope < 0 and low_slope > 0: triangle_type = 'symmetrical' elif high_slope < 0 and abs(low_slope) < 0.001: triangle_type = 'descending' elif low_slope > 0 and abs(high_slope) < 0.001: triangle_type = 'ascending' # 验证价格在趋势线范围内 if triangle_type and np.all(highs <= high_line * 1.02) and np.all(lows >= low_line * 0.98): return { 'type': triangle_type, 'upper': high_line[-1], 'lower': low_line[-1], 'apex_index': int(-high_intercept / high_slope) # 计算交汇点 } return None def populate_indicators(self, dataframe: pd.DataFrame, metadata: dict) -> pd.DataFrame: # 每根K线分析前N根K线形成的市场背景 window = self.pattern_window.value # 初始化新列 dataframe['market_state'] = 'neutral' dataframe['pattern_type'] = '' dataframe['trend_strength'] = 0.0 # 使用rolling窗口分析形态 (避免未来函数) for i in range(window, len(dataframe)): window_data = dataframe.iloc[i-window:i] analysis = self.detect_market_background(window_data, window) # 将分析结果存入当前K线 dataframe.at[dataframe.index[i], 'market_state'] = analysis['market_state'] dataframe.at[dataframe.index[i], 'pattern_type'] = analysis.get('pattern', '') dataframe.at[dataframe.index[i], 'trend_strength'] = analysis['trend_strength'] return dataframe def populate_entry_trend(self, dataframe: pd.DataFrame, metadata: dict) -> pd.DataFrame: # 通道下沿买入策略 dataframe.loc[ (dataframe['market_state'] == 'channel') & (dataframe['close'] < dataframe['close'].rolling(5).mean() * 1.01) & # 接近通道底部 (dataframe['volume'] > dataframe['volume'].rolling(20).mean()), 'enter_long'] = 1 # 头肩底突破策略 dataframe.loc[ (dataframe['pattern_type'] == 'inverse_head_shoulder') & (dataframe['close'] > dataframe['close'].shift(1) * 1.02), # 突破颈线 'enter_long'] = 1 return dataframe def populate_exit_trend(self, dataframe: pd.DataFrame, metadata: dict) -> pd.DataFrame: # 通道上沿卖出 dataframe.loc[ (dataframe['market_state'] == 'channel') & (dataframe['close'] > dataframe['close'].rolling(5).mean() * 0.99), # 接近通道顶部 'exit_long'] = 1 # 趋势逆转退出 dataframe.loc[ (dataframe['trend_strength'] < -0.03) & # 趋势转弱 (dataframe['market_state'].shift(1) == 'uptrend'), # 前一状态是上涨 'exit_long'] = 1 return dataframe ``` ### 关键实现技术 1. ​**滚动窗口分析**​ ``` for i in range(window, len(dataframe)): window_data = dataframe.iloc[i-window:i] analysis = detect_market_background(window_data) # 将结果存入当前K线 (无未来函数) ``` 2. ​**多形态检测系统**​ - 价格通道检测：使用线性回归和偏差分析 - 头肩形态：通过极值点定位和时间顺序验证 - 三角形识别：分析高低点趋势线收敛角度 3. ​**市场状态分类**​ ``` market_state = { 'neutral': 无明确方向, 'uptrend/downtrend': 单边趋势, 'channel': 价格通道, 'reversal': 反转形态, 'consolidation': 整理形态 } ``` 4. ​**几何特征量化**​ - 趋势强度: `slope * len(df) / mean_price` - 波动率: `df['close'].pct_change().std()` - 关键水平: 颈线、通道边界、三角形顶点 ### 高级技巧：动态形态检测 对于更复杂的形态，可以结合机器学习： ``` from sklearn.ensemble import IsolationForest def detect_anomaly_patterns(df): """使用异常检测识别非常规形态""" features = df[['close', 'volume', 'range']].pct_change().dropna() model = IsolationForest(contamination=0.05) anomalies = model.fit_predict(features) if np.sum(anomalies == -1) > 3: # 至少3个异常点 # 分析异常点的几何分布 anomaly_idx = np.where(anomalies == -1)[0] if is_cluster(anomaly_idx): # 检查是否形成集群 return 'cluster_anomaly' return None def is_cluster(indices, max_gap=3): """检查异常点是否形成集群""" gaps = np.diff(indices) return np.all(gaps <= max_gap) ``` ### 性能优化方案 处理长窗口时需考虑效率： ``` def populate_indicators(self, dataframe): # 仅对最近6个月数据做全量分析 if len(dataframe) > 4320: # 6个月(4h) dataframe = dataframe.tail(4320) # 每10根K线分析一次 for i in range(window, len(dataframe), 10): # 分析逻辑... # 填充空白 dataframe['market_state'].fillna(method='ffill', inplace=True) ``` ### 实际应用建议 1. ​**参数优化组合**​ ``` self.pattern_window.optimize = True self.channel_deviation.optimize = True # Hyperopt命令 # freqtrade hyperopt --strategy ComplexPatternStrategy -e 500 --timerange 20220101-20230101 ``` 2. ​**多时间框架验证**​ ``` @informative('1d', name='daily') def populate_daily(self, df, metadata): df['daily_state'] = ... # 日线形态分析 return df # 在4小时策略中: def populate_entry_trend(self, df): df.loc[ (df['market_state'] == 'uptrend') & (df['daily_state'] == 'channel') & # 日线在通道中 ..., 'enter_long'] = 1 ``` 3. ​**可视化调试**​ ``` # 策略中增加调试输出 if self.dp and self.dp.runmode.value in ('plot'): from freqtrade.plot.plotting import store_plot_file # 生成形态识别图表 self.plot_patterns(dataframe, metadata) ``` 这种基于几何形态的市场背景分析特别适用于： 1. 中长线趋势跟踪（通道策略） 2. 重大反转点捕捉（头肩形态） 3. 突破交易（三角形末端） 4. 市场状态自适应策略（根据背景调整参数） 关键是要平衡形态识别的复杂度和计算效率，建议从20-30根K线的窗口开始，重点识别2-3种高概率形态。