|
|
|
|
| """
|
| 经典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 = {}
|
|
|
|
|
| results['old_duck_head'] = self._detect_old_duck_head(
|
| close, open_p, high, low, volume, ma5, ma10, ma20, ma60, vol_ma5, df
|
| )
|
|
|
|
|
| results['three_white_soldiers'] = self._detect_three_white_soldiers(
|
| close, open_p, high, low, volume, vol_ma5, df
|
| )
|
|
|
|
|
| results['immortal_pointing'] = self._detect_immortal_pointing(
|
| close, open_p, high, low, volume, vol_ma5, df
|
| )
|
|
|
|
|
| results['double_volume'] = self._detect_double_volume(
|
| close, open_p, high, low, volume, vol_ma5, vol_ma20, df
|
| )
|
|
|
|
|
| results['ladder_volume'] = self._detect_ladder_volume(
|
| close, open_p, high, low, volume, vol_ma5, df
|
| )
|
|
|
|
|
| results['distribution_top'] = self._detect_distribution_top(
|
| close, open_p, high, low, volume, vol_ma5, vol_ma20, df
|
| )
|
|
|
|
|
| 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),
|
| }
|
|
|
| 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)
|
|
|
|
|
| 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,
|
| }
|
|
|
|
|
|
|
|
|
| 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
|
|
|
|
|
|
|
|
|
| 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
|
|
|
|
|
|
|
|
|
| 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
|
|
|
|
|
|
|
|
|
| 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}
|
|
|
|
|
| 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:
|
|
|
| result = dv_signals[-1]
|
| result['_all_signals'] = dv_signals
|
| return result
|
|
|
| result['description'] = '未出现倍量柱（需成交量 >= 前日1.9倍）'
|
| return result
|
|
|
|
|
|
|
|
|
| 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
|
|
|
|
|
| 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]
|
|
|
|
|
| if vol_ma20[i] <= 0:
|
| continue
|
| vol_ratio_20 = curr_vol / vol_ma20[i]
|
| if vol_ratio_20 < 2.0:
|
| continue
|
|
|
|
|
| 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
|
|
|
|
|
|
|
| body_top = max(curr_close, curr_open)
|
| upper_shadow = curr_high - body_top
|
| body_size = abs(curr_close - curr_open)
|
|
|
| has_reversal = (body_size > 0 and upper_shadow > body_size * 0.5) or (curr_close < curr_open)
|
|
|
|
|
| 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
|
|
|
|
|
|
|
|
|
| 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
|
|
|
|
|
|
|
|
|
| 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 = []
|
|
|
|
|
| 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)
|
|
|
