Blame for engine/validation.py

"""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           # a FAIL on a blocking check stops the stage
    kind: str = "identity"           # identity | sanity | data | policy | presentation
    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:  # noqa: BLE001
            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)


# ═══════════════════════════════════════════════════════════════════════════════════════
# DATA_VALIDATED
# ═══════════════════════════════════════════════════════════════════════════════════════
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}
        # different bases by design: this records the gap, it does not compare them as equals
        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})

    # ── guidance semantics (S13, S14) ──
    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")

    # ── segments (S3, S4): the canonical segment state is internally consistent ──
    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:
            # not an error: the state says so, and every consumer is disabled (S5)
            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))


# ═══════════════════════════════════════════════════════════════════════════════════════
# FORECAST_VALIDATED
# ═══════════════════════════════════════════════════════════════════════════════════════
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:
            # no statistical forecast at all (too little history, e.g. a recent IPO): nothing to test.
            # A PARTIAL forecast (one line present, another missing) is still a failure below.
            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"):
            # after a separation the story values the CONTINUING company: its start must be that base,
            # documented, and never the pre-separation trailing figures
            # the continuing-company base is the typed StoryBase (same period as the trailing statements);
            # the current-year consensus is a forward cross-check, not the base
            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}"
        # every industry the mix used (for its EV/Sales weight and its beta) must have passed parent-sector compatibility
        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"):       # no story ran: the intrinsic value IS the mechanical value
            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))


# ═══════════════════════════════════════════════════════════════════════════════════════
# ANALYSIS_FINALIZED
# ═══════════════════════════════════════════════════════════════════════════════════════
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}")
            # a trigger without a declared source must not restate a guided change in another unit (35 bps → 35%). A trigger
            # WITH a typed source (e.g. the story's maturity margin, a LEVEL) is checked against that source, never matched
            # to a guidance row by its words ("operating margin" is not "operating profit margin expansion")
            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))


# plain-language user limitations for warnings (never validator ids)
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:
            # D3 warns for two reasons; only an actual disagreement is described as one (an FPI whose
            # statements cannot be verified carries its own reconciliation note)
            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