crucible / engine /definitions.py
s-ttp's picture
Root-cause remediation: reconciled forecast basis, canonical segments, PricingState, company guidance, ownership/insider permissions, S1-S12 validators
931437f verified
Raw History Blame Contribute Delete
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, ""