Population gate: trade only real gappers — the year-long +EV boundary

The gapper-history reconstruction (n=3,008 symbol-days / 14 months) REFUTED the 15-min
confirmation gate (late crosses beat early 12/14 months incl June 2026 — the 19-day
"+0.42R early" was a definition artifact) and found the first at-scale +EV configuration:
top-20 ranked gappers (gap >= 10%, $1-20, day volume >= 300k) are +0.05..+0.15R under the
live BE@+5% ratchet exits in EVERY rank x timing segment — including rank 6-20 ("D tier",
owner-approved: +0.153R early, the best cell). The live book bled by trading OUTSIDE that
population (low-gap, thin names), not by mistiming.

- ExecutionSettings.min_entry_gap_pct / min_day_volume (TRADING_MIN_ENTRY_GAP_PCT=0.10,
  TRADING_MIN_DAY_VOLUME=300000): POPULATION gate on BOTH entry paths — auto path skips +
  logs rule_type=population_below; fast path returns skipped=population_below. Fail-open
  on missing data (house rule: never block on unknowns).
- evaluate_setup now surfaces the true overnight gap_pct in signal_data; ready-signal
  snapshot passes it through.
- fast path: submit_breakout_now takes gap_pct (trigger book stores % units; call site
  normalizes to fraction); TRIGGER_GAP_MIN raised 3 -> 10 to align the arm floor.
- confirmation gate flipped OFF via env (code stays, 0=disabled).
- rank concentration stays OFF (owner call, data-consistent: rank 6-20 converts fine).

Tests: population skip+log, real gapper passes, fail-open on missing data. 52 pass.
Monday config: population gate + cap 10 fills + quality C+ + VWAP + BE@+5% ratchet + SIP.

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

run_live_paper.py

@@ -862,6 +862,7 @@ def main(argv: list[str] | None = None) -> int:
                     t.symbol, t.trigger, t.stop, last_price=t.price,
                     cum_volume=t.cum_volume, halted=_recently_halted(t.symbol),
                     day_open=t.day_open, rank=t.rank,
+                    gap_pct=(t.gap_pct / 100.0) if t.gap_pct else None,  # book stores % units
                 )
                 if res.get("ok"):
                     book.mark_fired(t.symbol)

storage/projections.py

@@ -231,6 +231,7 @@ def query_ready_signals_snapshot(store, session_id: str | None = None) -> list[d
             "above_vwap": signal_data.get("above_vwap"),
             "day_open": signal_data.get("day_open"),
             "cum_volume": signal_data.get("cum_volume"),
+            "gap_pct": signal_data.get("gap_pct"),
             "timestamp": event.get("timestamp"),
         }
     return sorted(

strategy/evaluation/setup_evaluator.py

@@ -163,6 +163,10 @@ def evaluate_setup(
                 "above_vwap": bool(above_vwap),
                 "day_open": round(float(open_price), 4) if open_price > 0 else None,
                 "cum_volume": float(bars["volume"].sum()),
+                # true overnight gap (open vs prev close) — input to the live POPULATION
+                # gate: the year-long gapper replay showed the +EV boundary is the
+                # population (gap>=10% + real volume), not entry timing
+                "gap_pct": round(float(gap_pct), 6),
             }
         )
 

tests/unit/test_trading_execution.py

@@ -67,12 +67,15 @@ class Clock:
 
 
 def emit_signal(store, symbol="GOOD", entry=14.0, stop=13.45,
-                above_vwap=None, vwap=None, day_open=None, cum_volume=None, quality_score=0.7):
+                above_vwap=None, vwap=None, day_open=None, cum_volume=None,
+                quality_score=0.7, gap_pct=None):
     signal_data = {
         "entry_price": entry,
         "stop_loss_price": stop,
         "quality_score": quality_score,
     }
+    if gap_pct is not None:
+        signal_data["gap_pct"] = gap_pct
     if above_vwap is not None:
         signal_data["above_vwap"] = above_vwap
         signal_data["vwap"] = vwap if vwap is not None else (
@@ -460,6 +463,31 @@ def test_unfilled_entry_is_armed_then_times_out(store):
     assert service._armed == {}
 
 
+def test_population_gate_skips_low_gap_and_thin(store):
+    # POPULATION gate (year-long +EV boundary): a ready signal from a non-gapper
+    # (gap 2%) or a thin name is skipped + logged; a real gapper passes.
+    emit_signal(store, symbol="THIN", gap_pct=0.02, cum_volume=1_000_000)
+    emit_signal(store, symbol="GAPR", gap_pct=0.25, cum_volume=1_000_000)
+    executor = FakeExecutor()
+    service = make_service(store, executor, auto_approve=True,
+                           min_entry_gap_pct=0.10, min_day_volume=300_000,
+                           max_orders_per_tick=10)
+    service.tick()
+    assert [r.symbol for r in executor.executed] == ["GAPR"]
+    rejects = [json.loads(r["payload_json"]) for r in
+               store.query_events(event_type="risk_rule_triggered")]
+    assert any(r.get("rule_type") == "population_below" for r in rejects)
+
+
+def test_population_gate_fails_open_on_missing_data(store):
+    emit_signal(store, symbol="NODATA")            # no gap_pct/cum_volume fields
+    executor = FakeExecutor()
+    service = make_service(store, executor, auto_approve=True,
+                           min_entry_gap_pct=0.10, min_day_volume=300_000)
+    service.tick()
+    assert len(executor.executed) == 1             # never block on missing data
+
+
 def test_late_confirmation_skips_signal(store):
     # Confirmation-speed gate: the signal went ready at T0=09:45 (minute 15 of the
     # session); with the gate at 10 minutes it must be SKIPPED and shadow-logged,

trading_execution.py

@@ -105,6 +105,14 @@ class ExecutionSettings:
     # gone READY to be tradeable; later signals (incl. midday re-triggers) are
     # skipped + shadow-logged (rule_type=late_confirmation). 0 = disabled.
     entry_confirm_by_minute: int = 0
+    # POPULATION GATE — the year-long gapper replay (research/gapper_history.py, n=3,008
+    # symbol-days / 14 months) found the first at-scale +EV boundary: top-ranked gappers
+    # with gap >= 10% and real day volume are +0.05..+0.15R under the live ratchet exits,
+    # in EVERY rank x timing segment — while the live book bled by trading OUTSIDE that
+    # population (low-gap / thin names). Floors are FRACTION (0.10 = 10%) and shares.
+    # 0 = disabled. Fail-open: a signal missing gap/volume data is never blocked.
+    min_entry_gap_pct: float = 0.0
+    min_day_volume: float = 0.0
     # UNIFIED ENTRY: run the fast path's shared anti-chase gates (over_extended,
     # day-extension, halt) on the LIVE auto path too. These previously ran ONLY on
     # the fast trigger path, leaving live auto entries unprotected (see the
@@ -266,6 +274,8 @@ class ExecutionSettings:
             require_above_vwap=flag("TRADING_REQUIRE_ABOVE_VWAP", "1"),
             min_quality_score=_qfloat(values.get("TRADING_MIN_QUALITY_SCORE", "0.0")),
             entry_confirm_by_minute=int(values.get("TRADING_ENTRY_CONFIRM_BY_MINUTE", "0")),
+            min_entry_gap_pct=_qfloat(values.get("TRADING_MIN_ENTRY_GAP_PCT", "0.0")),
+            min_day_volume=float(values.get("TRADING_MIN_DAY_VOLUME", "0.0")),
             unified_entry=flag("TRADING_UNIFIED_ENTRY", "1"),
             entry_fill_model=values.get("TRADING_ENTRY_FILL_MODEL", "resting"),
             liquidity_max_volume_pct=float(
@@ -670,6 +680,24 @@ class TradingExecutionService:
         m = self._confirm_minute_of(ts)
         return m is not None and m > limit
 
+    def _population_reject(self, gap_pct, cum_volume) -> str | None:
+        """Reason string when the signal is OUTSIDE the +EV population (gap/volume floors);
+        None when tradeable. Fail-open: missing values never block."""
+        if self.settings.min_entry_gap_pct > 0 and gap_pct is not None:
+            try:
+                if float(gap_pct) < self.settings.min_entry_gap_pct:
+                    return f"gap {float(gap_pct):.1%} < {self.settings.min_entry_gap_pct:.0%}"
+            except (TypeError, ValueError):
+                pass
+        if self.settings.min_day_volume > 0 and cum_volume is not None:
+            try:
+                if float(cum_volume) < self.settings.min_day_volume:
+                    return (f"day volume {float(cum_volume):,.0f} < "
+                            f"{self.settings.min_day_volume:,.0f}")
+            except (TypeError, ValueError):
+                pass
+        return None
+
     def request_approvals_for_ready_signals(self) -> list[str]:
         """Turn fresh ready signals into approval requests. Returns order ids."""
         if not self.settings.enabled:
@@ -739,6 +767,28 @@ class TradingExecutionService:
                 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.
+            pop_reason = self._population_reject(signal.get("gap_pct"),
+                                                 signal.get("cum_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
@@ -1001,6 +1051,7 @@ class TradingExecutionService:
         halted: bool = False,
         day_open: float | None = None,
         rank: int = 0,
+        gap_pct: float | None = None,
     ) -> dict:
         """Fire a breakout entry immediately on a LIVE price cross.
 
@@ -1022,6 +1073,9 @@ class TradingExecutionService:
         # 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._population_reject(gap_pct, cum_volume or None):
+            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