360 lines
13 KiB
Python
360 lines
13 KiB
Python
"""Unit tests for v3 macro and competitive layers.
|
||
|
||
Tests the noisy-OR normalized macro exposure, resilience dampener,
|
||
macro LLR computation, shrunk correlation convergence, competitive LLR
|
||
clamping, and graph-distance attenuation.
|
||
|
||
Requirements validated: 9.1–9.5, 10.1–10.5
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import math
|
||
|
||
import pytest
|
||
|
||
from services.aggregation.interpolation import (
|
||
compute_macro_llr,
|
||
compute_normalized_macro_exposure,
|
||
)
|
||
from services.aggregation.signal_propagation import (
|
||
compute_competitive_llr,
|
||
compute_shrunk_correlation,
|
||
)
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Noisy-OR: compute_normalized_macro_exposure
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestNoisyORExposure:
|
||
"""Tests for noisy-OR normalized macro exposure (Req 9.1, 9.2, 9.3)."""
|
||
|
||
def test_all_overlaps_max_regional(self):
|
||
"""All O_k = 1.0 with regional tier → E_macro = 1.0."""
|
||
overlaps = {"geo": 1.0, "supply": 1.0, "commodity": 1.0, "sector": 1.0}
|
||
result = compute_normalized_macro_exposure(overlaps, tier="regional")
|
||
assert result == pytest.approx(1.0, abs=1e-9)
|
||
|
||
def test_all_overlaps_zero(self):
|
||
"""All O_k = 0 → E_macro = 0.0."""
|
||
overlaps = {"geo": 0.0, "supply": 0.0, "commodity": 0.0, "sector": 0.0}
|
||
result = compute_normalized_macro_exposure(overlaps, tier="regional")
|
||
assert result == pytest.approx(0.0, abs=1e-9)
|
||
|
||
def test_empty_overlaps(self):
|
||
"""Empty overlaps dict → E_macro = 0.0."""
|
||
result = compute_normalized_macro_exposure({}, tier="regional")
|
||
assert result == pytest.approx(0.0, abs=1e-9)
|
||
|
||
def test_single_dimension_geo(self):
|
||
"""Only geo overlap → partial exposure."""
|
||
overlaps = {"geo": 1.0}
|
||
result = compute_normalized_macro_exposure(overlaps, tier="regional")
|
||
# E_raw = 1 - (1-0.35*1)(1-0.25*0)(1-0.25*0)(1-0.15*0) = 1 - 0.65 = 0.35
|
||
# E_max = 1 - (0.65)(0.75)(0.75)(0.85) = 1 - 0.311484375 ≈ 0.688515625
|
||
# E_macro = 0.35 / 0.688515625 ≈ 0.508
|
||
expected_e_raw = 0.35
|
||
e_max = 1.0 - (0.65 * 0.75 * 0.75 * 0.85)
|
||
expected = expected_e_raw / e_max
|
||
assert result == pytest.approx(expected, rel=1e-6)
|
||
|
||
def test_partial_overlaps(self):
|
||
"""Partial overlaps produce intermediate exposure."""
|
||
overlaps = {"geo": 0.5, "supply": 0.3, "commodity": 0.0, "sector": 0.8}
|
||
result = compute_normalized_macro_exposure(overlaps, tier="regional")
|
||
# Should be between 0 and 1
|
||
assert 0.0 < result < 1.0
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Resilience dampener per tier
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestResilienceDampener:
|
||
"""Tests for resilience dampener application (Req 9.3)."""
|
||
|
||
def test_global_leader_dampener(self):
|
||
"""Global leader tier dampens exposure by 0.70."""
|
||
overlaps = {"geo": 1.0, "supply": 1.0, "commodity": 1.0, "sector": 1.0}
|
||
result = compute_normalized_macro_exposure(overlaps, tier="global_leader")
|
||
# E_macro = 1.0 * 0.70 = 0.70
|
||
assert result == pytest.approx(0.70, abs=1e-9)
|
||
|
||
def test_multinational_dampener(self):
|
||
"""Multinational tier dampens exposure by 0.85."""
|
||
overlaps = {"geo": 1.0, "supply": 1.0, "commodity": 1.0, "sector": 1.0}
|
||
result = compute_normalized_macro_exposure(overlaps, tier="multinational")
|
||
assert result == pytest.approx(0.85, abs=1e-9)
|
||
|
||
def test_regional_dampener(self):
|
||
"""Regional tier has no dampening (1.00)."""
|
||
overlaps = {"geo": 1.0, "supply": 1.0, "commodity": 1.0, "sector": 1.0}
|
||
result = compute_normalized_macro_exposure(overlaps, tier="regional")
|
||
assert result == pytest.approx(1.00, abs=1e-9)
|
||
|
||
def test_domestic_amplifier(self):
|
||
"""Domestic tier amplifies exposure by 1.20."""
|
||
overlaps = {"geo": 1.0, "supply": 1.0, "commodity": 1.0, "sector": 1.0}
|
||
result = compute_normalized_macro_exposure(overlaps, tier="domestic")
|
||
assert result == pytest.approx(1.20, abs=1e-9)
|
||
|
||
def test_unknown_tier_no_dampening(self):
|
||
"""Unknown tier defaults to 1.0 dampener."""
|
||
overlaps = {"geo": 1.0, "supply": 1.0, "commodity": 1.0, "sector": 1.0}
|
||
result = compute_normalized_macro_exposure(overlaps, tier="unknown_tier")
|
||
assert result == pytest.approx(1.00, abs=1e-9)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Macro LLR at boundary values
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestMacroLLR:
|
||
"""Tests for macro LLR computation (Req 9.4, 9.5)."""
|
||
|
||
def test_max_positive_inputs(self):
|
||
"""macro_impact=1, event_conf=1, q_recency=1, direction=+1 → p_macro=0.80."""
|
||
llr = compute_macro_llr(
|
||
macro_impact=1.0,
|
||
event_confidence=1.0,
|
||
q_recency=1.0,
|
||
macro_direction=1,
|
||
)
|
||
# p_macro = 0.50 + 0.30*1*1*1 = 0.80
|
||
expected = math.log(0.80 / 0.20) # ≈ 1.386
|
||
assert llr == pytest.approx(expected, rel=1e-6)
|
||
|
||
def test_max_negative_inputs(self):
|
||
"""All max with direction=-1 → negative LLR."""
|
||
llr = compute_macro_llr(
|
||
macro_impact=1.0,
|
||
event_confidence=1.0,
|
||
q_recency=1.0,
|
||
macro_direction=-1,
|
||
)
|
||
expected = -math.log(0.80 / 0.20)
|
||
assert llr == pytest.approx(expected, rel=1e-6)
|
||
|
||
def test_neutral_direction_zero_llr(self):
|
||
"""direction=0 → LLR=0.0 regardless of other inputs."""
|
||
llr = compute_macro_llr(
|
||
macro_impact=1.0,
|
||
event_confidence=1.0,
|
||
q_recency=1.0,
|
||
macro_direction=0,
|
||
)
|
||
assert llr == 0.0
|
||
|
||
def test_minimum_p_macro_clamp(self):
|
||
"""All impact factors zero → p_macro clamped to 0.501."""
|
||
llr = compute_macro_llr(
|
||
macro_impact=0.0,
|
||
event_confidence=0.0,
|
||
q_recency=0.0,
|
||
macro_direction=1,
|
||
)
|
||
# p_macro = 0.50 + 0 = 0.50, clamped up to 0.501
|
||
expected = math.log(0.501 / (1.0 - 0.501))
|
||
assert llr == pytest.approx(expected, rel=1e-6)
|
||
|
||
def test_mid_range_inputs(self):
|
||
"""Intermediate inputs produce reasonable LLR."""
|
||
llr = compute_macro_llr(
|
||
macro_impact=0.5,
|
||
event_confidence=0.7,
|
||
q_recency=0.8,
|
||
macro_direction=1,
|
||
)
|
||
# p_macro = 0.50 + 0.30 * 0.5 * 0.7 * 0.8 = 0.50 + 0.084 = 0.584
|
||
p_macro = 0.584
|
||
expected = math.log(p_macro / (1.0 - p_macro))
|
||
assert llr == pytest.approx(expected, rel=1e-6)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Shrunk correlation convergence
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestShrunkCorrelation:
|
||
"""Tests for shrinkage-adjusted correlation (Req 10.1, 10.2)."""
|
||
|
||
def test_large_n_approaches_rho_rolling(self):
|
||
"""n=1000 with same_sector → result ≈ rho_rolling."""
|
||
rho_rolling = 0.65
|
||
result = compute_shrunk_correlation(
|
||
rho_rolling=rho_rolling,
|
||
n_observations=1000,
|
||
same_sector=True,
|
||
)
|
||
# (1000/1030) × 0.65 + (30/1030) × 0.30 ≈ 0.6311 + 0.00874 ≈ 0.6398
|
||
weight_data = 1000 / 1030
|
||
weight_prior = 30 / 1030
|
||
expected = weight_data * rho_rolling + weight_prior * 0.30
|
||
assert result == pytest.approx(expected, rel=1e-6)
|
||
# Should be close to rho_rolling
|
||
assert abs(result - rho_rolling) < 0.02
|
||
|
||
def test_zero_observations_returns_prior(self):
|
||
"""n=0 → result = prior (same_sector: 0.30, cross_sector: 0.10)."""
|
||
# same sector
|
||
result_same = compute_shrunk_correlation(
|
||
rho_rolling=0.9,
|
||
n_observations=0,
|
||
same_sector=True,
|
||
)
|
||
# (0/30) × 0.9 + (30/30) × 0.30 = 0.30
|
||
assert result_same == pytest.approx(0.30, abs=1e-9)
|
||
|
||
# cross sector
|
||
result_cross = compute_shrunk_correlation(
|
||
rho_rolling=0.9,
|
||
n_observations=0,
|
||
same_sector=False,
|
||
)
|
||
# (0/30) × 0.9 + (30/30) × 0.10 = 0.10
|
||
assert result_cross == pytest.approx(0.10, abs=1e-9)
|
||
|
||
def test_cross_sector_prior(self):
|
||
"""Cross-sector uses prior = 0.10."""
|
||
result = compute_shrunk_correlation(
|
||
rho_rolling=0.50,
|
||
n_observations=30,
|
||
same_sector=False,
|
||
)
|
||
# (30/60) × 0.50 + (30/60) × 0.10 = 0.25 + 0.05 = 0.30
|
||
assert result == pytest.approx(0.30, abs=1e-9)
|
||
|
||
def test_negative_rolling_floored_at_zero(self):
|
||
"""Negative rolling correlation → rho_effective floored at 0."""
|
||
result = compute_shrunk_correlation(
|
||
rho_rolling=-0.50,
|
||
n_observations=100,
|
||
same_sector=False,
|
||
)
|
||
# (100/130)×(-0.50) + (30/130)×0.10 = -0.3846 + 0.0231 ≈ -0.3615
|
||
# Floored at 0
|
||
assert result == 0.0
|
||
|
||
def test_n_30_equal_weight(self):
|
||
"""n=30 → data and prior have equal weight."""
|
||
rho_rolling = 0.80
|
||
result = compute_shrunk_correlation(
|
||
rho_rolling=rho_rolling,
|
||
n_observations=30,
|
||
same_sector=True,
|
||
)
|
||
# (30/60)×0.80 + (30/60)×0.30 = 0.40 + 0.15 = 0.55
|
||
assert result == pytest.approx(0.55, abs=1e-9)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Competitive LLR clamp at ±1.25
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestCompetitiveLLRClamp:
|
||
"""Tests for competitive LLR clamping (Req 10.3, 10.4, 10.5)."""
|
||
|
||
def test_large_positive_clamped(self):
|
||
"""Large positive inputs → clamped to +1.25."""
|
||
result = compute_competitive_llr(
|
||
llr_source=10.0,
|
||
rho_effective=0.9,
|
||
d_network=1,
|
||
pattern_confidence=1.0,
|
||
)
|
||
assert result == pytest.approx(1.25, abs=1e-9)
|
||
|
||
def test_large_negative_clamped(self):
|
||
"""Large negative inputs → clamped to -1.25."""
|
||
result = compute_competitive_llr(
|
||
llr_source=-10.0,
|
||
rho_effective=0.9,
|
||
d_network=1,
|
||
pattern_confidence=1.0,
|
||
)
|
||
assert result == pytest.approx(-1.25, abs=1e-9)
|
||
|
||
def test_within_bounds_not_clamped(self):
|
||
"""Small inputs produce unclamped result."""
|
||
# attenuation = 0.5 × exp(-0.85 × 1) ≈ 0.5 × 0.4274 ≈ 0.2137
|
||
# LLR_competitive = 1.0 × 0.2137 × 0.8 ≈ 0.1710
|
||
result = compute_competitive_llr(
|
||
llr_source=1.0,
|
||
rho_effective=0.5,
|
||
d_network=1,
|
||
pattern_confidence=0.8,
|
||
)
|
||
expected = 1.0 * 0.5 * math.exp(-0.85 * 1) * 0.8
|
||
assert result == pytest.approx(expected, rel=1e-6)
|
||
assert abs(result) < 1.25
|
||
|
||
def test_zero_rho_gives_zero(self):
|
||
"""Zero correlation → zero competitive LLR."""
|
||
result = compute_competitive_llr(
|
||
llr_source=5.0,
|
||
rho_effective=0.0,
|
||
d_network=1,
|
||
pattern_confidence=1.0,
|
||
)
|
||
assert result == pytest.approx(0.0, abs=1e-9)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Distance > 3 → zero attenuation
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestDistanceAttenuation:
|
||
"""Tests for graph distance cutoff (Req 10.5)."""
|
||
|
||
def test_distance_4_returns_zero(self):
|
||
"""d_network=4 → LLR_competitive = 0.0."""
|
||
result = compute_competitive_llr(
|
||
llr_source=5.0,
|
||
rho_effective=0.9,
|
||
d_network=4,
|
||
pattern_confidence=1.0,
|
||
)
|
||
assert result == 0.0
|
||
|
||
def test_distance_10_returns_zero(self):
|
||
"""Very large distance → LLR_competitive = 0.0."""
|
||
result = compute_competitive_llr(
|
||
llr_source=5.0,
|
||
rho_effective=0.9,
|
||
d_network=10,
|
||
pattern_confidence=1.0,
|
||
)
|
||
assert result == 0.0
|
||
|
||
def test_distance_3_still_active(self):
|
||
"""d_network=3 (max allowed) → non-zero result."""
|
||
result = compute_competitive_llr(
|
||
llr_source=2.0,
|
||
rho_effective=0.8,
|
||
d_network=3,
|
||
pattern_confidence=0.9,
|
||
)
|
||
# attenuation = 0.8 × exp(-0.85 × 3) ≈ 0.8 × 0.0776 ≈ 0.0621
|
||
# LLR_competitive = 2.0 × 0.0621 × 0.9 ≈ 0.1118
|
||
expected = 2.0 * 0.8 * math.exp(-0.85 * 3) * 0.9
|
||
assert result == pytest.approx(expected, rel=1e-6)
|
||
assert result != 0.0
|
||
|
||
def test_distance_1_strongest(self):
|
||
"""d_network=1 gives strongest attenuation (least decay)."""
|
||
result_d1 = compute_competitive_llr(
|
||
llr_source=2.0, rho_effective=0.8, d_network=1, pattern_confidence=0.9,
|
||
)
|
||
result_d2 = compute_competitive_llr(
|
||
llr_source=2.0, rho_effective=0.8, d_network=2, pattern_confidence=0.9,
|
||
)
|
||
result_d3 = compute_competitive_llr(
|
||
llr_source=2.0, rho_effective=0.8, d_network=3, pattern_confidence=0.9,
|
||
)
|
||
assert result_d1 > result_d2 > result_d3 > 0.0
|