tests/test_tone_drift.py

"""tests/test_tone_drift.py — unit tests for analysis/tone_drift.py and
analysis/transcript_parse.py.

Tests are purely deterministic: they do NOT call the sentence-transformer
model (mocked), do NOT hit sections_db (mocked), and do NOT require any
ingested data. Fixtures mimic the flattened Alpha Vantage format produced
by ingestion/transcript.py (one "Speaker: content" line per segment).
"""
from __future__ import annotations

from unittest.mock import patch

import numpy as np

from analysis.signals import QuarterDelta
from analysis.tone_drift import (
    _is_evasive,
    _phrase_rate,
    _question_topic,
    compute,
    compute_recurring_evasions,
    compute_tone_trend,
    compute_topic_arcs,
    compute_topic_fades,
)
from analysis.transcript_parse import ParsedCall, QAExchange, parse_call

# ---------------------------------------------------------------------------
# Fixtures — flattened AV-format transcript
# ---------------------------------------------------------------------------

CALL_TEXT = """\
Operator: Good afternoon, and welcome to the Examplecorp third quarter 2025 earnings conference call. At this time all participants are in a listen-only mode.
Jane Smith: Thank you, operator, and good afternoon everyone. Revenue for the quarter came in at the high end of our outlook, driven by data center strength across all regions and continued adoption of our newest platform by enterprise customers worldwide.
John Doe: Thanks, Jane. Gross margin was consistent with our prior commentary and operating expenses were well controlled across the organization during the period under review.
Operator: We will now begin the question-and-answer session. The first question comes from the line of Alex Carter with Big Bank.
Alex Carter: Thanks for taking my question. Can you quantify the China headwind to data center revenue this quarter and how should we think about it going forward?
Jane Smith: It is too early to say how that dynamic plays out and we are not going to get into specifics on any single region today.
Operator: The next question comes from the line of Morgan Lee with Other Firm.
Morgan Lee: Great, thank you. Could you walk us through the drivers of gross margin in the quarter and the puts and takes for next quarter?
John Doe: Sure. Gross margin was 54.3% in the quarter, up 120 basis points sequentially, driven by mix and better unit costs, and we expect roughly 54% next quarter.
"""

CALL_TEXT_WITH_TITLES = CALL_TEXT.replace(
    "Jane Smith:", "Jane Smith (CEO):"
).replace(
    "John Doe:", "John Doe (CFO):"
).replace(
    "Alex Carter:", "Alex Carter (Analyst):"
)


def _mgmt_call(period: str, n_words: int, hedge_hits: int) -> ParsedCall:
    """ParsedCall whose management text has an exact hedge-phrase count."""
    filler_words = n_words - 2 * hedge_hits
    text = ("alpha " * filler_words) + ("i think " * hedge_hits)
    return ParsedCall(period=period, prepared_text=text, management_text=text)


def _prepared_call(period: str, topic_mentions: int) -> ParsedCall:
    base = "alpha " * 320
    topic = "Our backlog grew again this period. " * topic_mentions
    text = base + topic
    return ParsedCall(period=period, prepared_text=text, management_text=text)


def _qa_call(period: str, exchanges: list[QAExchange]) -> ParsedCall:
    return ParsedCall(
        period=period, prepared_text="alpha " * 400,
        management_text="alpha " * 400, qa=exchanges,
    )


_CHINA_Q = (
    "Can you quantify the China headwind to data center revenue this quarter "
    "and how should we think about the trajectory going forward into next year?"
)
_MARGIN_Q = (
    "Could you walk us through the drivers of gross margin in the quarter and "
    "the puts and takes that we should model for the next several quarters please?"
)
_EVASIVE_A = "It is too early to say and we are not going to guide on that level of detail."
_QUANT_A = (
    "Gross margin was 54.3% in the quarter, up 120 basis points sequentially, "
    "driven by product mix and better unit costs, and we expect roughly 54% next quarter "
    "as the new platform ramps through the year."
)


def _keyword_embed(texts):
    """Deterministic embedding: china-questions → e1, margin-questions → e2."""
    vecs = []
    for t in texts:
        if "china" in t.lower():
            vecs.append([1.0, 0.0])
        else:
            vecs.append([0.0, 1.0])
    return np.array(vecs)


# ---------------------------------------------------------------------------
# Tests: transcript parser
# ---------------------------------------------------------------------------

def test_parse_call_finds_qa_boundary_and_roles():
    call = parse_call("Q32025", CALL_TEXT)
    assert call.n_segments == 9
    assert "data center strength" in call.prepared_text
    # Operator and analyst speech never lands in management text
    assert "listen-only" not in call.management_text
    assert "taking my question" not in call.management_text
    # Q&A answers are management speech
    assert "too early to say" in call.management_text


def test_parse_call_pairs_exchanges():
    call = parse_call("Q32025", CALL_TEXT)
    assert len(call.qa) == 2
    first, second = call.qa
    assert first.analyst == "Alex Carter"
    assert "China headwind" in first.question
    assert "too early to say" in first.answer
    assert second.analyst == "Morgan Lee"
    assert "54.3%" in second.answer


def test_parse_call_tolerates_title_format():
    call = parse_call("Q32025", CALL_TEXT_WITH_TITLES)
    assert len(call.qa) == 2
    assert call.qa[0].analyst == "Alex Carter"
    assert "data center strength" in call.prepared_text


def test_parse_call_fallback_without_structure():
    text = "This is a raw transcript blob without any speaker structure at all."
    call = parse_call("Q32025", text)
    assert call.qa == []
    assert call.prepared_text == text
    assert call.management_text == text


def test_parse_call_unknown_qa_speaker_with_question_mark_is_analyst():
    text = CALL_TEXT + (
        "Sam Park: What is your capital expenditure plan for the next fiscal year "
        "given the capacity constraints you mentioned earlier in the call today?\n"
        "Jane Smith: We plan to invest ahead of demand as we have said before.\n"
    )
    call = parse_call("Q32025", text)
    assert len(call.qa) == 3
    assert call.qa[2].analyst == "Sam Park"


def test_parse_call_never_raises_on_garbage():
    for garbage in ["", "::::\n::::", "1234\n5678", None and "" or "?? ?? ??"]:
        call = parse_call("Q12026", garbage)
        assert isinstance(call, ParsedCall)


# ---------------------------------------------------------------------------
# Tests: tone trend
# ---------------------------------------------------------------------------

def test_tone_trend_detects_rising_hedge_rate():
    calls = [
        _mgmt_call("Q32025", 1000, 2),   # rate 20 per 10k
        _mgmt_call("Q42025", 1000, 4),   # rate 40
        _mgmt_call("Q12026", 1000, 6),   # rate 60
    ]
    deltas = compute_tone_trend(calls)
    hedge = [d for d in deltas if d.term == "hedging language"]
    assert len(hedge) == 1
    d = hedge[0]
    assert d.kind == "tone_trend"
    assert d.source == "transcript"
    assert "more cautious" in d.computed_metric
    assert "rising 3 quarters" in d.computed_metric
    assert d.significance == "HIGH"  # net change (60-20)/20 = 200% ≥ 50%
    assert d.period_from == "Q32025" and d.period_to == "Q12026"


def test_tone_trend_falling_hedge_rate_reads_more_confident():
    calls = [
        _mgmt_call("Q32025", 1000, 6),
        _mgmt_call("Q42025", 1000, 4),
        _mgmt_call("Q12026", 1000, 2),
    ]
    deltas = compute_tone_trend(calls)
    hedge = [d for d in deltas if d.term == "hedging language"]
    assert len(hedge) == 1
    assert "more confident" in hedge[0].computed_metric


def test_tone_trend_requires_three_calls():
    calls = [_mgmt_call("Q42025", 1000, 2), _mgmt_call("Q12026", 1000, 6)]
    assert compute_tone_trend(calls) == []


def test_tone_trend_skips_flat_series():
    calls = [_mgmt_call(p, 1000, 3) for p in ("Q32025", "Q42025", "Q12026")]
    assert compute_tone_trend(calls) == []


def test_phrase_rate_zero_on_empty():
    assert _phrase_rate("", ["i think"]) == 0


# ---------------------------------------------------------------------------
# Tests: topic arcs
# ---------------------------------------------------------------------------

def test_topic_arc_detects_rising_inventory_mentions():
    calls = [
        ParsedCall(period=p, prepared_text="", management_text="")
        for p in ("Q32025", "Q42025", "Q12026")
    ]
    raw = [
        "We watch inventory closely. " * 1,
        "We watch inventory closely. " * 4,
        "We watch inventory closely. " * 7,
    ]
    deltas = compute_topic_arcs(calls, raw)
    inv = [d for d in deltas if d.term == "inventory"]
    assert len(inv) == 1
    d = inv[0]
    assert d.kind == "topic_arc"
    assert "1→7 mentions" in d.computed_metric
    assert "rising 3 quarters" in d.computed_metric
    assert d.significance == "MEDIUM"


def test_topic_arc_noise_floor():
    calls = [
        ParsedCall(period=p, prepared_text="", management_text="")
        for p in ("Q32025", "Q42025", "Q12026")
    ]
    raw = ["no mention", "inventory once", "inventory twice inventory"]
    # max count is 2 < 3 → below noise floor
    assert compute_topic_arcs(calls, raw) == []


# ---------------------------------------------------------------------------
# Tests: recurring evasions
# ---------------------------------------------------------------------------

def test_recurring_evasion_detected_across_two_calls():
    calls = [
        _qa_call("Q32025", [
            QAExchange("Alex Carter", _CHINA_Q, _EVASIVE_A),
            QAExchange("Morgan Lee", _MARGIN_Q, _QUANT_A),
        ]),
        _qa_call("Q12026", [
            QAExchange("Alex Carter", _CHINA_Q, _EVASIVE_A),
            QAExchange("Morgan Lee", _MARGIN_Q, _QUANT_A),
        ]),
    ]
    with patch("analysis.tone_drift._embed", side_effect=_keyword_embed):
        deltas = compute_recurring_evasions(calls)
    assert len(deltas) == 1
    d = deltas[0]
    assert d.kind == "recurring_evasion"
    assert d.significance == "MEDIUM"  # 2 distinct periods
    assert "asked in Q32025, Q12026" in d.computed_metric
    assert "2/2 answers non-quantitative" in d.computed_metric
    assert "deflection: '" in d.computed_metric  # first matching phrase from _DEFLECTIONS
    assert d.before_text.startswith("Alex Carter:")
    assert "china" in d.term.lower()


def test_recurring_evasion_high_on_three_calls():
    exch = QAExchange("Alex Carter", _CHINA_Q, _EVASIVE_A)
    calls = [_qa_call(p, [exch]) for p in ("Q22025", "Q32025", "Q12026")]
    with patch("analysis.tone_drift._embed", side_effect=_keyword_embed):
        deltas = compute_recurring_evasions(calls)
    assert len(deltas) == 1
    assert deltas[0].significance == "HIGH"


def test_no_evasion_when_answers_are_quantitative():
    calls = [
        _qa_call("Q32025", [QAExchange("Morgan Lee", _MARGIN_Q, _QUANT_A)]),
        _qa_call("Q12026", [QAExchange("Morgan Lee", _MARGIN_Q, _QUANT_A)]),
    ]
    with patch("analysis.tone_drift._embed", side_effect=_keyword_embed):
        assert compute_recurring_evasions(calls) == []


def test_no_evasion_when_question_topic_not_recurring():
    calls = [
        _qa_call("Q32025", [QAExchange("Morgan Lee", _MARGIN_Q, _EVASIVE_A)]),
        _qa_call("Q12026", [QAExchange("Alex Carter", _CHINA_Q, _EVASIVE_A)]),
    ]
    with patch("analysis.tone_drift._embed", side_effect=_keyword_embed):
        assert compute_recurring_evasions(calls) == []


def test_recurring_evasion_requires_two_qa_calls():
    calls = [_qa_call("Q12026", [QAExchange("Alex Carter", _CHINA_Q, _EVASIVE_A)])]
    assert compute_recurring_evasions(calls) == []


# ---------------------------------------------------------------------------
# Tests: evasion scoring
# ---------------------------------------------------------------------------

def test_is_evasive_on_deflection_phrase():
    evasive, phrase = _is_evasive("Honestly it is too early to say anything about that.")
    assert evasive and phrase == "too early to say"


def test_is_evasive_on_short_nonquantitative_answer():
    evasive, phrase = _is_evasive(
        "We feel good about the trajectory and remain focused on execution across the portfolio."
    )
    assert evasive and phrase == ""


def test_not_evasive_when_quantitative():
    evasive, _ = _is_evasive(_QUANT_A)
    assert not evasive


def test_question_topic_extracts_keywords():
    topic = _question_topic(_CHINA_Q)
    assert "china" in topic


# ---------------------------------------------------------------------------
# Tests: topic fades
# ---------------------------------------------------------------------------

def test_topic_fade_detects_dropped_backlog():
    calls = [
        _prepared_call("Q22025", 3),
        _prepared_call("Q32025", 2),
        _prepared_call("Q12026", 0),  # backlog gone
    ]
    deltas = compute_topic_fades(calls)
    backlog = [d for d in deltas if d.term == "backlog"]
    assert len(backlog) == 1
    d = backlog[0]
    assert d.kind == "topic_fade"
    assert "absent in Q12026" in d.computed_metric
    assert "Q22025 and Q32025" in d.computed_metric
    assert d.after_text == ""


def test_topic_fade_skips_when_still_present():
    calls = [
        _prepared_call("Q22025", 3),
        _prepared_call("Q32025", 2),
        _prepared_call("Q12026", 1),  # still mentioned
    ]
    assert [d for d in compute_topic_fades(calls) if d.term == "backlog"] == []


def test_topic_fade_requires_prominence_in_two_priors():
    calls = [
        _prepared_call("Q22025", 0),
        _prepared_call("Q32025", 2),  # only one prominent prior
        _prepared_call("Q12026", 0),
    ]
    assert [d for d in compute_topic_fades(calls) if d.term == "backlog"] == []


# ---------------------------------------------------------------------------
# Tests: compute() top level
# ---------------------------------------------------------------------------

def test_compute_empty_with_fewer_than_two_transcripts():
    with patch("analysis.tone_drift.get_recent_transcripts", return_value=[("Q12026", CALL_TEXT)]):
        assert compute("FAKE") == []


def test_compute_never_raises():
    with patch("analysis.tone_drift.get_recent_transcripts", side_effect=RuntimeError("DB gone")):
        assert compute("FAKE") == []


def test_compute_handles_garbage_transcripts():
    garbage = [("Q32025", "?!? ###"), ("Q42025", "no structure here"), ("Q12026", "still nothing")]
    with patch("analysis.tone_drift.get_recent_transcripts", return_value=garbage):
        result = compute("FAKE")
    assert isinstance(result, list)


def test_compute_caps_at_six_and_dedupes():
    def _many(kind: str, n: int, term_prefix: str) -> list[QuarterDelta]:
        return [
            QuarterDelta(
                kind=kind, period_from="Q32025", period_to="Q12026",
                source="transcript", significance="MEDIUM", term=f"{term_prefix}{i}",
            )
            for i in range(n)
        ]

    transcripts = [("Q32025", CALL_TEXT), ("Q42025", CALL_TEXT), ("Q12026", CALL_TEXT)]
    high = QuarterDelta(
        kind="recurring_evasion", period_from="Q32025", period_to="Q12026",
        source="transcript", significance="HIGH", term="china",
    )
    dupe = high.model_copy()
    with patch("analysis.tone_drift.get_recent_transcripts", return_value=transcripts), \
         patch("analysis.tone_drift.compute_tone_trend", return_value=_many("tone_trend", 3, "t")), \
         patch("analysis.tone_drift.compute_topic_arcs", return_value=_many("topic_arc", 3, "a")), \
         patch("analysis.tone_drift.compute_recurring_evasions", return_value=[high, dupe]), \
         patch("analysis.tone_drift.compute_topic_fades", return_value=_many("topic_fade", 3, "f")):
        result = compute("FAKE")

    assert len(result) == 6
    assert result[0].significance == "HIGH"
    assert sum(1 for d in result if d.kind == "recurring_evasion") == 1  # deduped


def test_quarter_delta_accepts_new_kinds():
    for kind in ("tone_trend", "topic_arc", "recurring_evasion", "topic_fade"):
        d = QuarterDelta(kind=kind, period_from="Q32025", period_to="Q12026", source="transcript")
        assert QuarterDelta.model_validate(d.model_dump()).kind == kind