| """The adjusted↔GAAP earnings bridge, measured like for like. |
| |
| An adjusted EPS forecast becomes GAAP net income only through a bridge: how much larger the adjusted |
| figure is than GAAP diluted EPS for the same period and the same share basis. The earlier bridge was |
| one trailing point that divided adjusted EPS × *today's* share count by GAAP net income, mixing a |
| per-share series computed on historical weighted shares with the current count, and it was carried |
| forward as if it could never change. |
| |
| Here the bridge is a series of like-for-like observations: |
| |
| ratio(period) = Σ adjusted EPS over the period ÷ GAAP diluted EPS for the same period |
| |
| using fiscal years (four provider quarters summed against the filed annual diluted EPS) and the |
| trailing twelve months. Both sides are per diluted share for the same period, so share-count |
| changes cancel. The bridge records the median, range, latest value and trend, and flags a probable |
| change in the adjustment methodology when the latest observation departs materially from its own |
| history. In that case the older observations are not carried forward: only the latest is used, with |
| a wider uncertainty band. |
| """ |
| from __future__ import annotations |
|
|
| import datetime as _dt |
| import statistics |
|
|
| import definitions as D |
|
|
| RATIO_BOUNDS = (0.5, 3.0) |
| CHANGE_THRESHOLD = 0.20 |
| MIN_UNCERTAINTY = 0.05 |
|
|
|
|
| def _d(x): |
| try: |
| return _dt.date.fromisoformat(str(x)[:10]) |
| except (TypeError, ValueError): |
| return None |
|
|
|
|
| def _f(v): |
| try: |
| x = float(v) |
| return None if x != x else x |
| except (TypeError, ValueError): |
| return None |
|
|
|
|
| def observations(quarterly_adjusted, gaap_eps_rows, ttm_adjusted=None, ttm_gaap=None, ttm_end=None): |
| """Like-for-like (adjusted, GAAP, ratio) observations, oldest first. |
| |
| ``quarterly_adjusted``: [{fiscalDateEnding, reportedEPS}] (provider quarters). |
| ``gaap_eps_rows``: [{start, end, value, days}] SEC diluted EPS rows (annual rows are used). |
| """ |
| q = sorted(((_d(x.get("fiscalDateEnding")), _f(x.get("reportedEPS"))) for x in quarterly_adjusted or []), |
| key=lambda t: (t[0] or _dt.date.min)) |
| q = [(e, v) for e, v in q if e and v is not None] |
| obs = [] |
| for r in gaap_eps_rows or []: |
| if not (350 <= int(r.get("days") or 0) <= 380): |
| continue |
| s, e, g = _d(r.get("start")), _d(r.get("end")), _f(r.get("value")) |
| if not (s and e and g) or g <= 0: |
| continue |
| inside = [v for (qe, v) in q if s < qe <= e + _dt.timedelta(days=10)] |
| if len(inside) != 4: |
| continue |
| adj = sum(inside) |
| ratio = adj / g |
| if RATIO_BOUNDS[0] <= ratio <= RATIO_BOUNDS[1]: |
| obs.append({"period": f"FY ending {e.isoformat()}", "start": s.isoformat(), "end": e.isoformat(), |
| "adjusted_eps": round(adj, 4), "gaap_eps": g, "ratio": ratio, "kind": "fiscal year"}) |
| ta, tg = _f(ttm_adjusted), _f(ttm_gaap) |
| if ta and tg and tg > 0 and RATIO_BOUNDS[0] <= ta / tg <= RATIO_BOUNDS[1]: |
| te = str(ttm_end or "")[:10] |
| if not any(o["end"] == te for o in obs): |
| obs.append({"period": f"TTM ending {te}", "start": None, "end": te, "adjusted_eps": ta, "gaap_eps": tg, |
| "ratio": ta / tg, "kind": "trailing twelve months"}) |
| obs.sort(key=lambda o: o["end"]) |
| return obs |
|
|
|
|
| def build(quarterly_adjusted, gaap_eps_rows, ttm_adjusted=None, ttm_gaap=None, ttm_end=None, |
| adjusted_definition=D.PROVIDER_REPORTED_EPS, adjusted_provider="Alpha Vantage"): |
| """AdjustmentBridge dict, or an unavailable record with the reason.""" |
| obs = observations(quarterly_adjusted, gaap_eps_rows, ttm_adjusted, ttm_gaap, ttm_end) |
| base = {"definition_id": adjusted_definition, "adjusted_provider": adjusted_provider, |
| "gaap_definition": "GAAP diluted EPS (SEC XBRL)", "numerator": "adjusted EPS (provider quarters summed)", |
| "denominator": "GAAP diluted EPS, same period", "share_basis": "per diluted share, same period (like for like)", |
| "historical_ratios": [{k: (round(v, 5) if isinstance(v, float) else v) for k, v in o.items()} for o in obs]} |
| if not obs: |
| return dict(base, available=False, selected_ratio=None, selected_method="unavailable", |
| reason="no period with positive GAAP EPS and a matching adjusted EPS series", |
| uncertainty=None, methodology_change=False) |
| ratios = [o["ratio"] for o in obs] |
| latest = ratios[-1] |
| prior = ratios[:-1][-4:] |
| change = False |
| if len(prior) >= 2: |
| med_prior = statistics.median(prior) |
| change = abs(latest / med_prior - 1.0) > CHANGE_THRESHOLD |
| if change: |
| sel, method = latest, (f"latest observation only: it departs {abs(latest / statistics.median(prior) - 1):.0%} from the " |
| "prior median: a structural break (the adjustment definition, or the size of the items it excludes, " |
| "changed), so older ratios are not carried forward") |
| lo, hi = latest * (1 - max(0.10, MIN_UNCERTAINTY)), latest * (1 + max(0.10, MIN_UNCERTAINTY)) |
| valid_from = obs[-1]["start"] or obs[-1]["end"] |
| elif len(ratios) >= 3: |
| recent = ratios[-3:] |
| sel, method = statistics.median(recent), "median of the last three like-for-like observations" |
| lo, hi = min(recent), max(recent) |
| valid_from = obs[-3]["start"] or obs[-3]["end"] |
| else: |
| sel = latest |
| method = "latest observation (fewer than three like-for-like observations)" |
| lo, hi = min(ratios), max(ratios) |
| valid_from = obs[0]["start"] or obs[0]["end"] |
| lo, hi = min(lo, sel * (1 - MIN_UNCERTAINTY)), max(hi, sel * (1 + MIN_UNCERTAINTY)) |
| trend = "flat" |
| if len(ratios) >= 3: |
| d = ratios[-1] - ratios[-3] |
| trend = "rising" if d > 0.03 else "falling" if d < -0.03 else "flat" |
| return dict(base, available=True, selected_ratio=sel, selected_method=method, |
| median=statistics.median(ratios), range=[min(ratios), max(ratios)], latest=latest, trend=trend, |
| uncertainty={"low": lo, "high": hi}, methodology_change=change, |
| valid_from=valid_from, valid_to=obs[-1]["end"], n_observations=len(obs), |
| definition_version=(f"{adjusted_definition}@{valid_from}" if change else adjusted_definition)) |
|
|
|
|
| def latest_diluted_shares(series): |
| """(value, source) — the most recent quarterly weighted-average diluted share count from the filings, |
| else the trailing annual one, else None.""" |
| rows = sorted(series or [], key=lambda r: (r.get("end") or "", -(r.get("days") or 0))) |
| q = [r for r in rows if 80 <= int(r.get("days") or 0) <= 100] |
| if q: |
| return q[-1]["value"], f"SEC weighted-average diluted shares, quarter ending {q[-1]['end']}" |
| a = [r for r in rows if 350 <= int(r.get("days") or 0) <= 380] |
| if a: |
| return a[-1]["value"], f"SEC weighted-average diluted shares, year ending {a[-1]['end']}" |
| return None, None |
|
|
