Blame for engine/state_builder.py

"""Adapters: legacy outputs → AnalysisState.

These functions are the ONLY code that reads the legacy dicts (``all_data``, the forecast dict,
the fair-value dict, the story result, the debate dict, the blend dict …). Each writes the
namespaces of one pipeline stage. Everything downstream — personas, validators, the summary
prompt, the renderer, the claim audit and the recommendation — reads the state only.
"""
from __future__ import annotations

import datetime as _dt

import bridge as _br
import definitions as _defs
import forecast_state as _fs
from state import (ADJUSTED, ADS, CONSOLIDATED, CONTINUING, DERIVED, FY, FY1, FY2, GAAP, IFRS, INTRINSIC, MARKET, MODEL,
                   MODEL_ESTIMATE, NTM, ORDINARY_SHARE, PARTIALLY_RECONCILED, POINT, PRE_SEPARATION, RECONCILED, SPOT, TTM, UNAVAILABLE,
                   UNVERIFIED, VALIDATED, metric)


def _f(v):
    try:
        if v in (None, "", "None", "-"):
            return None
        x = float(v)
        return None if x != x else x
    except (TypeError, ValueError):
        return None


def _mv(metrics, k):
    return _f(((metrics or {}).get(k) or {}).get("value"))


def _today():
    return _dt.date.today().isoformat()


# ═══════════════════════════════════════════════════════════════════════════════════════
# DATA stage: identity, issuer, market, historical, consensus, provenance
# ═══════════════════════════════════════════════════════════════════════════════════════
def build_data_stage(state, all_data, price, metrics_pre=None, price_as_of=None, code_version=None, tech=None):
    d = all_data or {}
    ov = d.get("overview") or {}
    iss = d.get("issuer") or {}
    fw = d.get("valuation_framework") or {}
    pol = d.get("policy")
    rh = d.get("reconciled_historicals") or {}
    is_adr = bool(iss.get("is_adr"))
    adr_ratio = _f(iss.get("adr_ratio"))
    sb = ADS if is_adr else ORDINARY_SHARE
    basis = IFRS if "IFRS" in str(iss.get("accounting_standard") or "") else GAAP
    period_end = rh.get("period_end")
    # perimeter: a completed separation inside the trailing window (sell-side consensus already models
    # the smaller continuing company, confirmed by an 8-K) puts history and the forward view on
    # different businesses. Every historical metric carries that perimeter, so compatible() refuses
    # cross-perimeter growth rates, multiples and comparisons.
    rst = d.get("restatement") or {}
    separated = bool(rst) and str(rst.get("kind") or "") in ("spin_off", "divestiture", "sale", "split_off", "separation")
    hist_perimeter = PRE_SEPARATION if separated else CONSOLIDATED
    fwd_perimeter = CONTINUING if separated else CONSOLIDATED
    import corporate_actions as _ca
    perimeter_aid = _ca.action_id(state.identity.ticker, rst.get("kind"), rst.get("business")) if separated else None
    s = state
    s.identity.company_name = ov.get("Name") or s.identity.ticker
    s.identity.cik = iss.get("cik") or (d.get("submissions") or {}).get("cik")
    s.identity.as_of = _today()
    s.identity.code_version = code_version
    s.identity.sector = ov.get("Sector")
    s.identity.industry = ov.get("Industry")

    # issuer
    s.issuer.issuer_class = iss.get("issuer_class")
    s.issuer.filer_type = iss.get("filer_type")
    s.issuer.regime = "FOREIGN_PRIVATE_ISSUER" if iss.get("is_fpi") else "US_DOMESTIC"
    s.issuer.accounting_standard = iss.get("accounting_standard")
    s.issuer.reporting_currency = iss.get("reporting_currency")
    s.issuer.quote_currency = iss.get("quote_currency") or "USD"
    s.issuer.is_financial = bool(iss.get("is_financial"))
    s.issuer.is_fpi = bool(iss.get("is_fpi"))
    s.issuer.is_adr = is_adr
    s.issuer.adr_ratio = adr_ratio
    s.issuer.adr_source = iss.get("adr_source")
    s.issuer.security_basis = sb
    s.issuer.framework = {"name": fw.get("framework"), "reason": fw.get("valuation_framework_reason"),
                          "ev_meaningful": fw.get("ev_meaningful")}
    s.issuer.policy = ({"issuer_class": getattr(pol, "issuer_class", None),
                        "valuation_engine": getattr(pol, "valuation_engine", None),
                        "pricing_methods": list(getattr(pol, "allowed_pricing_methods", ()) or ()),
                        "disallowed_scorecard_rows": list(getattr(pol, "disallowed_scorecard_rows", ()) or ()),
                        "excluded_topics": list(getattr(pol, "persona_excluded_topics", ()) or ()),
                        "reason": getattr(pol, "reason", None)} if pol is not None else {})
    s.issuer.revenue_concept = (d.get("canonical_concepts") or {}).get("revenue")

    # segments: the canonical (reconciled) segment state is the only segment source for every consumer
    _canon = d.get("segments_canonical") or {"status": "UNAVAILABLE", "reason": "no segment data"}
    s.segments.canonical = _canon
    s.segments.status = _canon.get("status")
    # company guidance: primary source only (guidance.company_guidance); each verified row carries its rendering
    import catalog as _cat0
    _g = dict(d.get("company_guidance") or {"status": "UNAVAILABLE", "reason": "guidance not retrieved", "metrics": []})
    _g["metrics"] = [dict(m, rendered_value=m.get("rendered_value") or _cat0.render_guidance(m, "value_only"),
                          rendered_full=m.get("rendered_full") or _cat0.render_guidance(m, "full"))
                     for m in _g.get("metrics") or []]
    if _g.get("concept_states") is None:
        # a release that was never retrieved leaves every concept UNKNOWN (never an absence)
        import guidance as _gd
        _g["concept_states"] = _gd.concept_states(_g.get("metrics") or [], _g.get("rejected") or [],
                                                  release_checked=bool(_g.get("source")))
    s.guidance.company = _g
    s.guidance.status = _g.get("status")

    # market
    s.market.price = metric("price", "PRICE", price, SPOT, accounting_basis=MARKET, per_share=True,
                            security_basis=sb, period=price_as_of, source="Alpha Vantage global quote",
                            source_date=price_as_of, unit="USD")
    s.market.price_as_of = price_as_of
    mcap = _f(ov.get("MarketCapitalization"))
    s.market.market_cap = metric("market_cap", "MARKET_CAP", mcap, SPOT, accounting_basis=MARKET,
                                 source="Alpha Vantage company profile", source_date=ov.get("LatestQuarter"))
    sh = _f(ov.get("SharesOutstanding"))
    s.market.shares_outstanding = metric("shares_outstanding", "SHARES", sh, POINT, unit="count",
                                         security_basis=sb, source="Alpha Vantage company profile")
    s.market.shares_outstanding_sec = metric("shares_outstanding_sec", "SHARES", _mv(metrics_pre, "shares_outstanding_sec"),
                                             POINT, unit="count", security_basis=ORDINARY_SHARE,
                                             source="SEC cover page (XBRL dei)")
    dy = _mv(metrics_pre, "dividend_yield_statements")
    s.market.dividend_yield = metric("dividend_yield", "DIVIDEND_YIELD", dy, TTM, unit="ratio",
                                     derivation="dividends paid (cash-flow statements, TTM) ÷ market cap",
                                     quality_status=DERIVED if dy is not None else UNAVAILABLE)
    s.market.beta = _f(ov.get("Beta"))
    s.market.high_52w = metric("high_52w", "PRICE", _f(ov.get("52WeekHigh")), SPOT, per_share=True, security_basis=sb)
    s.market.low_52w = metric("low_52w", "PRICE", _f(ov.get("52WeekLow")), SPOT, per_share=True, security_basis=sb)
    mom = d.get("momentum_canon") or {}
    s.market.momentum = {"state": mom.get("label"), "detail": mom.get("detail"), "rule": mom.get("rule")}
    s.market.sma50 = metric("sma50", "PRICE", _f(ov.get("50DayMovingAverage")), SPOT, per_share=True, security_basis=sb)
    s.market.sma200 = metric("sma200", "PRICE", _f(ov.get("200DayMovingAverage")), SPOT, per_share=True, security_basis=sb)

    tsig = (tech or {}).get("signals") or {}
    s.market.return_3m = metric("return_3m", "PRICE_RETURN", _f(tsig.get("momentum_3m")), TTM, accounting_basis=MARKET, unit="ratio",
                                derivation="price change over 3 months (price history)")
    s.market.return_12m = metric("return_12m", "PRICE_RETURN", _f(tsig.get("momentum_12m")), TTM, accounting_basis=MARKET, unit="ratio",
                                 derivation="price change over 12 months (price history)")

    # historical (reconciled basis)
    # the TTM metrics' quality is their four quarters' (HistoricalMetricQuality), not the series-level status: a
    # series whose latest quarter could not be checked is partially reconciled, never simply "reconciled"
    _tq = (rh.get("ttm_quality") or {}).get("status")
    hq = {"SEC_RECONCILED": RECONCILED, "PARTIALLY_RECONCILED": PARTIALLY_RECONCILED}.get(
        _tq, RECONCILED if (_tq is None and rh.get("status") in ("RECONCILED", "SEC_ONLY")) else UNVERIFIED)
    hkw = dict(period_type=TTM, accounting_basis=basis, period=period_end, source=rh.get("basis") or "provider statements",
               quality_status=hq, perimeter=hist_perimeter)
    s.historical.perimeter = {"historical": hist_perimeter, "forward": fwd_perimeter, "action_id": perimeter_aid,
                              "event": (dict({k: rst.get(k) for k in ("kind", "business", "effective", "revenue_share", "basis")},
                                             action_id=perimeter_aid) if separated else None)}
    s.historical.reconciliation = {k: rh.get(k) for k in ("status", "basis", "period_end", "overrides", "disagreements",
                                                          "unverified_quarters", "candidates", "user_limitations", "pipeline_notes", "sec",
                                                          "ttm_quality")}
    s.historical.revenue_ttm = metric("revenue_ttm", "REVENUE", rh.get("revenue_ttm"), unit="USD", **hkw)
    s.historical.operating_income_ttm = metric("operating_income_ttm", "OPERATING_INCOME", rh.get("operating_income_ttm"), unit="USD", **hkw)
    s.historical.pretax_income_ttm = metric("pretax_income_ttm", "PRETAX_INCOME", rh.get("pretax_income_ttm"), unit="USD", **hkw)
    s.historical.net_income_ttm = metric("net_income_ttm", "NET_INCOME", rh.get("net_income_ttm"), unit="USD", **hkw)
    s.historical.operating_margin_ttm = metric("operating_margin_ttm", "OPERATING_MARGIN", rh.get("operating_margin_ttm"),
                                               unit="ratio", derivation="operating income ÷ revenue, reconciled TTM", **hkw)
    s.historical.net_margin_ttm = metric("net_margin_ttm", "NET_MARGIN", rh.get("net_margin_ttm"), unit="ratio",
                                         derivation="net income ÷ revenue, reconciled TTM", **hkw)
    # EPS: three roles, each labelled; use cases select via SELECTION_POLICY
    q_e = ((d.get("earnings") or {}).get("quarterlyEarnings") or [])[:4]
    adj = [_f(q.get("reportedEPS")) for q in q_e]
    eps_adj = sum(adj) if len(adj) == 4 and all(v is not None for v in adj) else None
    s.historical.eps_adjusted_ttm = metric("eps_adjusted_ttm", "EPS", eps_adj, TTM, accounting_basis=ADJUSTED,
                                           per_share=True, security_basis=sb, period=(q_e[0].get("fiscalDateEnding") if q_e else None),
                                           period_end=(q_e[0].get("fiscalDateEnding") if q_e else None),
                                           adjustment_definition_id=_defs.PROVIDER_REPORTED_EPS, adjustment_provider="Alpha Vantage",
                                           adjustment_version=_defs.PROVIDER_REPORTED_EPS,
                                           source="Alpha Vantage EARNINGS reportedEPS (provider / sell-side basis)",
                                           derivation="Σ last four reported quarterly EPS", perimeter=hist_perimeter,
                                           quality_status=VALIDATED if eps_adj is not None else UNAVAILABLE)
    sec_eps = ((rh.get("candidates") or {}).get("sec_ttm") or {}).get("eps_diluted")
    sec_end = ((rh.get("candidates") or {}).get("sec_ttm") or {}).get("period_end")
    gaap_eps, gaap_src = None, None
    if sec_eps is not None and period_end and sec_end and abs((_dt.date.fromisoformat(sec_end[:10]) - _dt.date.fromisoformat(str(period_end)[:10])).days) <= 10:
        gaap_eps = sec_eps * (adr_ratio if (is_adr and adr_ratio) else 1.0)
        gaap_src = "SEC XBRL diluted EPS, trailing twelve months"
    elif rh.get("net_income_ttm") is not None and sh:
        gaap_eps = rh["net_income_ttm"] / sh
        gaap_src = "reconciled TTM net income ÷ shares outstanding"
    s.historical.eps_gaap_ttm = metric("eps_gaap_ttm", "EPS", gaap_eps, TTM, accounting_basis=basis, per_share=True,
                                       security_basis=sb, period=period_end, source=gaap_src or "", perimeter=hist_perimeter,
                                       quality_status=VALIDATED if gaap_eps is not None else UNAVAILABLE)
    # the adjusted↔GAAP bridge, like for like (bridge.py), and the canonical diluted share count
    _ser = rh.get("sec_series") or {}
    _q_all = ((d.get("earnings") or {}).get("quarterlyEarnings") or [])[:28]
    _eps_rows = [dict(r, value=r["value"] * (adr_ratio if (is_adr and adr_ratio) else 1.0)) for r in (_ser.get("eps_diluted") or [])]
    bridge = _br.build(_q_all, _eps_rows, ttm_adjusted=eps_adj, ttm_gaap=gaap_eps, ttm_end=period_end)
    s.historical.adjustment_bridge = bridge
    if bridge.get("methodology_change"):
        # the trailing adjusted EPS spans the old definition only partly: it is labelled with the new version
        s.historical.eps_adjusted_ttm = s.historical.eps_adjusted_ttm.with_value(
            s.historical.eps_adjusted_ttm.value, adjustment_version=bridge.get("definition_version"))
    dsh, dsh_src = _br.latest_diluted_shares(_ser.get("diluted_shares"))
    if dsh and is_adr and adr_ratio:
        dsh, dsh_src = dsh / adr_ratio, (dsh_src or "") + f" ÷ {adr_ratio:g} ordinary shares per ADS"
    if not dsh and rh.get("net_income_ttm") and gaap_eps and gaap_eps > 0:
        dsh, dsh_src = rh["net_income_ttm"] / gaap_eps, "trailing GAAP net income ÷ trailing diluted EPS (weighted diluted count)"
    dsh_q = DERIVED
    if not dsh:
        dsh, dsh_src, dsh_q = sh, "basic shares outstanding (no diluted count available)", UNVERIFIED
    s.historical.diluted_shares = metric("diluted_shares", "DILUTED_SHARES", dsh, POINT, unit="count", security_basis=sb,
                                         source=dsh_src or "", quality_status=dsh_q if dsh else UNAVAILABLE)
    _bq = ((d.get("balance") or {}).get("quarterlyReports") or [{}])[0] or {}
    _eq = _f(_bq.get("totalShareholderEquity"))
    s.historical.book_equity = metric("book_equity", "BOOK_EQUITY", _eq, POINT, accounting_basis=basis, unit="USD",
                                      period=_bq.get("fiscalDateEnding"), source="latest quarterly balance sheet (provider statements)")
    _ni = rh.get("net_income_ttm")
    s.historical.roe_ttm = metric("roe_ttm", "ROE", (_ni / _eq) if (_ni is not None and _eq and _eq > 0) else None, TTM,
                                  accounting_basis=basis, unit="ratio", derivation="reconciled TTM net income ÷ latest book equity",
                                  quality_status=DERIVED if (_ni is not None and _eq and _eq > 0) else UNAVAILABLE)
    _ta = _f(_bq.get("totalAssets"))
    s.historical.roa_ttm = metric("roa_ttm", "ROA", (_ni / _ta) if (_ni is not None and _ta and _ta > 0) else None, TTM,
                                  accounting_basis=basis, unit="ratio", derivation="reconciled TTM net income ÷ latest total assets",
                                  period=_bq.get("fiscalDateEnding"),
                                  quality_status=DERIVED if (_ni is not None and _ta and _ta > 0) else UNAVAILABLE)
    s.historical.eps_provider_ttm = metric("eps_provider_ttm", "EPS", _f(ov.get("EPS")), TTM, accounting_basis="PROVIDER",
                                           per_share=True, security_basis=sb, source="Alpha Vantage company profile",
                                           source_date=ov.get("LatestQuarter"), quality_status=UNVERIFIED, perimeter=hist_perimeter)
    s.historical.revenue_fy_sec = metric("revenue_fy_sec", "REVENUE", _mv(metrics_pre, "revenue_fy_sec"), FY,
                                         accounting_basis=basis, source="SEC XBRL company facts")
    surp = [_f(q.get("surprisePercentage")) for q in ((d.get("earnings") or {}).get("quarterlyEarnings") or [])[:8]]
    surp = [x for x in surp if x is not None]
    s.historical.beat_rate_8q = metric("beat_rate_8q", "BEAT_RATE", (sum(1 for x in surp if x > 0) / len(surp)) if surp else None,
                                       TTM, unit="ratio", derivation=f"share of the last {len(surp)} quarters with a positive surprise",
                                       quality_status=DERIVED if surp else UNAVAILABLE)
    s.historical.beat_window = len(surp)
    s.historical.statements_as_of = period_end
    s.historical.profile_as_of = ov.get("LatestQuarter")

    # consensus (already FX-normalised at the boundary)
    ee = d.get("earnings_estimates_consensus") or {}
    n = int(_f(ee.get("analyst_count")) or 0)
    cq = VALIDATED if n else UNAVAILABLE
    ckw = dict(accounting_basis=ADJUSTED, source="Alpha Vantage EARNINGS_ESTIMATES", quality_status=cq, perimeter=fwd_perimeter)
    edef = dict(adjustment_definition_id=_defs.SELL_SIDE_CONSENSUS_EPS, adjustment_provider="Alpha Vantage",
                adjustment_version=_defs.SELL_SIDE_CONSENSUS_EPS)
    fy1_s, fy1_e = _defs.fiscal_year_bounds(ee.get("current_year_date"))
    fy2_s, fy2_e = _defs.fiscal_year_bounds(ee.get("next_year_date"))
    s.consensus.n_analysts = n
    s.consensus.eps_fy1 = metric("consensus_eps_fy1", "EPS", _f(ee.get("current_year_eps")) if n else None, FY1,
                                 per_share=True, security_basis=sb, period=ee.get("current_year_date"),
                                 period_start=fy1_s, period_end=fy1_e, **edef, **ckw)
    s.consensus.eps_fy2 = metric("consensus_eps_fy2", "EPS", _f(ee.get("next_year_eps")) if n else None, FY2,
                                 per_share=True, security_basis=sb, period=ee.get("next_year_date"),
                                 period_start=fy2_s, period_end=fy2_e, **edef, **ckw)
    s.consensus.revenue_fy1 = metric("consensus_revenue_fy1", "REVENUE", _f(ee.get("current_year_revenue")) if n else None,
                                     FY1, period=ee.get("current_year_date"), period_start=fy1_s, period_end=fy1_e,
                                     **{**ckw, "accounting_basis": basis})
    s.consensus.revenue_fy2 = metric("consensus_revenue_fy2", "REVENUE", _f(ee.get("next_year_revenue")) if n else None,
                                     FY2, period=ee.get("next_year_date"), period_start=fy2_s, period_end=fy2_e,
                                     **{**ckw, "accounting_basis": basis})
    # the explicit period conversion: consensus for the SAME twelve months the engine forecasts
    _w = _defs.ntm_weights(period_end, fy1_s, fy1_e, fy2_s, fy2_e) if n else None
    for _key, _c1, _c2, _concept, _bk in (("eps_ntm", s.consensus.eps_fy1, s.consensus.eps_fy2, "EPS", "eps"),
                                          ("revenue_ntm", s.consensus.revenue_fy1, s.consensus.revenue_fy2, "REVENUE", "rev")):
        _v, _der = None, None
        if _w and _c1.value is not None and (_c2.value is not None or _w[1] < 1e-9):
            _v = _w[0] * _c1.value + _w[1] * (_c2.value or 0.0)
            _der = (f"time-weighted consensus for the twelve months after {period_end}: {_w[0]:.0%} of FY1 "
                    f"(ending {fy1_e}) + {_w[1]:.0%} of FY2 (ending {fy2_e})")
        _extra = dict(per_share=True, security_basis=sb, **edef) if _bk == "eps" else {"accounting_basis": basis}
        setattr(s.consensus, _key, metric(f"consensus_{_key}", _concept, _v, NTM, period_start=(_w[2] if _w else None),
                                          period_end=(_w[3] if _w else None), derivation=_der,
                                          **{**ckw, **({"quality_status": DERIVED} if _v is not None else {"quality_status": UNAVAILABLE}),
                                             **_extra}))
    s.consensus.target_price = metric("analyst_target", "TARGET_PRICE", _f(ov.get("AnalystTargetPrice")), POINT,
                                      accounting_basis=MARKET, per_share=True, security_basis=sb,
                                      source="Alpha Vantage company profile (mean sell-side target)")
    s.consensus.n_ratings = sum(int(_f(ov.get(k)) or 0) for k in ("AnalystRatingStrongBuy", "AnalystRatingBuy",
                                                                   "AnalystRatingHold", "AnalystRatingSell", "AnalystRatingStrongSell"))
    s.consensus.normalization = ee.get("_normalization") or []
    s.consensus.reconciliation_notes = list(rh.get("consensus_notes") or [])

    # provenance
    s.provenance.sources = {
        "price": {"source": "Alpha Vantage", "as_of": price_as_of},
        "statements": {"source": rh.get("basis"), "as_of": period_end},
        "profile": {"source": "Alpha Vantage company profile", "as_of": ov.get("LatestQuarter"),
                    "note": "cross-check only; its derived fields are replaced by the reconciled basis"},
        "consensus": {"source": "Alpha Vantage EARNINGS_ESTIMATES", "n": n},
        "sec": {"available": bool((rh.get("sec") or {}).get("available")), "latest_ttm_end": sec_end},
    }
    s.provenance.fx = {"rate_to_usd": _f(d.get("fx_usd")), "source": d.get("fx_source")}
    _t10 = d.get("treasury_10y") or []
    _rf = _f(_t10[0].get("value")) if (isinstance(_t10, list) and _t10 and isinstance(_t10[0], dict)) else None
    s.provenance.risk_free = {"rate": (_rf / 100.0) if _rf is not None else None,
                              "source": "Alpha Vantage TREASURY_YIELD 10y (daily)" if _rf is not None else None,
                              "as_of": (_t10[0].get("date") if _rf is not None else None)}
    # every metric on a separated perimeter names the corporate action that split it
    from state import stamp_perimeter as _stamp
    _stamp(s, s.historical.perimeter.get("action_id"), ("historical", "consensus", "market"))
    return state


def reconcile_forecast_from_state(s, pred_fund):
    """The single reconciliation stage (forecast_state.reconcile_forecast) fed only from canonical state:
    reconciled history, the adjusted↔GAAP bridge, the canonical diluted share count, perimeter and period."""
    pf = pred_fund or {}
    hist = {k: s.historical.get(k) for k in ("revenue_ttm", "operating_income_ttm", "pretax_income_ttm",
                                              "net_income_ttm", "eps_adjusted_ttm")}
    per = s.historical.get("perimeter") or {}
    dsh = s.historical.get("diluted_shares")
    pf["_diluted_shares_source"] = getattr(dsh, "source", None)
    return _fs.reconcile_forecast(pf, hist, shares=s.value("market.shares_outstanding"),
                                  bridge=dict(s.historical.get("adjustment_bridge") or {}),
                                  diluted_shares=getattr(dsh, "value", None), is_financial=s.issuer.is_financial,
                                  accounting_standard=s.issuer.accounting_standard, per_share_basis=s.issuer.security_basis,
                                  period_end=s.historical.get("statements_as_of"),
                                  perimeter=per.get("forward") or CONSOLIDATED, base_perimeter=per.get("historical") or CONSOLIDATED)


# ═══════════════════════════════════════════════════════════════════════════════════════
# FORECAST / VALUATION stage
# ═══════════════════════════════════════════════════════════════════════════════════════
def build_forecast_stage(state, *, pred_fund, fair_val, story, engine_run, engine_meta, tech, fund, sent,
                         macro, ml, quant_rel, panel_rel, panel_row, all_data):
    s = state
    d = all_data or {}
    sb = s.issuer.security_basis
    pf = pred_fund or {}
    per = s.historical.get("perimeter") or {}
    # the ReconciledForecastState pricing already consumed (one object), else reconciled here from the same inputs
    fs = pf.get("_reconciled") or reconcile_forecast_from_state(s, pf)
    for k, v in fs.items():
        setattr(s.forecast, k, v)
    s.forecast.identities = [{"name": n, "ok": ok, "detail": det} for n, ok, det in _fs.identity_check(fs)]
    # the change a persona wants to state, as a claim (trailing → forecast), only on one basis and perimeter
    from state import compatible as _cmp
    _om_f, _om_h = s.forecast.get("operating_margin"), s.historical.get("operating_margin_ttm")
    s.forecast.operating_margin_change = (
        (_om_f.value - _om_h.value) if (_om_f is not None and _om_h is not None and _om_f.value is not None
                                        and _om_h.value is not None and _cmp(_om_f, _om_h, "growth")[0]
                                        and (per.get("forward") or CONSOLIDATED) == (per.get("historical") or CONSOLIDATED)) else None)
    s.forecast.consensus_gates = [g for g in (pf.get("eps_consensus_gate"), pf.get("rev_consensus_gate")) if g]
    # typed lineage: the share of each forecast that is the model's own (model_weight 0 = the forecast IS consensus)
    s.forecast.consensus_anchor = {"eps": pf.get("eps_consensus_anchor"), "revenue": pf.get("rev_consensus_anchor")}

    fv = fair_val or {}
    st = story or {}
    em = engine_meta or {}
    er = engine_run or {}
    # valuation engine: the ONLY source of valuation vocabulary
    s.valuation.engine = {k: em.get(k) for k in ("engine_id", "engine_family", "display_name", "short_display_name",
                                                  "methodology_description", "label_intrinsic", "terms",
                                                  "allowed_renderer_terms", "forbidden_renderer_terms")}
    s.valuation.engine_status = er.get("status")
    s.valuation.engine_components = er.get("components") or []
    story_value = _f(st.get("value"))
    mech = _f(((fv.get("methods") or {}).get("DCF") or {}).get("price"))
    fam = em.get("engine_family")
    # intrinsic value: the selected engine decides. For the FCFF engine the story value is the
    # intrinsic estimate and the mechanical DCF is its cross-check; other engines publish their own.
    if fam == "intrinsic_cashflow":
        intrinsic, same_model = (story_value, False) if story_value else (mech, True)
        intrinsic_src = "story-driven FCFF (median storyteller)" if story_value else "mechanical earnings DCF (no story ran)"
    else:
        intrinsic = _f(er.get("value")) if er.get("status") == "OK" else None
        same_model = True
        intrinsic_src = em.get("display_name") or "valuation engine"
    s.valuation.intrinsic_value = metric("intrinsic_value", "INTRINSIC_VALUE", intrinsic, INTRINSIC, accounting_basis=MODEL,
                                         per_share=True, security_basis=sb, source=intrinsic_src,
                                         quality_status=MODEL_ESTIMATE if intrinsic else UNAVAILABLE)
    s.valuation.mechanical_value = metric("mechanical_value", "INTRINSIC_VALUE", mech if not same_model else None, INTRINSIC,
                                          accounting_basis=MODEL, per_share=True, security_basis=sb,
                                          source="mechanical earnings DCF", quality_status=MODEL_ESTIMATE)
    s.valuation.same_model = same_model
    # the words for the intrinsic figure follow the model that actually produced it
    _terms = em.get("terms") or {}
    s.valuation.intrinsic_label = (_terms.get("intrinsic") if (fam != "intrinsic_cashflow" or story_value)
                                   else (_terms.get("cross_check") or "mechanical earnings DCF")) or "intrinsic value"
    s.valuation.lifecycle_stage = fv.get("lifecycle_stage")
    s.valuation.damodaran_inputs = dict(d.get("damodaran_inputs") or {"available": False})
    lb = fv.get("lifecycle_basis") or {}
    s.valuation.lifecycle = {"stage": fv.get("lifecycle_stage"),
                             "trailing_revenue_growth": _f(lb.get("trailing_revenue_growth")),
                             "forward_revenue_growth": _f(lb.get("forward_revenue_growth")),
                             "forward_eps_growth": _f(lb.get("forward_eps_growth")),
                             "forward_source": lb.get("forward_source"),
                             "basis": "Damodaran corporate life cycle: revenue growth (trailing and forward), profitability, payout"}
    s.valuation.gaap_distortion = dict(fv.get("stage_check") or {})      # deterministic test; see composite steps
    dm = fv.get("dcf_meta") or {}
    s.valuation.mechanical_inputs = {k: dm.get(k) for k in ("basis", "rf", "erp", "beta", "ke_initial", "g1", "g_terminal",
                                                            "high_growth_years", "roe")}
    # MechanicalDCFState (valuation_engines.two_stage_per_share): everything the mechanical value reproduces from
    s.valuation.mechanical_dcf = dm.get("dcf_state")
    stories = list(st.get("stories") or [])
    A = st.get("assumptions") or {}
    s.valuation.story = {
        "ran": bool(story_value),
        "stories": stories,
        "n_storytellers": len(stories),
        "adopted": next((x.get("index") for x in stories if x.get("adopted")), None),
        "aggregation": st.get("consensus_method"),
        "assumptions": {k: A.get(k) for k in ("tam_usd_b", "peak_share", "revenue_N_usd_b", "implied_rev_cagr", "growth_stage1",
                                              "stage1_years", "years", "margin_now", "margin_target", "sales_to_capital",
                                              "failure_probability", "wacc_initial", "wacc_terminal", "terminal_growth",
                                              "terminal_roc", "pv_terminal_share", "revenue_reported_usd_b", "revenue_start_usd_b",
                                              "revenue_base_note", "margin_base_note", "cost_of_debt",
                                              "debt_weight", "beta", "beta_stable", "wacc_terminal_basis", "wacc_hold_years", "stable_rate_kept",
                                              "defaults_used", "discount_rate_path", "margin_proposed", "margin_cap",
                                              "margin_floor")},
        "segments": A.get("segments") or [],
        "segment_build": dict((d.get("segment_brief") or {}).get("required_evidence") or {"required": False}),
        "dispersion": st.get("dispersion"),
        "breakeven": st.get("breakeven"),
        "bridge": st.get("bridge"),
        "text": st.get("text") or st.get("story"),
        "key_risk": st.get("key_risk"),
        # the revenue the FCFF actually starts from (the continuing company after a separation)
        "start_revenue": ((A.get("revenue_start_usd_b") or A.get("revenue_reported_usd_b")) * 1e9
                          if (A.get("revenue_start_usd_b") or A.get("revenue_reported_usd_b")) is not None else None),
        "revenue_base_note": A.get("revenue_base_note"),
        "margin_base_note": A.get("margin_base_note"),
        "start_margin": A.get("margin_now"),
        "margin_change": ((A["margin_target"] - A["margin_now"])
                          if (A.get("margin_target") is not None and A.get("margin_now") is not None) else None),
    }
    s.valuation.sensitivity = [dict(x) for x in (st.get("sensitivity") or [])]

    # ── typed claims: no material financial float reaches a persona without concept, period and perimeter ──
    _per = s.historical.get("perimeter") or {}
    _hp, _fp, _aid = _per.get("historical") or CONSOLIDATED, _per.get("forward") or CONSOLIDATED, _per.get("action_id")
    s.valuation.lifecycle_trailing_growth = metric(
        "revenue_growth_ttm", "REVENUE_GROWTH", _f(lb.get("trailing_revenue_growth")), TTM, accounting_basis=GAAP, unit="ratio",
        display_unit="PERCENT", perimeter=_hp, period=s.historical.get("statements_as_of"),
        source="reconciled quarterly revenue, trailing four quarters over the four before",
        derivation="trailing twelve months ÷ the prior twelve months − 1", quality_status=DERIVED)
    s.valuation.lifecycle_forward_growth = metric(
        "revenue_growth_fy2", "REVENUE_GROWTH", _f(lb.get("forward_revenue_growth")), FY2, accounting_basis=GAAP, unit="ratio",
        display_unit="PERCENT", perimeter=_fp, source=str(lb.get("forward_source") or "sell-side consensus"),
        derivation="consensus revenue, next fiscal year ÷ this fiscal year − 1", quality_status=DERIVED)
    _di0 = s.valuation.get("damodaran_inputs") or {}
    s.valuation.industry_operating_margin = metric(
        "industry_operating_margin", "OPERATING_MARGIN", _f(_di0.get("industry_operating_margin")), TTM, accounting_basis=GAAP,
        unit="ratio", display_unit="PERCENT", scope="INDUSTRY", perimeter="industry aggregate",
        source=f"Damodaran margin dataset: pre-tax unadjusted operating margin, {_di0.get('industry')} ({_di0.get('region')}, "
               f"{_di0.get('industry_as_of')})", quality_status=VALIDATED if _di0.get("industry_operating_margin") is not None else UNAVAILABLE)
    # StoryBase: the business the story values, over one stated period (story_base.py)
    _sb = d.get("story_base") or {}
    _skw = dict(period_type=TTM, accounting_basis=GAAP, perimeter=_sb.get("perimeter") or CONSOLIDATED,
                period=_sb.get("period_end"), period_start=_sb.get("period_start"), period_end=_sb.get("period_end"),
                corporate_action_id=_sb.get("corporate_action_id"), derivation=_sb.get("derivation"),
                quality_status=_sb.get("quality_status") or UNAVAILABLE)
    s.valuation.story_base = {
        "revenue": metric("story_base_revenue", "REVENUE", _sb.get("revenue"), unit="USD", **_skw),
        "operating_income": metric("story_base_operating_income", "OPERATING_INCOME", _sb.get("operating_income"), unit="USD", **_skw),
        "operating_margin": metric("story_base_operating_margin", "OPERATING_MARGIN", _sb.get("operating_margin"), unit="ratio",
                                   display_unit="PERCENT", **_skw),
        "label": _sb.get("label") or "reconciled trailing operating margin", "revenue_label": _sb.get("revenue_label"),
        "perimeter": _sb.get("perimeter"), "corporate_action_id": _sb.get("corporate_action_id"),
        "perimeter_id": __import__("business").perimeter_id(_sb),
        "quality_status": _sb.get("quality_status"), "derivation": _sb.get("derivation"), "cross_check": _sb.get("cross_check")}
    s.valuation.perimeter_adjustment = dict(d.get("perimeter_adjustment") or {})
    s.valuation.story_maturity_margin = metric(
        "story_maturity_margin", "OPERATING_MARGIN", _f(A.get("margin_target")), INTRINSIC, accounting_basis=MODEL, unit="ratio",
        display_unit="PERCENT", perimeter=_sb.get("perimeter") or CONSOLIDATED, corporate_action_id=_sb.get("corporate_action_id"),
        source="adopted storyteller (after guardrails)", quality_status=MODEL_ESTIMATE if A.get("margin_target") is not None else UNAVAILABLE)
    # story assumptions and the break-even solve, typed: no raw financial float reaches a persona
    _tkw = dict(period_type=INTRINSIC, accounting_basis=MODEL, perimeter=_sb.get("perimeter") or CONSOLIDATED,
                corporate_action_id=_sb.get("corporate_action_id"))
    _bn = lambda x: (float(x) * 1e9) if x is not None else None
    s.valuation.story_revenue_maturity = metric(
        "story_revenue_maturity", "REVENUE", _bn(A.get("revenue_N_usd_b")), unit="USD", source="adopted storyteller",
        derivation="market size at maturity × peak share", quality_status=MODEL_ESTIMATE if A.get("revenue_N_usd_b") else UNAVAILABLE, **_tkw)
    s.valuation.story_tam = metric("story_tam", "MARKET_SIZE", _bn(A.get("tam_usd_b")), unit="USD", source="adopted storyteller",
                                   quality_status=MODEL_ESTIMATE if A.get("tam_usd_b") else UNAVAILABLE, **_tkw)
    s.valuation.story_share = metric("story_share", "MARKET_SHARE", _f(A.get("peak_share")), unit="ratio", display_unit="PERCENT",
                                     source="adopted storyteller", quality_status=MODEL_ESTIMATE, **_tkw)
    s.valuation.story_years = metric("story_years", "YEARS", _f(A.get("years")), unit="count", currency=None,
                                     source="adopted storyteller", quality_status=MODEL_ESTIMATE, **_tkw)
    _be = st.get("breakeven") or {}
    _bm = _be.get("story_margin")
    s.valuation.breakeven_revenue = metric(
        "breakeven_revenue", "BREAK_EVEN_REVENUE", _bn(_be.get("revenue_needed_N_usd_b")) if _be.get("revenue_needed_reachable") else None,
        unit="USD", source="break-even solve (story DCF = today's price)",
        derivation=f"revenue at maturity at which the story DCF equals the price, other story assumptions held"
                   + (f" (operating margin {float(_bm) * 100:.1f}%)" if _bm is not None else ""),
        quality_status=MODEL_ESTIMATE if _be.get("revenue_needed_reachable") else UNAVAILABLE, **_tkw)
    s.valuation.breakeven_margin = metric(
        "breakeven_margin", "BREAK_EVEN_OPERATING_MARGIN", _f(_be.get("margin_needed")) if _be.get("margin_needed_reachable") else None,
        unit="ratio", display_unit="PERCENT", source="break-even solve (story DCF = today's price)",
        derivation="operating margin at maturity at which the story DCF equals the price on the story's revenue path",
        quality_status=MODEL_ESTIMATE if _be.get("margin_needed_reachable") else UNAVAILABLE, **_tkw)
    s.valuation.breakeven_revenue_cagr = metric(
        "breakeven_revenue_cagr", "BREAK_EVEN_REVENUE_CAGR", _f(_be.get("rev_cagr_needed")) if _be.get("revenue_needed_reachable") else None,
        unit="ratio", display_unit="PERCENT", source="break-even solve (story DCF = today's price)",
        derivation="(breakeven_revenue / story_base_revenue) ^ (1 / story_years) − 1",
        quality_status=MODEL_ESTIMATE if _be.get("revenue_needed_reachable") else UNAVAILABLE, **_tkw)
    # the CAGR is bound to ONE pair: it measures growth from the starting revenue to the break-even revenue
    s.valuation.breakeven_growth_binding = {"start_revenue_metric_id": "story_base_revenue",
                                            "target_revenue_metric_id": "breakeven_revenue", "years_metric_id": "story_years"}
    # DiscountRatePath: the exact vector the valuation discounted with (story_dcf), and its claims
    _dp = dict(A.get("discount_rate_path") or {})
    _ckw = dict(period_type=INTRINSIC, accounting_basis=MODEL, unit="ratio", display_unit="PERCENT", quantity_type="YIELD",
                perimeter=_sb.get("perimeter") or CONSOLIDATED, corporate_action_id=_sb.get("corporate_action_id"))
    s.valuation.discount_rate_path = dict(
        _dp, initial_wacc_metric=metric("initial_wacc", "COST_OF_CAPITAL", _dp.get("initial_wacc"), derivation="rf + beta × ERP, "
                                        "with the company's debt ratio and synthetic cost of debt", source="story valuation", **_ckw),
        terminal_wacc_metric=metric("terminal_wacc", "COST_OF_CAPITAL", _dp.get("terminal_wacc"), derivation=_dp.get("rule"),
                                    source="story valuation (the rate used in the terminal value)", **_ckw),
        benchmark_wacc_metric=metric("mature_benchmark_wacc", "COST_OF_CAPITAL", _dp.get("mature_benchmark_wacc"),
                                     derivation="Damodaran Total Market (without financials) rebuilt at today's rf and ERP",
                                     source="Damodaran datasets (benchmark; used only if the rule selects it)", **_ckw))

    # pricing
    comp = _f(fv.get("fair_price"))
    s.pricing.market_composite = metric("market_composite", "MARKET_PRICING", comp, POINT, accounting_basis=MARKET,
                                        per_share=True, security_basis=sb, source="multiples and analyst targets")
    # PricingState: every method with its input, multiple, basis, weights and adjustments; raw → trim → cap → final
    import pricing_state as _pst
    ps = _pst.build(fv, s.value("market.price"))
    s.pricing.state = ps
    s.pricing.methods = [{"name": m["method"], "label": m["label"], "price": _f(m.get("raw_price")),
                          "share": _f(m.get("share_final")), "share_raw": _f(m.get("share_raw")),
                          "confidence": _f(m.get("confidence")), "stage_weight": _f(m.get("stage_weight")),
                          "effective_weight": _f(m.get("effective_weight")), "target_multiple": _f(m.get("target_multiple")),
                          "multiple_basis": m.get("multiple_basis"), "raw_input": _f(m.get("raw_input")),
                          "raw_input_basis": m.get("raw_input_basis"), "in_composite": bool(m.get("in_composite")),
                          "dropped_reason": m.get("dropped_reason"), "adjustments": list(m.get("adjustments") or [])}
                         for m in ps["methods"] if m["method"] != "DCF"]
    s.pricing.composite_steps = [dict(x) for x in ps["steps"]]
    _pkw = dict(accounting_basis=MARKET, per_share=True, security_basis=sb, unit="USD")
    s.pricing.raw_composite = metric("raw_composite", "MARKET_PRICING", _f(ps.get("raw_composite")), POINT,
                                     derivation="Σ method price × (confidence × lifecycle weight) ÷ Σ weights", **_pkw)
    s.pricing.post_trim_composite = metric("post_trim_composite", "MARKET_PRICING", _f(ps.get("post_trim_composite")), POINT,
                                           derivation="raw composite after outlier trimming", **_pkw)
    _cap = ps.get("cap") or {}
    s.pricing.cap = {"applied": bool(_cap.get("applied")), "upper": _f(_cap.get("upper")), "lower": _f(_cap.get("lower"))}
    _atm = next((m for m in ps["methods"] if m["method"] == "Analyst Target"), None)
    s.pricing.analyst_target_adjusted = metric(
        "analyst_target_adjusted", "TARGET_PRICE", _f((_atm or {}).get("risk_adjusted_analyst_target_contribution")), POINT,
        derivation="mean target with 55% of its implied upside kept (ANALYST_TARGET_OPTIMISM)", **_pkw)
    s.pricing.analyst_target_in_composite = bool(_atm and _atm.get("in_composite"))
    s.pricing.adjustments = [dict(a) for a in ps["adjustments"]]
    eps_f = s.forecast.eps
    eps_a = s.select("TRAILING_PE_DENOMINATOR")
    eps_g = s.select("GAAP_PE_DENOMINATOR")
    pm = s.market.price
    from state import compatible as _compat0
    pol_methods = set((s.issuer.get("policy") or {}).get("pricing_methods") or ())
    pe_applies = not pol_methods or "P/E" in pol_methods     # the issuer policy excludes earnings multiples for some classes

    def _pe(den, mid, label, basis):
        ok = pe_applies and den is not None and bool(den.value) and den.value > 0 and _compat0(pm, den, "ratio")[0]
        v = (pm.value / den.value) if ok else None
        why = (f"price ÷ {label}" if pe_applies else "not applicable: this issuer's policy excludes earnings multiples")
        return metric(mid, "PE", v, den.period_type if den is not None else NTM, accounting_basis=basis, unit="x",
                      derivation=why, quality_status=DERIVED if v else UNAVAILABLE)
    s.pricing.pe_forward = _pe(eps_f, "pe_forward", "forecast EPS (next 12 months)", ADJUSTED)
    s.pricing.pe_trailing = _pe(eps_a, "pe_trailing", "trailing adjusted EPS", ADJUSTED)
    s.pricing.pe_trailing_gaap = _pe(eps_g, "pe_trailing_gaap", "trailing reported EPS", s.historical.eps_gaap_ttm.accounting_basis)
    g = s.value("forecast.eps_growth")
    peg = (s.value("pricing.pe_forward") / (g * 100.0)) if (s.value("pricing.pe_forward") and g and g > 0.02) else None
    s.pricing.peg_forward = metric("peg_forward", "PEG", peg, NTM, accounting_basis=ADJUSTED, unit="ratio",
                                   derivation="forward P/E ÷ forecast EPS growth (in percent)",
                                   quality_status=DERIVED if peg else UNAVAILABLE)
    # price-to-value relationships: computed once, here, through the compatibility gate. The
    # renderer only formats them; a direction word ("above"/"below") comes from the sign.
    from state import compatible as _compat

    def _gap(ref, mid, what):
        ok, _why = _compat(pm, ref, "ratio") if ref is not None else (False, [])
        v = (pm.value / ref.value - 1.0) if (ok and ref.value) else None
        return metric(mid, "PRICE_GAP", v, SPOT, accounting_basis=MARKET, unit="ratio",
                      derivation=f"price ÷ {what} − 1", quality_status=DERIVED if v is not None else UNAVAILABLE)
    s.pricing.price_vs_composite = _gap(s.pricing.market_composite, "price_vs_composite", "market composite")
    s.pricing.target_vs_price = metric("target_vs_price", "TARGET_GAP",
                                       (s.consensus.target_price.value / pm.value - 1.0)
                                       if (s.consensus.target_price.value and pm.value) else None, SPOT,
                                       accounting_basis=MARKET, unit="ratio", derivation="mean analyst target ÷ price − 1")
    s.valuation.price_vs_intrinsic = _gap(s.valuation.intrinsic_value, "price_vs_intrinsic", "intrinsic value")
    s.valuation.price_vs_mechanical = _gap(s.valuation.mechanical_value, "price_vs_mechanical", "mechanical value")
    # engine vs consensus only on the same twelve months (time-weighted FY1/FY2) and a compatible definition
    _fe, _ce = s.forecast.eps, s.consensus.get("eps_ntm")
    _ok, _why = _compat(_fe, _ce, "level") if _ce is not None else (False, ["no consensus for the forecast period"])
    s.forecast.eps_vs_consensus = metric("eps_vs_consensus", "EPS_GAP", (_fe.value / _ce.value - 1.0) if (_ok and _ce.value) else None,
                                         NTM, accounting_basis=ADJUSTED, unit="ratio",
                                         quality_status=DERIVED if (_ok and _ce.value) else UNAVAILABLE,
                                         derivation=("forecast adjusted EPS ÷ time-weighted consensus EPS for the same twelve months − 1; "
                                                     + (_defs.reconciliation_evidence(_fe, _ce) or "one adjustment definition"))
                                         if _ok else "not comparable: " + "; ".join(_why))
    pc = d.get("peer_comps") or {}

    PRICE_MULTIPLES = ("pe", "ev_ebitda", "ps", "fcf_yield", "div_yield")

    def _implausible_scale(r):
        # the panel keeps foreign filers in their reporting currency while market cap is in USD, so every
        # price-based multiple of such a row is off by the exchange rate; two or more impossible values
        # identify the row (withholding only — nothing is corrected or re-derived)
        hits = [(_f(r.get("pe")) is not None and 0 < _f(r.get("pe")) < 2),
                (_f(r.get("ev_ebitda")) is not None and 0 < _f(r.get("ev_ebitda")) < 1.5),
                (_f(r.get("ps")) is not None and _f(r.get("ps")) < 0.3 and (_f(r.get("op_margin")) or 0) > 0.10),
                (_f(r.get("fcf_yield")) is not None and _f(r.get("fcf_yield")) > 0.5),
                (_f(r.get("div_yield")) is not None and _f(r.get("div_yield")) > 0.2)]
        return sum(hits) >= 2

    def _meaningful(row):
        r = dict(row or {})
        if r.get("pe_unreconciled") or _implausible_scale(r):
            for k in PRICE_MULTIPLES:
                r[k] = None
            r["multiples_withheld"] = "price-based multiples on an inconsistent currency or scale basis"
        # a trailing P/E on negative earnings is not a multiple; it is published as unavailable
        if _f(r.get("pe")) is not None and _f(r.get("pe")) <= 0:
            r["pe"] = None
        return r
    _sep = (s.historical.get("perimeter") or {}).get("event")
    _subj = _meaningful(pc.get("subject")) if (pc.get("subject") and not _sep) else None
    # a derived statistic inherits the eligibility of every figure it is computed from: a percentile of a withheld
    # subject multiple is withheld with it (RelativeValuationClaim)
    _pct = {k: v for k, v in (pc.get("percentiles") or {}).items() if _subj and _subj.get(k) is not None}
    _why_withheld = ("its trailing figures include the separated business" if _sep else
                     (None if _subj else "the subject is not in the peer panel"))
    s.pricing.peers = {"subject": _subj, "subject_withheld": _why_withheld if not _subj else None,
                       "peers": [_meaningful(p) for p in (pc.get("peers") or [])], "percentiles": _pct,
                       "percentiles_withheld": (sorted(set(pc.get("percentiles") or {}) - set(_pct)) or None),
                       "as_of": pc.get("as_of"), "basis": pc.get("basis"), "sector": pc.get("sector"),
                       "source": "Sharadar panel (trailing multiples)"}
    _meta = {"pe": ("P/E", "TRAILING", "GAAP"), "ev_ebitda": ("EV/EBITDA", "TRAILING", "GAAP"), "ps": ("P/S", "TRAILING", "GAAP"),
             "fcf_yield": ("FCF yield", "TRAILING", "GAAP"), "op_margin": ("operating margin", "TRAILING", "GAAP"),
             "rev_growth": ("revenue growth", "TRAILING", "GAAP")}
    s.pricing.relative = {
        k: {"metric": lab, "subject_value": (_subj or {}).get(k), "peer_values": [p.get(k) for p in (pc.get("peers") or [])
                                                                                   if p.get(k) is not None],
            "percentile": _pct.get(k), "period_basis": per_b, "accounting_basis": acc_b,
            "perimeter_basis": ("pre-separation" if _sep else "consolidated"), "peer_perimeter_basis": "as reported",
            "comparison_allowed": _pct.get(k) is not None and not _sep,
            "reason": (None if (_pct.get(k) is not None and not _sep) else (_why_withheld or "no subject value"))}
        for k, (lab, per_b, acc_b) in _meta.items()}

    # quant: reliability decides what exists outside diagnostics
    s.quant.price_model = {"reliability": dict(quant_rel or {}),
                           "diagnostics_only": {"expected_return": _f((ml or {}).get("expected_return")),
                                                "direction_accuracy": _f((ml or {}).get("backtest_dir_accuracy")),
                                                "rank_ic": _f((ml or {}).get("backtest_rank_ic")),
                                                "mae": _f((ml or {}).get("backtest_mae")),
                                                "n_oos_eff": _f((ml or {}).get("n_oos_effective"))}}
    _pfig = (dict(panel_row, rank_share=(_f(panel_row.get("xs_rank_pct")) or 0.0) / 100.0)
             if (panel_row and (panel_rel or {}).get("persona_visibility")) else None)
    s.quant.panel_rank = {"reliability": dict(panel_rel or {}),
                          "figures": _pfig,
                          "diagnostics_only": dict(panel_row) if panel_row else None}

    # macro, technical, sentiment
    mr = macro or {}
    s.macro.view = {"return": _f(mr.get("expected_return")), "confidence": _f(mr.get("confidence"))}
    s.macro.signals = {k: v for k, v in (mr.get("signals") or {}).items() if isinstance(v, (int, float, str))}
    tr = tech or {}
    s.technical.view = {"return": _f(tr.get("expected_return")), "confidence": _f(tr.get("confidence"))}
    s.technical.signals = {k: v for k, v in (tr.get("signals") or {}).items() if isinstance(v, (int, float, str))}
    fd = fund or {}
    s.technical.scorecard = fd.get("scorecard")
    rc = d.get("risk_card") or {}
    s.technical.risk = {k: rc.get(k) for k in ("vol_1y", "max_drawdown_1y", "dist_52w_high", "corr_spy_1y")}
    sc = sent or {}
    comp_s = sc.get("components") or {}
    s.sentiment.news = {"score": _f(comp_s.get("news_sentiment")), "n_articles": comp_s.get("news_n_articles")}

    # insiders, leadership, corporate actions
    ic = d.get("insider_canon") or {}
    import insider as _ins0
    _ican = {k: ic.get(k) for k in ("characterization", "discretionary_activity_signal", "governance_signal",
                                    "plan_only", "plan_sale_value", "open_market_sale_value", "purchase_value",
                                    "window_months", "n_transactions", "note")}
    _ican["motive_inference_allowed"] = False
    _mc = (s.value("market.price") or 0) * (s.value("market.shares_outstanding") or 0)
    _ican["materiality"] = _ins0.materiality(_ican, _mc or None) if ic else None
    s.insiders.canon = _ican
    s.insiders.factor = {k: (ic.get("factor") or {}).get(k) for k in ("score", "basis")}
    # 13F ownership: positions, not intent, reported ~45 days after quarter end. A snapshot older than
    # two filing cycles is not current ownership and is withheld (kept in QA).
    o13 = d.get("ownership_13f") or {}
    age = None
    try:
        age = (_dt.date.today() - _dt.date.fromisoformat(str(o13.get("as_of"))[:10])).days if o13.get("as_of") else None
    except ValueError:
        age = None
    fresh = bool(o13) and age is not None and age <= 200
    names = [filer_display_name(x) for x in str(o13.get("guru_names") or "").split(",") if x.strip()]
    # the name list is a sample of the tracked holders (the snapshot keeps at most a few names): "and N others" counts
    # against the tracked-holder COUNT, never against the length of the sample
    shown = names[:6]
    n_tracked = int(round(o13["n_gurus"])) if o13.get("n_gurus") is not None else len(names)
    others = max(n_tracked - len(shown), 0)
    s.ownership.f13 = ({**{k: o13.get(k) for k in ("as_of", "n_holders", "n_holders_prev", "n_gurus", "guru_new_buys",
                                                   "guru_adds", "guru_trims", "guru_exits", "source")},
                        "guru_names_shown": shown, "guru_names_others": others,
                        "guru_names_short": (", ".join(shown) + (f" and {others} other{'s' if others != 1 else ''}" if others else ""))
                        if shown else None,
                        "n_holders_change": ((o13.get("n_holders") - o13.get("n_holders_prev"))
                                             if (o13.get("n_holders") is not None and o13.get("n_holders_prev") is not None) else None)}
                       if fresh else None)
    # OwnershipEvidence permissions: a 13F reports positions ~45 days after quarter end; it never records intent
    s.ownership.permissions = {"state_positions": True, "state_changes": True, "infer_intent": False,
                               "infer_motive": False, "use_as_return_signal": False}
    s.ownership.status = ("CURRENT" if fresh else ("STALE" if o13 else "UNAVAILABLE"))
    s.ownership.age_days = age
    s.ownership.diagnostics_only = dict(o13) if (o13 and not fresh) else None
    lc = d.get("leadership_canon") or {}
    s.leadership.roles = {r: {k: (lc.get(r) or {}).get(k) for k in ("name", "title", "source", "status", "filed")}
                          for r in ("CEO", "CFO", "COO") if isinstance(lc.get(r), dict) and (lc.get(r) or {}).get("name")}
    s.leadership.status = dict(d.get("leadership_status") or {})
    s.leadership.regime = lc.get("_regime")
    s.corporate_actions.states = [dict(x) for x in (d.get("corp_states") or []) if isinstance(x, dict)]
    s.corporate_actions.restatement = d.get("restatement")
    s.corporate_actions.pending = d.get("pending_deals") or []
    # the canonical CorporateActionState: one record per action id, status from filings only (corporate_actions.py)
    import corporate_actions as _ca
    _ev0 = dict((s.historical.get("perimeter") or {}).get("event") or {})
    _rsd = d.get("restatement") or {}
    if _ev0 and _rsd.get("business") == _ev0.get("business"):
        _ev0.update({k: _rsd.get(k) for k in ("edgar", "announced") if _rsd.get(k)})
    s.corporate_actions.state = _ca.build(s.identity.ticker, _ev0 or None, d.get("edgar_actions"), d.get("corp_actions"),
                                          d.get("segments_canonical"))
    # BusinessDecompositionState: the businesses a story may value, from the filed segment state and the canonical
    # corporate actions only (business.py). Rebuilt here from the state's own inputs, never taken from a storyteller.
    import business as _bz
    s.valuation.business_decomposition = _bz.build(
        s.identity.ticker, d.get("segments_canonical"), s.corporate_actions.state, d.get("story_base"),
        (d.get("perimeter_adjustment") or {}).get("segment_ids"))

    # qualitative context for the personas (text evidence, never numbers of record)
    s.context.news = [{"title": n.get("title"), "source": n.get("source"), "sentiment": n.get("overall_sentiment_label")}
                      for n in (d.get("news") or [])[:8] if isinstance(n, dict)]
    s.context.transcript_excerpt = (d.get("transcript_excerpt") or "")[:4000]
    s.context.edgar_actions = [dict(x) for x in (d.get("edgar_actions") or [])[:5] if isinstance(x, dict)]
    _cn_ctx = d.get("segments_canonical") or {}
    s.context.segments = ([{"segment_id": g.get("segment_id"), "name": g.get("name"), "revenue": g.get("revenue"),
                            "operating_income": g.get("operating_income"), "revenue_share": g.get("revenue_share")}
                           for g in _cn_ctx.get("segments") or []] if _cn_ctx.get("status") == "RECONCILED" else [])
    s.context.earnings_date = ((d.get("earnings_calendar") or [{}])[0] or {}).get("reportDate") if isinstance(d.get("earnings_calendar"), list) else None
    from state import stamp_perimeter as _stamp
    _stamp(s, (s.historical.get("perimeter") or {}).get("action_id"), ("forecast", "valuation", "pricing"))
    # segment counts per perimeter: the filed count describes the business AS FILED (before a separation); the
    # continuing count removes the separated segment(s). A count is never cited without its perimeter.
    _can = (s.segments.get("canonical") or {})
    _nseg = len(_can.get("segments") or []) if _can.get("status") == "RECONCILED" or _can.get("segments") else None
    _sep = list((s.valuation.get("perimeter_adjustment") or {}).get("segment_ids") or [])
    _hper = (s.historical.get("perimeter") or {})
    _paid = _hper.get("action_id")
    s.valuation.segment_count_filed = metric(
        "segment_count_filed", "SEGMENT_COUNT", _nseg, FY, unit="count", currency=None, quantity_type="LEVEL",
        perimeter=(_hper.get("historical") or CONSOLIDATED) if _paid else CONSOLIDATED, corporate_action_id=_paid,
        period=_can.get("period_end"), source=_can.get("source") or "filed segment note", quality_status=VALIDATED if _nseg else UNAVAILABLE)
    s.valuation.segment_count_continuing = metric(
        "segment_count_continuing", "SEGMENT_COUNT", (_nseg - len(_sep)) if (_nseg and _paid and _sep) else None, FY,
        unit="count", currency=None, quantity_type="LEVEL", perimeter=(_hper.get("forward") or CONSOLIDATED),
        corporate_action_id=_paid, period=_can.get("period_end"),
        derivation="filed segments less the separated segment(s)", quality_status=DERIVED if (_nseg and _paid and _sep) else UNAVAILABLE)
    # typed relations between typed quantities, and the scope of every model-derived claim
    import evidence as _evd
    import relations as _rel
    s.relations.state = _rel.build(s)
    s.evidence.state = _evd.build(s)
    return state


# ═══════════════════════════════════════════════════════════════════════════════════════
# SCENARIOS / BLEND / RECOMMENDATION (after the panel's structured judgement)
# ═══════════════════════════════════════════════════════════════════════════════════════
def build_scenarios(state, debate):
    """Structured panel judgement → ScenarioState. Prose is NOT stored here."""
    s = state
    db = debate or {}
    seats = db.get("seats") or {}
    rows = {}
    for key in ("bear", "neutral", "bull"):
        e = seats.get(key) or {}
        ret = _f(e.get("return"))
        # one rating table for every label in the document: a seat's label is the band of its return
        rows[key] = {"return": ret, "label": rating_for(round(ret, 3)) if ret is not None else None, "model": e.get("model"),
                     "confidence": _f(e.get("confidence"))}
    pb, pn, pu = _f(db.get("p_bear")), _f(db.get("p_neutral")), _f(db.get("p_bull"))
    for key, p in (("bear", pb), ("neutral", pn), ("bull", pu)):
        rows[key]["probability"] = p
    s.scenarios.bear = rows["bear"]
    s.scenarios.neutral = rows["neutral"]
    s.scenarios.bull = rows["bull"]
    complete = all(rows[k]["return"] is not None and rows[k]["probability"] is not None for k in rows)
    pm_ret = (sum(rows[k]["return"] * rows[k]["probability"] for k in rows)) if complete else None
    s.scenarios.complete = complete
    s.scenarios.pm_return = metric("pm_return", "PM_SCENARIO_RETURN", pm_ret, NTM, accounting_basis=MODEL, unit="ratio",
                                   derivation="Σ scenario probability × scenario price return",
                                   quality_status=MODEL_ESTIMATE if complete else UNAVAILABLE)
    s.scenarios.pm_label = rating_for(round(pm_ret, 3)) if pm_ret is not None else None
    s.scenarios.pm_confidence = _f(db.get("pm_confidence"))
    s.scenarios.panel_complete = bool(db.get("panel_complete")) and complete
    s.scenarios.missing_roles = list(db.get("missing_roles") or [])
    s.scenarios.probability_audit = db.get("scenario_probability_audit")
    s.scenarios.evidence = db.get("evidence_lineage")
    # {seat: {accepted, rejected}}: the evidence state the portfolio manager was given when it set the probabilities
    s.scenarios.pm_evidence_state = db.get("pm_evidence_state")
    typed = [(t, _typed_trigger(t, s)) for t in (db.get("thesis_triggers") or [])]
    # a "what would change the view" condition must not already hold in the current state: such a trigger is omitted
    s.scenarios.triggers = [x for _t, x in typed if x and not x.get("currently_satisfied")]
    s.scenarios.triggers_dropped = [{"metric_id": t.get("metric_id"), "threshold": t.get("threshold"), "operator": t.get("condition_operator"),
                                     "source_claim": t.get("source_claim"),
                                     "reason": ("already satisfied by the current state: " + ", ".join(
                                         f"{c} = {v:g}" for c, v in (x.get("current_values") or []) if trigger_holds(x, v))
                                                if x else (trigger_source_refusal(s, t.get("source_claim"))
                                                           or "untyped or unit/magnitude inconsistent with its source"))}
                                    for t, x in typed if not x or x.get("currently_satisfied")]
    return state


_OPS = {">": "above", ">=": "above", "<": "below", "<=": "below"}


_UNIT_STYPE = {"BASIS_POINTS": "BPS", "PERCENTAGE_POINTS": "PP", "PERCENT": "PCT", "DECIMAL_FRACTION": "PCT",
               "MULTIPLE": "MULTIPLE", "USD_PER_SHARE": "CURRENCY", "USD": "CURRENCY_BIG", "USD_MILLION": "CURRENCY_BIG",
               "USD_BILLION": "CURRENCY_BIG"}
_STYPE_UNIT = {"PCT": "PERCENT", "PCT_SIGNED": "PERCENT", "MULTIPLE": "MULTIPLE", "CURRENCY": "USD_PER_SHARE",
               "CURRENCY_BIG": "USD", "PP": "PERCENTAGE_POINTS", "BPS": "BASIS_POINTS"}


def _written_unit(raw):
    """(number, unit written or None) for a threshold string: '35 bps', '6%', '17x', '$3.2B', '35'."""
    import re as _re
    m = _re.fullmatch(r"(-|−)?\s*(\$)?\s*(\d+(?:\.\d+)?)\s*(%|bps|bp|basis points?|pp|pts|percentage points?|x|B|bn|M)?",
                      raw.replace(",", "").strip(), _re.I)
    if not m:
        return None, None
    v = float(m.group(3)) * (-1 if m.group(1) else 1)
    u = (m.group(4) or "").lower()
    if m.group(2):
        return (v * 1e9, "USD") if u in ("b", "bn") else ((v * 1e6, "USD") if u == "m" else (v, "USD_PER_SHARE"))
    unit = {"%": "PERCENT", "bps": "BASIS_POINTS", "bp": "BASIS_POINTS", "basis point": "BASIS_POINTS",
            "basis points": "BASIS_POINTS", "pp": "PERCENTAGE_POINTS", "pts": "PERCENTAGE_POINTS",
            "percentage point": "PERCENTAGE_POINTS", "percentage points": "PERCENTAGE_POINTS", "x": "MULTIPLE"}.get(u)
    return v, unit


def _source_unit(state, cid):
    """(unit, quantity_type, guidance_id) of the claim a trigger is measured against."""
    import catalog as _cat
    if not cid or state is None:
        return None, None, None
    if cid.startswith("GUIDANCE_"):
        g = state.get("guidance.company") or {}
        try:
            m = (g.get("metrics") or [])[int(cid.split("_")[1]) - 1]
        except (ValueError, IndexError):
            return None, None, None
        return m.get("unit"), m.get("quantity_type"), m.get("metric_id")
    d = _cat.BY_ID.get(cid)
    if d is None:
        return None, None, None
    m = state.get(d.path)
    qt = getattr(m, "quantity_type", None)
    return (getattr(m, "display_unit", None) or _STYPE_UNIT.get(d.stype)), qt, None


def _source_range(state, cid):
    """(low, high) canonical value of a trigger's source claim, or (None, None)."""
    import catalog as _cat
    if not cid or state is None:
        return None, None
    if cid.startswith("GUIDANCE_"):
        try:
            m = ((state.get("guidance.company") or {}).get("metrics") or [])[int(cid.split("_")[1]) - 1]
        except (ValueError, IndexError):
            return None, None
        return m.get("low"), m.get("high")
    d = _cat.BY_ID.get(cid)
    if d is None:
        return None, None
    v = state.value(d.path) if hasattr(state, "value") else None
    return (v, None) if isinstance(v, (int, float)) else (None, None)


def trigger_holds(tr, v):
    """Does value v already meet the trigger's condition?"""
    th = tr.get("threshold")
    if v is None or th is None:
        return False
    return float(v) >= float(th) if tr.get("side") == "above" else float(v) <= float(th)


def trigger_current_values(state, source_claim):
    """[(claim id, value)]: every current value of the metric a trigger measures — the source claim and every other
    claim of the same concept, period class and perimeter (e.g. the engine's and the consensus' NTM EPS). Guidance is a
    target, not a current value of the realised metric, and is excluded."""
    import catalog as _cat
    if not source_claim or state is None or str(source_claim).startswith("GUIDANCE_"):
        return []
    src = _cat._claim_metric(state, source_claim)
    if src is None or src.value is None:
        return []
    out = []
    for cid, d in _cat.BY_ID.items():
        if d.stype in (_cat.REL, _cat.PHRASE, _cat.TEXT) or d.deferred:
            continue
        m = _cat._claim_metric(state, cid)
        if m is None or m.value is None:
            continue
        if (m.economic_concept, m.period_class, m.perimeter) == (src.economic_concept, src.period_class, src.perimeter) \
                and (m.per_share == src.per_share):
            out.append((cid, float(m.value)))
    return out


_PRICE_MULTIPLES = frozenset({"PE", "PEG", "PS", "EV_EBITDA", "EV_SALES", "P_FCF", "PRICE_TO_BOOK", "PB"})


def trigger_source_refusal(state, cid):
    """Why a claim cannot be a trigger's source, or None: a figure of another business (a trailing figure that still
    includes a separated one), or a price-derived multiple — it moves with the price itself, so "P/E falls to 17x" can
    read as cheaper shares or as the bear case arriving (all three reviewers flagged run 45cf885f88330ebb's)."""
    if trigger_perimeter_mismatch(state, cid):
        return "its source is a trailing figure that includes a separated business"
    if state is None or not cid:
        return None
    import catalog as _cat
    try:
        m = _cat._claim_metric(state, cid)
    except Exception:  # noqa: BLE001
        return None
    if m is not None and str(m.economic_concept).upper() in _PRICE_MULTIPLES:
        return "its source is a price-derived multiple, which moves with the price itself"
    return None


def trigger_perimeter_mismatch(state, cid):
    """True when a trigger's source claim describes a different business from the one the forecast describes: a
    trailing figure that still includes a separated business moves mechanically as that business rolls out of the
    window, so a threshold on it does not measure the thesis (run 2a5c13806e3f0b7f: a trigger on the pre-separation
    trailing margin, 46.3%, while the continuing company runs at 49.4%)."""
    if state is None or not cid:
        return False
    import catalog as _cat
    try:
        m = _cat._claim_metric(state, cid)
    except Exception:  # noqa: BLE001
        return False
    return bool(m is not None and m.perimeter == "pre-separation"
                and ((state.historical.get("perimeter") or {}).get("event") or {}).get("business"))


_LEGAL_SUFFIX = {"LLC": "LLC", "LLP": "LLP", "LP": "LP", "PLC": "PLC", "UK": "UK", "USA": "USA", "NA": "NA", "AG": "AG",
                 "SA": "SA", "NV": "NV", "BV": "BV", "AB": "AB", "ASA": "ASA", "SE": "SE", "LTD": "Ltd", "INC": "Inc",
                 "CORP": "Corp", "CO": "Co"}


def filer_display_name(name):
    """A 13F filer's name as a reader writes it: legal-form and country tokens in their standard case ("Brave Warrior
    Advisors Llc" → "… LLC", "Egerton Capital Uk Llp" → "… UK LLP"); every other word as the source gives it."""
    words = str(name or "").strip().split()
    return " ".join(_LEGAL_SUFFIX.get(w.strip(".,").upper(), w) if i and w.strip(".,").upper() in _LEGAL_SUFFIX else w
                    for i, w in enumerate(words))


def _typed_trigger(t, state=None):
    """A thesis trigger is a structured judgement: a metric, a comparison and a typed threshold. A trigger that names
    its source claim INHERITS that claim's unit and quantity type (35 against a bps value is 35 bps); a threshold whose
    written unit contradicts its source (35% against a bps value) is not publishable and stays in QA."""
    import quantities as _q
    op = str(t.get("condition_operator") or "").strip()
    raw = str(t.get("threshold") or "").strip()
    if op not in _OPS or not raw:
        return None
    num, written = _written_unit(raw)
    if num is None:
        return None
    src_unit, qt, gid = _source_unit(state, t.get("source_claim"))
    if t.get("source_claim") and src_unit is None:
        return None                                   # names a claim that does not exist
    if trigger_source_refusal(state, t.get("source_claim")):
        return None                                   # a source that cannot measure the thesis (see the reason)
    if src_unit:
        money = ("USD", "USD_MILLION", "USD_BILLION")
        same = (written is None or written == src_unit or {written, src_unit} <= {"PERCENT", "DECIMAL_FRACTION"}
                or (written in money and src_unit in money))
        if not same:
            return None                               # unit detached from its source: 35% against 35 bps
        unit = src_unit
        value = num if (written in ("USD", "USD_PER_SHARE") and src_unit in money + ("USD_PER_SHARE",)) \
            else _q.to_canonical(num, unit)
    else:
        if written is None:
            return None
        unit = written
        value = num if unit in ("USD", "USD_PER_SHARE") else _q.to_canonical(num, unit)
    if src_unit and not _q.magnitude_ok(value, *_source_range(state, t.get("source_claim")))[0]:
        return None                                   # 3.6 against 335–360 bps: the right unit, the wrong number
    cur = trigger_current_values(state, t.get("source_claim"))
    probe = {"side": _OPS[op], "threshold": value}
    sat = [c for c, v in cur if trigger_holds(probe, v)]
    # a guidance source is a RANGE (a target, not a realised value): encode where the threshold sits in it, never a scalar
    rng, rng_state = None, None
    if str(t.get("source_claim") or "").startswith("GUIDANCE_"):
        lo, hi = _source_range(state, t.get("source_claim"))
        if lo is not None:
            hi = lo if hi is None else hi
            rng = {"kind": "GUIDANCE_RANGE", "low": lo, "high": hi, "source_claim_id": t.get("source_claim")}
            ends = [trigger_holds(probe, lo), trigger_holds(probe, hi)]
            if all(ends):
                rng_state = "SATISFIED_ACROSS_RANGE"
            elif not any(ends):
                rng_state = "OUTSIDE_RANGE"
            elif abs(value - lo) < 1e-12 or abs(value - hi) < 1e-12:
                rng_state = "AT_RANGE_ENDPOINT"
            else:
                rng_state = "INSIDE_RANGE"
            if rng_state == "SATISFIED_ACROSS_RANGE":
                sat = sat or [str(t.get("source_claim"))]
    return {"metric_id": t.get("metric_id"), "side": _OPS[op], "threshold": value, "threshold_kind": _UNIT_STYPE.get(unit, "TEXT"),
            "current_value": (dict(cur).get(sat[0]) if (sat and cur) else (cur[0][1] if cur else None)), "current_values": cur,
            "current_range": rng, "range_position": rng_state,
            "currently_satisfied": bool(sat),
            "threshold_unit": unit, "quantity_type": qt, "source_claim_id": t.get("source_claim"), "source_guidance_id": gid,
            "scenario_destination": t.get("scenario_destination"), "polarity": t.get("polarity"), "status": t.get("status")}


RATING_BANDS = ((0.20, "STRONG BUY"), (0.10, "BUY"), (-0.05, "HOLD"), (-0.20, "SELL"))


def rating_for(ret):
    """The one rating table. Applied to the ROUNDED headline, so the label always matches the
    number the reader sees."""
    if ret is None:
        return None
    for lo, lab in RATING_BANDS:
        if ret >= lo - 1e-12:
            return lab
    return "STRONG SELL"


BLEND_KINDS = ("MARKET_PRICING", "INTRINSIC_VALUE", "MACRO", "PRICE_MODEL", "ANALYST_PANEL", "OTHER")
BLEND_KIND_LABEL = {"MARKET_PRICING": "Market pricing", "INTRINSIC_VALUE": "Intrinsic value", "MACRO": "Macro backdrop",
                    "PRICE_MODEL": "Machine-learning price model", "ANALYST_PANEL": "Analyst panel", "OTHER": "Other source"}


def build_blend(state, blend):
    s = state
    b = blend or {}
    audit = b.get("blend_audit") or {}
    rows, excluded = [], []
    for r in audit.get("rows") or []:
        row = {k: r.get(k) for k in ("name", "return", "raw_return", "cap_note", "confidence", "base_weight",
                                     "corr_penalty", "eff_weight", "share", "contribution")}
        # BlendSource: kind and provenance as declared by the source's creator; the label follows the kind
        row["kind"] = r.get("kind") if r.get("kind") in BLEND_KINDS else "OTHER"
        row["underlying_object_ids"] = list(r.get("underlying_object_ids") or [])
        row["label"] = BLEND_KIND_LABEL[row["kind"]] if row["kind"] != "OTHER" else f"Other source ({row.get('name')})"
        row["source_id"] = row.get("name")
        # a source with no weight contributes nothing and its view is not a figure of record: it is
        # listed by name only, so an excluded (e.g. unreliable) model's number never reaches a reader
        if not (_f(row.get("share")) or 0.0) > 0:
            excluded.append({"name": row["label"], "kind": row["kind"], "reason": row.get("cap_note") or "zero weight"})
            continue
        rows.append(row)
    s.blend.sources = rows
    s.blend.excluded_sources = excluded
    raw = _f(b.get("expected_return"))
    head = round(raw, 3) if raw is not None else None       # one precision: 0.1 percentage point
    s.blend.headline_raw = raw
    s.blend.headline_return = metric("headline_return", "HEADLINE_RETURN", head, NTM, accounting_basis=MODEL, unit="ratio",
                                     derivation="clip(Σ source return × normalised share); rounded to 0.1 pt before rating",
                                     quality_status=MODEL_ESTIMATE)
    s.blend.weighted_sum = _f(audit.get("weighted_sum"))
    s.blend.adjustments = [{"name": a.get("name"), "factor": _f(a.get("factor")), "note": a.get("note")}
                           for a in (audit.get("adjustments") or []) if isinstance(a, dict)]
    s.blend.formula = audit.get("formula")
    s.blend.return_kind = "price return (dividends excluded)"
    s.blend.agreement = _f(b.get("confidence"))
    s.recommendation.rating = rating_for(head)
    s.recommendation.bands = [{"from": lo, "label": lab} for lo, lab in RATING_BANDS] + [{"from": None, "label": "STRONG SELL"}]
    s.recommendation.basis = "fixed bands on the headline return"
    # uncertainty: the scenario range, never an unreliable model's error
    bear, bull = s.scenarios.get("bear") or {}, s.scenarios.get("bull") or {}
    lo, hi = bear.get("return"), bull.get("return")
    s.uncertainty.scenario_range = {"low": min(lo, hi) if (lo is not None and hi is not None) else None,
                                    "high": max(lo, hi) if (lo is not None and hi is not None) else None,
                                    "basis": "bear and bull scenario returns from the analyst panel"}
    return state