fix: close remaining ingestion review gaps

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.

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:

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,
+                # A new index row keeps the normal grace period before promotion;
-                # keep the issuer blocked, and retry/escalate through the queue.
+                # 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=(

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([

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}

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)