Spaces:
Sleeping
Sleeping
| """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" | |
| ) | |
| 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() | |
| 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 {} | |