analysis/theme_sentiment.py

"""Quarter-by-quarter management sentiment by investor-relevant theme.

The analyser is deliberately deterministic.  It reads management speech from
stored earnings-call transcripts, isolates sentences that concern one of five
fixed themes, and measures three separate things:

* sentiment: positive minus negative/uncertain language, scaled -100..+100;
* attention: theme-bearing sentences per 10,000 management words;
* evidence: the number of theme-bearing sentences behind the score.

Only aggregates leave this module.  The source sentences are never included in
the returned object, which lets the dashboard explain a change without showing
the original transcript text.
"""
from __future__ import annotations

import re
from collections import Counter
from dataclasses import dataclass, field
from typing import Iterable

from analysis import theme_terms
from analysis.transcript_parse import parse_call
from storage.sections_db import get_recent_transcripts


THEMES: tuple[dict, ...] = (
    {
        "key": "products",
        "label": "Products",
        "patterns": (
            r"\bproducts?\b", r"\bplatforms?\b", r"\bservices?\b",
            r"\bportfolio\b", r"\broadmap\b", r"\blaunch(?:es|ed|ing)?\b",
            r"\badoption\b", r"\binnovation\b", r"\bfeatures?\b",
        ),
    },
    {
        "key": "geography",
        "label": "Geography",
        "patterns": (
            r"\bgeograph(?:y|ic|ies)\b", r"\bregions?\b", r"\bcountries\b",
            r"\bamericas?\b", r"\beurope\b", r"\bemea\b", r"\bapac\b",
            r"\basia\b", r"\bchina\b", r"\bjapan\b", r"\bindia\b",
            r"\binternational\b", r"\bdomestic\b", r"\boverseas\b",
        ),
    },
    {
        "key": "geopolitics",
        "label": "Geopolitics",
        "patterns": (
            r"\bgeopolitic\w*\b", r"\bexport controls?\b", r"\bsanctions?\b",
            r"\btariffs?\b", r"\btrade restrictions?\b", r"\btrade war\b",
            r"\bregulat(?:ion|ory|ors?)\b", r"\bgovernment\w*\b",
            r"\bconflicts?\b", r"\bwar\b", r"\bembargo\b",
        ),
    },
    {
        "key": "growth",
        "label": "Growth",
        "patterns": (
            r"\bgrowth\b", r"\brevenue\b", r"\bdemand\b", r"\bbookings?\b",
            r"\bbacklog\b", r"\borders?\b", r"\bexpansion\b",
            r"\bmarket share\b", r"\bvolume\b", r"\bunits?\b",
        ),
    },
    {
        "key": "margin",
        "label": "Margin",
        "patterns": (
            r"\bmargins?\b", r"\bprofitab\w*\b", r"\bgross profit\b",
            r"\boperating leverage\b", r"\bcosts?\b", r"\bexpenses?\b",
            r"\bopex\b", r"\bmix\b", r"\befficienc\w*\b",
        ),
    },
)


# Stems are intentional: "accelerating" and "accelerated" should be the same
# signal.  The list is compact enough to be auditable and sector-neutral.
_POSITIVE_STEMS = (
    "accelerat", "ahead", "benefit", "confident", "expand", "exceed",
    "growth", "improv", "momentum", "opportun", "outperform", "record",
    "resilien", "robust", "strong", "strength", "successful", "upside",
)
_NEGATIVE_STEMS = (
    "challeng", "constrain", "declin", "deteriorat", "disappoint", "downturn",
    "headwind", "pressure", "risk", "shortage", "slow", "soft", "uncertain",
    "weak", "worsen", "volatile", "delay", "loss",
)
_HEDGE_PHRASES = (
    "i think", "we believe", "too early", "hard to say", "it depends",
    "we'll see", "roughly", "somewhat", "a bit", "remains to be seen",
)
_NEGATIONS = frozenset({"not", "no", "never", "neither", "without"})


EMERGING_TOPICS: tuple[tuple[str, tuple[str, ...]], ...] = (
    ("AI", (r"\bartificial intelligence\b", r"\bgenerative ai\b", r"\bai\b")),
    ("Pricing", (r"\bpricing\b", r"\bprice increases?\b", r"\bdiscount\w*\b")),
    ("China", (r"\bchina\b", r"\bchinese\b")),
    ("Export controls", (r"\bexport controls?\b", r"\bexport restrictions?\b")),
    ("Tariffs", (r"\btariffs?\b", r"\bimport duties\b")),
    ("Capacity", (r"\bcapacity\b", r"\bcapacity constraints?\b")),
    ("Supply chain", (r"\bsupply chain\b", r"\bsupply constraints?\b")),
    ("Inventory", (r"\binventor(?:y|ies)\b",)),
    ("Competition", (r"\bcompetit(?:ion|ive|ors?)\b", r"\bmarket share\b")),
    ("Capital expenditure", (r"\bcapital expenditures?\b", r"\bcapex\b")),
    ("Currency", (r"\bforeign exchange\b", r"\bcurrency\b", r"\bfx\b")),
    ("Restructuring", (r"\brestructur\w*\b", r"\bworkforce reduction\b")),
)


GEOGRAPHY_PATTERNS: tuple[str, ...] = next(
    theme["patterns"] for theme in THEMES if theme["key"] == "geography"
)


@dataclass(frozen=True)
class Vocabulary:
    """The company's own names, and the ones that carry no information.

    ``filed`` maps a theme key to the segment names the issuer reports revenue
    under.  They are added to that theme's patterns, which is the point: a
    sector-neutral lexicon matches the word "products" and never the iPhone, so
    without this the Products row for Apple is scored on generic vocabulary
    alone.  Because each name comes from a filing, the enrichment stays sourced.

    ``exclude`` holds the issuer's own name and the speakers'.  Both are said
    constantly and neither is a subject anyone chose to raise.
    """

    filed: dict = field(default_factory=dict)
    exclude: frozenset = frozenset()

    @property
    def filed_terms(self) -> frozenset:
        return frozenset(
            term.lower() for terms in self.filed.values() for term in terms
        )


def _label_theme(label: str) -> str:
    """Route one filed segment label to the theme it belongs to.

    Business-segment axes are not all alike: Apple reports Americas, Europe and
    Greater China where NVIDIA reports Compute and Graphics.  The existing
    geography patterns already know a region when they see one, so they decide
    rather than a second list that could drift away from the first.
    """
    return "geography" if _matches_any(label, GEOGRAPHY_PATTERNS) else "products"


def build_vocabulary(
    company_name: str,
    ticker: str,
    labels: list,
    transcripts: list,
    speakers: object = (),
) -> Vocabulary:
    """Filed names the calls actually use, plus the names to ignore.

    A label is only kept when it is spoken.  XBRL labels arrive machine-shaped
    -- "Microsoft Three Six Five Commercial Products And Cloud Services" -- and
    adding "Three" as a product name would inflate the theme with a word that
    means nothing.  Requiring the corpus to contain the term closes that door
    without a hand-written exception list.
    """
    exclude = theme_terms.build_exclusions(company_name, ticker, speakers)
    spoken = theme_terms.named_terms(" ".join(transcripts), exclude)
    lowered = {term.lower(): count for term, count in spoken.items()}

    filed: dict = {}
    for axis, label in labels:
        theme_key = _label_theme(str(label))
        for term in theme_terms.label_terms(label):
            if lowered.get(term.lower(), 0) >= theme_terms.MIN_TOTAL_MENTIONS:
                filed.setdefault(theme_key, set()).add(term)
    return Vocabulary(
        filed={key: tuple(sorted(values)) for key, values in filed.items()},
        exclude=exclude,
    )


def _themes_with(vocabulary: Vocabulary) -> tuple:
    """THEMES, each carrying the filed names routed to it."""
    if not vocabulary.filed:
        return THEMES
    enriched = []
    for theme in THEMES:
        extra = vocabulary.filed.get(theme["key"], ())
        patterns = theme["patterns"] + tuple(
            r"\b" + re.escape(term) + r"\b" for term in extra
        )
        enriched.append({**theme, "patterns": patterns, "filed": extra})
    return tuple(enriched)


def _sentences(text: str) -> list[str]:
    return [
        part.strip()
        for part in re.split(r"(?<=[.!?])\s+|\n+", str(text or ""))
        if len(part.split()) >= 4
    ]


def _matches_any(text: str, patterns: Iterable[str]) -> bool:
    return any(re.search(pattern, text, re.IGNORECASE) for pattern in patterns)


def _stem_hits(tokens: list[str], stems: tuple[str, ...]) -> tuple[int, int]:
    """Return (ordinary hits, negated hits) for *stems* in a token sequence."""
    hits = negated = 0
    for index, token in enumerate(tokens):
        if not any(token.startswith(stem) for stem in stems):
            continue
        window = tokens[max(0, index - 3):index]
        if any(word in _NEGATIONS for word in window):
            negated += 1
        else:
            hits += 1
    return hits, negated


def score_sentence(sentence: str) -> tuple[float, int, int, int]:
    """Return valence plus auditable positive/negative/uncertainty counts."""
    lower = sentence.lower()
    tokens = re.findall(r"[a-z']+", lower)
    pos, negated_pos = _stem_hits(tokens, _POSITIVE_STEMS)
    neg, negated_neg = _stem_hits(tokens, _NEGATIVE_STEMS)
    hedges = sum(lower.count(phrase) for phrase in _HEDGE_PHRASES)
    positive = pos + negated_neg
    negative = neg + negated_pos
    denominator = max(1.0, positive + negative + hedges * 0.5)
    valence = (positive - negative - hedges * 0.35) / denominator
    return max(-1.0, min(1.0, valence)), positive, negative, hedges


def _theme_period(text: str, patterns: tuple[str, ...]) -> dict:
    relevant = [sentence for sentence in _sentences(text) if _matches_any(sentence, patterns)]
    words = max(1, len(text.split()))
    scored = [score_sentence(sentence) for sentence in relevant]
    pos = sum(item[1] for item in scored)
    neg = sum(item[2] for item in scored)
    hedges = sum(item[3] for item in scored)

    # One isolated sentence is too fragile to call sentiment, but its attention
    # count is still useful.  Missing sentiment is preferable to false precision.
    score = None
    if len(relevant) >= 2:
        score = round(100 * sum(item[0] for item in scored) / len(relevant))

    evidence = len(relevant)
    if evidence >= 8 and pos + neg + hedges >= 4:
        confidence = "HIGH"
    elif evidence >= 3:
        confidence = "MEDIUM"
    elif evidence:
        confidence = "LOW"
    else:
        confidence = "NONE"

    return {
        "score": score,
        "attention_per_10k": round(evidence / words * 10_000, 1),
        "evidence_sentences": evidence,
        "positive_signals": pos,
        "negative_signals": neg,
        "uncertainty_signals": hedges,
        "confidence": confidence,
    }


def _topic_series(texts: list[str], patterns: tuple[str, ...]) -> tuple[list[int], list[float]]:
    counts: list[int] = []
    rates: list[float] = []
    for text in texts:
        count = sum(len(re.findall(pattern, text, re.IGNORECASE)) for pattern in patterns)
        counts.append(count)
        rates.append(round(count / max(1, len(text.split())) * 10_000, 1))
    return counts, rates


def _emerging_topics(periods: list[str], texts: list[str]) -> list[dict]:
    if len(texts) < 2:
        return []
    found: list[dict] = []
    for label, patterns in EMERGING_TOPICS:
        counts, rates = _topic_series(texts, patterns)
        latest_count, prior_count = counts[-1], counts[-2]
        latest_rate, prior_rate = rates[-1], rates[-2]
        recent_prior_counts = counts[max(0, len(counts) - 3):-1]
        prior_avg = sum(rates[:-1]) / max(1, len(rates) - 1)

        status = ""
        if latest_count >= 2 and prior_count == 0 and sum(recent_prior_counts) == 0:
            status = "NEW"
        elif (
            latest_count >= 3
            and latest_rate >= 8
            and latest_rate >= max(prior_avg * 2.0, prior_rate + 5)
        ):
            status = "SURGING"
        if not status:
            continue
        found.append({
            "theme": label,
            "status": status,
            "period": periods[-1],
            "mentions": latest_count,
            "attention_per_10k": latest_rate,
            "prior_attention_per_10k": prior_rate,
            "change_per_10k": round(latest_rate - prior_rate, 1),
        })
    return sorted(found, key=lambda item: (-item["change_per_10k"], item["theme"]))[:6]


def analyse_transcripts(
    transcripts: list[tuple[str, str]],
    vocabulary: "Vocabulary | None" = None,
) -> dict:
    """Analyse chronological ``(period, transcript)`` pairs.  Pure and testable.

    *vocabulary* is optional so the analyser stays callable from a test with
    nothing but text.  Supplied, it both sharpens the scores -- the issuer's
    own segment names join the theme patterns -- and lets the term breakdown
    mark which names carry a filing behind them.
    """
    usable = [(period, text) for period, text in transcripts if str(text or "").strip()]
    if len(usable) < 2:
        return {}

    vocabulary = vocabulary or Vocabulary()
    periods: list[str] = []
    management_texts: list[str] = []
    calls: list[theme_terms.Call] = []
    for period, raw_text in usable:
        parsed = parse_call(period, raw_text)
        management = parsed.management_text or parsed.prepared_text or raw_text
        if not management.strip():
            continue
        periods.append(period)
        management_texts.append(management)
        calls.append(theme_terms.Call(
            period=period,
            prepared_text=parsed.prepared_text or management,
            answers_text=parsed.answers_text,
        ))
    if len(periods) < 2:
        return {}

    themes = _themes_with(vocabulary)
    breakdown = theme_terms.analyse(
        calls,
        themes,
        score_sentence,
        exclude=vocabulary.exclude,
        filed=vocabulary.filed_terms,
    )

    theme_rows: list[dict] = []
    for theme in themes:
        values = [_theme_period(text, theme["patterns"]) for text in management_texts]
        latest_score = values[-1]["score"]
        previous_score = values[-2]["score"]
        change = (
            latest_score - previous_score
            if latest_score is not None and previous_score is not None
            else None
        )
        theme_rows.append({
            "key": theme["key"],
            "label": theme["label"],
            "periods": values,
            "latest_score": latest_score,
            "change": change,
            "latest_attention_per_10k": values[-1]["attention_per_10k"],
            "latest_evidence_sentences": values[-1]["evidence_sentences"],
            "confidence": values[-1]["confidence"],
            "terms": breakdown.get(theme["key"], []),
        })

    measurable = [row for row in theme_rows if row["change"] is not None]
    biggest_improvement = max(measurable, key=lambda row: row["change"], default=None)
    biggest_deterioration = min(measurable, key=lambda row: row["change"], default=None)

    return {
        "method": "deterministic_lexicon_v2",
        "filed_vocabulary": sorted(vocabulary.filed_terms),
        "periods": periods,
        "themes": theme_rows,
        "biggest_improvement": (
            {"theme": biggest_improvement["label"], "change": biggest_improvement["change"]}
            if biggest_improvement and biggest_improvement["change"] > 0 else None
        ),
        "biggest_deterioration": (
            {"theme": biggest_deterioration["label"], "change": biggest_deterioration["change"]}
            if biggest_deterioration and biggest_deterioration["change"] < 0 else None
        ),
        "emerging_themes": _emerging_topics(periods, management_texts),
    }


def _stored_vocabulary(ticker: str, transcripts: list) -> Vocabulary:
    """Build the vocabulary from the databases, degrading to an empty one.

    Every source here is an enhancement: a missing segments table or an
    unresolved LFS pointer must cost the section its product names, never the
    section itself.
    """
    company_name = ''
    labels: list = []
    try:
        from storage import metrics_db

        rows = metrics_db.get_all_metrics(ticker) or []
        company_name = next(
            (str(row.get('company_name') or '') for row in rows if row.get('company_name')),
            '',
        )
    except Exception:
        pass
    try:
        labels = _filed_labels(ticker)
    except Exception:
        labels = []

    speakers: list = []
    for period, text in transcripts:
        speakers.extend(parse_call(period, text).speakers)
    return build_vocabulary(
        company_name, ticker, labels, [text for _period, text in transcripts], speakers
    )


def _filed_labels(ticker: str) -> list:
    """Every distinct (axis, label) the issuer has reported a segment under."""
    import sqlite3

    from storage.segments_db import DB_PATH

    if not DB_PATH.exists():
        return []
    with sqlite3.connect(DB_PATH) as conn:
        return [
            (axis, label)
            for axis, label in conn.execute(
                "SELECT DISTINCT axis, label FROM segments WHERE ticker=?",
                (ticker.upper(),),
            )
            if label
        ]


def compute(ticker: str, n: int = 6) -> dict:
    """Load and analyse the latest calls for *ticker*.  Never raises."""
    try:
        transcripts = get_recent_transcripts(ticker.upper(), n=n)
        return analyse_transcripts(
            transcripts, _stored_vocabulary(ticker.upper(), transcripts)
        )
    except Exception as exc:
        import sys

        print(f"[theme_sentiment] Error computing sentiment for {ticker}: {exc}", file=sys.stderr)
        return {}