population: learn_params + eod_replay consume the shared gate (kills look-ahead)

Both offline consumers now screen through strategy.population.population_reject
instead of divergent, look-ahead admits:

- learn_params.load_sessions: replaces the `gap>=0.03 OR day_rvol>=2.0` admit,
  whose day_rvol was a FULL-DAY volume sum (outcome-time look-ahead, the
  2026-07-08 class). gap is already a FRACTION (open vs prev close) — passed
  as-is; the OPEN price is passed for the price band. Stored daily bars expose
  no open-knowable volume, so volume=None (fail-open on the day-volume floor) —
  no day-end volume is smuggled into admission.
- eod_replay._daily_gappers: screens on the OPEN price + STATIC open gap (was
  t_close look-ahead) via the shared gate; the local gap is converted to a
  FRACTION before the call, so it inherits the >=150% squeeze exception. Deletes
  the stale $1-5 / 3-60% WATCHER_/TRIGGER_ constants (KEEPS RISK_PCT). The
  second-stage scan_gappers is made permissive (membership already decided by
  the shared gate), removing the `gap_pct <= 60%` cap that wrongly dropped
  squeezes. Prod thresholds passed explicitly (bare from_env would disable gap).

Unit test replaces the live-DB smoke run: $15/+12% qualifies, $4/+45% rejected
unless >=150%+rvol5. research/labeler.py deferred to plan 006 (no locatable
price/gap admission screen); sweep_exits/sweep_filters likewise deferred.

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

eod_replay.py

@@ -39,19 +39,22 @@ from research.query import query_avg_daily_volume, query_minute_bars, query_prev
 from strategy.backtest.engine import BacktestEngine
 from strategy.exits import ExitConfig, TRAIL_PRIOR_LOW, parse_profit_tiers
 from strategy.models import SetupCriteria
+from strategy.population import PopulationConfig, population_reject
 
-# the live verified config
-PRICE_MIN = float(os.environ.get("WATCHER_PRICE_MIN", "1"))
-PRICE_MAX = float(os.environ.get("WATCHER_PRICE_MAX", "5"))
-GAP_MIN = float(os.environ.get("TRIGGER_GAP_MIN", "3"))
-GAP_MAX = float(os.environ.get("TRIGGER_GAP_MAX", "60"))
-RVOL_MIN = float(os.environ.get("TRIGGER_RVOL_MIN", "2"))
+# Population is defined ONCE, in strategy/population.py — the SAME gate the live
+# loop trades ($2-20 band + gap floor + 35-150% dead band + >=150% squeeze). This
+# replaces the stale $1-5 / 3-60% look-ahead screen. Prod thresholds are passed
+# explicitly (a bare from_env() shell would disable the gap gate).
+_POP = PopulationConfig(min_entry_gap_pct=0.10)
 RISK_PCT = float(os.environ.get("TRADING_RISK_PER_TRADE_PCT", "0.01"))
 
 
 def _daily_gappers(con, sess, universe):
     """Identify today's qualifying gappers from DAILY bars (cheap) so we only
-    minute-backfill the handful that matter, not all 100 names."""
+    minute-backfill the handful that matter, not all 100 names. Screens through
+    the SHARED live population gate (strategy/population.py) on the OPEN price and
+    STATIC open gap — both open-knowable; the old screen used t_close (look-ahead)
+    and a hard 60% cap that dropped the >=150% squeezes production re-admits."""
     out = []
     for sym in universe:
         # Bars ON OR BEFORE the session date, newest first — so rows[0] is the
@@ -65,13 +68,14 @@ def _daily_gappers(con, sess, universe):
             "AND trade_date <= ? ORDER BY trade_date DESC LIMIT 25", [sym, sess]).fetchall()
         if len(rows) < 2 or str(rows[0][0]) != str(sess):
             continue
-        t_open, t_close, t_vol = float(rows[0][1]), float(rows[0][2]), float(rows[0][3])
+        t_open, t_vol = float(rows[0][1]), float(rows[0][3])
         prev_close = float(rows[1][2])
         prior = [float(r[3]) for r in rows[1:21]]
         adv = sum(prior) / len(prior) if prior else 0.0
-        gap = (t_open / prev_close - 1) * 100 if prev_close else 0.0
+        gap = (t_open / prev_close - 1) if prev_close else 0.0   # FRACTION (population contract)
         rvol = (t_vol / adv) if adv else 0.0
-        if PRICE_MIN <= t_close <= PRICE_MAX and GAP_MIN <= gap <= GAP_MAX and rvol >= RVOL_MIN:
+        # OPEN price + open gap are open-knowable; rvol re-admits real >=150% squeezes.
+        if population_reject(_POP, gap_pct=gap, price=t_open, volume=None, rvol=rvol) is None:
             out.append(sym)
     return out
 
@@ -104,8 +108,10 @@ def main(argv=None):
     #    minute bars for all 100; only the gappers need them)
     ingest_daily_history(con, client, universe, days=30)
     gapper_syms = _daily_gappers(con, sess, universe)
-    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)}")
+    print(f"daily-screen gappers (shared population gate: ${_POP.price_min:g}-{_POP.price_max:g}, "
+          f"gap >={_POP.min_entry_gap_pct:.0%} ex {_POP.gap_dead_band_lo:.0%}-{_POP.gap_dead_band_hi:.0%} "
+          f"dead-band, squeeze >={_POP.gap_dead_band_hi:.0%}+rvol{_POP.squeeze_rvol_min:g}): "
+          f"{len(gapper_syms)}")
 
     # 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
@@ -116,9 +122,13 @@ def main(argv=None):
     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)
-               if g.gap_pct <= GAP_MAX and g.symbol in keep]
+    # Population membership was already decided by the shared gate in
+    # _daily_gappers; scan_gappers here only computes gap/rvol/vwap for ranking, so
+    # pass permissive bounds and let `keep` be the sole authority (this also keeps
+    # the >=150% squeeze names the old `gap_pct <= GAP_MAX` cap wrongly dropped).
+    gappers = [g for g in scan_gappers(con, sess, min_gap_pct=0.0, min_relative_volume=0.0,
+                                       price_min=0.0, price_max=1e9, limit=200)
+               if g.symbol in keep]
     if not gappers:
         print("no qualifying gappers today.")
         return
@@ -127,7 +137,8 @@ def main(argv=None):
     cfg = _exit_cfg()
     eng = BacktestEngine(Config(), target_r=cfg.target_r, warmup_bars=3, eval_every=1,
                          ready_score_pct=55, min_bars=3, exit_config=cfg,
-                         criteria=SetupCriteria(gap_pct_min=0.05, relative_volume_min=RVOL_MIN))
+                         # intrabar ORB SETUP criteria (not population) — preserved
+                         criteria=SetupCriteria(gap_pct_min=0.05, relative_volume_min=2.0))
     equity = float(client.get_account().get("equity") or 0) or 100000.0
     risk_dollars = equity * RISK_PCT
 

learn_params.py

@@ -35,9 +35,16 @@ from research.query import (
 )
 from strategy.backtest.engine import BacktestEngine
 from strategy.models import SetupCriteria
+from strategy.population import PopulationConfig, population_reject
 
 load_dotenv()
 
+# Screen tuner sessions through the SHARED live population gate (strategy/population.py)
+# so the tuner fits the EXACT universe the live loop trades. Prod gap floor passed
+# explicitly (a bare from_env() shell would disable the gap gate). Price band = the
+# module's $2-20 default (production values).
+_POP = PopulationConfig(min_entry_gap_pct=0.10)
+
 
 def load_sessions(con):
     """[(symbol, session_date, rth_bars, prev_close, adv), ...] for stored data."""
@@ -53,15 +60,18 @@ def load_sessions(con):
             continue
         pc = query_previous_close(con, symbol, sess)
         adv = query_avg_daily_volume(con, symbol, sess)
-        # Mimic the LIVE scanner: only keep sessions that actually gapped up or
-        # traded elevated volume. Otherwise the backtest trades non-movers the
-        # live discovery would never surface, burying the real gappers' edge.
+        # Screen through the SHARED live population gate so the tuner fits exactly
+        # the universe the live loop trades. `gap` is the STATIC open gap (a
+        # FRACTION) and `open_px` the OPEN price — both open-knowable. The old
+        # admit used a full-day volume sum (day-end look-ahead, the 2026-07-08
+        # class); stored daily bars expose no open-knowable volume, so we pass
+        # volume=None (fail-open on the day-volume floor) rather than smuggle it in.
         try:
-            gap = (float(bars["open"].iloc[0]) - pc) / pc if pc else 0.0
-            day_rvol = (float(bars["volume"].sum()) / adv) if adv else 0.0
+            open_px = float(bars["open"].iloc[0])
+            gap = (open_px - pc) / pc if pc else 0.0
         except Exception:  # noqa: BLE001
-            gap, day_rvol = 0.0, 0.0
-        if not (gap >= 0.03 or day_rvol >= 2.0):
+            open_px, gap = 0.0, 0.0
+        if population_reject(_POP, gap_pct=gap, price=open_px, volume=None, rvol=None) is not None:
             continue
         out.append((symbol, sess, bars, pc, adv))
     return out

tests/unit/test_eod_replay_population.py

@@ -0,0 +1,55 @@
+"""Unit test standing in for the `python eod_replay.py <date>` smoke verify
+(the live DB must never be touched). Pins that eod_replay._daily_gappers now
+screens through the shared population gate on the OPEN price + STATIC open gap,
+inheriting the >=150% squeeze exception and dropping the old $1-5 / 3-60% /
+t_close look-ahead screen.
+
+Verify (plan step 3): a $15 name with a 12% open gap qualifies; a $4 name with a
+45% gap does not unless >=150% + rvol>=5.
+"""
+
+from datetime import date
+
+import eod_replay
+
+SESS = date(2026, 7, 10)
+PRIOR = date(2026, 7, 9)
+
+
+def _rows(open_px, prev_close, day_vol=1000.0, prior_vol=1000.0):
+    """daily_bars rows newest-first: (trade_date, open, close, volume).
+    row[0] = session day, row[1..] = prior days (for prev_close + 20-day adv)."""
+    day = (SESS, open_px, open_px, day_vol)
+    prior = [(PRIOR, prev_close, prev_close, prior_vol) for _ in range(20)]
+    return [day] + prior
+
+
+class _Result:
+    def __init__(self, rows):
+        self._rows = rows
+
+    def fetchall(self):
+        return self._rows
+
+
+class _FakeCon:
+    def __init__(self, by_sym):
+        self.by_sym = by_sym
+
+    def execute(self, sql, params):  # params == [symbol, session_date]
+        return _Result(self.by_sym.get(params[0], []))
+
+
+def test_daily_gappers_uses_shared_population_gate():
+    by_sym = {
+        # open price, prev_close chosen so open/prev_close - 1 == target gap
+        "BIG15": _rows(15.0, 15.0 / 1.12),                       # $15, +12% -> IN
+        "DEAD4": _rows(4.0, 4.0 / 1.45),                         # $4, +45% dead band -> OUT
+        "SQZ6": _rows(4.0, 4.0 / 3.0, day_vol=6000, prior_vol=1000),  # +200% rvol6 -> IN
+        "SQZ2": _rows(4.0, 4.0 / 3.0, day_vol=2000, prior_vol=1000),  # +200% rvol2 -> OUT
+        "CHEAP": _rows(0.98, 0.98 / 1.20),                      # $0.98 below band -> OUT
+        "LOWGAP": _rows(10.0, 10.0 / 1.05),                     # +5% below floor -> OUT
+    }
+    con = _FakeCon(by_sym)
+    keep = eod_replay._daily_gappers(con, SESS, list(by_sym))
+    assert set(keep) == {"BIG15", "SQZ6"}