"""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 # non-operating income a margin model does not see 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 # ── typed capacity functions: every identity declares its input and output basis ───────── 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")) # the operating-income model was trained on the same series as the raw revenue model; express # it as a margin and apply it to the published revenue so both use one revenue figure 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 # trained on the old perimeter: not comparable 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 # 2. adjusted EPS vs the GAAP capacity CONVERTED to the adjusted basis (never a GAAP ceiling on adjusted EPS) 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") # 3.–4. the ONE conversion: GAAP EPS, then GAAP income lines 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: # no bridge: GAAP lines come from the GAAP models; adjusted EPS stays on its own basis, unreconciled 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 # 5. coherence against the independent models (only meaningful when the lines were rebuilt) 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] # 6. the bridge's uncertainty, carried into GAAP EPS and net income 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