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).
- 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 the cover-page `dei` fact whose own `end` (cover date) is stored separately.
- Cash and debt composites are aggregate-first and mutually exclusive (each
  source tag counted at most once).
"""

from __future__ import annotations

import logging
from dataclasses import dataclass, field
from datetime import date, datetime
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}
_YTD_TOLERANCE_DAYS = 20  # covers 52/53-week fiscal calendars

# us-gaap duration concepts (money), priority order; first present wins.
_DURATION_USD = {
    "revenue": [
        "RevenueFromContractWithCustomerExcludingAssessedTax",
        "Revenues",
        "SalesRevenueNet",
    ],
    "net_income": ["NetIncomeLoss"],
    "operating_income": ["OperatingIncomeLoss"],
    "cfo": [
        "NetCashProvidedByUsedInOperatingActivities",
        "NetCashProvidedByUsedInOperatingActivitiesContinuingOperations",
    ],
    "capex": [
        "PaymentsToAcquirePropertyPlantAndEquipment",
        "PaymentsToAcquireProductiveAssets",
    ],
    "depreciation_amortization": [
        "DepreciationDepletionAndAmortization",
        "DepreciationAmortizationAndAccretionNet",
        "DepreciationAndAmortization",
    ],
}
_EPS_CONCEPTS = ["EarningsPerShareDiluted"]  # unit USD/shares
# 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


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


def parse_snapshots(
    companyfacts: dict[str, Any],
    filings: dict[str, FilingMeta],
    accessions: set[str],
) -> tuple[list[SnapshotRow], list[dict[str, str]]]:
    """Build snapshot rows for ``accessions`` (those with facts + filing meta).

    Returns (rows, skips) where each skip is {accession, reason} for filings
    with no usable period identity — the caller counts these in validation_json.
    """
    cik = f"{int(companyfacts['cik']):010d}"
    by_accn = _index_by_accession(companyfacts)
    rows: list[SnapshotRow] = []
    skips: list[dict[str, str]] = []
    for accn in accessions:
        meta = filings.get(accn)
        facts = by_accn.get(accn)
        if meta is None or not facts:
            skips.append({"accession": accn, "reason": "no facts or filing metadata"})
            continue
        row = _parse_one(cik, accn, facts, meta)
        if row is None:
            skips.append({"accession": accn, "reason": "no usable period identity"})
            continue
        rows.append(row)
    return rows, skips


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")
                    if not accn:
                        continue
                    out.setdefault(accn, []).append(
                        Fact(
                            taxonomy=taxonomy,
                            concept=concept,
                            unit=unit,
                            start=_d(f.get("start")),
                            end=_d(f.get("end")),
                            val=f.get("val"),
                            fy=f.get("fy"),
                            fp=f.get("fp"),
                        )
                    )
    return out


def _parse_one(cik: str, accn: str, facts: list[Fact], meta: FilingMeta) -> SnapshotRow | None:
    fy, fp = _fiscal_context(facts)
    if fy is None or fp not in _EXPECTED_YTD_DAYS:
        return None

    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)
    row.shares_outstanding, row.shares_outstanding_date = _select_shares(facts)
    return row


def _fiscal_context(facts: list[Fact]) -> tuple[int | None, str | None]:
    """A filing's own (fy, fp) — shared by all its facts; take the first set."""
    for f in facts:
        if f.fy is not None and f.fp:
            return f.fy, f.fp
    return None, None


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.concept != concept
                or f.unit != unit
                or f.start is None
                or f.end != report_date
                or f.val is None
            ):
                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.concept == concept
                and f.unit == "USD"
                and f.start is None
                and f.end == report_date
                and f.val is not None
            ):
                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]) -> tuple[float | None, date | None]:
    """dei:EntityCommonStockSharesOutstanding — cover-page instant. Store its own
    end (the cover date, which differs from period_end)."""
    candidates = [
        f
        for f in facts
        if f.taxonomy == "dei"
        and f.concept == "EntityCommonStockSharesOutstanding"
        and f.unit == "shares"
        and f.start is None
        and f.val is not None
    ]
    if not candidates:
        return None, None
    best = max(candidates, key=lambda f: f.end)
    return float(best.val), best.end


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