fix(tickers): honor the effective date instead of retiring on the mark

active_only tested delisted_on IS NULL, so a symbol dropped out of signals the
moment a Form 25 was detected — ten days before Rule 12d2-2 makes the removal
effective, while it was demonstrably still trading. A manual future-dated mark
behaved the same way. It now compares against the database's own date, so a
pending delisting stays live until the day it takes effect.

That exposes a second problem the fix would otherwise create. Trading typically
stops before the ten-day delay expires, so across that window the symbol is
correctly active yet produces no bars — and confirm_delisting returned None for
an already-marked row, which would have fired the staleness warning daily for
ten days, the exact noise this flow exists to remove. It now reports the known
effective date on every path where the delisting is established, so the caller
warns only about gaps that are still unexplained.

CURRENT_DATE renders identically on postgres and sqlite, and the OR is
parenthesized when callers chain further where clauses.

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

alembic/versions/032_ticker_delisting.py

@@ -0,0 +1,49 @@
+"""Record delisting on tickers instead of deleting them
+
+Revision ID: 032
+Revises: 031
+Create Date: 2026-08-11 00:00:00.000000
+
+Until now the only way to retire a symbol was ``delete_ticker`` (or
+``bootstrap_universe(prune_missing=True)``), both of which cascade through
+OHLCV, setups and scores. That destroys exactly the history four research
+documents already apologise for: today's tracked universe projected backward
+is survivorship-biased, and hard-deleting every delisted name is what causes
+it. Keeping the rows preserves the option to fix that later — it does not fix
+it by itself, which needs the replay to model a delisting as an exit event.
+
+``delisted_on`` is the effective date (from SEC Form 25/25-NSE/15 where we can
+confirm it, else the day it was marked); ``delisted_reason`` is a short code
+for how we learned. NULL in both means actively traded — the live signal path
+filters on that, while list and admin views keep showing the row so the
+delisting is visible rather than silently absent.
+
+Nullable and reversible by design: clearing ``delisted_on`` un-retires a
+symbol, which is what makes automatic marking safe where a delete would not be.
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+
+revision: str = "032"
+down_revision: Union[str, None] = "031"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.add_column("tickers", sa.Column("delisted_on", sa.Date(), nullable=True))
+    op.add_column(
+        "tickers", sa.Column("delisted_reason", sa.String(length=32), nullable=True)
+    )
+    # The live path filters "actively traded" on every universe scan; the index
+    # keeps that predicate cheap as delisted rows accumulate.
+    op.create_index("ix_tickers_delisted_on", "tickers", ["delisted_on"])
+
+
+def downgrade() -> None:
+    op.drop_index("ix_tickers_delisted_on", table_name="tickers")
+    op.drop_column("tickers", "delisted_reason")
+    op.drop_column("tickers", "delisted_on")

app/models/ticker.py

@@ -1,6 +1,6 @@
-from datetime import datetime
+from datetime import date, datetime
 
-from sqlalchemy import String, DateTime
+from sqlalchemy import Date, String, DateTime
 from sqlalchemy.orm import Mapped, mapped_column, relationship
 
 from app.database import Base
@@ -21,6 +21,13 @@ class Ticker(Base):
     cik: Mapped[str | None] = mapped_column(String(10), nullable=True)
     sic: Mapped[str | None] = mapped_column(String(4), nullable=True)
     sic_description: Mapped[str | None] = mapped_column(String(160), nullable=True)
+    # Delisting is recorded, never deleted: the rows carry the price history that
+    # makes a backtest less survivorship-biased, and a delete cascades it away.
+    # NULL == actively traded. The live signal path filters on this (see
+    # ticker_service.active_only); list/admin views keep the row and show it.
+    delisted_on: Mapped[date | None] = mapped_column(Date, nullable=True, index=True)
+    # How we learned: "form_25" (SEC confirmed), "manual" (operator).
+    delisted_reason: Mapped[str | None] = mapped_column(String(32), nullable=True)
     created_at: Mapped[datetime] = mapped_column(
         DateTime(timezone=True), default=datetime.utcnow, nullable=False
     )

app/routers/tickers.py

@@ -6,7 +6,7 @@ from sqlalchemy.ext.asyncio import AsyncSession
 from app.dependencies import get_db, require_access
 from app.models.user import User
 from app.schemas.common import APIEnvelope
-from app.schemas.ticker import TickerCreate, TickerResponse
+from app.schemas.ticker import TickerCreate, TickerDelistingUpdate, TickerResponse
 from app.services import ticker_service
 
 router = APIRouter(tags=["tickers"])
@@ -51,3 +51,35 @@ async def delete_ticker(
     """Delete a ticker and all associated data."""
     await ticker_service.delete_ticker(db, symbol)
     return APIEnvelope(status="success", data=None)
+
+
+@router.post("/tickers/{symbol}/delisting", response_model=APIEnvelope)
+async def mark_ticker_delisted(
+    symbol: str,
+    body: TickerDelistingUpdate,
+    _user: User = Depends(require_access),
+    db: AsyncSession = Depends(get_db),
+):
+    """Retire a symbol: excluded from signals, price history kept.
+
+    The non-destructive alternative to DELETE, which cascades the history away.
+    """
+    changed = await ticker_service.mark_delisted(
+        db, symbol, delisted_on=body.delisted_on, reason=ticker_service.REASON_MANUAL
+    )
+    return APIEnvelope(status="success", data={"changed": changed})
+
+
+@router.delete("/tickers/{symbol}/delisting", response_model=APIEnvelope)
+async def clear_ticker_delisting(
+    symbol: str,
+    _user: User = Depends(require_access),
+    db: AsyncSession = Depends(get_db),
+):
+    """Un-retire a symbol wrongly marked delisted.
+
+    Automatic marking is only defensible because this exists: a false positive
+    costs one row update rather than the price history a delete would take.
+    """
+    changed = await ticker_service.clear_delisted(db, symbol)
+    return APIEnvelope(status="success", data={"changed": changed})

app/scheduler.py

@@ -66,6 +66,7 @@ from app.services.event_study_service import run_and_store as run_event_study_an
 from app.services.outcome_service import evaluate_pending_setups
 from app.services.rr_scanner_service import scan_all_tickers
 from app.services.sentiment_provider_service import build_sentiment_provider
+from app.services import ticker_service
 from app.services.ticker_universe_service import bootstrap_universe
 
 logger = logging.getLogger(__name__)
@@ -396,8 +397,10 @@ async def _is_job_enabled(db: AsyncSession, job_name: str) -> bool:
 
 
 async def _get_all_tickers(db: AsyncSession) -> list[str]:
-    """Return all tracked ticker symbols sorted alphabetically."""
-    result = await db.execute(select(Ticker.symbol).order_by(Ticker.symbol))
+    """Return all actively-traded ticker symbols sorted alphabetically."""
+    result = await db.execute(
+        ticker_service.active_only(select(Ticker.symbol).order_by(Ticker.symbol))
+    )
     return list(result.scalars().all())
 
 
@@ -412,8 +415,10 @@ async def _get_ohlcv_priority_tickers(db: AsyncSession) -> list[str]:
     latest_date = func.max(OHLCVRecord.date)
     missing_first = case((latest_date.is_(None), 0), else_=1)
     result = await db.execute(
+        ticker_service.active_only(
             select(Ticker.symbol)
             .outerjoin(OHLCVRecord, OHLCVRecord.ticker_id == Ticker.id)
+        )
         .group_by(Ticker.id, Ticker.symbol)
         .order_by(missing_first.asc(), latest_date.asc(), Ticker.symbol.asc())
     )
@@ -662,6 +667,26 @@ async def collect_ohlcv(
                     _runtime_progress(job_name, processed=processed, total=total, current_ticker=symbol)
                     _log_event(logging.INFO, "ticker_collected", job=job_name, ticker=symbol, status=result.status, records=result.records_ingested)
                     if result.status == "stale":
+                        # "No new bars" cannot distinguish a delisting from a halt
+                        # or a rename, so ask SEC before warning again. A confirmed
+                        # delisting retires the symbol (keeping its history) and
+                        # ends the alert; anything unproven keeps warning.
+                        delisted_on = await ticker_service.confirm_delisting(
+                            db, symbol, last_bar=result.last_date
+                        )
+                        if delisted_on is not None:
+                            await _record_system_event(
+                                severity="info",
+                                source=job_name,
+                                code="ticker_delisted",
+                                message=(
+                                    f"{symbol} delisted on {delisted_on} (SEC Form 25/15). "
+                                    "Retired from signals; price history retained."
+                                ),
+                                symbol=symbol,
+                                dedup_key=f"ticker_delisted:{symbol}",
+                            )
+                        else:
                             await _record_system_event(
                                 severity="warning",
                                 source=job_name,

app/schemas/ticker.py

@@ -1,6 +1,6 @@
 """Ticker request/response schemas."""
 
-from datetime import datetime
+from datetime import date, datetime
 
 from pydantic import BaseModel, Field
 
@@ -14,5 +14,16 @@ class TickerResponse(BaseModel):
     symbol: str
     name: str | None = None
     created_at: datetime
+    # NULL == actively traded. Delisted symbols stay in the registry with their
+    # history and are excluded from signals — the date is what makes that
+    # visible instead of the row silently disappearing.
+    delisted_on: date | None = None
+    delisted_reason: str | None = None
 
     model_config = {"from_attributes": True}
+
+
+class TickerDelistingUpdate(BaseModel):
+    delisted_on: date = Field(
+        ..., description="Effective date the symbol stopped trading"
+    )

app/services/backtest_service.py

@@ -1460,6 +1460,21 @@ def _mp_context():
     return None
 
 
+async def _rollback_quietly(db: AsyncSession, context: str) -> None:
+    """Discard a failed unit of work so later statements on this session survive.
+
+    Every DB call in ``run_backtest`` is best-effort — one unreadable ticker must
+    not abort the whole replay. But swallowing the exception alone leaves asyncpg
+    in "current transaction is aborted": every later statement then fails the same
+    way until the first unguarded one (the report write) surfaces it as the job
+    error, long after the real cause. Same guard as ``price_service``.
+    """
+    try:
+        await db.rollback()
+    except Exception:
+        logger.exception("Session rollback after %s also failed", context)
+
+
 async def _fetch_columns(db: AsyncSession, symbol: str) -> tuple | None:
     """Read one ticker's OHLCV and detach it to primitive column arrays in the
     event loop (safe ORM access), ready to hand to a worker. None if no data."""
@@ -4037,9 +4052,12 @@ async def run_backtest(
     config = await get_recommendation_config(db)
     activation = await get_activation_config(db)
 
-    result = await db.execute(select(Ticker).order_by(Ticker.symbol))
-    tickers = list(result.scalars().all())
-    total = len(tickers)
+    # Plain strings, not Ticker instances: the rollbacks below expire any ORM
+    # objects held across them, and touching an expired attribute afterwards
+    # triggers sync lazy-loading, which raises on an AsyncSession.
+    result = await db.execute(select(Ticker.symbol).order_by(Ticker.symbol))
+    symbols = list(result.scalars().all())
+    total = len(symbols)
     rank_only_symbols = await _load_research_rank_only_symbols(db)
     if rank_only_symbols:
         logger.info(json.dumps({
@@ -4063,6 +4081,7 @@ async def run_backtest(
         )
     except Exception:
         logger.exception("Benchmark load for residual momentum failed")
+        await _rollback_quietly(db, "benchmark load")
 
     def _merge(result: tuple[list[dict], dict]) -> None:
         cands, series = result
@@ -4094,26 +4113,27 @@ async def run_backtest(
         done = 0
         with pool:
             for start in range(0, total, chunk):
-                batch = tickers[start : start + chunk]
+                batch = symbols[start : start + chunk]
                 futures = []
-                for ticker in batch:
+                for symbol in batch:
                     try:
-                        columns = await _fetch_columns(db, ticker.symbol)
+                        columns = await _fetch_columns(db, symbol)
                     except Exception:
-                        logger.exception("Backtest fetch failed for %s", ticker.symbol)
+                        logger.exception("Backtest fetch failed for %s", symbol)
+                        await _rollback_quietly(db, f"fetch for {symbol}")
                         continue
                     if columns is not None:
                         futures.append(loop.run_in_executor(
                             pool,
                             _replay_and_signals,
-                            ticker.symbol,
+                            symbol,
                             columns,
                             config,
                             activation,
                             benchmark_closes,
                             target_model,
                             cadence,
-                            ticker.symbol in rank_only_symbols,
+                            symbol in rank_only_symbols,
                         ))
                 for result in await asyncio.gather(*futures, return_exceptions=True):
                     if isinstance(result, Exception):
@@ -4126,25 +4146,26 @@ async def run_backtest(
     else:
         # Sequential fallback (Windows / 1 worker): run each replay in a worker
         # thread so the event loop — and the API server — stays responsive.
-        for index, ticker in enumerate(tickers):
+        for index, symbol in enumerate(symbols):
             if progress_cb is not None:
-                progress_cb(index, total, ticker.symbol)
+                progress_cb(index, total, symbol)
             try:
-                columns = await _fetch_columns(db, ticker.symbol)
+                columns = await _fetch_columns(db, symbol)
                 if columns is not None:
                     _merge(await asyncio.to_thread(
                         _replay_and_signals,
-                        ticker.symbol,
+                        symbol,
                         columns,
                         config,
                         activation,
                         benchmark_closes,
                         target_model,
                         cadence,
-                        ticker.symbol in rank_only_symbols,
+                        symbol in rank_only_symbols,
                     ))
             except Exception:
-                logger.exception("Backtest replay failed for %s", ticker.symbol)
+                logger.exception("Backtest replay failed for %s", symbol)
+                await _rollback_quietly(db, f"replay for {symbol}")
 
     if progress_cb is not None and total:
         progress_cb(total, total, "")
@@ -4209,6 +4230,7 @@ async def run_backtest(
                 )
             except Exception:
                 logger.exception("Benchmark load for the portfolio sim failed")
+                await _rollback_quietly(db, "portfolio-sim benchmark load")
 
             for policy in ("target", "hold"):
                 sim = _simulate_portfolio(
@@ -4229,6 +4251,7 @@ async def run_backtest(
                 live_exit_policy = await get_exit_policy(db)
             except Exception:
                 logger.exception("Live exit policy load failed; monitor uses defaults")
+                await _rollback_quietly(db, "exit policy load")
             portfolio_monitor_report = _portfolio_monitor(
                 candidates, price_columns, spy_closes, hold_horizon,
                 live_exit_policy=live_exit_policy,
@@ -4248,6 +4271,11 @@ async def run_backtest(
                 )
         except Exception:
             logger.exception("Portfolio simulation failed")
+            # Catches the price_columns fetch loop, which has no handler of its
+            # own. The inner handlers above may already have rolled back; a
+            # rollback on a clean session is a no-op, so this stays safe as the
+            # backstop for whichever DB call actually failed.
+            await _rollback_quietly(db, "portfolio simulation")
 
     report = {
         "generated_at": datetime.now(timezone.utc).isoformat(),

app/services/breadth_service.py

@@ -25,6 +25,7 @@ from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from app.models.ticker import Ticker
+from app.services import ticker_service
 from app.services.price_service import query_ohlcv
 
 logger = logging.getLogger(__name__)
@@ -112,7 +113,7 @@ def compute_divergence_series(
 async def _load_universe_closes(
     db: AsyncSession, symbols: list[str] | None = None
 ) -> dict[str, Series]:
-    stmt = select(Ticker).order_by(Ticker.symbol)
+    stmt = ticker_service.active_only(select(Ticker).order_by(Ticker.symbol))
     if symbols is not None:
         stmt = stmt.where(Ticker.symbol.in_(symbols))
     result = await db.execute(stmt)

app/services/dolt_earnings_importer.py

@@ -32,7 +32,7 @@ from app.config import settings
 from app.database import insert_for_session
 from app.models.earnings_event import EarningsEvent
 from app.models.ticker import Ticker
-from app.services import dolt_client, earnings_alignment
+from app.services import dolt_client, earnings_alignment, ticker_service
 from app.services.data_import import ValidationResult
 
 logger = logging.getLogger(__name__)
@@ -289,7 +289,11 @@ class DoltEarningsImporter:
     # -- helpers -----------------------------------------------------------
 
     async def _load_universe(self, db) -> dict[str, int]:
-        rows = (await db.execute(select(Ticker.id, Ticker.symbol))).all()
+        rows = (
+            await db.execute(
+                ticker_service.active_only(select(Ticker.id, Ticker.symbol))
+            )
+        ).all()
         return {
             earnings_alignment.normalise_symbol(symbol): tid
             for tid, symbol in rows

app/services/fundamentals_candidate_service.py

@@ -22,6 +22,7 @@ from app.models.fundamental_snapshot import FundamentalSnapshot
 from app.models.ohlcv import OHLCVRecord
 from app.models.ticker import Ticker
 from app.services import fundamentals_derivation as deriv
+from app.services import ticker_service
 
 
 @dataclass(frozen=True)
@@ -46,7 +47,11 @@ async def build_candidates(
     """Derive current cache candidates using only already-stored data."""
     today = today or datetime.now(ZoneInfo("America/New_York")).date()
     tickers = list(
-        (await db.execute(select(Ticker).order_by(Ticker.symbol))).scalars()
+        (
+            await db.execute(
+                ticker_service.active_only(select(Ticker).order_by(Ticker.symbol))
+            )
+        ).scalars()
     )
     if not tickers:
         return []

app/services/momentum_service.py

@@ -18,6 +18,7 @@ from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from app.models.ticker import Ticker
+from app.services import ticker_service
 from app.services.price_service import query_ohlcv
 
 logger = logging.getLogger(__name__)
@@ -169,7 +170,9 @@ async def compute_activation_ranks(db: AsyncSession) -> dict[str, dict[str, floa
     before scanning; the research backtest ranked each weekly setup-candidate
     cross-section, so this is the deliberate production approximation.
     """
-    result = await db.execute(select(Ticker).order_by(Ticker.symbol))
+    result = await db.execute(
+        ticker_service.active_only(select(Ticker).order_by(Ticker.symbol))
+    )
     tickers = list(result.scalars().all())
 
     benchmark_closes = await _load_activation_benchmark(db)

app/services/rr_scanner_service.py

@@ -31,7 +31,7 @@ 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
-from app.services import settings_store
+from app.services import settings_store, ticker_service
 from app.services.trade_policy import (
     MANUAL_BOOK,
     SHADOW_BOOK,
@@ -735,7 +735,11 @@ async def scan_all_tickers(
     # Plain ids/strings, not Ticker instances: the rollbacks below expire any
     # ORM objects held across them, and touching an expired attribute afterwards
     # triggers sync lazy-loading, which raises on an AsyncSession.
-    result = await db.execute(select(Ticker.id, Ticker.symbol).order_by(Ticker.symbol))
+    result = await db.execute(
+        ticker_service.active_only(
+            select(Ticker.id, Ticker.symbol).order_by(Ticker.symbol)
+        )
+    )
     ticker_rows = [(int(ticker_id), symbol) for ticker_id, symbol in result.all()]
     total = len(ticker_rows)
 

app/services/scoring_service.py

@@ -20,7 +20,7 @@ from app.database import insert_for_session
 from app.exceptions import NotFoundError, ValidationError
 from app.models.score import CompositeScore, DimensionScore
 from app.models.ticker import Ticker
-from app.services import settings_store
+from app.services import settings_store, ticker_service
 
 logger = logging.getLogger(__name__)
 
@@ -883,7 +883,11 @@ async def get_rankings(db: AsyncSession) -> dict:
     Returns dict suitable for RankingResponse.
     """
     weights = await _get_weights(db)
-    tickers = (await db.execute(select(Ticker).order_by(Ticker.symbol))).scalars().all()
+    tickers = (
+        await db.execute(
+            ticker_service.active_only(select(Ticker).order_by(Ticker.symbol))
+        )
+    ).scalars().all()
 
     async def _load_scores() -> tuple[dict[int, CompositeScore], dict[int, dict[str, DimensionScore]]]:
         comps = {
@@ -947,7 +951,7 @@ async def update_weights(
     await _save_weights(db, full_weights)
 
     # Recompute all composite scores
-    result = await db.execute(select(Ticker))
+    result = await db.execute(ticker_service.active_only(select(Ticker)))
     tickers = list(result.scalars().all())
 
     for ticker in tickers:

app/services/sec_client.py

@@ -45,6 +45,33 @@ _CA_VERIFY: str | bool = _CA if _CA and Path(_CA).exists() else True
 
 _FORMS_10 = frozenset({"10-K", "10-Q", "10-K/A", "10-Q/A"})
 
+# Notification of removal from listing. "25" is issuer-filed, "25-NSE" exchange-
+# filed. The Form 15 family is deliberately absent: it ends a *reporting*
+# obligation and does not mean the security stopped trading.
+_DELISTING_FORMS = frozenset({"25", "25-NSE"})
+
+# ``descriptionClassSecurity`` is free text ("Common Stock", "Class A Common
+# Stock, $0.01 par value", "6.25% Notes due 2030", "Warrants", "Depositary
+# Shares"). Only a common-equity class means the ticker itself stopped trading.
+_NON_COMMON_CLASS = re.compile(
+    r"\b(note|bond|debenture|preferred|warrant|right|unit|depositary|"
+    r"subordinated|debt|trust)s?\b",
+    re.IGNORECASE,
+)
+
+
+def _is_common_stock(description: str) -> bool:
+    """Does this Form 25 security class describe common equity?
+
+    Requires an explicit common-stock match AND no debt/preferred/warrant marker,
+    so "Depositary Shares each representing 1/1000th of Preferred" cannot pass on
+    the word "shares" alone. Unrecognised text is rejected — a symbol is retired
+    on this answer, so ambiguity must not read as yes.
+    """
+    if _NON_COMMON_CLASS.search(description):
+        return False
+    return re.search(r"\bcommon\s+(stock|share)", description, re.IGNORECASE) is not None
+
 
 class SecError(ProviderError):
     """SEC request failed (403, exhausted 429/5xx, timeout, transport, parse)."""
@@ -253,6 +280,91 @@ class SecClient:
             "filings": filings,
         }
 
+    async def delisting_filing(
+        self, cik: int | str, *, not_before: date | None = None
+    ) -> dict[str, Any] | None:
+        """Newest Form 25 removing this issuer's COMMON stock from listing.
+
+        Deliberately narrow, because the caller retires a symbol on the answer:
+
+        - **Form 25 only.** The Form 15 family terminates a reporting obligation
+          (often just a class falling under the holder threshold) and is no
+          evidence that trading stopped.
+        - **Class-checked.** Form 25 is filed per security class — an issuer
+          delisting its notes, preferred, warrants or an ADR class while the
+          common keeps trading files one too. The filing's own
+          ``descriptionClassSecurity`` is what separates those, so the primary
+          document is fetched and read rather than trusting the form type.
+        - **``not_before``** rejects a historical filing for some long-gone
+          class. Without it a 2019 Form 25 would retire a symbol whose bars
+          stopped in 2026, and stamp 2019 as the date.
+
+        Anything unreadable — no primary document (pre-2009 filings have none),
+        malformed XML, unrecognised class — returns ``None``. Fail closed: the
+        caller keeps warning instead of retiring on a guess.
+
+        Reads ``filings.recent`` directly; ``submissions()`` keeps only the
+        10-K/10-Q family, so Form 25 never survives its parser.
+        """
+        base = await self.get_json(f"{_DATA}/submissions/CIK{cik10(cik)}.json")
+        arrays = (base.get("filings") or {}).get("recent") or {}
+        forms = arrays.get("form") or []
+        dates = arrays.get("filingDate") or []
+        accessions = arrays.get("accessionNumber") or []
+        docs = arrays.get("primaryDocument") or []
+
+        candidates: list[tuple[date, str, str, str]] = []
+        for i, form in enumerate(forms):
+            if form not in _DELISTING_FORMS or i >= len(dates) or not dates[i]:
+                continue
+            try:
+                filed = date.fromisoformat(dates[i])
+            except ValueError:
+                continue
+            if not_before is not None and filed < not_before:
+                continue
+            if i >= len(accessions) or not accessions[i]:
+                continue
+            candidates.append((filed, form, accessions[i], docs[i] if i < len(docs) else ""))
+
+        for filed, form, accession, _doc in sorted(candidates, reverse=True):
+            security = await self._form25_security_class(cik, accession)
+            if security is None:
+                continue
+            if not _is_common_stock(security):
+                continue
+            return {
+                "form": form,
+                "filing_date": filed,
+                "security_class": security,
+            }
+        return None
+
+    async def _form25_security_class(
+        self, cik: int | str, accession: str
+    ) -> str | None:
+        """``descriptionClassSecurity`` from a Form 25's primary XML, or None.
+
+        The rendered ``primaryDocument`` is an XSL view of this file; the raw
+        ``primary_doc.xml`` beside it is the structured original.
+        """
+        folder = accession.replace("-", "")
+        url = (
+            f"{_WWW}/Archives/edgar/data/{int(cik)}/{folder}/primary_doc.xml"
+        )
+        try:
+            body = await self.get_text(url)
+        except SecNotFoundError:
+            return None
+        match = re.search(
+            r"<descriptionClassSecurity>(.*?)</descriptionClassSecurity>",
+            body,
+            re.IGNORECASE | re.DOTALL,
+        )
+        if match is None:
+            return None
+        return " ".join(match.group(1).split()) or None
+
     async def companyfacts(self, cik: int | str) -> dict[str, Any]:
         """Raw companyfacts JSON ({cik, entityName, facts})."""
         return await self.get_json(f"{_DATA}/api/xbrl/companyfacts/CIK{cik10(cik)}.json")

app/services/sec_fundamentals_importer.py

@@ -51,10 +51,10 @@ import json
 import logging
 from collections import Counter, defaultdict
 from dataclasses import dataclass, field, replace
-from datetime import date, datetime, timedelta, timezone
+from datetime import date, datetime, time, timedelta, timezone
 from typing import Any, Callable
 
-from sqlalchemy import delete, select, update
+from sqlalchemy import delete, exists, select, update
 
 from app.database import insert_for_session
 from app.models.data_import_run import DataImportRun
@@ -81,6 +81,26 @@ 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
+
+# Aggregate ceiling on deferral. MISSING_XBRL_RETRY_DAYS bounds how long ONE
+# filing blocks; it does not bound how long the import as a whole can stay
+# deferred. Those differ because a blocking filing is only queued by promote(),
+# which a deferred run never reaches — so during a rolling supply of
+# unresolvable filings (earnings season, when SEC's Company-Facts aggregation is
+# furthest behind) each new arrival restarts the clock before the previous one
+# clears, and nothing is written at all: not the good rows, not the gap rows
+# that would stop those filings blocking again.
+#
+# Once promotions have been stale this long, every unresolved filing is treated
+# as past the window. promote() then queues them all (see the _past_retry_window
+# call there), source_max_date advances, and _missing() forces queued rows
+# aged-out on later runs so they never block again — the import self-heals
+# through the paths that already exist.
+#
+# Well above MISSING_XBRL_RETRY_DAYS so ordinary overlapping blocks never trip
+# it. Affected symbols stay barred from setups either way: setup_blocked_ciks is
+# built from every missing filing regardless of window.
+PROMOTION_CEILING_DAYS = 7
 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
@@ -157,6 +177,9 @@ class SecFundamentalsImporter:
         self._retry_rows: list[dict[str, Any]] = []
         self._latest_index_date: date | None = None
         self._backfill = False
+        # Set by validate() when the aggregate ceiling forced the block open;
+        # read by promote() to alert that it did.
+        self._ceiling_tripped: dict[str, Any] | None = None
 
     # -- SourceImporter protocol -------------------------------------------
 
@@ -421,6 +444,24 @@ class SecFundamentalsImporter:
         # reconstructible by re-walking the index.
         blocking = _within_retry_window(staged.missing_xbrl)
         aged_out = _past_retry_window(staged.missing_xbrl)
+
+        # ...unless promotions have been stale past the aggregate ceiling, in
+        # which case the deferral has cost more than the filings it withholds.
+        # Ageing them here (not just locally) is deliberate: promote() re-derives
+        # the queue from the same list, so this is what gets them queued.
+        self._ceiling_tripped = None
+        if blocking and db is not None and await self._promotions_stale(db):
+            for item in staged.missing_xbrl:
+                item["age_days"] = max(
+                    item.get("age_days", 0), MISSING_XBRL_RETRY_DAYS + 1
+                )
+            self._ceiling_tripped = {
+                "forced": len(blocking),
+                "unresolved": len(staged.missing_xbrl),
+            }
+            blocking = _within_retry_window(staged.missing_xbrl)
+            aged_out = _past_retry_window(staged.missing_xbrl)
+
         if blocking:
             messages.append(
                 f"{len(blocking)} tracked XBRL filing(s) unresolved within the "
@@ -464,6 +505,9 @@ class SecFundamentalsImporter:
             "missing_xbrl": staged.missing_xbrl[:50],
             "missing_xbrl_count": len(staged.missing_xbrl),
             "missing_xbrl_blocking": len(blocking),
+            # Present only when the aggregate ceiling forced this run through, so
+            # a promoted run that carries known-unresolved filings says so.
+            "promotion_ceiling_tripped": self._ceiling_tripped,
             "recovered_from_coregistrant": staged.recovered[:50],
             "recovered_count": len(staged.recovered),
             # Complete compact gate input; detailed audit lists above stay capped.
@@ -616,6 +660,25 @@ class SecFundamentalsImporter:
                 created_at=_now(),
             ))
 
+        # A ceiling-forced promotion is the safety valve firing — it must be
+        # visible, or the import silently starts carrying known-unresolved
+        # filings. The affected symbols stay barred from setups regardless.
+        if self._ceiling_tripped:
+            db.add(SystemEvent(
+                severity="warning",
+                source="sec_facts",
+                code="promotion_ceiling_forced",
+                message=(
+                    f"Promoted with {self._ceiling_tripped['unresolved']} unresolved "
+                    f"filing(s) — {self._ceiling_tripped['forced']} still inside the "
+                    f"{MISSING_XBRL_RETRY_DAYS}-day retry window — because nothing had "
+                    f"promoted in {PROMOTION_CEILING_DAYS} days. They are queued for "
+                    "retry and their symbols remain blocked from setups."
+                )[:4000],
+                dedup_key=f"sec_facts:promotion_ceiling_forced:{run_id}",
+                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)
@@ -757,6 +820,38 @@ class SecFundamentalsImporter:
             if accession not in resolved
         ]
 
+    async def _promotions_stale(self, db) -> bool:
+        """Has nothing promoted within ``PROMOTION_CEILING_DAYS``?
+
+        Only true for a source that HAS promoted before. A never-promoted import
+        is initial setup, not a wedge: forcing its first promotion through would
+        mask a misconfiguration rather than recover from a transient SEC gap.
+
+        Measured from ``self.today`` rather than the wall clock, so the ceiling
+        honors the same injected date that ages the filings it releases.
+        """
+        cutoff = datetime.combine(
+            self.today - timedelta(days=PROMOTION_CEILING_DAYS),
+            time.min,
+            tzinfo=timezone.utc,
+        )
+        ever, recent = (
+            await db.execute(
+                select(
+                    exists().where(
+                        DataImportRun.source == SOURCE,
+                        DataImportRun.status == STATUS_PROMOTED,
+                    ),
+                    exists().where(
+                        DataImportRun.source == SOURCE,
+                        DataImportRun.status == STATUS_PROMOTED,
+                        DataImportRun.started_at >= cutoff,
+                    ),
+                )
+            )
+        ).one()
+        return bool(ever) and not bool(recent)
+
     async def _last_processed_index_date(self, db) -> date | None:
         return (
             await db.execute(

app/services/sec_universe.py

@@ -24,7 +24,7 @@ from typing import Iterable
 from sqlalchemy import select, update
 
 from app.models.ticker import Ticker
-from app.services import settings_store
+from app.services import settings_store, ticker_service
 from app.services.earnings_alignment import normalise_symbol
 from app.services.sec_client import SecClient
 
@@ -55,7 +55,11 @@ async def resolve_ciks(db, client: SecClient) -> ResolvedUniverse:
     returns the mapping + proposed `tickers.cik` writes; mutates nothing."""
     ticker_to_cik = await client.company_tickers()
     overrides = await cik_overrides(db)
-    rows = (await db.execute(select(Ticker.id, Ticker.symbol, Ticker.cik))).all()
+    rows = (
+        await db.execute(
+            ticker_service.active_only(select(Ticker.id, Ticker.symbol, Ticker.cik))
+        )
+    ).all()
 
     result = ResolvedUniverse()
     for tid, symbol, current_cik in rows:

app/services/ticker_service.py

@@ -1,13 +1,65 @@
-"""Ticker Registry service: add, delete, and list tracked tickers."""
+"""Ticker Registry service: add, delete, list, and retire tracked tickers."""
 
+import logging
 import re
+from datetime import date, timedelta
 
-from sqlalchemy import select
+from sqlalchemy import func, or_, select, update
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from app.exceptions import DuplicateError, NotFoundError, ValidationError
 from app.models.ticker import Ticker
 
+logger = logging.getLogger(__name__)
+
+# Reasons a symbol may be marked delisted, narrowest first.
+REASON_FORM_25 = "form_25"   # SEC Form 25/25-NSE/15 confirmed the exchange exit
+REASON_MANUAL = "manual"     # an operator decided
+
+# How long a symbol must be without bars before we spend an SEC request asking
+# whether it delisted. Guards against a market-data outage probing the whole
+# universe at once; a real delisting is still stale days later.
+MIN_STALE_DAYS_BEFORE_PROBE = 3
+
+# Rule 12d2-2: a Form 25 removal takes effect ten days after filing, so the
+# filing date is not the date the security stopped trading.
+FORM_25_EFFECTIVE_DAYS = 10
+
+# How far before the last bar a Form 25 may be filed and still explain this gap.
+# An exchange can file shortly before trading actually stops; anything older
+# concerns a class that was already gone while the symbol kept printing bars.
+FILING_LOOKBACK_DAYS = 30
+
+
+def _sec_client_factory():
+    """Build the SEC client for a delisting probe (patched in tests).
+
+    Imported lazily so the SEC/httpx stack stays off the import path of every
+    module that only wants ``active_only``.
+    """
+    from app.services.sec_client import SecClient
+
+    return SecClient()
+
+
+def active_only(stmt, *, as_of: date | None = None):
+    """Restrict a Ticker query to symbols that still trade.
+
+    Opt-in on purpose rather than folded into a shared getter: list and admin
+    views deliberately keep delisted rows so the delisting is *visible*, which a
+    silent default would undo. Apply this on the live signal path — scanning,
+    ranking, scoring, breadth, ingestion — and nowhere else.
+
+    ``delisted_on`` is an *effective* date, and a Form 25 is known ten days
+    before it takes effect, so a future date must not drop the symbol yet — it
+    is still trading and still worth scanning and ingesting. Compared in SQL
+    against the database's own date; ``as_of`` overrides it for tests.
+    """
+    cutoff = func.current_date() if as_of is None else as_of
+    return stmt.where(
+        or_(Ticker.delisted_on.is_(None), Ticker.delisted_on > cutoff)
+    )
+
 
 async def add_ticker(db: AsyncSession, symbol: str) -> Ticker:
     """Add a new ticker after validation.
@@ -52,6 +104,150 @@ async def delete_ticker(db: AsyncSession, symbol: str) -> None:
 
 
 async def list_tickers(db: AsyncSession) -> list[Ticker]:
-    """Return all tracked tickers sorted alphabetically by symbol."""
+    """Return all tracked tickers sorted alphabetically by symbol.
+
+    Delisted symbols are included and carry ``delisted_on`` — the registry is
+    where an operator needs to *see* that a symbol retired, not where it should
+    quietly disappear.
+    """
     result = await db.execute(select(Ticker).order_by(Ticker.symbol.asc()))
     return list(result.scalars().all())
+
+
+async def mark_delisted(
+    db: AsyncSession,
+    symbol: str,
+    *,
+    delisted_on: date,
+    reason: str = REASON_MANUAL,
+) -> bool:
+    """Record that a symbol stopped trading. True if this changed anything.
+
+    Idempotent, so the staleness path can call it every run without churning the
+    row: re-marking is a no-op. The one exception is an SEC confirmation landing
+    on a row an operator marked by hand — Form 25 carries the real effective
+    date, so it replaces the operator's estimate. Nothing downgrades a confirmed
+    row back to a manual one.
+    """
+    normalised = symbol.strip().upper()
+    result = await db.execute(select(Ticker).where(Ticker.symbol == normalised))
+    ticker = result.scalar_one_or_none()
+    if ticker is None:
+        raise NotFoundError(f"Ticker not found: {normalised}")
+    if ticker.delisted_on is not None:
+        upgrading = (
+            reason == REASON_FORM_25 and ticker.delisted_reason != REASON_FORM_25
+        )
+        if not upgrading:
+            return False
+
+    await db.execute(
+        update(Ticker)
+        .where(Ticker.id == ticker.id)
+        .values(delisted_on=delisted_on, delisted_reason=reason)
+    )
+    await db.commit()
+    logger.info(
+        "ticker %s marked delisted on %s (%s)", normalised, delisted_on, reason
+    )
+    return True
+
+
+async def confirm_delisting(
+    db: AsyncSession,
+    symbol: str,
+    *,
+    last_bar: date | None,
+    today: date | None = None,
+) -> date | None:
+    """Ask SEC whether ``symbol`` actually delisted; mark it if so.
+
+    Called when OHLCV goes stale, because "no new bars" alone cannot tell a
+    delisting from a halt or a rename. Returns the effective date whenever the
+    symbol is known to have delisted — whether this call established that or an
+    earlier one did — and ``None`` while it remains unproven, so the caller warns
+    only about gaps that still have no explanation.
+
+    Returning the already-known date matters between filing and effect: trading
+    usually stops before the ten-day Rule 12d2-2 delay expires, so the symbol is
+    correctly still active (see ``active_only``) while producing no bars. Without
+    this the staleness warning would fire daily across that window — the exact
+    noise the delisting flow exists to remove.
+
+    Deliberately driven by staleness rather than by the SEC fundamentals import:
+    that importer stalls for days at a time on unrelated Company-Facts gaps, and
+    detection wired into it would stall with it.
+
+    The probe waits for ``MIN_STALE_DAYS_BEFORE_PROBE``. A delisted symbol stays
+    stale forever, so the delay costs nothing, and it keeps a broad market-data
+    outage — where every tracked symbol reports stale at once — from turning into
+    one SEC request per symbol per run.
+    """
+    from app.services.sec_client import SecError
+
+    normalised = symbol.strip().upper()
+    result = await db.execute(select(Ticker).where(Ticker.symbol == normalised))
+    ticker = result.scalar_one_or_none()
+    if ticker is None:
+        return None
+    known = ticker.delisted_on
+    # Already confirmed by SEC — nothing left to learn, but the caller still
+    # needs the date to know this gap is explained. A row an operator marked by
+    # hand is worth probing: Form 25 upgrades the estimated date.
+    if ticker.delisted_reason == REASON_FORM_25:
+        return known
+    if not ticker.cik:
+        return known
+    # No bars at all is an ingestion problem, not evidence of a delisting.
+    if last_bar is None:
+        return known
+    if ((today or date.today()) - last_bar).days < MIN_STALE_DAYS_BEFORE_PROBE:
+        return known
+
+    try:
+        async with _sec_client_factory() as client:
+            # Only a Form 25 filed around or after the last bar can explain THIS
+            # gap. An older one belongs to a class that stopped trading before
+            # the symbol was still printing bars, and must not retire it.
+            filing = await client.delisting_filing(
+                ticker.cik, not_before=last_bar - timedelta(days=FILING_LOOKBACK_DAYS)
+            )
+    except SecError:
+        # Never let a probe failure escalate a routine staleness warning.
+        logger.warning("delisting probe failed for %s", normalised, exc_info=True)
+        return known
+
+    if filing is None:
+        return known
+    # Removal takes effect ten days after filing, so the filing date is not the
+    # date the symbol stopped trading.
+    effective = filing["filing_date"] + timedelta(days=FORM_25_EFFECTIVE_DAYS)
+    if await mark_delisted(
+        db, normalised, delisted_on=effective, reason=REASON_FORM_25
+    ):
+        return effective
+    return known
+
+
+async def clear_delisted(db: AsyncSession, symbol: str) -> bool:
+    """Un-retire a symbol. True if it had been marked.
+
+    The counterpart that makes automatic marking acceptable: a false positive
+    costs one row update, where a delete would have cost the price history.
+    """
+    normalised = symbol.strip().upper()
+    result = await db.execute(select(Ticker).where(Ticker.symbol == normalised))
+    ticker = result.scalar_one_or_none()
+    if ticker is None:
+        raise NotFoundError(f"Ticker not found: {normalised}")
+    if ticker.delisted_on is None:
+        return False
+
+    await db.execute(
+        update(Ticker)
+        .where(Ticker.id == ticker.id)
+        .values(delisted_on=None, delisted_reason=None)
+    )
+    await db.commit()
+    logger.info("ticker %s un-marked as delisted", normalised)
+    return True

app/services/ticker_universe_service.py

@@ -357,8 +357,25 @@ async def bootstrap_universe(
         db.add(Ticker(symbol=symbol))
 
     deleted_count = 0
+    skipped_delisted: list[str] = []
     if symbols_to_delete:
-        result = await db.execute(delete(Ticker).where(Ticker.symbol.in_(symbols_to_delete)))
+        # A delisted row was retained on purpose — its price history is exactly
+        # what a survivorship-honest backtest needs, and the delete cascades it
+        # away. Pruning must not undo that. (Pruning a symbol that is merely no
+        # longer an index constituent still destroys history; that needs a
+        # tracked/membership state separate from delisting.)
+        protected = (
+            await db.execute(
+                select(Ticker.symbol).where(
+                    Ticker.symbol.in_(symbols_to_delete),
+                    Ticker.delisted_on.is_not(None),
+                )
+            )
+        ).scalars().all()
+        skipped_delisted = sorted(protected)
+        deletable = [s for s in symbols_to_delete if s not in set(protected)]
+        if deletable:
+            result = await db.execute(delete(Ticker).where(Ticker.symbol.in_(deletable)))
             deleted_count = int(result.rowcount or 0)
 
     await db.commit()
@@ -378,4 +395,8 @@ async def bootstrap_universe(
         "already_tracked": len(target_symbols & existing_symbols),
         "deleted": deleted_count,
         "added_symbols": symbols_to_add[:50],
+        # Delisted rows a prune declined to destroy, so the caller can see the
+        # count did not match what they asked to remove.
+        "kept_delisted": skipped_delisted[:50],
+        "kept_delisted_count": len(skipped_delisted),
     }

frontend/src/lib/types.ts

@@ -858,6 +858,10 @@ export interface Ticker {
   symbol: string;
   name: string | null;
   created_at: string;
+  /** Set once the symbol stopped trading: excluded from signals, history kept. */
+  delisted_on: string | null;
+  /** How the delisting was learned: "form_25" (SEC confirmed) | "manual". */
+  delisted_reason: string | null;
 }
 
 // Admin

tests/unit/test_backtest_service.py

@@ -1620,3 +1620,71 @@ async def test_run_backtest_smoke(session):
     sweep = sorted(report["sweep"], key=lambda r: r["min_momentum_percentile"], reverse=True)
     counts = [r["total"] for r in sweep]
     assert counts == sorted(counts)  # ascending as threshold descends
+
+
+async def test_run_backtest_rolls_back_a_failed_ticker_fetch(session, monkeypatch):
+    """A failed per-ticker read must not leave the session mid-failed-transaction.
+
+    Every DB call in the replay loop is best-effort, but swallowing the error
+    without a rollback leaves asyncpg in "current transaction is aborted": every
+    later statement fails the same way until the first unguarded one — the report
+    write — surfaces it as the job error, long after the real cause.
+    """
+    await _seed_oscillating_ticker(session, "AAA")
+    await _seed_oscillating_ticker(session, "OSC")
+
+    real_fetch = bt._fetch_columns
+    rolled_back: list[str] = []
+
+    async def failing_fetch(db, symbol):
+        if symbol == "AAA":
+            raise RuntimeError("simulated OHLCV read failure")
+        return await real_fetch(db, symbol)
+
+    real_rollback = session.rollback
+
+    async def tracking_rollback():
+        rolled_back.append("x")
+        await real_rollback()
+
+    monkeypatch.setattr(bt, "_fetch_columns", failing_fetch)
+    monkeypatch.setattr(session, "rollback", tracking_rollback)
+
+    report = await bt.run_backtest(session)
+
+    assert rolled_back, "a failed ticker fetch left the session un-rolled-back"
+    # the surviving ticker is still replayed after the rollback
+    assert report["tickers"] == 2
+    assert report["candidates"] >= 1
+
+
+async def test_run_backtest_rolls_back_a_failed_portfolio_sim_load(session, monkeypatch):
+    """The portfolio-sim block loads the benchmark and the live exit policy from
+    the same session, well after the replay loop. A failure there poisons the
+    transaction exactly as one in the loop does, and the report write pays for it.
+    """
+    await _seed_oscillating_ticker(session, "OSC")
+
+    rolled_back: list[str] = []
+    called: list[str] = []
+
+    async def failing_exit_policy(db):
+        called.append("x")
+        raise RuntimeError("simulated exit-policy read failure")
+
+    real_rollback = session.rollback
+
+    async def tracking_rollback():
+        rolled_back.append("x")
+        await real_rollback()
+
+    monkeypatch.setattr(
+        "app.services.paper_trade_service.get_exit_policy", failing_exit_policy
+    )
+    monkeypatch.setattr(session, "rollback", tracking_rollback)
+
+    report = await bt.run_backtest(session)
+
+    assert called, "the portfolio-sim block never ran; test proves nothing"
+    assert rolled_back, "a failed portfolio-sim load left the session un-rolled-back"
+    assert report["tickers"] == 1

tests/unit/test_sec_fundamentals_importer.py

@@ -6,7 +6,7 @@ from __future__ import annotations
 import json
 import os
 import tempfile
-from datetime import date, datetime, timezone
+from datetime import date, datetime, timedelta, timezone
 
 import pytest
 from sqlalchemy import func, select
@@ -1128,3 +1128,158 @@ async def test_malformed_cik_override_is_ignored_not_fatal(engine):
         resolved = await resolve_ciks(db, client)
 
     assert resolved.symbol_to_cik["AAPL"] == 320193  # fell back to company_tickers
+
+
+# --- aggregate deferral ceiling -------------------------------------------
+
+def _blocking_staged(today: date) -> StagedFundamentals:
+    """One filing still inside the per-filing retry window."""
+    return StagedFundamentals(
+        resolved=ResolvedUniverse(),
+        missing_xbrl=[{
+            "cik": "0000320193",
+            "accession": "YOUNG-1",
+            "form": "10-Q",
+            "index_date": today,
+            "age_days": 0,
+            "reason": "not_in_companyfacts",
+        }],
+    )
+
+
+async def _add_run(factory, *, status: str, started_at: datetime) -> None:
+    from app.models.data_import_run import DataImportRun
+
+    async with factory() as db:
+        db.add(DataImportRun(
+            source="sec_facts", status=status, started_at=started_at,
+        ))
+        await db.commit()
+
+
+async def _validate_with_history(engine, *, promoted_days_ago: int | None):
+    factory = _factory(engine)
+    today = date(2026, 5, 20)
+    now = datetime(2026, 5, 20, 12, 0, tzinfo=timezone.utc)
+    if promoted_days_ago is not None:
+        await _add_run(
+            factory,
+            status=STATUS_PROMOTED,
+            started_at=now - timedelta(days=promoted_days_ago),
+        )
+    importer = SecFundamentalsImporter(today=today)
+    importer._latest_index_date = date(2026, 5, 19)
+    staged = _blocking_staged(today)
+    async with factory() as db:
+        return await importer.validate(db, staged), staged, importer
+
+
+async def test_ceiling_forces_a_promotion_once_deferral_outlasts_it(engine):
+    """The per-filing window bounds one filing; this bounds the whole import."""
+    result, staged, importer = await _validate_with_history(engine, promoted_days_ago=10)
+
+    assert result.ok is True
+    assert result.summary["missing_xbrl_blocking"] == 0
+    assert result.summary["promotion_ceiling_tripped"] == {"forced": 1, "unresolved": 1}
+    # Aged in place, so promote() re-derives the same verdict and queues it.
+    assert staged.missing_xbrl[0]["age_days"] > 3
+    # The symbol stays barred from setups — promoting is not trusting the data.
+    assert result.summary["setup_blocked_ciks"] == ["0000320193"]
+
+
+async def test_a_recent_promotion_keeps_the_normal_block(engine):
+    result, staged, _ = await _validate_with_history(engine, promoted_days_ago=1)
+
+    assert result.ok is False
+    assert result.summary["missing_xbrl_blocking"] == 1
+    assert result.summary["promotion_ceiling_tripped"] is None
+    assert result.retryable is True
+    assert staged.missing_xbrl[0]["age_days"] == 0
+
+
+async def test_ceiling_never_fires_before_a_first_promotion(engine):
+    """No baseline means initial setup, not a wedge — forcing it through would
+    mask a misconfiguration instead of recovering from an SEC gap."""
+    result, _, _ = await _validate_with_history(engine, promoted_days_ago=None)
+
+    assert result.ok is False
+    assert result.summary["promotion_ceiling_tripped"] is None
+
+
+async def test_ceiling_promotes_queues_and_alerts_end_to_end(engine, monkeypatch):
+    """The self-heal claim, end to end: a filing that would block forever gets
+    promoted through, queued for retry, and announced."""
+    from app.models.data_import_run import DataImportRun
+    from app.services.sec_facts_parser import ParseResult
+    from sqlalchemy import update as sa_update
+
+    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),
+    )
+    assert (await run_import(_importer(backfill), engine=engine)).status == STATUS_PROMOTED
+
+    # Age the only promotion past the ceiling: this is the wedge the ceiling exists
+    # for — the filing below stays young, so nothing else would ever release it.
+    async with factory() as db:
+        await db.execute(
+            sa_update(DataImportRun)
+            .where(DataImportRun.source == "sec_facts")
+            .values(started_at=datetime(2026, 4, 20, tzinfo=timezone.utc))
+        )
+        await db.commit()
+
+    # Present in Company Facts but unparseable — one missing_xbrl entry, not the
+    # two an absent-from-facts accession would also raise.
+    stuck_fact = _rev("2025-09-28", "2026-03-28", 254940, 2026, "Q2", "STUCK")
+    stuck_share = _shares("2026-04-17", 14687, "STUCK", 2026, "Q2")
+    client = FakeSecClient(
+        tickers={"AAPL": 320193},
+        companyfacts={
+            320193: _companyfacts(
+                [CF_K, CF_Q1, stuck_fact], [SH_K, SH_Q1, stuck_share]
+            )
+        },
+        submissions={320193: _submissions(SUB_FILINGS + [
+            _filing("STUCK", "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": "STUCK"},
+        ]},
+    )
+    monkeypatch.setattr(
+        "app.services.sec_facts_parser.parse_snapshots",
+        lambda *a, **k: ParseResult(
+            skipped_filings=[{"accession": "STUCK", "reason": "unparseable"}]
+        ),
+    )
+
+    # index_date 2026-05-01 vs today 2026-05-03 => 2 days old, still inside the
+    # per-filing window, so only the aggregate ceiling can let this through.
+    run = await run_import(_importer(client, today=date(2026, 5, 3)), engine=engine)
+    assert run.status == STATUS_PROMOTED
+
+    summary = json.loads(run.validation_json)
+    assert summary["promotion_ceiling_tripped"] == {"forced": 1, "unresolved": 1}
+
+    async with factory() as db:
+        gap = (await db.execute(select(SecFilingGap))).scalar_one()
+        events = (
+            await db.execute(
+                select(SystemEvent).where(
+                    SystemEvent.code == "promotion_ceiling_forced"
+                )
+            )
+        ).scalars().all()
+    # Queued, so later runs retry it without it ever blocking again...
+    assert gap.accession == "STUCK"
+    # ...and the safety valve firing is visible, not silent.
+    assert len(events) == 1
+    assert events[0].severity == "warning"
+    assert "7 days" in events[0].message

tests/unit/test_ticker_delisting.py

@@ -0,0 +1,416 @@
+"""Delisting lifecycle: marking, the active_only filter, and SEC confirmation.
+
+The behaviour under test is that a delisted symbol leaves the *live* path while
+its rows stay put — deleting it instead is what makes the backtest universe
+survivorship-biased, so retention is the point, not a side effect.
+"""
+
+from __future__ import annotations
+
+import json
+from collections.abc import AsyncGenerator
+from datetime import date
+
+import httpx
+import pytest
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
+
+from app.database import Base
+from app.models.ticker import Ticker
+from app.services import ticker_service
+from app.services.sec_client import SecClient
+
+_engine = create_async_engine("sqlite+aiosqlite://", echo=False)
+_session_factory = async_sessionmaker(_engine, class_=AsyncSession, expire_on_commit=False)
+
+
+@pytest.fixture(autouse=True)
+async def _setup_tables() -> AsyncGenerator[None, None]:
+    async with _engine.begin() as conn:
+        await conn.run_sync(Base.metadata.create_all)
+    yield
+    async with _engine.begin() as conn:
+        await conn.run_sync(Base.metadata.drop_all)
+
+
+@pytest.fixture
+async def session() -> AsyncGenerator[AsyncSession, None]:
+    async with _session_factory() as s:
+        yield s
+
+
+def _submissions(forms: list[str], dates: list[str]) -> dict:
+    return {
+        "cik": 712515,
+        "name": "ELECTRONIC ARTS INC.",
+        "filings": {
+            "recent": {
+                "form": forms,
+                "filingDate": dates,
+                "accessionNumber": [f"0001354457-26-{i:06d}" for i in range(len(forms))],
+                "primaryDocument": ["xslF25X02/primary_doc.xml"] * len(forms),
+            }
+        },
+    }
+
+
+def _sec_client(payload: dict, security: str | None = "Common Stock") -> SecClient:
+    """Mock submissions + the Form 25 primary document the class check reads."""
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        if request.url.path.endswith("primary_doc.xml"):
+            if security is None:
+                return httpx.Response(404)
+            body = (
+                "<?xml version='1.0'?><notificationOfRemoval>"
+                f"<descriptionClassSecurity>{security}</descriptionClassSecurity>"
+                "</notificationOfRemoval>"
+            )
+            return httpx.Response(200, content=body.encode())
+        return httpx.Response(200, content=json.dumps(payload).encode())
+
+    return SecClient(transport=httpx.MockTransport(handler), spacing_seconds=0)
+
+
+async def test_mark_delisted_is_idempotent(session: AsyncSession):
+    session.add(Ticker(symbol="EA"))
+    await session.commit()
+
+    assert await ticker_service.mark_delisted(
+        session, "EA", delisted_on=date(2026, 8, 4)
+    ) is True
+    # A second call must not churn the row — the staleness path retries daily.
+    assert await ticker_service.mark_delisted(
+        session, "EA", delisted_on=date(2026, 9, 1)
+    ) is False
+
+    row = (await session.execute(select(Ticker).where(Ticker.symbol == "EA"))).scalar_one()
+    assert row.delisted_on == date(2026, 8, 4)  # first date wins, not the retry
+    assert row.delisted_reason == ticker_service.REASON_MANUAL
+
+
+async def test_clear_delisted_restores_the_symbol(session: AsyncSession):
+    session.add(Ticker(symbol="EA"))
+    await session.commit()
+    await ticker_service.mark_delisted(session, "EA", delisted_on=date(2026, 8, 4))
+
+    assert await ticker_service.clear_delisted(session, "EA") is True
+    assert await ticker_service.clear_delisted(session, "EA") is False
+
+    row = (await session.execute(select(Ticker).where(Ticker.symbol == "EA"))).scalar_one()
+    assert row.delisted_on is None and row.delisted_reason is None
+
+
+async def test_active_only_filters_but_the_row_survives(session: AsyncSession):
+    session.add_all([Ticker(symbol="AAPL"), Ticker(symbol="EA")])
+    await session.commit()
+    await ticker_service.mark_delisted(session, "EA", delisted_on=date(2026, 8, 4))
+
+    active = (
+        await session.execute(ticker_service.active_only(select(Ticker.symbol)))
+    ).scalars().all()
+    assert list(active) == ["AAPL"]
+
+    # The whole point: the row — and everything cascading off it — is still there.
+    everything = [t.symbol for t in await ticker_service.list_tickers(session)]
+    assert everything == ["AAPL", "EA"]
+
+
+async def test_confirm_delisting_marks_on_a_form_25(session: AsyncSession, monkeypatch):
+    session.add(Ticker(symbol="EA", cik="0000712515"))
+    await session.commit()
+    monkeypatch.setattr(
+        ticker_service,
+        "_sec_client_factory",
+        lambda: _sec_client(_submissions(["8-K", "25-NSE"], ["2026-07-01", "2026-08-04"])),
+        raising=False,
+    )
+
+    marked = await ticker_service.confirm_delisting(
+        session, "EA", last_bar=date(2026, 8, 4), today=date(2026, 8, 11)
+    )
+    # Removal is effective ten days after the 2026-08-04 filing, not on it.
+    assert marked == date(2026, 8, 14)
+
+    row = (await session.execute(select(Ticker).where(Ticker.symbol == "EA"))).scalar_one()
+    assert row.delisted_reason == ticker_service.REASON_FORM_25
+
+
+async def test_confirm_delisting_leaves_a_halt_alone(session: AsyncSession, monkeypatch):
+    """A halt or a rename files no Form 25 — those must keep warning, not retire."""
+    session.add(Ticker(symbol="SATS", cik="0000012345"))
+    await session.commit()
+    monkeypatch.setattr(
+        ticker_service,
+        "_sec_client_factory",
+        lambda: _sec_client(_submissions(["8-K", "10-Q"], ["2026-07-01", "2026-08-04"])),
+        raising=False,
+    )
+
+    assert await ticker_service.confirm_delisting(
+        session, "SATS", last_bar=date(2026, 8, 4), today=date(2026, 8, 11)
+    ) is None
+    row = (await session.execute(select(Ticker).where(Ticker.symbol == "SATS"))).scalar_one()
+    assert row.delisted_on is None
+
+
+async def test_confirm_delisting_skips_a_symbol_without_a_cik(session: AsyncSession):
+    """No CIK, no SEC lookup — must not raise, and must not mark."""
+    session.add(Ticker(symbol="ADRX"))
+    await session.commit()
+    assert await ticker_service.confirm_delisting(
+        session, "ADRX", last_bar=date(2026, 8, 4), today=date(2026, 8, 11)
+    ) is None
+
+
+async def test_delisting_filing_picks_the_newest_match():
+    client = _sec_client(
+        _submissions(
+            ["25", "8-K", "25-NSE", "15-12B"],
+            ["2024-01-02", "2026-08-01", "2026-08-04", "2025-05-05"],
+        )
+    )
+    async with client as c:
+        found = await c.delisting_filing("0000712515")
+    assert found["form"] == "25-NSE"
+    assert found["filing_date"] == date(2026, 8, 4)
+
+
+async def test_delisting_filing_returns_none_without_one():
+    client = _sec_client(_submissions(["10-K", "8-K"], ["2026-01-02", "2026-08-01"]))
+    async with client as c:
+        assert await c.delisting_filing("0000320193") is None
+
+
+async def test_confirm_delisting_waits_before_spending_a_request(session: AsyncSession, monkeypatch):
+    """A one-day gap is a weekend or a hiccup. Probing every stale symbol during a
+    market-data outage would be one SEC request per symbol per run."""
+    session.add(Ticker(symbol="EA", cik="0000712515"))
+    await session.commit()
+
+    def _explode():
+        raise AssertionError("must not reach SEC before the stale threshold")
+
+    monkeypatch.setattr(ticker_service, "_sec_client_factory", _explode, raising=False)
+
+    assert await ticker_service.confirm_delisting(
+        session, "EA", last_bar=date(2026, 8, 10), today=date(2026, 8, 11)
+    ) is None
+    # ...and no bars at all is an ingestion problem, not a delisting.
+    assert await ticker_service.confirm_delisting(
+        session, "EA", last_bar=None, today=date(2026, 8, 11)
+    ) is None
+
+
+async def test_sec_confirmation_upgrades_a_manual_mark(session: AsyncSession, monkeypatch):
+    """An operator's estimated date is a guess; Form 25 carries the real one."""
+    session.add(Ticker(symbol="EA", cik="0000712515"))
+    await session.commit()
+    await ticker_service.mark_delisted(session, "EA", delisted_on=date(2026, 8, 11))
+
+    monkeypatch.setattr(
+        ticker_service,
+        "_sec_client_factory",
+        lambda: _sec_client(_submissions(["25-NSE"], ["2026-08-04"])),
+        raising=False,
+    )
+    assert await ticker_service.confirm_delisting(
+        session, "EA", last_bar=date(2026, 8, 4), today=date(2026, 8, 11)
+    ) == date(2026, 8, 14)
+
+    row = (await session.execute(select(Ticker).where(Ticker.symbol == "EA"))).scalar_one()
+    assert row.delisted_on == date(2026, 8, 14)
+    assert row.delisted_reason == ticker_service.REASON_FORM_25
+
+
+async def test_a_confirmed_row_is_never_reprobed(session: AsyncSession, monkeypatch):
+    session.add(Ticker(symbol="EA", cik="0000712515"))
+    await session.commit()
+    await ticker_service.mark_delisted(
+        session, "EA", delisted_on=date(2026, 8, 4),
+        reason=ticker_service.REASON_FORM_25,
+    )
+
+    def _explode():
+        raise AssertionError("a SEC-confirmed row must not cost another request")
+
+    monkeypatch.setattr(ticker_service, "_sec_client_factory", _explode, raising=False)
+    # Costs no request, and still reports the date so the caller knows this gap
+    # is explained and must not warn about it again.
+    assert await ticker_service.confirm_delisting(
+        session, "EA", last_bar=date(2026, 8, 4), today=date(2026, 9, 1)
+    ) == date(2026, 8, 4)
+
+
+async def test_ohlcv_priority_ordering_skips_delisted(session: AsyncSession):
+    """Covers the one statement where active_only wraps a compound select."""
+    from app.scheduler import _get_ohlcv_priority_tickers
+
+    session.add_all([Ticker(symbol="AAPL"), Ticker(symbol="EA"), Ticker(symbol="MSFT")])
+    await session.commit()
+    await ticker_service.mark_delisted(session, "EA", delisted_on=date(2026, 8, 4))
+
+    symbols = await _get_ohlcv_priority_tickers(session)
+    assert "EA" not in symbols
+    assert sorted(symbols) == ["AAPL", "MSFT"]
+
+
+async def test_a_form_25_for_another_security_class_is_ignored(session: AsyncSession, monkeypatch):
+    """Form 25 is per security class. An issuer delisting its notes, preferred or
+    warrants files one while the common keeps trading — retiring the ticker on
+    that would remove an actively traded symbol from every signal."""
+    session.add(Ticker(symbol="EA", cik="0000712515"))
+    await session.commit()
+    monkeypatch.setattr(
+        ticker_service,
+        "_sec_client_factory",
+        lambda: _sec_client(
+            _submissions(["25-NSE"], ["2026-08-04"]),
+            security="6.25% Notes due 2030",
+        ),
+        raising=False,
+    )
+
+    assert await ticker_service.confirm_delisting(
+        session, "EA", last_bar=date(2026, 8, 4), today=date(2026, 8, 11)
+    ) is None
+    row = (await session.execute(select(Ticker).where(Ticker.symbol == "EA"))).scalar_one()
+    assert row.delisted_on is None
+
+
+async def test_a_stale_historical_form_25_cannot_retire_a_symbol(session: AsyncSession, monkeypatch):
+    """A 2019 filing for a long-gone class must not retire a symbol whose bars
+    ran until 2026 — and must certainly not stamp 2019 as the date."""
+    session.add(Ticker(symbol="EA", cik="0000712515"))
+    await session.commit()
+    monkeypatch.setattr(
+        ticker_service,
+        "_sec_client_factory",
+        lambda: _sec_client(_submissions(["25"], ["2019-03-01"])),
+        raising=False,
+    )
+
+    assert await ticker_service.confirm_delisting(
+        session, "EA", last_bar=date(2026, 8, 4), today=date(2026, 8, 11)
+    ) is None
+
+
+async def test_form_15_alone_never_retires_a_symbol(session: AsyncSession, monkeypatch):
+    """Form 15 ends a reporting obligation; it is not evidence trading stopped."""
+    session.add(Ticker(symbol="EA", cik="0000712515"))
+    await session.commit()
+    monkeypatch.setattr(
+        ticker_service,
+        "_sec_client_factory",
+        lambda: _sec_client(_submissions(["15-12B", "15-12G"], ["2026-08-04", "2026-08-05"])),
+        raising=False,
+    )
+
+    assert await ticker_service.confirm_delisting(
+        session, "EA", last_bar=date(2026, 8, 4), today=date(2026, 8, 11)
+    ) is None
+
+
+async def test_an_unreadable_form_25_fails_closed(session: AsyncSession, monkeypatch):
+    """Pre-2009 filings have no primary_doc.xml. Unknown class must read as no."""
+    session.add(Ticker(symbol="EA", cik="0000712515"))
+    await session.commit()
+    monkeypatch.setattr(
+        ticker_service,
+        "_sec_client_factory",
+        lambda: _sec_client(_submissions(["25"], ["2026-08-04"]), security=None),
+        raising=False,
+    )
+
+    assert await ticker_service.confirm_delisting(
+        session, "EA", last_bar=date(2026, 8, 4), today=date(2026, 8, 11)
+    ) is None
+
+
+@pytest.mark.parametrize(
+    "description,expected",
+    [
+        ("Common Stock", True),
+        ("Class A Common Stock, $0.01 par value", True),
+        ("Common Shares, no par value", True),
+        ("6.25% Notes due 2030", False),
+        ("7.5% Series B Cumulative Preferred Stock", False),
+        ("Warrants to purchase Common Stock", False),
+        ("Depositary Shares each representing 1/1000th interest", False),
+        ("", False),
+    ],
+)
+def test_common_stock_classification(description: str, expected: bool):
+    from app.services.sec_client import _is_common_stock
+
+    assert _is_common_stock(description) is expected
+
+
+async def test_prune_keeps_delisted_rows(session: AsyncSession, monkeypatch):
+    """A prune must not destroy rows the delisting flow deliberately retained —
+    their price history is the whole reason those rows still exist."""
+    from app.services import ticker_universe_service as tus
+
+    session.add_all([Ticker(symbol="AAPL"), Ticker(symbol="EA"), Ticker(symbol="GONE")])
+    await session.commit()
+    await ticker_service.mark_delisted(session, "EA", delisted_on=date(2026, 8, 14))
+
+    async def fake_fetch(db, universe):
+        return ["AAPL"], "test"
+
+    monkeypatch.setattr(tus, "fetch_universe_symbols", fake_fetch)
+
+    summary = await tus.bootstrap_universe(session, "sp500", prune_missing=True)
+
+    remaining = sorted(t.symbol for t in await ticker_service.list_tickers(session))
+    assert remaining == ["AAPL", "EA"]  # GONE pruned, EA protected
+    assert summary["deleted"] == 1
+    assert summary["kept_delisted"] == ["EA"]
+
+
+async def test_a_future_effective_date_keeps_the_symbol_live(session: AsyncSession):
+    """Form 25 is known ten days before removal takes effect. The symbol is still
+    trading in that window and must keep being scanned and ingested."""
+    session.add_all([Ticker(symbol="AAPL"), Ticker(symbol="EA")])
+    await session.commit()
+    await ticker_service.mark_delisted(session, "EA", delisted_on=date(2026, 8, 14))
+
+    def active(as_of: date) -> list[str]:
+        return ticker_service.active_only(select(Ticker.symbol), as_of=as_of)
+
+    before = (await session.execute(active(date(2026, 8, 11)))).scalars().all()
+    on_the_day = (await session.execute(active(date(2026, 8, 14)))).scalars().all()
+    after = (await session.execute(active(date(2026, 8, 15)))).scalars().all()
+
+    assert sorted(before) == ["AAPL", "EA"]      # still trading
+    assert sorted(on_the_day) == ["AAPL"]        # removal effective
+    assert sorted(after) == ["AAPL"]
+
+
+async def test_the_pending_window_does_not_re_warn(session: AsyncSession, monkeypatch):
+    """Between filing and effect the symbol is active but produces no bars. That
+    must not resurrect the daily staleness warning this flow exists to end."""
+    session.add(Ticker(symbol="EA", cik="0000712515"))
+    await session.commit()
+    monkeypatch.setattr(
+        ticker_service,
+        "_sec_client_factory",
+        lambda: _sec_client(_submissions(["25-NSE"], ["2026-08-04"])),
+        raising=False,
+    )
+    first = await ticker_service.confirm_delisting(
+        session, "EA", last_bar=date(2026, 8, 4), today=date(2026, 8, 11)
+    )
+    assert first == date(2026, 8, 14)
+
+    def _explode():
+        raise AssertionError("must not re-probe a confirmed row")
+
+    monkeypatch.setattr(ticker_service, "_sec_client_factory", _explode, raising=False)
+    # Every later run inside the window still reports the delisting, so the
+    # caller keeps emitting "delisted" rather than "no new bars".
+    for day in (date(2026, 8, 12), date(2026, 8, 13)):
+        assert await ticker_service.confirm_delisting(
+            session, "EA", last_bar=date(2026, 8, 4), today=day
+        ) == date(2026, 8, 14)