diff --git a/app/scheduler.py b/app/scheduler.py index 7bc4092..68a13f7 100644 --- a/app/scheduler.py +++ b/app/scheduler.py @@ -1491,7 +1491,7 @@ async def sync_ticker_universe() -> None: _FINAL_REFETCH_DAYS = 5 _DAILY_PIPELINE_STEPS = [ - ("data_collector", "collect_ohlcv_for_scan"), + ("data_collector", "collect_ohlcv"), ("benchmark_collector", "collect_benchmark"), ("sentiment_collector", "collect_sentiment"), ("market_regime", "compute_market_regime"), @@ -1515,7 +1515,7 @@ _DAILY_PIPELINE_STEPS = [ _NEAR_CLOSE_PIPELINE_STEPS = [ # Must land today's in-progress bar (~20 min behind live), or the scan falls # back to the previous close and execution degrades to the stale_close floor. - ("data_collector", "collect_ohlcv"), + ("data_collector", "collect_ohlcv_for_scan"), ("rr_scanner", "scan_rr"), # Straight after the scan so shadow entries mark at the same near-close # prices the discretionary book is looking at. diff --git a/app/services/fundamentals_quality_service.py b/app/services/fundamentals_quality_service.py index 78ffcec..aa76600 100644 --- a/app/services/fundamentals_quality_service.py +++ b/app/services/fundamentals_quality_service.py @@ -5,7 +5,7 @@ from __future__ import annotations import json from dataclasses import dataclass -from sqlalchemy import exists, or_, select +from sqlalchemy import exists, func, select from sqlalchemy.ext.asyncio import AsyncSession from app.models.data_import_run import DataImportRun @@ -32,14 +32,18 @@ async def active_gaps( matching_snapshot = exists().where( FundamentalSnapshot.accession == SecFilingGap.accession ) + gap_date = func.coalesce( + SecFilingGap.index_date, + func.date(SecFilingGap.first_seen_at), + ) later_snapshot = exists().where( FundamentalSnapshot.cik == SecFilingGap.cik, FundamentalSnapshot.form.in_(_SEC_FORMS), - FundamentalSnapshot.filed_date > SecFilingGap.index_date, + FundamentalSnapshot.filed_date > gap_date, ) stmt = select(SecFilingGap).where( ~matching_snapshot, - or_(SecFilingGap.index_date.is_(None), ~later_snapshot), + ~later_snapshot, ) if ciks is not None: if not ciks: diff --git a/app/services/sec_fundamentals_importer.py b/app/services/sec_fundamentals_importer.py index 696b81a..6152b4c 100644 --- a/app/services/sec_fundamentals_importer.py +++ b/app/services/sec_fundamentals_importer.py @@ -342,11 +342,12 @@ class SecFundamentalsImporter: 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. + # A new index row keeps the normal grace period before promotion; + # a row already read from the queue retains its _retry_queue marker + # so later imports promote and retry without wedging the index. staged.missing_xbrl.append(_missing( cik, - {**index_row, "_retry_queue": True}, + index_row, "parser_unusable", self.today, self._coregistrants.get(skipped["accession"]), @@ -488,7 +489,10 @@ class SecFundamentalsImporter: retryable=( len(messages) == 1 and bool(blocking) - and all(m.get("reason") == "not_in_companyfacts" for m in blocking) + and all( + m.get("reason") in {"not_in_companyfacts", "parser_unusable"} + for m in blocking + ) ), deferred_alert_after_days=MISSING_XBRL_RETRY_DAYS, deferred_alert_messages=( diff --git a/tests/unit/test_fundamentals_quality_service.py b/tests/unit/test_fundamentals_quality_service.py index 1547f27..96ea3ab 100644 --- a/tests/unit/test_fundamentals_quality_service.py +++ b/tests/unit/test_fundamentals_quality_service.py @@ -109,6 +109,50 @@ async def test_active_gap_is_blocked_until_a_later_filing_supersedes_it(db_sessi assert await fundamentals_quality_service.blocked_ticker_ids(db_session) == set() +async def test_gap_without_index_date_uses_first_seen_date_for_supersession( + db_session, +): + ticker = Ticker(symbol="DATELESS", cik="0000000045") + first_seen = datetime(2026, 5, 1, 12, tzinfo=timezone.utc) + db_session.add_all([ + ticker, + SystemSetting( + key="fundamental_data_sec_dolt_cutover_enabled", + value="true", + ), + SecFilingGap( + cik=ticker.cik, + accession="DATELESS-Q", + form="10-Q", + index_date=None, + reason="not_in_companyfacts", + first_seen_at=first_seen, + last_attempted_at=first_seen, + ), + ]) + await db_session.flush() + + assert await fundamentals_quality_service.blocked_ticker_ids(db_session) == { + ticker.id + } + + db_session.add( + FundamentalSnapshot( + cik=ticker.cik, + accession="LATER-DATELESS-Q", + form="10-Q", + filed_date=date(2026, 5, 2), + accepted_at=datetime(2026, 5, 2, 12, tzinfo=timezone.utc), + period_end=date(2026, 3, 31), + fiscal_year=2026, + fiscal_period="Q1", + ) + ) + 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([ diff --git a/tests/unit/test_scheduler.py b/tests/unit/test_scheduler.py index 7816195..e63359a 100644 --- a/tests/unit/test_scheduler.py +++ b/tests/unit/test_scheduler.py @@ -5,6 +5,8 @@ from types import SimpleNamespace import pytest from app.scheduler import ( + _DAILY_PIPELINE_STEPS, + _NEAR_CLOSE_PIPELINE_STEPS, _consume_backtest_options, _consume_backtest_target_model, _parse_frequency, @@ -43,6 +45,14 @@ def test_manual_backtest_options_are_one_shot_and_default_back_to_weekly(): assert _consume_backtest_options() == ("production_gtl", "weekly") +def test_only_near_close_fetch_skips_redundant_sr_refresh(): + assert dict(_DAILY_PIPELINE_STEPS)["data_collector"] == "collect_ohlcv" + assert ( + dict(_NEAR_CLOSE_PIPELINE_STEPS)["data_collector"] + == "collect_ohlcv_for_scan" + ) + + class TestParseFrequency: def test_hourly(self): assert _parse_frequency("hourly") == {"hours": 1} diff --git a/tests/unit/test_sec_fundamentals_importer.py b/tests/unit/test_sec_fundamentals_importer.py index 220420d..d4f3bf2 100644 --- a/tests/unit/test_sec_fundamentals_importer.py +++ b/tests/unit/test_sec_fundamentals_importer.py @@ -650,6 +650,76 @@ async def test_queued_parser_skip_stays_blocked_with_actionable_reason( assert gap.reason == "parser_unusable" +async def test_new_parser_skip_gets_grace_then_enters_retry_queue( + 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) + + bad_fact = _rev( + "2025-09-28", "2026-03-28", 254940, 2026, "Q2", "NEWBAD" + ) + bad_share = _shares("2026-04-17", 14687, "NEWBAD", 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( + "NEWBAD", + "10-Q", + "2026-03-28", + "2026-05-01", + "2026-05-01T10:01:00.000Z", + ) + ])}, + latest_index=date(2026, 5, 2), + daily={ + date(2026, 5, 1): [{ + "form": "10-Q", + "cik": 320193, + "accession": "NEWBAD", + }] + }, + ) + + def skip_parse(*args, **kwargs): + return ParseResult(skipped_filings=[{ + "accession": "NEWBAD", + "reason": "unparseable", + }]) + + monkeypatch.setattr("app.services.sec_facts_parser.parse_snapshots", skip_parse) + + young = await run_import( + _importer(client, today=date(2026, 5, 3)), engine=engine + ) + assert young.status == STATUS_DEFERRED + assert "parser_unusable" in (young.error_details or "") + assert await _count(factory, SecFilingGap) == 0 + + aged = await run_import( + _importer(client, today=date(2026, 5, 5)), engine=engine + ) + assert aged.status == STATUS_PROMOTED + async with factory() as db: + gap = (await db.execute(select(SecFilingGap))).scalar_one() + assert gap.accession == "NEWBAD" + 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)