fix(sec): name aged filings in deferred warnings
Deploy / lint (push) Successful in 10s
Deploy / test (push) Successful in 2m3s
Deploy / deploy (push) Successful in 42s

This commit is contained in:
2026-07-31 14:58:22 +02:00
parent c8c660e63d
commit 7bcdf77ef9
4 changed files with 96 additions and 17 deletions
+28 -13
View File
@@ -33,7 +33,7 @@ from dataclasses import dataclass, field
from datetime import date, datetime, timedelta, timezone from datetime import date, datetime, timedelta, timezone
from typing import Any, Protocol, runtime_checkable from typing import Any, Protocol, runtime_checkable
from sqlalchemy import select, text from sqlalchemy import exists, select, text
from sqlalchemy.engine import Engine # noqa: F401 (typing only) from sqlalchemy.engine import Engine # noqa: F401 (typing only)
from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession
@@ -74,6 +74,7 @@ class ValidationResult:
# when ok=False. # when ok=False.
retryable: bool = False retryable: bool = False
deferred_alert_after_days: int | None = None deferred_alert_after_days: int | None = None
deferred_alert_messages: list[str] = field(default_factory=list)
@runtime_checkable @runtime_checkable
@@ -129,17 +130,22 @@ async def _last_promoted_revision(db: AsyncSession, source: str) -> str | None:
return row.scalar_one_or_none() return row.scalar_one_or_none()
async def _has_promoted_since(db: AsyncSession, source: str, cutoff: datetime) -> bool: async def _promotion_state_since(
row = await db.execute( db: AsyncSession, source: str, cutoff: datetime
select(DataImportRun.id) ) -> str:
.where( promoted = (
DataImportRun.source == source, DataImportRun.source == source,
DataImportRun.status == STATUS_PROMOTED, DataImportRun.status == STATUS_PROMOTED,
DataImportRun.started_at >= cutoff,
) )
.limit(1) ever, recent = (
await db.execute(
select(
exists().where(*promoted),
exists().where(*promoted, DataImportRun.started_at >= cutoff),
) )
return row.scalar_one_or_none() is not None )
).one()
return "recent" if recent else "stale" if ever else "never"
def _now() -> datetime: def _now() -> datetime:
@@ -246,16 +252,25 @@ async def run_import(
if alert_days is not None: if alert_days is not None:
alert_days = max(1, alert_days) alert_days = max(1, alert_days)
cutoff = run.started_at - timedelta(days=alert_days) cutoff = run.started_at - timedelta(days=alert_days)
if not await _has_promoted_since(session, source, cutoff): promotion_state = await _promotion_state_since(
session, source, cutoff
)
if promotion_state != "recent":
history = (
f"{source} import has never promoted successfully"
if promotion_state == "never"
else f"{source} import has not promoted successfully "
f"within {alert_days} day(s)"
)
await _alert( await _alert(
session, session,
source, source,
"deferred_stale", "deferred_stale",
[ [
f"No successful promotion for at least " f"{history}; import remains deferred",
f"{alert_days} day(s); deferred source lag may " *result.deferred_alert_messages,
f"now hide aged-out unresolved items: " f"Current deferral: "
f"{run.error_details or 'validation deferred'}" f"{run.error_details or 'validation deferred'}",
], ],
severity="warning", severity="warning",
dedup_hours=alert_days * 24, dedup_hours=alert_days * 24,
+11
View File
@@ -362,6 +362,7 @@ class SecFundamentalsImporter:
# the filings: "which ones" has to be in the alert itself, not merely # the filings: "which ones" has to be in the alert itself, not merely
# reconstructible by re-walking the index. # reconstructible by re-walking the index.
blocking = _within_retry_window(staged.missing_xbrl) blocking = _within_retry_window(staged.missing_xbrl)
aged_out = _past_retry_window(staged.missing_xbrl)
if blocking: if blocking:
messages.append( messages.append(
f"{len(blocking)} tracked XBRL filing(s) unresolved within the " f"{len(blocking)} tracked XBRL filing(s) unresolved within the "
@@ -422,6 +423,16 @@ class SecFundamentalsImporter:
and all(m.get("reason") == "not_in_companyfacts" for m in blocking) and all(m.get("reason") == "not_in_companyfacts" for m in blocking)
), ),
deferred_alert_after_days=MISSING_XBRL_RETRY_DAYS, deferred_alert_after_days=MISSING_XBRL_RETRY_DAYS,
deferred_alert_messages=(
[
f"{len(aged_out)} tracked SEC filing(s) remain unresolved past "
f"the {MISSING_XBRL_RETRY_DAYS}-day retry window and risk being "
f"promoted around without automatic retry: "
f"{_missing_detail(aged_out)}"
]
if aged_out
else []
),
) )
async def promote(self, db, staged: StagedFundamentals, run_id: int) -> dict[str, int]: async def promote(self, db, staged: StagedFundamentals, run_id: int) -> dict[str, int]:
+21 -1
View File
@@ -71,11 +71,13 @@ class FakeImporter:
def __init__( def __init__(
self, revision, *, ok=True, retryable=False, alert_days=None, self, revision, *, ok=True, retryable=False, alert_days=None,
n_rows=3, raise_in="none", n_rows=3, raise_in="none",
alert_messages=None,
): ):
self.revision = revision self.revision = revision
self.ok = ok self.ok = ok
self.retryable = retryable self.retryable = retryable
self.alert_days = alert_days self.alert_days = alert_days
self.alert_messages = alert_messages or []
self.n_rows = n_rows self.n_rows = n_rows
self.raise_in = raise_in self.raise_in = raise_in
self.staged_called = False self.staged_called = False
@@ -102,6 +104,7 @@ class FakeImporter:
messages=[] if self.ok else ["coverage below threshold"], messages=[] if self.ok else ["coverage below threshold"],
retryable=self.retryable, retryable=self.retryable,
deferred_alert_after_days=self.alert_days, deferred_alert_after_days=self.alert_days,
deferred_alert_messages=self.alert_messages,
) )
async def promote(self, db, staged, run_id): async def promote(self, db, staged, run_id):
@@ -231,7 +234,9 @@ async def test_stale_deferred_validation_emits_deduplicated_warning(engine):
await s.commit() await s.commit()
importer = FakeImporter( importer = FakeImporter(
"rev2", ok=False, retryable=True, alert_days=3, n_rows=5 "rev2", ok=False, retryable=True, alert_days=3,
alert_messages=["source detail names OLD-ACCESSION"],
n_rows=5,
) )
first = await run_import(importer, engine=engine) first = await run_import(importer, engine=engine)
second = await run_import(importer, engine=engine) second = await run_import(importer, engine=engine)
@@ -244,6 +249,21 @@ async def test_stale_deferred_validation_emits_deduplicated_warning(engine):
assert events[0].severity == "warning" assert events[0].severity == "warning"
assert events[0].code == "sec_facts_deferred_stale" assert events[0].code == "sec_facts_deferred_stale"
assert "OLD-ACCESSION" in events[0].message
assert "aged-out" not in events[0].message
async def test_never_promoted_deferred_warning_says_never(engine):
factory = _factory(engine)
run = await run_import(
FakeImporter("rev1", ok=False, retryable=True, alert_days=3),
engine=engine,
)
assert run is not None and run.status == STATUS_DEFERRED
async with factory() as s:
event = (await s.execute(select(SystemEvent))).scalar_one()
assert "has never promoted successfully" in event.message
async def test_exception_in_promote_rolls_back(engine): async def test_exception_in_promote_rolls_back(engine):
factory = _factory(engine) factory = _factory(engine)
@@ -262,6 +262,39 @@ async def test_companyfacts_lag_does_not_mask_second_validation_failure():
assert len(result.messages) == 2 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): 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 """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.""" failure from a Company-Facts lag, and must not be reported as one."""