diff --git a/alembic/versions/034_sec_filing_gap_exemption.py b/alembic/versions/034_sec_filing_gap_exemption.py new file mode 100644 index 0000000..2ca3f9e --- /dev/null +++ b/alembic/versions/034_sec_filing_gap_exemption.py @@ -0,0 +1,41 @@ +"""Track when a filing gap stops pausing setups + +Revision ID: 034 +Revises: 033 +Create Date: 2026-08-21 00:00:00.000000 + +An escalated gap stops pausing setups while the issuer's own fundamentals are +still recent (``GAP_GATE_RECENT_FILING_DAYS``). That reprieve is not permanent: +the stored filings age out, or a newer gap appears, and the pause returns — +silently, because ``filing_gap_aged`` only escalates gaps whose ``escalated_at`` +is NULL and so never fires twice for the same gap. + +``exempted_at`` is the state marker that makes the transition observable. It is +set (quietly) while the issuer is exempt and cleared when the exemption lapses, +which is when ``filing_gap_repaused`` fires — once per lapse, re-arming if the +issuer's data recovers and ages out again. + +Nullable, and carrying no meaning of its own beyond that state: an existing gap +starts NULL and is stamped on the next import that finds it exempt. +""" +from typing import Sequence, Union + +from alembic import op +import sqlalchemy as sa + + +revision: str = "034" +down_revision: Union[str, None] = "033" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + op.add_column( + "sec_filing_gaps", + sa.Column("exempted_at", sa.DateTime(timezone=True), nullable=True), + ) + + +def downgrade() -> None: + op.drop_column("sec_filing_gaps", "exempted_at") diff --git a/app/models/sec_filing_gap.py b/app/models/sec_filing_gap.py index a475d07..0e77200 100644 --- a/app/models/sec_filing_gap.py +++ b/app/models/sec_filing_gap.py @@ -30,3 +30,8 @@ class SecFilingGap(Base): 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) + # Set while this gap's issuer is exempt from the setup pause (escalated, and + # its own fundamentals still recent — see fundamentals_quality_service). + # Cleared when the exemption lapses, which is the moment the pause silently + # comes back and the only moment worth alerting on. + exempted_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) diff --git a/app/services/fundamentals_quality_service.py b/app/services/fundamentals_quality_service.py index 705e040..f55d1a1 100644 --- a/app/services/fundamentals_quality_service.py +++ b/app/services/fundamentals_quality_service.py @@ -70,13 +70,16 @@ async def active_gaps( return list((await db.execute(stmt)).scalars().all()) -async def _gap_exempt_ciks( +async def gap_exempt_ciks( db: AsyncSession, gaps: list[SecFilingGap] ) -> set[str]: """CIKs whose gaps have stopped pausing setups (see GAP_GATE_RECENT_FILING_DAYS). Every one of a CIK's active gaps must be escalated: one fresh gap alongside an old one still means a filing we might yet ingest, which is worth pausing for. + + Public because the importer alerts on this exact transition (a CIK dropping + out of this set is a pause coming back on) and the rule must not exist twice. """ by_cik: dict[str, list[SecFilingGap]] = defaultdict(list) for gap in gaps: @@ -135,7 +138,7 @@ async def blocked_reasons_by_cik( gaps = await active_gaps(db, ciks) # Escalated gaps on issuers that still have recent fundamentals no longer # pause setups, on either path below — the summary mirrors the same filings. - exempt = await _gap_exempt_ciks(db, gaps) + exempt = await gap_exempt_ciks(db, gaps) reasons = { gap.cik: "sec_filing_gap" for gap in gaps if gap.cik not in exempt } diff --git a/app/services/sec_fundamentals_importer.py b/app/services/sec_fundamentals_importer.py index 7789dbf..02207da 100644 --- a/app/services/sec_fundamentals_importer.py +++ b/app/services/sec_fundamentals_importer.py @@ -776,6 +776,60 @@ class SecFundamentalsImporter: .values(escalated_at=now) ) + # The escalation above fires once per gap, so nothing would report the + # *end* of the reprieve it grants. An escalated gap stops pausing setups + # while the issuer's own fundamentals are still recent, and that lapses + # on its own — the stored filings age past the window, or a newer gap + # appears — putting the pause back on with no alert anywhere. Track the + # exemption as state and alert on the transition, once per lapse. + current_gaps = await fundamentals_quality_service.active_gaps(db) + escalated_gaps = [g for g in current_gaps if g.escalated_at is not None] + if escalated_gaps: + exempt_ciks = await fundamentals_quality_service.gap_exempt_ciks( + db, escalated_gaps + ) + newly_exempt = [ + g for g in escalated_gaps + if g.cik in exempt_ciks and g.exempted_at is None + ] + lapsed = [ + g for g in escalated_gaps + if g.cik not in exempt_ciks and g.exempted_at is not None + ] + if newly_exempt: + # Silent on purpose: filing_gap_aged already announced this gap, + # and setups resuming is the behaviour that alert describes. + await db.execute( + update(SecFilingGap) + .where(SecFilingGap.id.in_([g.id for g in newly_exempt])) + .values(exempted_at=now) + ) + if lapsed: + named = ", ".join( + f"{gap.cik}/{gap.accession}" for gap in lapsed[:10] + ) + db.add(SystemEvent( + severity="warning", + source="sec_facts", + code="filing_gap_repaused", + message=( + f"{len(lapsed)} SEC filing gap(s) pause setups again: the " + "issuer's own fundamentals have aged out of the " + f"{fundamentals_quality_service.GAP_GATE_RECENT_FILING_DAYS}" + "-day window, or a newer gap arrived, so there is nothing " + f"recent left to score on: {named}" + )[:4000], + dedup_key=f"sec_facts:filing_gap_repaused:{run_id}", + created_at=now, + )) + # Cleared, not stamped: the issuer can recover and age out again, + # and each lapse is worth its own alert. + await db.execute( + update(SecFilingGap) + .where(SecFilingGap.id.in_([g.id for g in lapsed])) + .values(exempted_at=None) + ) + # 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: diff --git a/tests/unit/test_fundamentals_quality_service.py b/tests/unit/test_fundamentals_quality_service.py index 51bbd5f..d93812e 100644 --- a/tests/unit/test_fundamentals_quality_service.py +++ b/tests/unit/test_fundamentals_quality_service.py @@ -257,3 +257,25 @@ async def test_summary_path_does_not_reblock_an_exempt_cik(db_session): await db_session.flush() assert await fundamentals_quality_service.blocked_ticker_ids(db_session) == set() + + +async def test_a_newer_gap_ends_the_exemption(db_session): + """Production's second exit path: Q3 also fails to ingest, so an un-escalated + gap joins the escalated one and the issuer pauses again immediately.""" + ticker = Ticker(symbol="NEWGAP", cik="0000000050") + db_session.add_all([ + ticker, _escalated_gap(ticker.cik), _prior_quarter(ticker.cik, age_days=120) + ]) + await db_session.flush() + assert await fundamentals_quality_service.blocked_ticker_ids(db_session) == set() + + fresh = datetime.now(timezone.utc) + db_session.add(SecFilingGap( + cik=ticker.cik, accession="NEWGAP-Q3", form="10-Q", index_date=date.today(), + reason="not_in_companyfacts", first_seen_at=fresh, last_attempted_at=fresh, + )) + await db_session.flush() + + assert await fundamentals_quality_service.blocked_ticker_ids(db_session) == { + ticker.id + } diff --git a/tests/unit/test_sec_fundamentals_importer.py b/tests/unit/test_sec_fundamentals_importer.py index b437d63..f8fb013 100644 --- a/tests/unit/test_sec_fundamentals_importer.py +++ b/tests/unit/test_sec_fundamentals_importer.py @@ -1388,3 +1388,79 @@ async def test_reparse_never_restamps_a_collision_onto_the_co_registrant(engine) async with factory() as s: row = (await s.execute(select(FundamentalSnapshot))).scalar_one() assert row.cik == "0000906107" # still the issuer that filed it + + +# --- the reprieve ending: an exemption that lapses must not do so silently --- + +def _stale_gap(cik, *, exempted: bool): + now = datetime.now(timezone.utc) + return SecFilingGap( + cik=cik, accession=f"{cik}-AGED-Q", form="10-Q", + index_date=(now - timedelta(days=30)).date(), reason="not_in_companyfacts", + first_seen_at=now - timedelta(days=30), last_attempted_at=now, + escalated_at=now - timedelta(days=16), + exempted_at=(now - timedelta(days=16)) if exempted else None, + ) + + +def _snapshot(cik, *, age_days): + filed = date.today() - timedelta(days=age_days) + return FundamentalSnapshot( + cik=cik, accession=f"{cik}-PRIOR", form="10-Q", filed_date=filed, + accepted_at=datetime.now(timezone.utc) - timedelta(days=age_days), + period_end=filed, fiscal_year=filed.year, fiscal_period="Q1", + ) + + +async def _promote_only(engine, seed): + """Run promote() alone against seeded gap/snapshot state.""" + factory = _factory(engine) + async with factory() as s: + for obj in seed: + s.add(obj) + await s.commit() + importer = _importer(FakeSecClient( + tickers={}, companyfacts={}, submissions={}, latest_index=date(2026, 3, 1) + )) + async with factory() as db: + await importer.promote(db, StagedFundamentals(resolved=ResolvedUniverse()), run_id=77) + await db.commit() + async with factory() as s: + gaps = (await s.execute(select(SecFilingGap))).scalars().all() + events = (await s.execute(select(SystemEvent))).scalars().all() + return gaps, events + + +async def test_a_lapsed_exemption_raises_its_own_alert(engine): + """filing_gap_aged fires once and never again, so nothing else would say the + pause came back when the issuer's own fundamentals aged out.""" + cik = "0000000060" + gaps, events = await _promote_only( + engine, [_stale_gap(cik, exempted=True), _snapshot(cik, age_days=400)] + ) + repaused = [e for e in events if e.code == "filing_gap_repaused"] + assert len(repaused) == 1 + assert f"{cik}/{cik}-AGED-Q" in repaused[0].message + # Cleared, so a later recovery can re-arm and lapse again. + assert gaps[0].exempted_at is None + + +async def test_an_exemption_taking_effect_is_stamped_silently(engine): + """Setups resuming is what filing_gap_aged already described — stamping the + state must not raise a second alert for it.""" + cik = "0000000061" + gaps, events = await _promote_only( + engine, [_stale_gap(cik, exempted=False), _snapshot(cik, age_days=120)] + ) + assert [e.code for e in events if e.code.startswith("filing_gap")] == [] + assert gaps[0].exempted_at is not None + + +async def test_a_still_paused_gap_is_not_reported_as_lapsing(engine): + """It never became exempt, so there is no transition to report.""" + cik = "0000000062" + gaps, events = await _promote_only( + engine, [_stale_gap(cik, exempted=False), _snapshot(cik, age_days=400)] + ) + assert [e for e in events if e.code == "filing_gap_repaused"] == [] + assert gaps[0].exempted_at is None