"""Unit tests for Bayesian accumulator (services/aggregation/bayesian.py). Tests uninformative prior, sigmoid gate values, entropy direction mapping, and core Bayesian posterior computation. Requirements: 1.1, 1.2, 1.3, 1.4, 1.5 """ from __future__ import annotations import pytest from services.aggregation.bayesian import ( PRIOR, compute_bayesian_posterior, compute_entropy, ) from services.aggregation.scoring import ( SignalWeight, WeightedSignal, sigmoid_gate, ) # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _make_signal( sentiment: float, combined_weight: float = 1.0, impact: float = 1.0, ) -> WeightedSignal: """Create a minimal WeightedSignal for testing.""" weight = SignalWeight( recency=1.0, credibility=1.0, novelty_bonus=0.0, confidence_gate=1.0, market_ctx_multiplier=1.0, combined=combined_weight, ) return WeightedSignal( document_id="test-doc", weight=weight, sentiment_value=sentiment, impact_score=impact, ) # --------------------------------------------------------------------------- # Uninformative prior (empty signals → P_bull=0.5, α=1, β=1, C=0) # --------------------------------------------------------------------------- class TestUninformativePrior: """Req 1.5: empty signals return the uninformative prior.""" def test_prior_p_bull(self): assert PRIOR.p_bull == 0.5 def test_prior_alpha(self): assert PRIOR.alpha == 1.0 def test_prior_beta(self): assert PRIOR.beta == 1.0 def test_prior_confidence(self): assert PRIOR.bayesian_confidence == 0.0 def test_prior_entropy(self): assert PRIOR.entropy == 1.0 def test_prior_signal_count(self): assert PRIOR.signal_count == 0 def test_empty_signals_return_prior(self): result = compute_bayesian_posterior([]) assert result == PRIOR def test_all_nan_signals_return_prior(self): sig = _make_signal(sentiment=float("nan"), combined_weight=1.0) result = compute_bayesian_posterior([sig]) assert result == PRIOR # --------------------------------------------------------------------------- # Sigmoid gate specific values (Req 2.1–2.4) # --------------------------------------------------------------------------- class TestSigmoidGateValues: """Test specific sigmoid gate values from the design doc.""" def test_midpoint_gives_half(self): """x=0.5 → gate=0.5 (sigmoid midpoint).""" assert sigmoid_gate(0.5, steepness=5.0, midpoint=0.5) == pytest.approx(0.5) def test_low_confidence_well_below_half(self): """x=0.2 → gate well below 0.5 (Req 2.3: below 0.2 → below 0.05). With default steepness=5.0, σ(5·(0.2-0.5)) = σ(-1.5) ≈ 0.18. The gate is significantly below the midpoint value of 0.5. For gate < 0.05, steepness would need to be higher or x lower. """ gate = sigmoid_gate(0.2, steepness=5.0, midpoint=0.5) assert gate < 0.5 # With higher steepness (e.g. 10), x=0.2 gives gate < 0.05 gate_steep = sigmoid_gate(0.2, steepness=10.0, midpoint=0.5) assert gate_steep < 0.05 def test_high_confidence_well_above_half(self): """x=0.8 → gate well above 0.5 (Req 2.4: above 0.8 → above 0.95). With default steepness=5.0, σ(5·(0.8-0.5)) = σ(1.5) ≈ 0.82. For gate > 0.95, steepness would need to be higher or x higher. """ gate = sigmoid_gate(0.8, steepness=5.0, midpoint=0.5) assert gate > 0.5 # With higher steepness (e.g. 10), x=0.8 gives gate > 0.95 gate_steep = sigmoid_gate(0.8, steepness=10.0, midpoint=0.5) assert gate_steep > 0.95 def test_zero_confidence(self): """x=0.0 → gate very close to 0.""" gate = sigmoid_gate(0.0, steepness=5.0, midpoint=0.5) assert gate < 0.1 def test_full_confidence(self): """x=1.0 → gate very close to 1.""" gate = sigmoid_gate(1.0, steepness=5.0, midpoint=0.5) assert gate > 0.9 # --------------------------------------------------------------------------- # Entropy direction mapping (Req 9.1–9.5) # --------------------------------------------------------------------------- class TestEntropyDirectionMapping: """Test entropy computation and the direction mapping rules.""" def test_entropy_at_half_is_one(self): """H(0.5) = 1.0 (maximum entropy).""" assert compute_entropy(0.5) == pytest.approx(1.0) def test_entropy_at_zero_is_zero(self): """H(0.0) = 0.0 (edge case).""" assert compute_entropy(0.0) == 0.0 def test_entropy_at_one_is_zero(self): """H(1.0) = 0.0 (edge case).""" assert compute_entropy(1.0) == 0.0 def test_entropy_symmetric(self): """H(p) = H(1-p) for all p.""" assert compute_entropy(0.3) == pytest.approx(compute_entropy(0.7)) def test_high_entropy_implies_mixed(self): """H > 0.9 → direction should be 'mixed'. When P_bull ≈ 0.5, entropy is near 1.0 → mixed. """ # P_bull = 0.5 → H = 1.0 > 0.9 → mixed h = compute_entropy(0.5) assert h > 0.9 def test_bullish_direction(self): """P_bull > 0.65 and H ≤ 0.9 → bullish. P_bull = 0.75 → H ≈ 0.811 < 0.9 → bullish. """ p_bull = 0.75 h = compute_entropy(p_bull) assert h <= 0.9 assert p_bull > 0.65 def test_bearish_direction(self): """P_bull < 0.35 and H ≤ 0.9 → bearish. P_bull = 0.2 → H ≈ 0.722 < 0.9 → bearish. """ p_bull = 0.2 h = compute_entropy(p_bull) assert h <= 0.9 assert p_bull < 0.35 def test_neutral_direction(self): """0.35 ≤ P_bull ≤ 0.65 and H ≤ 0.9 → neutral. P_bull = 0.4 → H ≈ 0.971 — actually > 0.9, so let's use 0.35. P_bull = 0.35 → H ≈ 0.934 — still > 0.9. P_bull = 0.65 → H ≈ 0.934 — still > 0.9. The neutral zone is narrow; use a value where H ≤ 0.9. Actually, H ≤ 0.9 requires P_bull ≤ ~0.28 or P_bull ≥ ~0.72. So the neutral zone (0.35–0.65 with H ≤ 0.9) is effectively empty in practice. This is by design — high entropy in the neutral zone forces 'mixed' classification. """ # Verify that the neutral zone with H ≤ 0.9 is very narrow # P_bull = 0.35 → H > 0.9 → would be classified as mixed, not neutral h_at_035 = compute_entropy(0.35) assert h_at_035 > 0.9 # confirms mixed, not neutral # --------------------------------------------------------------------------- # Bayesian posterior computation # --------------------------------------------------------------------------- class TestBayesianPosterior: """Test core Bayesian posterior computation.""" def test_single_bullish_signal(self): """One positive signal shifts P_bull above 0.5.""" sig = _make_signal(sentiment=1.0, combined_weight=1.0) result = compute_bayesian_posterior([sig]) assert result.p_bull > 0.5 assert result.alpha > 1.0 assert result.beta == 1.0 # no bearish weight assert result.signal_count == 1 def test_single_bearish_signal(self): """One negative signal shifts P_bull below 0.5.""" sig = _make_signal(sentiment=-1.0, combined_weight=1.0) result = compute_bayesian_posterior([sig]) assert result.p_bull < 0.5 assert result.alpha == 1.0 # no bullish weight assert result.beta > 1.0 assert result.signal_count == 1 def test_balanced_signals_near_prior(self): """Equal bullish and bearish signals keep P_bull near 0.5.""" signals = [ _make_signal(sentiment=1.0, combined_weight=1.0), _make_signal(sentiment=-1.0, combined_weight=1.0), ] result = compute_bayesian_posterior(signals) assert result.p_bull == pytest.approx(0.5, abs=0.01) def test_confidence_zero_when_balanced(self): """Equal α and β → confidence near 0.""" signals = [ _make_signal(sentiment=1.0, combined_weight=1.0), _make_signal(sentiment=-1.0, combined_weight=1.0), ] result = compute_bayesian_posterior(signals) # α = 2, β = 2 → C = 1 - 4*2*2/(2+2)^2 = 1 - 16/16 = 0 assert result.bayesian_confidence == pytest.approx(0.0, abs=0.01) def test_confidence_increases_with_agreement(self): """More agreeing signals → higher confidence.""" one_sig = compute_bayesian_posterior([ _make_signal(sentiment=1.0, combined_weight=1.0), ]) three_sigs = compute_bayesian_posterior([ _make_signal(sentiment=1.0, combined_weight=1.0), _make_signal(sentiment=1.0, combined_weight=1.0), _make_signal(sentiment=1.0, combined_weight=1.0), ]) assert three_sigs.bayesian_confidence > one_sig.bayesian_confidence def test_nan_weight_signal_skipped(self): """Signals with NaN weight are skipped.""" signals = [ _make_signal(sentiment=1.0, combined_weight=float("nan")), _make_signal(sentiment=1.0, combined_weight=1.0), ] result = compute_bayesian_posterior(signals) assert result.signal_count == 1 def test_entropy_decreases_with_strong_evidence(self): """Strong bullish evidence → low entropy.""" signals = [ _make_signal(sentiment=1.0, combined_weight=3.0), _make_signal(sentiment=1.0, combined_weight=3.0), ] result = compute_bayesian_posterior(signals) assert result.entropy < 0.5 # strong evidence → low entropy