Compare commits
3
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
7bcdf77ef9 | ||
|
|
c8c660e63d | ||
|
|
f58f8b0818 |
@@ -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)
|
||||
@@ -39,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)
|
||||
|
||||
+12
-2
@@ -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
|
||||
@@ -945,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)
|
||||
@@ -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)
|
||||
|
||||
@@ -30,10 +30,10 @@ 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
|
||||
from sqlalchemy import exists, select, text
|
||||
from sqlalchemy.engine import Engine # noqa: F401 (typing only)
|
||||
from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession
|
||||
|
||||
@@ -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,12 @@ 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 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
|
||||
deferred_alert_messages: list[str] = field(default_factory=list)
|
||||
|
||||
|
||||
@runtime_checkable
|
||||
@@ -123,19 +130,46 @@ async def _last_promoted_revision(db: AsyncSession, source: str) -> str | None:
|
||||
return row.scalar_one_or_none()
|
||||
|
||||
|
||||
async def _promotion_state_since(
|
||||
db: AsyncSession, source: str, cutoff: datetime
|
||||
) -> str:
|
||||
promoted = (
|
||||
DataImportRun.source == source,
|
||||
DataImportRun.status == STATUS_PROMOTED,
|
||||
)
|
||||
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:
|
||||
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)
|
||||
@@ -149,7 +183,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
|
||||
@@ -208,9 +243,46 @@ 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()
|
||||
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)
|
||||
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"{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,
|
||||
)
|
||||
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(
|
||||
|
||||
@@ -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 "
|
||||
@@ -413,6 +414,25 @@ class SecFundamentalsImporter:
|
||||
summary=summary,
|
||||
source_max_date=self._latest_index_date,
|
||||
messages=messages,
|
||||
# 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,
|
||||
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]:
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
@@ -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,16 @@ 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, 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
|
||||
@@ -94,6 +102,9 @@ 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,
|
||||
deferred_alert_after_days=self.alert_days,
|
||||
deferred_alert_messages=self.alert_messages,
|
||||
)
|
||||
|
||||
async def promote(self, db, staged, run_id):
|
||||
@@ -200,6 +211,60 @@ 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_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,
|
||||
alert_messages=["source detail names OLD-ACCESSION"],
|
||||
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"
|
||||
|
||||
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)
|
||||
await run_import(FakeImporter("rev1", n_rows=3), engine=engine) # baseline
|
||||
|
||||
@@ -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():
|
||||
@@ -250,6 +251,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=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"] == 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
|
||||
|
||||
@@ -17,8 +17,17 @@ 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.sec_fundamentals_importer import SecFundamentalsImporter
|
||||
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
|
||||
@@ -191,7 +200,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 +222,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 +233,66 @@ 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_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):
|
||||
|
||||
Reference in New Issue
Block a user