| """ReconciledForecastState: one 12-month forecast, reconciled once, on explicit accounting bases. |
| |
| The models forecast revenue, an operating margin, net income and ADJUSTED EPS independently. The |
| adjusted EPS is on a non-GAAP definition (the provider's reported-EPS series the model is trained |
| on); revenue, operating income and net income are GAAP. They are tied together here, once: |
| |
| 1. Revenue comes from the revenue model (on the reconciled historical base). |
| 2. Adjusted EPS is the authoritative earnings forecast (it incorporates the sell-side anchor). It |
| is checked against the GAAP earnings capacity of the revenue and margin models only after that |
| capacity has been CONVERTED to the adjusted basis with the adjusted↔GAAP bridge |
| (``derive_gaap_eps_capacity`` → ``derive_adjusted_eps_capacity``). A GAAP ceiling is never |
| applied to adjusted EPS directly. |
| 3. GAAP EPS = adjusted EPS ÷ the bridge (bridge.py: like-for-like ratio of adjusted to GAAP |
| diluted EPS). This is the ONE place the adjusted/GAAP difference is removed. |
| 4. GAAP net income = GAAP EPS × the canonical diluted share count; pre-tax and operating income are |
| rebuilt through the company's historical pre-tax and non-operating bridges; margins are |
| definitions (income ÷ revenue). |
| 5. The independent operating-income and net-income models are kept and their distance from the |
| rebuilt values is reported as forecast coherence; it lowers confidence and is never hidden. |
| 6. The bridge's uncertainty band is carried into GAAP EPS and net income as a range. |
| |
| Where a bridge cannot be measured (losses, missing lines, a financial issuer) the GAAP lines are |
| not derived from adjusted EPS; the state says so. |
| """ |
| from __future__ import annotations |
|
|
| import definitions as _defs |
| from state import (ADJUSTED, DERIVED, GAAP, IFRS, MODEL_ESTIMATE, NTM, TTM, UNAVAILABLE, metric) |
|
|
| CONSISTENT_GAP = 0.15 |
| CONFLICTING_GAP = 0.40 |
| CAPACITY_HEADROOM = 1.05 |
|
|
|
|
| def _f(v): |
| try: |
| if v is None: |
| return None |
| x = float(v) |
| return None if x != x else x |
| except (TypeError, ValueError): |
| return None |
|
|
|
|
| def _gap(a, b): |
| if a is None or b is None: |
| return None |
| den = max(abs(a), abs(b), 1e-9) |
| return abs(a - b) / den |
|
|
|
|
| |
| def derive_gaap_eps_capacity(revenue, operating_margin, non_operating, net_income_per_pretax, diluted_shares, |
| headroom=CAPACITY_HEADROOM): |
| """GAAP inputs → GAAP EPS capacity: the most GAAP EPS the revenue and operating-margin forecasts can carry. |
| |
| GAAP revenue × GAAP operating margin + historical non-operating income → GAAP pre-tax income; |
| × historical GAAP net income per pre-tax dollar → GAAP net income; ÷ diluted shares → GAAP EPS.""" |
| if not all(x is not None for x in (revenue, operating_margin, non_operating, net_income_per_pretax)) or not diluted_shares: |
| return None |
| pt = revenue * operating_margin + non_operating |
| ni = pt * net_income_per_pretax |
| return {"value": ni * headroom / diluted_shares, "basis": "GAAP", "input_basis": "GAAP", |
| "inputs": {"revenue": revenue, "operating_margin": operating_margin, "non_operating": non_operating, |
| "net_income_per_pretax": net_income_per_pretax, "diluted_shares": diluted_shares, "headroom": headroom}} |
|
|
|
|
| def derive_adjusted_eps_capacity(gaap_capacity, bridge): |
| """GAAP EPS capacity → ADJUSTED EPS capacity, through the bridge (upper bound of its uncertainty band, so |
| the bridge's own uncertainty never forces a cut). None when there is no bridge: an adjusted figure is |
| then not checked against a GAAP identity at all.""" |
| if not gaap_capacity or not bridge or not bridge.get("available"): |
| return None |
| hi = (bridge.get("uncertainty") or {}).get("high") or bridge.get("selected_ratio") |
| return {"value": gaap_capacity["value"] * hi, "basis": "ADJUSTED", "input_basis": "GAAP", |
| "definition": bridge.get("definition_id"), "conversion": f"× {hi:.4f} (bridge upper bound)"} |
|
|
|
|
| def reconcile_forecast(pf, hist, *, shares, bridge=None, diluted_shares=None, is_financial=False, |
| accounting_standard="US GAAP", per_share_basis="ORDINARY_SHARE", period_end=None, |
| perimeter="consolidated", base_perimeter="consolidated"): |
| """Build the ReconciledForecastState (a dict of Metrics plus audit records) from the raw model outputs, |
| the reconciled history, the adjusted↔GAAP bridge and the canonical diluted share count. |
| |
| ``pf`` legacy forecast dict (raw model outputs; read here and nowhere else) |
| ``hist`` dict of historical Metrics: revenue_ttm, operating_income_ttm, pretax_income_ttm, |
| net_income_ttm, eps_adjusted_ttm |
| """ |
| pf = pf or {} |
| basis = IFRS if "IFRS" in str(accounting_standard or "") else GAAP |
| raw = pf.get("raw_results") or {} |
| bridge = bridge or {} |
|
|
| def hv(k): |
| m = hist.get(k) |
| return m.value if m is not None else None |
|
|
| R_h, OI_h, PT_h, NI_h = hv("revenue_ttm"), hv("operating_income_ttm"), hv("pretax_income_ttm"), hv("net_income_ttm") |
| EPSadj_h = hv("eps_adjusted_ttm") |
| D = _f(diluted_shares) or _f(shares) |
| R_f = _f(pf.get("revenue_annual")) |
| EPS_raw = _f(pf.get("eps_annual")) |
| OI_ml = _f((raw.get("operating_income") or {}).get("annual_sum")) |
| NI_ml = _f((raw.get("net_income") or {}).get("annual_sum")) |
| R_ml_raw = _f((raw.get("revenue") or {}).get("annual_sum")) |
| |
| |
| om_ml = (OI_ml / R_ml_raw) if (OI_ml is not None and R_ml_raw) else _f(pf.get("op_margin_predicted")) |
| OI_ml_on_R = (om_ml * R_f) if (om_ml is not None and R_f) else None |
| nm_ml = (NI_ml / R_ml_raw) if (NI_ml is not None and R_ml_raw) else _f(pf.get("net_margin_predicted")) |
| NI_ml_on_R = (nm_ml * R_f) if (nm_ml is not None and R_f) else None |
|
|
| notes, adjustments = [], [] |
| same_base = (perimeter == base_perimeter) or {perimeter, base_perimeter} <= {"consolidated", "continuing company"} |
| if not same_base: |
| notes.append(f"the trailing statements describe the {base_perimeter} business and the forecast the " |
| f"{perimeter}; historical bridges, growth rates and model coherence are not carried across") |
| ni_per_pt = (NI_h / PT_h) if (NI_h and PT_h and PT_h > 0 and NI_h > 0 and 0.3 <= NI_h / PT_h <= 1.2) else None |
| nonop = (PT_h - OI_h) if (PT_h is not None and OI_h is not None) else None |
| if not same_base: |
| ni_per_pt = nonop = None |
| OI_ml_on_R = NI_ml_on_R = None |
| br_ok = bool(bridge.get("available")) and same_base |
| r = bridge.get("selected_ratio") if br_ok else None |
| unc = bridge.get("uncertainty") if br_ok else None |
|
|
| |
| EPS_adj = EPS_raw |
| gaap_cap = derive_gaap_eps_capacity(R_f, om_ml, nonop, ni_per_pt, D) if (not is_financial and same_base) else None |
| adj_cap = derive_adjusted_eps_capacity(gaap_cap, bridge if br_ok else None) |
| if EPS_adj is not None and adj_cap is not None and EPS_adj > adj_cap["value"]: |
| adjustments.append({"kind": "adjusted EPS limited to the adjusted-basis earnings capacity", |
| "from": EPS_adj, "to": adj_cap["value"], "basis": "ADJUSTED", |
| "capacity_gaap_eps": gaap_cap["value"], "conversion": adj_cap["conversion"], |
| "reason": "the revenue and operating-margin forecasts cannot carry more earnings, " |
| "even at the top of the bridge's range"}) |
| EPS_adj = adj_cap["value"] |
| if EPS_adj is not None and gaap_cap is not None and adj_cap is None: |
| notes.append("no adjusted↔GAAP bridge: the adjusted EPS forecast is not checked against GAAP capacity") |
|
|
| |
| EPS_gaap = (EPS_adj / r) if (EPS_adj is not None and r) else None |
| NI_f = (EPS_gaap * D) if (EPS_gaap is not None and D) else None |
| OI_f = PT_f = None |
| rebuilt = False |
| if is_financial: |
| notes.append("financial issuer: operating income is not a meaningful line; net income only") |
| elif NI_f is not None and ni_per_pt and nonop is not None and OI_h and OI_h > 0: |
| PT_f = NI_f / ni_per_pt |
| OI_f = PT_f - nonop |
| if OI_f <= 0 < NI_f: |
| OI_f = NI_f * (OI_h / NI_h) |
| PT_f = OI_f + nonop |
| notes.append("non-operating bridge exceeded rebuilt pre-tax income; used the historical operating-to-net proportion") |
| rebuilt = True |
| elif NI_f is not None and OI_h and NI_h and OI_h > 0 and NI_h > 0 and same_base: |
| OI_f = NI_f * (OI_h / NI_h) |
| rebuilt = True |
| notes.append("pre-tax income unavailable; operating income rebuilt from the historical operating-to-net proportion") |
| elif NI_f is None: |
| |
| NI_f, OI_f = NI_ml_on_R, OI_ml_on_R |
| EPS_gaap = (NI_f / D) if (NI_f is not None and D) else None |
| notes.append("adjusted and GAAP forecasts are not reconciled (no measurable bridge); GAAP lines are the " |
| "independent models'") |
| else: |
| OI_f = OI_ml_on_R |
| notes.append("historical bridge not measurable (losses or missing lines); operating income is the independent model's") |
|
|
| op_margin = (OI_f / R_f) if (OI_f is not None and R_f and not is_financial) else None |
| net_margin = (NI_f / R_f) if (NI_f is not None and R_f) else None |
|
|
| |
| g_oi = _gap(OI_f, OI_ml_on_R) if (rebuilt and not is_financial) else None |
| g_ni = _gap(NI_f, NI_ml_on_R) if rebuilt else None |
| gaps = [g for g in (g_oi, g_ni) if g is not None] |
| if not gaps: |
| coherence = "NOT_VERIFIABLE" |
| else: |
| worst = max(gaps) |
| coherence = "CONSISTENT" if worst < CONSISTENT_GAP else ("DIVERGENT" if worst < CONFLICTING_GAP else "CONFLICTING") |
| conf_raw = _f(pf.get("eps_confidence")) or _f(pf.get("avg_confidence")) |
| conf_mult = {"CONSISTENT": 1.0, "DIVERGENT": 0.85, "CONFLICTING": 0.70, "NOT_VERIFIABLE": 0.90}[coherence] |
|
|
| |
| uncertainty = None |
| if EPS_adj is not None and unc and unc.get("low") and unc.get("high"): |
| g_lo, g_hi = EPS_adj / unc["high"], EPS_adj / unc["low"] |
| uncertainty = {"source": "adjusted↔GAAP bridge range", "eps_gaap_low": g_lo, "eps_gaap_high": g_hi, |
| "net_income_low": (g_lo * D) if D else None, "net_income_high": (g_hi * D) if D else None} |
|
|
| eps_growth = (EPS_adj / EPSadj_h - 1.0) if (same_base and EPS_adj is not None and EPSadj_h and EPSadj_h > 0) else None |
| rev_growth = (R_f / R_h - 1.0) if (same_base and R_f is not None and R_h and R_h > 0) else None |
|
|
| p_s, p_e = _defs.period_bounds_after(period_end) |
| kw = dict(period_type=NTM, accounting_basis=basis, quality_status=MODEL_ESTIMATE, source="Crucible 12-month model", |
| period=period_end, perimeter=perimeter, period_start=p_s, period_end=p_e, forecast_horizon_days=365) |
| edef = dict(adjustment_definition_id=bridge.get("definition_id") or _defs.PROVIDER_REPORTED_EPS, |
| adjustment_provider=f"Crucible model (on {bridge.get('adjusted_provider') or 'Alpha Vantage'} reportedEPS series)", |
| adjustment_version=bridge.get("definition_version") or _defs.PROVIDER_REPORTED_EPS) |
| fs = { |
| "revenue": metric("forecast_revenue", "REVENUE", R_f, unit="USD", derivation="revenue model on the reconciled quarterly series", **kw), |
| "operating_income": metric("forecast_operating_income", "OPERATING_INCOME", OI_f, unit="USD", |
| derivation=("GAAP net income rebuilt through the historical pre-tax and non-operating bridges" |
| if rebuilt else "independent operating-income model"), **kw), |
| "pretax_income": metric("forecast_pretax_income", "PRETAX_INCOME", PT_f, unit="USD", |
| derivation="GAAP net income ÷ historical net-income-per-pre-tax-dollar", **kw), |
| "net_income": metric("forecast_net_income", "NET_INCOME", NI_f, unit="USD", |
| derivation=("GAAP EPS × diluted shares (GAAP EPS = adjusted EPS ÷ bridge)" if (rebuilt or r) |
| else "independent net-income model"), **kw), |
| "operating_margin": metric("forecast_operating_margin", "OPERATING_MARGIN", op_margin, unit="ratio", |
| derivation="forecast operating income ÷ forecast revenue", **kw), |
| "net_margin": metric("forecast_net_margin", "NET_MARGIN", net_margin, unit="ratio", |
| derivation="forecast GAAP net income ÷ forecast revenue", **kw), |
| "diluted_shares": metric("forecast_diluted_shares", "DILUTED_SHARES", D, unit="count", period_type=NTM, |
| accounting_basis=basis, source=str(pf.get("_diluted_shares_source") or "canonical diluted share count"), |
| quality_status=DERIVED, perimeter=perimeter), |
| "eps": metric("forecast_eps", "EPS", EPS_adj, unit="USD", period_type=NTM, accounting_basis=ADJUSTED, |
| per_share=True, security_basis=per_share_basis, quality_status=MODEL_ESTIMATE, perimeter=perimeter, |
| source="Crucible 12-month earnings model", period=period_end, period_start=p_s, period_end=p_e, |
| forecast_horizon_days=365, derivation="statistical adjusted-EPS forecast blended with the sell-side anchor", |
| **edef), |
| "eps_gaap": metric("forecast_eps_gaap", "EPS", EPS_gaap, unit="USD", period_type=NTM, accounting_basis=basis, |
| per_share=True, security_basis=per_share_basis, perimeter=perimeter, period=period_end, |
| period_start=p_s, period_end=p_e, forecast_horizon_days=365, |
| quality_status=MODEL_ESTIMATE if EPS_gaap is not None else UNAVAILABLE, |
| derivation=(f"adjusted EPS ÷ bridge {r:.4f} ({bridge.get('selected_method')})" if r |
| else "independent net-income model ÷ diluted shares")), |
| "eps_growth": metric("forecast_eps_growth", "EPS_GROWTH", eps_growth, unit="ratio", period_type=NTM, |
| accounting_basis=ADJUSTED, quality_status=DERIVED if eps_growth is not None else UNAVAILABLE, |
| derivation="forecast adjusted EPS ÷ trailing adjusted EPS − 1 (same definition)", **edef), |
| "revenue_growth": metric("forecast_revenue_growth", "REVENUE_GROWTH", rev_growth, unit="ratio", period_type=NTM, |
| accounting_basis=basis, quality_status=DERIVED if rev_growth is not None else UNAVAILABLE, |
| derivation="forecast revenue ÷ reconciled trailing revenue − 1"), |
| "coherence": {"status": coherence, "operating_income_gap": g_oi, "net_income_gap": g_ni, |
| "independent_operating_income": OI_ml_on_R, "independent_net_income": NI_ml_on_R, |
| "independent_operating_margin": (om_ml if same_base else None), "rebuilt": rebuilt, |
| "bridge": {"adjusted_to_gaap": r, "net_income_per_pretax": ni_per_pt, "non_operating_ttm": nonop, |
| "definition": bridge.get("definition_id"), "method": bridge.get("selected_method")}, |
| "notes": notes, "perimeter_change": not same_base}, |
| "reconciliation": {"order": ["revenue model", "adjusted EPS model (after its consensus anchor)", |
| "capacity check on the adjusted basis (GAAP capacity × bridge)", |
| "GAAP EPS = adjusted EPS ÷ bridge (once)", "GAAP net income = GAAP EPS × diluted shares", |
| "pre-tax and operating income through historical bridges", "margins = income ÷ revenue"], |
| "adjustments": adjustments, "gaap_capacity": gaap_cap, "adjusted_capacity": adj_cap, |
| "raw_model_outputs": {"revenue": R_f, "eps_adjusted": EPS_raw, "operating_margin": om_ml, |
| "net_margin": nm_ml, "operating_income": OI_ml_on_R, "net_income": NI_ml_on_R}, |
| "bridge_applied_once": True}, |
| "uncertainty": uncertainty, |
| "confidence": (conf_raw * conf_mult) if conf_raw is not None else None, |
| "confidence_basis": f"model confidence × {conf_mult:.2f} for forecast coherence {coherence}", |
| "anchor": {"flag": pf.get("eps_anchor_flag"), "cyclical_adjustment": pf.get("eps_cyclical_adjustment"), |
| "ml_raw": _f(pf.get("eps_ml_raw")), "analyst_anchor": _f(pf.get("eps_analyst_adjusted")), |
| "ml_weight": _f(pf.get("eps_ml_weight")), "anchor_weight": _f(pf.get("eps_anchor_weight"))}, |
| } |
| if is_financial: |
| fs["operating_income"] = fs["operating_income"].with_value(None, quality_status=UNAVAILABLE) |
| fs["operating_margin"] = fs["operating_margin"].with_value(None, quality_status=UNAVAILABLE) |
| return fs |
|
|
|
|
| def identity_check(fs, hist=None, tol=0.005): |
| """The accounting identities a reconciled forecast must satisfy (evaluated, never assumed). |
| |
| Returns a list of (name, ok, detail). Only definitional relationships are identities: margins |
| are income ÷ revenue, EPS × shares reproduces net income on the stated bridge, operating income |
| cannot exceed revenue, and revenue is positive. 'Net income ≤ operating income' is NOT an |
| identity (non-operating gains exist) and is not tested here. |
| """ |
| out = [] |
| v = lambda k: (fs.get(k).value if fs.get(k) is not None else None) |
| R, OI, NI, EPS, S = v("revenue"), v("operating_income"), v("net_income"), v("eps"), v("diluted_shares") |
| om, nm = v("operating_margin"), v("net_margin") |
| br = ((fs.get("coherence") or {}).get("bridge") or {}) |
| eg = (fs.get("eps_gaap").value if fs.get("eps_gaap") is not None else None) |
| if R is None: |
| out.append(("revenue present", False, "forecast revenue is missing")) |
| else: |
| out.append(("revenue positive", R > 0, f"forecast revenue {R:,.0f}")) |
| if om is not None and OI is not None and R: |
| out.append(("operating margin = operating income ÷ revenue", abs(om - OI / R) <= tol, |
| f"{om:.4f} vs {OI / R:.4f}")) |
| if nm is not None and NI is not None and R: |
| out.append(("net margin = net income ÷ revenue", abs(nm - NI / R) <= tol, f"{nm:.4f} vs {NI / R:.4f}")) |
| if eg is not None and S and NI is not None: |
| implied = eg * S |
| out.append(("net income = GAAP EPS × diluted shares", abs(implied - NI) <= max(abs(NI), 1.0) * tol, |
| f"{implied:,.0f} vs {NI:,.0f}")) |
| if EPS is not None and eg is not None and br.get("adjusted_to_gaap"): |
| out.append(("GAAP EPS = adjusted EPS ÷ bridge", abs(EPS / br["adjusted_to_gaap"] - eg) <= max(abs(eg), 1e-9) * tol, |
| f"{EPS:.4f} ÷ {br['adjusted_to_gaap']:.4f} vs {eg:.4f}")) |
| if OI is not None and R: |
| out.append(("operating income ≤ revenue", OI <= R * (1 + tol), f"{OI:,.0f} vs {R:,.0f}")) |
| if om is not None: |
| out.append(("operating margin within ±100%", -1.0 <= om <= 1.0, f"{om:.4f}")) |
| return out |
|
|
