diff --git a/app/models/data_import_run.py b/app/models/data_import_run.py index ae25a14..6fb7969 100644 --- a/app/models/data_import_run.py +++ b/app/models/data_import_run.py @@ -40,4 +40,5 @@ class DataImportRun(Base): completed_at: Mapped[datetime | None] = mapped_column( DateTime(timezone=True), nullable=True ) + # Failure detail, or the non-error reason when status is deferred. error_details: Mapped[str | None] = mapped_column(Text, nullable=True) diff --git a/app/scheduler.py b/app/scheduler.py index 8a4cd4c..65ec390 100644 --- a/app/scheduler.py +++ b/app/scheduler.py @@ -950,7 +950,7 @@ async def _run_shadow_import(job_name: str, importer: SourceImporter) -> bool: """Run an importer and return whether its scheduled job was enabled. The SEC wrapper uses the return value to run its activated local cache step - after failed, no-op, promoted, or source-locked attempts while still honoring + after deferred, failed, no-op, promoted, or source-locked attempts while honoring the job-level disable switch. """ _log_event(logging.INFO, "job_start", job=job_name) diff --git a/app/services/data_import.py b/app/services/data_import.py index ed021c8..13837bd 100644 --- a/app/services/data_import.py +++ b/app/services/data_import.py @@ -30,7 +30,7 @@ import hashlib import json import logging from dataclasses import dataclass, field -from datetime import date, datetime, timezone +from datetime import date, datetime, timedelta, timezone from typing import Any, Protocol, runtime_checkable from sqlalchemy import select, text @@ -69,9 +69,11 @@ class ValidationResult: summary: dict[str, Any] = field(default_factory=dict) source_max_date: date | None = None messages: list[str] = field(default_factory=list) - # Expected source-side lag: retry next run without an operational alert. - # This is only meaningful when ok=False. + # Expected source-side lag: retry without an immediate error alert. Sources + # can bound the quiet period with deferred_alert_after_days. Only meaningful + # when ok=False. retryable: bool = False + deferred_alert_after_days: int | None = None @runtime_checkable @@ -127,19 +129,41 @@ 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) + ) + return row.scalar_one_or_none() is not None + + def _now() -> datetime: return datetime.now(timezone.utc) -async def _alert(db: AsyncSession, source: str, code: str, messages: list[str]) -> None: +async def _alert( + db: AsyncSession, + source: str, + code: str, + messages: list[str], + *, + severity: str = "error", + dedup_hours: int = 24, +) -> None: try: await system_event_service.log_event( db, - severity="error", + severity=severity, source="data_import", code=f"{source}_{code}", message=(("; ".join(messages)) or code)[:_MAX_ERROR_LEN], dedup_key=f"data_import:{source}:{code}", + dedup_hours=dedup_hours, ) except Exception: # noqa: BLE001 — alerting must never mask the real outcome logger.exception("Failed to emit data_import alert %s/%s", source, code) @@ -153,7 +177,8 @@ async def run_import( ) -> DataImportRun | None: """Run one import for ``importer``. - Returns the recorded ``DataImportRun`` (promoted / no_op / failed), or None + Returns the recorded ``DataImportRun`` (promoted / no_op / deferred / + failed), or None when the per-source advisory lock is already held (another run is active). ``force`` runs even when the revision is unchanged. The revision tracks the @@ -217,6 +242,24 @@ async def run_import( if result.retryable: run.status = STATUS_DEFERRED await session.commit() + alert_days = result.deferred_alert_after_days + 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): + 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'}" + ], + severity="warning", + dedup_hours=alert_days * 24, + ) logger.info( "data_import %s: deferred for retry: %s", source, diff --git a/app/services/sec_fundamentals_importer.py b/app/services/sec_fundamentals_importer.py index 2449fed..b3bf0e1 100644 --- a/app/services/sec_fundamentals_importer.py +++ b/app/services/sec_fundamentals_importer.py @@ -413,14 +413,15 @@ class SecFundamentalsImporter: summary=summary, source_max_date=self._latest_index_date, messages=messages, - # Pure Company-Facts publication lag is expected and self-healing: - # keep the cursor in place and retry, but do not page operators. Any - # other missing reason or validation message remains a real failure. + # Company-Facts absence is usually publication lag, but can also be a + # permanent co-registrant misfile that the daily index did not expose. + # Defer quietly at first; the framework warns if promotions stay stale. retryable=( len(messages) == 1 and bool(blocking) and all(m.get("reason") == "not_in_companyfacts" for m in blocking) ), + deferred_alert_after_days=MISSING_XBRL_RETRY_DAYS, ) async def promote(self, db, staged: StagedFundamentals, run_id: int) -> dict[str, int]: diff --git a/frontend/src/components/admin/JobControls.tsx b/frontend/src/components/admin/JobControls.tsx index d4ef092..b063f2e 100644 --- a/frontend/src/components/admin/JobControls.tsx +++ b/frontend/src/components/admin/JobControls.tsx @@ -25,7 +25,7 @@ function formatAgo(iso: string | null | undefined): string { function lastRunColor(status: string | null | undefined): string { if (status === 'error') return 'text-red-300'; - if (status === 'rate_limited') return 'text-amber-300'; + if (status === 'rate_limited' || status === 'deferred') return 'text-amber-300'; return 'text-gray-500'; } @@ -127,7 +127,7 @@ export function JobControls() { className={`text-[11px] font-medium ${ job.running ? 'text-blue-300' - : job.runtime_status === 'rate_limited' + : job.runtime_status === 'rate_limited' || job.runtime_status === 'deferred' ? 'text-amber-300' : job.runtime_status === 'error' ? 'text-red-300' @@ -140,6 +140,8 @@ export function JobControls() { ? 'Running' : job.runtime_status === 'rate_limited' ? 'Paused (rate-limited)' + : job.runtime_status === 'deferred' + ? 'Deferred (retrying)' : job.runtime_status === 'error' ? 'Last run error' : job.enabled diff --git a/tests/unit/test_data_import_framework.py b/tests/unit/test_data_import_framework.py index dcdc0c8..396d88b 100644 --- a/tests/unit/test_data_import_framework.py +++ b/tests/unit/test_data_import_framework.py @@ -17,7 +17,7 @@ from __future__ import annotations import asyncio import os import tempfile -from datetime import date, datetime, timezone +from datetime import date, datetime, timedelta, timezone import pytest from sqlalchemy import func, select @@ -68,10 +68,14 @@ class FakeImporter: source = "sec_facts" - def __init__(self, revision, *, ok=True, retryable=False, n_rows=3, raise_in="none"): + def __init__( + self, revision, *, ok=True, retryable=False, alert_days=None, + n_rows=3, raise_in="none", + ): self.revision = revision self.ok = ok self.retryable = retryable + self.alert_days = alert_days self.n_rows = n_rows self.raise_in = raise_in self.staged_called = False @@ -97,6 +101,7 @@ class FakeImporter: source_max_date=date(2026, 7, 21), messages=[] if self.ok else ["coverage below threshold"], retryable=self.retryable, + deferred_alert_after_days=self.alert_days, ) async def promote(self, db, staged, run_id): @@ -217,6 +222,29 @@ async def test_retryable_validation_defers_without_alerting(engine): assert await _count(factory, SystemEvent) == 0 # expected retry does not alert +async def test_stale_deferred_validation_emits_deduplicated_warning(engine): + factory = _factory(engine) + promoted = await run_import(FakeImporter("rev1", n_rows=3), engine=engine) + async with factory() as s: + promoted.started_at = datetime.now(timezone.utc) - timedelta(days=4) + await s.merge(promoted) + await s.commit() + + importer = FakeImporter( + "rev2", ok=False, retryable=True, alert_days=3, n_rows=5 + ) + first = await run_import(importer, engine=engine) + second = await run_import(importer, engine=engine) + + assert first is not None and first.status == STATUS_DEFERRED + assert second is not None and second.status == STATUS_DEFERRED + async with factory() as s: + events = (await s.execute(select(SystemEvent))).scalars().all() + assert len(events) == 1 + assert events[0].severity == "warning" + assert events[0].code == "sec_facts_deferred_stale" + + async def test_exception_in_promote_rolls_back(engine): factory = _factory(engine) await run_import(FakeImporter("rev1", n_rows=3), engine=engine) # baseline diff --git a/tests/unit/test_scheduler.py b/tests/unit/test_scheduler.py index 09c0266..7816195 100644 --- a/tests/unit/test_scheduler.py +++ b/tests/unit/test_scheduler.py @@ -20,6 +20,7 @@ from app.scheduler import ( queue_backtest_target_model, scheduler, ) +from app.services.data_import import STATUS_DEFERRED def test_manual_backtest_target_model_is_one_shot(): @@ -256,7 +257,7 @@ class TestShadowImportJobs: async def imported(importer): return SimpleNamespace( - status="deferred", + status=STATUS_DEFERRED, revision="abcdef1234567890", error_details="Company Facts publication lag; retrying", ) @@ -268,7 +269,7 @@ class TestShadowImportJobs: await _run_shadow_import("sec_fundamentals_import", object()) runtime = get_job_runtime_snapshot("sec_fundamentals_import") - assert runtime["status"] == "deferred" + assert runtime["status"] == STATUS_DEFERRED assert runtime["processed"] == 0 assert runtime["message"] == "Company Facts publication lag; retrying" diff --git a/tests/unit/test_sec_fundamentals_importer.py b/tests/unit/test_sec_fundamentals_importer.py index 73191ac..cf64111 100644 --- a/tests/unit/test_sec_fundamentals_importer.py +++ b/tests/unit/test_sec_fundamentals_importer.py @@ -23,7 +23,11 @@ from app.services.data_import import ( STATUS_PROMOTED, run_import, ) -from app.services.sec_fundamentals_importer import SecFundamentalsImporter +from app.services.sec_fundamentals_importer import ( + SecFundamentalsImporter, + StagedFundamentals, +) +from app.services.sec_universe import ResolvedUniverse @pytest.fixture @@ -232,6 +236,32 @@ async def test_consistency_gate_defers_without_alert_when_facts_lag_index(engine 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_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."""