S/R alerts: nearest zone only, scoped to watchlist + qualified, merged levels

Three fixes to over-firing S/R proximity alerts:
- Route through cluster_sr_zones (the same merger the chart uses) instead of raw
  SRLevel rows, so near-duplicate levels (e.g. CVX 183 + 185) collapse into one
  zone and one alert.
- Alert only the single NEAREST strong zone per ticker, not every nearby level.
- Scope to watchlist + qualified-setup tickers via _alert_scope_tickers (was
  iterating all watchlist entries only; qualified setups are now included too).

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

app/services/alert_service.py

@@ -34,6 +34,7 @@ from app.models.watchlist import WatchlistEntry
 from app.services.admin_service import get_activation_config, update_setting
 from app.services.qualification import best_target_probability, setup_qualifies
 from app.services.rr_scanner_service import get_trade_setups
+from app.services.sr_service import cluster_sr_zones
 
 logger = logging.getLogger(__name__)
 
@@ -55,8 +56,9 @@ _BOOL_DEFAULTS = {
 }
 
 # Tunables (kept as constants for now; promote to settings if needed)
-SR_PROXIMITY_PCT = 2.0      # within this % of a strong level → alert
-SR_MIN_STRENGTH = 60        # only strong levels are alert-worthy
+SR_PROXIMITY_PCT = 2.0      # within this % of a strong zone → alert
+SR_MIN_STRENGTH = 60        # only strong zones are alert-worthy
+SR_CLUSTER_TOLERANCE = 0.02  # merge levels within 2% into one zone (matches chart)
 SCORE_DROP_POINTS = 15.0    # composite drop vs watermark that triggers an alert
 COOLDOWN_HOURS = 72         # don't re-send the same key within this window
 DIGEST_HOUR_UTC = 22        # send the daily digest on the first run at/after this hour
@@ -222,6 +224,25 @@ async def _watchlist_tickers(db: AsyncSession) -> list[tuple[int, str]]:
     return [(tid, sym) for tid, sym in result.all()]
 
 
+async def _alert_scope_tickers(db: AsyncSession) -> list[tuple[int, str]]:
+    """Tickers eligible for S/R-proximity alerts: watchlist ∪ qualified setups.
+
+    A ticker only matters for a proximity heads-up if you're watching it or it
+    has an actionable setup — not the whole universe.
+    """
+    scope: dict[int, str] = {tid: sym for tid, sym in await _watchlist_tickers(db)}
+
+    qualified_symbols = {s["symbol"] for s in await _qualified_setups(db)}
+    missing = qualified_symbols - set(scope.values())
+    if missing:
+        result = await db.execute(
+            select(Ticker.id, Ticker.symbol).where(Ticker.symbol.in_(missing))
+        )
+        for tid, sym in result.all():
+            scope[tid] = sym
+    return list(scope.items())
+
+
 async def _qualified_setups(db: AsyncSession) -> list[dict]:
     setups = await get_trade_setups(db)
     config = await get_activation_config(db)
@@ -259,26 +280,48 @@ async def _latest_close(db: AsyncSession, ticker_id: int) -> float | None:
 
 
 async def _collect_sr_proximity(db: AsyncSession) -> list[tuple[str, str]]:
+    """One alert per ticker for the NEAREST strong S/R zone within range.
+
+    Levels are merged into zones with the same clusterer the chart uses, so a
+    cluster of near-duplicate levels (e.g. 183 + 185) is a single zone and a
+    single alert. Scoped to watchlist ∪ qualified tickers.
+    """
     out: list[tuple[str, str]] = []
-    for tid, symbol in await _watchlist_tickers(db):
+    for tid, symbol in await _alert_scope_tickers(db):
         price = await _latest_close(db, tid)
         if not price:
             continue
-        levels_result = await db.execute(
-            select(SRLevel).where(
-                SRLevel.ticker_id == tid,
-                SRLevel.strength >= SR_MIN_STRENGTH,
-            )
+
+        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()
+        ]
+        if not levels:
+            continue
+
+        zones = cluster_sr_zones(levels, price, tolerance=SR_CLUSTER_TOLERANCE)
+        strong = [z for z in zones if z["strength"] >= SR_MIN_STRENGTH]
+        if not strong:
+            continue
+
+        # Nearest strong zone only.
+        nearest = min(strong, key=lambda z: abs(price - z["midpoint"]))
+        dist_pct = abs(price - nearest["midpoint"]) / price * 100
+        if dist_pct > SR_PROXIMITY_PCT:
+            continue
+
+        label = (
+            f"{nearest['low']:.2f}–{nearest['high']:.2f}"
+            if nearest["level_count"] > 1
+            else f"{nearest['midpoint']:.2f}"
         )
-        for lv in levels_result.scalars().all():
-            dist_pct = abs(price - lv.price_level) / price * 100
-            if dist_pct <= SR_PROXIMITY_PCT:
-                key = f"sr:{symbol}:{lv.price_level:.2f}"
-                out.append((
-                    key,
-                    f"📍 <b>{symbol}</b> approaching {lv.type} at {lv.price_level:.2f} "
-                    f"(now {price:.2f}, {dist_pct:.1f}% away)",
-                ))
+        key = f"sr:{symbol}:{nearest['type']}"  # one per side per ticker per cooldown
+        out.append((
+            key,
+            f"📍 <b>{symbol}</b> approaching {nearest['type']} {label} "
+            f"(now {price:.2f}, {dist_pct:.1f}% away)",
+        ))
     return out
 
 

tests/unit/test_alert_service.py

@@ -2,12 +2,14 @@
 
 from __future__ import annotations
 
-from datetime import datetime, timedelta, timezone
+from datetime import date, datetime, timedelta, timezone
 
 import pytest
 
 from app.models.alert import AlertLog
+from app.models.ohlcv import OHLCVRecord
 from app.models.score import CompositeScore
+from app.models.sr_level import SRLevel
 from app.models.ticker import Ticker
 from app.models.user import User
 from app.models.watchlist import WatchlistEntry
@@ -109,6 +111,54 @@ async def test_score_drop_seeds_then_alerts(session):
     assert await svc._watermark(session, "AAA") == 60.0
 
 
+async def _add_ticker(session, symbol: str, *, watchlisted: bool, close: float,
+                      levels: list[tuple[float, str, int]]) -> int:
+    user = await session.get(User, 1)
+    if user is None:
+        session.add(User(id=1, username="u", password_hash="x", role="user", has_access=True))
+        await session.flush()
+    t = Ticker(symbol=symbol)
+    session.add(t)
+    await session.flush()
+    if watchlisted:
+        session.add(WatchlistEntry(user_id=1, ticker_id=t.id, entry_type="manual",
+                                   added_at=datetime.now(timezone.utc)))
+    session.add(OHLCVRecord(ticker_id=t.id, date=date.today(),
+                            open=close, high=close, low=close, close=close, volume=1))
+    for price, kind, strength in levels:
+        session.add(SRLevel(ticker_id=t.id, price_level=price, type=kind,
+                            strength=strength, detection_method="test"))
+    await session.commit()
+    return t.id
+
+
+async def test_sr_proximity_merges_close_levels_to_one_alert(session):
+    # Two resistance levels ~1% apart just above price → one merged zone, one alert
+    await _add_ticker(session, "CVX", watchlisted=True, close=182.0,
+                      levels=[(183.0, "resistance", 60), (185.0, "resistance", 60)])
+    msgs = await svc._collect_sr_proximity(session)
+    cvx = [m for m in msgs if "CVX" in m[1]]
+    assert len(cvx) == 1
+    assert "183.00–185.00" in cvx[0][1]
+
+
+async def test_sr_proximity_skips_non_watchlist_unqualified(session):
+    await _add_ticker(session, "ZZZ", watchlisted=False, close=182.0,
+                      levels=[(183.0, "resistance", 80)])
+    msgs = await svc._collect_sr_proximity(session)
+    assert not any("ZZZ" in m[1] for m in msgs)
+
+
+async def test_sr_proximity_only_nearest_zone(session):
+    # A near resistance (~1%) and a far one (~10%); only the near one alerts
+    await _add_ticker(session, "AAA", watchlisted=True, close=100.0,
+                      levels=[(101.0, "resistance", 70), (110.0, "resistance", 90)])
+    msgs = await svc._collect_sr_proximity(session)
+    aaa = [m for m in msgs if "AAA" in m[1]]
+    assert len(aaa) == 1
+    assert "101.00" in aaa[0][1]
+
+
 async def test_dispatch_disabled_short_circuits(session):
     res = await svc.dispatch_alerts(session)
     assert res["status"] == "disabled"