test: drop redundant scanner primary-target suites

test_rr_scanner_bug_exploration.py and test_rr_scanner_fix_check.py both
assert one invariant: the headline target is the probability-based near
level, not the far max-R:R lottery. That is already covered directly by
test_recommendation_service.py's _select_primary_target tests, which also
reach cases these never did (empty list, probability floor, activation vs
scanner floor), and end to end by test_rr_scanner_integration.py's
full-flow test -- a strict superset of their deterministic cases: three
resistance and three support levels, both directions, plus persistence
and rr_ratio consistency.

The two files also duplicated each other, and their docstrings had gone
stale: test_deterministic_long_three_levels documented a hand-computed
_compute_quality_score winner even though the assertion is about the
probability primary that supersedes it.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

alembic/versions/028_sec_filing_retry_queue.py

@@ -0,0 +1,147 @@
+"""SEC filing retry queue and setup-quality gate
+
+Revision ID: 028
+Revises: 027
+Create Date: 2026-08-03 00:00:00.000000
+"""
+from datetime import date, datetime, timezone
+import json
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+
+revision: str = "028"
+down_revision: Union[str, None] = "027"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        "sec_filing_gaps",
+        sa.Column("id", sa.Integer(), primary_key=True),
+        sa.Column("cik", sa.String(length=10), nullable=False),
+        sa.Column("accession", sa.String(length=25), nullable=False),
+        sa.Column("form", sa.String(length=12), nullable=True),
+        sa.Column("index_date", sa.Date(), nullable=True),
+        sa.Column("reason", sa.String(length=64), nullable=False),
+        sa.Column("coregistrant_ciks_json", sa.Text(), nullable=True),
+        sa.Column("first_seen_at", sa.DateTime(timezone=True), nullable=False),
+        sa.Column("last_attempted_at", sa.DateTime(timezone=True), nullable=False),
+        sa.Column("escalated_at", sa.DateTime(timezone=True), nullable=True),
+        sa.UniqueConstraint("accession", name="uq_sec_filing_gaps_accession"),
+    )
+    op.create_index("ix_sec_filing_gaps_cik", "sec_filing_gaps", ["cik"])
+    _backfill_retry_queue()
+
+
+def downgrade() -> None:
+    op.drop_index("ix_sec_filing_gaps_cik", table_name="sec_filing_gaps")
+    op.drop_table("sec_filing_gaps")
+
+
+def _as_date(value) -> date | None:
+    if isinstance(value, datetime):
+        return value.date()
+    if isinstance(value, date):
+        return value
+    if isinstance(value, str):
+        try:
+            return date.fromisoformat(value)
+        except ValueError:
+            return None
+    return None
+
+
+def _backfill_retry_queue() -> None:
+    """Materialize pre-queue promoted gaps once; runtime never scans history."""
+    bind = op.get_bind()
+    runs = sa.table(
+        "data_import_runs",
+        sa.column("source", sa.String()),
+        sa.column("status", sa.String()),
+        sa.column("validation_json", sa.Text()),
+        sa.column("source_max_date", sa.Date()),
+        sa.column("started_at", sa.DateTime(timezone=True)),
+    )
+    snapshots = sa.table(
+        "fundamental_snapshots",
+        sa.column("cik", sa.String()),
+        sa.column("accession", sa.String()),
+        sa.column("filed_date", sa.Date()),
+    )
+    gaps = sa.table(
+        "sec_filing_gaps",
+        sa.column("cik", sa.String()),
+        sa.column("accession", sa.String()),
+        sa.column("form", sa.String()),
+        sa.column("index_date", sa.Date()),
+        sa.column("reason", sa.String()),
+        sa.column("coregistrant_ciks_json", sa.Text()),
+        sa.column("first_seen_at", sa.DateTime(timezone=True)),
+        sa.column("last_attempted_at", sa.DateTime(timezone=True)),
+        sa.column("escalated_at", sa.DateTime(timezone=True)),
+    )
+
+    snapshot_rows = bind.execute(
+        sa.select(snapshots.c.cik, snapshots.c.accession, snapshots.c.filed_date)
+    ).all()
+    resolved_accessions = {row.accession for row in snapshot_rows}
+    latest_filed_by_cik: dict[str, date] = {}
+    for row in snapshot_rows:
+        if row.filed_date is not None:
+            current = latest_filed_by_cik.get(row.cik)
+            if current is None or row.filed_date > current:
+                latest_filed_by_cik[row.cik] = row.filed_date
+
+    audit_rows = bind.execute(
+        sa.select(
+            runs.c.validation_json,
+            runs.c.source_max_date,
+            runs.c.started_at,
+        ).where(
+            runs.c.source == "sec_facts",
+            runs.c.status == "promoted",
+            runs.c.validation_json.is_not(None),
+        )
+    ).all()
+    now = datetime.now(timezone.utc)
+    candidates: dict[str, dict] = {}
+    for audit in audit_rows:
+        try:
+            summary = json.loads(audit.validation_json)
+        except (TypeError, ValueError):
+            continue
+        if not isinstance(summary, dict):
+            continue
+        for item in summary.get("missing_xbrl") or []:
+            accession = item.get("accession")
+            raw_cik = item.get("cik")
+            if not accession or raw_cik is None or accession in resolved_accessions:
+                continue
+            cik = str(raw_cik).zfill(10)
+            index_date = _as_date(item.get("index_date")) or _as_date(
+                audit.source_max_date
+            )
+            later_filed = latest_filed_by_cik.get(cik)
+            if index_date is not None and later_filed is not None and later_filed > index_date:
+                continue
+            first_seen = audit.started_at or now
+            existing = candidates.get(accession)
+            if existing is not None and existing["first_seen_at"] <= first_seen:
+                continue
+            candidates[accession] = {
+                "cik": cik,
+                "accession": accession,
+                "form": item.get("form"),
+                "index_date": index_date,
+                "reason": item.get("reason") or "not_in_companyfacts",
+                "coregistrant_ciks_json": json.dumps(item.get("coregistrants") or []),
+                "first_seen_at": first_seen,
+                "last_attempted_at": first_seen,
+                "escalated_at": None,
+            }
+    if candidates:
+        op.bulk_insert(gaps, list(candidates.values()))

app/models/__init__.py

@@ -17,6 +17,7 @@ from app.models.regime_snapshot import RegimeSnapshot
 from app.models.benchmark_price import BenchmarkPrice
 from app.models.signal_context_snapshot import SignalContextSnapshot
 from app.models.system_event import SystemEvent
+from app.models.sec_filing_gap import SecFilingGap
 
 __all__ = [
     "Ticker",
@@ -40,4 +41,5 @@ __all__ = [
     "BenchmarkPrice",
     "SignalContextSnapshot",
     "SystemEvent",
+    "SecFilingGap",
 ]

app/models/sec_filing_gap.py

@@ -0,0 +1,32 @@
+from datetime import date, datetime
+
+from sqlalchemy import Date, DateTime, Index, String, Text, UniqueConstraint
+from sqlalchemy.orm import Mapped, mapped_column
+
+from app.database import Base
+
+
+class SecFilingGap(Base):
+    """Active SEC filing that could not yet be reconstructed.
+
+    Rows form a small retry queue. Successful snapshot ingestion deletes the
+    matching row; a later valid filing supersedes it. While a current row remains,
+    tickers mapped to its CIK are not eligible for actionable trade setups.
+    """
+
+    __tablename__ = "sec_filing_gaps"
+    __table_args__ = (
+        UniqueConstraint("accession", name="uq_sec_filing_gaps_accession"),
+        Index("ix_sec_filing_gaps_cik", "cik"),
+    )
+
+    id: Mapped[int] = mapped_column(primary_key=True)
+    cik: Mapped[str] = mapped_column(String(10), nullable=False)
+    accession: Mapped[str] = mapped_column(String(25), nullable=False)
+    form: Mapped[str | None] = mapped_column(String(12), nullable=True)
+    index_date: Mapped[date | None] = mapped_column(Date, nullable=True)
+    reason: Mapped[str] = mapped_column(String(64), nullable=False)
+    coregistrant_ciks_json: Mapped[str | None] = mapped_column(Text, nullable=True)
+    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)

app/routers/fundamentals.py

@@ -10,6 +10,7 @@ from app.schemas.common import APIEnvelope
 from app.schemas.fundamental import FundamentalResponse
 from app.services.fundamental_service import get_fundamental
 from app.services.fundamentals_api_service import build_fundamentals_v1
+from app.services import fundamentals_quality_service
 
 router = APIRouter(tags=["fundamentals"])
 
@@ -34,6 +35,7 @@ async def read_fundamentals(
     """Get latest fundamental data for a symbol (legacy fields + additive v1)."""
     record = await get_fundamental(db, symbol)
     v1 = await build_fundamentals_v1(db, symbol)
+    quality = await fundamentals_quality_service.ticker_quality(db, symbol)
 
     legacy: dict = {}
     if record is not None:
@@ -47,5 +49,12 @@ async def read_fundamentals(
             unavailable_fields=_parse_unavailable_fields(record.unavailable_fields_json),
         )
 
-    data = FundamentalResponse(symbol=symbol.strip().upper(), **legacy, **v1)
+    data = FundamentalResponse(
+        symbol=symbol.strip().upper(),
+        setup_eligible=quality.eligible,
+        setup_block_code=quality.code,
+        setup_block_reason=quality.message,
+        **legacy,
+        **v1,
+    )
     return APIEnvelope(status="success", data=data.model_dump())

app/scheduler.py

@@ -512,13 +512,14 @@ async def collect_ohlcv(
     job_name: str = "data_collector",
     *,
     refetch_days: int = 0,
+    refresh_sr: bool = True,
 ) -> None:
     """Fetch latest daily OHLCV for all tracked tickers.
 
     Uses AlpacaOHLCVProvider. Processes each ticker independently.
     On rate limit, records last successful ticker for resume.
         Start date is resolved by ingestion progress:
-            - existing ticker: resume from last_ingested_date + 1
+            - existing ticker: overlap last_ingested_date so partial bars refresh
             - new ticker: backfill the configured history window
 
     ``full_backfill`` forces every ticker to re-fetch the full
@@ -580,6 +581,7 @@ async def collect_ohlcv(
                 try:
                     result = await ingestion_service.fetch_and_ingest(
                         db, provider, symbol, start_date=backfill_start, end_date=end_date,
+                        refresh_sr=refresh_sr,
                     )
                     _last_successful[job_name] = symbol
                     processed += 1
@@ -619,6 +621,11 @@ async def collect_ohlcv(
         _runtime_finish(job_name, "error", processed=processed, total=total, message=str(exc))
 
 
+async def collect_ohlcv_for_scan() -> None:
+    """Near-close fetch; the scanner immediately rebuilds S/R per ticker."""
+    await collect_ohlcv(refresh_sr=False)
+
+
 async def backfill_ohlcv() -> None:
     """Deep historical backfill: re-fetch the full ``settings.ohlcv_history_days``
     window for every ticker, ignoring incremental resume.
@@ -1496,8 +1503,8 @@ _DAILY_PIPELINE_STEPS = [
     ("alerts", "dispatch_alerts_job"),
 ]
 
-# Near-close (~15:30 ET Mon–Fri): refresh in-progress day-t bars (already how
-# the intraday pipeline keeps the dashboard live), then the only daily
+# Near-close (~15:30 ET Mon–Fri): refresh in-progress day-t bars (incremental
+# ingestion overlaps the latest stored session), then the only daily
 # qualifying R:R scan, then Telegram immediately so manual fills can still hit
 # MOC cutoffs (~15:50/15:55). Under a 15-minute delayed SIP feed a 15:30 scan
 # may see ~15:15 prices — immaterial for a 12-1 momentum signal.
@@ -1508,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/schemas/fundamental.py

@@ -91,3 +91,6 @@ class FundamentalResponse(BaseModel):
     metrics: list[MetricItem] | None = None
     valuation: Valuation | None = None
     reads: FundamentalsReads | None = None
+    setup_eligible: bool = True
+    setup_block_code: str | None = None
+    setup_block_reason: str | None = None

app/schemas/paper_trade.py

@@ -53,3 +53,7 @@ class PaperTradeResponse(BaseModel):
     # when the trailing exit policy is active.
     trailing_stop: float | None = None
     trailing_distance_pct: float | None = None
+    # Trading sessions represented by post-entry OHLCV bars. These are populated
+    # only while the active exit policy has a max-hold rule.
+    sessions_held: int | None = None
+    sessions_remaining: int | None = None

app/services/fundamentals_quality_service.py

@@ -0,0 +1,164 @@
+"""Actionability gate for incomplete SEC fundamentals."""
+
+from __future__ import annotations
+
+import json
+from dataclasses import dataclass
+
+from sqlalchemy import exists, func, select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.models.data_import_run import DataImportRun
+from app.models.fundamental_snapshot import FundamentalSnapshot
+from app.models.sec_filing_gap import SecFilingGap
+from app.models.ticker import Ticker
+from app.services import fundamental_data_refresh_service
+
+_SEC_FORMS = ("10-K", "10-Q", "10-K/A", "10-Q/A")
+
+
+@dataclass(frozen=True)
+class SetupQuality:
+    eligible: bool
+    code: str | None = None
+    message: str | None = None
+
+
+async def active_gaps(
+    db: AsyncSession,
+    ciks: set[str] | None = None,
+) -> list[SecFilingGap]:
+    """Unresolved gaps that have not been superseded by a later filing."""
+    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 > gap_date,
+    )
+    stmt = select(SecFilingGap).where(
+        ~matching_snapshot,
+        ~later_snapshot,
+    )
+    if ciks is not None:
+        if not ciks:
+            return []
+        stmt = stmt.where(SecFilingGap.cik.in_(ciks))
+    return list((await db.execute(stmt)).scalars().all())
+
+
+async def _latest_validation(db: AsyncSession) -> dict:
+    payload = (
+        await db.execute(
+            select(DataImportRun.validation_json)
+            .where(
+                DataImportRun.source == "sec_facts",
+                DataImportRun.validation_json.is_not(None),
+            )
+            .order_by(DataImportRun.id.desc())
+            .limit(1)
+        )
+    ).scalar_one_or_none()
+    if not payload:
+        return {}
+    try:
+        summary = json.loads(payload)
+    except (TypeError, ValueError):
+        return {}
+    return summary if isinstance(summary, dict) else {}
+
+
+async def blocked_reasons_by_cik(
+    db: AsyncSession,
+    ciks: set[str] | None = None,
+) -> dict[str, str]:
+    """Current SEC blocker code by CIK; no historical audit scan."""
+    if not await fundamental_data_refresh_service.is_enabled(db):
+        return {}
+    if ciks is not None and not ciks:
+        return {}
+
+    reasons = {
+        gap.cik: "sec_filing_gap" for gap in await active_gaps(db, ciks)
+    }
+    summary = await _latest_validation(db)
+
+    def wanted(cik: str) -> bool:
+        return ciks is None or cik in ciks
+
+    # New summaries carry the complete compact CIK set while the detailed lists
+    # 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):
+            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):
+            reasons.setdefault(normalized, "sec_filing_gap")
+    for cik in summary.get("no_xbrl_ciks") or []:
+        normalized = str(cik) if cik else ""
+        if normalized and wanted(normalized):
+            reasons[normalized] = "no_xbrl_filings"
+    for item in summary.get("no_xbrl_filings") or []:
+        normalized = str(item.get("cik") or "")
+        if normalized and wanted(normalized):
+            reasons[normalized] = "no_xbrl_filings"
+    return reasons
+
+
+async def blocked_ciks(db: AsyncSession) -> set[str]:
+    return set(await blocked_reasons_by_cik(db))
+
+
+async def blocked_ticker_ids(db: AsyncSession) -> set[int]:
+    ciks = await blocked_ciks(db)
+    if not ciks:
+        return set()
+    rows = await db.execute(select(Ticker.id).where(Ticker.cik.in_(ciks)))
+    return {int(ticker_id) for ticker_id in rows.scalars()}
+
+
+async def ticker_quality(db: AsyncSession, symbol: str) -> SetupQuality:
+    ticker = (
+        await db.execute(
+            select(Ticker).where(Ticker.symbol == symbol.strip().upper())
+        )
+    ).scalar_one_or_none()
+    if ticker is None or not ticker.cik:
+        return SetupQuality(eligible=True)
+    reason = (await blocked_reasons_by_cik(db, {ticker.cik})).get(ticker.cik)
+    if reason == "no_xbrl_filings":
+        return SetupQuality(
+            eligible=False,
+            code=reason,
+            message=(
+                "No SEC 10-K/10-Q is available for this registrant, so new setups "
+                "are paused. New registrants clear automatically after their first "
+                "filing; a successor shell needs an SEC CIK override."
+            ),
+        )
+    if reason:
+        return SetupQuality(
+            eligible=False,
+            code=reason,
+            message=(
+                "A recent SEC filing is still being reconciled, so new setups are "
+                "paused. The scheduled fundamentals import retries it automatically."
+            ),
+        )
+    return SetupQuality(eligible=True)
+
+
+async def ticker_is_eligible(db: AsyncSession, ticker_id: int) -> bool:
+    cik = (
+        await db.execute(select(Ticker.cik).where(Ticker.id == ticker_id))
+    ).scalar_one_or_none()
+    if not cik:
+        return True
+    return cik not in await blocked_reasons_by_cik(db, {cik})

app/services/ingestion_service.py

@@ -100,6 +100,8 @@ async def fetch_and_ingest(
     symbol: str,
     start_date: date | None = None,
     end_date: date | None = None,
+    *,
+    refresh_sr: bool = True,
 ) -> IngestionResult:
     """Fetch OHLCV data from provider and upsert into Price Store.
 
@@ -129,7 +131,12 @@ async def fetch_and_ingest(
         if bar_count < minimum_backfill_bars:
             start_date = backfill_start
         elif progress is not None:
-            start_date = progress.last_ingested_date + timedelta(days=1)
+            # Re-fetch the latest stored session so an in-progress daily bar can
+            # be overwritten as the market moves. Starting one day later makes
+            # every subsequent intraday, near-close, and manual refresh skip
+            # today's bar once the first partial snapshot has been stored.
+            # The price-store upsert keeps this one-session overlap idempotent.
+            start_date = progress.last_ingested_date
         else:
             start_date = backfill_start
 
@@ -239,7 +246,7 @@ async def fetch_and_ingest(
                 ticker.symbol,
                 ingested_count,
             )
-            if ingested_count > 0:
+            if ingested_count > 0 and refresh_sr:
                 await _refresh_structural_sr(db, ticker.symbol)
             return IngestionResult(
                 symbol=ticker.symbol,
@@ -249,9 +256,28 @@ async def fetch_and_ingest(
                 message=f"Rate limited. Ingested {ingested_count} records. Resume available.",
             )
 
-    if ingested_count > 0:
+    if ingested_count > 0 and refresh_sr:
         await _refresh_structural_sr(db, ticker.symbol)
 
+    # Incremental fetches deliberately overlap the latest stored session so an
+    # in-progress bar can be updated. A halted/delisted symbol can therefore
+    # return one old bar forever; non-empty no longer means fresh. Judge stale
+    # state from the newest stored session after the upserts instead.
+    latest = await _get_latest_ohlcv_date(db, ticker.id)
+    gap_days = (end_date - latest).days if latest is not None else None
+    if gap_days is not None and gap_days > _STALE_OHLCV_GAP_DAYS:
+        return IngestionResult(
+            symbol=ticker.symbol,
+            records_ingested=ingested_count,
+            last_date=latest,
+            status="stale",
+            message=(
+                f"No new bars since {latest.isoformat()} ({gap_days}d gap). "
+                "The symbol may be halted, delisted, or renamed under a new ticker — "
+                "check the listing and add/fetch the current symbol if it changed."
+            ),
+        )
+
     return IngestionResult(
         symbol=ticker.symbol,
         records_ingested=ingested_count,

app/services/paper_trade_service.py

@@ -352,6 +352,7 @@ def _to_dict(
     current_price: float | None,
     benchmark_closes: dict[date, float] | None = None,
     trailing: tuple[float, float | None] | None = None,
+    holding_sessions: tuple[int, int] | None = None,
 ) -> dict:
     # For open trades, mark to market; for closed, the realized exit price.
     ref = current_price if trade.status == "open" else trade.close_price
@@ -395,6 +396,8 @@ def _to_dict(
         "fill_mode": trade.fill_mode,
         "trailing_stop": trailing[0] if trailing else None,
         "trailing_distance_pct": trailing[1] if trailing else None,
+        "sessions_held": holding_sessions[0] if holding_sessions else None,
+        "sessions_remaining": holding_sessions[1] if holding_sessions else None,
     }
 
 
@@ -435,6 +438,35 @@ async def list_trades(
     # Current trailing-stop level + distance for open trades (when a trailing
     # policy is active).
     policy = await get_exit_policy(db)
+    holding_sessions: dict[int, tuple[int, int]] = {}
+    if policy["mode"] in ("time", "atr_trailing"):
+        hold_days = int(policy["hold_days"])
+        open_trades = [trade for trade, _ in rows if trade.status == "open"]
+        if open_trades:
+            ticker_ids = {trade.ticker_id for trade in open_trades}
+            earliest_opened = min(trade.opened_at.date() for trade in open_trades)
+            session_rows = (
+                await db.execute(
+                    select(OHLCVRecord.ticker_id, OHLCVRecord.date)
+                    .where(
+                        OHLCVRecord.ticker_id.in_(ticker_ids),
+                        OHLCVRecord.date > earliest_opened,
+                    )
+                    .order_by(OHLCVRecord.ticker_id, OHLCVRecord.date)
+                )
+            ).all()
+            dates_by_ticker: dict[int, list[date]] = {}
+            for ticker_id, session_date in session_rows:
+                dates_by_ticker.setdefault(int(ticker_id), []).append(session_date)
+            for trade in open_trades:
+                dates = dates_by_ticker.get(trade.ticker_id, [])
+                held = len(dates) - bisect.bisect_right(
+                    dates, trade.opened_at.date()
+                )
+                # Do not clamp: a policy shortened below the current holding
+                # period must remain visible as overdue until the exit pass runs.
+                holding_sessions[trade.id] = (held, hold_days - held)
+
     trailing_info: dict[int, tuple[float, float | None]] = {}
     if policy["mode"] == "trailing":
         trail_frac = policy["trailing_pct"] / 100.0
@@ -483,7 +515,14 @@ async def list_trades(
             trailing_info[t.id] = (level, dist)
 
     return [
-        _to_dict(t, sym, prices.get(t.ticker_id), benchmark_closes, trailing_info.get(t.id))
+        _to_dict(
+            t,
+            sym,
+            prices.get(t.ticker_id),
+            benchmark_closes,
+            trailing_info.get(t.id),
+            holding_sessions.get(t.id),
+        )
         for t, sym in rows
     ]
 

app/services/rr_scanner_service.py

@@ -27,6 +27,7 @@ from app.models.signal_context_snapshot import SignalContextSnapshot
 from app.models.ticker import Ticker
 from app.models.trade_setup import TradeSetup
 from app.services.indicator_service import _extract_ohlcv, compute_atr
+from app.services import fundamentals_quality_service, system_event_service
 from app.services.price_service import query_ohlcv
 from app.services.qualification import setup_qualifies
 from app.services.sr_service import detect_gate_target_ladder
@@ -526,6 +527,7 @@ async def scan_ticker(
     primary_min_rr: float | None = None,
     gate_levels_override: list[Any] | None = None,
     scan_run_id: str | None = None,
+    fundamentals_eligible: bool | None = None,
 ) -> list[TradeSetup]:
     """Scan a single ticker for trade setups meeting the R:R threshold.
 
@@ -542,6 +544,17 @@ async def scan_ticker(
     """
     ticker = await _get_ticker(db, symbol)
 
+    if fundamentals_eligible is None:
+        fundamentals_eligible = await fundamentals_quality_service.ticker_is_eligible(
+            db, ticker.id
+        )
+    if not fundamentals_eligible:
+        logger.info(
+            "Skipping %s: unresolved or unavailable SEC fundamentals",
+            ticker.symbol,
+        )
+        return []
+
     if primary_min_rr is None:
         primary_min_rr = PRIMARY_TARGET_MIN_RR
 
@@ -726,6 +739,29 @@ async def scan_all_tickers(
     ticker_rows = [(int(ticker_id), symbol) for ticker_id, symbol in result.all()]
     total = len(ticker_rows)
 
+    # Data-quality failures are not weak signals: they make a ticker ineligible.
+    # Resolve once for the universe scan and pass the decision into scan_ticker.
+    try:
+        fundamentals_blocked_ids = (
+            await fundamentals_quality_service.blocked_ticker_ids(db)
+        )
+    except Exception:
+        await db.rollback()
+        logger.exception(
+            "Could not resolve fundamentals quality; blocking this scan closed"
+        )
+        await system_event_service.log_event_standalone(
+            severity="error",
+            source="rr_scanner",
+            code="fundamentals_quality_unavailable",
+            message=(
+                "The fundamentals quality gate could not be evaluated; the "
+                "universe scan was blocked to avoid issuing unchecked setups."
+            ),
+            dedup_key="rr_scanner:fundamentals_quality_unavailable",
+        )
+        fundamentals_blocked_ids = {ticker_id for ticker_id, _ in ticker_rows}
+
     # Gate-reset observations must use the same runtime activation settings as
     # the live setup list. If the config cannot be loaded, scan normally but do
     # not mutate reset state from an evaluation whose rules are unknown.
@@ -765,6 +801,12 @@ async def scan_all_tickers(
     for index, (ticker_id, symbol) in enumerate(ticker_rows):
         if progress_callback is not None:
             progress_callback(index, total, symbol)
+        if ticker_id in fundamentals_blocked_ids:
+            logger.info(
+                "Skipping %s: unresolved or unavailable SEC fundamentals",
+                symbol,
+            )
+            continue
         # Refresh Structural S/R once, then scores. get_sr_levels is read-only;
         # without this recalculate the score path would see yesterday's zones.
         # A refresh failure still scans the ticker: qualification re-gates on
@@ -795,6 +837,7 @@ async def scan_all_tickers(
                 volatility_percentile=(ranks.get(symbol) or {}).get("volatility_percentile"),
                 primary_min_rr=PRIMARY_TARGET_MIN_RR,
                 scan_run_id=scan_run_id,
+                fundamentals_eligible=True,
             )
             all_setups.extend(setups)
             if activation is not None:
@@ -882,6 +925,26 @@ async def get_trade_setups(
         stmt = stmt.where(TradeSetup.recommended_action == recommended_action)
     excluded_ticker_ids: set[int] = set()
     reentry_gate_locks: dict[int, datetime] = {}
+    try:
+        excluded_ticker_ids.update(
+            await fundamentals_quality_service.blocked_ticker_ids(db)
+        )
+    except Exception:
+        await db.rollback()
+        logger.exception(
+            "Could not resolve fundamentals quality; hiding actionable setups"
+        )
+        await system_event_service.log_event_standalone(
+            severity="error",
+            source="rr_scanner",
+            code="fundamentals_quality_unavailable",
+            message=(
+                "The fundamentals quality gate could not be evaluated; actionable "
+                "setups were hidden until the metadata check recovers."
+            ),
+            dedup_key="rr_scanner:fundamentals_quality_unavailable",
+        )
+        return []
     if exclude_open_trade_tickers:
         # Manual book only. The shadow book holds the *top-ranked* names by
         # construction, so letting its positions hide setups would leave the

app/services/sec_fundamentals_importer.py

@@ -31,10 +31,9 @@ Guardrails (design + reviews):
   stored as the parent's. Confirmed 2026-07-27 (NEE via FPL, DOW via Dow Chemical)
   and it is not transient: an NEE filing misattributed in 2014 is still misfiled.
 - **Bounded blocking.** Anything still unresolvable after ``MISSING_XBRL_RETRY_DAYS``
-  stops failing the run and is promoted around, with a named ``unresolved_filing``
-  warning. One filing SEC misfiled must not wedge every later import; the index
-  only moves forward, so a tolerated accession returns only via a reparse, and
-  only once SEC has re-filed it under the filer's own CIK.
+  stops failing the whole import and enters a durable retry queue. The scheduled
+  importer retries queued accessions automatically, while the affected issuer is
+  excluded from actionable setups until its filing is recovered.
 - ``promote`` inserts snapshots ``ON CONFLICT (accession) DO NOTHING`` (immutable),
   reports differing existing accessions, and applies ticker updates in the same
   transaction.
@@ -48,18 +47,21 @@ Guardrails (design + reviews):
 
 from __future__ import annotations
 
+import json
 import logging
 from collections import Counter, defaultdict
 from dataclasses import dataclass, field, replace
 from datetime import date, datetime, timedelta, timezone
 from typing import Any, Callable
 
-from sqlalchemy import select, update
+from sqlalchemy import delete, select, update
 
 from app.database import insert_for_session
 from app.models.data_import_run import DataImportRun
 from app.models.fundamental_snapshot import FundamentalSnapshot
+from app.models.sec_filing_gap import SecFilingGap
 from app.models.system_event import SystemEvent
+from app.services import fundamentals_quality_service
 from app.services import sec_facts_parser as parser
 from app.services import sec_universe
 from app.services.data_import import STATUS_PROMOTED, ValidationResult
@@ -79,6 +81,7 @@ 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
+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
 # wide enough to let a subsidiary shell's token float through (see _shares_continuous).
@@ -151,6 +154,7 @@ class SecFundamentalsImporter:
         # accession -> the OTHER CIKs the daily index lists it under (co-registrants
         # of a combined filing). Only populated for accessions a tracked issuer filed.
         self._coregistrants: dict[str, list[int]] = {}
+        self._retry_rows: list[dict[str, Any]] = []
         self._latest_index_date: date | None = None
         self._backfill = False
 
@@ -176,9 +180,19 @@ class SecFundamentalsImporter:
                     client, last_processed, self._latest_index_date
                 )
         content = sec_universe.index_content_hash(self._index_rows)
-        return sec_universe.compose_revision(
+        revision = sec_universe.compose_revision(
             self._latest_index_date, content, self._resolved.symbol_to_cik
         )
+        self._retry_rows = []
+        if not self._backfill:
+            self._retry_rows = await self._retry_backlog(
+                db,
+                set(self._resolved.cik_to_ticker_ids),
+            )
+        # Company Facts can change while the daily index revision stays fixed.
+        # Returning None deliberately bypasses the framework's no-op gate so a
+        # scheduled run retries every active gap.
+        return None if self._retry_rows else revision
 
     async def stage(self, db) -> StagedFundamentals:
         assert self._resolved is not None, "detect_revision must run first"
@@ -193,6 +207,26 @@ class SecFundamentalsImporter:
             if r["cik"] in cik_to_tids:
                 filed_by_cik[r["cik"]].append(r)
 
+        # Promoted-around filings live in a small durable retry queue, including
+        # the one-time migration backfill. Merge them into normal incremental
+        # work so the scheduled importer heals them without operator action.
+        if not self._backfill:
+            seen = {
+                (int(cik), row["accession"])
+                for cik, rows in filed_by_cik.items()
+                for row in rows
+            }
+            for row in self._retry_rows:
+                cik = int(row["cik"])
+                key = (cik, row["accession"])
+                if key in seen:
+                    continue
+                filed_by_cik[cik].append(row)
+                seen.add(key)
+                coregistrants = [int(value) for value in row.get("coregistrants") or []]
+                if coregistrants:
+                    self._coregistrants[row["accession"]] = coregistrants
+
         existing = await self._ciks_with_snapshots(db, set(cik_to_tids))
         if self._backfill:
             backfill_ciks = set(cik_to_tids)
@@ -247,6 +281,9 @@ class SecFundamentalsImporter:
 
         fiscal_year_end = sub.get("fiscal_year_end")
         recovered_rows: list[SnapshotRow] = []
+        index_rows = {
+            row["accession"]: row for row in filed_by_cik.get(cik, [])
+        }
         if is_backfill:
             accns = set(xbrl_meta)
         else:
@@ -262,7 +299,13 @@ class SecFundamentalsImporter:
                     # or a co-registrant filing). NOT a Company-Facts lag — separate
                     # cause, separate fix, so it gets its own reason.
                     staged.missing_xbrl.append(
-                        _missing(cik, index_row, "not_in_submissions", self.today)
+                        _missing(
+                            cik,
+                            index_row,
+                            "not_in_submissions",
+                            self.today,
+                            self._coregistrants.get(accn),
+                        )
                     )
                 elif accn in present:
                     accns.add(accn)
@@ -289,11 +332,26 @@ class SecFundamentalsImporter:
                             "coregistrant_facts_rejected" if source_cik
                             else "not_in_companyfacts",
                             self.today,
+                            self._coregistrants.get(accn),
                         ))
 
         # fiscalYearEnd (MMDD) is what lets the parser derive period identity from
         # reportDate instead of SEC's unreliable fy/fp fields.
         result = parser.parse_snapshots(cf, xbrl_meta, accns, fiscal_year_end=fiscal_year_end)
+        for skipped in result.skipped_filings:
+            index_row = index_rows.get(skipped["accession"])
+            if index_row is not None:
+                # Facts are present but our parser cannot construct a snapshot.
+                # 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,
+                    "parser_unusable",
+                    self.today,
+                    self._coregistrants.get(skipped["accession"]),
+                ))
         staged.rows.extend(result.rows)
         staged.rows.extend(recovered_rows)
         staged.skipped_filings.extend(result.skipped_filings)
@@ -398,11 +456,22 @@ class SecFundamentalsImporter:
             "skipped_non_xbrl": len(staged.skipped_non_xbrl),
             "no_xbrl_filings": staged.no_xbrl_filings[:50],
             "no_xbrl_filings_count": len(staged.no_xbrl_filings),
+            "no_xbrl_ciks": sorted({
+                str(item["cik"])
+                for item in staged.no_xbrl_filings
+                if item.get("cik")
+            }),
             "missing_xbrl": staged.missing_xbrl[:50],
             "missing_xbrl_count": len(staged.missing_xbrl),
             "missing_xbrl_blocking": len(blocking),
             "recovered_from_coregistrant": staged.recovered[:50],
             "recovered_count": len(staged.recovered),
+            # Complete compact gate input; detailed audit lists above stay capped.
+            "setup_blocked_ciks": sorted({
+                str(item["cik"])
+                for item in [*staged.missing_xbrl, *staged.no_xbrl_filings]
+                if item.get("cik")
+            }),
             "invalid_payloads": staged.invalid_payloads,
             "cik_updates": len(staged.resolved.cik_updates),
             # differing existing accessions (immutable — kept, reported here)
@@ -420,14 +489,17 @@ 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=(
                 [
                     f"{len(aged_out)} tracked SEC filing(s) remain unresolved past "
-                    f"the {MISSING_XBRL_RETRY_DAYS}-day retry window and risk being "
-                    f"promoted around without automatic retry: "
+                    f"the {MISSING_XBRL_RETRY_DAYS}-day retry window. They will "
+                    f"enter automatic retry and block affected symbols from setups: "
                     f"{_missing_detail(aged_out)}"
                 ]
                 if aged_out
@@ -462,6 +534,69 @@ class SecFundamentalsImporter:
             await db.execute(stmt)
             inserted += 1
 
+        # Synchronize the retry queue in the snapshot-promotion transaction.
+        existing_gaps = (await db.execute(select(SecFilingGap))).scalars().all()
+        existing_gap_accessions = {gap.accession for gap in existing_gaps}
+        resolved_accessions = {row.accession for row in staged.rows}
+        # A filing now classified non-XBRL can never yield a snapshot and is no
+        # longer a fundamentals completeness gap.
+        resolved_accessions.update(
+            item["accession"] for item in staged.skipped_non_xbrl
+        )
+        queue_resolved = 0
+        if resolved_accessions:
+            result = await db.execute(
+                delete(SecFilingGap).where(
+                    SecFilingGap.accession.in_(resolved_accessions)
+                )
+            )
+            queue_resolved = int(result.rowcount or 0)
+
+        now = _now()
+        tolerated = _past_retry_window(staged.missing_xbrl)
+        for gap in tolerated:
+            stmt = insert_for_session(db, SecFilingGap).values(
+                cik=gap["cik"],
+                accession=gap["accession"],
+                form=gap.get("form"),
+                index_date=gap.get("index_date"),
+                reason=gap["reason"],
+                coregistrant_ciks_json=json.dumps(gap.get("coregistrants") or []),
+                first_seen_at=now,
+                last_attempted_at=now,
+            )
+            await db.execute(
+                stmt.on_conflict_do_update(
+                    index_elements=["accession"],
+                    set_={
+                        "cik": stmt.excluded.cik,
+                        "form": stmt.excluded.form,
+                        "index_date": stmt.excluded.index_date,
+                        "reason": stmt.excluded.reason,
+                        "coregistrant_ciks_json": stmt.excluded.coregistrant_ciks_json,
+                        "last_attempted_at": stmt.excluded.last_attempted_at,
+                    },
+                )
+            )
+
+        # Remove gaps made irrelevant by a later valid 10-K/10-Q. Quality reads
+        # already ignore them; physical cleanup keeps the queue small.
+        active_ids = {gap.id for gap in await fundamentals_quality_service.active_gaps(db)}
+        obsolete_ids = {
+            gap.id for gap in existing_gaps
+            if gap.id not in active_ids and gap.accession not in resolved_accessions
+        }
+        if obsolete_ids:
+            result = await db.execute(
+                delete(SecFilingGap).where(SecFilingGap.id.in_(obsolete_ids))
+            )
+            queue_resolved += int(result.rowcount or 0)
+
+        newly_queued = [
+            gap for gap in tolerated
+            if gap["accession"] not in existing_gap_accessions
+        ]
+
         # Warn (in-transaction, so it commits atomically with the promotion) when
         # any existing accession reconstructed differently — kept immutable.
         if staged.discrepancies:
@@ -481,69 +616,87 @@ class SecFundamentalsImporter:
                 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)
+        aged_gaps = (
+            await db.execute(
+                select(SecFilingGap).where(
+                    SecFilingGap.first_seen_at <= escalation_cutoff,
+                    SecFilingGap.escalated_at.is_(None),
+                )
+            )
+        ).scalars().all()
+        if aged_gaps:
+            named = ", ".join(
+                f"{gap.cik}/{gap.accession} ({gap.reason})"
+                for gap in aged_gaps[:10]
+            )
+            db.add(SystemEvent(
+                severity="warning",
+                source="sec_facts",
+                code="filing_gap_aged",
+                message=(
+                    f"{len(aged_gaps)} SEC filing gap(s) remain unresolved after "
+                    f"{FILING_GAP_ESCALATE_DAYS} days; affected setups remain paused. "
+                    f"Review the filing/CIK mapping or parser: {named}"
+                )[:4000],
+                dedup_key=f"sec_facts:filing_gap_aged:{run_id}",
+                created_at=now,
+            ))
+            await db.execute(
+                update(SecFilingGap)
+                .where(SecFilingGap.id.in_([gap.id for gap in aged_gaps]))
+                .values(escalated_at=now)
+            )
+
         # 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:
             named = ", ".join(
                 f"{r['accession']} <- CIK {r['source_cik']}" for r in staged.recovered[:10]
             )
-            db.add(SystemEvent(
-                severity="warning",
-                source="sec_facts",
-                code="coregistrant_recovery",
-                message=(
-                    f"{len(staged.recovered)} filing(s) were absent from the filer's "
-                    f"own Company Facts and were parsed from a co-registrant's file "
-                    f"instead (share count checked against the issuer's history): {named}"
-                )[:4000],
-                dedup_key=f"sec_facts:coregistrant_recovery:{run_id}",
-                created_at=_now(),
-            ))
+            logger.info(
+                "sec_facts: recovered %d filing(s) from co-registrants: %s",
+                len(staged.recovered),
+                named,
+            )
 
-        # Filings past the retry window: promoted WITHOUT them so one misfiled
-        # filing cannot wedge every later import. This is the deliberate trade —
-        # loud and named, because the index only moves forward and nothing will
-        # revisit them on its own.
-        tolerated = _past_retry_window(staged.missing_xbrl)
-        if tolerated:
+        # One warning when a gap first enters automatic retry. Repeating it every
+        # day adds noise; the queue remains the durable actionable state.
+        if newly_queued:
+            symbols_by_cik: dict[str, list[str]] = defaultdict(list)
+            for symbol, cik in staged.resolved.symbol_to_cik.items():
+                symbols_by_cik[cik10(cik)].append(symbol)
+            named = ", ".join(
+                f"{'/'.join(symbols_by_cik.get(gap['cik'], [])) or gap['cik']}"
+                f"/{gap['accession']}"
+                for gap in newly_queued[:10]
+            )
             db.add(SystemEvent(
                 severity="warning",
                 source="sec_facts",
                 code="unresolved_filing",
                 message=(
-                    f"{len(tolerated)} tracked filing(s) still unresolvable after "
-                    f"{MISSING_XBRL_RETRY_DAYS} days; promoting without them rather "
-                    f"than blocking every later import. They are NOT retried. "
-                    f"scripts/reparse_fundamentals.py recovers them ONLY once SEC "
-                    f"re-files under the filer's own CIK — it walks no index, so it "
-                    f"cannot reach facts still sitting under a co-registrant: "
-                    f"{_missing_detail(tolerated)}"
+                    f"{len(newly_queued)} filing(s) entered automatic SEC retry. "
+                    f"Affected symbols are blocked from new actionable setups until "
+                    f"their filing is recovered: {named}"
                 )[:4000],
                 dedup_key=f"sec_facts:unresolved_filing:{run_id}",
                 created_at=_now(),
             ))
 
-        # A tracked issuer whose registrant has no XBRL filings can never produce a
-        # snapshot, and it is restaged on every run forever. That is a resolution
-        # problem, not missing data, and it is silent without this.
+        # A new registrant may have no XBRL filing yet. Keep it out of actionable
+        # setups, but log it instead of raising a recurring operator warning.
         if staged.no_xbrl_filings:
             named = ", ".join(
                 f"{e['cik']} ({e.get('name') or '?'})" for e in staged.no_xbrl_filings[:10]
             )
-            db.add(SystemEvent(
-                severity="warning",
-                source="sec_facts",
-                code="no_xbrl_filings",
-                message=(
-                    f"{len(staged.no_xbrl_filings)} tracked issuer(s) resolved to a "
-                    f"registrant with no XBRL 10-K/10-Q. Either a successor shell "
-                    f"(pin the real filer via the '{sec_universe.CIK_OVERRIDES_KEY}' "
-                    f"setting) or a new registrant that has not filed its first "
-                    f"10-K/10-Q yet, which needs nothing and clears itself: {named}"
-                )[:4000],
-                dedup_key=f"sec_facts:no_xbrl_filings:{run_id}",
-                created_at=_now(),
-            ))
+            logger.info(
+                "sec_facts: %d registrant(s) have no XBRL history yet: %s",
+                len(staged.no_xbrl_filings),
+                named,
+            )
 
         ticker_counts = await sec_universe.apply_ticker_updates(
             db, staged.resolved, staged.sic_updates
@@ -553,11 +706,57 @@ class SecFundamentalsImporter:
             "updated": updated,
             "existing_unchanged": len(staged.existing_accessions) - updated,
             "discrepancies": len(staged.discrepancies),
+            "retry_queue_added": len(newly_queued),
+            "retry_queue_resolved": queue_resolved,
             **ticker_counts,
         }
 
     # -- helpers -----------------------------------------------------------
 
+    async def _retry_backlog(
+        self,
+        db,
+        tracked_ciks: set[int],
+    ) -> list[dict[str, Any]]:
+        """Active typed gaps; migration 028 owns historical bootstrap."""
+        if not tracked_ciks:
+            return []
+        tracked = {cik10(cik) for cik in tracked_ciks}
+        candidates: dict[str, dict[str, Any]] = {}
+
+        queued = await fundamentals_quality_service.active_gaps(db, tracked)
+        for gap in queued:
+            try:
+                coregistrants = json.loads(gap.coregistrant_ciks_json or "[]")
+            except (TypeError, ValueError):
+                coregistrants = []
+            candidates[gap.accession] = {
+                "cik": gap.cik,
+                "accession": gap.accession,
+                "form": gap.form,
+                "index_date": gap.index_date,
+                "reason": gap.reason,
+                "coregistrants": coregistrants,
+                "_retry_queue": True,
+            }
+
+        if not candidates:
+            return []
+        resolved = set(
+            (
+                await db.execute(
+                    select(FundamentalSnapshot.accession).where(
+                        FundamentalSnapshot.accession.in_(list(candidates))
+                    )
+                )
+            ).scalars().all()
+        )
+        return [
+            item
+            for accession, item in candidates.items()
+            if accession not in resolved
+        ]
+
     async def _last_processed_index_date(self, db) -> date | None:
         return (
             await db.execute(
@@ -666,19 +865,32 @@ def _companyfacts_structure_error(cf: Any) -> str | None:
     return None
 
 
-def _missing(cik: int, row: dict[str, Any], reason: str, today: date) -> dict[str, Any]:
+def _missing(
+    cik: int,
+    row: dict[str, Any],
+    reason: str,
+    today: date,
+    coregistrants: list[int] | None = None,
+) -> dict[str, Any]:
     """One unresolvable index row, carrying everything needed to look the filing
     up by hand (EDGAR accession + the index date it was seen on) and to decide
     whether it is still young enough to be worth blocking on."""
     index_date = row.get("index_date")
+    age_days = (
+        (today - index_date).days if isinstance(index_date, date) else 0
+    )
+    if row.get("_retry_queue"):
+        age_days = max(age_days, MISSING_XBRL_RETRY_DAYS + 1)
     return {
         "cik": cik10(cik),
         "accession": row["accession"],
         "form": row.get("form"),
         "index_date": index_date,
-        # No index date (older cached rows) => age 0 => blocks, the safe default.
-        "age_days": (today - index_date).days if isinstance(index_date, date) else 0,
+        # A newly observed row without a date blocks safely. A durable queue row
+        # has already passed the bounded window and is forced aged-out above.
+        "age_days": age_days,
         "reason": reason,
+        "coregistrants": list(coregistrants or []),
     }
 
 

docs/fundamentals-deployment.md

@@ -15,6 +15,11 @@ not add OS cron entries: the application scheduler owns both jobs.
 - Cron expressions are editable in Admin → Schedule.
 - Every attempt is recorded in `data_import_runs`; failures also create a system
   event. A failed validation does not promote partial data.
+- An SEC filing still missing after the short publication-lag window enters
+  `sec_filing_gaps`. The daily importer retries it automatically; affected
+  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.
 
 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
@@ -83,7 +88,8 @@ In Admin → Jobs, wait until no other job is running, then:
 3. Check Admin → System Events. There should be no new import error.
 4. Confirm the next-run times correspond to 02:30 and 04:00 New York time.
 5. Open several ticker pages and confirm the fundamentals panel has populated
-   data and still handles partial/missing issuers cleanly.
+   data and still handles partial/missing issuers cleanly. A ticker held by the
+   quality gate should show **New setups paused** with the specific SEC reason.
 
 ## A5 parity observation window
 
@@ -231,6 +237,14 @@ least several scheduled cycles before A6 removes the legacy providers.
   must stop. Existing promoted snapshots/events remain available.
 - Inspect the job runtime, latest `data_import_runs.validation_json`, service
   logs, and Admin → System Events before retrying.
+- `unresolved_filing` is emitted once when a filing enters automatic retry. It
+  does not require a server command. If the gap is still current after 14 days,
+  `filing_gap_aged` is emitted once with the CIK, accession, and parser/mapping
+  reason. A later valid 10-K/10-Q retires the gap even when the original SEC
+  accession never becomes usable.
+- Successful co-registrant recovery is logged without a warning. New registrants
+  with no XBRL history are also logged quietly, but their ticker page explains
+  that setups remain paused and that successor shells may need `sec_cik_overrides`.
 - Re-run `sudo -u deploy bash ./deploy/provision_fundamentals.sh --check` for
   binary, clone, permission, disk, or environment failures.
 - The Dolt clone is a reproducible cache and does not need a bespoke backup.

frontend/src/components/dashboard/OpenTradesPanel.tsx

@@ -22,6 +22,24 @@ function pnlColor(v: number): string {
   return 'text-gray-300';
 }
 
+function maxHoldText(trade: PaperTrade, compact = false): string | null {
+  const remaining = trade.sessions_remaining;
+  if (remaining == null) return null;
+  const held = trade.sessions_held ?? 0;
+  if (remaining < 0) return compact ? 'past max hold' : `${held} held · past max hold`;
+  if (remaining === 0) return compact ? 'max hold reached' : `${held} held · max hold reached`;
+  if (compact) return `${remaining} ${remaining === 1 ? 'session' : 'sessions'} left`;
+  return `${held} held · ${remaining} remaining`;
+}
+
+function maxHoldColor(trade: PaperTrade): string {
+  const remaining = trade.sessions_remaining;
+  if (remaining == null) return 'text-gray-400';
+  const holdDays = Math.max(1, (trade.sessions_held ?? 0) + remaining);
+  const warningAt = Math.max(1, Math.ceil(holdDays * 0.2));
+  return remaining <= warningAt ? 'text-amber-300' : 'text-gray-400';
+}
+
 function DirTag({ direction }: { direction: string }) {
   const isLong = direction === 'long';
   return (
@@ -116,6 +134,13 @@ function TradeDetail({ trade, exitLabel, exitMode, atrMultiplier, trailingPct, o
           }
         />
         <Detail label="exit rule" value={exitLabel ?? 'target/stop'} />
+        {maxHoldText(trade) && (
+          <Detail
+            label="max hold"
+            value={maxHoldText(trade)}
+            valueClass={maxHoldColor(trade)}
+          />
+        )}
         <div className="flex items-end">
           <button
             onClick={onClose}
@@ -173,13 +198,14 @@ export function OpenTradesPanel() {
   const trailingPct = policy?.trailing_pct ?? 12;
   const exitLabel = policy
     ? policy.mode === 'atr_trailing'
-      ? `${atrMultiplier.toFixed(1)}x ATR trailing stop / ${policy.hold_days}d max`
+      ? `${atrMultiplier.toFixed(1)}x ATR trailing stop / ${policy.hold_days} sessions max`
       : policy.mode === 'trailing'
         ? `trailing ${Math.round(trailingPct)}%`
         : policy.mode === 'time'
-          ? `${policy.hold_days}d hold`
+          ? `${policy.hold_days}-session hold`
           : 'target/stop'
     : null;
+  const hasMaxHold = exitMode === 'atr_trailing' || exitMode === 'time';
 
   const rows = trades ?? [];
 
@@ -217,6 +243,7 @@ export function OpenTradesPanel() {
             {rows.map((t) => {
               const p = tradePnl(t);
               const open = expandedId === t.id;
+              const holdText = maxHoldText(t, true);
               return (
                 <li key={t.id}>
                   <div
@@ -230,7 +257,11 @@ export function OpenTradesPanel() {
                       }
                     }}
                     aria-expanded={open}
-                    className="grid w-full cursor-pointer grid-cols-[110px_1fr_60px_16px] items-center gap-3 rounded-lg px-2 py-2.5 text-left transition-colors hover:bg-white/[0.03] sm:grid-cols-[130px_150px_1fr_70px_16px]"
+                    className={`grid w-full cursor-pointer grid-cols-[110px_1fr_60px_16px] items-center gap-3 rounded-lg px-2 py-2.5 text-left transition-colors hover:bg-white/[0.03] ${
+                      hasMaxHold
+                        ? 'sm:grid-cols-[130px_150px_1fr_70px_16px] lg:grid-cols-[130px_150px_1fr_110px_70px_16px]'
+                        : 'sm:grid-cols-[130px_150px_1fr_70px_16px]'
+                    }`}
                   >
                     <span className="flex items-center gap-2">
                       <Link
@@ -246,6 +277,14 @@ export function OpenTradesPanel() {
                       {formatPrice(t.entry_price)} → {t.current_price != null ? formatPrice(t.current_price) : '—'}
                     </span>
                     <RBar r={p?.r ?? null} max={rMax} />
+                    {hasMaxHold && (
+                      <span
+                        className={`num hidden text-right text-[11px] lg:block ${maxHoldColor(t)}`}
+                        title="Maximum holding period; the stop may close this trade sooner."
+                      >
+                        {holdText ?? '—'}
+                      </span>
+                    )}
                     <span className={`num text-right text-[13px] font-semibold ${p?.r != null ? pnlColor(p.r) : 'text-gray-500'}`}>
                       {p?.r != null ? `${p.r >= 0 ? '+' : ''}${p.r.toFixed(2)}R` : '—'}
                     </span>

frontend/src/dev/harness.tsx

@@ -38,6 +38,7 @@ const ind = (median: number, favorable_percentile: number) =>
 const legacy = {
   pe_ratio: null, revenue_growth: null, earnings_surprise: null, market_cap: null,
   next_earnings_date: null, fetched_at: null, unavailable_fields: {},
+  setup_eligible: true, setup_block_code: null, setup_block_reason: null,
 };
 
 const full: FundamentalResponse = {

frontend/src/lib/types.ts

@@ -237,6 +237,8 @@ export interface PaperTrade {
   close_reason: 'time' | 'trailing' | 'stop' | 'target' | 'manual' | null;
   trailing_stop: number | null;
   trailing_distance_pct: number | null;
+  sessions_held: number | null;
+  sessions_remaining: number | null;
 }
 
 export interface ExitPolicy {
@@ -827,6 +829,9 @@ export interface FundamentalResponse {
   metrics: MetricItem[] | null;
   valuation: Valuation | null;
   reads: FundamentalsReads | null;
+  setup_eligible: boolean;
+  setup_block_code: string | null;
+  setup_block_reason: string | null;
 }
 
 // Indicators

frontend/src/pages/TickerDetailPage.tsx

@@ -64,10 +64,44 @@ function timeAgo(iso: string): string {
   return `${days}d ago`;
 }
 
+function marketDate(date = new Date()): string {
+  const parts = new Intl.DateTimeFormat('en-US', {
+    timeZone: 'America/New_York',
+    year: 'numeric',
+    month: '2-digit',
+    day: '2-digit',
+  }).formatToParts(date);
+  const value = (type: Intl.DateTimeFormatPartTypes) =>
+    parts.find((part) => part.type === type)?.value ?? '';
+  return value('year') + '-' + value('month') + '-' + value('day');
+}
+
+function formatSessionDate(isoDate: string): string {
+  const currentMarketDate = marketDate();
+  if (isoDate === currentMarketDate) return 'Today';
+
+  // Parse date-only market sessions explicitly. Parsing YYYY-MM-DD directly as
+  // a Date means midnight UTC and makes today's bar look many hours old.
+  const [year, month, day] = isoDate.split('-').map(Number);
+  if (!year || !month || !day) return isoDate;
+  return new Intl.DateTimeFormat(undefined, {
+    month: 'short',
+    day: 'numeric',
+    year: year === Number(currentMarketDate.slice(0, 4)) ? undefined : 'numeric',
+    timeZone: 'UTC',
+  }).format(new Date(Date.UTC(year, month - 1, day)));
+}
+
+function formatOHLCVFreshness(sessionDate: string, updatedAt?: string | null): string {
+  const session = formatSessionDate(sessionDate);
+  return updatedAt ? session + ' · updated ' + timeAgo(updatedAt) : session;
+}
+
 interface DataStatusItem {
   label: string;
   available: boolean;
   timestamp?: string | null;
+  timestampLabel?: string | null;
   selector: FetchSelector;   // what a refresh of this row fetches
   paid?: boolean;            // provider call that may cost money/quota
 }
@@ -100,7 +134,7 @@ function DataFreshnessBar({
           }`} />
           <span className="text-xs text-gray-400">{item.label}</span>
           {item.available && item.timestamp ? (
-            <span className="text-[10px] text-gray-500">{timeAgo(item.timestamp)}</span>
+            <span className="text-[10px] text-gray-500">{item.timestampLabel ?? timeAgo(item.timestamp)}</span>
           ) : !item.available ? (
             <span className="text-[10px] text-gray-600">no data</span>
           ) : null}
@@ -171,10 +205,16 @@ export default function TickerDetailPage() {
   const dataStatus: DataStatusItem[] = useMemo(() => [
     {
       label: 'OHLCV',
-      // Market age of the latest bar (session date), not DB insert time —
-      // created_at stays frozen when the provider returns no new sessions.
+      // Keep the market session date distinct from the last successful bar
+      // write; treating YYYY-MM-DD as an instant makes today's session look old.
       available: !!ohlcv.data && ohlcv.data.length > 0,
       timestamp: ohlcv.data?.[ohlcv.data.length - 1]?.date,
+      timestampLabel: ohlcv.data?.length
+        ? formatOHLCVFreshness(
+            ohlcv.data[ohlcv.data.length - 1].date,
+            ohlcv.data[ohlcv.data.length - 1].created_at,
+          )
+        : null,
       selector: ['ohlcv'] as FetchSelector,
       paid: true,
     },
@@ -319,6 +359,15 @@ export default function TickerDetailPage() {
             busy={ingestion.isPending}
           />
         </div>
+        {fundamentals.data && !fundamentals.data.setup_eligible && (
+          <div className="border-b border-white/[0.06] px-6 py-3 sm:px-7">
+            <Callout variant="warning">
+              <span className="font-medium">New setups paused.</span>{' '}
+              {fundamentals.data.setup_block_reason ??
+                'SEC fundamentals are incomplete for this ticker.'}
+            </Callout>
+          </div>
+        )}
         <div className="p-6 pb-5 sm:p-7 sm:pb-5">
         <div className="flex flex-wrap items-start justify-between gap-x-8 gap-y-5">
           <div className="min-w-0">

tests/conftest.py

@@ -15,6 +15,8 @@ from sqlalchemy.ext.asyncio import (
     create_async_engine,
 )
 
+from sqlalchemy import delete
+
 from app.database import Base
 from app.providers.protocol import OHLCVData
 
@@ -32,14 +34,29 @@ _test_session_factory = async_sessionmaker(
 )
 
 
+_schema_created = False
+
+
 @pytest.fixture(autouse=True)
 async def _setup_db():
-    """Create all tables before each test and drop them after."""
+    """Hand every test an empty database.
+
+    The schema is built once and then truncated per test rather than dropped and
+    recreated. A create_all/drop_all cycle costs ~49ms against these 22 tables and
+    ran for every test in the suite — including the many that never open a session
+    — where deleting every row costs ~6ms for the same guarantee. No model sets
+    ``sqlite_autoincrement``, so SQLite reuses rowids after a full delete and
+    generated ids still restart at 1.
+    """
+    global _schema_created
     async with _test_engine.begin() as conn:
+        if not _schema_created:
             await conn.run_sync(Base.metadata.create_all)
+            _schema_created = True
+        else:
+            for table in reversed(Base.metadata.sorted_tables):
+                await conn.execute(delete(table))
     yield
-    async with _test_engine.begin() as conn:
-        await conn.run_sync(Base.metadata.drop_all)
 
 
 @pytest.fixture

tests/unit/test_fundamentals_quality_service.py

@@ -0,0 +1,183 @@
+from __future__ import annotations
+
+import json
+from datetime import date, datetime, timezone
+
+from app.models.data_import_run import DataImportRun
+from app.models.fundamental_snapshot import FundamentalSnapshot
+from app.models.sec_filing_gap import SecFilingGap
+from app.models.settings import SystemSetting
+from app.models.ticker import Ticker
+from app.services import fundamentals_quality_service
+
+
+async def test_latest_sec_validation_blocks_deferred_and_no_history_ciks(
+    db_session,
+):
+    missing = Ticker(symbol="MISSING", cik="0000000001")
+    no_history = Ticker(symbol="NEWREG", cik="0000000002")
+    healthy = Ticker(symbol="HEALTHY", cik="0000000003")
+    db_session.add_all([missing, no_history, healthy])
+    await db_session.flush()
+    db_session.add(
+        SystemSetting(
+            key="fundamental_data_sec_dolt_cutover_enabled",
+            value="true",
+        )
+    )
+    db_session.add(
+        DataImportRun(
+            source="sec_facts",
+            status="deferred",
+            validation_json=json.dumps({
+                "missing_xbrl": [{"cik": missing.cik, "accession": "MISSING-Q"}],
+                "no_xbrl_filings": [{"cik": no_history.cik}],
+            }),
+            started_at=datetime.now(timezone.utc),
+        )
+    )
+    await db_session.flush()
+
+    assert await fundamentals_quality_service.blocked_ticker_ids(db_session) == {
+        missing.id,
+        no_history.id,
+    }
+
+
+async def test_sec_quality_gate_is_inactive_before_cutover(db_session):
+    ticker = Ticker(symbol="SHADOW", cik="0000000042")
+    db_session.add(ticker)
+    await db_session.flush()
+    now = datetime.now(timezone.utc)
+    db_session.add(
+        SecFilingGap(
+            cik=ticker.cik,
+            accession="SHADOW-Q",
+            form="10-Q",
+            index_date=date.today(),
+            reason="not_in_companyfacts",
+            first_seen_at=now,
+            last_attempted_at=now,
+        )
+    )
+    await db_session.flush()
+
+    assert await fundamentals_quality_service.blocked_ticker_ids(db_session) == set()
+
+
+
+
+async def test_active_gap_is_blocked_until_a_later_filing_supersedes_it(db_session):
+    ticker = Ticker(symbol="HIST", cik="0000000043")
+    now = datetime.now(timezone.utc)
+    db_session.add_all([
+        ticker,
+        SystemSetting(
+            key="fundamental_data_sec_dolt_cutover_enabled",
+            value="true",
+        ),
+        SecFilingGap(
+            cik=ticker.cik,
+            accession="HIST-Q",
+            form="10-Q",
+            index_date=date.today().replace(day=1),
+            reason="coregistrant_facts_rejected",
+            first_seen_at=now,
+            last_attempted_at=now,
+        ),
+    ])
+    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-Q",
+            form="10-Q",
+            filed_date=date.today(),
+            accepted_at=datetime.now(timezone.utc),
+            period_end=date.today(),
+            fiscal_year=date.today().year,
+            fiscal_period="Q2",
+        )
+    )
+    await db_session.flush()
+
+    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([
+        ticker,
+        SystemSetting(
+            key="fundamental_data_sec_dolt_cutover_enabled",
+            value="true",
+        ),
+        DataImportRun(
+            source="sec_facts",
+            status="promoted",
+            validation_json=json.dumps({
+                "setup_blocked_ciks": [ticker.cik],
+                "no_xbrl_ciks": [ticker.cik],
+                "no_xbrl_filings": [],
+            }),
+            started_at=datetime.now(timezone.utc),
+        ),
+    ])
+    await db_session.flush()
+
+    quality = await fundamentals_quality_service.ticker_quality(db_session, "NEWREG")
+    assert quality.eligible is False
+    assert quality.code == "no_xbrl_filings"
+    assert "CIK override" in (quality.message or "")
+    assert await fundamentals_quality_service.ticker_is_eligible(
+        db_session, ticker.id
+    ) is False

tests/unit/test_ingestion_service.py

@@ -6,6 +6,8 @@ from datetime import date, timedelta
 
 import pytest
 
+from app.models.ohlcv import OHLCVRecord
+from app.models.settings import IngestionProgress
 from app.models.ticker import Ticker
 from app.providers.protocol import OHLCVData
 from app.services import ingestion_service as svc
@@ -18,9 +20,12 @@ async def session():
         yield s
 
 
-async def _add_ticker(session, symbol: str) -> None:
-    session.add(Ticker(symbol=symbol))
+async def _add_ticker(session, symbol: str) -> Ticker:
+    ticker = Ticker(symbol=symbol)
+    session.add(ticker)
     await session.commit()
+    await session.refresh(ticker)
+    return ticker
 
 
 def _bars(symbol: str, n: int) -> list[OHLCVData]:
@@ -51,6 +56,50 @@ async def test_happy_path_ingests_bars(session):
     assert result.records_ingested == 3
 
 
+async def test_incremental_fetch_overlaps_latest_session_and_updates_partial_bar(session):
+    """Once today exists, a live refresh must fetch and overwrite it again."""
+    ticker = await _add_ticker(session, "LIVE")
+    today = date.today()
+    session.add_all([
+        OHLCVRecord(
+            ticker_id=ticker.id,
+            date=today - timedelta(days=i),
+            open=100.0,
+            high=101.0,
+            low=99.0,
+            close=100.0,
+            volume=1000,
+        )
+        for i in range(200)
+    ])
+    session.add(IngestionProgress(ticker_id=ticker.id, last_ingested_date=today))
+    await session.commit()
+
+    provider = MockMarketDataProvider(ohlcv_data=[
+        OHLCVData(
+            ticker="LIVE",
+            date=today,
+            open=100.0,
+            high=124.0,
+            low=99.0,
+            close=123.0,
+            volume=2000,
+        )
+    ])
+    result = await svc.fetch_and_ingest(session, provider, "LIVE")
+
+    assert provider.calls == [{
+        "ticker": "LIVE",
+        "start_date": today,
+        "end_date": today,
+    }]
+    assert result.status == "complete"
+    assert result.records_ingested == 1
+    records = await svc.price_service.query_ohlcv(session, "LIVE", today, today)
+    assert records[0].close == 123.0
+    assert records[0].volume == 2000
+
+
 async def test_empty_fetch_with_existing_history_is_up_to_date(session):
     # Covered ticker, just no new bars in the window → complete, not no_data.
     await _add_ticker(session, "BBB")
@@ -81,9 +130,34 @@ async def test_empty_fetch_with_stale_history_reports_stale(session):
     ]
     await svc.fetch_and_ingest(session, MockMarketDataProvider(ohlcv_data=old), "SATS")
 
-    result = await svc.fetch_and_ingest(session, MockMarketDataProvider(ohlcv_data=[]), "SATS")
+    # Incremental overlap means Alpaca can keep returning the final historical
+    # bar. That is still stale: the latest session did not advance.
+    result = await svc.fetch_and_ingest(
+        session,
+        MockMarketDataProvider(ohlcv_data=[old[-1]]),
+        "SATS",
+    )
 
     assert result.status == "stale"
-    assert result.records_ingested == 0
+    assert result.records_ingested == 1
     assert result.last_date is not None
     assert "renamed" in (result.message or "").lower() or "halted" in (result.message or "").lower()
+
+
+async def test_ingest_can_skip_sr_refresh_when_scanner_follows(session, monkeypatch):
+    await _add_ticker(session, "SCAN")
+    calls: list[str] = []
+
+    async def fake_refresh(db, symbol):
+        calls.append(symbol)
+
+    monkeypatch.setattr(svc, "_refresh_structural_sr", fake_refresh)
+    result = await svc.fetch_and_ingest(
+        session,
+        MockMarketDataProvider(ohlcv_data=_bars("SCAN", 3)),
+        "SCAN",
+        refresh_sr=False,
+    )
+
+    assert result.status == "complete"
+    assert calls == []

tests/unit/test_paper_trade_service.py

@@ -56,6 +56,116 @@ async def test_create_and_list_open(session):
     assert row["symbol"] == "AAA"
     assert row["status"] == "open"
     assert row["current_price"] == 110.0  # marked to the latest close
+    assert row["sessions_held"] == 0
+    assert row["sessions_remaining"] == 30
+
+
+async def test_list_open_counts_post_entry_sessions_for_max_hold(session):
+    await svc.set_exit_policy(session, mode="atr_trailing", hold_days=5)
+    ticker_id = await _seed(session, "COUNT", close=110.0)
+    trade = await svc.create_trade(
+        session,
+        1,
+        symbol="COUNT",
+        direction="long",
+        entry_price=100.0,
+        shares=10,
+        stop_loss=95.0,
+        target=120.0,
+    )
+    today = _today()
+    trade.opened_at = datetime.combine(
+        today - timedelta(days=5), datetime.min.time(), tzinfo=timezone.utc
+    )
+    session.add_all([
+        OHLCVRecord(
+            ticker_id=ticker_id,
+            date=today - timedelta(days=4),
+            open=101,
+            high=102,
+            low=100,
+            close=101,
+            volume=1,
+        ),
+        OHLCVRecord(
+            ticker_id=ticker_id,
+            date=today - timedelta(days=2),
+            open=102,
+            high=103,
+            low=101,
+            close=102,
+            volume=1,
+        ),
+    ])
+    await session.commit()
+
+    row = (await svc.list_trades(session, 1, status="open"))[0]
+    # Two added bars plus today's seeded bar; skipped calendar dates do not count.
+    assert row["sessions_held"] == 3
+    assert row["sessions_remaining"] == 2
+
+
+async def test_list_open_exposes_past_max_hold_after_policy_is_shortened(session):
+    await svc.set_exit_policy(session, mode="time", hold_days=2)
+    ticker_id = await _seed(session, "OVERDUE", close=110.0)
+    trade = await svc.create_trade(
+        session,
+        1,
+        symbol="OVERDUE",
+        direction="long",
+        entry_price=100.0,
+        shares=10,
+        stop_loss=95.0,
+        target=120.0,
+    )
+    today = _today()
+    trade.opened_at = datetime.combine(
+        today - timedelta(days=5), datetime.min.time(), tzinfo=timezone.utc
+    )
+    session.add_all([
+        OHLCVRecord(
+            ticker_id=ticker_id,
+            date=today - timedelta(days=4),
+            open=101,
+            high=102,
+            low=100,
+            close=101,
+            volume=1,
+        ),
+        OHLCVRecord(
+            ticker_id=ticker_id,
+            date=today - timedelta(days=2),
+            open=102,
+            high=103,
+            low=101,
+            close=102,
+            volume=1,
+        ),
+    ])
+    await session.commit()
+
+    row = (await svc.list_trades(session, 1, status="open"))[0]
+    assert row["sessions_held"] == 3
+    assert row["sessions_remaining"] == -1
+
+
+async def test_list_open_omits_countdown_without_max_hold_policy(session):
+    await svc.set_exit_policy(session, mode="trailing")
+    await _seed(session, "NOHOLD", close=110.0)
+    await svc.create_trade(
+        session,
+        1,
+        symbol="NOHOLD",
+        direction="long",
+        entry_price=100.0,
+        shares=10,
+        stop_loss=95.0,
+        target=120.0,
+    )
+
+    row = (await svc.list_trades(session, 1, status="open"))[0]
+    assert row["sessions_held"] is None
+    assert row["sessions_remaining"] is None
 
 
 async def test_create_trade_enforces_post_stop_gate_reset_at_service_boundary(session):

tests/unit/test_rr_scanner_bug_exploration.py

@@ -1,285 +0,0 @@
-"""Regression: scanner must not headline the most distant (max raw R:R) level.
-
-Historical bug: provisional candidate pick used max R:R / quality only. Production
-headline is probability-based primary after enhance_trade_setup — near levels
-with real reach-probability beat far lotteries.
-
-**Validates: Requirements 1.1, 1.3, 1.4, 2.1, 2.3, 2.4**
-"""
-
-from __future__ import annotations
-
-from datetime import date, timedelta
-
-import pytest
-from hypothesis import given, settings, HealthCheck, strategies as st
-from sqlalchemy.ext.asyncio import AsyncSession
-
-from app.models.ohlcv import OHLCVRecord
-from app.models.sr_level import SRLevel
-from app.models.ticker import Ticker
-from app.services.rr_scanner_service import scan_ticker
-
-
-# ---------------------------------------------------------------------------
-# Session fixture that allows scan_ticker to commit
-# ---------------------------------------------------------------------------
-# The default db_session fixture wraps in session.begin() which conflicts
-# with scan_ticker's internal commit().  We use a plain session instead.
-
-@pytest.fixture
-async def scan_session() -> AsyncSession:
-    """Provide a DB session compatible with scan_ticker (which commits)."""
-    from tests.conftest import _test_session_factory
-
-    async with _test_session_factory() as session:
-        yield session
-
-
-# ---------------------------------------------------------------------------
-# Helpers
-# ---------------------------------------------------------------------------
-
-def _make_ohlcv_bars(
-    ticker_id: int,
-    num_bars: int = 20,
-    base_close: float = 100.0,
-) -> list[OHLCVRecord]:
-    """Generate realistic OHLCV bars with small daily variation.
-
-    Produces bars where close ≈ base_close, with enough range for ATR
-    computation (needs >= 15 bars).  The ATR will be roughly 2.0.
-    """
-    bars: list[OHLCVRecord] = []
-    start = date(2024, 1, 1)
-    for i in range(num_bars):
-        close = base_close + (i % 3 - 1) * 0.5  # oscillate ±0.5
-        bars.append(OHLCVRecord(
-            ticker_id=ticker_id,
-            date=start + timedelta(days=i),
-            open=close - 0.3,
-            high=close + 1.0,
-            low=close - 1.0,
-            close=close,
-            volume=100_000,
-        ))
-    return bars
-
-
-# ---------------------------------------------------------------------------
-# Deterministic test: strong-near vs weak-far (long setup)
-# ---------------------------------------------------------------------------
-
-@pytest.mark.asyncio
-async def test_long_prefers_strong_near_over_weak_far(scan_session: AsyncSession):
-    """With a strong nearby resistance and a weak distant resistance, the
-    probability primary should be the nearby level — NOT the far lottery.
-    """
-    ticker = Ticker(symbol="EXPLR")
-    scan_session.add(ticker)
-    await scan_session.flush()
-
-    # 20 bars closing around 100
-    bars = _make_ohlcv_bars(ticker.id, num_bars=20, base_close=100.0)
-    scan_session.add_all(bars)
-
-    # With ATR=2.0 and multiplier=1.5, risk=3.0.
-    # R:R threshold=1.5 → min reward=4.5 → min target=104.5
-    # Strong nearby resistance: price=105, strength=90 (R:R≈1.67, quality≈0.66)
-    near_level = SRLevel(
-        ticker_id=ticker.id,
-        price_level=105.0,
-        type="resistance",
-        strength=90,
-        detection_method="volume_profile",
-    )
-    # Weak distant resistance: price=130, strength=5 (R:R=10, quality≈0.58)
-    far_level = SRLevel(
-        ticker_id=ticker.id,
-        price_level=130.0,
-        type="resistance",
-        strength=5,
-        detection_method="volume_profile",
-    )
-    scan_session.add_all([near_level, far_level])
-    await scan_session.flush()
-
-    setups = await scan_ticker(
-        scan_session,
-        "EXPLR",
-        rr_threshold=1.5,
-        gate_levels_override=[near_level, far_level],
-    )
-
-    long_setups = [s for s in setups if s.direction == "long"]
-    assert len(long_setups) == 1, "Expected exactly one long setup"
-
-    selected_target = long_setups[0].target
-    # The scanner must NOT pick the most distant level (130)
-    assert selected_target != pytest.approx(130.0, abs=0.01), (
-        "Bug: scanner picked the weak distant level (130) instead of the "
-        "strong nearby level (105)"
-    )
-    # Probability primary should pick the strong nearby level
-    assert selected_target == pytest.approx(105.0, abs=0.01)
-    primaries = [t for t in long_setups[0].targets if t.get("is_primary")]
-    assert len(primaries) == 1
-    assert primaries[0]["price"] == pytest.approx(105.0, abs=0.01)
-
-
-# ---------------------------------------------------------------------------
-# Deterministic test: strong-near vs weak-far (short setup)
-# ---------------------------------------------------------------------------
-
-@pytest.mark.asyncio
-async def test_short_prefers_strong_near_over_weak_far(scan_session: AsyncSession):
-    """Short-side mirror: strong nearby support should be preferred over
-    weak distant support.
-    """
-    ticker = Ticker(symbol="EXPLS")
-    scan_session.add(ticker)
-    await scan_session.flush()
-
-    bars = _make_ohlcv_bars(ticker.id, num_bars=20, base_close=100.0)
-    scan_session.add_all(bars)
-
-    # With ATR=2.0 and multiplier=1.5, risk=3.0.
-    # R:R threshold=1.5 → min reward=4.5 → min target below 95.5
-    # Strong nearby support: price=95, strength=85 (R:R≈1.67, quality≈0.64)
-    near_level = SRLevel(
-        ticker_id=ticker.id,
-        price_level=95.0,
-        type="support",
-        strength=85,
-        detection_method="pivot_point",
-    )
-    # Weak distant support: price=70, strength=5 (R:R=10, quality≈0.58)
-    far_level = SRLevel(
-        ticker_id=ticker.id,
-        price_level=70.0,
-        type="support",
-        strength=5,
-        detection_method="pivot_point",
-    )
-    scan_session.add_all([near_level, far_level])
-    await scan_session.flush()
-
-    setups = await scan_ticker(
-        scan_session,
-        "EXPLS",
-        rr_threshold=1.5,
-        gate_levels_override=[near_level, far_level],
-    )
-
-    short_setups = [s for s in setups if s.direction == "short"]
-    assert len(short_setups) == 1, "Expected exactly one short setup"
-
-    selected_target = short_setups[0].target
-    assert selected_target != pytest.approx(70.0, abs=0.01), (
-        "Bug: scanner picked the weak distant level (70) instead of the "
-        "strong nearby level (95)"
-    )
-    assert selected_target == pytest.approx(95.0, abs=0.01)
-
-
-# ---------------------------------------------------------------------------
-# Hypothesis property test: selection is NOT always the most distant level
-# ---------------------------------------------------------------------------
-
-@st.composite
-def strong_near_weak_far_pair(draw: st.DrawFn) -> dict:
-    """Generate a (strong-near, weak-far) resistance pair above entry=100.
-
-    Guarantees:
-    - near_price < far_price  (both above entry)
-    - near_strength >> far_strength
-    - Both meet the R:R threshold of 1.5 given typical ATR ≈ 2 → risk ≈ 3
-    """
-    # Near level: 5–15 above entry (R:R ≈ 1.7–5.0 with risk≈3)
-    near_dist = draw(st.floats(min_value=5.0, max_value=15.0))
-    near_strength = draw(st.integers(min_value=70, max_value=100))
-
-    # Far level: 25–60 above entry (R:R ≈ 8.3–20 with risk≈3)
-    far_dist = draw(st.floats(min_value=25.0, max_value=60.0))
-    far_strength = draw(st.integers(min_value=1, max_value=15))
-
-    return {
-        "near_price": 100.0 + near_dist,
-        "near_strength": near_strength,
-        "far_price": 100.0 + far_dist,
-        "far_strength": far_strength,
-    }
-
-
-@pytest.mark.asyncio
-@given(pair=strong_near_weak_far_pair())
-@settings(
-    max_examples=15,
-    deadline=None,
-    suppress_health_check=[HealthCheck.function_scoped_fixture],
-)
-async def test_property_scanner_does_not_always_pick_most_distant(
-    pair: dict,
-    scan_session: AsyncSession,
-):
-    """**Validates: Requirements 1.1, 1.3, 1.4, 2.1, 2.3, 2.4**
-
-    Property: when a strong nearby resistance exists alongside a weak distant
-    resistance, the scanner does NOT always select the most distant level.
-
-    On unfixed code this would fail for every example because max-R:R always
-    picks the farthest level.
-    """
-    from tests.conftest import _test_engine, _test_session_factory
-
-    # Each hypothesis example needs a fresh DB state
-    async with _test_engine.begin() as conn:
-        from app.database import Base
-        await conn.run_sync(Base.metadata.drop_all)
-        await conn.run_sync(Base.metadata.create_all)
-
-    async with _test_session_factory() as session:
-        ticker = Ticker(symbol="PROP")
-        session.add(ticker)
-        await session.flush()
-
-        bars = _make_ohlcv_bars(ticker.id, num_bars=20, base_close=100.0)
-        session.add_all(bars)
-
-        near_level = SRLevel(
-            ticker_id=ticker.id,
-            price_level=pair["near_price"],
-            type="resistance",
-            strength=pair["near_strength"],
-            detection_method="volume_profile",
-        )
-        far_level = SRLevel(
-            ticker_id=ticker.id,
-            price_level=pair["far_price"],
-            type="resistance",
-            strength=pair["far_strength"],
-            detection_method="volume_profile",
-        )
-        session.add_all([near_level, far_level])
-        await session.commit()
-
-        setups = await scan_ticker(
-            session,
-            "PROP",
-            rr_threshold=1.5,
-            gate_levels_override=[near_level, far_level],
-        )
-
-        long_setups = [s for s in setups if s.direction == "long"]
-        assert len(long_setups) == 1, "Expected exactly one long setup"
-
-        selected_target = long_setups[0].target
-        most_distant = round(pair["far_price"], 4)
-
-        # The fixed scanner should prefer the strong nearby level, not the
-        # most distant weak one.
-        assert selected_target != pytest.approx(most_distant, abs=0.01), (
-            f"Bug: scanner picked the most distant level ({most_distant}) "
-            f"with strength={pair['far_strength']} over the nearby level "
-            f"({round(pair['near_price'], 4)}) with strength={pair['near_strength']}"
-        )

tests/unit/test_rr_scanner_fix_check.py

@@ -1,375 +0,0 @@
-"""Fix-checking tests for R:R scanner probability-based primary selection.
-
-Verify that after enhance_trade_setup the headline target is the most likely
-worthwhile primary (R:R + probability floors), for both long and short setups.
-The pre-enhance quality loop only seeds a provisional target.
-
-**Validates: Requirements 2.1, 2.2, 2.3, 2.4**
-"""
-
-from __future__ import annotations
-
-from datetime import date, timedelta
-
-import pytest
-from hypothesis import given, settings, HealthCheck, strategies as st
-from sqlalchemy.ext.asyncio import AsyncSession
-
-from app.models.ohlcv import OHLCVRecord
-from app.models.sr_level import SRLevel
-from app.models.ticker import Ticker
-from app.services.rr_scanner_service import scan_ticker
-
-
-def _assert_primary_is_most_likely_worthwhile(setup) -> None:
-    """Headline = starred primary = max(probability, rr) among floor-clearing targets."""
-    targets = setup.targets
-    assert targets, "expected generated targets"
-    primaries = [t for t in targets if t.get("is_primary")]
-    assert len(primaries) == 1, "exactly one primary target expected"
-    primary = primaries[0]
-    assert setup.target == pytest.approx(primary["price"], abs=0.01)
-
-    # Mirrors recommendation_service._select_primary_target floors.
-    worthwhile = [
-        t for t in targets
-        if float(t["rr_ratio"]) >= 1.5 and float(t["probability"]) >= 20.0
-    ]
-    pool = worthwhile or targets
-    best = max(pool, key=lambda t: (t["probability"], t["rr_ratio"]))
-    assert primary["price"] == pytest.approx(best["price"], abs=0.01)
-
-
-# ---------------------------------------------------------------------------
-# Session fixture (plain session, not wrapped in begin())
-# ---------------------------------------------------------------------------
-
-@pytest.fixture
-async def scan_session() -> AsyncSession:
-    """Provide a DB session compatible with scan_ticker (which commits)."""
-    from tests.conftest import _test_session_factory
-
-    async with _test_session_factory() as session:
-        yield session
-
-
-# ---------------------------------------------------------------------------
-# Helpers
-# ---------------------------------------------------------------------------
-
-def _make_ohlcv_bars(
-    ticker_id: int,
-    num_bars: int = 20,
-    base_close: float = 100.0,
-) -> list[OHLCVRecord]:
-    """Generate OHLCV bars closing around base_close with ATR ≈ 2.0."""
-    bars: list[OHLCVRecord] = []
-    start = date(2024, 1, 1)
-    for i in range(num_bars):
-        close = base_close + (i % 3 - 1) * 0.5  # oscillate ±0.5
-        bars.append(OHLCVRecord(
-            ticker_id=ticker_id,
-            date=start + timedelta(days=i),
-            open=close - 0.3,
-            high=close + 1.0,
-            low=close - 1.0,
-            close=close,
-            volume=100_000,
-        ))
-    return bars
-
-
-# ---------------------------------------------------------------------------
-# Hypothesis strategy: multiple resistance levels above entry for longs
-# ---------------------------------------------------------------------------
-
-@st.composite
-def long_candidate_levels(draw: st.DrawFn) -> list[dict]:
-    """Generate 2-5 resistance levels above entry_price=100.
-
-    All levels meet the R:R threshold of 1.5 given ATR≈2, risk≈3,
-    so min reward=4.5, min target=104.5.
-    """
-    num_levels = draw(st.integers(min_value=2, max_value=5))
-    levels = []
-    for _ in range(num_levels):
-        # Distance from entry: 5 to 50 (all above 4.5 threshold)
-        distance = draw(st.floats(min_value=5.0, max_value=50.0))
-        strength = draw(st.integers(min_value=0, max_value=100))
-        levels.append({
-            "price": 100.0 + distance,
-            "strength": strength,
-        })
-    return levels
-
-
-@st.composite
-def short_candidate_levels(draw: st.DrawFn) -> list[dict]:
-    """Generate 2-5 support levels below entry_price=100.
-
-    All levels meet the R:R threshold of 1.5 given ATR≈2, risk≈3,
-    so min reward=4.5, max target=95.5.
-    """
-    num_levels = draw(st.integers(min_value=2, max_value=5))
-    levels = []
-    for _ in range(num_levels):
-        # Distance below entry: 5 to 50 (all above 4.5 threshold)
-        distance = draw(st.floats(min_value=5.0, max_value=50.0))
-        strength = draw(st.integers(min_value=0, max_value=100))
-        levels.append({
-            "price": 100.0 - distance,
-            "strength": strength,
-        })
-    return levels
-
-
-# ---------------------------------------------------------------------------
-# Property test: long setup selects probability-based primary
-# ---------------------------------------------------------------------------
-
-@pytest.mark.asyncio
-@given(levels=long_candidate_levels())
-@settings(
-    max_examples=20,
-    deadline=None,
-    suppress_health_check=[HealthCheck.function_scoped_fixture],
-)
-async def test_property_long_selects_probability_primary(
-    levels: list[dict],
-    scan_session: AsyncSession,
-):
-    """**Validates: Requirements 2.1, 2.3, 2.4**
-
-    Property: when multiple resistance levels meet the R:R threshold,
-    the headline after enhance is the probability-based primary.
-    """
-    from tests.conftest import _test_engine, _test_session_factory
-    from app.database import Base
-
-    # Fresh DB state per hypothesis example
-    async with _test_engine.begin() as conn:
-        await conn.run_sync(Base.metadata.drop_all)
-        await conn.run_sync(Base.metadata.create_all)
-
-    async with _test_session_factory() as session:
-        ticker = Ticker(symbol="FIXL")
-        session.add(ticker)
-        await session.flush()
-
-        bars = _make_ohlcv_bars(ticker.id, num_bars=20, base_close=100.0)
-        session.add_all(bars)
-
-        sr_levels = []
-        for lv in levels:
-            sr_levels.append(SRLevel(
-                ticker_id=ticker.id,
-                price_level=lv["price"],
-                type="resistance",
-                strength=lv["strength"],
-                detection_method="volume_profile",
-            ))
-        session.add_all(sr_levels)
-        await session.commit()
-
-        setups = await scan_ticker(
-            session,
-            "FIXL",
-            rr_threshold=1.5,
-            gate_levels_override=sr_levels,
-        )
-
-        long_setups = [s for s in setups if s.direction == "long"]
-        assert len(long_setups) == 1, "Expected exactly one long setup"
-
-        _assert_primary_is_most_likely_worthwhile(long_setups[0])
-
-
-# ---------------------------------------------------------------------------
-# Property test: short setup selects probability-based primary
-# ---------------------------------------------------------------------------
-
-@pytest.mark.asyncio
-@given(levels=short_candidate_levels())
-@settings(
-    max_examples=20,
-    deadline=None,
-    suppress_health_check=[HealthCheck.function_scoped_fixture],
-)
-async def test_property_short_selects_probability_primary(
-    levels: list[dict],
-    scan_session: AsyncSession,
-):
-    """**Validates: Requirements 2.2, 2.3, 2.4**
-
-    Property: when multiple support levels meet the R:R threshold,
-    the headline after enhance is the probability-based primary.
-    """
-    from tests.conftest import _test_engine, _test_session_factory
-    from app.database import Base
-
-    # Fresh DB state per hypothesis example
-    async with _test_engine.begin() as conn:
-        await conn.run_sync(Base.metadata.drop_all)
-        await conn.run_sync(Base.metadata.create_all)
-
-    async with _test_session_factory() as session:
-        ticker = Ticker(symbol="FIXS")
-        session.add(ticker)
-        await session.flush()
-
-        bars = _make_ohlcv_bars(ticker.id, num_bars=20, base_close=100.0)
-        session.add_all(bars)
-
-        sr_levels = []
-        for lv in levels:
-            sr_levels.append(SRLevel(
-                ticker_id=ticker.id,
-                price_level=lv["price"],
-                type="support",
-                strength=lv["strength"],
-                detection_method="pivot_point",
-            ))
-        session.add_all(sr_levels)
-        await session.commit()
-
-        setups = await scan_ticker(
-            session,
-            "FIXS",
-            rr_threshold=1.5,
-            gate_levels_override=sr_levels,
-        )
-
-        short_setups = [s for s in setups if s.direction == "short"]
-        assert len(short_setups) == 1, "Expected exactly one short setup"
-
-        _assert_primary_is_most_likely_worthwhile(short_setups[0])
-
-
-# ---------------------------------------------------------------------------
-# Deterministic test: 3 levels with known quality scores (long)
-# ---------------------------------------------------------------------------
-
-@pytest.mark.asyncio
-async def test_deterministic_long_three_levels(scan_session: AsyncSession):
-    """**Validates: Requirements 2.1, 2.3, 2.4**
-
-    Concrete example with 3 resistance levels of known quality scores.
-    Entry=100, ATR≈2, risk≈3.
-
-    Level A: price=105, strength=90 → rr=5/3≈1.67, dist=5
-      quality = 0.35*(1.67/10) + 0.35*(90/100) + 0.30*(1-5/100)
-             = 0.35*0.167 + 0.35*0.9 + 0.30*0.95
-             = 0.0585 + 0.315 + 0.285 = 0.6585
-
-    Level B: price=112, strength=50 → rr=12/3=4.0, dist=12
-      quality = 0.35*(4/10) + 0.35*(50/100) + 0.30*(1-12/100)
-             = 0.35*0.4 + 0.35*0.5 + 0.30*0.88
-             = 0.14 + 0.175 + 0.264 = 0.579
-
-    Level C: price=130, strength=10 → rr=30/3=10.0, dist=30
-      quality = 0.35*(10/10) + 0.35*(10/100) + 0.30*(1-30/100)
-             = 0.35*1.0 + 0.35*0.1 + 0.30*0.7
-             = 0.35 + 0.035 + 0.21 = 0.595
-
-    Expected winner: Level A (quality=0.6585)
-    """
-    ticker = Ticker(symbol="DET3L")
-    scan_session.add(ticker)
-    await scan_session.flush()
-
-    bars = _make_ohlcv_bars(ticker.id, num_bars=20, base_close=100.0)
-    scan_session.add_all(bars)
-
-    level_a = SRLevel(
-        ticker_id=ticker.id, price_level=105.0, type="resistance",
-        strength=90, detection_method="volume_profile",
-    )
-    level_b = SRLevel(
-        ticker_id=ticker.id, price_level=112.0, type="resistance",
-        strength=50, detection_method="volume_profile",
-    )
-    level_c = SRLevel(
-        ticker_id=ticker.id, price_level=130.0, type="resistance",
-        strength=10, detection_method="volume_profile",
-    )
-    scan_session.add_all([level_a, level_b, level_c])
-    await scan_session.flush()
-
-    setups = await scan_ticker(
-        scan_session,
-        "DET3L",
-        rr_threshold=1.5,
-        gate_levels_override=[level_a, level_b, level_c],
-    )
-
-    long_setups = [s for s in setups if s.direction == "long"]
-    assert len(long_setups) == 1, "Expected exactly one long setup"
-
-    _assert_primary_is_most_likely_worthwhile(long_setups[0])
-    # Near/strong level A wins on reach-probability over far lottery C.
-    assert long_setups[0].target == pytest.approx(105.0, abs=0.01), (
-        f"Expected primary=105.0 (near, high reach-prob), got {long_setups[0].target}"
-    )
-
-
-# ---------------------------------------------------------------------------
-# Deterministic test: 3 levels with known quality scores (short)
-# ---------------------------------------------------------------------------
-
-@pytest.mark.asyncio
-async def test_deterministic_short_three_levels(scan_session: AsyncSession):
-    """**Validates: Requirements 2.2, 2.3, 2.4**
-
-    Concrete example with 3 support levels of known quality scores.
-    Entry=100, ATR≈2, risk≈3.
-
-    Level A: price=95, strength=85 → rr=5/3≈1.67, dist=5
-      quality = 0.35*(1.67/10) + 0.35*(85/100) + 0.30*(1-5/100)
-             = 0.0585 + 0.2975 + 0.285 = 0.641
-
-    Level B: price=88, strength=45 → rr=12/3=4.0, dist=12
-      quality = 0.35*(4/10) + 0.35*(45/100) + 0.30*(1-12/100)
-             = 0.14 + 0.1575 + 0.264 = 0.5615
-
-    Level C: price=70, strength=8 → rr=30/3=10.0, dist=30
-      quality = 0.35*(10/10) + 0.35*(8/100) + 0.30*(1-30/100)
-             = 0.35 + 0.028 + 0.21 = 0.588
-
-    Expected winner: Level A (quality=0.641)
-    """
-    ticker = Ticker(symbol="DET3S")
-    scan_session.add(ticker)
-    await scan_session.flush()
-
-    bars = _make_ohlcv_bars(ticker.id, num_bars=20, base_close=100.0)
-    scan_session.add_all(bars)
-
-    level_a = SRLevel(
-        ticker_id=ticker.id, price_level=95.0, type="support",
-        strength=85, detection_method="pivot_point",
-    )
-    level_b = SRLevel(
-        ticker_id=ticker.id, price_level=88.0, type="support",
-        strength=45, detection_method="pivot_point",
-    )
-    level_c = SRLevel(
-        ticker_id=ticker.id, price_level=70.0, type="support",
-        strength=8, detection_method="pivot_point",
-    )
-    scan_session.add_all([level_a, level_b, level_c])
-    await scan_session.flush()
-
-    setups = await scan_ticker(
-        scan_session,
-        "DET3S",
-        rr_threshold=1.5,
-        gate_levels_override=[level_a, level_b, level_c],
-    )
-
-    short_setups = [s for s in setups if s.direction == "short"]
-    assert len(short_setups) == 1, "Expected exactly one short setup"
-
-    _assert_primary_is_most_likely_worthwhile(short_setups[0])
-    assert short_setups[0].target == pytest.approx(95.0, abs=0.01), (
-        f"Expected primary=95.0 (near, high reach-prob), got {short_setups[0].target}"
-    )

tests/unit/test_rr_scanner_preservation.py

@@ -23,6 +23,8 @@ from sqlalchemy.ext.asyncio import AsyncSession
 from app.models.ohlcv import OHLCVRecord
 from app.models.paper_trade import PaperTrade
 from app.models.signal_context_snapshot import SignalContextSnapshot
+from app.models.sec_filing_gap import SecFilingGap
+from app.models.settings import SystemSetting
 from app.models.sr_level import SRLevel
 from app.models.ticker import Ticker
 from app.models.trade_setup import TradeSetup
@@ -513,6 +515,45 @@ async def test_get_trade_setups_excludes_stale_rows(db_session: AsyncSession):
     assert stale_rows == []
 
 
+@pytest.mark.asyncio
+async def test_get_trade_setups_hides_active_sec_filing_gap(
+    db_session: AsyncSession,
+):
+    now = datetime.now(timezone.utc)
+    ticker = Ticker(symbol="SECWAIT", cik="0000000042")
+    db_session.add(ticker)
+    await db_session.flush()
+    db_session.add_all([
+        SystemSetting(
+            key="fundamental_data_sec_dolt_cutover_enabled",
+            value="true",
+        ),
+        SecFilingGap(
+            cik=ticker.cik,
+            accession="0000000042-26-000001",
+            form="10-Q",
+            index_date=date.today(),
+            reason="not_in_companyfacts",
+            first_seen_at=now,
+            last_attempted_at=now,
+        ),
+        TradeSetup(
+            ticker_id=ticker.id,
+            direction="long",
+            entry_price=100.0,
+            stop_loss=97.0,
+            target=109.0,
+            rr_ratio=3.0,
+            composite_score=70.0,
+            confidence_score=80.0,
+            detected_at=now,
+        ),
+    ])
+    await db_session.flush()
+
+    assert await get_trade_setups(db_session, symbol="SECWAIT") == []
+
+
 @pytest.mark.asyncio
 async def test_get_trade_setups_can_exclude_tickers_with_open_paper_trades(
     db_session: AsyncSession,

tests/unit/test_rr_scanner_scan_all.py

@@ -108,3 +108,61 @@ async def test_scan_error_does_not_stop_later_tickers(session, monkeypatch):
     await rr_scanner_service.scan_all_tickers(session)
 
     assert scanned == ["AAA", "BBB"]
+
+
+async def test_scan_skips_ticker_with_incomplete_sec_fundamentals(
+    session, monkeypatch
+):
+    ticker = Ticker(symbol="BLOCKED", cik="0000000001")
+    session.add(ticker)
+    await session.commit()
+
+    async def _blocked(db):
+        return {ticker.id}
+
+    async def _unexpected_scan(*args, **kwargs):
+        raise AssertionError("fundamentals-incomplete ticker was scanned")
+
+    monkeypatch.setattr(
+        rr_scanner_service.fundamentals_quality_service,
+        "blocked_ticker_ids",
+        _blocked,
+    )
+    monkeypatch.setattr(rr_scanner_service, "scan_ticker", _unexpected_scan)
+
+    assert await rr_scanner_service.scan_all_tickers(session) == []
+
+
+async def test_scan_quality_failure_blocks_closed_and_emits_event(
+    session, monkeypatch
+):
+    session.add(Ticker(symbol="BLOCKED"))
+    await session.commit()
+
+    async def _boom(db):
+        raise ValueError("bad quality metadata")
+
+    async def _unexpected_scan(*args, **kwargs):
+        raise AssertionError("ticker was scanned without a quality decision")
+
+    events: list[dict] = []
+
+    async def _capture_event(**kwargs):
+        events.append(kwargs)
+
+    monkeypatch.setattr(
+        rr_scanner_service.fundamentals_quality_service,
+        "blocked_ticker_ids",
+        _boom,
+    )
+    monkeypatch.setattr(rr_scanner_service, "scan_ticker", _unexpected_scan)
+    monkeypatch.setattr(
+        rr_scanner_service.system_event_service,
+        "log_event_standalone",
+        _capture_event,
+    )
+
+    assert await rr_scanner_service.scan_all_tickers(session) == []
+    assert [event["code"] for event in events] == [
+        "fundamentals_quality_unavailable"
+    ]

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

@@ -15,6 +15,7 @@ from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_asyn
 from app.database import Base
 import app.models  # noqa: F401
 from app.models.fundamental_snapshot import FundamentalSnapshot
+from app.models.sec_filing_gap import SecFilingGap
 from app.models.system_event import SystemEvent
 from app.models.ticker import Ticker
 from app.services.data_import import (
@@ -374,7 +375,7 @@ async def test_recovers_facts_misfiled_under_coregistrant(engine):
     # Stamped to the issuer that filed, NOT the co-registrant whose file it came from.
     assert q2.cik == "0000320193"
     assert q2.revenue == 254940 and q2.shares_outstanding == 14687
-    assert "coregistrant_recovery" in codes
+    assert "coregistrant_recovery" not in codes
 
 
 async def test_coregistrant_recovery_rejects_discontinuous_share_count(engine):
@@ -425,11 +426,333 @@ async def test_unresolved_filing_stops_blocking_after_retry_window(engine):
     assert run.source_max_date == date(2026, 5, 2)  # and the index advances
     summary = json.loads(run.validation_json or "{}")
     assert summary["missing_xbrl_count"] == 1 and summary["missing_xbrl_blocking"] == 0
+    assert await _count(factory, SecFilingGap) == 1
 
     async with factory() as s:
         events = (await s.execute(select(SystemEvent))).scalars().all()
     unresolved = [e for e in events if e.code == "unresolved_filing"]
     assert len(unresolved) == 1 and "GHOST" in unresolved[0].message
+    assert "automatic SEC retry" in unresolved[0].message
+
+    # The next scheduled run retries even though the SEC daily-index revision
+    # has not changed. Company Facts is a separate SEC product and may catch up
+    # independently, so the generic revision no-op must not suppress this work.
+    still_missing_run = await run_import(
+        _importer(incr, today=date(2026, 5, 11)),
+        engine=engine,
+    )
+    assert still_missing_run.status == STATUS_PROMOTED
+    assert still_missing_run.revision is None
+    assert await _count(factory, SecFilingGap) == 1
+
+    # A later normal scheduled import retries only the queued issuer. Once SEC
+    # publishes the accession in Company Facts, it is inserted and unblocked
+    # without a full-universe reparse or operator action.
+    cf_ghost = _rev("2025-09-28", "2026-03-28", 254940, 2026, "Q2", "GHOST")
+    sh_ghost = _shares("2026-04-17", 14687, "GHOST", 2026, "Q2")
+    healed = FakeSecClient(
+        tickers={"AAPL": 320193},
+        companyfacts={
+            320193: _companyfacts(
+                [CF_K, CF_Q1, cf_ghost],
+                [SH_K, SH_Q1, sh_ghost],
+            )
+        },
+        submissions={
+            320193: _submissions(SUB_FILINGS + [
+                _filing(
+                    "GHOST",
+                    "10-Q",
+                    "2026-03-28",
+                    "2026-05-01",
+                    "2026-05-01T10:01:00.000Z",
+                )
+            ])
+        },
+        latest_index=date(2026, 5, 2),
+    )
+    healed_run = await run_import(
+        _importer(healed, today=date(2026, 5, 12)),
+        engine=engine,
+    )
+
+    assert healed_run.status == STATUS_PROMOTED
+    assert await _count(factory, FundamentalSnapshot) == 3
+    assert await _count(factory, SecFilingGap) == 0
+
+
+async def test_queued_gap_without_index_date_retries_without_wedging_and_escalates_once(
+    engine,
+):
+    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)
+
+    old = datetime(2026, 4, 1, tzinfo=timezone.utc)
+    async with factory() as db:
+        db.add(SecFilingGap(
+            cik="0000320193",
+            accession="DATELESS",
+            form="10-Q",
+            index_date=None,
+            reason="not_in_companyfacts",
+            first_seen_at=old,
+            last_attempted_at=old,
+        ))
+        await db.commit()
+
+    missing = FakeSecClient(
+        tickers={"AAPL": 320193},
+        companyfacts={320193: _companyfacts([CF_K, CF_Q1], [SH_K, SH_Q1])},
+        submissions={320193: _submissions(SUB_FILINGS + [
+            _filing(
+                "DATELESS",
+                "10-Q",
+                "2026-03-28",
+                "2026-05-01",
+                "2026-05-01T10:01:00.000Z",
+            )
+        ])},
+        latest_index=date(2026, 1, 31),
+    )
+    first = await run_import(
+        _importer(missing, today=date(2026, 5, 20)), engine=engine
+    )
+    second = await run_import(
+        _importer(missing, today=date(2026, 5, 21)), engine=engine
+    )
+
+    assert first.status == STATUS_PROMOTED
+    assert second.status == STATUS_PROMOTED
+    async with factory() as db:
+        gap = (await db.execute(select(SecFilingGap))).scalar_one()
+        events = (
+            await db.execute(
+                select(SystemEvent).where(SystemEvent.code == "filing_gap_aged")
+            )
+        ).scalars().all()
+    assert gap.escalated_at is not None
+    assert len(events) == 1
+
+
+async def test_queued_filing_reclassified_non_xbrl_is_removed(engine):
+    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)
+    now = datetime.now(timezone.utc)
+    async with factory() as db:
+        db.add(SecFilingGap(
+            cik="0000320193",
+            accession="NONX",
+            form="10-Q/A",
+            index_date=date(2026, 5, 1),
+            reason="not_in_companyfacts",
+            first_seen_at=now,
+            last_attempted_at=now,
+        ))
+        await db.commit()
+
+    client = FakeSecClient(
+        tickers={"AAPL": 320193},
+        companyfacts={320193: _companyfacts([CF_K, CF_Q1], [SH_K, SH_Q1])},
+        submissions={320193: _submissions(SUB_FILINGS + [
+            _filing(
+                "NONX",
+                "10-Q/A",
+                "2026-03-28",
+                "2026-05-01",
+                "2026-05-01T10:01:00.000Z",
+                is_xbrl=False,
+            )
+        ])},
+        latest_index=date(2026, 1, 31),
+    )
+    run = await run_import(_importer(client, today=date(2026, 5, 20)), engine=engine)
+
+    assert run.status == STATUS_PROMOTED
+    assert await _count(factory, SecFilingGap) == 0
+
+
+async def test_queued_parser_skip_stays_blocked_with_actionable_reason(
+    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)
+    now = datetime.now(timezone.utc)
+    async with factory() as db:
+        db.add(SecFilingGap(
+            cik="0000320193",
+            accession="BADPARSE",
+            form="10-Q",
+            index_date=date(2026, 5, 1),
+            reason="not_in_companyfacts",
+            first_seen_at=now,
+            last_attempted_at=now,
+        ))
+        await db.commit()
+
+    bad_fact = _rev(
+        "2025-09-28", "2026-03-28", 254940, 2026, "Q2", "BADPARSE"
+    )
+    bad_share = _shares("2026-04-17", 14687, "BADPARSE", 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(
+                "BADPARSE",
+                "10-Q",
+                "2026-03-28",
+                "2026-05-01",
+                "2026-05-01T10:01:00.000Z",
+            )
+        ])},
+        latest_index=date(2026, 1, 31),
+    )
+
+    def skip_parse(*args, **kwargs):
+        return ParseResult(skipped_filings=[{
+            "accession": "BADPARSE",
+            "reason": "unparseable",
+        }])
+
+    monkeypatch.setattr("app.services.sec_facts_parser.parse_snapshots", skip_parse)
+    run = await run_import(_importer(client, today=date(2026, 5, 20)), engine=engine)
+
+    assert run.status == STATUS_PROMOTED
+    async with factory() as db:
+        gap = (await db.execute(select(SecFilingGap))).scalar_one()
+    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)
+    staged = StagedFundamentals(
+        resolved=ResolvedUniverse(),
+        missing_xbrl=[
+            {
+                "cik": f"{i:010d}",
+                "accession": f"MISS-{i}",
+                "form": "10-Q",
+                "index_date": date(2026, 5, 1),
+                "age_days": 19,
+                "reason": "not_in_companyfacts",
+            }
+            for i in range(60)
+        ],
+        no_xbrl_filings=[
+            {"cik": f"{i + 100:010d}", "name": f"New {i}"}
+            for i in range(60)
+        ],
+        recovered=[
+            {"cik": f"{i:010d}", "accession": f"REC-{i}", "source_cik": "1"}
+            for i in range(60)
+        ],
+    )
+
+    result = await importer.validate(None, staged)
+
+    assert len(result.summary["missing_xbrl"]) == 50
+    assert len(result.summary["no_xbrl_filings"]) == 50
+    assert len(result.summary["no_xbrl_ciks"]) == 60
+    assert len(result.summary["recovered_from_coregistrant"]) == 50
+    assert len(result.summary["setup_blocked_ciks"]) == 120
 
 
 async def test_non_xbrl_amendment_skipped_not_failed(engine):