diff --git a/app/services/sec_facts_parser.py b/app/services/sec_facts_parser.py index 8052a9a..99239cc 100644 --- a/app/services/sec_facts_parser.py +++ b/app/services/sec_facts_parser.py @@ -11,17 +11,24 @@ The load-bearing rules (design Decision 2 + review): - 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. +- 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. - 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 +from dataclasses import dataclass, field from datetime import date, datetime from typing import Any, NamedTuple @@ -107,34 +114,49 @@ class FilingMeta: 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], -) -> tuple[list[SnapshotRow], list[dict[str, str]]]: +) -> ParseResult: """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. + ``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) - rows: list[SnapshotRow] = [] - skips: list[dict[str, str]] = [] + result = ParseResult() 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"}) + result.skipped_filings.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"}) + result.skipped_filings.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 + 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]]: diff --git a/app/services/sec_fundamentals_importer.py b/app/services/sec_fundamentals_importer.py new file mode 100644 index 0000000..d494335 --- /dev/null +++ b/app/services/sec_fundamentals_importer.py @@ -0,0 +1,343 @@ +"""SEC fundamentals importer (workstream A, phase A3). + +A ``SourceImporter`` (see ``app/services/data_import.py``) that populates the +immutable ``fundamental_snapshots`` from SEC Company Facts and back-fills +``tickers.cik/sic/sic_description``. EDGAR-daily-index driven: it fetches +companyfacts only for tracked issuers that filed since the last run (full-history +backfill on the first run / for newly-added issuers). Shadow only — nothing reads +snapshots until A4. + +Guardrails (design + reviews): +- ``detect_revision`` caches the resolved universe + the exact tracked index rows + and composes the revision from them; ``stage`` consumes those same cached inputs + (it does not refetch the index/universe) so promoted data always matches the + computed revision. +- Resolution is read-only in ``stage`` (proposals only); ticker writes happen in + ``promote`` via ``apply_ticker_updates``. +- ``validate`` runs the **index↔Company-Facts consistency gate** before any write: + a tracked XBRL index accession missing from Company Facts fails the run (the two + are separate SEC products that can lag) so we retry rather than record a + null/partial snapshot. Non-XBRL amendments are skipped with a recorded reason. +- ``promote`` inserts snapshots ``ON CONFLICT (accession) DO NOTHING`` (immutable), + reports differing existing accessions, and applies ticker updates in the same + transaction. +""" + +from __future__ import annotations + +import logging +from collections import defaultdict +from dataclasses import dataclass, field +from datetime import date, datetime, timedelta, timezone +from typing import Any, Callable + +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.services import sec_facts_parser as parser +from app.services import sec_universe +from app.services.data_import import STATUS_PROMOTED, ValidationResult +from app.services.sec_client import SecClient, SecError, cik10 +from app.services.sec_facts_parser import FilingMeta, SnapshotRow +from app.services.sec_universe import ResolvedUniverse + +logger = logging.getLogger(__name__) + +SOURCE = "sec_facts" +_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", + "period_end", "fiscal_year", "fiscal_period", "revenue", "net_income", + "operating_income", "diluted_eps", "cfo", "capex", "depreciation_amortization", + "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") + + +@dataclass +class StagedFundamentals: + resolved: ResolvedUniverse + sic_updates: list[tuple[int, str | None, str | None]] = field(default_factory=list) + rows: list[SnapshotRow] = field(default_factory=list) + skipped_filings: list[dict[str, str]] = field(default_factory=list) + 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) + backfill: bool = False + issuers_fetched: int = 0 + issuers_with_rows: int = 0 + + +def _now() -> datetime: + return datetime.now(timezone.utc) + + +class SecFundamentalsImporter: + source = SOURCE + + def __init__( + self, + *, + client_factory: Callable[[], SecClient] | None = None, + today: date | None = None, + ) -> None: + self._client_factory = client_factory or (lambda: SecClient()) + self.today = today or _now().date() + # cached by detect_revision, consumed by stage: + self._resolved: ResolvedUniverse | None = None + self._index_rows: list[dict[str, Any]] = [] + self._latest_index_date: date | None = None + self._backfill = False + + # -- SourceImporter protocol ------------------------------------------- + + async def detect_revision(self, db) -> str | None: + async with self._client_factory() as client: + self._resolved = await sec_universe.resolve_ciks(db, client) + last_processed = await self._last_processed_index_date(db) + self._latest_index_date = await client.latest_index_date(self.today) + if self._latest_index_date is None: + raise SecError("no EDGAR daily index available") + if last_processed is None: + self._backfill = True + self._index_rows = [] + else: + self._backfill = False + self._index_rows = await self._collect_index_rows( + client, last_processed, self._latest_index_date + ) + content = sec_universe.index_content_hash(self._index_rows) + return sec_universe.compose_revision( + self._latest_index_date, content, self._resolved.symbol_to_cik + ) + + async def stage(self, db) -> StagedFundamentals: + assert self._resolved is not None, "detect_revision must run first" + resolved = self._resolved + staged = StagedFundamentals(resolved=resolved, backfill=self._backfill) + + cik_to_tids = resolved.cik_to_ticker_ids + filed_by_cik: dict[int, list[str]] = defaultdict(list) + for r in self._index_rows: + if r["cik"] in cik_to_tids: + filed_by_cik[r["cik"]].append(r["accession"]) + + existing = await self._ciks_with_snapshots(db, set(cik_to_tids)) + if self._backfill: + backfill_ciks = set(cik_to_tids) + else: + # Newly added issuers (resolved but no snapshots yet) get a full-history + # backfill; issuers that already have history are handled incrementally. + backfill_ciks = {c for c in cik_to_tids if c not in existing} + incremental_ciks = set(filed_by_cik) - backfill_ciks + + async with self._client_factory() as client: + 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) + return staged + + async def _stage_issuer(self, client, cik, is_backfill, filed_by_cik, staged) -> None: + cf = await client.companyfacts(cik) + sub = await client.submissions(cik, include_history=is_backfill) + xbrl_meta, nonxbrl = _filing_meta(sub) + + if is_backfill: + accns = set(xbrl_meta) + else: + present = parser.companyfacts_accessions(cf) + accns = set() + for accn in filed_by_cik.get(cik, []): + if accn in nonxbrl: + staged.skipped_non_xbrl.append({"cik": cik10(cik), "accession": accn}) + elif accn in xbrl_meta and accn in present: + accns.add(accn) + else: + # XBRL (or unknown) filing not yet in Company Facts → the products + # have lagged; fail+retry rather than record nothing for it. + staged.missing_xbrl.append({"cik": cik10(cik), "accession": accn}) + + result = parser.parse_snapshots(cf, xbrl_meta, accns) + staged.rows.extend(result.rows) + staged.skipped_filings.extend(result.skipped_filings) + staged.field_issues.extend(result.field_issues) + staged.issuers_fetched += 1 + if result.rows: + staged.issuers_with_rows += 1 + + # SIC proposal for this issuer's tickers (read-only; applied in promote). + sic = str(sub["sic"]) if sub.get("sic") else None + desc = sub.get("sic_description") + for tid in staged.resolved.cik_to_ticker_ids.get(cik, []): + staged.sic_updates.append((tid, sic, desc)) + + async def validate(self, db, staged: StagedFundamentals) -> ValidationResult: + messages: list[str] = [] + + # Consistency gate — before any write. + if staged.missing_xbrl: + messages.append( + f"{len(staged.missing_xbrl)} tracked XBRL filing(s) not yet in " + "Company Facts (index/facts lag) — retry" + ) + + accns = [r.accession for r in staged.rows] + if len(accns) != len(set(accns)): + messages.append("duplicate accession in staged snapshots") + + if staged.backfill: + n_issuers = len(staged.resolved.cik_to_ticker_ids) + coverage = staged.issuers_with_rows / n_issuers if n_issuers else 0.0 + if coverage < MIN_BACKFILL_COVERAGE: + messages.append( + f"backfill coverage {coverage:.0%} < {MIN_BACKFILL_COVERAGE:.0%}" + ) + + summary = { + "backfill": staged.backfill, + "issuers_fetched": staged.issuers_fetched, + "issuers_with_rows": staged.issuers_with_rows, + "snapshot_rows": len(staged.rows), + "skipped_filings": len(staged.skipped_filings), + "field_issues": len(staged.field_issues), + "skipped_non_xbrl": len(staged.skipped_non_xbrl), + "missing_xbrl": len(staged.missing_xbrl), + "cik_updates": len(staged.resolved.cik_updates), + } + return ValidationResult( + ok=not messages, + summary=summary, + source_max_date=self._latest_index_date, + messages=messages, + ) + + 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 + + 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, + **ticker_counts, + } + + # -- helpers ----------------------------------------------------------- + + async def _last_processed_index_date(self, db) -> date | None: + return ( + await db.execute( + select(DataImportRun.source_max_date) + .where(DataImportRun.source == SOURCE, DataImportRun.status == STATUS_PROMOTED) + .order_by(DataImportRun.id.desc()) + .limit(1) + ) + ).scalar_one_or_none() + + async def _collect_index_rows( + self, client: SecClient, last_processed: date, latest: date + ) -> list[dict[str, Any]]: + 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) + rows: list[dict[str, Any]] = [] + day = start + while day <= latest: + for r in await client.daily_index(day): + if r["form"] in _XBRL_FORMS and r["cik"] in tracked: + rows.append(r) + day += timedelta(days=1) + return rows + + async def _ciks_with_snapshots(self, db, ciks: set[int]) -> set[int]: + if not ciks: + return set() + cik_strs = [cik10(c) for c in ciks] + found = ( + await db.execute( + select(FundamentalSnapshot.cik) + .where(FundamentalSnapshot.cik.in_(cik_strs)) + .distinct() + ) + ).scalars().all() + return {int(c) for c in found} + + async def _existing_by_accession(self, db, accessions: list[str]) -> dict[str, FundamentalSnapshot]: + if not accessions: + return {} + rows = ( + await db.execute( + select(FundamentalSnapshot).where(FundamentalSnapshot.accession.in_(accessions)) + ) + ).scalars().all() + return {r.accession: r for r in rows} + + +def _filing_meta(sub: dict[str, Any]) -> tuple[dict[str, FilingMeta], set[str]]: + """(xbrl_meta, nonxbrl_accessions) from a submissions payload. xbrl_meta only + includes 10-K/10-Q(/A) filings that are XBRL and have full period metadata.""" + xbrl: dict[str, FilingMeta] = {} + nonxbrl: set[str] = set() + for f in sub.get("filings", []): + if f["form"] not in _XBRL_FORMS: + continue + if not f.get("is_xbrl"): + nonxbrl.add(f["accession"]) + continue + if not (f.get("report_date") and f.get("filing_date") and f.get("acceptance_datetime")): + continue + xbrl[f["accession"]] = FilingMeta( + report_date=date.fromisoformat(f["report_date"]), + filing_date=date.fromisoformat(f["filing_date"]), + accepted_at=_parse_dt(f["acceptance_datetime"]), + form=f["form"], + ) + return xbrl, nonxbrl + + +def _parse_dt(value: str) -> datetime: + return datetime.fromisoformat(value.replace("Z", "+00:00")) + + +def _row_values(row: SnapshotRow, run_id: int) -> dict[str, Any]: + values = {col: getattr(row, col) for col in _SNAPSHOT_COLS} + values["import_run_id"] = run_id + values["created_at"] = _now() + return values + + +def _differs(row: SnapshotRow, old: FundamentalSnapshot) -> bool: + return any(getattr(row, col) != getattr(old, col) for col in _DISCREPANCY_COLS) diff --git a/tests/unit/test_sec_facts_parser.py b/tests/unit/test_sec_facts_parser.py index df5e380..90c4df2 100644 --- a/tests/unit/test_sec_facts_parser.py +++ b/tests/unit/test_sec_facts_parser.py @@ -66,9 +66,9 @@ def _by_accn(rows): def test_parses_ytd_not_discrete_and_cover_date_shares(): - rows, skips = parse_snapshots(COMPANYFACTS, FILINGS, {"A", "B"}) - assert not skips - b = _by_accn(rows)["B"] + res = parse_snapshots(COMPANYFACTS, FILINGS, {"A", "B"}) + assert not res.skipped_filings and not res.field_issues + b = _by_accn(res.rows)["B"] assert (b.cik, b.fiscal_year, b.fiscal_period) == ("0000320193", 2026, "Q2") assert b.period_end == date(2026, 3, 28) @@ -88,8 +88,8 @@ def test_parses_ytd_not_discrete_and_cover_date_shares(): def test_q1_discrete_is_the_ytd(): - rows, _ = parse_snapshots(COMPANYFACTS, FILINGS, {"A"}) - a = _by_accn(rows)["A"] + res = parse_snapshots(COMPANYFACTS, FILINGS, {"A"}) + a = _by_accn(res.rows)["A"] assert a.fiscal_period == "Q1" assert a.revenue == 143756 # Q1 YTD == Q1 discrete assert a.period_start == date(2025, 9, 28) @@ -103,15 +103,15 @@ def test_skips_filing_without_usable_period(): ]}}}}, } filings = {"X": FilingMeta(date(2026, 3, 28), date(2026, 5, 1), datetime(2026, 5, 1, tzinfo=UTC), "10-Q")} - rows, skips = parse_snapshots(cf, filings, {"X"}) - assert rows == [] - assert skips == [{"accession": "X", "reason": "no usable period identity"}] + res = parse_snapshots(cf, filings, {"X"}) + assert res.rows == [] + assert res.skipped_filings == [{"accession": "X", "reason": "no usable period identity"}] def test_missing_accession_is_skipped(): - rows, skips = parse_snapshots(COMPANYFACTS, FILINGS, {"NOPE"}) - assert rows == [] - assert skips == [{"accession": "NOPE", "reason": "no facts or filing metadata"}] + res = parse_snapshots(COMPANYFACTS, FILINGS, {"NOPE"}) + assert res.rows == [] + assert res.skipped_filings == [{"accession": "NOPE", "reason": "no facts or filing metadata"}] def test_debt_prefers_aggregate_over_parts(): @@ -186,8 +186,9 @@ def test_conflicting_context_skips_row(): "NetIncomeLoss": {"units": {"USD": [_dur("2025-09-28", "2026-03-28", 2, "B", fy=2025, fp="Q3")]}}, }}} filings = {"B": FilingMeta(RD, date(2026, 5, 1), datetime(2026, 5, 1, tzinfo=UTC), "10-Q")} - rows, skips = parse_snapshots(cf, filings, {"B"}) - assert rows == [] and skips == [{"accession": "B", "reason": "no usable period identity"}] + res = parse_snapshots(cf, filings, {"B"}) + assert res.rows == [] + assert res.skipped_filings == [{"accession": "B", "reason": "no usable period identity"}] def test_foreign_taxonomy_and_malformed_facts_ignored(): @@ -206,10 +207,10 @@ def test_foreign_taxonomy_and_malformed_facts_ignored(): ]}}}, }} filings = {"B": FilingMeta(RD, date(2026, 5, 1), datetime(2026, 5, 1, tzinfo=UTC), "10-Q")} - rows, _ = parse_snapshots(cf, filings, {"B"}) - assert rows[0].revenue == 500 # us-gaap Revenues, not the acme concept - assert rows[0].net_income is None # val None ignored - assert rows[0].operating_income is None # NaN ignored + res = parse_snapshots(cf, filings, {"B"}) + assert res.rows[0].revenue == 500 # us-gaap Revenues, not the acme concept + assert res.rows[0].net_income is None # val None ignored + assert res.rows[0].operating_income is None # NaN ignored def test_ambiguous_shares_produces_row_plus_note(): @@ -221,9 +222,10 @@ def test_ambiguous_shares_produces_row_plus_note(): ]}}}, }} filings = {"B": FilingMeta(RD, date(2026, 5, 1), datetime(2026, 5, 1, tzinfo=UTC), "10-Q")} - rows, skips = parse_snapshots(cf, filings, {"B"}) - assert len(rows) == 1 and rows[0].shares_outstanding is None # row kept, shares null - assert {"accession": "B", "reason": "ambiguous shares outstanding"} in skips + res = parse_snapshots(cf, filings, {"B"}) + assert len(res.rows) == 1 and res.rows[0].shares_outstanding is None # row kept, shares null + assert res.field_issues == [{"accession": "B", "reason": "ambiguous shares outstanding"}] + assert res.skipped_filings == [] # a field issue is NOT a skipped filing # Opt-in live check against real Apple companyfacts. Skips unless SEC_LIVE=1 and a @@ -252,7 +254,7 @@ async def test_live_apple_parse_invariants(): for f in sub["filings"] if f["report_date"] and f["filing_date"] and f["acceptance_datetime"] } - rows, _ = parse_snapshots(cf, filings, set(filings)) + rows = parse_snapshots(cf, filings, set(filings)).rows assert len(rows) > 20 assert all(r.period_end and r.fiscal_year and r.fiscal_period for r in rows) # YTD revenue is non-decreasing within a fiscal year diff --git a/tests/unit/test_sec_fundamentals_importer.py b/tests/unit/test_sec_fundamentals_importer.py new file mode 100644 index 0000000..67306c6 --- /dev/null +++ b/tests/unit/test_sec_fundamentals_importer.py @@ -0,0 +1,304 @@ +"""Integration tests for the SEC fundamentals importer, driven through the real +import framework with a fake SEC client (no network).""" + +from __future__ import annotations + +import os +import tempfile +from datetime import date, datetime, timezone + +import pytest +from sqlalchemy import func, select +from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine + +from app.database import Base +import app.models # noqa: F401 +from app.models.fundamental_snapshot import FundamentalSnapshot +from app.models.ticker import Ticker +from app.services.data_import import STATUS_FAILED, STATUS_PROMOTED, run_import +from app.services.sec_fundamentals_importer import SecFundamentalsImporter + + +@pytest.fixture +async def engine(): + fd, path = tempfile.mkstemp(suffix=".db") + os.close(fd) + eng = create_async_engine(f"sqlite+aiosqlite:///{path}") + async with eng.begin() as conn: + await conn.run_sync(Base.metadata.create_all) + try: + yield eng + finally: + await eng.dispose() + try: + os.unlink(path) + except OSError: + pass + + +def _factory(engine): + return async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False) + + +# --- fixture SEC data (AAPL, cik 320193) ----------------------------------- + +def _rev(start, end, val, fy, fp, accn): + return {"start": start, "end": end, "val": val, "fy": fy, "fp": fp, "accn": accn, "form": "10-K"} + + +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): + return { + "cik": 320193, + "facts": { + "us-gaap": {"RevenueFromContractWithCustomerExcludingAssessedTax": {"units": {"USD": rev_facts}}}, + "dei": {"EntityCommonStockSharesOutstanding": {"units": {"shares": share_facts}}}, + }, + } + + +def _filing(accn, form, report, filed, accepted, is_xbrl=True): + return {"accession": accn, "form": form, "report_date": report, "filing_date": filed, + "acceptance_datetime": accepted, "is_xbrl": is_xbrl} + + +CF_K = _rev("2024-09-29", "2025-09-27", 416161, 2025, "FY", "K") +CF_Q1 = _rev("2025-09-28", "2025-12-27", 143756, 2026, "Q1", "Q") +SH_K = _shares("2025-10-17", 14776, "K", 2025, "FY") +SH_Q1 = _shares("2026-01-16", 14681, "Q", 2026, "Q1") + +SUB_FILINGS = [ + _filing("K", "10-K", "2025-09-27", "2025-10-31", "2025-10-31T10:01:26.000Z"), + _filing("Q", "10-Q", "2025-12-27", "2026-01-30", "2026-01-30T11:01:00.000Z"), +] + + +def _submissions(filings): + return {"cik": 320193, "sic": "3571", "sic_description": "Electronic Computers", + "fiscal_year_end": "0926", "tickers": ["AAPL"], "filings": filings} + + +class FakeSecClient: + def __init__(self, *, tickers, companyfacts, submissions, latest_index, daily=None): + self._tickers = tickers + self._cf = companyfacts + self._sub = submissions + self._latest = latest_index + self._daily = daily or {} + + async def __aenter__(self): + return self + + async def __aexit__(self, *a): + return False + + async def company_tickers(self): + return dict(self._tickers) + + async def latest_index_date(self, today=None): + return self._latest + + async def daily_index(self, day): + return list(self._daily.get(day, [])) + + async def companyfacts(self, cik): + return self._cf[int(cik)] + + async def submissions(self, cik, *, include_history=False): + return self._sub[int(cik)] + + +def _importer(client, today=date(2026, 2, 1)): + return SecFundamentalsImporter(client_factory=lambda: client, today=today) + + +async def _seed(factory, symbols): + async with factory() as s: + for sym in symbols: + s.add(Ticker(symbol=sym)) + await s.commit() + + +async def _count(factory, model): + async with factory() as s: + return (await s.execute(select(func.count()).select_from(model))).scalar_one() + + +# --------------------------------------------------------------------------- + + +async def test_backfill_inserts_snapshots_and_ticker_meta(engine): + factory = _factory(engine) + await _seed(factory, ["AAPL"]) + client = 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), + ) + run = await run_import(_importer(client), engine=engine) + + assert run.status == STATUS_PROMOTED + assert run.source_max_date == date(2026, 1, 31) + assert await _count(factory, FundamentalSnapshot) == 2 + + async with factory() as s: + t = (await s.execute(select(Ticker))).scalar_one() + assert t.cik == "0000320193" and t.sic == "3571" + snaps = (await s.execute(select(FundamentalSnapshot))).scalars().all() + assert {x.fiscal_period for x in snaps} == {"FY", "Q1"} + assert all(x.import_run_id == run.id for x in snaps) + fy = next(x for x in snaps if x.fiscal_period == "FY") + assert fy.revenue == 416161 and fy.shares_outstanding == 14776 + + +async def test_incremental_adds_only_new_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) + assert await _count(factory, FundamentalSnapshot) == 2 + + # A new Q2 10-Q appears in the daily index and Company Facts. + cf_q2 = _rev("2025-09-28", "2026-03-28", 254940, 2026, "Q2", "Q2A") + sh_q2 = _shares("2026-04-17", 14687, "Q2A", 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("Q2A", "10-Q", "2026-03-28", "2026-05-01", "2026-05-01T10:01:00.000Z")])}, + latest_index=date(2026, 5, 2), + daily={date(2026, 5, 1): [{"form": "10-Q", "cik": 320193, "accession": "Q2A"}]}, + ) + run = await run_import(_importer(incr, today=date(2026, 5, 3)), engine=engine) + + assert run.status == STATUS_PROMOTED + assert await _count(factory, FundamentalSnapshot) == 3 # only Q2A added + async with factory() as s: + q2 = (await s.execute( + select(FundamentalSnapshot).where(FundamentalSnapshot.accession == "Q2A") + )).scalar_one() + assert q2.fiscal_period == "Q2" and q2.revenue == 254940 + + +async def test_consistency_gate_fails_when_facts_lag_index(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) + + # Index + submissions list an XBRL filing "GHOST" that Company Facts lacks. + incr = FakeSecClient( + tickers={"AAPL": 320193}, + companyfacts={320193: _companyfacts([CF_K, CF_Q1], [SH_K, SH_Q1])}, # no GHOST + submissions={320193: _submissions(SUB_FILINGS + [ + _filing("GHOST", "10-Q", "2026-03-28", "2026-05-01", "2026-05-01T10:01:00.000Z")])}, + latest_index=date(2026, 5, 2), + daily={date(2026, 5, 1): [{"form": "10-Q", "cik": 320193, "accession": "GHOST"}]}, + ) + run = await run_import(_importer(incr, today=date(2026, 5, 3)), engine=engine) + + assert run.status == STATUS_FAILED + assert "Company Facts" in (run.error_details or "") + assert await _count(factory, FundamentalSnapshot) == 2 # nothing new written + + +async def test_non_xbrl_amendment_skipped_not_failed(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) + + incr = FakeSecClient( + tickers={"AAPL": 320193}, + companyfacts={320193: _companyfacts([CF_K, CF_Q1], [SH_K, SH_Q1])}, + submissions={320193: _submissions(SUB_FILINGS + [ + _filing("AMD", "10-K/A", "2025-09-27", "2026-05-01", "2026-05-01T10:01:00.000Z", is_xbrl=False)])}, + latest_index=date(2026, 5, 2), + daily={date(2026, 5, 1): [{"form": "10-K/A", "cik": 320193, "accession": "AMD"}]}, + ) + run = await run_import(_importer(incr, today=date(2026, 5, 3)), engine=engine) + + assert run.status == STATUS_PROMOTED # non-XBRL amendment is skipped, not a failure + assert "skipped_non_xbrl" in (run.validation_json or "") + assert await _count(factory, FundamentalSnapshot) == 2 + + +async def test_failed_backfill_leaves_tickers_unwritten(engine): + factory = _factory(engine) + await _seed(factory, ["AAPL", "MSFT", "NVDA"]) # 3 resolve, only AAPL yields rows + client = FakeSecClient( + tickers={"AAPL": 320193, "MSFT": 789019, "NVDA": 1045810}, + companyfacts={ + 320193: _companyfacts([CF_K, CF_Q1], [SH_K, SH_Q1]), + 789019: {"cik": 789019, "facts": {}}, # no facts -> no rows + 1045810: {"cik": 1045810, "facts": {}}, + }, + submissions={ + 320193: _submissions(SUB_FILINGS), + 789019: {"cik": 789019, "sic": "7372", "sic_description": "x", "filings": []}, + 1045810: {"cik": 1045810, "sic": "3674", "sic_description": "y", "filings": []}, + }, + latest_index=date(2026, 1, 31), + ) + run = await run_import(_importer(client), engine=engine) + + assert run.status == STATUS_FAILED # coverage 1/3 < 50% + assert "coverage" in (run.error_details or "") + assert await _count(factory, FundamentalSnapshot) == 0 + # read-only resolution: no ticker cik/sic written on a failed run + async with factory() as s: + 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 + + factory = _factory(engine) + utc = timezone.utc + async with factory() as s: # pre-existing immutable snapshot K (revenue 100, run 1) + 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))) + 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]) + + 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 + 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