fix(sec): A3 slice-2b review — no index cap, full discrepancy + malformed gate

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 <noreply@anthropic.com>

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.
     """

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"])
+            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)]

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

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"