Fix unbounded universe -> over-cap ingest 400 (audit HIGH #2)

The live 'symbols' list only ever grew (discover appends every 5 min, dashboard
injects add more) and was comma-joined into ONE bars request; once >~100 symbols
Alpaca 400s the whole batch -> ALL minute-bar ingestion silently stops mid-session.

- client.get_minute_bars / get_daily_bars / get_latest_trades batch symbols in
  chunks of 100 (cap-safe) and merge; get_daily_bars also paginates now.
- run_live_paper bounds the watchlist (WATCHER_MAX_UNIVERSE, default 80) so it
  rotates toward current movers instead of growing all day.

Found by the silent-failure audit workflow. Verified _chunked(250)->[100,100,50],
live calls intact. 71/71 tests green.

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

alpaca_paper/client.py

@@ -65,6 +65,12 @@ class AlpacaPaperClient:
     def _data(self, method: str, path: str, **kw):
         return self._request(method, f"{self.settings.data_base_url}{path}", **kw)
 
+    @staticmethod
+    def _chunked(items: list[str], size: int = 100):
+        """Yield symbol batches; Alpaca caps multi-symbol requests (~100)."""
+        for i in range(0, len(items), size):
+            yield items[i : i + size]
+
     # -- trading API -------------------------------------------------------
 
     def get_account(self) -> dict:
@@ -127,35 +133,39 @@ class AlpacaPaperClient:
         limit: int = 10_000,
         feed: str | None = None,
     ) -> dict[str, list[dict]]:
-        """Fetch 1-minute bars for symbols, following pagination tokens."""
+        """Fetch 1-minute bars, batching symbols (cap-safe) + following pagination."""
         all_bars: dict[str, list[dict]] = {s: [] for s in symbols}
-        page_token = None
-        while True:
-            params = {
-                "symbols": ",".join(symbols),
-                "timeframe": "1Min",
-                "start": start_iso,
-                "end": end_iso,
-                "limit": limit,
-                "feed": feed or self.settings.feed,
-                "adjustment": "raw",
-                "page_token": page_token,
-            }
-            payload = self._data("GET", "/v2/stocks/bars", params=params)
-            for symbol, bars in (payload.get("bars") or {}).items():
-                all_bars.setdefault(symbol, []).extend(bars or [])
-            page_token = payload.get("next_page_token")
-            if not page_token:
-                break
+        for chunk in self._chunked(symbols, 100):
+            page_token = None
+            while True:
+                params = {
+                    "symbols": ",".join(chunk),
+                    "timeframe": "1Min",
+                    "start": start_iso,
+                    "end": end_iso,
+                    "limit": limit,
+                    "feed": feed or self.settings.feed,
+                    "adjustment": "raw",
+                    "page_token": page_token,
+                }
+                payload = self._data("GET", "/v2/stocks/bars", params=params)
+                for symbol, bars in (payload.get("bars") or {}).items():
+                    all_bars.setdefault(symbol, []).extend(bars or [])
+                page_token = payload.get("next_page_token")
+                if not page_token:
+                    break
         return all_bars
 
     def get_latest_trades(self, symbols: list[str]) -> dict[str, dict]:
-        payload = self._data(
-            "GET",
-            "/v2/stocks/trades/latest",
-            params={"symbols": ",".join(symbols), "feed": self.settings.feed},
-        )
-        return payload.get("trades") or {}
+        trades: dict[str, dict] = {}
+        for chunk in self._chunked(symbols, 100):
+            payload = self._data(
+                "GET",
+                "/v2/stocks/trades/latest",
+                params={"symbols": ",".join(chunk), "feed": self.settings.feed},
+            )
+            trades.update(payload.get("trades") or {})
+        return trades
 
     def get_most_actives(self, top: int = 20, by: str = "volume") -> list[dict]:
         payload = self._data(
@@ -168,17 +178,28 @@ class AlpacaPaperClient:
     def get_daily_bars(
         self, symbols: list[str], start_iso: str, end_iso: str | None = None
     ) -> dict[str, list[dict]]:
-        payload = self._data(
-            "GET",
-            "/v2/stocks/bars",
-            params={
-                "symbols": ",".join(symbols),
-                "timeframe": "1Day",
-                "start": start_iso,
-                "end": end_iso,
-                "limit": 10_000,
-                "feed": self.settings.feed,
-                "adjustment": "raw",
-            },
-        )
-        return payload.get("bars") or {}
+        """Daily bars, batching symbols (cap-safe) + following pagination."""
+        all_bars: dict[str, list[dict]] = {}
+        for chunk in self._chunked(symbols, 100):
+            page_token = None
+            while True:
+                payload = self._data(
+                    "GET",
+                    "/v2/stocks/bars",
+                    params={
+                        "symbols": ",".join(chunk),
+                        "timeframe": "1Day",
+                        "start": start_iso,
+                        "end": end_iso,
+                        "limit": 10_000,
+                        "feed": self.settings.feed,
+                        "adjustment": "raw",
+                        "page_token": page_token,
+                    },
+                )
+                for symbol, bars in (payload.get("bars") or {}).items():
+                    all_bars.setdefault(symbol, []).extend(bars or [])
+                page_token = payload.get("next_page_token")
+                if not page_token:
+                    break
+        return all_bars

run_live_paper.py

@@ -373,6 +373,12 @@ def main(argv: list[str] | None = None) -> int:
         # pulls their bars and the watcher evaluates them
         added = [s for s in res.universe if s not in symbols]
         symbols.extend(added)
+        # Keep the watchlist BOUNDED so it rotates toward current movers instead
+        # of growing all session. (Ingestion is now batched so an over-cap list
+        # no longer 400s, but an unbounded list is still wasteful and stale.)
+        max_uni = int(os.environ.get("WATCHER_MAX_UNIVERSE", "80"))
+        if len(symbols) > max_uni:
+            del symbols[: len(symbols) - max_uni]
         return (f"universe={len(res.universe)} gappers={len(res.gappers)}"
                 + (f" +{len(added)} new" if added else ""))