"""Bayesian accumulator for probabilistic sentiment aggregation. Accumulates weighted signals into a Bayesian posterior using log-likelihood accumulation, Beta distribution parameters, and Shannon entropy for mixed-signal detection. Requirements: 1.1, 1.2, 1.3, 1.4, 1.5, 1.6, 9.1, 9.7 V3 posterior assembly: 5.1, 5.2, 5.3, 5.4, 5.5, 5.6, 5.7 """ from __future__ import annotations import math from dataclasses import dataclass from typing import TYPE_CHECKING from services.aggregation.scoring import WeightedSignal if TYPE_CHECKING: from services.aggregation.regime import V3RegimeClassification from services.aggregation.worker import EvidenceCluster @dataclass(frozen=True) class BayesianPosterior: """Bayesian posterior state from signal accumulation.""" p_bull: float # σ(L_t), bullish probability [0, 1] alpha: float # Beta distribution α parameter (≥ 1.0) beta: float # Beta distribution β parameter (≥ 1.0) log_likelihood: float # Raw log-likelihood accumulation L_t bayesian_confidence: float # 1 - 4αβ/(α+β)², [0, 1] entropy: float # Shannon entropy H, [0, 1] signal_count: int # Number of signals processed # Uninformative prior (no evidence) PRIOR = BayesianPosterior( p_bull=0.5, alpha=1.0, beta=1.0, log_likelihood=0.0, bayesian_confidence=0.0, entropy=1.0, signal_count=0, ) def compute_entropy(p_bull: float) -> float: """Shannon entropy H = -p·log₂(p) - (1-p)·log₂(1-p). Returns value in [0, 1]. Maximum at p=0.5, zero at p=0 or p=1. Handles edge cases p≤0 and p≥1 by returning 0.0. """ if p_bull <= 0.0 or p_bull >= 1.0: return 0.0 q = 1.0 - p_bull return -(p_bull * math.log2(p_bull) + q * math.log2(q)) def compute_bayesian_posterior( signals: list[WeightedSignal], ) -> BayesianPosterior: """Accumulate weighted signals into a Bayesian posterior. Computes: - Log-likelihood: L_t = Σ(w_i · s_i) - Bullish probability: P_bull = σ(L_t) - Beta posterior: α = 1 + W_bull, β = 1 + W_bear - Bayesian confidence: C = 1 - 4αβ/(α+β)² - Shannon entropy: H = -p·log₂(p) - (1-p)·log₂(1-p) Returns PRIOR for empty signal lists. Skips signals with NaN weight or sentiment. """ if not signals: return PRIOR log_likelihood = 0.0 w_bull = 0.0 w_bear = 0.0 count = 0 for sig in signals: combined = sig.weight.combined sentiment = sig.sentiment_value # Skip signals with NaN weight or sentiment if math.isnan(combined) or math.isnan(sentiment): continue log_likelihood += combined * sentiment if sentiment > 0.0: w_bull += combined elif sentiment < 0.0: w_bear += combined count += 1 if count == 0: return PRIOR # P_bull via sigmoid: σ(L_t) = 1 / (1 + exp(-L_t)) # Guard against overflow in exp for very large |L_t| if log_likelihood > 500.0: p_bull = 1.0 elif log_likelihood < -500.0: p_bull = 0.0 else: p_bull = 1.0 / (1.0 + math.exp(-log_likelihood)) # Beta posterior parameters alpha = 1.0 + w_bull beta_param = 1.0 + w_bear # Bayesian confidence: C = 1 - 4αβ/(α+β)² ab_sum = alpha + beta_param bayesian_confidence = 1.0 - (4.0 * alpha * beta_param) / (ab_sum * ab_sum) # Clamp to [0, 1] to guard against floating-point rounding bayesian_confidence = max(0.0, min(1.0, bayesian_confidence)) # Shannon entropy entropy = compute_entropy(p_bull) return BayesianPosterior( p_bull=p_bull, alpha=alpha, beta=beta_param, log_likelihood=log_likelihood, bayesian_confidence=bayesian_confidence, entropy=entropy, signal_count=count, ) # --------------------------------------------------------------------------- # V3 Posterior Assembly — Calibrated Evidence Engine # Requirements: 5.1, 5.2, 5.3, 5.4, 5.5, 5.6, 5.7 # --------------------------------------------------------------------------- @dataclass(frozen=True) class V3Posterior: """V3 posterior result from log-odds Bayesian assembly. Attributes: p_up: Posterior probability of upward move, (0, 1). p_down: 1 - p_up. log_odds: Raw log-odds (logit) of P_up. strength: abs(2 × P_up - 1), signal conviction [0, 1]. direction: Classified direction string ('bullish', 'bearish', 'neutral'). n_eff_total: Total effective evidence count across all clusters. regime: Market regime string used for this computation. """ p_up: float p_down: float log_odds: float strength: float direction: str n_eff_total: float regime: str # Regime-specific direction thresholds: (bullish_threshold, bearish_threshold) # P_up >= bullish → "bullish"; P_up <= bearish → "bearish"; else "neutral" _V3_DIRECTION_THRESHOLDS: dict[str, tuple[float, float]] = { "panic": (0.68, 0.32), "trend_following": (0.60, 0.40), "mean_reversion": (0.63, 0.37), "uncertainty": (0.65, 0.35), } def _logit(p: float) -> float: """Compute logit = ln(p / (1-p)) with boundary guard.""" p = max(1e-10, min(1 - 1e-10, p)) return math.log(p / (1 - p)) def _sigmoid(x: float) -> float: """Compute sigmoid = 1 / (1 + exp(-x)) with overflow guard.""" if x > 500: return 1.0 if x < -500: return 0.0 return 1.0 / (1.0 + math.exp(-x)) def compute_v3_posterior( clusters: list[EvidenceCluster], regime: V3RegimeClassification, p_prior: float = 0.50, ) -> V3Posterior: """Assemble v3 posterior via log-odds accumulation. Computes: logit(P_up) = logit(P_prior) + sum(gamma_regime × LLR_c) P_up = sigmoid(log_odds), clamped to [1e-10, 1 - 1e-10] strength = abs(2 × P_up - 1) direction via regime-specific thresholds Args: clusters: List of EvidenceCluster objects with computed cluster_llr. regime: V3RegimeClassification providing evidence_multiplier and regime. p_prior: Calibrated prior probability, clamped to [0.40, 0.60]. Returns: V3Posterior with computed posterior fields. Requirements: 5.1, 5.2, 5.3, 5.4, 5.5, 5.6, 5.7 """ # Clamp prior to [0.40, 0.60] (Req 5.6) p_prior = max(0.40, min(0.60, p_prior)) # Gamma: regime-specific evidence multiplier (Req 5.2) gamma = regime.evidence_multiplier # Compute log-odds: logit(P_prior) + sum(gamma × LLR_c) (Req 5.1, 5.2) log_odds = _logit(p_prior) + sum(gamma * c.cluster_llr for c in clusters) # Compute P_up via sigmoid (Req 5.3) p_up = _sigmoid(log_odds) # Clamp to open interval (Req 5.3) p_up = max(1e-10, min(1 - 1e-10, p_up)) p_down = 1.0 - p_up # Strength = |2 × P_up - 1| (Req 5.4) strength = abs(2.0 * p_up - 1.0) # n_eff_total = sum of cluster n_eff (Req 5.5) n_eff_total = sum(c.n_eff for c in clusters) # Classify direction using regime-specific thresholds (Req 5.5) regime_key = regime.regime.value # MarketRegime enum → string bull_thresh, bear_thresh = _V3_DIRECTION_THRESHOLDS.get( regime_key, (0.65, 0.35) ) if p_up >= bull_thresh: direction = "bullish" elif p_up <= bear_thresh: direction = "bearish" else: direction = "neutral" return V3Posterior( p_up=p_up, p_down=p_down, log_odds=log_odds, strength=strength, direction=direction, n_eff_total=n_eff_total, regime=regime_key, )