350 lines
11 KiB
Python
350 lines
11 KiB
Python
"""Unit tests for v3 multiplicative confidence and data quality.
|
||
|
||
Tests for compute_v3_confidence, compute_v3_data_quality, and
|
||
should_force_informational_v3 functions.
|
||
|
||
Requirements validated: 8.1–8.5, 17.1–17.8
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import math
|
||
from datetime import datetime, timezone
|
||
|
||
import pytest
|
||
|
||
from services.aggregation.scoring import EvidenceUnit
|
||
from services.aggregation.worker import (
|
||
compute_v3_confidence,
|
||
compute_v3_data_quality,
|
||
should_force_informational_v3,
|
||
)
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Helpers
|
||
# ---------------------------------------------------------------------------
|
||
|
||
_NOW = datetime(2025, 1, 15, 12, 0, 0, tzinfo=timezone.utc)
|
||
|
||
|
||
def _make_unit(
|
||
layer: str = "company",
|
||
extraction_conf: float = 0.8,
|
||
impact: float = 0.7,
|
||
) -> EvidenceUnit:
|
||
"""Create a minimal EvidenceUnit for testing."""
|
||
return EvidenceUnit(
|
||
symbol="AAPL",
|
||
layer=layer,
|
||
event_type="earnings",
|
||
source_id="doc-1",
|
||
source_group="company",
|
||
timestamp=_NOW,
|
||
horizon="7d",
|
||
direction=1,
|
||
sentiment_strength=0.8,
|
||
impact=impact,
|
||
extraction_conf=extraction_conf,
|
||
source_cred=0.85,
|
||
novelty=0.9,
|
||
event_base_rate=0.25,
|
||
cluster_id="test-cluster",
|
||
)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Test: Zero data quality → zero confidence
|
||
# Requirement: 8.1, 8.5
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestZeroDataQualityConfidence:
|
||
"""When data_quality = 0.0, confidence must be 0.0."""
|
||
|
||
def test_zero_data_quality_produces_zero_confidence(self):
|
||
confidence = compute_v3_confidence(
|
||
n_eff_total=5.0,
|
||
q_values=[0.8, 0.7],
|
||
llrs=[1.0, 0.5],
|
||
strength=0.6,
|
||
regime_confidence_mult=1.0,
|
||
contradiction=0.0,
|
||
data_quality=0.0,
|
||
)
|
||
assert confidence == 0.0
|
||
|
||
def test_near_zero_data_quality_suppresses_confidence(self):
|
||
"""Very low data_quality → near-zero confidence."""
|
||
confidence = compute_v3_confidence(
|
||
n_eff_total=10.0,
|
||
q_values=[0.9, 0.9],
|
||
llrs=[1.5, 1.5],
|
||
strength=0.8,
|
||
regime_confidence_mult=1.0,
|
||
contradiction=0.0,
|
||
data_quality=0.01,
|
||
)
|
||
assert confidence < 0.05
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Test: Full contradiction (1.0) → zero confidence
|
||
# Requirement: 8.4
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestFullContradictionConfidence:
|
||
"""When contradiction = 1.0, confidence must be 0.0."""
|
||
|
||
def test_full_contradiction_produces_zero_confidence(self):
|
||
confidence = compute_v3_confidence(
|
||
n_eff_total=10.0,
|
||
q_values=[0.9, 0.8],
|
||
llrs=[1.0, 1.2],
|
||
strength=0.7,
|
||
regime_confidence_mult=1.0,
|
||
contradiction=1.0,
|
||
data_quality=0.9,
|
||
)
|
||
assert confidence == 0.0
|
||
|
||
def test_high_contradiction_suppresses_confidence(self):
|
||
"""Contradiction = 0.9 → confidence heavily suppressed."""
|
||
conf_no_contra = compute_v3_confidence(
|
||
n_eff_total=5.0,
|
||
q_values=[0.8],
|
||
llrs=[1.0],
|
||
strength=0.6,
|
||
regime_confidence_mult=1.0,
|
||
contradiction=0.0,
|
||
data_quality=0.8,
|
||
)
|
||
conf_high_contra = compute_v3_confidence(
|
||
n_eff_total=5.0,
|
||
q_values=[0.8],
|
||
llrs=[1.0],
|
||
strength=0.6,
|
||
regime_confidence_mult=1.0,
|
||
contradiction=0.9,
|
||
data_quality=0.8,
|
||
)
|
||
assert conf_high_contra < conf_no_contra * 0.15
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Test: Low n_eff → suppressed C_evidence
|
||
# Requirement: 8.2
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestLowNEffConfidence:
|
||
"""Low n_eff_total → C_evidence is suppressed."""
|
||
|
||
def test_very_low_n_eff_suppresses_c_evidence(self):
|
||
"""n_eff_total=0.5 → C_evidence = 1 - exp(-0.1) ≈ 0.095."""
|
||
# C_evidence = 1 - exp(-0.5 / 5.0) = 1 - exp(-0.1) ≈ 0.0952
|
||
expected_c_evidence = 1.0 - math.exp(-0.1)
|
||
assert expected_c_evidence == pytest.approx(0.0952, rel=1e-2)
|
||
|
||
confidence = compute_v3_confidence(
|
||
n_eff_total=0.5,
|
||
q_values=[0.8],
|
||
llrs=[1.0],
|
||
strength=0.8,
|
||
regime_confidence_mult=1.0,
|
||
contradiction=0.0,
|
||
data_quality=0.9,
|
||
)
|
||
# Confidence is bounded by C_evidence ≈ 0.095
|
||
assert confidence < 0.15
|
||
|
||
def test_high_n_eff_yields_higher_confidence(self):
|
||
"""Higher n_eff → higher C_evidence → higher overall confidence."""
|
||
conf_low = compute_v3_confidence(
|
||
n_eff_total=1.0,
|
||
q_values=[0.8],
|
||
llrs=[1.0],
|
||
strength=0.6,
|
||
regime_confidence_mult=1.0,
|
||
contradiction=0.0,
|
||
data_quality=0.8,
|
||
)
|
||
conf_high = compute_v3_confidence(
|
||
n_eff_total=10.0,
|
||
q_values=[0.8],
|
||
llrs=[1.0],
|
||
strength=0.6,
|
||
regime_confidence_mult=1.0,
|
||
contradiction=0.0,
|
||
data_quality=0.8,
|
||
)
|
||
assert conf_high > conf_low
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Test: Data quality computed from known inputs
|
||
# Requirement: 17.1, 17.2, 17.3, 17.4, 17.5, 17.6
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestDataQualityComputation:
|
||
"""Verify data quality formula with known inputs."""
|
||
|
||
def test_high_quality_inputs(self):
|
||
"""Zero failure, fresh signal, many sources → high quality."""
|
||
units = [_make_unit() for _ in range(10)]
|
||
dq = compute_v3_data_quality(
|
||
units=units,
|
||
extraction_failure_rate=0.0,
|
||
age_newest_hours=1.0,
|
||
n_source_types=4,
|
||
)
|
||
# Q_parse=1.0, Q_fresh=exp(-1/168)≈0.994, Q_coverage=1-exp(-2)≈0.865,
|
||
# Q_diversity=min(1, log2(5)/log2(4))=1.0
|
||
assert dq > 0.60
|
||
|
||
def test_high_extraction_failure_rate(self):
|
||
"""extraction_failure_rate=0.8 → Q_parse=0.2 → low quality."""
|
||
units = [_make_unit() for _ in range(5)]
|
||
dq = compute_v3_data_quality(
|
||
units=units,
|
||
extraction_failure_rate=0.8,
|
||
age_newest_hours=1.0,
|
||
n_source_types=3,
|
||
)
|
||
# Q_parse = 0.2 → heavily suppresses data_quality
|
||
assert dq < 0.30
|
||
|
||
def test_zero_sources_zero_diversity(self):
|
||
"""No source types → Q_diversity = 0 → data_quality = 0."""
|
||
units = [_make_unit() for _ in range(5)]
|
||
dq = compute_v3_data_quality(
|
||
units=units,
|
||
extraction_failure_rate=0.0,
|
||
age_newest_hours=1.0,
|
||
n_source_types=0,
|
||
)
|
||
assert dq == 0.0
|
||
|
||
def test_empty_units_low_coverage(self):
|
||
"""No valid units → Q_coverage = 1 - exp(0) = 0 → data_quality = 0."""
|
||
dq = compute_v3_data_quality(
|
||
units=[],
|
||
extraction_failure_rate=0.0,
|
||
age_newest_hours=1.0,
|
||
n_source_types=3,
|
||
)
|
||
assert dq == 0.0
|
||
|
||
def test_stale_signal_decays_quality(self):
|
||
"""Very old signal → Q_fresh low → suppresses data_quality."""
|
||
units = [_make_unit() for _ in range(5)]
|
||
dq_fresh = compute_v3_data_quality(
|
||
units=units,
|
||
extraction_failure_rate=0.0,
|
||
age_newest_hours=1.0,
|
||
n_source_types=3,
|
||
)
|
||
dq_stale = compute_v3_data_quality(
|
||
units=units,
|
||
extraction_failure_rate=0.0,
|
||
age_newest_hours=500.0,
|
||
n_source_types=3,
|
||
)
|
||
assert dq_stale < dq_fresh
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Test: Force informational mode
|
||
# Requirement: 17.7, 17.8
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestForceInformational:
|
||
"""Test should_force_informational_v3 forcing conditions."""
|
||
|
||
def test_data_quality_below_threshold_forces(self):
|
||
"""data_quality < 0.50 → forces informational."""
|
||
units = [_make_unit() for _ in range(5)]
|
||
forced, reason = should_force_informational_v3(
|
||
data_quality=0.49,
|
||
units=units,
|
||
extraction_failure_rate=0.0,
|
||
)
|
||
assert forced is True
|
||
assert reason == "data_quality_below_threshold"
|
||
|
||
def test_data_quality_at_threshold_does_not_force(self):
|
||
"""data_quality = 0.50 → does NOT force informational."""
|
||
units = [_make_unit() for _ in range(5)]
|
||
forced, reason = should_force_informational_v3(
|
||
data_quality=0.50,
|
||
units=units,
|
||
extraction_failure_rate=0.0,
|
||
)
|
||
assert forced is False
|
||
|
||
def test_insufficient_evidence_forces(self):
|
||
"""N_valid < 2 → forces informational."""
|
||
units = [_make_unit()]
|
||
forced, reason = should_force_informational_v3(
|
||
data_quality=0.80,
|
||
units=units,
|
||
extraction_failure_rate=0.0,
|
||
)
|
||
assert forced is True
|
||
assert reason == "insufficient_evidence_count"
|
||
|
||
def test_high_extraction_failure_forces(self):
|
||
"""extraction_failure_rate > 0.50 → Q_parse < 0.50 → forces."""
|
||
units = [_make_unit() for _ in range(5)]
|
||
forced, reason = should_force_informational_v3(
|
||
data_quality=0.80,
|
||
units=units,
|
||
extraction_failure_rate=0.51,
|
||
)
|
||
assert forced is True
|
||
assert reason == "extraction_parse_rate_below_threshold"
|
||
|
||
def test_only_macro_signals_forces(self):
|
||
"""Only macro/competitive evidence (no company) → forces."""
|
||
units = [_make_unit(layer="macro"), _make_unit(layer="competitive")]
|
||
forced, reason = should_force_informational_v3(
|
||
data_quality=0.80,
|
||
units=units,
|
||
extraction_failure_rate=0.0,
|
||
)
|
||
assert forced is True
|
||
assert reason == "macro_competitive_only_evidence"
|
||
|
||
def test_only_macro_with_macro_only_enabled_does_not_force(self):
|
||
"""Only macro evidence WITH macro_only_enabled → does NOT force."""
|
||
units = [_make_unit(layer="macro"), _make_unit(layer="macro")]
|
||
forced, reason = should_force_informational_v3(
|
||
data_quality=0.80,
|
||
units=units,
|
||
extraction_failure_rate=0.0,
|
||
macro_only_enabled=True,
|
||
)
|
||
assert forced is False
|
||
|
||
def test_mixed_signals_does_not_force(self):
|
||
"""Company + macro evidence → does NOT force."""
|
||
units = [_make_unit(layer="company"), _make_unit(layer="macro")]
|
||
forced, reason = should_force_informational_v3(
|
||
data_quality=0.80,
|
||
units=units,
|
||
extraction_failure_rate=0.0,
|
||
)
|
||
assert forced is False
|
||
|
||
def test_good_inputs_do_not_force(self):
|
||
"""All good → no forcing."""
|
||
units = [_make_unit() for _ in range(5)]
|
||
forced, reason = should_force_informational_v3(
|
||
data_quality=0.80,
|
||
units=units,
|
||
extraction_failure_rate=0.1,
|
||
)
|
||
assert forced is False
|
||
assert reason == ""
|