346 lines
12 KiB
Markdown
346 lines
12 KiB
Markdown
#交易 #量化交易 #策略研究 #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种高概率形态。 |