diff --git a/alembic/versions/028_sec_filing_retry_queue.py b/alembic/versions/028_sec_filing_retry_queue.py new file mode 100644 index 0000000..e9735c2 --- /dev/null +++ b/alembic/versions/028_sec_filing_retry_queue.py @@ -0,0 +1,38 @@ +"""SEC filing retry queue and setup-quality gate + +Revision ID: 028 +Revises: 027 +Create Date: 2026-08-03 00:00:00.000000 +""" +from typing import Sequence, Union + +from alembic import op +import sqlalchemy as sa + + +revision: str = "028" +down_revision: Union[str, None] = "027" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + op.create_table( + "sec_filing_gaps", + sa.Column("id", sa.Integer(), primary_key=True), + sa.Column("cik", sa.String(length=10), nullable=False), + sa.Column("accession", sa.String(length=25), nullable=False), + sa.Column("form", sa.String(length=12), nullable=True), + sa.Column("index_date", sa.Date(), nullable=True), + sa.Column("reason", sa.String(length=64), nullable=False), + 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.UniqueConstraint("accession", name="uq_sec_filing_gaps_accession"), + ) + op.create_index("ix_sec_filing_gaps_cik", "sec_filing_gaps", ["cik"]) + + +def downgrade() -> None: + op.drop_index("ix_sec_filing_gaps_cik", table_name="sec_filing_gaps") + op.drop_table("sec_filing_gaps") diff --git a/app/models/__init__.py b/app/models/__init__.py index eedcf09..198dfb0 100644 --- a/app/models/__init__.py +++ b/app/models/__init__.py @@ -17,6 +17,7 @@ from app.models.regime_snapshot import RegimeSnapshot from app.models.benchmark_price import BenchmarkPrice from app.models.signal_context_snapshot import SignalContextSnapshot from app.models.system_event import SystemEvent +from app.models.sec_filing_gap import SecFilingGap __all__ = [ "Ticker", @@ -40,4 +41,5 @@ __all__ = [ "BenchmarkPrice", "SignalContextSnapshot", "SystemEvent", + "SecFilingGap", ] diff --git a/app/models/sec_filing_gap.py b/app/models/sec_filing_gap.py new file mode 100644 index 0000000..2612b68 --- /dev/null +++ b/app/models/sec_filing_gap.py @@ -0,0 +1,31 @@ +from datetime import date, datetime + +from sqlalchemy import Date, DateTime, Index, String, Text, UniqueConstraint +from sqlalchemy.orm import Mapped, mapped_column + +from app.database import Base + + +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. + """ + + __tablename__ = "sec_filing_gaps" + __table_args__ = ( + UniqueConstraint("accession", name="uq_sec_filing_gaps_accession"), + Index("ix_sec_filing_gaps_cik", "cik"), + ) + + id: Mapped[int] = mapped_column(primary_key=True) + cik: Mapped[str] = mapped_column(String(10), nullable=False) + accession: Mapped[str] = mapped_column(String(25), nullable=False) + form: Mapped[str | None] = mapped_column(String(12), nullable=True) + index_date: Mapped[date | None] = mapped_column(Date, nullable=True) + reason: Mapped[str] = mapped_column(String(64), nullable=False) + 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) diff --git a/app/services/fundamentals_quality_service.py b/app/services/fundamentals_quality_service.py new file mode 100644 index 0000000..91e3ba4 --- /dev/null +++ b/app/services/fundamentals_quality_service.py @@ -0,0 +1,143 @@ +"""Actionability gate for incomplete SEC fundamentals.""" + +from __future__ import annotations + +import json + +from sqlalchemy import exists, select +from sqlalchemy.ext.asyncio import AsyncSession + +from app.models.data_import_run import DataImportRun +from app.models.fundamental_snapshot import FundamentalSnapshot +from app.models.sec_filing_gap import SecFilingGap +from app.models.ticker import Ticker +from app.services import fundamental_data_refresh_service + + +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() + + 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)) + + payload = ( + await db.execute( + select(DataImportRun.validation_json) + .where( + DataImportRun.source == "sec_facts", + DataImportRun.validation_json.is_not(None), + ) + .order_by(DataImportRun.id.desc()) + .limit(1) + ) + ).scalar_one_or_none() + if not payload: + return blocked + try: + summary = json.loads(payload) + except (TypeError, ValueError): + return blocked + + for item in summary.get("missing_xbrl") or []: + if item.get("cik"): + blocked.add(str(item["cik"])) + for item in summary.get("no_xbrl_filings") or []: + if item.get("cik"): + blocked.add(str(item["cik"])) + return blocked + + +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_ticker_ids(db: AsyncSession) -> set[int]: + ciks = await blocked_ciks(db) + if not ciks: + return set() + rows = await db.execute(select(Ticker.id).where(Ticker.cik.in_(ciks))) + return {int(ticker_id) for ticker_id in rows.scalars()} + + +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 7c1503b..ce5158b 100644 --- a/app/services/rr_scanner_service.py +++ b/app/services/rr_scanner_service.py @@ -27,6 +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.price_service import query_ohlcv from app.services.qualification import setup_qualifies from app.services.sr_service import detect_gate_target_ladder @@ -526,6 +527,7 @@ async def scan_ticker( primary_min_rr: float | None = None, gate_levels_override: list[Any] | None = None, scan_run_id: str | None = None, + fundamentals_eligible: bool | None = None, ) -> list[TradeSetup]: """Scan a single ticker for trade setups meeting the R:R threshold. @@ -542,6 +544,17 @@ async def scan_ticker( """ ticker = await _get_ticker(db, symbol) + if fundamentals_eligible is None: + fundamentals_eligible = await fundamentals_quality_service.ticker_is_eligible( + db, ticker.id + ) + if not fundamentals_eligible: + logger.info( + "Skipping %s: unresolved or unavailable SEC fundamentals", + ticker.symbol, + ) + return [] + if primary_min_rr is None: primary_min_rr = PRIMARY_TARGET_MIN_RR @@ -726,6 +739,19 @@ async def scan_all_tickers( ticker_rows = [(int(ticker_id), symbol) for ticker_id, symbol in result.all()] total = len(ticker_rows) + # Data-quality failures are not weak signals: they make a ticker ineligible. + # Resolve once for the universe scan and pass the decision into scan_ticker. + try: + fundamentals_blocked_ids = ( + await fundamentals_quality_service.blocked_ticker_ids(db) + ) + except Exception: + await db.rollback() + logger.exception( + "Could not resolve fundamentals quality; blocking this scan closed" + ) + fundamentals_blocked_ids = {ticker_id for ticker_id, _ in ticker_rows} + # Gate-reset observations must use the same runtime activation settings as # the live setup list. If the config cannot be loaded, scan normally but do # not mutate reset state from an evaluation whose rules are unknown. @@ -765,6 +791,12 @@ async def scan_all_tickers( for index, (ticker_id, symbol) in enumerate(ticker_rows): if progress_callback is not None: progress_callback(index, total, symbol) + if ticker_id in fundamentals_blocked_ids: + logger.info( + "Skipping %s: unresolved or unavailable SEC fundamentals", + symbol, + ) + continue # Refresh Structural S/R once, then scores. get_sr_levels is read-only; # without this recalculate the score path would see yesterday's zones. # A refresh failure still scans the ticker: qualification re-gates on @@ -795,6 +827,7 @@ async def scan_all_tickers( volatility_percentile=(ranks.get(symbol) or {}).get("volatility_percentile"), primary_min_rr=PRIMARY_TARGET_MIN_RR, scan_run_id=scan_run_id, + fundamentals_eligible=True, ) all_setups.extend(setups) if activation is not None: @@ -882,6 +915,16 @@ async def get_trade_setups( stmt = stmt.where(TradeSetup.recommended_action == recommended_action) excluded_ticker_ids: set[int] = set() reentry_gate_locks: dict[int, datetime] = {} + try: + excluded_ticker_ids.update( + await fundamentals_quality_service.blocked_ticker_ids(db) + ) + except Exception: + await db.rollback() + logger.exception( + "Could not resolve fundamentals quality; hiding actionable setups" + ) + return [] if exclude_open_trade_tickers: # Manual book only. The shadow book holds the *top-ranked* names by # construction, so letting its positions hide setups would leave the diff --git a/app/services/sec_fundamentals_importer.py b/app/services/sec_fundamentals_importer.py index 579327b..2dbbb25 100644 --- a/app/services/sec_fundamentals_importer.py +++ b/app/services/sec_fundamentals_importer.py @@ -31,10 +31,9 @@ Guardrails (design + reviews): stored as the parent's. Confirmed 2026-07-27 (NEE via FPL, DOW via Dow Chemical) and it is not transient: an NEE filing misattributed in 2014 is still misfiled. - **Bounded blocking.** Anything still unresolvable after ``MISSING_XBRL_RETRY_DAYS`` - stops failing the run and is promoted around, with a named ``unresolved_filing`` - warning. One filing SEC misfiled must not wedge every later import; the index - only moves forward, so a tolerated accession returns only via a reparse, and - only once SEC has re-filed it under the filer's own CIK. + stops failing the whole import and enters a durable retry queue. The scheduled + importer retries queued accessions automatically, while the affected issuer is + excluded from actionable setups until its filing is recovered. - ``promote`` inserts snapshots ``ON CONFLICT (accession) DO NOTHING`` (immutable), reports differing existing accessions, and applies ticker updates in the same transaction. @@ -48,18 +47,21 @@ Guardrails (design + reviews): from __future__ import annotations +import json import logging from collections import Counter, defaultdict from dataclasses import dataclass, field, replace from datetime import date, datetime, timedelta, timezone from typing import Any, Callable -from sqlalchemy import select, update +from sqlalchemy import delete, select, update from app.database import insert_for_session from app.models.data_import_run import DataImportRun from app.models.fundamental_snapshot import FundamentalSnapshot +from app.models.sec_filing_gap import SecFilingGap from app.models.system_event import SystemEvent +from app.services import fundamentals_quality_service from app.services import sec_facts_parser as parser from app.services import sec_universe from app.services.data_import import STATUS_PROMOTED, ValidationResult @@ -151,6 +153,8 @@ class SecFundamentalsImporter: # accession -> the OTHER CIKs the daily index lists it under (co-registrants # 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 @@ -176,9 +180,23 @@ class SecFundamentalsImporter: client, last_processed, self._latest_index_date ) content = sec_universe.index_content_hash(self._index_rows) - return sec_universe.compose_revision( + revision = sec_universe.compose_revision( self._latest_index_date, content, self._resolved.symbol_to_cik ) + 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 async def stage(self, db) -> StagedFundamentals: assert self._resolved is not None, "detect_revision must run first" @@ -193,6 +211,27 @@ 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. + if not self._backfill: + seen = { + (int(cik), row["accession"]) + for cik, rows in filed_by_cik.items() + for row in rows + } + for row in self._retry_rows: + cik = int(row["cik"]) + key = (cik, row["accession"]) + if key in seen: + continue + filed_by_cik[cik].append(row) + seen.add(key) + coregistrants = [int(value) for value in row.get("coregistrants") or []] + if coregistrants: + self._coregistrants[row["accession"]] = coregistrants + existing = await self._ciks_with_snapshots(db, set(cik_to_tids)) if self._backfill: backfill_ciks = set(cik_to_tids) @@ -262,7 +301,13 @@ class SecFundamentalsImporter: # or a co-registrant filing). NOT a Company-Facts lag — separate # cause, separate fix, so it gets its own reason. staged.missing_xbrl.append( - _missing(cik, index_row, "not_in_submissions", self.today) + _missing( + cik, + index_row, + "not_in_submissions", + self.today, + self._coregistrants.get(accn), + ) ) elif accn in present: accns.add(accn) @@ -289,6 +334,7 @@ class SecFundamentalsImporter: "coregistrant_facts_rejected" if source_cik else "not_in_companyfacts", self.today, + self._coregistrants.get(accn), )) # fiscalYearEnd (MMDD) is what lets the parser derive period identity from @@ -396,12 +442,12 @@ 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[:50], + "no_xbrl_filings": staged.no_xbrl_filings, "no_xbrl_filings_count": len(staged.no_xbrl_filings), - "missing_xbrl": staged.missing_xbrl[:50], + "missing_xbrl": staged.missing_xbrl, "missing_xbrl_count": len(staged.missing_xbrl), "missing_xbrl_blocking": len(blocking), - "recovered_from_coregistrant": staged.recovered[:50], + "recovered_from_coregistrant": staged.recovered, "recovered_count": len(staged.recovered), "invalid_payloads": staged.invalid_payloads, "cik_updates": len(staged.resolved.cik_updates), @@ -426,8 +472,8 @@ class SecFundamentalsImporter: 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"the {MISSING_XBRL_RETRY_DAYS}-day retry window. They will " + f"enter automatic retry and block affected symbols from setups: " f"{_missing_detail(aged_out)}" ] if aged_out @@ -462,6 +508,53 @@ 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() + ) + resolved_accessions = {row.accession for row in staged.rows} + queue_resolved = 0 + if resolved_accessions: + result = await db.execute( + delete(SecFilingGap).where( + SecFilingGap.accession.in_(resolved_accessions) + ) + ) + queue_resolved = int(result.rowcount or 0) + + tolerated = _past_retry_window(staged.missing_xbrl) + now = _now() + for gap in tolerated: + stmt = insert_for_session(db, SecFilingGap).values( + cik=gap["cik"], + accession=gap["accession"], + form=gap.get("form"), + index_date=gap.get("index_date"), + reason=gap["reason"], + coregistrant_ciks_json=json.dumps(gap.get("coregistrants") or []), + first_seen_at=now, + last_attempted_at=now, + ) + await db.execute( + stmt.on_conflict_do_update( + index_elements=["accession"], + set_={ + "cik": stmt.excluded.cik, + "form": stmt.excluded.form, + "index_date": stmt.excluded.index_date, + "reason": stmt.excluded.reason, + "coregistrant_ciks_json": stmt.excluded.coregistrant_ciks_json, + "last_attempted_at": stmt.excluded.last_attempted_at, + }, + ) + ) + newly_queued = [ + gap for gap in tolerated + if gap["accession"] not in existing_gap_accessions + ] + # Warn (in-transaction, so it commits atomically with the promotion) when # any existing accession reconstructed differently — kept immutable. if staged.discrepancies: @@ -487,63 +580,47 @@ class SecFundamentalsImporter: named = ", ".join( f"{r['accession']} <- CIK {r['source_cik']}" for r in staged.recovered[:10] ) - db.add(SystemEvent( - severity="warning", - source="sec_facts", - code="coregistrant_recovery", - message=( - f"{len(staged.recovered)} filing(s) were absent from the filer's " - f"own Company Facts and were parsed from a co-registrant's file " - f"instead (share count checked against the issuer's history): {named}" - )[:4000], - dedup_key=f"sec_facts:coregistrant_recovery:{run_id}", - created_at=_now(), - )) + logger.info( + "sec_facts: recovered %d filing(s) from co-registrants: %s", + len(staged.recovered), + named, + ) - # Filings past the retry window: promoted WITHOUT them so one misfiled - # filing cannot wedge every later import. This is the deliberate trade — - # loud and named, because the index only moves forward and nothing will - # revisit them on its own. - tolerated = _past_retry_window(staged.missing_xbrl) - if tolerated: + # One warning when a gap first enters automatic retry. Repeating it every + # day adds noise; the queue remains the durable actionable state. + if newly_queued: + symbols_by_cik: dict[str, list[str]] = defaultdict(list) + for symbol, cik in staged.resolved.symbol_to_cik.items(): + symbols_by_cik[cik10(cik)].append(symbol) + named = ", ".join( + f"{'/'.join(symbols_by_cik.get(gap['cik'], [])) or gap['cik']}" + f"/{gap['accession']}" + for gap in newly_queued[:10] + ) db.add(SystemEvent( severity="warning", source="sec_facts", code="unresolved_filing", message=( - f"{len(tolerated)} tracked filing(s) still unresolvable after " - f"{MISSING_XBRL_RETRY_DAYS} days; promoting without them rather " - f"than blocking every later import. They are NOT retried. " - f"scripts/reparse_fundamentals.py recovers them ONLY once SEC " - f"re-files under the filer's own CIK — it walks no index, so it " - f"cannot reach facts still sitting under a co-registrant: " - f"{_missing_detail(tolerated)}" + f"{len(newly_queued)} filing(s) entered automatic SEC retry. " + f"Affected symbols are blocked from new actionable setups until " + f"their filing is recovered: {named}" )[:4000], dedup_key=f"sec_facts:unresolved_filing:{run_id}", created_at=_now(), )) - # A tracked issuer whose registrant has no XBRL filings can never produce a - # snapshot, and it is restaged on every run forever. That is a resolution - # problem, not missing data, and it is silent without this. + # A new registrant may have no XBRL filing yet. Keep it out of actionable + # setups, but log it instead of raising a recurring operator warning. if staged.no_xbrl_filings: named = ", ".join( f"{e['cik']} ({e.get('name') or '?'})" for e in staged.no_xbrl_filings[:10] ) - db.add(SystemEvent( - severity="warning", - source="sec_facts", - code="no_xbrl_filings", - message=( - f"{len(staged.no_xbrl_filings)} tracked issuer(s) resolved to a " - f"registrant with no XBRL 10-K/10-Q. Either a successor shell " - f"(pin the real filer via the '{sec_universe.CIK_OVERRIDES_KEY}' " - f"setting) or a new registrant that has not filed its first " - f"10-K/10-Q yet, which needs nothing and clears itself: {named}" - )[:4000], - dedup_key=f"sec_facts:no_xbrl_filings:{run_id}", - created_at=_now(), - )) + logger.info( + "sec_facts: %d registrant(s) have no XBRL history yet: %s", + len(staged.no_xbrl_filings), + named, + ) ticker_counts = await sec_universe.apply_ticker_updates( db, staged.resolved, staged.sic_updates @@ -553,11 +630,105 @@ class SecFundamentalsImporter: "updated": updated, "existing_unchanged": len(staged.existing_accessions) - updated, "discrepancies": len(staged.discrepancies), + "retry_queue_added": len(newly_queued), + "retry_queue_resolved": queue_resolved, **ticker_counts, } # -- helpers ----------------------------------------------------------- + async def _retry_backlog( + self, + db, + tracked_ciks: set[int], + *, + include_history: bool, + ) -> list[dict[str, Any]]: + """Active gaps plus pre-queue gaps from promoted validation history.""" + 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() + for gap in queued: + try: + coregistrants = json.loads(gap.coregistrant_ciks_json or "[]") + except (TypeError, ValueError): + coregistrants = [] + candidates[gap.accession] = { + "cik": gap.cik, + "accession": gap.accession, + "form": gap.form, + "index_date": gap.index_date, + "reason": gap.reason, + "coregistrants": coregistrants, + } + + # 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( + ( + await db.execute( + select(FundamentalSnapshot.accession).where( + FundamentalSnapshot.accession.in_(list(candidates)) + ) + ) + ).scalars().all() + ) + return [ + item + for accession, item in candidates.items() + if accession not in resolved + ] + async def _last_processed_index_date(self, db) -> date | None: return ( await db.execute( @@ -666,7 +837,13 @@ def _companyfacts_structure_error(cf: Any) -> str | None: return None -def _missing(cik: int, row: dict[str, Any], reason: str, today: date) -> dict[str, Any]: +def _missing( + cik: int, + row: dict[str, Any], + reason: str, + today: date, + coregistrants: list[int] | None = None, +) -> dict[str, Any]: """One unresolvable index row, carrying everything needed to look the filing 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.""" @@ -679,6 +856,7 @@ def _missing(cik: int, row: dict[str, Any], reason: str, today: date) -> dict[st # 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, "reason": reason, + "coregistrants": list(coregistrants or []), } diff --git a/docs/fundamentals-deployment.md b/docs/fundamentals-deployment.md index 1b0ea7f..f68d3f3 100644 --- a/docs/fundamentals-deployment.md +++ b/docs/fundamentals-deployment.md @@ -15,6 +15,9 @@ not add OS cron entries: the application scheduler owns both jobs. - Cron expressions are editable in Admin → Schedule. - Every attempt is recorded in `data_import_runs`; failures also create a system 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. 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 @@ -231,6 +234,9 @@ least several scheduled cycles before A6 removes the legacy providers. must stop. Existing promoted snapshots/events remain available. - 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. - 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/tests/unit/test_fundamentals_quality_service.py b/tests/unit/test_fundamentals_quality_service.py new file mode 100644 index 0000000..fdb85f3 --- /dev/null +++ b/tests/unit/test_fundamentals_quality_service.py @@ -0,0 +1,113 @@ +from __future__ import annotations + +import json +from datetime import date, datetime, timezone + +from app.models.data_import_run import DataImportRun +from app.models.fundamental_snapshot import FundamentalSnapshot +from app.models.sec_filing_gap import SecFilingGap +from app.models.settings import SystemSetting +from app.models.ticker import Ticker +from app.services import fundamentals_quality_service + + +async def test_latest_sec_validation_blocks_deferred_and_no_history_ciks( + db_session, +): + missing = Ticker(symbol="MISSING", cik="0000000001") + no_history = Ticker(symbol="NEWREG", cik="0000000002") + healthy = Ticker(symbol="HEALTHY", cik="0000000003") + db_session.add_all([missing, no_history, healthy]) + await db_session.flush() + db_session.add( + SystemSetting( + key="fundamental_data_sec_dolt_cutover_enabled", + value="true", + ) + ) + db_session.add( + DataImportRun( + source="sec_facts", + status="deferred", + validation_json=json.dumps({ + "missing_xbrl": [{"cik": missing.cik, "accession": "MISSING-Q"}], + "no_xbrl_filings": [{"cik": no_history.cik}], + }), + started_at=datetime.now(timezone.utc), + ) + ) + await db_session.flush() + + assert await fundamentals_quality_service.blocked_ticker_ids(db_session) == { + missing.id, + no_history.id, + } + + +async def test_sec_quality_gate_is_inactive_before_cutover(db_session): + ticker = Ticker(symbol="SHADOW", cik="0000000042") + db_session.add(ticker) + await db_session.flush() + now = datetime.now(timezone.utc) + db_session.add( + SecFilingGap( + cik=ticker.cik, + accession="SHADOW-Q", + form="10-Q", + index_date=date.today(), + reason="not_in_companyfacts", + first_seen_at=now, + last_attempted_at=now, + ) + ) + await db_session.flush() + + assert await fundamentals_quality_service.blocked_ticker_ids(db_session) == set() + + + + +async def test_promoted_gap_is_blocked_during_queue_migration_bootstrap(db_session): + ticker = Ticker(symbol="HIST", cik="0000000043") + 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), + ), + ]) + await db_session.flush() + + assert await fundamentals_quality_service.blocked_ticker_ids(db_session) == { + ticker.id + } + + db_session.add( + FundamentalSnapshot( + cik=ticker.cik, + accession="HIST-Q", + form="10-Q", + filed_date=date.today(), + accepted_at=datetime.now(timezone.utc), + period_end=date.today(), + fiscal_year=date.today().year, + fiscal_period="Q2", + ) + ) + await db_session.flush() + + assert await fundamentals_quality_service.blocked_ticker_ids(db_session) == set() diff --git a/tests/unit/test_rr_scanner_preservation.py b/tests/unit/test_rr_scanner_preservation.py index c5f6cd5..b9fba53 100644 --- a/tests/unit/test_rr_scanner_preservation.py +++ b/tests/unit/test_rr_scanner_preservation.py @@ -23,6 +23,8 @@ from sqlalchemy.ext.asyncio import AsyncSession from app.models.ohlcv import OHLCVRecord from app.models.paper_trade import PaperTrade from app.models.signal_context_snapshot import SignalContextSnapshot +from app.models.sec_filing_gap import SecFilingGap +from app.models.settings import SystemSetting from app.models.sr_level import SRLevel from app.models.ticker import Ticker from app.models.trade_setup import TradeSetup @@ -513,6 +515,45 @@ async def test_get_trade_setups_excludes_stale_rows(db_session: AsyncSession): assert stale_rows == [] +@pytest.mark.asyncio +async def test_get_trade_setups_hides_active_sec_filing_gap( + db_session: AsyncSession, +): + now = datetime.now(timezone.utc) + ticker = Ticker(symbol="SECWAIT", cik="0000000042") + db_session.add(ticker) + await db_session.flush() + db_session.add_all([ + SystemSetting( + key="fundamental_data_sec_dolt_cutover_enabled", + value="true", + ), + SecFilingGap( + cik=ticker.cik, + accession="0000000042-26-000001", + form="10-Q", + index_date=date.today(), + reason="not_in_companyfacts", + first_seen_at=now, + last_attempted_at=now, + ), + TradeSetup( + ticker_id=ticker.id, + direction="long", + entry_price=100.0, + stop_loss=97.0, + target=109.0, + rr_ratio=3.0, + composite_score=70.0, + confidence_score=80.0, + detected_at=now, + ), + ]) + await db_session.flush() + + assert await get_trade_setups(db_session, symbol="SECWAIT") == [] + + @pytest.mark.asyncio async def test_get_trade_setups_can_exclude_tickers_with_open_paper_trades( db_session: AsyncSession, diff --git a/tests/unit/test_rr_scanner_scan_all.py b/tests/unit/test_rr_scanner_scan_all.py index 27f6775..c46b475 100644 --- a/tests/unit/test_rr_scanner_scan_all.py +++ b/tests/unit/test_rr_scanner_scan_all.py @@ -108,3 +108,26 @@ async def test_scan_error_does_not_stop_later_tickers(session, monkeypatch): await rr_scanner_service.scan_all_tickers(session) assert scanned == ["AAA", "BBB"] + + +async def test_scan_skips_ticker_with_incomplete_sec_fundamentals( + session, monkeypatch +): + ticker = Ticker(symbol="BLOCKED", cik="0000000001") + session.add(ticker) + await session.commit() + + async def _blocked(db): + return {ticker.id} + + async def _unexpected_scan(*args, **kwargs): + raise AssertionError("fundamentals-incomplete ticker was scanned") + + monkeypatch.setattr( + rr_scanner_service.fundamentals_quality_service, + "blocked_ticker_ids", + _blocked, + ) + monkeypatch.setattr(rr_scanner_service, "scan_ticker", _unexpected_scan) + + assert await rr_scanner_service.scan_all_tickers(session) == [] diff --git a/tests/unit/test_sec_fundamentals_importer.py b/tests/unit/test_sec_fundamentals_importer.py index 8f10dd3..b2b1f3e 100644 --- a/tests/unit/test_sec_fundamentals_importer.py +++ b/tests/unit/test_sec_fundamentals_importer.py @@ -15,6 +15,7 @@ from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_asyn from app.database import Base import app.models # noqa: F401 from app.models.fundamental_snapshot import FundamentalSnapshot +from app.models.sec_filing_gap import SecFilingGap from app.models.system_event import SystemEvent from app.models.ticker import Ticker from app.services.data_import import ( @@ -374,7 +375,7 @@ async def test_recovers_facts_misfiled_under_coregistrant(engine): # Stamped to the issuer that filed, NOT the co-registrant whose file it came from. assert q2.cik == "0000320193" assert q2.revenue == 254940 and q2.shares_outstanding == 14687 - assert "coregistrant_recovery" in codes + assert "coregistrant_recovery" not in codes async def test_coregistrant_recovery_rejects_discontinuous_share_count(engine): @@ -425,11 +426,59 @@ async def test_unresolved_filing_stops_blocking_after_retry_window(engine): assert run.source_max_date == date(2026, 5, 2) # and the index advances summary = json.loads(run.validation_json or "{}") assert summary["missing_xbrl_count"] == 1 and summary["missing_xbrl_blocking"] == 0 + assert await _count(factory, SecFilingGap) == 1 async with factory() as s: events = (await s.execute(select(SystemEvent))).scalars().all() unresolved = [e for e in events if e.code == "unresolved_filing"] assert len(unresolved) == 1 and "GHOST" in unresolved[0].message + assert "automatic SEC retry" in unresolved[0].message + + # The next scheduled run retries even though the SEC daily-index revision + # has not changed. Company Facts is a separate SEC product and may catch up + # independently, so the generic revision no-op must not suppress this work. + still_missing_run = await run_import( + _importer(incr, today=date(2026, 5, 11)), + engine=engine, + ) + assert still_missing_run.status == STATUS_PROMOTED + assert still_missing_run.revision is None + assert await _count(factory, SecFilingGap) == 1 + + # A later normal scheduled import retries only the queued issuer. Once SEC + # publishes the accession in Company Facts, it is inserted and unblocked + # without a full-universe reparse or operator action. + cf_ghost = _rev("2025-09-28", "2026-03-28", 254940, 2026, "Q2", "GHOST") + sh_ghost = _shares("2026-04-17", 14687, "GHOST", 2026, "Q2") + healed = FakeSecClient( + tickers={"AAPL": 320193}, + companyfacts={ + 320193: _companyfacts( + [CF_K, CF_Q1, cf_ghost], + [SH_K, SH_Q1, sh_ghost], + ) + }, + submissions={ + 320193: _submissions(SUB_FILINGS + [ + _filing( + "GHOST", + "10-Q", + "2026-03-28", + "2026-05-01", + "2026-05-01T10:01:00.000Z", + ) + ]) + }, + latest_index=date(2026, 5, 2), + ) + healed_run = await run_import( + _importer(healed, today=date(2026, 5, 12)), + engine=engine, + ) + + assert healed_run.status == STATUS_PROMOTED + assert await _count(factory, FundamentalSnapshot) == 3 + assert await _count(factory, SecFilingGap) == 0 async def test_non_xbrl_amendment_skipped_not_failed(engine):