feat: always-fresh sentiment for top picks, watchlist & open trades

Tiered, uncapped sentiment scope so the names that matter are never shown
without sentiment.

- Priority (always fully refreshed): top-pick feeders — momentum leaders with a
  tradeable long setup over the R:R floor (the tickers that are, or could become
  with positive sentiment, the dashboard top pick) — plus the curated watchlist
  and open paper trades.
- Filler: top-N by composite, a discovery net, fetched after the priority set so
  a mid-run rate limit lands the important names first.
- Removed the per-run cap (sentiment_max_per_run): the relevant set is naturally
  bounded (watchlist <= 20, composite <= top_composite), so a full refresh stays
  inside the free tier. extra="ignore" keeps a stale env var from breaking startup.
- Refresh window 72h -> 120h (5 days): sentiment shifts slowly, score window is 7d.

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

app/scheduler.py

@@ -20,7 +20,7 @@ from datetime import date, datetime, timedelta, timezone
 
 from apscheduler.schedulers.asyncio import AsyncIOScheduler
 from apscheduler.triggers.cron import CronTrigger
-from sqlalchemy import case, func, or_, select
+from sqlalchemy import and_, case, func, or_, select
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from app.config import settings
@@ -218,78 +218,141 @@ async def _get_ohlcv_priority_tickers(db: AsyncSession) -> list[str]:
     return list(result.scalars().all())
 
 
-async def _get_sentiment_priority_tickers(db: AsyncSession) -> list[str]:
-    """Symbols to fetch sentiment for, budgeted to stay in the free search tier.
+async def _get_top_pick_feeder_ids(db: AsyncSession) -> set[int]:
+    """Ticker ids whose latest LONG setup makes them a top-pick feeder.
 
-    Scope: only tickers that matter — watchlist + open paper trades + top-N by
-    composite score + the momentum leaders the activation gate qualifies on. Skip
-    any refreshed within ``sentiment_fresh_hours``. Cap the run at
-    ``sentiment_max_per_run``, oldest/missing first. Once the relevant set is
-    fresh, runs make zero grounded searches until it ages out.
+    A dashboard 'top pick' is the highest-momentum *qualified* setup. Sentiment
+    can never move a ticker's momentum percentile (the gate's core axis) — only
+    its confidence and EV ranking. So the only tickers that are, or could become
+    with positive sentiment, a top pick are momentum leaders that already have a
+    tradeable long setup clearing the R:R floor. That set is exactly:
+
+        latest long setup with momentum_percentile >= gate AND rr_ratio >= floor.
+
+    It contains both the currently-qualified setups and the near-miss ones held
+    back only by a neutral/missing sentiment — the cases the user saw surface as
+    top picks with no sentiment. Only meaningful with the momentum gate on
+    (min_momentum_percentile > 0); off, there is no leader axis to anchor on and we
+    defer to the filler set. Best-effort: a config failure must not stop collection.
+    """
+    from app.models.trade_setup import TradeSetup
+
+    try:
+        from app.services.admin_service import get_activation_config
+
+        activation = await get_activation_config(db)
+        min_pct = float(activation.get("min_momentum_percentile", 0.0))
+        min_rr = float(activation.get("min_rr", 0.0))
+    except Exception:
+        logger.exception("Sentiment top-pick scoping failed; using filler set only")
+        return set()
+
+    if min_pct <= 0:
+        return set()
+
+    # Latest long setup per ticker, then keep those clearing the gate's momentum
+    # percentile and R:R floor. (Sentiment runs before the day's scan, so this
+    # reads the previous scan's setups — momentum is a slow, cross-sectional signal,
+    # so yesterday's leaders are the right anchor.)
+    latest_long = (
+        select(TradeSetup.ticker_id, func.max(TradeSetup.detected_at).label("md"))
+        .where(TradeSetup.direction == "long")
+        .group_by(TradeSetup.ticker_id)
+        .subquery()
+    )
+    rows = await db.execute(
+        select(TradeSetup.ticker_id)
+        .join(
+            latest_long,
+            and_(
+                TradeSetup.ticker_id == latest_long.c.ticker_id,
+                TradeSetup.detected_at == latest_long.c.md,
+            ),
+        )
+        .where(
+            TradeSetup.direction == "long",
+            TradeSetup.rr_ratio >= min_rr,
+            TradeSetup.momentum_percentile.is_not(None),
+            TradeSetup.momentum_percentile >= min_pct,
+        )
+    )
+    return {r[0] for r in rows.all()}
+
+
+async def _stale_sentiment_symbols(
+    db: AsyncSession, ticker_ids: set[int], cutoff: datetime
+) -> list[str]:
+    """Symbols among ``ticker_ids`` whose newest sentiment is missing or older than
+    ``cutoff``, ordered missing-first → oldest → alphabetical."""
+    if not ticker_ids:
+        return []
+    latest_ts = func.max(SentimentScore.timestamp)
+    missing_first = case((latest_ts.is_(None), 0), else_=1)
+    stmt = (
+        select(Ticker.symbol)
+        .outerjoin(SentimentScore, SentimentScore.ticker_id == Ticker.id)
+        .where(Ticker.id.in_(ticker_ids))
+        .group_by(Ticker.id, Ticker.symbol)
+        .having(or_(latest_ts.is_(None), latest_ts < cutoff))
+        .order_by(missing_first.asc(), latest_ts.asc(), Ticker.symbol.asc())
+    )
+    result = await db.execute(stmt)
+    return list(result.scalars().all())
+
+
+async def _get_sentiment_priority_tickers(db: AsyncSession) -> list[str]:
+    """Symbols to fetch sentiment for, skipping anything refreshed within
+    ``sentiment_fresh_hours``.
+
+    No per-run cap: the relevant set is naturally bounded (curated watchlist <= 20,
+    a handful of open trades and top-pick feeders, top-N composite), so refreshing
+    all of it stays well inside the free search tier — and everything that matters
+    is always fully covered. The two tiers only affect ORDER, so a mid-run provider
+    rate limit still lands the names we care about first:
+
+      Priority: top-pick feeders (momentum leaders with a tradeable long setup, see
+        ``_get_top_pick_feeder_ids``) + the curated watchlist + open paper trades —
+        the set we never want shown without sentiment.
+      Filler: top-N by composite — a cheap discovery net for names not yet covered.
+
+    Once the set is fresh, runs make zero grounded searches until it ages out.
     """
     from app.models.paper_trade import PaperTrade
     from app.models.score import CompositeScore
     from app.models.watchlist import WatchlistEntry
 
-    relevant: set[int] = set()
+    cutoff = datetime.now(timezone.utc) - timedelta(hours=settings.sentiment_fresh_hours)
+
+    # Priority: the set we always want fresh — top-pick feeders, the curated
+    # watchlist, and open positions.
+    priority_ids = await _get_top_pick_feeder_ids(db)
     wl = await db.execute(
         select(WatchlistEntry.ticker_id)
         .where(WatchlistEntry.entry_type != "dismissed")
         .distinct()
     )
-    relevant.update(r[0] for r in wl.all())
+    priority_ids.update(r[0] for r in wl.all())
     pt = await db.execute(
         select(PaperTrade.ticker_id).where(PaperTrade.status == "open").distinct()
     )
-    relevant.update(r[0] for r in pt.all())
+    priority_ids.update(r[0] for r in pt.all())
+
+    # Filler: top-N by composite, a discovery net for names not already covered.
     top = await db.execute(
         select(CompositeScore.ticker_id)
         .order_by(CompositeScore.score.desc())
         .limit(settings.sentiment_top_composite)
     )
-    relevant.update(r[0] for r in top.all())
+    filler_ids = {r[0] for r in top.all()} - priority_ids
 
-    # Momentum leaders: the tickers that can clear the activation gate, which
-    # selects the top ``min_momentum_percentile`` slice by 12-1 momentum — a
-    # different axis than composite score. The gate qualifies setups on this
-    # percentile, so without including them a freshly-qualifying ticker carries no
-    # sentiment and gets enhanced as neutral. Pre-fetching their sentiment here (in
-    # the daily pipeline, sentiment runs right after the OHLCV refresh) means the
-    # following R:R scan reads real sentiment for the setups it qualifies.
-    # Best-effort: a momentum/config failure must not stop sentiment collection.
-    try:
-        from app.services import momentum_service
-        from app.services.admin_service import get_activation_config
-
-        activation = await get_activation_config(db)
-        min_pct = float(activation.get("min_momentum_percentile", 0.0))
-        if min_pct > 0:
-            percentiles = await momentum_service.compute_momentum_percentiles(db)
-            leaders = [sym for sym, pct in percentiles.items() if pct >= min_pct]
-            if leaders:
-                rows = await db.execute(
-                    select(Ticker.id).where(Ticker.symbol.in_(leaders))
-                )
-                relevant.update(r[0] for r in rows.all())
-    except Exception:
-        logger.exception("Sentiment momentum-leader scoping failed; using base relevant set")
-
-    if not relevant:
+    if not priority_ids and not filler_ids:
         return []
 
-    cutoff = datetime.now(timezone.utc) - timedelta(hours=settings.sentiment_fresh_hours)
-    latest_ts = func.max(SentimentScore.timestamp)
-    missing_first = case((latest_ts.is_(None), 0), else_=1)
-    result = await db.execute(
-        select(Ticker.symbol)
-        .outerjoin(SentimentScore, SentimentScore.ticker_id == Ticker.id)
-        .where(Ticker.id.in_(relevant))
-        .group_by(Ticker.id, Ticker.symbol)
-        .having(or_(latest_ts.is_(None), latest_ts < cutoff))
-        .order_by(missing_first.asc(), latest_ts.asc(), Ticker.symbol.asc())
-        .limit(settings.sentiment_max_per_run)
-    )
-    return list(result.scalars().all())
+    # No cap — fetch every stale name. Priority first so a rate limit mid-run still
+    # covers the curated/at-risk set before the discovery net.
+    priority_syms = await _stale_sentiment_symbols(db, priority_ids, cutoff)
+    filler_syms = await _stale_sentiment_symbols(db, filler_ids, cutoff)
+    return priority_syms + filler_syms
 
 
 async def _get_fundamental_priority_tickers(db: AsyncSession) -> list[str]: