"""Prose ingestion: LLM text → claim-tagged fragments, in ONE pass, with no rewriting. LLMs write explanation; they do not author numbers. Every figure in persona, PM and summary text must be a ``[[CLAIM_ID]]`` token, which is replaced by the claim's rendered value and keeps the claim's identity as a fragment. A sentence is REJECTED (removed, and recorded for internal QA) when it * contains a material number typed by the model instead of a token, * uses an unknown token, * places a claim next to a label that claim forbids (e.g. "the headline return of [[PM_RETURN]]"), * uses valuation vocabulary the selected engine forbids, * leans on a quantitative model whose reliability does not permit it as evidence, * describes scheduled Rule 10b5-1 sales as conviction or a governance problem, * names an executive who is not the canonical officer for that role. Nothing is rewritten. There is no second correction pass, so there is no correction loop: a sentence is either accepted as written or removed. """ from __future__ import annotations import re from units import format_percent from dataclasses import dataclass, field TOKEN = re.compile(r"\[\[([A-Z0-9_]+)\]\]") # a material number typed by the model: currency, percentages, multiples, basis points, decimals, # thousands-separated figures, magnitudes MATERIAL_NUMBER = re.compile( r"(\$\s?\d)|(\d\s?%)|(\b\d+(?:\.\d+)?\s?x\b)|(\b\d+(?:\.\d+)?\s?times\b)|(\b\d+\s?bps\b)|" r"(\b\d+\.\d+\b)|(\b\d{1,3}(?:,\d{3})+\b)|(\b\d+(?:\.\d+)?\s?(?:bn|billion|million|mn|trillion)\b)|" r"(\bpercent\b)|(\bpercentile\b\s*\d)|(\d+(?:st|nd|rd|th)\s+percentile)|" # basis points / percentage points / turns written out, and bare quantitative integers of two or more digits r"(\b\d+\s*(?:basis\s+points?|bps?|percentage\s+points?|points?|pts|turns?|notches)\b)|((? b_hi) if up else (a_hi < b_lo) if not true: word = (up or down).group(0) return x.group(1), y.group(1), [f"'{word}' is false for these values (the ranges overlap or the order is reversed)"] return None # ── structural relation rules (validator ids: RELATION_IDENTITY, RELATION_MAGNITUDE, RELATION_REQUIRED_FOR_PAIR, # MODEL_EVIDENCE_SCOPE, MODEL_EVIDENCE_VISIBILITY, GUIDANCE_EPISTEMIC_STATE, RETURN_DIRECTION_RENDER, # EXPECTATIONS_POLARITY). They read claim identities, never comparison words. def offered_violation(sentence, rules): """A token the author was not offered: a relation's magnitude without its pair, a signed quantity without its direction, or a hidden model's figure.""" if rules.offered is None: return None for m in TOKEN.finditer(sentence): cid = m.group(1) if cid == "OWN_RETURN" or cid in rules.offered: continue if cid in (rules.magnitude_bindings or {}) and cid not in ("PRICE_VS_INTRINSIC", "PRICE_VS_MECHANICAL", "PRICE_VS_COMPOSITE", "TARGET_VS_PRICE", "FORECAST_EPS_VS_CONSENSUS"): a, b = rules.magnitude_bindings[cid] return "RELATION_MAGNITUDE", f"[[{cid}]] is the magnitude of {a} vs {b} only; cite its relationship token" if cid in (rules.presentation_claims or ()): return "RELATION_IDENTITY", (f"[[{cid}]] is a relation's size and direction without its subject; cite the " "relationship token, which states both sides") if cid in (rules.phrase_of or {}): return "RETURN_DIRECTION_RENDER", f"[[{cid}]] is a signed figure; cite [[{rules.phrase_of[cid]}]], which carries its direction" if cid.endswith("__FULL_SENTENCE"): return "RELATION_CLAUSE_GRAMMAR", f"[[{cid}]] is a report-template form; authors write the relation as a clause" mid = rules.claim_model(cid) if rules.claim_model else None if mid: return "MODEL_EVIDENCE_VISIBILITY", f"[[{cid}]] belongs to {mid}, whose reliability does not allow it to be offered" return "RELATION_IDENTITY", f"[[{cid}]] was not offered to this author" return None def relation_violation(sentence, rules): """(validator, reason): a magnitude used for another pair, or two related quantities in one sentence without the approved relation that connects them (or with no relation possible at all).""" toks = list(dict.fromkeys(m.group(1) for m in TOKEN.finditer(sentence))) for mcid, pair in (rules.magnitude_bindings or {}).items(): if mcid not in toks: continue others = [t for t in toks if t != mcid] for i, a in enumerate(others): for b in others[i + 1:]: if frozenset({a, b}) != frozenset(pair) and rules.pair_fn and rules.pair_fn(a, b) is not None: return "RELATION_MAGNITUDE", (f"[[{mcid}]] measures {pair[0]} vs {pair[1]}; it cannot state the relation " f"between [[{a}]] and [[{b}]]") if rules.pair_fn is None: return None for i, a in enumerate(toks): for b in toks[i + 1:]: if a.startswith("GUIDANCE_") and b.startswith("GUIDANCE_"): continue # rows of one guidance table, as the company presents them st = rules.pair_fn(a, b) if st is None or st[0] == "OK": continue if st[0] == "RELATION": if st[1] not in toks: return "RELATION_REQUIRED_FOR_PAIR", (f"[[{a}]] and [[{b}]] share a sentence without their relationship " f"token [[{st[1]}]]") continue return "RELATION_REQUIRED_FOR_PAIR", (f"[[{a}]] and [[{b}]] cannot be related ({'; '.join(st[1])[:100]}); " "state them in separate sentences") return None # ── clause grammar around relationship tokens (function words only: where a clause may start and end) ── CLAUSE_CONJ = ("and", "but", "while", "whereas", "because", "since", "as", "so", "although", "though", "yet", "that", "or", "if", "when", "given", "today", "meanwhile", "however", "also", "thus", "hence", "now", "which", "where") MAX_FULL_PER_SENTENCE = 2 PREPOSITIONS = ("for", "of", "over", "in", "on", "at", "across", "through", "from", "to", "with", "within", "during") # a relative pronoun needs a gap in its clause ("which leaves …"); a complete clause after it has none ("which the price is …") CLAUSE_OPENERS = tuple(w for w in CLAUSE_CONJ if w != "which") # what may follow "[[FULL]], …": a conjunction, a relative clause, an appositive or a participial phrase — never a verb # whose subject would be the relation ("[[FULL]], destroy value") CLAUSE_CONTINUATIONS = CLAUSE_CONJ + ("who", "whose", "a", "an", "the", "leaving", "making", "meaning", "implying", "with") def clause_grammar_violation(sentence, full_clause): """A FULL relationship (or any clause-rendered claim) is a complete clause: it must open the sentence or follow a clause boundary (never a relative pronoun, which needs a gap the clause does not have), and be followed by the end of the sentence, a boundary, or a comma plus a continuation that is not a verb whose subject would be the relation ("[[FULL]], destroy value"); at most two per sentence.""" if full_clause is None: return None toks = [m for m in TOKEN.finditer(sentence) if full_clause(m.group(1))] if len(toks) > MAX_FULL_PER_SENTENCE: return f"{len(toks)} complete relationship clauses in one sentence (at most {MAX_FULL_PER_SENTENCE}); split it" for m in toks: before = sentence[:m.start()].rstrip() after = sentence[m.end():].lstrip() last = (re.findall(r"[A-Za-z']+", before) or [""])[-1].lower() ok_b = not before or before[-1] in ",;:(—–-\"'" or last in CLAUSE_OPENERS if not after or after[0] in ".;:)!?—–-\"'": ok_a = True elif after[0] == ",": nxt = (re.findall(r"[A-Za-z'’]+", after[1:30]) or [""])[0].lower() ok_a = not nxt or nxt in CLAUSE_CONTINUATIONS or nxt.endswith(("ing", "ed", "ly")) else: ok_a = (re.findall(r"[A-Za-z']+", after[:20]) or [""])[0].lower() in CLAUSE_CONJ if not ok_b: return (f"[[{m.group(1)}]] is a complete clause and cannot be the object of '{before.split()[-1]}'; use its " "__CLAUSE or __COMPACT form after the subject, or start a new clause") if not ok_a: return (f"[[{m.group(1)}]] is a complete clause and cannot be the subject of " f"'{after.lstrip(', ').split()[0]}'; end the clause after it") return None def operand_doubled(sentence, catalog): """A relation renders its own operands ("the price ($403.30) is 9.1% above …"): the same operand cited again as its own token in that sentence states the figure twice ("Buybacks at $403.30, the price ($403.30) is …"). The operands are the relation's typed lhs / rhs claims; a mode that does not render an operand does not own it.""" toks = [m.group(1) for m in TOKEN.finditer(sentence)] for rid in toks: rc = catalog.get(rid) if getattr(rc, "stype", None) != "REL" or not isinstance(getattr(rc, "value", None), dict): continue rendered = str(rc.rendered or "") for side in ("lhs_claim_id", "rhs_claim_id"): op = rc.value.get(side) oc = catalog.get(op) if op else None if op in toks and op != rid and oc is not None and oc.rendered and str(oc.rendered) in rendered: return (f"CLAIM_QUALIFIER_DOUBLED: [[{op}]] ({oc.rendered}) is already stated inside [[{rid}]]; cite the " "relation alone") return None def mode_subject_violation(sentence, rel_subjects): """A CLAUSE / COMPACT relation token is a predicate about its lhs: the words of its own clause before it must name the lhs and must not name the rhs (the relation can never be attached to the opposite subject).""" for m in TOKEN.finditer(sentence): cid = m.group(1) if "__" not in cid or cid.endswith("__FULL_SENTENCE"): continue base = cid.split("__")[0] lhs, rhs = (rel_subjects or {}).get(base, ((), ())) # coordinated predicates share the subject: "the price [[X__PREDICATE]] and [[Y__PREDICATE]]" if re.search(r"\[\[[A-Z0-9_]+__(?:PREDICATE|CLAUSE|COMPACT)\]\]\s*,?\s*(?:and|but|yet|while)\s*$", sentence[:m.start()]): prev_base = re.findall(r"\[\[([A-Z0-9_]+)__", sentence[:m.start()])[-1] if (rel_subjects or {}).get(prev_base, ((), ()))[0] == lhs: continue clause = re.split(r"[,;:(—–]|\b(?:and|but|while|whereas|because|since|so)\b", TOKEN.sub(lambda t: " " + t.group(1).lower() .replace("_", " ") + " ", sentence[:m.start()]))[-1].lower() if rhs and any(w.lower() in clause for w in rhs): return f"[[{cid}]] describes {base.split('_', 1)[1].lower().replace('_', ' ')}'s left-hand side; its subject here is the right-hand side" if lhs and not any(w.lower() in clause for w in lhs): return f"[[{cid}]] must follow the subject it describes ({', '.join(lhs[:3])})" return None def trend_violation(sentence, rules): """A trend conclusion (accelerating, decelerating, improving, deteriorating, expanding, contracting) about a financial metric needs an approved TrendRelationship in the sentence, or a typed change claim that carries it.""" import relations as _rel found = _rel.trend_asserted(TOKEN.sub(" ", sentence)) if not found: return None toks = {m.group(1) for m in TOKEN.finditer(sentence)} cited = {rules.trends_allowed[t] for t in toks if t in (rules.trends_allowed or {})} has_change = bool(toks & set(rules.change_claims or ())) for cls, word in found: if cls in cited: continue if cls in (_rel.IMPROVING, _rel.DETERIORATING, _rel.EXPANDING, _rel.CONTRACTING) and has_change: continue return (f"'{word}' asserts a {cls.lower()} trend without an approved trend relationship (no compatible earlier " "comparator on the same perimeter)") return None def interpretation_violation(sentence, interp): """Evidence a family may only DESCRIBE (e.g. 13F when it is not a return signal) used to carry an interpretation.""" import evidence as _ev low = TOKEN.sub(lambda t: " " + t.group(1).lower() + " ", sentence).lower() toks = {m.group(1) for m in TOKEN.finditer(sentence)} if re.search(r"\b(?:not|no|never)\b[^.]{0,30}\bsignal", low): return None # "not a signal" states the permission itself for fam, d in (interp or {}).items(): if d.get("permission") != _ev.DESCRIPTIVE_ONLY: continue refers = bool(toks & set(d.get("claims") or ())) or any(w in low for w in d.get("subjects") or ()) if not refers: continue hit = next((w for w in _ev.INTERPRETIVE_SURFACE if w in low), None) # ", aligning / signalling / suggesting …" or "which shows …": a consequence drawn from the evidence cons = re.search(r",\s*(?:\w+ly\s+)?(?!(?:including|excluding|according|following|totaling|totalling|representing|" r"holding|reporting|opening|adding|trimming|exiting)\b)(\w+ing)\b(?!\s+(?:to|in|from|over|across|" r"(?:the\s+|a\s+)?[a-z0-9_]*_[a-z0-9_]+\b))|" r"\bwhich\s+(?:shows?|signals?|suggests?|indicates?|means?|implies|reflects?)\b", low) if cons and not hit and any(re.search(r"\b" + re.escape(c.lower()) + r"\b", low) for c in d.get("claims") or ()): return (f"{fam} evidence is DESCRIPTIVE_ONLY here; '{cons.group(0).strip(', ')}' draws a consequence from it " "(state the positions or trades only)") if hit: return f"{fam} evidence is DESCRIPTIVE_ONLY here; '{hit.strip()}' interprets it (state the positions or counts only)" return None def guidance_binding_violation(sentence, names): """A GUIDANCE_n value attached to the wrong subject: its own clause (the text since the previous claim token, and up to the next one) must name its concept (a share-repurchase value is never a "private-markets base").""" toks = list(TOKEN.finditer(sentence)) for i, m in enumerate(toks): cid = m.group(1) if cid not in (names or {}): continue concept, words = names[cid][0], names[cid][1] basis_words = names[cid][2] if len(names[cid]) > 2 else () lo = toks[i - 1].end() if i > 0 else 0 hi = toks[i + 1].start() if i + 1 < len(toks) else len(sentence) # its own clause: back to the previous clause boundary or claim, forward to the next boundary or claim before = re.split(r"[,;:(—–]", sentence[lo:m.start()])[-1] after = re.split(r"[,;:)—–.]", sentence[m.end():hi])[0] window = (before[-90:] + " " + after[:45]).lower() if not any(w in window for w in words): return (f"[[{cid}]] is the {concept.lower().replace('_', ' ')} guidance value; its clause must name it " f"({', '.join(words[:3])})") if basis_words and not any(w in window for w in basis_words): return (f"[[{cid}]] is one of several {concept.lower().replace('_', ' ')} rows with different values; its clause " f"must name its basis ({', '.join(basis_words)})") return None def relation_logic_violation(sentence, cases, variables=None): """An ALTERNATIVE_CONDITIONALS proposition restated as a joint one: a claim from each case in one sentence, or a paraphrase that says the price requires both case variables. Only [[REL_EXPECTATIONS_STORY]] states both cases, and it renders them as alternatives. The check reads the relation's own cases and variables, not the word "and".""" import relations as _rel if not cases: return None toks = {m.group(1).split("__")[0] for m in TOKEN.finditer(sentence)} if "REL_EXPECTATIONS_STORY" in toks: return None hit = [c for c, ids in cases.items() if toks & set(ids)] if len(hit) > 1: return ("the break-even revenue case and margin case are ALTERNATIVE conditional solves (each holds the other " "variable constant); stating both in one sentence reads as a joint requirement — cite " "[[REL_EXPECTATIONS_STORY]] or use separate sentences") # the requirement and both case variables must share one clause ("…high margins and revenue growth, but the stock # already prices in strong economics" describes the business, then states a requirement naming neither) for seg in re.split(r"[;:—–]|,?\s*\b(?:but|while|whereas|although|though|yet)\b", sentence): low = TOKEN.sub(" ", seg).lower() stoks = {m.group(1).split("__")[0] for m in TOKEN.finditer(seg)} if variables and any(p in low for p in _rel.REQUIREMENT_PREDICATE) and any(w in low for w in _rel.REQUIREMENT_SUBJECT) \ and not any(mk in low for mk in _rel.ALTERNATIVE_MARKERS): named = [c for c, words in variables.items() if any(re.search(r"\b" + re.escape(w), low) for w in words) or (stoks & set(cases.get(c) or ()))] if len(named) > 1: return ("a paraphrase of what the price requires names both alternative cases (revenue and margin) as one " "requirement; only [[REL_EXPECTATIONS_STORY]] may state both") return None LINKING = ("is", "are", "was", "were", "be", "been", "being", "sits", "sit", "sitting", "stands", "stand", "trades", "trade", "trading", "remains", "remain", "lies", "runs", "looks", "look", "comes", "come", "lands", "land", "stays", "stay", "still", "now", "already", "roughly", "about", "nearly", "currently", "at") def relation_slot_violation(sentence, rel_subjects, catalog=None): """Typed render slots: CLAUSE / COMPACT follow a linking verb; PREDICATE carries its own verb and follows its subject directly; no mode may be followed by its right-hand side again ("7.8% above consensus than sell-side consensus").""" for m in TOKEN.finditer(sentence): cid = m.group(1) if "__" not in cid: continue base, mode = cid.split("__", 1) if mode == "FULL_SENTENCE": if TOKEN.sub("", sentence).strip(" .!?;:—–-\"'"): return f"[[{cid}]] is a whole sentence; write nothing else in that sentence" continue prev = (re.findall(r"[A-Za-z'’]+", sentence[:m.start()]) or [""])[-1].lower() if mode in ("CLAUSE", "COMPACT") and prev not in LINKING: return f"[[{cid}]] needs a verb before it (\"the price is …\"); use the __PREDICATE form directly after the subject" _lhs, rhs = (rel_subjects or {}).get(base, ((), ())) if mode == "PREDICATE": # it carries its own verb: it must follow the subject noun itself ("the price [[…__PREDICATE]]") tail = sentence[:m.start()].rstrip().lower() def _subject_then_modifier(t): # the subject noun, then at most a short prepositional modifier with no verb ("the engine's forecast for # the next twelve months [[…__PREDICATE]]"); earlier tokens are masked so their ids never read as words t = re.sub(r"\[\[[a-z0-9_]+\]\]", " ⟨tok⟩ ", t) for w in (_lhs or ()): i = t.rfind(w.lower()) if i < 0: continue rest = t[i + len(w):] words = re.findall(r"[a-z'’]+", rest) if not rest.strip() or (len(words) <= 7 and words and (words[0] in PREPOSITIONS or (len(words) > 1 and words[1] in PREPOSITIONS)) and not any(x in LINKING for x in words) and not re.search(r"[,;:]", rest)): return True # coordinated predicates: "the price [[X__PREDICATE]] yet [[Y__PREDICATE]]" share one subject if re.fullmatch(r"\s*⟨tok⟩\s*,?\s*(?:and|but|yet|while)\s*", rest): return True return False # the predicate carries a singular verb ("is 9.0% below …"): a plural subject breaks agreement ("the shares is") if prev in ("shares", "stocks", "they", "multiples", "valuations") and \ re.match(r"\s*(?:is|sits|stands|trades|remains)\b", str(getattr((catalog or {}).get(cid), "rendered", "") or "")): return f"[[{cid}]] carries a singular verb; its subject '{prev}' is plural (write \"the stock\" or \"the price\")" if prev in LINKING or not _subject_then_modifier(tail): return f"[[{cid}]] carries its own verb and must directly follow its subject ({', '.join((_lhs or ('the subject',))[:3])})" after = re.split(r"[,;:.(—–]|\b(?:and|but|while|which|because|so)\b", sentence[m.end():])[0].lower() if rhs and any(w.lower() in after for w in rhs): return f"[[{cid}]] already names its right-hand side; do not repeat it after the token" return None PRICE_TERMS = ("price", "shares", "stock") def valuation_relation_violation(sentence, vrels): """Direction language about a valuation anchor (story value, mechanical DCF, market composite) must state its subject and object and agree with the anchor's typed relation: "the price is above the DCF" and "the DCF sits below the price" both hold; "the mechanical DCF discount" names no side and is rejected.""" import relations as _rel if not vrels: return None toks = {m.group(1).split("__")[0] for m in TOKEN.finditer(sentence)} full = TOKEN.sub(lambda t: " " + t.group(1).lower() + " ", sentence).lower() price_named = any(re.search(r"\b" + p + r"\b", full) for p in PRICE_TERMS) flip = {_rel.ABOVE: _rel.BELOW, _rel.BELOW: _rel.ABOVE} all_anchors = [a for rid in _rel.VALUATION_ANCHORS for a in _rel.VALUATION_ANCHORS[rid]] # a direction word binds within its own clause, to the nearest anchor ("below market pricing while it sits above # story value" states two relations, one per clause) for low in re.split(r"[;:,(—–]|\b(?:while|but|whereas|and|although|though|yet|even as)\b", full): for rid, anchors in _rel.VALUATION_ANCHORS.items(): if rid not in vrels: continue for a in anchors: for am in re.finditer(r"\b" + re.escape(a) + r"\b", low): for rtype, words in _rel.VALUATION_DIRECTION.items(): for w in words: for dm in re.finditer(r"\b" + w + r"\b", low): if abs(dm.start() - am.start()) > 45: continue lo_, hi_ = sorted((dm.end(), am.start())) if dm.start() < am.start() else sorted((am.end(), dm.start())) if any(re.search(r"\b" + re.escape(o) + r"\b", low[lo_:hi_]) for o in all_anchors if o != a): continue # another anchor sits between them: the word is not about this one # "story value premium" next to its relation token: a premium OVER the anchor (a bare # compound names no side and stays unexplicit) compound = w in ("premium", "discount") and 0 <= dm.start() - am.end() <= 3 if compound: explicit, said = rid in toks, rtype elif dm.start() < am.start(): # " … premium/above … ": price is the subject before = low[max(0, dm.start() - 80):dm.start()] # "Shares trade below X but above Y": a clause with no subject of its own shares the # sentence's subject (the price named in the first clause) bare = not re.sub(r"\b(?:is|are|sits?|stands?|trades?|trading|still|now|also|well|" r"only|slightly|modestly|far|much|at|a)\b|[^a-z]", "", before) explicit = any(p in before for p in PRICE_TERMS) or \ (price_named and (re.search(r"\b(?:it|they)\b", before) or bare)) said = rtype if not compound and dm.start() >= am.start() and w in ("cheap", "expensive", "rich", "undervalued", "overvalued"): # an evaluative adjective describes the stock ("the composite argues cheap") explicit, said = (price_named or rid in toks), rtype elif not compound and dm.start() >= am.start(): # " … below … ": anchor is subject between = low[am.end():dm.start()] if any(re.search(r"\b" + p + r"\b", between) for p in PRICE_TERMS): explicit, said = True, rtype # "composite-vs-price discount" else: after_ = low[dm.end():dm.end() + 40] explicit = any(p in after_ for p in PRICE_TERMS) or \ (price_named and re.match(r"\W*(?:the\s+)?(?:it|them)\b", after_)) said = flip[rtype] if not explicit and rid not in toks: return (f"'{w}' about the {a} names no subject and object (whose {w}?); state it as the " f"price relative to the {a}, or cite [[{rid}]]") if said != vrels[rid]: return f"'{w}' contradicts [[{rid}]]: the price is {vrels[rid].lower()} the {a}" return None UNIVERSAL_ANCHORS = r"\b(?:on|by|across|against|versus|vs\.?)\s+(?:every|all|each|any)\s+(?:\w+\s+)?(?:anchors?|measures?|metrics?|" \ r"bases|basis|yardsticks?|methods?|valuation (?:anchors?|measures?|methods?))\b|\bon all (?:counts|fronts)\b" def universal_valuation_violation(sentence, vrels): """"Expensive on every anchor": a direction claimed for ALL valuation anchors must hold for each anchor's relation.""" import relations as _rel low = TOKEN.sub(" ", sentence).lower() if not vrels or not re.search(UNIVERSAL_ANCHORS, low): return None for rtype, words in _rel.VALUATION_DIRECTION.items(): if any(re.search(r"\b" + w + r"\b", low) for w in words): against = [rid for rid, t in vrels.items() if t in (_rel.ABOVE, _rel.BELOW) and t != rtype] if against: return (f"claims the price is {rtype.lower()} value on every anchor, but [[{against[0]}]] holds " f"{vrels[against[0]].lower()}") return None def evidence_domain_violation(sentence, rules): """Evidence from one domain (e.g. MOMENTUM: price vs the 200-day average) used to establish a conclusion in another (a FUNDAMENTALS trend such as revenue deceleration) without evidence from that domain in the sentence.""" import evidence as _ev import relations as _rel toks = {m.group(1) for m in TOKEN.finditer(sentence)} low = TOKEN.sub(" ", sentence).lower() concl = [] for stems in _rel.TREND_SURFACE.values(): for st in stems: for cm in re.finditer(r"\b" + re.escape(st), low): concl.append((st, _ev.conclusion_domain(low[max(0, cm.start() - 80):cm.end() + 60]))) if not concl: return None if any(t.startswith(_ev.FUNDAMENTAL_EVIDENCE_PREFIXES) or t in (rules.trends_allowed or {}) for t in toks): return None # fundamentals / guidance evidence (or a trend relation) is present for dom, d in _ev.EVIDENCE_DOMAINS.items(): if not (toks & set(d["claims"]) or any(w in low for w in d["subjects"])): continue for word, cdom in concl: if cdom not in d["allowed_target_domains"]: return (f"{dom.lower()} evidence cannot establish a {cdom.lower().replace('_', ' ')} conclusion ('{word}'); " f"it may support only {', '.join(x.lower() for x in d['allowed_target_domains'])}") return None TYPED_SEGMENT_COUNT = re.compile(r"\b(\d+|two|three|four|five|six|seven|eight|nine|ten)\s+(?:reportable\s+|reported\s+|" r"operating\s+|core\s+|business\s+)?(?:segments|divisions|reporting units)\b", re.I) def segment_perimeter_violation(sentence, sp): """A segment count typed as text, or the filed (pre-separation) count or coverage cited without saying it describes the business as filed: a continuing-company sentence may not borrow the historical count.""" low = TOKEN.sub(" ", sentence).lower() if TYPED_SEGMENT_COUNT.search(low): return "a segment count must be cited as [[SEGMENT_COUNT_FILED]] or [[SEGMENT_COUNT_CONTINUING]], never typed" if not sp: return None toks = {m.group(1) for m in TOKEN.finditer(sentence)} for cid in toks & set(sp.get("historical_claims") or ()): if not any(q in low for q in sp.get("qualifiers") or ()): return (f"[[{cid}]] describes the business as filed, before the separation; say so (e.g. 'before the separation', " "'as filed') or use the continuing-company count") return None def evidence_conclusion_violation(sentence, uses): """(validator, reason): evidence used for a conclusion type its EvidenceUse does not allow (beta → an entry point, insider purchases → market pressure, a third-party rating headline → the issuer's moat). A conclusion stated only as a denial ("not a catalyst") is descriptive.""" import evidence as _ev if not uses: return None toks = {m.group(1).split("__")[0] for m in TOKEN.finditer(sentence)} low = TOKEN.sub(" ", sentence).lower() present = [k for k, u in uses.items() if toks & set(u.get("evidence_claim_ids") or ()) or any(w in low for w in u.get("subjects") or ())] if not present: return None other = any(t.startswith(_ev.FUNDAMENTAL_EVIDENCE_PREFIXES) or t.startswith("REL_") or t in ("PRICE", "INTRINSIC_VALUE", "MARKET_COMPOSITE", "ANALYST_TARGET", "MECHANICAL_VALUE") for t in toks) for ctype, surface in _ev.CONCLUSION_TYPES.items(): for w in surface: for m in re.finditer(re.escape(w), low): if DENIAL.search(low[max(0, m.start() - 28):m.start()]): continue if other and ctype in (_ev.MOAT_OR_SECULAR_GROWTH, _ev.STOCK_UPSIDE): continue # fundamentals / valuation evidence in the sentence may support it if any(ctype in (uses[k].get("allowed_conclusion_types") or ()) for k in present): continue vid = "NEWS_EVIDENCE_SCOPE" if present == ["NEWS"] or ("NEWS" in present and len(present) == 1) else \ "EVIDENCE_CONCLUSION_SCOPE" return vid, (f"{', '.join(p.lower() for p in present)} evidence may support only " f"{', '.join(sorted({c.lower() for k in present for c in uses[k]['allowed_conclusion_types']}))}; " f"'{w.strip()}' asserts a {ctype.lower().replace('_', ' ')} conclusion") return None def performance_violation(sentence, returns, issuer_words=()): """A statement about how the stock performed ("a weak year for the stock") without the signed return claim it describes, or against its sign.""" import relations as _rel found = _rel.performance_asserted(TOKEN.sub(" ", sentence), issuer_words) if not found: return None cited = {m.group(1): (returns or {}).get(m.group(1)) for m in TOKEN.finditer(sentence) if m.group(1) in (returns or {})} for sign, word, year in found: ok = [c for c, v in cited.items() if v is not None and ((v < 0) == (sign == _rel.NEGATIVE_PERF)) and (not year or _rel.STOCK_RETURN_CLAIMS.get(c) == "12M")] if not ok: want = "[[RETURN_12M_PHRASE]]" if year else "[[RETURN_12M_PHRASE]] or [[RETURN_3M_PHRASE]]" if cited: return f"'{word}' describes a {sign.lower()} stock move, but the cited return has the other sign or period" return f"'{word}' describes how the stock performed without the return it describes ({want})" return None def assessment_violation(sentence, assessments, assumption_claims): """'Conservative' / 'aggressive' about a story assumption without, or against, its typed AssumptionAssessment; or about the assumptions collectively, which have no assessment.""" import relations as _rel low_full = sentence.lower() # "high" / "low" about an assumption, in its clause or the relative clause right after it ("…of [[X]], which is high") for m in TOKEN.finditer(sentence): cid = m.group(1) if cid not in (assumption_claims or ()): continue before = re.split(r"[,;:—–(]|\b(?:and|but|while|whereas|plus)\b", sentence[:m.start()])[-1].lower() rel = re.match(r"\s*(?:,?\s*(?:which|that)\s+)?(?:is|looks|seems|appears|remains)\s+([^,;.]{0,40})", sentence[m.end():]) scope = before[-40:] + " " + (rel.group(1).lower() if rel else "") for level, words in _rel.LEVEL_WORDS.items(): w = next((x for x in words if re.search(r"(?= 3] if not quotes: return ("attributes a statement to management without quoting its words from the call; quote the speaker " "(at least three words) or describe the company's filings instead") if not any((_ev.quote_speaker(turns, q) or {}).get("role") == _ev.MANAGEMENT for q in quotes): return "the quoted words are not management's words in the call transcript" return None STATEMENT_NOUNS = {"balance sheet": "BALANCE_SHEET", "income statement": "INCOME_STATEMENT", "p&l": "INCOME_STATEMENT", "profit and loss": "INCOME_STATEMENT", "cash flow statement": "CASH_FLOW", "cash-flow statement": "CASH_FLOW", "statement of cash flows": "CASH_FLOW"} def statement_domain_violation(sentence, statements): """A figure placed on the wrong financial statement ("the balance sheet carries amortization of $785M"): its own clause may name only the statement its concept belongs to.""" for clause in re.split(r"[,;:—–(.]", sentence): low = clause.lower() named = {st for noun, st in STATEMENT_NOUNS.items() if noun in low} if not named: continue for t in TOKEN.finditer(clause): own = (statements or {}).get(t.group(1)) if own and own not in named: return (f"[[{t.group(1)}]] is an {own.lower().replace('_', ' ')} item; its clause places it on the " f"{' / '.join(sorted(n.lower().replace('_', ' ') for n in named))}") return None def family_comparison_violation(sentence): """A comparison between two metric families ("earnings compound faster than revenue") with no typed relation.""" import relations as _rel for a, b, w in _rel.family_comparison_asserted(TOKEN.sub(" ", sentence)): return (f"compares {a.lower().replace('_', ' ')} with {b.lower().replace('_', ' ')} ('{w}') but no typed relation " "connects the two (different metrics, and no compatible growth pair on one perimeter)") return None def proximity_violation(sentence, magnitudes): """'Close to fair value' / 'fairly valued' without a cited valuation anchor inside the closeness band.""" import relations as _rel low = TOKEN.sub(" ", sentence).lower() m = re.search(_rel.PROXIMITY_SURFACE, low) if not m or re.search(r"\b(?:not|far from|no longer|hardly)\W+(?:\w+\W+){0,2}$", low[:m.start()]): return None cited = {t.group(1).split("__")[0] for t in TOKEN.finditer(sentence)} & set(magnitudes or {}) if not cited: return f"'{m.group(0)}' is a valuation judgement without a valuation anchor's relationship token" far = [c for c in cited if magnitudes[c] is not None and abs(magnitudes[c]) > _rel.CLOSENESS_BAND] if far: return (f"'{m.group(0)}' but the cited anchor gap is {format_percent(abs(magnitudes[far[0]]), 1)}, outside the " f"{format_percent(_rel.CLOSENESS_BAND, 0)} closeness band") return None RETURN_SUBJECT = r"\b(?:expected return|probability[- ]weighted(?: return)?|weighted return|scenario[- ]weighted|overall return|" \ r"(?:my|our|the panel'?s?|the pm'?s?) (?:expected )?return|expected 12[- ]month return|net expectation)\b" RETURN_DIRECTION = {"NEGATIVE": r"\b(?:bearish|negative|downside|a loss|below zero|lower)\b", "POSITIVE": r"\b(?:bullish|positive|upside|a gain|above zero)\b"} def return_direction_violation(sentence, own_value): """A direction written about the author's own expected return that its signed value contradicts ("expected return … tilted slightly bearish" for +6.9%).""" if own_value is None: return None for seg in re.split(r"[;:—–]|\b(?:but|while|whereas|because|since|although)\b", TOKEN.sub(" ", sentence).lower()): if not re.search(RETURN_SUBJECT, seg): continue for sign, rx in RETURN_DIRECTION.items(): m = re.search(rx, seg) if m and not re.search(r"\b(?:not|no|never)\W+(?:\w+\W+){0,2}$", seg[:m.start()]) \ and ((sign == "NEGATIVE") != (own_value < 0)) and abs(own_value) > 0.0005: return (f"'{m.group(0)}' describes the expected return, whose value is " f"{format_percent(own_value, 1, sign=True)}") return None def phrase_slot_violation(sentence, catalog): """Typed render slots for claims that are not noun phrases: a PREDICATE phrase ("down 11.8% over twelve months") follows a linking verb or its subject ("shares are …", "with shares …"), never a bare preposition ("With down …"); a LIST rendering (items joined by semicolons) ends its sentence.""" for m in TOKEN.finditer(sentence): cl = catalog.get(m.group(1)) slot = getattr(C_BY_ID.get(m.group(1)), "slot", "") if C_BY_ID else "" before = sentence[:m.start()].rstrip() if slot == "PREDICATE" and before and before[-1] not in ",;:(—–": prev = (re.findall(r"[A-Za-z'’]+", before) or [""])[-1].lower() if prev not in LINKING + PREDICATE_SUBJECTS: return (f"[[{m.group(1)}]] is a predicate ('{cl.rendered}'); it follows a verb or its subject " f"(\"shares are …\"), not '{prev}'") r = str(getattr(cl, "rendered", "") or "") if getattr(cl, "stype", None) == "TEXT" and ";" in r and sentence[m.end():].strip(" .!?\"')") != "": return f"[[{m.group(1)}]] is a list ('{r[:40]}…'); it must end its sentence" return None PREDICATE_SUBJECTS = ("shares", "stock", "price", "equity", "it", "they", "which", "been", "stay", "stays") # the concept a money figure measures, from its claim path: its own clause must name it ("requires $33.00B" names no # concept; "revenue of $33.00B" does) MONEY_CONCEPT_NAMES = (("revenue", ("revenue", "sales", "top line", "top-line")), ("tam", ("market", "addressable", "tam")), ("operating_income", ("operating income", "operating profit", "ebit")), ("net_income", ("net income", "earnings", "profit")), ("market_cap", ("market cap", "capitali", "market value")), ("insiders", ("insider", "purchase", "bought", "sale", "sold", "buying", "selling"))) MULTIPLE_NAMES = (("pe_", ("p/e", "pe ", "price-to-earnings", "earnings multiple", "multiple", "earnings")), ("peg", ("peg",)), ("ev_ebitda", ("ev/ebitda", "ebitda")), ("p_fcf", ("p/fcf", "free cash flow"))) def claim_label_violation(sentence, catalog): """A money figure (CURRENCY_BIG) or a multiple whose clause does not name what it measures.""" toks = list(TOKEN.finditer(sentence)) for i, m in enumerate(toks): cl = catalog.get(m.group(1)) st_ = getattr(cl, "stype", None) if st_ not in ("CURRENCY_BIG", "MULTIPLE"): continue path = str(getattr(cl, "path", "")).lower().split(".")[-1] + "_" table = MONEY_CONCEPT_NAMES if st_ == "CURRENCY_BIG" else MULTIPLE_NAMES names = next((ws for key, ws in table if key in (str(getattr(cl, "path", "")).lower() if st_ == "CURRENCY_BIG" else path)), None) if not names: continue lo = toks[i - 1].end() if i > 0 else 0 hi = toks[i + 1].start() if i + 1 < len(toks) else len(sentence) before = re.split(r"[;:(—–]", sentence[lo:m.start()])[-1] after = re.split(r"[,;:)—–.]", sentence[m.end():hi])[0] window = (before[-90:] + " " + after[:45]).lower() if not any(w in window for w in names): return f"[[{m.group(1)}]] ({cl.rendered}) is a {names[0].strip()} figure; its clause must name it ({', '.join(n.strip() for n in names[:3])})" return None def named_counterpart_violation(sentence, rules): """A cited claim compared with a quantity the sentence names but does not cite ("guides growth of [[GUIDANCE_4]], far below the break-even revenue growth the price demands"): the named quantity is that claim, and the pair must be compatible and connected by its relation.""" if rules.pair_fn is None or not C_BY_ID: return None low = _dash(TOKEN.sub(" ", sentence).lower().replace("’", "'")) if not (COMPARATIVE.search(low) or ORDER_UP.search(low) or ORDER_DOWN.search(low)): return None cited = [m.group(1) for m in TOKEN.finditer(sentence)] for y, d in C_BY_ID.items(): if not d.names or y in cited or y not in rules.catalog or not any(n in low for n in d.names): continue for x in cited: st = rules.pair_fn(x, y) if st is None: continue if st[0] == "OK": # a compatible pair still has a direction: the comparative must match the values up, dn = ORDER_UP.search(low) or re.search(r"\bhigher\b", low), ORDER_DOWN.search(low) or re.search(r"\blower\b", low) rng, yv = (rules.range_fn(x) if rules.range_fn else None), getattr(rules.catalog.get(y), "value", None) xv = getattr(rules.catalog.get(x), "value", None) lo_, hi_ = ((rng[0], rng[1]) if rng else (xv, xv)) if (rng or isinstance(xv, (int, float))) else (None, None) if isinstance(yv, (int, float)) and lo_ is not None and bool(up) != bool(dn): x_first = sentence.find(f"[[{x}]]") < min((low.find(n) for n in d.names if n in low), default=10 ** 6) says_x_above = bool(up) == x_first if (says_x_above and hi_ <= yv) or (not says_x_above and lo_ >= yv): return (f"states [[{x}]] is {'above' if says_x_above else 'below'} the {d.names[0]} ([[{y}]]), " "but the values say otherwise") continue if st[0] == "RELATION" and st[1] in cited: continue why = f"their relation is [[{st[1]}]]" if st[0] == "RELATION" else "; ".join(st[1])[:100] return f"compares [[{x}]] with the {d.names[0]} ([[{y}]]) without citing it ({why})" return None # ── verbal magnitudes ("high-teens EPS growth") are numbers: they must fit a claim of that metric ────────────── MAGNITUDE_BANDS = (("low-single-digit", 0.0, 0.04), ("mid-single-digit", 0.03, 0.07), ("high-single-digit", 0.06, 0.10), ("single-digit", 0.0, 0.10), ("low-double-digit", 0.09, 0.14), ("low-teens", 0.09, 0.14), ("mid-teens", 0.13, 0.17), ("high-teens", 0.16, 0.20), ("double-digit", 0.095, 1.0), ("low-twenties", 0.19, 0.24), ("mid-twenties", 0.23, 0.27), ("high-twenties", 0.26, 0.30)) MAGNITUDE_FAMILIES = {"RETURN": (("total return", "return", "upside", "downside"), ("SCENARIO_BEAR_RETURN", "SCENARIO_NEUTRAL_RETURN", "SCENARIO_BULL_RETURN", "PM_RETURN", "HEADLINE_RETURN", "TARGET_VS_PRICE", "RETURN_12M", "RETURN_3M"), ()), "EPS_GROWTH": (("eps", "earnings", "profits", "profit"), ("FORECAST_EPS_GROWTH",), ("EPS_GROWTH",)), "REVENUE_GROWTH": (("revenue", "sales", "top-line", "top line", "growth"), ("LIFECYCLE_FORWARD_GROWTH", "LIFECYCLE_TRAILING_GROWTH", "FORECAST_REVENUE_GROWTH"), ("REVENUE_GROWTH", "ORGANIC_REVENUE_GROWTH"))} def magnitude_violation(sentence, catalog): """A verbal magnitude band about a metric that no claim of that metric supports.""" low = _dash(TOKEN.sub(" ", sentence).lower()).replace(" to ", "-to-") for band, lo, hi in MAGNITUDE_BANDS: for m in re.finditer(r"(?:mid|low|high)?-?(?:to-)?" + re.escape(band) + r"|" + re.escape(band.replace("-", " ")), low): after = low[m.end():m.end() + 40] # the band binds to the nearest metric word after it ("mid-single-digit growth and elite profitability") pos = [(mm.start(), f) for f, (words, _c, _g) in MAGNITUDE_FAMILIES.items() for w in words for mm in re.finditer(r"\b" + re.escape(w) + r"\b", after)] if not pos: continue fam = min(pos)[1] words, cids, concepts = MAGNITUDE_FAMILIES[fam] vals = [abs(float(catalog[c].value)) if fam == "RETURN" else float(catalog[c].value) for c in cids if c in catalog and isinstance(catalog[c].value, (int, float))] for cid, c in catalog.items(): v = c.value if cid.startswith("GUIDANCE_") and isinstance(c.value, dict) else None if v and v.get("concept") in concepts: vals += [float(x) for x in (v.get("low"), v.get("high")) if isinstance(x, (int, float))] # a range like "mid-to-high single digit" widens to both bands wide_lo = lo - 0.01 - (0.03 if "to-" in m.group(0) else 0.0) if not any(wide_lo <= x <= hi + 0.01 for x in vals): shown = ", ".join(format_percent(x, 1, sign=True) for x in vals[:3]) or "none" return (f"'{m.group(0).replace('-to-', ' to ')}' fits no {fam.lower().replace('_', ' ')} claim " f"(available: {shown})") return None # spelled-out counts of the things COUNT claims count ("two adds and four new positions") are numbers without lineage SPELLED = r"\b(one|two|three|four|five|six|seven|eight|nine|ten|eleven|twelve|thirteen|fourteen|fifteen|sixteen|seventeen|" \ r"eighteen|nineteen|twenty|thirty|forty|fifty|hundred|thousand)\b" COUNTED = ("investor", "institution", "holder", "position", "filer", "add", "addition", "exit", "trim", "purchase", "exercise", "share", "analyst", "contributor", "opened", "added", "trimmed", "exited", "bought", "sold") # periods ("eight quarters", "ten years") are windows, not counts; segment counts are owned by SEGMENT_PERIMETER_LANGUAGE def spelled_count_violation(sentence): low = TOKEN.sub(" ", sentence).lower() for m in re.finditer(SPELLED, low): nxt = re.findall(r"[a-z]+", low[m.end():m.end() + 30])[:3] if m.group(1) == "one" and (not nxt or nxt[0] in ("of", "or", "another", "side", "way", "hand", "day", "time")): continue if any(w.startswith(c) for w in nxt for c in COUNTED): return f"'{m.group(0)} {' '.join(nxt[:2])}' spells out a count; cite its claim token instead" return None # a breakdown whose parts only make sense together: citing some parts and omitting the rest misstates the whole COMPLETE_SETS = (("OWN_13F_NEW", "OWN_13F_ADDS", "OWN_13F_TRIMS", "OWN_13F_EXITS"),) def complete_set_violation(sentence, catalog): toks = {m.group(1) for m in TOKEN.finditer(sentence)} for group in COMPLETE_SETS: present = [c for c in group if c in catalog] cited = [c for c in present if c in toks] if cited and len(cited) < len(present): missing = [c for c in present if c not in toks] return (f"cites part of a breakdown ({', '.join(cited)}) without the rest ({', '.join(missing)}); state all " "of it or none") return None # a price move "to " whose direction the two values contradict ("compression to $443.00" above $403.30) _MOVE_DOWN = r"\b(?:compress\w*|compression|fall\w*|fell|declin\w*|drop\w*|de-?rat\w*|slid\w*|slump\w*|contract\w*)\b" _MOVE_UP = r"\b(?:ris\w*|rose|rall\w*|re-?rat\w*|climb\w*|expan\w*|recover\w*|gain\w*)\b" def price_move_violation(sentence, catalog): price = getattr(catalog.get("PRICE"), "value", None) if not isinstance(price, (int, float)): return None for m in TOKEN.finditer(sentence): c = catalog.get(m.group(1)) if getattr(c, "stype", None) != "CURRENCY" or m.group(1) == "PRICE" or not isinstance(c.value, (int, float)): continue before = sentence[max(0, m.start() - 45):m.start()].lower() if not re.search(r"\bto\s*$", before): continue if re.search(_MOVE_DOWN, before) and c.value > price: return f"a decline 'to [[{m.group(1)}]]' ({c.rendered}) would be a rise: it is above the price ({catalog['PRICE'].rendered})" if re.search(_MOVE_UP, before) and c.value < price: return f"a rise 'to [[{m.group(1)}]]' ({c.rendered}) would be a fall: it is below the price ({catalog['PRICE'].rendered})" return None # an internal identifier (UPPER_SNAKE claim ids, lower_snake state fields): one class for authored objects (arguments.py # T2), legacy prose ingestion (check_sentence) and the report audit, so an object that would fail the audit is dropped # (and can be repaired) at authoring instead of withholding the whole report IDENTIFIER = re.compile(r"(?= 0: # "mature‑growth" (U+2011) is "mature-growth" return f"CLAIM_QUALIFIER_DOUBLED: [[{m.group(1)}]] already renders '{r}'; do not repeat it around the token" if getattr(cl, "stype", None) in ("PHRASE", "REL") and r: words = [w for w in re.findall(r"[a-z]+", r.lower()) if len(w) > 2] around = TOKEN.sub(" ", sentence[max(0, m.start() - 60):m.start()] + " " + sentence[m.end():m.end() + 60]).lower() tri = {" ".join(words[i:i + 3]) for i in range(len(words) - 2)} # the same tokenisation on both sides ("today's" is "today" on both) if any(t in " ".join(w for w in re.findall(r"[a-z]+", around) if len(w) > 2) for t in tri): return f"CLAIM_QUALIFIER_DOUBLED: [[{m.group(1)}]] already renders that descriptor; do not repeat it" if getattr(cl, "stype", None) == "PHRASE" and cl.rendered: before = sentence[max(0, m.start() - 14):m.start()].lower() if re.search(r"\b(?:a|an)\s*$", before) and re.match(r"(?:a|an)\s", cl.rendered): return f"CLAIM_QUALIFIER_DOUBLED: [[{m.group(1)}]] already carries its article ({cl.rendered})" if re.search(r"\b(?:" + "|".join(PHRASE_DIRECTION_WORDS) + r")\b[^.;:]*$", before): return f"CLAIM_QUALIFIER_DOUBLED: [[{m.group(1)}]] already carries its direction ({cl.rendered})" continue op = (getattr(cl, "value", None) or {}).get("operator") if getattr(cl, "stype", None) == "GUIDANCE" else None if op in _g._BOUND_WORDS and re.search(r"(?:" + "|".join(_g._BOUND_WORDS.values()) + r")\W*$", sentence[max(0, m.start() - 24):m.start()], re.I): return f"CLAIM_QUALIFIER_DOUBLED: [[{m.group(1)}]] already states its bound ({cl.rendered}); do not write it again" return None def boundary_repeat(sentence, catalog): """The word before a token repeats the first word its rendering starts with ("Today, [[REL_EXPECTATIONS_STORY]]" → "Today, today's price requires …").""" for m in TOKEN.finditer(sentence): r = str(getattr(catalog.get(m.group(1)), "rendered", "") or "") first = (re.findall(r"[a-z]+", r.lower()) or [""])[0] prev = (re.findall(r"[a-z]+", sentence[:m.start()].lower()) or [""])[-1] if first and len(first) > 3 and prev == first: return f"CLAIM_QUALIFIER_DOUBLED: [[{m.group(1)}]] already starts with '{first}'; do not write it before the token" return None LIFECYCLE_STAGES = ("start-up", "young-growth", "high-growth", "mature-growth", "mature-stable", "decline") _STAGE_NOUNS = r"(?:stage|phase|company|franchise|business|compounder|firm|name|profile|story)" def lifecycle_violation(sentence, catalog, company_words=()): """A life-cycle stage term for the company other than its typed stage ("durable high growth" for a mature-growth company). The stage words are the LIFECYCLE_STAGE claim's own vocabulary; a stage term scoped to a segment ("high growth in private markets") is not about the company.""" stage = str(getattr(catalog.get("LIFECYCLE_STAGE"), "rendered", "") or "").lower() if stage not in LIFECYCLE_STAGES: return None low = _dash(TOKEN.sub(lambda t: " " + t.group(1).lower() + " ", sentence).lower()) for other in LIFECYCLE_STAGES: if other == stage: continue spaced = other.replace("-", "[ -]") for m in re.finditer(r"\b" + spaced + r"\b", low): after, before = low[m.end():m.end() + 30], low[:m.start()] if re.match(r"\s+(?:in|from|across|at|within|for)\b", after) or re.search(r"\b(?:not|no longer|unlike|past|from)\W+(?:\w+\W+){0,1}$", before): continue noun_after = re.match(r"\s*" + _STAGE_NOUNS, after) if other == "decline" and not noun_after: continue # "a revenue decline" is not the decline stage about_company = noun_after or any(w and w in before for w in company_words) or "lifecycle_stage" in low if about_company: return f"'{m.group(0)}' names a {other} company; its typed life-cycle stage is {stage}" return None def directional_violation(sentence, restricted): """The restricted evidence family a sentence uses to carry a direction, or None.""" low = sentence.lower() if not DIRECTIONAL.search(sentence) or DENIAL.search(low) and re.search(r"not (?:a )?(?:return )?signal|no (?:return )?signal", low): return None for fam, rx in (("13F", EVIDENCE_13F), ("insider", EVIDENCE_INSIDER)): if fam in (restricted or ()) and rx.search(sentence): return fam return None def relative_violation(sentence, allowed): """A rich/cheap/premium/discount-versus-peers statement without an allowed relative-valuation claim.""" if not (RELVAL.search(sentence) and PEER_WORDS.search(sentence)): return None named = [k for k, rx in _REL_METRIC.items() if re.search(rx, sentence, re.I)] if named: bad = [k for k in named if k not in (allowed or ())] return bad[0] if bad else None return None if allowed else "any" DENIAL = re.compile(r"\b(?:not|no|never|neither|nor|without|isn't|is not|are not|does not|do not apply|inapplicable|excluded|disabled)\b", re.I) try: from catalog import BY_ID as C_BY_ID # render slots are declared on the claim definitions except Exception: # noqa: BLE001 C_BY_ID = {} @dataclass class Fragment: text: str claim_id: str | None = None value: object = None origin: str = "system" # "llm" for fragments ingested from an author; templates are "system" def to_dict(self): return {"text": self.text, "claim_id": self.claim_id, "value": self.value} @dataclass class Rejection: source: str sentence: str reason: str def to_dict(self): return {"source": self.source, "sentence": self.sentence, "reason": self.reason} @dataclass class Rules: catalog: dict engine_forbidden: tuple = () quant_evidence_allowed: bool = False # the single-stock price model's own permission panel_evidence_allowed: bool = False # the cross-sectional rank's own permission (never OR'd) panel_forbidden_phrases: tuple = () quant_forbidden_phrases: tuple = () plan_sales_present: bool = False # governance framing of scheduled sales is rejected guidance_available: bool = False # informational; the rule itself keys on GUIDANCE_n tokens compare_fn: object = None # (claim_a, claim_b) -> (ok, reasons) | None (catalog.comparison) counterparts: dict = field(default_factory=dict) # claim -> incomparable counterparts (catalog.implicit_counterparts) descriptors: dict = field(default_factory=dict) # claim -> (family, words describing it) (catalog.claim_descriptors) range_fn: object = None # claim -> (low, high, kind) | None (catalog.claim_range) restricted_evidence: tuple = () # evidence families that may not carry a direction: "13F", "insider" relative_allowed: tuple = () # peer metrics with an allowed RelativeValuationClaim offered: object = None # claim ids an author may cite (catalog.offered); None = unrestricted pair_fn: object = None # (a, b) -> None | ("RELATION", rel_id) | ("OK",) | ("INCOMPATIBLE", why) magnitude_bindings: dict = field(default_factory=dict) # magnitude claim -> (lhs claim, rhs claim) it measures claim_model: object = None # claim id -> model id (evidence.claim_model) evidence_targets: object = None # claim id -> (source model, [allowed target models]) | None guidance_terms: dict = field(default_factory=dict) # concept -> (epistemic state, (names a reader uses)) expectations_polarity: str | None = None # REL_EXPECTATIONS_STORY relation_type, or None when none exists presentation_claims: tuple = () # report-only relation magnitudes (catalog.RELATION_PRESENTATION) phrase_of: dict = field(default_factory=dict) # signed claim -> the phrase claim that carries its direction full_clause: object = None # claim id -> True when it renders a complete clause (catalog.is_full_clause) rel_subjects: dict = field(default_factory=dict) # REL claim id -> (lhs subject words, rhs subject words) trends_allowed: dict = field(default_factory=dict) # TREND claim id -> trend class, for approved TrendRelationships only change_claims: tuple = () # typed CHANGE / MARGIN_CHANGE claims that carry a level change themselves interpretation: dict = field(default_factory=dict) # evidence family -> {permission, claims, subjects} (evidence.interpretation) guidance_names: dict = field(default_factory=dict) # GUIDANCE_n -> (concept, names a reader uses for that concept) alternative_cases: dict = field(default_factory=dict) # case -> claim ids (ImpliedExpectations ALTERNATIVE_CONDITIONALS) case_variables: dict = field(default_factory=dict) # case -> the variable names a paraphrase of that case uses segment_perimeter: dict = field(default_factory=dict) # {"historical_claims", "qualifiers"} when a separation exists evidence_uses: dict = field(default_factory=dict) # EvidenceUse per evidence family (evidence.evidence_uses) internal_terms: tuple = () # internal QA/state vocabulary that may not reach a reader directional_context: bool = False # the sentence is a reason FOR or AGAINST the view ("Why" / risks) valuation_relations: dict = field(default_factory=dict) # REL id -> relation type, for the valuation anchors valuation_magnitudes: dict = field(default_factory=dict) # REL id -> signed gap, for the valuation anchors stock_returns: dict = field(default_factory=dict) # stock-return claim -> signed value (StockPerformance) assessments: dict = field(default_factory=dict) # story assumption -> typed AssumptionAssessment assumption_claims: tuple = () # the story-assumption claim ids transcript_turns: list | None = None # [{speaker, role, text}] of the call excerpt (TranscriptEvidence) company_words: tuple = () # the issuer's own names (a quote attributed to the company) statements: dict = field(default_factory=dict) # claim id -> financial statement it belongs to officers: dict = field(default_factory=dict) # role → canonical name own_return_claim: str | None = None # the claim [[OWN_RETURN]] resolves to own_label: str | None = None # this author's rating label (one rating table) held_labels: tuple = () # every rating label the analysis actually holds def _split(text): t = re.sub(r"\s+", " ", str(text or "")).strip() return [x for x in SENT.split(t) if x.strip()] if t else [] def _residual(sentence): """The sentence with tokens and benign numerals removed — what must contain no material number.""" return BENIGN.sub(" ", TOKEN.sub(" ", sentence)) LABEL_WINDOW = 48 # a label never reaches across a sentence end, a clause boundary, a comma, or a comparison word: in # "[[A]] trailing vs [[B]]" or "from [[A]] to [[B]]" the words before the conjunction belong to A CLAUSE = re.compile(r"[:;—–(),]|[.!?](?:\s|$)|\b(?:vs\.?|versus|against|compared with|from|to|and)\b", re.I) RATING_WORD = re.compile(r"\b((?i:strong[ -]buy|strong[ -]sell)|BUY|SELL|HOLD)\b|\b((?:strong[ -])?(?:buy|sell|hold))[ -](?:view|call|rating|recommendation|stance)\b") FIRST_PERSON = re.compile(r"\b(?:I|my|we|our|me)\b") GAP_THEN_DIRECTION = re.compile(r"\[\[([A-Z0-9_]+)\]\]\s+(?:above|below|over|under)\b", re.I) def label_violations(fragments, prohibited_for): """[(claim_id, label)] for every claim fragment preceded by a label that claim forbids. ONE rule, used by prose ingestion and by the final report audit alike: the words that label a claim are the text between the previous claim (or the start of the phrase) and this claim, at most LABEL_WINDOW characters. A label that belongs to an earlier claim is never attributed to a later one, and the rule reads rendered text, so it gives the same answer at both points.""" out, ctx = [], "" for f in fragments: if not f.claim_id: ctx += f.text continue before = CLAUSE.split(ctx[-LABEL_WINDOW:])[-1].lower() for bad in (prohibited_for(f.claim_id) or ()): if bad.lower() in before and not DENIAL.search(before[-24:]): out.append((f.claim_id, bad)) ctx = "" return out def check_sentence(sentence, rules: Rules): """(ok, reason). One evaluation; the sentence is never modified here.""" for m in TOKEN.finditer(sentence): cid = m.group(1) if cid == "OWN_RETURN": if not rules.own_return_claim or rules.own_return_claim not in rules.catalog: return False, "own-return token used where no own return exists" cid = rules.own_return_claim if rules.catalog.get(cid) is None: return False, f"unknown or unavailable claim token [[{m.group(1)}]]" rv1 = relation_violation(sentence, Rules(catalog=rules.catalog, magnitude_bindings=rules.magnitude_bindings, pair_fn=rules.pair_fn)) if rules.magnitude_bindings else None if rv1 and rv1[0] == "RELATION_MAGNITUDE": return False, f"{rv1[0]}: {rv1[1]}" ov = offered_violation(sentence, rules) if ov: return False, f"{ov[0]}: {ov[1]}" for cid, bad in label_violations(to_fragments(sentence, rules), lambda c: rules.catalog[c].prohibited): return False, f"[[{cid}]] labelled '{bad}', a label that claim forbids" dq = qualifier_doubled(sentence, rules.catalog, rules.own_return_claim) or operand_doubled(sentence, rules.catalog) \ or boundary_repeat(sentence, rules.catalog) if dq: return False, dq for m in RATING_WORD.finditer(sentence): lab = re.sub(r"[ -]+", " ", (m.group(1) or m.group(2) or "")).upper() if rules.held_labels and lab not in rules.held_labels: return False, f"rating word '{m.group(0)}' is not a rating this analysis holds" if rules.own_label and FIRST_PERSON.search(sentence) and lab != rules.own_label: return False, f"states its own view as '{lab}' but its rating is {rules.own_label}" if re.search(r"(?:…|\.\.\.)\W*$", sentence.strip()) or not re.search(r"[.!?][\"'”’)\]]*$", sentence.strip()) \ and len(sentence.split()) > 12: return False, "INCOMPLETE_SENTENCE: the sentence is cut off; only complete sentences are published" mgv = magnitude_violation(sentence, rules.catalog) if mgv: return False, f"VERBAL_MAGNITUDE: {mgv}" scv = spelled_count_violation(sentence) if scv: return False, f"FINAL_MATERIAL_CLAIM_LINEAGE: {scv}" lc = lifecycle_violation(sentence, rules.catalog, rules.company_words) if lc: return False, f"CLAIM_QUALIFIER_DOUBLED: {lc}" cs = complete_set_violation(sentence, rules.catalog) if cs: return False, f"EVIDENCE_INTERPRETATION: {cs}" lb = claim_label_violation(sentence, rules.catalog) if lb: return False, f"CLAIM_LABEL_BINDING: {lb}" for m in TOKEN.finditer(sentence): c1 = rules.catalog.get(m.group(1)) nxt = (re.findall(r"[A-Za-z]+", sentence[m.end():m.end() + 25]) or [""])[0].lower() if getattr(c1, "stype", None) == "COUNT" and getattr(c1, "value", None) is not None and round(float(c1.value)) == 1 \ and re.match(r"[a-z]{3,}[^su]s$", nxt) and nxt not in ("was", "has", "is", "this", "its"): return False, f"[[{m.group(1)}]] is 1; '{nxt}' is plural" res = _residual(sentence) mm = MATERIAL_NUMBER.search(res) if mm: return False, f"material number typed as text ('{mm.group(0).strip()}') instead of a claim token" low = sentence.lower() for t in rules.engine_forbidden or (): if re.search(r"\b" + re.escape(t.lower()) + r"\b", low) and not DENIAL.search(low): return False, f"valuation term '{t}' is not used by this issuer's valuation engine" panel_words = PANEL_TOPIC.search(sentence) price_words = PRICE_MODEL_TOPIC.search(PANEL_TOPIC.sub(" ", sentence)) if panel_words and not rules.panel_evidence_allowed and not DENIAL.search(low): return False, "MODEL_EVIDENCE_VISIBILITY: cites the cross-sectional rank, which is not usable as evidence for this analysis" if price_words and not rules.quant_evidence_allowed and not DENIAL.search(low): return False, "MODEL_EVIDENCE_VISIBILITY: cites the price model, which is not usable as evidence for this analysis" for words, phrases in ((price_words, rules.quant_forbidden_phrases), (panel_words, rules.panel_forbidden_phrases)): for p in phrases or (): if words and re.search(r"\b" + re.escape(p.lower()) + r"\b", low): return False, f"describes the quantitative model as '{p}', stronger than its reliability allows" if rules.plan_sales_present: for p in PLAN_SALE_BAD: for m in re.finditer(p, sentence, re.I): # "no … negative conviction signal" states the canonical finding; only an assertion counts if not DENIAL.search(CLAUSE.split(sentence[max(0, m.start() - 40):m.start()])[-1]): return False, "describes scheduled Rule 10b5-1 sales as conviction or a governance problem" if GUIDANCE_WORDS.search(sentence) and not GUIDANCE_TOKEN.search(sentence) and not DENIAL.search(low): return False, ("describes company guidance without citing a verified [[GUIDANCE_n]] claim" + ("" if rules.guidance_available else " (company guidance is unavailable in canonical data)")) for m in TOKEN.finditer(sentence): cid = m.group(1) if cid != "OWN_RETURN" and len(re.findall(re.escape(m.group(0)), sentence)) > 1: return False, f"[[{cid}]] repeated in one sentence" jm = re.search(r"\]\]\s*\[\[", sentence) if jm: return False, "two claims written back to back with nothing between them" tv = token_structure_violation(sentence, rules.catalog) if tv: return False, tv rv0 = relation_violation(sentence, rules) if rv0: return False, f"{rv0[0]}: {rv0[1]}" ms = mode_subject_violation(sentence, rules.rel_subjects) if ms: return False, f"RELATION_IDENTITY: {ms}" cg = clause_grammar_violation(sentence, rules.full_clause) if cg: return False, f"RELATION_CLAUSE_GRAMMAR: {cg}" ed0 = evidence_domain_violation(sentence, rules) if ed0: return False, f"EVIDENCE_DOMAIN_SCOPE: {ed0}" tr = trend_violation(sentence, rules) if tr: return False, f"TREND_RELATION_REQUIRED: {tr}" iv0 = interpretation_violation(sentence, rules.interpretation) if iv0: return False, f"EVIDENCE_INTERPRETATION: {iv0}" if rules.directional_context: dc = placement_violation(sentence, rules) if dc: return False, f"EVIDENCE_INTERPRETATION: {dc}" gb = guidance_binding_violation(sentence, rules.guidance_names) if gb: return False, f"GUIDANCE_LABEL_BINDING: {gb}" rl = relation_logic_violation(sentence, rules.alternative_cases, rules.case_variables) if rl: return False, f"RELATION_LOGIC_PRESERVATION: {rl}" rs_ = relation_slot_violation(sentence, rules.rel_subjects, rules.catalog) if rs_: return False, f"RELATION_CLAUSE_GRAMMAR: {rs_}" vr = valuation_relation_violation(sentence, rules.valuation_relations) or \ universal_valuation_violation(sentence, rules.valuation_relations) if vr: return False, f"VALUATION_RELATION_LANGUAGE: {vr}" ed = evidence_domain_violation(sentence, rules) if ed: return False, f"EVIDENCE_DOMAIN_SCOPE: {ed}" sg = segment_perimeter_violation(sentence, rules.segment_perimeter) if sg: return False, f"SEGMENT_PERIMETER_LANGUAGE: {sg}" ec = evidence_conclusion_violation(sentence, rules.evidence_uses) if ec: return False, f"{ec[0]}: {ec[1]}" ps = phrase_slot_violation(sentence, rules.catalog) if ps: return False, f"RELATION_CLAUSE_GRAMMAR: {ps}" fc = family_comparison_violation(sentence) if fc: return False, f"RELATION_REQUIRED_FOR_PAIR: {fc}" px = proximity_violation(sentence, rules.valuation_magnitudes) if px: return False, f"VALUATION_RELATION_LANGUAGE: {px}" own_cid = rules.own_return_claim or ("HEADLINE_RETURN" if "HEADLINE_RETURN" in rules.catalog else None) own_v = getattr(rules.catalog.get(own_cid), "value", None) if own_cid else None rd = return_direction_violation(sentence, own_v if isinstance(own_v, (int, float)) else None) if rd: return False, f"RETURN_DIRECTION_CONSISTENCY: {rd}" pf = performance_violation(sentence, rules.stock_returns, rules.company_words) if pf: return False, f"STOCK_PERFORMANCE_CLAIM: {pf}" av = assessment_violation(sentence, rules.assessments, rules.assumption_claims) if av: return False, f"ASSUMPTION_ASSESSMENT: {av}" qv = quote_violation(sentence, rules.transcript_turns, rules.company_words) or \ paraphrase_attribution_violation(sentence, rules.transcript_turns, rules.company_words) if qv: return False, f"QUOTE_ATTRIBUTION: {qv}" sd = statement_domain_violation(sentence, rules.statements) if sd: return False, f"STATEMENT_DOMAIN: {sd}" mv_ = price_move_violation(sentence, rules.catalog) if mv_: return False, f"VALUATION_RELATION_LANGUAGE: {mv_}" raw = IDENTIFIER.findall(TOKEN.sub(" ", sentence)) if raw: return False, f"PUBLIC_LANGUAGE_ONLY: internal identifier '{raw[0]}' written as text" pl = public_language_violation(sentence, rules.internal_terms) if pl: return False, f"PUBLIC_LANGUAGE_ONLY: {pl}" ev = evidence_scope_violation(sentence, rules) if ev: return False, f"MODEL_EVIDENCE_SCOPE: {ev}" gv = guidance_epistemic_violation(sentence, rules.guidance_terms) if gv: return False, f"GUIDANCE_EPISTEMIC_STATE: {gv}" pv = polarity_violation(sentence, rules.expectations_polarity) if pv: return False, f"EXPECTATIONS_POLARITY: {pv}" ov = ordering_violation(sentence, rules.range_fn) if ov: return False, f"orders [[{ov[0]}]] against [[{ov[1]}]]: {ov[2][0]}" nc = named_counterpart_violation(sentence, rules) if nc: return False, f"RELATION_REQUIRED_FOR_PAIR: {nc}" iv = implicit_comparison_violation(sentence, rules.counterparts, rules.descriptors) if iv: return False, (f"compares [[{iv[0]}]] with its counterpart [[{iv[1]}]] without citing it, and they are not comparable " f"({'; '.join(iv[2])[:120]})") dv = directional_violation(sentence, rules.restricted_evidence) if dv: return False, (f"uses {'13F ownership' if dv == '13F' else 'insider trading'} as a directional signal, which its evidence " "permissions do not allow (state the positions or trades only)") rv = relative_violation(sentence, rules.relative_allowed) if rv: return False, (f"describes the stock as rich/cheap against peers on {'a multiple' if rv == 'any' else rv} without a " "comparable peer basis") mv = motive_violation(sentence) if mv: return False, (f"infers a motive ('{mv[1]}') from {'Form 13F ownership' if mv[0] == '13F' else 'insider (Form 4)'} " "filings, which record positions and trades, not intent") for m in TOKEN.finditer(sentence): unit = getattr(C_BY_ID.get(m.group(1)), "unit_after", "") if C_BY_ID else "" if unit and not re.match(r"[\s\-\u2011]*(?:\w+\s+){0,1}" + unit, sentence[m.end():m.end() + 30], re.I): return False, f"[[{m.group(1)}]] is a count of {unit}s; write the unit after it (\"[[{m.group(1)}]] {unit}s\")" for m in TOKEN.finditer(sentence): cid = rules.own_return_claim if m.group(1) == "OWN_RETURN" else m.group(1) req = getattr(rules.catalog.get(cid), "required_qualifiers", ()) or () if req: before = sentence[max(0, m.start() - 60):m.start()].lower() if not any(q in before for q in req): return False, f"[[{m.group(1)}]] must be written as {req[0]} (one of: {', '.join(req[:4])})" for m in ROLE_NAME.finditer(sentence): role = m.group(1).upper() role = {"CHIEF EXECUTIVE OFFICER": "CEO", "CHIEF FINANCIAL OFFICER": "CFO", "CHIEF OPERATING OFFICER": "COO"}.get(role, role) canon = (rules.officers or {}).get(role) name = m.group(2).strip() if canon and not _same_person(name, canon): return False, f"names {name} as {role}; the canonical {role} is {canon}" if not canon: return False, f"names a {role} although no {role} is verified for this company" return True, "" def motive_violation(sentence): """('13F' | 'insider', word) when a sentence attributes a motive to institutional holders or insiders; else None. One rule for prose ingestion and the report audit (S10, S11).""" for m in MOTIVE_WORD.finditer(sentence): if DENIAL.search(CLAUSE.split(sentence[max(0, m.start() - 40):m.start()])[-1]): continue win = sentence[max(0, m.start() - MOTIVE_WINDOW): m.end() + MOTIVE_WINDOW] if MOTIVE_ACTOR_13F.search(win): return "13F", m.group(0) if MOTIVE_ACTOR_INSIDER.search(win) or (MOTIVE_ACTOR_MGMT.search(win) and TRADE_WORD.search(win)): return "insider", m.group(0) return None def guidance_violation(fragments): """True when a phrase describes company guidance without a GUIDANCE_n claim fragment (S9).""" text = "".join(f.text for f in fragments) if not GUIDANCE_WORDS.search(text) or DENIAL.search(text): return False return not any(str(f.claim_id or "").startswith("GUIDANCE_") for f in fragments) def qualifier_violation(fragments, required_for): """[(claim_id, qualifier list)] for claims written without their required qualifier (S12).""" out, ctx = [], "" for f in fragments: if f.claim_id: req = required_for(f.claim_id) or () if req and not any(q in ctx[-60:].lower() for q in req): out.append((f.claim_id, req)) ctx += f.text return out def _same_person(a, b): ta = {w.strip(".").lower() for w in a.split() if len(w.strip(".")) > 1} tb = {w.strip(".").lower() for w in b.split() if len(w.strip(".")) > 1} return bool(ta & tb) and (max(ta, key=len) in tb or max(tb, key=len) in ta) def to_fragments(sentence, rules: Rules): out, pos = [], 0 for m in TOKEN.finditer(sentence): if m.start() > pos: out.append(Fragment(sentence[pos:m.start()])) cid = m.group(1) if cid == "OWN_RETURN": cid = rules.own_return_claim c = rules.catalog[cid] txt = c.rendered import catalog as _cat if (c.stype == "REL" or getattr(_cat.BY_ID.get(cid), "clause", False)) and txt and not sentence[:m.start()].strip(): txt = txt[0].upper() + txt[1:] # a relation clause that opens the sentence starts with a capital out.append(Fragment(txt, c.claim_id, c.value)) pos = m.end() if pos < len(sentence): out.append(Fragment(sentence[pos:])) return out # a sentence that opens by pointing back ("Neither is credible.", "This means …") depends on what came before it ANAPHORIC_OPENING = re.compile(r"^\W*(?:this|that|these|those|it|its|they|their|them|neither|both|either|such|which|" r"the latter|the former|hence|thus|therefore|so|also|likewise|similarly|conversely|" r"otherwise|instead|then)\b", re.I) def ingest(text, rules: Rules, source): """(paragraph fragments, rejections). Each accepted sentence keeps its claim fragments. Once a sentence of the passage is removed, a later sentence that opens by referring back is removed too: its antecedent may be gone.""" frags, rej = [], [] removed = False for sent in _split(text): ok, why = check_sentence(sent, rules) if ok and removed and ANAPHORIC_OPENING.search(sent): ok, why = False, "DANGLING_REFERENCE: refers back to a sentence that was removed" if not ok: rej.append(Rejection(source, sent[:240], why)) removed = True continue if frags: frags.append(Fragment(" ", origin="llm")) for f in to_fragments(sent, rules): f.origin = "llm" frags.append(f) return frags, rej