| """UserReport, the canonical renderer, the final claim audit, and the InternalQARecord. |
| |
| One structured report is built once, from the frozen AnalysisState and the ingested prose. HTML |
| and markdown are two serialisations of the same fragment list, so they cannot say different |
| things. Every number is a claim fragment that resolves to one state path; the audit re-resolves |
| each fragment against the frozen state and rejects any material number that is not a claim. |
| The audited fragments ARE the delivered document: the web page, the download and the saved report |
| all use these bytes. Internal QA (validator ids, lineage, rejected sentences, engine logs) lives in |
| a separate record and never enters the user report. |
| """ |
| from __future__ import annotations |
|
|
| from collections.abc import Mapping |
| import html as _html |
| import re |
| from dataclasses import dataclass, field |
|
|
| import catalog as C |
| from prose import BENIGN, DENIAL, IDENTIFIER, MATERIAL_NUMBER, Fragment |
|
|
| |
| |
| |
|
|
|
|
| @dataclass |
| class Block: |
| kind: str |
| section: str |
| content: object = field(default_factory=list) |
| title: str | None = None |
| css: str | None = None |
|
|
| def units(self): |
| """Fragment sequences that read as one phrase: a paragraph, one bullet, one table row. |
| Label adjacency is judged inside a unit, never across table cells or bullets.""" |
| if self.kind in ("para", "callout", "heading"): |
| return [list(self.content)] |
| if self.kind in ("bullets", "points"): |
| return [list(item) for item in self.content] |
| if self.kind in ("table", "kv", "tiles"): |
| return [[f for cell in row for f in cell] for row in self.content.get("rows", [])] |
| return [] |
|
|
| def fragments(self): |
| if self.kind in ("para", "callout", "heading"): |
| return list(self.content) |
| if self.kind in ("bullets", "points"): |
| return [f for item in self.content for f in item] |
| if self.kind in ("table", "kv", "tiles"): |
| out = [Fragment(h) for h in self.content.get("head", [])] |
| for row in self.content.get("rows", []): |
| for cell in row: |
| out.extend(cell) |
| return out |
| return [] |
|
|
|
|
| @dataclass |
| class UserReport: |
| ticker: str |
| company: str |
| as_of: str |
| version: str |
| blocks: list |
| claims: dict |
| limitations: list |
| arguments: dict = field(default_factory=dict) |
|
|
| def fragments(self): |
| return [f for b in self.blocks for f in b.fragments()] |
|
|
|
|
| class _R: |
| """Fragment helpers bound to a frozen state and its catalog. The renderer never computes a |
| number: it formats a state value or it writes words.""" |
|
|
| def __init__(self, state, cat): |
| self.s, self.cat, self.used = state, cat, {} |
| self.seen_evidence = set() |
| self.authored_seen = set() |
|
|
| def c(self, cid, cap=False): |
| cl = self.cat.get(cid) |
| if cl is None: |
| return Fragment("unavailable") |
| self.used[cid] = cl |
| txt = cl.rendered |
| if cap and cl.stype == C.REL and txt: |
| txt = txt[0].upper() + txt[1:] |
| return Fragment(txt, cid, cl.value) |
|
|
| def p(self, path, stype, label): |
| cl = C.path_claim(self.s, path, stype, label) |
| if cl.rendered is None: |
| return Fragment("—") |
| self.used[cl.claim_id] = cl |
| return Fragment(cl.rendered, cl.claim_id, cl.value) |
|
|
| @staticmethod |
| def t(text): |
| return Fragment(text) |
|
|
|
|
| def _para(section, *frags, css=None, title=None): |
| return Block("para", section, [f if isinstance(f, Fragment) else Fragment(str(f)) for f in frags], title=title, css=css) |
|
|
|
|
| |
| def historical_quality(s): |
| """HistoricalMetricQuality of the trailing figures (reconcile.ttm_quality). A state built before it existed is read |
| conservatively: any unchecked quarter makes the series at most partially reconciled.""" |
| rec = s.historical.get("reconciliation") or {} |
| q = rec.get("ttm_quality") |
| if isinstance(q, Mapping) and q.get("status"): |
| return q |
| st, pe = rec.get("status"), str(rec.get("period_end") or "") |
| import datetime as _dt |
| try: |
| lo = (_dt.date.fromisoformat(pe[:10]) - _dt.timedelta(days=360)).isoformat() |
| except ValueError: |
| lo = "" |
| unv = [e for e in rec.get("unverified_quarters") or [] if str(e) > lo] |
| status = ("SEC_RECONCILED" if st in ("RECONCILED", "SEC_ONLY") and not unv else |
| "PARTIALLY_RECONCILED" if st == "RECONCILED" else "PROVIDER_ONLY" if st and st != "NO_STATEMENTS" else "UNVERIFIED") |
| return {"status": status, "verified_through": None, "unverified_periods": unv, "source_mix": [], "quarters": []} |
|
|
|
|
| def ttm_header(q, separated=False): |
| """The trailing column's label from its quality state: only SEC_RECONCILED reads as reconciled without qualification.""" |
| st, unv = q.get("status"), list(q.get("unverified_periods") or []) |
| ends = " and ".join(unv) |
| if st == "SEC_RECONCILED": |
| parts = ["reconciled to SEC filings"] |
| elif st == "PARTIALLY_RECONCILED": |
| parts = [f"reconciled to SEC filings through {q['verified_through']}" if q.get("verified_through") |
| else "partly reconciled to SEC filings", |
| f"quarter{'s' if len(unv) > 1 else ''} ended {ends} from the data provider, unverified"] |
| elif st == "PROVIDER_ONLY": |
| parts = ["the data provider's figures, not checked against SEC filings"] |
| else: |
| parts = ["unverified"] |
| if separated: |
| parts.append("before the separation") |
| return "Trailing twelve months (" + "; ".join(parts) + ")" |
|
|
|
|
| def anchored_forecasts(s): |
| """The forecast keys set from sell-side consensus (model weight 0), and those shrunk toward it (0 < weight < 1).""" |
| anc = s.forecast.get("consensus_anchor") or {} |
| full, part = [], [] |
| for k in ("revenue", "eps"): |
| a = anc.get(k) if isinstance(anc, Mapping) else None |
| if isinstance(a, Mapping) and a.get("model_weight") is not None: |
| (full if float(a["model_weight"]) <= 0.0 else part if float(a["model_weight"]) < 1.0 else []).append(k) |
| return full, part |
|
|
|
|
| def forecast_header(s, separated=False): |
| full, part = anchored_forecasts(s) |
| name = {"revenue": "revenue", "eps": "EPS"} |
| notes = (["continuing company"] if separated else []) + \ |
| ([" and ".join(name[k] for k in full) + " set from sell-side consensus"] if full else []) + \ |
| ([" and ".join(name[k] for k in part) + " shrunk toward sell-side consensus"] if part else []) |
| return "Engine forecast, next 12 months" + (" (" + "; ".join(notes) + ")" if notes else "") |
|
|
|
|
| |
| |
| HIST_MARK, FCST_MARK = "\u2020", "\u2021" |
|
|
|
|
| def historical_quality_render(q, separated=False): |
| """{full_description, table_header, footnote} for the trailing column, all from HistoricalMetricQuality.""" |
| st, unv = q.get("status"), list(q.get("unverified_periods") or []) |
| ends = " and ".join(unv) |
| what = "Trailing twelve months" + (", before the separation" if separated else "") |
| one = len(unv) == 1 |
| unchecked = (f"the quarter{'' if one else 's'} ended {ends} could not be verified against SEC filings and " |
| f"use{'s' if one else ''} the data provider's figures") |
| if st == "SEC_RECONCILED": |
| body = "reconciled to SEC filings" |
| elif st == "PARTIALLY_RECONCILED": |
| body = (f"reconciled to SEC filings through {q['verified_through']}" if q.get("verified_through") |
| else "partly reconciled to SEC filings") + f"; {unchecked}" |
| elif st == "PROVIDER_ONLY": |
| body = "the data provider's figures, not checked against SEC filings" |
| else: |
| body = "not verified against SEC filings" |
| return {"full_description": ttm_header(q, separated), |
| "table_header": ("TTM, pre-separation" if separated else "TTM") + HIST_MARK, |
| "footnote": f"{HIST_MARK} {what}: {body}."} |
|
|
|
|
| def forecast_lineage_render(s, separated=False): |
| """{full_description, table_header, footnote} for the forecast column, from the typed consensus anchor.""" |
| full, part = anchored_forecasts(s) |
| name = {"revenue": "revenue", "eps": "EPS"} |
| anc = s.forecast.get("consensus_anchor") or {} |
| why = {"RESTATEMENT": "the reported statements still include a separated business", |
| "MISFIRE": "the model's own estimate was set aside"} |
| reasons = [why.get(str((anc.get(k) or {}).get("reason"))) for k in full] |
| reasons = [x for x in dict.fromkeys(reasons) if x] |
| parts = ["Engine forecast" + (" for the continuing company" if separated else "") + "."] |
| if full: |
| subj = " and ".join(name[k] for k in full) |
| parts.append(f"{subj[:1].upper() + subj[1:]} {'are' if len(full) > 1 else 'is'} set from sell-side consensus in " |
| f"this run" + (f" ({'; '.join(reasons)})" if reasons else "") + ", not independent engine estimates.") |
| if part: |
| subj = " and ".join(name[k] for k in part) |
| parts.append(f"{subj[:1].upper() + subj[1:]} {'are' if len(part) > 1 else 'is'} shrunk toward sell-side consensus.") |
| if not full and not part: |
| parts.append("A model estimate, not guidance or consensus.") |
| return {"full_description": forecast_header(s, separated), |
| "table_header": ("Next 12 months, continuing company" if separated else "Next 12 months") + FCST_MARK, |
| "footnote": f"{FCST_MARK} " + " ".join(parts)} |
|
|
|
|
| _ANCHOR_WHY = {"RESTATEMENT": "the reported statements still include a separated business, so it is set from sell-side " |
| "consensus for the continuing company", |
| "MISFIRE": "the model's own estimate was set aside and replaced by sell-side consensus"} |
| _DIMENSION_WORDS = {"DEFINITION": "earnings definition", "ACCOUNTING_BASIS": "accounting basis", "PERIOD": "period", |
| "PERIMETER": "business perimeter", "CURRENCY": "currency", "SECURITY_BASIS": "share basis"} |
|
|
|
|
| def consensus_absence_sentence(s, rid, noun): |
| """Why the engine's figure is not compared with consensus, from relations.absence (typed), or None.""" |
| import relations as _RL |
| ab = _RL.absence(s, rid) |
| if ab is None: |
| return None |
| if ab["kind"] == "ANCHORED": |
| why = _ANCHOR_WHY.get(str(ab["anchor"].get("reason")), "it is set from sell-side consensus") |
| return (f"It is not compared with the engine's forecast: the engine's {noun} forecast is not an independent " |
| f"estimate ({why}).") |
| if ab["kind"] == "INCOMPATIBLE": |
| dims = [_DIMENSION_WORDS.get(d, "basis") for d in ab["dimensions"]] |
| return ("It is not compared with the engine's forecast: the two are not on the same " |
| + " or ".join(dict.fromkeys(dims)) + ".") |
| return "It is not compared with the engine's forecast." |
|
|
|
|
| def build_user_report(s, cat, prose, limitations): |
| """``prose`` holds ingested fragment lists: summary, thesis[], risks[], seats{role: {...}}, pm{...}.""" |
| r = _R(s, cat) |
| B = [] |
| eng = s.valuation.get("engine") or {} |
| terms = eng.get("terms") or {} |
| iv_word = s.valuation.get("intrinsic_label") or terms.get("intrinsic") or "intrinsic value" |
| per = s.historical.get("perimeter") or {} |
| sep = per.get("event") |
| pol_methods = set((s.issuer.get("policy") or {}).get("pricing_methods") or ()) |
|
|
| |
| B.append(Block("heading", "verdict", [r.t(f"{s.identity.get('company_name')} ({s.identity.ticker})")])) |
| B.append(_para("verdict", r.c("RATING"), r.t(" · "), r.c("HEADLINE_RETURN"), |
| r.t(" model-implied 12-month price return · share price "), r.c("PRICE"), css="verdict")) |
| |
| tiles = [] |
| if "INTRINSIC_VALUE" in cat: |
| tiles.append([[r.t(iv_word[:1].upper() + iv_word[1:])], [r.c("INTRINSIC_VALUE")], [r.t("price "), r.c("PRICE_VS_INTRINSIC")]]) |
| if "MARKET_COMPOSITE" in cat: |
| tiles.append([[r.t("Market pricing")], [r.c("MARKET_COMPOSITE")], [r.t("price "), r.c("PRICE_VS_COMPOSITE")]]) |
| if "RANGE_LOW" in cat and "RANGE_HIGH" in cat: |
| tiles.append([[r.t("Analyst scenarios")], [r.c("RANGE_LOW"), r.t(" to "), r.c("RANGE_HIGH")], |
| [r.t("probability-weighted "), r.c("PM_RETURN")]]) |
| if "MOMENTUM" in cat: |
| tiles.append([[r.t("Momentum")], [r.c("MOMENTUM")], []]) |
| if tiles: |
| B.append(Block("tiles", "verdict", {"rows": tiles})) |
| args = prose.get("arguments") or {} |
| s_args = args.get("summary") or {} |
| if s_args.get("thesis") or (not prose.get("summary") and (args.get("pm") or {}).get("thesis")): |
| B.append(Block("heading", "summary", [r.t("In brief")])) |
| _bt = s_args.get("thesis") or args["pm"]["thesis"] |
| r.authored_seen.add(_bt) |
| B.append(Block("para", "summary", [_authored(_bt)])) |
| elif prose.get("summary"): |
| B.append(Block("heading", "summary", [r.t("In brief")])) |
| B.append(Block("para", "summary", prose["summary"])) |
| else: |
| |
| |
| brief = [cid for cid in ("REL_PRICE_INTRINSIC__FULL_SENTENCE", "REL_PRICE_MARKET_COMPOSITE__FULL_SENTENCE") if cid in cat] |
| if brief: |
| B.append(Block("heading", "summary", [r.t("In brief")])) |
| frs = [] |
| for i, cid in enumerate(brief): |
| frs += ([r.t(" ")] if i else []) + [r.c(cid)] |
| B.append(Block("para", "summary", frs)) |
| _why = _admit(r, s_args.get("items_thesis")) |
| _wrong = _admit(r, s_args.get("items_risks")) |
| if _why: |
| B.append(Block("bullets", "summary", [_arg_item(r, s, cat, args, p) for p in _why], title="Why")) |
| elif prose.get("thesis"): |
| B.append(Block("bullets", "summary", prose["thesis"], title="Why")) |
| if _wrong: |
| B.append(Block("bullets", "summary", [_arg_item(r, s, cat, args, p) for p in _wrong], |
| title="What would make this wrong")) |
| elif prose.get("risks"): |
| B.append(Block("bullets", "summary", prose["risks"], title="What would make this wrong")) |
|
|
| |
| pm = (prose.get("seats") or {}).get("pm") or {} |
| seats = prose.get("seats") or {} |
| order = (("neutral", "Neutral analyst", "SCENARIO_NEUTRAL_RETURN", "SCENARIO_NEUTRAL_PROB"), |
| ("bear", "Bear", "SCENARIO_BEAR_RETURN", "SCENARIO_BEAR_PROB"), |
| ("bull", "Bull", "SCENARIO_BULL_RETURN", "SCENARIO_BULL_PROB")) |
| B.append(Block("heading", "debate", [r.t("The analyst debate")] + |
| ([r.t(" · "), r.c("RANGE_LOW"), r.t(" to "), r.c("RANGE_HIGH")] if ("RANGE_LOW" in cat and "RANGE_HIGH" in cat) else []))) |
|
|
| def _notes(entry, title=None, opponent=None): |
| """One seat's reasoning and key points; a title labels the round (opening case, rebuttal). A rebuttal point's |
| topic is, by the prompt's contract, the OPPOSING claim: it is labelled as the opponent's, never shown as if the |
| author asserted it.""" |
| if entry.get("reasoning"): |
| B.append(Block("para", "debate", entry["reasoning"], title=title)) |
| if entry.get("key_points"): |
| pts = entry["key_points"] |
| if opponent: |
| pts = [[Fragment(f"The {opponent}'s claim — ")] + list(kp) for kp in pts] |
| |
| B.append(Block("points", "debate", pts, title=None if entry.get("reasoning") else title)) |
|
|
| def _has(e): |
| return bool((e or {}).get("reasoning") or (e or {}).get("key_points")) |
|
|
| if any(k in args for k in ("neutral", "bear", "bull", "bear_open", "bull_open", "pm")): |
| _structured_debate(B, r, s, cat, args, seats, order, _notes) |
| order = () |
| for role, name, ret_id, p_id in order: |
| st = seats.get(role) or {} |
| B.append(_para("debate", r.t(f"{name} — "), r.c(ret_id), |
| r.t(f" · {(s.scenarios.get(role) or {}).get('label') or ''} · probability "), r.c(p_id), css="seat-head")) |
| op = seats.get(f"{role}_open") or {} |
| if not (_has(op) or _has(st)): |
| |
| B.append(_para("debate", r.t(_empty_seat_note(s, role)))) |
| elif op: |
| _notes(op, "Opening case") |
| _notes(st, "Rebuttal", opponent={"bear": "bull", "bull": "bear"}.get(role)) |
| else: |
| _notes(st) |
| if order or not args: |
| B.append(_para("debate", r.t("Portfolio manager — probability-weighted "), r.c("PM_RETURN"), |
| r.t(f" · {s.scenarios.get('pm_label') or ''}"), css="seat-head")) |
| if (order or not args) and all(c in cat for c in ("SCENARIO_BEAR_PROB", "SCENARIO_NEUTRAL_PROB", "SCENARIO_BULL_PROB", |
| "PM_RETURN_PHRASE")): |
| |
| B.append(_para("debate", r.t("The portfolio manager weights the bear case at "), r.c("SCENARIO_BEAR_PROB"), |
| r.t(", the neutral case at "), r.c("SCENARIO_NEUTRAL_PROB"), r.t(" and the bull case at "), |
| r.c("SCENARIO_BULL_PROB"), r.t(", for "), r.c("PM_RETURN_PHRASE"), r.t("."))) |
| if (order or not args) and (pm.get("reasoning") or pm.get("key_points")): |
| _notes(pm) |
| trg = s.scenarios.get("triggers") or [] |
| if trg: |
| items = [] |
| for i, t in enumerate(trg[:3]): |
| |
| |
| src = t.get("source_claim_id") or t.get("source_guidance_id") |
| subj = trigger_subject(s, cat, src) |
| if not subj: |
| continue |
| dest = "the bull case" if t.get("scenario_destination") == "BULL" else "the bear case" |
| verb = "rises to" if t.get("side") == "above" else "falls to" |
| thr = r.p(f"scenarios.triggers.{i}.threshold", t.get("threshold_kind") or C.TEXT, "trigger threshold (hypothetical)") |
| side = f" or {'above' if t.get('side') == 'above' else 'below'}" |
| cl = cat.get(src) |
| if getattr(cl, "stype", None) == C.GUIDANCE: |
| |
| items.append([r.t(f"If {subj} {verb} "), thr, r.t(f"{side} (company guidance: "), r.c(src), |
| r.t(f"), the view moves toward {dest}.")]) |
| elif set(getattr(C.claim_def(s, src), "prohibited", ()) or ()) & {"now", "current", "today"}: |
| items.append([r.t(f"If {subj} {verb} "), thr, r.t(f"{side}, the view moves toward {dest}.")]) |
| else: |
| items.append([r.t(f"If {subj}, now "), r.c(src), r.t(f", {verb} "), thr, |
| r.t(f"{side}, the view moves toward {dest}.")]) |
| if items: |
| B.append(Block("bullets", "debate", items, title="What would change the view")) |
|
|
| |
| B.append(Block("heading", "valuation", [r.t("Intrinsic value")] + ([r.t(" · "), r.c("INTRINSIC_VALUE")] if "INTRINSIC_VALUE" in cat else []))) |
| lc = s.valuation.get("lifecycle") or {} |
| if "LIFECYCLE_STAGE" in cat: |
| lp = [r.t("Life-cycle stage: "), r.c("LIFECYCLE_STAGE")] |
| |
| _cmp = C.comparison(s, "LIFECYCLE_TRAILING_GROWTH", "LIFECYCLE_FORWARD_GROWTH") |
| if "LIFECYCLE_TRAILING_GROWTH" in cat and (_cmp is None or _cmp[0] or "LIFECYCLE_FORWARD_GROWTH" not in cat): |
| lp += [r.t(", with trailing revenue growth of "), r.c("LIFECYCLE_TRAILING_GROWTH")] |
| if "LIFECYCLE_FORWARD_GROWTH" in cat: |
| |
| |
| import arguments as _A |
| _fg = _A._READER["LIFECYCLE_FORWARD_GROWTH"][0].strip() |
| lp += [r.t(f" and {_fg} of " if (lp[-1:] and len(lp) > 2) else f", with {_fg} of "), |
| r.c("LIFECYCLE_FORWARD_GROWTH")] |
| lp += [r.t(".")] |
| B.append(_para("valuation", *lp)) |
| if "INTRINSIC_VALUE" in cat: |
| |
| if "REL_PRICE_INTRINSIC" in cat: |
| vp = [r.c("REL_PRICE_INTRINSIC", cap=True), r.t(".")] |
| else: |
| vp = [r.t(f"The {iv_word} is "), r.c("INTRINSIC_VALUE"), r.t(".")] |
| if "REL_PRICE_MECHANICAL" in cat and not s.valuation.get("same_model"): |
| vp += [r.t(" "), r.c("REL_PRICE_MECHANICAL", cap=True), r.t(".")] |
| B.append(_para("valuation", *vp)) |
| stories = prose.get("stories") or [] |
| adopted = next((x for x in stories if x.get("adopted")), None) |
| if adopted and adopted.get("story"): |
| risk = ([r.t(" Key risk: ")] + list(adopted["risk"]) + [r.t(".")]) if adopted.get("risk") else [] |
| B.append(Block("para", "valuation", list(adopted["story"]) + risk, title="The story behind the value")) |
| |
| |
| implied = [cid for cid in ("REL_BREAK_EVEN_REVENUE_STORY", "REL_BREAK_EVEN_REVENUE_GROWTH", "REL_BREAK_EVEN_MARGIN_STORY", |
| "REL_EXPECTATIONS_STORY") if cid in cat] |
| if implied: |
| B.append(Block("bullets", "valuation", [[r.c(cid, cap=True), r.t(".")] for cid in implied], title="What the price implies")) |
| dmi = s.valuation.get("damodaran_inputs") or {} |
| if dmi.get("available") and "DMD_ERP" in cat: |
| rows = [] |
| if "DMD_INDUSTRY" in cat: |
| rows.append([[r.t("Industry (his classification)")], [r.c("DMD_INDUSTRY")]]) |
| if "DMD_BETA" in cat: |
| basis = "business-mix" if dmi.get("business_mix") else "industry" |
| rows.append([[r.t("Beta, unlevered → levered" + (" (business mix)" if basis == "business-mix" else ""))], |
| [r.c("DMD_UNLEVERED_BETA"), r.t(" → "), r.c("DMD_BETA")]]) |
| erp_cell = [r.c("DMD_ERP"), r.t(" (implied "), r.c("DMD_IMPLIED_ERP"), r.t(", "), r.c("DMD_ERP_MONTH")] |
| if (dmi.get("crp") or 0) > 0: |
| erp_cell += [r.t(", plus country risk premium "), r.c("DMD_CRP"), r.t(" for "), r.c("DMD_COUNTRY")] |
| rows.append([[r.t("Equity risk premium")], erp_cell + [r.t(")")]]) |
| def _applicability(ind, own, own_words): |
| """The industry figure beside the company's own when they differ by more than 2x either way: a benchmark |
| that does not describe the business is labelled as such, never presented as its norm. ``own`` is a claim id |
| or a (state path, stype, label) triple.""" |
| iv = getattr(cat.get(ind), "value", None) |
| if isinstance(own, tuple): |
| ov = s.value(own[0]) |
| own_frag = lambda: r.p(*own) |
| else: |
| ov = getattr(cat.get(own), "value", None) |
| own_frag = lambda: r.c(own) |
| try: |
| ratio = float(ov) / float(iv) if iv and ov else None |
| except (TypeError, ValueError): |
| ratio = None |
| if ratio is None or 0.5 <= ratio <= 2.0: |
| return [r.c(ind)] |
| return [r.c(ind), r.t(f" (not representative of this company: {own_words} "), own_frag(), r.t(")")] |
| if "DMD_S2C" in cat: |
| rows.append([[r.t("Industry sales-to-capital")], |
| _applicability("DMD_S2C", ("valuation.story.assumptions.sales_to_capital", C.RATIO, "sales-to-capital"), |
| "the story uses")]) |
| if "DMD_MARGIN" in cat: |
| rows.append([[r.t("Industry operating margin")], |
| _applicability("DMD_MARGIN", "STORY_START_MARGIN", "the continuing company's starting margin is")]) |
| if "WACC_BENCHMARK" in cat: |
| rows.append([[r.t("Mature-market cost-of-capital benchmark")], [r.c("WACC_BENCHMARK")]]) |
| B.append(Block("table", "valuation", {"head": ["Input (Damodaran's data)", "Value"], "rows": rows}, |
| title="Market and industry inputs")) |
| mix = dmi.get("business_mix") or [] |
| if mix: |
| base = "valuation.damodaran_inputs.business_mix." |
| by_value = all(g.get("estimated_value") for g in mix) |
| head = ["Filed segment", "Share of revenue"] + (["Industry EV/Sales", "Weight (estimated value)"] if by_value else []) + \ |
| ["Industry (his)", "Unlevered beta"] |
| B.append(Block("table", "valuation", {"head": head, |
| "rows": [[[r.t(str(g.get("name")))], |
| [r.p(f"{base}{i}." + ("revenue_weight" if by_value else "weight"), C.PCT, "segment share of revenue")]] |
| + ([[r.p(f"{base}{i}.industry_ev_sales", C.MULTIPLE, "segment industry EV/Sales")], |
| [r.p(f"{base}{i}.weight", C.PCT, "segment value weight")]] if by_value else []) |
| + [[r.t(str(g.get("industry")))], [r.p(f"{base}{i}.unlevered_beta", C.RATIO, "segment industry unlevered beta")]] |
| for i, g in enumerate(mix)]}, |
| title="Business mix")) |
| stry = s.valuation.get("story") or {} |
| if stry.get("ran"): |
| A = "valuation.story.assumptions." |
| rows = [[[r.t("Market size at maturity")], [r.c("STORY_TAM")]], |
| [[r.t("Peak market share")], [r.c("STORY_SHARE")]], |
| [[r.t("Revenue at maturity")], [r.c("STORY_REVENUE_MATURITY")]], |
| [[r.t("Years to maturity")], [r.c("STORY_YEARS")]], |
| ] + ([[[r.t(str((s.valuation.get("story_base") or {}).get("revenue_label") or "Starting revenue").capitalize())], |
| [r.c("STORY_START_REVENUE")]]] if "STORY_START_REVENUE" in cat else []) + [ |
| ] + ([[[r.t(str((s.valuation.get("story_base") or {}).get("label") or "Starting operating margin").capitalize())], |
| [r.c("STORY_START_MARGIN")]]] if "STORY_START_MARGIN" in cat else []) + [ |
| [[r.t("Operating margin at maturity (story assumption)")], [r.c("STORY_MARGIN")]], |
| [[r.t("Sales to capital")], [r.p(A + "sales_to_capital", C.RATIO, "sales-to-capital")]], |
| ] + ([[[r.t("Initial cost of capital")], [r.c("STORY_WACC")]]] if "STORY_WACC" in cat else []) |
| |
| _dp = s.valuation.get("discount_rate_path") or {} |
| if "STORY_WACC_STABLE" in cat and "STORY_WACC_HOLD" in cat: |
| if abs((_dp.get("terminal_wacc") or 0) - (_dp.get("initial_wacc") or 0)) < 1e-9: |
| rows.append([[r.t("Cost of capital path")], [r.t("the initial rate in every year")]]) |
| else: |
| rows.append([[r.t("Cost of capital path")], [r.c("STORY_WACC_HOLD"), r.t(" years at the initial rate, then a straight line over "), |
| r.p("valuation.discount_rate_path.transition_years", C.COUNT, "transition years"), |
| r.t(" years to the terminal rate")]]) |
| rows.append([[r.t("Terminal cost of capital used")], [r.c("STORY_WACC_STABLE")]]) |
| _yr = _dp.get("yearly_rates") or [] |
| if _yr: |
| cell = [] |
| for i in range(len(_yr)): |
| cell += ([r.t(" · ")] if i else []) + [r.p(f"valuation.discount_rate_path.yearly_rates.{i}", C.PCT, f"discount rate year {i + 1}")] |
| rows.append([[r.t("Discount rate by year")], cell]) |
| rows += [[[r.t("Terminal growth")], [r.c("STORY_TERMINAL_GROWTH")]]] |
| if _dp.get("terminal_spread") is not None: |
| rows.append([[r.t("Terminal cost of capital less terminal growth")], |
| [r.p("valuation.discount_rate_path.terminal_spread", C.PP, "terminal spread")]]) |
| rows += [[[r.t("Failure probability")], [r.c("STORY_FAILURE")]]] |
| B.append(Block("table", "valuation", {"head": ["Story assumption (adopted storyteller)", "Value"], "rows": rows}, |
| title="Story assumptions")) |
| segs = stry.get("segments") or [] |
| if segs: |
| SB = "valuation.story.segments." |
| _a0 = stry.get("assumptions") or {} |
| _capped = (_a0.get("margin_proposed") is not None and _a0.get("margin_target") is not None |
| and abs(float(_a0["margin_proposed"]) - float(_a0["margin_target"])) > 0.0005) |
| B.append(Block("table", "valuation", {"head": ["Business (adopted story)", "Revenue at maturity (USD bn)", |
| "Operating margin at maturity" + (" (storyteller's proposal)" if _capped else ""), |
| "Sales to capital"], |
| "rows": [[[r.t(str(g.get("name")))], [r.p(f"{SB}{i}.rev_usd_b", C.RATIO, "segment revenue at maturity (USD bn)")], |
| ([r.p(f"{SB}{i}.margin", C.PCT, "segment maturity margin")] if g.get("margin") is not None else [r.t("n/a")]), |
| ([r.p(f"{SB}{i}.sales_to_capital", C.RATIO, "segment sales-to-capital")] if g.get("sales_to_capital") else [r.t("story-level")])] |
| for i, g in enumerate(segs)]}, |
| title="Segment build")) |
| _a = stry.get("assumptions") or {} |
| if _a.get("margin_proposed") is not None and _a.get("margin_target") is not None \ |
| and abs(float(_a["margin_proposed"]) - float(_a["margin_target"])) > 0.0005: |
| _raised = _a.get("margin_floor") is not None and float(_a["margin_proposed"]) < float(_a["margin_target"]) |
| B.append(_para("valuation", r.t("The segments' revenue-weighted operating margin at maturity ("), |
| r.p("valuation.story.assumptions.margin_proposed", C.PCT, "segment-weighted maturity margin"), |
| r.t(") is raised to " if _raised else ") is capped at "), |
| r.p("valuation.story.assumptions.margin_target", C.PCT, "maturity operating margin used"), |
| r.t(", the maturity operating margin the valuation uses" |
| + (": a currently profitable company is not valued below the lower of today's margin and " |
| "the best sustained margin in its sector." if _raised else ".")))) |
| srows = [] |
| for i, x in enumerate(stry.get("stories") or []): |
| srows.append([[r.t(f"Storyteller {i + 1}" + (" (adopted)" if x.get("adopted") else "") |
| + (f", {x['n_segments']} businesses" if x.get("n_segments") else ""))], |
| [r.p(f"valuation.story.stories.{i}.value", C.CURRENCY, "storyteller value")], |
| [r.p(f"valuation.story.stories.{i}.derived.margin_target", C.PCT, "storyteller maturity margin")], |
| [r.p(f"valuation.story.stories.{i}.derived.years", C.COUNT, "storyteller years")]]) |
| if srows: |
| B.append(Block("table", "valuation", {"head": ["Storyteller", "Value", "Maturity margin", "Years"], "rows": srows}, |
| title="The storytellers")) |
| items = [[r.t(f"Storyteller {i + 1}: ")] + x["story"] for i, x in enumerate(stories) if x.get("story")] |
| if len(items) > 1: |
| B.append(Block("bullets", "valuation", items, title="Their stories")) |
| div = s.valuation.get("storyteller_diversity") or {} |
| if div.get("status") == "DUPLICATE_STORY": |
| B.append(_para("valuation", r.t("Two storytellers used effectively identical assumptions, so the range reflects fewer " |
| "independent views than the number of storytellers."))) |
| sens = s.valuation.get("sensitivity") or [] |
| if sens: |
| def _case(i, x): |
| if x.get("driver") and x.get("shift") is not None: |
| kind = C.PP if x.get("shift_kind") == "PP" else C.PCT_SIGNED |
| return [r.t(f"{x['driver']} "), r.p(f"valuation.sensitivity.{i}.shift", kind, "sensitivity shift")] |
| return [r.t(str(x.get("case")))] |
| B.append(Block("table", "valuation", {"head": ["Change", "Value", "Change vs base"], |
| "rows": [[_case(i, x), |
| [r.p(f"valuation.sensitivity.{i}.value", C.CURRENCY, "sensitivity value")], |
| [r.p(f"valuation.sensitivity.{i}.vs_base", C.PCT_SIGNED, "change vs base")]] |
| for i, x in enumerate(sens)]}, title="How sensitive is the value?")) |
|
|
| |
| B.append(Block("heading", "pricing", [r.t("Market pricing")] + ([r.t(" · "), r.c("MARKET_COMPOSITE")] if "MARKET_COMPOSITE" in cat else []))) |
| if "REL_PRICE_MARKET_COMPOSITE" in cat: |
| B.append(_para("pricing", r.c("REL_PRICE_MARKET_COMPOSITE", cap=True), r.t("."))) |
| meth = s.pricing.get("methods") or [] |
| if meth: |
| B.append(Block("table", "pricing", {"head": ["Method", "Implied price", "Multiple", "Share of composite"], |
| "rows": [[[r.t(m.get("label") or m.get("name"))], |
| [r.p(f"pricing.methods.{i}.price", C.CURRENCY, "method price")], |
| ([r.p(f"pricing.methods.{i}.target_multiple", C.MULTIPLE, "target multiple"), |
| r.t(_method_basis(m))] |
| if m.get("target_multiple") else [r.t("—")]), |
| [r.p(f"pricing.methods.{i}.share", C.PCT, "method share") if m.get("in_composite") else r.t("excluded")]] |
| for i, m in enumerate(meth)]})) |
| |
| steps = [[[r.t("Weighted average of the methods")], [r.c("MARKET_COMPOSITE_RAW")]]] |
| _moved = (s.value("pricing.raw_composite") is not None and s.value("pricing.market_composite") is not None |
| and abs(s.value("pricing.raw_composite") - s.value("pricing.market_composite")) > 0.005) |
| if s.value("pricing.post_trim_composite") is not None and s.value("pricing.raw_composite") is not None \ |
| and abs(s.value("pricing.post_trim_composite") - s.value("pricing.raw_composite")) > 1e-6: |
| steps.append([[r.t("After dropping outlier methods")], [r.c("MARKET_COMPOSITE_POST_TRIM")]]) |
| if (s.pricing.get("cap") or {}).get("applied"): |
| steps.append([[r.t("Bounded to "), r.p("pricing.cap.lower", C.MULTIPLE, "cap lower"), r.t("–"), |
| r.p("pricing.cap.upper", C.MULTIPLE, "cap upper"), r.t(" the price")], [r.c("MARKET_COMPOSITE")]]) |
| else: |
| steps.append([[r.t("Market composite")], [r.c("MARKET_COMPOSITE")]]) |
| if _moved: |
| B.append(Block("table", "pricing", {"head": ["Step", "Value"], "rows": steps}, title="How the composite is built")) |
| if not pol_methods or "P/E" in pol_methods: |
| pe = [r.t("Forward P/E on the engine's EPS forecast: "), r.c("PE_FORWARD")] |
| if "PE_TRAILING" in cat: |
| pe += [r.t("; trailing P/E "), r.c("PE_TRAILING"), r.t(" on adjusted EPS and "), r.c("PE_TRAILING_GAAP"), r.t(" on reported EPS")] |
| elif sep: |
| pe += [r.t("; the company's own trailing multiples are not shown because its trailing earnings include the " |
| "separated business")] |
| if "PEG" in cat: |
| pe += [r.t("; forward PEG "), r.c("PEG")] |
| B.append(_para("pricing", *pe, r.t("."))) |
| if "ANALYST_TARGET" in cat: |
| at = [r.t("Mean sell-side price target "), r.c("ANALYST_TARGET"), r.t(" ("), r.c("TARGET_VS_PRICE"), r.t(" from the price)")] |
| if "ANALYST_TARGET_ADJUSTED" in cat and s.pricing.get("analyst_target_in_composite"): |
| at += [r.t("; it enters the composite at "), r.c("ANALYST_TARGET_ADJUSTED")] |
| B.append(_para("pricing", *at, r.t("."))) |
| peers = (s.pricing.get("peers") or {}).get("peers") or [] |
| if peers: |
| prow = [] |
| for i, pe in enumerate(peers[:8]): |
| base = f"pricing.peers.peers.{i}." |
| prow.append([[r.t(str(pe.get("ticker")))], [r.p(base + "pe", C.MULTIPLE, "peer P/E")], |
| [r.p(base + "ev_ebitda", C.MULTIPLE, "peer EV/EBITDA")], [r.p(base + "op_margin", C.PCT, "peer operating margin")], |
| [r.p(base + "rev_growth", C.PCT_SIGNED, "peer revenue growth")]]) |
| sj = "pricing.peers.subject." |
| if (s.pricing.get("peers") or {}).get("subject"): |
| prow.insert(0, [[r.t(f"{s.identity.ticker} (subject)")], [r.p(sj + "pe", C.MULTIPLE, "subject P/E (peer panel)")], |
| [r.p(sj + "ev_ebitda", C.MULTIPLE, "subject EV/EBITDA")], [r.p(sj + "op_margin", C.PCT, "subject operating margin (peer panel)")], |
| [r.p(sj + "rev_growth", C.PCT_SIGNED, "subject revenue growth (peer panel)")]]) |
| B.append(Block("table", "pricing", {"head": ["Company", "P/E (trailing)", "EV/EBITDA", "Operating margin", "Revenue growth"], |
| "rows": prow}, title="Relative to peers (Sharadar panel, trailing)")) |
| notes = [] |
| if (s.pricing.get("peers") or {}).get("subject_withheld"): |
| notes.append(f"{s.identity.ticker}'s own row is not shown: {(s.pricing.get('peers') or {}).get('subject_withheld')}.") |
| subj = (s.pricing.get("peers") or {}).get("subject") or {} |
| wh = ([s.identity.ticker] if subj.get("multiples_withheld") else []) + \ |
| [str(pe.get("ticker")) for pe in peers[:8] if pe.get("multiples_withheld")] |
| if wh: |
| notes.append("Valuation multiples are not shown for " + ", ".join(wh) + ": the panel reports them in a different currency.") |
| if notes: |
| B.append(_para("pricing", r.t(" ".join(notes)))) |
|
|
| |
| B.append(Block("heading", "forecast", [r.t("Fundamentals and forecast")])) |
| rows = [[[r.t("Revenue")], [r.c("REVENUE_TTM")], [r.c("FORECAST_REVENUE")]], |
| [[r.t("Operating margin")], [r.c("OPERATING_MARGIN_TTM")], [r.c("FORECAST_OPERATING_MARGIN")]], |
| [[r.t("Net margin")], [r.c("NET_MARGIN_TTM")], [r.c("FORECAST_NET_MARGIN")]], |
| [[r.t("EPS (adjusted basis)")], [r.c("EPS_ADJUSTED_TTM")], [r.c("FORECAST_EPS")]]] |
| _hq, _fl = historical_quality_render(historical_quality(s), bool(sep)), forecast_lineage_render(s, bool(sep)) |
| B.append(Block("table", "forecast", {"head": ["", _hq["table_header"], _fl["table_header"]], "rows": rows}, css="rp-fund")) |
| B.append(_para("forecast", r.t(_hq["footnote"]), css="rp-foot")) |
| B.append(_para("forecast", r.t(_fl["footnote"]), css="rp-foot")) |
| if sep: |
| B.append(_para("forecast", r.t(f"The trailing figures include a business separated from the company on {sep.get('effective')}. " |
| "The forecast and sell-side estimates describe the continuing company, so the two columns " |
| "describe different businesses: no growth rate is computed between them."))) |
| fc = (s.forecast.get("coherence") or {}).get("status") |
| words = {"CONSISTENT": "The revenue, earnings and operating-income models agree.", |
| "DIVERGENT": "The independent operating-income and net-income models diverge from the reconciled forecast; confidence is reduced.", |
| "CONFLICTING": "The independent operating-income and net-income models conflict with the reconciled forecast; treat the margin path as uncertain.", |
| "NOT_VERIFIABLE": "The forecast components could not be cross-checked against each other."}.get(fc) |
| if not sep and not s.issuer.get("is_financial") and fc in ("DIVERGENT", "CONFLICTING", "NOT_VERIFIABLE"): |
| B.append(_para("forecast", r.t(words))) |
| if "CONSENSUS_EPS_FY1" in cat and "REL_FORECAST_EPS_CONSENSUS" in cat: |
| B.append(_para("forecast", r.t("Sell-side consensus EPS is "), r.c("CONSENSUS_EPS_FY1"), r.t(" for the current fiscal year ("), |
| r.c("CONSENSUS_ANALYSTS"), r.t(" contributors). "), r.c("REL_FORECAST_EPS_CONSENSUS", cap=True), r.t("."))) |
| elif "CONSENSUS_EPS_FY1" in cat: |
| B.append(_para("forecast", r.t("Sell-side consensus EPS is "), r.c("CONSENSUS_EPS_FY1"), r.t(" for the current fiscal year ("), |
| r.c("CONSENSUS_ANALYSTS"), r.t(" contributors). " |
| + consensus_absence_sentence(s, "REL_FORECAST_EPS_CONSENSUS", "EPS")))) |
| if "REL_ENGINE_REVENUE_CONSENSUS" in cat: |
| B.append(_para("forecast", r.c("REL_ENGINE_REVENUE_CONSENSUS", cap=True), r.t("."))) |
| |
| if "REL_GUIDANCE_EPS_CONSENSUS" in cat: |
| B.append(_para("forecast", r.c("REL_GUIDANCE_EPS_CONSENSUS", cap=True), r.t("."))) |
| |
| g = s.guidance.get("company") or {} |
| if g.get("status") == "AVAILABLE": |
| |
| rows_g = guidance_table_rows(s, cat) |
| if rows_g: |
| src = g.get("source") or {} |
| |
| |
| merged = {} |
| for i, m in rows_g: |
| _cl = cat.get(f"GUIDANCE_{i}") |
| key = (str(m.get("label")), str(m.get("period_label")), str(_cl.rendered) if _cl is not None else f"#{i}") |
| merged.setdefault(key, (i, m, []))[2].append( |
| {"GAAP": "GAAP", "ADJUSTED": "non-GAAP"}.get(m.get("accounting_basis"), "—")) |
| rows_out = [] |
| for (_lab, _per, _v), (i, m, bases) in merged.items(): |
| bs = [b for b in dict.fromkeys(bases) if b != "—"] |
| rows_out.append([[r.t(str(m.get("label")))], [r.t(str(m.get("period_label")))], [r.c(f"GUIDANCE_{i}")], |
| [r.t(" and ".join(bs) if bs else "—")]]) |
| B.append(Block("table", "forecast", {"head": ["Company guidance", "Period", "Guided", "Basis"], "rows": rows_out}, |
| title=f"Company guidance ({src.get('form')} filed {src.get('filed')})")) |
| ins = s.insiders.get("canon") or {} |
| if ins.get("characterization"): |
| B.append(_para("forecast", r.t("Insider transactions (SEC Form 4): "), r.c("INSIDER_CHARACTERIZATION"), r.t("."))) |
| if s.ownership.get("status") == "CURRENT" and "OWN_13F_TRACKED" in cat: |
| prev = [r.t(", against "), r.c("OWN_13F_HOLDERS_PREV"), r.t(" the quarter before")] if "OWN_13F_HOLDERS_PREV" in cat else [] |
| B.append(_para("forecast", r.t("Institutional ownership (SEC Form 13F, quarter ended "), r.c("OWN_13F_AS_OF"), r.t("): "), |
| r.c("OWN_13F_HOLDERS"), r.t(" institutions reported a position"), *prev, r.t(". "), r.c("OWN_13F_TRACKED"), r.t(" tracked long-term investors hold the stock, including "), |
| r.c("OWN_13F_NAMES"), r.t("; last quarter "), r.c("OWN_13F_NEW"), r.t(" opened a position, "), r.c("OWN_13F_ADDS"), |
| r.t(" added, "), r.c("OWN_13F_TRIMS"), r.t(" trimmed and "), r.c("OWN_13F_EXITS"), |
| r.t(" exited."))) |
| if (s.leadership.get("roles") or {}).get("CEO"): |
| B.append(_para("forecast", r.t("Chief executive: "), r.c("CEO"), r.t("."))) |
|
|
| |
| B.append(Block("heading", "blend", [r.t("What drives the headline")])) |
| src = s.blend.get("sources") or [] |
| if src: |
| B.append(Block("table", "blend", {"head": ["Source", "View", "Share of the blend"], |
| "rows": [[[r.t(str(x.get("label") or x.get("name")))], |
| [r.p(f"blend.sources.{i}.return", C.PCT_SIGNED, "source view")], |
| [r.p(f"blend.sources.{i}.share", C.PCT, "source share")]] for i, x in enumerate(src)]})) |
| for i, a in enumerate(s.blend.get("adjustments") or ()): |
| B.append(_para("blend", r.t(f"Adjustment applied to the weighted sum — {a.get('name')}: ×"), |
| r.p(f"blend.adjustments.{i}.factor", C.RATIO, "blend adjustment factor"), r.t("."))) |
| exc = [str(x.get("name")) for x in (s.blend.get("excluded_sources") or ()) |
| if x.get("name") and x.get("kind") != "PRICE_MODEL"] |
| if exc: |
| B.append(_para("blend", r.t("Carrying no weight in this analysis: " + ", ".join(exc) + "."))) |
| B += _ml_paragraphs(s, r, cat) |
|
|
| |
| B.append(Block("heading", "limits", [r.t("Sources and limitations")])) |
| if limitations: |
| B.append(Block("bullets", "limits", [[r.t(x)] for x in limitations])) |
| B.append(_para("limits", r.t(f"Statements through {s.historical.get('statements_as_of')} · price as of {s.market.get('price_as_of')} · " |
| "not investment advice."))) |
| B.append(_para("limits", r.t("Analysis version "), Fragment(s.version, "VERSION", s.version), r.t("."), css="version")) |
|
|
| claims = {cid: cl.to_dict() for cid, cl in r.used.items()} |
| |
| for b in B: |
| for f in b.fragments(): |
| if f.claim_id and f.claim_id not in claims and f.claim_id in cat: |
| claims[f.claim_id] = cat[f.claim_id].to_dict() |
| claims["VERSION"] = {"claim_id": "VERSION", "path": "version", "stype": "TEXT", "label": "analysis version", |
| "value": s.version, "rendered": s.version} |
| return UserReport(s.identity.ticker, s.identity.get("company_name"), s.identity.get("as_of"), s.version, B, claims, limitations, |
| arguments=dict(prose.get("arguments") or {})) |
|
|
|
|
| |
|
|
|
|
| def _ml_paragraphs(s, r, cat): |
| """Machine learning, in plain words. Two models are checked on every run; each counts only as far |
| as its out-of-sample record allows. A model that is not admitted is described without its figures.""" |
| pm = (s.quant.get("price_model") or {}).get("reliability") or {} |
| pr = (s.quant.get("panel_rank") or {}).get("reliability") or {} |
| out = [] |
| if pr.get("persona_visibility") and "ML_RANK_SHARE" in cat: |
| out.append(_para("blend", r.t("Machine-learning rank: above "), r.c("ML_RANK_SHARE"), r.t(" of the "), r.c("ML_RANK_UNIVERSE"), |
| r.t(" largest US companies (decile "), r.c("ML_RANK_DECILE"), r.t("); walk-forward rank correlation "), |
| r.c("ML_RANK_IC"), r.t(". Weak evidence only."))) |
| elif pr.get("state") == "UNRELIABLE": |
| why = ("failed in the most recent test period" if pr.get("failed_recently") else "out-of-sample record too weak") |
| out.append(_para("blend", r.t(f"Machine-learning rank: not used ({why})."))) |
| if pm.get("state") in ("LIMITED", "VALIDATED"): |
| out.append(_para("blend", r.t("Machine-learning price model: " + ("used with a capped share (limited out-of-sample record)." |
| if pm.get("state") == "LIMITED" else "used (validated out of sample).")))) |
| elif pm.get("state") in ("UNRELIABLE", "UNAVAILABLE"): |
| out.append(_para("blend", r.t("Machine-learning price model: not used (" |
| + ("direction calls no better than chance)." if pm.get("state") == "UNRELIABLE" |
| else "not enough history to test).")))) |
| return out |
|
|
|
|
| |
| |
| |
| def _frag_md(f): |
| return f.text |
|
|
|
|
| def _frag_html(f): |
| t = _html.escape(f.text) |
| return f'<span class="cl" data-claim="{_html.escape(f.claim_id)}">{t}</span>' if f.claim_id else t |
|
|
|
|
| def render_markdown(rep: UserReport): |
| L = [] |
| for b in rep.blocks: |
| if b.kind == "heading": |
| L.append(("# " if b.section == "verdict" else "## ") + "".join(_frag_md(f) for f in b.content)) |
| elif b.kind in ("para", "callout"): |
| txt = "".join(_frag_md(f) for f in b.content) |
| if b.title: |
| L.append(f"*{b.title}*") |
| L.append(f"**{txt}**" if b.css in ("verdict", "seat-head") else txt) |
| elif b.kind in ("bullets", "points"): |
| if b.title: |
| L.append(f"**{b.title}**") |
| L.extend("- " + "".join(_frag_md(f) for f in item) for item in b.content) |
| elif b.kind == "tiles": |
| for row in b.content.get("rows", []): |
| lab, val, note = ("".join(_frag_md(f) for f in cell) for cell in (row + [[], [], []])[:3]) |
| L.append(f"- **{lab}:** {val}" + (f" ({note})" if note else "")) |
| elif b.kind == "table": |
| if b.title: |
| L.append(f"**{b.title}**") |
| head = b.content.get("head", []) |
| L.append("| " + " | ".join(head) + " |") |
| L.append("|" + "---|" * len(head)) |
| for row in b.content.get("rows", []): |
| L.append("| " + " | ".join("".join(_frag_md(f) for f in cell) for cell in row) + " |") |
| L.append("") |
| return "\n".join(L).strip() + "\n" |
|
|
|
|
| |
| HEAD_END = "head_end" |
| _SEAT_NAMES = {"neutral": "neutral analyst", "bear": "bear", "bull": "bull"} |
|
|
|
|
| def _authored(text): |
| return Fragment(str(text), origin="authored") |
|
|
|
|
| QUOTE = "quote" |
|
|
|
|
| def trigger_subject(s, cat, cid): |
| """The reader's name for the quantity a trigger watches: its source claim's label (a guidance row by its own label, |
| period and basis). None when the claim is not in the catalog.""" |
| cl = cat.get(cid) if cid else None |
| if cl is None or cl.rendered is None: |
| return None |
| if getattr(cl, "stype", None) == C.GUIDANCE: |
| |
| |
| |
| import guidance as _G |
| v = cl.value if isinstance(cl.value, dict) else {} |
| rows = ((s.guidance.get("company") or {}).get("metrics")) or [] |
| try: |
| label = str((rows[int(str(cid).split("_")[1]) - 1] or {}).get("label") or "").strip() |
| except (ValueError, IndexError): |
| label = "" |
| label = label or _G.concept_label(v.get("concept")) |
| basis = C.guidance_basis_label(cat, cid) |
| basis = None if (basis and basis.lower() in label.lower()) else basis |
| qual = ", ".join(x for x in (v.get("period_label"), basis) if x) |
| return label + (f" ({qual})" if qual else "") |
| label = re.sub(r"\s*\((?:complete relation|[^)]*form:[^)]*)\)", "", str(cl.label or "")).strip() |
| import arguments as _A |
| note = _A._perimeter_note(s) if cid in _A._TRAILING else None |
| return (("the " + label) + (f" ({note})" if note else "")) if label else None |
|
|
|
|
| def _rels(s): |
| return {rid: {"lhs": x.get("lhs_claim_id"), "rhs": x.get("rhs_claim_id"), "magnitude_claim": x.get("presentation_claim_id"), |
| "type": x.get("relation_type")} |
| for rid, x in (((s.get("relations.state") or {}).get("items")) or {}).items()} |
|
|
|
|
| def _arg_item(r, s, cat, args, p, head=None): |
| """One structured point / item: [head…, view, HEAD_END, " — ", evidence…]. Evidence the report has already printed |
| in full is printed compactly (the report's evidence registry).""" |
| import arguments as A |
| r.authored_seen.add(p["view"]) |
| frs = list(head or []) + [_authored(p["view"]), Fragment("", origin=HEAD_END)] |
| ev = A.render_evidence(r, p.get("evidence") or [], s, cat, _rels(s), seen=r.seen_evidence) |
| return frs + ([Fragment(" — ")] + ev if ev else []) |
|
|
|
|
| def _fresh(r, text): |
| """True the first time an authored text is printed in this report (a text is never printed twice).""" |
| return bool(text) and text not in r.authored_seen |
|
|
|
|
| def _admit(r, objs): |
| """The objects whose text has not been printed yet, each registered as it is admitted (two identical views in |
| one list print once).""" |
| out = [] |
| for p in objs or []: |
| if _fresh(r, p.get("view")): |
| r.authored_seen.add(p["view"]) |
| out.append(p) |
| return out |
|
|
|
|
| def _stance_text(role, stance, label): |
| who = _SEAT_NAMES.get(role, role) |
| if stance == "REVISED_LABEL": |
| return f"The {who} changes its rating after the exchange, to {label}." if label else f"The {who} changes its rating after the exchange." |
| if stance == "REVISED_RETURN": |
| return f"The {who} revises its return after the exchange; its rating stays {label}." if label else \ |
| f"The {who} revises its return after the exchange." |
| return f"The {who} holds its case." |
|
|
|
|
| def _empty_seat_note(s, role): |
| """A seat with no publishable reasoning: whose the return and the probability are, and what the portfolio manager |
| knew when it set the probability (scenarios.pm_evidence_state, recorded before the PM round).""" |
| st = (s.scenarios.get("pm_evidence_state") or {}).get(role) or {} |
| told = ", who was told this seat had no accepted supporting arguments" if st.get("accepted") == 0 else "" |
| return ("This analyst's written reasoning is not shown: none of it met the report's evidence standards. The scenario " |
| f"return is the analyst's own; its probability was assigned by the portfolio manager{told}.") |
|
|
|
|
| def _structured_debate(B, r, s, cat, args, seats, order, _notes): |
| """The debate in round order: opening cases, rebuttals (each answer quotes the point it answers), the decision.""" |
| import arguments as A |
|
|
| def opening(role): |
| o = args.get(f"{role}_open") |
| if o is None and args.get(role) is not None and "answers" not in args[role]: |
| o = args[role] |
| return o |
|
|
| for role, name, ret_id, p_id in order: |
| if role not in seats and opening(role) is None: |
| continue |
| B.append(_para("debate", r.t(f"{name} — "), r.c(ret_id), |
| r.t(f" · {(s.scenarios.get(role) or {}).get('label') or ''} · probability "), r.c(p_id), css="seat-head")) |
| o = opening(role) |
| if o is not None: |
| th = o.get("thesis") if _fresh(r, o.get("thesis")) else None |
| if th: |
| r.authored_seen.add(th) |
| pts = _admit(r, o.get("points")) |
| if th: |
| B.append(Block("para", "debate", [_authored(th)], title="Opening case" if role != "neutral" else None)) |
| if pts: |
| B.append(Block("points", "debate", [_arg_item(r, s, cat, args, p) for p in pts], |
| title=None if th else ("Opening case" if role != "neutral" else None))) |
| if not o.get("thesis") and not o.get("points"): |
| B.append(_para("debate", r.t(_empty_seat_note(s, role)))) |
| elif not th and not pts: |
| B.append(_para("debate", r.t("This analyst's validated points are printed above."))) |
| else: |
| leg = seats.get(f"{role}_open") or seats.get(role) or {} |
| if leg.get("reasoning") or leg.get("key_points"): |
| _notes(leg) |
| else: |
| B.append(_para("debate", r.t(_empty_seat_note(s, role)))) |
| replies = [(role, opp) for role, opp in (("bear", "bull"), ("bull", "bear")) |
| if (args.get(role) or {}).get("answers") is not None and f"{role}_open" in args] |
| for role, opp in replies: |
| a = args[role] |
| B.append(_para("debate", r.t(f"The {role} replies to the {opp}"), css="seat-head")) |
| B.append(_para("debate", r.t(_stance_text(role, a.get("stance"), (s.scenarios.get(role) or {}).get("label"))))) |
| if _fresh(r, a.get("thesis")): |
| r.authored_seen.add(a["thesis"]) |
| B.append(Block("para", "debate", [_authored(a["thesis"])])) |
| items = [] |
| for x in a.get("answers") or []: |
| if not _fresh(r, x.get("view")): |
| continue |
| head = [Fragment(f"On the {opp}'s point “"), Fragment(str(x["to_view"]), origin=QUOTE), |
| Fragment(f"” — {A.VERDICT_CHIP.get(x['verdict'], x['verdict'])}: ")] |
| items.append(_arg_item(r, s, cat, args, x, head=head)) |
| if items: |
| B.append(Block("points", "debate", items)) |
| pmx = args.get("pm") |
| B.append(_para("debate", r.t("Portfolio manager — probability-weighted "), r.c("PM_RETURN"), |
| r.t(f" · {s.scenarios.get('pm_label') or ''}"), css="seat-head")) |
| |
| if all(c in cat for c in ("SCENARIO_BEAR_PROB", "SCENARIO_NEUTRAL_PROB", "SCENARIO_BULL_PROB", "PM_RETURN_PHRASE")): |
| B.append(_para("debate", r.t("The portfolio manager weights the bear case at "), r.c("SCENARIO_BEAR_PROB"), |
| r.t(", the neutral case at "), r.c("SCENARIO_NEUTRAL_PROB"), r.t(" and the bull case at "), |
| r.c("SCENARIO_BULL_PROB"), r.t(", for "), r.c("PM_RETURN_PHRASE"), r.t("."))) |
| if pmx and (pmx.get("thesis") or pmx.get("points")): |
| if _fresh(r, pmx.get("thesis")): |
| r.authored_seen.add(pmx["thesis"]) |
| B.append(Block("para", "debate", [_authored(pmx["thesis"])])) |
| pts = _admit(r, pmx.get("points")) |
| if pts: |
| B.append(Block("points", "debate", [_arg_item(r, s, cat, args, p) for p in pts], title="Deciding factors")) |
| elif (seats.get("pm") or {}).get("reasoning") or (seats.get("pm") or {}).get("key_points"): |
| _notes(seats["pm"]) |
|
|
|
|
| def guidance_table_rows(s, cat): |
| """[(n, row)] of the company-level guidance rows the report prints (segment-named rows are left out). A guidance |
| value an author cites must be one of these.""" |
| g = s.guidance.get("company") or {} |
| if g.get("status") != "AVAILABLE": |
| return [] |
| segn = {str(x).lower() for sg in ((s.segments.get("canonical") or {}).get("segments") or []) |
| for x in [sg.get("name")] + [p.get("name") for p in sg.get("product_lines") or []] if x} |
| return [(i, m) for i, m in enumerate(g.get("metrics") or [], 1) |
| if f"GUIDANCE_{i}" in cat and not any(n and n in str(m.get("label", "")).lower() for n in segn)] |
|
|
|
|
| def _split_topic(frags): |
| """("Topic", rest) for a key point written as "Topic: point" (the analysts' format), else (None, frags).""" |
| if frags and not frags[0].claim_id: |
| t = frags[0].text |
| k = t.find(": ") |
| if 0 < k <= 80 and (t[k + 2:].strip() or len(frags) > 1): |
| return t[:k], ([Fragment(t[k + 2:])] if t[k + 2:] else []) + list(frags[1:]) |
| return None, frags |
|
|
|
|
| OPEN_SECTIONS = ("verdict", "summary") |
| HEADLINE_SECTIONS = ("valuation",) |
|
|
|
|
| def render_html(rep: UserReport): |
| H = [f'<article class="crucible-report" data-version="{_html.escape(rep.version)}">'] |
| cur, fold = None, False |
| for b in rep.blocks: |
| if b.section != cur: |
| if cur is not None: |
| H.append("</details>" if fold else "</section>") |
| fold = b.section not in OPEN_SECTIONS and b.kind == "heading" |
| H.append(f'<details class="rp-sec rp-more rp-{b.section}">' if fold else f'<section class="rp-sec rp-{b.section}">') |
| cur = b.section |
| if b.kind == "heading": |
| tag = "h1" if b.section == "verdict" else "h2" |
| inner = f"<{tag}>" + "".join(_frag_html(f) for f in b.content) + f"</{tag}>" |
| H.append(f"<summary>{inner}</summary>" if (fold and tag == "h2") else inner) |
| elif b.kind == "points": |
| if b.title: |
| H.append(f"<h3>{_html.escape(b.title)}</h3>") |
| items = [] |
| for item in b.content: |
| k = next((i for i, f in enumerate(item) if f.origin == HEAD_END), None) |
| if k is not None: |
| hh = "".join(_frag_html(f) for f in item[:k]) |
| rest = [f for f in item[k + 1:]] |
| if rest and not rest[0].claim_id and rest[0].text.strip() == "—": |
| rest = rest[1:] |
| body = "".join(_frag_html(f) for f in rest) |
| items.append(f"<li><details><summary>{hh}</summary><p>{body}</p></details></li>" if body.strip() |
| else f"<li>{hh}</li>") |
| continue |
| topic, rest = _split_topic(item) |
| if topic: |
| items.append(f'<li><details><summary>{_html.escape(topic)}</summary><p>' |
| + "".join(_frag_html(f) for f in rest) + "</p></details></li>") |
| else: |
| items.append("<li>" + "".join(_frag_html(f) for f in item) + "</li>") |
| H.append('<ul class="rp-points">' + "".join(items) + "</ul>") |
| elif b.kind == "tiles": |
| cells = [] |
| for row in b.content.get("rows", []): |
| lab, val, note = ("".join(_frag_html(f) for f in cell) for cell in (row + [[], [], []])[:3]) |
| cells.append(f'<div class="rp-tile"><div class="rp-tile-l">{lab}</div><div class="rp-tile-v">{val}</div>' |
| f'<div class="rp-tile-n">{note}</div></div>') |
| H.append('<div class="rp-tiles">' + "".join(cells) + "</div>") |
| elif b.kind in ("para", "callout", "bullets", "table") and b.title and b.section in HEADLINE_SECTIONS: |
| H.append(f'<details class="rp-sub"><summary>{_html.escape(b.title)}</summary>') |
| H.append(_block_body(b)) |
| H.append("</details>") |
| elif b.kind in ("para", "callout"): |
| if b.title: |
| H.append(f'<h4 class="rp-round">{_html.escape(b.title)}</h4>') |
| H.append(_block_body(b)) |
| elif b.kind == "bullets": |
| if b.title: |
| H.append(f"<h3>{_html.escape(b.title)}</h3>") |
| H.append(_block_body(b)) |
| elif b.kind == "table": |
| if b.title: |
| H.append(f"<h3>{_html.escape(b.title)}</h3>") |
| H.append(_block_body(b)) |
| if cur is not None: |
| H.append("</details>" if fold else "</section>") |
| H.append("</article>") |
| return "\n".join(H) |
|
|
|
|
| def _block_body(b): |
| """A paragraph, list or table without its title.""" |
| if b.kind in ("para", "callout"): |
| return f'<p class="{b.css or ""}">' + "".join(_frag_html(f) for f in b.content) + "</p>" |
| if b.kind == "bullets": |
| return "<ul>" + "".join("<li>" + "".join(_frag_html(f) for f in item) + "</li>" for item in b.content) + "</ul>" |
| if b.kind == "table": |
| heads = b.content.get("head", []) |
| head = "".join(f"<th>{_html.escape(h)}</th>" for h in heads) |
| |
| rows = "".join("<tr>" + "".join(f'<td data-label="{_html.escape(heads[j] if j < len(heads) else "")}">' |
| + "".join(_frag_html(f) for f in cell) + "</td>" for j, cell in enumerate(row)) + "</tr>" |
| for row in b.content.get("rows", [])) |
| kv = " rp-kv" if len(b.content.get("head", [])) == 2 else "" |
| css = f" {b.css}" if b.css else "" |
| return f'<div class="rp-scroll"><table class="rp-t{kv}{css}"><thead><tr>{head}</tr></thead><tbody>{rows}</tbody></table></div>' |
| return "" |
|
|
|
|
| def _method_basis(m): |
| """What a composite multiple is applied to, and the basis it was set on: a target multiple set on a trailing basis |
| and applied to next-twelve-month figures is not the company's trailing multiple.""" |
| raw = str(m.get("raw_input_basis") or "").lower() |
| period = ("next-12-month " if ("ntm" in raw or "next" in raw) else "forecast " if "forecast" in raw |
| else "trailing " if ("ttm" in raw or "trailing" in raw) else "") |
| noun = ("adjusted EPS" if "adjusted eps" in raw else "EPS" if "eps" in raw else "free cash flow per share" if "fcf" in raw |
| else "revenue per share" if "revenue" in raw else "ebitda" if "ebitda" in raw else None) |
| set_on = {"TRAILING_ADJUSTED": "trailing", "TRAILING": "trailing", "TRAILING_GAAP": "trailing", |
| "FORWARD_ADJUSTED": "forward"}.get(m.get("multiple_basis")) |
| if noun is None: |
| return _BASIS_WORDS.get(m.get("multiple_basis"), "") |
| |
| |
| whose = "the engine's " if ("reconciledforecaststate" in raw or raw.startswith("forecast")) else "" |
| return f" on {whose}{period}{noun}" + (f" (multiple set on a {set_on} basis)" if set_on and not period.startswith(set_on) else "") |
|
|
|
|
| _BASIS_WORDS = {"TRAILING_ADJUSTED": " trailing, on adjusted EPS", "FORWARD_ADJUSTED": " forward, on adjusted EPS", |
| "TRAILING": " trailing", "TRAILING_GAAP": " trailing, on reported EPS"} |
| _SPAN = re.compile(r'<span class="cl" data-claim="([^"]+)">(.*?)</span>') |
| |
| |
| _QA_ID = r"[ADFXRS]\d{1,2}" |
| QA_VOCAB = re.compile(r"\bSEMANTIC_[A-Z_]+|\bFINAL_RENDER\w*|publication[- ](?:blocked|ready)|\bStage [AB]\b|" |
| r"\bvalidator\b|(?-i:\bPASS\b|\bFAIL\b)|NOT_APPLICABLE|" |
| rf"\b(?:check|checks|gate|audit|rule|validation)\s+{_QA_ID}\b|" |
| rf"\b{_QA_ID}\s+(?:check|gate|audit|passed|failed|warning|blocked|blocking)\b", re.I) |
|
|
|
|
| |
| |
| |
| _IDENT = IDENTIFIER |
|
|
|
|
| def audit(rep: UserReport, state, md: str, html_doc: str, quant_hidden_values=()): |
| from validation import FAIL, NA_, PASS, CheckResult, ValidationResult, _run |
| frags = rep.fragments() |
|
|
| def numbers(): |
| bad = [f.text for f in frags if not f.claim_id and MATERIAL_NUMBER.search(BENIGN.sub(" ", f.text))] |
| return (FAIL, f"{len(bad)} material number(s) outside a claim: {bad[:3]}") if bad else \ |
| (PASS, f"every material number is one of {len(rep.claims)} registered claims") |
|
|
| def values(): |
| bad = [] |
| for f in frags: |
| if not f.claim_id or f.claim_id == "VERSION": |
| continue |
| cd = C.claim_def(state, f.claim_id) |
| if cd is not None: |
| want = C.claim_for(state, cd).rendered |
| elif f.claim_id.startswith("P:"): |
| want = C.path_claim(state, f.claim_id[2:], rep.claims[f.claim_id]["stype"], "").rendered |
| else: |
| bad.append(f"{f.claim_id}: unknown claim") |
| continue |
| if want != f.text and not (cd is not None and (cd.stype == C.REL or cd.clause) and want and f.text == want[0].upper() + want[1:]): |
| bad.append(f"{f.claim_id}: shows '{f.text}', state gives '{want}'") |
| vf = [f for f in frags if f.claim_id == "VERSION"] |
| if not vf or any(f.text != state.version for f in vf): |
| bad.append("the document does not carry the frozen state's version") |
| return (FAIL, "; ".join(bad[:4])) if bad else (PASS, "every claim fragment equals its value in the frozen state") |
|
|
| def identity(): |
| from prose import label_violations |
| bad = [] |
| for unit in (u for b in rep.blocks for u in b.units()): |
| bad += [f"{cid} labelled '{w}'" for cid, w in |
| label_violations(unit, lambda c: getattr(C.claim_def(state, c), "prohibited", ()))] |
| return (FAIL, "; ".join(bad[:4])) if bad else (PASS, "no claim carries a label its type forbids") |
|
|
| def terminology(): |
| terms = (state.valuation.get("engine") or {}).get("forbidden_renderer_terms") or [] |
| bad = set() |
| for unit in (u for b in rep.blocks for u in b.units()): |
| for sent in re.split(r"(?<=[.!?])\s+", "".join(f.text for f in unit).lower()): |
| for t in terms: |
| |
| if re.search(r"\b" + re.escape(t.lower()) + r"(?![a-z])", sent) and not DENIAL.search(sent): |
| bad.add(t) |
| return (FAIL, f"forbidden valuation terms present: {sorted(bad)}") if bad else (PASS, "valuation vocabulary is the selected engine's") |
|
|
| def no_qa(): |
| text = "".join(f.text for f in frags if not f.claim_id) |
| m = QA_VOCAB.search(text) |
| return (FAIL, f"internal QA vocabulary in the user report: '{m.group(0)}'") if m else (PASS, "no internal QA vocabulary") |
|
|
| def quant_hidden(): |
| leaks = [] |
| pm_vis = (((state.quant.get("price_model") or {}).get("reliability") or {}).get("persona_visibility")) |
| pr_vis = (((state.quant.get("panel_rank") or {}).get("reliability") or {}).get("persona_visibility")) |
| for f in frags: |
| if f.claim_id and f.claim_id.startswith("P:quant.") and not (pm_vis if "price_model" in f.claim_id else pr_vis): |
| leaks.append(f.claim_id) |
| for i, src in enumerate(state.blend.get("sources") or ()): |
| if src.get("kind") == "PRICE_MODEL" and not pm_vis: |
| leaks.append(f"blend source {src.get('name')}") |
| prose_text = "".join(f.text for f in frags if not f.claim_id) |
| leaks += [v for v in quant_hidden_values if v and v in prose_text] |
| return (FAIL, f"an unreliable model's figures appear: {leaks}") if leaks else (PASS, "unreliable models contribute no user-visible figure") |
|
|
| def same_bytes(): |
| spans = _SPAN.findall(html_doc) |
| want = [(f.claim_id, _html.escape(f.text)) for f in frags if f.claim_id] |
| if spans != want: |
| return FAIL, f"HTML claim spans ({len(spans)}) differ from the report fragments ({len(want)})" |
| if render_markdown(rep) != md: |
| return FAIL, "markdown is not the serialisation of the audited fragments" |
| return PASS, f"HTML and markdown carry the same {len(want)} claim fragments" |
|
|
| def required(): |
| need = ["RATING", "HEADLINE_RETURN", "PRICE", "PM_RETURN", "VERSION"] |
| need += [cid for cid, path in (("MARKET_COMPOSITE", "pricing.market_composite"), ("FORECAST_EPS", "forecast.eps")) |
| if state.value(path) is not None] |
| if state.value("valuation.intrinsic_value") is not None: |
| need.append("INTRINSIC_VALUE") |
| if state.value("consensus.eps_fy1") is not None: |
| need.append("CONSENSUS_EPS_FY1") |
| have = {f.claim_id for f in frags if f.claim_id} |
| miss = [x for x in need if x not in have] |
| return (FAIL, f"required claims not rendered: {miss}") if miss else (PASS, f"{len(need)} required claims rendered") |
|
|
| |
| from prose import guidance_violation, motive_violation, qualifier_violation, SENT |
|
|
| def _sentences(): |
| for b in rep.blocks: |
| for unit in b.units(): |
| if any(f.origin in ("authored", HEAD_END, QUOTE) for f in unit): |
| continue |
| cur = [] |
| for f in unit: |
| cur.append(f) |
| if not f.claim_id and re.search(r"[.!?](\s|$)", f.text): |
| yield cur |
| cur = [] |
| if cur: |
| yield cur |
|
|
| def guidance_evidence(): |
| bad = ["".join(f.text for f in u)[:120] for u in _sentences() if guidance_violation(u)] |
| g = (state.guidance.get("company") or {}).get("status") |
| return (FAIL, f"guidance described without a guidance claim: {bad[:2]}") if bad else \ |
| (PASS, f"company guidance {g}; every mention cites a verified guidance claim") |
|
|
| def _motive(kind): |
| bad = [] |
| for u in _sentences(): |
| mv = motive_violation("".join(f.text for f in u)) |
| if mv and mv[0] == kind: |
| bad.append("".join(f.text for f in u)[:120]) |
| return (FAIL, f"motive attributed to {kind} filings: {bad[:2]}") if bad else (PASS, f"no motive inferred from {kind} filings") |
|
|
| def _pairs_ok(): |
| from prose import implicit_comparison_violation, ordering_violation |
| _cat = C.build_catalog(state, include_deferred=False) |
| cp, ds = C.implicit_counterparts(state, _cat), C.claim_descriptors(state, _cat) |
| bad = [] |
| for u in _sentences(): |
| tok = "".join(f"[[{f.claim_id}]]" if (f.claim_id and not f.claim_id.startswith("P:")) else f.text for f in u) |
| cv = implicit_comparison_violation(tok, cp, ds) or ordering_violation(tok, lambda c: C.claim_range(state, c)) |
| if cv: |
| bad.append(f"{cv[0]} vs {cv[1]}: {'; '.join(cv[2])[:80]}") |
| return (FAIL, f"incompatible comparisons in the report: {bad[:2]}") if bad else (PASS, "no stated ordering contradicts the values") |
|
|
| |
| import evidence as _EV |
| import guidance as _GD |
| import prose as _P |
| import relations as _RL |
| _rs = state.relations.get("state") or {} |
| _pairs = _RL.pair_index(_rs) |
| _R = _P.Rules(catalog=C.build_catalog(state, include_deferred=True), pair_fn=lambda a, b: C.pair_status(state, a, b, _pairs), |
| magnitude_bindings=_RL.magnitude_bindings(_rs), claim_model=_EV.claim_model, |
| evidence_targets=lambda c: _EV.evidence_targets(state, c), |
| guidance_terms={c: (_GD.epistemic_state(state.guidance.get("company"), c), n) for c, n in _GD.CONCEPT_TERMS.items()}, |
| expectations_polarity=((_rs.get("items") or {}).get("REL_EXPECTATIONS_STORY") or {}).get("relation_type")) |
| import pipeline as _PL |
| for _k, _v in _PL.REL_RULES(state, _R.catalog).items(): |
| setattr(_R, _k, _v) |
|
|
| def _authored(): |
| for u in _sentences(): |
| if any(getattr(f, "origin", "system") == "llm" for f in u): |
| yield u, "".join(f"[[{f.claim_id}]]" if (f.claim_id and not f.claim_id.startswith("P:")) else f.text for f in u) |
|
|
| def _rel_kind(kind): |
| def run(): |
| bad = [] |
| for u, tok in _authored(): |
| v = _P.relation_violation(tok, _R) |
| if v and v[0] == kind: |
| bad.append(v[1][:110]) |
| if kind == "RELATION_IDENTITY": |
| pres = set(C.RELATION_PRESENTATION) |
| bad += [f"[[{f.claim_id}]] in authored prose (a relation's size without its subject)" |
| for u, _t in _authored() for f in u if f.claim_id in pres and f.origin == "llm"] |
| bad += [p for k, p in _RL.check(state) if k == "REL1"] |
| if kind == "RELATION_MAGNITUDE": |
| bad += [p for k, p in _RL.check(state) if k == "REL2"] |
| return (FAIL, "; ".join(bad[:2])) if bad else (PASS, f"{len(_rs.get('items') or {})} typed relations hold in the text") |
| return run |
|
|
| def _evidence_scope(): |
| bad = [(_P.evidence_scope_violation(tok, _R) or "")[:110] for u, tok in _authored() if _P.evidence_scope_violation(tok, _R)] |
| return (FAIL, "; ".join(bad[:2])) if bad else (PASS, "model evidence describes only its own model") |
|
|
| def _evidence_visibility(): |
| bad = [f.claim_id for u, _t in _authored() for f in u |
| if f.claim_id and f.origin == "llm" and _EV.claim_model(f.claim_id) |
| and not _EV.model_visible(state, _EV.claim_model(f.claim_id))] |
| return (FAIL, f"hidden models' figures in authored prose: {sorted(set(bad))}") if bad else (PASS, "no hidden model's figure in authored prose") |
|
|
| def _epistemic(): |
| bad = [tok[:110] for u, tok in _authored() if _P.guidance_epistemic_violation(tok, _R.guidance_terms)] |
| return (FAIL, f"guidance presence/absence asserted beyond its epistemic state: {bad[:2]}") if bad else \ |
| (PASS, "no guidance presence or absence asserted beyond its epistemic state") |
|
|
| def _polarity(): |
| bad = [tok[:110] for u, tok in _authored() if _P.polarity_violation(tok, _R.expectations_polarity)] |
| return (FAIL, f"expectation polarity not from the structured relation: {bad[:2]}") if bad else \ |
| (PASS, "expectation polarity only as REL_EXPECTATIONS_STORY states it") |
|
|
| def _returns(): |
| bad = [f.claim_id for u, _t in _authored() for f in u if f.origin == "llm" and f.claim_id in C.PHRASE_OF] |
| for f in frags: |
| cd = C.claim_def(state, f.claim_id) if f.claim_id and not f.claim_id.startswith("P:") else None |
| if cd is not None and cd.stype == C.PHRASE: |
| rc = C.return_claim(state.value(cd.path), cd.phrase) |
| if rc is None or f.text not in (rc["rendered"], rc["rendered"][:1].upper() + rc["rendered"][1:]): |
| bad.append(f"{f.claim_id}: phrase does not carry the sign of its value") |
| return (FAIL, f"signed figures without their own direction: {bad[:3]}") if bad else (PASS, "signed quantities carry their own direction") |
|
|
| def _doubled(): |
| bad = [tok[:110] for u, tok in _authored() if _P.qualifier_doubled(tok, _R.catalog) or _P.operand_doubled(tok, _R.catalog)] |
| return (FAIL, f"a claim's own qualifier written twice: {bad[:2]}") if bad else (PASS, "no qualifier written twice") |
|
|
| def _clause_grammar(): |
| bad = [tok[:110] for u, tok in _authored() |
| if _P.clause_grammar_violation(tok, _R.full_clause) or _P.mode_subject_violation(tok, _R.rel_subjects) |
| or _P.relation_slot_violation(tok, _R.rel_subjects, _R.catalog)] |
| return (FAIL, f"relationship clauses used as noun phrases or attached to another subject: {bad[:2]}") if bad else \ |
| (PASS, "every relationship token sits in a clause position, at most two per sentence") |
|
|
| def _trends(): |
| bad = [tok[:110] for u, tok in _authored() if _P.trend_violation(tok, _R)] |
| return (FAIL, f"trend conclusions without an approved trend relationship: {bad[:2]}") if bad else \ |
| (PASS, f"trend conclusions only from approved trend relationships ({len(_R.trends_allowed)} available)") |
|
|
| def _interpretation(): |
| bad = [tok[:110] for u, tok in _authored() if _P.interpretation_violation(tok, _R.interpretation)] |
| perm = {f: d.get("permission") for f, d in (_R.interpretation or {}).items()} |
| return (FAIL, f"descriptive-only evidence interpreted: {bad[:2]}") if bad else (PASS, f"evidence interpretation within permission {perm}") |
|
|
| def _guidance_binding(): |
| bad = [tok[:110] for u, tok in _authored() if _P.guidance_binding_violation(tok, _R.guidance_names)] |
| return (FAIL, f"guidance values attached to another subject: {bad[:2]}") if bad else \ |
| (PASS, "every authored guidance value is named by its own concept") |
|
|
| def _final_claim_only(): |
| """Cover and summary: every material number is a claim fragment whose text and value are the frozen state's.""" |
| bad = [] |
| for b in rep.blocks: |
| if b.section not in ("verdict", "summary"): |
| continue |
| for f in b.fragments(): |
| if not f.claim_id: |
| if MATERIAL_NUMBER.search(BENIGN.sub(" ", f.text)): |
| bad.append(f"authored number '{MATERIAL_NUMBER.search(BENIGN.sub(' ', f.text)).group(0).strip()}'") |
| continue |
| cd = C.claim_def(state, f.claim_id) if not f.claim_id.startswith("P:") else None |
| if f.claim_id.startswith("P:") or f.claim_id == "VERSION": |
| continue |
| if cd is None: |
| bad.append(f"{f.claim_id}: not in the final claim registry") |
| continue |
| cl = C.claim_for(state, cd) |
| want = cl.rendered or "" |
| if f.text != want and not ((cd.stype == C.REL or cd.clause) and want and f.text == want[0].upper() + want[1:]): |
| bad.append(f"{f.claim_id}: shows '{f.text}', final registry has '{want}'") |
| elif f.value is not None and cl.value is not None and not isinstance(cl.value, dict) and f.value != cl.value: |
| bad.append(f"{f.claim_id}: value {f.value} ≠ final {cl.value}") |
| return (FAIL, "; ".join(bad[:3])) if bad else (PASS, "cover and summary numbers are final claims only") |
|
|
| def _ownership_phrase(): |
| bad = [f.claim_id for u, _t in _authored() for f in u if f.origin == "llm" and f.claim_id == "OWN_13F_HOLDERS_CHANGE"] |
| d = C.BY_ID.get("OWN_13F_HOLDERS_CHANGE_PHRASE") |
| for f in frags: |
| if f.claim_id == "OWN_13F_HOLDERS_CHANGE_PHRASE": |
| rc = C.return_claim(state.value(d.path), d.phrase) |
| if rc is None or f.text not in (rc["rendered"], rc["rendered"][:1].upper() + rc["rendered"][1:]): |
| bad.append("the ownership-change phrase does not own its direction, magnitude and period") |
| return (FAIL, "; ".join(map(str, bad[:3]))) if bad else (PASS, "ownership change appears only as its complete phrase") |
|
|
| def _relation_logic(): |
| bad = [tok[:110] for u, tok in _authored() if _P.relation_logic_violation(tok, _R.alternative_cases, _R.case_variables)] |
| return (FAIL, f"alternative conditional cases stated jointly: {bad[:2]}") if bad else \ |
| (PASS, "alternative break-even cases are never stated as a joint requirement") |
|
|
| def _valuation_relations(): |
| bad = [tok[:110] for u, tok in _authored() if _P.valuation_relation_violation(tok, _R.valuation_relations) |
| or _P.universal_valuation_violation(tok, _R.valuation_relations)] |
| return (FAIL, f"valuation direction words without their relation: {bad[:2]}") if bad else \ |
| (PASS, "valuation direction words cite and agree with their relation") |
|
|
| def _segment_perimeter(): |
| bad = [tok[:110] for u, tok in _authored() if _P.segment_perimeter_violation(tok, _R.segment_perimeter)] |
| return (FAIL, f"segment counts outside their perimeter: {bad[:2]}") if bad else \ |
| (PASS, "segment counts cited with their perimeter") |
|
|
| def _evidence_domain(): |
| bad = [tok[:110] for u, tok in _authored() if _P.evidence_domain_violation(tok, _R)] |
| return (FAIL, f"evidence used outside its domain: {bad[:2]}") if bad else (PASS, "evidence establishes only its own domain") |
|
|
| def _lineage(): |
| """Every material number in the document is a claim of the FINAL frozen state (text and value), a guidance value |
| is an AVAILABLE row of the final guidance state, and every guidance value an author cites is in the table.""" |
| bad = [] |
| rows = (state.guidance.get("company") or {}).get("metrics") or [] |
| for f in frags: |
| if not f.claim_id: |
| mm = MATERIAL_NUMBER.search(BENIGN.sub(" ", f.text)) |
| if mm: |
| bad.append(f"'{mm.group(0).strip()}' has no claim") |
| continue |
| if f.claim_id.startswith("P:") or f.claim_id == "VERSION": |
| continue |
| cd = C.claim_def(state, f.claim_id) |
| if cd is None: |
| bad.append(f"{f.claim_id}: not in the final state") |
| continue |
| if f.claim_id.startswith("GUIDANCE_"): |
| try: |
| row = rows[int(f.claim_id.split("_")[1]) - 1] |
| except (ValueError, IndexError): |
| bad.append(f"{f.claim_id}: no such final guidance row") |
| continue |
| if (state.guidance.get("company") or {}).get("status") != "AVAILABLE" or not row.get("verified"): |
| bad.append(f"{f.claim_id}: not an available, verified row") |
| table = {f.claim_id for b in rep.blocks if b.kind == "table" and b.section == "forecast" |
| for f in b.fragments() if f.claim_id and f.claim_id.startswith("GUIDANCE_")} |
| cited = {f.claim_id for u, _t in _authored() for f in u if f.claim_id and f.claim_id.startswith("GUIDANCE_")} |
| bad += [f"{c}: cited in prose but absent from the guidance table" for c in sorted(cited - table)] |
| return (FAIL, "; ".join(bad[:3])) if bad else (PASS, "every material number resolves to a final claim") |
|
|
| def _typed_rule(fn, what, ok): |
| def run(): |
| bad = [tok[:110] for u, tok in _authored() if fn(tok)] |
| return (FAIL, f"{what}: {bad[:2]}") if bad else (PASS, ok) |
| return run |
|
|
| |
| _lin = {} |
|
|
| def _lineage_hits(): |
| """Every authored object (own evidence) and thesis (named counterparts) through arguments.lineage_reasons, and |
| every template sentence by co-citation: (code, where, text).""" |
| if "v" in _lin: |
| return _lin["v"] |
| import arguments as _A |
| hits = [] |
| cat0 = C.build_catalog(state, include_deferred=False) |
| ctx = _A.context(state, cat0, _PL.rules_for(state, cat0), _PL.INTERNAL_TERMS, QA_VOCAB) |
| for role, a in (rep.arguments or {}).items(): |
| if not a: |
| continue |
| for key in ("points", "items_thesis", "items_risks", "answers"): |
| for p in a.get(key) or []: |
| ev = list(p.get("evidence") or []) + list(p.get("to_evidence") or []) |
| hits += [(x.split(":")[0], f"{role}.{key}", p.get("view") or "") for x in _A.lineage_reasons(ctx, p.get("view"), ev)] |
| if a.get("thesis"): |
| sc = _A._scope(_A._rest_objects(role, a)) |
| hits += [(x.split(":")[0], f"{role}.thesis", a["thesis"]) |
| for x in _A.lineage_reasons(ctx, a["thesis"], list(sc), cocite=False)] |
| for u in _sentences(): |
| ids = [f.claim_id for f in u if f.claim_id and not f.claim_id.startswith("P:")] |
| txt = "".join(f.text for f in u) |
| hits += [(x.split(":")[0], "template", txt) for x in _A.lineage_reasons(ctx, txt, ids, named=False)] |
| _lin["v"] = hits |
| return hits |
|
|
| def _guidance_consensus(): |
| bad = [] |
| it = (_rs.get("items") or {}).get("REL_GUIDANCE_EPS_CONSENSUS") |
| if it: |
| bad += [f"REL_GUIDANCE_EPS_CONSENSUS: {w}" for w in _RL.verify_guidance_relation(state, it)] |
| bad += [f"{w}: {x[:90]}" for c, w, x in _lineage_hits() if c == "GUIDANCE_CONSENSUS_COMPATIBILITY"] |
| return (FAIL, f"guidance compared with consensus without a compatible typed relation: {bad[:2]}") if bad else \ |
| (PASS, "guidance meets consensus only through a relation on one fiscal period, horizon and basis" |
| + ("" if it else " (none is compatible: stated separately)")) |
|
|
| def _forecast_relation(): |
| bad = [] |
| for rid in ("REL_FORECAST_EPS_CONSENSUS", "REL_ENGINE_REVENUE_CONSENSUS"): |
| it = (_rs.get("items") or {}).get(rid) |
| if it: |
| bad += [f"{rid}: {w}" for w in _RL.verify_forecast_relation(state, rid, it)] |
| bad += [f"{w}: {x[:90]}" for c, w, x in _lineage_hits() if c == "FORECAST_RELATION_REQUIRED"] |
| return (FAIL, f"engine forecasts compared with consensus without their relation: {bad[:2]}") if bad else \ |
| (PASS, "engine-vs-consensus comparisons only through independent, same-horizon relations") |
|
|
| def _historical_quality(): |
| """The trailing column's label and the TTM metrics' quality derive from HistoricalMetricQuality; a series with an |
| unchecked quarter is never labelled simply reconciled.""" |
| q = historical_quality(state) |
| bad = [] |
| stored = (state.historical.get("reconciliation") or {}).get("ttm_quality") |
| if stored is not None and dict(q) != dict(stored): |
| bad.append("a quality state exists but the labels were derived without it") |
| qs = q.get("quarters") or [] |
| if qs: |
| chk = [x.get("source") not in ("PROVIDER_ONLY", "UNAVAILABLE") for x in qs] |
| want = "SEC_RECONCILED" if all(chk) else "PARTIALLY_RECONCILED" if any(chk) else "PROVIDER_ONLY" |
| if q.get("status") != want: |
| bad.append(f"quality status {q.get('status')} but its quarters give {want}") |
| unv = [x["period"] for x, ok in zip(qs, chk) if not ok] |
| if list(q.get("unverified_periods") or []) != unv: |
| bad.append(f"unverified periods {q.get('unverified_periods')} ≠ {unv}") |
| if q.get("status") == "SEC_RECONCILED" and q.get("unverified_periods"): |
| bad.append("SEC_RECONCILED with unverified periods") |
| fund = [i for i, b in enumerate(rep.blocks) if b.kind == "table" and b.css == "rp-fund"] |
| if not fund: |
| bad.append("the fundamentals table is missing") |
| for i in fund: |
| h = rep.blocks[i].content.get("head") or ["", "", ""] |
| want = [historical_quality_render(q, sep_) for sep_ in (False, True)] |
| if h[1] not in {w["table_header"] for w in want}: |
| bad.append(f"trailing column labelled '{h[1]}'") |
| foot = "" |
| for nxt in rep.blocks[i + 1:]: |
| if nxt.css != "rp-foot": |
| break |
| foot += " ".join(f.text for f in nxt.content) + " " |
| if not any(w["footnote"] in foot for w in want): |
| bad.append("the trailing column's quality footnote is not printed under the table") |
| if FCST_MARK in str(h[2]) and FCST_MARK + " " not in foot: |
| bad.append("the forecast column's lineage footnote is not printed under the table") |
| mq = {"SEC_RECONCILED": "RECONCILED", "PARTIALLY_RECONCILED": "PARTIALLY_RECONCILED"}.get(q.get("status"), "UNVERIFIED") |
| for mid in ("revenue_ttm", "operating_income_ttm", "net_income_ttm"): |
| m = state.historical.get(mid) |
| if m is not None and m.value is not None and m.quality_status != mq: |
| bad.append(f"{mid} carries {m.quality_status}, its quarters give {mq}") |
| return (FAIL, "; ".join(bad[:3])) if bad else \ |
| (PASS, f"{q.get('status')} (verified through {q.get('verified_through')}; unverified {q.get('unverified_periods') or 'none'})") |
|
|
| def _segment_note(): |
| a = ((state.valuation.get("story") or {}).get("assumptions")) or {} |
| if a.get("margin_proposed") is None or a.get("margin_target") is None or \ |
| abs(float(a["margin_proposed"]) - float(a["margin_target"])) <= 0.0005 or not (state.valuation.get("story") or {}).get("segments"): |
| return PASS, "segment margins are the margin the valuation uses" |
| shown = {f.claim_id for f in frags if f.claim_id} |
| ok = {"P:valuation.story.assumptions.margin_proposed", "P:valuation.story.assumptions.margin_target"} <= shown |
| return (PASS, "the segments' weighted margin and the margin used (capped or floored) are both shown") if ok else \ |
| (FAIL, "segment margins shown without the cap or floor to the margin the valuation uses") |
|
|
| def _structured(): |
| """Every authored fragment belongs to a validated structured object, and every object still validates with |
| the pipeline's own rule inputs (the same functions ingestion used).""" |
| sets = rep.arguments or {} |
| texts = set() |
| for a in sets.values(): |
| if not a: |
| continue |
| if a.get("thesis"): |
| texts.add(a["thesis"]) |
| for key in ("points", "items_thesis", "items_risks", "answers"): |
| for p in a.get(key) or []: |
| texts.add(p["view"]) |
| if p.get("to_view"): |
| texts.add(p["to_view"]) |
| quotes = {p["to_view"] for a in sets.values() if a for p in a.get("answers") or [] if p.get("to_view")} |
| bad = [f.text[:80] for f in frags if f.origin == "authored" and f.text not in texts] |
| bad += [f"quote: {f.text[:70]}" for f in frags if f.origin == QUOTE and f.text not in quotes] |
| if sets: |
| import arguments as _A |
| import pipeline as _PLs |
| try: |
| cat0 = C.build_catalog(state, include_deferred=False) |
| ctx = _A.context(state, cat0, _PLs.rules_for(state, cat0), _PLs.INTERNAL_TERMS, QA_VOCAB) |
| for role, a in sets.items(): |
| if not a or role == "summary": |
| continue |
| _a, drops = _A.revalidate(ctx, a, own={"value": None, "label": None, "return_claim": None}) |
| bad += [f"{role}: {d['reason'][:90]}" for d in drops if not str(d["reason"]).startswith( |
| ("REVALIDATION: VIEW_RATING", "REVALIDATION: VIEW_RETURN_DIRECTION"))] |
| except Exception as e: |
| bad.append(f"revalidation raised {type(e).__name__}: {str(e)[:80]}") |
| return (FAIL, f"structured arguments that do not validate: {bad[:3]}") if bad else \ |
| (PASS, f"{sum(1 for f in frags if f.origin == 'authored')} authored fragments, all from validated objects") |
|
|
| def _identifiers(): |
| """No internal identifier (snake_case or UPPER_SNAKE) reaches the reader, in template or authored text: the |
| PM's triggers printed "If revenue_growth_next_fy falls to 5.0%" in run 8d12481bffa6e2e6.""" |
| bad = [] |
| for f in frags: |
| if f.claim_id == "VERSION": |
| continue |
| m = _IDENT.search(f.text or "") |
| if m: |
| bad.append(m.group(0)) |
| return (FAIL, f"internal identifiers in the document: {sorted(set(bad))[:4]}") if bad else \ |
| (PASS, "no internal identifiers") |
|
|
| def _authored_once(): |
| from collections import Counter |
| cnt = Counter(f.text for f in frags if f.origin == "authored") |
| dup = [t[:70] for t, n in cnt.items() if n > 1] |
| return (FAIL, f"authored text printed more than once: {dup[:2]}") if dup else \ |
| (PASS, "every authored text is printed once") |
|
|
| def _public_language(): |
| bad = [tok[:110] for u, tok in _authored() if _P.public_language_violation(tok, _PL.INTERNAL_TERMS)] |
| return (FAIL, f"internal QA language in the document: {bad[:2]}") if bad else (PASS, "public language only") |
|
|
| def _conclusion(kind): |
| def run(): |
| bad = [] |
| for u, tok in _authored(): |
| v = _P.evidence_conclusion_violation(tok, _R.evidence_uses) |
| if v and v[0] == kind: |
| bad.append(tok[:110]) |
| return (FAIL, f"evidence used beyond its allowed conclusion types: {bad[:2]}") if bad else \ |
| (PASS, "evidence supports only its allowed conclusion types") |
| return run |
|
|
| def _blend_labels(): |
| from state_builder import BLEND_KIND_LABEL |
| rows = [b for b in rep.blocks if b.kind == "table" and b.section == "blend"] |
| shown = ["".join(f.text for f in row[0]) for b in rows for row in b.content.get("rows", [])] |
| bad = [] |
| for x, lab in zip(state.blend.get("sources") or [], shown): |
| want = BLEND_KIND_LABEL.get(x.get("kind")) if x.get("kind") != "OTHER" else f"Other source ({x.get('name')})" |
| if lab != want: |
| bad.append(f"'{lab}' shown for a {x.get('kind')} source (want '{want}')") |
| return (FAIL, "; ".join(bad[:2])) if bad else (PASS, "blend labels derive from typed source kinds") |
|
|
| def _directional(): |
| from prose import directional_violation |
| restricted = tuple(x for x, blocked in ( |
| ("13F", not (state.ownership.get("permissions") or {}).get("use_as_return_signal", False)), |
| ("insider", not (((state.insiders.get("canon") or {}).get("materiality") or {}).get("material")))) if blocked) |
| bad = [] |
| for u in _sentences(): |
| tok = "".join(f"[[{f.claim_id}]]" if f.claim_id else f.text for f in u) |
| if directional_violation(tok, restricted): |
| bad.append(tok[:100]) |
| return (FAIL, f"restricted evidence used directionally: {bad[:2]}") if bad else (PASS, "no restricted evidence carries a direction") |
|
|
| def _relative(): |
| from prose import relative_violation |
| allowed = tuple(k for k, v in (state.pricing.get("relative") or {}).items() if v.get("comparison_allowed")) |
| bad = [("".join(f.text for f in u))[:100] for u in _sentences() if relative_violation("".join(f.text for f in u), allowed)] |
| return (FAIL, f"relative valuation without a comparable peer basis: {bad[:2]}") if bad else (PASS, "relative valuation only on comparable bases") |
|
|
| def stable_wacc(): |
| bad = [] |
| for u in _sentences(): |
| bad += [cid for cid, _q in qualifier_violation(u, lambda c: getattr(C.claim_def(state, c), "required_qualifiers", ()))] |
| return (FAIL, f"stable-period rate written without its qualifier: {bad[:3]}") if bad else \ |
| (PASS, "every stable-period rate carries its stable / mature / terminal qualifier") |
|
|
| checks = [(numbers, "R1", "every material number is a registered claim", True, "presentation"), |
| (values, "R2", "every rendered claim equals the frozen state", True, "identity"), |
| (identity, "R3", "claims keep their semantic identity", True, "identity"), |
| (terminology, "R4", "valuation vocabulary from the selected engine", True, "policy"), |
| (no_qa, "R5", "no internal QA content in the user report", True, "presentation"), |
| (quant_hidden, "R6", "unreliable models invisible to users", True, "policy"), |
| (same_bytes, "R7", "HTML and markdown are the audited fragments", True, "identity"), |
| (required, "R8", "required claims present", True, "presentation"), |
| (guidance_evidence, "S9", "guidance only with canonical guidance evidence", True, "policy"), |
| (lambda: _motive("13F"), "S10", "no motive inferred from 13F ownership", True, "policy"), |
| (lambda: _motive("insider"), "S11", "no motive inferred from insider filings", True, "policy"), |
| (stable_wacc, "S12", "stable-period cost of capital never presented as current", True, "identity"), |
| (_pairs_ok, "S16", "prose comparisons are between compatible typed claims", True, "identity"), |
| (_directional, "S24", "evidence without return-signal permission carries no direction", True, "policy"), |
| (_relative, "S26", "relative valuation requires the same basis and perimeter", True, "policy"), |
| (_rel_kind("RELATION_IDENTITY"), "RELATION_IDENTITY", "relations keep their subject and object", True, "identity"), |
| (_rel_kind("RELATION_MAGNITUDE"), "RELATION_MAGNITUDE", "magnitudes stay bound to the pair they measure", True, "identity"), |
| (_rel_kind("RELATION_REQUIRED_FOR_PAIR"), "RELATION_REQUIRED_FOR_PAIR", |
| "related quantities share a sentence only through an approved relation", True, "identity"), |
| (_evidence_scope, "MODEL_EVIDENCE_SCOPE", "model evidence evaluates only its allowed target models", True, "policy"), |
| (_evidence_visibility, "MODEL_EVIDENCE_VISIBILITY", "hidden models contribute no authored figure", True, "policy"), |
| (_epistemic, "GUIDANCE_EPISTEMIC_STATE", "no absence asserted from rejected or unfound guidance", True, "policy"), |
| (_polarity, "EXPECTATIONS_POLARITY", "expectation polarity only from the structured relation", True, "identity"), |
| (_returns, "RETURN_DIRECTION_RENDER", "signed quantities carry their own direction", True, "presentation"), |
| (_blend_labels, "BLEND_SOURCE_KIND", "blend labels derive from typed source kinds", True, "presentation"), |
| (_doubled, "CLAIM_QUALIFIER_DOUBLED", "no claim's own direction or bound is written twice", True, "presentation"), |
| (_clause_grammar, "RELATION_CLAUSE_GRAMMAR", "relationship clauses keep clause positions", True, "presentation"), |
| (_trends, "TREND_RELATION_REQUIRED", "trend conclusions need an approved trend relationship", True, "identity"), |
| (_interpretation, "EVIDENCE_INTERPRETATION", "descriptive-only evidence carries no interpretation", True, "policy"), |
| (_guidance_binding, "GUIDANCE_LABEL_BINDING", "guidance values are named by their own concept", True, "identity"), |
| (_final_claim_only, "FINAL_CLAIM_ONLY", "cover and summary numbers resolve to final claims", True, "identity"), |
| (_ownership_phrase, "OWNERSHIP_CHANGE_PHRASE", "ownership change only as its complete phrase", True, "presentation"), |
| (_relation_logic, "RELATION_LOGIC_PRESERVATION", "paraphrase preserves each relation's logical structure", True, "identity"), |
| (_valuation_relations, "VALUATION_RELATION_LANGUAGE", "valuation direction language follows its relation", True, "identity"), |
| (_segment_perimeter, "SEGMENT_PERIMETER_LANGUAGE", "segment counts carry their perimeter", True, "identity"), |
| (_evidence_domain, "EVIDENCE_DOMAIN_SCOPE", "evidence establishes only its own domain", True, "policy"), |
| (_lineage, "FINAL_MATERIAL_CLAIM_LINEAGE", "every material number resolves to a final claim", True, "identity"), |
| (_public_language, "PUBLIC_LANGUAGE_ONLY", "no internal QA language in the document", True, "presentation"), |
| (_conclusion("EVIDENCE_CONCLUSION_SCOPE"), "EVIDENCE_CONCLUSION_SCOPE", "evidence supports only allowed conclusions", |
| True, "policy"), |
| (_conclusion("NEWS_EVIDENCE_SCOPE"), "NEWS_EVIDENCE_SCOPE", "news supports only its classified uses", True, "policy"), |
| (_typed_rule(lambda t: _P.performance_violation(t, _R.stock_returns, _R.company_words), "stock performance without its return claim", |
| "stock performance stated only with its signed return claim"), |
| "STOCK_PERFORMANCE_CLAIM", "stock performance only from the signed return claim", True, "identity"), |
| (_typed_rule(lambda t: _P.assessment_violation(t, _R.assessments, _R.assumption_claims), |
| "assumptions judged beyond their typed assessment", |
| "assumptions called conservative or aggressive only as assessed"), |
| "ASSUMPTION_ASSESSMENT", "assumption judgements follow their typed assessment", True, "identity"), |
| (_typed_rule(lambda t: _P.quote_violation(t, _R.transcript_turns, _R.company_words), |
| "call quotations attributed to the wrong speaker", "call quotations keep their speaker"), |
| "QUOTE_ATTRIBUTION", "call quotations keep their speaker's role", True, "identity"), |
| (_typed_rule(lambda t: _P.statement_domain_violation(t, _R.statements), "figures placed on the wrong statement", |
| "figures stay on their own financial statement"), |
| "STATEMENT_DOMAIN", "figures stay on their own financial statement", True, "identity"), |
| (_typed_rule(lambda t: _P.phrase_slot_violation(t, _R.catalog), "predicate or list claims out of their slot", |
| "predicate and list claims sit in their render slots"), |
| "CLAIM_RENDER_SLOT", "predicate and list claims keep their render slot", True, "presentation"), |
| (_typed_rule(lambda t: _P.family_comparison_violation(t), "comparisons across metric families", |
| "no comparison across metric families without a typed relation"), |
| "METRIC_FAMILY_RELATION", "comparisons across metric families need a typed relation", True, "identity"), |
| (_typed_rule(lambda t: _P.proximity_violation(t, _R.valuation_magnitudes), "closeness to value outside the band", |
| "closeness to value only within the band of a cited anchor"), |
| "VALUE_PROXIMITY", "'close to fair value' only within the closeness band of a cited anchor", True, "identity"), |
| (_typed_rule(lambda t: _P.named_counterpart_violation(t, _R) or _P.claim_label_violation(t, _R.catalog), |
| "a figure compared with an uncited quantity, or not named by its concept", |
| "figures are named by their concept and compared only with cited, compatible claims"), |
| "CLAIM_LABEL_BINDING", "figures are named by their concept; named comparisons cite their claim", True, "identity"), |
| (_typed_rule(lambda t: _P.magnitude_violation(t, _R.catalog) or _P.spelled_count_violation(t), |
| "verbal magnitudes or spelled counts without a claim", "verbal magnitudes fit their claims"), |
| "VERBAL_MAGNITUDE", "verbal magnitudes and counts fit a claim of their metric", True, "identity"), |
| (_typed_rule(lambda t: _P.paraphrase_attribution_violation(t, _R.transcript_turns, _R.company_words), |
| "an analyst's words attributed to management", "paraphrased call content keeps its speaker"), |
| "PARAPHRASE_ATTRIBUTION", "paraphrased call content keeps its speaker's role", True, "identity"), |
| (_structured, "STRUCTURED_ARGUMENTS", "authored judgement comes only from validated structured objects", |
| True, "identity"), |
| (_identifiers, "NO_INTERNAL_IDENTIFIERS", "no internal identifier reaches the reader", True, "presentation"), |
| (_authored_once, "AUTHORED_ONCE", "no authored text is printed twice", True, "presentation"), |
| (_segment_note, "SEGMENT_MARGIN_DISCLOSURE", "a capped segment margin is disclosed with the margin used", True, |
| "presentation"), |
| (_guidance_consensus, "GUIDANCE_CONSENSUS_COMPATIBILITY", |
| "guidance against consensus only on one fiscal period, horizon and basis", True, "identity"), |
| (_forecast_relation, "FORECAST_RELATION_REQUIRED", |
| "engine forecasts against consensus only through their typed relation", True, "identity"), |
| (_historical_quality, "HISTORICAL_QUALITY_LABEL", "trailing-figure labels derive from their quality state", |
| True, "presentation")] |
| return ValidationResult("REPORT_VALIDATED", _run(checks)) |
|
|
|
|
| |
| |
| |
| USER_SAFE_MESSAGES = { |
| |
| |
| "INITIAL": ("Some source data for this company could not be reconciled, so a complete analysis could not be " |
| "published. This usually lasts until the company's next filing or data update: running the analysis again " |
| "now will most likely give the same result."), |
| "DATA_VALIDATED": ("The forecast or valuation for this company did not pass its consistency checks, so the analysis " |
| "was not published. This comes from how the model handles this company's figures, so running it again " |
| "will usually give the same result; the case is recorded for review."), |
| |
| "FORECAST_VALIDATED": ("The analyst panel's conclusions could not be combined into a consistent result, so the analysis " |
| "was not published. Please try again later."), |
| "ANALYSIS_FINALIZED": ("The finished report did not pass its final accuracy check, so it was not published. " |
| "Please try again later."), |
| } |
| |
| WITHHELD_REASON_MESSAGES = { |
| "PRE_REVENUE": ("This company has little or no revenue yet: its operating losses are larger than its revenue. The model " |
| "values a business from an established revenue base, so pre-revenue and early-revenue companies are not " |
| "covered. Running the analysis again will give the same result."), |
| } |
| USER_SAFE_MESSAGE = USER_SAFE_MESSAGES["INITIAL"] |
|
|
|
|
| def user_payload(rep: UserReport, md: str, html_doc: str, state): |
| """The ONLY object the public API, the web page and the download use.""" |
| return {"schema": "crucible.user_report/1", "status": "PUBLISHED", "ticker": rep.ticker, "company": rep.company, |
| "as_of": rep.as_of, "version": rep.version, |
| "rating": state.value("recommendation.rating"), |
| "headline": {"value": state.value("blend.headline_return"), "rendered": (rep.claims.get("HEADLINE_RETURN") or {}).get("rendered")}, |
| "price": {"value": state.value("market.price"), "rendered": (rep.claims.get("PRICE") or {}).get("rendered")}, |
| "limitations": rep.limitations, "html": html_doc, "markdown": md, |
| "claims": {k: {"rendered": v.get("rendered"), "value": v.get("value"), "label": v.get("label")} for k, v in rep.claims.items()}} |
|
|
|
|
| def withheld_payload(ticker, company=None, stage_reached="INITIAL", reason=None): |
| """The stage reached is the last one that PASSED; the message describes the step after it. A known, specific cause |
| (``reason``) replaces the stage's general message.""" |
| out = {"schema": "crucible.user_report/1", "status": "WITHHELD", "ticker": ticker, "company": company, |
| "message": WITHHELD_REASON_MESSAGES.get(reason) or USER_SAFE_MESSAGES.get(stage_reached, USER_SAFE_MESSAGE)} |
| if reason in WITHHELD_REASON_MESSAGES: |
| out["reason"] = reason |
| return out |
|
|
|
|
| def qa_record(*, ticker, run_id, stage_reached, validations, state_snapshot, rejections, reconciliation, notes=None, |
| report_audit=None, timings=None, arguments=None): |
| """Internal QA only: never served publicly, never rendered into the user report.""" |
| return {"schema": "crucible.internal_qa/1", "ticker": ticker, "run_id": run_id, "stage_reached": stage_reached, |
| "arguments": arguments, |
| "validations": [v.to_dict() for v in validations if v is not None], |
| "report_audit": report_audit.to_dict() if report_audit is not None else None, |
| "rejected_sentences": [r.to_dict() for r in rejections], "reconciliation": reconciliation, |
| "state": state_snapshot, "notes": notes or [], "timings": timings or {}} |
|
|
