stonks-oracle/services/intelligence_pipeline_v3/sentiment/mixed_sentiment.py

"""Mixed sentiment detection from evidence-group disagreement.

When evidence groups for the same company disagree (some positive,
some negative), the resulting label is "mixed" rather than an
unconstrained model output. This follows the design requirement that
mixed sentiment comes from conflicting supported evidence, not from
a fourth softmax label.

NOTE: This module provides the legacy compute_mixed_sentiment function
for backward compatibility. New code should prefer
aggregation.aggregate_evidence_sentiments which accepts TextSentiment
objects directly.
"""

from __future__ import annotations

from services.intelligence_pipeline_v3.sentiment.models import CompanySentimentResult, TextSentiment

# Disagreement threshold: both positive and negative probabilities
# must be >= this value across evidence groups to signal disagreement
DISAGREEMENT_THRESHOLD = 0.3


def compute_mixed_sentiment(
    company_id: str,
    group_results: list[tuple[float, float, float]],
    evidence_ids: list[str],
    model_version: str,
    calibration_version: str = "uncalibrated",
) -> CompanySentimentResult:
    """Compute company sentiment from multiple evidence group results.

    Applies disagreement detection: if evidence groups for the same
    company show both positive and negative signals above the threshold,
    the label is "mixed".

    Parameters
    ----------
    company_id
        Resolved company identifier.
    group_results
        List of (positive_prob, negative_prob, neutral_prob) tuples,
        one per evidence group for this company.
    evidence_ids
        All evidence span IDs contributing to this result.
    model_version
        Sentiment model version string.
    calibration_version
        Calibration artifact version.

    Returns
    -------
    CompanySentimentResult
        Aggregated sentiment result with disagreement-based mixed detection.
    """
    if not group_results:
        # No evidence — neutral by default
        return CompanySentimentResult(
            company_id=company_id,
            label="neutral",
            positive_prob=0.0,
            negative_prob=0.0,
            neutral_prob=1.0,
            evidence_ids=evidence_ids,
            is_mixed=False,
            per_text_scores=[],
            model_version=model_version,
            calibration_version=calibration_version,
        )

    # Check for disagreement across groups
    is_mixed = _detect_disagreement(group_results)

    # Compute weighted average of group probabilities
    n = len(group_results)
    avg_pos = sum(r[0] for r in group_results) / n
    avg_neg = sum(r[1] for r in group_results) / n
    avg_neu = sum(r[2] for r in group_results) / n

    # Normalize to ensure probabilities sum to 1.0
    total = avg_pos + avg_neg + avg_neu
    if total > 0:
        avg_pos /= total
        avg_neg /= total
        avg_neu /= total
    else:
        avg_pos = 0.0
        avg_neg = 0.0
        avg_neu = 1.0

    # Determine label
    if is_mixed:
        label = "mixed"
    else:
        label = _argmax_label(avg_pos, avg_neg, avg_neu)

    # Build per-text scores from tuples for provenance
    per_text_scores: list[TextSentiment] = []
    for i, (pos, neg, neu) in enumerate(group_results):
        eid = evidence_ids[i] if i < len(evidence_ids) else f"unknown_{i}"
        per_text_scores.append(
            TextSentiment(
                evidence_id=eid,
                positive_prob=pos,
                negative_prob=neg,
                neutral_prob=neu,
            )
        )

    return CompanySentimentResult(
        company_id=company_id,
        label=label,
        positive_prob=round(avg_pos, 6),
        negative_prob=round(avg_neg, 6),
        neutral_prob=round(avg_neu, 6),
        evidence_ids=evidence_ids,
        is_mixed=is_mixed,
        per_text_scores=per_text_scores,
        model_version=model_version,
        calibration_version=calibration_version,
    )


def _detect_disagreement(group_results: list[tuple[float, float, float]]) -> bool:
    """Detect if evidence groups disagree on sentiment direction.

    Disagreement is detected when across all groups, the maximum
    positive probability is >= threshold AND the maximum negative
    probability is >= threshold. This means some evidence strongly
    suggests positive while other evidence strongly suggests negative.
    """
    if len(group_results) < 2:
        return False

    max_pos = max(r[0] for r in group_results)
    max_neg = max(r[1] for r in group_results)

    return max_pos >= DISAGREEMENT_THRESHOLD and max_neg >= DISAGREEMENT_THRESHOLD


def _argmax_label(pos: float, neg: float, neu: float) -> str:
    """Return the label with the highest probability."""
    if pos >= neg and pos >= neu:
        return "positive"
    elif neg >= pos and neg >= neu:
        return "negative"
    else:
        return "neutral"