|
|
| """
|
| 波浪理论分析模块 (Wave Analysis Module) - v3
|
| 基于艾略特波浪理论 · 冯雷《股票波浪理论揭秘》
|
|
|
| 核心修正（v3）：
|
| - 严格区分「5浪推动」与「调整区(A/B/C…)」
|
| - 5浪推动 = 一个完整的中期趋势结构（涨跌均可）
|
| - 调整区 = 5浪完成后的独立修正阶段，不属于推动结构
|
| - 禁止输出1-2-3-4-5-A-B-C的8浪封闭循环
|
| - 支持周线主级别 / 日线子级别的嵌套框架
|
| """
|
|
|
| import numpy as np
|
| import pandas as pd
|
| from scipy.signal import argrelextrema
|
|
|
|
|
| class WaveAnalyzer:
|
| """
|
| 单周期浪型分析器
|
|
|
| 架构：
|
| impulse_phase（推动浪）：浪1-浪5，构成一个完整的趋势结构
|
| corrective_phase（调整区）：A浪/B浪/C浪/…，5浪完成后的修正阶段
|
| """
|
|
|
| PHASE_IMPULSE = 'impulse'
|
| PHASE_CORRECTIVE = 'corrective'
|
|
|
| def __init__(self, df):
|
| self.df = df.copy()
|
| if 'timestamps' in self.df.columns and not isinstance(self.df.index, pd.DatetimeIndex):
|
| self.df.set_index('timestamps', inplace=True)
|
| self.close = self.df['close'].values.astype(float)
|
| self.high = self.df['high'].values.astype(float)
|
| self.low = self.df['low'].values.astype(float)
|
| self.volume = self.df['volume'].values.astype(float) if 'volume' in self.df.columns else np.zeros(len(self.close))
|
| self.n = len(self.close)
|
| self.dates = self.df.index
|
|
|
| def find_pivots(self, order=5):
|
| peaks_idx = argrelextrema(self.high, np.greater_equal, order=order)[0]
|
| valleys_idx = argrelextrema(self.low, np.less_equal, order=order)[0]
|
| return list(peaks_idx), list(valleys_idx)
|
|
|
| def refine_pivots(self, peaks, valleys, min_swing_pct=0.02):
|
| all_pivots = [(idx, 'peak', self.high[idx]) for idx in peaks] + \
|
| [(idx, 'valley', self.low[idx]) for idx in valleys]
|
| all_pivots.sort(key=lambda x: x[0])
|
|
|
| filtered = []
|
| for i, (idx, ptype, price) in enumerate(all_pivots):
|
| if i == 0 or i == len(all_pivots) - 1:
|
| filtered.append((idx, ptype, price))
|
| continue
|
| if filtered:
|
| swing_pct = abs(price - filtered[-1][2]) / filtered[-1][2]
|
| if swing_pct < min_swing_pct:
|
| continue
|
| filtered.append((idx, ptype, price))
|
|
|
| alternating = []
|
| for idx, ptype, price in filtered:
|
| if not alternating:
|
| alternating.append((idx, ptype, price))
|
| elif ptype != alternating[-1][1]:
|
| alternating.append((idx, ptype, price))
|
| else:
|
| if ptype == 'peak' and price > alternating[-1][2]:
|
| alternating[-1] = (idx, ptype, price)
|
| elif ptype == 'valley' and price < alternating[-1][2]:
|
| alternating[-1] = (idx, ptype, price)
|
| return [p[0] for p in alternating if p[1] == 'peak'], \
|
| [p[0] for p in alternating if p[1] == 'valley']
|
|
|
| def build_swing_points(self, peaks, valleys):
|
| all_pivots = [(idx, 'high', self.high[idx]) for idx in peaks] + \
|
| [(idx, 'low', self.low[idx]) for idx in valleys]
|
| all_pivots.sort(key=lambda x: x[0])
|
| result = []
|
| for idx, ptype, price in all_pivots:
|
| if not result:
|
| result.append((idx, ptype, price))
|
| elif ptype != result[-1][1]:
|
| result.append((idx, ptype, price))
|
| else:
|
| if ptype == 'high' and price > result[-1][2]:
|
| result[-1] = (idx, ptype, price)
|
| elif ptype == 'low' and price < result[-1][2]:
|
| result[-1] = (idx, ptype, price)
|
| return result
|
|
|
| def identify_waves(self, order=5, min_swing_pct=0.02):
|
| """
|
| 主入口：识别浪型结构
|
|
|
| 返回 {
|
| 'waves': [...], # 浪型列表（推动+调整区分开）
|
| 'impulse_waves': [...], # 推动浪 1-5
|
| 'corrective_waves': [...], # 调整区 A/B/C
|
| 'phase': 'impulse'|'corrective'|'unknown',
|
| 'swing_points': [...],
|
| 'fib_levels': {...},
|
| 'buy_signals': [...],
|
| 'sell_signals': [...],
|
| 'characteristics': {...},
|
| 'current_wave': {...},
|
| 'wave_count': int,
|
| 'is_complete_5wave': bool,
|
| 'iron_rule_checks': {...},
|
| }
|
| """
|
| peaks, valleys = self.find_pivots(order=order)
|
| peaks, valleys = self.refine_pivots(peaks, valleys, min_swing_pct)
|
| swings = self.build_swing_points(peaks, valleys)
|
|
|
| if len(swings) < 3:
|
| return self._empty_result()
|
|
|
|
|
| trend = self._detect_trend(swings)
|
|
|
|
|
| strict_match = self._find_5wave_strict(swings, trend)
|
|
|
| loose_match = self._find_5wave_loose(swings, trend)
|
|
|
| best_match = None
|
| is_strict = False
|
| if strict_match and (not loose_match or strict_match[1] >= loose_match[1]):
|
| best_match = strict_match[0]
|
| is_strict = True
|
| elif loose_match:
|
| best_match = loose_match[0]
|
| is_strict = False
|
|
|
| if best_match:
|
| waves = self._build_waves(swings, best_match)
|
| fib_levels = self._calc_fib_levels(waves)
|
| buy_signals, sell_signals = self._identify_signals(waves)
|
| characteristics = self._analyze_characteristics(waves, is_strict)
|
| iron_rules = self._check_iron_rules(waves)
|
| else:
|
| waves = self._label_partial_waves(swings, trend)
|
| fib_levels = {}
|
| buy_signals, sell_signals = self._identify_signals_simple(swings)
|
| characteristics = {}
|
| iron_rules = {}
|
|
|
|
|
| impulse_waves = [w for w in waves if w['phase'] == self.PHASE_IMPULSE]
|
| corrective_waves = [w for w in waves if w['phase'] == self.PHASE_CORRECTIVE]
|
|
|
|
|
| if corrective_waves:
|
| phase = self.PHASE_CORRECTIVE
|
| elif impulse_waves and len(impulse_waves) >= 5:
|
| phase = self.PHASE_CORRECTIVE
|
| elif impulse_waves:
|
| phase = self.PHASE_IMPULSE
|
| else:
|
| phase = 'unknown'
|
|
|
| return {
|
| 'waves': waves,
|
| 'impulse_waves': impulse_waves,
|
| 'corrective_waves': corrective_waves,
|
| 'phase': phase,
|
| 'swing_points': [(int(s[0]), s[1], round(float(s[2]), 2)) for s in swings],
|
| 'fib_levels': fib_levels,
|
| 'buy_signals': buy_signals,
|
| 'sell_signals': sell_signals,
|
| 'characteristics': characteristics,
|
| 'current_wave': self._current_wave_position(waves, phase),
|
| 'wave_count': len(waves),
|
| 'is_complete_5wave': best_match is not None,
|
| 'iron_rule_checks': iron_rules,
|
| }
|
|
|
| def _detect_trend(self, swings):
|
| """检测总体趋势方向（用于确定推动浪是上涨还是下跌）"""
|
| if len(swings) < 2:
|
| return 'up'
|
| first_price = swings[0][2]
|
| last_price = swings[-1][2]
|
|
|
| mid_idx = len(swings) // 2
|
| mid_price = swings[mid_idx][2]
|
| if last_price > first_price and mid_price > first_price:
|
| return 'up'
|
| elif last_price < first_price and mid_price < first_price:
|
| return 'down'
|
|
|
| up_sum = 0
|
| down_sum = 0
|
| for i in range(len(swings) - 1):
|
| diff = swings[i + 1][2] - swings[i][2]
|
| if diff > 0:
|
| up_sum += diff
|
| else:
|
| down_sum += abs(diff)
|
| return 'up' if up_sum >= down_sum else 'down'
|
|
|
| def _find_5wave_strict(self, swings, trend):
|
| """严格匹配5浪推动：铁律1浪4底>浪1顶，铁律2浪3非最短"""
|
| if len(swings) < 6:
|
| return None
|
| best_score = -1
|
| best = None
|
| for s in range(len(swings)):
|
| if swings[s][1] != 'low' or s + 5 >= len(swings):
|
| continue
|
|
|
|
|
| exp = ['low', 'high', 'low', 'high', 'low', 'high']
|
| if any(swings[s + k][1] != exp[k] for k in range(6)):
|
| continue
|
| p = [swings[s + k][2] for k in range(6)]
|
| a1, a3, a5 = p[1] - p[0], p[3] - p[2], p[5] - p[4]
|
|
|
| if trend == 'up':
|
| if a1 <= 0 or a3 <= 0 or a5 <= 0:
|
| continue
|
| else:
|
|
|
| if a1 <= 0 or a3 <= 0:
|
| continue
|
|
|
| if p[4] <= p[1]:
|
| continue
|
| if a3 < a1 and a3 < a5:
|
| continue
|
|
|
| score = self._score_5wave(p, a1, a3, a5, swings, s, trend)
|
| if score > best_score:
|
| best_score = score
|
| best = (s, s + 1, s + 2, s + 3, s + 4, s + 5)
|
| return (best, best_score) if best and best_score >= 3 else None
|
|
|
| def _find_5wave_loose(self, swings, trend):
|
| """宽松匹配：铁律违反时扣分但不排除"""
|
| if len(swings) < 6:
|
| return None
|
| best_score = -1
|
| best = None
|
| for s in range(len(swings)):
|
| if swings[s][1] != 'low' or s + 5 >= len(swings):
|
| continue
|
| exp = ['low', 'high', 'low', 'high', 'low', 'high']
|
| if any(swings[s + k][1] != exp[k] for k in range(6)):
|
| continue
|
| p = [swings[s + k][2] for k in range(6)]
|
| a1, a3, a5 = p[1] - p[0], p[3] - p[2], p[5] - p[4]
|
|
|
| if trend == 'up' and (a1 <= 0 or a3 <= 0 or a5 <= 0):
|
| continue
|
| elif trend == 'down' and (a1 <= 0 or a3 <= 0):
|
| continue
|
|
|
| score = self._score_5wave(p, a1, a3, a5, swings, s, trend)
|
|
|
| if p[4] <= p[1]:
|
| score -= 3
|
| if a3 < a1 and a3 < a5:
|
| score -= 3
|
|
|
| if score > best_score:
|
| best_score = score
|
| best = (s, s + 1, s + 2, s + 3, s + 4, s + 5)
|
| return (best, best_score) if best and best_score >= 5 else None
|
|
|
| def _score_5wave(self, p, a1, a3, a5, swings, s, trend):
|
| """5浪结构评分"""
|
| score = 0
|
|
|
| if a1 > 0:
|
| r3 = a3 / a1
|
| if 0.8 <= r3 <= 2.618:
|
| score += 3
|
| if 1.1 <= r3 <= 2.2:
|
| score += 2
|
|
|
| if a1 > 0:
|
| r2 = (p[1] - p[2]) / a1
|
| if 0.15 <= r2 <= 0.75:
|
| score += 2
|
| if 0.3 <= r2 <= 0.7:
|
| score += 1
|
|
|
| r4_denom = p[3] - p[2]
|
| if r4_denom > 0:
|
| r4 = (p[3] - p[4]) / r4_denom
|
| if 0.1 <= r4 <= 0.75:
|
| score += 2
|
|
|
| if a1 > 0:
|
| r5 = a5 / a1
|
| if 0.5 <= r5 <= 3.5:
|
| score += 1
|
|
|
| coverage = (swings[s + 5][0] - swings[s][0]) / self.n
|
| score += coverage * 3
|
|
|
| if p[5] > p[3] > p[1]:
|
| score += 2
|
| if a3 >= a1 and a3 >= a5:
|
| score += 1
|
| return score
|
|
|
| def _build_waves(self, swings, indices):
|
| """
|
| 构建5浪推动 + 调整区（非8浪合并）
|
|
|
| 关键区别：浪5之后的修正标记为「调整区」而非第6/7/8浪。
|
| 一个完整的中期趋势 = 推动浪(1-5) + 调整区(A/B/C...)，
|
| 其中推动浪是独立完整的结构，调整区是5浪完成后的修正阶段。
|
| """
|
| i1, i2, i3, i4, i5, i6 = indices
|
| start_idx = max(0, swings[i1][0] - 1)
|
| waves = []
|
|
|
|
|
| impulse_defs = [
|
| (start_idx, swings[i2][0], '1', self.PHASE_IMPULSE, 'impulse'),
|
| (swings[i2][0], swings[i3][0], '2', self.PHASE_IMPULSE, 'corrective'),
|
| (swings[i3][0], swings[i4][0], '3', self.PHASE_IMPULSE, 'impulse'),
|
| (swings[i4][0], swings[i5][0], '4', self.PHASE_IMPULSE, 'corrective'),
|
| (swings[i5][0], swings[i6][0], '5', self.PHASE_IMPULSE, 'impulse'),
|
| ]
|
| for si, ei, label, phase, wtype in impulse_defs:
|
| amp = self.close[ei] - self.close[si]
|
| waves.append({
|
| 'index': int(label),
|
| 'label': label,
|
| 'phase': phase,
|
| 'type': wtype,
|
| 'start_idx': int(si),
|
| 'end_idx': int(ei),
|
| 'start_price': round(float(self.close[si]), 2),
|
| 'end_price': round(float(self.close[ei]), 2),
|
| 'amplitude': round(float(amp), 2),
|
| 'amplitude_pct': round(float(amp / self.close[si] * 100), 2) if self.close[si] > 0 else 0,
|
| })
|
|
|
|
|
|
|
| remaining = swings[i6 + 1:] if i6 + 1 < len(swings) else []
|
| if len(remaining) >= 2:
|
| for k in range(min(5, len(remaining) - 1)):
|
| si = remaining[k][0]
|
| ei = remaining[k + 1][0] if k + 1 < len(remaining) else self.n - 1
|
| amp = self.close[ei] - self.close[si]
|
| label = chr(ord('A') + k)
|
| waves.append({
|
| 'index': label,
|
| 'label': label,
|
| 'phase': self.PHASE_CORRECTIVE,
|
| 'type': 'corrective',
|
| 'start_idx': int(si),
|
| 'end_idx': int(ei),
|
| 'start_price': round(float(self.close[si]), 2),
|
| 'end_price': round(float(self.close[ei]), 2),
|
| 'amplitude': round(float(amp), 2),
|
| 'amplitude_pct': round(float(amp / self.close[si] * 100), 2) if self.close[si] > 0 else 0,
|
| })
|
| return waves
|
|
|
| def _label_partial_waves(self, swings, trend):
|
| """
|
| 局部标注（无法匹配完整5浪时）
|
|
|
| 核心规则：
|
| - 推动浪阶段只标注1-5
|
| - 当满足切换条件时转入调整区，标注A/B/C
|
| - 严禁在一个结构中同时标注推动浪编号和调整区字母
|
| """
|
| waves = []
|
| phase = self.PHASE_IMPULSE
|
| impulse_count = 0
|
| corrective_count = 0
|
|
|
| for i in range(len(swings) - 1):
|
| si = swings[i][0]
|
| ei = swings[i + 1][0]
|
| amp = self.close[ei] - self.close[si]
|
| rel = amp / self.close[si] * 100 if self.close[si] > 0 else 0
|
| is_up = amp > 0
|
|
|
| if phase == self.PHASE_IMPULSE:
|
| if is_up:
|
|
|
| impulse_count += 1
|
| label = str(min(impulse_count, 5))
|
| else:
|
|
|
| impulse_count += 1
|
| label = str(min(impulse_count, 5))
|
|
|
|
|
|
|
| up_in_impulse = sum(1 for j in range(len(waves)) if waves[j]['amplitude'] > 0)
|
| down_in_impulse = sum(1 for j in range(len(waves)) if waves[j]['amplitude'] <= 0)
|
| if up_in_impulse >= 3 and down_in_impulse >= 2 and impulse_count >= 5:
|
| phase = self.PHASE_CORRECTIVE
|
| corrective_count = 0
|
|
|
| label = 'A'
|
| corrective_count = 1
|
|
|
| wtype = 'impulse' if is_up else 'corrective'
|
| waves.append({
|
| 'index': label,
|
| 'label': label,
|
| 'phase': phase if label != 'A' else self.PHASE_CORRECTIVE,
|
| 'type': wtype,
|
| 'start_idx': int(si),
|
| 'end_idx': int(ei),
|
| 'start_price': round(float(self.close[si]), 2),
|
| 'end_price': round(float(self.close[ei]), 2),
|
| 'amplitude': round(float(amp), 2),
|
| 'amplitude_pct': round(float(rel), 2),
|
| })
|
|
|
| else:
|
|
|
| corrective_count += 1
|
| label = chr(ord('A') + min(corrective_count - 1, 4))
|
| waves.append({
|
| 'index': label,
|
| 'label': label,
|
| 'phase': self.PHASE_CORRECTIVE,
|
| 'type': 'corrective' if not is_up else 'impulse',
|
| 'start_idx': int(si),
|
| 'end_idx': int(ei),
|
| 'start_price': round(float(self.close[si]), 2),
|
| 'end_price': round(float(self.close[ei]), 2),
|
| 'amplitude': round(float(amp), 2),
|
| 'amplitude_pct': round(float(rel), 2),
|
| })
|
|
|
| return waves
|
|
|
| def _check_iron_rules(self, waves):
|
| """检查波浪铁律"""
|
| imp = [w for w in waves if w['phase'] == self.PHASE_IMPULSE and w['label'].isdigit()]
|
| checks = {}
|
|
|
| if len(imp) >= 5:
|
| w1, w2, w3, w4, w5 = imp[0], imp[1], imp[2], imp[3], imp[4]
|
|
|
|
|
| rule1_ok = w4['end_price'] > w1['end_price']
|
| checks['rule1_w4_above_w1'] = {
|
| 'rule': '浪4底 > 浪1顶',
|
| 'w4_bottom': w4['end_price'],
|
| 'w1_top': w1['end_price'],
|
| 'pass': rule1_ok,
|
| 'detail': f'浪4底部({w4["end_price"]:.2f}) {">" if rule1_ok else "≤"} 浪1顶部({w1["end_price"]:.2f})'
|
| }
|
|
|
|
|
| a1 = abs(w1['amplitude'])
|
| a3 = abs(w3['amplitude'])
|
| a5 = abs(w5['amplitude'])
|
| rule2_ok = not (a3 < a1 and a3 < a5)
|
| checks['rule2_w3_not_shortest'] = {
|
| 'rule': '浪3非最短',
|
| 'amps': {'w1': round(a1, 2), 'w3': round(a3, 2), 'w5': round(a5, 2)},
|
| 'pass': rule2_ok,
|
| 'detail': f'浪1:{a1:.1f} 浪3:{a3:.1f} 浪5:{a5:.1f} → {"浪3非最短" if rule2_ok else "浪3最短(违反!)"}'
|
| }
|
|
|
|
|
| rule3_ok = w2['end_price'] > w1['start_price']
|
| checks['rule2_above_w1_start'] = {
|
| 'rule': '浪2底 > 浪1起点',
|
| 'pass': rule3_ok,
|
| 'detail': f'浪2底部({w2["end_price"]:.2f}) {">" if rule3_ok else "≤"} 浪1起点({w1["start_price"]:.2f})'
|
| }
|
|
|
| return checks
|
|
|
| def _calc_fib_levels(self, waves):
|
| imp = [w for w in waves if w['phase'] == self.PHASE_IMPULSE and w['label'].isdigit()]
|
| fib = {}
|
| if len(imp) >= 2:
|
| w1 = imp[0]
|
| w1a = w1['amplitude']
|
| if w1a > 0:
|
| cor = [w for w in imp if w['type'] == 'corrective']
|
| if cor:
|
| fib['wave3_target'] = round(cor[0]['end_price'] + w1a * 1.618, 2)
|
| fib['wave5_low'] = round(w1['start_price'], 2)
|
| fib['wave5_high'] = round(w1['end_price'] + w1a * 3.236, 2)
|
| last = imp[-1]
|
| ta = last['end_price'] - last['start_price']
|
| if ta > 0:
|
| fib['correction_levels'] = {
|
| '0.236': round(last['end_price'] - ta * 0.236, 2),
|
| '0.382': round(last['end_price'] - ta * 0.382, 2),
|
| '0.500': round(last['end_price'] - ta * 0.5, 2),
|
| '0.618': round(last['end_price'] - ta * 0.618, 2),
|
| }
|
| return fib
|
|
|
| def _identify_signals(self, waves):
|
| """根据浪型阶段生成交易信号"""
|
| buy_signals = []
|
| sell_signals = []
|
| for i, w in enumerate(waves):
|
| ei = w['end_idx']
|
| if ei >= self.n - 1:
|
| continue
|
| label = w['label']
|
| ep = w['end_price']
|
| phase = w.get('phase', self.PHASE_IMPULSE)
|
|
|
| if phase == self.PHASE_IMPULSE:
|
| if label == '2':
|
| if i >= 1:
|
| w1 = waves[i - 1]
|
| r = (w1['end_price'] - ep) / w1['amplitude'] if w1['amplitude'] > 0 else 0
|
| if 0.15 <= r <= 0.85:
|
| buy_signals.append({
|
| 'date': str(self.dates[ei])[:10], 'price': ep, 'index': int(ei),
|
| 'type': 'buy', 'label': '第2浪低点买入', 'wave': '2', 'phase': 'impulse',
|
| 'strength': 85 if 0.35 <= r <= 0.65 else 60,
|
| 'reason': f'回调{(r * 100):.1f}%，黄金分割回调带'
|
| })
|
| elif label == '4':
|
| if i >= 3:
|
| w1 = waves[i - 3]
|
| strength = 75 if ep > w1['end_price'] else 50
|
| reason = f'第4浪底部({"铁律满足" if ep > w1["end_price"] else "注意风险"})'
|
| buy_signals.append({
|
| 'date': str(self.dates[ei])[:10], 'price': ep, 'index': int(ei),
|
| 'type': 'buy', 'label': '第4浪低点买入', 'wave': '4', 'phase': 'impulse',
|
| 'strength': strength, 'reason': reason
|
| })
|
| elif label == '3':
|
| sell_signals.append({
|
| 'date': str(self.dates[ei])[:10], 'price': ep, 'index': int(ei),
|
| 'type': 'sell', 'label': '第3浪高点卖出', 'wave': '3', 'phase': 'impulse',
|
| 'strength': 90, 'reason': '最具爆炸性上涨后'
|
| })
|
| elif label == '5':
|
| strength = 95 if self._check_vol_div(waves, i) == 'bearish_div' else 80
|
| sell_signals.append({
|
| 'date': str(self.dates[ei])[:10], 'price': ep, 'index': int(ei),
|
| 'type': 'sell', 'label': '第5浪高点卖出', 'wave': '5', 'phase': 'impulse',
|
| 'strength': strength, 'reason': '第5浪末端'
|
| })
|
|
|
| elif phase == self.PHASE_CORRECTIVE:
|
|
|
| if label == 'C':
|
| buy_signals.append({
|
| 'date': str(self.dates[ei])[:10], 'price': ep, 'index': int(ei),
|
| 'type': 'buy', 'label': '调整区C浪末端', 'wave': 'C', 'phase': 'corrective',
|
| 'strength': 80, 'reason': '调整区C浪结束，等待新推动浪'
|
| })
|
| elif label == 'B':
|
| sell_signals.append({
|
| 'date': str(self.dates[ei])[:10], 'price': ep, 'index': int(ei),
|
| 'type': 'sell', 'label': '调整区B浪高点', 'wave': 'B', 'phase': 'corrective',
|
| 'strength': 85, 'reason': '调整区B浪反弹高点'
|
| })
|
|
|
| return buy_signals, sell_signals
|
|
|
| def _identify_signals_simple(self, swings):
|
| """局部浪型信号（无明确5浪匹配时）"""
|
| buy_signals = []
|
| sell_signals = []
|
| for i in range(1, len(swings) - 1):
|
| idx = swings[i][0]
|
| if idx >= self.n - 1:
|
| continue
|
| if swings[i][1] == 'low':
|
| prev = swings[i - 1][2]
|
| pct = (prev - swings[i][2]) / prev * 100
|
| if pct > 3:
|
| buy_signals.append({
|
| 'date': str(self.dates[idx])[:10], 'price': round(float(swings[i][2]), 2),
|
| 'index': int(idx), 'type': 'buy', 'label': '波谷买入', 'wave': '?',
|
| 'phase': 'unknown', 'strength': min(85, int(pct * 4)),
|
| 'reason': f'回调{pct:.1f}%后的波谷'
|
| })
|
| elif swings[i][1] == 'high':
|
| prev = swings[i - 1][2]
|
| pct = (swings[i][2] - prev) / prev * 100
|
| if pct > 3:
|
| sell_signals.append({
|
| 'date': str(self.dates[idx])[:10], 'price': round(float(swings[i][2]), 2),
|
| 'index': int(idx), 'type': 'sell', 'label': '波峰卖出', 'wave': '?',
|
| 'phase': 'unknown', 'strength': min(85, int(pct * 4)),
|
| 'reason': f'上涨{pct:.1f}%后的波峰'
|
| })
|
| return buy_signals, sell_signals
|
|
|
| def _analyze_characteristics(self, waves, is_strict=True):
|
| chars = {}
|
| impulse_traits = {
|
| '1': '底部营造浪，量能温和放大',
|
| '2': '强势回调0.382/0.618',
|
| '3': '最具爆炸性，成交量最大，主升浪核心',
|
| '4': '与浪2交替，常为三角形/平台整理',
|
| '5': '末升浪，常伴随量价背离'
|
| }
|
| corrective_traits = {
|
| 'A': '5浪确认后的第一波下跌',
|
| 'B': '反弹修正（多头陷阱）',
|
| 'C': '杀伤力最强的下跌段',
|
| 'D': '下跌中继（三角形）',
|
| 'E': '末端下跌'
|
| }
|
| for w in waves:
|
| if w['phase'] == self.PHASE_IMPULSE:
|
| chars[w['label']] = {'description': impulse_traits.get(w['label'], ''), 'amplitude': w['amplitude_pct']}
|
| else:
|
| chars[w['label']] = {'description': corrective_traits.get(w['label'], ''), 'amplitude': w['amplitude_pct']}
|
|
|
| imp = [w for w in waves if w['label'].isdigit() and w['phase'] == self.PHASE_IMPULSE]
|
| if len(imp) >= 4:
|
| chars['iron_rule_1'] = f'浪4底({imp[3]["end_price"]:.2f}) {">" if imp[3]["end_price"] > imp[0]["end_price"] else "≤"} 浪1顶({imp[0]["end_price"]:.2f})'
|
| if len(imp) >= 3:
|
| amps = [abs(imp[i]['amplitude']) for i in range(min(3, len(imp)))]
|
| chars['iron_rule_2'] = f'浪3({"非最短" if len(amps) >= 3 and amps[1] != min(amps) else "最短!"}) 1:{amps[0]:.0f} 3:{amps[1]:.0f} 5:{amps[2] if len(amps) > 2 else "?":.0f}'
|
| chars['is_strict_match'] = is_strict
|
| return chars
|
|
|
| def _current_wave_position(self, waves, phase):
|
| """判断当前价格所处的浪型位置"""
|
| if not waves:
|
| return {'wave': 'unknown', 'position': 'unknown', 'label': '无法判断', 'phase': 'unknown'}
|
|
|
| lw = waves[-1]
|
| label = lw['label']
|
| wphase = lw.get('phase', phase)
|
|
|
| is_corr = label.isalpha()
|
| prefix = '调整' if is_corr else '第'
|
| if lw['end_idx'] >= self.n - 3:
|
| phase_cn = '推动浪' if wphase == self.PHASE_IMPULSE else '调整区'
|
| return {
|
| 'wave': label, 'position': 'late',
|
| 'label': f'处于{prefix}{label}浪尾段（{phase_cn}阶段）',
|
| 'phase': wphase
|
| }
|
| else:
|
| phase_cn = '推动浪' if wphase == self.PHASE_IMPULSE else '调整区'
|
| return {
|
| 'wave': label, 'position': 'after',
|
| 'label': f'{prefix}{label}浪之后（{phase_cn}阶段）',
|
| 'phase': wphase
|
| }
|
|
|
| def _check_vol_div(self, waves, wi):
|
| w = waves[wi]
|
| s = max(0, w['start_idx'])
|
| e = min(self.n - 1, w['end_idx'])
|
| if e <= s or np.sum(self.volume[s:e + 1]) == 0:
|
| return 'unknown'
|
| mid = (s + e) // 2
|
| vf = np.mean(self.volume[s:mid + 1]) if mid > s else np.mean(self.volume[s:e + 1])
|
| vs = np.mean(self.volume[mid + 1:e + 1]) if mid < e else vf
|
| return 'bearish_div' if self.close[e] > self.close[s] and vs < vf * 0.8 else 'normal'
|
|
|
| def get_chart_annotations(self):
|
| result = self.identify_waves()
|
| annotations = []
|
| shapes = []
|
| dates_iso = [pd.Timestamp(d).isoformat() if not isinstance(d, str) else d for d in self.dates]
|
|
|
|
|
| impulse_colors = {'1': '#FFD700', '2': '#87CEEB', '3': '#FF6B6B', '4': '#87CEEB', '5': '#FFD700'}
|
| corrective_colors = {'A': '#69f0ae', 'B': '#DDA0DD', 'C': '#FF8A65', 'D': '#B0BEC5', 'E': '#90CAF9'}
|
|
|
| for w in result['waves']:
|
| mid = (w['start_idx'] + w['end_idx']) // 2
|
| if mid < self.n:
|
| if w.get('phase') == self.PHASE_CORRECTIVE:
|
| c = corrective_colors.get(w['label'], '#B0BEC5')
|
| prefix = '调整'
|
| else:
|
| c = impulse_colors.get(w['label'], '#FFFFFF')
|
| prefix = '第'
|
| y = self.high[max(0, mid - 2):min(self.n, mid + 3)].max() * 1.025
|
| text = f"{prefix}{w['label']}浪"
|
| annotations.append({
|
| 'x': dates_iso[mid], 'y': y, 'text': text,
|
| 'showarrow': True, 'arrowhead': 0, 'arrowcolor': c,
|
| 'font': {'size': 13, 'color': c, 'family': 'Microsoft YaHei, sans-serif'},
|
| 'bordercolor': c, 'borderwidth': 1, 'borderpad': 2
|
| })
|
|
|
|
|
| drawn = set()
|
| for w in result['waves']:
|
| si = w['start_idx']
|
| if si not in drawn and 0 < si < self.n - 1:
|
|
|
| phase = w.get('phase', self.PHASE_IMPULSE)
|
| if phase == self.PHASE_CORRECTIVE:
|
| dash = 'dash'
|
| color = 'rgba(105, 240, 174, 0.25)'
|
| else:
|
| dash = 'dot'
|
| color = 'rgba(255, 255, 255, 0.12)'
|
| shapes.append({
|
| 'type': 'line', 'x0': dates_iso[si], 'x1': dates_iso[si],
|
| 'y0': 0, 'y1': 1, 'yref': 'paper',
|
| 'line': {'color': color, 'width': 1, 'dash': dash}
|
| })
|
| drawn.add(si)
|
|
|
|
|
| for sig in result.get('buy_signals', []):
|
| idx = sig['index']
|
| if idx < self.n:
|
| y = self.low[max(0, idx - 1):min(self.n, idx + 2)].min() * 0.975
|
| annotations.append({
|
| 'x': dates_iso[idx], 'y': y, 'text': '▲买',
|
| 'showarrow': True, 'arrowhead': 2, 'arrowcolor': '#EF5350',
|
| 'font': {'size': 12, 'color': '#EF5350', 'family': 'Microsoft YaHei, sans-serif'},
|
| 'bgcolor': 'rgba(239,83,80,0.15)'
|
| })
|
|
|
| for sig in result.get('sell_signals', []):
|
| idx = sig['index']
|
| if idx < self.n:
|
| y = self.high[max(0, idx - 1):min(self.n, idx + 2)].max() * 1.025
|
| annotations.append({
|
| 'x': dates_iso[idx], 'y': y, 'text': '▼卖',
|
| 'showarrow': True, 'arrowhead': 2, 'arrowcolor': '#66bb6a',
|
| 'font': {'size': 12, 'color': '#66bb6a', 'family': 'Microsoft YaHei, sans-serif'},
|
| 'bgcolor': 'rgba(102,187,106,0.15)'
|
| })
|
|
|
|
|
| fib = result.get('fib_levels', {})
|
| for ratio, price in fib.get('correction_levels', {}).items():
|
| if self.low.min() <= price <= self.high.max():
|
| shapes.append({
|
| 'type': 'line', 'x0': dates_iso[0], 'x1': dates_iso[-1],
|
| 'y0': price, 'y1': price,
|
| 'line': {'color': 'rgba(255,165,0,0.3)', 'width': 1, 'dash': 'dashdot'}
|
| })
|
| annotations.append({
|
| 'x': dates_iso[-1], 'y': price, 'text': f'F{ratio}={price}',
|
| 'showarrow': False, 'font': {'size': 9, 'color': 'rgba(255,165,0,0.7)'},
|
| 'xanchor': 'left'
|
| })
|
|
|
| return {'annotations': annotations, 'shapes': shapes, 'analysis': result}
|
|
|
| def _empty_result(self):
|
| return {
|
| 'waves': [], 'impulse_waves': [], 'corrective_waves': [],
|
| 'phase': 'unknown',
|
| 'swing_points': [], 'fib_levels': {},
|
| 'buy_signals': [], 'sell_signals': [],
|
| 'characteristics': {},
|
| 'current_wave': {'wave': 'unknown', 'position': 'unknown', 'label': '数据不足', 'phase': 'unknown'},
|
| 'wave_count': 0, 'is_complete_5wave': False, 'iron_rule_checks': {}
|
| }
|
|
|
|
|
| def analyze_wave_patterns(df, order=5):
|
| return WaveAnalyzer(df).get_chart_annotations()
|
|
|
