Optimize signal read paths and enforce score invariants

alembic/versions/019_enforce_score_snapshot_invariants.py

@@ -0,0 +1,61 @@
+"""Enforce singleton score and fundamental snapshots.
+
+Revision ID: 019
+Revises: 018
+Create Date: 2026-07-11 00:00:00.000000
+"""
+
+from __future__ import annotations
+
+from alembic import op
+import sqlalchemy as sa
+
+
+revision = "019"
+down_revision = "018"
+branch_labels = None
+depends_on = None
+
+
+def _remove_duplicates(table: str, partition_by: str, order_by: str) -> None:
+    op.execute(
+        sa.text(
+            f"""
+            DELETE FROM {table}
+            WHERE id IN (
+                SELECT id FROM (
+                    SELECT id, ROW_NUMBER() OVER (
+                        PARTITION BY {partition_by}
+                        ORDER BY {order_by} DESC, id DESC
+                    ) AS row_number
+                    FROM {table}
+                ) AS ranked
+                WHERE row_number > 1
+            )
+            """
+        )
+    )
+
+
+def upgrade() -> None:
+    _remove_duplicates("dimension_scores", "ticker_id, dimension", "computed_at")
+    _remove_duplicates("composite_scores", "ticker_id", "computed_at")
+    _remove_duplicates("fundamental_data", "ticker_id", "fetched_at")
+
+    op.create_unique_constraint(
+        "uq_dimension_score_ticker_dimension",
+        "dimension_scores",
+        ["ticker_id", "dimension"],
+    )
+    op.create_unique_constraint("uq_composite_score_ticker", "composite_scores", ["ticker_id"])
+    op.create_unique_constraint("uq_fundamental_data_ticker", "fundamental_data", ["ticker_id"])
+    op.create_index("ix_sr_levels_ticker_id", "sr_levels", ["ticker_id"])
+    op.create_index("ix_trade_setups_ticker_rr", "trade_setups", ["ticker_id", "rr_ratio"])
+
+
+def downgrade() -> None:
+    op.drop_index("ix_trade_setups_ticker_rr", table_name="trade_setups")
+    op.drop_index("ix_sr_levels_ticker_id", table_name="sr_levels")
+    op.drop_constraint("uq_fundamental_data_ticker", "fundamental_data", type_="unique")
+    op.drop_constraint("uq_composite_score_ticker", "composite_scores", type_="unique")
+    op.drop_constraint("uq_dimension_score_ticker_dimension", "dimension_scores", type_="unique")

app/database.py

@@ -1,5 +1,8 @@
 from collections.abc import AsyncGenerator
+from typing import Any
 
+from sqlalchemy.dialects.postgresql import insert as postgresql_insert
+from sqlalchemy.dialects.sqlite import insert as sqlite_insert
 from sqlalchemy.ext.asyncio import (
     AsyncSession,
     async_sessionmaker,
@@ -28,6 +31,13 @@ class Base(DeclarativeBase):
     pass
 
 
+def insert_for_session(session: AsyncSession, table: Any) -> Any:
+    """Build a dialect-native INSERT that supports conflict handling."""
+    if session.get_bind().dialect.name == "postgresql":
+        return postgresql_insert(table)
+    return sqlite_insert(table)
+
+
 async def get_session() -> AsyncGenerator[AsyncSession, None]:
     async with async_session_factory() as session:
         yield session

app/models/fundamental.py

@@ -1,6 +1,6 @@
 from datetime import date, datetime
 
-from sqlalchemy import Date, DateTime, Float, ForeignKey, Text
+from sqlalchemy import Date, DateTime, Float, ForeignKey, Text, UniqueConstraint
 from sqlalchemy.orm import Mapped, mapped_column, relationship
 
 from app.database import Base
@@ -8,6 +8,9 @@ from app.database import Base
 
 class FundamentalData(Base):
     __tablename__ = "fundamental_data"
+    __table_args__ = (
+        UniqueConstraint("ticker_id", name="uq_fundamental_data_ticker"),
+    )
 
     id: Mapped[int] = mapped_column(primary_key=True)
     ticker_id: Mapped[int] = mapped_column(

app/models/score.py

@@ -1,6 +1,6 @@
 from datetime import datetime
 
-from sqlalchemy import Boolean, DateTime, Float, ForeignKey, String, Text
+from sqlalchemy import Boolean, DateTime, Float, ForeignKey, String, Text, UniqueConstraint
 from sqlalchemy.orm import Mapped, mapped_column, relationship
 
 from app.database import Base
@@ -8,6 +8,9 @@ from app.database import Base
 
 class DimensionScore(Base):
     __tablename__ = "dimension_scores"
+    __table_args__ = (
+        UniqueConstraint("ticker_id", "dimension", name="uq_dimension_score_ticker_dimension"),
+    )
 
     id: Mapped[int] = mapped_column(primary_key=True)
     ticker_id: Mapped[int] = mapped_column(
@@ -25,6 +28,9 @@ class DimensionScore(Base):
 
 class CompositeScore(Base):
     __tablename__ = "composite_scores"
+    __table_args__ = (
+        UniqueConstraint("ticker_id", name="uq_composite_score_ticker"),
+    )
 
     id: Mapped[int] = mapped_column(primary_key=True)
     ticker_id: Mapped[int] = mapped_column(

app/models/sr_level.py

@@ -1,6 +1,6 @@
 from datetime import datetime
 
-from sqlalchemy import DateTime, Float, ForeignKey, Integer, String
+from sqlalchemy import DateTime, Float, ForeignKey, Index, Integer, String
 from sqlalchemy.orm import Mapped, mapped_column, relationship
 
 from app.database import Base
@@ -8,6 +8,7 @@ from app.database import Base
 
 class SRLevel(Base):
     __tablename__ = "sr_levels"
+    __table_args__ = (Index("ix_sr_levels_ticker_id", "ticker_id"),)
 
     id: Mapped[int] = mapped_column(primary_key=True)
     ticker_id: Mapped[int] = mapped_column(

app/models/trade_setup.py

@@ -2,7 +2,7 @@ from datetime import date, datetime
 
 import json
 
-from sqlalchemy import Date, DateTime, Float, ForeignKey, String, Text
+from sqlalchemy import Date, DateTime, Float, ForeignKey, Index, String, Text
 from sqlalchemy.orm import Mapped, mapped_column, relationship
 
 from app.database import Base
@@ -10,6 +10,7 @@ from app.database import Base
 
 class TradeSetup(Base):
     __tablename__ = "trade_setups"
+    __table_args__ = (Index("ix_trade_setups_ticker_rr", "ticker_id", "rr_ratio"),)
 
     id: Mapped[int] = mapped_column(primary_key=True)
     ticker_id: Mapped[int] = mapped_column(

app/services/alert_service.py

@@ -17,11 +17,12 @@ from __future__ import annotations
 
 import logging
 import math
+from collections import defaultdict
 from datetime import datetime, timedelta, timezone
 from types import SimpleNamespace
 
 import httpx
-from sqlalchemy import select
+from sqlalchemy import func, select
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from app.config import settings
@@ -257,17 +258,6 @@ def _log_alert(db: AsyncSession, alert_type: str, key: str, value: float | None
     )
 
 
-async def _watermark(db: AsyncSession, symbol: str) -> float | None:
-    result = await db.execute(
-        select(AlertLog.value)
-        .where(AlertLog.alert_type == WATERMARK_TYPE, AlertLog.dedup_key == symbol)
-        .order_by(AlertLog.created_at.desc())
-        .limit(1)
-    )
-    row = result.first()
-    return row[0] if row else None
-
-
 # ---------------------------------------------------------------------------
 # Trigger collectors
 # ---------------------------------------------------------------------------
@@ -407,17 +397,49 @@ async def _collect_sr_proximity(db: AsyncSession) -> list[tuple[str, str]]:
     single alert. Scoped to the watchlist only — qualified tickers already get
     their own 'qualified setup' alert, so S/R on them would be redundant.
     """
+    watchlist = await _watchlist_tickers(db)
+    if not watchlist:
+        return []
+
+    ticker_ids = [ticker_id for ticker_id, _ in watchlist]
+    latest_dates = (
+        select(
+            OHLCVRecord.ticker_id,
+            func.max(OHLCVRecord.date).label("latest_date"),
+        )
+        .where(OHLCVRecord.ticker_id.in_(ticker_ids))
+        .group_by(OHLCVRecord.ticker_id)
+        .subquery()
+    )
+    prices_result = await db.execute(
+        select(OHLCVRecord.ticker_id, OHLCVRecord.close).join(
+            latest_dates,
+            (OHLCVRecord.ticker_id == latest_dates.c.ticker_id)
+            & (OHLCVRecord.date == latest_dates.c.latest_date),
+        )
+    )
+    prices = {ticker_id: float(close) for ticker_id, close in prices_result.all()}
+
+    levels_result = await db.execute(
+        select(SRLevel).where(SRLevel.ticker_id.in_(ticker_ids))
+    )
+    levels_by_ticker: dict[int, list[dict]] = defaultdict(list)
+    for level in levels_result.scalars():
+        levels_by_ticker[level.ticker_id].append(
+            {
+                "price_level": level.price_level,
+                "strength": level.strength,
+                "type": level.type,
+            }
+        )
+
     out: list[tuple[str, str]] = []
-    for tid, symbol in await _watchlist_tickers(db):
-        price = await _latest_close(db, tid)
+    for tid, symbol in watchlist:
+        price = prices.get(tid)
         if not price:
             continue
 
-        levels_result = await db.execute(select(SRLevel).where(SRLevel.ticker_id == tid))
-        levels = [
-            {"price_level": lv.price_level, "strength": lv.strength, "type": lv.type}
-            for lv in levels_result.scalars().all()
-        ]
+        levels = levels_by_ticker[tid]
         if not levels:
             continue
 
@@ -445,17 +467,54 @@ async def _collect_score_drops(db: AsyncSession) -> list[tuple[str, str]]:
     doesn't re-fire; let the watermark rise with the score so the next drop is
     measured from the new high.
     """
-    out: list[tuple[str, str]] = []
-    for tid, symbol in await _watchlist_tickers(db):
-        comp_result = await db.execute(
-            select(CompositeScore.score).where(CompositeScore.ticker_id == tid)
-        )
-        row = comp_result.first()
-        if row is None or row[0] is None:
-            continue
-        current = float(row[0])
+    watchlist = await _watchlist_tickers(db)
+    if not watchlist:
+        return []
 
-        base = await _watermark(db, symbol)
+    ticker_ids = [ticker_id for ticker_id, _ in watchlist]
+    symbols = [symbol for _, symbol in watchlist]
+    scores_result = await db.execute(
+        select(CompositeScore.ticker_id, CompositeScore.score).where(
+            CompositeScore.ticker_id.in_(ticker_ids)
+        )
+    )
+    scores = {ticker_id: float(score) for ticker_id, score in scores_result.all()}
+
+    ranked_watermarks = (
+        select(
+            AlertLog.dedup_key,
+            AlertLog.value,
+            func.row_number()
+            .over(
+                partition_by=AlertLog.dedup_key,
+                order_by=(AlertLog.created_at.desc(), AlertLog.id.desc()),
+            )
+            .label("rank"),
+        )
+        .where(
+            AlertLog.alert_type == WATERMARK_TYPE,
+            AlertLog.dedup_key.in_(symbols),
+        )
+        .subquery()
+    )
+    watermarks_result = await db.execute(
+        select(ranked_watermarks.c.dedup_key, ranked_watermarks.c.value).where(
+            ranked_watermarks.c.rank == 1
+        )
+    )
+    watermarks = {
+        symbol: float(value)
+        for symbol, value in watermarks_result.all()
+        if value is not None
+    }
+
+    out: list[tuple[str, str]] = []
+    for tid, symbol in watchlist:
+        current = scores.get(tid)
+        if current is None:
+            continue
+
+        base = watermarks.get(symbol)
         if base is None:
             _log_alert(db, WATERMARK_TYPE, symbol, value=current)  # seed, no alert
             continue

app/services/fundamental_service.py

@@ -10,9 +10,10 @@ import json
 import logging
 from datetime import datetime, timezone
 
-from sqlalchemy import select
+from sqlalchemy import select, update
 from sqlalchemy.ext.asyncio import AsyncSession
 
+from app.database import insert_for_session
 from app.exceptions import NotFoundError
 from app.models.fundamental import FundamentalData
 from app.models.score import DimensionScore
@@ -48,51 +49,45 @@ async def store_fundamental(
     """
     ticker = await _get_ticker(db, symbol)
 
-    # Check for existing record
-    result = await db.execute(
-        select(FundamentalData).where(FundamentalData.ticker_id == ticker.id)
-    )
-    existing = result.scalar_one_or_none()
-
     now = datetime.now(timezone.utc)
     unavailable_fields_json = json.dumps(unavailable_fields or {})
 
-    if existing is not None:
-        existing.pe_ratio = pe_ratio
-        existing.revenue_growth = revenue_growth
-        existing.earnings_surprise = earnings_surprise
-        existing.market_cap = market_cap
-        existing.next_earnings_date = next_earnings_date
-        existing.fetched_at = now
-        existing.unavailable_fields_json = unavailable_fields_json
-        record = existing
-    else:
-        record = FundamentalData(
-            ticker_id=ticker.id,
-            pe_ratio=pe_ratio,
-            revenue_growth=revenue_growth,
-            earnings_surprise=earnings_surprise,
-            market_cap=market_cap,
-            next_earnings_date=next_earnings_date,
-            fetched_at=now,
-            unavailable_fields_json=unavailable_fields_json,
-        )
-        db.add(record)
+    stmt = insert_for_session(db, FundamentalData).values(
+        ticker_id=ticker.id,
+        pe_ratio=pe_ratio,
+        revenue_growth=revenue_growth,
+        earnings_surprise=earnings_surprise,
+        market_cap=market_cap,
+        next_earnings_date=next_earnings_date,
+        fetched_at=now,
+        unavailable_fields_json=unavailable_fields_json,
+    )
+    stmt = stmt.on_conflict_do_update(
+        index_elements=["ticker_id"],
+        set_={
+            "pe_ratio": stmt.excluded.pe_ratio,
+            "revenue_growth": stmt.excluded.revenue_growth,
+            "earnings_surprise": stmt.excluded.earnings_surprise,
+            "market_cap": stmt.excluded.market_cap,
+            "next_earnings_date": stmt.excluded.next_earnings_date,
+            "fetched_at": stmt.excluded.fetched_at,
+            "unavailable_fields_json": stmt.excluded.unavailable_fields_json,
+        },
+    ).returning(FundamentalData)
+    record = (await db.execute(stmt)).scalar_one()
 
     # Mark fundamental dimension score as stale if it exists
     # TODO: Use DimensionScore service when built
-    dim_result = await db.execute(
-        select(DimensionScore).where(
+    await db.execute(
+        update(DimensionScore)
+        .where(
             DimensionScore.ticker_id == ticker.id,
             DimensionScore.dimension == "fundamental",
         )
+        .values(is_stale=True)
     )
-    dim_score = dim_result.scalar_one_or_none()
-    if dim_score is not None:
-        dim_score.is_stale = True
 
     await db.commit()
-    await db.refresh(record)
     return record
 
 

app/services/price_service.py

@@ -3,9 +3,9 @@
 from datetime import date, datetime
 
 from sqlalchemy import select
-from sqlalchemy.dialects.postgresql import insert as pg_insert
 from sqlalchemy.ext.asyncio import AsyncSession
 
+from app.database import insert_for_session
 from app.exceptions import NotFoundError, ValidationError
 from app.models.ohlcv import OHLCVRecord
 from app.models.ticker import Ticker
@@ -53,7 +53,7 @@ async def upsert_ohlcv(
     _validate_ohlcv(high, low, open_, close, volume, record_date)
     ticker = await _get_ticker(db, symbol)
 
-    stmt = pg_insert(OHLCVRecord).values(
+    stmt = insert_for_session(db, OHLCVRecord).values(
         ticker_id=ticker.id,
         date=record_date,
         open=open_,
@@ -64,7 +64,7 @@ async def upsert_ohlcv(
         created_at=datetime.utcnow(),
     )
     stmt = stmt.on_conflict_do_update(
-        constraint="uq_ohlcv_ticker_date",
+        index_elements=["ticker_id", "date"],
         set_={
             "open": stmt.excluded.open,
             "high": stmt.excluded.high,

app/services/rr_scanner_service.py

@@ -566,6 +566,7 @@ async def scan_all_tickers(
 
         ranks = await momentum_service.compute_activation_ranks(db)
     except Exception:
+        await db.rollback()
         logger.exception("Activation ranking refresh failed")
         ranks = {}
 
@@ -573,19 +574,21 @@ async def scan_all_tickers(
     for index, ticker in enumerate(tickers):
         if progress_callback is not None:
             progress_callback(index, total, ticker.symbol)
+        # Refresh scores first so the scheduled scan works off current data.
+        # Nothing else marks scores stale, so without this they'd never update
+        # for tickers the user doesn't manually fetch.
         try:
-            # Refresh scores first so the scheduled scan works off current data.
-            # Nothing else marks scores stale, so without this they'd never
-            # update for tickers the user doesn't manually fetch.
-            try:
-                from app.services import scoring_service
+            from app.services import scoring_service
 
-                await scoring_service.compute_all_dimensions(db, ticker.symbol)
-                await scoring_service.compute_composite_score(db, ticker.symbol)
-                await db.commit()
-            except Exception:
-                logger.exception("Error refreshing scores for %s", ticker.symbol)
+            await scoring_service.compute_all_dimensions(db, ticker.symbol)
+            await scoring_service.compute_composite_score(db, ticker.symbol)
+            await db.commit()
+        except Exception:
+            await db.rollback()
+            logger.exception("Error refreshing scores for %s", ticker.symbol)
+            continue
 
+        try:
             setups = await scan_ticker(
                 db, ticker.symbol, rr_threshold, atr_multiplier,
                 momentum_percentile=(ranks.get(ticker.symbol) or {}).get("momentum_percentile"),
@@ -594,6 +597,7 @@ async def scan_all_tickers(
             )
             all_setups.extend(setups)
         except Exception:
+            await db.rollback()
             logger.exception("Error scanning ticker %s", ticker.symbol)
 
     if progress_callback is not None and total:

app/services/scoring_service.py

@@ -16,6 +16,7 @@ from datetime import datetime, timezone
 from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
 
+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
@@ -661,14 +662,23 @@ async def compute_dimension_score(
             # Can't compute — mark stale
             existing.is_stale = True
     elif score_val is not None:
-        dim = DimensionScore(
+        stmt = insert_for_session(db, DimensionScore).values(
             ticker_id=ticker.id,
             dimension=dimension,
             score=score_val,
             is_stale=False,
             computed_at=now,
         )
-        db.add(dim)
+        await db.execute(
+            stmt.on_conflict_do_update(
+                index_elements=["ticker_id", "dimension"],
+                set_={
+                    "score": stmt.excluded.score,
+                    "is_stale": False,
+                    "computed_at": stmt.excluded.computed_at,
+                },
+            )
+        )
 
     return score_val
 
@@ -738,25 +748,24 @@ async def compute_composite_score(
 
     # Persist composite score
     now = datetime.now(timezone.utc)
-    comp_result = await db.execute(
-        select(CompositeScore).where(CompositeScore.ticker_id == ticker.id)
+    stmt = insert_for_session(db, CompositeScore).values(
+        ticker_id=ticker.id,
+        score=composite,
+        is_stale=False,
+        weights_json=json.dumps(weights),
+        computed_at=now,
     )
-    existing = comp_result.scalar_one_or_none()
-
-    if existing is not None:
-        existing.score = composite
-        existing.is_stale = False
-        existing.weights_json = json.dumps(weights)
-        existing.computed_at = now
-    else:
-        comp = CompositeScore(
-            ticker_id=ticker.id,
-            score=composite,
-            is_stale=False,
-            weights_json=json.dumps(weights),
-            computed_at=now,
+    await db.execute(
+        stmt.on_conflict_do_update(
+            index_elements=["ticker_id"],
+            set_={
+                "score": stmt.excluded.score,
+                "is_stale": False,
+                "weights_json": stmt.excluded.weights_json,
+                "computed_at": stmt.excluded.computed_at,
+            },
         )
-        db.add(comp)
+    )
 
     return composite, missing
 

app/services/watchlist_service.py

@@ -8,6 +8,7 @@ best trade setup, active S/R levels, and latest price + day-over-day move.
 from __future__ import annotations
 
 import logging
+from collections import defaultdict
 from datetime import datetime, timezone
 
 from sqlalchemy import func, select
@@ -102,87 +103,122 @@ async def remove_entry(
     await db.commit()
 
 
-async def _enrich_entry(
+async def _enrich_entries(
     db: AsyncSession,
-    entry: WatchlistEntry,
-    symbol: str,
-) -> dict:
-    """Build enriched watchlist entry dict with scores, R:R, SR levels, price."""
-    ticker_id = entry.ticker_id
+    rows: list[tuple[WatchlistEntry, str]],
+) -> list[dict]:
+    """Build watchlist rows from a fixed set of bulk lookups."""
+    if not rows:
+        return []
 
-    # Composite score
-    comp_result = await db.execute(
-        select(CompositeScore).where(CompositeScore.ticker_id == ticker_id)
+    ticker_ids = [entry.ticker_id for entry, _ in rows]
+    comps_result = await db.execute(
+        select(CompositeScore).where(CompositeScore.ticker_id.in_(ticker_ids))
     )
-    comp = comp_result.scalar_one_or_none()
+    comps = {score.ticker_id: score for score in comps_result.scalars()}
 
-    # Dimension scores
-    dim_result = await db.execute(
-        select(DimensionScore).where(DimensionScore.ticker_id == ticker_id)
+    dims_result = await db.execute(
+        select(DimensionScore).where(DimensionScore.ticker_id.in_(ticker_ids))
     )
-    dims = [
-        {"dimension": ds.dimension, "score": ds.score}
-        for ds in dim_result.scalars().all()
-    ]
+    dims_by_ticker: dict[int, list[dict]] = defaultdict(list)
+    for score in dims_result.scalars():
+        dims_by_ticker[score.ticker_id].append(
+            {"dimension": score.dimension, "score": score.score}
+        )
 
-    # Best trade setup (highest R:R) for this ticker
+    ranked_setups = (
+        select(
+            TradeSetup.id,
+            func.row_number()
+            .over(
+                partition_by=TradeSetup.ticker_id,
+                order_by=TradeSetup.rr_ratio.desc(),
+            )
+            .label("rank"),
+        )
+        .where(TradeSetup.ticker_id.in_(ticker_ids))
+        .subquery()
+    )
     setup_result = await db.execute(
         select(TradeSetup)
-        .where(TradeSetup.ticker_id == ticker_id)
-        .order_by(TradeSetup.rr_ratio.desc())
-        .limit(1)
+        .join(ranked_setups, TradeSetup.id == ranked_setups.c.id)
+        .where(ranked_setups.c.rank == 1)
     )
-    setup = setup_result.scalar_one_or_none()
+    best_setups = {setup.ticker_id: setup for setup in setup_result.scalars()}
 
-    # Active SR levels
-    sr_result = await db.execute(
+    levels_result = await db.execute(
         select(SRLevel)
-        .where(SRLevel.ticker_id == ticker_id)
-        .order_by(SRLevel.strength.desc())
+        .where(SRLevel.ticker_id.in_(ticker_ids))
+        .order_by(SRLevel.ticker_id, SRLevel.strength.desc())
     )
-    sr_levels = [
-        {
-            "price_level": lv.price_level,
-            "type": lv.type,
-            "strength": lv.strength,
-        }
-        for lv in sr_result.scalars().all()
-    ]
+    levels_by_ticker: dict[int, list[dict]] = defaultdict(list)
+    for level in levels_result.scalars():
+        levels_by_ticker[level.ticker_id].append(
+            {
+                "price_level": level.price_level,
+                "type": level.type,
+                "strength": level.strength,
+            }
+        )
 
-    # Latest two daily closes → current price + day-over-day move
-    price_result = await db.execute(
-        select(OHLCVRecord.close, OHLCVRecord.date)
-        .where(OHLCVRecord.ticker_id == ticker_id)
-        .order_by(OHLCVRecord.date.desc())
-        .limit(2)
+    ranked_prices = (
+        select(
+            OHLCVRecord.ticker_id,
+            OHLCVRecord.close,
+            OHLCVRecord.date,
+            func.row_number()
+            .over(
+                partition_by=OHLCVRecord.ticker_id,
+                order_by=OHLCVRecord.date.desc(),
+            )
+            .label("rank"),
+        )
+        .where(OHLCVRecord.ticker_id.in_(ticker_ids))
+        .subquery()
     )
-    bars = price_result.all()
-    last_close = bars[0].close if bars else None
-    prev_close = bars[1].close if len(bars) > 1 else None
-    change_pct = (
-        (last_close - prev_close) / prev_close * 100
-        if last_close is not None and prev_close
-        else None
+    prices_result = await db.execute(
+        select(
+            ranked_prices.c.ticker_id,
+            ranked_prices.c.close,
+            ranked_prices.c.date,
+        )
+        .where(ranked_prices.c.rank <= 2)
+        .order_by(ranked_prices.c.ticker_id, ranked_prices.c.rank)
     )
-    price_date = bars[0].date if bars else None
+    prices_by_ticker: dict[int, list[tuple[float, datetime]]] = defaultdict(list)
+    for ticker_id, close, price_date in prices_result.all():
+        prices_by_ticker[ticker_id].append((close, price_date))
 
-    return {
-        "symbol": symbol,
-        "entry_type": entry.entry_type,
-        "composite_score": comp.score if comp else None,
-        "dimensions": dims,
-        "rr_ratio": setup.rr_ratio if setup else None,
-        "rr_direction": setup.direction if setup else None,
-        # Residual 12-1 activation percentile gates qualification; strategy_rank
-        # is the promoted top-pick ordering score.
-        "momentum_percentile": setup.momentum_percentile if setup else None,
-        "strategy_rank": setup.strategy_rank if setup else None,
-        "sr_levels": sr_levels,
-        "last_close": last_close,
-        "change_pct": change_pct,
-        "price_date": price_date,
-        "added_at": entry.added_at,
-    }
+    entries: list[dict] = []
+    for entry, symbol in rows:
+        ticker_id = entry.ticker_id
+        comp = comps.get(ticker_id)
+        setup = best_setups.get(ticker_id)
+        bars = prices_by_ticker[ticker_id]
+        last_close = bars[0][0] if bars else None
+        prev_close = bars[1][0] if len(bars) > 1 else None
+        entries.append(
+            {
+                "symbol": symbol,
+                "entry_type": entry.entry_type,
+                "composite_score": comp.score if comp else None,
+                "dimensions": dims_by_ticker[ticker_id],
+                "rr_ratio": setup.rr_ratio if setup else None,
+                "rr_direction": setup.direction if setup else None,
+                "momentum_percentile": setup.momentum_percentile if setup else None,
+                "strategy_rank": setup.strategy_rank if setup else None,
+                "sr_levels": levels_by_ticker[ticker_id],
+                "last_close": last_close,
+                "change_pct": (
+                    (last_close - prev_close) / prev_close * 100
+                    if last_close is not None and prev_close
+                    else None
+                ),
+                "price_date": bars[0][1] if bars else None,
+                "added_at": entry.added_at,
+            }
+        )
+    return entries
 
 
 async def get_watchlist(
@@ -203,10 +239,7 @@ async def get_watchlist(
     result = await db.execute(stmt)
     rows = result.all()
 
-    entries: list[dict] = []
-    for entry, symbol in rows:
-        enriched = await _enrich_entry(db, entry, symbol)
-        entries.append(enriched)
+    entries = await _enrich_entries(db, rows)
 
     # Sort
     if sort_by == "composite":

tests/unit/test_alert_service.py

@@ -95,7 +95,17 @@ async def test_score_drop_seeds_then_alerts(session):
     msgs = await svc._collect_score_drops(session)
     await session.commit()
     assert msgs == []
-    assert await svc._watermark(session, "AAA") == 80.0
+    watermarks = (
+        await session.execute(
+            select(AlertLog.value)
+            .where(
+                AlertLog.alert_type == svc.WATERMARK_TYPE,
+                AlertLog.dedup_key == "AAA",
+            )
+            .order_by(AlertLog.created_at.desc(), AlertLog.id.desc())
+        )
+    ).scalars().all()
+    assert watermarks == [80.0]
 
     # Drop the composite well past the threshold
     row = (await session.execute(
@@ -111,7 +121,71 @@ async def test_score_drop_seeds_then_alerts(session):
     assert key == "scoredrop:AAA"
     assert "AAA" in text
     # rebaselined to the new (lower) level
-    assert await svc._watermark(session, "AAA") == 60.0
+    watermarks = (
+        await session.execute(
+            select(AlertLog.value)
+            .where(
+                AlertLog.alert_type == svc.WATERMARK_TYPE,
+                AlertLog.dedup_key == "AAA",
+            )
+            .order_by(AlertLog.created_at.desc(), AlertLog.id.desc())
+        )
+    ).scalars().all()
+    assert watermarks[0] == 60.0
+
+
+async def test_score_drop_uses_latest_watermark_when_timestamps_tie(session):
+    await _seed_watchlisted_ticker(session, "AAA", 50.0)
+    timestamp = datetime.now(timezone.utc)
+    session.add_all([
+        AlertLog(
+            alert_type=svc.WATERMARK_TYPE,
+            dedup_key="AAA",
+            value=90.0,
+            created_at=timestamp,
+        ),
+        AlertLog(
+            alert_type=svc.WATERMARK_TYPE,
+            dedup_key="AAA",
+            value=70.0,
+            created_at=timestamp,
+        ),
+    ])
+    await session.commit()
+
+    msgs = await svc._collect_score_drops(session)
+
+    assert len(msgs) == 1
+    assert "from 70" in msgs[0][1]
+
+
+async def test_score_drop_seeds_when_latest_watermark_has_no_value(session):
+    await _seed_watchlisted_ticker(session, "AAA", 80.0)
+    session.add(
+        AlertLog(
+            alert_type=svc.WATERMARK_TYPE,
+            dedup_key="AAA",
+            value=None,
+            created_at=datetime.now(timezone.utc),
+        )
+    )
+    await session.commit()
+
+    msgs = await svc._collect_score_drops(session)
+    await session.commit()
+
+    assert msgs == []
+    values = (
+        await session.execute(
+            select(AlertLog.value)
+            .where(
+                AlertLog.alert_type == svc.WATERMARK_TYPE,
+                AlertLog.dedup_key == "AAA",
+            )
+            .order_by(AlertLog.created_at.desc(), AlertLog.id.desc())
+        )
+    ).scalars().all()
+    assert values[0] == 80.0
 
 
 def test_format_qualified_includes_current_price_and_target_move():