#!/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)