entry-gate: extract shared _evaluate_entry_gates for both paths (plan 005 step 2)

One ordered gate sequence (session, confirmation-speed, population, VWAP,
quality, concentrate-by-rank, daily cap [fast-only], dedup/cooldown/reentry,
dilution veto, anti-chase) that BOTH request_approvals_for_ready_signals and
submit_breakout_now call. Gate bodies lifted verbatim from whichever path
owned them; slow-path risk events preserved source-guarded so the fast path
gains no new events. The two documented asymmetries now close:
  - fast path routes through the VWAP gate (was ungated),
  - slow path honours concentrate-by-rank via a rank_map threaded from the
    trigger book into tick()/request_approvals (no-op at concentrate_top_n=0,
    which is the production config — STOP #3 does not apply).

Price-sanity/entry_ref coercion stays path-native (different data sources);
slow-path dedup pre-filter kept first so per-tick re-scans don't re-emit.
16 characterization tests pin both paths; the only decision diffs vs
pre-extraction are the two closures above.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

run_live_paper.py

@@ -735,7 +735,13 @@ def main(argv: list[str] | None = None) -> int:
     def step_execute():
         if not rt["execution"]:
             return "skipped (no keys)"
-        res = rt["execution"].tick()
+        # Hand the trigger book's current ranks to the slow auto path so it honours
+        # concentrate-by-rank identically to the fast path (plan 005). None-safe: an
+        # unranked/absent symbol -> factor 1.0 (no-op). Built inline from the book's
+        # thread-safe trigger dict — no accessor added to runtime/triggers.py.
+        _bk = rt.get("book")
+        rank_map = ({sym: t.rank for sym, t in _bk.triggers.items()} if _bk else None)
+        res = rt["execution"].tick(rank_map=rank_map)
         backed = len(res.get("backed_out") or [])
         # `auto` is the raw approve_order() results: a REAL broker submission carries a
         # broker_order_id; rejections (e.g. the daily-entry cap) are tagged rejected=True.

tests/unit/test_entry_gate_parity.py

@@ -142,15 +142,54 @@ def test_slow_late_confirmation_skips(store):
     assert "late_confirmation" in _reasons(store)
 
 
-def test_slow_ignores_rank_today(store):
-    # STEP 2 CLOSURE: today the slow path has NO rank gate, so a name that would
-    # rank outside the top-N still gets an approval. Step 2 threads a rank_map and
-    # flips this to a rank_concentration skip.
-    _emit(store, gap_pct=0.25)
-    created = _make(store, concentrate_top_n=3).request_approvals_for_ready_signals()
+def test_slow_rank_gate_closure(store):
+    # STEP 2 CLOSURE (plan 005): the slow path now honours concentrate-by-rank when
+    # a rank_map is threaded in from the trigger book. A name ranked outside the
+    # top-N is rejected — before extraction the slow path had no rank gate and this
+    # same call created an approval (see git history of this test).
+    _emit(store, symbol="RANKED", gap_pct=0.25)
+    svc = _make(store, concentrate_top_n=3)
+    created = svc.request_approvals_for_ready_signals(rank_map={"RANKED": 5})
+    assert created == []
+
+
+def test_slow_rank_within_top_n_still_creates(store):
+    # A ranked name INSIDE the top-N still gets an approval (rank-1 = full size).
+    _emit(store, symbol="RANKED", gap_pct=0.25)
+    svc = _make(store, concentrate_top_n=3)
+    created = svc.request_approvals_for_ready_signals(rank_map={"RANKED": 1})
     assert len(created) == 1
 
 
+def test_fast_vwap_gate_closure(store):
+    # STEP 2 CLOSURE (plan 005): the fast path now routes through the shared gate,
+    # which contains the VWAP check. Demonstrated directly on _evaluate_entry_gates
+    # with source="fast" (the fast public signature carries no above_vwap, so the
+    # closure is only reachable when a below-VWAP value is supplied). Before
+    # extraction the fast path had no VWAP gate at all.
+    from trading_execution import EntryContext
+    svc = _make(store, require_above_vwap=True)
+    ctx = EntryContext(symbol="X", source="fast", entry=14.0, stop=13.45,
+                       trigger=14.0, entry_ref=14.0, gap_pct=0.25, above_vwap=False,
+                       vwap=15.0, confirm_ts=T0)
+    decision = svc._evaluate_entry_gates(ctx)
+    assert decision.passed is False
+    assert decision.reason == "vwap_below"
+
+
+def test_gate_parity_below_vwap_enforced_both_paths(store):
+    # With above_vwap supplied, BOTH sources now reach the same vwap_below verdict —
+    # the asymmetry is closed. (The auto source also emits the shadow/skip event;
+    # the fast source stays event-free, only the decision unifies.)
+    from trading_execution import EntryContext
+    svc = _make(store, require_above_vwap=True)
+    for src in ("auto", "fast"):
+        d = svc._evaluate_entry_gates(EntryContext(
+            symbol="X", source=src, entry=14.0, stop=13.45, trigger=14.0,
+            entry_ref=14.0, gap_pct=0.25, above_vwap=False, vwap=15.0, confirm_ts=T0))
+        assert d.reason == "vwap_below", src
+
+
 # ---- FAST path (submit_breakout_now): decision = skipped string ----
 
 def test_fast_in_population_fires(store):

trading_execution.py

@@ -19,7 +19,7 @@ import json
 import logging
 import os
 import threading
-from dataclasses import dataclass
+from dataclasses import dataclass, field
 from datetime import datetime, timedelta
 
 from runtime.flatten import (buy_fills_from_orders, cancel_protective_and_close,
@@ -331,6 +331,46 @@ class ExecutionSettings:
         )
 
 
+@dataclass
+class EntryContext:
+    """Everything the shared entry-gate sequence needs about one candidate.
+
+    Built by each path from its own data source (the slow path from a ready-signal
+    snapshot, the fast path from a live trigger), then handed to
+    ``_evaluate_entry_gates``. Fields the source lacks stay None -> fail-open."""
+
+    symbol: str
+    source: str                    # "auto" (slow) | "fast"
+    entry: float                   # entry/limit reference level
+    stop: float
+    trigger: float                 # anti-chase trigger (slow: == entry_ref)
+    entry_ref: float               # price used for anti-chase distance
+    gap_pct: float | None = None
+    price: float | None = None     # price for the population band
+    cum_volume: float | None = None
+    rvol: float | None = None
+    above_vwap: bool | None = None
+    vwap: float | None = None
+    quality_score: float | None = None
+    rank: int | None = None
+    day_open: float | None = None
+    halted: bool = False
+    confirm_ts: object = None       # ready time (slow) / now() (fast) for late gate
+    held: set | None = None         # precomputed once-per-tick on the slow path
+    pending: set | None = None
+
+
+@dataclass
+class GateDecision:
+    """First-reject-or-pass outcome of the shared entry-gate sequence."""
+
+    passed: bool
+    reason: str | None = None       # reject reason / fast-path `skipped` string
+    rank_factor: float = 1.0        # risk scale from concentrate-by-rank
+    throttle: bool = False          # slow path: bench symbol from re-eval this session
+    gates: dict = field(default_factory=dict)  # per-gate {value,threshold,outcome} receipt
+
+
 class TradingExecutionService:
     """Drives the signal -> approval -> order pipeline each tick."""
 
@@ -844,7 +884,186 @@ class TradingExecutionService:
         return population_reject(
             cfg, gap_pct=gap_pct, price=price, volume=cum_volume, rvol=rvol)
 
-    def request_approvals_for_ready_signals(self) -> list[str]:
+    def _evaluate_entry_gates(self, ctx: EntryContext) -> GateDecision:
+        """The ONE ordered entry-gate sequence, shared by the slow auto-approval
+        path and the fast trigger path (plan 005). Returns the first reject or a
+        pass; each gate body is the one that lived on whichever path owned it.
+
+        Emission convention (topology, not policy): the slow path's per-gate risk
+        events are preserved verbatim, source=="auto"-guarded, so the fast path
+        gains no new events — its observable decision stays the returned reason /
+        `skipped` string. The dilution veto emits on both paths (as it did today).
+        Price-sanity / entry_ref derivation stays in each caller (path-specific
+        float coercion from different data sources)."""
+        gates: dict = {}
+
+        def _done(passed, reason=None, throttle=False, rank_factor=1.0):
+            decision = GateDecision(passed=passed, reason=reason,
+                                    rank_factor=rank_factor, throttle=throttle,
+                                    gates=gates)
+            self._emit_gate_receipt(ctx, decision)
+            return decision
+
+        is_auto = ctx.source == "auto"
+
+        # SESSION — quiet skip on both paths (outside-session is normal, not a risk event)
+        if self._outside_session():
+            gates["session"] = {"outcome": "reject"}
+            return _done(False, "outside_session", throttle=False)
+        gates["session"] = {"outcome": "pass"}
+
+        # CONFIRMATION-SPEED gate — only early-confirming breakouts convert.
+        if self._late_confirmation(ctx.confirm_ts):
+            gates["late_confirmation"] = {
+                "value": self._confirm_minute_of(ctx.confirm_ts),
+                "threshold": self.settings.entry_confirm_by_minute, "outcome": "reject"}
+            if is_auto:
+                self.store.emit(RiskRuleTriggeredEvent(
+                    timestamp=datetime.now(), mode=self.mode,
+                    correlation_id=self.session_id,
+                    message=(f"{ctx.symbol} ready too late "
+                             f"(confirm minute {self._confirm_minute_of(ctx.confirm_ts):.0f}"
+                             f" > {self.settings.entry_confirm_by_minute}) — skipped"),
+                    rule_type="late_confirmation",
+                    rule_value=float(self.settings.entry_confirm_by_minute),
+                    current_state={"symbol": ctx.symbol, "ready_at": str(ctx.confirm_ts)},
+                    action_taken="skipped_entry"))
+            return _done(False, "late_confirmation", throttle=is_auto)
+        gates["late_confirmation"] = {"outcome": "pass"}
+
+        # POPULATION gate — the year-long +EV boundary (gap band, volume, price).
+        pop_reason = self._population_reject(
+            ctx.gap_pct, ctx.cum_volume, price=ctx.price, rvol=ctx.rvol)
+        if pop_reason:
+            gates["population"] = {"value": ctx.gap_pct,
+                                   "threshold": self.settings.min_entry_gap_pct,
+                                   "outcome": "reject"}
+            if is_auto:
+                self.store.emit(RiskRuleTriggeredEvent(
+                    timestamp=datetime.now(), mode=self.mode,
+                    correlation_id=self.session_id,
+                    message=f"{ctx.symbol} outside tradeable population ({pop_reason}) — skipped",
+                    rule_type="population_below",
+                    rule_value=float(self.settings.min_entry_gap_pct),
+                    current_state={"symbol": ctx.symbol, "gap_pct": ctx.gap_pct,
+                                   "cum_volume": ctx.cum_volume},
+                    action_taken="skipped_entry"))
+            return _done(False, "population_below", throttle=is_auto)
+        gates["population"] = {"outcome": "pass"}
+
+        # VWAP selection gate — below-VWAP is always shadow-logged (slow path);
+        # entries are only SKIPPED when require_above_vwap is on. Fail-open on None.
+        # CLOSURE (plan 005): the fast path now routes through this gate too — when
+        # above_vwap is supplied it enforces (the fast public signature carries no
+        # above_vwap today, so in practice it stays fail-open there).
+        if ctx.above_vwap is False:
+            enforced = self.settings.require_above_vwap
+            gates["vwap"] = {"value": ctx.vwap, "threshold": "above_vwap",
+                             "outcome": "reject" if enforced else "shadow"}
+            if is_auto:
+                self.store.emit(RiskRuleTriggeredEvent(
+                    timestamp=datetime.now(), mode=self.mode,
+                    correlation_id=self.session_id,
+                    message=(f"{ctx.symbol} ready BELOW session VWAP "
+                             f"(vwap={ctx.vwap}, entry={ctx.entry}) — "
+                             f"{'skipped' if enforced else 'shadow-logged'}"),
+                    rule_type="vwap_below", rule_value=float(ctx.vwap or 0.0),
+                    current_state={"symbol": ctx.symbol, "entry": float(ctx.entry),
+                                   "enforced": enforced},
+                    action_taken="skipped_entry" if enforced else "shadow_logged"))
+            if enforced:
+                return _done(False, "vwap_below", throttle=True)
+
+        # QUALITY-GRADE gate — shadow-logged; only SKIPPED when min_quality_score>0.
+        qs = ctx.quality_score
+        if qs is not None and self.settings.min_quality_score > 0 and qs < self.settings.min_quality_score:
+            gates["quality"] = {"value": qs,
+                                "threshold": self.settings.min_quality_score,
+                                "outcome": "reject"}
+            if is_auto:
+                self.store.emit(RiskRuleTriggeredEvent(
+                    timestamp=datetime.now(), mode=self.mode,
+                    correlation_id=self.session_id,
+                    message=(f"{ctx.symbol} ready below quality gate "
+                             f"(score {qs:.2f} < {self.settings.min_quality_score:.2f}) — skipped"),
+                    rule_type="quality_below", rule_value=float(qs),
+                    current_state={"symbol": ctx.symbol, "quality_score": float(qs),
+                                   "min": self.settings.min_quality_score},
+                    action_taken="skipped_entry"))
+            return _done(False, "quality_below", throttle=True)
+
+        # CONCENTRATE-BY-RANK — only the top-N armed names enter; risk scales down by
+        # rank. CLOSURE (plan 005): now runs on the slow path too when a rank is
+        # present (rank is None on the slow path unless a rank_map is threaded in ->
+        # factor 1.0, no-op). No event on either path (matches the fast path today).
+        rank_factor = rank_risk_factor(ctx.rank or 0, self.settings.concentrate_top_n)
+        gates["rank"] = {"value": ctx.rank, "threshold": self.settings.concentrate_top_n,
+                         "outcome": "reject" if rank_factor <= 0.0 else "pass"}
+        if rank_factor <= 0.0:
+            return _done(False, "rank_concentration", throttle=is_auto,
+                         rank_factor=rank_factor)
+
+        # DAILY FRESH-ENTRY CAP — fast path only (the slow path's cap is enforced
+        # later, in approve_order, and stays there untouched).
+        if ctx.source == "fast":
+            cap = self.settings.max_fresh_entries_per_day
+            if cap > 0:
+                today = self._now().date().isoformat() if hasattr(self._now(), "date") else None
+                if today is not None:
+                    if self._fresh_entries.get("date") != today:
+                        self._fresh_entries = {"date": today, "n": 0}
+                    if self._fresh_entries["n"] >= cap:
+                        gates["daily_cap"] = {"value": self._fresh_entries["n"],
+                                              "threshold": cap, "outcome": "reject"}
+                        return _done(False, "daily_entry_cap", throttle=False,
+                                     rank_factor=rank_factor)
+
+        # DEDUP / BENCH / RE-ENTRY — never double-enter a held/pending/requested name,
+        # a benched (recently backed-out) name, or one already round-tripped today.
+        held = ctx.held if ctx.held is not None else self._held_symbols()
+        pending = (ctx.pending if ctx.pending is not None
+                   else {row["symbol"] for row in query_approval_queue(self.store)})
+        if ctx.symbol in held or ctx.symbol in pending or ctx.symbol in self._requested_symbols:
+            return _done(False, "already_active", throttle=False, rank_factor=rank_factor)
+        if self._in_cooldown(ctx.symbol):
+            return _done(False, "cooldown", throttle=False, rank_factor=rank_factor)
+        if ctx.symbol in self._exited_today:
+            return _done(False, "reentry_blocked", throttle=False, rank_factor=rank_factor)
+
+        # DILUTION VETO — emits on BOTH paths (as today).
+        vetoed = self._dilution_vetoed(ctx.symbol)
+        if vetoed is not None:
+            gates["dilution"] = {"value": vetoed.get("conviction"),
+                                 "threshold": self.settings.catalyst_veto_min_conviction,
+                                 "outcome": "reject"}
+            self._emit_dilution_veto(ctx.symbol, vetoed)
+            return _done(False, "dilution_veto", throttle=is_auto, rank_factor=rank_factor)
+
+        # ANTI-CHASE / HALT — unconditional on the fast path; behind TRADING_UNIFIED_ENTRY
+        # on the slow path (the documented rollback switch; fast stays unconditional).
+        if ctx.source == "fast" or self.settings.unified_entry:
+            skip = self._anti_chase_skip(ctx.entry_ref, ctx.trigger, ctx.day_open, ctx.halted)
+            if skip is not None:
+                gates["anti_chase"] = {"value": ctx.entry_ref, "threshold": ctx.day_open,
+                                       "outcome": "reject"}
+                if is_auto:
+                    self.store.emit(RiskRuleTriggeredEvent(
+                        timestamp=datetime.now(), mode=self.mode,
+                        correlation_id=self.session_id,
+                        message=(f"{ctx.symbol} entry blocked: {skip} "
+                                 f"(entry={ctx.entry}, day_open={ctx.day_open})"),
+                        rule_type=skip, rule_value=float(ctx.day_open or 0.0),
+                        current_state={"symbol": ctx.symbol, "entry": ctx.entry_ref,
+                                       "day_open": ctx.day_open},
+                        action_taken="skipped_entry"))
+                return _done(False, skip, throttle=is_auto, rank_factor=rank_factor)
+
+        return _done(True, rank_factor=rank_factor)
+
+    def _emit_gate_receipt(self, ctx: EntryContext, decision: GateDecision) -> None:
+        """Decision receipt hook — populated in plan 005 step 3."""
+
+    def request_approvals_for_ready_signals(self, rank_map: dict | None = None) -> list[str]:
         """Turn fresh ready signals into approval requests. Returns order ids."""
         if not self.settings.enabled:
             return []
@@ -876,149 +1095,41 @@ class TradingExecutionService:
             if len(created) >= self.settings.max_orders_per_tick:
                 break
             symbol = signal["symbol"]
+            # Dedup FIRST (as before extraction): the slow path re-scans every ready
+            # signal each tick, so already-handled names must be skipped silently
+            # here — this is what keeps _requested_symbols from re-emitting a policy
+            # event every tick. The shared gate's own dedup then only bites the fast
+            # trigger path (per-fire, no re-scan).
             if (symbol in held or symbol in pending
                     or symbol in self._requested_symbols
                     or self._in_cooldown(symbol)        # benched after recent backouts
                     or symbol in self._exited_today):   # already round-tripped today
                 continue
-            vetoed = self._dilution_vetoed(symbol)
-            if vetoed is not None:
-                self._emit_dilution_veto(symbol, vetoed)
-                self._requested_symbols.add(symbol)  # don't re-evaluate every tick
-                continue
             entry = signal.get("entry_price")
             stop = signal.get("stop_loss_price")
             if not entry or not stop or stop >= entry:
-                continue
-
-            # CONFIRMATION-SPEED gate — the one backtested +EV filter: only setups
-            # whose breakout confirmed EARLY convert (+0.42R by ~bar 10); late
-            # confirmations and midday re-triggers are dead (-0.8R). Skip + log.
-            if self._late_confirmation(signal.get("timestamp")):
-                self.store.emit(
-                    RiskRuleTriggeredEvent(
-                        timestamp=datetime.now(),
-                        mode=self.mode,
-                        correlation_id=self.session_id,
-                        message=(f"{symbol} ready too late "
-                                 f"(confirm minute {self._confirm_minute_of(signal.get('timestamp')):.0f}"
-                                 f" > {self.settings.entry_confirm_by_minute}) — skipped"),
-                        rule_type="late_confirmation",
-                        rule_value=float(self.settings.entry_confirm_by_minute),
-                        current_state={"symbol": symbol,
-                                       "ready_at": str(signal.get("timestamp"))},
-                        action_taken="skipped_entry",
-                    )
-                )
-                self._requested_symbols.add(symbol)   # don't re-log every tick
-                continue
-
-            # POPULATION gate — the year-long +EV boundary: real gappers with real
-            # volume. Fail-open on missing data; skips are logged and visible.
-            sess_reason = self._outside_session()
-            if sess_reason:
-                continue  # quiet skip: outside-session evaluation is normal, not a risk event
-            pop_reason = self._population_reject(
-                signal.get("gap_pct"), signal.get("cum_volume"),
-                price=signal.get("entry_price"),
-                rvol=signal.get("relative_volume"))
-            if pop_reason:
-                self.store.emit(
-                    RiskRuleTriggeredEvent(
-                        timestamp=datetime.now(),
-                        mode=self.mode,
-                        correlation_id=self.session_id,
-                        message=f"{symbol} outside tradeable population ({pop_reason}) — skipped",
-                        rule_type="population_below",
-                        rule_value=float(self.settings.min_entry_gap_pct),
-                        current_state={"symbol": symbol,
-                                       "gap_pct": signal.get("gap_pct"),
-                                       "cum_volume": signal.get("cum_volume")},
-                        action_taken="skipped_entry",
-                    )
-                )
-                self._requested_symbols.add(symbol)   # don't re-log every tick
-                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:
+                continue  # price-sanity (silent; path-specific float coercion)
+            entry_f0 = float(entry)
+            # ONE shared gate sequence (plan 005). The slow path rests a limit AT the
+            # level, so entry_ref == trigger == the entry level. rank arrives from the
+            # trigger book via rank_map (None -> factor 1.0, no-op).
+            decision = self._evaluate_entry_gates(EntryContext(
+                symbol=symbol, source="auto", entry=entry_f0, stop=float(stop),
+                trigger=entry_f0, entry_ref=entry_f0,
+                gap_pct=signal.get("gap_pct"), price=entry_f0,
+                cum_volume=signal.get("cum_volume"),
+                rvol=signal.get("relative_volume"),
+                above_vwap=signal.get("above_vwap"), vwap=signal.get("vwap"),
+                quality_score=signal.get("quality_score"),
+                rank=(rank_map or {}).get(symbol),
+                day_open=signal.get("day_open"), halted=False,
+                confirm_ts=signal.get("timestamp"),
+                held=held, pending=pending))
+            if not decision.passed:
+                if decision.throttle:
                     self._requested_symbols.add(symbol)  # don't re-evaluate every tick
-                    continue
-
-            # QUALITY-GRADE gate — trade fewer, higher-quality setups. Shadow-logged
-            # always; entries only SKIPPED when min_quality_score > 0. Fail-open: a
-            # signal with no quality_score is never blocked.
-            qs = signal.get("quality_score")
-            if qs is not None and self.settings.min_quality_score > 0 and qs < self.settings.min_quality_score:
-                self.store.emit(
-                    RiskRuleTriggeredEvent(
-                        timestamp=datetime.now(),
-                        mode=self.mode,
-                        correlation_id=self.session_id,
-                        message=(f"{symbol} ready below quality gate "
-                                 f"(score {qs:.2f} < {self.settings.min_quality_score:.2f}) — skipped"),
-                        rule_type="quality_below",
-                        rule_value=float(qs),
-                        current_state={"symbol": symbol, "quality_score": float(qs),
-                                       "min": self.settings.min_quality_score},
-                        action_taken="skipped_entry",
-                    )
-                )
-                self._requested_symbols.add(symbol)  # don't re-evaluate every tick
                 continue
 
-            # Anti-chase / halt gates on the LIVE auto path (the entry-path
-            # unification — these ran ONLY on the fast trigger path before). The
-            # auto path rests a limit AT the trigger, so it can't chase above it
-            # (over_extended is a no-op here); the gate that bites is day-extension:
-            # don't enter a name whose breakout level is already parabolic vs the
-            # session open. halted=False (a resting limit won't fill mid-halt).
-            # Fail-open: a missing day_open skips only the day-extension check.
-            if self.settings.unified_entry:
-                entry_f0 = float(entry)
-                skip = self._anti_chase_skip(entry_f0, entry_f0,
-                                             signal.get("day_open"), False)
-                if skip is not None:
-                    self.store.emit(
-                        RiskRuleTriggeredEvent(
-                            timestamp=datetime.now(),
-                            mode=self.mode,
-                            correlation_id=self.session_id,
-                            message=(f"{symbol} entry blocked: {skip} "
-                                     f"(entry={entry}, day_open={signal.get('day_open')})"),
-                            rule_type=skip,
-                            rule_value=float(signal.get("day_open") or 0.0),
-                            current_state={"symbol": symbol, "entry": entry_f0,
-                                           "day_open": signal.get("day_open")},
-                            action_taken="skipped_entry",
-                        )
-                    )
-                    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%).
@@ -1033,17 +1144,19 @@ class TradingExecutionService:
             max_shares = (int(liq * float(cum_vol))
                           if (self.settings.unified_entry and liq > 0 and cum_vol > 0)
                           else None)
+            # scale risk by rank (rank-1 full, rank-2 half) exactly as the fast path;
+            # at concentrate_top_n=0 the factor is 1.0, so this is a no-op today.
             sizing = calculate_position_size(
                 float(entry),
                 float(stop),
                 equity=eq,
                 config=PositionSizingConfig(
-                    risk_per_trade_pct=self.settings.risk_per_trade_pct,
+                    risk_per_trade_pct=self.settings.risk_per_trade_pct * decision.rank_factor,
                     default_equity=self.settings.default_equity,
                 ),
                 max_position_value=pos_cap,
                 max_shares=max_shares,
-                max_risk_dollars=self.settings.max_risk_dollars,
+                max_risk_dollars=self.settings.max_risk_dollars * decision.rank_factor,
             )
             if sizing.position_size <= 0:
                 continue
@@ -1222,51 +1335,11 @@ class TradingExecutionService:
             return {"ok": False, "skipped": "halted"}
         if self._open_position_count() >= self.settings.max_concurrent_positions:
             return {"ok": False, "skipped": "max_positions"}
-        # CONFIRMATION-SPEED gate (same as the auto path): a live trigger fires in
-        # near-real-time, so now() IS the confirmation time. Late breaks are dead.
-        if self._late_confirmation(self._now()):
-            return {"ok": False, "skipped": "late_confirmation"}
-        # POPULATION gate (same as the auto path): the year-long +EV boundary
-        if self._outside_session():
-            return {"ok": False, "skipped": "outside_session"}
-        if self._population_reject(gap_pct, cum_volume or None,
-                                    price=price if price else last_price, rvol=rvol):
-            return {"ok": False, "skipped": "population_below"}
-
-        # CONCENTRATE-BY-RANK: only the top-N armed names may enter, and risk is
-        # scaled down by rank. Stops the equal-size spray across mediocre gappers
-        # (the human's edge is concentration). rank_factor 0 => outside the top-N.
-        rank_factor = rank_risk_factor(rank, self.settings.concentrate_top_n)
-        if rank_factor <= 0.0:
-            return {"ok": False, "skipped": "rank_concentration"}
-        # hard per-DAY fresh-entry cap (a day-trader takes a few good shots, not 12)
-        cap = self.settings.max_fresh_entries_per_day
-        if cap > 0:
-            today = self._now().date().isoformat() if hasattr(self._now(), "date") else None
-            if today is not None:
-                if self._fresh_entries.get("date") != today:
-                    self._fresh_entries = {"date": today, "n": 0}
-                if self._fresh_entries["n"] >= cap:
-                    return {"ok": False, "skipped": "daily_entry_cap"}
-
-        # per-symbol dedup shared with the slow path: never double-enter a name
-        # that's already held, pending approval, or requested this session.
-        held = self._held_symbols()
-        pending = {row["symbol"] for row in query_approval_queue(self.store)}
-        if symbol in held or symbol in pending or symbol in self._requested_symbols:
-            return {"ok": False, "skipped": "already_active"}
-        if self._in_cooldown(symbol):  # benched after recent backouts (anti-thrash)
-            return {"ok": False, "skipped": "cooldown"}
-        if symbol in self._exited_today:  # already round-tripped today — no re-entry
-            return {"ok": False, "skipped": "reentry_blocked"}
-        vetoed = self._dilution_vetoed(symbol)  # gated catalyst dilution veto
-        if vetoed is not None:
-            self._emit_dilution_veto(symbol, vetoed)
-            return {"ok": False, "skipped": "dilution_veto"}
-
-        # Use last_price only when it's a real positive tick. `last_price or
-        # trigger` would coerce a 0.0/garbage price to the trigger — silently
-        # skipping the extension guard AND sizing on a phantom price.
+        # Price-sanity + entry_ref derivation stays here (path-specific coercion from
+        # a live tick). `last_price or trigger` would coerce a 0.0/garbage price to
+        # the trigger — silently skipping the extension guard AND sizing on a phantom
+        # price. The extension gate downstream needs a real entry_ref, so resolve it
+        # before the shared gate.
         if last_price is not None and float(last_price) <= 0:
             return {"ok": False, "skipped": "bad_price"}
         entry_ref = float(last_price) if last_price is not None else float(trigger)
@@ -1275,16 +1348,22 @@ class TradingExecutionService:
         if stop_f >= entry_ref or trigger_f <= 0:
             return {"ok": False, "skipped": "bad_geometry"}
 
-        # Anti-chase ceiling: distance of the entry price ABOVE the breakout
-        # trigger (ORB high). A clean break crosses the level smoothly (~0%) and
-        # passes; a violent gap-THROUGH — an intraday parabolic, or a halt
-        # reopening past the level — lands far above it, and chasing that buys
-        # the top tick that reverses/halts down. (Overnight +100% gappers are
-        # already excluded upstream by the trigger book's gap_max; this catches
-        # the intraday chase / gap-through that gap_max can't see.)
-        skip = self._anti_chase_skip(entry_ref, trigger_f, day_open, halted)
-        if skip is not None:
-            return {"ok": False, "skipped": skip}
+        # ONE shared gate sequence (plan 005): confirmation-speed, session,
+        # population, VWAP, quality, concentrate-by-rank, daily cap, dedup/cooldown/
+        # reentry, dilution veto, anti-chase/halt — the identical logic the slow
+        # auto path runs. The fast public signature carries no VWAP/quality inputs,
+        # so those gates stay fail-open here (None -> never block).
+        decision = self._evaluate_entry_gates(EntryContext(
+            symbol=symbol, source="fast", entry=entry_ref, stop=stop_f,
+            trigger=trigger_f, entry_ref=entry_ref,
+            gap_pct=gap_pct, price=(price if price else last_price),
+            cum_volume=(cum_volume or None), rvol=rvol,
+            above_vwap=None, vwap=None, quality_score=None,
+            rank=rank, day_open=day_open, halted=halted,
+            confirm_ts=self._now()))
+        if not decision.passed:
+            return {"ok": False, "skipped": decision.reason}
+        rank_factor = decision.rank_factor
 
         eq = self._current_equity()
         liq = self.settings.liquidity_max_volume_pct
@@ -1763,10 +1842,13 @@ class TradingExecutionService:
         return actions
 
     @_locked
-    def tick(self) -> dict:
-        """One execution pass: expire stale entries, request approvals, auto-execute."""
+    def tick(self, rank_map: dict | None = None) -> dict:
+        """One execution pass: expire stale entries, request approvals, auto-execute.
+
+        ``rank_map`` ({symbol: rank}) is threaded from the trigger book so the slow
+        auto path can honour concentrate-by-rank identically to the fast path."""
         backed_out = self.expire_stale_entries()
-        requested = self.request_approvals_for_ready_signals()
+        requested = self.request_approvals_for_ready_signals(rank_map=rank_map)
         auto = self.process_auto_approvals() if self.settings.auto_approve else []
         return {
             "approvals_requested": requested,