feat: ticker search, watchlist momentum column, alpha vs S&P 500

Three usability fixes:

1. Global ticker search in the sidebar (TickerSearch) — typeahead over the
   tracked universe that opens a ticker's detail page without adding it to the
   watchlist. Also wired into the mobile nav.

2. Watchlist table shows the ticker's 12-1 momentum percentile (the top-pick
   selector) instead of the noisy full S/R-level list. Enriched from the setup
   already loaded in watchlist_service._enrich_entry — no extra query.

3. Alpha vs the S&P 500 on paper trades (open + closed). New benchmark_prices
   table + benchmark_service store SPY daily closes (a standalone series, not a
   Ticker, so it never enters the scanner / momentum ranking / rankings) via a
   new daily-pipeline step. paper_trade_service computes per-trade
   benchmark_return / alpha_pct / alpha_usd over each holding period; the open-
   trades table, dashboard, and closed-trades panel surface per-trade and total
   alpha. The list read path never makes a provider call.

Deploy: alembic upgrade head, then run the benchmark/daily job once to populate
SPY closes (alpha shows "—" until then).

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

app/scheduler.py

@@ -36,6 +36,7 @@ from app.providers.protocol import SentimentData
 from app.services import fundamental_service, ingestion_service, sentiment_service, settings_store
 from app.services.alert_service import dispatch_alerts
 from app.services.backtest_service import run_and_store as run_backtest_and_store
+from app.services.benchmark_service import refresh_benchmark_prices
 from app.services.market_regime_service import update_market_regime
 from app.services.regime_monitor_service import update_regime_monitor
 from app.services.event_study_service import run_and_store as run_event_study_and_store
@@ -866,6 +867,34 @@ async def compute_market_regime() -> None:
         _log_event(logging.ERROR, "job_error", job=job_name, error_type=type(exc).__name__, message=str(exc))
 
 
+# ---------------------------------------------------------------------------
+# Job: Benchmark Collector (SPY closes for paper-trade alpha)
+# ---------------------------------------------------------------------------
+
+
+async def collect_benchmark() -> None:
+    """Refresh the stored benchmark (SPY) daily closes used for paper-trade alpha."""
+    job_name = "benchmark_collector"
+    _log_event(logging.INFO, "job_start", job=job_name)
+    _runtime_start(job_name, total=1)
+
+    try:
+        async with async_session_factory() as db:
+            if not await _is_job_enabled(db, job_name):
+                _log_event(logging.INFO, "job_skipped", job=job_name, reason="disabled")
+                _runtime_finish(job_name, "skipped", processed=0, total=1, message="Disabled")
+                return
+
+            written = await refresh_benchmark_prices(db)
+
+        _runtime_progress(job_name, processed=1, total=1)
+        _runtime_finish(job_name, "completed", processed=1, total=1, message=f"{written} rows")
+        _log_event(logging.INFO, "job_complete", job=job_name, rows=written)
+    except Exception as exc:
+        _runtime_finish(job_name, "error", processed=0, total=1, message=str(exc))
+        _log_event(logging.ERROR, "job_error", job=job_name, error_type=type(exc).__name__, message=str(exc))
+
+
 # ---------------------------------------------------------------------------
 # Job: Regime Monitor
 # ---------------------------------------------------------------------------
@@ -1016,6 +1045,7 @@ async def sync_ticker_universe() -> None:
 # Daily (full): the complete data→signal refresh, once a day.
 _DAILY_PIPELINE_STEPS = [
     ("data_collector", "collect_ohlcv"),
+    ("benchmark_collector", "collect_benchmark"),
     ("sentiment_collector", "collect_sentiment"),
     ("rr_scanner", "scan_rr"),
     ("outcome_evaluator", "evaluate_outcomes"),
@@ -1068,8 +1098,8 @@ async def _run_pipeline(job_name: str, steps: list[tuple[str, str]]) -> None:
 
 
 async def run_daily_pipeline() -> None:
-    """Full daily flow: OHLCV → sentiment → R:R scan → outcome eval (+paper
-    close) → market regime."""
+    """Full daily flow: OHLCV → benchmark → sentiment → R:R scan → outcome eval
+    (+paper close) → market regime."""
     await _run_pipeline("daily_pipeline", _DAILY_PIPELINE_STEPS)
 
 
@@ -1176,6 +1206,7 @@ def configure_scheduler(schedule_config: dict[str, str] | None = None) -> None:
     # interval job). They stay manually triggerable from Admin → Jobs.
     _members = [
         (collect_ohlcv, "data_collector", "Data Collector (OHLCV)"),
+        (collect_benchmark, "benchmark_collector", "Benchmark Collector"),
         (collect_sentiment, "sentiment_collector", "Sentiment Collector"),
         (scan_rr, "rr_scanner", "R:R Scanner"),
         (evaluate_outcomes, "outcome_evaluator", "Outcome Evaluator"),