| """Structured arguments (contract ``crucible.args/2``): authors choose EVIDENCE, templates render every figure. |
| |
| Why this exists (measured on 14 live SPGI runs, 1,047 labelled sentences): when authors wrote free English with |
| number tokens spliced in, 52% of what they wrote was defective, a word-list validator published 30% defects and threw |
| away 35% of good prose, and 62% of the published defects were semantic — no rule over wording could decide them. The |
| loop closed only by changing what authors are asked to write. |
| |
| The contract: |
| - an author returns typed objects: ``{"kind", "evidence": [claim ids], "view": "<figure-free judgement>"}``, a |
| figure-free ``thesis``, rebuttal ``answers`` that point at the opponent's VALIDATED points, PM ``factors``; |
| - every figure, comparison and direction in the report is rendered by templates from the cited typed claims; |
| - validation is typed: ids offered and allowed for the kind, pair compatibility, figure-free text, rating and return |
| direction, assessments and valuation relations of the point's OWN evidence, quote lineage by speaker role; |
| - what a typed check cannot decide (does the judgement follow from its evidence?) goes to one semantic gate |
| (``gate_prompt`` / ``apply_gate``) with the reviewers' rubric, never to a word list. |
| |
| Nothing is rewritten: an object that fails is dropped with its reason codes. Parsing applies only the recorded, |
| meaning-preserving normalizations in ``_normalize_*`` (id spelling, kind and verdict aliases). |
| """ |
| from __future__ import annotations |
|
|
| import re |
| from dataclasses import dataclass, field |
|
|
| import catalog as C |
| import evidence as EV |
| import prose as P |
|
|
| SCHEMA = "crucible.args/2" |
| KINDS = ("OUTLOOK", "EARNINGS", "COMPETITION", "RISK", "MANAGEMENT", "VALUATION", "EXPECTATIONS", "OWNERSHIP") |
| KIND_LABEL = {"OUTLOOK": "Outlook", "EARNINGS": "Earnings", "COMPETITION": "Competition", "RISK": "Risk", |
| "MANAGEMENT": "Management", "VALUATION": "Valuation", "EXPECTATIONS": "What the price requires", |
| "OWNERSHIP": "Ownership"} |
| VERDICTS = ("FAILS", "PARTLY", "HOLDS") |
| VERDICT_CHIP = {"FAILS": "Disputed", "PARTLY": "Partly conceded", "HOLDS": "Conceded"} |
| LIMITS = {"view": 30, "thesis": 35, "pm_thesis": 50, "rebuttal_thesis": 35, "summary": 45, "item": 22, |
| "points": 4, "answers": 3, "factors": 4, "items": 3, "evidence": 3, "answer_evidence": 3} |
|
|
| |
| _STORY_ASSUMPTIONS = ("STORY_TAM", "STORY_SHARE", "STORY_REVENUE_MATURITY", "STORY_YEARS", "STORY_MARGIN", "STORY_WACC", |
| "STORY_WACC_STABLE", "STORY_TERMINAL_GROWTH", "STORY_FAILURE", "STORY_START_MARGIN", |
| "STORY_START_REVENUE", "STORY_MARGIN_CHANGE", "STORY_S2C") |
| _BREAK_EVEN = ("REL_BREAK_EVEN_REVENUE_STORY", "REL_BREAK_EVEN_MARGIN_STORY", "REL_BREAK_EVEN_REVENUE_GROWTH") |
| _TRENDS = ("TREND_REVENUE", "TREND_REVENUE_GROWTH", "TREND_OPERATING_MARGIN", "TREND_EPS") |
| KIND_IDS = { |
| "OUTLOOK": {"FORECAST_REVENUE", "FORECAST_REVENUE_GROWTH", "REVENUE_TTM", "CONSENSUS_REVENUE_FY1", "CONSENSUS_REVENUE_FY2", |
| "CONSENSUS_REVENUE_NTM", "REL_ENGINE_REVENUE_CONSENSUS", "LIFECYCLE_STAGE", "LIFECYCLE_FORWARD_GROWTH", |
| "TREND_REVENUE", "TREND_REVENUE_GROWTH"}, |
| "EARNINGS": {"OPERATING_MARGIN_TTM", "NET_MARGIN_TTM", "EPS_ADJUSTED_TTM", "EPS_GAAP_TTM", "FORECAST_EPS", |
| "FORECAST_EPS_GROWTH", "FORECAST_OPERATING_MARGIN", "FORECAST_REVENUE_GROWTH", "CONSENSUS_EPS_NTM", |
| "CONSENSUS_EPS_FY1", "CONSENSUS_EPS_FY2", "REL_FORECAST_EPS_CONSENSUS", "REL_GUIDANCE_EPS_CONSENSUS", "BEAT_RATE", |
| "TREND_OPERATING_MARGIN", "TREND_EPS"}, |
| "VALUATION": {"REL_PRICE_INTRINSIC", "REL_PRICE_MECHANICAL", "REL_PRICE_MARKET_COMPOSITE", "REL_TARGET_PRICE", |
| "PE_FORWARD", "PE_TRAILING", "PEG", "LIFECYCLE_STAGE", "DMD_BETA", "DMD_UNLEVERED_BETA", "DMD_ERP", |
| "DMD_IMPLIED_ERP", "DMD_CRP", "DMD_S2C", "REL_EXPECTATIONS_STORY", *_STORY_ASSUMPTIONS, *_BREAK_EVEN}, |
| "EXPECTATIONS": {"REL_EXPECTATIONS_STORY", *_BREAK_EVEN}, |
| "RISK": {"STORY_FAILURE", "MOMENTUM", "SMA200", "REL_EXPECTATIONS_STORY", *_BREAK_EVEN, *_TRENDS}, |
| "COMPETITION": {"REL_OPERATING_MARGIN_INDUSTRY"}, |
| "MANAGEMENT": set(), |
| "OWNERSHIP": set(), |
| } |
| |
| GUIDANCE_KIND = {"REVENUE": ("OUTLOOK", "RISK"), "REVENUE_GROWTH": ("OUTLOOK", "RISK"), "ORGANIC_REVENUE_GROWTH": ("OUTLOOK", "RISK"), |
| "EPS": ("EARNINGS", "RISK"), "EPS_GROWTH": ("EARNINGS", "RISK"), "OPERATING_MARGIN": ("EARNINGS", "RISK"), |
| "OPERATING_MARGIN_CHANGE": ("EARNINGS", "RISK"), "GROSS_MARGIN": ("EARNINGS",), |
| "GROSS_MARGIN_CHANGE": ("EARNINGS",), "NET_MARGIN": ("EARNINGS",), "OPERATING_INCOME": ("EARNINGS", "RISK"), |
| "OPERATING_EXPENSES": ("EARNINGS",), "SHARE_REPURCHASE": ("MANAGEMENT",), "DIVIDEND": ("MANAGEMENT",), |
| "CAPEX": ("MANAGEMENT",), "FREE_CASH_FLOW": ("EARNINGS", "MANAGEMENT")} |
| |
| |
| |
| |
| WITHHELD = frozenset({"WACC_BENCHMARK", "MARKET_COMPOSITE_RAW", "MARKET_COMPOSITE_POST_TRIM", "ANALYST_TARGET_ADJUSTED", |
| "DMD_ERP_MONTH", "DMD_COUNTRY", "OWN_13F_AS_OF", "SEGMENTS_COVERAGE", "ADJ_GAAP_BRIDGE", "CEO", |
| "CFO", "SEGMENT_COUNT_FILED", "SEGMENT_COUNT_CONTINUING", "DMD_MARGIN", "DMD_INDUSTRY", "DMD_DE", |
| "RETURN_12M_PHRASE", "RETURN_3M_PHRASE"}) |
|
|
|
|
| def kinds_for(cid, cat): |
| """The kinds a claim may support (empty = not author evidence).""" |
| base = str(cid).split("__")[0] |
| out = {k for k, ids in KIND_IDS.items() if base in ids} |
| cl = cat.get(base) if cat is not None else None |
| if getattr(cl, "stype", None) == C.GUIDANCE: |
| out |= set(GUIDANCE_KIND.get(((cl.value or {}) if isinstance(cl.value, dict) else {}).get("concept"), ())) |
| return out |
|
|
|
|
| |
| @dataclass |
| class Context: |
| state: object |
| cat: dict |
| rules: object |
| offer: frozenset |
| kind_evidence: dict |
| relations: dict |
| internal_terms: tuple = () |
| qa_vocab: object = None |
| kinds: tuple = () |
| cases: dict = field(default_factory=dict) |
|
|
|
|
| def author_offer(state, cat): |
| """The claim ids an author may cite: offered ∩ KIND_EVIDENCE − WITHHELD (+ ownership only when directional).""" |
| import report_model as RM |
| interp = ((state.get("evidence.state") or {}).get("interpretation") or {}) |
| table = {f"GUIDANCE_{i}" for i, _m in RM.guidance_table_rows(state, cat)} |
| trailing_forward_ok = True |
| try: |
| cmp_ = C.comparison(state, "LIFECYCLE_TRAILING_GROWTH", "LIFECYCLE_FORWARD_GROWTH") |
| trailing_forward_ok = cmp_ is None or bool(cmp_[0]) |
| except Exception: |
| trailing_forward_ok = False |
| directional = {fam for fam, key in (("OWN_13F_", "13F"), ("INSIDER_", "insider")) |
| if (interp.get(key) or {}).get("permission") == EV.DIRECTIONAL_ALLOWED} |
| out = set() |
| for cid in C.offered(state, cat, include_deferred=False): |
| if "__" in cid or cid in WITHHELD: |
| continue |
| cl = cat.get(cid) |
| if cl is None or cl.rendered is None: |
| continue |
| if any(cid.startswith(pre) for pre in directional): |
| out.add(cid) |
| continue |
| if not kinds_for(cid, cat): |
| continue |
| if getattr(cl, "stype", None) == C.GUIDANCE and cid not in table: |
| continue |
| if cid == "LIFECYCLE_TRAILING_GROWTH" and not trailing_forward_ok: |
| continue |
| out.add(cid) |
| return frozenset(out) |
|
|
|
|
| def context(state, cat, rules, internal_terms=(), qa_vocab=None): |
| offer = author_offer(state, cat) |
| kev = {} |
| for k in KINDS: |
| if k == "OWNERSHIP": |
| ids = {c for c in offer if c.startswith(("OWN_13F_", "INSIDER_"))} |
| else: |
| ids = {c for c in offer if k in kinds_for(c, cat)} |
| if ids: |
| kev[k] = frozenset(ids) |
| items = ((state.get("relations.state") or {}).get("items")) or {} |
| rels = {rid: {"lhs": r.get("lhs_claim_id"), "rhs": r.get("rhs_claim_id"), "magnitude_claim": r.get("presentation_claim_id"), |
| "years": r.get("years_claim_id"), "type": r.get("relation_type")} for rid, r in items.items()} |
| ex = items.get("REL_EXPECTATIONS_STORY") or {} |
| cases = dict(ex.get("alternative_case_claims") or {}) if ex.get("logical_structure") == "ALTERNATIVE_CONDITIONALS" else {} |
| return Context(state, cat, rules, offer, kev, rels, tuple(internal_terms or ()), qa_vocab, tuple(kev), cases) |
|
|
|
|
| def constituents(ctx, rid): |
| """The claims a relation states (its sides, magnitude and years); for the expectations frontier, its cases'.""" |
| if rid == "REL_EXPECTATIONS_STORY": |
| out = set() |
| for ids in ctx.cases.values(): |
| for c in ids: |
| out |= constituents(ctx, c) if c in ctx.relations else {c} |
| return out |
| r = ctx.relations.get(rid) or {} |
| return {x for x in (r.get("lhs"), r.get("rhs"), r.get("magnitude_claim"), r.get("years")) if x} |
|
|
|
|
| |
| _KIND_ALIASES = {"OUTLOOK": "OUTLOOK", "BUSINESS OUTLOOK": "OUTLOOK", "GROWTH": "OUTLOOK", "DEMAND": "OUTLOOK", |
| "EARNINGS": "EARNINGS", "MARGINS": "EARNINGS", "PROFITABILITY": "EARNINGS", |
| "COMPETITION": "COMPETITION", "COMPETITIVE LANDSCAPE": "COMPETITION", "MOAT": "COMPETITION", |
| "COMPETITIVE POSITION": "COMPETITION", "RISK": "RISK", "RISKS": "RISK", "BUSINESS RISKS": "RISK", |
| "MANAGEMENT": "MANAGEMENT", "CAPITAL ALLOCATION": "MANAGEMENT", "VALUATION": "VALUATION", |
| "MULTIPLES": "VALUATION", "INTRINSIC VALUE": "VALUATION", "EXPECTATIONS": "EXPECTATIONS", |
| "PRICED IN": "EXPECTATIONS", "WHAT THE PRICE REQUIRES": "EXPECTATIONS", "OWNERSHIP": "OWNERSHIP"} |
| _VERDICT_ALIASES = {"FAILS": "FAILS", "FAIL": "FAILS", "REFUTED": "FAILS", "WRONG": "FAILS", "FALSE": "FAILS", |
| "DISPUTED": "FAILS", "PARTLY": "PARTLY", "PARTIAL": "PARTLY", "MIXED": "PARTLY", |
| "PARTLY HOLDS": "PARTLY", "HOLDS": "HOLDS", "VALID": "HOLDS", "STANDS": "HOLDS", |
| "CONCEDED": "HOLDS", "AGREE": "HOLDS", "CONCEDE": "HOLDS"} |
| _ID = re.compile(r"[A-Za-z][A-Za-z0-9_]*") |
|
|
|
|
| def coerce_ids(x): |
| """A list of base claim ids from a list or "A, B" or "[[A]]" or "A (label)"; never fuzzy-matched.""" |
| items = x if isinstance(x, list) else ([x] if isinstance(x, str) else []) |
| out = [] |
| for it in items: |
| if not isinstance(it, str): |
| continue |
| for part in re.split(r"[,;\s]+(?=\[?\[?[A-Za-z])", it.strip()): |
| m = _ID.search(part.replace("[", " ").replace("]", " ").replace("{", " ").replace("}", " ")) |
| if not m: |
| continue |
| cid = m.group(0).upper().split("__")[0] |
| cid = C.PHRASE_OF.get(cid, cid) |
| if cid not in out: |
| out.append(cid) |
| return out |
|
|
|
|
| def _kind(x): |
| return _KIND_ALIASES.get(re.sub(r"\s+", " ", str(x or "")).strip().upper()) |
|
|
|
|
| def _text(x): |
| if not isinstance(x, str): |
| return None |
| t = re.sub(r"\s+", " ", x).strip().strip('"').strip() |
| t = re.sub(r"^[-•*]\s+", "", t) |
| return t or None |
|
|
|
|
| def _view_of(d): |
| for k in ("view", "point", "text", "judgement", "judgment", "argument"): |
| if isinstance(d.get(k), str): |
| return _text(d[k]) |
| return None |
|
|
|
|
| def _evidence_of(d): |
| for k in ("evidence", "evidence_ids", "claim_ids", "claims", "ids", "tokens", "citations"): |
| if k in d: |
| return coerce_ids(d[k]) |
| return [] |
|
|
|
|
| def is_structured(raw): |
| """True when a persona reply carries structured-argument objects.""" |
| st = (raw or {}).get("structured") or {} |
| pts = st.get("points") or st.get("factors") or st.get("answers") |
| return isinstance(pts, list) and any(isinstance(p, dict) and _view_of(p) for p in pts) |
|
|
|
|
| |
| _RAW_ID = P.IDENTIFIER |
| _QUOTES = re.compile(r"[\"“”]|(?<![A-Za-z])['‘][^'’]{12,}['’](?![A-Za-z])") |
| _BAND_WORDS = tuple(b for b, _lo, _hi in P.MAGNITUDE_BANDS) |
| _SPELLED_CARDINAL = re.compile(r"\b(two|three|four|five|six|seven|eight|nine|ten|eleven|twelve|thirteen|fourteen|fifteen|" |
| r"sixteen|seventeen|eighteen|nineteen|twenty|thirty|forty|fifty|sixty|seventy|eighty|ninety|" |
| r"hundred|thousand|million|billion|trillion)\b", re.I) |
|
|
|
|
| def text_reasons(ctx, text, *, max_words, form=True): |
| """T1–T5 on an authored text: shape, numerals, rating labels, internal vocabulary, form.""" |
| r = [] |
| if not text: |
| return ["EMPTY"] |
| if "[[" in text or "]]" in text or "{" in text or "}" in text: |
| r.append("TOKEN_IN_VIEW: claim ids go in evidence, never in the text") |
| m = _RAW_ID.search(text) |
| if m or any(re.search(r"\b" + re.escape(c) + r"\b", text) for c in ctx.offer): |
| r.append(f"ID_TEXT_IN_VIEW: '{m.group(0) if m else 'claim id'}'") |
| if _QUOTES.search(text): |
| r.append("QUOTE_IN_VIEW: quotations are not published in authored text") |
| if re.search(r"\d", text): |
| r.append("NUMERAL: the report prints every figure from the evidence") |
| else: |
| low = P._dash(text.lower()) |
| wm = P.MATERIAL_NUMBER.search(P.BENIGN.sub(" ", text)) |
| band = next((b for b in _BAND_WORDS if b in low or b.replace("-", " ") in low), None) |
| sc = _SPELLED_CARDINAL.search(re.sub(r"\b[a-z]+[-\u2010\u2011 ](?:month|year|quarter|day|week)s?\b", " ", low)) |
| if wm or band or sc: |
| r.append(f"NUMERAL: '{(wm.group(0) if wm else band or sc.group(0)).strip()}'") |
| if P.RATING_WORD.search(text): |
| r.append("RATING_WORD: the report prints each rating itself") |
| pl = P.public_language_violation(text, ctx.internal_terms) |
| if pl or (ctx.qa_vocab is not None and ctx.qa_vocab.search(text)): |
| r.append(f"INTERNAL_VOCAB: {pl or 'QA vocabulary'}") |
| low = text.lower() |
| for t in getattr(ctx.rules, "engine_forbidden", ()) or (): |
| if re.search(r"\b" + re.escape(t.lower()) + r"\b", low) and not P.DENIAL.search(low): |
| r.append(f"INTERNAL_VOCAB: '{t}' is not this issuer's valuation vocabulary") |
| break |
| for topic, phrases in ((P.PRICE_MODEL_TOPIC, getattr(ctx.rules, "quant_forbidden_phrases", ()) or ()), |
| (P.PANEL_TOPIC, getattr(ctx.rules, "panel_forbidden_phrases", ()) or ())): |
| if topic.search(text) and any(re.search(r"\b" + re.escape(ph.lower()) + r"\b", low) for ph in phrases): |
| r.append("INTERNAL_VOCAB: describes a quantitative model more strongly than its reliability allows") |
| break |
| n = len([w for w in re.split(r"\s+", text.strip()) if w]) |
| if n > max_words: |
| r.append(f"WORD_LIMIT: {n} words (at most {max_words})") |
| if re.search(r"(?:…|\.\.\.)\W*$", text.strip()) or not re.search(r"[.!?][\"'”’)\]]*$", text.strip()): |
| r.append("INCOMPLETE: not a complete sentence") |
| if form and P.ANAPHORIC_OPENING.search(text): |
| r.append("ANAPHORIC_OPENING: a point stands on its own") |
| return r |
|
|
|
|
| |
| |
| _CONSENSUS_IDS = frozenset({"CONSENSUS_EPS_FY1", "CONSENSUS_EPS_FY2", "CONSENSUS_EPS_NTM", "CONSENSUS_REVENUE_FY1", |
| "CONSENSUS_REVENUE_FY2", "CONSENSUS_REVENUE_NTM", "LIFECYCLE_FORWARD_GROWTH"}) |
| _SUBJECT_OF_CONCEPT = {"REVENUE": "revenue", "REVENUE_GROWTH": "revenue", "ORGANIC_REVENUE_GROWTH": "revenue", |
| "EPS": "eps", "EPS_GROWTH": "eps"} |
| |
| _CONNECT = {("ENGINE", "revenue"): "REL_ENGINE_REVENUE_CONSENSUS", ("ENGINE", "eps"): "REL_FORECAST_EPS_CONSENSUS", |
| ("GUIDANCE", "eps"): "REL_GUIDANCE_EPS_CONSENSUS", ("GUIDANCE", "revenue"): None} |
| |
| |
| _CLASS_NAMES = {"CONSENSUS": tuple(w for w in C._FORECAST_WORDS if "consensus" in w) + ("street",), |
| "ENGINE": tuple(w for w in C._CONSENSUS_WORDS if w != "guidance") + ("engine's forecast", "our estimate")} |
|
|
|
|
| def provenance(cid, cat): |
| """ENGINE / GUIDANCE / CONSENSUS for a claim, or None.""" |
| if getattr(cat.get(cid), "stype", None) == C.GUIDANCE: |
| return "GUIDANCE" |
| if cid in _CONSENSUS_IDS: |
| return "CONSENSUS" |
| if str(cid).startswith("FORECAST_"): |
| return "ENGINE" |
| return None |
|
|
|
|
| def _subject(ctx, cid): |
| try: |
| m = C._claim_metric(ctx.state, cid) |
| except Exception: |
| return None |
| return _SUBJECT_OF_CONCEPT.get(getattr(m, "economic_concept", None)) if m is not None else None |
|
|
|
|
| def _names_class(low, cls): |
| if cls == "GUIDANCE": |
| return bool(P.GUIDANCE_WORDS.search(low)) |
| return any(re.search(r"\b" + re.escape(n) + r"\b", low) for n in _CLASS_NAMES.get(cls, ())) |
|
|
|
|
| def lineage_reasons(ctx, text, scope, cocite=True, named=True): |
| """An engine or company-guidance figure compared with consensus only through the typed relation that establishes |
| the comparison (same horizon, perimeter, basis; an engine figure independent of consensus). Two checks, both typed, |
| neither reading the predicate (aligned, below, close to ... are never enumerated): |
| CO-CITATION — the text's evidence holds an engine/guidance claim and a consensus claim of the same subject; |
| NAMED COUNTERPART — the text cites one side and names the other source class by the catalog's own class labels |
| (catalog._FORECAST_WORDS / _CONSENSUS_WORDS, prose.GUIDANCE_WORDS) without citing it: two sources of one subject in |
| one statement are a comparison. Either way the connecting relation must be cited; where none exists the figures may |
| be stated in separate statements. |
| ``cocite`` is off for a thesis, whose scope is the union of its points' evidence rather than its own citation; |
| ``named`` is off for template sentences (typed renderings, audited by co-citation only).""" |
| ids = list(dict.fromkeys(scope or [])) |
| rels = [e for e in ids if e in ctx.relations] |
| own = {} |
| for e in ids: |
| members = covered(ctx, e) if e in ctx.relations else {e} |
| for x in members: |
| own.setdefault(x, e) |
| typed = {x: (provenance(x, ctx.cat), _subject(ctx, x)) for x in own} |
|
|
| def need(side, subj, a, b=None): |
| rel = _CONNECT.get((side, subj)) |
| |
| if rel and rel in rels and {a, b} - {None} <= covered(ctx, rel): |
| return None |
| code = "GUIDANCE_CONSENSUS_COMPATIBILITY" if side == "GUIDANCE" else "FORECAST_RELATION_REQUIRED" |
| what = f"{a} and {b}" if b else f"{a} and consensus" |
| if rel and rel in ctx.relations: |
| return f"{code}: {what} are compared only through {rel}" |
| need_ = ("the same fiscal period and horizon, a compatible accounting and security basis" if side == "GUIDANCE" |
| else "the same horizon, perimeter and basis, and an engine figure independent of consensus") |
| return f"{code}: no typed relation compares {what} (it needs {need_}); state them separately" |
| for a, (pa, sa) in (typed.items() if cocite else ()): |
| for b, (pb, sb) in typed.items(): |
| if pa in ("ENGINE", "GUIDANCE") and pb == "CONSENSUS" and sa and sa == sb: |
| r = need(pa, sa, a, b) |
| if r: |
| return [r] |
| low = P._dash(str(text or "").lower()) |
| if named: |
| present = {(p, s) for p, s in typed.values() if p and s} |
| for side, subj in present: |
| if side in ("ENGINE", "GUIDANCE") and ("CONSENSUS", subj) not in present and _names_class(low, "CONSENSUS"): |
| r = need(side, subj, next(x for x, v in typed.items() if v == (side, subj))) |
| if r: |
| return [r] |
| if side == "CONSENSUS": |
| for other in ("GUIDANCE", "ENGINE"): |
| if (other, subj) not in present and _names_class(low, other): |
| r = need(other, subj, next(x for x, v in typed.items() if v == (side, subj))) |
| if r: |
| return [r] |
| return [] |
|
|
|
|
| def _probe(text, scope): |
| return text + " ; " + " ".join(f"[[{e}]]" for e in scope) |
|
|
|
|
| def direction_reasons(ctx, text, scope, own, *, cocite=True): |
| """T6–T11: the text against its evidence SCOPE and the author's own return (typed values only).""" |
| R = ctx.rules |
| r = [] |
| vrels = R.valuation_relations or {} |
| probe = _probe(text, scope) |
| vr = P.valuation_relation_violation(probe, vrels) or P.universal_valuation_violation(text, vrels) |
| if vr: |
| r.append(f"VALUATION_CONTRADICTS: {vr}") |
| in_scope = [e for e in scope if e in vrels] |
| words = [w for t in vrels.values() for w in _VALUATION_WORDS.get(t, ())] |
| low = text.lower() |
| hit = next((w for w in _DIR_OF_WORD if re.search(r"\b" + w + r"\b", low)), None) |
| if hit and not in_scope and not P.DENIAL.search(low): |
| r.append(f"VALUATION_UNCITED: '{hit}' needs a cited valuation relation") |
| elif hit and in_scope and not P.DENIAL.search(low): |
| want = _DIR_OF_WORD.get(hit) |
| if want and all(vrels[e] != want for e in in_scope): |
| r.append(f"VALUATION_CONTRADICTS: '{hit}' but every cited anchor holds the other direction") |
| av = P.assessment_violation(probe, R.assessments, R.assumption_claims) |
| if av: |
| r.append(f"ASSESSMENT_CONTRADICTS: {av}") |
| if own.get("value") is not None: |
| rd = P.return_direction_violation(text, own["value"]) |
| if rd: |
| r.append(f"RETURN_SIGN: {rd}") |
| |
| |
| sp = re.search(P._SPEECH, low) |
| if sp and (re.search(_SPEAKERS, low[:sp.start()]) or any(w and w in low[:sp.start()] for w in R.company_words)): |
| r.append("ATTRIBUTION: a statement about what management said is not publishable without its quoted words") |
| for m in P.ROLE_NAME.finditer(text): |
| role = {"CHIEF EXECUTIVE OFFICER": "CEO", "CHIEF FINANCIAL OFFICER": "CFO", |
| "CHIEF OPERATING OFFICER": "COO"}.get(m.group(1).upper(), m.group(1).upper()) |
| canon = (R.officers or {}).get(role) |
| if not canon or not P._same_person(m.group(2).strip(), canon): |
| r.append(f"OFFICER: names {m.group(2).strip()} as {role}") |
| break |
| if P.GUIDANCE_WORDS.search(text) and not any(str(e).startswith("GUIDANCE_") for e in scope) and not P.DENIAL.search(low): |
| r.append("GUIDANCE_UNCITED: a statement about guidance needs a cited guidance row") |
| lc = P.lifecycle_violation(text, ctx.cat, R.company_words) |
| if lc: |
| r.append(f"LIFECYCLE: {lc}") |
| r += lineage_reasons(ctx, text, scope, cocite=cocite) |
| return r |
|
|
|
|
| |
| _SPEAKERS = r"\b(?:management|the company|company|executives?|ceo|cfo|coo|chief executive|chief financial|leadership)\b" |
| |
| |
| _DIR_OF_WORD = {"cheap": "BELOW", "undervalued": "BELOW", "expensive": "ABOVE", "overvalued": "ABOVE"} |
| _VALUATION_WORDS = {"ABOVE": ("expensive", "overvalued", "premium"), "BELOW": ("cheap", "undervalued", "discount")} |
|
|
|
|
| def covered(ctx, rid): |
| """What a cited relation already states: its constituents and, for the expectations frontier, its case relations.""" |
| out = set(constituents(ctx, rid)) |
| if rid == "REL_EXPECTATIONS_STORY": |
| out |= {c for ids in ctx.cases.values() for c in ids} |
| return out |
|
|
|
|
| def evidence_reasons(ctx, ids, kind, *, allowed=None, required=True, max_ids=None): |
| """Evidence rules (typed): known, offered, allowed, pairs, single case. Any offered id may support any kind: the kind |
| labels the point, and the pair and relation rules protect what the evidence can show (the per-kind lists dropped |
| valid evidence: a buyback guidance row under MANAGEMENT, an answer outside its target's kind). An id that a cited |
| relation already states adds no evidence, so it does not count toward the limit.""" |
| r = [] |
| cited = [e for e in ids if e in ctx.relations] |
| cov = set().union(*[covered(ctx, x) for x in cited]) if cited else set() |
| n = len([e for e in ids if e not in cov]) |
| if n > (max_ids or LIMITS["evidence"]): |
| r.append(f"EVIDENCE_COUNT: at most {max_ids or LIMITS['evidence']} ids") |
| for e in ids: |
| if ctx.cat.get(e) is None: |
| r.append(f"UNKNOWN_ID: {e}") |
| elif e not in ctx.offer: |
| r.append(f"WITHHELD_EVIDENCE: {e}") |
| elif allowed is not None and e not in allowed: |
| r.append(f"ID_NOT_ALLOWED: {e} cannot support this point") |
| if required and not ids: |
| r.append("NO_EVIDENCE: every point cites at least one id") |
| if r: |
| return r |
| rels_c = [e for e in ids if e in ctx.relations] |
| facts = [e for e in ids if e not in ctx.relations] |
| K = set().union(*[constituents(ctx, x) for x in rels_c]) if rels_c else set() |
| pair_fn = getattr(ctx.rules, "pair_fn", None) |
| if pair_fn is not None: |
| for i, a in enumerate(facts): |
| for b in facts[i + 1:] + sorted(K - {a}): |
| if a == b or (b in K and a in K): |
| continue |
| try: |
| st = pair_fn(a, b) |
| except Exception: |
| st = None |
| if st is None: |
| continue |
| if st[0] == "RELATION": |
| if st[1] in rels_c: |
| continue |
| return [f"PAIR_RELATION_REQUIRED: {a} and {b} are compared only through {st[1]}"] |
| if st[0] == "OK": |
| return [f"PAIR_UNRELATED: {a} and {b} have no typed relation (cite the relation, or put them in " |
| "separate points)"] |
| if benign_cocitation(ctx, a, b): |
| continue |
| return [f"PAIR_INCOMPATIBLE: {a} and {b} ({'; '.join(st[1])[:90]})"] |
| for i, x in enumerate(rels_c): |
| for y in rels_c[i + 1:]: |
| for a in constituents(ctx, x): |
| for b in constituents(ctx, y): |
| if a == b: |
| continue |
| try: |
| st = pair_fn(a, b) |
| except Exception: |
| st = None |
| if st and st[0] == "INCOMPATIBLE": |
| return [f"PAIR_INCOMPATIBLE: {x} and {y} ({'; '.join(st[1])[:90]})"] |
| |
| |
| if ctx.cases and "REL_EXPECTATIONS_STORY" not in ids: |
| hit = [c for c, cids in ctx.cases.items() if set(ids) & set(cids)] |
| if len(hit) > 1: |
| return ["EXPECTATIONS_JOINT_CASES: the revenue and margin solves are alternatives; cite REL_EXPECTATIONS_STORY"] |
| return r |
|
|
|
|
| def benign_cocitation(ctx, a, b): |
| """Two facts that may not be COMPARED can still share a point when citing them together cannot read as a |
| like-for-like comparison of one quantity: different concepts on the same perimeter, basis and quantity type |
| (operating and net margin; reported and organic growth), or the same concept for consecutive fiscal years (a |
| trajectory, FY1 then FY2). A perimeter, accounting-basis or level/change mismatch, or partly overlapping periods, |
| still drop: those are the measured defects (reviewers judged the rest of the pair drops good content).""" |
| import datetime as _dt |
| try: |
| ma, mb = C._claim_metric(ctx.state, a), C._claim_metric(ctx.state, b) |
| except Exception: |
| return False |
| if ma is None or mb is None or ma.perimeter != mb.perimeter: |
| return False |
| na = ("NA", None, "") |
| if ma.accounting_basis not in na and mb.accounting_basis not in na and ma.accounting_basis != mb.accounting_basis: |
| return False |
| if ma.quantity_type and mb.quantity_type and ma.quantity_type != mb.quantity_type: |
| return False |
| if ma.economic_concept != mb.economic_concept: |
| return True |
| try: |
| pairs = [(_dt.date.fromisoformat(str(x.period_start)[:10]), _dt.date.fromisoformat(str(x.period_end)[:10])) |
| for x in (ma, mb)] |
| except (TypeError, ValueError): |
| return False |
| (s1, e1), (s2, e2) = sorted(pairs) |
| return 0 <= (s2 - e1).days <= 5 |
|
|
|
|
| def cross_reasons(ctx, ids, target_ids): |
| """An answer's evidence against the evidence of the point it answers: only an INCOMPATIBLE pair drops (different |
| period, perimeter or basis); unrelated quantities may answer each other.""" |
| pair_fn = getattr(ctx.rules, "pair_fn", None) |
| if pair_fn is None: |
| return [] |
|
|
| def expand(xs): |
| out = set() |
| for x in xs: |
| out |= covered(ctx, x) if x in ctx.relations else {x} |
| return out |
| for a in sorted(expand(ids)): |
| for b in sorted(expand(target_ids)): |
| if a == b: |
| continue |
| try: |
| st = pair_fn(a, b) |
| except Exception: |
| st = None |
| if st and st[0] == "INCOMPATIBLE" and not benign_cocitation(ctx, a, b): |
| return [f"PAIR_INCOMPATIBLE_WITH_TARGET: {a} and {b} ({'; '.join(st[1])[:90]})"] |
| return [] |
|
|
|
|
| def _labels(x): |
| """[(prefix, number)] from [1, 3] / ["P1", "A2"] / "1, 3": each entry a bare number or ONE letter + number.""" |
| items = x if isinstance(x, list) else ([x] if isinstance(x, (str, int)) and not isinstance(x, bool) else []) |
| out = [] |
| for it in items: |
| if isinstance(it, bool): |
| continue |
| if isinstance(it, int): |
| out.append(("", it)) |
| continue |
| for part in re.split(r"[,;]\s*", str(it)): |
| m = re.fullmatch(r"\s*([A-Za-z]?)\s*(\d+)\s*", part) |
| if m: |
| out.append((m.group(1).upper(), int(m.group(2)))) |
| return out |
|
|
|
|
| def _positions(x): |
| return [k for _pre, k in _labels(x)] |
|
|
|
|
| def rests_on(raw, objs, own="P", by_target=False): |
| """The validated numbers (n) of the objects a thesis rests on. A bare number or the author's own prefix names its |
| own object by position (P for a seat's points, A for a rebuttal's answers, F for the PM's factors); in a rebuttal an |
| opponent P-label names the answer addressed to that point. Any other label names nothing (never a digit match).""" |
| by_src = {o.get("src"): o["n"] for o in objs or []} |
| by_to = {o.get("to"): o["n"] for o in objs or []} if by_target else {} |
| out = [] |
| for pre, k in _labels(raw): |
| if pre in ("", own): |
| n = by_src.get(k) |
| elif by_target and pre == "P": |
| n = by_to.get(f"P{k}") |
| else: |
| n = None |
| if n is not None and n not in out: |
| out.append(n) |
| return out |
|
|
|
|
| def _norm(t): |
| return re.sub(r"[^a-z0-9 ]", "", re.sub(r"\s+", " ", str(t or "").lower())).strip() |
|
|
|
|
| def _contains(big, small): |
| n = len(small) |
| return any(big[i:i + n] == small for i in range(len(big) - n + 1)) if n else False |
|
|
|
|
| def same_text(reply, opening): |
| """Typed identity on whole tokens: the reply IS the opening, a contiguous fragment of it, or the opening with at |
| most a fifth of new words. A longer reply that merely contains a short opening says something new.""" |
| x, y = _norm(reply).split(), _norm(opening).split() |
| if not x or not y: |
| return False |
| return x == y or _contains(y, x) or (_contains(x, y) and len(y) >= 0.8 * len(x)) |
|
|
|
|
| def _own(value, label, return_claim=None): |
| return {"value": value, "label": label, "return_claim": return_claim} |
|
|
|
|
| def validate_points(ctx, raw_points, *, own, limit, source, allowed=None): |
| """[validated point dicts], [drops]. Validated points are numbered P1… in order (frozen when forwarded).""" |
| out, drops, seen = [], [], set() |
| for i, p in enumerate(raw_points if isinstance(raw_points, list) else []): |
| if len(out) >= limit: |
| drops.append({"source": f"{source}.{i}", "reason": "OVERFLOW", "raw": p}) |
| continue |
| if not isinstance(p, dict): |
| drops.append({"source": f"{source}.{i}", "reason": "SCHEMA: not an object", "raw": p}) |
| continue |
| kind, view, ev = _kind(p.get("kind") or p.get("topic")), _view_of(p), _evidence_of(p) |
| rs = [] |
| if kind is None or kind not in ctx.kinds: |
| rs.append(f"KIND_UNKNOWN: {p.get('kind') or p.get('topic')!r}") |
| else: |
| rs += evidence_reasons(ctx, ev, kind, allowed=None if allowed is None else set(allowed) & set(ctx.offer)) |
| rs += text_reasons(ctx, view, max_words=LIMITS["view"]) |
| if not rs: |
| rs += direction_reasons(ctx, view, ev, own) |
| key = (kind, tuple(sorted(ev))) |
| if not rs and key in seen: |
| rs.append("DUPLICATE: same kind and evidence as an earlier point") |
| if not rs and any(_norm(view) == _norm(o["view"]) for o in out): |
| rs.append("DUPLICATE: the same words as an earlier point") |
| if rs: |
| drops.append({"source": f"{source}.{i}", "reason": "; ".join(rs)[:400], "raw": p}) |
| continue |
| seen.add(key) |
| out.append({"n": len(out) + 1, "src": i + 1, "kind": kind, "evidence": ev, "view": view}) |
| return out, drops |
|
|
|
|
| def validate_thesis(ctx, text, *, own, max_words, source, scope=(), rests=None, named=True, repeat_of=None): |
| """A thesis is judged against the evidence of the points it RESTS ON (``rests``: their validated numbers). A thesis |
| whose points were all removed has nothing under it (run 8d12481bffa6e2e6 published "earnings quality is overstated |
| by adjustments and buyback dependency" after both supporting points were dropped).""" |
| t = _text(text) |
| if not t: |
| return None, [] |
| rs = text_reasons(ctx, t, max_words=max_words, form=False) |
| if rests is not None and not rests: |
| rs.append("THESIS_ORPHANED: " + ("name the points it rests on in thesis_rests_on" if not named |
| else "every point it rests on was removed")) |
| if repeat_of and same_text(t, repeat_of): |
| rs.append("REPEATS_OPENING: the reply restates the opening thesis; say what the exchange changed") |
| if not rs: |
| rs = direction_reasons(ctx, t, list(scope), own, cocite=False) |
| if rs: |
| return None, [{"source": source, "reason": "; ".join(rs)[:400], "raw": t}] |
| return t, [] |
|
|
|
|
| def _scope(points): |
| return [e for p in points or [] for e in p.get("evidence") or []] |
|
|
|
|
| def validate_seat(ctx, role, reply, *, own): |
| """A round-1 seat (neutral / bear / bull) → the argument set and the drops.""" |
| st = (reply or {}).get("structured") or {} |
| pts, d2 = validate_points(ctx, st.get("points"), own=own, limit=LIMITS["points"], source=f"{role}.points") |
| rest = rests_on(st.get("thesis_rests_on"), pts, own="P") |
| thesis, d1 = validate_thesis(ctx, st.get("thesis"), own=own, max_words=LIMITS["thesis"], source=f"{role}.thesis", |
| scope=_scope([p for p in pts if p["n"] in rest]), rests=rest, |
| named=bool(_labels(st.get("thesis_rests_on")))) |
| return {"schema": SCHEMA, "role": role, "format": "v2", "thesis": thesis, "thesis_rests_on": rest if thesis else [], |
| "points": pts}, d1 + d2 |
|
|
|
|
| def validate_rebuttal(ctx, role, reply, *, own, opponent, opponent_role, r1_value, r1_label, opening_thesis=None): |
| """A rebuttal: answers point at the opponent's VALIDATED points (P-labels); the stance is derived from the numbers.""" |
| st = (reply or {}).get("structured") or {} |
| targets = {f"P{p['n']}": p for p in (opponent or {}).get("points") or []} |
| stance = stance_of(r1_value, r1_label, own.get("value"), own.get("label")) |
| if not targets: |
| drops = [{"source": f"{role}.rebuttal", "reason": "NO_TARGETS: the opponent has no validated points", "raw": st}] \ |
| if (st.get("answers") or st.get("thesis")) else [] |
| return {"schema": SCHEMA, "role": role, "format": "v2", "thesis": None, "thesis_rests_on": [], "answers": [], |
| "stance": stance}, drops |
| answers, used, drops = [], set(), [] |
| for i, a in enumerate(st.get("answers") if isinstance(st.get("answers"), list) else []): |
| src = f"{role}.rebuttal.answers.{i}" |
| if len(answers) >= LIMITS["answers"]: |
| drops.append({"source": src, "reason": "OVERFLOW", "raw": a}) |
| continue |
| if not isinstance(a, dict): |
| drops.append({"source": src, "reason": "SCHEMA: not an object", "raw": a}) |
| continue |
| m = re.search(r"(\d+)", str(a.get("to") or "")) |
| lab = f"P{int(m.group(1))}" if m else None |
| tgt = targets.get(lab) |
| verdict = _VERDICT_ALIASES.get(str(a.get("verdict") or "").strip().upper()) |
| ev, view = _evidence_of(a), _view_of(a) |
| rs = [] |
| if tgt is None: |
| rs.append(f"BAD_TARGET: {a.get('to')!r} is not one of the opponent's points") |
| elif lab in used: |
| rs.append("BAD_TARGET: the point was already answered") |
| if verdict is None: |
| rs.append(f"VERDICT_UNKNOWN: {a.get('verdict')!r}") |
| if tgt is not None: |
| |
| rs += evidence_reasons(ctx, ev, None, required=(verdict == "FAILS"), max_ids=LIMITS["answer_evidence"]) or \ |
| cross_reasons(ctx, ev, tgt.get("evidence") or []) |
| rs += text_reasons(ctx, view, max_words=LIMITS["view"]) |
| if not rs: |
| rs += direction_reasons(ctx, view, ev + list((tgt or {}).get("evidence") or []), own) |
| if rs: |
| drops.append({"source": src, "reason": "; ".join(rs)[:400], "raw": a}) |
| continue |
| used.add(lab) |
| answers.append({"n": len(answers) + 1, "src": i + 1, "to": lab, "to_role": opponent_role, "to_kind": tgt["kind"], |
| "to_view": tgt["view"], "to_evidence": list(tgt.get("evidence") or []), "verdict": verdict, |
| "evidence": ev, "view": view}) |
| rest = rests_on(st.get("thesis_rests_on"), answers, own="A", by_target=True) |
| thesis, d1 = validate_thesis(ctx, st.get("thesis"), own=own, max_words=LIMITS["rebuttal_thesis"], |
| source=f"{role}.rebuttal.thesis", scope=_scope([x for x in answers if x["n"] in rest]), |
| rests=rest, named=bool(_labels(st.get("thesis_rests_on"))), repeat_of=opening_thesis) |
| return {"schema": SCHEMA, "role": role, "format": "v2", "thesis": thesis, "thesis_rests_on": rest if thesis else [], |
| "answers": answers, "stance": stance}, d1 + drops |
|
|
|
|
| def stance_of(r1_value, r1_label, value, label): |
| """HOLD / REVISED_RETURN / REVISED_LABEL from the numbers (the typed stance, never authored).""" |
| if r1_label and label and r1_label != label: |
| return "REVISED_LABEL" |
| if r1_value is not None and value is not None and abs(float(value) - float(r1_value)) >= 0.005: |
| return "REVISED_RETURN" |
| return "HOLD" |
|
|
|
|
| def debate_evidence(sets): |
| """The ids the validated seats cited (the PM and the summary may not introduce new evidence).""" |
| return {e for a in (sets or {}).values() if a for key in ("points", "answers") for p in a.get(key) or [] |
| for e in p.get("evidence") or []} |
|
|
|
|
| def validate_pm(ctx, reply, *, own, debate_ids=None): |
| st = (reply or {}).get("structured") or {} |
| if debate_ids is not None and not debate_ids: |
| return {"schema": SCHEMA, "role": "pm", "format": "v2", "thesis": None, "thesis_rests_on": [], "points": []}, \ |
| ([{"source": "pm", "reason": "NO_ALLOWED_EVIDENCE: the validated debate cited no evidence", "raw": st}] if st else []) |
| pts, d2 = validate_points(ctx, st.get("factors") or st.get("points"), own=own, limit=LIMITS["factors"], |
| source="pm.factors", allowed=debate_ids) |
| rest = rests_on(st.get("thesis_rests_on"), pts, own="F") |
| thesis, d1 = validate_thesis(ctx, st.get("thesis"), own=own, max_words=LIMITS["pm_thesis"], source="pm.thesis", |
| scope=_scope([p for p in pts if p["n"] in rest]), rests=rest, |
| named=bool(_labels(st.get("thesis_rests_on")))) |
| return {"schema": SCHEMA, "role": "pm", "format": "v2", "thesis": thesis, "thesis_rests_on": rest if thesis else [], |
| "points": pts}, d1 + d2 |
|
|
|
|
| def validate_summary(ctx, summ, *, own, debate_ids=None, debate_texts=()): |
| """The executive summary: a figure-free text and view-first items citing only debate evidence, each in the |
| summary's own words (a verbatim copy of a debate object would be printed twice).""" |
| summ = summ or {} |
| drops, items = [], {} |
| if debate_ids is not None and not debate_ids: |
| return {"schema": SCHEMA, "role": "summary", "format": "v2", "thesis": None, "items_thesis": [], "items_risks": []}, \ |
| ([{"source": "summary", "reason": "NO_ALLOWED_EVIDENCE: the validated debate cited no evidence", "raw": summ}] |
| if summ else []) |
| for key in ("thesis", "risks"): |
| out = [] |
| for i, it in enumerate(summ.get(key) if isinstance(summ.get(key), list) else []): |
| src = f"{key}.{i}" |
| if len(out) >= LIMITS["items"]: |
| break |
| if not isinstance(it, dict): |
| drops.append({"source": src, "reason": "SCHEMA: not an object", "raw": it}) |
| continue |
| ev, view = _evidence_of(it), _view_of(it) |
| rs = evidence_reasons(ctx, ev, None, allowed=(set(debate_ids) if debate_ids is not None else None)) + \ |
| text_reasons(ctx, view, max_words=LIMITS["item"]) |
| if not rs and any(_norm(view) == _norm(d) for d in debate_texts or ()): |
| rs.append("COPIES_DEBATE: write the reason in the summary's own words") |
| if not rs: |
| rs = direction_reasons(ctx, view, ev, own) |
| if rs: |
| drops.append({"source": src, "reason": "; ".join(rs)[:400], "raw": it}) |
| continue |
| out.append({"n": len(out) + 1, "kind": None, "evidence": ev, "view": view}) |
| items[key] = out |
| alli = items["thesis"] + items["risks"] |
| text, d0 = validate_thesis(ctx, summ.get("summary"), own=own, max_words=LIMITS["summary"], source="summary", |
| scope=_scope(alli), rests=[1] if alli else []) |
| return {"schema": SCHEMA, "role": "summary", "format": "v2", "thesis": text, "items_thesis": items["thesis"], |
| "items_risks": items["risks"]}, d0 + drops |
|
|
|
|
| |
| def _perimeter_note(state): |
| ev = ((state.historical.get("perimeter") or {}).get("event") or {}) |
| return f"before the separation of {ev.get('business')}" if ev.get("business") else None |
|
|
|
|
| _READER = { |
| "REVENUE_TTM": ("trailing revenue ", ""), "OPERATING_MARGIN_TTM": ("trailing operating margin ", ""), |
| "NET_MARGIN_TTM": ("trailing net margin ", ""), "NET_INCOME_TTM": ("trailing net income ", ""), |
| "OPERATING_INCOME_TTM": ("trailing operating income ", ""), "EPS_ADJUSTED_TTM": ("trailing adjusted EPS ", ""), |
| "EPS_GAAP_TTM": ("trailing reported EPS ", ""), |
| "BEAT_RATE": ("EPS beat estimates in ", " of recent quarters"), |
| "ADJ_GAAP_BRIDGE": ("adjusted EPS at ", " times reported EPS (trailing)"), |
| "FORECAST_REVENUE": ("the engine's revenue forecast for the next twelve months ", ""), |
| "FORECAST_REVENUE_GROWTH": ("the engine's forecast revenue growth for the next twelve months ", ""), |
| "FORECAST_OPERATING_MARGIN": ("the engine's operating margin forecast for the next twelve months ", ""), |
| "FORECAST_EPS": ("the engine's EPS forecast for the next twelve months ", ""), |
| "CONSENSUS_EPS_FY1": ("consensus EPS for the current fiscal year ", ""), |
| "CONSENSUS_EPS_FY2": ("consensus EPS for the next fiscal year ", ""), |
| "CONSENSUS_REVENUE_FY1": ("consensus revenue for the current fiscal year ", ""), |
| "CONSENSUS_REVENUE_NTM": ("consensus revenue for the next twelve months ", ""), |
| "ANALYST_TARGET": ("mean analyst price target ", ""), "PRICE": ("share price ", ""), |
| "MARKET_CAP": ("market capitalisation ", ""), "DIVIDEND_YIELD": ("dividend yield ", ""), |
| "PE_FORWARD": ("forward P/E ", ""), "PEG": ("forward PEG ", ""), |
| "INTRINSIC_VALUE": ("story value ", ""), "MECHANICAL_VALUE": ("mechanical DCF value ", ""), |
| "MARKET_COMPOSITE": ("market composite ", ""), "LIFECYCLE_STAGE": ("life-cycle stage ", ""), |
| "LIFECYCLE_FORWARD_GROWTH": ("consensus revenue growth for the next fiscal year ", ""), |
| "LIFECYCLE_TRAILING_GROWTH": ("trailing revenue growth ", ""), |
| "BE_REVENUE": ("revenue the price needs at maturity ", ""), "BE_MARGIN": ("operating margin the price needs ", ""), |
| "STORY_MARGIN": ("story maturity operating margin ", ""), "STORY_TAM": ("story market size at maturity ", ""), |
| "STORY_SHARE": ("story peak market share ", ""), "STORY_REVENUE_MATURITY": ("story revenue at maturity ", ""), |
| "STORY_WACC": ("initial cost of capital ", ""), "STORY_WACC_STABLE": ("terminal (stable-period) cost of capital ", ""), |
| "STORY_TERMINAL_GROWTH": ("terminal growth ", ""), "STORY_FAILURE": ("failure probability ", ""), |
| "STORY_START_MARGIN": ("continuing-company starting operating margin ", ""), |
| "STORY_START_REVENUE": ("continuing-company starting revenue ", ""), |
| "DMD_BETA": ("bottom-up beta ", ""), "DMD_MARGIN": ("industry operating margin ", " (Damodaran industry data)"), |
| "DMD_INDUSTRY": ("industry ", ""), "DMD_ERP": ("equity risk premium ", ""), "DMD_S2C": ("industry sales-to-capital ", ""), |
| "SEGMENT_COUNT_CONTINUING": ("", " businesses in the continuing company"), "CEO": ("chief executive ", ""), |
| "CFO": ("chief financial officer ", ""), "SMA200": ("200-day average ", ""), "MOMENTUM": ("momentum ", ""), |
| "RETURN_12M_PHRASE": ("shares ", ""), "RETURN_3M_PHRASE": ("shares ", ""), |
| } |
| _TRAILING = ("REVENUE_TTM", "OPERATING_MARGIN_TTM", "NET_MARGIN_TTM", "NET_INCOME_TTM", "OPERATING_INCOME_TTM", |
| "EPS_ADJUSTED_TTM", "EPS_GAAP_TTM", "ADJ_GAAP_BRIDGE", "LIFECYCLE_TRAILING_GROWTH") |
|
|
|
|
| def _fact(R, cid, state, cat): |
| """Fragments for one fact: reader label + the claim fragment (+ perimeter qualifier).""" |
| cl = cat.get(cid) |
| if cl is None: |
| return [] |
| if getattr(cl, "stype", None) == C.GUIDANCE: |
| v = cl.value if isinstance(cl.value, dict) else {} |
| rows = ((state.guidance.get("company") or {}).get("metrics")) or [] |
| try: |
| label = str((rows[int(cid.split("_")[1]) - 1] or {}).get("label") or "").strip() |
| except (ValueError, IndexError): |
| label = "" |
| label = label or str(v.get("concept") or "guidance").lower().replace("_", " ") |
| basis = C.guidance_basis_label(cat, cid) |
| basis = None if (basis and basis.lower() in label.lower()) else basis |
| tail = f" ({v.get('period_label')}{', ' + basis if basis else ''})" if v.get("period_label") else \ |
| (f" ({basis})" if basis else "") |
| return [P.Fragment(f"company guidance, {label}: "), R.c(cid), P.Fragment(tail)] |
| pre, post = _READER.get(cid) or ((re.sub(r"\s*\([^)]*\)", "", cl.label or cid).strip() + " "), "") |
| if not post and pre.endswith(" "): |
| pre = pre.rstrip() + ": " |
| note = _perimeter_note(state) if cid in _TRAILING else None |
| if note: |
| post = post + f" ({note})" |
| return [P.Fragment(pre), R.c(cid), P.Fragment(post)] |
|
|
|
|
| |
| |
| |
| |
| _SUBJECT = {"PRICE": ("The price ", "PREDICATE"), "ANALYST_TARGET": ("The mean analyst target ", "PREDICATE"), |
| "STORY_START_MARGIN": ("The continuing company's operating margin ", "PREDICATE"), |
| "FORECAST_EPS": ("The engine's EPS forecast for the next twelve months is ", "CLAUSE"), |
| "FORECAST_REVENUE": ("The engine's revenue forecast for the next twelve months is ", "CLAUSE"), |
| "BE_MARGIN": ("The operating margin today's price needs at the story's revenue is ", "CLAUSE")} |
|
|
|
|
| def _relation(R, rid, cat, rels, compact=False, cases_cited=False): |
| """A relation as one sentence: the price relations in their predicate form after "The price", else the clause (or, |
| when the report has already printed it in full, its subject and predicate).""" |
| rel = rels.get(rid) or {} |
| if rel.get("lhs") == "PRICE" and cat.get(f"{rid}__PREDICATE") is not None: |
| return [P.Fragment("The price "), R.c(f"{rid}__PREDICATE"), P.Fragment(".")] |
| if rid == "REL_EXPECTATIONS_STORY" and cat.get(f"{rid}__COMPACT") is not None: |
| |
| return [R.c(f"{rid}__COMPACT", cap=True), P.Fragment("." if cases_cited else |
| " (the valuation section sets out the revenue and the margin " |
| "the price needs).")] |
| subj, form = _SUBJECT.get(rel.get("lhs"), (None, None)) |
| if compact and subj and cat.get(f"{rid}__{form}") is not None: |
| return [P.Fragment(subj), R.c(f"{rid}__{form}"), P.Fragment(".")] |
| return [R.c(rid, cap=True), P.Fragment(".")] |
|
|
|
|
| def render_evidence(R, ids, state, cat, rels, seen=None): |
| """Evidence fragments: relations as sentences; the remaining facts (minus any relation's own operands) as one. |
| ``seen`` is the report's evidence registry: a relation printed in full earlier in the report is printed compactly.""" |
| ids = list(dict.fromkeys(ids)) |
| cited_rels = [e for e in ids if cat.get(e) is not None and cat[e].stype == C.REL] |
| operands = {x for rid in cited_rels for x in ((rels.get(rid) or {}).get("lhs"), (rels.get(rid) or {}).get("rhs")) if x} |
| out = [] |
| case_ids = {"REL_BREAK_EVEN_REVENUE_STORY", "REL_BREAK_EVEN_MARGIN_STORY", "REL_BREAK_EVEN_REVENUE_GROWTH"} |
| for rid in cited_rels: |
| out += ([P.Fragment(" ")] if out else []) + _relation(R, rid, cat, rels, compact=seen is not None and rid in seen, |
| cases_cited=bool(case_ids & set(ids))) |
| if seen is not None: |
| seen.add(rid) |
| facts = [e for e in ids if e not in cited_rels and e not in operands] |
| parts = [_fact(R, e, state, cat) for e in facts] |
| parts = [p for p in parts if p] |
| if parts: |
| sent = [] |
| for i, p in enumerate(parts): |
| sent += ([P.Fragment("; ")] if i else []) + p |
| first = sent[0] |
| if first.text and not first.claim_id: |
| sent[0] = P.Fragment(first.text[:1].upper() + first.text[1:]) |
| out += ([P.Fragment(" ")] if out else []) + sent + [P.Fragment(".")] |
| return out |
|
|
|
|
| def evidence_plain(ids, cat, rels): |
| """The evidence as plain text (for forwarding to the next round and for the semantic gate).""" |
| bits = [] |
| for e in dict.fromkeys(ids): |
| cl = cat.get(e) |
| if cl is None: |
| continue |
| if cl.stype == C.REL: |
| bits.append(str(cl.rendered)) |
| else: |
| pre, post = _READER.get(e) or ((re.sub(r"\s*\([^)]*\)", "", cl.label or e).strip() + " "), "") |
| pre = pre.rstrip() + ": " if not post and pre.endswith(" ") else pre |
| bits.append(f"{pre}{cl.rendered}{post}".strip()) |
| return "; ".join(bits) |
|
|
|
|
| def forward_text(argset, cat, rels, *, indexed=False, label=None): |
| """What the next round reads: the VALIDATED thesis and points only (never raw prose or risk flags).""" |
| if not argset: |
| return "" |
| lines = [] |
| if argset.get("thesis"): |
| lines.append(f" Thesis: {P.TOKEN.sub('its own return', argset['thesis'])}") |
| for p in argset.get("points") or []: |
| tag = f"P{p['n']} " if indexed else "- " |
| ev = evidence_plain(p.get("evidence") or [], cat, rels) |
| lines.append(f" {tag}[{KIND_LABEL.get(p.get('kind'), 'Point')}] {p['view']}" + (f" (evidence: {ev})" if ev else "")) |
| head = f"{label}:\n" if label else "" |
| return head + "\n".join(lines) |
|
|
|
|
| def evidence_menu(ctx): |
| """The author's evidence menu (replaces the token header): ids by kind, with values and meanings.""" |
| lines = ["EVIDENCE MENU — cite these IDs in \"evidence\"; the report prints each one's figure and wording itself.", |
| "ID | value | meaning"] |
| for cid in sorted(ctx.offer): |
| cl = ctx.cat[cid] |
| meaning = re.sub(r"\s*\([^)]*complete relation[^)]*\)", "", cl.label or "") |
| basis = C.guidance_basis_label(ctx.cat, cid) if cl.stype == C.GUIDANCE else None |
| meaning += f" [{basis} basis]" if basis else "" |
| val = f"\"{cl.rendered}\"" if cl.stype == C.REL else str(cl.rendered) |
| lines.append(f" {cid} | {val} | {meaning}") |
| lines.append(f"KINDS (a label for what the point is about): {', '.join(ctx.kinds)}. Every point cites 1-3 IDs from this " |
| "menu that support it; two facts in one point need the relation that connects them; the two break-even " |
| "solves are alternatives, so cite REL_EXPECTATIONS_STORY to discuss them together.") |
| return "\n".join(lines) |
|
|
|
|
| |
| GATE_VERDICTS = ("OK", "UNSUPPORTED", "CONTRADICTED", "MISATTRIBUTED", "EVIDENCE_MISUSE") |
|
|
| |
| |
| |
| |
| GATE_RUBRIC = """You check an equity report before it is published, to the standard its reviewers apply. Each OBJECT below is |
| a short judgement written by an analyst persona, printed next to the EVIDENCE it cites (figures rendered by the system |
| from verified data). An investor reads the words as claims. Decide for EACH object whether an investor reading it would |
| be misled or the report embarrassed. |
| |
| verdict: |
| - CONTRADICTED: the evidence, the FACTS or the REST OF THE REPORT contradict it: a wrong direction, or a universal or |
| absolute claim ("no company sustains ...", "impossible", "never", a "premium" or "discount" against peers) that any |
| figure printed in the report refutes. Check such claims against every table in the report, not only the object's |
| evidence. |
| - UNSUPPORTED: it asserts as FACT something material that no evidence, fact or call line shows: a cause, motive or |
| intention read from a price series (momentum or a moving average do not show what investors think or plan); a claim |
| about a quantity none of its evidence measures (a comparison with consensus for a metric that has no consensus |
| figure); closeness or distance ("near", "far above") to a figure it does not cite; events, trends or history not in |
| the brief. |
| - MISATTRIBUTED: it presents the model's own assumptions or forecasts as the company's plans or guidance, or says |
| management, the company or the call said something the call excerpt does not show that speaker saying. |
| - EVIDENCE_MISUSE: it draws a conclusion its evidence cannot carry: figures of a different period or business perimeter, |
| an industry aggregate as the company's own figure, ownership counts as conviction. |
| - OK: honest opinion or judgement that is hedged or plainly a view ("looks demanding", "appears attainable", "if margins |
| stall ...") and is consistent with the evidence and the report. Do not remove a view only because it is strongly |
| worded or because you would weigh the evidence differently. |
| One short reason per object. Unless the verdict is OK, "quote" gives the exact words of that object's own "text" the |
| verdict is about.""" |
|
|
| |
| |
| |
| GATE_THESIS_RUBRIC = """You check an equity report before it is published. Each OBJECT below is an analyst persona's THESIS: a short |
| summary of its case. "evidence" lists the figures the system printed for the author's validated points; the author may |
| also draw on any figure printed elsewhere in the REPORT. Judge ONLY the thesis text. Summarising and characterising is |
| a thesis's job. |
| |
| verdict: |
| - CONTRADICTED: a figure in its evidence, the FACTS or the REPORT contradicts it: a wrong direction (upside against a |
| value the price already exceeds; cheap where every anchor shows a premium), or an absolute claim that a printed figure |
| refutes. |
| - UNSUPPORTED: it asserts a specific material fact that nothing printed in the report supports: an event, a motive or |
| intention nobody states, a problem the report's figures do not show, or plans for the company that its guidance does |
| not contain. |
| - MISATTRIBUTED: it attributes words or plans to management or the call that the call excerpt does not show. |
| - OK: a summary judgement or characterisation consistent with the report, including general claims its figures bear |
| out (a high-quality franchise where margins are high, capital returns where return guidance is printed, a demanding |
| valuation where the price sits above value). Strong wording alone is not a reason to remove. |
| Be strict about contradictions and invented specifics, and lenient about characterisation. One short reason per object. |
| Unless the verdict is OK, "quote" gives the exact words of that thesis's own "text".""" |
|
|
|
|
| def _rest_objects(role, a): |
| """The validated objects a thesis rests on (all items for the summary; all points when no list was recorded).""" |
| if role == "summary": |
| return (a.get("items_thesis") or []) + (a.get("items_risks") or []) |
| pool = (a.get("points") or []) + (a.get("answers") or []) |
| if "thesis_rests_on" not in a: |
| return pool |
| want = set(a.get("thesis_rests_on") or []) |
| return [p for p in pool if p.get("n") in want] |
|
|
|
|
| def gate_objects(sets, cat, rels, phase="objects"): |
| """[{id, author, text, evidence}]: phase "objects" = every point, item and answer; phase "theses" = every thesis, |
| with the surviving points it rests on as its evidence (judged after the objects, so a thesis is never approved |
| over points the same gate removed).""" |
| out = [] |
| for role, a in (sets or {}).items(): |
| if not a: |
| continue |
| if phase == "theses": |
| if a.get("thesis"): |
| |
| |
| ids = list(dict.fromkeys(e for p in _rest_objects(role, a) for e in p.get("evidence") or [])) |
| out.append({"id": f"{role}:thesis", "author": role, "text": P.TOKEN.sub("its own return", a["thesis"]), |
| "evidence": evidence_plain(ids, cat, rels) or "(no evidence)"}) |
| continue |
| for key in ("points", "items_thesis", "items_risks"): |
| for p in a.get(key) or []: |
| out.append({"id": f"{role}:{key}:{p['n']}", "author": role, "text": p["view"], |
| "evidence": evidence_plain(p.get("evidence") or [], cat, rels)}) |
| for i, x in enumerate(a.get("answers") or []): |
| out.append({"id": f"{role}:answers:{i + 1}", "author": role, |
| "text": f"(answering the {x['to_role']}'s point \"{x['to_view']}\", verdict {x['verdict']}) {x['view']}", |
| "evidence": evidence_plain(x.get("evidence") or [], cat, rels)}) |
| return out |
|
|
|
|
| def gate_prompt(objects, facts_text, excerpt, report_text=None, phase="objects"): |
| import json as _json |
| return ((GATE_THESIS_RUBRIC if phase == "theses" else GATE_RUBRIC) |
| + "\n\nFACTS (verified, rendered by the system):\n" + facts_text |
| + ("\n\nTHE REST OF THE REPORT (printed by the system: every figure the reader sees beside these objects):\n" |
| + report_text[:16000] if report_text else "") |
| + "\n\nCALL EXCERPT (the end of the latest earnings call; speaker names precede their words):\n" |
| + (excerpt or "(none)")[:6000] |
| + "\n\nOBJECTS:\n" + "\n".join(_json.dumps(o, ensure_ascii=False) for o in objects) |
| + '\n\nRespond with ONLY this JSON: {"verdicts": [{"id": "<object id>", "verdict": "OK|UNSUPPORTED|CONTRADICTED|' |
| 'MISATTRIBUTED|EVIDENCE_MISUSE", "quote": "<unless OK: the exact words of this object\'s text>", ' |
| '"reason": "<one short sentence>"}]} with one entry per object.') |
|
|
|
|
| def facts_text(ctx): |
| """The verified facts a gate reads: every offered claim, the relations, the assessments and the perimeter.""" |
| lines = [f"- {cid}: {ctx.cat[cid].rendered} — {ctx.cat[cid].label}" for cid in sorted(ctx.offer)] |
| for cid, a in ((ctx.rules.assessments or {}).items()): |
| lines.append(f"- assessment of {cid}: {a.get('assessment')} ({a.get('basis')})") |
| note = _perimeter_note(ctx.state) |
| if note: |
| lines.append(f"- trailing (twelve-month) figures are {note}; forecasts and the story describe the continuing company") |
| return "\n".join(lines) |
|
|
|
|
| def _attributed(quote, text): |
| """A non-OK verdict names the words it is about; they must be the object's own (typed lineage of the verdict).""" |
| q = _norm(quote) |
| return bool(q) and q in _norm(text) |
|
|
|
|
| def apply_gate(sets, verdicts, phase="objects", unattributed=None): |
| """Remove every object of this phase the gate did not mark OK (a missing verdict is not OK); objects of the other |
| phase are untouched. A non-OK verdict whose quote is not in the object's own text is about something else: the |
| object is kept and its id is added to ``unattributed`` (the caller asks once more for it alone). Returns |
| (sets, drops).""" |
| vmap = {} |
| for v in verdicts or []: |
| if isinstance(v, dict) and isinstance(v.get("id"), str): |
| vmap[v["id"]] = (str(v.get("verdict") or "").upper(), str(v.get("reason") or "")[:200], v.get("quote")) |
| unattributed = unattributed if unattributed is not None else [] |
|
|
| def verdict(oid, text): |
| vd = vmap.get(oid, ("MISSING", "no verdict", None)) |
| if vd[0] in ("OK", "MISSING") or _attributed(vd[2], text): |
| return vd |
| unattributed.append(oid) |
| return ("OK", "", None) |
| drops, out = [], {} |
| for role, a in (sets or {}).items(): |
| if not a: |
| out[role] = a |
| continue |
| a = dict(a) |
| if phase == "theses": |
| if a.get("thesis"): |
| vd = verdict(f"{role}:thesis", a["thesis"]) |
| if vd[0] != "OK": |
| drops.append({"source": f"{role}.thesis", "reason": f"SEMANTIC_{vd[0]}: {vd[1]}", "raw": a["thesis"]}) |
| a["thesis"] = None |
| if "thesis_rests_on" in a: |
| a["thesis_rests_on"] = [] |
| out[role] = a |
| continue |
| for key in ("points", "items_thesis", "items_risks"): |
| kept = [] |
| for p in a.get(key) or []: |
| vd = verdict(f"{role}:{key}:{p['n']}", p["view"]) |
| if vd[0] == "OK": |
| kept.append(p) |
| else: |
| drops.append({"source": f"{role}.{key}.P{p['n']}", "reason": f"SEMANTIC_{vd[0]}: {vd[1]}", "raw": p}) |
| if key in a: |
| a[key] = kept |
| if a.get("answers"): |
| kept = [] |
| for i, x in enumerate(a["answers"]): |
| vd = verdict(f"{role}:answers:{i + 1}", x["view"]) |
| if vd[0] == "OK": |
| kept.append(x) |
| else: |
| drops.append({"source": f"{role}.answers.A{i + 1}", "reason": f"SEMANTIC_{vd[0]}: {vd[1]}", "raw": x}) |
| a["answers"] = kept |
| out[role] = a |
| out, d2 = drop_orphan_answers(out) |
| return out, drops + d2 |
|
|
|
|
| def drop_orphan_answers(sets): |
| """An answer whose target point was removed no longer answers anything the reader can see: drop it (recorded).""" |
| out, drops = dict(sets or {}), [] |
| for role, a in list(out.items()): |
| if not a or not a.get("answers"): |
| continue |
| tgt = out.get(a["answers"][0].get("to_role")) or {} |
| visible = {f"P{p['n']}" for p in tgt.get("points") or []} |
| kept = [x for x in a["answers"] if x["to"] in visible] |
| drops += [{"source": f"{role}.answers.{x.get('to')}", "reason": "TARGET_REMOVED: the point it answers is not " |
| "published", "raw": x} for x in a["answers"] if x["to"] not in visible] |
| if len(kept) != len(a["answers"]): |
| out[role] = dict(a, answers=kept) |
| return out, drops |
|
|
|
|
| def orphan_theses(sets): |
| """Drop every thesis whose points were all removed after it validated (typed revalidation or the gate); keep the |
| rest list to the survivors (after dropping answers to removed points). Returns (sets, drops).""" |
| sets, drops = drop_orphan_answers(sets) |
| out = {} |
| for role, a in (sets or {}).items(): |
| if not a or not a.get("thesis"): |
| out[role] = a |
| continue |
| a = dict(a) |
| if role == "summary": |
| alive = bool((a.get("items_thesis") or []) + (a.get("items_risks") or [])) |
| elif "thesis_rests_on" in a: |
| pool = {p.get("n") for p in (a.get("points") or []) + (a.get("answers") or [])} |
| a["thesis_rests_on"] = [n for n in a.get("thesis_rests_on") or [] if n in pool] |
| alive = bool(a["thesis_rests_on"]) |
| else: |
| alive = True |
| if not alive: |
| drops.append({"source": f"{role}.thesis", "reason": "THESIS_ORPHANED: every point it rests on was removed", |
| "raw": a["thesis"]}) |
| a["thesis"] = None |
| out[role] = a |
| return out, drops |
|
|
|
|
| |
| _NOT_REPAIRABLE = ("INTERNAL_ERROR", "OVERFLOW", "SEMANTIC_", "REBUTTAL_NO_NUMBERS", "REVALIDATION", "NO_TARGETS", |
| "NO_ALLOWED_EVIDENCE", "TARGET_REMOVED") |
|
|
|
|
| def repairable(drops): |
| """The drops an author can fix by re-authoring (typed reasons); never a crash, an overflow or a gate verdict.""" |
| return [d for d in drops or [] if d.get("reason") and not str(d["reason"]).startswith(_NOT_REPAIRABLE)] |
|
|
|
|
| def repair_prompt(original_prompt, structured, drops): |
| """The original task, the author's previous structured reply and each removed object with the rule it broke. The |
| reply to this is validated exactly like the first one; code never edits an object.""" |
| import json as _json |
| lines = [f"- {d.get('source')}: {str(d.get('reason'))[:300]}\n was: " |
| + _json.dumps(d.get("raw"), ensure_ascii=False, default=str)[:400] for d in drops] |
| return (original_prompt + "\n\n=== REPAIR (one attempt) ===\nYOUR PREVIOUS REPLY (its structured part):\n" |
| + _json.dumps(structured or {}, ensure_ascii=False, default=str)[:6000] |
| + "\n\nTHESE OBJECTS WERE REMOVED, WITH THE RULE EACH BROKE:\n" + "\n".join(lines) |
| + "\n\nReturn the COMPLETE JSON again, in the same format as before. Keep every number exactly as before. " |
| "Rewrite each removed object so it meets the rule named, or leave it out; keep the objects that were not " |
| "removed as they were, and renumber thesis_rests_on to the positions in your new reply.") |
|
|
|
|
| def object_count(argset): |
| """How many publishable objects an argument set holds.""" |
| a = argset or {} |
| return (1 if a.get("thesis") else 0) + sum(len(a.get(k) or []) for k in ("points", "answers", "items_thesis", |
| "items_risks")) |
|
|
|
|
| |
| SEAT_SETS = {"neutral": ("neutral",), "bear": ("bear_open", "bear"), "bull": ("bull_open", "bull")} |
| SEAT_ROLES = tuple(r for roles in SEAT_SETS.values() for r in roles) |
|
|
|
|
| def supporting_count(argset): |
| """The supporting arguments an argument set holds: its points and answers (a thesis is a conclusion, not support).""" |
| a = argset or {} |
| return len(a.get("points") or []) + len(a.get("answers") or []) |
|
|
|
|
| def seat_evidence_state(sets, drops=()): |
| """{seat: {accepted, rejected}}: the supporting arguments each seat still has after typed validation and the |
| semantic gate, and how many were removed (first attempts superseded by a repair are not counted).""" |
| out = {} |
| for seat, roles in SEAT_SETS.items(): |
| rej = 0 |
| for d in drops or (): |
| src = str((d or {}).get("source") or "") |
| if src.startswith(("attempt1:", "repair")) or src.endswith(".thesis"): |
| continue |
| if src.split(".")[0].split("_")[0] == seat: |
| rej += 1 |
| out[seat] = {"accepted": sum(supporting_count((sets or {}).get(r)) for r in roles), "rejected": rej} |
| return out |
|
|
|
|
| def evidence_state_line(seat, st): |
| """One line of the PM's brief: the seat's evidence state, stated explicitly when nothing survived.""" |
| n, m = int((st or {}).get("accepted") or 0), int((st or {}).get("rejected") or 0) |
| removed = f"{m} removed by evidence validation" if m else "none removed" |
| if n == 0: |
| return (f" EVIDENCE STATE: {seat.capitalize()} seat has no accepted supporting arguments ({removed}). Its scenario " |
| "return stands without accepted support.") |
| return f" EVIDENCE STATE: {n} supporting argument{'s' if n != 1 else ''} accepted ({removed})." |
|
|
|
|
| def repair_keeps(first, repaired): |
| """A repair is kept only when it loses nothing in ANY class: the thesis (and summary text) stays if it was there, |
| and no list of points, answers or items gets shorter (a total count could trade a thesis for a point).""" |
| a, b = first or {}, repaired or {} |
| if a.get("thesis") and not b.get("thesis"): |
| return False |
| if any(len(b.get(k) or []) < len(a.get(k) or []) for k in ("points", "answers", "items_thesis", "items_risks")): |
| return False |
| return object_count(b) > object_count(a) |
|
|
|
|
| |
| def revalidate(ctx, argset, *, own): |
| """Re-run the typed checks on a stored argument set (compose and the report audit). Returns (argset, drops).""" |
| if not argset: |
| return argset, [] |
| a, drops = dict(argset), [] |
| role = a.get("role") or "?" |
| for key in ("points", "items_thesis", "items_risks"): |
| if key not in a: |
| continue |
| kept = [] |
| for p in a.get(key) or []: |
| kind = p.get("kind") |
| rs = (evidence_reasons(ctx, p.get("evidence") or [], kind) if kind else |
| evidence_reasons(ctx, p.get("evidence") or [], None, allowed=None)) + \ |
| text_reasons(ctx, p.get("view"), max_words=LIMITS["view"]) |
| if not rs: |
| rs = direction_reasons(ctx, p["view"], p.get("evidence") or [], own) |
| (drops.append({"source": f"{role}.{key}.{p.get('n')}", "reason": "REVALIDATION: " + "; ".join(rs)[:380], |
| "raw": p}) if rs else kept.append(p)) |
| a[key] = kept |
| if a.get("answers"): |
| kept = [] |
| for x in a["answers"]: |
| rs = (evidence_reasons(ctx, x.get("evidence") or [], None, required=(x.get("verdict") == "FAILS"), |
| max_ids=LIMITS["answer_evidence"]) or cross_reasons(ctx, x.get("evidence") or [], |
| x.get("to_evidence") or [])) + \ |
| text_reasons(ctx, x.get("view"), max_words=LIMITS["view"]) |
| if not rs: |
| rs = direction_reasons(ctx, x["view"], (x.get("evidence") or []) + (x.get("to_evidence") or []), own) |
| (drops.append({"source": f"{role}.answers.{x.get('to')}", "reason": "REVALIDATION: " + "; ".join(rs)[:380], |
| "raw": x}) if rs else kept.append(x)) |
| a["answers"] = kept |
| if a.get("thesis"): |
| lim = LIMITS["pm_thesis"] if role == "pm" else LIMITS["summary"] if role == "summary" else LIMITS["thesis"] |
| rest = _rest_objects(role, a) |
| rests = None if (role != "summary" and "thesis_rests_on" not in a) else [p.get("n") or 1 for p in rest] |
| t, d = validate_thesis(ctx, a["thesis"], own=own, max_words=lim, source=f"{role}.thesis", scope=_scope(rest), |
| rests=rests) |
| a["thesis"], drops = t, drops + d |
| if "thesis_rests_on" in a: |
| a["thesis_rests_on"] = [p.get("n") for p in rest] if t else [] |
| return a, drops |
|
|
