| """Canonical inputs for the multiples-based pricing composite. |
| |
| The pricing composite (predict_yearly.derive_fair_value) historically read the provider's company |
| profile directly (ReturnOnEquityTTM, ProfitMargin, PriceToSalesRatioTTM, PERatio, AnalystTargetPrice, |
| MarketCapitalization, Beta) and the raw forecast dict (eps_annual, rev_per_share, shares_outstanding). |
| Those figures are computed by the provider on its own share count, period and basis, and the raw |
| forecast is the one the reconciliation stage exists to correct. |
| |
| ``canonical_view`` returns the same shapes the composite already consumes, built only from the |
| validated AnalysisState and the ReconciledForecastState, plus a lineage record for every field it |
| supplies. A field with no canonical source is set to None (the method that needs it drops out) rather |
| than falling back to the provider figure. Descriptors that are not numbers (Sector, Industry, name) |
| pass through unchanged. |
| """ |
| from __future__ import annotations |
|
|
|
|
| def _v(m): |
| return getattr(m, "value", m) if m is not None else None |
|
|
|
|
| def _f(x): |
| try: |
| x = float(x) |
| return None if x != x else x |
| except (TypeError, ValueError): |
| return None |
|
|
|
|
| |
| OVERVIEW_FIELDS = { |
| "ReturnOnEquityTTM": "historical.roe_ttm (reconciled TTM net income ÷ latest book equity)", |
| "ProfitMargin": "historical.net_margin_ttm (reconciled TTM net income ÷ reconciled TTM revenue)", |
| "PriceToSalesRatioTTM": "market.price × market.shares_outstanding ÷ historical.revenue_ttm", |
| "PERatio": "market.price ÷ historical.eps_gaap_ttm (trailing GAAP diluted EPS)", |
| "AnalystTargetPrice": "consensus.target_price", |
| "MarketCapitalization": "market.price × market.shares_outstanding", |
| "Beta": "market.beta", |
| "ReturnOnAssetsTTM": "historical.roa_ttm (reconciled TTM net income ÷ latest total assets)", |
| "DividendYield": "market.dividend_yield (dividends paid TTM ÷ market cap)", |
| } |
| |
| |
| |
| UNSOURCED = ("QuarterlyRevenueGrowthYOY", "QuarterlyEarningsGrowthYOY", "PriceToCashFlowRatio", "EPS", "DilutedEPSTTM", |
| "ForwardPE", "TrailingPE", "PEGRatio", "PriceToBookRatio", "EVToRevenue", "EVToEBITDA") |
|
|
|
|
| def canonical_view(state, overview, pred_fund, *, reconciled=None): |
| """(overview_view, forecast_view, lineage) for the pricing composite.""" |
| ov = dict(overview or {}) |
| pf = dict(pred_fund or {}) |
| lin = {"overview": {}, "forecast": {}, "unsourced": []} |
| if state is None: |
| return ov, pf, dict(lin, status="legacy: no AnalysisState supplied") |
|
|
| price = _f(state.value("market.price")) |
| sh = _f(state.value("market.shares_outstanding")) |
| rev = _f(state.value("historical.revenue_ttm")) |
| eps_g = _f(state.value("historical.eps_gaap_ttm")) |
| mcap = (price * sh) if (price and sh) else None |
| vals = { |
| "ReturnOnEquityTTM": _f(state.value("historical.roe_ttm")), |
| "ProfitMargin": _f(state.value("historical.net_margin_ttm")), |
| "PriceToSalesRatioTTM": (mcap / rev) if (mcap and rev and rev > 0) else None, |
| "PERatio": (price / eps_g) if (price and eps_g and eps_g > 0) else None, |
| "AnalystTargetPrice": _f(state.value("consensus.target_price")), |
| "MarketCapitalization": mcap, |
| "Beta": _f(state.get("market.beta")), |
| "ReturnOnAssetsTTM": _f(state.value("historical.roa_ttm")), |
| "DividendYield": _f(state.value("market.dividend_yield")), |
| } |
| for k, v in vals.items(): |
| ov[k] = v |
| lin["overview"][k] = {"value": v, "source": OVERVIEW_FIELDS[k], "provider_value": _f((overview or {}).get(k))} |
| for k in UNSOURCED: |
| if k in ov: |
| lin["unsourced"].append(k) |
| ov[k] = None |
|
|
| fs = reconciled or {} |
| eps = _v(fs.get("eps")) |
| R = _v(fs.get("revenue")) |
| pf_map = { |
| "eps_annual": (eps, "ReconciledForecastState.eps (adjusted, NTM)"), |
| "eps_growth": (_v(fs.get("eps_growth")), "ReconciledForecastState.eps_growth (same definition)"), |
| "rev_growth": (_v(fs.get("revenue_growth")), "ReconciledForecastState.revenue_growth"), |
| "rev_per_share": ((R / sh) if (R and sh) else None, |
| "ReconciledForecastState.revenue ÷ market.shares_outstanding (the share count in market cap)"), |
| "shares_outstanding": (sh, "market.shares_outstanding"), |
| "eps_ttm_current": (_f(state.value("historical.eps_adjusted_ttm")), |
| "historical.eps_adjusted_ttm (same definition as the forecast EPS)"), |
| } |
| for k, (v, src) in pf_map.items(): |
| if not fs and k not in ("shares_outstanding", "eps_ttm_current"): |
| continue |
| lin["forecast"][k] = {"value": v, "source": src if v is not None else f"unavailable ({src})", |
| "raw_model_value": _f((pred_fund or {}).get(k))} |
| if v is None: |
| |
| |
| pf.pop(k, None) |
| else: |
| pf[k] = v |
| lin["status"] = "canonical" if fs else "canonical inputs; forecast not reconciled" |
| return ov, pf, lin |
|
|
