diff --git a/app/services/sec_fundamentals_importer.py b/app/services/sec_fundamentals_importer.py index 6152b4c..b3322ba 100644 --- a/app/services/sec_fundamentals_importer.py +++ b/app/services/sec_fundamentals_importer.py @@ -51,10 +51,10 @@ import json import logging from collections import Counter, defaultdict from dataclasses import dataclass, field, replace -from datetime import date, datetime, timedelta, timezone +from datetime import date, datetime, time, timedelta, timezone from typing import Any, Callable -from sqlalchemy import delete, select, update +from sqlalchemy import delete, exists, select, update from app.database import insert_for_session from app.models.data_import_run import DataImportRun @@ -81,6 +81,26 @@ 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 + +# Aggregate ceiling on deferral. MISSING_XBRL_RETRY_DAYS bounds how long ONE +# filing blocks; it does not bound how long the import as a whole can stay +# deferred. Those differ because a blocking filing is only queued by promote(), +# which a deferred run never reaches — so during a rolling supply of +# unresolvable filings (earnings season, when SEC's Company-Facts aggregation is +# furthest behind) each new arrival restarts the clock before the previous one +# clears, and nothing is written at all: not the good rows, not the gap rows +# that would stop those filings blocking again. +# +# Once promotions have been stale this long, every unresolved filing is treated +# as past the window. promote() then queues them all (see the _past_retry_window +# call there), source_max_date advances, and _missing() forces queued rows +# aged-out on later runs so they never block again — the import self-heals +# through the paths that already exist. +# +# Well above MISSING_XBRL_RETRY_DAYS so ordinary overlapping blocks never trip +# it. Affected symbols stay barred from setups either way: setup_blocked_ciks is +# built from every missing filing regardless of window. +PROMOTION_CEILING_DAYS = 7 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 @@ -157,6 +177,9 @@ class SecFundamentalsImporter: self._retry_rows: list[dict[str, Any]] = [] self._latest_index_date: date | None = None self._backfill = False + # Set by validate() when the aggregate ceiling forced the block open; + # read by promote() to alert that it did. + self._ceiling_tripped: dict[str, Any] | None = None # -- SourceImporter protocol ------------------------------------------- @@ -421,6 +444,24 @@ class SecFundamentalsImporter: # reconstructible by re-walking the index. blocking = _within_retry_window(staged.missing_xbrl) aged_out = _past_retry_window(staged.missing_xbrl) + + # ...unless promotions have been stale past the aggregate ceiling, in + # which case the deferral has cost more than the filings it withholds. + # Ageing them here (not just locally) is deliberate: promote() re-derives + # the queue from the same list, so this is what gets them queued. + self._ceiling_tripped = None + if blocking and db is not None and await self._promotions_stale(db): + for item in staged.missing_xbrl: + item["age_days"] = max( + item.get("age_days", 0), MISSING_XBRL_RETRY_DAYS + 1 + ) + self._ceiling_tripped = { + "forced": len(blocking), + "unresolved": len(staged.missing_xbrl), + } + 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 " @@ -464,6 +505,9 @@ class SecFundamentalsImporter: "missing_xbrl": staged.missing_xbrl[:50], "missing_xbrl_count": len(staged.missing_xbrl), "missing_xbrl_blocking": len(blocking), + # Present only when the aggregate ceiling forced this run through, so + # a promoted run that carries known-unresolved filings says so. + "promotion_ceiling_tripped": self._ceiling_tripped, "recovered_from_coregistrant": staged.recovered[:50], "recovered_count": len(staged.recovered), # Complete compact gate input; detailed audit lists above stay capped. @@ -616,6 +660,25 @@ class SecFundamentalsImporter: created_at=_now(), )) + # A ceiling-forced promotion is the safety valve firing — it must be + # visible, or the import silently starts carrying known-unresolved + # filings. The affected symbols stay barred from setups regardless. + if self._ceiling_tripped: + db.add(SystemEvent( + severity="warning", + source="sec_facts", + code="promotion_ceiling_forced", + message=( + f"Promoted with {self._ceiling_tripped['unresolved']} unresolved " + f"filing(s) — {self._ceiling_tripped['forced']} still inside the " + f"{MISSING_XBRL_RETRY_DAYS}-day retry window — because nothing had " + f"promoted in {PROMOTION_CEILING_DAYS} days. They are queued for " + "retry and their symbols remain blocked from setups." + )[:4000], + dedup_key=f"sec_facts:promotion_ceiling_forced:{run_id}", + 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) @@ -757,6 +820,38 @@ class SecFundamentalsImporter: if accession not in resolved ] + async def _promotions_stale(self, db) -> bool: + """Has nothing promoted within ``PROMOTION_CEILING_DAYS``? + + Only true for a source that HAS promoted before. A never-promoted import + is initial setup, not a wedge: forcing its first promotion through would + mask a misconfiguration rather than recover from a transient SEC gap. + + Measured from ``self.today`` rather than the wall clock, so the ceiling + honors the same injected date that ages the filings it releases. + """ + cutoff = datetime.combine( + self.today - timedelta(days=PROMOTION_CEILING_DAYS), + time.min, + tzinfo=timezone.utc, + ) + ever, recent = ( + await db.execute( + select( + exists().where( + DataImportRun.source == SOURCE, + DataImportRun.status == STATUS_PROMOTED, + ), + exists().where( + DataImportRun.source == SOURCE, + DataImportRun.status == STATUS_PROMOTED, + DataImportRun.started_at >= cutoff, + ), + ) + ) + ).one() + return bool(ever) and not bool(recent) + async def _last_processed_index_date(self, db) -> date | None: return ( await db.execute( diff --git a/tests/unit/test_sec_fundamentals_importer.py b/tests/unit/test_sec_fundamentals_importer.py index d4f3bf2..7ed2ac2 100644 --- a/tests/unit/test_sec_fundamentals_importer.py +++ b/tests/unit/test_sec_fundamentals_importer.py @@ -6,7 +6,7 @@ from __future__ import annotations import json import os import tempfile -from datetime import date, datetime, timezone +from datetime import date, datetime, timedelta, timezone import pytest from sqlalchemy import func, select @@ -1128,3 +1128,79 @@ async def test_malformed_cik_override_is_ignored_not_fatal(engine): resolved = await resolve_ciks(db, client) assert resolved.symbol_to_cik["AAPL"] == 320193 # fell back to company_tickers + + +# --- aggregate deferral ceiling ------------------------------------------- + +def _blocking_staged(today: date) -> StagedFundamentals: + """One filing still inside the per-filing retry window.""" + return StagedFundamentals( + resolved=ResolvedUniverse(), + missing_xbrl=[{ + "cik": "0000320193", + "accession": "YOUNG-1", + "form": "10-Q", + "index_date": today, + "age_days": 0, + "reason": "not_in_companyfacts", + }], + ) + + +async def _add_run(factory, *, status: str, started_at: datetime) -> None: + from app.models.data_import_run import DataImportRun + + async with factory() as db: + db.add(DataImportRun( + source="sec_facts", status=status, started_at=started_at, + )) + await db.commit() + + +async def _validate_with_history(engine, *, promoted_days_ago: int | None): + factory = _factory(engine) + today = date(2026, 5, 20) + now = datetime(2026, 5, 20, 12, 0, tzinfo=timezone.utc) + if promoted_days_ago is not None: + await _add_run( + factory, + status=STATUS_PROMOTED, + started_at=now - timedelta(days=promoted_days_ago), + ) + importer = SecFundamentalsImporter(today=today) + importer._latest_index_date = date(2026, 5, 19) + staged = _blocking_staged(today) + async with factory() as db: + return await importer.validate(db, staged), staged, importer + + +async def test_ceiling_forces_a_promotion_once_deferral_outlasts_it(engine): + """The per-filing window bounds one filing; this bounds the whole import.""" + result, staged, importer = await _validate_with_history(engine, promoted_days_ago=10) + + assert result.ok is True + assert result.summary["missing_xbrl_blocking"] == 0 + assert result.summary["promotion_ceiling_tripped"] == {"forced": 1, "unresolved": 1} + # Aged in place, so promote() re-derives the same verdict and queues it. + assert staged.missing_xbrl[0]["age_days"] > 3 + # The symbol stays barred from setups — promoting is not trusting the data. + assert result.summary["setup_blocked_ciks"] == ["0000320193"] + + +async def test_a_recent_promotion_keeps_the_normal_block(engine): + result, staged, _ = await _validate_with_history(engine, promoted_days_ago=1) + + assert result.ok is False + assert result.summary["missing_xbrl_blocking"] == 1 + assert result.summary["promotion_ceiling_tripped"] is None + assert result.retryable is True + assert staged.missing_xbrl[0]["age_days"] == 0 + + +async def test_ceiling_never_fires_before_a_first_promotion(engine): + """No baseline means initial setup, not a wedge — forcing it through would + mask a misconfiguration instead of recovering from an SEC gap.""" + result, _, _ = await _validate_with_history(engine, promoted_days_ago=None) + + assert result.ok is False + assert result.summary["promotion_ceiling_tripped"] is None