Celes Renata
a72f336ad1
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.
412 lines
17 KiB
Python
412 lines
17 KiB
Python
"""Unit tests and property tests for the sentence-aware segmenter.
|
|
|
|
Tests cover:
|
|
- Basic segmentation with correct offsets (21.1)
|
|
- Document-type-specific strategies (21.2)
|
|
- Filing section and transcript speaker preservation (21.3)
|
|
- Boilerplate scoring (21.4)
|
|
- No truncation for long documents (21.5)
|
|
- Property tests for offset mapping, reconstruction, and checksums (21.6)
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
|
|
import pytest
|
|
from hypothesis import given, settings
|
|
from hypothesis import strategies as st
|
|
|
|
from services.intelligence_pipeline_v3.segmenter import (
|
|
ArticleStrategy,
|
|
FilingStrategy,
|
|
MacroEventStrategy,
|
|
Segmenter,
|
|
TranscriptStrategy,
|
|
score_boilerplate,
|
|
)
|
|
from services.intelligence_pipeline_v3.segmenter.strategies import get_strategy
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Fixtures
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@pytest.fixture
|
|
def segmenter() -> Segmenter:
|
|
return Segmenter()
|
|
|
|
|
|
SAMPLE_ARTICLE = (
|
|
"Apple reported record revenue of $123.9 billion for Q1 2024. "
|
|
"The company beat analyst expectations by a wide margin. "
|
|
"CEO Tim Cook said growth was driven by iPhone and services. "
|
|
"Shares rose 3% in after-hours trading.\n\n"
|
|
"Meanwhile, Microsoft also reported strong results. "
|
|
"Azure cloud revenue grew 28% year-over-year. "
|
|
"The company expects continued momentum in AI workloads."
|
|
)
|
|
|
|
SAMPLE_FILING = (
|
|
"Item 1. Business\n\n"
|
|
"The company is a global technology leader. "
|
|
"We operate in three segments. "
|
|
"Our products serve enterprise customers.\n\n"
|
|
"Item 2. Properties\n\n"
|
|
"We own facilities in 15 countries. "
|
|
"Our headquarters is in San Jose, California. "
|
|
"We lease approximately 5 million square feet.\n\n"
|
|
"Item 7. Management's Discussion and Analysis\n\n"
|
|
"Revenue increased 15% to $50 billion. "
|
|
"Operating expenses grew 8% driven by R&D investment. "
|
|
"Net income was $12 billion, up from $10 billion. "
|
|
"We expect continued growth in our cloud segment."
|
|
)
|
|
|
|
SAMPLE_TRANSCRIPT = (
|
|
"OPERATOR: Welcome to the Q4 2024 earnings call. "
|
|
"I would now like to turn the call over to Tim Cook.\n\n"
|
|
"Tim Cook - CEO: Thank you. "
|
|
"We are pleased to report another record quarter. "
|
|
"Revenue reached $123.9 billion. "
|
|
"Services revenue hit an all-time high.\n\n"
|
|
"Luca Maestri - CFO: Looking at our financials, "
|
|
"gross margin expanded to 46.6%. "
|
|
"Operating cash flow was $40 billion.\n\n"
|
|
"OPERATOR: We will now take questions from analysts."
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 21.1 — Preserve source offsets and checksums
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestSourceOffsetsAndChecksums:
|
|
"""Test that chunks preserve exact source offsets and have valid checksums."""
|
|
|
|
def test_chunk_text_matches_source_offsets(self, segmenter: Segmenter) -> None:
|
|
"""Each chunk.text must exactly equal source[start_char:end_char]."""
|
|
chunks = segmenter.segment(SAMPLE_ARTICLE, "article", "doc-001")
|
|
for chunk in chunks:
|
|
assert chunk.text == SAMPLE_ARTICLE[chunk.start_char:chunk.end_char]
|
|
|
|
def test_chunk_checksum_is_sha256(self, segmenter: Segmenter) -> None:
|
|
"""Checksum must be SHA-256 of chunk text."""
|
|
chunks = segmenter.segment(SAMPLE_ARTICLE, "article", "doc-001")
|
|
for chunk in chunks:
|
|
expected = hashlib.sha256(chunk.text.encode("utf-8")).hexdigest()
|
|
assert chunk.checksum == expected
|
|
|
|
def test_chunk_id_is_deterministic(self, segmenter: Segmenter) -> None:
|
|
"""chunk_id is {document_id}:{start_char}."""
|
|
chunks = segmenter.segment(SAMPLE_ARTICLE, "article", "doc-001")
|
|
for chunk in chunks:
|
|
assert chunk.chunk_id == f"doc-001:{chunk.start_char}"
|
|
|
|
def test_empty_text_returns_no_chunks(self, segmenter: Segmenter) -> None:
|
|
"""Empty string produces no chunks."""
|
|
assert segmenter.segment("", "article") == []
|
|
|
|
def test_single_sentence_produces_one_chunk(self, segmenter: Segmenter) -> None:
|
|
"""A short text produces exactly one chunk."""
|
|
text = "Apple stock rose 5% today."
|
|
chunks = segmenter.segment(text, "article", "short-doc")
|
|
assert len(chunks) == 1
|
|
assert chunks[0].text == text
|
|
assert chunks[0].start_char == 0
|
|
assert chunks[0].end_char == len(text)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 21.2 — Document-type-specific chunk strategies
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestDocumentTypeStrategies:
|
|
"""Test that each document type uses its own strategy."""
|
|
|
|
def test_article_uses_article_strategy(self) -> None:
|
|
strategy = get_strategy("article")
|
|
assert strategy is ArticleStrategy
|
|
|
|
def test_news_uses_article_strategy(self) -> None:
|
|
strategy = get_strategy("news")
|
|
assert strategy is ArticleStrategy
|
|
|
|
def test_filing_uses_filing_strategy(self) -> None:
|
|
strategy = get_strategy("filing")
|
|
assert strategy is FilingStrategy
|
|
|
|
def test_transcript_uses_transcript_strategy(self) -> None:
|
|
strategy = get_strategy("transcript")
|
|
assert strategy is TranscriptStrategy
|
|
|
|
def test_macro_event_uses_macro_strategy(self) -> None:
|
|
strategy = get_strategy("macro_event")
|
|
assert strategy is MacroEventStrategy
|
|
|
|
def test_unknown_type_uses_default(self) -> None:
|
|
strategy = get_strategy("unknown_type_xyz")
|
|
assert strategy is ArticleStrategy
|
|
|
|
def test_macro_chunks_are_smaller(self, segmenter: Segmenter) -> None:
|
|
"""Macro event strategy produces smaller chunks than filing strategy."""
|
|
# Generate a long text
|
|
long_text = "This is a sentence about macro events. " * 200
|
|
macro_chunks = segmenter.segment(long_text, "macro_event", "macro-1")
|
|
filing_chunks = segmenter.segment(long_text, "filing", "filing-1")
|
|
|
|
if len(macro_chunks) > 1 and len(filing_chunks) > 1:
|
|
avg_macro = sum(len(c.text) for c in macro_chunks) / len(macro_chunks)
|
|
avg_filing = sum(len(c.text) for c in filing_chunks) / len(filing_chunks)
|
|
assert avg_macro < avg_filing
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 21.3 — Preserve filing sections and transcript speakers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestFilingSectionsAndSpeakers:
|
|
"""Test that filing sections and transcript speakers are preserved."""
|
|
|
|
def test_filing_section_path_assigned(self, segmenter: Segmenter) -> None:
|
|
"""Filing chunks should have section_path based on Item headers."""
|
|
chunks = segmenter.segment(SAMPLE_FILING, "filing", "filing-001")
|
|
# At least one chunk should have a section path
|
|
sections_found = [c for c in chunks if c.section_path]
|
|
assert len(sections_found) > 0
|
|
|
|
def test_filing_section_contains_item_headers(self, segmenter: Segmenter) -> None:
|
|
"""Filing section paths should reference Item headers."""
|
|
chunks = segmenter.segment(SAMPLE_FILING, "filing", "filing-001")
|
|
all_sections = set()
|
|
for c in chunks:
|
|
for s in c.section_path:
|
|
all_sections.add(s)
|
|
# Should find at least some of the Item headers
|
|
assert any("Item 1" in s for s in all_sections) or any("Item 2" in s for s in all_sections)
|
|
|
|
def test_transcript_speaker_assigned(self, segmenter: Segmenter) -> None:
|
|
"""Transcript chunks should have speaker labels."""
|
|
chunks = segmenter.segment(SAMPLE_TRANSCRIPT, "transcript", "tx-001")
|
|
speakers_found = [c for c in chunks if c.speaker]
|
|
assert len(speakers_found) > 0
|
|
|
|
def test_transcript_speaker_names_correct(self, segmenter: Segmenter) -> None:
|
|
"""Speaker names should match those in the transcript."""
|
|
chunks = segmenter.segment(SAMPLE_TRANSCRIPT, "transcript", "tx-001")
|
|
all_speakers = {c.speaker for c in chunks if c.speaker}
|
|
# Should find at least one of the speakers
|
|
assert any("Tim Cook" in s or "OPERATOR" in s or "Luca Maestri" in s for s in all_speakers)
|
|
|
|
def test_article_has_no_speaker(self, segmenter: Segmenter) -> None:
|
|
"""Article chunks should not have speaker metadata."""
|
|
chunks = segmenter.segment(SAMPLE_ARTICLE, "article", "art-001")
|
|
for chunk in chunks:
|
|
assert chunk.speaker is None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 21.4 — Mark boilerplate and duplicate chunks
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestBoilerplateDetection:
|
|
"""Test boilerplate scoring."""
|
|
|
|
def test_forward_looking_boilerplate(self) -> None:
|
|
"""Forward-looking statements disclaimer scores high."""
|
|
text = (
|
|
"This press release contains forward-looking statements. "
|
|
"Actual results may differ materially from expectations. "
|
|
"All rights reserved. © 2024 Company Inc."
|
|
)
|
|
score = score_boilerplate(text)
|
|
assert score >= 0.5
|
|
|
|
def test_factual_content_scores_low(self) -> None:
|
|
"""Factual financial content scores low."""
|
|
text = (
|
|
"Revenue increased 23% year-over-year to $45.2 billion. "
|
|
"Earnings per share were $2.18, beating consensus of $2.05. "
|
|
"The company raised full-year guidance to $180 billion."
|
|
)
|
|
score = score_boilerplate(text)
|
|
assert score < 0.3
|
|
|
|
def test_boilerplate_score_capped_at_one(self) -> None:
|
|
"""Score never exceeds 1.0."""
|
|
text = (
|
|
"Forward-looking statements disclaimer. Safe harbor. "
|
|
"Copyright 2024. All rights reserved. Disclaimer applies. "
|
|
"This press release contains certain information. "
|
|
"Actual results may differ materially. Not an offer or solicitation."
|
|
)
|
|
score = score_boilerplate(text)
|
|
assert score <= 1.0
|
|
|
|
def test_empty_text_scores_zero(self) -> None:
|
|
"""Empty text scores 0.0."""
|
|
assert score_boilerplate("") == 0.0
|
|
assert score_boilerplate(" \n\t ") == 0.0
|
|
|
|
def test_segmenter_assigns_boilerplate_scores(self, segmenter: Segmenter) -> None:
|
|
"""Chunks from segmenter have boilerplate_score populated."""
|
|
text = (
|
|
"Revenue grew 20% this quarter. Strong performance across all segments.\n\n"
|
|
"This press release contains forward-looking statements. "
|
|
"Actual results may differ materially from those anticipated."
|
|
)
|
|
chunks = segmenter.segment(text, "article", "bp-001")
|
|
# All chunks should have a score between 0 and 1
|
|
for chunk in chunks:
|
|
assert 0.0 <= chunk.boilerplate_score <= 1.0
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 21.5 — Remove the 8,000-character truncation from v3
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestNoTruncation:
|
|
"""Test that long documents are NOT truncated."""
|
|
|
|
def test_long_document_produces_many_chunks(self, segmenter: Segmenter) -> None:
|
|
"""A 50,000-char document should produce multiple chunks, not be truncated."""
|
|
# Create a document well beyond 8,000 chars
|
|
sentences = [f"Sentence number {i} with some financial data about revenue growth. " for i in range(1000)]
|
|
long_text = " ".join(sentences)
|
|
assert len(long_text) > 50000
|
|
|
|
chunks = segmenter.segment(long_text, "article", "long-doc")
|
|
|
|
# Should have many chunks covering the full document
|
|
assert len(chunks) > 5
|
|
|
|
# Last chunk should reach near the end of the document
|
|
assert chunks[-1].end_char == len(long_text)
|
|
|
|
def test_full_coverage_of_long_document(self, segmenter: Segmenter) -> None:
|
|
"""Every character in a long document should be covered by at least one chunk."""
|
|
sentences = [f"Market analysis point {i} shows interesting trends. " for i in range(500)]
|
|
long_text = " ".join(sentences)
|
|
|
|
chunks = segmenter.segment(long_text, "article", "coverage-doc")
|
|
|
|
# First chunk starts at 0 or very near it
|
|
assert chunks[0].start_char == 0
|
|
# Last chunk ends at document end
|
|
assert chunks[-1].end_char == len(long_text)
|
|
|
|
def test_beyond_8000_chars_content_preserved(self, segmenter: Segmenter) -> None:
|
|
"""Content after 8000 chars is preserved in chunks (not truncated)."""
|
|
# Build text where important content is after 8000 chars
|
|
padding = "Filler content for padding. " * 400 # ~11,200 chars
|
|
important = "CRITICAL EARNINGS BEAT $5.00 EPS versus $4.50 expected."
|
|
text = padding + important
|
|
|
|
chunks = segmenter.segment(text, "article", "no-trunc")
|
|
|
|
# The important content should appear in at least one chunk
|
|
all_text = "".join(c.text[c.overlap_left:] for c in chunks)
|
|
assert "CRITICAL EARNINGS BEAT" in all_text
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 21.6 — Property tests proving chunk/evidence span mapping
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestPropertyBasedSegmenter:
|
|
"""Property-based tests for segmenter invariants.
|
|
|
|
**Validates: Requirements 4.1, 4.2, 4.6**
|
|
"""
|
|
|
|
@given(text=st.text(min_size=1, max_size=20000, alphabet=st.characters(
|
|
categories=("L", "N", "P", "Z", "S"),
|
|
include_characters=".!? \n",
|
|
)))
|
|
@settings(max_examples=100)
|
|
def test_every_chunk_maps_to_source_text(self, text: str) -> None:
|
|
"""Property: For any text, chunk.text == source[chunk.start_char:chunk.end_char].
|
|
|
|
**Validates: Requirements 4.1, 4.6**
|
|
"""
|
|
segmenter = Segmenter()
|
|
chunks = segmenter.segment(text, "article", "prop-test")
|
|
|
|
for chunk in chunks:
|
|
assert chunk.text == text[chunk.start_char:chunk.end_char], (
|
|
f"Chunk at [{chunk.start_char}:{chunk.end_char}] does not match source"
|
|
)
|
|
|
|
@given(text=st.text(min_size=1, max_size=20000, alphabet=st.characters(
|
|
categories=("L", "N", "P", "Z", "S"),
|
|
include_characters=".!? \n",
|
|
)))
|
|
@settings(max_examples=100)
|
|
def test_chunks_cover_full_document(self, text: str) -> None:
|
|
"""Property: The non-overlapping core of chunks covers the entire source.
|
|
|
|
**Validates: Requirements 4.1, 4.2**
|
|
"""
|
|
segmenter = Segmenter()
|
|
chunks = segmenter.segment(text, "article", "cover-test")
|
|
|
|
if not chunks:
|
|
# Empty/whitespace-only text may produce no chunks
|
|
assert not text.strip()
|
|
return
|
|
|
|
# First chunk starts at 0
|
|
assert chunks[0].start_char == 0
|
|
|
|
# Last chunk ends at document length
|
|
assert chunks[-1].end_char == len(text)
|
|
|
|
# Chunks must be ordered and cover the full range
|
|
# The core (non-overlap) portions should cover without gaps
|
|
# Due to overlap, adjacent chunks' starts may be <= previous chunk's end
|
|
for i in range(1, len(chunks)):
|
|
# Each chunk's start (adjusted for overlap) should not leave gaps
|
|
core_start = chunks[i].start_char + chunks[i].overlap_left
|
|
prev_end = chunks[i - 1].end_char
|
|
assert core_start <= prev_end, (
|
|
f"Gap between chunk {i-1} end ({prev_end}) and chunk {i} core start ({core_start})"
|
|
)
|
|
|
|
@given(text=st.text(min_size=1, max_size=10000, alphabet=st.characters(
|
|
categories=("L", "N", "P", "Z", "S"),
|
|
include_characters=".!? \n",
|
|
)))
|
|
@settings(max_examples=100)
|
|
def test_checksum_matches_sha256_of_text(self, text: str) -> None:
|
|
"""Property: Checksum is always SHA-256 of chunk.text.
|
|
|
|
**Validates: Requirements 4.1**
|
|
"""
|
|
segmenter = Segmenter()
|
|
chunks = segmenter.segment(text, "article", "checksum-test")
|
|
|
|
for chunk in chunks:
|
|
expected = hashlib.sha256(chunk.text.encode("utf-8")).hexdigest()
|
|
assert chunk.checksum == expected, (
|
|
f"Checksum mismatch for chunk {chunk.chunk_id}"
|
|
)
|
|
|
|
@given(
|
|
text=st.text(min_size=10, max_size=15000, alphabet=st.characters(
|
|
categories=("L", "N", "P", "Z", "S"),
|
|
include_characters=".!? \n",
|
|
)),
|
|
doc_type=st.sampled_from(["article", "filing", "transcript", "macro_event"]),
|
|
)
|
|
@settings(max_examples=100)
|
|
def test_all_document_types_preserve_offsets(self, text: str, doc_type: str) -> None:
|
|
"""Property: Offset invariant holds for all document types.
|
|
|
|
**Validates: Requirements 4.2, 4.3**
|
|
"""
|
|
segmenter = Segmenter()
|
|
chunks = segmenter.segment(text, doc_type, "multi-type-test")
|
|
|
|
for chunk in chunks:
|
|
assert chunk.text == text[chunk.start_char:chunk.end_char]
|
|
assert chunk.document_type == doc_type
|