stonks-oracle/tests/test_contradiction.py

"""Tests for contradiction detection and disagreement representation.

Requirements: 6.4, 6.5
"""
from datetime import datetime, timezone

from services.aggregation.contradiction import (
    CatalystEntry,
    ContradictionResult,
    detect_contradictions,
)
from services.aggregation.scoring import WeightedSignal, compute_signal_weight

NOW = datetime(2026, 4, 11, 12, 0, 0, tzinfo=timezone.utc)


def _sw(doc_id: str, sentiment: float, impact: float = 0.5) -> WeightedSignal:
    """Helper to build a WeightedSignal with default scoring."""
    w = compute_signal_weight(NOW, NOW, "7d", 0.8, extraction_confidence=0.8)
    return WeightedSignal(doc_id, w, sentiment_value=sentiment, impact_score=impact)


# ---------------------------------------------------------------------------
# Overall score (backward compat with compute_contradiction_score)
# ---------------------------------------------------------------------------


def test_no_signals_returns_zero():
    result = detect_contradictions([])
    assert result.score == 0.0
    assert result.details == []


def test_all_positive_no_contradiction():
    signals = [_sw("d1", 1.0), _sw("d2", 1.0)]
    result = detect_contradictions(signals)
    assert result.score == 0.0
    assert len(result.details) == 0


def test_equal_opposing_gives_half():
    signals = [_sw("d1", 1.0, 0.5), _sw("d2", -1.0, 0.5)]
    result = detect_contradictions(signals)
    assert abs(result.score - 0.5) < 1e-4


def test_neutral_signals_ignored():
    signals = [_sw("d1", 0.0), _sw("d2", 0.0)]
    result = detect_contradictions(signals)
    assert result.score == 0.0
    assert result.details == []


# ---------------------------------------------------------------------------
# Sentiment disagreement detail
# ---------------------------------------------------------------------------


def test_sentiment_disagreement_detail_present():
    signals = [_sw("d1", 1.0, 0.6), _sw("d2", -1.0, 0.4)]
    result = detect_contradictions(signals)
    sentiments = [d for d in result.details if d.dimension == "sentiment"]
    assert len(sentiments) == 1
    detail = sentiments[0]
    assert detail.positive_doc_ids == ["d1"]
    assert detail.negative_doc_ids == ["d2"]
    assert detail.positive_weight > 0
    assert detail.negative_weight > 0
    assert "positive" in detail.description.lower() or "sentiment" in detail.description.lower()


def test_no_sentiment_detail_when_all_agree():
    signals = [_sw("d1", 1.0), _sw("d2", 1.0)]
    result = detect_contradictions(signals)
    sentiments = [d for d in result.details if d.dimension == "sentiment"]
    assert len(sentiments) == 0


# ---------------------------------------------------------------------------
# Catalyst disagreement detail
# ---------------------------------------------------------------------------


def test_catalyst_disagreement_detected():
    signals = [_sw("d1", 1.0, 0.7), _sw("d2", -1.0, 0.5)]
    entries = [
        CatalystEntry("d1", "earnings"),
        CatalystEntry("d2", "earnings"),
    ]
    result = detect_contradictions(signals, entries)
    catalyst_details = [d for d in result.details if d.dimension.startswith("catalyst:")]
    assert len(catalyst_details) == 1
    assert catalyst_details[0].dimension == "catalyst:earnings"
    assert catalyst_details[0].positive_doc_ids == ["d1"]
    assert catalyst_details[0].negative_doc_ids == ["d2"]


def test_no_catalyst_disagreement_when_same_sentiment():
    signals = [_sw("d1", 1.0), _sw("d2", 1.0)]
    entries = [
        CatalystEntry("d1", "earnings"),
        CatalystEntry("d2", "earnings"),
    ]
    result = detect_contradictions(signals, entries)
    catalyst_details = [d for d in result.details if d.dimension.startswith("catalyst:")]
    assert len(catalyst_details) == 0


def test_catalyst_disagreement_across_types():
    """Different catalyst types with internal disagreement each get a detail."""
    signals = [
        _sw("d1", 1.0, 0.5),
        _sw("d2", -1.0, 0.5),
        _sw("d3", 1.0, 0.5),
        _sw("d4", -1.0, 0.5),
    ]
    entries = [
        CatalystEntry("d1", "earnings"),
        CatalystEntry("d2", "earnings"),
        CatalystEntry("d3", "product"),
        CatalystEntry("d4", "product"),
    ]
    result = detect_contradictions(signals, entries)
    catalyst_details = [d for d in result.details if d.dimension.startswith("catalyst:")]
    dims = {d.dimension for d in catalyst_details}
    assert "catalyst:earnings" in dims
    assert "catalyst:product" in dims


# ---------------------------------------------------------------------------
# Integration with assemble_trend_summary
# ---------------------------------------------------------------------------


def test_trend_summary_includes_disagreement_details():
    """assemble_trend_summary should populate disagreement_details."""
    from datetime import timedelta

    from services.aggregation.worker import (
        ImpactRow,
        assemble_trend_summary,
        build_weighted_signals,
    )

    impacts = [
        ImpactRow(
            document_id="d1", confidence=0.8, novelty_score=0.5,
            source_credibility=0.8, sentiment="positive", impact_score=0.7,
            catalyst_type="earnings", key_facts=[], risks=[],
            published_at=NOW - timedelta(hours=1),
        ),
        ImpactRow(
            document_id="d2", confidence=0.8, novelty_score=0.5,
            source_credibility=0.8, sentiment="negative", impact_score=0.7,
            catalyst_type="earnings", key_facts=[], risks=[],
            published_at=NOW - timedelta(hours=2),
        ),
    ]
    signals = build_weighted_signals(impacts, NOW, "7d")
    summary = assemble_trend_summary("AAPL", "7d", signals, impacts, reference_time=NOW)

    assert summary.contradiction_score > 0
    assert len(summary.disagreement_details) > 0
    dims = {d.dimension for d in summary.disagreement_details}
    assert "sentiment" in dims