cyb50-quant/cyb50-pro/core/signal_fusion/bayesian_fusion.py

"""
贝叶斯信号融合引擎

使用贝叶斯网络进行非线性信号融合
P(上涨|信号,生态) ∝ P(生态) × ∏ P(信号i|上涨,生态) / P(信号i|生态)
"""

from dataclasses import dataclass, field
from typing import Dict, List, Optional, Tuple, Any
from datetime import datetime

import numpy as np
import pandas as pd
from scipy import stats


@dataclass
class FusedSignal:
    """融合后的信号"""
    up_probability: float      # 上涨概率
    down_probability: float    # 下跌概率
    neutral_probability: float # 横盘概率
    overall_confidence: float  # 整体置信度
    signal_grade: str          # 信号等级：strong/medium/weak/none
    recommended_action: str    # 建议动作
    fusion_metadata: Dict[str, Any] = field(default_factory=dict)
    timestamp: datetime = field(default_factory=datetime.now)


class BayesianSignalFusion:
    """
    贝叶斯信号融合器

    融合逻辑：
    1. 一级信号 → 贝叶斯网络输入
    2. 生态作为隐变量调节条件概率
    3. 输出后验概率分布
    """

    def __init__(
        self,
        prior_up: float = 0.33,
        prior_down: float = 0.33,
        prior_neutral: float = 0.34,
        min_sample_size: int = 1000
    ):
        self.prior = {
            "up": prior_up,
            "down": prior_down,
            "neutral": prior_neutral
        }
        self.min_sample_size = min_sample_size

        # 历史条件概率表（简化实现）
        self.conditional_probs: Dict[str, Dict] = {}

    def fuse(
        self,
        primary_signals: List[Any],
        ecosystem: Any,
        fallback_to_weighted: bool = True
    ) -> FusedSignal:
        """
        执行贝叶斯融合

        Args:
            primary_signals: 一级信号列表
            ecosystem: 市场生态
            fallback_to_weighted: 小样本时是否回退到加权平均

        Returns:
            FusedSignal: 融合后的信号
        """
        if not primary_signals:
            return FusedSignal(
                up_probability=self.prior["up"],
                down_probability=self.prior["down"],
                neutral_probability=self.prior["neutral"],
                overall_confidence=0.0,
                signal_grade="none",
                recommended_action="hold"
            )

        # 检查样本量
        if fallback_to_weighted:
            sample_size = self._estimate_sample_size(ecosystem)
            if sample_size < self.min_sample_size:
                return self._weighted_fusion(primary_signals, ecosystem, sample_size)

        # 贝叶斯推断
        posterior = self._bayesian_inference(primary_signals, ecosystem)

        # 确定信号等级和建议
        grade, action = self._determine_signal_grade(posterior)

        return FusedSignal(
            up_probability=posterior["up"],
            down_probability=posterior["down"],
            neutral_probability=posterior["neutral"],
            overall_confidence=self._calculate_confidence(posterior, primary_signals),
            signal_grade=grade,
            recommended_action=action,
            fusion_metadata={
                "method": "bayesian",
                "signal_count": len(primary_signals),
                "prior": self.prior.copy(),
                "posterior": posterior
            }
        )

    def _bayesian_inference(
        self,
        signals: List[Any],
        ecosystem: Any
    ) -> Dict[str, float]:
        """执行贝叶斯推断"""
        # 初始化后验为 prior
        posterior = self.prior.copy()

        # 获取生态条件
        regime = getattr(ecosystem.macro, 'regime', None) if ecosystem else None

        # 对每个信号更新后验
        for signal in signals:
            likelihood = self._calculate_likelihood(signal, regime)

            # 贝叶斯更新: P(H|D) ∝ P(D|H) * P(H)
            for direction in ["up", "down", "neutral"]:
                posterior[direction] *= likelihood.get(direction, 0.33)

        # 归一化
        total = sum(posterior.values())
        if total > 0:
            posterior = {k: v/total for k, v in posterior.items()}

        return posterior

    def _calculate_likelihood(
        self,
        signal: Any,
        regime: Any
    ) -> Dict[str, float]:
        """计算似然函数 P(信号|方向,生态)"""
        value = getattr(signal, 'value', 0)

        # 基于信号值和生态计算似然
        # 简化模型：信号值越正，上涨概率越高
        likelihood = {
            "up": max(0.1, min(0.9, 0.5 + value * 0.4)),
            "down": max(0.1, min(0.9, 0.5 - value * 0.4)),
            "neutral": max(0.1, min(0.9, 1 - abs(value) * 0.5))
        }

        # 生态调节
        if regime:
            regime_boost = {
                "summer": {"up": 1.2, "down": 0.8, "neutral": 0.9},
                "winter": {"up": 0.8, "down": 1.2, "neutral": 0.9},
                "spring": {"up": 1.1, "down": 0.9, "neutral": 0.95},
                "autumn": {"up": 0.9, "down": 0.9, "neutral": 1.1}
            }.get(regime.value, {"up": 1, "down": 1, "neutral": 1})

            for direction in likelihood:
                likelihood[direction] *= regime_boost[direction]

        # 重新归一化
        total = sum(likelihood.values())
        return {k: v/total for k, v in likelihood.items()}

    def _weighted_fusion(
        self,
        signals: List[Any],
        ecosystem: Any,
        sample_size: int
    ) -> FusedSignal:
        """小样本回退：加权平均融合"""
        if not signals:
            return self._create_neutral_signal("no_signals")

        # 加权平均
        weighted_sum = sum(s.value * getattr(s, 'confidence', 0.5) for s in signals)
        total_weight = sum(getattr(s, 'confidence', 0.5) for s in signals)

        if total_weight == 0:
            return self._create_neutral_signal("zero_weight")

        avg_signal = weighted_sum / total_weight

        # 转换为概率
        up_prob = max(0, min(1, 0.5 + avg_signal * 0.5))
        down_prob = max(0, min(1, 0.5 - avg_signal * 0.5))
        neutral_prob = 1 - abs(avg_signal)

        # 归一化
        total = up_prob + down_prob + neutral_prob
        probs = {
            "up": up_prob / total,
            "down": down_prob / total,
            "neutral": neutral_prob / total
        }

        grade, action = self._determine_signal_grade(probs)

        return FusedSignal(
            up_probability=probs["up"],
            down_probability=probs["down"],
            neutral_probability=probs["neutral"],
            overall_confidence=abs(avg_signal) * (sample_size / self.min_sample_size),
            signal_grade=grade,
            recommended_action=action,
            fusion_metadata={
                "method": "weighted_fallback",
                "reason": f"sample_size {sample_size} < {self.min_sample_size}",
                "signal_count": len(signals)
            }
        )

    def _determine_signal_grade(self, probs: Dict[str, float]) -> Tuple[str, str]:
        """确定信号等级"""
        max_prob = max(probs.values())
        dominant = max(probs, key=probs.get)

        if max_prob > 0.7 and probs["neutral"] < 0.3:
            grade = "strong"
        elif max_prob > 0.55:
            grade = "medium"
        elif max_prob > 0.45:
            grade = "weak"
        else:
            grade = "none"

        # 建议动作
        if dominant == "up" and grade in ["strong", "medium"]:
            action = "buy"
        elif dominant == "down" and grade in ["strong", "medium"]:
            action = "sell"
        else:
            action = "hold"

        return grade, action

    def _calculate_confidence(
        self,
        posterior: Dict[str, float],
        signals: List[Any]
    ) -> float:
        """计算整体置信度"""
        # 基于概率分布的熵计算置信度
        entropy = -sum(p * np.log(p + 1e-10) for p in posterior.values())
        max_entropy = np.log(3)  # 三分类最大熵

        confidence = 1 - (entropy / max_entropy)

        # 信号数量加成
        signal_bonus = min(0.2, len(signals) * 0.02)

        return min(1.0, confidence + signal_bonus)

    def _estimate_sample_size(self, ecosystem: Any) -> int:
        """估算当前生态的历史样本量"""
        # 简化：返回固定值，实际应从数据库查询
        return 2000  # 假设充足样本

    def _create_neutral_signal(self, reason: str) -> FusedSignal:
        """创建中性信号"""
        return FusedSignal(
            up_probability=self.prior["up"],
            down_probability=self.prior["down"],
            neutral_probability=self.prior["neutral"],
            overall_confidence=0.0,
            signal_grade="none",
            recommended_action="hold",
            fusion_metadata={"reason": reason}
        )