Blame for model/classic_patterns.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""

经典K线形态识别模块

识别老鸭头、红三兵、仙人指路、倍量柱等经典形态

结合《量柱擒涨停》《涨停板战法研究合集》等经典理论

"""

import numpy as np
import pandas as pd
import warnings
warnings.filterwarnings('ignore')


class ClassicPatternDetector:
    """经典形态识别器"""

    def __init__(self):
        self.pattern_names = [
            'old_duck_head',      # 老鸭头
            'three_white_soldiers', # 红三兵
            'immortal_pointing',   # 仙人指路
            'double_volume',       # 倍量柱
            'ladder_volume',       # 梯量柱
            'golden_cross',        # 黄金交叉(均线)
        ]

    def detect_all(self, df):
        """

        识别所有经典形态，返回检测结果和分析摘要



        Args:

            df: DataFrame，需包含 open/high/low/close/volume 列，建议已设置 DatetimeIndex



        Returns:

            dict: {

                'patterns': { 形态名: { detected, signal, details... } },

                'summary': str,          # 综合分析摘要

                'score': float,          # 综合形态评分 0~100

                'detected_list': [...]   # 已检测到的形态列表

            }

        """
        df = df.copy()
        if len(df) < 10:
            return self._empty_result('数据不足，至少需要10根K线')

        # 确保有足够的列
        for col in ['open', 'high', 'low', 'close', 'volume']:
            if col not in df.columns:
                return self._empty_result(f'缺少 {col} 列')

        close = df['close'].values.astype(float)
        open_p = df['open'].values.astype(float)
        high = df['high'].values.astype(float)
        low = df['low'].values.astype(float)
        volume = df['volume'].values.astype(float)

        # 预计算常用指标
        ma5 = pd.Series(close).rolling(5).mean().values
        ma10 = pd.Series(close).rolling(10).mean().values
        ma20 = pd.Series(close).rolling(20).mean().values
        ma60 = pd.Series(close).rolling(60).mean().values
        vol_ma5 = pd.Series(volume).rolling(5).mean().values
        vol_ma20 = pd.Series(volume).rolling(20).mean().values

        results = {}

        # ---- 1. 老鸭头形态 ----
        results['old_duck_head'] = self._detect_old_duck_head(
            close, open_p, high, low, volume, ma5, ma10, ma20, ma60, vol_ma5, df
        )

        # ---- 2. 红三兵形态 ----
        results['three_white_soldiers'] = self._detect_three_white_soldiers(
            close, open_p, high, low, volume, vol_ma5, df
        )

        # ---- 3. 仙人指路形态 ----
        results['immortal_pointing'] = self._detect_immortal_pointing(
            close, open_p, high, low, volume, vol_ma5, df
        )

        # ---- 4. 倍量柱形态 ----
        results['double_volume'] = self._detect_double_volume(
            close, open_p, high, low, volume, vol_ma5, vol_ma20, df
        )

        # ---- 5. 梯量柱形态 ----
        results['ladder_volume'] = self._detect_ladder_volume(
            close, open_p, high, low, volume, vol_ma5, df
        )

        # ---- 5b. 波段高位卖点 ----
        results['distribution_top'] = self._detect_distribution_top(
            close, open_p, high, low, volume, vol_ma5, vol_ma20, df
        )

        # ---- 6. 黄金交叉 ----
        results['golden_cross'] = self._detect_golden_cross(
            close, ma5, ma10, ma20, ma60, df
        )

        # 汇总
        detected_list = []
        total_score = 0
        score_count = 0
        details_parts = []

        name_map = {
            'old_duck_head': '老鸭头',
            'three_white_soldiers': '红三兵',
            'immortal_pointing': '仙人指路',
            'double_volume': '倍量柱',
            'ladder_volume': '梯量柱',
            'golden_cross': '均线金叉',
            'distribution_top': '波段高位',
        }

        for key, val in results.items():
            if val.get('detected', False):
                item = {
                    'name': name_map.get(key, key),
                    'key': key,
                    'signal': val.get('signal', 'neutral'),
                    'description': val.get('description', ''),
                    'strength': val.get('strength', 0),
                }
                # 传递 index/date/price 用于图表标注
                for extra_key in ('index', 'date', 'price', 'vol_ratio'):
                    if extra_key in val:
                        item[extra_key] = val[extra_key]
                detected_list.append(item)
                if 'score' in val:
                    total_score += val['score']
                    score_count += 1
                details_parts.append(f"【{name_map.get(key, key)}】{val.get('description', '')}")

        avg_score = total_score / score_count if score_count > 0 else 0

        # 综合信号判断
        bull_signals = sum(1 for d in detected_list if d['signal'] == 'bullish')
        bear_signals = sum(1 for d in detected_list if d['signal'] == 'bearish')

        if bull_signals >= 2:
            overall_signal = 'bullish'
            summary = f"检测到 {len(detected_list)} 个经典形态，多方信号占优（{bull_signals}个看多），市场偏强势"
        elif bear_signals >= 2:
            overall_signal = 'bearish'
            summary = f"检测到 {len(detected_list)} 个经典形态，空方信号占优（{bear_signals}个看空），需注意风险"
        elif len(detected_list) > 0:
            overall_signal = 'neutral'
            summary = f"检测到 {len(detected_list)} 个经典形态，多空力量较为均衡"
        else:
            overall_signal = 'none'
            summary = "近期未检测到明显的经典K线形态，市场处于无明确形态状态"

        if details_parts:
            summary += "。" + "; ".join(details_parts)

        # 生成图表标注用的 markers
        buy_markers = []
        sell_markers = []
        for item in detected_list:
            if item.get('key') == 'double_volume' and 'index' in item:
                buy_markers.append({
                    'index': int(item['index']),
                    'date': item.get('date', ''),
                    'price': item.get('price', 0),
                    'label': '倍量' + ('强势' if item.get('signal') == 'bullish' else '关注'),
                    'strength': item.get('strength', 0),
                })
            if item.get('key') == 'distribution_top' and 'index' in item:
                sell_markers.append({
                    'index': int(item['index']),
                    'date': item.get('date', ''),
                    'price': item.get('price', 0),
                    'label': item.get('label', '波段高位'),
                    'strength': item.get('strength', 0),
                })

        return {
            'patterns': results,
            'summary': summary,
            'score': round(avg_score, 1),
            'overall_signal': overall_signal,
            'detected_list': detected_list,
            'name_map': name_map,
            'buy_markers': buy_markers,
            'sell_markers': sell_markers,
        }

    # ========================================================
    #  1. 老鸭头形态
    # ========================================================
    def _detect_old_duck_head(self, close, open_p, high, low, volume,

                               ma5, ma10, ma20, ma60, vol_ma5, df):
        """

        老鸭头形态识别：

        - 股价经过上涨后回调，5日均线、10日均线走平或下行（鸭脖子+鸭头）

        - 随后放量突破前高，5日均线重新上穿10日均线（鸭嘴张开）

        - 60日均线保持上行作为上升趋势的支撑

        """
        n = len(close)
        if n < 60:
            return {'detected': False, 'description': '数据不足60根K线，无法判断老鸭头'}

        lookback = min(40, n - 20)
        result = {'detected': False, 'signal': 'neutral', 'description': '', 'strength': 0}

        for i in range(n - lookback, n - 5):
            if i < 20:
                continue

            pre_start = max(0, i - 60)
            pre_end = max(0, i - 20)
            if pre_end <= pre_start:
                continue

            region_high = np.max(high[pre_start:pre_end])
            region_high_idx = pre_start + np.argmax(high[pre_start:pre_end])

            if region_high_idx < pre_start + 5:
                continue

            trough_start = region_high_idx
            trough_end = min(n, region_high_idx + 30)
            if trough_end - trough_start < 10:
                continue

            trough_min = np.min(low[trough_start:trough_end])
            trough_min_idx = trough_start + np.argmin(low[trough_start:trough_end])

            pullback_pct = (region_high - trough_min) / region_high if region_high > 0 else 0
            if pullback_pct < 0.03 or pullback_pct > 0.30:
                continue

            for j in range(max(0, n - 5), n):
                if j <= trough_min_idx:
                    continue

                curr_close = close[j]
                curr_vol = volume[j]

                if curr_close > region_high:
                    if vol_ma5[j] > 0 and curr_vol > vol_ma5[j] * 1.2:
                        if ma60[j] > ma60[max(0, j - 5)] if ma60[j] > 0 else False:
                            strength = min(100, 50 + pullback_pct * 200 + (curr_vol / vol_ma5[j] - 1) * 30)
                            result = {
                                'detected': True,
                                'signal': 'bullish',
                                'description': f'老鸭头形态：前高{region_high:.2f}回调{pullback_pct*100:.1f}%后放量突破至{curr_close:.2f}，60日均线保持上行',
                                'strength': round(min(100, strength), 1),
                                'score': round(min(100, strength), 1),
                                'prev_high': round(region_high, 2),
                                'pullback_pct': round(pullback_pct * 100, 1),
                                'breakout_price': round(curr_close, 2),
                            }
                            return result

        result['description'] = '未检测到老鸭头形态（需满足：前期上涨-回调-放量突破）'
        return result

    # ========================================================
    #  2. 红三兵形态
    # ========================================================
    def _detect_three_white_soldiers(self, close, open_p, high, low, volume, vol_ma5, df):
        """

        红三兵形态识别（标准 + 变体）：

        - 连续3根阳线（收盘>开盘）

        - 每根阳线的开盘价在前一根实体之内或之上

        - 收盘价逐步抬高

        - 可选：伴随温和放量

        """
        n = len(close)
        result = {'detected': False, 'signal': 'neutral', 'description': '', 'strength': 0}

        for i in range(n - 3, n - 1):
            if i < 2:
                continue

            c0, c1, c2 = close[i-2], close[i-1], close[i]
            o0, o1, o2 = open_p[i-2], open_p[i-1], open_p[i]
            v0, v1, v2 = volume[i-2], volume[i-1], volume[i]

            if not (c0 > o0 and c1 > o1 and c2 > o2):
                continue
            if not (c2 > c1 > c0):
                continue

            body_low_0 = min(o0, c0)
            body_low_1 = min(o1, c1)

            if not (o1 >= body_low_0):
                continue
            if not (o2 >= body_low_1):
                continue

            def upper_shadow(o, c, h):
                body_top = max(o, c)
                return (h - body_top) / (body_top - min(o, c)) if (body_top - min(o, c)) > 0 else 0

            if upper_shadow(o0, c0, high[i-2]) > 0.6:
                continue
            if upper_shadow(o1, c1, high[i-1]) > 0.6:
                continue
            if upper_shadow(o2, c2, high[i]) > 0.6:
                continue

            total_gain = (c2 - o0) / o0 * 100 if o0 > 0 else 0
            vol_increasing = v2 > v1 > v0
            vol_ratio = v2 / vol_ma5[i] if vol_ma5[i] > 0 else 1
            strength = min(100, 40 + total_gain * 8 + (20 if vol_increasing else 0) + min(20, (vol_ratio - 1) * 30))

            desc = f'红三兵形态：连续3根阳线，累计涨幅{total_gain:.1f}%'
            if vol_increasing:
                desc += '，温和放量确认'

            result = {
                'detected': True,
                'signal': 'bullish',
                'description': desc,
                'strength': round(min(100, strength), 1),
                'score': round(min(100, strength), 1),
                'total_gain_pct': round(total_gain, 2),
                'vol_increasing': vol_increasing,
            }
            return result

        result['description'] = '近3日未出现标准红三兵形态'
        return result

    # ========================================================
    #  3. 仙人指路形态
    # ========================================================
    def _detect_immortal_pointing(self, close, open_p, high, low, volume, vol_ma5, df):
        """

        仙人指路形态识别：

        - 前期处于上升趋势

        - 出现一根长上影线K线（上影线长度 > 实体2倍，且 > 下影线2倍）

        - 当日收阳或小阴（实体较小，说明上方抛压被消化）

        - 随后1~3天内股价突破该上影线高点

        """
        n = len(close)
        result = {'detected': False, 'signal': 'neutral', 'description': '', 'strength': 0}

        for i in range(n - 4, n - 1):
            if i < 10:
                continue

            o = open_p[i]
            c = close[i]
            h = high[i]
            l = low[i]

            body = abs(c - o)
            upper_shadow = h - max(o, c)
            lower_shadow = min(o, c) - l

            if upper_shadow < body * 1.5:
                continue
            if lower_shadow > upper_shadow * 0.5:
                continue
            if body > upper_shadow * 0.8:
                continue

            pre_10_close = close[max(0, i - 10)]
            if pre_10_close <= 0 or c / pre_10_close < 1.0:
                continue

            for j in range(i + 1, min(n, i + 4)):
                if close[j] > h:
                    gain = (close[j] - c) / c * 100 if c > 0 else 0
                    strength = min(100, 50 + (upper_shadow / body - 1.5) * 30 + gain * 3)
                    result = {
                        'detected': True,
                        'signal': 'bullish',
                        'description': f'仙人指路：长上影线试盘后突破影线高点{h:.2f}，当前{close[j]:.2f}',
                        'strength': round(min(100, strength), 1),
                        'score': round(min(100, strength), 1),
                        'shadow_high': round(h, 2),
                        'breakout_price': round(close[j], 2),
                    }
                    return result

            if i == n - 2 or i == n - 1:
                result = {
                    'detected': True,
                    'signal': 'bullish',
                    'description': f'仙人指路（待确认）：出现长上影线试探{h:.2f}，等待突破确认',
                    'strength': round(45, 1),
                    'score': round(45, 1),
                    'shadow_high': round(h, 2),
                    'pending': True,
                }
                return result

        result['description'] = '近5日未检测到仙人指路形态'
        return result

    # ========================================================
    #  4. 倍量柱形态
    # ========================================================
    def _detect_double_volume(self, close, open_p, high, low, volume, vol_ma5, vol_ma20, df):
        """

        倍量柱形态识别（参考《量柱擒涨停》第10讲）：

        - 当日成交量 >= 前一日成交量的1.9倍（接近或超过2倍）

        - 当日为阳线（收盘>开盘），表明主力资金大幅流入

        - 倍量柱出现后若缩量不破倍量柱最低价，则是黄金柱的前身

        """
        n = len(close)
        result = {'detected': False, 'signal': 'neutral', 'description': '', 'strength': 0}

        # Scan all data, collect all double_volume signals
        dv_signals = []
        for i in range(1, n):
            prev_vol = volume[i - 1]
            curr_vol = volume[i]
            curr_close = close[i]
            curr_open = open_p[i]
            curr_low = low[i]

            if prev_vol <= 0:
                continue

            vol_ratio = curr_vol / prev_vol
            if vol_ratio < 1.9:
                continue

            is_bull = curr_close > curr_open
            price_change_pct = (curr_close - curr_open) / curr_open * 100 if curr_open > 0 else 0

            protected = True
            for j in range(i + 1, min(n, i + 10)):
                if low[j] < curr_low:
                    protected = False
                    break

            signal = 'bullish' if is_bull else 'neutral'
            strength = min(100, 40 + (vol_ratio - 1.9) * 30 + (20 if is_bull else 0) + (20 if protected else 0))

            dv_signals.append({
                'detected': True,
                'signal': signal,
                'description': f'倍量柱：成交量是前一日的{vol_ratio:.1f}倍' + (f'，阳线涨幅{price_change_pct:.1f}%' if is_bull else '，阴线需警惕') + ('，回调未跌破倍量柱最低价（黄金柱潜力）' if protected else ''),
                'strength': round(min(100, strength), 1),
                'score': round(min(100, strength), 1),
                'vol_ratio': round(vol_ratio, 2),
                'price_change_pct': round(price_change_pct, 2),
                'protected': protected,
                'index': i,
                'date': str(df.index[i])[:10] if hasattr(df.index[i], '__format__') else str(df.index[i])[:10],
                'price': round(float(curr_close), 2),
            })

        if dv_signals:
            # Return the most recent signal as the primary result
            result = dv_signals[-1]
            result['_all_signals'] = dv_signals  # Pass all for chart markers
            return result

        result['description'] = '未出现倍量柱（需成交量 >= 前日1.9倍）'
        return result

    # ========================================================
    #  4b. 波段高位卖点检测（放量+高换手+浪型顶点）
    # ========================================================
    def _detect_distribution_top(self, close, open_p, high, low, volume, vol_ma5, vol_ma20, df):
        """

        波段高位卖点识别：

        - 成交量显著放大（超过20日均量2倍以上）

        - 价格处于近期高点（近20日最高价的95%以上）

        - 高换手特征（成交量/流通股本 proxies）

        - 配合浪型顶点判断（接近上升浪末端）

        """
        n = len(close)
        result = {'detected': False, 'signal': 'neutral', 'description': '', 'strength': 0}

        if n < 30:
            return result

        # Scan all data for distribution top signals
        dt_signals = []
        for i in range(20, n):
            curr_vol = volume[i]
            curr_close = close[i]
            curr_high = high[i]
            curr_open = open_p[i]

            # 条件1: 成交量放大（超过20日均量2倍）
            if vol_ma20[i] <= 0:
                continue
            vol_ratio_20 = curr_vol / vol_ma20[i]
            if vol_ratio_20 < 2.0:
                continue

            # 条件2: 价格处于近20日高点区域
            window_high = max(high[max(0, i-19):i+1])
            price_ratio = curr_high / window_high if window_high > 0 else 0
            if price_ratio < 0.95:
                continue

            # 条件3: 当日或近3日有冲高回落特征
            # （最高价远高于收盘价，上影线较长）
            body_top = max(curr_close, curr_open)
            upper_shadow = curr_high - body_top
            body_size = abs(curr_close - curr_open)
            # 上影线大于实体的0.5倍视为有回落特征
            has_reversal = (body_size > 0 and upper_shadow > body_size * 0.5) or (curr_close < curr_open)

            # 条件4: 近期有放量滞涨特征（价涨但量更大）
            if i >= 2:
                recent_vol_avg = (volume[i-2] + volume[i-1] + curr_vol) / 3
                recent_price_change = (close[i] - close[i-3]) / close[i-3] * 100 if close[i-3] > 0 else 0
                # 放量但涨幅不大（滞涨）
                stagflation = vol_ratio_20 >= 2.5 and abs(recent_price_change) < 5
            else:
                stagflation = False

            # 综合评分
            if has_reversal or stagflation:
                strength = min(100, 50 + (vol_ratio_20 - 2.0) * 15 + (20 if has_reversal else 0) + (15 if stagflation else 0))
                signal = 'bearish'
            else:
                strength = min(100, 35 + (vol_ratio_20 - 2.0) * 10)
                signal = 'bearish'

            label = '放量高位'
            desc = f'波段高位：成交量是20日均量的{vol_ratio_20:.1f}倍'
            if has_reversal:
                desc += '，冲高回落上影线较长'
                label = '放量回落'
            if stagflation:
                desc += '，放量滞涨需警惕'
                label = '放量滞涨'

            dt_signals.append({
                'detected': True,
                'signal': signal,
                'description': desc,
                'strength': round(min(100, strength), 1),
                'score': round(min(100, strength), 1),
                'vol_ratio': round(vol_ratio_20, 2),
                'index': i,
                'date': str(df.index[i])[:10] if hasattr(df.index[i], '__format__') else str(df.index[i])[:10],
                'price': round(float(curr_close), 2),
                'label': label,
            })

        if dt_signals:
            result = dt_signals[-1]
            result['_all_signals'] = dt_signals
            return result

        result['description'] = '未出现波段高位放量信号'
        return result

    # ========================================================
    #  5. 梯量柱形态
    # ========================================================
    def _detect_ladder_volume(self, close, open_p, high, low, volume, vol_ma5, df):
        """

        梯量柱形态识别（参考《量柱擒涨停》第11讲）：

        - 连续3~5天成交量逐步放大（阶梯式递增）

        - 价格同步上涨或维持

        - "盯三防四"：关注第3根梯量柱，防范第4根变化

        """
        n = len(close)
        result = {'detected': False, 'signal': 'neutral', 'description': '', 'strength': 0}

        for start in range(max(0, n - 15), n - 3):
            count = 0
            for k in range(start, min(n, start + 6)):
                if k > 0 and volume[k] > volume[k - 1]:
                    count += 1
                else:
                    break

            if count < 3:
                continue

            price_up_count = 0
            for k in range(start, start + count):
                if close[k] > close[k - 1] if k > 0 else True:
                    price_up_count += 1

            end_idx = start + count - 1
            total_vol_increase = volume[end_idx] / volume[start] if volume[start] > 0 else 1
            total_price_change = (close[end_idx] - close[start]) / close[start] * 100 if close[start] > 0 else 0

            after_ladder = ''
            signal = 'neutral'
            strength = 40

            if end_idx + 1 < n:
                if end_idx + 1 < n and volume[end_idx + 1] > volume[end_idx]:
                    signal = 'bullish'
                    strength = 75
                    after_ladder = '，梯量第四柱继续放量（变态行为，偏多）'
                elif close[end_idx + 1] >= close[end_idx]:
                    signal = 'bullish'
                    strength = 65
                    after_ladder = '，缩量但价格维持（主力控盘良好）'
                else:
                    signal = 'neutral'
                    strength = 40
                    after_ladder = '，梯量后回落（正常衰竭）'
            else:
                after_ladder = '，梯量进行中'

            price_sync = price_up_count / count * 100
            if price_sync >= 70:
                signal = 'bullish' if signal != 'bearish' else signal
                strength += 10

            desc = f'梯量柱：连续{count}天放量递增（{total_vol_increase:.1f}倍），价格变化{total_price_change:.1f}%{after_ladder}'

            result = {
                'detected': True,
                'signal': signal,
                'description': desc,
                'strength': round(min(100, strength), 1),
                'score': round(min(100, strength), 1),
                'consecutive_days': count,
                'vol_increase_ratio': round(total_vol_increase, 2),
                'price_change_pct': round(total_price_change, 2),
            }
            return result

        result['description'] = '近15日未出现梯量柱形态（需连续3天以上成交量递增）'
        return result

    # ========================================================
    #  6. 均线金叉
    # ========================================================
    def _detect_golden_cross(self, close, ma5, ma10, ma20, ma60, df):
        """

        均线金叉识别：

        - 5日上穿10日（短期金叉）

        - 10日上穿20日（中期金叉）

        - 20日上穿60日（长期金叉）

        """
        n = len(close)
        result = {'detected': False, 'signal': 'neutral', 'description': '', 'strength': 0}

        crosses = []

        # Scan all data, collect all double_volume signals
        dv_signals = []
        for i in range(1, n):
            if ma5[i] > 0 and ma10[i] > 0:
                if ma5[i - 1] <= ma10[i - 1] and ma5[i] > ma10[i]:
                    crosses.append(('MA5上穿MA10', i, 'short'))

            if ma10[i] > 0 and ma20[i] > 0:
                if ma10[i - 1] <= ma20[i - 1] and ma10[i] > ma20[i]:
                    crosses.append(('MA10上穿MA20', i, 'mid'))

            if ma20[i] > 0 and ma60[i] > 0:
                if ma20[i - 1] <= ma60[i - 1] and ma20[i] > ma60[i]:
                    crosses.append(('MA20上穿MA60', i, 'long'))

        if crosses:
            latest = crosses[-1]
            strength_map = {'short': 50, 'mid': 65, 'long': 80}
            strength = strength_map.get(latest[2], 50)

            if len(crosses) >= 2:
                strength = min(95, strength + 15)

            cross_names = '、'.join(c[0] for c in crosses)
            desc = f'均线金叉：{cross_names}'

            result = {
                'detected': True,
                'signal': 'bullish',
                'description': desc,
                'strength': round(min(100, strength), 1),
                'score': round(min(100, strength), 1),
                'crosses': [{'name': c[0], 'type': c[2]} for c in crosses],
            }
            return result

        if n >= 2 and ma5[n-1] > 0 and ma10[n-1] > 0:
            if ma5[n-1] > ma10[n-1] and ma10[n-1] > ma20[n-1] if ma20[n-1] > 0 else True:
                result['description'] = '均线多头排列中（未出现新金叉）'
            else:
                result['description'] = '近10日未出现均线金叉'
        else:
            result['description'] = '数据不足，无法判断均线金叉'

        return result

    def _empty_result(self, reason):
        return {
            'patterns': {},
            'summary': reason,
            'score': 0,
            'overall_signal': 'none',
            'detected_list': [],
            'name_map': {},
        }


def detect_classic_patterns(df):
    """

    便捷入口：传入 DataFrame，返回形态检测结果

    """
    detector = ClassicPatternDetector()
    return detector.detect_all(df)