dennisthiessenandClaude Opus 4.8
4754dbc17b
1. Multi-class shares: prefer the single dei:EntityCommonStockSharesOutstanding cover-page fact; else fall back to us-gaap:CommonStockSharesOutstanding at period end (Alphabet has no dei fact). Never sum class facts (companyfacts is non-dimensional) and never use weighted-average/diluted; conflicting values -> null, counted as an "ambiguous shares outstanding" note in validation. Plan's "sum class-specific" wording corrected. Verified live: Alphabet shares now populate (12.1B), Apple still uses its dei cover date. 2. Fiscal context is the majority (fy, fp) among facts ending at reportDate, with ties rejected — no longer the arbitrary first fact. 3. Hardening: catalog selectors require taxonomy == "us-gaap"; indexing drops malformed facts (missing accession/end, non-finite value) so a custom concept or bad date can't be selected. Tests: +8 (dei precedence, us-gaap fallback, conflict->null, no weighted-average, tie-context skip, foreign-taxonomy/malformed ignored, ambiguous-shares note). 14 passed, 1 skipped. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
345 lines
13 KiB
Python
345 lines
13 KiB
Python
"""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
|
|
import math
|
|
from dataclasses import dataclass
|
|
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, note = _parse_one(cik, accn, facts, meta)
|
|
if row is None:
|
|
skips.append({"accession": accn, "reason": note or "unparseable"})
|
|
continue
|
|
rows.append(row)
|
|
if note: # row produced, but a field-level issue to count in validation
|
|
skips.append({"accession": accn, "reason": note})
|
|
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")
|
|
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
|
|
) -> 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 = _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
|
|
return row, ("ambiguous shares outstanding" if ambiguous else None)
|
|
|
|
|
|
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 _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)
|