Download engine/definitions.py from s-ttp/crucible: direct link, hf CLI and curl.
- Browser
- Download file 7.63 kB
-
https://huggingface.co/spaces/s-ttp/crucible/resolve/main/engine/definitions.py
- Command line
-
hf download hf://spaces/s-ttp/crucible/engine/definitions.py
-
curl -L -o definitions.py https://huggingface.co/spaces/s-ttp/crucible/resolve/main/engine/definitions.py
7.63 kB
| """Non-GAAP definitions and fiscal periods: what makes two figures comparable. | |
| ``accounting_basis = "ADJUSTED"`` says only that a figure is not GAAP. It does not say whose | |
| adjustment it is or what it excludes, so it cannot establish that two adjusted figures describe | |
| the same quantity. Every adjusted metric therefore carries an ``adjustment_definition_id`` (plus | |
| provider and version), and two adjusted metrics compare only when: | |
| * their definitions are identical, or | |
| * an explicit reconciliation between the two definitions is registered below, with the evidence | |
| that makes them comparable. | |
| Forward figures also carry their actual fiscal period (``period_start`` / ``period_end``). "Next | |
| twelve months" and "fiscal year 1" are both forward, but they are the same period only when they | |
| overlap; otherwise the comparison needs an explicit conversion. | |
| """ | |
| from __future__ import annotations | |
| import datetime as _dt | |
| # ── adjustment definitions ──────────────────────────────────────────────────────────── | |
| PROVIDER_REPORTED_EPS = "PROVIDER_REPORTED_EPS" # a data provider's "reported" (Street) EPS series | |
| SELL_SIDE_CONSENSUS_EPS = "SELL_SIDE_CONSENSUS_EPS" # a provider's consensus estimate series | |
| COMPANY_NON_GAAP_EPS = "COMPANY_NON_GAAP_EPS" # the company's own non-GAAP EPS (from its release) | |
| INTERNAL_ADJUSTED_MODEL = "INTERNAL_ADJUSTED_MODEL" # a model output on no external definition | |
| DEFINITIONS = { | |
| PROVIDER_REPORTED_EPS: "the data provider's reported (Street) EPS for each quarter, as published by the provider", | |
| SELL_SIDE_CONSENSUS_EPS: "the provider's aggregated sell-side estimate for the fiscal year", | |
| COMPANY_NON_GAAP_EPS: "the company's own non-GAAP EPS, per its earnings release", | |
| INTERNAL_ADJUSTED_MODEL: "an internal adjusted-earnings model with no external definition", | |
| } | |
| # Explicit reconciliations: pairs of definitions that may be compared, only for the same provider, | |
| # with the evidence that puts them on one basis. Anything not listed is incomparable. | |
| RECONCILIATIONS = { | |
| frozenset({PROVIDER_REPORTED_EPS, SELL_SIDE_CONSENSUS_EPS}): { | |
| "same_provider_required": True, | |
| "evidence": "the provider publishes its reported EPS and its estimate on one basis and computes the " | |
| "earnings surprise between them (reported − estimated), so the two series are defined alike", | |
| }, | |
| } | |
| # forward periods compare only when they overlap at least this much (share of the longer period) | |
| MIN_PERIOD_OVERLAP = 0.75 | |
| def definitions_compatible(a, b): | |
| """(ok, reason) for two metrics where at least one is on an adjusted basis.""" | |
| da, db = getattr(a, "adjustment_definition_id", None), getattr(b, "adjustment_definition_id", None) | |
| ba, bb = getattr(a, "accounting_basis", None), getattr(b, "accounting_basis", None) | |
| if ba != bb: | |
| return False, f"accounting basis {ba} ≠ {bb}" | |
| if not da or not db: | |
| return False, (f"adjusted figure without an adjustment definition " | |
| f"({getattr(a, 'metric_id', '?') if not da else getattr(b, 'metric_id', '?')})") | |
| va, vb = getattr(a, "adjustment_version", None), getattr(b, "adjustment_version", None) | |
| if da == db: | |
| if va and vb and va != vb: | |
| return False, f"adjustment definition {da} changed between versions ({va} ≠ {vb})" | |
| return True, "" | |
| rec = RECONCILIATIONS.get(frozenset({da, db})) | |
| if rec is None: | |
| return False, f"adjustment definitions differ ({da} ≠ {db}) and no reconciliation is registered" | |
| pa, pb = getattr(a, "adjustment_provider", None), getattr(b, "adjustment_provider", None) | |
| if rec.get("same_provider_required") and (not pa or not pb or _provider_root(pa) != _provider_root(pb)): | |
| return False, f"adjustment definitions {da} and {db} are reconciled only for one provider ({pa} ≠ {pb})" | |
| return True, "" | |
| def reconciliation_evidence(a, b): | |
| """The registered evidence for comparing two different definitions (or None).""" | |
| rec = RECONCILIATIONS.get(frozenset({getattr(a, "adjustment_definition_id", None), | |
| getattr(b, "adjustment_definition_id", None)})) | |
| return (rec or {}).get("evidence") | |
| def _provider_root(p): | |
| """'Crucible model (on Alpha Vantage reportedEPS)' → 'alpha vantage': a model of a provider's series | |
| carries that provider's definition.""" | |
| p = str(p or "").lower() | |
| for root in ("alpha vantage", "sharadar", "sec", "company"): | |
| if root in p: | |
| return root | |
| return p | |
| # ── fiscal periods ──────────────────────────────────────────────────────────────────── | |
| def _d(x): | |
| try: | |
| return _dt.date.fromisoformat(str(x)[:10]) | |
| except (TypeError, ValueError): | |
| return None | |
| def period_bounds_after(end_of_base, days=365): | |
| """(start, end) of the twelve months after a reporting date, as ISO strings.""" | |
| e = _d(end_of_base) | |
| if e is None: | |
| return None, None | |
| return (e + _dt.timedelta(days=1)).isoformat(), (e + _dt.timedelta(days=days)).isoformat() | |
| def fiscal_year_bounds(fy_end): | |
| """(start, end) of the fiscal year ending on ``fy_end``.""" | |
| e = _d(fy_end) | |
| if e is None: | |
| return None, None | |
| try: | |
| s = e.replace(year=e.year - 1) + _dt.timedelta(days=1) | |
| except ValueError: # 29 February | |
| s = e - _dt.timedelta(days=365) | |
| return s.isoformat(), e.isoformat() | |
| def overlap_share(a_start, a_end, b_start, b_end): | |
| sa, ea, sb, eb = _d(a_start), _d(a_end), _d(b_start), _d(b_end) | |
| if not all((sa, ea, sb, eb)): | |
| return None | |
| inter = (min(ea, eb) - max(sa, sb)).days + 1 | |
| longer = max((ea - sa).days + 1, (eb - sb).days + 1) | |
| return max(0.0, inter / longer) if longer > 0 else None | |
| def ntm_weights(base_end, fy1_start, fy1_end, fy2_start, fy2_end): | |
| """(w1, w2, start, end): the share of the twelve months after ``base_end`` that falls in fiscal year 1 and | |
| fiscal year 2. The explicit conversion that lets fiscal-year estimates describe the forecast's period.""" | |
| s, e = period_bounds_after(base_end) | |
| if not s: | |
| return None | |
| o1 = overlap_share(s, e, fy1_start, fy1_end) or 0.0 | |
| o2 = overlap_share(s, e, fy2_start, fy2_end) or 0.0 | |
| tot = o1 + o2 | |
| if tot < 0.95: # the two fiscal years do not cover the window | |
| return None | |
| return o1 / tot, o2 / tot, s, e | |
| def periods_overlap(a, b, minimum=MIN_PERIOD_OVERLAP): | |
| """(ok, reason) for two forward figures: they must state their periods and overlap materially.""" | |
| share = overlap_share(getattr(a, "period_start", None), getattr(a, "period_end", None), | |
| getattr(b, "period_start", None), getattr(b, "period_end", None)) | |
| if share is None: | |
| return False, (f"forward period not stated ({getattr(a, 'metric_id', '?')} " | |
| f"{getattr(a, 'period_start', None)}–{getattr(a, 'period_end', None)} vs {getattr(b, 'metric_id', '?')} " | |
| f"{getattr(b, 'period_start', None)}–{getattr(b, 'period_end', None)})") | |
| if share < minimum: | |
| return False, (f"forward periods overlap only {share:.0%} ({a.period_start}–{a.period_end} vs " | |
| f"{b.period_start}–{b.period_end}); an explicit conversion is required") | |
| return True, "" | |