From 8dcdcac2a6d0e260522ce262c79aa562a53f3e99 Mon Sep 17 00:00:00 2001 From: Dennis Thiessen Date: Wed, 22 Jul 2026 19:55:58 +0200 Subject: [PATCH] =?UTF-8?q?fix(sec):=20A3=20slice-2b=20review=20=E2=80=94?= =?UTF-8?q?=20no=20index=20cap,=20full=20discrepancy=20+=20malformed=20gat?= =?UTF-8?q?e?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 1. Removed the 45-day index-walk cap: it discarded the older part of a long outage while still advancing source_max_date, permanently losing filings. The walk now covers every unprocessed date (a large gap is one-time cost). 2. Discrepancy detection meets the immutability contract: it compares ALL source snapshot fields (not five), read-only during stage/validate, reports the differing accessions + fields in validation_json, and promote emits a warning system event (in-transaction) — never mutating the stored row. 3. Malformed companyfacts (missing facts/units structure) are recorded separately and FAIL validation, instead of silently degrading to skipped rows that the 50% backfill coverage floor could still pass. Also corrected the stale "sum share classes" / DEI-only wording in the snapshot model docstring and the A3 design doc to describe the us-gaap fallback. Tests: +4 regressions (>45-day gap loses nothing, newly-added issuer backfills without filing, malformed payload fails, shares discrepancy detected + evented). 23 passed, 1 skipped. Co-Authored-By: Claude Opus 4.8 --- app/models/fundamental_snapshot.py | 12 +- app/services/sec_fundamentals_importer.py | 104 +++++++++----- docs/dolt-sec-a3-design.md | 8 +- tests/unit/test_sec_fundamentals_importer.py | 135 +++++++++++++++---- 4 files changed, 194 insertions(+), 65 deletions(-) diff --git a/app/models/fundamental_snapshot.py b/app/models/fundamental_snapshot.py index addfee4..afdc469 100644 --- a/app/models/fundamental_snapshot.py +++ b/app/models/fundamental_snapshot.py @@ -20,11 +20,13 @@ class FundamentalSnapshot(Base): capex, depreciation_amortization) hold the filing's normalized **cumulative YTD/FY** value over (period_start -> period_end). Balance-sheet facts (cash_and_st_investments, total_debt, shares_outstanding) are **period-end** - values. ``shares_outstanding`` is a point-in-time count - (``dei:EntityCommonStockSharesOutstanding``, summed across share classes for - a multi-class issuer) — deliberately not the weighted-average diluted share - count, since both consumers (estimated market cap, YoY dilution read) want a - point-in-time value. Discrete quarters (10-Q YTD deltas, Q4 = FY - Q1..Q3), TTM, YoY and + values. ``shares_outstanding`` is a single consolidated point-in-time count — + the ``dei:EntityCommonStockSharesOutstanding`` cover-page fact, or + ``us-gaap:CommonStockSharesOutstanding`` at period end when no dei fact exists + (e.g. Alphabet). It is never a class sum (companyfacts is non-dimensional) nor + the weighted-average diluted count, since both consumers (estimated market cap, + YoY dilution read) want a point-in-time value. Discrete quarters (10-Q YTD + deltas, Q4 = FY - Q1..Q3), TTM, YoY and the quarter tape are all derived at read time — so non-calendar fiscal years resolve correctly and a later amendment never leaves a stale frozen quarter. """ diff --git a/app/services/sec_fundamentals_importer.py b/app/services/sec_fundamentals_importer.py index d494335..c44497a 100644 --- a/app/services/sec_fundamentals_importer.py +++ b/app/services/sec_fundamentals_importer.py @@ -36,6 +36,7 @@ from sqlalchemy import func, select from app.database import insert_for_session from app.models.data_import_run import DataImportRun from app.models.fundamental_snapshot import FundamentalSnapshot +from app.models.system_event import SystemEvent from app.services import sec_facts_parser as parser from app.services import sec_universe from app.services.data_import import STATUS_PROMOTED, ValidationResult @@ -50,9 +51,6 @@ _XBRL_FORMS = {"10-K", "10-Q", "10-K/A", "10-Q/A"} # On the one-time backfill, require this fraction of tracked issuers to yield at # least one snapshot (guards a broken fetch/parse from promoting a hollow table). MIN_BACKFILL_COVERAGE = 0.5 -# Bound how far back the incremental index walk goes if the job hasn't run in a -# while (each day = one small request); older gaps are logged, not silently lost. -_MAX_INDEX_WALK_DAYS = 45 _SNAPSHOT_COLS = ( "cik", "accession", "form", "filed_date", "accepted_at", "period_start", @@ -61,8 +59,9 @@ _SNAPSHOT_COLS = ( "cash_and_st_investments", "total_debt", "shares_outstanding", "shares_outstanding_date", ) -# Fields compared to flag a differing existing accession (immutable → report, not mutate). -_DISCREPANCY_COLS = ("period_end", "fiscal_year", "fiscal_period", "revenue", "net_income") +# Compare ALL source fields (every column except the accession key) to flag a +# differing existing accession — immutable, so we report, never mutate. +_COMPARE_COLS = tuple(c for c in _SNAPSHOT_COLS if c != "accession") @dataclass @@ -74,6 +73,9 @@ class StagedFundamentals: field_issues: list[dict[str, str]] = field(default_factory=list) skipped_non_xbrl: list[dict[str, str]] = field(default_factory=list) missing_xbrl: list[dict[str, str]] = field(default_factory=list) + invalid_payloads: list[dict[str, str]] = field(default_factory=list) + existing_accessions: set[str] = field(default_factory=set) + discrepancies: list[dict[str, Any]] = field(default_factory=list) backfill: bool = False issuers_fetched: int = 0 issuers_with_rows: int = 0 @@ -146,10 +148,30 @@ class SecFundamentalsImporter: for cik in sorted(backfill_ciks | incremental_ciks): is_backfill = cik in backfill_ciks await self._stage_issuer(client, cik, is_backfill, filed_by_cik, staged) + + # Read-only discrepancy detection: an accession we reconstructed that is + # already stored, differing in ANY source field (immutable → report in + # validation, event on promote, never mutate). Also gives promote the + # existing set so its insert count is dialect-independent. + if staged.rows: + existing = await self._existing_by_accession(db, [r.accession for r in staged.rows]) + staged.existing_accessions = set(existing) + for row in staged.rows: + old = existing.get(row.accession) + if old is not None: + fields = _diff_fields(row, old) + if fields: + staged.discrepancies.append({"accession": row.accession, "fields": fields}) return staged async def _stage_issuer(self, client, cik, is_backfill, filed_by_cik, staged) -> None: cf = await client.companyfacts(cik) + if not isinstance(cf, dict) or not isinstance(cf.get("facts"), dict): + # Malformed payload (missing facts/units structure) — record separately + # and fail validation, rather than letting it degrade to skipped rows. + staged.invalid_payloads.append({"cik": cik10(cik), "reason": "missing facts structure"}) + staged.issuers_fetched += 1 + return sub = await client.submissions(cik, include_history=is_backfill) xbrl_meta, nonxbrl = _filing_meta(sub) @@ -191,6 +213,12 @@ class SecFundamentalsImporter: f"{len(staged.missing_xbrl)} tracked XBRL filing(s) not yet in " "Company Facts (index/facts lag) — retry" ) + # Malformed companyfacts payloads must fail, not degrade to skipped rows. + if staged.invalid_payloads: + messages.append( + f"{len(staged.invalid_payloads)} issuer(s) returned a malformed " + "companyfacts payload (missing facts structure)" + ) accns = [r.accession for r in staged.rows] if len(accns) != len(set(accns)): @@ -213,7 +241,11 @@ class SecFundamentalsImporter: "field_issues": len(staged.field_issues), "skipped_non_xbrl": len(staged.skipped_non_xbrl), "missing_xbrl": len(staged.missing_xbrl), + "invalid_payloads": staged.invalid_payloads, "cik_updates": len(staged.resolved.cik_updates), + # differing existing accessions (immutable — kept, reported here) + "discrepancies": staged.discrepancies[:50], + "discrepancy_count": len(staged.discrepancies), } return ValidationResult( ok=not messages, @@ -224,33 +256,37 @@ class SecFundamentalsImporter: async def promote(self, db, staged: StagedFundamentals, run_id: int) -> dict[str, int]: inserted = 0 - discrepancies = 0 - if staged.rows: - existing = await self._existing_by_accession(db, [r.accession for r in staged.rows]) - for row in staged.rows: - old = existing.get(row.accession) - if old is not None: - if _differs(row, old): - discrepancies += 1 - logger.warning( - "sec_facts: accession %s reconstructed differently than " - "stored (immutable — not overwriting)", row.accession - ) - continue # ON CONFLICT DO NOTHING (below) leaves it untouched - stmt = insert_for_session(db, FundamentalSnapshot).values( - **_row_values(row, run_id) - ) - stmt = stmt.on_conflict_do_nothing(index_elements=["accession"]) - await db.execute(stmt) - inserted += 1 + for row in staged.rows: + if row.accession in staged.existing_accessions: + continue # immutable — keep the original row + stmt = insert_for_session(db, FundamentalSnapshot).values(**_row_values(row, run_id)) + stmt = stmt.on_conflict_do_nothing(index_elements=["accession"]) # race belt-and-suspenders + await db.execute(stmt) + inserted += 1 + + # Warn (in-transaction, so it commits atomically with the promotion) when + # any existing accession reconstructed differently — kept immutable. + if staged.discrepancies: + accns = ", ".join(d["accession"] for d in staged.discrepancies[:10]) + db.add(SystemEvent( + severity="warning", + source="sec_facts", + code="snapshot_discrepancy", + message=( + f"{len(staged.discrepancies)} stored accession(s) reconstructed " + f"differently; kept immutable: {accns}" + )[:4000], + dedup_key=f"sec_facts:discrepancy:{run_id}", + created_at=_now(), + )) ticker_counts = await sec_universe.apply_ticker_updates( db, staged.resolved, staged.sic_updates ) return { "inserted": inserted, - "existing_unchanged": len(staged.rows) - inserted, - "discrepancies": discrepancies, + "existing_unchanged": len(staged.existing_accessions), + "discrepancies": len(staged.discrepancies), **ticker_counts, } @@ -269,12 +305,15 @@ class SecFundamentalsImporter: async def _collect_index_rows( self, client: SecClient, last_processed: date, latest: date ) -> list[dict[str, Any]]: + # Walk EVERY unprocessed date. No cap — dropping the older part of a long + # outage while still advancing source_max_date would permanently lose + # those filings. A large gap is one-time cost, not silent data loss. tracked = set(self._resolved.cik_to_ticker_ids) if self._resolved else set() - start = max(last_processed + timedelta(days=1), latest - timedelta(days=_MAX_INDEX_WALK_DAYS)) - if start > last_processed + timedelta(days=1): - logger.warning("sec_facts: index gap > %d days; walking from %s", _MAX_INDEX_WALK_DAYS, start) + gap = (latest - last_processed).days + if gap > 60: + logger.warning("sec_facts: %d-day index gap since %s; walking all", gap, last_processed) rows: list[dict[str, Any]] = [] - day = start + day = last_processed + timedelta(days=1) while day <= latest: for r in await client.daily_index(day): if r["form"] in _XBRL_FORMS and r["cik"] in tracked: @@ -339,5 +378,6 @@ def _row_values(row: SnapshotRow, run_id: int) -> dict[str, Any]: return values -def _differs(row: SnapshotRow, old: FundamentalSnapshot) -> bool: - return any(getattr(row, col) != getattr(old, col) for col in _DISCREPANCY_COLS) +def _diff_fields(row: SnapshotRow, old: FundamentalSnapshot) -> list[str]: + """Source fields where a re-parsed row differs from the stored (immutable) row.""" + return [col for col in _COMPARE_COLS if getattr(row, col) != getattr(old, col)] diff --git a/docs/dolt-sec-a3-design.md b/docs/dolt-sec-a3-design.md index 6a4ea5b..cd54949 100644 --- a/docs/dolt-sec-a3-design.md +++ b/docs/dolt-sec-a3-design.md @@ -101,8 +101,12 @@ One `fundamental_snapshots` row per accession, representing the filing's Q3≈9mo, FY≈12mo). **If the YTD fact is absent, store null — never a discrete masquerading as cumulative** (that would poison read-time differencing). - **Balance-sheet instants → at `end == reportDate`.** `shares_outstanding` is - the exception: take `dei:EntityCommonStockSharesOutstanding` for that accession - and store *its own* `end` in `shares_outstanding_date` (cover date ≠ period_end). + the exception: prefer the `dei:EntityCommonStockSharesOutstanding` cover-page + fact and store *its own* `end` in `shares_outstanding_date` (cover date ≠ + period_end); when there is no dei fact (e.g. Alphabet) fall back to + `us-gaap:CommonStockSharesOutstanding` at `reportDate`. A single consolidated + value — never a class sum (companyfacts is non-dimensional) nor + weighted-average/diluted; conflicting values → null. - **Amendments:** a real `10-K/A` / `10-Q/A` is a new accession → a new immutable row for the same `(cik, fy, fp)`; readers pick the newest valid `accepted_at`. - **Out of scope (stated, not silent):** restatements that appear *only* as diff --git a/tests/unit/test_sec_fundamentals_importer.py b/tests/unit/test_sec_fundamentals_importer.py index 67306c6..9a28b72 100644 --- a/tests/unit/test_sec_fundamentals_importer.py +++ b/tests/unit/test_sec_fundamentals_importer.py @@ -50,9 +50,9 @@ def _shares(end, val, accn, fy, fp): return {"end": end, "val": val, "fy": fy, "fp": fp, "accn": accn, "form": "10-K"} -def _companyfacts(rev_facts, share_facts): +def _companyfacts(rev_facts, share_facts, cik=320193): return { - "cik": 320193, + "cik": cik, "facts": { "us-gaap": {"RevenueFromContractWithCustomerExcludingAssessedTax": {"units": {"USD": rev_facts}}}, "dei": {"EntityCommonStockSharesOutstanding": {"units": {"shares": share_facts}}}, @@ -269,36 +269,119 @@ async def test_failed_backfill_leaves_tickers_unwritten(engine): assert all(t.cik is None and t.sic is None for t in (await s.execute(select(Ticker))).scalars()) -async def test_promote_conflict_reports_discrepancy_without_mutation(engine): - from app.services.sec_facts_parser import SnapshotRow - from app.services.sec_fundamentals_importer import StagedFundamentals - from app.services.sec_universe import ResolvedUniverse - +async def test_index_gap_over_45_days_loses_no_filings(engine): factory = _factory(engine) + await _seed(factory, ["AAPL"]) + backfill = FakeSecClient( + tickers={"AAPL": 320193}, + companyfacts={320193: _companyfacts([CF_K, CF_Q1], [SH_K, SH_Q1])}, + submissions={320193: _submissions(SUB_FILINGS)}, + latest_index=date(2026, 1, 31), + ) + await run_import(_importer(backfill), engine=engine) + + # 74-day gap; the filing sits in the OLD part (>45d before latest). + cf_q2 = _rev("2025-09-28", "2026-03-28", 254940, 2026, "Q2", "OLD") + sh_q2 = _shares("2026-04-17", 14687, "OLD", 2026, "Q2") + incr = FakeSecClient( + tickers={"AAPL": 320193}, + companyfacts={320193: _companyfacts([CF_K, CF_Q1, cf_q2], [SH_K, SH_Q1, sh_q2])}, + submissions={320193: _submissions(SUB_FILINGS + [ + _filing("OLD", "10-Q", "2026-03-28", "2026-02-10", "2026-02-10T10:01:00.000Z")])}, + latest_index=date(2026, 4, 15), + daily={date(2026, 2, 10): [{"form": "10-Q", "cik": 320193, "accession": "OLD"}]}, + ) + run = await run_import(_importer(incr, today=date(2026, 4, 16)), engine=engine) + assert run.status == STATUS_PROMOTED + assert await _count(factory, FundamentalSnapshot) == 3 # the old-gap filing was NOT lost + + +async def test_newly_added_issuer_backfills_without_filing(engine): + factory = _factory(engine) + await _seed(factory, ["AAPL"]) + backfill = FakeSecClient( + tickers={"AAPL": 320193}, + companyfacts={320193: _companyfacts([CF_K, CF_Q1], [SH_K, SH_Q1])}, + submissions={320193: _submissions(SUB_FILINGS)}, + latest_index=date(2026, 1, 31), + ) + await run_import(_importer(backfill), engine=engine) + + # MSFT added to the universe later; it did NOT file (not in the daily index). + await _seed(factory, ["MSFT"]) + msft_rev = _rev("2024-07-01", "2025-06-30", 270000, 2025, "FY", "M") + msft_sh = _shares("2025-07-15", 7400, "M", 2025, "FY") + incr = FakeSecClient( + tickers={"AAPL": 320193, "MSFT": 789019}, + companyfacts={ + 320193: _companyfacts([CF_K, CF_Q1], [SH_K, SH_Q1]), + 789019: _companyfacts([msft_rev], [msft_sh], cik=789019), + }, + submissions={ + 320193: _submissions(SUB_FILINGS), + 789019: {"cik": 789019, "sic": "7372", "sic_description": "Prepackaged Software", + "filings": [_filing("M", "10-K", "2025-06-30", "2025-07-30", "2025-07-30T10:00:00.000Z")]}, + }, + latest_index=date(2026, 2, 3), + daily={}, # MSFT did not file + ) + run = await run_import(_importer(incr, today=date(2026, 2, 4)), engine=engine) + assert run.status == STATUS_PROMOTED + async with factory() as s: + msft = (await s.execute( + select(FundamentalSnapshot).where(FundamentalSnapshot.cik == "0000789019") + )).scalars().all() + assert len(msft) == 1 and msft[0].revenue == 270000 # full-history backfill despite no filing + + +async def test_malformed_companyfacts_fails_validation(engine): + factory = _factory(engine) + await _seed(factory, ["AAPL", "MSFT"]) + client = FakeSecClient( + tickers={"AAPL": 320193, "MSFT": 789019}, + companyfacts={ + 320193: _companyfacts([CF_K, CF_Q1], [SH_K, SH_Q1]), + 789019: {"cik": 789019}, # malformed — no "facts" structure + }, + submissions={ + 320193: _submissions(SUB_FILINGS), + 789019: {"cik": 789019, "sic": "7372", "sic_description": "x", "filings": []}, + }, + latest_index=date(2026, 1, 31), + ) + run = await run_import(_importer(client), engine=engine) + assert run.status == STATUS_FAILED + assert "malformed" in (run.error_details or "") + assert await _count(factory, FundamentalSnapshot) == 0 # nothing promoted + + +async def test_discrepancy_in_shares_is_detected_and_reported(engine): + from app.models.system_event import SystemEvent utc = timezone.utc - async with factory() as s: # pre-existing immutable snapshot K (revenue 100, run 1) + factory = _factory(engine) + await _seed(factory, ["AAPL"]) + # Pre-store accession K matching what the parser will produce EXCEPT shares. + async with factory() as s: s.add(FundamentalSnapshot( cik="0000320193", accession="K", form="10-K", filed_date=date(2025, 10, 31), - accepted_at=datetime(2025, 10, 31, tzinfo=utc), period_end=date(2025, 9, 27), - fiscal_year=2025, fiscal_period="FY", revenue=100.0, import_run_id=1, - created_at=datetime(2025, 10, 31, tzinfo=utc))) + accepted_at=datetime(2025, 10, 31, 10, 1, 26, tzinfo=utc), period_start=date(2024, 9, 29), + period_end=date(2025, 9, 27), fiscal_year=2025, fiscal_period="FY", revenue=416161.0, + shares_outstanding=999.0, import_run_id=1, created_at=datetime(2025, 10, 31, tzinfo=utc))) await s.commit() - k_diff = SnapshotRow(cik="0000320193", accession="K", form="10-K", filed_date=date(2025, 10, 31), - accepted_at=datetime(2025, 10, 31, tzinfo=utc), period_end=date(2025, 9, 27), - fiscal_year=2025, fiscal_period="FY", revenue=999.0) # differs - n_new = SnapshotRow(cik="0000320193", accession="N", form="10-Q", filed_date=date(2026, 1, 30), - accepted_at=datetime(2026, 1, 30, tzinfo=utc), period_end=date(2025, 12, 27), - fiscal_year=2026, fiscal_period="Q1", revenue=143.0) - staged = StagedFundamentals(resolved=ResolvedUniverse(), rows=[k_diff, n_new]) + client = FakeSecClient( + tickers={"AAPL": 320193}, + companyfacts={320193: _companyfacts([CF_K, CF_Q1], [SH_K, SH_Q1])}, # SH_K = 14776 != 999 + submissions={320193: _submissions(SUB_FILINGS)}, + latest_index=date(2026, 1, 31), + ) + run = await run_import(_importer(client), engine=engine) - imp = SecFundamentalsImporter(client_factory=lambda: None) - async with factory() as s: - counts = await imp.promote(s, staged, run_id=2) - await s.commit() - - assert counts["inserted"] == 1 and counts["discrepancies"] == 1 + assert run.status == STATUS_PROMOTED # a discrepancy is reported, not a failure + assert '"discrepancy_count": 1' in (run.validation_json or "") + assert "shares_outstanding" in (run.validation_json or "") async with factory() as s: k = (await s.execute(select(FundamentalSnapshot).where(FundamentalSnapshot.accession == "K"))).scalar_one() - assert k.revenue == 100.0 and k.import_run_id == 1 # immutable — not overwritten - assert (await s.execute(select(func.count()).select_from(FundamentalSnapshot))).scalar_one() == 2 + assert k.shares_outstanding == 999.0 and k.import_run_id == 1 # immutable — not overwritten + events = (await s.execute(select(SystemEvent).where(SystemEvent.code == "snapshot_discrepancy"))).scalars().all() + assert len(events) == 1 and events[0].severity == "warning"