| """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() |
|
|
|
|
| |
| |
| |
| 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") |
| |
| |
| |
| |
| 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") |
|
|
| |
| 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") |
|
|
| |
| _canon = d.get("segments_canonical") or {"status": "UNAVAILABLE", "reason": "no segment data"} |
| s.segments.canonical = _canon |
| s.segments.status = _canon.get("status") |
| |
| 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: |
| |
| 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") |
|
|
| |
| 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)") |
|
|
| |
| |
| |
| _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) |
| |
| 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) |
| |
| _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"): |
| |
| 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") |
|
|
| |
| 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}) |
| |
| _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 []) |
|
|
| |
| 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)} |
| |
| 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) |
|
|
|
|
| |
| |
| |
| 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 {} |
| |
| 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)] |
| |
| 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] |
| |
| 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 {} |
| |
| 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") |
| |
| |
| 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 |
| |
| _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 {}) |
| 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")} |
| |
| 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"), |
| |
| "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 [])] |
|
|
| |
| _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) |
| |
| _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) |
| |
| _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) |
| |
| s.valuation.breakeven_growth_binding = {"start_revenue_metric_id": "story_base_revenue", |
| "target_revenue_metric_id": "breakeven_revenue", "years_metric_id": "story_years"} |
| |
| _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)) |
|
|
| |
| 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") |
| |
| 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 |
|
|
| 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) |
| |
| |
| 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") |
| |
| _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): |
| |
| |
| |
| 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" |
| |
| 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 |
| |
| |
| _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()} |
|
|
| |
| 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} |
|
|
| |
| 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")} |
|
|
| |
| 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")} |
| |
| |
| 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()] |
| |
| |
| 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) |
| |
| 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 [] |
| |
| 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")) |
| |
| |
| 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")) |
|
|
| |
| 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")) |
| |
| |
| _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) |
| |
| import evidence as _evd |
| import relations as _rel |
| s.relations.state = _rel.build(s) |
| s.evidence.state = _evd.build(s) |
| return state |
|
|
|
|
| |
| |
| |
| 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")) |
| |
| 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") |
| |
| s.scenarios.pm_evidence_state = db.get("pm_evidence_state") |
| typed = [(t, _typed_trigger(t, s)) for t in (db.get("thesis_triggers") or [])] |
| |
| 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: |
| 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: |
| 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 |
| if trigger_source_refusal(state, t.get("source_claim")): |
| return None |
| 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 = 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 |
| 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)] |
| |
| 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")} |
| |
| 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") |
| |
| |
| 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 |
| 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" |
| |
| 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 |
|
|
