"""analysis/tone_drift.py — multi-quarter transcript drift signals for the Analyst Edge layer. Pure Python + sentence-transformers, zero LLM calls. Extends the textdiff approach (filings) to earnings-call transcripts across the last N quarters: 1. tone_trend — hedge/certainty word rate trajectory in management speech 2. topic_arc — lexicon term frequency rising/falling across 3+ calls 3. recurring_evasion — analyst question asked on 2+ calls, answered evasively 4. topic_fade — topic prominent in prior prepared remarks, absent now Usage: from analysis.tone_drift import compute signals = compute("NVDA") """ from __future__ import annotations import re from collections import Counter from analysis.signals import QuarterDelta from analysis.textdiff import ( _LEXICON, _KPI_PATTERNS, _detect_trend, _embed, _find_context_sentence, _split_sentences, _truncate, ) from analysis.transcript_parse import ParsedCall, parse_call from storage.sections_db import _transcript_sort_key, get_recent_transcripts # --------------------------------------------------------------------------- # Config # --------------------------------------------------------------------------- # Spoken hedging — looser register than the written _HEDGE_WORDS in textdiff. _SPOKEN_HEDGES = [ "i think", "we believe", "sort of", "kind of", "we'll see", "hard to say", "too early", "it depends", "uncertain", "cautious", "headwind", "challenge", "moderate", "soften", "roughly", "somewhat", "a bit of", "remains to be seen", ] _CERTAINTY_WORDS = [ "we will", "confident", "strong", "record", "robust", "momentum", "accelerat", "very pleased", "outstanding", "exceed", "ahead of plan", "better than expected", ] # Phrases that mark a non-answer to an analyst question. _DEFLECTIONS = [ "we don't guide", "we do not guide", "not going to guide", "too early to say", "too early to tell", "as i said", "as we said", "we don't disclose", "we do not disclose", "won't break out", "don't break out", "not going to get into", "stay tuned", "more to come", "we'll see how", ] # Transcript-specific topics beyond the shared filing lexicon. _TRANSCRIPT_EXTRA_TERMS: list[tuple[str, str]] = [ (r"\bdemand\b", "demand"), (r"\binventory\b", "inventory"), (r"\bpricing\b", "pricing"), (r"\bvisibility\b", "visibility"), (r"\bsupply\b", "supply"), (r"\bchina\b", "China"), (r"\bbacklog\b", "backlog"), ] _EVASION_SIM_THRESHOLD = 0.60 # cosine: two questions considered the same topic _MIN_QUESTION_WORDS = 15 _MAX_QUESTIONS_PER_CALL = 30 _QUESTION_EMBED_WORDS = 120 _DIGIT_RATIO_THRESHOLD = 0.005 # answers with fewer digits than this are "non-quantitative" _SHORT_ANSWER_WORDS = 80 _MAX_TOTAL = 6 _STOPWORDS = frozenset( "a an and are as at be but by can could for from has have how i if in is it " "just like me my of on or our so that the then there this to was we what when " "which will with would you your about more very really them they those these " "going get got want wanted maybe think know kind sort little also any do does " "your guys thanks thank question congrats curious wondering color give us " "side year years quarter quarters talk talked talking look looking lot bit " "say said see seeing help understand grew growing mentioned should".split() ) # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _word_count(text: str) -> int: return len(text.split()) def _phrase_rate(text: str, phrases: list[str]) -> int: """Occurrences of any phrase per 10k words, rounded to int.""" words = _word_count(text) if words == 0: return 0 lower = text.lower() hits = sum(lower.count(p) for p in phrases) return round(hits / words * 10_000) def _rep_sentence(text: str, phrases: list[str]) -> str: """Shortest quotable sentence containing one of the phrases.""" matches = [ s for s in _split_sentences(text) if any(p in s.lower() for p in phrases) ] if not matches: return "" return _truncate(min(matches, key=lambda s: len(s.split())), 60) def _question_topic(question: str) -> str: """2-3 most frequent non-stopword tokens as a compact topic label.""" tokens = re.findall(r"[a-z]{3,}", question.lower()) counts = Counter(t for t in tokens if t not in _STOPWORDS) return " / ".join(t for t, _ in counts.most_common(3)) def _is_evasive(answer: str) -> tuple[bool, str]: """(evasive?, deflection phrase found or '').""" lower = answer.lower() for phrase in _DEFLECTIONS: if phrase in lower: return True, phrase words = answer.split() if words and len(words) < _SHORT_ANSWER_WORDS: digit_tokens = sum(1 for w in words if any(c.isdigit() for c in w)) if digit_tokens / len(words) < _DIGIT_RATIO_THRESHOLD: return True, "" return False, "" # --------------------------------------------------------------------------- # 1. Management tone trend # --------------------------------------------------------------------------- def compute_tone_trend(calls: list[ParsedCall]) -> list[QuarterDelta]: """Hedge/certainty word rate trajectory across 3+ calls (management speech only).""" usable = [c for c in calls if _word_count(c.management_text) >= 500] if len(usable) < 3: return [] deltas: list[QuarterDelta] = [] series_specs = [ (_SPOKEN_HEDGES, "hedging language", "hedge-word", {"rising": "more cautious", "falling": "more confident"}), (_CERTAINTY_WORDS, "confidence language", "certainty-word", {"rising": "more confident", "falling": "more cautious"}), ] for phrases, term, rate_label, direction_map in series_specs: rates = [_phrase_rate(c.management_text, phrases) for c in usable] trend = _detect_trend(rates) if trend is None: continue direction, run_str, _ = trend.split() run_quarters = int(run_str) reading = direction_map[direction] first, last = rates[0], rates[-1] net_change = abs(last - first) / (first or 1) sig = "HIGH" if (run_quarters >= 4 or net_change >= 0.5) else "MEDIUM" trajectory = "→".join(str(r) for r in rates) metric = ( f"{rate_label} rate {trajectory} per 10k words over " f"{usable[0].period}→{usable[-1].period} ({trend}) → {reading}" ) deltas.append(QuarterDelta( kind="tone_trend", period_from=usable[0].period, period_to=usable[-1].period, before_text=_rep_sentence(usable[0].management_text, phrases), after_text=_rep_sentence(usable[-1].management_text, phrases), computed_metric=metric, source="transcript", significance=sig, term=term, )) return deltas # --------------------------------------------------------------------------- # 2. Topic emphasis arcs # --------------------------------------------------------------------------- def compute_topic_arcs( calls: list[ParsedCall], raw_texts: list[str], ) -> list[QuarterDelta]: """Lexicon terms whose mention count rises/falls monotonically across 3+ calls.""" if len(calls) < 3: return [] deltas: list[QuarterDelta] = [] for pattern, label in _LEXICON + _TRANSCRIPT_EXTRA_TERMS: counts = [len(re.findall(pattern, t, re.IGNORECASE)) for t in raw_texts] if max(counts) < 3: continue # noise floor trend = _detect_trend(counts) if trend is None: continue run_quarters = int(trend.split()[1]) metric = ( f"{counts[0]}→{counts[-1]} mentions over " f"{calls[0].period}→{calls[-1].period} ({trend})" ) deltas.append(QuarterDelta( kind="topic_arc", period_from=calls[0].period, period_to=calls[-1].period, before_text=_find_context_sentence(raw_texts[0], pattern) if counts[0] else "", after_text=_find_context_sentence(raw_texts[-1], pattern) if counts[-1] else "", computed_metric=metric, source="transcript", significance="HIGH" if run_quarters >= 4 else "MEDIUM", term=label, )) deltas.sort(key=lambda d: {"HIGH": 0, "MEDIUM": 1}.get(d.significance, 2)) return deltas[:3] # --------------------------------------------------------------------------- # 3. Recurring Q&A evasions # --------------------------------------------------------------------------- def compute_recurring_evasions(calls: list[ParsedCall]) -> list[QuarterDelta]: """Analyst question topics raised on 2+ calls where answers stay non-quantitative.""" qa_calls = [c for c in calls if c.qa] if len(qa_calls) < 2: return [] latest = qa_calls[-1] priors = qa_calls[:-1] def _select(call: ParsedCall) -> list: picked = [x for x in call.qa if _word_count(x.question) >= _MIN_QUESTION_WORDS] return picked[:_MAX_QUESTIONS_PER_CALL] latest_qs = _select(latest) prior_qs: list[tuple[str, object]] = [] # (period, QAExchange) for call in priors: prior_qs.extend((call.period, x) for x in _select(call)) if not latest_qs or not prior_qs: return [] texts = ( [_truncate(x.question, _QUESTION_EMBED_WORDS) for x in latest_qs] + [_truncate(x.question, _QUESTION_EMBED_WORDS) for _, x in prior_qs] ) vecs = _embed(texts) latest_vecs = vecs[: len(latest_qs)] prior_vecs = vecs[len(latest_qs):] sim = latest_vecs @ prior_vecs.T # (n_latest, n_prior) deltas: list[QuarterDelta] = [] used_prior: set[int] = set() for li, lx in enumerate(latest_qs): matched = [ pi for pi in range(len(prior_qs)) if pi not in used_prior and sim[li, pi] >= _EVASION_SIM_THRESHOLD ] if not matched: continue # prior_qs is in chronological call order, so cluster[0] is the earliest. cluster = [(prior_qs[pi][0], prior_qs[pi][1]) for pi in matched] cluster.append((latest.period, lx)) periods = sorted({p for p, _ in cluster}, key=_transcript_sort_key) if len(periods) < 2: continue evasive_flags = [_is_evasive(x.answer) for _, x in cluster] n_evasive = sum(1 for flag, _ in evasive_flags if flag) # Require a majority of evasive answers, not just two outliers in a # large cluster of recurring questions. if n_evasive < 2 or n_evasive * 2 < len(cluster): continue used_prior.update(matched) deflection = next((p for flag, p in evasive_flags if flag and p), "") deflection_str = f" (deflection: '{deflection}')" if deflection else "" metric = ( f"asked in {', '.join(periods)}; " f"{n_evasive}/{len(cluster)} answers non-quantitative{deflection_str}" ) earliest_period, earliest_x = cluster[0] latest_evasive = next( (x for (_, x), (flag, _) in zip(reversed(cluster), reversed(evasive_flags)) if flag), lx, ) answer_quote = ( _rep_sentence(latest_evasive.answer, [deflection]) if deflection else "" ) or _truncate(latest_evasive.answer, 60) deltas.append(QuarterDelta( kind="recurring_evasion", period_from=periods[0], period_to=latest.period, before_text=_truncate(f"{earliest_x.analyst}: {earliest_x.question}", 60), after_text=answer_quote, computed_metric=metric, source="transcript", significance="HIGH" if len(periods) >= 3 else "MEDIUM", term=_question_topic(lx.question), )) deltas.sort(key=lambda d: {"HIGH": 0, "MEDIUM": 1}.get(d.significance, 2)) return deltas[:2] # --------------------------------------------------------------------------- # 4. Prepared-remarks topic fades # --------------------------------------------------------------------------- def compute_topic_fades(calls: list[ParsedCall]) -> list[QuarterDelta]: """Topic with 2+ mentions in 2+ prior prepared remarks, absent from the latest.""" usable = [c for c in calls if _word_count(c.prepared_text) >= 300] if len(usable) < 3: return [] latest = usable[-1] priors = usable[:-1] deltas: list[QuarterDelta] = [] seen_labels: set[str] = set() for pattern, label in _KPI_PATTERNS + _TRANSCRIPT_EXTRA_TERMS: if label in seen_labels: continue # 'backlog' appears in both lists seen_labels.add(label) prior_counts = [ (c, len(re.findall(pattern, c.prepared_text, re.IGNORECASE))) for c in priors ] prominent = [(c, n) for c, n in prior_counts if n >= 2] if len(prominent) < 2: continue if re.search(pattern, latest.prepared_text, re.IGNORECASE): continue periods_str = " and ".join(c.period for c, _ in prominent) counts_str = ", ".join(str(n) for _, n in prominent) last_prominent = prominent[-1][0] deltas.append(QuarterDelta( kind="topic_fade", period_from=prominent[0][0].period, period_to=latest.period, before_text=_find_context_sentence(last_prominent.prepared_text, pattern), after_text="", computed_metric=( f"'{label}' in prepared remarks of {periods_str} " f"({counts_str} mentions), absent in {latest.period}" ), source="transcript", significance="MEDIUM", term=label, )) return deltas[:2] # --------------------------------------------------------------------------- # Main entry point # --------------------------------------------------------------------------- def compute(ticker: str, n: int = 4) -> list[QuarterDelta]: """Compute all transcript drift signals for a ticker. Examines the last *n* non-empty transcripts. Returns an empty list if fewer than 2 are available or on any error — never raises. """ try: return _compute_inner(ticker, n) except Exception as exc: import sys print(f"[tone_drift] Error computing deltas for {ticker}: {exc}", file=sys.stderr) return [] def _compute_inner(ticker: str, n: int) -> list[QuarterDelta]: transcripts = get_recent_transcripts(ticker.upper(), n=n) if len(transcripts) < 2: return [] raw_texts = [text for _, text in transcripts] calls = [parse_call(period, text) for period, text in transcripts] all_deltas: list[QuarterDelta] = [] all_deltas.extend(compute_tone_trend(calls)) all_deltas.extend(compute_topic_arcs(calls, raw_texts)) all_deltas.extend(compute_recurring_evasions(calls)) all_deltas.extend(compute_topic_fades(calls)) # Dedupe by (kind, term), HIGH first, global cap. seen: set[tuple[str, str]] = set() deduped: list[QuarterDelta] = [] order = {"HIGH": 0, "MEDIUM": 1, "LOW": 2} all_deltas.sort(key=lambda d: (order[d.significance], d.kind)) for d in all_deltas: key = (d.kind, d.term) if key not in seen: seen.add(key) deduped.append(d) return deduped[:_MAX_TOTAL]