# -*- coding: utf-8 -*- """ 蒸鱼三层融合引擎 v1.0 - Fusion Engine 融合 QuantMind 三层融合架构（粗筛->精筛->风控闸门）+ 市场状态自适应权重 到蒸鱼现有终极评分体系。 """ import numpy as np FUSION_CONFIG = { "version": "2026-06-09", "enabled": True, "layer1_consensus": { "enabled": True, "min_models_required": 3, "min_consensus_ratio": 0.3, "max_neutral_ratio": 0.6, }, "layer2_regime_weights": { "enabled": True, "regime_weights": { "TREND_UP": { "ml_gbdt": 0.25, "ml_lr": 0.10, "ml_hc": 0.20, "fv5": 0.10, "fv5_hc": 0.15, "plan_e": 0.05, "v4_composite": 0.05, "factor": 0.03, "wyckoff": 0.02, "wave": 0.02, "classic_pattern": 0.01, "multi_dimension": 0.01, "talang": 0.01, }, "TREND_DOWN": { "ml_gbdt": 0.30, "ml_lr": 0.15, "ml_hc": 0.25, "fv5": 0.08, "fv5_hc": 0.12, "plan_e": 0.03, "v4_composite": 0.03, "factor": 0.02, "wyckoff": 0.01, "wave": 0.01, "classic_pattern": 0.00, "multi_dimension": 0.00, "talang": 0.00, }, "RANGE": { "ml_gbdt": 0.15, "ml_lr": 0.08, "ml_hc": 0.10, "fv5": 0.15, "fv5_hc": 0.12, "plan_e": 0.10, "v4_composite": 0.10, "factor": 0.08, "wyckoff": 0.04, "wave": 0.03, "classic_pattern": 0.02, "multi_dimension": 0.02, "talang": 0.01, }, }, "high_confidence_gate": { "enabled": True, "min_confidence": 0.55, "gate_mode": "soft", }, }, "layer3_risk_gate": { "enabled": False, "max_single_weight": 0.30, "max_volatility_20d": 0.06, }, "merge": { "mode": "weighted", "consensus_bonus": True, "divergence_penalty": True, }, } class FusionEngine: def __init__(self, config: dict = None): self.config = config or FUSION_CONFIG self._regime_detector = None def _get_regime_detector(self): if self._regime_detector is None: try: from market_regime import MarketRegimeDetector self._regime_detector = MarketRegimeDetector() except ImportError: self._regime_detector = None return self._regime_detector def detect_market_regime(self, df) -> dict: detector = self._get_regime_detector() if detector is None: return {"regime": "RANGE", "strength": 0.0} try: close = df['close'].values.astype(float) high = df['high'].values.astype(float) low = df['low'].values.astype(float) volume = df['volume'].values.astype(float) if 'volume' in df.columns else None result = detector.detect(close, high, low, volume) return result except Exception: return {"regime": "RANGE", "strength": 0.0} def _layer1_consensus_filter(self, model_scores: dict) -> dict: cfg = self.config["layer1_consensus"] if not cfg["enabled"]: return model_scores total = len(model_scores) if total < cfg["min_models_required"]: return model_scores neutral_count = sum(1 for s in model_scores.values() if 40 <= s <= 60) if neutral_count / total > cfg["max_neutral_ratio"]: return {n: 50 + (s - 50) * 0.6 for n, s in model_scores.items()} return model_scores def _layer2_regime_weights(self, model_scores: dict, regime: str) -> dict: cfg = self.config["layer2_regime_weights"] if not cfg["enabled"]: return {m: 1.0 / len(model_scores) for m in model_scores} rw = cfg["regime_weights"] if regime not in rw: regime = "RANGE" weights = {m: rw[regime].get(m, 0.05) for m in model_scores} total = sum(weights.values()) if total > 0: weights = {k: v / total for k, v in weights.items()} gate = cfg["high_confidence_gate"] if gate["enabled"]: for m in list(weights.keys()): conf = abs(model_scores[m] - 50) / 50 if conf < gate["min_confidence"]: if gate["gate_mode"] == "hard": del weights[m] else: weights[m] *= conf total = sum(weights.values()) if total > 0: weights = {k: v / total for k, v in weights.items()} return weights def _layer3_risk_gate(self, score: float, df=None) -> float: cfg = self.config["layer3_risk_gate"] if not cfg["enabled"] or df is None or len(df) < 20: return score close = df['close'].values.astype(float) ret = np.diff(close) / close[:-1] vol = np.std(ret[-20:]) * np.sqrt(252) if vol > cfg["max_volatility_20d"]: score = 50 + (score - 50) * 0.7 return score def _apply_consensus_modulation(self, score: float, model_scores: dict) -> float: cfg = self.config["merge"] scores = list(model_scores.values()) if len(scores) < 3: return score std = np.std(scores) if cfg["consensus_bonus"] and std < 15: score += (15 - std) * 0.3 if cfg["divergence_penalty"] and std > 25: score -= (std - 25) * 0.2 return np.clip(score, 0, 100) def fuse(self, ultimate_result: dict, df=None) -> dict: if not ultimate_result: return ultimate_result contributions = ultimate_result.get('model_contributions', []) model_scores = {c['model']: c['score'] for c in contributions} if not model_scores: return ultimate_result filtered = self._layer1_consensus_filter(model_scores) regime = "RANGE" regime_strength = 0.0 if df is not None: rr = self.detect_market_regime(df) regime = rr.get('regime', 'RANGE') regime_strength = rr.get('adx', 0) weights = self._layer2_regime_weights(filtered, regime) fusion_score = sum(filtered[m] * weights.get(m, 0) for m in filtered) if weights else ultimate_result.get('ultimate_score', 50) fusion_score = self._apply_consensus_modulation(fusion_score, filtered) fusion_score = self._layer3_risk_gate(fusion_score, df) if fusion_score >= 68: signal, advice, action = '强烈买入', '多模型高度共振看多, 建议积极介入', 'BUY' elif fusion_score >= 60: signal, advice, action = '买入', '多数模型指向看多, 可分批建仓', 'BUY' elif fusion_score >= 54: signal, advice, action = '偏多观望', '略偏积极, 可少量试探或等待更好入场点', 'HOLD' elif fusion_score >= 46: signal, advice, action = '中性', '多空分歧较大, 建议暂缓操作等待方向明确', 'HOLD' elif fusion_score <= 32: signal, advice, action = '强烈卖出', '多模型高度共振看空, 建议果断离场', 'SELL' elif fusion_score <= 40: signal, advice, action = '卖出', '多数模型指向看空, 建议减仓或离场', 'SELL' else: signal, advice, action = '偏空观望', '略偏悲观, 已有持仓考虑减仓', 'HOLD' original = ultimate_result.get('ultimate_score', 50) result = dict(ultimate_result) result.update({ 'fusion_score': round(fusion_score, 1), 'fusion_signal': signal, 'fusion_action': action, 'fusion_advice': advice, 'fusion_score_delta': round(fusion_score - original, 1), 'fusion_enabled': True, 'fusion_config_version': self.config['version'], 'market_regime': regime, 'regime_strength': round(regime_strength, 1), 'regime_weights': {k: round(v, 4) for k, v in sorted(weights.items(), key=lambda x: -x[1])}, 'layer1_active': self.config['layer1_consensus']['enabled'], 'layer2_active': self.config['layer2_regime_weights']['enabled'], 'layer3_active': self.config['layer3_risk_gate']['enabled'], }) return result _fusion_engine = None def get_fusion_engine() -> FusionEngine: global _fusion_engine if _fusion_engine is None: _fusion_engine = FusionEngine() return _fusion_engine def fuse_ultimate_score(ultimate_result: dict, df=None) -> dict: return get_fusion_engine().fuse(ultimate_result, df)