| """Stage validators over AnalysisState. |
| |
| Rules every check follows |
| * It reads the state only. |
| * Its status comes from evaluating evidence: PASS, FAIL, WARNING or NOT_APPLICABLE. A check never |
| reports PASS because an input was missing (that is NOT_APPLICABLE with a reason, or FAIL when |
| the input is required), and an exception inside a check is a FAIL, not a PASS. |
| * ACCOUNTING IDENTITIES may block. ECONOMIC SANITY heuristics warn; they never masquerade as |
| identities. |
| * Any comparison of two metrics goes through ``state.compatible``; an incompatible pair is not |
| compared. |
| |
| Stages: DATA → FORECAST → ANALYSIS → REPORT. Each returns a ``ValidationResult``; the pipeline may |
| only ``advance`` when ``passed`` is true. |
| """ |
| from __future__ import annotations |
|
|
| import datetime as _dt |
| from dataclasses import dataclass, field |
|
|
| from state import compatible |
|
|
| PASS, FAIL, WARNING, NA_ = "PASS", "FAIL", "WARNING", "NOT_APPLICABLE" |
|
|
|
|
| @dataclass |
| class CheckResult: |
| id: str |
| name: str |
| status: str |
| detail: str |
| blocking: bool = False |
| kind: str = "identity" |
| evidence: dict = field(default_factory=dict) |
|
|
| def to_dict(self): |
| return {"id": self.id, "name": self.name, "status": self.status, "detail": self.detail, |
| "blocking": self.blocking, "kind": self.kind, "evidence": self.evidence} |
|
|
|
|
| @dataclass |
| class ValidationResult: |
| stage: str |
| checks: list |
|
|
| @property |
| def blocking(self): |
| return [c for c in self.checks if c.status == FAIL and c.blocking] |
|
|
| @property |
| def warnings(self): |
| return [c for c in self.checks if c.status == WARNING or (c.status == FAIL and not c.blocking)] |
|
|
| @property |
| def passed(self): |
| return not self.blocking and bool(self.checks) |
|
|
| def to_dict(self): |
| return {"stage": self.stage, "passed": self.passed, "blocking": [c.id for c in self.blocking], |
| "warnings": [c.id for c in self.warnings], "checks": [c.to_dict() for c in self.checks]} |
|
|
|
|
| def _run(checks): |
| """Evaluate check functions; an exception is a FAIL with the error as evidence.""" |
| out = [] |
| for fn, cid, name, blocking, kind in checks: |
| try: |
| r = fn() |
| status, detail = r[0], r[1] |
| ev = r[2] if len(r) > 2 else {} |
| except Exception as e: |
| status, detail, ev = FAIL, f"check raised {type(e).__name__}: {str(e)[:120]}", {} |
| if status not in (PASS, FAIL, WARNING, NA_): |
| status, detail = FAIL, f"invalid status {status!r}" |
| out.append(CheckResult(cid, name, status, detail, blocking, kind, ev)) |
| return out |
|
|
|
|
| def _v(m): |
| return None if m is None else getattr(m, "value", m) |
|
|
|
|
| |
| |
| |
| def validate_data(s): |
| def price(): |
| p = _v(s.market.get("price")) |
| if p is None: |
| return FAIL, "no current price" |
| return (PASS, f"price {p:,.2f}") if p > 0 else (FAIL, f"non-positive price {p}") |
|
|
| def mcap(): |
| p, sh, mc = _v(s.market.get("price")), _v(s.market.get("shares_outstanding")), _v(s.market.get("market_cap")) |
| if not (p and sh and mc): |
| return NA_, "price, shares or market cap unavailable" |
| dev = p * sh / mc - 1.0 |
| ev = {"implied": p * sh, "reported": mc, "deviation": dev} |
| if abs(dev) > 0.5: |
| return FAIL, f"price × shares differs from market cap by {dev:+.0%} — a share-count or unit error", ev |
| if abs(dev) > 0.10: |
| return WARNING, f"price × shares differs from market cap by {dev:+.0%}", ev |
| return PASS, f"price × shares within {abs(dev):.1%} of market cap", ev |
|
|
| def base(): |
| rec = s.historical.get("reconciliation") or {} |
| st = rec.get("status") |
| r = _v(s.historical.get("revenue_ttm")) |
| if st == "NO_STATEMENTS" or (r is None and not s.issuer.get("is_financial")): |
| return FAIL, "no historical revenue base could be established" |
| ev = {"status": st, "overrides": len(rec.get("overrides") or []), "disagreements": rec.get("disagreements")} |
| if rec.get("disagreements"): |
| return WARNING, (f"{st}: reconciled basis selected ({rec.get('basis')}); remaining disagreement with the " |
| f"provider summary: " + "; ".join(x.get("what", "") for x in rec["disagreements"])), ev |
| if st == "RECONCILED": |
| return PASS, f"one historical basis: {rec.get('basis')} ({ev['overrides']} provider values replaced)", ev |
| if st == "SEC_ONLY": |
| return PASS, f"one historical basis from the primary source: {rec.get('basis')}", ev |
| return WARNING, f"historical basis is {rec.get('basis')} — not verifiable against filings", ev |
|
|
| def per_share_basis(): |
| sb = s.issuer.get("security_basis") |
| bad = [m.metric_id for m in (s.market.get("price"), s.historical.get("eps_adjusted_ttm"), |
| s.historical.get("eps_gaap_ttm"), s.consensus.get("eps_fy1")) |
| if m is not None and m.value is not None and m.per_share and m.security_basis != sb] |
| return (FAIL, f"per-share metrics not on the quoted {sb} basis: {bad}") if bad else (PASS, f"all per-share metrics on the {sb} basis") |
|
|
| def currency(): |
| bad = [k for k, m in s.historical.items() if hasattr(m, "currency") and m.value is not None |
| and m.unit == "USD" and (m.currency or "USD") != "USD"] |
| if s.consensus.get("normalization") and any("UNIT_UNRESOLVED" in x for x in s.consensus.normalization): |
| bad.append("consensus") |
| return (FAIL, f"not in the report currency: {bad}") if bad else (PASS, "every historical and consensus figure in USD") |
|
|
| def recency(): |
| pe = s.historical.get("statements_as_of") |
| if not pe: |
| return FAIL, "latest statement date unknown" |
| age = (_dt.date.today() - _dt.date.fromisoformat(str(pe)[:10])).days |
| ev = {"period_end": pe, "age_days": age} |
| if age > 400: |
| return FAIL, f"latest statements are {age} days old", ev |
| return (WARNING if age > 200 else PASS), f"latest quarter {pe} ({age} days old)", ev |
|
|
| def freshness(): |
| pe, prof = s.historical.get("statements_as_of"), s.historical.get("profile_as_of") |
| if not (pe and prof): |
| return NA_, "profile or statement date unavailable" |
| gap = (_dt.date.fromisoformat(str(pe)[:10]) - _dt.date.fromisoformat(str(prof)[:10])).days |
| ev = {"statements": pe, "profile": prof, "gap_days": gap} |
| if abs(gap) > 100: |
| return WARNING, (f"profile summary dated {prof} vs statements {pe}; its derived fields are replaced by " |
| f"the reconciled basis and it is used as a cross-check only"), ev |
| return PASS, f"profile ({prof}) and statements ({pe}) describe the same period", ev |
|
|
| def consensus(): |
| n = s.consensus.get("n_analysts") or 0 |
| if not n: |
| return NA_, "no sell-side estimate contributors — consensus treated as unavailable" |
| return PASS, f"{n} estimate contributors" |
|
|
| def risk_free(): |
| rf = (s.provenance.get("risk_free") or {}).get("rate") |
| if rf is None: |
| return FAIL, "risk-free rate unavailable — no discount rate can be built" |
| return (PASS, f"10-year Treasury {rf:.2%}") if 0 < rf < 0.2 else (FAIL, f"implausible risk-free rate {rf}") |
|
|
| def adr(): |
| if not s.issuer.get("is_adr"): |
| return NA_, "ordinary shares are quoted" |
| r = s.issuer.get("adr_ratio") |
| return (PASS, f"ADS ratio {r}:1 ({s.issuer.get('adr_source')})") if r else (FAIL, "ADS quoted but the ADS ratio is unknown") |
|
|
| def issuer(): |
| return (PASS, f"{s.issuer.get('issuer_class')} · {(s.issuer.get('framework') or {}).get('name')}") \ |
| if s.issuer.get("issuer_class") and (s.issuer.get("framework") or {}).get("name") else (FAIL, "issuer not classified") |
|
|
| def gaap_vs_adjusted(): |
| a, g = s.historical.get("eps_adjusted_ttm"), s.historical.get("eps_gaap_ttm") |
| if not (a and g and a.value and g.value): |
| return NA_, "one EPS basis unavailable" |
| ok, why = compatible(a, g, "level") |
| ev = {"adjusted": a.value, "gaap": g.value} |
| |
| return (WARNING if abs(a.value / g.value - 1) > 0.25 else PASS), \ |
| (f"adjusted EPS {a.value:.2f} vs reported {g.value:.2f} — different bases ({'; '.join(why) or 'same basis'}); " |
| "each use case selects one explicitly"), ev |
|
|
| def perimeter(): |
| per = s.historical.get("perimeter") or {} |
| ev = per.get("event") |
| if not ev: |
| return PASS, "history and the forward view describe the same business" |
| return WARNING, (f"{ev.get('kind')} of {ev.get('business')} (effective {ev.get('effective')}): trailing figures are " |
| f"{per.get('historical')}, consensus and the forecast are {per.get('forward')}; cross-perimeter " |
| "growth rates and multiples are not compared"), {"event": ev} |
|
|
| def consensus_scale(): |
| notes = s.consensus.get("reconciliation_notes") or [] |
| c, h = s.consensus.get("revenue_fy1"), s.historical.get("revenue_ttm") |
| if c is None or c.value is None or h is None or not h.value: |
| return (WARNING, "; ".join(notes)) if notes else (NA_, "no consensus revenue to check") |
| r = c.value / h.value |
| if not 0.2 <= r <= 5.0: |
| return FAIL, f"consensus revenue is {r:,.1f}x the reconciled trailing revenue — a unit or currency error", {"ratio": r} |
| return (WARNING, "; ".join(notes), {"ratio": r}) if notes else (PASS, f"consensus revenue {r:.2f}x trailing", {"ratio": r}) |
|
|
| |
| def guidance_units(): |
| import guidance as _g |
| g = s.guidance.get("company") or {} |
| rows = g.get("metrics") or [] |
| if not rows: |
| return NA_, f"company guidance {g.get('status')}" |
| bad = [] |
| for m in rows: |
| prop = {"concept": m.get("concept"), "unit": m.get("raw_unit") or m.get("unit"), "label": m.get("label"), |
| "low": m.get("raw_low"), "high": m.get("raw_high"), "quote": m.get("quote"), "operator": m.get("operator")} |
| ok, why = _g.verify(prop, m.get("quote") or "") |
| if not ok: |
| bad.append(f"{m.get('metric_id')}: {why}") |
| rv = _g.render(m, "value_only") |
| if rv != m.get("rendered_value"): |
| bad.append(f"{m.get('metric_id')}: rendered '{m.get('rendered_value')}' ≠ unit rendering '{rv}'") |
| return (FAIL, "; ".join(bad[:3])) if bad else (PASS, f"{len(rows)} guidance rows carry the unit written in their quote") |
|
|
| def change_not_level(): |
| import quantities as _q |
| bad = [] |
| for m in (s.guidance.get("company") or {}).get("metrics") or []: |
| qt = _q.CONCEPT_QUANTITY.get(m.get("concept")) |
| if qt and m.get("quantity_type") != qt: |
| bad.append(f"{m.get('metric_id')}: concept {m.get('concept')} but quantity {m.get('quantity_type')}") |
| if m.get("unit") in (_q.BASIS_POINTS, _q.PERCENTAGE_POINTS) and m.get("quantity_type") not in _q.CHANGE_TYPES: |
| bad.append(f"{m.get('metric_id')}: a {m.get('unit')} figure stored as {m.get('quantity_type')}") |
| if "growth" in str(m.get("label") or "").lower() and m.get("quantity_type") != _q.GROWTH_RATE: |
| bad.append(f"{m.get('metric_id')}: a growth rate stored under {m.get('concept')}") |
| for ns in ("historical", "consensus"): |
| for k in list(getattr(s, ns).keys()): |
| v = getattr(s, ns).get(k) |
| if hasattr(v, "quantity_type") and v.quantity_type in ("MARGIN_CHANGE", "CHANGE") and not str(v.economic_concept).endswith(("_CHANGE", "_GAP")): |
| bad.append(f"{ns}.{k}: change stored under the level concept {v.economic_concept}") |
| return (FAIL, "; ".join(bad[:3])) if bad else (PASS, "levels, changes and growth rates are typed apart") |
|
|
| |
| def segment_revenue(): |
| import segments as _sg |
| c = s.segments.get("canonical") or {} |
| st = c.get("status") |
| if st in (None, _sg.UNAVAILABLE, _sg.SINGLE): |
| return NA_, f"segments {st or 'unavailable'}" |
| if st == _sg.NOT_RECONCILED: |
| |
| return WARNING, f"filed segments do not reconcile: {str(c.get('reason'))[:200]}", {"coverage": c.get("coverage")} |
| bad = [(n, d) for n, ok, d in _sg.check(c) if not ok and "allocations" not in n] |
| return (FAIL, "; ".join(f"{n}: {d}" for n, d in bad), {"coverage": c.get("coverage")}) if bad else \ |
| (PASS, f"Σ {len(c.get('segments') or [])} segments = {c.get('coverage'):.1%} of consolidated revenue", |
| {"coverage": c.get("coverage")}) |
|
|
| def segment_allocations(): |
| c = s.segments.get("canonical") or {} |
| if c.get("status") != "RECONCILED": |
| return NA_, "no reconciled segments" |
| segs = c.get("segments") or [] |
| if not any(g.get("allocations") is not None for g in segs): |
| return NA_, "industries not yet allocated" |
| bad = [f"{g.get('segment_id')}: {sum(a.get('fraction') or 0 for a in g.get('allocations') or []):.6f}" |
| for g in segs if abs(sum(a.get("fraction") or 0 for a in g.get("allocations") or []) - 1.0) > 1e-6] |
| return (FAIL, "allocation fractions do not sum to 1: " + "; ".join(bad)) if bad else \ |
| (PASS, f"{len(segs)} segments, each allocated in full") |
|
|
| def guidance_canonical(): |
| import guidance as _g |
| import quantities as _q |
| g = s.guidance.get("company") or {} |
| rows = g.get("metrics") or [] |
| if not rows: |
| return NA_, "no verified guidance rows" |
| bad = [] |
| for i, m in enumerate(rows, 1): |
| op = m.get("operator") or "RANGE" |
| if op not in _g.OPERATORS: |
| bad.append(f"GUIDANCE_{i}: operator {op} unknown") |
| ru, rl = m.get("raw_unit"), m.get("raw_low") |
| if ru and rl is not None and abs(_q.to_canonical(rl, ru) - float(m.get("low"))) > 1e-9 * max(1.0, abs(float(m.get("low")))): |
| bad.append(f"GUIDANCE_{i}: canonical value ≠ raw value × unit") |
| w = _g.written_operator({"quote": m.get("quote"), "low": rl, "high": m.get("raw_high")}) |
| if (op in _g._BOUND_WORDS and w != op) or (op in ("POINT", "RANGE") and w): |
| bad.append(f"GUIDANCE_{i}: operator {op} ≠ written bound {w}") |
| return (FAIL, "; ".join(bad[:3])) if bad else (PASS, f"{len(rows)} rows: canonical values and bound operators match their quotes") |
|
|
| def guidance_epistemic(): |
| g = s.guidance.get("company") or {} |
| st = g.get("concept_states") |
| if st is None: |
| return FAIL, "no per-concept epistemic state (only a document-level status)" |
| bad = [] |
| for rj in g.get("rejected") or (): |
| c = rj.get("concept") |
| if c in st and rj.get("quote") and not any(r.get("concept") == c for r in g.get("metrics") or ()) \ |
| and st[c].get("state") != "FOUND_BUT_REJECTED": |
| bad.append(f"{c}: rejected source text recorded as {st[c].get('state')}") |
| for x in (g.get("diagnostics") or {}).get("candidate_rows") or (): |
| if (st.get(x.get("concept")) or {}).get("state") == "NOT_FOUND": |
| bad.append(f"{x.get('concept')}: the outlook section has a row for it ('{str(x.get('quote'))[:60]}') but it is NOT_FOUND") |
| for c, v in st.items(): |
| if v.get("state") == "VERIFIED_ABSENT" and not v.get("evidence"): |
| bad.append(f"{c}: VERIFIED_ABSENT without an explicit statement of absence") |
| rej = sorted(c for c, v in st.items() if v.get("state") == "FOUND_BUT_REJECTED") |
| return (FAIL, "; ".join(bad[:3])) if bad else (PASS, f"{len(st)} concepts typed; found but rejected: {', '.join(rej) or 'none'}") |
|
|
| def ownership_names(): |
| f = s.ownership.get("f13") or {} |
| if not f or f.get("guru_names_short") is None: |
| return NA_, "no tracked-holder names" |
| shown, others, n = f.get("guru_names_shown") or [], f.get("guru_names_others"), f.get("n_gurus") |
| if n is None or others is None: |
| return FAIL, "tracked-holder names without a count to reconcile against" |
| if len(shown) + int(others) != int(round(n)): |
| return FAIL, f"{len(shown)} names + {others} others ≠ {int(round(n))} tracked holders" |
| return PASS, f"{len(shown)} names + {others} others = {int(round(n))} tracked holders" |
|
|
| checks = [ |
| (price, "D1", "current price present", True, "data"), |
| (mcap, "D2", "price × shares reconciles with market cap", True, "data"), |
| (base, "D3", "one reconciled historical basis", True, "data"), |
| (per_share_basis, "D4", "per-share metrics on the quoted security basis", True, "identity"), |
| (currency, "D5", "all figures in the report currency", True, "identity"), |
| (recency, "D6", "statements are recent", True, "data"), |
| (freshness, "D7", "source dates align", False, "data"), |
| (consensus, "D8", "sell-side consensus has contributors", False, "data"), |
| (risk_free, "D9", "risk-free rate sourced", True, "data"), |
| (adr, "D10", "ADS ratio sourced for ADS-quoted issuers", True, "identity"), |
| (issuer, "D11", "issuer classified and framework selected", True, "policy"), |
| (gaap_vs_adjusted, "D12", "reported and adjusted EPS bases recorded", False, "data"), |
| (perimeter, "D13", "history and forecast on one business perimeter", False, "data"), |
| (consensus_scale, "D14", "consensus on the same scale and currency as the reconciled base", True, "identity"), |
| (guidance_units, "S13", "guidance unit in the quote matches the stored unit", True, "identity"), |
| (change_not_level, "S14", "a change or growth rate is never stored under a level concept", True, "identity"), |
| (segment_revenue, "S3", "segment revenue reconciles to consolidated revenue", True, "identity"), |
| (segment_allocations, "S4", "segment industry allocations sum to 1", True, "identity"), |
| (guidance_canonical, "GUIDANCE_CANONICAL_VALUE", "guidance values verified on canonical value, unit and bound operator", |
| True, "identity"), |
| (guidance_epistemic, "GUIDANCE_EPISTEMIC_STATE", "every guidance concept carries an epistemic state", True, "identity"), |
| (ownership_names, "OWNERSHIP_NAME_COUNT", "named tracked holders plus 'others' equal the tracked-holder count", True, |
| "identity"), |
| ] |
| return ValidationResult("DATA_VALIDATED", _run(checks)) |
|
|
|
|
| |
| |
| |
| def validate_forecast(s): |
| def identities(): |
| ids = s.forecast.get("identities") or [] |
| if _v(s.forecast.get("revenue")) is None and _v(s.forecast.get("eps")) is None: |
| |
| |
| return NA_, "no statistical 12-month forecast (insufficient history); the analysis rests on the intrinsic valuation" |
| if not ids: |
| return FAIL, "no identity was evaluated" |
| bad = [i for i in ids if not i["ok"]] |
| return (FAIL, "violated: " + "; ".join(f"{i['name']} ({i['detail']})" for i in bad), {"identities": ids}) if bad \ |
| else (PASS, f"{len(ids)} accounting identities hold", {"identities": ids}) |
|
|
| def sanity(): |
| om, nm = s.forecast.get("operating_margin"), s.forecast.get("net_margin") |
| if s.issuer.get("is_financial"): |
| return NA_, "operating margin is not meaningful for a balance-sheet financial" |
| ok, why = compatible(nm, om, "margin_relationship") |
| if not ok: |
| return NA_, "margins not comparable: " + "; ".join(why) |
| same = compatible(om, s.historical.get("operating_margin_ttm"), "margin_relationship")[0] or \ |
| not (s.historical.get("perimeter") or {}).get("event") |
| hom, hnm = (_v(s.historical.get("operating_margin_ttm")), _v(s.historical.get("net_margin_ttm"))) if same else (None, None) |
| ev = {"forward_operating": om.value, "forward_net": nm.value, "trailing_operating": hom, "trailing_net": hnm} |
| notes = [] |
| if nm.value > om.value + 0.05 and not (hnm is not None and hom is not None and hnm > hom): |
| notes.append("forecast net margin exceeds operating margin although history does not show non-operating gains") |
| if hom is not None and abs(om.value - hom) > 0.25: |
| notes.append(f"forecast operating margin departs {om.value - hom:+.0%} from the trailing margin") |
| return (WARNING, "; ".join(notes), ev) if notes else (PASS, "forward margins are economically plausible", ev) |
|
|
| def coherence(): |
| c = s.forecast.get("coherence") or {} |
| st = c.get("status") |
| if st in (None, "NOT_VERIFIABLE"): |
| return NA_, "independent models could not be compared with the reconciled forecast" |
| ev = {k: c.get(k) for k in ("operating_income_gap", "net_income_gap")} |
| return (PASS, "independent models agree with the reconciled forecast", ev) if st == "CONSISTENT" else \ |
| (WARNING, f"independent models {st.lower()} from the reconciled forecast; forecast confidence reduced", ev) |
|
|
| def growth(): |
| eps, base, g = s.forecast.get("eps"), s.select("EPS_GROWTH_BASE"), s.forecast.get("eps_growth") |
| if g is None or g.value is None: |
| return NA_, "no EPS growth rate is published (" + ("perimeter changed" if (s.historical.get("perimeter") or {}).get("event") |
| else "EPS or its base unavailable") + ")" |
| if not (eps and base and eps.value and base.value): |
| return FAIL, "a growth rate is published without a measurable base" |
| ok, why = compatible(eps, base, "growth") |
| if not ok: |
| return FAIL, "growth base incompatible: " + "; ".join(why) |
| want = eps.value / base.value - 1.0 |
| return (PASS, f"{eps.value:.2f} ÷ {base.value:.2f} − 1 = {want:+.1%}") if (g and abs(g.value - want) < 1e-6) \ |
| else (FAIL, f"stated growth {getattr(g, 'value', None)} ≠ {want:+.4f}") |
|
|
| def consensus_concept(): |
| f, c = s.forecast.get("revenue"), s.consensus.get("revenue_ntm") |
| if not (c and c.value): |
| return NA_, "no revenue consensus" |
| ok, why = compatible(f, c, "level") |
| return (PASS, "forecast and consensus revenue are the same concept and basis") if ok else \ |
| (WARNING, "forecast and consensus revenue not comparable: " + "; ".join(why)) |
|
|
| def engine(): |
| st = s.valuation.get("engine_status") |
| name = (s.valuation.get("engine") or {}).get("display_name") |
| if not name: |
| return FAIL, "no valuation engine metadata" |
| if st != "OK" and (s.valuation.get("story") or {}).get("ran") and _v(s.valuation.get("intrinsic_value")) is not None: |
| return PASS, f"{name}: story valuation produced the intrinsic value (the mechanical cross-check is {st})" |
| return (PASS, f"{name}: {st}") if st == "OK" else (FAIL, f"{name}: {st}") |
|
|
| def story_base(): |
| stry = s.valuation.get("story") or {} |
| if not stry.get("ran"): |
| return NA_, "no story valuation ran" |
| r, m = _v(s.historical.get("revenue_ttm")), _v(s.historical.get("operating_margin_ttm")) |
| sr, sm = stry.get("start_revenue"), stry.get("start_margin") |
| ev = {"story_revenue": sr, "canonical_revenue": r, "story_margin": sm, "canonical_margin": m} |
| if (s.historical.get("perimeter") or {}).get("event"): |
| |
| |
| |
| |
| sbv = s.valuation.get("story_base") or {} |
| base_rev = _v(sbv.get("revenue")) if sbv.get("perimeter") == "continuing company" else None |
| c1 = _v(s.consensus.get("revenue_fy1")) |
| ev.update(continuing_revenue=base_rev or c1, consensus_fy1_revenue=c1, story_base_perimeter=sbv.get("perimeter"), |
| revenue_note=stry.get("revenue_base_note"), margin_note=stry.get("margin_base_note")) |
| bad = [] |
| if base_rev and sr and abs(sr / base_rev - 1) > 0.02: |
| bad.append(f"story starts from revenue {sr:,.0f}, not the StoryBase {base_rev:,.0f}") |
| elif not base_rev and c1 and sr and abs(sr / c1 - 1) > 0.02 and not stry.get("revenue_base_note"): |
| bad.append(f"story starts from revenue {sr:,.0f}, not the continuing-company base {c1:,.0f}") |
| if r and sr and abs(sr / r - 1) <= 0.02: |
| bad.append("story starts from the pre-separation revenue") |
| if not stry.get("margin_base_note"): |
| bad.append("the continuing-company starting margin is undocumented") |
| if bad: |
| return FAIL, "; ".join(bad), ev |
| proxy = "no segment margin" in str(stry.get("margin_base_note") or "") |
| return ((WARNING if proxy else PASS), |
| "story starts from the continuing-company base (" + str(stry.get("revenue_base_note") or "")[:90] + ")" |
| + ("; its starting margin is the consolidated figure used as a proxy" if proxy else ""), ev) |
| bad = [] |
| if r and sr and abs(sr / r - 1) > 0.02: |
| bad.append(f"story starts from revenue {sr:,.0f} vs reconciled {r:,.0f}") |
| if m is not None and sm is not None and abs(sm - m) > 0.005: |
| bad.append(f"story starts from margin {sm:.3f} vs reconciled {m:.3f}") |
| return (FAIL, "; ".join(bad), ev) if bad else (PASS, "story valuation starts from the reconciled historical basis", ev) |
|
|
| def quant_exclusion(): |
| rel = (s.quant.get("price_model") or {}).get("reliability") or {} |
| st = rel.get("state") |
| if not st: |
| return FAIL, "price-model reliability not evaluated" |
| if st in ("UNRELIABLE", "UNAVAILABLE"): |
| bad = [k for k in ("max_blend_weight", "scenario_adjustment_cap") if (rel.get(k) or 0) != 0] + \ |
| (["persona_visibility"] if rel.get("persona_visibility") else []) |
| return (FAIL, f"{st} model still permitted: {bad}") if bad else (PASS, f"{st}: excluded from blend, scenarios and personas") |
| return PASS, f"{st}: {rel.get('allowed_claim_strength')}" |
|
|
| def policy(): |
| pol = s.issuer.get("policy") or {} |
| rows = {m.get("name") if isinstance(m, dict) else (m[1] if isinstance(m, (list, tuple)) and len(m) > 1 else None) |
| for m in ((s.technical.get("scorecard") or {}).get("metrics") or [])} |
| leaked = sorted(r for r in rows if r and r in set(pol.get("disallowed_scorecard_rows") or [])) |
| if not pol: |
| return FAIL, "no applicability policy" |
| return (FAIL, f"excluded rows present: {leaked}") if leaked else (PASS, f"{pol.get('issuer_class')} policy enforced") |
|
|
| def intrinsic(): |
| iv = s.valuation.get("intrinsic_value") |
| if iv is None or iv.value is None: |
| return WARNING, "no intrinsic value was produced" |
| ok, why = compatible(s.market.price, iv, "ratio") |
| return (PASS, f"intrinsic value {iv.value:,.2f} on the {iv.security_basis} basis") if ok else (FAIL, "; ".join(why)) |
|
|
| def ownership(): |
| st = s.ownership.get("status") |
| if st == "CURRENT": |
| return PASS, f"13F ownership as of {(s.ownership.get('f13') or {}).get('as_of')} ({s.ownership.get('age_days')} days old)" |
| if st == "STALE": |
| return WARNING, f"13F snapshot is {s.ownership.get('age_days')} days old — ownership withheld from the analysis" |
| return NA_, "no 13F ownership data for this ticker" |
|
|
| def damodaran_inputs(): |
| di = s.valuation.get("damodaran_inputs") or {} |
| if not di.get("available"): |
| return WARNING, "Damodaran datasets unavailable — the documented ERP constant and sector betas were used" |
| today = _dt.date.today() |
| age = lambda d: (today - _dt.date.fromisoformat(str(d)[:10])).days if d else 9999 |
| stale = [f"implied ERP {di.get('erp_month')}" for _ in [0] if age(di.get("erp_month")) > 62] + \ |
| [f"country risk premiums {di.get('crp_as_of')}" for _ in [0] if di.get("crp") and age(di.get("crp_as_of")) > 240] + \ |
| [f"industry data {di.get('industry_as_of')}" for _ in [0] if di.get("industry") and age(di.get("industry_as_of")) > 450] |
| if not di.get("industry"): |
| stale.append("no industry match (sector beta map used)") |
| ev = {k: di.get(k) for k in ("industry", "industry_match", "erp_month", "crp_as_of", "industry_as_of")} |
| return (WARNING, "out of date or incomplete: " + "; ".join(stale), ev) if stale else \ |
| (PASS, f"{di.get('industry')} ({di.get('industry_match')}); ERP {di.get('erp_month')}", ev) |
|
|
| def eps_basis_clamp(): |
| rec = s.forecast.get("reconciliation") or {} |
| adj = [a for a in rec.get("adjustments") or [] if "EPS" in str(a.get("kind"))] |
| bad = [a for a in adj if a.get("basis") != "ADJUSTED" or not a.get("conversion")] |
| if bad: |
| return FAIL, f"adjusted EPS limited against a capacity without basis conversion: {bad[0].get('kind')}" |
| if not rec.get("bridge_applied_once", True): |
| return FAIL, "the adjusted↔GAAP bridge was applied more than once" |
| return PASS, (f"{len(adj)} EPS limit(s), each against the adjusted-basis capacity" if adj |
| else "adjusted EPS was not limited; the bridge is applied once (GAAP EPS = adjusted ÷ bridge)") |
|
|
| def bridge_definition(): |
| br = s.historical.get("adjustment_bridge") or {} |
| e = s.forecast.get("eps") |
| if not br.get("available"): |
| return NA_, "no adjusted↔GAAP bridge: GAAP lines come from independent models" |
| if e is None or e.value is None: |
| return NA_, "no forecast EPS" |
| import definitions as _d |
| if br.get("definition_id") != e.adjustment_definition_id: |
| return FAIL, f"bridge definition {br.get('definition_id')} ≠ forecast EPS definition {e.adjustment_definition_id}" |
| if _d._provider_root(br.get("adjusted_provider")) != _d._provider_root(e.adjustment_provider): |
| return FAIL, f"bridge provider {br.get('adjusted_provider')} ≠ forecast EPS provider {e.adjustment_provider}" |
| return PASS, f"{br.get('definition_id')} ({br.get('selected_method')})" |
|
|
| def business_mix_gate(): |
| di = s.valuation.get("damodaran_inputs") or {} |
| bm = di.get("business_mix") or [] |
| c = s.segments.get("canonical") or {} |
| if not bm: |
| return PASS, f"no business-mix beta ({di.get('business_mix_status') or 'single industry'})" |
| if c.get("status") != "RECONCILED": |
| return FAIL, f"a business-mix beta was computed from segments that are {c.get('status')}" |
| ids = {g.get("segment_id") for g in c.get("segments") or []} |
| stray = [r.get("name") for r in bm if r.get("segment_id") not in ids] |
| if stray: |
| return FAIL, f"business-mix rows not backed by a canonical segment: {stray}" |
| |
| import damodaran_data as _dd |
| bad = _dd.business_mix_problems(bm, di.get("industry"), bool(s.issuer.get("is_financial"))) |
| if bad: |
| return FAIL, f"business-mix beta built on an incompatible segment mapping: {'; '.join(bad)[:300]}" |
| return PASS, f"business mix of {len(ids)} reconciled segments, every industry mapping compatible" |
|
|
| def pricing_reproducible(): |
| import pricing_state as _ps |
| ps = s.pricing.get("state") or {} |
| if not ps.get("methods"): |
| return NA_, "no pricing composite" |
| ok, det = _ps.reproduce(ps) |
| cd = ps.get("cap_diagnostics") or {} |
| if cd.get("cap_applied"): |
| det += (f"; bound applied ({cd.get('direction')}): raw {cd.get('raw_vs_price', 0):.2f}× price" |
| + (f", methods above the cap: {[m['method'] for m in cd.get('methods_above_cap') or []]}" |
| if cd.get("methods_above_cap") else "")) |
| return (PASS, det, {"cap_diagnostics": cd}) if ok else (FAIL, det, {"cap_diagnostics": cd}) |
|
|
| def no_double_adjustment(): |
| import pricing_state as _ps |
| ps = s.pricing.get("state") or {} |
| if not ps.get("methods"): |
| return NA_, "no pricing composite" |
| dd = _ps.double_adjustments(ps.get("adjustments")) |
| arch = _ps.multiple_architecture(ps) |
| if dd: |
| return FAIL, "; ".join(f"{d['type']}: {d['characteristic']} on {d['component']} ({', '.join(d['ids'])})" for d in dd[:3]) |
| if arch: |
| return FAIL, "; ".join(arch[:3]) |
| return PASS, ("no characteristic counted twice; multiples rebuild as own × regime × (1 − w) + " |
| "industry × industry-relative premium × w") |
|
|
| def analyst_target_distinct(): |
| ps = s.pricing.get("state") or {} |
| at = next((m for m in ps.get("methods") or [] if m.get("method") == "Analyst Target"), None) |
| if not at: |
| return NA_, "analyst target not a pricing method" |
| raw, adj = at.get("raw_analyst_target"), at.get("risk_adjusted_analyst_target_contribution") |
| if raw is None or adj is None: |
| return FAIL, "raw target or its risk-adjusted contribution is missing from the pricing state" |
| if at.get("label") == "Analyst Target": |
| return FAIL, "the method is labelled as the published target, not as its risk-adjusted contribution" |
| if raw > (s.value("market.price") or 0) * 1.0001 and not adj < raw: |
| return FAIL, f"a target above the price entered unadjusted ({raw} → {adj})" |
| return PASS, f"raw {raw:,.2f} → contribution {adj:,.2f}" |
|
|
| def multiple_lineage(): |
| ps = s.pricing.get("state") or {} |
| need = [m for m in ps.get("methods") or [] if m.get("method") in ("P/E", "PEG", "P/FCF", "P/S") and m.get("in_raw")] |
| if not need: |
| return NA_, "no multiple-based method" |
| bad = [m["method"] for m in need if m.get("target_multiple") is None or not m.get("multiple_basis") |
| or (m.get("method") in ("P/E", "P/S") and not ((m.get("multiple_lineage") or {}).get("sector_anchor") or {}).get("source"))] |
| return (FAIL, f"target multiple without lineage: {bad}") if bad else \ |
| (PASS, "; ".join(f"{m['method']} {m['target_multiple']:.1f}x ({m['multiple_basis']})" for m in need)) |
|
|
| def perimeter_ids(): |
| per = s.historical.get("perimeter") or {} |
| aid = per.get("action_id") |
| split = ("pre-separation", "continuing company") |
| bad, n = [], 0 |
| for ns in ("historical", "consensus", "forecast", "valuation", "pricing", "market"): |
| space = getattr(s, ns) |
| for k in list(space.keys()): |
| vals = space.get(k) |
| items = vals.items() if isinstance(vals, dict) else [(None, vals)] |
| for kk, v in items: |
| if getattr(v, "perimeter", None) in split: |
| n += 1 |
| if not aid: |
| bad.append(f"{ns}.{k}{'.' + kk if kk else ''} on {v.perimeter} with no perimeter event") |
| elif v.corporate_action_id != aid: |
| bad.append(f"{ns}.{k}{'.' + kk if kk else ''}: action {v.corporate_action_id} ≠ {aid}") |
| if not aid and not n: |
| return NA_, "no perimeter change" |
| return (FAIL, "; ".join(bad[:3])) if bad else (PASS, f"{n} figures carry {aid}") |
|
|
| def action_status(): |
| import corporate_actions as _ca |
| st = s.corporate_actions.get("state") or {} |
| if not st.get("actions"): |
| return NA_, "no corporate actions" |
| bad = _ca.check(st) |
| per = (s.historical.get("perimeter") or {}).get("action_id") |
| if per and per != st.get("perimeter_action_id"): |
| bad.append(f"perimeter action {per} ≠ canonical {st.get('perimeter_action_id')}") |
| return (FAIL, "; ".join(bad[:3])) if bad else (PASS, "; ".join(f"{a['action_id']} {a['status']}" for a in st["actions"][:4])) |
|
|
| def story_base_ok(): |
| import story_base as _sb |
| b = s.valuation.get("story_base") or {} |
| if not (s.valuation.get("story") or {}).get("ran"): |
| return NA_, "no story valuation" |
| flat = {"perimeter": b.get("perimeter"), "corporate_action_id": b.get("corporate_action_id"), "label": b.get("label"), |
| "quality_status": b.get("quality_status"), "period_end": getattr(b.get("revenue"), "period_end", None)} |
| bad = _sb.check(flat, s.valuation.get("perimeter_adjustment")) |
| per = s.historical.get("perimeter") or {} |
| if per.get("action_id") and b.get("perimeter") == "consolidated": |
| bad.append("the company separated a business but the story base is consolidated") |
| return (FAIL, "; ".join(bad[:3])) if bad else (PASS, f"{b.get('perimeter')} ({b.get('quality_status')}): {b.get('label')}") |
|
|
| def story_segments_ok(): |
| import story_base as _sb |
| stry = s.valuation.get("story") or {} |
| adj = s.valuation.get("perimeter_adjustment") or {} |
| if not stry.get("ran") or not adj.get("segment_ids"): |
| return NA_, "no separated segment" |
| bad = _sb.check({"perimeter": "consolidated"}, adj, stry.get("segments"), adj.get("segment_ids")) |
| return (FAIL, "; ".join(bad[:3])) if bad else (PASS, f"story segments exclude {', '.join(adj.get('segment_names') or [])}") |
|
|
| def story_business(code, ok_text): |
| """STORY_BUSINESS_IDENTITY / STORY_SEGMENT_PERIMETER / DISPOSED_BUSINESS_EXCLUSION / STORY_BUSINESS_SOURCE / |
| CORPORATE_ACTION_COMPLETION_LINEAGE (business.py): the story's segments against the canonical business set of |
| the valuation perimeter, and that set against the corporate-action state. They run for every story, whether or |
| not a perimeter adjustment is active (a disposal completed before the trailing window leaves none).""" |
| def run(): |
| import business as _bz |
| stry = s.valuation.get("story") or {} |
| if not stry.get("ran"): |
| return NA_, "no story valuation" |
| dec = s.valuation.get("business_decomposition") |
| found = _bz.check_segments(stry.get("segments"), dec) + \ |
| _bz.check_state(dec, s.corporate_actions.get("state"), s.corporate_actions.get("states")) |
| if code == _bz.PERIMETER and dec and (s.valuation.get("story_base") or {}).get("perimeter_id") not in (None, dec.get("perimeter_id")): |
| found.append((code, f"the business set is on perimeter {dec.get('perimeter_id')}, the story base on " |
| f"{(s.valuation.get('story_base') or {}).get('perimeter_id')}")) |
| bad = [m for c, m in found if c == code] |
| n = len(stry.get("segments") or []) |
| ev = {"mode": (dec or {}).get("mode"), "source_type": (dec or {}).get("source_type"), "story_segments": n, |
| "eligible": [b["business_id"] for b in _bz.eligible(dec)], |
| "not_eligible": [b["business_id"] for b in (dec or {}).get("businesses") or [] if not b.get("story_eligible")]} |
| return (FAIL, "; ".join(bad)[:400], ev) if bad else \ |
| (PASS, ok_text + (f" ({n} story segments)" if n else " (one-business story, no segments)"), ev) |
| return run |
|
|
| def mechanical_discount_path(): |
| """MECHANICAL_DCF_DISCOUNT_PATH: the mechanical value recomputed from its stored rate path by cumulative |
| compounding (valuation_engines.check_mechanical_dcf).""" |
| import valuation_engines as _ve |
| md = s.valuation.get("mechanical_dcf") |
| mv = _v(s.valuation.get("mechanical_value")) |
| if mv is None and s.valuation.get("same_model"): |
| mv = _v(s.valuation.get("intrinsic_value")) |
| if not md: |
| return NA_, "no two-stage mechanical DCF path (" + str((s.valuation.get("mechanical_inputs") or {}).get("basis") or "no mechanical value") + ")" |
| bad = _ve.check_mechanical_dcf(md, mv) |
| ev = {"ke_path": [round(k, 6) for k in md.get("ke_path") or []], |
| "cumulative_discount_factors": [round(d, 6) for d in md.get("cumulative_discount_factors") or []], |
| "pv_explicit": round(sum(md.get("pv_flows") or []), 4), "pv_terminal_value": md.get("pv_terminal_value"), |
| "total_value": md.get("total_value")} |
| return (FAIL, "; ".join(bad)[:400], ev) if bad else \ |
| (PASS, f"mechanical value {md.get('total_value'):.2f} reproduces from its rate path by cumulative discounting", ev) |
|
|
| def rate_path_ok(): |
| dp = s.valuation.get("discount_rate_path") or {} |
| if not dp.get("yearly_rates"): |
| return (NA_, "no story valuation") if not (s.valuation.get("story") or {}).get("ran") else (FAIL, "the story ran without a persisted discount-rate path") |
| y = dp["yearly_rates"] |
| bad = [] |
| if abs(y[0] - dp["initial_wacc"]) > 1e-9: |
| bad.append("year 1 ≠ initial rate") |
| if abs(y[-1] - dp["terminal_wacc"]) > 1e-9: |
| bad.append("final year ≠ terminal rate") |
| if len(y) != int(dp.get("hold_years") or 0) + int(dp.get("transition_years") or 0): |
| bad.append("path length ≠ hold + transition years") |
| a = (s.valuation.get("story") or {}).get("assumptions") or {} |
| if a.get("wacc_initial") is not None and abs(round(y[0], 4) - a["wacc_initial"]) > 1e-4: |
| bad.append("story assumptions' initial rate ≠ the path") |
| return (FAIL, "; ".join(bad)) if bad else (PASS, f"{len(y)} yearly rates, {y[0]:.4f} → {y[-1]:.4f}") |
|
|
| def terminal_rate_ok(): |
| dp = s.valuation.get("discount_rate_path") or {} |
| if dp.get("terminal_wacc") is None: |
| return NA_, "no discount-rate path" |
| a = (s.valuation.get("story") or {}).get("assumptions") or {} |
| claim = _v(dp.get("terminal_wacc_metric")) |
| bad = [] |
| if claim is None or abs(claim - dp["terminal_wacc"]) > 1e-9: |
| bad.append(f"terminal claim {claim} ≠ terminal rate used {dp['terminal_wacc']}") |
| if a.get("wacc_terminal") is not None and abs(a["wacc_terminal"] - round(dp["terminal_wacc"], 4)) > 1e-4: |
| bad.append(f"story assumptions' terminal {a['wacc_terminal']} ≠ the path's {dp['terminal_wacc']}") |
| if a.get("stable_rate_kept"): |
| bad.append("the retired permanent-low-rate rule was applied") |
| return (FAIL, "; ".join(bad)) if bad else (PASS, f"terminal rate used {dp['terminal_wacc']:.4f} (benchmark " |
| f"{(dp.get('mature_benchmark_wacc') or 0):.4f}); spread {dp.get('terminal_spread') or 0:.4f}") |
|
|
| def sensitivity_path_ok(): |
| dp = s.valuation.get("discount_rate_path") or {} |
| cases = [c for c in s.valuation.get("sensitivity") or [] if c.get("driver") == "Risk-free rate"] |
| if not cases or not dp: |
| return NA_, "no risk-free sensitivity" |
| bad = [] |
| for c in cases: |
| p = c.get("discount_rate_path") or {} |
| if not p.get("yearly_rates"): |
| bad.append(f"{c.get('case')}: no rebuilt path") |
| continue |
| if p.get("rf") is None or abs(p["rf"] - (dp.get("rf") + c.get("shift"))) > 1e-6 and not (c.get("shift") < 0 and p.get("rf") == 0.01): |
| bad.append(f"{c.get('case')}: rf {p.get('rf')} ≠ base {dp.get('rf')} + {c.get('shift')}") |
| if dp.get("mature_benchmark_wacc") is not None and p.get("mature_benchmark_wacc") is not None \ |
| and abs(p["mature_benchmark_wacc"] - dp["mature_benchmark_wacc"]) < 1e-6: |
| bad.append(f"{c.get('case')}: the benchmark was not rebuilt at the shifted rf") |
| return (FAIL, "; ".join(bad[:3])) if bad else (PASS, f"{len(cases)} rate cases rebuild the benchmark and the path") |
|
|
| def peer_stats_ok(): |
| pc = s.pricing.get("peers") or {} |
| rel = s.pricing.get("relative") or {} |
| bad = [] |
| if not pc.get("subject") and pc.get("percentiles"): |
| bad.append(f"percentiles {sorted(pc['percentiles'])} survive a withheld subject ({pc.get('subject_withheld')})") |
| for k, v in rel.items(): |
| if v.get("comparison_allowed") and (v.get("subject_value") is None or v.get("perimeter_basis") == "pre-separation"): |
| bad.append(f"{k}: comparison allowed on a withheld or pre-separation subject") |
| return (FAIL, "; ".join(bad[:3])) if bad else (PASS, f"{sum(1 for v in rel.values() if v.get('comparison_allowed'))} " |
| "relative claims allowed") |
|
|
| def rel_check(kind): |
| import relations as _rel |
| rs = s.relations.get("state") or {} |
| if not (rs.get("items") or {}): |
| return NA_, "no relations (a required endpoint is missing)" |
| bad = [p for k, p in _rel.check(s) if k == kind] |
| n = len(rs.get("items") or {}) |
| rej = rs.get("rejected") or [] |
| if kind == "REL3" and not bad: |
| return PASS, f"{n} relations, all between compatible quantities" + (f"; {len(rej)} incompatible pair(s) never created" if rej else "") |
| return (FAIL, "; ".join(bad[:3])) if bad else (PASS, f"{n} relations consistent") |
|
|
| def expectations_basis(): |
| rs = s.relations.get("state") or {} |
| b = rs.get("basis") or {} |
| ex = (rs.get("items") or {}).get("REL_EXPECTATIONS_STORY") |
| if not ex: |
| return NA_, "no implied-expectations relation" |
| piv = s.value("valuation.price_vs_intrinsic") |
| notes = [] |
| if b.get("breakeven_story_margin") is not None and b.get("adopted_story_margin") is not None \ |
| and abs(float(b["breakeven_story_margin"]) - float(b["adopted_story_margin"])) > 0.0005: |
| notes.append(f"the break-even revenue is solved at a {float(b['breakeven_story_margin']) * 100:.1f}% margin, the adopted " |
| f"story uses {float(b['adopted_story_margin']) * 100:.1f}%") |
| _be = (s.get("valuation.story") or {}).get("breakeven") or {} |
| _iv = s.value("valuation.intrinsic_value") |
| if _be.get("story_value") and _iv and abs(float(_be["story_value"]) / float(_iv) - 1.0) > 0.02: |
| notes.append(f"the break-even base value (${float(_be['story_value']):,.2f}) is not the story value (${float(_iv):,.2f})") |
| import relations as _rl |
| if piv is not None and ((piv > 0 and ex["relation_type"] == _rl.LESS_DEMANDING) |
| or (piv < 0 and ex["relation_type"] == _rl.MORE_DEMANDING)): |
| notes.append(f"price vs intrinsic value ({piv:+.1%}) and the expectations polarity ({ex['relation_type']}) disagree") |
| return (WARNING, "; ".join(notes), b) if notes else (PASS, f"{ex['relation_type']} on one basis", b) |
|
|
| def evidence_visibility(): |
| import catalog as _c |
| import evidence as _e |
| es = s.evidence.get("state") or {} |
| bad = [cid for cid, e in (es.get("claims") or {}).items() if e.get("persona_visible") != _e.model_visible(s, e["source_model_id"])] |
| cat = _c.build_catalog(s, include_deferred=False) |
| leak = [cid for cid in _c.offered(s, cat) if _e.claim_model(cid) and not _e.model_visible(s, _e.claim_model(cid))] |
| if bad or leak: |
| return FAIL, f"visibility not per model: {bad + leak}" |
| hidden = [m for m, v in (es.get("models") or {}).items() if not v.get("persona_visible")] |
| return PASS, f"per-model visibility; hidden models: {', '.join(hidden) or 'none'}" |
|
|
| def segment_build(): |
| st = s.valuation.get("story") or {} |
| segs, a = st.get("segments") or [], st.get("assumptions") or {} |
| if not segs: |
| return NA_, "no segment build" |
| rev = sum(float(g.get("rev_usd_b") or 0) for g in segs) |
| bad = [] |
| if a.get("revenue_N_usd_b") and abs(rev - float(a["revenue_N_usd_b"])) > 0.01 * float(a["revenue_N_usd_b"]): |
| bad.append(f"segment revenue {rev:.2f}B ≠ story revenue at maturity {float(a['revenue_N_usd_b']):.2f}B") |
| mrows = [g for g in segs if g.get("margin") is not None] |
| if mrows and a.get("margin_target") is not None: |
| wm = sum(float(g["rev_usd_b"]) * float(g["margin"]) for g in mrows) / sum(float(g["rev_usd_b"]) for g in mrows) |
| prop, tgt, cap, flo = a.get("margin_proposed"), float(a["margin_target"]), a.get("margin_cap"), a.get("margin_floor") |
| if prop is None or abs(wm - float(prop)) > 0.0015: |
| bad.append(f"segments weight to {wm:.3f}, recorded proposal {prop}") |
| elif abs(float(prop) - tgt) > 0.0005 and not (cap is not None and abs(tgt - float(cap)) <= 0.0005) \ |
| and not (flo is not None and abs(tgt - float(flo)) <= 0.0005 and float(prop) < tgt): |
| bad.append(f"the valuation margin {tgt:.3f} is neither the segments' {float(prop):.3f}, the recorded cap {cap} " |
| f"nor a recorded floor {flo}") |
| return (FAIL, "; ".join(bad)) if bad else (PASS, "segment revenue and margins reconcile to the story (a cap or floor, " |
| "when applied, is recorded and disclosed)") |
|
|
| checks = [ |
| (identities, "F1", "forecast accounting identities hold", True, "identity"), |
| (sanity, "F2", "forward margins economically plausible", False, "sanity"), |
| (coherence, "F3", "independent forecast models agree", False, "sanity"), |
| (growth, "F4", "EPS growth measured on a compatible base", True, "identity"), |
| (consensus_concept, "F5", "forecast and consensus comparable", False, "data"), |
| (engine, "F6", "selected valuation engine produced its value", True, "policy"), |
| (story_base, "F7", "story uses the reconciled historical basis", True, "identity"), |
| (quant_exclusion, "F8", "unreliable quant models excluded", True, "policy"), |
| (policy, "F9", "issuer applicability policy enforced", True, "policy"), |
| (intrinsic, "F10", "intrinsic value on the quoted basis", False, "identity"), |
| (ownership, "F11", "13F ownership current", False, "data"), |
| (damodaran_inputs, "F12", "Damodaran market and industry inputs current", False, "data"), |
| (eps_basis_clamp, "S1", "adjusted EPS never limited against a GAAP capacity without conversion", True, "identity"), |
| (bridge_definition, "S2", "adjusted↔GAAP bridge on the forecast EPS definition", True, "identity"), |
| (business_mix_gate, "S5", "business-mix beta only from reconciled segments", True, "identity"), |
| (pricing_reproducible, "S6", "market composite reproducible from state", True, "identity"), |
| (analyst_target_distinct, "S7", "raw analyst target and its risk-adjusted contribution distinct", True, "presentation"), |
| (multiple_lineage, "S8", "every pricing multiple has lineage", True, "identity"), |
| (no_double_adjustment, "S6B", "no characteristic counted twice (premiums and penalties)", True, "identity"), |
| (perimeter_ids, "S17", "every perimeter-dependent figure carries the perimeter's corporate action", True, "identity"), |
| (action_status, "S18", "corporate-action status verified by a filing, or UNVERIFIED and hidden", True, "policy"), |
| (story_base_ok, "S19", "story base has an explicit perimeter and an aligned period", True, "identity"), |
| (story_segments_ok, "S20", "story segment set matches the story perimeter", True, "identity"), |
| (story_business("STORY_BUSINESS_IDENTITY", "every story business is an eligible business of record"), |
| "STORY_BUSINESS_IDENTITY", "a storyteller values only businesses of the canonical state", True, "identity"), |
| (story_business("STORY_SEGMENT_PERIMETER", "every story business is on the story base's perimeter"), |
| "STORY_SEGMENT_PERIMETER", "story businesses share the story base's perimeter", True, "identity"), |
| (story_business("DISPOSED_BUSINESS_EXCLUSION", "no completed disposal is in the story"), |
| "DISPOSED_BUSINESS_EXCLUSION", "a business with a completed disposal never enters the story", True, "identity"), |
| (story_business("STORY_BUSINESS_SOURCE", "business identity comes from a primary company source"), |
| "STORY_BUSINESS_SOURCE", "no business relies on model recall", True, "policy"), |
| (story_business("CORPORATE_ACTION_COMPLETION_LINEAGE", "completion evidence found in filings is in the canonical state"), |
| "CORPORATE_ACTION_COMPLETION_LINEAGE", "filing-verified completions reach the canonical action state", True, "identity"), |
| (mechanical_discount_path, "MECHANICAL_DCF_DISCOUNT_PATH", |
| "the mechanical DCF discounts along its year-by-year rate path", True, "identity"), |
| (rate_path_ok, "S21", "discount-rate path rendered is the path used", True, "identity"), |
| (terminal_rate_ok, "S22", "terminal cost-of-capital claim is the terminal rate used", True, "identity"), |
| (sensitivity_path_ok, "S23", "sensitivity cases rebuild the whole discount-rate path", True, "identity"), |
| (peer_stats_ok, "S25", "derived peer statistics inherit their source's eligibility", True, "policy"), |
| (lambda: rel_check("REL1"), "RELATION_IDENTITY", "every relation's lhs, rhs, operator and rendering match the state", True, "identity"), |
| (lambda: rel_check("REL2"), "RELATION_MAGNITUDE", "every relation magnitude follows its declared definition and pair", True, "identity"), |
| (lambda: rel_check("REL3"), "RELATION_COMPATIBILITY", "relations exist only between compatible quantities", True, "identity"), |
| (expectations_basis, "EXPECTATIONS_BASIS", "the break-even solve and the adopted story share one basis", False, "sanity"), |
| (evidence_visibility, "MODEL_EVIDENCE_VISIBILITY", "model-derived claims follow their own model's reliability", True, "policy"), |
| (segment_build, "SEGMENT_BUILD_RECONCILES", "segment build reconciles to the story's revenue and margin", True, |
| "identity"), |
| ] |
| return ValidationResult("FORECAST_VALIDATED", _run(checks)) |
|
|
|
|
| |
| |
| |
| def validate_analysis(s): |
| def panel(): |
| if not s.scenarios.get("panel_complete"): |
| return FAIL, "the analyst panel is incomplete: " + ", ".join(s.scenarios.get("missing_roles") or ["a scenario return or probability"]) |
| return PASS, "all four seats produced a result" |
|
|
| def probabilities(): |
| ps = [(s.scenarios.get(k) or {}).get("probability") for k in ("bear", "neutral", "bull")] |
| if any(p is None for p in ps): |
| return FAIL, "a scenario probability is missing" |
| if any(p < 0 or p > 1 for p in ps) or abs(sum(ps) - 1.0) > 0.005: |
| return FAIL, f"probabilities {ps} do not form a distribution" |
| return PASS, f"probabilities sum to {sum(ps):.3f}" |
|
|
| def scenario_arithmetic(): |
| rows = [s.scenarios.get(k) or {} for k in ("bear", "neutral", "bull")] |
| pm = _v(s.scenarios.get("pm_return")) |
| if pm is None: |
| return FAIL, "PM return not computed" |
| ev = sum(r["return"] * r["probability"] for r in rows) |
| return (PASS, f"Σ p × r = {ev:+.4f}") if abs(ev - pm) < 0.001 else (FAIL, f"Σ p × r = {ev:+.4f} ≠ PM {pm:+.4f}") |
|
|
| def headline(): |
| rows = s.blend.get("sources") or [] |
| h, raw = _v(s.blend.get("headline_return")), s.blend.get("headline_raw") |
| if h is None or not rows: |
| return FAIL, "headline or its sources missing" |
| total = sum((r.get("contribution") or 0.0) for r in rows) |
| factor = 1.0 |
| for a in s.blend.get("adjustments") or (): |
| factor *= (a.get("factor") if a.get("factor") is not None else 1.0) |
| want = max(min(total * factor, 3.0), -0.85) |
| if abs(want - raw) > 0.0005: |
| return FAIL, f"Σ contributions {total:+.4f} × adjustments {factor:.3f} = {want:+.4f} ≠ headline {raw:+.4f}" |
| return (PASS, f"headline {h:+.3f} = Σ contributions") if abs(round(raw, 3) - h) < 1e-9 else (FAIL, "rounding mismatch") |
|
|
| def rating(): |
| from state_builder import rating_for |
| h, r = _v(s.blend.get("headline_return")), s.recommendation.get("rating") |
| want = rating_for(h) |
| return (PASS, f"{r} = band({h:+.3f})") if (r and r == want) else (FAIL, f"rating {r} ≠ band {want}") |
|
|
| def quant_zero(): |
| rel = (s.quant.get("price_model") or {}).get("reliability") or {} |
| if rel.get("state") not in ("UNRELIABLE", "UNAVAILABLE"): |
| return NA_, f"price model {rel.get('state')}" |
| share = sum((r.get("share") or 0) for r in (s.blend.get("sources") or []) if "Price Model" in str(r.get("name"))) |
| tilt = ((s.scenarios.get("probability_audit") or {}).get("quant_adjustment") or {}) |
| applied = any(abs(v or 0) > 0 for v in tilt.values()) |
| adj = [a.get("name") for a in (s.blend.get("adjustments") or ()) if "price model" in str(a.get("note") or "").lower()] |
| if share > 0 or applied or adj: |
| return FAIL, (f"unreliable model influenced the result (blend share {share:.4f}, scenario tilt {applied}" |
| + (f", blend adjustment {adj}" if adj else "") + ")") |
| return PASS, "unreliable model has zero blend share and zero scenario influence" |
|
|
| def evidence(): |
| ev = s.scenarios.get("evidence") or {} |
| if not ev: |
| return NA_, "the PM cited no evidence list" |
| return (PASS, "every cited evidence item was eligible") if ev.get("ok") else (FAIL, "; ".join(ev.get("problems") or [])) |
|
|
| def triggers(): |
| trg = s.scenarios.get("triggers") or [] |
| bad = [t for t in trg if t.get("status") == "POLARITY_REVERSED"] |
| if not trg: |
| return NA_, "no structured triggers" |
| return (FAIL, f"{len(bad)} trigger(s) with reversed polarity") if bad else (PASS, f"{len(trg)} trigger(s) polarity-checked") |
|
|
| def trigger_units(): |
| import re as _re |
| import quantities as _q |
| trs = s.scenarios.get("triggers") or [] |
| if not trs: |
| return NA_, "no triggers" |
| rows = (s.guidance.get("company") or {}).get("metrics") or [] |
| bad = [] |
| for t in trs: |
| src = t.get("source_claim_id") |
| if src and src.startswith("GUIDANCE_"): |
| try: |
| g = rows[int(src.split("_")[1]) - 1] |
| except (ValueError, IndexError): |
| bad.append(f"{t.get('metric_id')}: unknown source {src}") |
| continue |
| if t.get("threshold_unit") != g.get("unit"): |
| bad.append(f"{t.get('metric_id')}: unit {t.get('threshold_unit')} ≠ its source's {g.get('unit')}") |
| ok_m, why_m = _q.magnitude_ok(t.get("threshold"), g.get("low"), g.get("high")) |
| if not ok_m: |
| bad.append(f"{t.get('metric_id')}: {why_m}") |
| |
| |
| |
| if src or t.get("quantity_type") in ("LEVEL", "MARGIN_LEVEL"): |
| continue |
| tw = {w for w in _re.findall(r"[a-z]+", str(t.get("metric_id") or "").lower()) if len(w) > 3} |
| for g in rows: |
| gw = {w for w in _re.findall(r"[a-z]+", str(g.get("label") or "").lower()) if len(w) > 3} |
| if len(tw & gw) >= 2 and g.get("unit") in ("BASIS_POINTS", "PERCENTAGE_POINTS") \ |
| and t.get("threshold_unit") not in ("BASIS_POINTS", "PERCENTAGE_POINTS"): |
| bad.append(f"{t.get('metric_id')}: a guided {g.get('unit').lower().replace('_', ' ')} change restated as " |
| f"{t.get('threshold_unit') or t.get('threshold_kind')}") |
| break |
| return (FAIL, "; ".join(bad[:3])) if bad else (PASS, f"{len(trs)} triggers keep their source units") |
|
|
| def trigger_not_satisfied(): |
| from state_builder import trigger_current_values, trigger_holds |
| trs = s.scenarios.get("triggers") or [] |
| if not trs: |
| return NA_, "no triggers" |
| bad = [] |
| for t in trs: |
| cur = trigger_current_values(s, t.get("source_claim_id")) |
| hit = [f"{c} = {v:g}" for c, v in cur if trigger_holds(t, v)] |
| if t.get("range_position") == "SATISFIED_ACROSS_RANGE": |
| hit.append(f"the whole guided range {t['current_range']['low']:g}–{t['current_range']['high']:g}") |
| if hit or t.get("currently_satisfied"): |
| bad.append(f"{t.get('metric_id')} {t.get('side')} {t.get('threshold')}: already holds ({', '.join(hit)})") |
| return (FAIL, "; ".join(bad[:3])) if bad else (PASS, f"{len(trs)} triggers; none already holds") |
|
|
| def blend_kind(): |
| from state_builder import BLEND_KIND_LABEL, BLEND_KINDS |
| src = s.blend.get("sources") or [] |
| if not src: |
| return NA_, "no blend sources" |
| bad = [] |
| price, comp = s.value("market.price"), s.value("pricing.market_composite") |
| for x in src: |
| k = x.get("kind") |
| if k not in BLEND_KINDS: |
| bad.append(f"{x.get('name')}: untyped") |
| continue |
| want = BLEND_KIND_LABEL[k] if k != "OTHER" else f"Other source ({x.get('name')})" |
| if x.get("label") != want: |
| bad.append(f"{x.get('name')}: label '{x.get('label')}' ≠ '{want}' for kind {k}") |
| if k == "MARKET_PRICING" and price and comp and x.get("return") is not None \ |
| and abs(float(x.get("raw_return") if x.get("raw_return") is not None else x["return"]) - (comp / price - 1.0)) > 0.0015: |
| bad.append(f"{x.get('name')}: a MARKET_PRICING view must be composite / price − 1 ({comp / price - 1:+.4f}), " |
| f"not {x['return']:+.4f}") |
| return (FAIL, "; ".join(bad[:3])) if bad else (PASS, "; ".join(f"{x['label']} ({x['kind']})" for x in src)) |
|
|
| checks = [ |
| (panel, "X1", "analyst panel complete", True, "data"), |
| (probabilities, "X2", "scenario probabilities form a distribution", True, "identity"), |
| (scenario_arithmetic, "X3", "PM return = Σ probability × scenario return", True, "identity"), |
| (headline, "X4", "headline reproduces from its sources", True, "identity"), |
| (rating, "X5", "rating = fixed band of the displayed headline", True, "identity"), |
| (quant_zero, "X6", "unreliable quant has zero influence", True, "policy"), |
| (evidence, "X7", "PM evidence eligible", True, "policy"), |
| (triggers, "X8", "thesis triggers polarity-consistent", True, "identity"), |
| (trigger_units, "S15", "thesis triggers inherit their source claim's unit", True, "identity"), |
| (blend_kind, "BLEND_SOURCE_KIND", "blend sources are typed; labels and views follow their kind", True, "identity"), |
| (trigger_not_satisfied, "TRIGGER_NOT_ALREADY_SATISFIED", "no published trigger already holds in the current state", |
| True, "identity"), |
| ] |
| return ValidationResult("ANALYSIS_FINALIZED", _run(checks)) |
|
|
|
|
| |
| USER_LIMITATION = { |
| "EXPECTATIONS_BASIS": ("What today's price requires is solved on the story's assumptions with a capped maturity margin, " |
| "which differs from the adopted story's margin; the break-even revenue is stated at that basis."), |
| "D3": "Provider figures disagreed with other sources; this analysis uses the reconciled figures described in the data notes.", |
| "D13": ("A business was separated from the company within the trailing twelve months: trailing figures include it, while the " |
| "forecast and sell-side estimates describe the continuing company, so growth rates and trailing multiples are not shown."), |
| "D14": "Some sell-side estimates arrived in a different currency or scale; they were converted where that was unambiguous and omitted otherwise.", |
| "D7": "The data provider's company summary is dated differently from the latest statements and is used only as a cross-check.", |
| "D8": "No sell-side estimates were available, so consensus comparisons are omitted.", |
| "D12": "Reported and adjusted earnings differ materially; each figure is labelled with its basis.", |
| "F2": "The earnings forecast implies a margin path well away from recent history.", |
| "F3": "The earnings, operating-income and net-income forecasts disagree materially; forecast confidence is reduced.", |
| "F5": "The revenue forecast and the sell-side revenue estimate are not on the same basis, so they are not compared.", |
| "F10": "No intrinsic value could be produced for this company.", |
| "F1": ("There is not enough history for a statistical 12-month forecast, so the analysis rests on the intrinsic " |
| "valuation, market pricing and the analyst panel."), |
| "F11": "The latest institutional-ownership (13F) snapshot is out of date, so ownership is not discussed.", |
| "F12": "Some of Damodaran's market or industry inputs used here are not the latest edition.", |
| "S3": ("The company's filed segment figures do not add up to its consolidated revenue, so segment economics and a " |
| "business-mix cost of capital are not used; the company is valued on its single industry."), |
| } |
|
|
|
|
| def user_limitations(*results): |
| out = [] |
| for r in results: |
| for c in r.checks: |
| |
| |
| if c.id == "D3" and not (c.evidence or {}).get("disagreements"): |
| continue |
| if c.id == "F1" and c.status == NA_ and "insufficient history" in c.detail and USER_LIMITATION["F1"] not in out: |
| out.append(USER_LIMITATION["F1"]) |
| continue |
| if c.status == WARNING and c.id in USER_LIMITATION and USER_LIMITATION[c.id] not in out: |
| out.append(USER_LIMITATION[c.id]) |
| return out |
|
|
