From f58f8b08185de6d87544095bce28989d5e835468 Mon Sep 17 00:00:00 2001 From: Dennis Thiessen Date: Fri, 31 Jul 2026 12:40:29 +0200 Subject: [PATCH] fix(sec): defer expected Company Facts lag without alerting --- app/models/data_import_run.py | 5 +++-- app/scheduler.py | 12 ++++++++++- app/services/data_import.py | 16 +++++++++++++- app/services/sec_fundamentals_importer.py | 8 +++++++ tests/unit/test_data_import_framework.py | 19 ++++++++++++++++- tests/unit/test_scheduler.py | 22 ++++++++++++++++++++ tests/unit/test_sec_fundamentals_importer.py | 16 +++++++++----- 7 files changed, 88 insertions(+), 10 deletions(-) diff --git a/app/models/data_import_run.py b/app/models/data_import_run.py index 92507ac..ae25a14 100644 --- a/app/models/data_import_run.py +++ b/app/models/data_import_run.py @@ -10,7 +10,8 @@ class DataImportRun(Base): """One row per bulk-import attempt (SEC facts / Dolt earnings / Dolt stocks). Lean audit record for the batch import framework: every attempt is logged, - whether it promoted, was a ``no_op`` (unchanged revision), or ``failed``. + whether it promoted, was a ``no_op`` (unchanged revision), was ``deferred`` + for an expected retry, or ``failed``. ``row_counts`` and ``validation`` hold JSON strings (repo convention — see ``fundamental_data.unavailable_fields_json``), not JSONB; the validation blob carries reconciliation/discrepancy summaries so no separate conflicts @@ -28,7 +29,7 @@ class DataImportRun(Base): source: Mapped[str] = mapped_column(String(32), nullable=False) # Dolt commit hash, or SEC archive SHA-256. Null until known. revision: Mapped[str | None] = mapped_column(String(64), nullable=True) - # running | validated | promoted | no_op | failed + # running | validated | promoted | no_op | deferred | failed status: Mapped[str] = mapped_column(String(16), nullable=False) source_max_date: Mapped[date | None] = mapped_column(Date, nullable=True) row_counts_json: Mapped[str | None] = mapped_column(Text, nullable=True) diff --git a/app/scheduler.py b/app/scheduler.py index 42a8d52..8a4cd4c 100644 --- a/app/scheduler.py +++ b/app/scheduler.py @@ -43,7 +43,12 @@ from app.services import ( fundamentals_parity_service, fundamental_data_refresh_service, ) -from app.services.data_import import STATUS_FAILED, SourceImporter, run_import +from app.services.data_import import ( + STATUS_DEFERRED, + STATUS_FAILED, + SourceImporter, + run_import, +) from app.services.dolt_earnings_importer import DoltEarningsImporter from app.services.sec_fundamentals_importer import SecFundamentalsImporter from app.services.alert_service import dispatch_alerts @@ -967,6 +972,11 @@ async def _run_shadow_import(job_name: str, importer: SourceImporter) -> bool: revision = f" · {run.revision[:12]}" if run.revision else "" message = f"{run.status}{revision}" + if run.status == STATUS_DEFERRED: + message = run.error_details or message + _log_event(logging.INFO, "job_deferred", job=job_name, message=message) + _runtime_finish(job_name, "deferred", processed=0, total=1, message=message) + return True if run.status == STATUS_FAILED: message = run.error_details or message _log_event(logging.ERROR, "job_error", job=job_name, message=message) diff --git a/app/services/data_import.py b/app/services/data_import.py index f65992b..ed021c8 100644 --- a/app/services/data_import.py +++ b/app/services/data_import.py @@ -48,6 +48,7 @@ STATUS_RUNNING = "running" STATUS_VALIDATED = "validated" STATUS_PROMOTED = "promoted" STATUS_NO_OP = "no_op" +STATUS_DEFERRED = "deferred" STATUS_FAILED = "failed" _MAX_ERROR_LEN = 4000 @@ -68,6 +69,9 @@ 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. + retryable: bool = False @runtime_checkable @@ -208,9 +212,19 @@ async def run_import( run.validation_json = json.dumps(result.summary, default=str) if not result.ok: - run.status = STATUS_FAILED run.error_details = ("; ".join(result.messages))[:_MAX_ERROR_LEN] run.completed_at = _now() + if result.retryable: + run.status = STATUS_DEFERRED + await session.commit() + logger.info( + "data_import %s: deferred for retry: %s", + source, + result.messages, + ) + return run + + run.status = STATUS_FAILED await session.commit() await _alert(session, source, "validation_failed", result.messages) logger.warning( diff --git a/app/services/sec_fundamentals_importer.py b/app/services/sec_fundamentals_importer.py index d516d11..2449fed 100644 --- a/app/services/sec_fundamentals_importer.py +++ b/app/services/sec_fundamentals_importer.py @@ -413,6 +413,14 @@ 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. + retryable=( + len(messages) == 1 + and bool(blocking) + and all(m.get("reason") == "not_in_companyfacts" for m in blocking) + ), ) 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 02a60a8..dcdc0c8 100644 --- a/tests/unit/test_data_import_framework.py +++ b/tests/unit/test_data_import_framework.py @@ -29,6 +29,7 @@ from app.models.data_import_run import DataImportRun from app.models.fundamental_snapshot import FundamentalSnapshot from app.models.system_event import SystemEvent from app.services.data_import import ( + STATUS_DEFERRED, STATUS_FAILED, STATUS_NO_OP, STATUS_PROMOTED, @@ -67,9 +68,10 @@ class FakeImporter: source = "sec_facts" - def __init__(self, revision, *, ok=True, n_rows=3, raise_in="none"): + def __init__(self, revision, *, ok=True, retryable=False, n_rows=3, raise_in="none"): self.revision = revision self.ok = ok + self.retryable = retryable self.n_rows = n_rows self.raise_in = raise_in self.staged_called = False @@ -94,6 +96,7 @@ class FakeImporter: summary={"staged_rows": len(staged)}, source_max_date=date(2026, 7, 21), messages=[] if self.ok else ["coverage below threshold"], + retryable=self.retryable, ) async def promote(self, db, staged, run_id): @@ -200,6 +203,20 @@ async def test_failed_validation_leaves_data_untouched(engine): assert await _count(factory, SystemEvent) == 1 # alerted +async def test_retryable_validation_defers_without_alerting(engine): + factory = _factory(engine) + await run_import(FakeImporter("rev1", n_rows=3), engine=engine) + + run = await run_import( + FakeImporter("rev2", ok=False, retryable=True, n_rows=5), engine=engine + ) + + assert run is not None and run.status == STATUS_DEFERRED + assert "coverage" in (run.error_details or "") + assert await _count(factory, FundamentalSnapshot) == 3 # untouched + assert await _count(factory, SystemEvent) == 0 # expected retry does not alert + + 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 d6acb68..09c0266 100644 --- a/tests/unit/test_scheduler.py +++ b/tests/unit/test_scheduler.py @@ -250,6 +250,28 @@ class TestShadowImportJobs: assert runtime["processed"] == 0 assert runtime["message"] == "validation failed" + async def test_deferred_run_is_visible_without_error_status(self, monkeypatch): + async def enabled(db, job_name): + return True + + async def imported(importer): + return SimpleNamespace( + status="deferred", + revision="abcdef1234567890", + error_details="Company Facts publication lag; retrying", + ) + + monkeypatch.setattr("app.scheduler.async_session_factory", self._session_factory) + monkeypatch.setattr("app.scheduler._is_job_enabled", enabled) + monkeypatch.setattr("app.scheduler.run_import", imported) + + await _run_shadow_import("sec_fundamentals_import", object()) + + runtime = get_job_runtime_snapshot("sec_fundamentals_import") + assert runtime["status"] == "deferred" + assert runtime["processed"] == 0 + assert runtime["message"] == "Company Facts publication lag; retrying" + async def test_source_lock_surfaces_skipped(self, monkeypatch): async def enabled(db, job_name): return True diff --git a/tests/unit/test_sec_fundamentals_importer.py b/tests/unit/test_sec_fundamentals_importer.py index 19cf5d7..73191ac 100644 --- a/tests/unit/test_sec_fundamentals_importer.py +++ b/tests/unit/test_sec_fundamentals_importer.py @@ -17,7 +17,12 @@ import app.models # noqa: F401 from app.models.fundamental_snapshot import FundamentalSnapshot from app.models.system_event import SystemEvent from app.models.ticker import Ticker -from app.services.data_import import STATUS_FAILED, STATUS_PROMOTED, run_import +from app.services.data_import import ( + STATUS_DEFERRED, + STATUS_FAILED, + STATUS_PROMOTED, + run_import, +) from app.services.sec_fundamentals_importer import SecFundamentalsImporter @@ -191,7 +196,7 @@ async def test_incremental_adds_only_new_filing(engine): assert q2.fiscal_period == "Q2" and q2.revenue == 254940 -async def test_consistency_gate_fails_when_facts_lag_index(engine): +async def test_consistency_gate_defers_without_alert_when_facts_lag_index(engine): factory = _factory(engine) await _seed(factory, ["AAPL"]) backfill = FakeSecClient( @@ -213,9 +218,9 @@ async def test_consistency_gate_fails_when_facts_lag_index(engine): ) run = await run_import(_importer(incr, today=date(2026, 5, 3)), engine=engine) - assert run.status == STATUS_FAILED - # The gate blocks every later run until it clears, so the alert itself has to - # name the filing and say why it could not be resolved. + 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 @@ -224,6 +229,7 @@ async def test_consistency_gate_fails_when_facts_lag_index(engine): 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_gate_separates_missing_submissions_from_missing_facts(engine):