dennisthiessen/signal-platform
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
main
signal-platform/tests/unit/test_sec_fundamentals_importer.py
T
dennisthiessenandClaude Opus 5 83fe76c506 fix(sec): alert when a filing gap's reprieve lapses instead of re-pausing quietly 
An escalated gap stops pausing setups while the issuer's own fundamentals are
still recent. That reprieve ends on its own — the stored filings age past
GAP_GATE_RECENT_FILING_DAYS, or a newer gap arrives and the all-escalated
condition fails — and nothing reported either, because filing_gap_aged only
escalates gaps whose escalated_at is NULL and so never fires twice for the same
gap. For the 43 issuers behind the previous commit that lands around
2026-10-26, when their late-April filings age out together.

sec_filing_gaps.exempted_at (migration 034) makes the transition observable:
stamped quietly while the issuer is exempt, cleared when the exemption lapses,
and the clear is what raises filing_gap_repaused — once per lapse, re-arming if
the issuer's data recovers and ages out again. A gap that was never exempt has
no transition and stays silent; it is simply still paused, which
filing_gap_aged already said.

gap_exempt_ciks is public so the importer alerts on membership changes in
exactly the set the gate reads, rather than restating the rule.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_016Lo99z3jWqu9X9ueBU3Z3D
2026-08-21 17:44:04 +02:00

1467 lines
56 KiB
Python
Raw Permalink Blame History

"""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 json
import os
import tempfile
from datetime import date, datetime, timedelta, 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.sec_filing_gap import SecFilingGap
from app.models.system_event import SystemEvent
from app.models.ticker import Ticker
from app.services.data_import import (
STATUS_DEFERRED,
STATUS_FAILED,
STATUS_PROMOTED,
run_import,
)
from app.services.sec_fundamentals_importer import (
SecFundamentalsImporter,
StagedFundamentals,
)
from app.services.sec_universe import ResolvedUniverse
@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, cik=320193):
return {
"cik": cik,
"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_defers_without_alert_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_DEFERRED
# The gate blocks every later run until it clears, so run history still has
# to name the filing and say why it could not be resolved.
details = run.error_details or ""
assert "GHOST" in details and "not_in_companyfacts" in details
assert "2026-05-01" in details # index date the filing was seen on
summary = json.loads(run.validation_json or "{}")
assert summary["missing_xbrl_count"] == 1
assert summary["missing_xbrl"][0]["accession"] == "GHOST"
assert summary["missing_xbrl"][0]["form"] == "10-Q"
assert await _count(factory, FundamentalSnapshot) == 2 # nothing new written
assert await _count(factory, SystemEvent) == 0 # expected SEC lag does not alert
async def test_companyfacts_lag_does_not_mask_second_validation_failure():
importer = SecFundamentalsImporter(today=date(2026, 5, 3))
importer._latest_index_date = date(2026, 5, 2)
staged = StagedFundamentals(
resolved=ResolvedUniverse(),
missing_xbrl=[{
"cik": "0000320193",
"accession": "GHOST",
"form": "10-Q",
"index_date": date(2026, 5, 1),
"age_days": 2,
"reason": "not_in_companyfacts",
}],
invalid_payloads=[{
"cik": "0000789019",
"reason": "missing facts structure",
}],
)
result = await importer.validate(None, staged)
assert not result.ok
assert not result.retryable
assert len(result.messages) == 2
async def test_deferred_alert_names_aged_out_accessions_separately():
importer = SecFundamentalsImporter(today=date(2026, 5, 6))
importer._latest_index_date = date(2026, 5, 5)
staged = StagedFundamentals(
resolved=ResolvedUniverse(),
missing_xbrl=[
{
"cik": "0000320193",
"accession": "YOUNG",
"form": "10-Q",
"index_date": date(2026, 5, 5),
"age_days": 1,
"reason": "not_in_companyfacts",
},
{
"cik": "0000789019",
"accession": "AGED-OUT",
"form": "10-Q",
"index_date": date(2026, 5, 1),
"age_days": 5,
"reason": "not_in_companyfacts",
},
],
)
result = await importer.validate(None, staged)
assert result.retryable
assert len(result.messages) == 1 and "YOUNG" in result.messages[0]
assert len(result.deferred_alert_messages) == 1
assert "AGED-OUT" in result.deferred_alert_messages[0]
async def test_gate_separates_missing_submissions_from_missing_facts(engine):
"""An index row the issuer's own filing list does not carry is a different
failure from a Company-Facts lag, and must not be reported as one."""
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)}, # submissions never lists ORPHAN
latest_index=date(2026, 5, 2),
daily={date(2026, 5, 1): [{"form": "10-Q", "cik": 320193, "accession": "ORPHAN"}]},
)
run = await run_import(_importer(incr, today=date(2026, 5, 3)), engine=engine)
assert run.status == STATUS_FAILED
assert "ORPHAN" in (run.error_details or "")
assert "not_in_submissions" in (run.error_details or "")
summary = json.loads(run.validation_json or "{}")
assert summary["missing_xbrl"][0]["reason"] == "not_in_submissions"
def _coregistrant_client(share_fact):
"""Incremental client where Q2A's facts landed in co-registrant 99999's file
instead of the filer's own — the NEE-via-FPL / DOW-via-Dow-Chemical shape."""
cf_q2 = _rev("2025-09-28", "2026-03-28", 254940, 2026, "Q2", "Q2A")
return FakeSecClient(
tickers={"AAPL": 320193},
companyfacts={
320193: _companyfacts([CF_K, CF_Q1], [SH_K, SH_Q1]), # filer's own: no Q2A
99999: _companyfacts([cf_q2], [share_fact], cik=99999),
},
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): [
# one combined filing, listed by the index under both co-registrants
{"form": "10-Q", "cik": 320193, "accession": "Q2A"},
{"form": "10-Q", "cik": 99999, "accession": "Q2A"},
]},
)
async def test_recovers_facts_misfiled_under_coregistrant(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)
# 14687 shares is continuous with the issuer's own history (14681 last quarter).
incr = _coregistrant_client(_shares("2026-04-17", 14687, "Q2A", 2026, "Q2"))
run = await run_import(_importer(incr, today=date(2026, 5, 3)), engine=engine)
assert run.status == STATUS_PROMOTED
summary = json.loads(run.validation_json or "{}")
assert summary["recovered_count"] == 1
assert summary["recovered_from_coregistrant"][0]["source_cik"] == "0000099999"
assert summary["missing_xbrl_count"] == 0
async with factory() as s:
q2 = (await s.execute(
select(FundamentalSnapshot).where(FundamentalSnapshot.accession == "Q2A")
)).scalar_one()
codes = (await s.execute(select(SystemEvent.code))).scalars().all()
# Stamped to the issuer that filed, NOT the co-registrant whose file it came from.
assert q2.cik == "0000320193"
assert q2.revenue == 254940 and q2.shares_outstanding == 14687
assert "coregistrant_recovery" not in codes
async def test_coregistrant_recovery_rejects_discontinuous_share_count(engine):
"""A co-registrant shell's standalone facts must never be stored as the
parent's — a token float is the signature and it has to be refused."""
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 = _coregistrant_client(_shares("2026-04-17", 100, "Q2A", 2026, "Q2"))
run = await run_import(_importer(incr, today=date(2026, 5, 3)), engine=engine)
assert run.status == STATUS_FAILED
assert "coregistrant_facts_rejected" in (run.error_details or "")
assert await _count(factory, FundamentalSnapshot) == 2 # nothing recovered
async def test_unresolved_filing_stops_blocking_after_retry_window(engine):
"""A filing SEC has misfiled must not wedge every later import forever."""
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])}, # 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"}]},
)
# 9 days after the index date — well past the retry window.
run = await run_import(_importer(incr, today=date(2026, 5, 10)), engine=engine)
assert run.status == STATUS_PROMOTED # promoted around it, not blocked by it
assert run.source_max_date == date(2026, 5, 2) # and the index advances
summary = json.loads(run.validation_json or "{}")
assert summary["missing_xbrl_count"] == 1 and summary["missing_xbrl_blocking"] == 0
assert await _count(factory, SecFilingGap) == 1
async with factory() as s:
events = (await s.execute(select(SystemEvent))).scalars().all()
unresolved = [e for e in events if e.code == "unresolved_filing"]
assert len(unresolved) == 1 and "GHOST" in unresolved[0].message
assert "automatic SEC retry" in unresolved[0].message
# The next scheduled run retries even though the SEC daily-index revision
# has not changed. Company Facts is a separate SEC product and may catch up
# independently, so the generic revision no-op must not suppress this work.
still_missing_run = await run_import(
_importer(incr, today=date(2026, 5, 11)),
engine=engine,
)
assert still_missing_run.status == STATUS_PROMOTED
assert still_missing_run.revision is None
assert await _count(factory, SecFilingGap) == 1
# A later normal scheduled import retries only the queued issuer. Once SEC
# publishes the accession in Company Facts, it is inserted and unblocked
# without a full-universe reparse or operator action.
cf_ghost = _rev("2025-09-28", "2026-03-28", 254940, 2026, "Q2", "GHOST")
sh_ghost = _shares("2026-04-17", 14687, "GHOST", 2026, "Q2")
healed = FakeSecClient(
tickers={"AAPL": 320193},
companyfacts={
320193: _companyfacts(
[CF_K, CF_Q1, cf_ghost],
[SH_K, SH_Q1, sh_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),
)
healed_run = await run_import(
_importer(healed, today=date(2026, 5, 12)),
engine=engine,
)
assert healed_run.status == STATUS_PROMOTED
assert await _count(factory, FundamentalSnapshot) == 3
assert await _count(factory, SecFilingGap) == 0
async def test_queued_gap_without_index_date_retries_without_wedging_and_escalates_once(
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)
old = datetime(2026, 4, 1, tzinfo=timezone.utc)
async with factory() as db:
db.add(SecFilingGap(
cik="0000320193",
accession="DATELESS",
form="10-Q",
index_date=None,
reason="not_in_companyfacts",
first_seen_at=old,
last_attempted_at=old,
))
await db.commit()
missing = FakeSecClient(
tickers={"AAPL": 320193},
companyfacts={320193: _companyfacts([CF_K, CF_Q1], [SH_K, SH_Q1])},
submissions={320193: _submissions(SUB_FILINGS + [
_filing(
"DATELESS",
"10-Q",
"2026-03-28",
"2026-05-01",
"2026-05-01T10:01:00.000Z",
)
])},
latest_index=date(2026, 1, 31),
)
first = await run_import(
_importer(missing, today=date(2026, 5, 20)), engine=engine
)
second = await run_import(
_importer(missing, today=date(2026, 5, 21)), engine=engine
)
assert first.status == STATUS_PROMOTED
assert second.status == STATUS_PROMOTED
async with factory() as db:
gap = (await db.execute(select(SecFilingGap))).scalar_one()
events = (
await db.execute(
select(SystemEvent).where(SystemEvent.code == "filing_gap_aged")
)
).scalars().all()
assert gap.escalated_at is not None
assert len(events) == 1
async def test_queued_filing_reclassified_non_xbrl_is_removed(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)
now = datetime.now(timezone.utc)
async with factory() as db:
db.add(SecFilingGap(
cik="0000320193",
accession="NONX",
form="10-Q/A",
index_date=date(2026, 5, 1),
reason="not_in_companyfacts",
first_seen_at=now,
last_attempted_at=now,
))
await db.commit()
client = FakeSecClient(
tickers={"AAPL": 320193},
companyfacts={320193: _companyfacts([CF_K, CF_Q1], [SH_K, SH_Q1])},
submissions={320193: _submissions(SUB_FILINGS + [
_filing(
"NONX",
"10-Q/A",
"2026-03-28",
"2026-05-01",
"2026-05-01T10:01:00.000Z",
is_xbrl=False,
)
])},
latest_index=date(2026, 1, 31),
)
run = await run_import(_importer(client, today=date(2026, 5, 20)), engine=engine)
assert run.status == STATUS_PROMOTED
assert await _count(factory, SecFilingGap) == 0
async def test_queued_parser_skip_stays_blocked_with_actionable_reason(
engine, monkeypatch
):
from app.services.sec_facts_parser import ParseResult
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)
now = datetime.now(timezone.utc)
async with factory() as db:
db.add(SecFilingGap(
cik="0000320193",
accession="BADPARSE",
form="10-Q",
index_date=date(2026, 5, 1),
reason="not_in_companyfacts",
first_seen_at=now,
last_attempted_at=now,
))
await db.commit()
bad_fact = _rev(
"2025-09-28", "2026-03-28", 254940, 2026, "Q2", "BADPARSE"
)
bad_share = _shares("2026-04-17", 14687, "BADPARSE", 2026, "Q2")
client = FakeSecClient(
tickers={"AAPL": 320193},
companyfacts={
320193: _companyfacts(
[CF_K, CF_Q1, bad_fact], [SH_K, SH_Q1, bad_share]
)
},
submissions={320193: _submissions(SUB_FILINGS + [
_filing(
"BADPARSE",
"10-Q",
"2026-03-28",
"2026-05-01",
"2026-05-01T10:01:00.000Z",
)
])},
latest_index=date(2026, 1, 31),
)
def skip_parse(*args, **kwargs):
return ParseResult(skipped_filings=[{
"accession": "BADPARSE",
"reason": "unparseable",
}])
monkeypatch.setattr("app.services.sec_facts_parser.parse_snapshots", skip_parse)
run = await run_import(_importer(client, today=date(2026, 5, 20)), engine=engine)
assert run.status == STATUS_PROMOTED
async with factory() as db:
gap = (await db.execute(select(SecFilingGap))).scalar_one()
assert gap.reason == "parser_unusable"
async def test_new_parser_skip_gets_grace_then_enters_retry_queue(
engine, monkeypatch
):
from app.services.sec_facts_parser import ParseResult
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)
bad_fact = _rev(
"2025-09-28", "2026-03-28", 254940, 2026, "Q2", "NEWBAD"
)
bad_share = _shares("2026-04-17", 14687, "NEWBAD", 2026, "Q2")
client = FakeSecClient(
tickers={"AAPL": 320193},
companyfacts={
320193: _companyfacts(
[CF_K, CF_Q1, bad_fact], [SH_K, SH_Q1, bad_share]
)
},
submissions={320193: _submissions(SUB_FILINGS + [
_filing(
"NEWBAD",
"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": "NEWBAD",
}]
},
)
def skip_parse(*args, **kwargs):
return ParseResult(skipped_filings=[{
"accession": "NEWBAD",
"reason": "unparseable",
}])
monkeypatch.setattr("app.services.sec_facts_parser.parse_snapshots", skip_parse)
young = await run_import(
_importer(client, today=date(2026, 5, 3)), engine=engine
)
assert young.status == STATUS_DEFERRED
assert "parser_unusable" in (young.error_details or "")
assert await _count(factory, SecFilingGap) == 0
aged = await run_import(
_importer(client, today=date(2026, 5, 5)), engine=engine
)
assert aged.status == STATUS_PROMOTED
async with factory() as db:
gap = (await db.execute(select(SecFilingGap))).scalar_one()
assert gap.accession == "NEWBAD"
assert gap.reason == "parser_unusable"
async def test_validation_caps_details_but_keeps_complete_blocked_cik_set():
importer = SecFundamentalsImporter(today=date(2026, 5, 20))
importer._latest_index_date = date(2026, 5, 19)
staged = StagedFundamentals(
resolved=ResolvedUniverse(),
missing_xbrl=[
{
"cik": f"{i:010d}",
"accession": f"MISS-{i}",
"form": "10-Q",
"index_date": date(2026, 5, 1),
"age_days": 19,
"reason": "not_in_companyfacts",
}
for i in range(60)
],
no_xbrl_filings=[
{"cik": f"{i + 100:010d}", "name": f"New {i}"}
for i in range(60)
],
recovered=[
{"cik": f"{i:010d}", "accession": f"REC-{i}", "source_cik": "1"}
for i in range(60)
],
)
result = await importer.validate(None, staged)
assert len(result.summary["missing_xbrl"]) == 50
assert len(result.summary["no_xbrl_filings"]) == 50
assert len(result.summary["no_xbrl_ciks"]) == 60
assert len(result.summary["recovered_from_coregistrant"]) == 50
assert len(result.summary["setup_blocked_ciks"]) == 120
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_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")]},
},
# SAME index date as the prior run and no filing: only the universe
# fingerprint (MSFT added) changes the revision, so this proves the
# fingerprint alone prevents starvation.
latest_index=date(2026, 1, 31),
daily={}, # MSFT did not file
)
run = await run_import(_importer(incr, today=date(2026, 2, 1)), 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_missing_units_structure_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]),
# facts present, but a concept is missing its units mapping
789019: {"cik": 789019, "facts": {"us-gaap": {"Revenues": {"label": "Revenues"}}}},
},
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 "units" in (run.validation_json or "")
assert await _count(factory, FundamentalSnapshot) == 0
async def test_discrepancy_in_shares_is_detected_and_reported(engine):
from app.models.system_event import SystemEvent
utc = timezone.utc
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, 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()
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)
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.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"
# The alert has to say WHICH column moved: a differing cik is a co-registrant
# attribution, a differing revenue is our numbers changing.
assert "K (shares_outstanding, shares_outstanding_date)" in events[0].message
# --- reparse: rewriting rows a fixed parser reconstructs differently --------
# A 4-4-5 filer's YTD-Q3 span (36 weeks = 251 days). The old 20-day tolerance
# around 273 rejected it and stored revenue=None; 25 accepts it. Reparsing with
# the fixed parser is exactly the situation this mode exists for.
CF_Q3_445 = _rev("2025-09-01", "2026-05-10", 207431, 2026, "Q3", "Q3F")
SUB_445 = [_filing("Q3F", "10-Q", "2026-05-10", "2026-06-01", "2026-06-01T10:01:00.000Z")]
def _445_client():
return FakeSecClient(
tickers={"AAPL": 320193},
companyfacts={320193: _companyfacts([CF_Q3_445], [_shares("2026-05-15", 100, "Q3F", 2026, "Q3")])},
submissions={320193: _submissions(SUB_445)},
latest_index=date(2026, 6, 1),
)
async def _import_with_old_tolerance(engine, monkeypatch):
"""Seed the DB the way the pre-fix parser did: Q3 revenue rejected -> null."""
from app.services import sec_facts_parser
monkeypatch.setattr(sec_facts_parser, "_YTD_TOLERANCE_DAYS", 20)
run = await run_import(_importer(_445_client(), today=date(2026, 6, 2)), engine=engine)
monkeypatch.undo()
return run
async def test_reparse_rewrites_rows_the_fixed_parser_reads_differently(engine, monkeypatch):
factory = _factory(engine)
await _seed(factory, ["AAPL"])
first = await _import_with_old_tolerance(engine, monkeypatch)
async with factory() as s:
stale = (await s.execute(select(FundamentalSnapshot))).scalar_one()
assert stale.revenue is None, "precondition: the old parser stored a null"
# Reparse with the current (fixed) parser. force=True because SEC has not
# changed -- the staleness is on our side, so the revision gate would no-op.
run = await run_import(
SecFundamentalsImporter(
client_factory=lambda: _445_client(), today=date(2026, 6, 2), reparse=True
),
engine=engine,
force=True,
)
assert run.status == STATUS_PROMOTED
assert '"updated": 1' in run.row_counts_json
async with factory() as s:
fixed = (await s.execute(select(FundamentalSnapshot))).scalar_one()
assert fixed.revenue == 207431 # rewritten in place
assert fixed.accession == stale.accession
assert fixed.import_run_id == run.id # rewrite is attributable
assert fixed.import_run_id != first.id
async def test_reparse_leaves_unchanged_rows_untouched(engine):
factory = _factory(engine)
await _seed(factory, ["AAPL"])
first = await run_import(_importer(_445_client(), today=date(2026, 6, 2)), engine=engine)
run = await run_import(
SecFundamentalsImporter(
client_factory=lambda: _445_client(), today=date(2026, 6, 2), reparse=True
),
engine=engine,
force=True,
)
assert '"updated": 0' in run.row_counts_json
async with factory() as s:
row = (await s.execute(select(FundamentalSnapshot))).scalar_one()
assert row.import_run_id == first.id # provenance preserved, no needless rewrite
async def test_without_reparse_a_differing_row_stays_immutable(engine, monkeypatch):
"""The default contract is unchanged: report the discrepancy, never mutate."""
factory = _factory(engine)
await _seed(factory, ["AAPL"])
await _import_with_old_tolerance(engine, monkeypatch)
importer = SecFundamentalsImporter(
client_factory=lambda: _445_client(), today=date(2026, 6, 2), reparse=True
)
async with _factory(engine)() as db:
await importer.detect_revision(db)
staged = await importer.stage(db)
importer.reparse = False # same staged diff, default disposition
counts = await importer.promote(db, staged, run_id=999)
await db.commit()
assert staged.discrepancies, "the diff should still be detected and reported"
assert counts["updated"] == 0
async with factory() as s:
row = (await s.execute(select(FundamentalSnapshot))).scalar_one()
assert row.revenue is None # untouched
async def test_force_bypasses_the_unchanged_revision_no_op(engine):
factory = _factory(engine)
await _seed(factory, ["AAPL"])
await run_import(_importer(_445_client(), today=date(2026, 6, 2)), engine=engine)
same = _importer(_445_client(), today=date(2026, 6, 2))
assert (await run_import(same, engine=engine)).status == "no_op"
forced = SecFundamentalsImporter(
client_factory=lambda: _445_client(), today=date(2026, 6, 2), reparse=True
)
assert (await run_import(forced, engine=engine, force=True)).status == STATUS_PROMOTED
# --- CIK resolution: successor registrants with no filings -----------------
async def test_issuer_with_no_xbrl_filings_is_reported_not_silent(engine):
"""XOM resolved to CIK 2115436 'ExxonMobil Holdings Corp', which has zero
filings, so it produced no snapshots and nothing said why."""
factory = _factory(engine)
await _seed(factory, ["AAPL"])
client = FakeSecClient(
tickers={"AAPL": 320193},
companyfacts={320193: _companyfacts([CF_K], [SH_K])},
submissions={320193: {**_submissions([]), "name": "Shell Holdings Corp"}},
latest_index=date(2026, 1, 31),
)
importer = _importer(client)
async with factory() as db:
await importer.detect_revision(db)
staged = await importer.stage(db)
result = await importer.validate(db, staged)
assert result.summary["no_xbrl_filings_count"] == 1
assert staged.no_xbrl_filings[0]["cik"] == "0000320193"
assert staged.no_xbrl_filings[0]["name"] == "Shell Holdings Corp"
async def test_cik_override_pins_a_ticker_to_the_real_filer(engine):
from app.models.settings import SystemSetting
from app.services.sec_universe import CIK_OVERRIDES_KEY, resolve_ciks
factory = _factory(engine)
await _seed(factory, ["AAPL"])
async with factory() as s:
s.add(SystemSetting(key=CIK_OVERRIDES_KEY, value='{"AAPL": 34088}'))
await s.commit()
client = FakeSecClient(
tickers={"AAPL": 320193}, # SEC points at the wrong registrant
companyfacts={}, submissions={}, latest_index=date(2026, 1, 31),
)
async with factory() as db:
resolved = await resolve_ciks(db, client)
assert resolved.symbol_to_cik["AAPL"] == 34088
assert resolved.cik_updates == [(1, "0000034088")]
async def test_malformed_cik_override_is_ignored_not_fatal(engine):
from app.models.settings import SystemSetting
from app.services.sec_universe import CIK_OVERRIDES_KEY, resolve_ciks
factory = _factory(engine)
await _seed(factory, ["AAPL"])
async with factory() as s:
s.add(SystemSetting(key=CIK_OVERRIDES_KEY, value="not json at all"))
await s.commit()
client = FakeSecClient(
tickers={"AAPL": 320193}, companyfacts={}, submissions={},
latest_index=date(2026, 1, 31),
)
async with factory() as db:
resolved = await resolve_ciks(db, client)
assert resolved.symbol_to_cik["AAPL"] == 320193 # fell back to company_tickers
# --- aggregate deferral ceiling -------------------------------------------
def _blocking_staged(today: date) -> StagedFundamentals:
"""One filing still inside the per-filing retry window."""
return StagedFundamentals(
resolved=ResolvedUniverse(),
missing_xbrl=[{
"cik": "0000320193",
"accession": "YOUNG-1",
"form": "10-Q",
"index_date": today,
"age_days": 0,
"reason": "not_in_companyfacts",
}],
)
async def _add_run(factory, *, status: str, started_at: datetime) -> None:
from app.models.data_import_run import DataImportRun
async with factory() as db:
db.add(DataImportRun(
source="sec_facts", status=status, started_at=started_at,
))
await db.commit()
async def _validate_with_history(engine, *, promoted_days_ago: int | None):
factory = _factory(engine)
today = date(2026, 5, 20)
now = datetime(2026, 5, 20, 12, 0, tzinfo=timezone.utc)
if promoted_days_ago is not None:
await _add_run(
factory,
status=STATUS_PROMOTED,
started_at=now - timedelta(days=promoted_days_ago),
)
importer = SecFundamentalsImporter(today=today)
importer._latest_index_date = date(2026, 5, 19)
staged = _blocking_staged(today)
async with factory() as db:
return await importer.validate(db, staged), staged, importer
async def test_ceiling_forces_a_promotion_once_deferral_outlasts_it(engine):
"""The per-filing window bounds one filing; this bounds the whole import."""
result, staged, importer = await _validate_with_history(engine, promoted_days_ago=10)
assert result.ok is True
assert result.summary["missing_xbrl_blocking"] == 0
assert result.summary["promotion_ceiling_tripped"] == {"forced": 1, "unresolved": 1}
# Aged in place, so promote() re-derives the same verdict and queues it.
assert staged.missing_xbrl[0]["age_days"] > 3
# The symbol stays barred from setups — promoting is not trusting the data.
assert result.summary["setup_blocked_ciks"] == ["0000320193"]
async def test_a_recent_promotion_keeps_the_normal_block(engine):
result, staged, _ = await _validate_with_history(engine, promoted_days_ago=1)
assert result.ok is False
assert result.summary["missing_xbrl_blocking"] == 1
assert result.summary["promotion_ceiling_tripped"] is None
assert result.retryable is True
assert staged.missing_xbrl[0]["age_days"] == 0
async def test_ceiling_never_fires_before_a_first_promotion(engine):
"""No baseline means initial setup, not a wedge — forcing it through would
mask a misconfiguration instead of recovering from an SEC gap."""
result, _, _ = await _validate_with_history(engine, promoted_days_ago=None)
assert result.ok is False
assert result.summary["promotion_ceiling_tripped"] is None
async def test_ceiling_promotes_queues_and_alerts_end_to_end(engine, monkeypatch):
"""The self-heal claim, end to end: a filing that would block forever gets
promoted through, queued for retry, and announced."""
from app.models.data_import_run import DataImportRun
from app.services.sec_facts_parser import ParseResult
from sqlalchemy import update as sa_update
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),
)
assert (await run_import(_importer(backfill), engine=engine)).status == STATUS_PROMOTED
# Age the only promotion past the ceiling: this is the wedge the ceiling exists
# for — the filing below stays young, so nothing else would ever release it.
async with factory() as db:
await db.execute(
sa_update(DataImportRun)
.where(DataImportRun.source == "sec_facts")
.values(started_at=datetime(2026, 4, 20, tzinfo=timezone.utc))
)
await db.commit()
# Present in Company Facts but unparseable — one missing_xbrl entry, not the
# two an absent-from-facts accession would also raise.
stuck_fact = _rev("2025-09-28", "2026-03-28", 254940, 2026, "Q2", "STUCK")
stuck_share = _shares("2026-04-17", 14687, "STUCK", 2026, "Q2")
client = FakeSecClient(
tickers={"AAPL": 320193},
companyfacts={
320193: _companyfacts(
[CF_K, CF_Q1, stuck_fact], [SH_K, SH_Q1, stuck_share]
)
},
submissions={320193: _submissions(SUB_FILINGS + [
_filing("STUCK", "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": "STUCK"},
]},
)
monkeypatch.setattr(
"app.services.sec_facts_parser.parse_snapshots",
lambda *a, **k: ParseResult(
skipped_filings=[{"accession": "STUCK", "reason": "unparseable"}]
),
)
# index_date 2026-05-01 vs today 2026-05-03 => 2 days old, still inside the
# per-filing window, so only the aggregate ceiling can let this through.
run = await run_import(_importer(client, today=date(2026, 5, 3)), engine=engine)
assert run.status == STATUS_PROMOTED
summary = json.loads(run.validation_json)
assert summary["promotion_ceiling_tripped"] == {"forced": 1, "unresolved": 1}
async with factory() as db:
gap = (await db.execute(select(SecFilingGap))).scalar_one()
events = (
await db.execute(
select(SystemEvent).where(
SystemEvent.code == "promotion_ceiling_forced"
)
)
).scalars().all()
# Queued, so later runs retry it without it ever blocking again...
assert gap.accession == "STUCK"
# ...and the safety valve firing is visible, not silent.
assert len(events) == 1
assert events[0].severity == "warning"
assert "7 days" in events[0].message
# --- attribution collisions: two tracked CIKs claiming one filing ----------
# A REIT and its operating partnership co-file one 10-K, and SEC's
# company_tickers.json points the old symbol at the partnership (EQR ->
# ERP Operating LP) while the issuer itself trades under a new one (VMRK).
_COMBINED = [_filing("COMBINED-K", "10-K", "2025-12-31", "2026-02-13",
"2026-02-13T21:00:00.000Z")]
_CF_COMBINED = _rev("2025-01-01", "2025-12-31", 2900000, 2025, "FY", "COMBINED-K")
_SH_COMBINED = _shares("2026-02-01", 380000, "COMBINED-K", 2025, "FY")
def _reit_submissions(cik, tickers):
return {"cik": cik, "sic": "6798", "sic_description": "REIT",
"fiscal_year_end": "1231", "tickers": tickers, "filings": _COMBINED}
def _reit_client(tickers):
return FakeSecClient(
tickers=tickers,
companyfacts={
cik: _companyfacts([_CF_COMBINED], [_SH_COMBINED], cik=cik)
for cik in tickers.values()
},
submissions={
cik: _reit_submissions(cik, [sym]) for sym, cik in tickers.items()
},
latest_index=date(2026, 3, 1),
)
async def test_cik_collision_is_reported_as_attribution_not_discrepancy(engine):
"""Only `cik` differs, so nothing was re-parsed differently — the universe
resolves a co-registrant it should not track, and the alert must say that."""
factory = _factory(engine)
await _seed(factory, ["VMRK"])
run = await run_import(
_importer(_reit_client({"VMRK": 906107}), today=date(2026, 3, 2)), engine=engine
)
assert run.status == STATUS_PROMOTED
# The stale symbol is added, resolving to the partnership's CIK.
await _seed(factory, ["EQR"])
run = await run_import(
_importer(_reit_client({"VMRK": 906107, "EQR": 931182}), today=date(2026, 3, 2)),
engine=engine,
)
assert run.status == STATUS_PROMOTED
# Production's shape: a run-level incremental in which the untracked-until-now
# CIK is individually backfilled (run 63 recorded exactly this).
assert '"backfill": false' in (run.validation_json or "")
async with factory() as s:
rows = (await s.execute(select(FundamentalSnapshot))).scalars().all()
events = (await s.execute(select(SystemEvent))).scalars().all()
# The filing stays with the issuer that filed it, stored once.
assert [(r.accession, r.cik) for r in rows] == [("COMBINED-K", "0000906107")]
codes = {e.code for e in events}
assert "accession_cik_collision" in codes
assert "snapshot_discrepancy" not in codes # not a reconstruction change
collision = next(e for e in events if e.code == "accession_cik_collision")
assert "stored 0000906107, parsed 0000931182" in collision.message
assert "sec_cik_overrides" in collision.message # names the actual fix
async def test_reparse_never_restamps_a_collision_onto_the_co_registrant(engine):
"""A reparse rewrites rows a fixed parser reconstructs differently. A cik-only
difference is not that: rewriting would hand the filing to the co-registrant."""
from app.services.sec_facts_parser import SnapshotRow
factory = _factory(engine)
await _seed(factory, ["VMRK"])
assert (await run_import(
_importer(_reit_client({"VMRK": 906107}), today=date(2026, 3, 2)), engine=engine
)).status == STATUS_PROMOTED
importer = _importer(_reit_client({"VMRK": 906107}), today=date(2026, 3, 2))
importer.reparse = True
staged = StagedFundamentals(
resolved=ResolvedUniverse(),
rows=[SnapshotRow(
cik="0000931182", accession="COMBINED-K", form="10-K",
filed_date=date(2026, 2, 13),
accepted_at=datetime(2026, 2, 13, 21, tzinfo=timezone.utc),
period_end=date(2025, 12, 31), fiscal_year=2025, fiscal_period="FY",
)],
existing_accessions={"COMBINED-K"},
discrepancies=[{
"accession": "COMBINED-K", "fields": ["cik"],
"cik": "0000931182", "stored_cik": "0000906107",
}],
)
async with _factory(engine)() as db:
counts = await importer.promote(db, staged, run_id=999)
await db.commit()
assert counts["updated"] == 0
async with factory() as s:
row = (await s.execute(select(FundamentalSnapshot))).scalar_one()
assert row.cik == "0000906107" # still the issuer that filed it
# --- the reprieve ending: an exemption that lapses must not do so silently ---
def _stale_gap(cik, *, exempted: bool):
now = datetime.now(timezone.utc)
return SecFilingGap(
cik=cik, accession=f"{cik}-AGED-Q", form="10-Q",
index_date=(now - timedelta(days=30)).date(), reason="not_in_companyfacts",
first_seen_at=now - timedelta(days=30), last_attempted_at=now,
escalated_at=now - timedelta(days=16),
exempted_at=(now - timedelta(days=16)) if exempted else None,
)
def _snapshot(cik, *, age_days):
filed = date.today() - timedelta(days=age_days)
return FundamentalSnapshot(
cik=cik, accession=f"{cik}-PRIOR", form="10-Q", filed_date=filed,
accepted_at=datetime.now(timezone.utc) - timedelta(days=age_days),
period_end=filed, fiscal_year=filed.year, fiscal_period="Q1",
)
async def _promote_only(engine, seed):
"""Run promote() alone against seeded gap/snapshot state."""
factory = _factory(engine)
async with factory() as s:
for obj in seed:
s.add(obj)
await s.commit()
importer = _importer(FakeSecClient(
tickers={}, companyfacts={}, submissions={}, latest_index=date(2026, 3, 1)
))
async with factory() as db:
await importer.promote(db, StagedFundamentals(resolved=ResolvedUniverse()), run_id=77)
await db.commit()
async with factory() as s:
gaps = (await s.execute(select(SecFilingGap))).scalars().all()
events = (await s.execute(select(SystemEvent))).scalars().all()
return gaps, events
async def test_a_lapsed_exemption_raises_its_own_alert(engine):
"""filing_gap_aged fires once and never again, so nothing else would say the
pause came back when the issuer's own fundamentals aged out."""
cik = "0000000060"
gaps, events = await _promote_only(
engine, [_stale_gap(cik, exempted=True), _snapshot(cik, age_days=400)]
)
repaused = [e for e in events if e.code == "filing_gap_repaused"]
assert len(repaused) == 1
assert f"{cik}/{cik}-AGED-Q" in repaused[0].message
# Cleared, so a later recovery can re-arm and lapse again.
assert gaps[0].exempted_at is None
async def test_an_exemption_taking_effect_is_stamped_silently(engine):
"""Setups resuming is what filing_gap_aged already described — stamping the
state must not raise a second alert for it."""
cik = "0000000061"
gaps, events = await _promote_only(
engine, [_stale_gap(cik, exempted=False), _snapshot(cik, age_days=120)]
)
assert [e.code for e in events if e.code.startswith("filing_gap")] == []
assert gaps[0].exempted_at is not None
async def test_a_still_paused_gap_is_not_reported_as_lapsing(engine):
"""It never became exempt, so there is no transition to report."""
cik = "0000000062"
gaps, events = await _promote_only(
engine, [_stale_gap(cik, exempted=False), _snapshot(cik, age_days=400)]
)
assert [e for e in events if e.code == "filing_gap_repaused"] == []
assert gaps[0].exempted_at is None
Reference in New Issue View Git Blame Copy Permalink