diff --git a/app/services/fundamentals_quality_service.py b/app/services/fundamentals_quality_service.py index 161e025..705e040 100644 --- a/app/services/fundamentals_quality_service.py +++ b/app/services/fundamentals_quality_service.py @@ -3,7 +3,9 @@ from __future__ import annotations import json +from collections import defaultdict from dataclasses import dataclass +from datetime import datetime, timedelta, timezone from sqlalchemy import exists, func, select from sqlalchemy.ext.asyncio import AsyncSession @@ -15,6 +17,23 @@ from app.models.ticker import Ticker _SEC_FORMS = ("10-K", "10-Q", "10-K/A", "10-Q/A") +# How recent the issuer's own newest filing must be for an *escalated* gap to +# stop pausing setups. A gap pauses an issuer until it is either resolved or +# superseded by a later ingested filing — which assumes the gap is temporary. +# It is not always: SEC's per-company Company-Facts files can go stale +# indefinitely (2026-08, 43 large caps whose Q2 10-Qs the frames API carried but +# whose companyfacts files never received), and since the supersede rule needs a +# *successfully ingested* later filing, a stale file also swallows the next +# quarter. The pause is then open-ended rather than seasonal. +# +# So the pause hands off to the alert: once `filing_gap_aged` has escalated a gap +# to an operator (`escalated_at`), the issuer resumes on the fundamentals it does +# have — provided those are recent. An issuer with nothing this fresh has no +# usable fundamentals at all and stays paused, which is the case the gate was +# built for. The retry queue is untouched: `active_gaps` still returns these, so +# the importer keeps retrying and a recovered filing still resolves normally. +GAP_GATE_RECENT_FILING_DAYS = 180 + @dataclass(frozen=True) class SetupQuality: @@ -51,6 +70,39 @@ async def active_gaps( return list((await db.execute(stmt)).scalars().all()) +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. + """ + by_cik: dict[str, list[SecFilingGap]] = defaultdict(list) + for gap in gaps: + by_cik[gap.cik].append(gap) + escalated = { + cik + for cik, items in by_cik.items() + if all(gap.escalated_at is not None for gap in items) + } + if not escalated: + return set() + cutoff = ( + datetime.now(timezone.utc) - timedelta(days=GAP_GATE_RECENT_FILING_DAYS) + ).date() + rows = await db.execute( + select(FundamentalSnapshot.cik) + .where( + FundamentalSnapshot.cik.in_(escalated), + FundamentalSnapshot.form.in_(_SEC_FORMS), + FundamentalSnapshot.filed_date >= cutoff, + ) + .distinct() + ) + return set(rows.scalars()) + + async def _latest_validation(db: AsyncSession) -> dict: payload = ( await db.execute( @@ -80,8 +132,12 @@ async def blocked_reasons_by_cik( if ciks is not None and not ciks: return {} + 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) reasons = { - gap.cik: "sec_filing_gap" for gap in await active_gaps(db, ciks) + gap.cik: "sec_filing_gap" for gap in gaps if gap.cik not in exempt } summary = await _latest_validation(db) @@ -92,11 +148,11 @@ async def blocked_reasons_by_cik( # stay capped for audit readability. Detailed entries supply the reason. for cik in summary.get("setup_blocked_ciks") or []: normalized = str(cik) if cik else "" - if normalized and wanted(normalized): + if normalized and wanted(normalized) and normalized not in exempt: reasons.setdefault(normalized, "sec_filing_gap") for item in summary.get("missing_xbrl") or []: normalized = str(item.get("cik") or "") - if normalized and wanted(normalized): + if normalized and wanted(normalized) and normalized not in exempt: reasons.setdefault(normalized, "sec_filing_gap") for cik in summary.get("no_xbrl_ciks") or []: normalized = str(cik) if cik else "" diff --git a/docs/fundamentals-deployment.md b/docs/fundamentals-deployment.md index e64172c..c65fbbe 100644 --- a/docs/fundamentals-deployment.md +++ b/docs/fundamentals-deployment.md @@ -24,6 +24,17 @@ entries: the application scheduler owns both jobs. 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. +- A gap that survives 14 days raises `filing_gap_aged` and, from that point, + stops pausing setups **if** the issuer's own newest stored 10-K/10-Q is less + than `GAP_GATE_RECENT_FILING_DAYS` (180) old. This is the hand-off from pause + to alert, and it exists because the pause would otherwise be open-ended: + SEC's per-company Company-Facts files can go stale indefinitely (2026-08: 43 + large caps whose Q2 10-Qs the `frames` API carried but whose + `companyfacts/CIK*.json` never received), and the supersede rule needs a + *successfully ingested* later filing, so a stale file swallows the next + quarter too. Retrying is unaffected — the gap stays queued and a recovered + filing still resolves it normally. An issuer with no filing that recent has no + usable fundamentals at all and stays paused. 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 diff --git a/tests/unit/test_fundamentals_quality_service.py b/tests/unit/test_fundamentals_quality_service.py index c9533c8..51bbd5f 100644 --- a/tests/unit/test_fundamentals_quality_service.py +++ b/tests/unit/test_fundamentals_quality_service.py @@ -1,7 +1,7 @@ from __future__ import annotations import json -from datetime import date, datetime, timezone +from datetime import date, datetime, timedelta, timezone from app.models.data_import_run import DataImportRun from app.models.fundamental_snapshot import FundamentalSnapshot @@ -139,3 +139,121 @@ async def test_ticker_quality_explains_no_xbrl_block(db_session): assert await fundamentals_quality_service.ticker_is_eligible( db_session, ticker.id ) is False + + +def _escalated_gap(cik: str, *, escalated: bool = True) -> SecFilingGap: + first_seen = datetime.now(timezone.utc) - timedelta(days=24) + return SecFilingGap( + cik=cik, + accession=f"{cik}-STALE-Q", + form="10-Q", + index_date=(first_seen.date()), + reason="not_in_companyfacts", + first_seen_at=first_seen, + last_attempted_at=datetime.now(timezone.utc), + escalated_at=( + datetime.now(timezone.utc) - timedelta(days=10) if escalated else None + ), + ) + + +def _prior_quarter(cik: str, *, age_days: int) -> FundamentalSnapshot: + """The issuer's last successfully ingested filing, older than the gap so it + cannot supersede it — exactly the production shape of a stale companyfacts + file: Q1 stored, Q2 missing.""" + filed = date.today() - timedelta(days=age_days) + return FundamentalSnapshot( + cik=cik, + accession=f"{cik}-PRIOR-Q", + 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 test_escalated_gap_stops_blocking_when_fundamentals_are_recent(db_session): + ticker = Ticker(symbol="STALEFACTS", cik="0000000046") + db_session.add(ticker) + db_session.add(_escalated_gap(ticker.cik)) + await db_session.flush() + assert await fundamentals_quality_service.blocked_ticker_ids(db_session) == { + ticker.id + } + + # The alert has run and the issuer still has last quarter to score on. + db_session.add(_prior_quarter(ticker.cik, age_days=120)) + await db_session.flush() + + assert await fundamentals_quality_service.blocked_ticker_ids(db_session) == set() + # ...but the filing is still queued, so the importer keeps retrying it. + assert len(await fundamentals_quality_service.active_gaps(db_session)) == 1 + + +async def test_escalated_gap_keeps_blocking_when_fundamentals_are_stale(db_session): + ticker = Ticker(symbol="NOTHINGFRESH", cik="0000000047") + db_session.add_all([ + ticker, + _escalated_gap(ticker.cik), + _prior_quarter( + ticker.cik, + age_days=fundamentals_quality_service.GAP_GATE_RECENT_FILING_DAYS + 30, + ), + ]) + await db_session.flush() + + assert await fundamentals_quality_service.blocked_ticker_ids(db_session) == { + ticker.id + } + + +async def test_unescalated_gap_still_blocks_alongside_an_escalated_one(db_session): + ticker = Ticker(symbol="TWOGAPS", cik="0000000048") + fresh = datetime.now(timezone.utc) + db_session.add_all([ + ticker, + _escalated_gap(ticker.cik), + SecFilingGap( + cik=ticker.cik, + accession="TWOGAPS-FRESH-Q", + form="10-Q", + index_date=date.today(), + reason="not_in_companyfacts", + first_seen_at=fresh, + last_attempted_at=fresh, + ), + _prior_quarter(ticker.cik, age_days=120), + ]) + await db_session.flush() + + assert await fundamentals_quality_service.blocked_ticker_ids(db_session) == { + ticker.id + } + + +async def test_summary_path_does_not_reblock_an_exempt_cik(db_session): + """The run summary mirrors the same filings as the queue — it must honour the + same hand-off, or the bound is inert in production.""" + ticker = Ticker(symbol="MIRRORED", cik="0000000049") + db_session.add(ticker) + db_session.add(_escalated_gap(ticker.cik)) + db_session.add(_prior_quarter(ticker.cik, age_days=120)) + await db_session.flush() + db_session.add( + DataImportRun( + source="sec_facts", + status="promoted", + validation_json=json.dumps({ + "setup_blocked_ciks": [ticker.cik], + "missing_xbrl": [ + {"cik": ticker.cik, "accession": f"{ticker.cik}-STALE-Q"} + ], + }), + started_at=datetime.now(timezone.utc), + ) + ) + await db_session.flush() + + assert await fundamentals_quality_service.blocked_ticker_ids(db_session) == set()