Resiliency + risk + monitoring: SIP backfill, freshness watchdog, gross cap, naked alarm

From the multi-agent audit (ship-now, low-overfit-risk cluster):

INGESTION RESILIENCY
- client._request: 429/5xx now retry with Retry-After + exponential backoff
  (previously hard-failed, feeding silent-stale ingest).
- SIP backfill feed (settings.backfill_feed='sip'): live realtime stays IEX, but
  backtest/backfill ingestion uses the full consolidated tape — IEX sees only
  ~2-5% of true volume (verified: AAPL 22,797 vs 452,704), which was silently
  corrupting every rvol/$-volume decision and backtest. eod_replay wired to it.
- step_ingest freshness watchdog: emits a STALE event + board note when the
  newest bar is older than INGEST_STALE_SECONDS during RTH (45% of passes
  historically returned 0 rows with no error). Optional hard arm-block
  (INGEST_STALE_BLOCK, off by default).

RISK
- Portfolio gross-notional cap (TRADING_MAX_GROSS_NOTIONAL_PCT=0.60): both sizing
  paths now cap each entry to the remaining gross budget, so 6 x 16% can't pile
  the book into ~96% gross of correlated small-caps. Blocks/shrinks late entries.
- Fixed the duplicate TRADING_MAX_CONCURRENT_POSITIONS in .env.example.

MONITORING (UI)
- Naked/unprotected-position alarm: _unprotected_positions joins the latest
  position + order snapshots (order sync now flattens legs) and the TUI board
  shows a red UNPROTECTED banner the instant a held long lacks stop coverage —
  the exact blind spot when APWC/NOWL went naked. Verified it flags today's
  3 overnight positions.

Tests: gross-notional cap blocks when budget full; 50 pass.

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

.env.example

@@ -49,7 +49,7 @@ TRADING_EXECUTION_ENABLED=1
 TRADING_EXECUTION_INTERVAL_SECONDS=30
 TRADING_AUTO_APPROVE=0          # 0 = manual approval in the dashboard, 1 = auto-arm (no approval)
 TRADING_MAX_ORDERS_PER_TICK=1
-TRADING_MAX_CONCURRENT_POSITIONS=3
+# NOTE: concurrency is set ONCE in the risk section below (was duplicated here).
 TRADING_RISK_PER_TRADE_PCT=0.01 # fraction of equity risked per trade
 # --- entry mechanism (Ross-Cameron-style auto-arm; all tunable) ---
 TRADING_REWARD_MULTIPLE=2.0     # target = entry + this * (entry - stop); 2R default
@@ -80,6 +80,16 @@ STRATEGY_SETUPS=opening_range_break
 LIVE_BARS_ENABLED=1
 LIVE_BARS_INTERVAL_SECONDS=60
 LIVE_BARS_LOOKBACK_MINUTES=120
+# Data feed: live realtime uses iex (free, realtime, but ~2-5% of true volume).
+# Backfill/backtest uses sip (full consolidated tape) so rvol/$-volume reflect
+# REAL volume — the iex thinness silently corrupts every volume-based decision.
+ALPACA_DATA_FEED=iex
+ALPACA_BACKFILL_FEED=sip
+# Freshness watchdog: warn (emit + display) when the newest bar is older than
+# this during RTH. INGEST_STALE_BLOCK=1 also refuses to ARM on stale data
+# (off by default so a mis-tuned threshold can't silently halt trading).
+INGEST_STALE_SECONDS=180
+INGEST_STALE_BLOCK=0
 
 # Broker state sync
 ALPACA_PAPER_SYNC_ENABLED=1
@@ -109,6 +119,10 @@ ORB_BARS=5                        # opening range = first N regular-hours minute
 TRADING_TRIGGER_SLIP_PCT=0.004    # cap the chase above the trigger on the fill
 # Account-aware sizing (matters most on a small REAL account):
 TRADING_MAX_POSITION_PCT=0.16     # one position's $ value <= this fraction of equity
+# PORTFOLIO gross-notional cap: total $ across ALL open+pending positions stays
+# under this fraction of equity. Without it, 6 concurrent x 16% = ~96% gross in
+# correlated small-cap gappers. With it, later entries shrink/blocked to fit.
+TRADING_MAX_GROSS_NOTIONAL_PCT=0.60
 # Liquidity-aware sizing: shares <= this fraction of the name's volume, so a
 # thin name is sized DOWN automatically instead of taking a full position it
 # can't absorb. 0 = off (fine at tiny size).

alpaca_paper/client.py

@@ -123,7 +123,18 @@ class AlpacaPaperClient:
                     raw = resp.read().decode()
                     return json.loads(raw) if raw else {}
             except urllib.error.HTTPError as exc:
-                # a real API response (4xx/5xx) — do not retry, surface it
+                # 429 (rate limit) and 5xx (server) are TRANSIENT — retry with
+                # backoff, honoring Retry-After. Previously these hard-failed,
+                # which fed silent-stale ingest (a rate-limited bar fetch looked
+                # like a clean "0 rows" pass). Other 4xx are real — surface them.
+                if exc.code in (429, 500, 502, 503, 504) and attempt < self.max_retries:
+                    retry_after = exc.headers.get("Retry-After") if exc.headers else None
+                    try:
+                        delay = float(retry_after) if retry_after else 0.0
+                    except (TypeError, ValueError):
+                        delay = 0.0
+                    _time.sleep(max(delay, 0.5 * (2 ** attempt)))  # exp backoff floor
+                    continue
                 raise AlpacaApiError(exc.code, exc.read().decode(errors="replace")) from exc
             except urllib.error.URLError:
                 # transient network/DNS (e.g. gaierror) — retry a couple times

alpaca_paper/settings.py

@@ -12,7 +12,13 @@ class AlpacaPaperSettings:
     secret_key: str = ""
     trading_base_url: str = "https://paper-api.alpaca.markets"
     data_base_url: str = "https://data.alpaca.markets"
+    # live realtime feed. IEX is free + realtime but THIN (~2-5% of true volume),
+    # which corrupts every rvol / $-volume decision computed off it.
     feed: str = "iex"
+    # historical/backfill feed. SIP is the consolidated tape (full volume) — use
+    # it for backtests/backfill so rvol & $-volume reflect reality. SIP realtime
+    # has a ~15-min embargo on free plans, which is why live stays on `feed`.
+    backfill_feed: str = "sip"
 
     @classmethod
     def from_env(cls, env: dict[str, str] | None = None) -> "AlpacaPaperSettings":
@@ -27,11 +33,13 @@ class AlpacaPaperSettings:
         )
         base = values.get("APCA_API_BASE_URL", "https://paper-api.alpaca.markets")
         feed = values.get("ALPACA_DATA_FEED", "iex")
+        backfill_feed = values.get("ALPACA_BACKFILL_FEED", "sip")
         return cls(
             api_key=api_key,
             secret_key=secret,
             trading_base_url=base.rstrip("/"),
             feed=feed,
+            backfill_feed=backfill_feed,
         )
 
     @property

eod_replay.py

@@ -107,9 +107,14 @@ def main(argv=None):
     print(f"daily-screen gappers (${PRICE_MIN:g}-{PRICE_MAX:g}, gap {GAP_MIN:g}-{GAP_MAX:g}%, "
           f"rvol>={RVOL_MIN:g}): {len(gapper_syms)}")
 
-    # 3) minute-backfill ONLY the gappers (fast), then rank them, and grow the dataset
-    ing = ingest_live_minute_bars(con, client, gapper_syms, lookback_minutes=440)
-    print(f"backfilled {ing.minute_rows} minute rows across {len(ing.symbols)} gappers\n")
+    # 3) minute-backfill ONLY the gappers (fast), then rank them, and grow the
+    # dataset. Use the SIP (full-volume) feed so rvol/$-volume in the replay
+    # reflect REAL volume — the live IEX feed sees only ~2-5% of it, which was
+    # silently corrupting every volume-based backtest conclusion.
+    ing = ingest_live_minute_bars(con, client, gapper_syms, lookback_minutes=440,
+                                  feed=client.settings.backfill_feed)
+    print(f"backfilled {ing.minute_rows} minute rows ({client.settings.backfill_feed}) "
+          f"across {len(ing.symbols)} gappers\n")
     keep = set(gapper_syms)
     gappers = [g for g in scan_gappers(con, sess, min_gap_pct=GAP_MIN, min_relative_volume=RVOL_MIN,
                                        price_min=PRICE_MIN, price_max=PRICE_MAX, limit=200)

momentum_cli.py

@@ -554,6 +554,46 @@ def _open_unrealized(con):
     return total
 
 
+_PROT_ACTIVE = {"held", "new", "accepted", "pending_new",
+                "accepted_for_bidding", "partially_filled"}
+
+
+def _unprotected_positions(con):
+    """Long positions whose RESTING protective sell coverage is short of the held
+    quantity — i.e. naked / under-protected. Computed from the latest synced
+    positions + orders snapshots (the order sync now flattens bracket legs, so a
+    held stop/TP leg is visible). This is the alarm we lacked when APWC/NOWL went
+    naked and the board showed nothing. Returns [{symbol, qty, protected}]."""
+    pos_ev = con.execute(
+        "SELECT payload_json FROM events WHERE event_type='account_positions_updated' "
+        "ORDER BY timestamp DESC LIMIT 1").fetchall()
+    ord_ev = con.execute(
+        "SELECT payload_json FROM events WHERE event_type='account_orders_updated' "
+        "ORDER BY timestamp DESC LIMIT 1").fetchall()
+    if not pos_ev:
+        return []
+    positions = json.loads(pos_ev[0][0]).get("positions") or []
+    orders = json.loads(ord_ev[0][0]).get("orders") or [] if ord_ev else []
+    sell_cover: dict = {}
+    for o in orders:
+        if (o.get("side") == "sell" and o.get("status") in _PROT_ACTIVE
+                and o.get("type") in ("stop", "stop_limit", "limit", "trailing_stop")):
+            sym = o.get("symbol")
+            qty = float(o.get("quantity") or 0) - float(o.get("filled_quantity") or 0)
+            sell_cover[sym] = sell_cover.get(sym, 0.0) + max(qty, 0.0)
+    out = []
+    for p in positions:
+        sym = p.get("symbol")
+        try:
+            qty = float(p.get("quantity") if p.get("quantity") is not None
+                        else p.get("qty") or 0)
+        except (TypeError, ValueError):
+            qty = 0.0
+        if qty > 0 and sell_cover.get(sym, 0.0) < qty - 1e-6:  # long, under-covered
+            out.append({"symbol": sym, "qty": qty, "protected": sell_cover.get(sym, 0.0)})
+    return out
+
+
 def _matched_realized(day_pnl, open_unreal, fifo_realized):
     """Broker-authoritative realized for the header: day P&L minus current open
     unrealized, so the parenthetical reconciles exactly (matched + open == day
@@ -650,16 +690,25 @@ def _render_board(con):
         rk.add_row(str(ts)[11:19], m or "")
 
     n_eval = len(_latest_eval_board(con))
+    # naked/under-protected alarm — the failure that cost us this session and the
+    # board never showed. A red banner the instant a held long lacks stop coverage.
+    naked = _unprotected_positions(con)
+    naked_line = ""
+    if naked:
+        names = ", ".join(f"{n['symbol']}({n['protected']:.0f}/{n['qty']:.0f})" for n in naked)
+        naked_line = (f"\n[bold white on red] ⚠ UNPROTECTED: {names} — held long with no/short "
+                      f"protective stop [/]")
     header = Panel(
         f"[bold]momentum live[/bold]   "
         f"[{pcolor}]{phase}[/{pcolor}] [dim]{detail}[/dim]   "
         f"equity [green]{eq}[/green]   "
         f"day P&L [bold {dcol}]{day_pnl:+,.0f}[/]   "
         f"[{hb_color}]♥ {hb_text}[/{hb_color}]   "
-        f"[bold cyan]{armed_n} armed[/bold cyan] · {n_eval} scanning\n"
-        f"[dim]hunting $1–20 gappers (≥ gap% on ≥ 2× volume); buy the break of the "
+        f"[bold cyan]{armed_n} armed[/bold cyan] · {n_eval} scanning"
+        f"{naked_line}\n"
+        f"[dim]hunting $1–5 gappers (≥ gap% on ≥ 2× volume); buy the break of the "
         f"opening-range high, stop at its low · Ctrl-C to exit[/dim]",
-        style="cyan")
+        style="red" if naked else "cyan")
     return Group(header, tt, st, pt, rk)
 
 

run_live_paper.py

@@ -347,6 +347,12 @@ def main(argv: list[str] | None = None) -> int:
     lookback = int(os.environ.get("LIVE_BARS_LOOKBACK_MINUTES", "240"))
     refresh_lb = int(os.environ.get("LIVE_BARS_REFRESH_MINUTES", "20"))
     _ingest_first = {"v": True}
+    # freshness watchdog: warn when the newest bar is older than this during RTH.
+    _ingest_stale_secs = float(os.environ.get("INGEST_STALE_SECONDS", "180"))
+    # block arming on stale data? OFF by default — observability first; a
+    # mis-tuned threshold must not silently halt trading.
+    _ingest_stale_block = os.environ.get("INGEST_STALE_BLOCK", "0").strip() in {"1", "true", "yes"}
+    _data_stale = {"v": False}
 
     def _first_pass_lookback() -> int:
         # The first ingest must cover the whole session-so-far so the
@@ -363,6 +369,31 @@ def main(argv: list[str] | None = None) -> int:
             since_open = 0.0
         return int(max(lookback, since_open + 15))
 
+    def _freshest_bar_age():
+        """Seconds since the newest stored minute bar (UTC), or None. The bar
+        timestamp is its START, and IEX lags ~1 bar, so ~90-120s is NORMAL during
+        RTH — only flag well beyond that."""
+        from datetime import timezone
+        try:
+            row = rt["research_con"].execute(
+                "SELECT max(timestamp) FROM minute_bars WHERE session_date = ?",
+                [_now_session_date()]).fetchone()
+            if not row or not row[0]:
+                return None
+            ts = row[0]
+            if isinstance(ts, str):
+                ts = datetime.fromisoformat(ts.replace("Z", "+00:00"))
+            if ts.tzinfo is None:
+                ts = ts.replace(tzinfo=timezone.utc)
+            return (datetime.now(timezone.utc) - ts).total_seconds()
+        except Exception:  # noqa: BLE001
+            return None
+
+    def _is_rth():
+        from zoneinfo import ZoneInfo
+        n = datetime.now(ZoneInfo("America/New_York"))
+        return n.weekday() < 5 and (9, 30) <= (n.hour, n.minute) < (16, 0)
+
     def step_ingest():
         if not client:
             return "skipped (no keys)"
@@ -372,6 +403,21 @@ def main(argv: list[str] | None = None) -> int:
         else:
             lb = refresh_lb
         res = ingest_live_minute_bars(rt["research_con"], client, symbols, lookback_minutes=lb)
+        # Freshness watchdog: 45% of passes historically returned 0 rows with NO
+        # error — indistinguishable from a feed outage. Make staleness VISIBLE
+        # (emit + display); hard-block arming only if explicitly enabled, so a
+        # mis-tuned threshold can't silently halt trading.
+        age = _freshest_bar_age()
+        stale = bool(_is_rth() and age is not None and age > _ingest_stale_secs)
+        _data_stale["v"] = stale
+        if stale:
+            rt["store"].emit(ModuleTickEvent(
+                timestamp=datetime.now(), mode=EventMode.PAPER,
+                correlation_id=rt["session_id"],
+                message=f"INGEST STALE: freshest bar {age:.0f}s old (> {_ingest_stale_secs:.0f}s)",
+                module="ingestion", stage="stale", duration_ms=0.0,
+                input_count=len(symbols), output_count=res.minute_rows,
+                metrics={"bar_age_s": age}, errors=[{"error": "stale_data"}]))
         rt["store"].emit(
             ModuleTickEvent(
                 timestamp=datetime.now(),
@@ -387,7 +433,9 @@ def main(argv: list[str] | None = None) -> int:
                 errors=[{"error": e} for e in res.errors],
             )
         )
-        return f"{res.minute_rows} rows (lookback {lb}m)" + (f", errors={res.errors}" if res.errors else "")
+        note = (f" STALE({age:.0f}s)" if stale else "")
+        return (f"{res.minute_rows} rows (lookback {lb}m){note}"
+                + (f", errors={res.errors}" if res.errors else ""))
 
     def step_watch():
         res = rt["watcher"].tick(_now_session_date())
@@ -465,6 +513,10 @@ def main(argv: list[str] | None = None) -> int:
         from research.ingestion.signals import scan_gappers
         from strategy.evaluation.structure import opening_range
 
+        # don't arm new triggers on stale data when blocking is enabled — a break
+        # computed off a frozen feed could fire into a name that has already moved.
+        if _ingest_stale_block and _data_stale["v"]:
+            return "skipped: data stale (INGEST_STALE_BLOCK)"
         session_date = _now_session_date()
         # fresh catalysts (the discover step ingests the feeds; we just read them)
         news_map = recent_news_map(

tests/unit/test_triggers.py

@@ -183,6 +183,17 @@ def test_breakout_now_dedups_second_attempt():
     assert res2["ok"] is False and res2["skipped"] == "already_active"
 
 
+def test_breakout_now_blocks_when_gross_notional_full():
+    """Portfolio gross-notional cap: when open positions already fill the budget,
+    a new entry is blocked instead of piling the book toward ~100% gross."""
+    # equity 100k, cap 0.60 -> 60k budget; an open position worth 60k exhausts it
+    client = _FakeClient(positions=[{"symbol": "HELD", "qty": "1",
+                                     "market_value": "60000"}])
+    store, svc = _svc(client)
+    res = svc.submit_breakout_now("AAA", trigger=10.0, stop=9.5, last_price=10.0)
+    assert res["ok"] is False and res["skipped"] == "gross_notional_cap"
+
+
 def test_breakout_now_rejects_bad_geometry():
     store, svc = _svc(_FakeClient())
     res = svc.submit_breakout_now("AAA", 10.0, 10.5, last_price=10.0)  # stop > entry

trading_execution.py

@@ -108,6 +108,12 @@ class ExecutionSettings:
     # liquidity cap: shares <= this fraction of the symbol's day volume so at
     # SIZE you don't move the market. 0 = off (irrelevant at small size).
     liquidity_max_volume_pct: float = 0.0
+    # PORTFOLIO gross-notional cap: total $ across ALL open+pending positions must
+    # stay under this fraction of equity. Without it, max_concurrent (6) x
+    # max_position_pct (0.16) = ~96% gross in correlated small-cap gappers — one
+    # regime move hits the whole book together. A later entry is shrunk to fit the
+    # remaining budget, or blocked if none remains. 0 = off.
+    max_gross_notional_pct: float = 0.60
     # --- backout cooldown (anti-thrash) ------------------------------------
     # after an entry backs out (timeout / price-break) the symbol is benched for
     # this many seconds before it can re-arm. Without it a breakout that won't
@@ -165,6 +171,9 @@ class ExecutionSettings:
             liquidity_max_volume_pct=float(
                 values.get("TRADING_LIQUIDITY_MAX_VOLUME_PCT", "0.0")
             ),
+            max_gross_notional_pct=float(
+                values.get("TRADING_MAX_GROSS_NOTIONAL_PCT", "0.60")
+            ),
             backout_cooldown_seconds=float(
                 values.get("TRADING_BACKOUT_COOLDOWN_SECONDS", "180")
             ),
@@ -286,6 +295,23 @@ class TradingExecutionService:
                 pass
         return self.equity
 
+    def _remaining_gross_budget(self, equity: float) -> float:
+        """$ of new notional still allowed under the portfolio gross-notional cap
+        (cap*equity minus the market value of currently-open positions). Inf when
+        the cap is disabled. Caps how big the NEXT entry can be so the whole book
+        can't pile into ~100% gross of correlated small-caps."""
+        cap = float(self.settings.max_gross_notional_pct or 0.0)
+        if cap <= 0:
+            return float("inf")
+        positions = self._broker_positions() or []
+        open_gross = 0.0
+        for p in positions:
+            try:
+                open_gross += abs(float(p.get("market_value") or 0.0))
+            except (TypeError, ValueError):
+                pass
+        return max(0.0, cap * equity - open_gross)
+
     def _broker_held_symbols(self) -> set[str] | None:
         """Symbols the broker reports as OPEN POSITIONS right now (truth).
 
@@ -504,6 +530,10 @@ class TradingExecutionService:
                 continue
 
             eq = self._current_equity()
+            # cap this entry to BOTH the per-position limit AND the portfolio's
+            # remaining gross-notional budget (so the book can't pile into ~100%).
+            pos_cap = min(eq * self.settings.max_position_pct,
+                          self._remaining_gross_budget(eq))
             sizing = calculate_position_size(
                 float(entry),
                 float(stop),
@@ -512,7 +542,7 @@ class TradingExecutionService:
                     risk_per_trade_pct=self.settings.risk_per_trade_pct,
                     default_equity=self.settings.default_equity,
                 ),
-                max_position_value=eq * self.settings.max_position_pct,
+                max_position_value=pos_cap,
             )
             if sizing.position_size <= 0:
                 continue
@@ -650,6 +680,11 @@ class TradingExecutionService:
         eq = self._current_equity()
         liq = self.settings.liquidity_max_volume_pct
         max_shares = int(liq * cum_volume) if (liq > 0 and cum_volume > 0) else None
+        # per-position cap AND remaining portfolio gross-notional budget
+        pos_cap = min(eq * self.settings.max_position_pct,
+                      self._remaining_gross_budget(eq))
+        if pos_cap <= 0:
+            return {"ok": False, "skipped": "gross_notional_cap"}
         sizing = calculate_position_size(
             entry_ref,
             stop_f,
@@ -658,7 +693,7 @@ class TradingExecutionService:
                 risk_per_trade_pct=self.settings.risk_per_trade_pct,
                 default_equity=self.settings.default_equity,
             ),
-            max_position_value=eq * self.settings.max_position_pct,
+            max_position_value=pos_cap,
             max_shares=max_shares,
         )
         if sizing.position_size <= 0: