From e74539d44938714036104535f22272fda39d90fc Mon Sep 17 00:00:00 2001 From: momentum dev Date: Thu, 18 Jun 2026 18:18:36 -0400 Subject: [PATCH] Resiliency + risk + monitoring: SIP backfill, freshness watchdog, gross cap, naked alarm MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- .env.example | 16 ++++++++++- alpaca_paper/client.py | 13 ++++++++- alpaca_paper/settings.py | 8 ++++++ eod_replay.py | 11 ++++++-- momentum_cli.py | 55 +++++++++++++++++++++++++++++++++++-- run_live_paper.py | 54 +++++++++++++++++++++++++++++++++++- tests/unit/test_triggers.py | 11 ++++++++ trading_execution.py | 39 ++++++++++++++++++++++++-- 8 files changed, 196 insertions(+), 11 deletions(-) diff --git a/.env.example b/.env.example index 674cb06..c8219a7 100644 --- a/.env.example +++ b/.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). diff --git a/alpaca_paper/client.py b/alpaca_paper/client.py index e25beaf..5e9d582 100644 --- a/alpaca_paper/client.py +++ b/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 diff --git a/alpaca_paper/settings.py b/alpaca_paper/settings.py index 86ed340..c6dae5f 100644 --- a/alpaca_paper/settings.py +++ b/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 diff --git a/eod_replay.py b/eod_replay.py index 2f3d494..98f6f20 100644 --- a/eod_replay.py +++ b/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) diff --git a/momentum_cli.py b/momentum_cli.py index 9206c08..65c149e 100644 --- a/momentum_cli.py +++ b/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) diff --git a/run_live_paper.py b/run_live_paper.py index 8e68801..ad4e608 100644 --- a/run_live_paper.py +++ b/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( diff --git a/tests/unit/test_triggers.py b/tests/unit/test_triggers.py index ce4178e..7808e0e 100644 --- a/tests/unit/test_triggers.py +++ b/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 diff --git a/trading_execution.py b/trading_execution.py index b9183ca..2a86a7b 100644 --- a/trading_execution.py +++ b/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: