initial vault sync

2026-06-23 00:24:32 +08:00
commit 8e7723c17a
795 changed files with 157458 additions and 0 deletions
--- a/投资交易/freqtrade是否支持自定义K线形态的识别监测？.md
+++ b/投资交易/freqtrade是否支持自定义K线形态的识别监测？.md
@ -0,0 +1,346 @@
+#交易 #量化交易 #策略研究 #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种高概率形态。