diff --git a/app/services/data_import.py b/app/services/data_import.py index 13837bd..4071fc5 100644 --- a/app/services/data_import.py +++ b/app/services/data_import.py @@ -33,7 +33,7 @@ from dataclasses import dataclass, field from datetime import date, datetime, timedelta, timezone 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.ext.asyncio import AsyncEngine, AsyncSession @@ -74,6 +74,7 @@ class ValidationResult: # when ok=False. retryable: bool = False deferred_alert_after_days: int | None = None + deferred_alert_messages: list[str] = field(default_factory=list) @runtime_checkable @@ -129,17 +130,22 @@ async def _last_promoted_revision(db: AsyncSession, source: str) -> str | None: return row.scalar_one_or_none() -async def _has_promoted_since(db: AsyncSession, source: str, cutoff: datetime) -> bool: - row = await db.execute( - select(DataImportRun.id) - .where( - DataImportRun.source == source, - DataImportRun.status == STATUS_PROMOTED, - DataImportRun.started_at >= cutoff, - ) - .limit(1) +async def _promotion_state_since( + db: AsyncSession, source: str, cutoff: datetime +) -> str: + promoted = ( + DataImportRun.source == source, + DataImportRun.status == STATUS_PROMOTED, ) - return row.scalar_one_or_none() is not None + ever, recent = ( + await db.execute( + select( + exists().where(*promoted), + exists().where(*promoted, DataImportRun.started_at >= cutoff), + ) + ) + ).one() + return "recent" if recent else "stale" if ever else "never" def _now() -> datetime: @@ -246,16 +252,25 @@ async def run_import( if alert_days is not None: alert_days = max(1, 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( session, source, "deferred_stale", [ - f"No successful promotion for at least " - f"{alert_days} day(s); deferred source lag may " - f"now hide aged-out unresolved items: " - f"{run.error_details or 'validation deferred'}" + f"{history}; import remains deferred", + *result.deferred_alert_messages, + f"Current deferral: " + f"{run.error_details or 'validation deferred'}", ], severity="warning", dedup_hours=alert_days * 24, diff --git a/app/services/sec_fundamentals_importer.py b/app/services/sec_fundamentals_importer.py index b3bf0e1..579327b 100644 --- a/app/services/sec_fundamentals_importer.py +++ b/app/services/sec_fundamentals_importer.py @@ -362,6 +362,7 @@ class SecFundamentalsImporter: # the filings: "which ones" has to be in the alert itself, not merely # reconstructible by re-walking the index. blocking = _within_retry_window(staged.missing_xbrl) + aged_out = _past_retry_window(staged.missing_xbrl) if blocking: messages.append( 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) ), 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]: diff --git a/tests/unit/test_data_import_framework.py b/tests/unit/test_data_import_framework.py index 396d88b..81313a4 100644 --- a/tests/unit/test_data_import_framework.py +++ b/tests/unit/test_data_import_framework.py @@ -71,11 +71,13 @@ class FakeImporter: def __init__( self, revision, *, ok=True, retryable=False, alert_days=None, n_rows=3, raise_in="none", + alert_messages=None, ): self.revision = revision self.ok = ok self.retryable = retryable self.alert_days = alert_days + self.alert_messages = alert_messages or [] self.n_rows = n_rows self.raise_in = raise_in self.staged_called = False @@ -102,6 +104,7 @@ class FakeImporter: messages=[] if self.ok else ["coverage below threshold"], retryable=self.retryable, deferred_alert_after_days=self.alert_days, + deferred_alert_messages=self.alert_messages, ) 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() 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) 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].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): factory = _factory(engine) diff --git a/tests/unit/test_sec_fundamentals_importer.py b/tests/unit/test_sec_fundamentals_importer.py index cf64111..8f10dd3 100644 --- a/tests/unit/test_sec_fundamentals_importer.py +++ b/tests/unit/test_sec_fundamentals_importer.py @@ -262,6 +262,39 @@ async def test_companyfacts_lag_does_not_mask_second_validation_failure(): 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."""