VWAP entry gate on the live auto path + concentrate boot-guard

Wire the one validated selection signal (above-VWAP breakouts reach +1R
~1.5x as often, n=3,110 from the labeler lift report) into the LIVE auto
entry path (request_approvals_for_ready_signals) — NOT the dead fast
submit_breakout_now path where the cap and rank logic got stranded.

- setup_evaluator already computes above_vwap/levels.vwap; surface them in
  the setup dict (also an observability win) and pass through
  query_ready_signals_snapshot.
- New ExecutionSettings.require_above_vwap (TRADING_REQUIRE_ABOVE_VWAP, env
  default on). When a ready signal is below session VWAP: always emit a
  rule_type=vwap_below event (shadow measurement); only SKIP the entry when
  the flag is on. Fail-open: above_vwap=None (old signals) is never blocked.
- Boot log now prints the entry-gate state, and WARNS if
  TRADING_CONCENTRATE_TOP_N>0 (that rank switch only affects the dead fast
  path — see the rank-scaling audit).

Tests: gate skips below-VWAP when enforced, shadow-logs+allows when off,
allows above-VWAP, and fail-opens on the missing field.

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

run_live_paper.py

@@ -802,6 +802,17 @@ def main(argv: list[str] | None = None) -> int:
         if client else None
     )
     print(f"[boot] exit management: {exit_cfg.describe()}")
+    print(
+        f"[boot] entry gates: VWAP {'ENFORCED' if exec_settings.require_above_vwap else 'shadow-only'}"
+        f", daily-cap {exec_settings.max_fresh_entries_per_day or 'off'}"
+        f", risk-cap ${exec_settings.max_risk_dollars:.0f}"
+    )
+    if exec_settings.concentrate_top_n > 0:
+        print(
+            "[boot] WARNING: TRADING_CONCENTRATE_TOP_N>0, but rank-concentration is "
+            "enforced ONLY on the dead fast-trigger path, NOT the live auto-approval "
+            "path — it will not affect live entries (see entry-path-fast-vs-auto)."
+        )
 
     def step_manage_exits():
         if exit_mgr is None:

storage/projections.py

@@ -227,6 +227,8 @@ def query_ready_signals_snapshot(store, session_id: str | None = None) -> list[d
             "entry_price": signal_data.get("entry_price"),
             "stop_loss_price": signal_data.get("stop_loss_price"),
             "quality_score": signal_data.get("quality_score"),
+            "vwap": signal_data.get("vwap"),
+            "above_vwap": signal_data.get("above_vwap"),
             "timestamp": event.get("timestamp"),
         }
     return sorted(

strategy/evaluation/setup_evaluator.py

@@ -157,6 +157,10 @@ def evaluate_setup(
                 "quality_score": quality.score,
                 "quality_grade": quality.grade,
                 "confidence": round(score_pct / 100.0, 4),
+                # surfaced for the live VWAP entry gate + observability
+                "vwap": (round(float(levels.vwap), 4)
+                         if levels.vwap is not None else None),
+                "above_vwap": bool(above_vwap),
             }
         )
 

tests/unit/test_trading_execution.py

@@ -66,7 +66,17 @@ class Clock:
         self.t += timedelta(minutes=minutes)
 
 
-def emit_signal(store, symbol="GOOD", entry=14.0, stop=13.45):
+def emit_signal(store, symbol="GOOD", entry=14.0, stop=13.45,
+                above_vwap=None, vwap=None):
+    signal_data = {
+        "entry_price": entry,
+        "stop_loss_price": stop,
+        "quality_score": 0.7,
+    }
+    if above_vwap is not None:
+        signal_data["above_vwap"] = above_vwap
+        signal_data["vwap"] = vwap if vwap is not None else (
+            round(entry - 1.0, 2) if above_vwap else round(entry + 1.0, 2))
     store.emit(
         SignalReadyEvent(
             timestamp=T0,
@@ -76,11 +86,7 @@ def emit_signal(store, symbol="GOOD", entry=14.0, stop=13.45):
             symbol=symbol,
             signal_type="bull_flag",
             confidence=0.9,
-            signal_data={
-                "entry_price": entry,
-                "stop_loss_price": stop,
-                "quality_score": 0.7,
-            },
+            signal_data=signal_data,
         )
     )
 
@@ -209,6 +215,43 @@ def test_daily_cap_never_blocks_manual_approval(store):
     assert len(executor.executed) == 2                   # cap ignored for manual override
 
 
+def test_vwap_gate_skips_below_vwap_when_enforced(store):
+    emit_signal(store, symbol="LOWV", entry=14.0, stop=13.45, above_vwap=False, vwap=15.0)
+    service = make_service(store, require_above_vwap=True)
+    created = service.request_approvals_for_ready_signals()
+    assert created == []                                  # entry skipped
+    assert query_approval_queue(store) == []
+    vb = [json.loads(e["payload_json"]) for e in store.query_events(event_type="risk_rule_triggered")
+          if json.loads(e["payload_json"])["rule_type"] == "vwap_below"]
+    assert len(vb) == 1 and vb[0]["action_taken"] == "skipped_entry"
+
+
+def test_vwap_gate_shadow_logs_but_allows_when_off(store):
+    emit_signal(store, symbol="LOWV", entry=14.0, stop=13.45, above_vwap=False, vwap=15.0)
+    service = make_service(store, require_above_vwap=False)
+    created = service.request_approvals_for_ready_signals()
+    assert len(created) == 1                              # still entered (shadow mode)
+    vb = [json.loads(e["payload_json"]) for e in store.query_events(event_type="risk_rule_triggered")
+          if json.loads(e["payload_json"])["rule_type"] == "vwap_below"]
+    assert len(vb) == 1 and vb[0]["action_taken"] == "shadow_logged"
+
+
+def test_vwap_gate_allows_above_vwap(store):
+    emit_signal(store, symbol="HIV", entry=14.0, stop=13.45, above_vwap=True, vwap=13.0)
+    service = make_service(store, require_above_vwap=True)
+    created = service.request_approvals_for_ready_signals()
+    assert len(created) == 1
+    assert not [e for e in store.query_events(event_type="risk_rule_triggered")
+                if json.loads(e["payload_json"])["rule_type"] == "vwap_below"]
+
+
+def test_vwap_gate_fail_open_on_missing_field(store):
+    # signals lacking the above_vwap field (old signals) must NEVER be blocked
+    emit_signal(store, symbol="OLDV", entry=14.0, stop=13.45)  # no above_vwap
+    service = make_service(store, require_above_vwap=True)
+    assert len(service.request_approvals_for_ready_signals()) == 1
+
+
 def test_no_duplicate_requests_for_same_symbol(store):
     emit_signal(store)
     service = make_service(store)

trading_execution.py

@@ -80,6 +80,12 @@ class ExecutionSettings:
     # armed name sizes equally as before). Plus a hard per-DAY fresh-entry cap.
     concentrate_top_n: int = 0
     max_fresh_entries_per_day: int = 0   # 0 = no daily cap
+    # VWAP SELECTION GATE — the one validated entry-quality signal (above-VWAP
+    # breakouts reach +1R ~1.5x as often, n=3,110 from the labeler lift report).
+    # Every below-VWAP ready signal is shadow-logged (rule_type=vwap_below) for
+    # measurement; entries are only SKIPPED when this is True. Fail-open: a signal
+    # with above_vwap=None (missing the field) is never blocked.
+    require_above_vwap: bool = False
     max_daily_loss_pct: float = 0.03
     # on a daily-loss breach, also flatten open positions + cancel unfilled entries
     flatten_on_breach: bool = True
@@ -218,6 +224,7 @@ class ExecutionSettings:
             max_risk_dollars=float(values.get("TRADING_MAX_RISK_DOLLARS", "0.0")),
             concentrate_top_n=int(values.get("TRADING_CONCENTRATE_TOP_N", "0")),
             max_fresh_entries_per_day=int(values.get("TRADING_MAX_FRESH_ENTRIES_PER_DAY", "0")),
+            require_above_vwap=flag("TRADING_REQUIRE_ABOVE_VWAP", "1"),
             liquidity_max_volume_pct=float(
                 values.get("TRADING_LIQUIDITY_MAX_VOLUME_PCT", "0.0")
             ),
@@ -644,6 +651,34 @@ class TradingExecutionService:
             if not entry or not stop or stop >= entry:
                 continue
 
+            # VWAP selection gate — the one validated entry-quality signal (above-
+            # VWAP breakouts reach +1R ~1.5x as often, n=3,110). Every below-VWAP
+            # ready signal is shadow-logged for measurement; the entry is only
+            # SKIPPED when require_above_vwap is on. above_vwap is None on signals
+            # lacking the field -> fail-open, never block on missing data.
+            if signal.get("above_vwap") is False:
+                enforced = self.settings.require_above_vwap
+                self.store.emit(
+                    RiskRuleTriggeredEvent(
+                        timestamp=datetime.now(),
+                        mode=self.mode,
+                        correlation_id=self.session_id,
+                        message=(
+                            f"{symbol} ready BELOW session VWAP "
+                            f"(vwap={signal.get('vwap')}, entry={entry}) — "
+                            f"{'skipped' if enforced else 'shadow-logged'}"
+                        ),
+                        rule_type="vwap_below",
+                        rule_value=float(signal.get("vwap") or 0.0),
+                        current_state={"symbol": symbol, "entry": float(entry),
+                                       "enforced": enforced},
+                        action_taken="skipped_entry" if enforced else "shadow_logged",
+                    )
+                )
+                if enforced:
+                    self._requested_symbols.add(symbol)  # don't re-evaluate every tick
+                    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%).