diff --git a/alembic/versions/028_sec_filing_retry_queue.py b/alembic/versions/028_sec_filing_retry_queue.py index e9735c2..04325d3 100644 --- a/alembic/versions/028_sec_filing_retry_queue.py +++ b/alembic/versions/028_sec_filing_retry_queue.py @@ -4,6 +4,8 @@ Revision ID: 028 Revises: 027 Create Date: 2026-08-03 00:00:00.000000 """ +from datetime import date, datetime, timezone +import json from typing import Sequence, Union from alembic import op @@ -28,11 +30,118 @@ def upgrade() -> None: sa.Column("coregistrant_ciks_json", sa.Text(), nullable=True), sa.Column("first_seen_at", sa.DateTime(timezone=True), nullable=False), sa.Column("last_attempted_at", sa.DateTime(timezone=True), nullable=False), + sa.Column("escalated_at", sa.DateTime(timezone=True), nullable=True), sa.UniqueConstraint("accession", name="uq_sec_filing_gaps_accession"), ) op.create_index("ix_sec_filing_gaps_cik", "sec_filing_gaps", ["cik"]) + _backfill_retry_queue() def downgrade() -> None: op.drop_index("ix_sec_filing_gaps_cik", table_name="sec_filing_gaps") op.drop_table("sec_filing_gaps") + + +def _as_date(value) -> date | None: + if isinstance(value, datetime): + return value.date() + if isinstance(value, date): + return value + if isinstance(value, str): + try: + return date.fromisoformat(value) + except ValueError: + return None + return None + + +def _backfill_retry_queue() -> None: + """Materialize pre-queue promoted gaps once; runtime never scans history.""" + bind = op.get_bind() + runs = sa.table( + "data_import_runs", + sa.column("source", sa.String()), + sa.column("status", sa.String()), + sa.column("validation_json", sa.Text()), + sa.column("source_max_date", sa.Date()), + sa.column("started_at", sa.DateTime(timezone=True)), + ) + snapshots = sa.table( + "fundamental_snapshots", + sa.column("cik", sa.String()), + sa.column("accession", sa.String()), + sa.column("filed_date", sa.Date()), + ) + gaps = sa.table( + "sec_filing_gaps", + sa.column("cik", sa.String()), + sa.column("accession", sa.String()), + sa.column("form", sa.String()), + sa.column("index_date", sa.Date()), + sa.column("reason", sa.String()), + sa.column("coregistrant_ciks_json", sa.Text()), + sa.column("first_seen_at", sa.DateTime(timezone=True)), + sa.column("last_attempted_at", sa.DateTime(timezone=True)), + sa.column("escalated_at", sa.DateTime(timezone=True)), + ) + + snapshot_rows = bind.execute( + sa.select(snapshots.c.cik, snapshots.c.accession, snapshots.c.filed_date) + ).all() + resolved_accessions = {row.accession for row in snapshot_rows} + latest_filed_by_cik: dict[str, date] = {} + for row in snapshot_rows: + if row.filed_date is not None: + current = latest_filed_by_cik.get(row.cik) + if current is None or row.filed_date > current: + latest_filed_by_cik[row.cik] = row.filed_date + + audit_rows = bind.execute( + sa.select( + runs.c.validation_json, + runs.c.source_max_date, + runs.c.started_at, + ).where( + runs.c.source == "sec_facts", + runs.c.status == "promoted", + runs.c.validation_json.is_not(None), + ) + ).all() + now = datetime.now(timezone.utc) + candidates: dict[str, dict] = {} + for audit in audit_rows: + try: + summary = json.loads(audit.validation_json) + except (TypeError, ValueError): + continue + if not isinstance(summary, dict): + continue + for item in summary.get("missing_xbrl") or []: + accession = item.get("accession") + raw_cik = item.get("cik") + if not accession or raw_cik is None or accession in resolved_accessions: + continue + cik = str(raw_cik).zfill(10) + index_date = _as_date(item.get("index_date")) or _as_date( + audit.source_max_date + ) + later_filed = latest_filed_by_cik.get(cik) + if index_date is not None and later_filed is not None and later_filed > index_date: + continue + first_seen = audit.started_at or now + existing = candidates.get(accession) + if existing is not None and existing["first_seen_at"] <= first_seen: + continue + candidates[accession] = { + "cik": cik, + "accession": accession, + "form": item.get("form"), + "index_date": index_date, + "reason": item.get("reason") or "not_in_companyfacts", + "coregistrant_ciks_json": json.dumps(item.get("coregistrants") or []), + "first_seen_at": first_seen, + "last_attempted_at": first_seen, + "escalated_at": None, + } + if candidates: + op.bulk_insert(gaps, list(candidates.values())) diff --git a/app/models/sec_filing_gap.py b/app/models/sec_filing_gap.py index 2612b68..a475d07 100644 --- a/app/models/sec_filing_gap.py +++ b/app/models/sec_filing_gap.py @@ -10,8 +10,8 @@ class SecFilingGap(Base): """Active SEC filing that could not yet be reconstructed. Rows form a small retry queue. Successful snapshot ingestion deletes the - matching row; while a row remains, tickers mapped to its CIK are not eligible - for actionable trade setups. + matching row; a later valid filing supersedes it. While a current row remains, + tickers mapped to its CIK are not eligible for actionable trade setups. """ __tablename__ = "sec_filing_gaps" @@ -29,3 +29,4 @@ class SecFilingGap(Base): coregistrant_ciks_json: Mapped[str | None] = mapped_column(Text, nullable=True) first_seen_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False) last_attempted_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False) + escalated_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) diff --git a/app/routers/fundamentals.py b/app/routers/fundamentals.py index 23c59e5..49c028c 100644 --- a/app/routers/fundamentals.py +++ b/app/routers/fundamentals.py @@ -10,6 +10,7 @@ from app.schemas.common import APIEnvelope from app.schemas.fundamental import FundamentalResponse from app.services.fundamental_service import get_fundamental from app.services.fundamentals_api_service import build_fundamentals_v1 +from app.services import fundamentals_quality_service router = APIRouter(tags=["fundamentals"]) @@ -34,6 +35,7 @@ async def read_fundamentals( """Get latest fundamental data for a symbol (legacy fields + additive v1).""" record = await get_fundamental(db, symbol) v1 = await build_fundamentals_v1(db, symbol) + quality = await fundamentals_quality_service.ticker_quality(db, symbol) legacy: dict = {} if record is not None: @@ -47,5 +49,12 @@ async def read_fundamentals( unavailable_fields=_parse_unavailable_fields(record.unavailable_fields_json), ) - data = FundamentalResponse(symbol=symbol.strip().upper(), **legacy, **v1) + data = FundamentalResponse( + symbol=symbol.strip().upper(), + setup_eligible=quality.eligible, + setup_block_code=quality.code, + setup_block_reason=quality.message, + **legacy, + **v1, + ) return APIEnvelope(status="success", data=data.model_dump()) diff --git a/app/schemas/fundamental.py b/app/schemas/fundamental.py index b4d97f5..bbbbf1c 100644 --- a/app/schemas/fundamental.py +++ b/app/schemas/fundamental.py @@ -91,3 +91,6 @@ class FundamentalResponse(BaseModel): metrics: list[MetricItem] | None = None valuation: Valuation | None = None reads: FundamentalsReads | None = None + setup_eligible: bool = True + setup_block_code: str | None = None + setup_block_reason: str | None = None diff --git a/app/services/fundamentals_quality_service.py b/app/services/fundamentals_quality_service.py index 91e3ba4..172c67f 100644 --- a/app/services/fundamentals_quality_service.py +++ b/app/services/fundamentals_quality_service.py @@ -3,8 +3,9 @@ from __future__ import annotations import json +from dataclasses import dataclass -from sqlalchemy import exists, select +from sqlalchemy import exists, or_, select from sqlalchemy.ext.asyncio import AsyncSession from app.models.data_import_run import DataImportRun @@ -13,35 +14,41 @@ from app.models.sec_filing_gap import SecFilingGap from app.models.ticker import Ticker from app.services import fundamental_data_refresh_service +_SEC_FORMS = ("10-K", "10-Q", "10-K/A", "10-Q/A") -async def blocked_ciks(db: AsyncSession) -> set[str]: - """CIKs whose SEC inputs are known incomplete. - The durable queue covers filings already promoted around. The latest - validation payload covers young filings still in the deferred retry window - and new registrants with no XBRL history. - """ - if not await fundamental_data_refresh_service.is_enabled(db): - return set() +@dataclass(frozen=True) +class SetupQuality: + eligible: bool + code: str | None = None + message: str | None = None - active = ( - await db.execute( - select(SecFilingGap.cik).where( - ~exists().where( - FundamentalSnapshot.accession == SecFilingGap.accession - ) - ) - ) - ).scalars().all() - blocked = set(active) - # Migration bootstrap: before the first scheduled import has materialized - # the durable queue, recover unresolved promoted gaps from the import audit. - # Once the queue has rows, the importer owns this state and this history scan - # is no longer needed on setup reads. - if not active and not await retry_queue_initialized(db): - blocked.update(await _historical_unresolved_ciks(db)) +async def active_gaps( + db: AsyncSession, + ciks: set[str] | None = None, +) -> list[SecFilingGap]: + """Unresolved gaps that have not been superseded by a later filing.""" + matching_snapshot = exists().where( + FundamentalSnapshot.accession == SecFilingGap.accession + ) + later_snapshot = exists().where( + FundamentalSnapshot.cik == SecFilingGap.cik, + FundamentalSnapshot.form.in_(_SEC_FORMS), + FundamentalSnapshot.filed_date > SecFilingGap.index_date, + ) + stmt = select(SecFilingGap).where( + ~matching_snapshot, + or_(SecFilingGap.index_date.is_(None), ~later_snapshot), + ) + if ciks is not None: + if not ciks: + return [] + stmt = stmt.where(SecFilingGap.cik.in_(ciks)) + return list((await db.execute(stmt)).scalars().all()) + +async def _latest_validation(db: AsyncSession) -> dict: payload = ( await db.execute( select(DataImportRun.validation_json) @@ -54,81 +61,38 @@ async def blocked_ciks(db: AsyncSession) -> set[str]: ) ).scalar_one_or_none() if not payload: - return blocked + return {} try: summary = json.loads(payload) except (TypeError, ValueError): - return blocked + return {} + return summary if isinstance(summary, dict) else {} + +async def blocked_reasons_by_cik(db: AsyncSession) -> dict[str, str]: + """Current SEC blocker code by CIK; no historical audit scan.""" + if not await fundamental_data_refresh_service.is_enabled(db): + return {} + + reasons = {gap.cik: "sec_filing_gap" for gap in await active_gaps(db)} + summary = await _latest_validation(db) + + # New summaries carry the complete compact CIK set while the detailed lists + # stay capped for audit readability. Detailed entries supply the reason. + for cik in summary.get("setup_blocked_ciks") or []: + if cik: + reasons.setdefault(str(cik), "sec_filing_gap") for item in summary.get("missing_xbrl") or []: if item.get("cik"): - blocked.add(str(item["cik"])) + reasons.setdefault(str(item["cik"]), "sec_filing_gap") for item in summary.get("no_xbrl_filings") or []: if item.get("cik"): - blocked.add(str(item["cik"])) - return blocked + reasons[str(item["cik"])] = "no_xbrl_filings" + return reasons -async def retry_queue_initialized(db: AsyncSession) -> bool: - """Whether a promoted run has synchronized the durable retry queue.""" - payload = ( - await db.execute( - select(DataImportRun.row_counts_json) - .where( - DataImportRun.source == "sec_facts", - DataImportRun.status == "promoted", - DataImportRun.row_counts_json.is_not(None), - ) - .order_by(DataImportRun.id.desc()) - .limit(1) - ) - ).scalar_one_or_none() - if not payload: - return False - try: - counts = json.loads(payload) - except (TypeError, ValueError): - return False - return "retry_queue_added" in counts - - -async def _historical_unresolved_ciks(db: AsyncSession) -> set[str]: - payloads = ( - await db.execute( - select(DataImportRun.validation_json).where( - DataImportRun.source == "sec_facts", - DataImportRun.status == "promoted", - DataImportRun.validation_json.is_not(None), - ) - ) - ).scalars().all() - - accession_to_cik: dict[str, str] = {} - for payload in payloads: - try: - summary = json.loads(payload) - except (TypeError, ValueError): - continue - for item in summary.get("missing_xbrl") or []: - accession = item.get("accession") - cik = item.get("cik") - if accession and cik: - accession_to_cik[str(accession)] = str(cik) - - if not accession_to_cik: - return set() - resolved = set( - ( - await db.execute( - select(FundamentalSnapshot.accession).where( - FundamentalSnapshot.accession.in_(accession_to_cik) - ) - ) - ).scalars() - ) - return { - cik for accession, cik in accession_to_cik.items() if accession not in resolved - } +async def blocked_ciks(db: AsyncSession) -> set[str]: + return set(await blocked_reasons_by_cik(db)) async def blocked_ticker_ids(db: AsyncSession) -> set[int]: @@ -139,5 +103,36 @@ async def blocked_ticker_ids(db: AsyncSession) -> set[int]: return {int(ticker_id) for ticker_id in rows.scalars()} +async def ticker_quality(db: AsyncSession, symbol: str) -> SetupQuality: + ticker = ( + await db.execute( + select(Ticker).where(Ticker.symbol == symbol.strip().upper()) + ) + ).scalar_one_or_none() + if ticker is None or not ticker.cik: + return SetupQuality(eligible=True) + reason = (await blocked_reasons_by_cik(db)).get(ticker.cik) + if reason == "no_xbrl_filings": + return SetupQuality( + eligible=False, + code=reason, + message=( + "No SEC 10-K/10-Q is available for this registrant, so new setups " + "are paused. New registrants clear automatically after their first " + "filing; a successor shell needs an SEC CIK override." + ), + ) + if reason: + return SetupQuality( + eligible=False, + code=reason, + message=( + "A recent SEC filing is still being reconciled, so new setups are " + "paused. The scheduled fundamentals import retries it automatically." + ), + ) + return SetupQuality(eligible=True) + + async def ticker_is_eligible(db: AsyncSession, ticker_id: int) -> bool: return ticker_id not in await blocked_ticker_ids(db) diff --git a/app/services/rr_scanner_service.py b/app/services/rr_scanner_service.py index ce5158b..012abbc 100644 --- a/app/services/rr_scanner_service.py +++ b/app/services/rr_scanner_service.py @@ -27,7 +27,7 @@ from app.models.signal_context_snapshot import SignalContextSnapshot from app.models.ticker import Ticker from app.models.trade_setup import TradeSetup from app.services.indicator_service import _extract_ohlcv, compute_atr -from app.services import fundamentals_quality_service +from app.services import fundamentals_quality_service, system_event_service from app.services.price_service import query_ohlcv from app.services.qualification import setup_qualifies from app.services.sr_service import detect_gate_target_ladder @@ -750,6 +750,16 @@ async def scan_all_tickers( logger.exception( "Could not resolve fundamentals quality; blocking this scan closed" ) + await system_event_service.log_event_standalone( + severity="error", + source="rr_scanner", + code="fundamentals_quality_unavailable", + message=( + "The fundamentals quality gate could not be evaluated; the " + "universe scan was blocked to avoid issuing unchecked setups." + ), + dedup_key="rr_scanner:fundamentals_quality_unavailable", + ) fundamentals_blocked_ids = {ticker_id for ticker_id, _ in ticker_rows} # Gate-reset observations must use the same runtime activation settings as @@ -924,6 +934,16 @@ async def get_trade_setups( logger.exception( "Could not resolve fundamentals quality; hiding actionable setups" ) + await system_event_service.log_event_standalone( + severity="error", + source="rr_scanner", + code="fundamentals_quality_unavailable", + message=( + "The fundamentals quality gate could not be evaluated; actionable " + "setups were hidden until the metadata check recovers." + ), + dedup_key="rr_scanner:fundamentals_quality_unavailable", + ) return [] if exclude_open_trade_tickers: # Manual book only. The shadow book holds the *top-ranked* names by diff --git a/app/services/sec_fundamentals_importer.py b/app/services/sec_fundamentals_importer.py index 2dbbb25..d001f43 100644 --- a/app/services/sec_fundamentals_importer.py +++ b/app/services/sec_fundamentals_importer.py @@ -81,6 +81,7 @@ MIN_BACKFILL_COVERAGE = 0.5 # three); past that it is misfiled, not late, and blocking forever costs more # than the missing filing does — see the unresolved-filing guardrail below. MISSING_XBRL_RETRY_DAYS = 3 +FILING_GAP_ESCALATE_DAYS = 14 # Share-count band a co-registrant-recovered row must land in, relative to the # issuer's own last snapshot. Wide enough for buybacks/issuance, nowhere near # wide enough to let a subsidiary shell's token float through (see _shares_continuous). @@ -154,7 +155,6 @@ class SecFundamentalsImporter: # of a combined filing). Only populated for accessions a tracked issuer filed. self._coregistrants: dict[str, list[int]] = {} self._retry_rows: list[dict[str, Any]] = [] - self._retry_queue_bootstrap_pending = False self._latest_index_date: date | None = None self._backfill = False @@ -185,18 +185,14 @@ class SecFundamentalsImporter: ) self._retry_rows = [] if not self._backfill: - self._retry_queue_bootstrap_pending = not ( - await fundamentals_quality_service.retry_queue_initialized(db) - ) self._retry_rows = await self._retry_backlog( db, set(self._resolved.cik_to_ticker_ids), - include_history=self._retry_queue_bootstrap_pending, ) # Company Facts can change while the daily index revision stays fixed. # Returning None deliberately bypasses the framework's no-op gate so a # scheduled run retries every active gap. - return None if self._retry_rows or self._retry_queue_bootstrap_pending else revision + return None if self._retry_rows else revision async def stage(self, db) -> StagedFundamentals: assert self._resolved is not None, "detect_revision must run first" @@ -211,10 +207,9 @@ class SecFundamentalsImporter: if r["cik"] in cik_to_tids: filed_by_cik[r["cik"]].append(r) - # Promoted-around filings live in a small durable retry queue. Historical - # validation payloads bootstrap gaps created before the queue existed. - # Merge them into the normal incremental work so the scheduled import, - # not an operator-run full reparse, heals them when SEC catches up. + # Promoted-around filings live in a small durable retry queue, including + # the one-time migration backfill. Merge them into normal incremental + # work so the scheduled importer heals them without operator action. if not self._backfill: seen = { (int(cik), row["accession"]) @@ -286,6 +281,9 @@ class SecFundamentalsImporter: fiscal_year_end = sub.get("fiscal_year_end") recovered_rows: list[SnapshotRow] = [] + index_rows = { + row["accession"]: row for row in filed_by_cik.get(cik, []) + } if is_backfill: accns = set(xbrl_meta) else: @@ -340,6 +338,19 @@ class SecFundamentalsImporter: # fiscalYearEnd (MMDD) is what lets the parser derive period identity from # reportDate instead of SEC's unreliable fy/fp fields. result = parser.parse_snapshots(cf, xbrl_meta, accns, fiscal_year_end=fiscal_year_end) + for skipped in result.skipped_filings: + index_row = index_rows.get(skipped["accession"]) + if index_row is not None: + # Facts are present but our parser cannot construct a snapshot. + # This is terminal for this run: promote around it immediately, + # keep the issuer blocked, and retry/escalate through the queue. + staged.missing_xbrl.append(_missing( + cik, + {**index_row, "_retry_queue": True}, + "parser_unusable", + self.today, + self._coregistrants.get(skipped["accession"]), + )) staged.rows.extend(result.rows) staged.rows.extend(recovered_rows) staged.skipped_filings.extend(result.skipped_filings) @@ -442,13 +453,19 @@ class SecFundamentalsImporter: "skipped_filings": len(staged.skipped_filings), "field_issues": len(staged.field_issues), "skipped_non_xbrl": len(staged.skipped_non_xbrl), - "no_xbrl_filings": staged.no_xbrl_filings, + "no_xbrl_filings": staged.no_xbrl_filings[:50], "no_xbrl_filings_count": len(staged.no_xbrl_filings), - "missing_xbrl": staged.missing_xbrl, + "missing_xbrl": staged.missing_xbrl[:50], "missing_xbrl_count": len(staged.missing_xbrl), "missing_xbrl_blocking": len(blocking), - "recovered_from_coregistrant": staged.recovered, + "recovered_from_coregistrant": staged.recovered[:50], "recovered_count": len(staged.recovered), + # Complete compact gate input; detailed audit lists above stay capped. + "setup_blocked_ciks": sorted({ + str(item["cik"]) + for item in [*staged.missing_xbrl, *staged.no_xbrl_filings] + if item.get("cik") + }), "invalid_payloads": staged.invalid_payloads, "cik_updates": len(staged.resolved.cik_updates), # differing existing accessions (immutable — kept, reported here) @@ -508,13 +525,15 @@ class SecFundamentalsImporter: await db.execute(stmt) inserted += 1 - # Synchronize the active retry queue in the same transaction as snapshot - # promotion. Reconstructed accessions leave the queue; aged-out gaps enter - # or refresh it and will be retried by the next scheduled import. - existing_gap_accessions = set( - (await db.execute(select(SecFilingGap.accession))).scalars().all() - ) + # Synchronize the retry queue in the snapshot-promotion transaction. + existing_gaps = (await db.execute(select(SecFilingGap))).scalars().all() + existing_gap_accessions = {gap.accession for gap in existing_gaps} resolved_accessions = {row.accession for row in staged.rows} + # A filing now classified non-XBRL can never yield a snapshot and is no + # longer a fundamentals completeness gap. + resolved_accessions.update( + item["accession"] for item in staged.skipped_non_xbrl + ) queue_resolved = 0 if resolved_accessions: result = await db.execute( @@ -524,8 +543,8 @@ class SecFundamentalsImporter: ) queue_resolved = int(result.rowcount or 0) - tolerated = _past_retry_window(staged.missing_xbrl) now = _now() + tolerated = _past_retry_window(staged.missing_xbrl) for gap in tolerated: stmt = insert_for_session(db, SecFilingGap).values( cik=gap["cik"], @@ -550,6 +569,20 @@ class SecFundamentalsImporter: }, ) ) + + # Remove gaps made irrelevant by a later valid 10-K/10-Q. Quality reads + # already ignore them; physical cleanup keeps the queue small. + active_ids = {gap.id for gap in await fundamentals_quality_service.active_gaps(db)} + obsolete_ids = { + gap.id for gap in existing_gaps + if gap.id not in active_ids and gap.accession not in resolved_accessions + } + if obsolete_ids: + result = await db.execute( + delete(SecFilingGap).where(SecFilingGap.id.in_(obsolete_ids)) + ) + queue_resolved += int(result.rowcount or 0) + newly_queued = [ gap for gap in tolerated if gap["accession"] not in existing_gap_accessions @@ -574,6 +607,40 @@ class SecFundamentalsImporter: created_at=_now(), )) + # Persistent current gaps get one actionable escalation rather than a + # daily warning. The nullable marker makes this durable and noise-free. + escalation_cutoff = now - timedelta(days=FILING_GAP_ESCALATE_DAYS) + aged_gaps = ( + await db.execute( + select(SecFilingGap).where( + SecFilingGap.first_seen_at <= escalation_cutoff, + SecFilingGap.escalated_at.is_(None), + ) + ) + ).scalars().all() + if aged_gaps: + named = ", ".join( + f"{gap.cik}/{gap.accession} ({gap.reason})" + for gap in aged_gaps[:10] + ) + db.add(SystemEvent( + severity="warning", + source="sec_facts", + code="filing_gap_aged", + message=( + f"{len(aged_gaps)} SEC filing gap(s) remain unresolved after " + f"{FILING_GAP_ESCALATE_DAYS} days; affected setups remain paused. " + f"Review the filing/CIK mapping or parser: {named}" + )[:4000], + dedup_key=f"sec_facts:filing_gap_aged:{run_id}", + created_at=now, + )) + await db.execute( + update(SecFilingGap) + .where(SecFilingGap.id.in_([gap.id for gap in aged_gaps])) + .values(escalated_at=now) + ) + # Recovered rows are real data from an unexpected place — record where they # came from, so a wrong recovery is auditable rather than invisible. if staged.recovered: @@ -641,20 +708,14 @@ class SecFundamentalsImporter: self, db, tracked_ciks: set[int], - *, - include_history: bool, ) -> list[dict[str, Any]]: - """Active gaps plus pre-queue gaps from promoted validation history.""" + """Active typed gaps; migration 028 owns historical bootstrap.""" if not tracked_ciks: return [] tracked = {cik10(cik) for cik in tracked_ciks} candidates: dict[str, dict[str, Any]] = {} - queued = ( - await db.execute( - select(SecFilingGap).where(SecFilingGap.cik.in_(tracked)) - ) - ).scalars().all() + queued = await fundamentals_quality_service.active_gaps(db, tracked) for gap in queued: try: coregistrants = json.loads(gap.coregistrant_ciks_json or "[]") @@ -667,51 +728,9 @@ class SecFundamentalsImporter: "index_date": gap.index_date, "reason": gap.reason, "coregistrants": coregistrants, + "_retry_queue": True, } - # Bootstrap warnings produced before sec_filing_gaps existed. Promoted - # runs contain only aged-out gaps; deferred/failed runs are naturally - # retried because source_max_date has not advanced. - if include_history: - histories = ( - await db.execute( - select(DataImportRun.validation_json) - .where( - DataImportRun.source == SOURCE, - DataImportRun.status == STATUS_PROMOTED, - DataImportRun.validation_json.is_not(None), - ) - .order_by(DataImportRun.id.asc()) - ) - ).scalars().all() - for payload in histories: - try: - summary = json.loads(payload) - except (TypeError, ValueError): - continue - for item in summary.get("missing_xbrl") or []: - accession = item.get("accession") - cik = str(item.get("cik") or "") - if not accession or cik not in tracked or accession in candidates: - continue - raw_date = item.get("index_date") - try: - index_date = ( - date.fromisoformat(raw_date) - if isinstance(raw_date, str) - else raw_date - ) - except ValueError: - index_date = None - candidates[accession] = { - "cik": cik, - "accession": accession, - "form": item.get("form"), - "index_date": index_date, - "reason": item.get("reason") or "not_in_companyfacts", - "coregistrants": item.get("coregistrants") or [], - } - if not candidates: return [] resolved = set( @@ -848,13 +867,19 @@ def _missing( up by hand (EDGAR accession + the index date it was seen on) and to decide whether it is still young enough to be worth blocking on.""" index_date = row.get("index_date") + age_days = ( + (today - index_date).days if isinstance(index_date, date) else 0 + ) + if row.get("_retry_queue"): + age_days = max(age_days, MISSING_XBRL_RETRY_DAYS + 1) return { "cik": cik10(cik), "accession": row["accession"], "form": row.get("form"), "index_date": index_date, - # No index date (older cached rows) => age 0 => blocks, the safe default. - "age_days": (today - index_date).days if isinstance(index_date, date) else 0, + # A newly observed row without a date blocks safely. A durable queue row + # has already passed the bounded window and is forced aged-out above. + "age_days": age_days, "reason": reason, "coregistrants": list(coregistrants or []), } diff --git a/docs/fundamentals-deployment.md b/docs/fundamentals-deployment.md index f68d3f3..a648a33 100644 --- a/docs/fundamentals-deployment.md +++ b/docs/fundamentals-deployment.md @@ -17,7 +17,9 @@ not add OS cron entries: the application scheduler owns both jobs. event. A failed validation does not promote partial data. - An SEC filing still missing after the short publication-lag window enters `sec_filing_gaps`. The daily importer retries it automatically; affected - tickers are excluded from actionable setups until a snapshot is recovered. + tickers are excluded from actionable setups until a snapshot is recovered or + a later valid 10-K/10-Q supersedes the gap. Migration `028` materializes older + promoted gaps into this queue once, so setup reads never scan import history. The systemd service uses one application worker. The import framework also holds a PostgreSQL advisory lock per source, so an overlapping manual/scheduled run is @@ -86,7 +88,8 @@ In Admin → Jobs, wait until no other job is running, then: 3. Check Admin → System Events. There should be no new import error. 4. Confirm the next-run times correspond to 02:30 and 04:00 New York time. 5. Open several ticker pages and confirm the fundamentals panel has populated - data and still handles partial/missing issuers cleanly. + data and still handles partial/missing issuers cleanly. A ticker held by the + quality gate should show **New setups paused** with the specific SEC reason. ## A5 parity observation window @@ -235,8 +238,13 @@ least several scheduled cycles before A6 removes the legacy providers. - Inspect the job runtime, latest `data_import_runs.validation_json`, service logs, and Admin → System Events before retrying. - `unresolved_filing` is emitted once when a filing enters automatic retry. It - does not require a server command. Successful co-registrant recovery and new - registrants with no XBRL history are logged without recurring warning events. + does not require a server command. If the gap is still current after 14 days, + `filing_gap_aged` is emitted once with the CIK, accession, and parser/mapping + reason. A later valid 10-K/10-Q retires the gap even when the original SEC + accession never becomes usable. +- Successful co-registrant recovery is logged without a warning. New registrants + with no XBRL history are also logged quietly, but their ticker page explains + that setups remain paused and that successor shells may need `sec_cik_overrides`. - Re-run `sudo -u deploy bash ./deploy/provision_fundamentals.sh --check` for binary, clone, permission, disk, or environment failures. - The Dolt clone is a reproducible cache and does not need a bespoke backup. diff --git a/frontend/src/dev/harness.tsx b/frontend/src/dev/harness.tsx index 4d9010e..3afdab4 100644 --- a/frontend/src/dev/harness.tsx +++ b/frontend/src/dev/harness.tsx @@ -38,6 +38,7 @@ const ind = (median: number, favorable_percentile: number) => const legacy = { pe_ratio: null, revenue_growth: null, earnings_surprise: null, market_cap: null, next_earnings_date: null, fetched_at: null, unavailable_fields: {}, + setup_eligible: true, setup_block_code: null, setup_block_reason: null, }; const full: FundamentalResponse = { diff --git a/frontend/src/lib/types.ts b/frontend/src/lib/types.ts index cc063a7..991a658 100644 --- a/frontend/src/lib/types.ts +++ b/frontend/src/lib/types.ts @@ -829,6 +829,9 @@ export interface FundamentalResponse { metrics: MetricItem[] | null; valuation: Valuation | null; reads: FundamentalsReads | null; + setup_eligible: boolean; + setup_block_code: string | null; + setup_block_reason: string | null; } // Indicators diff --git a/frontend/src/pages/TickerDetailPage.tsx b/frontend/src/pages/TickerDetailPage.tsx index 1b799e0..d6900c2 100644 --- a/frontend/src/pages/TickerDetailPage.tsx +++ b/frontend/src/pages/TickerDetailPage.tsx @@ -359,6 +359,15 @@ export default function TickerDetailPage() { busy={ingestion.isPending} /> + {fundamentals.data && !fundamentals.data.setup_eligible && ( +
+ + New setups paused.{' '} + {fundamentals.data.setup_block_reason ?? + 'SEC fundamentals are incomplete for this ticker.'} + +
+ )}
diff --git a/tests/unit/test_fundamentals_quality_service.py b/tests/unit/test_fundamentals_quality_service.py index fdb85f3..942ba2e 100644 --- a/tests/unit/test_fundamentals_quality_service.py +++ b/tests/unit/test_fundamentals_quality_service.py @@ -67,27 +67,23 @@ async def test_sec_quality_gate_is_inactive_before_cutover(db_session): -async def test_promoted_gap_is_blocked_during_queue_migration_bootstrap(db_session): +async def test_active_gap_is_blocked_until_a_later_filing_supersedes_it(db_session): ticker = Ticker(symbol="HIST", cik="0000000043") + now = datetime.now(timezone.utc) db_session.add_all([ ticker, SystemSetting( key="fundamental_data_sec_dolt_cutover_enabled", value="true", ), - DataImportRun( - source="sec_facts", - status="promoted", - validation_json=json.dumps({ - "missing_xbrl": [{"cik": ticker.cik, "accession": "HIST-Q"}], - }), - started_at=datetime.now(timezone.utc), - ), - DataImportRun( - source="sec_facts", - status="promoted", - validation_json=json.dumps({"missing_xbrl": []}), - started_at=datetime.now(timezone.utc), + SecFilingGap( + cik=ticker.cik, + accession="HIST-Q", + form="10-Q", + index_date=date.today().replace(day=1), + reason="coregistrant_facts_rejected", + first_seen_at=now, + last_attempted_at=now, ), ]) await db_session.flush() @@ -99,7 +95,7 @@ async def test_promoted_gap_is_blocked_during_queue_migration_bootstrap(db_sessi db_session.add( FundamentalSnapshot( cik=ticker.cik, - accession="HIST-Q", + accession="LATER-Q", form="10-Q", filed_date=date.today(), accepted_at=datetime.now(timezone.utc), @@ -111,3 +107,29 @@ async def test_promoted_gap_is_blocked_during_queue_migration_bootstrap(db_sessi await db_session.flush() assert await fundamentals_quality_service.blocked_ticker_ids(db_session) == set() + + +async def test_ticker_quality_explains_no_xbrl_block(db_session): + ticker = Ticker(symbol="NEWREG", cik="0000000044") + db_session.add_all([ + ticker, + SystemSetting( + key="fundamental_data_sec_dolt_cutover_enabled", + value="true", + ), + DataImportRun( + source="sec_facts", + status="promoted", + validation_json=json.dumps({ + "setup_blocked_ciks": [ticker.cik], + "no_xbrl_filings": [{"cik": ticker.cik}], + }), + started_at=datetime.now(timezone.utc), + ), + ]) + await db_session.flush() + + quality = await fundamentals_quality_service.ticker_quality(db_session, "NEWREG") + assert quality.eligible is False + assert quality.code == "no_xbrl_filings" + assert "CIK override" in (quality.message or "") diff --git a/tests/unit/test_rr_scanner_scan_all.py b/tests/unit/test_rr_scanner_scan_all.py index c46b475..49404bd 100644 --- a/tests/unit/test_rr_scanner_scan_all.py +++ b/tests/unit/test_rr_scanner_scan_all.py @@ -131,3 +131,38 @@ async def test_scan_skips_ticker_with_incomplete_sec_fundamentals( monkeypatch.setattr(rr_scanner_service, "scan_ticker", _unexpected_scan) assert await rr_scanner_service.scan_all_tickers(session) == [] + + +async def test_scan_quality_failure_blocks_closed_and_emits_event( + session, monkeypatch +): + session.add(Ticker(symbol="BLOCKED")) + await session.commit() + + async def _boom(db): + raise ValueError("bad quality metadata") + + async def _unexpected_scan(*args, **kwargs): + raise AssertionError("ticker was scanned without a quality decision") + + events: list[dict] = [] + + async def _capture_event(**kwargs): + events.append(kwargs) + + monkeypatch.setattr( + rr_scanner_service.fundamentals_quality_service, + "blocked_ticker_ids", + _boom, + ) + monkeypatch.setattr(rr_scanner_service, "scan_ticker", _unexpected_scan) + monkeypatch.setattr( + rr_scanner_service.system_event_service, + "log_event_standalone", + _capture_event, + ) + + assert await rr_scanner_service.scan_all_tickers(session) == [] + assert [event["code"] for event in events] == [ + "fundamentals_quality_unavailable" + ] diff --git a/tests/unit/test_sec_fundamentals_importer.py b/tests/unit/test_sec_fundamentals_importer.py index b2b1f3e..7c06613 100644 --- a/tests/unit/test_sec_fundamentals_importer.py +++ b/tests/unit/test_sec_fundamentals_importer.py @@ -481,6 +481,209 @@ async def test_unresolved_filing_stops_blocking_after_retry_window(engine): assert await _count(factory, SecFilingGap) == 0 +async def test_queued_gap_without_index_date_retries_without_wedging_and_escalates_once( + engine, +): + factory = _factory(engine) + await _seed(factory, ["AAPL"]) + backfill = FakeSecClient( + tickers={"AAPL": 320193}, + companyfacts={320193: _companyfacts([CF_K, CF_Q1], [SH_K, SH_Q1])}, + submissions={320193: _submissions(SUB_FILINGS)}, + latest_index=date(2026, 1, 31), + ) + await run_import(_importer(backfill), engine=engine) + + old = datetime(2026, 4, 1, tzinfo=timezone.utc) + async with factory() as db: + db.add(SecFilingGap( + cik="0000320193", + accession="DATELESS", + form="10-Q", + index_date=None, + reason="not_in_companyfacts", + first_seen_at=old, + last_attempted_at=old, + )) + await db.commit() + + missing = FakeSecClient( + tickers={"AAPL": 320193}, + companyfacts={320193: _companyfacts([CF_K, CF_Q1], [SH_K, SH_Q1])}, + submissions={320193: _submissions(SUB_FILINGS + [ + _filing( + "DATELESS", + "10-Q", + "2026-03-28", + "2026-05-01", + "2026-05-01T10:01:00.000Z", + ) + ])}, + latest_index=date(2026, 1, 31), + ) + first = await run_import( + _importer(missing, today=date(2026, 5, 20)), engine=engine + ) + second = await run_import( + _importer(missing, today=date(2026, 5, 21)), engine=engine + ) + + assert first.status == STATUS_PROMOTED + assert second.status == STATUS_PROMOTED + async with factory() as db: + gap = (await db.execute(select(SecFilingGap))).scalar_one() + events = ( + await db.execute( + select(SystemEvent).where(SystemEvent.code == "filing_gap_aged") + ) + ).scalars().all() + assert gap.escalated_at is not None + assert len(events) == 1 + + +async def test_queued_filing_reclassified_non_xbrl_is_removed(engine): + factory = _factory(engine) + await _seed(factory, ["AAPL"]) + backfill = FakeSecClient( + tickers={"AAPL": 320193}, + companyfacts={320193: _companyfacts([CF_K, CF_Q1], [SH_K, SH_Q1])}, + submissions={320193: _submissions(SUB_FILINGS)}, + latest_index=date(2026, 1, 31), + ) + await run_import(_importer(backfill), engine=engine) + now = datetime.now(timezone.utc) + async with factory() as db: + db.add(SecFilingGap( + cik="0000320193", + accession="NONX", + form="10-Q/A", + index_date=date(2026, 5, 1), + reason="not_in_companyfacts", + first_seen_at=now, + last_attempted_at=now, + )) + await db.commit() + + client = FakeSecClient( + tickers={"AAPL": 320193}, + companyfacts={320193: _companyfacts([CF_K, CF_Q1], [SH_K, SH_Q1])}, + submissions={320193: _submissions(SUB_FILINGS + [ + _filing( + "NONX", + "10-Q/A", + "2026-03-28", + "2026-05-01", + "2026-05-01T10:01:00.000Z", + is_xbrl=False, + ) + ])}, + latest_index=date(2026, 1, 31), + ) + run = await run_import(_importer(client, today=date(2026, 5, 20)), engine=engine) + + assert run.status == STATUS_PROMOTED + assert await _count(factory, SecFilingGap) == 0 + + +async def test_queued_parser_skip_stays_blocked_with_actionable_reason( + engine, monkeypatch +): + from app.services.sec_facts_parser import ParseResult + + factory = _factory(engine) + await _seed(factory, ["AAPL"]) + backfill = FakeSecClient( + tickers={"AAPL": 320193}, + companyfacts={320193: _companyfacts([CF_K, CF_Q1], [SH_K, SH_Q1])}, + submissions={320193: _submissions(SUB_FILINGS)}, + latest_index=date(2026, 1, 31), + ) + await run_import(_importer(backfill), engine=engine) + now = datetime.now(timezone.utc) + async with factory() as db: + db.add(SecFilingGap( + cik="0000320193", + accession="BADPARSE", + form="10-Q", + index_date=date(2026, 5, 1), + reason="not_in_companyfacts", + first_seen_at=now, + last_attempted_at=now, + )) + await db.commit() + + bad_fact = _rev( + "2025-09-28", "2026-03-28", 254940, 2026, "Q2", "BADPARSE" + ) + bad_share = _shares("2026-04-17", 14687, "BADPARSE", 2026, "Q2") + client = FakeSecClient( + tickers={"AAPL": 320193}, + companyfacts={ + 320193: _companyfacts( + [CF_K, CF_Q1, bad_fact], [SH_K, SH_Q1, bad_share] + ) + }, + submissions={320193: _submissions(SUB_FILINGS + [ + _filing( + "BADPARSE", + "10-Q", + "2026-03-28", + "2026-05-01", + "2026-05-01T10:01:00.000Z", + ) + ])}, + latest_index=date(2026, 1, 31), + ) + + def skip_parse(*args, **kwargs): + return ParseResult(skipped_filings=[{ + "accession": "BADPARSE", + "reason": "unparseable", + }]) + + monkeypatch.setattr("app.services.sec_facts_parser.parse_snapshots", skip_parse) + run = await run_import(_importer(client, today=date(2026, 5, 20)), engine=engine) + + assert run.status == STATUS_PROMOTED + async with factory() as db: + gap = (await db.execute(select(SecFilingGap))).scalar_one() + assert gap.reason == "parser_unusable" + + +async def test_validation_caps_details_but_keeps_complete_blocked_cik_set(): + importer = SecFundamentalsImporter(today=date(2026, 5, 20)) + importer._latest_index_date = date(2026, 5, 19) + staged = StagedFundamentals( + resolved=ResolvedUniverse(), + missing_xbrl=[ + { + "cik": f"{i:010d}", + "accession": f"MISS-{i}", + "form": "10-Q", + "index_date": date(2026, 5, 1), + "age_days": 19, + "reason": "not_in_companyfacts", + } + for i in range(60) + ], + no_xbrl_filings=[ + {"cik": f"{i + 100:010d}", "name": f"New {i}"} + for i in range(60) + ], + recovered=[ + {"cik": f"{i:010d}", "accession": f"REC-{i}", "source_cik": "1"} + for i in range(60) + ], + ) + + result = await importer.validate(None, staged) + + assert len(result.summary["missing_xbrl"]) == 50 + assert len(result.summary["no_xbrl_filings"]) == 50 + assert len(result.summary["recovered_from_coregistrant"]) == 50 + assert len(result.summary["setup_blocked_ciks"]) == 120 + + async def test_non_xbrl_amendment_skipped_not_failed(engine): factory = _factory(engine) await _seed(factory, ["AAPL"])