12 KiB
12 KiB
#交易 #量化交易 #策略研究 #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
关键实现技术
- 滚动窗口分析
for i in range(window, len(dataframe)):
window_data = dataframe.iloc[i-window:i]
analysis = detect_market_background(window_data)
# 将结果存入当前K线 (无未来函数)
-
多形态检测系统
- 价格通道检测：使用线性回归和偏差分析
- 头肩形态：通过极值点定位和时间顺序验证
- 三角形识别：分析高低点趋势线收敛角度
-
市场状态分类
market_state = {
'neutral': 无明确方向,
'uptrend/downtrend': 单边趋势,
'channel': 价格通道,
'reversal': 反转形态,
'consolidation': 整理形态
}
- 几何特征量化
- 趋势强度:
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)
实际应用建议
- 参数优化组合
self.pattern_window.optimize = True
self.channel_deviation.optimize = True
# Hyperopt命令
# freqtrade hyperopt --strategy ComplexPatternStrategy -e 500 --timerange 20220101-20230101
- 多时间框架验证
@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
- 可视化调试
# 策略中增加调试输出
if self.dp and self.dp.runmode.value in ('plot'):
from freqtrade.plot.plotting import store_plot_file
# 生成形态识别图表
self.plot_patterns(dataframe, metadata)
这种基于几何形态的市场背景分析特别适用于：
- 中长线趋势跟踪（通道策略）
- 重大反转点捕捉（头肩形态）
- 突破交易（三角形末端）
- 市场状态自适应策略（根据背景调整参数）
关键是要平衡形态识别的复杂度和计算效率，建议从20-30根K线的窗口开始，重点识别2-3种高概率形态。