Compare commits

...
8 Commits
Author SHA1 Message Date
dennisthiessenandClaude Opus 5 247a7889b9 fix(tickers): honor the effective date instead of retiring on the mark
Deploy / lint (push) Successful in 11s
Deploy / test (push) Successful in 1m22s
Deploy / deploy (push) Successful in 38s
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>
2026-08-11 16:58:51 +02:00
dennisthiessenandClaude Opus 5 1d4ed39fd2 fix(tickers): close the delisting review findings
Detection could retire an actively traded symbol — silently, since it then
vanishes from every signal. Three causes:

- Form 25 is filed per security class. An issuer removing its notes, preferred
  or warrants files one while the common keeps trading. The filing's own
  descriptionClassSecurity distinguishes them, so the primary document is now
  fetched and read; anything not recognisably common equity is rejected, as is
  anything unreadable (pre-2009 filings have no primary_doc.xml). Fail closed.
- Form 15 ends a reporting obligation and is no evidence trading stopped. The
  whole family is dropped.
- A historical filing for a long-gone class could retire a symbol whose bars ran
  years later, stamping the old date. Filings before the last bar (less a 30-day
  lead for the exchange) are now ignored.

Rule 12d2-2 makes removal effective ten days after filing, so delisted_on is the
effective date rather than the filing date.

bootstrap_universe(prune_missing=True) still ran a cascading delete over
delisted rows, undoing the retention this branch exists for; it now skips them
and reports kept_delisted so the count is explicable.

clear_delisted had no route, which made "safe to automate because it is
reversible" false — reversal needed SQL. POST/DELETE /tickers/{symbol}/delisting
now mark and un-mark, giving an operator a non-destructive alternative to the
cascading DELETE that was the only option.

Shared-CIK siblings (GOOG/GOOGL) stay safe by construction: the probe is
per-symbol and gated on that symbol's own staleness, so a class that still
trades is never probed.

Not addressed: pruning a symbol merely dropped from the index still destroys its
history — the same survivorship problem in a different costume, needing a
tracked/membership state separate from delisting.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
2026-08-11 16:58:51 +02:00
dennisthiessenandClaude Opus 5 d950fcf70e fix(tickers): let an SEC confirmation upgrade a manual delisting mark
mark_delisted returned early on any already-delisted row, so the sequence an
operator actually hits — mark EA by hand today, Form 25-NSE surfaces three days
later dated 2026-08-04 — left the estimated date and "manual" reason in place
permanently. Form 25 carries the real effective date, so it now replaces an
operator's estimate; a confirmed row is never downgraded or re-probed.

Also cover _get_ohlcv_priority_tickers, the one place active_only wraps a
compound select rather than a bare one — the unit suite reached none of it.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
2026-08-11 16:58:51 +02:00
dennisthiessenandClaude Opus 5 6501b7e9a0 feat(tickers): record delisting instead of deleting the symbol
Retiring a symbol meant delete_ticker or bootstrap_universe(prune_missing),
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 — it does not fix it, which needs the replay to model a delisting as an
exit event.

tickers gains delisted_on / delisted_reason (migration 032). NULL means
actively traded.

The filter is opt-in via ticker_service.active_only rather than folded into a
shared getter: the registry and admin views deliberately keep delisted rows so
the delisting is visible, and a silent default would undo that. Applied to the
live path only — scanner, momentum ranking, scoring, breadth, fundamentals
candidates, SEC universe, earnings import, ingestion loops. run_backtest keeps
them on purpose.

Detection runs off OHLCV staleness, not off the SEC fundamentals import: that
importer stalls for days on unrelated Company-Facts gaps and would take
detection down with it. On a stale symbol the scheduler asks SEC for a Form
25/25-NSE/15 and retires it only on a hit, so a halt or a rename (SATS->ECHO)
keeps the existing warning. The probe waits 3 stale days so a market-data
outage cannot turn into one SEC request per symbol per run.

Safe to automate because it is reversible: clear_delisted un-retires a false
positive, where a delete had already taken the history.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
2026-08-11 16:58:51 +02:00
dennisthiessenandClaude Opus 5 486fb500d1 test(sec): cover the ceiling's promote/queue/alert path end to end
The ceiling tests asserted validate()'s verdict but nothing proved the claim
the design rests on: that a forced promotion actually queues the filings it
released and says that it did. Drive it through run_import with a filing that
stays inside the per-filing window, so only the aggregate ceiling can release
it, and assert the SecFilingGap row and the promotion_ceiling_forced event.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
2026-08-11 16:58:46 +02:00
dennisthiessenandClaude Opus 5 77570557db feat(sec): cap how long the fundamentals import can stay deferred
MISSING_XBRL_RETRY_DAYS bounds how long ONE filing blocks promotion. It does
not bound the import as a whole, and the two come apart because a blocking
filing is only queued by promote(), which a deferred run never reaches. During
a rolling supply of unresolvable filings — earnings season, when SEC's
Company-Facts aggregation lags furthest — each new arrival restarts the 3-day
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.

Add an aggregate ceiling. Once promotions have been stale for
PROMOTION_CEILING_DAYS (7), every unresolved filing is aged past the retry
window in place, so promote() queues them all through the path that already
exists, source_max_date advances, and _missing() keeps queued rows aged-out on
later runs. The import self-heals instead of compounding.

Deliberately not the alternative of queueing gap rows on a deferred run: that
would drop the grace period to a single run for every filing, including the
common case of a Company-Facts lag that resolves in a day, and it needs a write
on a run that failed validation.

The per-filing window is untouched, a never-promoted source never trips (that
is initial setup, not a wedge), and affected symbols stay barred from setups
either way since setup_blocked_ciks ignores the window. A forced promotion
raises promotion_ceiling_forced so the safety valve is never silent.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
2026-08-11 16:58:46 +02:00
dennisthiessenandClaude Opus 5 fbca38e144 fix(backtest): roll back the portfolio-sim DB failures too
The first pass guarded the replay loop but not the portfolio-simulation block,
which re-fetches price columns and loads the benchmark and the live exit policy
from the same session much later. A failure in any of those swallows the
exception without clearing the transaction — the identical failure mode, with
the identical symptom: the report write is the first unguarded statement and
takes the blame.

The outer handler is the backstop for the price_columns loop, which has no
handler of its own; rolling back a session an inner handler already cleared is
a no-op.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
2026-08-11 16:10:57 +02:00
dennisthiessenandClaude Opus 5 6ca7f13779 fix(backtest): roll back the session after a swallowed DB failure
Every DB call in run_backtest is best-effort so one unreadable ticker cannot
abort the whole replay, but the handlers swallowed the exception without
clearing the transaction. asyncpg then reports "current transaction is
aborted" for every later statement, and the first unguarded one — the report
write — surfaced it as the job error, long after the real cause.

Add _rollback_quietly at the three swallowing sites (benchmark load, parallel
fetch, sequential replay), matching the guard price_service already uses.

Load plain symbols instead of Ticker instances: a rollback expires ORM objects
held across it, and touching an expired attribute afterwards triggers sync
lazy-loading, which raises on an AsyncSession. rr_scanner_service hit this
same trap. Only .symbol was ever used.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
2026-08-11 14:20:19 +02:00
21 changed files with 1294 additions and 50 deletions
+49
View File
@@ -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")
+9 -2
View File
@@ -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 sqlalchemy.orm import Mapped, mapped_column, relationship
from app.database import Base from app.database import Base
@@ -21,6 +21,13 @@ class Ticker(Base):
cik: Mapped[str | None] = mapped_column(String(10), nullable=True) cik: Mapped[str | None] = mapped_column(String(10), nullable=True)
sic: Mapped[str | None] = mapped_column(String(4), nullable=True) sic: Mapped[str | None] = mapped_column(String(4), nullable=True)
sic_description: Mapped[str | None] = mapped_column(String(160), 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( created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), default=datetime.utcnow, nullable=False DateTime(timezone=True), default=datetime.utcnow, nullable=False
) )
+33 -1
View File
@@ -6,7 +6,7 @@ from sqlalchemy.ext.asyncio import AsyncSession
from app.dependencies import get_db, require_access from app.dependencies import get_db, require_access
from app.models.user import User from app.models.user import User
from app.schemas.common import APIEnvelope 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 from app.services import ticker_service
router = APIRouter(tags=["tickers"]) router = APIRouter(tags=["tickers"])
@@ -51,3 +51,35 @@ async def delete_ticker(
"""Delete a ticker and all associated data.""" """Delete a ticker and all associated data."""
await ticker_service.delete_ticker(db, symbol) await ticker_service.delete_ticker(db, symbol)
return APIEnvelope(status="success", data=None) 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})
+36 -11
View File
@@ -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.outcome_service import evaluate_pending_setups
from app.services.rr_scanner_service import scan_all_tickers from app.services.rr_scanner_service import scan_all_tickers
from app.services.sentiment_provider_service import build_sentiment_provider 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 from app.services.ticker_universe_service import bootstrap_universe
logger = logging.getLogger(__name__) 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]: async def _get_all_tickers(db: AsyncSession) -> list[str]:
"""Return all tracked ticker symbols sorted alphabetically.""" """Return all actively-traded ticker symbols sorted alphabetically."""
result = await db.execute(select(Ticker.symbol).order_by(Ticker.symbol)) result = await db.execute(
ticker_service.active_only(select(Ticker.symbol).order_by(Ticker.symbol))
)
return list(result.scalars().all()) 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) latest_date = func.max(OHLCVRecord.date)
missing_first = case((latest_date.is_(None), 0), else_=1) missing_first = case((latest_date.is_(None), 0), else_=1)
result = await db.execute( result = await db.execute(
select(Ticker.symbol) ticker_service.active_only(
.outerjoin(OHLCVRecord, OHLCVRecord.ticker_id == Ticker.id) select(Ticker.symbol)
.outerjoin(OHLCVRecord, OHLCVRecord.ticker_id == Ticker.id)
)
.group_by(Ticker.id, Ticker.symbol) .group_by(Ticker.id, Ticker.symbol)
.order_by(missing_first.asc(), latest_date.asc(), Ticker.symbol.asc()) .order_by(missing_first.asc(), latest_date.asc(), Ticker.symbol.asc())
) )
@@ -662,14 +667,34 @@ async def collect_ohlcv(
_runtime_progress(job_name, processed=processed, total=total, current_ticker=symbol) _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) _log_event(logging.INFO, "ticker_collected", job=job_name, ticker=symbol, status=result.status, records=result.records_ingested)
if result.status == "stale": if result.status == "stale":
await _record_system_event( # "No new bars" cannot distinguish a delisting from a halt
severity="warning", # or a rename, so ask SEC before warning again. A confirmed
source=job_name, # delisting retires the symbol (keeping its history) and
code="ohlcv_stale", # ends the alert; anything unproven keeps warning.
message=result.message or f"No new OHLCV bars for {symbol}", delisted_on = await ticker_service.confirm_delisting(
symbol=symbol, db, symbol, last_bar=result.last_date
dedup_key=f"ohlcv_stale:{symbol}",
) )
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,
code="ohlcv_stale",
message=result.message or f"No new OHLCV bars for {symbol}",
symbol=symbol,
dedup_key=f"ohlcv_stale:{symbol}",
)
if result.status == "partial": if result.status == "partial":
# Rate limited — stop and resume next run # Rate limited — stop and resume next run
_log_event(logging.WARNING, "rate_limited", job=job_name, ticker=symbol, processed=processed) _log_event(logging.WARNING, "rate_limited", job=job_name, ticker=symbol, processed=processed)
+12 -1
View File
@@ -1,6 +1,6 @@
"""Ticker request/response schemas.""" """Ticker request/response schemas."""
from datetime import datetime from datetime import date, datetime
from pydantic import BaseModel, Field from pydantic import BaseModel, Field
@@ -14,5 +14,16 @@ class TickerResponse(BaseModel):
symbol: str symbol: str
name: str | None = None name: str | None = None
created_at: datetime 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} model_config = {"from_attributes": True}
class TickerDelistingUpdate(BaseModel):
delisted_on: date = Field(
..., description="Effective date the symbol stopped trading"
)
+43 -15
View File
@@ -1460,6 +1460,21 @@ def _mp_context():
return None 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: async def _fetch_columns(db: AsyncSession, symbol: str) -> tuple | None:
"""Read one ticker's OHLCV and detach it to primitive column arrays in the """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.""" 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) config = await get_recommendation_config(db)
activation = await get_activation_config(db) activation = await get_activation_config(db)
result = await db.execute(select(Ticker).order_by(Ticker.symbol)) # Plain strings, not Ticker instances: the rollbacks below expire any ORM
tickers = list(result.scalars().all()) # objects held across them, and touching an expired attribute afterwards
total = len(tickers) # 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) rank_only_symbols = await _load_research_rank_only_symbols(db)
if rank_only_symbols: if rank_only_symbols:
logger.info(json.dumps({ logger.info(json.dumps({
@@ -4063,6 +4081,7 @@ async def run_backtest(
) )
except Exception: except Exception:
logger.exception("Benchmark load for residual momentum failed") logger.exception("Benchmark load for residual momentum failed")
await _rollback_quietly(db, "benchmark load")
def _merge(result: tuple[list[dict], dict]) -> None: def _merge(result: tuple[list[dict], dict]) -> None:
cands, series = result cands, series = result
@@ -4094,26 +4113,27 @@ async def run_backtest(
done = 0 done = 0
with pool: with pool:
for start in range(0, total, chunk): for start in range(0, total, chunk):
batch = tickers[start : start + chunk] batch = symbols[start : start + chunk]
futures = [] futures = []
for ticker in batch: for symbol in batch:
try: try:
columns = await _fetch_columns(db, ticker.symbol) columns = await _fetch_columns(db, symbol)
except Exception: 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 continue
if columns is not None: if columns is not None:
futures.append(loop.run_in_executor( futures.append(loop.run_in_executor(
pool, pool,
_replay_and_signals, _replay_and_signals,
ticker.symbol, symbol,
columns, columns,
config, config,
activation, activation,
benchmark_closes, benchmark_closes,
target_model, target_model,
cadence, cadence,
ticker.symbol in rank_only_symbols, symbol in rank_only_symbols,
)) ))
for result in await asyncio.gather(*futures, return_exceptions=True): for result in await asyncio.gather(*futures, return_exceptions=True):
if isinstance(result, Exception): if isinstance(result, Exception):
@@ -4126,25 +4146,26 @@ async def run_backtest(
else: else:
# Sequential fallback (Windows / 1 worker): run each replay in a worker # Sequential fallback (Windows / 1 worker): run each replay in a worker
# thread so the event loop — and the API server — stays responsive. # 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: if progress_cb is not None:
progress_cb(index, total, ticker.symbol) progress_cb(index, total, symbol)
try: try:
columns = await _fetch_columns(db, ticker.symbol) columns = await _fetch_columns(db, symbol)
if columns is not None: if columns is not None:
_merge(await asyncio.to_thread( _merge(await asyncio.to_thread(
_replay_and_signals, _replay_and_signals,
ticker.symbol, symbol,
columns, columns,
config, config,
activation, activation,
benchmark_closes, benchmark_closes,
target_model, target_model,
cadence, cadence,
ticker.symbol in rank_only_symbols, symbol in rank_only_symbols,
)) ))
except Exception: 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: if progress_cb is not None and total:
progress_cb(total, total, "") progress_cb(total, total, "")
@@ -4209,6 +4230,7 @@ async def run_backtest(
) )
except Exception: except Exception:
logger.exception("Benchmark load for the portfolio sim failed") logger.exception("Benchmark load for the portfolio sim failed")
await _rollback_quietly(db, "portfolio-sim benchmark load")
for policy in ("target", "hold"): for policy in ("target", "hold"):
sim = _simulate_portfolio( sim = _simulate_portfolio(
@@ -4229,6 +4251,7 @@ async def run_backtest(
live_exit_policy = await get_exit_policy(db) live_exit_policy = await get_exit_policy(db)
except Exception: except Exception:
logger.exception("Live exit policy load failed; monitor uses defaults") logger.exception("Live exit policy load failed; monitor uses defaults")
await _rollback_quietly(db, "exit policy load")
portfolio_monitor_report = _portfolio_monitor( portfolio_monitor_report = _portfolio_monitor(
candidates, price_columns, spy_closes, hold_horizon, candidates, price_columns, spy_closes, hold_horizon,
live_exit_policy=live_exit_policy, live_exit_policy=live_exit_policy,
@@ -4248,6 +4271,11 @@ async def run_backtest(
) )
except Exception: except Exception:
logger.exception("Portfolio simulation failed") 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 = { report = {
"generated_at": datetime.now(timezone.utc).isoformat(), "generated_at": datetime.now(timezone.utc).isoformat(),
+2 -1
View File
@@ -25,6 +25,7 @@ from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.ext.asyncio import AsyncSession
from app.models.ticker import Ticker from app.models.ticker import Ticker
from app.services import ticker_service
from app.services.price_service import query_ohlcv from app.services.price_service import query_ohlcv
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
@@ -112,7 +113,7 @@ def compute_divergence_series(
async def _load_universe_closes( async def _load_universe_closes(
db: AsyncSession, symbols: list[str] | None = None db: AsyncSession, symbols: list[str] | None = None
) -> dict[str, Series]: ) -> 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: if symbols is not None:
stmt = stmt.where(Ticker.symbol.in_(symbols)) stmt = stmt.where(Ticker.symbol.in_(symbols))
result = await db.execute(stmt) result = await db.execute(stmt)
+6 -2
View File
@@ -32,7 +32,7 @@ from app.config import settings
from app.database import insert_for_session from app.database import insert_for_session
from app.models.earnings_event import EarningsEvent from app.models.earnings_event import EarningsEvent
from app.models.ticker import Ticker 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 from app.services.data_import import ValidationResult
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
@@ -289,7 +289,11 @@ class DoltEarningsImporter:
# -- helpers ----------------------------------------------------------- # -- helpers -----------------------------------------------------------
async def _load_universe(self, db) -> dict[str, int]: 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 { return {
earnings_alignment.normalise_symbol(symbol): tid earnings_alignment.normalise_symbol(symbol): tid
for tid, symbol in rows for tid, symbol in rows
@@ -22,6 +22,7 @@ from app.models.fundamental_snapshot import FundamentalSnapshot
from app.models.ohlcv import OHLCVRecord from app.models.ohlcv import OHLCVRecord
from app.models.ticker import Ticker from app.models.ticker import Ticker
from app.services import fundamentals_derivation as deriv from app.services import fundamentals_derivation as deriv
from app.services import ticker_service
@dataclass(frozen=True) @dataclass(frozen=True)
@@ -46,7 +47,11 @@ async def build_candidates(
"""Derive current cache candidates using only already-stored data.""" """Derive current cache candidates using only already-stored data."""
today = today or datetime.now(ZoneInfo("America/New_York")).date() today = today or datetime.now(ZoneInfo("America/New_York")).date()
tickers = list( 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: if not tickers:
return [] return []
+4 -1
View File
@@ -18,6 +18,7 @@ from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.ext.asyncio import AsyncSession
from app.models.ticker import Ticker from app.models.ticker import Ticker
from app.services import ticker_service
from app.services.price_service import query_ohlcv from app.services.price_service import query_ohlcv
logger = logging.getLogger(__name__) 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 before scanning; the research backtest ranked each weekly setup-candidate
cross-section, so this is the deliberate production approximation. 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()) tickers = list(result.scalars().all())
benchmark_closes = await _load_activation_benchmark(db) benchmark_closes = await _load_activation_benchmark(db)
+6 -2
View File
@@ -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.price_service import query_ohlcv
from app.services.qualification import setup_qualifies from app.services.qualification import setup_qualifies
from app.services.sr_service import detect_gate_target_ladder 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 ( from app.services.trade_policy import (
MANUAL_BOOK, MANUAL_BOOK,
SHADOW_BOOK, SHADOW_BOOK,
@@ -735,7 +735,11 @@ async def scan_all_tickers(
# Plain ids/strings, not Ticker instances: the rollbacks below expire any # Plain ids/strings, not Ticker instances: the rollbacks below expire any
# ORM objects held across them, and touching an expired attribute afterwards # ORM objects held across them, and touching an expired attribute afterwards
# triggers sync lazy-loading, which raises on an AsyncSession. # 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()] ticker_rows = [(int(ticker_id), symbol) for ticker_id, symbol in result.all()]
total = len(ticker_rows) total = len(ticker_rows)
+7 -3
View File
@@ -20,7 +20,7 @@ from app.database import insert_for_session
from app.exceptions import NotFoundError, ValidationError from app.exceptions import NotFoundError, ValidationError
from app.models.score import CompositeScore, DimensionScore from app.models.score import CompositeScore, DimensionScore
from app.models.ticker import Ticker from app.models.ticker import Ticker
from app.services import settings_store from app.services import settings_store, ticker_service
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
@@ -883,7 +883,11 @@ async def get_rankings(db: AsyncSession) -> dict:
Returns dict suitable for RankingResponse. Returns dict suitable for RankingResponse.
""" """
weights = await _get_weights(db) 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]]]: async def _load_scores() -> tuple[dict[int, CompositeScore], dict[int, dict[str, DimensionScore]]]:
comps = { comps = {
@@ -947,7 +951,7 @@ async def update_weights(
await _save_weights(db, full_weights) await _save_weights(db, full_weights)
# Recompute all composite scores # 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()) tickers = list(result.scalars().all())
for ticker in tickers: for ticker in tickers:
+112
View File
@@ -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"}) _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): class SecError(ProviderError):
"""SEC request failed (403, exhausted 429/5xx, timeout, transport, parse).""" """SEC request failed (403, exhausted 429/5xx, timeout, transport, parse)."""
@@ -253,6 +280,91 @@ class SecClient:
"filings": filings, "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]: async def companyfacts(self, cik: int | str) -> dict[str, Any]:
"""Raw companyfacts JSON ({cik, entityName, facts}).""" """Raw companyfacts JSON ({cik, entityName, facts})."""
return await self.get_json(f"{_DATA}/api/xbrl/companyfacts/CIK{cik10(cik)}.json") return await self.get_json(f"{_DATA}/api/xbrl/companyfacts/CIK{cik10(cik)}.json")
+97 -2
View File
@@ -51,10 +51,10 @@ import json
import logging import logging
from collections import Counter, defaultdict from collections import Counter, defaultdict
from dataclasses import dataclass, field, replace 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 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.database import insert_for_session
from app.models.data_import_run import DataImportRun 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 # three); past that it is misfiled, not late, and blocking forever costs more
# than the missing filing does — see the unresolved-filing guardrail below. # than the missing filing does — see the unresolved-filing guardrail below.
MISSING_XBRL_RETRY_DAYS = 3 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 FILING_GAP_ESCALATE_DAYS = 14
# Share-count band a co-registrant-recovered row must land in, relative to the # 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 # 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._retry_rows: list[dict[str, Any]] = []
self._latest_index_date: date | None = None self._latest_index_date: date | None = None
self._backfill = False 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 ------------------------------------------- # -- SourceImporter protocol -------------------------------------------
@@ -421,6 +444,24 @@ class SecFundamentalsImporter:
# reconstructible by re-walking the index. # reconstructible by re-walking the index.
blocking = _within_retry_window(staged.missing_xbrl) blocking = _within_retry_window(staged.missing_xbrl)
aged_out = _past_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: if blocking:
messages.append( messages.append(
f"{len(blocking)} tracked XBRL filing(s) unresolved within the " f"{len(blocking)} tracked XBRL filing(s) unresolved within the "
@@ -464,6 +505,9 @@ class SecFundamentalsImporter:
"missing_xbrl": staged.missing_xbrl[:50], "missing_xbrl": staged.missing_xbrl[:50],
"missing_xbrl_count": len(staged.missing_xbrl), "missing_xbrl_count": len(staged.missing_xbrl),
"missing_xbrl_blocking": len(blocking), "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_from_coregistrant": staged.recovered[:50],
"recovered_count": len(staged.recovered), "recovered_count": len(staged.recovered),
# Complete compact gate input; detailed audit lists above stay capped. # Complete compact gate input; detailed audit lists above stay capped.
@@ -616,6 +660,25 @@ class SecFundamentalsImporter:
created_at=_now(), 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 # Persistent current gaps get one actionable escalation rather than a
# daily warning. The nullable marker makes this durable and noise-free. # daily warning. The nullable marker makes this durable and noise-free.
escalation_cutoff = now - timedelta(days=FILING_GAP_ESCALATE_DAYS) escalation_cutoff = now - timedelta(days=FILING_GAP_ESCALATE_DAYS)
@@ -757,6 +820,38 @@ class SecFundamentalsImporter:
if accession not in resolved 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: async def _last_processed_index_date(self, db) -> date | None:
return ( return (
await db.execute( await db.execute(
+6 -2
View File
@@ -24,7 +24,7 @@ from typing import Iterable
from sqlalchemy import select, update from sqlalchemy import select, update
from app.models.ticker import Ticker 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.earnings_alignment import normalise_symbol
from app.services.sec_client import SecClient 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.""" returns the mapping + proposed `tickers.cik` writes; mutates nothing."""
ticker_to_cik = await client.company_tickers() ticker_to_cik = await client.company_tickers()
overrides = await cik_overrides(db) 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() result = ResolvedUniverse()
for tid, symbol, current_cik in rows: for tid, symbol, current_cik in rows:
+199 -3
View File
@@ -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 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 sqlalchemy.ext.asyncio import AsyncSession
from app.exceptions import DuplicateError, NotFoundError, ValidationError from app.exceptions import DuplicateError, NotFoundError, ValidationError
from app.models.ticker import Ticker 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: async def add_ticker(db: AsyncSession, symbol: str) -> Ticker:
"""Add a new ticker after validation. """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]: 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())) result = await db.execute(select(Ticker).order_by(Ticker.symbol.asc()))
return list(result.scalars().all()) 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
+23 -2
View File
@@ -357,9 +357,26 @@ async def bootstrap_universe(
db.add(Ticker(symbol=symbol)) db.add(Ticker(symbol=symbol))
deleted_count = 0 deleted_count = 0
skipped_delisted: list[str] = []
if symbols_to_delete: 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
deleted_count = int(result.rowcount or 0) # 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() await db.commit()
@@ -378,4 +395,8 @@ async def bootstrap_universe(
"already_tracked": len(target_symbols & existing_symbols), "already_tracked": len(target_symbols & existing_symbols),
"deleted": deleted_count, "deleted": deleted_count,
"added_symbols": symbols_to_add[:50], "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),
} }
+4
View File
@@ -858,6 +858,10 @@ export interface Ticker {
symbol: string; symbol: string;
name: string | null; name: string | null;
created_at: string; 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 // Admin
+68
View File
@@ -1620,3 +1620,71 @@ async def test_run_backtest_smoke(session):
sweep = sorted(report["sweep"], key=lambda r: r["min_momentum_percentile"], reverse=True) sweep = sorted(report["sweep"], key=lambda r: r["min_momentum_percentile"], reverse=True)
counts = [r["total"] for r in sweep] counts = [r["total"] for r in sweep]
assert counts == sorted(counts) # ascending as threshold descends 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
+156 -1
View File
@@ -6,7 +6,7 @@ from __future__ import annotations
import json import json
import os import os
import tempfile import tempfile
from datetime import date, datetime, timezone from datetime import date, datetime, timedelta, timezone
import pytest import pytest
from sqlalchemy import func, select 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) resolved = await resolve_ciks(db, client)
assert resolved.symbol_to_cik["AAPL"] == 320193 # fell back to company_tickers 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
+416
View File
@@ -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)