Optimize signal read paths and enforce score invariants

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
@@ -407,17 +408,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 +478,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

@@ -13,6 +13,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
 from app.models.fundamental import FundamentalData
 from app.models.score import DimensionScore
@@ -67,7 +68,7 @@ async def store_fundamental(
         existing.unavailable_fields_json = unavailable_fields_json
         record = existing
     else:
-        record = FundamentalData(
+        stmt = insert_for_session(db, FundamentalData).values(
             ticker_id=ticker.id,
             pe_ratio=pe_ratio,
             revenue_growth=revenue_growth,
@@ -77,7 +78,24 @@ async def store_fundamental(
             fetched_at=now,
             unavailable_fields_json=unavailable_fields_json,
         )
-        db.add(record)
+        await db.execute(
+            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,
+                },
+            )
+        )
+        result = await db.execute(
+            select(FundamentalData).where(FundamentalData.ticker_id == ticker.id)
+        )
+        record = result.scalar_one()
 
     # Mark fundamental dimension score as stale if it exists
     # TODO: Use DimensionScore service when built

app/services/price_service.py

@@ -97,7 +97,13 @@ async def query_ohlcv(
     Returns records sorted by date ascending.
     Raises NotFoundError if the ticker does not exist.
     """
-    ticker = await _get_ticker(db, symbol)
+    normalised = symbol.strip().upper()
+    cache = db.info.get("ohlcv_cache")
+    cache_key = (normalised, start_date, end_date)
+    if cache is not None and cache_key in cache:
+        return list(cache[cache_key])
+
+    ticker = await _get_ticker(db, normalised)
 
     stmt = select(OHLCVRecord).where(OHLCVRecord.ticker_id == ticker.id)
     if start_date is not None:
@@ -107,4 +113,7 @@ async def query_ohlcv(
     stmt = stmt.order_by(OHLCVRecord.date.asc())
 
     result = await db.execute(stmt)
-    return list(result.scalars().all())
+    records = list(result.scalars().all())
+    if cache is not None:
+        cache[cache_key] = records
+    return list(records)

app/services/rr_scanner_service.py

@@ -556,6 +556,9 @@ async def scan_all_tickers(
     result = await db.execute(select(Ticker).order_by(Ticker.symbol))
     tickers = list(result.scalars().all())
     total = len(tickers)
+    # Ranking, score refresh, and setup detection repeatedly read the same
+    # immutable OHLCV series during one scan. Scope the cache to this run only.
+    db.info["ohlcv_cache"] = {}
 
     # Rank the universe up front so each new setup carries both the residual
     # activation gate percentile and the promoted production ordering score.
@@ -582,7 +585,6 @@ async def scan_all_tickers(
 
                 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)
 
@@ -596,6 +598,11 @@ async def scan_all_tickers(
         except Exception:
             logger.exception("Error scanning ticker %s", ticker.symbol)
 
+    # scan_ticker commits successful setup writes. This final commit persists
+    # refreshed scores for tickers that produced no setup or hit a scan error.
+    await db.commit()
+
+    db.info.pop("ohlcv_cache", None)
     if progress_callback is not None and total:
         progress_callback(total, 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
 
@@ -749,14 +759,24 @@ async def compute_composite_score(
         existing.weights_json = json.dumps(weights)
         existing.computed_at = now
     else:
-        comp = CompositeScore(
+        stmt = insert_for_session(db, CompositeScore).values(
             ticker_id=ticker.id,
             score=composite,
             is_stale=False,
             weights_json=json.dumps(weights),
             computed_at=now,
         )
-        db.add(comp)
+        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,
+                },
+            )
+        )
 
     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
@@ -185,6 +186,124 @@ async def _enrich_entry(
     }
 
 
+async def _enrich_entries(
+    db: AsyncSession,
+    rows: list[tuple[WatchlistEntry, str]],
+) -> list[dict]:
+    """Build watchlist rows from a fixed set of bulk lookups."""
+    if not rows:
+        return []
+
+    ticker_ids = [entry.ticker_id for entry, _ in rows]
+    comps_result = await db.execute(
+        select(CompositeScore).where(CompositeScore.ticker_id.in_(ticker_ids))
+    )
+    comps = {score.ticker_id: score for score in comps_result.scalars()}
+
+    dims_result = await db.execute(
+        select(DimensionScore).where(DimensionScore.ticker_id.in_(ticker_ids))
+    )
+    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}
+        )
+
+    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)
+        .join(ranked_setups, TradeSetup.id == ranked_setups.c.id)
+        .where(ranked_setups.c.rank == 1)
+    )
+    best_setups = {setup.ticker_id: setup for setup in setup_result.scalars()}
+
+    levels_result = await db.execute(
+        select(SRLevel)
+        .where(SRLevel.ticker_id.in_(ticker_ids))
+        .order_by(SRLevel.ticker_id, SRLevel.strength.desc())
+    )
+    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,
+            }
+        )
+
+    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()
+    )
+    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)
+    )
+    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))
+
+    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(
     db: AsyncSession,
     user_id: int,
@@ -203,10 +322,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":