"""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. Known limitation: ``fiscalYearEnd`` is the issuer's *current* calendar, so a company that has changed its fiscal year end gets its historical periods measured against the new one. The quarter tolerance shunts most of those to the fy/fp fallback, and a same-key collision resolves newest-wins, so the failure mode is a degraded old year rather than a scrambled current one. """ 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)