analysis/theme_terms.py

"""What management actually named, inside each theme it was talking about.

The heatmap next door scores five themes on lexical tone.  On its own that
score is unfalsifiable: a reader shown "Products +23" has no way to learn what
management was calling a product.  Worse, the theme patterns are sector-neutral
by design -- they match the word "products", never the iPhone -- so the cell can
be built entirely on generic vocabulary.

This module counts the named things inside the sentences each theme already
matched: iPhone, Apple Intelligence, Blackwell, Greater China, H20.  Two sources
feed it, and they are kept apart because their standing differs:

* **filed** terms appear in the company's own segment disclosure, which carries
  a SEC accession behind it;
* **discovered** terms are proper nouns the call itself used, found with no
  per-ticker curation at all.

Only counts leave this module.  A frequency is an aggregate statistic, not a
quotation, so the section keeps its promise to render no transcript sentence.
"""
from __future__ import annotations

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

# Words, keeping the internal punctuation that carries meaning in a name:
# "U.S", "R&D", "China-based", "TSMC/Foxconn".  Apostrophes are deliberately
# left out, which is what stops "I'll" and "we're" from reading as names.
_TOKEN_RE = re.compile(r"[A-Za-z][A-Za-z0-9]*(?:[.&/-][A-Za-z0-9]+)*")

# Capitalised in every transcript, informative in none: call furniture, the
# safe-harbour paragraph, and the titles of the people speaking.
_BOILERPLATE = frozenset("""
form forms sec gaap non-gaap ir investor investors relations safe harbor act
securities exchange commission litigation reform private research report
operator thank thanks welcome hello everyone please question questions answer
good morning afternoon evening today tomorrow yesterday
january february march april may june july august september october november
december monday tuesday wednesday thursday friday saturday sunday
president chief executive officer financial chairman chairwoman director
company companies corporation corporations incorporated
section sections item note notes page slide slides appendix
q1 q2 q3 q4 h1 h2 fy fy24 fy25 fy26 fy27 cy
""".split())

# Words a filed label carries for structure rather than for meaning. Kept out
# of the vocabulary because "Rest" of Asia Pacific and "Home" in Wearables,
# Home & Accessories would otherwise be scored as things the company sells.
_LABEL_FILLER = frozenset(
    "rest other others all and the total member members segment segments "
    "reportable corporate unallocated home sales revenue revenues".split()
)

# A corporate suffix says nothing about the issuer; the rest of the registered
# name does, and is the one term guaranteed to top every theme if kept.
_CORPORATE_SUFFIXES = frozenset(
    "inc inc. corp corp. corporation co co. company ltd ltd. llc plc "
    "holdings holding group the and".split()
)

# Spellings of the same subject that would otherwise occupy two rows of the
# same table. Kept deliberately short: this is punctuation, not a synonym list.
_ALIASES = {"us": "U.S.", "u.s": "U.S.", "usa": "U.S.", "usg": "U.S. government"}

# Below this a term is an accident of one sentence rather than a subject.
MIN_TOTAL_MENTIONS = 2

# How many terms a single theme shows.  Enough to be a breakdown, few enough to
# stay readable under a heatmap row.
TERMS_PER_THEME = 8

# A tone reading on one sentence is noise; the parent module draws the same line.
MIN_SENTENCES_FOR_TONE = 2


@dataclass(frozen=True)
class Call:
    """One earnings call, already split into who chose to say what.

    The split matters more than it looks.  Prepared remarks are what management
    decided to put in front of investors; answers are what analysts made it
    address.  A subject that lives only in the answer column is a subject the
    company is not volunteering.
    """

    period: str
    prepared_text: str = ""
    answers_text: str = ""

    @property
    def management_text(self) -> str:
        return f"{self.prepared_text}\n{self.answers_text}".strip()


# ── Extraction ───────────────────────────────────────────────────────────────

def _core(token: str) -> str:
    return token.strip(".-&/")


def _singular(word: str) -> str:
    """Fold the plural of an acronym into the acronym.

    "GPUs" and "GPU" are one subject and must not occupy two rows. The test is
    deliberately narrow -- the stem has to be entirely upper case -- so that
    AirPods stays AirPods rather than becoming an AirPod.
    """
    stem = word[:-1]
    return stem if word.endswith("s") and len(stem) >= 2 and stem.isupper() else word


def _flush(found: Counter, run: list, exclude: frozenset[str]) -> None:
    """Record a finished run of capitalised words as one name."""
    if not run:
        return
    term = " ".join(_ALIASES.get(word.lower(), _singular(word)) for word in run)
    if _keep(term, exclude):
        found[term] += 1


def _is_named(token: str, first_in_sentence: bool) -> bool:
    """True when a token reads as the name of a thing rather than as prose."""
    core = _core(token)
    if len(core) < 2 or core.lower() in _BOILERPLATE:
        return False
    # Internal capitals settle it on their own: iPhone, NVLink, GPUs, CUDA.
    if any(character.isupper() for character in core[1:]):
        return True
    if not core[0].isupper():
        return False
    # A capital with a digit is a part number, not a sentence opening: H20, A18.
    if any(character.isdigit() for character in core):
        return True
    # An ordinary capitalised word is only a name away from the first slot,
    # where every sentence capitalises whatever happens to start it.
    return not first_in_sentence and len(core) >= 3


def _keep(term: str, exclude: frozenset[str]) -> bool:
    """Drop a finished name that is only the issuer's or a speaker's.

    The test runs on the whole name rather than on each word, which is the
    difference between "Apple Intelligence" surviving intact and being cut down
    to a phantom "Intelligence" because the issuer's own name opens it.
    """
    words = term.lower().split()
    return bool(words) and not all(word in exclude for word in words)


def named_terms(text: str, exclude: frozenset[str] = frozenset()) -> Counter:
    """Count the named terms in *text*, merging adjacent words into one name.

    Runs are joined so that "Apple Watch" and "Greater China" survive as
    themselves, and broken on anything that is not whitespace so that the list
    "iPhone, Mac, iPad and Apple Watch" yields four names rather than one.
    """
    found: Counter = Counter()
    for sentence in re.split(r"(?<=[.!?])\s+|\n+", str(text or "")):
        run: list[str] = []
        previous_end = 0
        for index, match in enumerate(_TOKEN_RE.finditer(sentence)):
            separated = sentence[previous_end:match.start()].strip() != ""
            previous_end = match.end()
            if separated and run:
                _flush(found, run, exclude)
                run = []
            if _is_named(match.group(0), index == 0):
                run.append(_core(match.group(0)))
            elif run:
                _flush(found, run, exclude)
                run = []
        _flush(found, run, exclude)
    return found


def build_exclusions(
    company_name: str = "",
    ticker: str = "",
    speakers: object = (),
) -> frozenset[str]:
    """Names that are always present and never informative.

    The issuer says its own name in every second sentence, and the people on the
    call are named at every hand-off.  Neither is a subject management chose.
    """
    excluded: set[str] = set()
    for word in re.findall(r"[A-Za-z][A-Za-z0-9.&-]*", str(company_name or "")):
        cleaned = _core(word).lower()
        if cleaned and cleaned not in _CORPORATE_SUFFIXES:
            excluded.add(cleaned)
    if ticker:
        excluded.add(str(ticker).lower())
    for speaker in speakers or ():
        for part in re.findall(r"[A-Za-z][A-Za-z0-9.&-]*", str(speaker)):
            cleaned = _core(part).lower()
            if len(cleaned) >= 2:
                excluded.add(cleaned)
    return frozenset(excluded)


def label_terms(label: str) -> tuple[str, ...]:
    """The searchable names inside one filed segment label.

    XBRL labels arrive machine-shaped -- "Wearables, Home & Accessories",
    "Microsoft Three Six Five Commercial Products And Cloud Services" -- so the
    whole phrase is rarely spoken.  Both the phrase and its distinctive words
    are returned; the caller keeps only those the calls actually used, which is
    what stops "Three" and "Commercial" from becoming product names.
    """
    cleaned = " ".join(str(label or "").split())
    if not cleaned:
        return ()
    terms = {cleaned}
    words = [word for word in re.split(r"[^A-Za-z0-9&.-]+", cleaned) if word]
    for word in words:
        if len(word) >= 3 and word.lower() not in _CORPORATE_SUFFIXES and word.lower() not in _LABEL_FILLER:
            terms.add(word)
    # Two adjacent words catch the names that only exist as a pair.
    for first, second in zip(words, words[1:]):
        if first.lower() not in _LABEL_FILLER and second.lower() not in _LABEL_FILLER:
            terms.add(f"{first} {second}")
    return tuple(sorted(terms))


# ── Aggregation ──────────────────────────────────────────────────────────────

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) -> bool:
    return any(re.search(pattern, text, re.IGNORECASE) for pattern in patterns)


@dataclass
class _Period:
    """Working state for one theme in one call."""

    counts: Counter = field(default_factory=Counter)
    prepared: Counter = field(default_factory=Counter)
    answers: Counter = field(default_factory=Counter)
    carriers: dict = field(default_factory=dict)
    words: int = 1


def _collect(text: str, patterns, exclude: frozenset[str]) -> tuple[Counter, dict]:
    """Term counts and the sentences carrying them, within one theme."""
    counts: Counter = Counter()
    carriers: dict[str, list[str]] = {}
    for sentence in _sentences(text):
        if not _matches_any(sentence, patterns):
            continue
        for term, hits in named_terms(sentence, exclude).items():
            counts[term] += hits
            carriers.setdefault(term, []).append(sentence)
    return counts, carriers


def _tone(sentences: list[str], score_sentence) -> int | None:
    if len(sentences) < MIN_SENTENCES_FOR_TONE:
        return None
    return round(100 * sum(score_sentence(item)[0] for item in sentences) / len(sentences))


def analyse(
    calls: list[Call],
    themes,
    score_sentence,
    exclude: frozenset[str] = frozenset(),
    filed: frozenset[str] = frozenset(),
) -> dict[str, list[dict]]:
    """Per theme, the named terms behind it, newest period last.

    *score_sentence* is injected rather than imported so that the tone of a term
    is measured by exactly the lexicon that produced the heatmap cell above it —
    one scorer, not two that can drift apart.
    """
    if not calls:
        return {}

    breakdown: dict[str, list[dict]] = {}
    for theme in themes:
        patterns = theme["patterns"]
        periods: list[_Period] = []
        for call in calls:
            state = _Period(words=max(1, len(call.management_text.split())))
            state.counts, state.carriers = _collect(call.management_text, patterns, exclude)
            state.prepared, _ = _collect(call.prepared_text, patterns, exclude)
            state.answers, _ = _collect(call.answers_text, patterns, exclude)
            periods.append(state)

        totals: Counter = Counter()
        for state in periods:
            totals.update(state.counts)

        rows: list[dict] = []
        for term, total in totals.items():
            if total < MIN_TOTAL_MENTIONS:
                continue
            counts = [state.counts.get(term, 0) for state in periods]
            rates = [
                round(state.counts.get(term, 0) / state.words * 10_000, 1)
                for state in periods
            ]
            latest, prior = periods[-1], periods[-2] if len(periods) >= 2 else None
            rows.append({
                "term": term,
                "filed": term.lower() in filed,
                "total_mentions": total,
                "counts": counts,
                "rates": rates,
                "mentions": counts[-1],
                "rate": rates[-1],
                "prior_rate": rates[-2] if prior is not None else None,
                "change": round(rates[-1] - rates[-2], 1) if prior is not None else None,
                "prepared": latest.prepared.get(term, 0),
                "answers": latest.answers.get(term, 0),
                "sentences": len(latest.carriers.get(term, [])),
                "score": _tone(latest.carriers.get(term, []), score_sentence),
            })

        # The latest call leads, because that is the quarter being read. Weight
        # across every call breaks the tie, which keeps a name management has
        # stopped saying visible in the tail rather than deleting the signal.
        # Filed names come first at equal weight: one carries an accession, the
        # other carries only the call.
        rows.sort(key=lambda row: (
            -row["mentions"], -row["total_mentions"], not row["filed"], row["term"],
        ))
        breakdown[theme["key"]] = rows[:TERMS_PER_THEME]
    return breakdown