Files
Celes Renata a72f336ad1 feat: Intelligence Pipeline v3 — full implementation
Multi-stage evidence-grounded inference architecture replacing the
monolithic 9B model extraction pipeline. CPU-first specialist services
handle routine extraction while the 9B vLLM model is preserved for
semantic adjudication of ambiguous cases.

Key components:
- Capability-aware inference gateway (OpenAI-compatible + Ollama)
- Endpoint registry with DB migrations and REST API
- Sentence-aware document segmenter (property tests)
- Deterministic financial parsing with offset integrity
- Symbol resolution with ambiguity detection
- Specialist service (GLiNER2, dynamic batching, K8s deployment)
- Company-specific sentiment (FinBERT, calibration)
- Retrieval-based novelty and duplicate detection
- Confidence calibration pipeline
- Deterministic routing engine (property tests)
- 9B adjudication layer with VRAM gating
- Stock-specific impact model (features, labels, baseline, trained)
- Pipeline orchestrator (state machine, queues, leases, feature flags)
- Bounded parallelism (async workers, semaphore, load shedding)
- Observability (tracing, metrics, alerts)
- Compatibility adapter (v3→v2 golden mapping tests)
- Shadow/canary promotion framework
- Active learning and fine-tuning pipeline

Test results: 1,161 tests pass, ruff lint clean.
All 282 spec tasks completed.
2026-07-13 02:14:59 +00:00

510 lines
21 KiB
Python

"""Golden mapping tests for the v3→v2 compatibility adapter.
Tests cover:
- Every legacy sentiment enum value is reachable
- impact_score stays in [-1, 1]
- impact_horizon is one of the valid strings
- novelty_score stays in [0, 1]
- confidence stays in [0, 1]
- Adapter disabled by default (mode=disabled raises)
- Adapter enabled in replay mode
- model_provider = 'hybrid' is always set
- Lineage includes adapter version
"""
from __future__ import annotations
import pytest
from services.intelligence_pipeline_v3.compatibility.adapter import (
ADAPTER_VERSION,
AdapterDisabledError,
CompatibilityAdapter,
)
from services.intelligence_pipeline_v3.compatibility.config import (
DEFAULT_ADAPTER_MODE,
AdapterMode,
is_adapter_enabled,
)
from services.intelligence_pipeline_v3.compatibility.models import (
V3CompanySignal,
V3DirectionProbabilities,
V3HorizonProbabilities,
V3IntelligenceRecord,
V3SentimentDistribution,
V3StageRun,
)
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
_SENTINEL = object()
def _make_signal(
*,
sentiment: V3SentimentDistribution | None = None,
horizon: V3HorizonProbabilities | None = None,
direction: V3DirectionProbabilities | None = None,
expected_magnitude: float | None = None,
event_classes: list[str] | None | object = _SENTINEL,
) -> V3CompanySignal:
"""Factory for a minimal v3 company signal with overrides."""
if event_classes is _SENTINEL:
event_classes = ["earnings_beat"]
return V3CompanySignal(
company_id="aaaaaaaa-1111-2222-3333-444444444444",
ticker="AAPL",
relevance_probability=0.9,
event_classes=event_classes or [],
sentiment=sentiment or V3SentimentDistribution(positive=0.7, negative=0.1, neutral=0.2),
direction_probabilities=direction or V3DirectionProbabilities(positive=0.6, negative=0.2, neutral=0.2),
horizon_probabilities=horizon or V3HorizonProbabilities(one_day=0.6, seven_day=0.3, thirty_day=0.1),
expected_magnitude=expected_magnitude,
evidence_spans=["span-1", "span-2"],
)
def _make_v3_record(signals: list[V3CompanySignal] | None = None) -> V3IntelligenceRecord:
"""Factory for a minimal v3 intelligence record."""
return V3IntelligenceRecord(
document_id="doc-001",
document_type="article",
summary="Test summary",
macro_themes=["earnings", "technology"],
novelty_score=0.7,
confidence=0.85,
company_signals=signals or [_make_signal()],
stage_runs=[
V3StageRun(stage="segmenter", schema_version="1.0.0", duration_ms=50),
V3StageRun(stage="specialist", model_version="gliner2-large-v1", schema_version="1.0.0", duration_ms=200),
V3StageRun(stage="sentiment", model_version="finbert-v1", schema_version="1.0.0", duration_ms=100),
],
pipeline_version="3.0.0",
)
# ---------------------------------------------------------------------------
# Task 22.4: Adapter disabled outside replay/shadow mode
# ---------------------------------------------------------------------------
class TestAdapterDisabled:
"""Verify adapter is gated by mode — disabled by default."""
def test_default_mode_is_disabled(self) -> None:
assert DEFAULT_ADAPTER_MODE == AdapterMode.DISABLED
def test_is_adapter_enabled_false_for_disabled(self) -> None:
assert is_adapter_enabled(AdapterMode.DISABLED) is False
def test_is_adapter_enabled_true_for_replay(self) -> None:
assert is_adapter_enabled(AdapterMode.REPLAY_ONLY) is True
def test_is_adapter_enabled_true_for_shadow(self) -> None:
assert is_adapter_enabled(AdapterMode.SHADOW_ONLY) is True
def test_is_adapter_enabled_true_for_canary(self) -> None:
assert is_adapter_enabled(AdapterMode.CANARY) is True
def test_is_adapter_enabled_true_for_production(self) -> None:
assert is_adapter_enabled(AdapterMode.PRODUCTION) is True
def test_disabled_adapter_raises_on_map(self) -> None:
adapter = CompatibilityAdapter(mode=AdapterMode.DISABLED)
with pytest.raises(AdapterDisabledError):
adapter.map_to_v2(_make_v3_record())
def test_replay_adapter_succeeds(self) -> None:
adapter = CompatibilityAdapter(mode=AdapterMode.REPLAY_ONLY)
v2_record, lineage = adapter.map_to_v2(_make_v3_record())
assert v2_record is not None
assert lineage is not None
def test_shadow_adapter_succeeds(self) -> None:
adapter = CompatibilityAdapter(mode=AdapterMode.SHADOW_ONLY)
v2_record, _lineage = adapter.map_to_v2(_make_v3_record())
assert v2_record is not None
# ---------------------------------------------------------------------------
# Task 22.1 / 22.2: Mapping and lineage
# ---------------------------------------------------------------------------
class TestModelProviderHybrid:
"""Verify model_provider is always 'hybrid'."""
def test_model_provider_is_hybrid(self) -> None:
adapter = CompatibilityAdapter(mode=AdapterMode.REPLAY_ONLY)
v2_record, _lineage = adapter.map_to_v2(_make_v3_record())
assert v2_record.model_provider == "hybrid"
def test_model_name_is_pipeline_v3(self) -> None:
adapter = CompatibilityAdapter(mode=AdapterMode.REPLAY_ONLY)
v2_record, _lineage = adapter.map_to_v2(_make_v3_record())
assert v2_record.model_name == "intelligence-pipeline-v3"
class TestLineage:
"""Verify lineage records adapter version and stage details."""
def test_lineage_includes_adapter_version(self) -> None:
adapter = CompatibilityAdapter(mode=AdapterMode.REPLAY_ONLY)
_v2_record, lineage = adapter.map_to_v2(_make_v3_record())
assert lineage.adapter_version == ADAPTER_VERSION
def test_lineage_includes_pipeline_version(self) -> None:
adapter = CompatibilityAdapter(mode=AdapterMode.REPLAY_ONLY)
_v2_record, lineage = adapter.map_to_v2(_make_v3_record())
assert lineage.pipeline_version == "3.0.0"
def test_lineage_includes_stage_runs(self) -> None:
adapter = CompatibilityAdapter(mode=AdapterMode.REPLAY_ONLY)
_v2_record, lineage = adapter.map_to_v2(_make_v3_record())
assert len(lineage.stage_runs) == 3
stages = [sr.stage for sr in lineage.stage_runs]
assert "segmenter" in stages
assert "specialist" in stages
assert "sentiment" in stages
def test_lineage_links_v3_to_v2(self) -> None:
adapter = CompatibilityAdapter(mode=AdapterMode.REPLAY_ONLY)
v2_record, lineage = adapter.map_to_v2(_make_v3_record())
assert lineage.v3_document_id == "doc-001"
assert lineage.v2_intelligence_id == v2_record.id
# ---------------------------------------------------------------------------
# Task 22.3: Golden mapping tests — sentiment enum
# ---------------------------------------------------------------------------
class TestSentimentMapping:
"""Every legacy sentiment enum value (positive/negative/neutral/mixed) is reachable."""
def test_positive_sentiment(self) -> None:
signal = _make_signal(
sentiment=V3SentimentDistribution(positive=0.8, negative=0.1, neutral=0.1)
)
adapter = CompatibilityAdapter(mode=AdapterMode.REPLAY_ONLY)
v2, _ = adapter.map_to_v2(_make_v3_record(signals=[signal]))
assert v2.impact_records[0].sentiment == "positive"
def test_negative_sentiment(self) -> None:
signal = _make_signal(
sentiment=V3SentimentDistribution(positive=0.1, negative=0.8, neutral=0.1)
)
adapter = CompatibilityAdapter(mode=AdapterMode.REPLAY_ONLY)
v2, _ = adapter.map_to_v2(_make_v3_record(signals=[signal]))
assert v2.impact_records[0].sentiment == "negative"
def test_neutral_sentiment(self) -> None:
signal = _make_signal(
sentiment=V3SentimentDistribution(positive=0.1, negative=0.1, neutral=0.8)
)
adapter = CompatibilityAdapter(mode=AdapterMode.REPLAY_ONLY)
v2, _ = adapter.map_to_v2(_make_v3_record(signals=[signal]))
assert v2.impact_records[0].sentiment == "neutral"
def test_mixed_sentiment(self) -> None:
signal = _make_signal(
sentiment=V3SentimentDistribution(positive=0.4, negative=0.4, neutral=0.2)
)
adapter = CompatibilityAdapter(mode=AdapterMode.REPLAY_ONLY)
v2, _ = adapter.map_to_v2(_make_v3_record(signals=[signal]))
assert v2.impact_records[0].sentiment == "mixed"
def test_mixed_threshold_boundary(self) -> None:
"""Both positive and negative at exactly 0.3 triggers mixed."""
signal = _make_signal(
sentiment=V3SentimentDistribution(positive=0.3, negative=0.3, neutral=0.4)
)
adapter = CompatibilityAdapter(mode=AdapterMode.REPLAY_ONLY)
v2, _ = adapter.map_to_v2(_make_v3_record(signals=[signal]))
assert v2.impact_records[0].sentiment == "mixed"
# ---------------------------------------------------------------------------
# Task 22.3: Golden mapping tests — impact_score range
# ---------------------------------------------------------------------------
class TestImpactScoreRange:
"""impact_score stays in [-1, 1]."""
def test_impact_score_from_magnitude(self) -> None:
signal = _make_signal(expected_magnitude=0.5)
adapter = CompatibilityAdapter(mode=AdapterMode.REPLAY_ONLY)
v2, _ = adapter.map_to_v2(_make_v3_record(signals=[signal]))
assert -1.0 <= v2.impact_records[0].impact_score <= 1.0
assert v2.impact_records[0].impact_score == 0.5
def test_impact_score_clamped_high(self) -> None:
signal = _make_signal(expected_magnitude=2.5)
adapter = CompatibilityAdapter(mode=AdapterMode.REPLAY_ONLY)
v2, _ = adapter.map_to_v2(_make_v3_record(signals=[signal]))
assert v2.impact_records[0].impact_score == 1.0
def test_impact_score_clamped_low(self) -> None:
signal = _make_signal(expected_magnitude=-3.0)
adapter = CompatibilityAdapter(mode=AdapterMode.REPLAY_ONLY)
v2, _ = adapter.map_to_v2(_make_v3_record(signals=[signal]))
assert v2.impact_records[0].impact_score == -1.0
def test_impact_score_negative_magnitude(self) -> None:
signal = _make_signal(expected_magnitude=-0.7)
adapter = CompatibilityAdapter(mode=AdapterMode.REPLAY_ONLY)
v2, _ = adapter.map_to_v2(_make_v3_record(signals=[signal]))
assert v2.impact_records[0].impact_score == -0.7
def test_impact_score_fallback_from_direction(self) -> None:
"""When expected_magnitude is None, derive from direction probabilities."""
signal = _make_signal(
expected_magnitude=None,
direction=V3DirectionProbabilities(positive=0.8, negative=0.1, neutral=0.1),
)
adapter = CompatibilityAdapter(mode=AdapterMode.REPLAY_ONLY)
v2, _ = adapter.map_to_v2(_make_v3_record(signals=[signal]))
score = v2.impact_records[0].impact_score
assert -1.0 <= score <= 1.0
# 0.8 - 0.1 = 0.7
assert abs(score - 0.7) < 1e-9
def test_impact_score_zero(self) -> None:
signal = _make_signal(expected_magnitude=0.0)
adapter = CompatibilityAdapter(mode=AdapterMode.REPLAY_ONLY)
v2, _ = adapter.map_to_v2(_make_v3_record(signals=[signal]))
assert v2.impact_records[0].impact_score == 0.0
# ---------------------------------------------------------------------------
# Task 22.3: Golden mapping tests — impact_horizon valid strings
# ---------------------------------------------------------------------------
VALID_HORIZONS = {"intraday", "1d", "7d", "30d", "90d"}
class TestImpactHorizonMapping:
"""impact_horizon is one of the valid legacy strings."""
def test_intraday_horizon(self) -> None:
signal = _make_signal(
horizon=V3HorizonProbabilities(intraday=0.9, one_day=0.05, seven_day=0.05)
)
adapter = CompatibilityAdapter(mode=AdapterMode.REPLAY_ONLY)
v2, _ = adapter.map_to_v2(_make_v3_record(signals=[signal]))
assert v2.impact_records[0].impact_horizon == "intraday"
assert v2.impact_records[0].impact_horizon in VALID_HORIZONS
def test_one_day_horizon(self) -> None:
signal = _make_signal(
horizon=V3HorizonProbabilities(intraday=0.1, one_day=0.7, seven_day=0.2)
)
adapter = CompatibilityAdapter(mode=AdapterMode.REPLAY_ONLY)
v2, _ = adapter.map_to_v2(_make_v3_record(signals=[signal]))
assert v2.impact_records[0].impact_horizon == "1d"
def test_seven_day_horizon(self) -> None:
signal = _make_signal(
horizon=V3HorizonProbabilities(seven_day=0.8, thirty_day=0.1, ninety_day=0.1)
)
adapter = CompatibilityAdapter(mode=AdapterMode.REPLAY_ONLY)
v2, _ = adapter.map_to_v2(_make_v3_record(signals=[signal]))
assert v2.impact_records[0].impact_horizon == "7d"
def test_thirty_day_horizon(self) -> None:
signal = _make_signal(
horizon=V3HorizonProbabilities(thirty_day=0.9, ninety_day=0.1)
)
adapter = CompatibilityAdapter(mode=AdapterMode.REPLAY_ONLY)
v2, _ = adapter.map_to_v2(_make_v3_record(signals=[signal]))
assert v2.impact_records[0].impact_horizon == "30d"
def test_ninety_day_horizon(self) -> None:
signal = _make_signal(
horizon=V3HorizonProbabilities(ninety_day=0.9)
)
adapter = CompatibilityAdapter(mode=AdapterMode.REPLAY_ONLY)
v2, _ = adapter.map_to_v2(_make_v3_record(signals=[signal]))
assert v2.impact_records[0].impact_horizon == "90d"
def test_horizon_always_valid(self) -> None:
"""Default horizon probs still produce a valid string."""
signal = _make_signal(horizon=V3HorizonProbabilities())
adapter = CompatibilityAdapter(mode=AdapterMode.REPLAY_ONLY)
v2, _ = adapter.map_to_v2(_make_v3_record(signals=[signal]))
assert v2.impact_records[0].impact_horizon in VALID_HORIZONS
# ---------------------------------------------------------------------------
# Task 22.3: Golden mapping tests — novelty_score range
# ---------------------------------------------------------------------------
class TestNoveltyScoreRange:
"""novelty_score stays in [0, 1]."""
def test_novelty_passes_through(self) -> None:
record = _make_v3_record()
record.novelty_score = 0.7
adapter = CompatibilityAdapter(mode=AdapterMode.REPLAY_ONLY)
v2, _ = adapter.map_to_v2(record)
assert v2.novelty_score == 0.7
assert 0.0 <= v2.novelty_score <= 1.0
def test_novelty_zero(self) -> None:
record = _make_v3_record()
record.novelty_score = 0.0
adapter = CompatibilityAdapter(mode=AdapterMode.REPLAY_ONLY)
v2, _ = adapter.map_to_v2(record)
assert v2.novelty_score == 0.0
def test_novelty_one(self) -> None:
record = _make_v3_record()
record.novelty_score = 1.0
adapter = CompatibilityAdapter(mode=AdapterMode.REPLAY_ONLY)
v2, _ = adapter.map_to_v2(record)
assert v2.novelty_score == 1.0
# ---------------------------------------------------------------------------
# Task 22.3: Golden mapping tests — confidence range
# ---------------------------------------------------------------------------
class TestConfidenceRange:
"""confidence stays in [0, 1]."""
def test_confidence_passes_through(self) -> None:
record = _make_v3_record()
record.confidence = 0.85
adapter = CompatibilityAdapter(mode=AdapterMode.REPLAY_ONLY)
v2, _ = adapter.map_to_v2(record)
assert v2.confidence == 0.85
assert 0.0 <= v2.confidence <= 1.0
def test_confidence_zero(self) -> None:
record = _make_v3_record()
record.confidence = 0.0
adapter = CompatibilityAdapter(mode=AdapterMode.REPLAY_ONLY)
v2, _ = adapter.map_to_v2(record)
assert v2.confidence == 0.0
def test_confidence_one(self) -> None:
record = _make_v3_record()
record.confidence = 1.0
adapter = CompatibilityAdapter(mode=AdapterMode.REPLAY_ONLY)
v2, _ = adapter.map_to_v2(record)
assert v2.confidence == 1.0
# ---------------------------------------------------------------------------
# Task 22.3: Golden mapping tests — catalyst type mapping
# ---------------------------------------------------------------------------
class TestCatalystTypeMapping:
"""Event taxonomy maps to legacy catalyst enum values."""
@pytest.mark.parametrize(
"event_class,expected_catalyst",
[
("earnings_beat", "earnings"),
("earnings_miss", "earnings"),
("guidance_raise", "earnings"),
("guidance_cut", "earnings"),
("product_launch", "product"),
("legal_regulatory", "legal"),
("ma_announcement", "m_and_a"),
("supply_chain", "supply_chain"),
("rating_change", "rating_change"),
("macro_event", "macro"),
("management_change", "other"),
("dividend_change", "other"),
("buyback", "other"),
],
)
def test_event_class_to_catalyst(self, event_class: str, expected_catalyst: str) -> None:
signal = _make_signal(event_classes=[event_class])
adapter = CompatibilityAdapter(mode=AdapterMode.REPLAY_ONLY)
v2, _ = adapter.map_to_v2(_make_v3_record(signals=[signal]))
assert v2.impact_records[0].catalyst_type == expected_catalyst
def test_unknown_event_class_falls_back_to_other(self) -> None:
signal = _make_signal(event_classes=["unknown_future_event"])
adapter = CompatibilityAdapter(mode=AdapterMode.REPLAY_ONLY)
v2, _ = adapter.map_to_v2(_make_v3_record(signals=[signal]))
assert v2.impact_records[0].catalyst_type == "other"
def test_empty_event_classes_falls_back_to_other(self) -> None:
signal = _make_signal(event_classes=[])
adapter = CompatibilityAdapter(mode=AdapterMode.REPLAY_ONLY)
v2, _ = adapter.map_to_v2(_make_v3_record(signals=[signal]))
assert v2.impact_records[0].catalyst_type == "other"
def test_first_matching_event_wins(self) -> None:
"""When multiple event classes, first match determines catalyst."""
signal = _make_signal(event_classes=["product_launch", "earnings_beat"])
adapter = CompatibilityAdapter(mode=AdapterMode.REPLAY_ONLY)
v2, _ = adapter.map_to_v2(_make_v3_record(signals=[signal]))
assert v2.impact_records[0].catalyst_type == "product"
# ---------------------------------------------------------------------------
# Task 22.1: Field mapping completeness
# ---------------------------------------------------------------------------
class TestFieldMappingCompleteness:
"""Verify all v2 fields are populated from v3 sources."""
def test_summary_mapped(self) -> None:
adapter = CompatibilityAdapter(mode=AdapterMode.REPLAY_ONLY)
v2, _ = adapter.map_to_v2(_make_v3_record())
assert v2.summary == "Test summary"
def test_macro_themes_mapped(self) -> None:
adapter = CompatibilityAdapter(mode=AdapterMode.REPLAY_ONLY)
v2, _ = adapter.map_to_v2(_make_v3_record())
assert v2.macro_themes == ["earnings", "technology"]
def test_evidence_spans_mapped(self) -> None:
adapter = CompatibilityAdapter(mode=AdapterMode.REPLAY_ONLY)
v2, _ = adapter.map_to_v2(_make_v3_record())
assert v2.impact_records[0].evidence_spans == ["span-1", "span-2"]
def test_relevance_mapped(self) -> None:
adapter = CompatibilityAdapter(mode=AdapterMode.REPLAY_ONLY)
v2, _ = adapter.map_to_v2(_make_v3_record())
assert v2.impact_records[0].relevance == 0.9
def test_ticker_mapped(self) -> None:
adapter = CompatibilityAdapter(mode=AdapterMode.REPLAY_ONLY)
v2, _ = adapter.map_to_v2(_make_v3_record())
assert v2.impact_records[0].ticker == "AAPL"
def test_multiple_companies(self) -> None:
signals = [
_make_signal(),
V3CompanySignal(
company_id="bbbbbbbb-1111-2222-3333-444444444444",
ticker="MSFT",
relevance_probability=0.7,
event_classes=["product_launch"],
sentiment=V3SentimentDistribution(positive=0.6, negative=0.2, neutral=0.2),
direction_probabilities=V3DirectionProbabilities(positive=0.5, negative=0.2, neutral=0.3),
horizon_probabilities=V3HorizonProbabilities(seven_day=0.6, thirty_day=0.4),
expected_magnitude=0.3,
evidence_spans=["span-3"],
),
]
adapter = CompatibilityAdapter(mode=AdapterMode.REPLAY_ONLY)
v2, _ = adapter.map_to_v2(_make_v3_record(signals=signals))
assert len(v2.impact_records) == 2
tickers = {r.ticker for r in v2.impact_records}
assert tickers == {"AAPL", "MSFT"}