#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ 多维度技术指标联动研判模块 包含12个固定参数指标、5大研判维度和统一信号共振逻辑 固定参数指标清单： MACD(5,10,3) KDJ(3,3,3) EMA5/10/20/60 BOLL(20,2) RSI(6,14) CCI(20) 成交量+5/10日均量 OBV DMI(14,6) SAR(0.02,0.2) MFI(14) 5大研判维度： 1. 趋势判定：MACD + 多级EMA + BOLL 2. 拐点判定：KDJ + RSI + CCI 3. 行情验证：成交量 + OBV 4. 强度确认：DMI + MFI 5. 风控配套：SAR 信号生效原则：多指标同向共振优先采信，背离则暂缓观望 """ import numpy as np import pandas as pd from typing import Dict, List, Tuple, Optional import warnings warnings.filterwarnings('ignore') # === 兼容层：优先尝试 talib，失败则用自带兼容模块 === try: import talib _HAS_TALIB = True except ImportError: _HAS_TALIB = False import importlib.util, sys, os _talib_path = os.path.join(os.path.dirname(__file__), 'talib.py') if os.path.exists(_talib_path): _spec = importlib.util.spec_from_file_location('talib_compat', _talib_path) talib = importlib.util.module_from_spec(_spec) sys.modules['talib'] = talib _spec.loader.exec_module(talib) else: raise ImportError("talib 或 talib.py 兼容层均不可用") # ============================================================================ # 第一部分：TechnicalIndicators —— 12个固定参数指标计算 # ============================================================================ class TechnicalIndicators: """技术指标计算类 —— 严格按固定参数执行""" def __init__(self): self.indicators = {} def calculate_all_indicators(self, df: pd.DataFrame) -> pd.DataFrame: """计算全部12类技术指标""" df = df.copy() if 'timestamps' in df.columns: df = df.sort_values('timestamps') df = self._calc_ema_system(df) # EMA5/10/20/60 df = self._calc_macd(df) # MACD(5,10,3) df = self._calc_kdj(df) # KDJ(3,3,3) df = self._calc_boll(df) # BOLL(20,2) df = self._calc_rsi(df) # RSI(6,14) df = self._calc_cci(df) # CCI(20) df = self._calc_volume(df) # 成交量+5/10日均量 df = self._calc_obv(df) # OBV df = self._calc_dmi(df) # DMI(14,6) df = self._calc_sar(df) # SAR(0.02,0.2) df = self._calc_mfi(df) # MFI(14) df = self._calc_auxiliary(df) # 辅助指标（ATR/ROC/MOM等） return df # ---------- 1. EMA 均线系统 ---------- def _calc_ema_system(self, df): c = df['close'].values.astype(float) for p in (5, 10, 20, 60): df[f'ema_{p}'] = talib.EMA(c, timeperiod=p) # SMA 作为辅助参考 for p in (5, 10, 20): df[f'sma_{p}'] = talib.SMA(c, timeperiod=p) return df # ---------- 2. MACD(5,10,3) ---------- def _calc_macd(self, df): c = df['close'].values.astype(float) df['macd'], df['macd_signal'], df['macd_hist'] = talib.MACD( c, fastperiod=5, slowperiod=10, signalperiod=3) return df # ---------- 3. KDJ(3,3,3) ---------- def _calc_kdj(self, df): h = df['high'].values.astype(float) l = df['low'].values.astype(float) c = df['close'].values.astype(float) df['k'], df['d'] = talib.STOCH( h, l, c, fastk_period=3, slowk_period=3, slowd_period=3) df['j'] = 3 * df['k'] - 2 * df['d'] return df # ---------- 4. BOLL(20,2) ---------- def _calc_boll(self, df): c = df['close'].values.astype(float) df['bb_upper'], df['bb_middle'], df['bb_lower'] = talib.BBANDS( c, timeperiod=20, nbdevup=2, nbdevdn=2) df['bb_width'] = (df['bb_upper'] - df['bb_lower']) / df['bb_middle'] df['bb_position'] = (c - df['bb_lower']) / (df['bb_upper'] - df['bb_lower']) return df # ---------- 5. RSI(6,14) ---------- def _calc_rsi(self, df): c = df['close'].values.astype(float) df['rsi_6'] = talib.RSI(c, timeperiod=6) df['rsi_14'] = talib.RSI(c, timeperiod=14) return df # ---------- 6. CCI(20) ---------- def _calc_cci(self, df): h = df['high'].values.astype(float) l = df['low'].values.astype(float) c = df['close'].values.astype(float) df['cci'] = talib.CCI(h, l, c, timeperiod=20) return df # ---------- 7. 成交量体系 ---------- def _calc_volume(self, df): v = df['volume'].values.astype(float) df['vol_ma5'] = talib.SMA(v, timeperiod=5) df['vol_ma10'] = talib.SMA(v, timeperiod=10) df['vol_ratio5'] = v / df['vol_ma5'] df['vol_ratio10'] = v / df['vol_ma10'] return df # ---------- 8. OBV ---------- def _calc_obv(self, df): c = df['close'].values.astype(float) v = df['volume'].values.astype(float) df['obv'] = talib.OBV(c, v) # OBV 均线 df['obv_ma10'] = talib.EMA(df['obv'].values.astype(float), timeperiod=10) return df # ---------- 9. DMI(14,6) ---------- def _calc_dmi(self, df): h = df['high'].values.astype(float) l = df['low'].values.astype(float) c = df['close'].values.astype(float) df['adx'] = talib.ADX(h, l, c, timeperiod=14) df['plus_di'] = talib.PLUS_DI(h, l, c, timeperiod=14) df['minus_di'] = talib.MINUS_DI(h, l, c, timeperiod=14) # ADX 平滑（6周期EMA）用于趋势确认 df['adx_smooth'] = talib.EMA(df['adx'].fillna(0).values.astype(float), timeperiod=6) df['di_diff'] = df['plus_di'] - df['minus_di'] # 多空力量差 return df # ---------- 10. SAR(0.02,0.2) ---------- def _calc_sar(self, df): h = df['high'].values.astype(float) l = df['low'].values.astype(float) df['sar'] = talib.SAR(h, l, acceleration=0.02, maximum=0.2) df['sar_signal'] = np.where(df['close'] > df['sar'], 1, -1) # 1多头 -1空头 return df # ---------- 11. MFI(14) ---------- def _calc_mfi(self, df): h = df['high'].values.astype(float) l = df['low'].values.astype(float) c = df['close'].values.astype(float) v = df['volume'].values.astype(float) df['mfi'] = talib.MFI(h, l, c, v, timeperiod=14) return df # ---------- 辅助指标 ---------- def _calc_auxiliary(self, df): h = df['high'].values.astype(float) l = df['low'].values.astype(float) c = df['close'].values.astype(float) v = df['volume'].values.astype(float) df['atr'] = talib.ATR(h, l, c, timeperiod=14) df['willr'] = talib.WILLR(h, l, c, timeperiod=14) df['roc_10'] = talib.ROC(c, timeperiod=10) df['mom_10'] = talib.MOM(c, timeperiod=10) df['cmo'] = talib.CMO(c, timeperiod=14) # AD 累积/派发 df['ad'] = talib.AD(h, l, c, v) # 威廉%R 辅助 df['aroon_up'], df['aroon_down'] = talib.AROON(h, l, timeperiod=14) df['ultosc'] = talib.ULTOSC(h, l, c, timeperiod1=7, timeperiod2=14, timeperiod3=28) # 量价关系 df['price_change'] = df['close'].pct_change() df['volume_change'] = df['volume'].pct_change() return df # ============================================================================ # 第二部分：保留原有 VolumePriceAnalyzer / WaveTheoryAnalyzer（向后兼容） # ============================================================================ class VolumePriceAnalyzer: """量价分析器""" def __init__(self): self.indicators = TechnicalIndicators() def analyze_volume_price_relationship(self, df): analysis = {} price_change = df['close'].pct_change() volume_change = df['volume'].pct_change() corr_val = price_change.corr(volume_change) analysis['volume_price_correlation'] = 0.0 if (corr_val is None or (isinstance(corr_val, float) and np.isnan(corr_val))) else corr_val analysis['volume_price_divergence'] = self._detect_divergence(df) analysis['volume_anomaly_score'] = self._calculate_volume_anomaly_score(df) analysis['volume_price_sync_score'] = self._calculate_sync_score(df) return analysis def _detect_divergence(self, df): pt = df['close'].rolling(window=10).apply(lambda x: np.polyfit(range(len(x)), x, 1)[0]) vt = df['volume'].rolling(window=10).apply(lambda x: np.polyfit(range(len(x)), x, 1)[0]) d = np.corrcoef(pt.dropna(), vt.dropna())[0, 1] return d if not np.isnan(d) else 0.0 def _calculate_volume_anomaly_score(self, df): vm = df['volume'].rolling(20).mean() vs = df['volume'].rolling(20).std() a = np.abs((df['volume'] - vm) / vs) return a.mean() if not a.isna().all() else 0.0 def _calculate_sync_score(self, df): pd_dir = np.sign(df['close'].diff()) vd_dir = np.sign(df['volume'].diff()) s = (pd_dir == vd_dir).mean() return s if not np.isnan(s) else 0.0 class WaveTheoryAnalyzer: """波浪理论分析器""" def __init__(self): self.indicators = TechnicalIndicators() def analyze_wave_patterns(self, df): analysis = {} analysis['wave_count'] = self._count_waves(df) analysis['wave_strength'] = self._calculate_wave_strength(df) analysis['fibonacci_levels'] = self._calculate_fibonacci_analysis(df) analysis['wave_phase'] = self._identify_wave_phase(df) return analysis def _count_waves(self, df): c = df['close'].values waves, trend = 0, 0 for i in range(1, len(c)): if c[i] > c[i-1]: if trend <= 0: waves += 1; trend = 1 else: if trend >= 0: waves += 1; trend = -1 return waves def _calculate_wave_strength(self, df): c = df['close'].values changes = np.diff(c) cc, mc = 0, 0 for ch in changes: if ch > 0: cc += 1; mc = max(mc, cc) else: cc = 0 return mc / len(changes) if len(changes) > 0 else 0.0 def _calculate_fibonacci_analysis(self, df): c = df['close'].values rh, rl = np.max(c[-20:]), np.min(c[-20:]) return {k: rl + (rh - rl) * v for k, v in [('fib_236', 0.236), ('fib_382', 0.382), ('fib_500', 0.5), ('fib_618', 0.618), ('fib_786', 0.786)]} def _identify_wave_phase(self, df): c = df['close'].values if len(c) < 10: return "insufficient_data" rt = np.polyfit(range(10), c[-10:], 1)[0] if rt > 0.01: return "impulse_wave" elif rt < -0.01: return "corrective_wave" return "consolidation" # ============================================================================ # 第三部分：MultiDimensionSignalAnalyzer —— 多维度联动研判 + 信号共振 # ============================================================================ class MultiDimensionSignalAnalyzer: """ 多维度技术指标联动研判体系 5大研判维度： 1. 趋势判定 (Trend) : MACD(5,10,3) + EMA5/10/20/60 + BOLL(20,2) 2. 拐点判定 (Inflection) : KDJ(3,3,3) + RSI(6,14) + CCI(20) 3. 行情验证 (Verification): 成交量+5/10日均量 + OBV 4. 强度确认 (Strength) : DMI(14,6) + MFI(14) 5. 风控配套 (RiskControl) : SAR(0.02,0.2) 信号生效原则： - 多指标同向共振 → 采信 - 指标相互背离 → 暂缓观望 """ def __init__(self): self.calc = TechnicalIndicators() def analyze(self, df: pd.DataFrame) -> Dict: """ 执行完整的多维度联动研判 Parameters ---------- df : DataFrame OHLCV 数据（需包含 open/high/low/close/volume 列） Returns ------- dict 包含 dimensions、resonance、overall_signal、indicators_detail """ df = self.calc.calculate_all_indicators(df.copy()) last = df.iloc[-1] prev = df.iloc[-2] if len(df) >= 2 else last # 提取最新值 iv = self._extract_latest(df, last, prev) # 5大维度逐一研判 d_trend = self._judge_trend(iv) d_inflection = self._judge_inflection(iv) d_verification = self._judge_verification(iv, df) d_strength = self._judge_strength(iv) d_risk = self._judge_risk(iv) dimensions = { 'trend': d_trend, 'inflection': d_inflection, 'verification': d_verification, 'strength': d_strength, 'risk_control': d_risk, } # 信号共振 resonance = self._calc_resonance(dimensions) # 综合研判 overall = self._calc_overall(dimensions, resonance) # 指标详情（供前端展示） detail = self._build_detail(iv, dimensions) return { 'dimensions': dimensions, 'resonance': resonance, 'overall_signal': overall, 'indicators_detail': detail, } # ------------------------------------------------------------------ # 提取最新指标值 # ------------------------------------------------------------------ def _extract_latest(self, df, last, prev): """从 DataFrame 中提取最新和前一日的指标值""" def _v(col): return last.get(col, np.nan) def _p(col): return prev.get(col, np.nan) return { # 价格 'close': _v('close'), 'open': _v('open'), 'high': _v('high'), 'low': _v('low'), # EMA 'ema5': _v('ema_5'), 'ema10': _v('ema_10'), 'ema20': _v('ema_20'), 'ema60': _v('ema_60'), # MACD 'macd': _v('macd'), 'macd_signal': _v('macd_signal'), 'macd_hist': _v('macd_hist'), 'macd_prev': _p('macd'), 'macd_signal_prev': _p('macd_signal'), # KDJ 'k': _v('k'), 'd': _v('d'), 'j': _v('j'), 'k_prev': _p('k'), 'd_prev': _p('d'), # BOLL 'bb_upper': _v('bb_upper'), 'bb_middle': _v('bb_middle'), 'bb_lower': _v('bb_lower'), 'bb_width': _v('bb_width'), 'bb_position': _v('bb_position'), # RSI 'rsi6': _v('rsi_6'), 'rsi14': _v('rsi_14'), # CCI 'cci': _v('cci'), # 成交量 'volume': _v('volume'), 'vol_ma5': _v('vol_ma5'), 'vol_ma10': _v('vol_ma10'), 'vol_ratio5': _v('vol_ratio5'), 'vol_ratio10': _v('vol_ratio10'), 'close_prev': _p('close'), # OBV 'obv': _v('obv'), 'obv_ma10': _v('obv_ma10'), # DMI 'adx': _v('adx'), 'adx_smooth': _v('adx_smooth'), 'plus_di': _v('plus_di'), 'minus_di': _v('minus_di'), 'di_diff': _v('di_diff'), # SAR 'sar': _v('sar'), 'sar_signal': _v('sar_signal'), # MFI 'mfi': _v('mfi'), # 辅助 'atr': _v('atr'), } # ------------------------------------------------------------------ # 维度1：趋势判定 (MACD + EMA + BOLL) # ------------------------------------------------------------------ def _judge_trend(self, iv): """趋势判定：MACD + 多级EMA均线定方向，BOLL确认运行格局""" signals = [] # --- MACD --- macd_score = 0 macd_text = [] # MACD柱状图方向 if not np.isnan(iv['macd_hist']): if iv['macd_hist'] > 0: macd_score += 1 macd_text.append('MACD柱>0，多头动能') else: macd_score -= 1 macd_text.append('MACD柱<0，空头动能') # MACD金叉/死叉 if not np.isnan(iv['macd']) and not np.isnan(iv['macd_signal']): if not np.isnan(iv['macd_prev']) and not np.isnan(iv['macd_signal_prev']): prev_diff = iv['macd_prev'] - iv['macd_signal_prev'] curr_diff = iv['macd'] - iv['macd_signal'] if prev_diff <= 0 and curr_diff > 0: macd_score += 2 macd_text.append('MACD金叉确认') elif prev_diff >= 0 and curr_diff < 0: macd_score -= 2 macd_text.append('MACD死叉确认') if iv['macd'] > iv['macd_signal']: macd_score += 0.5 else: macd_score -= 0.5 # MACD零轴位置 if not np.isnan(iv['macd']): if iv['macd'] > 0: macd_score += 1 macd_text.append('MACD运行在零轴上方') else: macd_score -= 1 macd_text.append('MACD运行在零轴下方') signals.append({ 'indicator': 'MACD(5,10,3)', 'score': macd_score, 'max_score': 4, 'signal': 'bullish' if macd_score > 0 else 'bearish' if macd_score < 0 else 'neutral', 'details': macd_text }) # --- EMA均线系统 --- ema_score = 0 ema_text = [] close = iv['close'] # 多头排列判定：EMA5 > EMA10 > EMA20 > EMA60 emas = [('EMA5', iv['ema5']), ('EMA10', iv['ema10']), ('EMA20', iv['ema20']), ('EMA60', iv['ema60'])] valid_emas = [(n, v) for n, v in emas if not np.isnan(v)] if len(valid_emas) >= 2: is_bull = all(valid_emas[i][1] >= valid_emas[i+1][1] for i in range(len(valid_emas)-1)) is_bear = all(valid_emas[i][1] <= valid_emas[i+1][1] for i in range(len(valid_emas)-1)) if is_bull: ema_score += 2 ema_text.append('均线多头排列') elif is_bear: ema_score -= 2 ema_text.append('均线空头排列') # 价格相对EMA5位置 if not np.isnan(iv['ema5']): if close > iv['ema5']: ema_score += 0.5 ema_text.append('价格站上EMA5') else: ema_score -= 0.5 ema_text.append('价格跌破EMA5') # EMA5/EMA10金叉死叉（短期） if not np.isnan(iv['ema5']) and not np.isnan(iv['ema10']): if iv['ema5'] > iv['ema10']: ema_score += 0.5 ema_text.append('EMA5>EMA10短线偏多') else: ema_score -= 0.5 ema_text.append('EMA5 iv['ema20']: ema_score += 0.5 ema_text.append('价格在EMA20上方运行') else: ema_score -= 0.5 ema_text.append('价格在EMA20下方运行') signals.append({ 'indicator': 'EMA均线系统', 'score': ema_score, 'max_score': 3.5, 'signal': 'bullish' if ema_score > 0 else 'bearish' if ema_score < 0 else 'neutral', 'details': ema_text }) # --- BOLL --- boll_score = 0 boll_text = [] if not np.isnan(iv['bb_position']): bp = iv['bb_position'] if bp > 0.8: boll_score -= 1 boll_text.append('价格接近上轨，注意回调风险') elif bp > 0.5: boll_score += 0.5 boll_text.append('价格在中上轨区间，偏强运行') elif bp < 0.2: boll_score += 1 boll_text.append('价格接近下轨，关注反弹机会') elif bp < 0.5: boll_score -= 0.5 boll_text.append('价格在中下轨区间，偏弱运行') # 轨道收窄/扩张 if not np.isnan(iv['bb_width']): # 简化：用bb_width值判断 bw = iv['bb_width'] if bw > 0.15: boll_text.append('布林带宽较大，波动剧烈') elif bw < 0.05: boll_text.append('布林带极度收窄，变盘在即') signals.append({ 'indicator': 'BOLL(20,2)', 'score': boll_score, 'max_score': 1.5, 'signal': 'bullish' if boll_score > 0 else 'bearish' if boll_score < 0 else 'neutral', 'details': boll_text }) # 维度总分 total = sum(s['score'] for s in signals) max_total = sum(s['max_score'] for s in signals) return { 'name': '趋势判定', 'description': 'MACD + 多级EMA均线定方向，BOLL确认运行格局', 'signals': signals, 'score': total, 'max_score': max_total, 'verdict': 'bullish' if total > max_total * 0.3 else 'bearish' if total < -max_total * 0.3 else 'neutral', 'verdict_text': '趋势偏多' if total > max_total * 0.3 else '趋势偏空' if total < -max_total * 0.3 else '趋势不明/震荡', } # ------------------------------------------------------------------ # 维度2：拐点判定 (KDJ + RSI + CCI) # ------------------------------------------------------------------ def _judge_inflection(self, iv): """拐点判定：KDJ+RSI+CCI组合捕捉短期高低点反转""" signals = [] # --- KDJ(3,3,3) --- kdj_score = 0 kdj_text = [] if not np.isnan(iv['k']) and not np.isnan(iv['d']): k, d, j = iv['k'], iv['d'], iv['j'] # 超买超卖 if k > 80 and d > 80: kdj_score -= 1.5 kdj_text.append('KDJ高位超买区(K={:.0f},D={:.0f})'.format(k, d)) elif k < 20 and d < 20: kdj_score += 1.5 kdj_text.append('KDJ低位超卖区(K={:.0f},D={:.0f})'.format(k, d)) elif k > 50: kdj_score += 0.5 kdj_text.append('KDJ处于强势区间') elif k < 50: kdj_score -= 0.5 kdj_text.append('KDJ处于弱势区间') # J值极端 if not np.isnan(j): if j > 100: kdj_score -= 1 kdj_text.append('J值>100，严重超买') elif j < 0: kdj_score += 1 kdj_text.append('J值<0，严重超卖') # 金叉死叉 if not np.isnan(iv['k_prev']) and not np.isnan(iv['d_prev']): if iv['k_prev'] <= iv['d_prev'] and k > d: kdj_score += 1.5 kdj_text.append('KDJ低位金叉，反转信号') elif iv['k_prev'] >= iv['d_prev'] and k < d: kdj_score -= 1.5 kdj_text.append('KDJ高位死叉，见顶信号') signals.append({ 'indicator': 'KDJ(3,3,3)', 'score': kdj_score, 'max_score': 4, 'signal': 'bullish' if kdj_score > 0 else 'bearish' if kdj_score < 0 else 'neutral', 'details': kdj_text }) # --- RSI(6,14) --- rsi_score = 0 rsi_text = [] # RSI6 短线 if not np.isnan(iv['rsi6']): if iv['rsi6'] > 80: rsi_score -= 1 rsi_text.append('RSI6={:.1f}，严重超买'.format(iv['rsi6'])) elif iv['rsi6'] > 70: rsi_score -= 0.5 rsi_text.append('RSI6={:.1f}，超买区间'.format(iv['rsi6'])) elif iv['rsi6'] < 20: rsi_score += 1 rsi_text.append('RSI6={:.1f}，严重超卖'.format(iv['rsi6'])) elif iv['rsi6'] < 30: rsi_score += 0.5 rsi_text.append('RSI6={:.1f}，超卖区间'.format(iv['rsi6'])) else: rsi_text.append('RSI6={:.1f}，正常区间'.format(iv['rsi6'])) # RSI14 中线 if not np.isnan(iv['rsi14']): if iv['rsi14'] > 70: rsi_score -= 0.5 rsi_text.append('RSI14={:.1f}，偏超买'.format(iv['rsi14'])) elif iv['rsi14'] < 30: rsi_score += 0.5 rsi_text.append('RSI14={:.1f}，偏超卖'.format(iv['rsi14'])) # RSI背离辅助（简化：用RSI6与RSI14的方向差异） if not np.isnan(iv['rsi6']) and not np.isnan(iv['rsi14']): if iv['rsi6'] > 70 and iv['rsi14'] < 50: rsi_text.append('RSI6/RSI14背离，注意假突破') elif iv['rsi6'] < 30 and iv['rsi14'] > 50: rsi_text.append('RSI6/RSI14背离，注意假跌破') signals.append({ 'indicator': 'RSI(6,14)', 'score': rsi_score, 'max_score': 2, 'signal': 'bullish' if rsi_score > 0 else 'bearish' if rsi_score < 0 else 'neutral', 'details': rsi_text }) # --- CCI(20) --- cci_score = 0 cci_text = [] if not np.isnan(iv['cci']): c = iv['cci'] if c > 200: cci_score -= 1 cci_text.append('CCI={:.0f}，极端强势多头'.format(c)) elif c > 100: cci_score -= 0.5 cci_text.append('CCI={:.0f}，强势多头区间'.format(c)) elif c < -200: cci_score += 1 cci_text.append('CCI={:.0f}，极端弱势空头'.format(c)) elif c < -100: cci_score += 0.5 cci_text.append('CCI={:.0f}，弱势空头区间'.format(c)) else: cci_text.append('CCI={:.0f}，正常波动区间'.format(c)) signals.append({ 'indicator': 'CCI(20)', 'score': cci_score, 'max_score': 1.5, 'signal': 'bullish' if cci_score > 0 else 'bearish' if cci_score < 0 else 'neutral', 'details': cci_text }) total = sum(s['score'] for s in signals) max_total = sum(s['max_score'] for s in signals) return { 'name': '拐点判定', 'description': 'KDJ+RSI+CCI组合捕捉短期高低点反转', 'signals': signals, 'score': total, 'max_score': max_total, 'verdict': 'bullish' if total > max_total * 0.3 else 'bearish' if total < -max_total * 0.3 else 'neutral', 'verdict_text': '拐点偏多/超卖反弹' if total > max_total * 0.3 else '拐点偏空/超买回调' if total < -max_total * 0.3 else '无明显拐点信号', } # ------------------------------------------------------------------ # 维度3：行情验证 (成交量 + OBV) # ------------------------------------------------------------------ def _judge_verification(self, iv, df): """行情验证：成交量+OBV验证趋势真实性""" signals = [] # --- 成交量体系 --- vol_score = 0 vol_text = [] close = iv['close'] close_prev = iv['close_prev'] # 量比判断 if not np.isnan(iv['vol_ratio5']): vr = iv['vol_ratio5'] if vr > 2.0: vol_text.append('5日量比={:.1f}，显著放量'.format(vr)) if close > close_prev: vol_score += 1.5 # 放量上涨 else: vol_score -= 1.0 # 放量下跌 elif vr > 1.3: vol_text.append('5日量比={:.1f}，温和放量'.format(vr)) if close > close_prev: vol_score += 0.5 elif vr < 0.5: vol_text.append('5日量比={:.1f}，显著缩量'.format(vr)) # 缩量下跌视为惜售，缩量上涨视为动力不足 if close < close_prev: vol_score += 0.5 # 缩量下跌=惜售 else: vol_score -= 0.5 else: vol_text.append('5日量比={:.1f}，成交量正常'.format(vr)) # 10日量比 if not np.isnan(iv['vol_ratio10']): vr10 = iv['vol_ratio10'] if vr10 > 1.5: vol_text.append('10日量比={:.1f}，中期放量'.format(vr10)) signals.append({ 'indicator': '成交量体系', 'score': vol_score, 'max_score': 2, 'signal': 'bullish' if vol_score > 0 else 'bearish' if vol_score < 0 else 'neutral', 'details': vol_text }) # --- OBV --- obv_score = 0 obv_text = [] if not np.isnan(iv['obv']) and not np.isnan(iv['obv_ma10']): obv = iv['obv'] obv_ma = iv['obv_ma10'] if obv > obv_ma: obv_score += 1 obv_text.append('OBV位于均线上方，资金流入') else: obv_score -= 1 obv_text.append('OBV位于均线下方，资金流出') # OBV趋势与价格趋势一致性 if not np.isnan(close_prev) and close > close_prev: if obv > obv_ma: obv_text.append('量价齐升，趋势健康') obv_score += 0.5 else: obv_text.append('价涨量缩，需警惕量价背离') obv_score -= 0.5 elif not np.isnan(close_prev) and close < close_prev: if obv < obv_ma: obv_text.append('量价齐跌，空头确认') obv_score -= 0.5 else: obv_text.append('价跌量升，主力吸筹可能') obv_score += 0.5 signals.append({ 'indicator': 'OBV能量潮', 'score': obv_score, 'max_score': 1.5, 'signal': 'bullish' if obv_score > 0 else 'bearish' if obv_score < 0 else 'neutral', 'details': obv_text }) total = sum(s['score'] for s in signals) max_total = sum(s['max_score'] for s in signals) return { 'name': '行情验证', 'description': '成交量+OBV验证趋势真实性，排除无量假突破/假回落', 'signals': signals, 'score': total, 'max_score': max_total, 'verdict': 'bullish' if total > max_total * 0.2 else 'bearish' if total < -max_total * 0.2 else 'neutral', 'verdict_text': '量能验证通过，趋势可信' if total > max_total * 0.2 else '量能不支持，警惕假信号' if total < -max_total * 0.2 else '量能中性', } # ------------------------------------------------------------------ # 维度4：强度确认 (DMI + MFI) # ------------------------------------------------------------------ def _judge_strength(self, iv): """强度确认：DMI+MFI确认行情动能强弱与资金认可度""" signals = [] # --- DMI(14,6) --- dmi_score = 0 dmi_text = [] if not np.isnan(iv['adx']): adx = iv['adx'] # ADX > 25 趋势明确 if adx > 50: dmi_text.append('ADX={:.1f}，极强趋势'.format(adx)) dmi_score += 1.5 elif adx > 25: dmi_text.append('ADX={:.1f}，趋势明确'.format(adx)) dmi_score += 1 elif adx > 20: dmi_text.append('ADX={:.1f}，趋势形成中'.format(adx)) dmi_score += 0.5 else: dmi_text.append('ADX={:.1f}，无明确趋势/震荡市'.format(adx)) # +DI/-DI 判断多空力量 if not np.isnan(iv['plus_di']) and not np.isnan(iv['minus_di']): pdi, mdi = iv['plus_di'], iv['minus_di'] if pdi > mdi: dmi_score += 1 dmi_text.append('+DI({:.1f})>-DI({:.1f})，多头主导'.format(pdi, mdi)) else: dmi_score -= 1 dmi_text.append('-DI({:.1f})>+DI({:.1f})，空头主导'.format(mdi, pdi)) # DI差值 if not np.isnan(iv['di_diff']): diff = iv['di_diff'] if diff > 20: dmi_text.append('多空力量差={:.1f}，多方优势明显'.format(diff)) dmi_score += 0.5 elif diff < -20: dmi_text.append('多空力量差={:.1f}，空方优势明显'.format(diff)) dmi_score -= 0.5 signals.append({ 'indicator': 'DMI(14,6)', 'score': dmi_score, 'max_score': 3, 'signal': 'bullish' if dmi_score > 0 else 'bearish' if dmi_score < 0 else 'neutral', 'details': dmi_text }) # --- MFI(14) --- mfi_score = 0 mfi_text = [] if not np.isnan(iv['mfi']): mfi = iv['mfi'] if mfi > 80: mfi_score -= 1 mfi_text.append('MFI={:.1f}，资金过度流入/超买'.format(mfi)) elif mfi > 60: mfi_score += 0.5 mfi_text.append('MFI={:.1f}，资金持续流入'.format(mfi)) elif mfi < 20: mfi_score += 1 mfi_text.append('MFI={:.1f}，资金过度流出/超卖'.format(mfi)) elif mfi < 40: mfi_score -= 0.5 mfi_text.append('MFI={:.1f}，资金持续流出'.format(mfi)) else: mfi_text.append('MFI={:.1f}，资金流向中性'.format(mfi)) signals.append({ 'indicator': 'MFI(14)', 'score': mfi_score, 'max_score': 1.5, 'signal': 'bullish' if mfi_score > 0 else 'bearish' if mfi_score < 0 else 'neutral', 'details': mfi_text }) total = sum(s['score'] for s in signals) max_total = sum(s['max_score'] for s in signals) return { 'name': '强度确认', 'description': 'DMI+MFI确认行情动能强弱与资金认可度', 'signals': signals, 'score': total, 'max_score': max_total, 'verdict': 'bullish' if total > max_total * 0.3 else 'bearish' if total < -max_total * 0.3 else 'neutral', 'verdict_text': '动能强劲，资金认可' if total > max_total * 0.3 else '动能衰竭，资金撤离' if total < -max_total * 0.3 else '动能中性', } # ------------------------------------------------------------------ # 维度5：风控配套 (SAR) # ------------------------------------------------------------------ def _judge_risk(self, iv): """风控配套：SAR自动锚定趋势止盈止损点位""" signals = [] sar_score = 0 sar_text = [] if not np.isnan(iv['sar']) and not np.isnan(iv['sar_signal']): close = iv['close'] sar = iv['sar'] if iv['sar_signal'] == 1: # SAR在价格下方 = 多头趋势 sar_score += 2 sar_text.append('SAR={:.2f}在价格下方，多头趋势运行'.format(sar)) sar_text.append('SAR可作为止损参考位: {:.2f}'.format(sar)) else: # SAR在价格上方 = 空头趋势 sar_score -= 2 sar_text.append('SAR={:.2f}在价格上方，空头趋势运行'.format(sar)) sar_text.append('SAR可作为止盈/止损参考位: {:.2f}'.format(sar)) # SAR与价格距离（衡量风险敞口） distance = abs(close - sar) / close * 100 if distance > 5: sar_text.append('价格偏离SAR {:.1f}%，趋势强劲但注意回撤'.format(distance)) elif distance < 1: sar_text.append('价格接近SAR {:.1f}%，趋势可能反转'.format(distance)) signals.append({ 'indicator': 'SAR(0.02,0.2)', 'score': sar_score, 'max_score': 2, 'signal': 'bullish' if sar_score > 0 else 'bearish' if sar_score < 0 else 'neutral', 'details': sar_text }) return { 'name': '风控配套', 'description': 'SAR自动锚定趋势止盈止损点位，辅助风控量化', 'signals': signals, 'score': sar_score, 'max_score': 2, 'verdict': 'bullish' if sar_score > 0 else 'bearish' if sar_score < 0 else 'neutral', 'verdict_text': 'SAR多头运行，可持仓' if sar_score > 0 else 'SAR空头运行，宜止损' if sar_score < 0 else 'SAR中性', } # ------------------------------------------------------------------ # 信号共振计算 # ------------------------------------------------------------------ def _calc_resonance(self, dimensions): """ 多指标同向共振优先采信，指标背离则暂缓判定 共振等级： - 强共振：5个维度全部同向 - 弱共振：4个维度同向 - 共振：3个维度同向 - 分歧：2/3 维度反向 - 强分歧：各维度方向混乱 - 背离：方向完全不一致 """ verdicts = {k: v['verdict'] for k, v in dimensions.items()} bullish_count = sum(1 for v in verdicts.values() if v == 'bullish') bearish_count = sum(1 for v in verdicts.values() if v == 'bearish') neutral_count = sum(1 for v in verdicts.values() if v == 'neutral') total = len(verdicts) # 5 if bullish_count == total: level = 'strong_bullish' text = '5维度强共振看多' elif bearish_count == total: level = 'strong_bearish' text = '5维度强共振看空' elif bullish_count >= 4: level = 'bullish' text = '4维度共振看多' elif bearish_count >= 4: level = 'bearish' text = '4维度共振看空' elif bullish_count >= 3: level = 'weak_bullish' text = '3维度偏多共振' elif bearish_count >= 3: level = 'weak_bearish' text = '3维度偏空共振' elif neutral_count >= 3: level = 'neutral' text = '多维度中性，方向不明' elif bullish_count == 2 and bearish_count == 2: level = 'divergence' text = '指标分歧，多空对峙' elif bullish_count == 2 and bearish_count == 1: level = 'slight_bullish' text = '略偏多，但缺乏共振确认' elif bearish_count == 2 and bullish_count == 1: level = 'slight_bearish' text = '略偏空，但缺乏共振确认' else: level = 'divergence' text = '指标严重分歧，暂缓操作' # 共振分数（-100 到 +100） score_map = { 'strong_bullish': 90, 'bullish': 70, 'weak_bullish': 40, 'slight_bullish': 20, 'neutral': 0, 'slight_bearish': -20, 'weak_bearish': -40, 'bearish': -70, 'strong_bearish': -90, 'divergence': 0, } return { 'level': level, 'text': text, 'score': score_map.get(level, 0), 'bullish_count': bullish_count, 'bearish_count': bearish_count, 'neutral_count': neutral_count, 'distribution': {k: v for k, v in verdicts.items()}, } # ------------------------------------------------------------------ # 综合研判 # ------------------------------------------------------------------ def _calc_overall(self, dimensions, resonance): """综合5维度和共振得出最终信号""" # 加权总分 weights = { 'trend': 0.30, # 趋势最重要 'inflection': 0.20, # 拐点次之 'verification': 0.20, # 行情验证 'strength': 0.15, # 强度确认 'risk_control': 0.15, # 风控 } weighted_score = 0 for dim_name, dim_data in dimensions.items(): w = weights.get(dim_name, 0.2) max_s = dim_data['max_score'] if dim_data['max_score'] > 0 else 1 normalized = dim_data['score'] / max_s # -1 到 +1 weighted_score += normalized * w # -1 到 +1 映射到分数 total_score = round(weighted_score * 100, 1) # 结合共振等级修正 res_level = resonance['level'] if 'strong' in res_level: # 强共振放大信号 total_score = total_score * 1.2 elif 'divergence' in res_level: # 分歧缩小信号 total_score = total_score * 0.5 total_score = max(-100, min(100, round(total_score, 1))) # 最终信号 if total_score > 40: signal = 'bullish' signal_text = '多头信号' elif total_score > 15: signal = 'slightly_bullish' signal_text = '偏多信号' elif total_score > -15: signal = 'neutral' signal_text = '观望信号' elif total_score > -40: signal = 'slightly_bearish' signal_text = '偏空信号' else: signal = 'bearish' signal_text = '空头信号' # 操作建议 if signal == 'bullish': action = '积极做多，注意SAR止损位' elif signal == 'slightly_bullish': action = '可轻仓参与，严格风控' elif signal == 'neutral': action = '暂缓操作，等待明确信号' elif signal == 'slightly_bearish': action = '减仓观望，不宜追高' else: action = '回避风险，考虑止损离场' return { 'signal': signal, 'signal_text': signal_text, 'score': total_score, 'action': action, 'resonance_level': res_level, 'resonance_text': resonance['text'], } # ------------------------------------------------------------------ # 指标详情构建（供前端展示） # ------------------------------------------------------------------ def _build_detail(self, iv, dimensions): """构建前端展示用的指标数值详情""" return { 'MACD': { 'DIF': self._safe_round(iv['macd'], 4), 'DEA': self._safe_round(iv['macd_signal'], 4), 'MACD柱': self._safe_round(iv['macd_hist'], 4), }, 'KDJ': { 'K': self._safe_round(iv['k'], 2), 'D': self._safe_round(iv['d'], 2), 'J': self._safe_round(iv['j'], 2), }, 'EMA': { 'EMA5': self._safe_round(iv['ema5'], 2), 'EMA10': self._safe_round(iv['ema10'], 2), 'EMA20': self._safe_round(iv['ema20'], 2), 'EMA60': self._safe_round(iv['ema60'], 2), }, 'BOLL': { '上轨': self._safe_round(iv['bb_upper'], 2), '中轨': self._safe_round(iv['bb_middle'], 2), '下轨': self._safe_round(iv['bb_lower'], 2), '位置': self._safe_round(iv['bb_position'], 3), }, 'RSI': { 'RSI6': self._safe_round(iv['rsi6'], 2), 'RSI14': self._safe_round(iv['rsi14'], 2), }, 'CCI': { 'CCI': self._safe_round(iv['cci'], 2), }, '成交量': { '5日均量': self._safe_round(iv['vol_ma5'], 0), '10日均量': self._safe_round(iv['vol_ma10'], 0), '5日量比': self._safe_round(iv['vol_ratio5'], 2), }, 'OBV': { 'OBV': self._safe_round(iv['obv'], 0), 'OBV_MA10': self._safe_round(iv['obv_ma10'], 0), }, 'DMI': { 'ADX': self._safe_round(iv['adx'], 2), '+DI': self._safe_round(iv['plus_di'], 2), '-DI': self._safe_round(iv['minus_di'], 2), }, 'SAR': { 'SAR': self._safe_round(iv['sar'], 2), '信号': '多头' if iv.get('sar_signal') == 1 else '空头', }, 'MFI': { 'MFI': self._safe_round(iv['mfi'], 2), }, } @staticmethod def _safe_round(val, digits): if val is None or (isinstance(val, float) and np.isnan(val)): return None return round(float(val), digits) # ============================================================================ # 向后兼容的增强入口 # ============================================================================ def enhance_kronos_features(df: pd.DataFrame) -> pd.DataFrame: """为Kronos模型增强特征（保持向后兼容）""" indicators = TechnicalIndicators() enhanced_df = indicators.calculate_all_indicators(df) vp_analyzer = VolumePriceAnalyzer() vp = vp_analyzer.analyze_volume_price_relationship(enhanced_df) for k, v in vp.items(): enhanced_df[f'vp_{k}'] = v w_analyzer = WaveTheoryAnalyzer() wa = w_analyzer.analyze_wave_patterns(enhanced_df) for k, v in wa.items(): if isinstance(v, dict): for sk, sv in v.items(): enhanced_df[f'wave_{k}_{sk}'] = sv else: enhanced_df[f'wave_{k}'] = v return enhanced_df def run_multidimension_analysis(df: pd.DataFrame) -> Dict: """ 执行多维度联动研判（供路由调用） Parameters ---------- df : DataFrame OHLCV 数据 Returns ------- dict 完整的多维度研判结果 """ analyzer = MultiDimensionSignalAnalyzer() return analyzer.analyze(df) # ============================================================================ # 测试 # ============================================================================ if __name__ == "__main__": dates = pd.date_range('2024-01-01', periods=120, freq='D') np.random.seed(42) base = 100 prices = [base] for i in range(119): change = np.random.normal(0, 0.02) prices.append(max(0.01, prices[-1] * (1 + change))) df = pd.DataFrame({ 'timestamps': dates, 'open': prices, 'high': [p * (1 + abs(np.random.normal(0, 0.015))) for p in prices], 'low': [p * (1 - abs(np.random.normal(0, 0.015))) for p in prices], 'close': prices, 'volume': np.random.normal(1000000, 300000, 120), }) # 测试多维度研判 result = run_multidimension_analysis(df) print("=" * 60) print("多维度技术指标联动研判结果") print("=" * 60) for dim_name, dim_data in result['dimensions'].items(): print(f"\n【{dim_data['name']}】 {dim_data['description']}") print(f" 判定: {dim_data['verdict_text']} (得分: {dim_data['score']:.1f}/{dim_data['max_score']:.1f})") for sig in dim_data['signals']: print(f" - {sig['indicator']}: {sig['signal']} ({sig['score']:.1f}/{sig['max_score']:.1f})") for d in sig['details']: print(f" {d}") print(f"\n{'=' * 60}") print(f"【信号共振】 {result['resonance']['text']}") print(f" 共振等级: {result['resonance']['level']}") print(f" 共振分数: {result['resonance']['score']}") print(f" 分布: 看多={result['resonance']['bullish_count']} " f"看空={result['resonance']['bearish_count']} " f"中性={result['resonance']['neutral_count']}") print(f"\n【综合研判】 {result['overall_signal']['signal_text']}") print(f" 综合分数: {result['overall_signal']['score']}") print(f" 操作建议: {result['overall_signal']['action']}") print(f"\n【指标详情】") for name, vals in result['indicators_detail'].items(): val_str = ' | '.join(f"{k}={v}" for k, v in vals.items() if v is not None) print(f" {name}: {val_str}")