| """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_]+)\]\]") |
| |
| |
| 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)|" |
| |
| r"(\b\d+\s*(?:basis\s+points?|bps?|percentage\s+points?|points?|pts|turns?|notches)\b)|((?<![\w.$/\-\u2010-\u2013])\d{2,}(?![\w%/\-\u2010-\u2013]))|" |
| |
| r"(\b(?:doubl(?:e|ed|es|ing)|tripl(?:e|ed|es|ing)|quadrupl\w*|halv(?:e|ed|es|ing)|twice|thrice|double-digit|triple-digit)\b)|" |
| r"((?<!first )(?<!second )(?<!back )(?<!front )(?<!latter )(?<!former )\bhalf\b(?!\s+(?:of\s+)?(?:the\s+)?(?:year|fiscal|calendar|decade)))", re.I) |
| |
| BENIGN = re.compile(r"\b(?:19|20)\d{2}\b|\bQ[1-4]\b|\bFY\s?\d{2,4}\b|\b\d{1,3}[-\u2010\u2011\u2012\u2013\s](?:month|year|quarter|day|week)s?\b|" |
| r"\b\d{1,2}\s+(?:months?|years?|quarters?|days?|analysts?|peers?|storytellers?|seats?)\b|" |
| r"\b10b5-1\b|\bRule\s+10b5-1\b|\bForm\s+4\b|\b8-K\b|\b10-[KQ]\b|\b20-F\b|\b6-K\b|\bItem\s+\d\.\d+\b|" |
| r"\bS&P\s+500\b|\bNasdaq(?:-|\s)100\b|\bRussell\s+(?:1000|2000|3000)\b|\bDow\s+30\b|\b13F\b|\bExhibit\s+99\.1\b", re.I) |
| SENT = re.compile(r"(?<=[.!?])\s+(?=[A-Z\[“\"])") |
|
|
| QUANT_TOPIC = re.compile(r"\b(quant(?:itative)? model|price model|machine[- ]learning|\bML\b|panel[- ]ML|cross-sectional " |
| r"(?:rank|model|screen)|factor model|backtest\w*|percentile|decile)\b", re.I) |
| |
| PANEL_TOPIC = re.compile(r"\b(panel[- ]ML|cross-sectional (?:rank|model|screen)|percentile|decile)\b", re.I) |
| PRICE_MODEL_TOPIC = re.compile(r"\b(quant(?:itative)? model|price model|machine[- ]learning|\bML\b|factor model|backtest\w*)\b", re.I) |
| PLAN_SALE_BAD = (r"governance\s+(?:drag|concern|risk|issue|red\s+flag)", |
| r"(?:bearish|negative)\s+(?:conviction|management\s+signal)", |
| r"(?:management|the\s+CEO|the\s+CFO|insiders?|executives?)\s+(?:is|are)\s+(?:exiting|bailing|fleeing|dumping)", |
| r"informed\s+(?:capital|money|selling)", r"smart\s+money\s+is\s+selling", r"loss\s+of\s+confidence") |
| ROLE_NAME = re.compile(r"\b(CEO|CFO|COO|chief (?:executive|financial|operating) officer)\s*,?\s+((?:[A-Z][a-z]+\.?\s?){1,3}[A-Z][a-z]+)") |
| |
| GUIDANCE_WORDS = re.compile(r"\b(guidance|guided|guides|reaffirm\w*|(?:company|management|its|the company's|management's)\s+" |
| r"(?:outlook|expects?|forecasts?|projects?|targets?)|(?:raised|lowered|cut|lifted|maintained)\s+" |
| r"(?:its|the|full-year|annual)\s+(?:outlook|forecast|targets?))\b", re.I) |
| GUIDANCE_TOKEN = re.compile(r"\[\[GUIDANCE_\d+\]\]") |
| |
| MOTIVE_ACTOR_INSIDER = re.compile(r"\b(insiders?|insider\s+(?:selling|buying|sales|purchases|activity|transactions)|Form\s+4)\b", re.I) |
| |
| |
| |
| MOTIVE_ACTOR_MGMT = re.compile(r"\b(management|executives?|directors?|the\s+CEO|the\s+CFO|officers?)\b", re.I) |
| TRADE_WORD = re.compile(r"\b(sell\w*|sold|sales?|buy\w*|bought|purchas\w*|stake|holdings?|shares|10b5-1|dispos\w*)\b", re.I) |
| MOTIVE_ACTOR_13F = re.compile(r"\b(institutions?|institutional\s+(?:investors?|holders?|ownership)|funds?|hedge\s+funds?|" |
| r"tracked\s+(?:long-term\s+)?investors?|long-term\s+investors?|smart\s+money|guru\w*|holders?|13F|" |
| r"super-?investors?)\b", re.I) |
| MOTIVE_WORD = re.compile(r"\b(conviction|confiden(?:ce|t)|vote\s+of\s+(?:no\s+)?confidence|betting|bet\s+on|believes?|believed|" |
| r"bullish|bearish|doubts?|doubting|sees?\s+value|knows?|informed|insight|signal(?:s|ed|ing|led|ling)?\s+" |
| r"(?:that|a|an|their|its|confidence|conviction|doubt|belief|optimism|pessimism|caution|concern)|" |
| r"optimis\w*|pessimis\w*|endorse\w*|vote\s+with\s+their\s+feet|skin\s+in\s+the\s+game)\b", re.I) |
| MOTIVE_WINDOW = 70 |
|
|
| |
| COMPARATIVE = re.compile(r"\b(vs\.?|versus|above|below|higher than|lower than|exceed\w*|lag\w*|premium|discount|richer|" |
| r"faster|quicker|stronger than|weaker than|outgrow\w*|more than|less than|" |
| r"cheaper|better than|worse than|compared (?:with|to)|against|outpac\w*|trail\w*|ahead of|behind|" |
| r"decelerat\w*|accelerat\w*|slow\w*|speed\w* up|down from|up from|drops? to|falls? to|rises? to|" |
| r"expands? (?:from|to)|contracts? (?:from|to))\b", re.I) |
| |
| DIRECTIONAL = re.compile(r"\b(accumulat\w*|distribution|floor|ceiling|support\w*|cushion\w*|tailwind|headwind|bullish|bearish|" |
| r"upside|downside|thinning|deepening|confirm\w*|validat\w*|endorse\w*|backing|vote|conviction|" |
| r"smart money|flows? point|point(?:s)? to|signal\w*|reassur\w*|red flag|warning sign)\b", re.I) |
| EVIDENCE_13F = re.compile(r"\[\[OWN_13F_[A-Z_]+\]\]|\b(13F|institutions?|institutional|holders?|tracked (?:long-term )?investors?|" |
| r"ownership|breadth|super-?investors?|funds? (?:hold|own|add|bought|sold))\b", re.I) |
| EVIDENCE_INSIDER = re.compile(r"\[\[INSIDER_[A-Z_]+\]\]|\b(insiders?|insider (?:buying|selling|purchases|sales|activity)|Form 4)\b", re.I) |
| |
| RELVAL = re.compile(r"\b(rich(?:ly|er)?|cheap(?:ly|er)?|expensive|inexpensive|premium|discount|undervalued|overvalued|" |
| r"stretched|lofty|bargain)\b", re.I) |
| PEER_WORDS = re.compile(r"\b(peers?|peer group|competitors?|rivals?|comparables?|comps|the sector|sector peers|industry peers)\b", re.I) |
| _REL_METRIC = {"pe": r"p/e|\bpe\b|price[- ]to[- ]earnings|earnings multiple|\bearnings\b", |
| "ps": r"p/s|price[- ]to[- ]sales|sales multiple|\bsales\b", "ev_ebitda": r"ebitda", |
| "fcf_yield": r"free[- ]cash[- ]flow yield|fcf yield|\bfcf\b"} |
|
|
|
|
| def comparison_violation(sentence, compare_fn): |
| """(claim_a, claim_b, reasons) for the first pair of consecutive claims joined by a comparative construct that |
| compatible() rejects; None otherwise.""" |
| if compare_fn is None: |
| return None |
| toks = list(TOKEN.finditer(sentence)) |
| for x, y in zip(toks, toks[1:]): |
| between = sentence[x.end():y.start()] |
| before = sentence[max(0, x.start() - 40):x.start()] |
| comparative = COMPARATIVE.search(between) or (re.search(r"\bfrom\b", before, re.I) and re.search(r"\bto\b", between, re.I)) |
| if not comparative: |
| continue |
| r = compare_fn(x.group(1), y.group(1)) |
| if r is not None and not r[0]: |
| return x.group(1), y.group(1), r[1] |
| return None |
|
|
|
|
| TREND_VERB = re.compile(r"\b(decelerat\w*|re-?accelerat\w*|accelerat\w*|slow(?:s|ed|ing|down|er)?|speed\w* up|" |
| r"pick(?:s|ed|ing)? up|moderat(?:e|es|ed|ing))\b", re.I) |
|
|
|
|
| PERIOD_DESCRIPTOR = re.compile(r"\b(trailing|forward)\b", re.I) |
|
|
|
|
| def implicit_comparison_violation(sentence, counterparts, descriptors=None): |
| """(claim, counterpart, reasons) when a sentence cites one claim and compares it with an incomparable counterpart |
| it does not cite: by naming the counterpart's period or perimeter next to a comparative ("[[X]] below trailing"), |
| or by describing a trend in a growth rate whose historical counterpart is incomparable ("decelerating").""" |
| if not counterparts: |
| return None |
| cited = {m.group(1) for m in TOKEN.finditer(sentence)} |
| res = PERIOD_DESCRIPTOR.sub(" ", TOKEN.sub(" ", sentence)) |
| for cid in cited: |
| for other, why, words, trend in counterparts.get(cid, ()): |
| if other in cited: |
| continue |
| if trend and TREND_VERB.search(res): |
| return cid, other, why |
| fam = (descriptors or {}).get(other, (None, ()))[0] |
| owned = {w for c in cited for f, ws in [(descriptors or {}).get(c, (None, ()))] if f == fam for w in ws} |
| hit = [w for w in words if w not in owned and re.search(r"\b" + re.escape(w) + r"\b", TOKEN.sub(" ", sentence), re.I)] |
| if hit: |
| rest = res |
| for w in hit: |
| rest = re.sub(r"\b" + re.escape(w) + r"\b", " ", rest, flags=re.I) |
| if COMPARATIVE.search(rest): |
| return cid, other, why |
| return None |
|
|
|
|
| |
| TOKEN_RANGE = re.compile(r"\[\[([A-Z0-9_]+)\]\]\s*(–|—|-|\bto\b|\bthrough\b)\s*\[\[([A-Z0-9_]+)\]\]") |
| SIGN_DIRECTION = re.compile(r"\b(down|up|fell|rose|declin\w*|increas\w*|decreas\w*|dropp?\w*|gain\w*|grew|shrank|" |
| r"lower|higher)\s+(?:by\s+)?\[\[([A-Z0-9_]+)\]\]", re.I) |
| SIGN_DIRECTION_AFTER = re.compile(r"\[\[([A-Z0-9_]+)\]\]\s+(slide|slump|decline|drop|fall|loss|gain|rise|increase|decrease|" |
| r"jump|rally|retreat)\b", re.I) |
| RECALL_TAG = re.compile(r"\bfrom memory\b", re.I) |
|
|
|
|
| def token_structure_violation(sentence, catalog): |
| """A reason when claim tokens are combined into a figure no claim states; None otherwise.""" |
| for m in TOKEN_RANGE.finditer(sentence): |
| a, b = m.group(1), m.group(3) |
| if m.group(2).lower() in ("to", "through") and not (a.startswith("GUIDANCE_") and b.startswith("GUIDANCE_")): |
| continue |
| return f"[[{a}]] and [[{b}]] written as one range; each claim is a complete value" |
| g = [m.group(1) for m in TOKEN.finditer(sentence) if m.group(1).startswith("GUIDANCE_")] |
| seen = {} |
| for cid in dict.fromkeys(g): |
| r = getattr(catalog.get(cid), "rendered", None) |
| if r is not None and r in seen: |
| return (f"[[{seen[r]}]] and [[{cid}]] render identically ({r}); cite one and state its basis in words") |
| seen[r] = cid |
| for m in SIGN_DIRECTION.finditer(sentence): |
| r = str(getattr(catalog.get(m.group(2)), "rendered", "") or "") |
| if r[:1] in ("+", "-", "−"): |
| return f"[[{m.group(2)}]] already carries its sign ({r}); '{m.group(1)}' doubles or inverts it" |
| for m in SIGN_DIRECTION_AFTER.finditer(sentence): |
| r = str(getattr(catalog.get(m.group(1)), "rendered", "") or "") |
| if r[:1] in ("+", "-", "−"): |
| return f"[[{m.group(1)}]] already carries its sign ({r}); '{m.group(2)}' doubles or inverts it" |
| if RECALL_TAG.search(sentence): |
| num = [c for c in TOKEN.findall(sentence) if getattr(catalog.get(c), "stype", "TEXT") != "TEXT"] |
| if num: |
| return f"a recalled fact ('from memory') is qualitative and may not carry the verified figure [[{num[0]}]]" |
| return None |
|
|
|
|
| |
| ORDER_UP = re.compile(r"\b(above|higher than|exceed\w*|more than|greater than|ahead of|over|outpac\w*|beyond)\b", re.I) |
| ORDER_DOWN = re.compile(r"\b(below|lower than|less than|short of|under|lag\w*|behind|trail\w*)\b", re.I) |
|
|
|
|
| def ordering_violation(sentence, range_fn): |
| """(a, b, why) when a sentence orders two numeric claims ([[A]] … below … [[B]]) in a way their values contradict; |
| a range that contains the other value is neither above nor below it.""" |
| if range_fn is None: |
| return None |
| toks = list(TOKEN.finditer(sentence)) |
| for x, y in zip(toks, toks[1:]): |
| between = PERIOD_DESCRIPTOR.sub(" ", sentence[x.end():y.start()]) |
| up, down = ORDER_UP.search(between), ORDER_DOWN.search(between) |
| if bool(up) == bool(down): |
| continue |
| ra, rb = range_fn(x.group(1)), range_fn(y.group(1)) |
| if not (ra and rb) or ra[2] != rb[2]: |
| continue |
| a_lo, a_hi, b_lo, b_hi = ra[0], ra[1], rb[0], rb[1] |
| true = (a_lo > 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 |
|
|
|
|
| |
| |
| |
| 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 |
| 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_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") |
| |
| CLAUSE_OPENERS = tuple(w for w in CLAUSE_CONJ if w != "which") |
| |
| |
| 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, ((), ())) |
| |
| 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 |
| 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) |
| |
| 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) |
| |
| 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") |
| |
| |
| 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": |
| |
| tail = sentence[:m.start()].rstrip().lower() |
|
|
| def _subject_then_modifier(t): |
| |
| |
| 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 |
| |
| if re.fullmatch(r"\s*⟨tok⟩\s*,?\s*(?:and|but|yet|while)\s*", rest): |
| return True |
| return False |
| |
| 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]] |
| |
| |
| 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 |
| |
| |
| 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(): |
| before = low[max(0, dm.start() - 80):dm.start()] |
| |
| |
| 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"): |
| |
| explicit, said = (price_named or rid in toks), rtype |
| elif not compound and dm.start() >= am.start(): |
| 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 |
| 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 |
| 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 |
| 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() |
| |
| 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"(?<![\w-])" + x + r"(?![\w-])", scope)), None) |
| if not w or re.search(r"\b(?:not|hardly|neither)\W+(?:\w+\W+)?" + w, scope): |
| continue |
| claimed = _rel.level_assessment(cid, level) |
| a = (assessments or {}).get(cid) or {} |
| if a.get("assessment") != claimed: |
| return (f"'{w}' makes [[{cid}]] {claimed.lower()}, but it is assessed " |
| f"{str(a.get('assessment') or 'UNASSESSED').lower().replace('_', ' ')}" |
| f"{' (' + a['basis'] + ')' if a.get('basis') else ''}") |
| for clause in re.split(r"[,;:—–(]|\b(?:and|but|while|whereas|plus)\b", sentence): |
| low = clause.lower() |
| for cls, words in _rel.ASSESSMENT_WORDS.items(): |
| m = next((re.search(r"\b" + w + r"\w*", low) for w in words if re.search(r"\b" + w + r"\w*", low)), None) |
| if not m or re.search(r"\b(?:not|hardly|far from|anything but|neither|nor)\W+(?:\w+\W+)?$", low[:m.start()]): |
| continue |
| toks = [t.group(1) for t in TOKEN.finditer(clause) if t.group(1) in (assumption_claims or ())] |
| for t in toks: |
| a = (assessments or {}).get(t) or {} |
| if a.get("assessment") != cls: |
| return (f"[[{t}]] is assessed {str(a.get('assessment') or 'UNASSESSED').lower().replace('_', ' ')}" |
| f"{' (' + a['basis'] + ')' if a.get('basis') else ''}; '{m.group(0)}' asserts {cls.lower()}") |
| if not toks: |
| |
| |
| for t in [x.group(1) for x in TOKEN.finditer(sentence) if x.group(1) in (assumption_claims or ())]: |
| a = (assessments or {}).get(t) or {} |
| if a.get("assessment") != cls: |
| return (f"'{m.group(0)}' judges [[{t}]], which is assessed " |
| f"{str(a.get('assessment') or 'UNASSESSED').lower().replace('_', ' ')}") |
| if not toks and re.search(r"\b" + re.escape(m.group(0)) + r"\W+(?:\w+\W+){0,2}(?:assumptions?|inputs?|parameters?|" |
| r"discounting|valuation inputs?)\b", low_full): |
| return (f"'{m.group(0)}' judges the story's assumptions collectively; only a single assumption with a typed " |
| "assessment may be called conservative or aggressive") |
| return None |
|
|
|
|
| _QUOTE = re.compile(r"[\"“]([^\"”]{12,300})[\"”]|(?<![A-Za-z])['‘]([^'’]{12,300}?)['’](?![A-Za-z])") |
| _COMPANY_SIDE = r"\b(?:management|the company|company|executives?|ceo|cfo|coo|chief executive|chief financial|leadership|guid\w*)\b" |
| _ADMISSION = r"\b(?:acknowledg\w*|admit\w*|admission|conced\w*|concession|confess\w*|conceding)\b" |
| _ANALYST_SIDE = r"\b(?:analysts?|questioner|a question|the question|asked)\b" |
|
|
|
|
| def quote_violation(sentence, turns, company_words=()): |
| """A call quotation attributed to the wrong side (an analyst's question presented as what the company acknowledges, |
| or management's words as an analyst's), or presented as from the call without being in it.""" |
| if turns is None: |
| return None |
| import evidence as _ev |
| for m in _QUOTE.finditer(sentence): |
| q = m.group(1) or m.group(2) |
| if len(re.findall(r"[A-Za-z]+", q)) < 4: |
| continue |
| turn = _ev.quote_speaker(turns, q) |
| before = sentence[:m.start()].lower() |
| framed_call = re.search(r"\b(?:call|transcript|said|says|noted|stated|per|remarked|told)\b", sentence.lower()) |
| if turn is None: |
| if framed_call: |
| return f"the quotation '{q[:50]}' is not in the call transcript the analysis used" |
| continue |
| company = bool(re.search(_COMPANY_SIDE, before) or re.search(_ADMISSION, before) |
| or any(w and w in before for w in company_words)) |
| if turn["role"] == _ev.ANALYST and company: |
| return (f"'{q[:50]}' was said by an analyst ({turn['speaker']}) asking a question; it cannot be presented as " |
| "what the company or management acknowledges") |
| if turn["role"] == _ev.MANAGEMENT and re.search(_ANALYST_SIDE, before) and not company: |
| return f"'{q[:50]}' was said by management ({turn['speaker']}), not by an analyst" |
| return None |
|
|
|
|
| _SPEECH = r"\b(?:acknowledg\w*|admit\w*|conced\w*|said|says|noted|notes|stated|states|told|highlighted|highlights|" \ |
| r"remarked|conceded|warned|flagged|signal(?:l)?ed|touted|touts|emphasi[sz]ed|cites|cited|claims|claimed|" \ |
| r"describes|described|points to|pointed to|argues|argued|reiterat\w*|boasts|boasted|promised|promises|insists|insisted)\b" |
| def paraphrase_attribution_violation(sentence, turns, company_words=()): |
| """A statement attributed to management ("the CEO acknowledged on the call that monetization remains challenging") |
| whose content comes from an analyst's turn and from no management turn.""" |
| if not turns: |
| return None |
| low = TOKEN.sub(" ", sentence).lower() |
| sp = re.search(_SPEECH, low) |
| if not sp or not (re.search(_COMPANY_SIDE, low[:sp.start()]) or any(w and w in low[:sp.start()] for w in company_words)): |
| return None |
| import evidence as _ev |
| |
| |
| |
| quotes = [q.group(1) or q.group(2) for q in _QUOTE.finditer(sentence)] + \ |
| [q.group(1) for q in re.finditer(r"['‘]([^'’]{6,120})['’]", sentence)] |
| quotes = [q for q in quotes if len(re.findall(r"[A-Za-z]+", q)) >= 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") |
|
|
|
|
| |
| |
| 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": |
| |
| 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 |
|
|
|
|
| |
| 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] |
| |
| 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))] |
| |
| 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 = 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") |
| |
|
|
|
|
| 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 |
|
|
|
|
| |
| 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 |
|
|
|
|
| |
| _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 |
|
|
|
|
| |
| |
| |
| IDENTIFIER = re.compile(r"(?<!\[\[)\b(?:[a-z][a-z0-9]*(?:_[a-z0-9]+)+|[A-Z][A-Z0-9]*(?:_[A-Z0-9]+)+)\b(?!\]\])") |
|
|
|
|
| def public_language_violation(sentence, terms): |
| """Internal QA or state vocabulary (coherence checks, verification status, validators, the claim registry) in text a |
| reader sees. The terms are the system's own internal states and fields, not a list of forbidden opinions.""" |
| low = TOKEN.sub(" ", sentence).lower() |
| hit = next((t for t in terms or () if re.search(r"\b" + re.escape(t) + r"\b", low)), None) |
| return f"'{hit}' is internal QA language; describe the public limitation instead" if hit else None |
|
|
|
|
| def placement_violation(sentence, rules): |
| """A sentence placed as a reason for or against the view is directional by position: descriptive-only evidence |
| (13F positions, immaterial insider trades, unclassified headlines) cannot be that reason.""" |
| import evidence as _ev |
| toks = {m.group(1) for m in TOKEN.finditer(sentence)} |
| low = TOKEN.sub(" ", sentence).lower() |
| for fam, d in (rules.interpretation or {}).items(): |
| if d.get("permission") == _ev.DESCRIPTIVE_ONLY and (toks & set(d.get("claims") or ()) or any(w in low for w in d.get("subjects") or ())): |
| return f"{fam} evidence is descriptive only; it cannot be listed as a reason for or against the view" |
| news = (rules.evidence_uses or {}).get("NEWS") or {} |
| if any(w in low for w in news.get("subjects") or ()): |
| return "headlines are descriptive context only; they cannot be listed as a reason for or against the view" |
| return None |
|
|
|
|
| def evidence_scope_violation(sentence, rules): |
| """Evidence about one model used in a sentence about another model.""" |
| if not (rules.evidence_targets and rules.claim_model): |
| return None |
| toks = [rules.own_return_claim if m.group(1) == "OWN_RETURN" else m.group(1) for m in TOKEN.finditer(sentence)] |
| for e in toks: |
| t = rules.evidence_targets(e) |
| if not t: |
| continue |
| src, targets = t |
| for o in toks: |
| mo = rules.claim_model(o) if o and o != e else None |
| if mo and mo != src and mo not in targets: |
| return f"[[{e}]] is evidence about {src}; it may not evaluate {mo} ([[{o}]])" |
| return None |
|
|
|
|
| def guidance_epistemic_violation(sentence, terms): |
| """Presence or absence of guidance asserted for an item whose extraction was rejected, or absence asserted for |
| an item that was only not found.""" |
| low = sentence.lower() |
| for concept, (state, names) in (terms or {}).items(): |
| if state in ("AVAILABLE", "VERIFIED_ABSENT") or not any(n in low for n in names): |
| continue |
| if state == "FOUND_BUT_REJECTED" and (DENIAL.search(low) or GUIDANCE_WORDS.search(sentence)): |
| return (f"{concept.lower().replace('_', ' ')} outlook: the filing's text failed verification, so neither its " |
| "presence nor its absence may be stated") |
| if state in ("NOT_FOUND", "UNKNOWN") and DENIAL.search(low) and GUIDANCE_WORDS.search(sentence): |
| return f"{concept.lower().replace('_', ' ')} guidance was not found, which is not a verified absence" |
| return None |
|
|
|
|
| def polarity_violation(sentence, polarity): |
| """Expectation polarity asserted without, or against, the ImpliedExpectations relation.""" |
| import relations as _rel |
| bare = TOKEN.sub(" ", sentence) |
| pols = _rel.polarity_asserted(bare, structural=False) |
| |
| |
| for sp in _rel.polarity_asserted(bare, structural=True): |
| if polarity is None: |
| return "compares a fundamental with what the price implies, but no expectations relation exists for this issuer" |
| if sp != polarity: |
| return (f"compares a fundamental with what the price implies as {sp.lower().replace('_', ' ')}, but the " |
| f"relation holds {polarity.lower().replace('_', ' ')}") |
| if not pols: |
| return None |
| if "[[REL_EXPECTATIONS_STORY]]" not in sentence: |
| return "asserts how demanding the price is relative to the story without citing [[REL_EXPECTATIONS_STORY]]" |
| if polarity not in pols: |
| return f"asserts {pols[0].replace('EXPECTATIONS_', '').lower().replace('_', ' ')} but the relation holds {polarity}" |
| return None |
|
|
|
|
| |
| def _dash(t): |
| """One hyphen: the non-breaking, figure and en dashes an author types are the same word joiner.""" |
| return re.sub(r"[\u2010\u2011\u2012\u2013\u00ad]", "-", t) |
|
|
|
|
| PHRASE_DIRECTION_WORDS = ("up", "down", "more", "fewer", "upside", "downside", "gain", "loss") |
|
|
|
|
| def qualifier_doubled(sentence, catalog, own_return_claim=None): |
| """A claim that renders its own qualifier written with it a second time: a direction after a gap or a relation |
| clause ("…is 7.8% above consensus ($19.04) above consensus"), or a bound before a guidance value that already |
| states it ("more than more than $7B"). The qualifier words are the claims' own (GAP/REL directions, guidance |
| operators), not a comparison list.""" |
| import guidance as _g |
| for m in GAP_THEN_DIRECTION.finditer(sentence): |
| cid = own_return_claim if m.group(1) == "OWN_RETURN" else m.group(1) |
| if getattr(catalog.get(cid), "stype", None) in ("GAP", "REL"): |
| return (f"CLAIM_QUALIFIER_DOUBLED: [[{m.group(1)}]] already states its direction; a second direction word inverts " |
| "or doubles it") |
| for m in TOKEN.finditer(sentence): |
| cl = catalog.get(own_return_claim if m.group(1) == "OWN_RETURN" else m.group(1)) |
| r = str(getattr(cl, "rendered", "") or "") |
| |
| b, a = sentence[:m.start()], sentence[m.end():] |
| if (r.lstrip("+-").startswith("$") and b.rstrip(" ").endswith("$")) or \ |
| (r.endswith("%") and a.lstrip(" ").startswith("%")) or (r.endswith("x") and re.match(r"\s*x\b", a)) or \ |
| (r.endswith("bps") and re.match(r"\s*bps\b", a, re.I)): |
| return f"CLAIM_QUALIFIER_DOUBLED: [[{m.group(1)}]] already renders its unit ({r}); do not type it again" |
| |
| if getattr(cl, "stype", None) == "TEXT" and r and len(r) <= 40 and not any(ch.isdigit() for ch in r): |
| around = (sentence[max(0, m.start() - 60):m.start()] + " " + sentence[m.end():m.end() + 40]).lower() |
| if _dash(TOKEN.sub(" ", around)).find(_dash(r.lower())) >= 0: |
| 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)} |
| |
| 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 |
| 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 |
| except Exception: |
| C_BY_ID = {} |
|
|
|
|
| @dataclass |
| class Fragment: |
| text: str |
| claim_id: str | None = None |
| value: object = None |
| origin: str = "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 |
| panel_evidence_allowed: bool = False |
| panel_forbidden_phrases: tuple = () |
| quant_forbidden_phrases: tuple = () |
| plan_sales_present: bool = False |
| guidance_available: bool = False |
| compare_fn: object = None |
| counterparts: dict = field(default_factory=dict) |
| descriptors: dict = field(default_factory=dict) |
| range_fn: object = None |
| restricted_evidence: tuple = () |
| relative_allowed: tuple = () |
| offered: object = None |
| pair_fn: object = None |
| magnitude_bindings: dict = field(default_factory=dict) |
| claim_model: object = None |
| evidence_targets: object = None |
| guidance_terms: dict = field(default_factory=dict) |
| expectations_polarity: str | None = None |
| presentation_claims: tuple = () |
| phrase_of: dict = field(default_factory=dict) |
| full_clause: object = None |
| rel_subjects: dict = field(default_factory=dict) |
| trends_allowed: dict = field(default_factory=dict) |
| change_claims: tuple = () |
| interpretation: dict = field(default_factory=dict) |
| guidance_names: dict = field(default_factory=dict) |
| alternative_cases: dict = field(default_factory=dict) |
| case_variables: dict = field(default_factory=dict) |
| segment_perimeter: dict = field(default_factory=dict) |
| evidence_uses: dict = field(default_factory=dict) |
| internal_terms: tuple = () |
| directional_context: bool = False |
| valuation_relations: dict = field(default_factory=dict) |
| valuation_magnitudes: dict = field(default_factory=dict) |
| stock_returns: dict = field(default_factory=dict) |
| assessments: dict = field(default_factory=dict) |
| assumption_claims: tuple = () |
| transcript_turns: list | None = None |
| company_words: tuple = () |
| statements: dict = field(default_factory=dict) |
| officers: dict = field(default_factory=dict) |
| own_return_claim: str | None = None |
| own_label: str | None = None |
| held_labels: tuple = () |
|
|
|
|
| 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 |
| |
| |
| 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): |
| |
| 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:] |
| out.append(Fragment(txt, c.claim_id, c.value)) |
| pos = m.end() |
| if pos < len(sentence): |
| out.append(Fragment(sentence[pos:])) |
| return out |
|
|
|
|
| |
| 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 |
|
|
|
|
|
|
