Optimize signal read paths and enforce score invariants

Three follow-ups to the gate probability floor (8f41143):

- Signals table shows the starred primary target (shared primaryTarget
  helper) instead of an independently computed max-probability best,
  so Overview, Signals and ticker details agree by construction.
- Targets pinned at the 3% probability clamp floor collapse to the
  nearest one (enhance_trade_setup + backtest candidates in parity):
  floor-pinned levels are indistinguishable to the model, so farther
  ones were duplicate 3% rows inviting lottery headlines.
- get_trade_setups only returns setups re-emitted within
  LIVE_SETUP_MAX_AGE_DAYS (3): an older latest row means the daily
  scan no longer confirms the setup, and such rows otherwise surface
  forever on Overview/Signals/ticker/alerts. History endpoints keep
  full history.

Backtest on the Jul-3 snapshot is metric-identical to the gate-floor
run on all qualified stats (1089 qualified, Sharpe 2.02, CAGR +49.6%,
DD -15.8%): the prune only removes noise the gate already rejected.

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

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
@@ -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/backtest_service.py

@@ -65,6 +65,7 @@ from app.services.qualification import (
 from app.services.recommendation_service import (
     _choose_recommended_action,
     _classify_by_probability,
+    _prune_floor_pinned_targets,
     _risk_level_from_conflicts,
     _select_primary_target,
     _zone_representative_levels,
@@ -179,6 +180,9 @@ def _window_setups(
                 t, dim_scores, None, direction, config
             )
             t["classification"] = _classify_by_probability(t["probability"])
+        # Collapse duplicate floor-pinned lottery targets (parity with
+        # enhance_trade_setup).
+        targets = _prune_floor_pinned_targets(targets)
         primary = _select_primary_target(targets)
         if primary is None:
             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/qualification.py

@@ -5,9 +5,11 @@ performance stats (server) and mirrored on the frontend. The core selection is
 residual cross-sectional momentum: a setup's ticker must rank in the top
 ``min_momentum_percentile`` of the universe by beta-adjusted 12-1 month momentum.
 R:R and confidence remain as floors, and conviction/conflict survive as optional
-tighteners (off by default). Qualified setups must also have a probability-backed
-target; otherwise a mathematically high R:R can be driven by a fragile target
-with no independent validation.
+tighteners (off by default). Qualified setups must also have a primary target
+with at least ``MIN_TARGET_PROBABILITY`` reach probability: a primary below the
+floor is a lottery target whose distance inflates R:R, so it would otherwise
+game the min_rr gate (the model clamps probabilities at 3%, and far targets pin
+there while their live R:R stays high forever).
 """
 
 from __future__ import annotations
@@ -16,6 +18,13 @@ from typing import Any
 
 HIGH_CONVICTION_ACTIONS = {"LONG_HIGH", "SHORT_HIGH"}
 
+# Floor for the primary target's reach probability, shared with the primary
+# target selection in recommendation_service and mirrored in the frontend
+# (qualification.ts). Under the two-barrier model a fair-race 1.5:1 target sits
+# near ~34% before drift adjustments, so 20% only excludes targets the model
+# itself considers long shots.
+MIN_TARGET_PROBABILITY = 20.0
+
 
 def _action_direction(action: str | None) -> str:
     if not action or action == "NEUTRAL":
@@ -85,7 +94,8 @@ def setup_qualifies(setup: Any, config: dict) -> bool:
         live_rr = live_risk_reward(setup, float(current_price))
         if live_rr is not None and live_rr < config["min_rr"]:
             return False
-    if primary_target_probability(setup) is None:
+    target_probability = primary_target_probability(setup)
+    if target_probability is None or target_probability < MIN_TARGET_PROBABILITY:
         return False
     if (setup.confidence_score or 0.0) < config["min_confidence"]:
         return False

app/services/recommendation_service.py

@@ -13,6 +13,7 @@ from app.models.settings import SystemSetting
 from app.models.sr_level import SRLevel
 from app.models.ticker import Ticker
 from app.models.trade_setup import TradeSetup
+from app.services.qualification import MIN_TARGET_PROBABILITY
 from app.services.sr_service import cluster_sr_zones
 
 logger = logging.getLogger(__name__)
@@ -44,6 +45,12 @@ _MODERATE_MAX_ATR = 4.6
 # the same tolerance the chart and alerts use, so S/R is one model app-wide.
 _SR_ZONE_TOLERANCE = 0.02
 
+# Reach-probability estimates are clamped to this band; a target at the floor
+# means "the model considers it essentially unreachable" and floor-pinned
+# targets are mutually indistinguishable.
+_PROBABILITY_CLAMP_LOW = 3.0
+_PROBABILITY_CLAMP_HIGH = 95.0
+
 
 def _clamp(value: float, low: float, high: float) -> float:
     return max(low, min(high, value))
@@ -407,7 +414,7 @@ class ProbabilityEstimator:
         elif opposed:
             probability -= signal_weight * 100.0
 
-        return round(_clamp(probability, 3.0, 95.0), 2)
+        return round(_clamp(probability, _PROBABILITY_CLAMP_LOW, _PROBABILITY_CLAMP_HIGH), 2)
 
 
 signal_conflict_detector = SignalConflictDetector()
@@ -575,10 +582,30 @@ def build_recommendation_snapshot(
 
 
 PRIMARY_TARGET_MIN_RR = 1.5
-# Below this the target is a lottery ticket: under the two-barrier model a
-# fair-race 1.5:1 target sits near ~34% before drift adjustments, so 20% only
-# excludes targets the model itself considers long shots.
-PRIMARY_TARGET_MIN_PROBABILITY = 20.0
+# Below this the target is a lottery ticket. Shared with the activation gate
+# (qualification.MIN_TARGET_PROBABILITY) so the primary selection and the gate
+# agree on what counts as a probability-backed target.
+PRIMARY_TARGET_MIN_PROBABILITY = MIN_TARGET_PROBABILITY
+
+
+def _prune_floor_pinned_targets(targets: list[dict]) -> list[dict]:
+    """Keep only the nearest target pinned at the probability clamp floor.
+
+    Floor-pinned targets are indistinguishable to the model (true probability
+    at/below the clamp), so farther ones add no information — they just fill
+    the table with duplicate "3%" rows whose inflated R:R invites lottery
+    picks. ``targets`` is distance-sorted by the generator, so the first
+    floor-pinned entry is the nearest (most reachable) representative.
+    """
+    pruned: list[dict] = []
+    seen_floor = False
+    for target in targets:
+        if float(target.get("probability", 0.0)) <= _PROBABILITY_CLAMP_LOW:
+            if seen_floor:
+                continue
+            seen_floor = True
+        pruned.append(target)
+    return pruned
 
 
 def _select_primary_target(
@@ -664,6 +691,9 @@ async def enhance_trade_setup(
         # Label follows from the reach-probability: high prob = Conservative.
         target["classification"] = _classify_by_probability(target["probability"])
 
+    # Collapse duplicate floor-pinned lottery targets to the nearest one.
+    targets = _prune_floor_pinned_targets(targets)
+
     # Primary target = most-likely target with real asymmetry (see
     # _select_primary_target), not the old quality-score pick that ignored
     # probability. Sync the setup's headline target/rr_ratio so the chart, gate

app/services/rr_scanner_service.py

@@ -11,7 +11,7 @@ from __future__ import annotations
 import json
 import logging
 from collections.abc import Callable
-from datetime import date, datetime, timezone
+from datetime import date, datetime, timedelta, timezone
 
 from sqlalchemy import and_, func, select
 from sqlalchemy.ext.asyncio import AsyncSession
@@ -39,6 +39,15 @@ logger = logging.getLogger(__name__)
 
 STRATEGY_VERSION = "residual_highvol_80_20_atr_trail3_v1"
 
+# A setup counts as live only while the daily scan keeps re-emitting it. The
+# scan runs every day (07:00 UTC cron), so anything older than this was NOT
+# re-confirmed — typically because no level clears the R:R threshold from the
+# current price anymore. Without this cutoff such rows stay "latest" forever
+# (the scanner never writes a replacement) and keep surfacing on the live
+# views. 3 days buffers a missed pipeline run or two; history endpoints are
+# unaffected.
+LIVE_SETUP_MAX_AGE_DAYS = 3
+
 
 async def _get_ticker(db: AsyncSession, symbol: str) -> Ticker:
     normalised = symbol.strip().upper()
@@ -547,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.
@@ -573,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)
 
@@ -587,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, "")
 
@@ -602,10 +618,17 @@ async def get_trade_setups(
     live_recommendation: bool = False,
     exclude_open_trade_tickers: bool = False,
 ) -> list[dict]:
-    """Get latest stored trade setups, optionally filtered."""
+    """Get latest stored trade setups, optionally filtered.
+
+    Only setups the daily scan re-emitted within ``LIVE_SETUP_MAX_AGE_DAYS``
+    are returned — an older "latest" row means the scanner no longer finds a
+    valid setup for that ticker, so it must not surface as current.
+    """
+    cutoff = datetime.now(timezone.utc) - timedelta(days=LIVE_SETUP_MAX_AGE_DAYS)
     stmt = (
         select(TradeSetup, Ticker.symbol)
         .join(Ticker, TradeSetup.ticker_id == Ticker.id)
+        .where(TradeSetup.detected_at >= cutoff)
     )
     if direction is not None:
         stmt = stmt.where(TradeSetup.direction == direction.lower())

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":

frontend/src/components/scanner/TradeTable.tsx

@@ -1,9 +1,10 @@
 import { Link } from 'react-router-dom';
 import type { TradeSetup } from '../../lib/types';
 import { formatPrice, formatPercent, formatDateTime } from '../../lib/format';
+import { primaryTarget } from '../../lib/qualification';
 import { recommendationActionDirection, recommendationActionLabel } from '../../lib/recommendation';
 
-export type SortColumn = 'symbol' | 'direction' | 'recommended_action' | 'confidence_score' | 'entry_price' | 'stop_loss' | 'target' | 'best_target_probability' | 'risk_amount' | 'reward_amount' | 'rr_ratio' | 'stop_pct' | 'target_pct' | 'risk_level' | 'composite_score' | 'detected_at';
+export type SortColumn = 'symbol' | 'direction' | 'recommended_action' | 'confidence_score' | 'entry_price' | 'stop_loss' | 'target' | 'primary_target_probability' | 'risk_amount' | 'reward_amount' | 'rr_ratio' | 'stop_pct' | 'target_pct' | 'risk_level' | 'composite_score' | 'detected_at';
 export type SortDirection = 'asc' | 'desc';
 
 interface TradeTableProps {
@@ -21,7 +22,7 @@ const columns: { key: SortColumn; label: string }[] = [
   { key: 'entry_price', label: 'Entry' },
   { key: 'stop_loss', label: 'Stop Loss' },
   { key: 'target', label: 'Target' },
-  { key: 'best_target_probability', label: 'Best Target' },
+  { key: 'primary_target_probability', label: 'Primary Target' },
   { key: 'risk_amount', label: 'Risk $' },
   { key: 'reward_amount', label: 'Reward $' },
   { key: 'rr_ratio', label: 'R:R' },
@@ -65,10 +66,12 @@ function riskLevelClass(riskLevel: TradeSetup['risk_level']) {
   return 'text-gray-400';
 }
 
-function bestTargetText(trade: TradeSetup) {
-  if (!trade.targets || trade.targets.length === 0) return '—';
-  const best = [...trade.targets].sort((a, b) => b.probability - a.probability)[0];
-  return `${formatPrice(best.price)} (${best.probability.toFixed(0)}%)`;
+// The starred primary — the same target the Overview and ticker details
+// headline, so every view agrees on which target a setup is "about".
+function primaryTargetText(trade: TradeSetup) {
+  const primary = primaryTarget(trade);
+  if (!primary) return '—';
+  return `${formatPrice(primary.price)} (${primary.probability.toFixed(0)}%)`;
 }
 
 export function TradeTable({ trades, sortColumn, sortDirection, onSort }: TradeTableProps) {
@@ -121,7 +124,7 @@ export function TradeTable({ trades, sortColumn, sortDirection, onSort }: TradeT
                 <td className="px-4 py-3.5 font-mono text-gray-200">{formatPrice(trade.entry_price)}</td>
                 <td className="px-4 py-3.5 font-mono text-gray-200">{formatPrice(trade.stop_loss)}</td>
                 <td className="px-4 py-3.5 font-mono text-gray-200">{formatPrice(trade.target)}</td>
-                <td className="px-4 py-3.5 font-mono text-gray-200">{bestTargetText(trade)}</td>
+                <td className="px-4 py-3.5 font-mono text-gray-200">{primaryTargetText(trade)}</td>
                 <td className="px-4 py-3.5 font-mono text-gray-200">{formatPrice(analysis.risk_amount)}</td>
                 <td className="px-4 py-3.5 font-mono text-gray-200">{formatPrice(analysis.reward_amount)}</td>
                 <td className={`px-4 py-3.5 font-mono font-semibold ${rrColorClass(trade.rr_ratio)}`}>{trade.rr_ratio.toFixed(2)}</td>

frontend/src/components/signals/SetupsPanel.tsx

@@ -2,7 +2,7 @@ import { useEffect, useMemo, useState } from 'react';
 import { useMutation, useQueryClient } from '@tanstack/react-query';
 import { useActivation } from '../../hooks/useActivation';
 import { useTrades } from '../../hooks/useTrades';
-import { qualifiesSetup, activationSummary } from '../../lib/qualification';
+import { qualifiesSetup, activationSummary, primaryTargetProbability } from '../../lib/qualification';
 import { TradeTable, type SortColumn, type SortDirection, computeTradeAnalysis } from '../scanner/TradeTable';
 import { SkeletonTable } from '../ui/Skeleton';
 import { useToast } from '../ui/Toast';
@@ -42,8 +42,8 @@ function getComputedValue(trade: TradeSetup, column: SortColumn): number {
     case 'stop_pct': return analysis.stop_pct;
     case 'target_pct': return analysis.target_pct;
     case 'confidence_score': return trade.confidence_score ?? -1;
-    case 'best_target_probability':
-      return trade.targets?.length ? Math.max(...trade.targets.map((t) => t.probability)) : -1;
+    case 'primary_target_probability':
+      return primaryTargetProbability(trade) ?? -1;
     case 'risk_level':
       if (trade.risk_level === 'Low') return 1;
       if (trade.risk_level === 'Medium') return 2;
@@ -78,7 +78,7 @@ function sortTrades(
       case 'stop_pct':
       case 'target_pct':
       case 'confidence_score':
-      case 'best_target_probability':
+      case 'primary_target_probability':
       case 'risk_level':
         cmp = getComputedValue(a, column) - getComputedValue(b, column);
         break;

frontend/src/lib/qualification.ts

@@ -1,7 +1,14 @@
-import type { ActivationConfig, TradeSetup } from './types';
+import type { ActivationConfig, TradeSetup, TradeTarget } from './types';
 
 const HIGH_CONVICTION_ACTIONS = new Set(['LONG_HIGH', 'SHORT_HIGH']);
 
+/**
+ * Floor for the primary target's reach probability — mirrors
+ * MIN_TARGET_PROBABILITY in app/services/qualification.py. A primary below
+ * this is a lottery target whose distance inflates R:R past the min_rr gate.
+ */
+export const MIN_TARGET_PROBABILITY = 20;
+
 function actionDirection(action: TradeSetup['recommended_action']): 'long' | 'short' | 'neutral' {
   if (!action || action === 'NEUTRAL') return 'neutral';
   if (action.startsWith('LONG')) return 'long';
@@ -9,15 +16,18 @@ function actionDirection(action: TradeSetup['recommended_action']): 'long' | 'sh
   return 'neutral';
 }
 
-export function bestTargetProbability(setup: TradeSetup): number {
-  return setup.targets?.length ? Math.max(...setup.targets.map((t) => t.probability)) : 0;
+/** The starred primary target (the one the headline R:R refers to), falling
+ * back to the most likely target when no star is stored. */
+export function primaryTarget(setup: TradeSetup): TradeTarget | null {
+  const starred = setup.targets?.find((t) => t.is_primary);
+  if (starred) return starred;
+  if (!setup.targets?.length) return null;
+  return [...setup.targets].sort((a, b) => b.probability - a.probability)[0];
 }
 
 /** Probability of the starred primary target (the one the headline R:R refers to). */
 export function primaryTargetProbability(setup: TradeSetup): number | null {
-  const primary = setup.targets?.find((t) => t.is_primary);
-  if (primary) return primary.probability;
-  return setup.targets?.length ? bestTargetProbability(setup) : null;
+  return primaryTarget(setup)?.probability ?? null;
 }
 
 /** R:R recomputed from the current price (0 if no reward/risk left). */
@@ -40,7 +50,7 @@ export function qualifiesSetup(setup: TradeSetup, config: ActivationConfig): boo
     return false;
   }
   const targetProbability = primaryTargetProbability(setup);
-  if (targetProbability == null || targetProbability <= 0) return false;
+  if (targetProbability == null || targetProbability < MIN_TARGET_PROBABILITY) return false;
   if ((setup.confidence_score ?? 0) < config.min_confidence) return false;
   // Residual cross-sectional momentum is the core selection (long-only). While
   // the gate is active, shorts never qualify; missing ranks do not qualify
@@ -77,6 +87,9 @@ export function disqualifyReason(setup: TradeSetup, config: ActivationConfig): s
   }
   const targetProbability = primaryTargetProbability(setup);
   if (targetProbability == null || targetProbability <= 0) return 'no target probability';
+  if (targetProbability < MIN_TARGET_PROBABILITY) {
+    return `target probability below ${MIN_TARGET_PROBABILITY}%`;
+  }
   if ((setup.confidence_score ?? 0) < config.min_confidence) {
     return `confidence below ${config.min_confidence.toFixed(0)}%`;
   }

tests/unit/test_qualification.py

@@ -82,6 +82,31 @@ class TestFloors:
             DEFAULT_GATE,
         ) is True
 
+    def test_lottery_primary_target_fails(self):
+        # A far target pinned at the model's 3% clamp floor: its distance keeps
+        # both stored and live R:R above the gate, so only the probability floor
+        # can reject it (the stale pre-fix lottery-headline case).
+        s = _setup(rr_ratio=3.09, targets=[{"probability": 3.0, "is_primary": True}])
+        assert setup_qualifies(s, DEFAULT_GATE) is False
+
+    def test_probability_at_floor_passes(self):
+        assert setup_qualifies(
+            _setup(targets=[{"probability": 20.0, "is_primary": True}]),
+            DEFAULT_GATE,
+        ) is True
+
+    def test_probability_just_below_floor_fails(self):
+        assert setup_qualifies(
+            _setup(targets=[{"probability": 19.9, "is_primary": True}]),
+            DEFAULT_GATE,
+        ) is False
+
+    def test_best_target_fallback_below_floor_fails(self):
+        # No starred primary: the fallback takes the best target, which must
+        # still clear the probability floor.
+        s = _setup(targets=[{"probability": 12.0}, {"probability": 8.0}])
+        assert setup_qualifies(s, DEFAULT_GATE) is False
+
 
 class TestMomentumGate:
     def test_top_momentum_passes(self):

tests/unit/test_recommendation_service.py

@@ -5,6 +5,7 @@ from dataclasses import dataclass
 from app.services.recommendation_service import (
     _build_reasoning,
     _choose_recommended_action,
+    _prune_floor_pinned_targets,
     _select_primary_target,
     direction_analyzer,
     probability_estimator,
@@ -154,6 +155,35 @@ def test_primary_target_requires_probability_floor():
     assert primary["price"] == 112.0
 
 
+def test_prune_keeps_only_nearest_floor_pinned_target():
+    # Two targets pinned at the 3% clamp floor are indistinguishable to the
+    # model — only the nearest survives; farther ones are duplicate noise.
+    targets = [
+        {"price": 204.0, "rr_ratio": 0.7, "probability": 25.6},
+        {"price": 241.0, "rr_ratio": 2.0, "probability": 3.0},
+        {"price": 272.0, "rr_ratio": 3.1, "probability": 3.0},
+    ]
+    pruned = _prune_floor_pinned_targets(targets)
+    assert [t["price"] for t in pruned] == [204.0, 241.0]
+
+
+def test_prune_leaves_targets_above_floor_untouched():
+    targets = [
+        {"price": 110.0, "rr_ratio": 2.0, "probability": 65.0},
+        {"price": 120.0, "rr_ratio": 3.5, "probability": 20.0},
+    ]
+    assert _prune_floor_pinned_targets(targets) == targets
+
+
+def test_prune_all_floor_pinned_keeps_nearest_only():
+    targets = [
+        {"price": 241.0, "rr_ratio": 2.0, "probability": 3.0},
+        {"price": 272.0, "rr_ratio": 3.1, "probability": 3.0},
+    ]
+    pruned = _prune_floor_pinned_targets(targets)
+    assert [t["price"] for t in pruned] == [241.0]
+
+
 def test_detects_sentiment_technical_conflict():
     conflicts = signal_conflict_detector.detect_conflicts(
         dimension_scores={"technical": 72.0, "momentum": 55.0, "fundamental": 50.0},

tests/unit/test_rr_scanner_preservation.py

@@ -29,7 +29,11 @@ from app.models.trade_setup import TradeSetup
 from app.models.score import CompositeScore, DimensionScore
 from app.models.sentiment import SentimentScore
 from app.models.user import User
-from app.services.rr_scanner_service import scan_ticker, get_trade_setups
+from app.services.rr_scanner_service import (
+    LIVE_SETUP_MAX_AGE_DAYS,
+    get_trade_setups,
+    scan_ticker,
+)
 
 
 def _as_utc(value: datetime) -> datetime:
@@ -69,11 +73,11 @@ def _make_ohlcv_bars(
     num_bars: int = 20,
     base_close: float = 100.0,
 ) -> list[OHLCVRecord]:
-    """Generate OHLCV bars closing around base_close with ATR ≈ 2.0."""
+    """Generate OHLCV bars closing around base_close with ATR ≈ 2.0."""
     bars: list[OHLCVRecord] = []
     start = date(2024, 1, 1)
     for i in range(num_bars):
-        close = base_close + (i % 3 - 1) * 0.5  # oscillate ±0.5
+        close = base_close + (i % 3 - 1) * 0.5  # oscillate ±0.5
         bars.append(OHLCVRecord(
             ticker_id=ticker_id,
             date=start + timedelta(days=i),
@@ -101,7 +105,7 @@ def zero_candidate_scenario(draw: st.DrawFn) -> dict:
       but all below the R:R threshold for their respective directions
     - Levels in the right direction but below R:R threshold
 
-    Note: scan_ticker does NOT filter by SR level type — it only checks whether
+    Note: scan_ticker does NOT filter by SR level type — it only checks whether
     the price_level is above or below entry. So "wrong side" means all levels
     are clustered near entry and below threshold in both directions.
     """
@@ -111,10 +115,10 @@ def zero_candidate_scenario(draw: st.DrawFn) -> dict:
         return {"variant": variant, "levels": []}
 
     else:  # below_threshold
-        # All levels close to entry so R:R < 1.5 with risk ≈ 3
-        # For longs: reward < 4.5 → price < 104.5
-        # For shorts: reward < 4.5 → price > 95.5
-        # Place all levels in the 96–104 band (below threshold both ways)
+        # All levels close to entry so R:R < 1.5 with risk ≈ 3
+        # For longs: reward < 4.5 → price < 104.5
+        # For shorts: reward < 4.5 → price > 95.5
+        # Place all levels in the 96–104 band (below threshold both ways)
         num = draw(st.integers(min_value=1, max_value=3))
         levels = []
         for _ in range(num):
@@ -139,7 +143,7 @@ def zero_candidate_scenario(draw: st.DrawFn) -> dict:
 def single_candidate_scenario(draw: st.DrawFn) -> dict:
     """Generate a scenario with exactly one S/R level that meets the R:R threshold.
 
-    For longs: one resistance above entry with R:R >= 1.5 (price >= 104.5 with risk ≈ 3).
+    For longs: one resistance above entry with R:R >= 1.5 (price >= 104.5 with risk ≈ 3).
     """
     direction = draw(st.sampled_from(["long", "short"]))
 
@@ -175,7 +179,7 @@ async def test_property_zero_candidates_produce_no_setup(
     """**Validates: Requirements 3.1, 3.2**
 
     Property: when zero candidate S/R levels exist (no levels, wrong side,
-    or below threshold), scan_ticker produces no setup — unchanged from
+    or below threshold), scan_ticker produces no setup — unchanged from
     original behavior.
     """
     from tests.conftest import _test_engine, _test_session_factory
@@ -225,7 +229,7 @@ async def test_property_single_candidate_selected_unchanged(
     """**Validates: Requirements 3.3**
 
     Property: when exactly one candidate S/R level meets the R:R threshold,
-    scan_ticker selects it — same as the original code would.
+    scan_ticker selects it — same as the original code would.
     """
     from tests.conftest import _test_engine, _test_session_factory
     from app.database import Base
@@ -270,7 +274,7 @@ async def test_property_single_candidate_selected_unchanged(
 
 
 # ===========================================================================
-# 7.2 Unit test: no S/R levels → no setup produced
+# 7.2 Unit test: no S/R levels → no setup produced
 # ===========================================================================
 
 @pytest.mark.asyncio
@@ -296,7 +300,7 @@ async def test_no_sr_levels_produces_no_setup(scan_session: AsyncSession):
 
 
 # ===========================================================================
-# 7.3 Unit test: single candidate meets threshold → selected
+# 7.3 Unit test: single candidate meets threshold → selected
 # ===========================================================================
 
 @pytest.mark.asyncio
@@ -306,7 +310,7 @@ async def test_single_resistance_above_threshold_selected(scan_session: AsyncSes
     When exactly one resistance level above entry meets the R:R threshold,
     it should be selected as the long setup target.
 
-    Entry ≈ 100, ATR ≈ 2, risk ≈ 3. Resistance at 110 → R:R ≈ 3.33 (>= 1.5).
+    Entry ≈ 100, ATR ≈ 2, risk ≈ 3. Resistance at 110 → R:R ≈ 3.33 (>= 1.5).
     """
     ticker = Ticker(symbol="SINGL")
     scan_session.add(ticker)
@@ -343,7 +347,7 @@ async def test_single_support_below_threshold_selected(scan_session: AsyncSessio
     When exactly one support level below entry meets the R:R threshold,
     it should be selected as the short setup target.
 
-    Entry ≈ 100, ATR ≈ 2, risk ≈ 3. Support at 90 → R:R ≈ 3.33 (>= 1.5).
+    Entry ≈ 100, ATR ≈ 2, risk ≈ 3. Support at 90 → R:R ≈ 3.33 (>= 1.5).
     """
     ticker = Ticker(symbol="SINGS")
     scan_session.add(ticker)
@@ -444,6 +448,42 @@ async def test_get_trade_setups_sorting_rr_desc_composite_desc(db_session: Async
     )
 
 
+@pytest.mark.asyncio
+async def test_get_trade_setups_excludes_stale_rows(db_session: AsyncSession):
+    """A "latest" row older than LIVE_SETUP_MAX_AGE_DAYS means the daily scan
+    stopped re-emitting the setup (nothing clears the R:R threshold from the
+    current price) — it must not surface on the live views."""
+    now = datetime.now(timezone.utc)
+    ticker_fresh = Ticker(symbol="FRESH")
+    ticker_stale = Ticker(symbol="STALE")
+    db_session.add_all([ticker_fresh, ticker_stale])
+    await db_session.flush()
+
+    db_session.add_all([
+        TradeSetup(
+            ticker_id=ticker_fresh.id, direction="long",
+            entry_price=100.0, stop_loss=97.0, target=109.0,
+            rr_ratio=3.0, composite_score=50.0,
+            detected_at=now - timedelta(days=1),
+        ),
+        TradeSetup(
+            ticker_id=ticker_stale.id, direction="long",
+            entry_price=100.0, stop_loss=97.0, target=109.0,
+            rr_ratio=3.0, composite_score=50.0,
+            detected_at=now - timedelta(days=LIVE_SETUP_MAX_AGE_DAYS, hours=1),
+        ),
+    ])
+    await db_session.flush()
+
+    results = await get_trade_setups(db_session)
+    symbols = [r["symbol"] for r in results]
+    assert symbols == ["FRESH"], f"Stale setup must be excluded, got {symbols}"
+
+    # The per-symbol view applies the same liveness rule.
+    stale_rows = await get_trade_setups(db_session, symbol="STALE")
+    assert stale_rows == []
+
+
 @pytest.mark.asyncio
 async def test_get_trade_setups_can_exclude_tickers_with_open_paper_trades(
     db_session: AsyncSession,
@@ -540,9 +580,12 @@ async def test_get_trade_setups_can_exclude_tickers_with_open_paper_trades(
 
 async def _seed_stale_setup_with_current_scores(db_session: AsyncSession) -> TradeSetup:
     """Stored setup frozen at scan time (conf 82, neutral) vs. current context
-    (bullish sentiment, composite 96) that yields live confidence 97."""
-    old_scan = datetime(2026, 7, 1, tzinfo=timezone.utc)
-    current = datetime(2026, 7, 3, tzinfo=timezone.utc)
+    (bullish sentiment, composite 96) that yields live confidence 97.
+
+    The scan date stays inside the LIVE_SETUP_MAX_AGE_DAYS liveness window —
+    these tests exercise the live overlay on a still-live row, not staleness."""
+    current = datetime.now(timezone.utc)
+    old_scan = current - timedelta(days=2)
     old_reasoning = (
         "LONG (high confidence): 82% with aligned signals "
         "(technical=88, momentum=60, sentiment=neutral)."
@@ -653,7 +696,7 @@ async def test_live_recommendation_filters_apply_to_live_values(
     """min_confidence must judge the overlaid live confidence, not the stored one."""
     await _seed_stale_setup_with_current_scores(db_session)
 
-    # Stored confidence is 82 — a stored-column filter would drop this row.
+    # Stored confidence is 82 — a stored-column filter would drop this row.
     # Live confidence is 97, so it must pass.
     rows = await get_trade_setups(
         db_session,
@@ -675,7 +718,7 @@ async def test_live_recommendation_filters_apply_to_live_values(
 
 
 async def _seed_two_direction_setup(db_session: AsyncSession) -> None:
-    current = datetime(2026, 7, 3, tzinfo=timezone.utc)
+    current = datetime.now(timezone.utc)
     ticker = Ticker(symbol="BOTH")
     db_session.add(ticker)
     await db_session.flush()
@@ -776,7 +819,7 @@ async def test_live_recommendation_action_independent_of_direction_filter(
 async def test_live_overlay_preserves_setup_specific_risk_and_context(
     db_session: AsyncSession,
 ):
-    current = datetime(2026, 7, 3, tzinfo=timezone.utc)
+    current = datetime.now(timezone.utc)
     ticker = Ticker(symbol="RISK")
     db_session.add(ticker)
     await db_session.flush()
@@ -883,7 +926,7 @@ async def test_live_trade_setup_read_does_not_recompute_scores(db_session: Async
 async def test_intraday_price_update_changes_live_price_without_new_signal_rows(
     db_session: AsyncSession,
 ):
-    current = datetime(2026, 7, 3, tzinfo=timezone.utc)
+    current = datetime.now(timezone.utc)
     ticker = Ticker(symbol="LIVEP")
     db_session.add(ticker)
     await db_session.flush()