566 lines
20 KiB
Python
566 lines
20 KiB
Python
"""Unit tests for EvidenceUnit normalization and LLR conversion.
|
||
|
||
Validates: Requirements 1.1–1.8, 2.1–2.9, 3.1–3.6
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import math
|
||
from datetime import datetime, timezone
|
||
|
||
from services.aggregation.scoring import (
|
||
EvidenceUnit,
|
||
SourceStats,
|
||
compute_llr,
|
||
compute_v3_reliability,
|
||
normalize_company_signal,
|
||
normalize_competitive_signal,
|
||
normalize_macro_signal,
|
||
)
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Fixtures
|
||
# ---------------------------------------------------------------------------
|
||
|
||
_NOW = datetime(2025, 1, 15, 12, 0, 0, tzinfo=timezone.utc)
|
||
|
||
|
||
def _make_company_signal(**overrides) -> dict:
|
||
"""Create a minimal valid company signal dict."""
|
||
base = {
|
||
"symbol": "AAPL",
|
||
"timestamp": _NOW,
|
||
"source_id": "doc-001",
|
||
"event_type": "earnings",
|
||
"source_group": "company",
|
||
"horizon": "7d",
|
||
"sentiment": "positive",
|
||
"sentiment_strength": 0.8,
|
||
"impact": 0.7,
|
||
"extraction_conf": 0.9,
|
||
"source_cred": 0.85,
|
||
"novelty": 0.9,
|
||
}
|
||
base.update(overrides)
|
||
return base
|
||
|
||
|
||
def _make_macro_signal(**overrides) -> dict:
|
||
"""Create a minimal valid macro signal dict."""
|
||
base = {
|
||
"symbol": "MSFT",
|
||
"timestamp": _NOW,
|
||
"source_id": "event-100",
|
||
"event_type": "regulatory",
|
||
"impact_direction": "positive",
|
||
"macro_impact_score": 0.6,
|
||
"event_confidence": 0.75,
|
||
"estimated_duration": "medium_term",
|
||
}
|
||
base.update(overrides)
|
||
return base
|
||
|
||
|
||
def _make_competitive_signal(**overrides) -> dict:
|
||
"""Create a minimal valid competitive signal dict."""
|
||
base = {
|
||
"symbol": "GOOG",
|
||
"timestamp": _NOW,
|
||
"source_id": "comp-doc-55",
|
||
"event_type": "product_launch",
|
||
"signal_direction": "bearish",
|
||
"signal_strength": 0.7,
|
||
"relationship_strength": 0.8,
|
||
"pattern_confidence": 0.65,
|
||
"time_horizon": "30d",
|
||
}
|
||
base.update(overrides)
|
||
return base
|
||
|
||
|
||
# ===========================================================================
|
||
# TestNormalizeCompanySignal
|
||
# ===========================================================================
|
||
|
||
|
||
class TestNormalizeCompanySignal:
|
||
"""Test normalize_company_signal mapping and validation."""
|
||
|
||
def test_full_company_signal(self):
|
||
"""A complete company signal maps all fields correctly."""
|
||
sig = _make_company_signal()
|
||
eu = normalize_company_signal(sig)
|
||
|
||
assert eu is not None
|
||
assert eu.symbol == "AAPL"
|
||
assert eu.layer == "company"
|
||
assert eu.event_type == "earnings"
|
||
assert eu.source_id == "doc-001"
|
||
assert eu.source_group == "company"
|
||
assert eu.timestamp == _NOW
|
||
assert eu.horizon == "7d"
|
||
assert eu.direction == 1 # "positive" → +1
|
||
assert eu.sentiment_strength == 0.8
|
||
assert eu.impact == 0.7
|
||
assert eu.extraction_conf == 0.9
|
||
assert eu.source_cred == 0.85
|
||
assert eu.novelty == 0.9
|
||
assert eu.event_base_rate == 0.25 # earnings base rate
|
||
assert len(eu.cluster_id) == 16 # sha256 hex prefix
|
||
|
||
def test_missing_symbol_rejected(self):
|
||
"""Missing symbol → returns None with warning."""
|
||
sig = _make_company_signal(symbol=None)
|
||
assert normalize_company_signal(sig) is None
|
||
|
||
def test_missing_timestamp_rejected(self):
|
||
"""Missing timestamp → returns None with warning."""
|
||
sig = _make_company_signal(timestamp=None)
|
||
assert normalize_company_signal(sig) is None
|
||
|
||
def test_missing_source_id_rejected(self):
|
||
"""Missing source_id → returns None with warning."""
|
||
sig = _make_company_signal(source_id=None)
|
||
assert normalize_company_signal(sig) is None
|
||
|
||
def test_empty_string_symbol_rejected(self):
|
||
"""Empty string symbol → returns None (falsy check)."""
|
||
sig = _make_company_signal(symbol="")
|
||
assert normalize_company_signal(sig) is None
|
||
|
||
def test_direction_mappings(self):
|
||
"""Direction string mappings: positive→+1, negative→-1, neutral→0."""
|
||
for sentiment, expected in [
|
||
("positive", 1),
|
||
("negative", -1),
|
||
("neutral", 0),
|
||
("bullish", 1),
|
||
("bearish", -1),
|
||
("mixed", 0),
|
||
]:
|
||
eu = normalize_company_signal(_make_company_signal(sentiment=sentiment))
|
||
assert eu is not None
|
||
assert eu.direction == expected, f"'{sentiment}' should map to {expected}"
|
||
|
||
def test_missing_optional_fields_default_0_5(self):
|
||
"""Missing optional numeric fields substitute 0.5."""
|
||
sig = {
|
||
"symbol": "TSLA",
|
||
"timestamp": _NOW,
|
||
"source_id": "doc-xyz",
|
||
}
|
||
eu = normalize_company_signal(sig)
|
||
|
||
assert eu is not None
|
||
assert eu.sentiment_strength == 0.5
|
||
assert eu.impact == 0.5
|
||
assert eu.extraction_conf == 0.5
|
||
assert eu.source_cred == 0.5
|
||
assert eu.novelty == 0.5
|
||
|
||
def test_invalid_horizon_defaults_to_7d(self):
|
||
"""Invalid horizon string falls back to '7d'."""
|
||
sig = _make_company_signal(horizon="invalid_horizon")
|
||
eu = normalize_company_signal(sig)
|
||
assert eu is not None
|
||
assert eu.horizon == "7d"
|
||
|
||
def test_timestamp_string_parsed(self):
|
||
"""ISO timestamp string is parsed to datetime."""
|
||
sig = _make_company_signal(timestamp="2025-01-10T08:00:00+00:00")
|
||
eu = normalize_company_signal(sig)
|
||
assert eu is not None
|
||
assert eu.timestamp == datetime(2025, 1, 10, 8, 0, 0, tzinfo=timezone.utc)
|
||
|
||
def test_unknown_event_type_uses_default_base_rate(self):
|
||
"""Unknown event_type uses default base rate of 0.10."""
|
||
sig = _make_company_signal(event_type="mysterious_event")
|
||
eu = normalize_company_signal(sig)
|
||
assert eu is not None
|
||
assert eu.event_base_rate == 0.10
|
||
|
||
|
||
# ===========================================================================
|
||
# TestNormalizeMacroSignal
|
||
# ===========================================================================
|
||
|
||
|
||
class TestNormalizeMacroSignal:
|
||
"""Test normalize_macro_signal mapping and validation."""
|
||
|
||
def test_full_macro_signal(self):
|
||
"""A complete macro signal maps all fields correctly."""
|
||
sig = _make_macro_signal()
|
||
eu = normalize_macro_signal(sig)
|
||
|
||
assert eu is not None
|
||
assert eu.symbol == "MSFT"
|
||
assert eu.layer == "macro"
|
||
assert eu.source_group == "macro"
|
||
assert eu.direction == 1 # "positive" → +1
|
||
assert eu.impact == 0.6 # macro_impact_score
|
||
assert eu.source_cred == 0.75 # event_confidence
|
||
assert eu.extraction_conf == 0.75 # event_confidence
|
||
assert eu.novelty == 1.0 # default for new events
|
||
|
||
def test_horizon_short_term(self):
|
||
"""short_term → 7d."""
|
||
sig = _make_macro_signal(estimated_duration="short_term")
|
||
eu = normalize_macro_signal(sig)
|
||
assert eu is not None
|
||
assert eu.horizon == "7d"
|
||
|
||
def test_horizon_medium_term(self):
|
||
"""medium_term → 30d."""
|
||
sig = _make_macro_signal(estimated_duration="medium_term")
|
||
eu = normalize_macro_signal(sig)
|
||
assert eu is not None
|
||
assert eu.horizon == "30d"
|
||
|
||
def test_horizon_long_term(self):
|
||
"""long_term → 90d."""
|
||
sig = _make_macro_signal(estimated_duration="long_term")
|
||
eu = normalize_macro_signal(sig)
|
||
assert eu is not None
|
||
assert eu.horizon == "90d"
|
||
|
||
def test_missing_symbol_rejected(self):
|
||
"""Missing symbol in macro signal → None."""
|
||
sig = _make_macro_signal()
|
||
del sig["symbol"]
|
||
assert normalize_macro_signal(sig) is None
|
||
|
||
def test_ticker_alias_accepted(self):
|
||
"""'ticker' key is accepted as alias for 'symbol'."""
|
||
sig = _make_macro_signal()
|
||
del sig["symbol"]
|
||
sig["ticker"] = "AMZN"
|
||
eu = normalize_macro_signal(sig)
|
||
assert eu is not None
|
||
assert eu.symbol == "AMZN"
|
||
|
||
def test_event_id_alias_accepted(self):
|
||
"""'event_id' key is accepted as alias for 'source_id'."""
|
||
sig = _make_macro_signal()
|
||
del sig["source_id"]
|
||
sig["event_id"] = "global-evt-42"
|
||
eu = normalize_macro_signal(sig)
|
||
assert eu is not None
|
||
assert eu.source_id == "global-evt-42"
|
||
|
||
def test_direction_mapping_negative(self):
|
||
"""Negative impact_direction → direction = -1."""
|
||
sig = _make_macro_signal(impact_direction="negative")
|
||
eu = normalize_macro_signal(sig)
|
||
assert eu is not None
|
||
assert eu.direction == -1
|
||
|
||
def test_direction_mapping_neutral(self):
|
||
"""Neutral impact_direction → direction = 0."""
|
||
sig = _make_macro_signal(impact_direction="neutral")
|
||
eu = normalize_macro_signal(sig)
|
||
assert eu is not None
|
||
assert eu.direction == 0
|
||
|
||
|
||
# ===========================================================================
|
||
# TestNormalizeCompetitiveSignal
|
||
# ===========================================================================
|
||
|
||
|
||
class TestNormalizeCompetitiveSignal:
|
||
"""Test normalize_competitive_signal mapping and validation."""
|
||
|
||
def test_full_competitive_signal(self):
|
||
"""A complete competitive signal maps all fields correctly."""
|
||
sig = _make_competitive_signal()
|
||
eu = normalize_competitive_signal(sig)
|
||
|
||
assert eu is not None
|
||
assert eu.symbol == "GOOG"
|
||
assert eu.layer == "competitive"
|
||
assert eu.source_group == "competitive"
|
||
assert eu.horizon == "30d"
|
||
assert eu.direction == -1 # "bearish" → -1
|
||
|
||
def test_impact_is_product_of_strengths(self):
|
||
"""Impact = signal_strength × relationship_strength."""
|
||
sig = _make_competitive_signal(signal_strength=0.7, relationship_strength=0.8)
|
||
eu = normalize_competitive_signal(sig)
|
||
assert eu is not None
|
||
assert abs(eu.impact - 0.56) < 1e-9 # 0.7 × 0.8
|
||
|
||
def test_source_cred_from_pattern_confidence(self):
|
||
"""source_cred mapped from pattern_confidence."""
|
||
sig = _make_competitive_signal(pattern_confidence=0.65)
|
||
eu = normalize_competitive_signal(sig)
|
||
assert eu is not None
|
||
assert eu.source_cred == 0.65
|
||
assert eu.extraction_conf == 0.65
|
||
|
||
def test_direction_bullish(self):
|
||
"""signal_direction='bullish' → direction = +1."""
|
||
sig = _make_competitive_signal(signal_direction="bullish")
|
||
eu = normalize_competitive_signal(sig)
|
||
assert eu is not None
|
||
assert eu.direction == 1
|
||
|
||
def test_direction_neutral(self):
|
||
"""signal_direction='neutral' → direction = 0."""
|
||
sig = _make_competitive_signal(signal_direction="neutral")
|
||
eu = normalize_competitive_signal(sig)
|
||
assert eu is not None
|
||
assert eu.direction == 0
|
||
|
||
def test_novelty_defaults_to_1(self):
|
||
"""Novelty defaults to 1.0 for competitive signals."""
|
||
sig = _make_competitive_signal()
|
||
# Ensure no explicit novelty key
|
||
sig.pop("novelty", None)
|
||
eu = normalize_competitive_signal(sig)
|
||
assert eu is not None
|
||
assert eu.novelty == 1.0
|
||
|
||
def test_missing_required_source_id_rejected(self):
|
||
"""Missing source_id in competitive signal → None."""
|
||
sig = _make_competitive_signal(source_id=None)
|
||
# Also ensure alias key is absent
|
||
sig.pop("source_document_id", None)
|
||
assert normalize_competitive_signal(sig) is None
|
||
|
||
def test_target_ticker_alias(self):
|
||
"""'target_ticker' key accepted as alias for 'symbol'."""
|
||
sig = _make_competitive_signal()
|
||
del sig["symbol"]
|
||
sig["target_ticker"] = "META"
|
||
eu = normalize_competitive_signal(sig)
|
||
assert eu is not None
|
||
assert eu.symbol == "META"
|
||
|
||
def test_time_horizon_short_term_maps_to_7d(self):
|
||
"""Competitive time_horizon='short_term' maps to '7d' via macro map."""
|
||
sig = _make_competitive_signal(time_horizon="short_term")
|
||
eu = normalize_competitive_signal(sig)
|
||
assert eu is not None
|
||
assert eu.horizon == "7d"
|
||
|
||
|
||
# ===========================================================================
|
||
# TestReliabilityPipeline
|
||
# ===========================================================================
|
||
|
||
|
||
class TestReliabilityPipeline:
|
||
"""Test compute_v3_reliability with known inputs."""
|
||
|
||
def test_known_inputs_perfect_signal(self):
|
||
"""Perfect inputs (source_cred=1, extraction_conf=1, novelty=1, fresh, no duplicates) → q_i close to 1."""
|
||
unit = EvidenceUnit(
|
||
symbol="AAPL",
|
||
layer="company",
|
||
event_type="earnings",
|
||
source_id="doc-perfect",
|
||
source_group="company",
|
||
timestamp=_NOW,
|
||
horizon="7d",
|
||
direction=1,
|
||
sentiment_strength=1.0,
|
||
impact=1.0,
|
||
extraction_conf=1.0,
|
||
source_cred=1.0,
|
||
novelty=1.0,
|
||
event_base_rate=0.25,
|
||
cluster_id="test-cluster",
|
||
)
|
||
# Source with strong track record
|
||
stats = SourceStats(source_id="doc-perfect", hits=50, misses=0)
|
||
# Fresh signal (0 age)
|
||
rel = compute_v3_reliability(unit, stats, cluster_position=0, reference_time=_NOW)
|
||
|
||
# q_ext: sigmoid(8.0 * (1.0 - 0.55)) = sigmoid(3.6) ≈ 0.9734
|
||
assert rel.q_ext > 0.95
|
||
|
||
# q_source: E[theta] = (3+50)/(3+3+50+0) = 53/56 ≈ 0.946
|
||
# clamp((0.946 - 0.50) / 0.35, 0, 1) = clamp(1.274, 0, 1) = 1.0
|
||
assert rel.q_source == 1.0
|
||
|
||
# q_recency: fresh signal → 2^0 = 1.0
|
||
assert rel.q_recency == 1.0
|
||
|
||
# q_uniqueness: clamp(0.5 + 0.5*1.0, 0.5, 1.0) * 1/sqrt(1) = 1.0
|
||
assert rel.q_uniqueness == 1.0
|
||
|
||
# q_i should be close to 1 (bounded by q_ext ≈ 0.97)
|
||
assert rel.q_i > 0.90
|
||
|
||
def test_zero_history_source_yields_zero_q_source(self):
|
||
"""A source with zero history (hits=0, misses=0) → q_source = 0.0 (Req 2.3)."""
|
||
unit = EvidenceUnit(
|
||
symbol="AAPL",
|
||
layer="company",
|
||
event_type="earnings",
|
||
source_id="new-source",
|
||
source_group="company",
|
||
timestamp=_NOW,
|
||
horizon="7d",
|
||
direction=1,
|
||
sentiment_strength=0.8,
|
||
impact=0.7,
|
||
extraction_conf=0.9,
|
||
source_cred=0.85,
|
||
novelty=0.9,
|
||
event_base_rate=0.25,
|
||
cluster_id="test-cluster",
|
||
)
|
||
stats = SourceStats(source_id="new-source", hits=0, misses=0)
|
||
rel = compute_v3_reliability(unit, stats, cluster_position=0, reference_time=_NOW)
|
||
|
||
# E[theta] = 3/(3+3) = 0.5; clamp((0.5-0.5)/0.35, 0, 1) = 0.0
|
||
assert rel.q_source == 0.0
|
||
# Therefore q_i = 0.0 (multiplied by zero)
|
||
assert rel.q_i == 0.0
|
||
|
||
def test_duplicate_signal_penalized(self):
|
||
"""Signals later in a cluster (high cluster_position) get lower q_uniqueness."""
|
||
unit = EvidenceUnit(
|
||
symbol="AAPL",
|
||
layer="company",
|
||
event_type="earnings",
|
||
source_id="doc-dup",
|
||
source_group="company",
|
||
timestamp=_NOW,
|
||
horizon="7d",
|
||
direction=1,
|
||
sentiment_strength=0.8,
|
||
impact=0.7,
|
||
extraction_conf=0.9,
|
||
source_cred=0.85,
|
||
novelty=0.9,
|
||
event_base_rate=0.25,
|
||
cluster_id="test-cluster",
|
||
)
|
||
stats = SourceStats(source_id="doc-dup", hits=20, misses=5)
|
||
|
||
rel_first = compute_v3_reliability(unit, stats, cluster_position=0, reference_time=_NOW)
|
||
rel_third = compute_v3_reliability(unit, stats, cluster_position=3, reference_time=_NOW)
|
||
|
||
# Third signal has lower q_uniqueness due to 1/sqrt(1+3) = 0.5
|
||
assert rel_third.q_uniqueness < rel_first.q_uniqueness
|
||
assert rel_third.q_i < rel_first.q_i
|
||
|
||
def test_stale_signal_low_recency(self):
|
||
"""A signal that is very old gets low q_recency."""
|
||
old_timestamp = datetime(2024, 1, 1, 0, 0, 0, tzinfo=timezone.utc)
|
||
unit = EvidenceUnit(
|
||
symbol="AAPL",
|
||
layer="company",
|
||
event_type="earnings",
|
||
source_id="doc-old",
|
||
source_group="company",
|
||
timestamp=old_timestamp,
|
||
horizon="7d",
|
||
direction=1,
|
||
sentiment_strength=0.8,
|
||
impact=0.7,
|
||
extraction_conf=0.9,
|
||
source_cred=0.85,
|
||
novelty=0.9,
|
||
event_base_rate=0.25,
|
||
cluster_id="test-cluster",
|
||
)
|
||
stats = SourceStats(source_id="doc-old", hits=20, misses=5)
|
||
rel = compute_v3_reliability(unit, stats, cluster_position=0, reference_time=_NOW)
|
||
|
||
# Over a year old with 7d horizon (tau_base=72h) → q_recency very low
|
||
# Display floor is 0.01
|
||
assert rel.q_recency == 0.01
|
||
|
||
|
||
# ===========================================================================
|
||
# TestLLRConversion
|
||
# ===========================================================================
|
||
|
||
|
||
class TestLLRConversion:
|
||
"""Test compute_llr boundary cases and sign behavior."""
|
||
|
||
def _make_unit(self, direction: int, impact: float = 0.7, sentiment_strength: float = 0.8) -> EvidenceUnit:
|
||
"""Helper to create an EvidenceUnit with specified direction."""
|
||
return EvidenceUnit(
|
||
symbol="TEST",
|
||
layer="company",
|
||
event_type="earnings",
|
||
source_id="llr-test",
|
||
source_group="company",
|
||
timestamp=_NOW,
|
||
horizon="7d",
|
||
direction=direction,
|
||
sentiment_strength=sentiment_strength,
|
||
impact=impact,
|
||
extraction_conf=0.9,
|
||
source_cred=0.85,
|
||
novelty=0.9,
|
||
event_base_rate=0.25,
|
||
cluster_id="test-cluster",
|
||
)
|
||
|
||
def test_neutral_signal_zero_llr(self):
|
||
"""Neutral signal (direction=0) → LLR = 0.0 exactly (Req 3.3)."""
|
||
unit = self._make_unit(direction=0)
|
||
llr = compute_llr(unit, q_i=0.9)
|
||
assert llr == 0.0
|
||
|
||
def test_bullish_positive_llr(self):
|
||
"""Bullish signal (direction=+1) → positive LLR (Req 3.6)."""
|
||
unit = self._make_unit(direction=1)
|
||
llr = compute_llr(unit, q_i=0.9)
|
||
assert llr > 0.0
|
||
|
||
def test_bearish_negative_llr(self):
|
||
"""Bearish signal (direction=-1) → negative LLR (Req 3.6)."""
|
||
unit = self._make_unit(direction=-1)
|
||
llr = compute_llr(unit, q_i=0.9)
|
||
assert llr < 0.0
|
||
|
||
def test_p_correct_max_clamp(self):
|
||
"""Maximum p_correct = 0.85 → LLR ≈ ln(0.85/0.15) ≈ 1.735 (Req 3.5)."""
|
||
# With direction=+1, q_i=1.0, impact=1.0, sentiment_strength=1.0:
|
||
# p_correct = clamp(0.50 + 0.35*1*1*1, 0.501, 0.85) = 0.85
|
||
unit = self._make_unit(direction=1, impact=1.0, sentiment_strength=1.0)
|
||
llr = compute_llr(unit, q_i=1.0)
|
||
expected = math.log(0.85 / 0.15) # ≈ 1.7346
|
||
assert abs(llr - expected) < 0.001
|
||
|
||
def test_p_correct_min_clamp(self):
|
||
"""Minimum p_correct = 0.501 → |LLR| ≈ ln(0.501/0.499) ≈ 0.004 (Req 3.4)."""
|
||
# With direction=-1, q_i very small → p_correct clamps to 0.501
|
||
# q_i=0 → 0.50 + 0.35*0*anything = 0.50 → clamped to 0.501
|
||
unit = self._make_unit(direction=-1, impact=0.0, sentiment_strength=0.0)
|
||
llr = compute_llr(unit, q_i=0.0)
|
||
expected = -math.log(0.501 / 0.499) # ≈ -0.004
|
||
assert abs(llr - expected) < 0.001
|
||
|
||
def test_llr_sign_always_matches_direction(self):
|
||
"""For directional signals, LLR sign must match direction (Req 3.6)."""
|
||
for direction in [1, -1]:
|
||
for q_i in [0.0, 0.1, 0.5, 0.9, 1.0]:
|
||
unit = self._make_unit(direction=direction)
|
||
llr = compute_llr(unit, q_i=q_i)
|
||
if direction == 1:
|
||
assert llr > 0.0, f"direction=+1, q_i={q_i} should give positive LLR"
|
||
else:
|
||
assert llr < 0.0, f"direction=-1, q_i={q_i} should give negative LLR"
|
||
|
||
def test_llr_magnitude_bounded(self):
|
||
"""LLR magnitude is bounded by [≈0.004, ≈1.735] for directional signals."""
|
||
min_mag = math.log(0.501 / 0.499) # ≈ 0.004
|
||
max_mag = math.log(0.85 / 0.15) # ≈ 1.735
|
||
|
||
# Test at both extremes
|
||
unit_max = self._make_unit(direction=1, impact=1.0, sentiment_strength=1.0)
|
||
llr_max = compute_llr(unit_max, q_i=1.0)
|
||
assert abs(llr_max) <= max_mag + 0.001
|
||
|
||
unit_min = self._make_unit(direction=1, impact=0.0, sentiment_strength=0.0)
|
||
llr_min = compute_llr(unit_min, q_i=0.0)
|
||
assert abs(llr_min) >= min_mag - 0.001
|