signal-platform/app/services/sec_facts_parser.py

"""Pure parser: SEC companyfacts -> fundamental_snapshots rows.

Turns one issuer's `companyfacts` JSON (+ its submissions filing metadata) into
per-accession snapshot rows for the filing's **primary period**, following the
A3 design (docs/dolt-sec-a3-design.md). No I/O, no DB — unit-testable against a
fixture and verifiable against a real companyfacts pull.

The load-bearing rules (design Decision 2 + review):
- Period identity comes from `end == submissions.reportDate`, never `fy/fp`
  (fy/fp is the *filing's* context; comparatives inside a filing repeat it).
  This applies to the stored `fiscal_year`/`fiscal_period` too: they are derived
  from `reportDate` against the issuer's `fiscalYearEnd` (see `_period_identity`),
  because SEC's fy/fp collide and invert often enough to break the quarter chain.
- Duration facts are stored as **cumulative YTD**: pick the fact whose span
  matches the fiscal-period-to-date length (Q1≈3mo … FY≈12mo) within tolerance.
  If no YTD-length fact exists, store null — never a discrete masquerading as YTD.
- Balance-sheet instants are taken at `end == reportDate`. `shares_outstanding`
  is a single consolidated value: the cover-page `dei` fact (its own cover-date
  `end` stored separately) if present, else `us-gaap:CommonStockSharesOutstanding`
  at period end (e.g. Alphabet has no `dei` fact) — never a class sum or the
  weighted-average/diluted count. Multi-class issuers report it per class, which
  is dimensional and therefore absent from companyfacts entirely, so
  `weighted_avg_diluted_shares` is stored alongside as an explicit fallback for
  market cap — a separate column, never backfilled into `shares_outstanding`.
- Cash and debt composites are aggregate-first and mutually exclusive (each
  source tag counted at most once).

`parse_snapshots` separates `skipped_filings` (no usable row produced) from
`field_issues` (a row was produced but a field is null/ambiguous) — callers must
not treat field issues as missing coverage.
"""

from __future__ import annotations

import logging
import math
from dataclasses import dataclass, field
from datetime import date, datetime, timedelta
from typing import Any, NamedTuple

logger = logging.getLogger(__name__)

# Expected YTD span (days) per fiscal period; a duration fact must land within
# tolerance of this to count as the period's cumulative value.
_EXPECTED_YTD_DAYS = {"Q1": 91, "Q2": 182, "Q3": 273, "FY": 365}
# Period identity (see _period_identity): how far a quarter end sits before its
# fiscal-year end, and how far a fiscal-year end may drift from the nominal MMDD.
# The quarter bands are 91 days apart, so ±35 stays unambiguous even for a 4-4-5
# filer whose 16-week Q4 puts Q3 112 days out.
_QUARTER_DAYS_TO_FY_END = {"Q1": 273, "Q2": 182, "Q3": 91}
_QUARTER_TOLERANCE_DAYS = 35
_FYE_DRIFT_TOLERANCE_DAYS = 21
# Covers 52/53-week calendars *and* 4-4-5 retail ones (12/12/12/16 weeks), whose
# YTD-Q3 is 36 weeks = 251-252 days and missed a 20-day tolerance by ~2 -- so
# COST/PEP lost Q3 every year, breaking the quarter chain and nulling TTM + YoY.
# Q1 84d, Q2 168d and FY 364d were always inside. Adjacent periods stay
# unambiguous at 25 (66-116, 157-207, 248-298, 340-390).
_YTD_TOLERANCE_DAYS = 25

# us-gaap duration concepts (money), priority order; first present wins.
_DURATION_USD = {
    # Order is load-bearing (first present wins) and the tail entries are
    # deliberately *appended*: every issuer that already resolved keeps the same
    # concept, and only issuers that resolved to nothing gain a value.
    #   - IncludingAssessedTax: REITs/consumer filers that tag only this variant
    #     (e.g. ARE, KHC) reported no revenue at all.
    #   - RevenuesNetOfInterestExpense: the banks' total-revenue tag. JPM/GS/WFC
    #     tag it in every 10-Q and `Revenues` only (if at all) in the 10-K.
    "revenue": [
        "RevenueFromContractWithCustomerExcludingAssessedTax",
        "Revenues",
        "SalesRevenueNet",
        "RevenueFromContractWithCustomerIncludingAssessedTax",
        "RevenuesNetOfInterestExpense",
    ],
    "net_income": ["NetIncomeLoss"],
    "operating_income": ["OperatingIncomeLoss"],
    "cfo": [
        "NetCashProvidedByUsedInOperatingActivities",
        "NetCashProvidedByUsedInOperatingActivitiesContinuingOperations",
    ],
    "capex": [
        "PaymentsToAcquirePropertyPlantAndEquipment",
        "PaymentsToAcquireProductiveAssets",
    ],
    "depreciation_amortization": [
        "DepreciationDepletionAndAmortization",
        "DepreciationAmortizationAndAccretionNet",
        "DepreciationAndAmortization",
    ],
}
# unit USD/shares. Appended (not reordered) so any issuer that already resolved
# keeps the same concept. REG tags only the continuing-operations variant on every
# filing; FCX switches by form type -- EarningsPerShareDiluted in its 10-Qs, the
# continuing-ops tag in its 10-K -- which nulled the FY row and killed Q4 + TTM.
# The basic variants are a last resort for a period that tags no diluted EPS at
# all (PPL's 2026 Q1). Basic ignores option/convert dilution so it slightly
# overstates EPS (~1.2% for PPL), but only fires when diluted is entirely absent,
# and high-dilution names always tag diluted -- so it never displaces a real one.
_EPS_CONCEPTS = [
    "EarningsPerShareDiluted",
    "IncomeLossFromContinuingOperationsPerDilutedShare",
    "EarningsPerShareBasic",
    "IncomeLossFromContinuingOperationsPerBasicShare",
]
# Weighted-average diluted share count (unit "shares"), the market-cap fallback
# for multi-class issuers whose cover-page count is dimensional and therefore
# absent from companyfacts. Always present, since EPS is computed from it.
_WEIGHTED_AVG_SHARE_CONCEPTS = [
    "WeightedAverageNumberOfDilutedSharesOutstanding",
    "WeightedAverageNumberOfSharesOutstandingBasicAndDiluted",
]
# us-gaap instant (balance-sheet) concepts, at end == reportDate.
_CASH = ["CashAndCashEquivalentsAtCarryingValue"]
_ST_INVESTMENTS = ["ShortTermInvestments", "MarketableSecuritiesCurrent"]  # pick one
_LONG_TERM_DEBT_AGG = ["LongTermDebt"]
_LONG_TERM_DEBT_PARTS = ["LongTermDebtNoncurrent", "LongTermDebtCurrent"]
_SHORT_TERM_DEBT = ["ShortTermBorrowings", "CommercialPaper"]  # pick one


class Fact(NamedTuple):
    taxonomy: str
    concept: str
    unit: str
    start: date | None  # None => instant
    end: date
    val: float
    fy: int | None
    fp: str | None


@dataclass
class SnapshotRow:
    cik: str
    accession: str
    form: str
    filed_date: date
    accepted_at: datetime
    period_end: date
    fiscal_year: int
    fiscal_period: str
    period_start: date | None = None
    revenue: float | None = None
    net_income: float | None = None
    operating_income: float | None = None
    diluted_eps: float | None = None
    cfo: float | None = None
    capex: float | None = None
    depreciation_amortization: float | None = None
    cash_and_st_investments: float | None = None
    total_debt: float | None = None
    shares_outstanding: float | None = None
    shares_outstanding_date: date | None = None
    weighted_avg_diluted_shares: float | None = None


@dataclass
class FilingMeta:
    report_date: date
    filing_date: date
    accepted_at: datetime
    form: str


@dataclass
class ParseResult:
    rows: list[SnapshotRow] = field(default_factory=list)
    # accessions for which NO row was produced (no facts / no usable period).
    skipped_filings: list[dict[str, str]] = field(default_factory=list)
    # accessions with a row but a field-level warning (e.g. ambiguous shares).
    field_issues: list[dict[str, str]] = field(default_factory=list)


def parse_snapshots(
    companyfacts: dict[str, Any],
    filings: dict[str, FilingMeta],
    accessions: set[str],
    fiscal_year_end: str | None = None,
) -> ParseResult:
    """Build snapshot rows for ``accessions`` (those with facts + filing meta).

    ``fiscal_year_end`` is the issuer's ``submissions.fiscalYearEnd`` (MMDD) and
    is what makes period identity independent of SEC's unreliable fy/fp fields
    (see ``_period_identity``). Omitting it falls back to the old fy/fp behaviour.

    ``skipped_filings`` = no row produced (missing facts/meta or no usable period
    identity); ``field_issues`` = a row was produced but a field is null/ambiguous.
    Callers must not use field issues as failed-row coverage.
    """
    cik = f"{int(companyfacts['cik']):010d}"
    by_accn = _index_by_accession(companyfacts)
    result = ParseResult()
    for accn in accessions:
        meta = filings.get(accn)
        facts = by_accn.get(accn)
        if meta is None or not facts:
            result.skipped_filings.append({"accession": accn, "reason": "no facts or filing metadata"})
            continue
        row, note = _parse_one(cik, accn, facts, meta, fiscal_year_end)
        if row is None:
            result.skipped_filings.append({"accession": accn, "reason": note or "unparseable"})
            continue
        result.rows.append(row)
        if note:
            result.field_issues.append({"accession": accn, "reason": note})
    return result


def companyfacts_accessions(companyfacts: dict[str, Any]) -> set[str]:
    """Every accession that appears anywhere in a companyfacts payload — used by
    the importer's index↔Company-Facts consistency gate."""
    return set(_index_by_accession(companyfacts).keys())


def _index_by_accession(companyfacts: dict[str, Any]) -> dict[str, list[Fact]]:
    """One pass over companyfacts -> {accession: [Fact, ...]}."""
    out: dict[str, list[Fact]] = {}
    for taxonomy, concepts in companyfacts.get("facts", {}).items():
        for concept, body in concepts.items():
            for unit, facts in body.get("units", {}).items():
                for f in facts:
                    accn = f.get("accn")
                    end = _d(f.get("end"))
                    val = f.get("val")
                    # Skip malformed facts so they can't be selected accidentally:
                    # every usable fact needs an accession, an end date, and a
                    # finite numeric value.
                    if not accn or end is None or not _finite(val):
                        continue
                    out.setdefault(accn, []).append(
                        Fact(
                            taxonomy=taxonomy,
                            concept=concept,
                            unit=unit,
                            start=_d(f.get("start")),
                            end=end,
                            val=val,
                            fy=f.get("fy"),
                            fp=f.get("fp"),
                        )
                    )
    return out


def _parse_one(
    cik: str, accn: str, facts: list[Fact], meta: FilingMeta,
    fiscal_year_end: str | None = None,
) -> tuple[SnapshotRow | None, str | None]:
    """Returns (row, note). row is None when there's no usable period identity;
    note is a validation reason (row-skip reason when row is None, else a
    field-level issue such as ambiguous shares)."""
    fy, fp = _period_identity(meta, fiscal_year_end)
    if fy is None or fp is None:
        # No fiscal calendar, or a period the calendar cannot place (a transition
        # period). Fall back to the filing's own context: an imperfect label still
        # beats dropping the filing entirely.
        fy, fp = _fiscal_context(facts, meta.report_date)
    if fy is None or fp not in _EXPECTED_YTD_DAYS:
        return None, "no usable period identity"

    row = SnapshotRow(
        cik=cik,
        accession=accn,
        form=meta.form,
        filed_date=meta.filing_date,
        accepted_at=meta.accepted_at,
        period_end=meta.report_date,
        fiscal_year=fy,
        fiscal_period=fp,
    )

    # duration YTD facts (money) + EPS
    for field_name, concepts in _DURATION_USD.items():
        val, start = _select_ytd(facts, concepts, meta.report_date, fp, "USD")
        setattr(row, field_name, val)
        if field_name == "revenue" and start is not None:
            row.period_start = start
    eps, eps_start = _select_ytd(facts, _EPS_CONCEPTS, meta.report_date, fp, "USD/shares")
    row.diluted_eps = eps
    if row.period_start is None and eps_start is not None:
        row.period_start = eps_start

    # balance-sheet instants at reportDate
    row.cash_and_st_investments = _compose_cash(facts, meta.report_date)
    row.total_debt = _compose_debt(facts, meta.report_date)
    shares, shares_date, ambiguous = _select_shares(facts, meta.report_date)
    row.shares_outstanding = shares
    row.shares_outstanding_date = shares_date
    row.weighted_avg_diluted_shares = _select_weighted_avg_shares(facts, meta.report_date)

    return row, ("ambiguous shares outstanding" if ambiguous else None)


def _period_identity(
    meta: FilingMeta, fiscal_year_end: str | None
) -> tuple[int | None, str | None]:
    """(fiscal_year, fiscal_period) from the period end and the issuer's fiscal
    calendar — never from the fy/fp fields.

    SEC's fy/fp describe the *filing*, and they are unreliable as period identity:
    observed in production, a 10-Q labelled ``FY`` (BXP), a year ending 2025-12-31
    labelled 2024 (FRT, a December filer), a year ending 2025-06-27 labelled 2027
    (STX), and four different period ends all labelled 2022 Q3 (PPL). Because
    readers key on (fiscal_year, fiscal_period), colliding labels silently discard
    a period and inverted ones scramble the quarter chain — nulling TTM and YoY.

    ``period_end`` is authoritative, so identity is derived from it: the form
    decides FY vs quarter, and distance to the fiscal-year end decides which
    quarter. Labels need not match the issuer's own naming — a filer whose year
    ends in early January (DPZ) shifts by one — they need to be unique, monotonic
    and YoY-aligned, which is all the derivation asks of them. Nothing outside the
    derivation reads these columns.
    """
    fy = _fiscal_year_of(meta.report_date, fiscal_year_end)
    if fy is None:
        return None, None
    if meta.form.startswith("10-K"):
        return fy, "FY"
    nominal_end = _nominal_fy_end(fy, fiscal_year_end)
    if nominal_end is None:
        return None, None
    remaining = (nominal_end - meta.report_date).days
    best = min(
        _QUARTER_DAYS_TO_FY_END,
        key=lambda k: abs(_QUARTER_DAYS_TO_FY_END[k] - remaining),
    )
    if abs(_QUARTER_DAYS_TO_FY_END[best] - remaining) > _QUARTER_TOLERANCE_DAYS:
        return None, None  # transition period or odd filing — let the caller fall back
    return fy, best


def _nominal_fy_end(year: int, fiscal_year_end: str | None) -> date | None:
    """The issuer's nominal fiscal-year end in ``year`` from a MMDD string."""
    if not fiscal_year_end or len(fiscal_year_end) != 4 or not fiscal_year_end.isdigit():
        return None
    month, day = int(fiscal_year_end[:2]), int(fiscal_year_end[2:])
    if not 1 <= month <= 12 or not 1 <= day <= 31:
        return None
    while day > 28:  # 52/53-week ends land on 0229/0230/0231 in some filings
        try:
            return date(year, month, day)
        except ValueError:
            day -= 1
    return date(year, month, day)


def _fiscal_year_of(period_end: date, fiscal_year_end: str | None) -> int | None:
    """Which fiscal year ``period_end`` belongs to.

    A 52/53-week calendar's real year end drifts around the nominal MMDD (and can
    cross the calendar year), so allow drift before rolling into the next year.
    """
    nominal = _nominal_fy_end(period_end.year, fiscal_year_end)
    if nominal is None:
        return None
    return (
        period_end.year
        if period_end <= nominal + timedelta(days=_FYE_DRIFT_TOLERANCE_DAYS)
        else period_end.year + 1
    )


def _fiscal_context(facts: list[Fact], report_date: date) -> tuple[int | None, str | None]:
    """The filing's (fy, fp) taken as the majority context among the facts that
    end at reportDate (the current-period facts, which share the filing's
    context). Reject a tie so a conflicting context is never chosen arbitrarily."""
    counts: dict[tuple[int, str], int] = {}
    for f in facts:
        if f.end == report_date and f.fy is not None and f.fp:
            counts[(f.fy, f.fp)] = counts.get((f.fy, f.fp), 0) + 1
    if not counts:
        return None, None
    ranked = sorted(counts.items(), key=lambda kv: kv[1], reverse=True)
    if len(ranked) > 1 and ranked[0][1] == ranked[1][1]:
        return None, None  # tie → conflicting contexts, reject
    return ranked[0][0]


def _select_ytd(
    facts: list[Fact], concepts: list[str], report_date: date, fp: str, unit: str
) -> tuple[float | None, date | None]:
    """First present concept whose duration fact ends at reportDate and whose span
    matches the fiscal-period-to-date length. Returns (val, period_start)."""
    expected = _EXPECTED_YTD_DAYS[fp]
    for concept in concepts:
        best: Fact | None = None
        best_diff: int | None = None
        for f in facts:
            if (
                f.taxonomy != "us-gaap"
                or f.concept != concept
                or f.unit != unit
                or f.start is None
                or f.end != report_date
            ):
                continue
            diff = abs((f.end - f.start).days - expected)
            if diff <= _YTD_TOLERANCE_DAYS and (best_diff is None or diff < best_diff):
                best, best_diff = f, diff
        if best is not None:
            return float(best.val), best.start
    return None, None


def _select_instant(facts: list[Fact], concepts: list[str], report_date: date) -> float | None:
    """First present instant (balance-sheet) fact at end == reportDate, unit USD."""
    for concept in concepts:
        for f in facts:
            if (
                f.taxonomy == "us-gaap"
                and f.concept == concept
                and f.unit == "USD"
                and f.start is None
                and f.end == report_date
            ):
                return float(f.val)
    return None


def _compose_cash(facts: list[Fact], report_date: date) -> float | None:
    cash = _select_instant(facts, _CASH, report_date)
    st = _select_instant(facts, _ST_INVESTMENTS, report_date)  # first present of the two
    if cash is None and st is None:
        return None
    return (cash or 0.0) + (st or 0.0)


def _compose_debt(facts: list[Fact], report_date: date) -> float | None:
    long_term = _select_instant(facts, _LONG_TERM_DEBT_AGG, report_date)
    if long_term is None:
        nc = _select_instant(facts, ["LongTermDebtNoncurrent"], report_date)
        cur = _select_instant(facts, ["LongTermDebtCurrent"], report_date)
        long_term = None if nc is None and cur is None else (nc or 0.0) + (cur or 0.0)
    short_term = _select_instant(facts, _SHORT_TERM_DEBT, report_date)
    if long_term is None and short_term is None:
        return None
    return (long_term or 0.0) + (short_term or 0.0)


def _select_shares(
    facts: list[Fact], report_date: date
) -> tuple[float | None, date | None, bool]:
    """Issuer-wide shares outstanding as a single consolidated value (never a
    class sum — companyfacts is non-dimensional — and never weighted-average/
    diluted). Returns (value, shares_date, ambiguous).

    1. Prefer the `dei:EntityCommonStockSharesOutstanding` cover-page instant;
       its own end is the shares date (cover date != period_end).
    2. Else fall back to `us-gaap:CommonStockSharesOutstanding` at period end
       (e.g. Alphabet has no dei fact); shares date = reportDate.
    Conflicting values within the chosen source → (None, None, True) to be
    counted in validation.
    """
    dei = [
        f
        for f in facts
        if f.taxonomy == "dei"
        and f.concept == "EntityCommonStockSharesOutstanding"
        and f.unit == "shares"
        and f.start is None
    ]
    if dei:
        if len({f.val for f in dei}) > 1:
            return None, None, True
        best = max(dei, key=lambda f: f.end)
        return float(best.val), best.end, False

    gaap = [
        f
        for f in facts
        if f.taxonomy == "us-gaap"
        and f.concept == "CommonStockSharesOutstanding"
        and f.unit == "shares"
        and f.start is None
        and f.end == report_date
    ]
    if gaap:
        if len({f.val for f in gaap}) > 1:
            return None, None, True
        return float(gaap[0].val), report_date, False

    return None, None, False  # simply absent — not a conflict


def _select_weighted_avg_shares(facts: list[Fact], report_date: date) -> float | None:
    """The most recent quarter's weighted-average diluted share count.

    Deliberately the **shortest** duration ending at reportDate, not the YTD one:
    the shorter the window the closer the average sits to the current count, which
    is what a market cap wants. Measured against issuers where the true
    point-in-time count is available, the quarter average is within ~0.6%.
    """
    best: tuple[int, float] | None = None
    for concept in _WEIGHTED_AVG_SHARE_CONCEPTS:
        for f in facts:
            if (
                f.taxonomy != "us-gaap"
                or f.concept != concept
                or f.unit != "shares"
                or f.start is None
                or f.end != report_date
                or f.val <= 0
            ):
                continue
            span = (f.end - f.start).days
            if best is None or span < best[0]:
                best = (span, float(f.val))
        if best is not None:
            return best[1]  # first present concept wins, as elsewhere
    return None


def _d(value: Any) -> date | None:
    if not value:
        return None
    try:
        return date.fromisoformat(str(value)[:10])
    except ValueError:
        return None


def _finite(value: Any) -> bool:
    """True for a finite numeric value (rejects None, bool, strings, NaN/inf)."""
    return isinstance(value, (int, float)) and not isinstance(value, bool) and math.isfinite(value)