Blame for engine/definitions.py

"""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, ""