perf: stop the watcher scoring names that can never trade

criteria_evaluated + signal_blocked held 3.1 GB of a 7.6 GB events table.
On 2026-09-10 alone the watcher wrote 191,458 criteria receipts across 57
symbols -- ~3,359 per symbol per day -- and 154,260 of them (80.6%) were
for names whose static open gap was outside the tradeable band. NOK, NIO,
PATH, PLUG, SOFI, AAL and F all sit inside the $2-20 price filter, so
WATCHER_PRICE_MIN/MAX admits them; only the gap floor excludes them, and
that was checked per-tick instead of once up front.

Apply the SHARED population gate (strategy/population.population_reject --
the same function, same config the entry path already enforces) before the
evaluation and its receipts. This is behaviour-preserving for trading: those
names were being scored, emitted, and then rejected downstream anyway --
429 "outside tradeable population" risk events across 221 symbols, led by
exactly the names that top the receipt counts. The decision just moves
upstream of the write.

Details that matter:

  - The verdict is recomputed each tick, never cached. A >150% gapper reads
    "squeeze unverified" on thin volume and is admitted once rvol confirms;
    a cached rejection would bench it for the session.
  - _transition only emits on CHANGE, so a rejected symbol costs one
    state event per session and is then silent.
  - Fail-open throughout: an unreadable open, a missing previous close or
    absent volume returns None, so a data gap can never bench a symbol.
  - Gap is the STATIC OPEN gap (session open vs previous close), fixed for
    the day, not the live price -- the verdict must not wander intraday.

Off by default (population=None) so every existing caller and test keeps
its current behaviour; the live loop opts in, and WATCHER_POPULATION_GATE=0
restores the old path.

Note for diagnostics: momentum_cli's criteria views now show only names
inside the population. That is the intended trade -- the excluded rows
described symbols the bot could never have taken.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01A2LepdJfPCfK5Q8uznjFBM

deploy/docker/overrides.live.seed.env

@@ -65,11 +65,12 @@ TRADING_MAX_FRESH_ENTRIES_PER_DAY=10
 TRADING_MAX_GROSS_NOTIONAL_PCT=0.60
 TRADING_MAX_ORDERS_PER_TICK=1
 TRADING_MAX_POSITION_PCT=0.16
-# Boot-time sizing fallback. calculate_position_size falls back to this only when
+# Boot-time sizing fallback. calculate_position_size falls back to this ONLY when
 # equity was never readable; at the old 100000 a boot inside Tradier's ~01:06
 # maintenance window (equity reads 0.0 there — 16 times in 14 days) would have
-# sized 5% of 100k on a ~$250 account. Keep it near real account size.
-TRADING_DEFAULT_EQUITY=300
+# sized 5% of 100k on a ~$250 account. Keep it BELOW real account equity so the
+# fallback can only ever under-size (live equity 241.75 on 2026-09-13).
+TRADING_DEFAULT_EQUITY=200
 TRADING_MAX_RISK_DOLLARS=50000
 TRADING_MIN_DAY_VOLUME=300000
 TRADING_MIN_ENTRY_GAP_PCT=0.10
@@ -96,6 +97,11 @@ TRIGGER_MAX_ARMED=10
 TRIGGER_MIN_DOLLAR_VOL=8000
 TRIGGER_PRICE_MAX_AGE=12
 TRIGGER_RVOL_MIN=2.0
+# Apply the shared tradeable-population gate before the watcher scores anything.
+# Without it the watcher wrote 191,458 criteria receipts across 57 symbols on
+# 2026-09-10, 80.6% of them for names outside the gap band. Set 0 to restore the
+# old evaluate-everything behaviour.
+WATCHER_POPULATION_GATE=1
 WATCHER_ENABLED=1
 WATCHER_INTERVAL_SECONDS=8
 WATCHER_PRICE_MAX=20

run_live_paper.py

@@ -65,6 +65,7 @@ from runtime.market_calendar import (
 )
 from runtime.triggers import ArmedTriggerBook
 from runtime.watcher import Watcher, WatcherConfig
+from strategy.population import PopulationConfig
 from strategy.exits import ExitConfig
 from storage.event_schema import EventMode, ModuleTickEvent
 from storage.event_store import EventStore
@@ -412,6 +413,11 @@ def build_runtime(args: argparse.Namespace) -> dict:
             ready_score_pct=float(os.environ.get(
                 "WATCHER_READY_SCORE", str(_learned.get("ready_score_pct", 60.0)))),
             min_quality=float(os.environ.get("WATCHER_MIN_QUALITY", "0.30")),
+            # Same population gate the entry path enforces, applied before the
+            # watcher scores anything — so it stops writing criteria receipts for
+            # names that can never trade. Opt out with WATCHER_POPULATION_GATE=0.
+            population=(PopulationConfig.from_env()
+                        if _flag("WATCHER_POPULATION_GATE", "1") else None),
         ),
         catalyst_score_provider=(
             catalyst_score_provider if ollama_cfg.catalyst_score_enabled else None

runtime/watcher.py

@@ -29,6 +29,7 @@ from storage.event_schema import (
 )
 from storage.event_store import EventStore
 from strategy.evaluation.setup_evaluator import evaluate_setup
+from strategy.population import PopulationConfig, population_reject
 
 logger = logging.getLogger(__name__)
 
@@ -60,6 +61,15 @@ class WatcherConfig:
     min_bars: int = 10
     ready_score_pct: float = 60.0
     min_quality: float = 0.30
+    # Shared tradeable-population gate, applied BEFORE the per-tick evaluation.
+    # None = off (the historical behaviour every existing caller and test relies
+    # on). The live loop passes PopulationConfig.from_env() so the watcher stops
+    # scoring names the entry path can never accept: on 2026-09-10 the watcher
+    # wrote 191,458 criteria receipts across 57 symbols and 154,260 of them
+    # (80.6%) were for gaps outside the tradeable band — large caps like NOK, F
+    # and SOFI that sit inside the $2-20 price filter but can never pass the gap
+    # floor. Those two event types were 3.1 GB of a 7.6 GB events table.
+    population: "PopulationConfig | None" = None
 
 
 @dataclass
@@ -69,6 +79,7 @@ class WatcherTickResult:
     ready: list[str] = field(default_factory=list)
     blocked: list[str] = field(default_factory=list)
     discovered: list[str] = field(default_factory=list)
+    skipped: int = 0          # rejected by the population gate before evaluation
     errors: list[str] = field(default_factory=list)
 
 
@@ -122,6 +133,39 @@ class Watcher:
             )
         )
 
+    def _population_reject(self, candidate: WatchCandidate, bars) -> str | None:
+        """Reject reason if this name is outside the tradeable population, else None.
+
+        Uses the STATIC OPEN gap (session open vs previous close) — the same
+        open-knowable quantity the live entry gate uses, and fixed for the day —
+        never the live price, which would make the verdict wander intraday.
+        Fail-open throughout: anything unreadable returns None so a data gap can
+        never silently bench a symbol.
+        """
+        try:
+            open_px = float(bars["open"].iloc[0])
+        except Exception:  # noqa: BLE001
+            return None
+        prev_close = candidate.previous_close
+        gap = None
+        if prev_close:
+            try:
+                gap = (open_px - float(prev_close)) / float(prev_close)
+            except (TypeError, ValueError, ZeroDivisionError):
+                gap = None
+        volume = rvol = None
+        try:
+            volume = float(bars["volume"].sum())
+            adv = candidate.avg_daily_volume
+            if adv and float(adv) > 0:
+                rvol = volume / float(adv)
+        except Exception:  # noqa: BLE001
+            volume = rvol = None
+        return population_reject(
+            self.config.population,
+            gap_pct=gap, price=open_px, volume=volume, rvol=rvol, as_of="open",
+        )
+
     def _discover(self, candidate: WatchCandidate) -> None:
         if candidate.symbol in self._states:
             return
@@ -172,6 +216,20 @@ class Watcher:
                     )
                     continue
 
+                # POPULATION PRE-FILTER — skip the expensive evaluation (and its
+                # two fat receipts) for a name the entry path could never accept.
+                # Recomputed every tick rather than cached: a >150% gapper starts
+                # "squeeze unverified" on thin volume and is admitted later once
+                # rvol confirms, so a cached rejection would bench it for the
+                # session. _transition only emits on CHANGE, so a rejected symbol
+                # costs exactly one event and then goes silent.
+                if self.config.population is not None:
+                    reason = self._population_reject(candidate, bars)
+                    if reason is not None:
+                        self._transition(symbol, "out_of_population", reason)
+                        result.skipped += 1
+                        continue
+
                 # Evaluate the setup as of the latest bar's timestamp, not the
                 # wall clock. Minute bars are stored UTC-naive, but the entry
                 # cutoff reasons in US/Eastern wall-clock, so convert. This

tests/unit/runtime/test_watcher_population_gate.py

@@ -0,0 +1,163 @@
+"""Population pre-filter on the intraday watcher (2026-09-13).
+
+The watcher scored every candidate on every tick and emitted a fat
+``criteria_evaluated`` receipt each time — including for names the entry path
+can never accept. On 2026-09-10 that was 191,458 receipts across 57 symbols,
+154,260 of them (80.6%) for gaps outside the tradeable band: large caps like
+NOK, F and SOFI that sit inside the $2-20 price filter but can never clear the
+gap floor. ``criteria_evaluated`` + ``signal_blocked`` held 3.1 GB of a 7.6 GB
+events table.
+
+These tests use a fake store rather than EventStore so they run without a
+Postgres — the rest of the watcher suite needs one.
+"""
+
+from datetime import date
+
+import pandas as pd
+
+from runtime.watcher import WatchCandidate, Watcher, WatcherConfig
+from storage.event_schema import EventMode
+from strategy.population import PopulationConfig
+
+SESSION = date(2026, 9, 10)
+
+
+class FakeStore:
+    """Records emits instead of writing to Postgres."""
+
+    def __init__(self):
+        self.events = []
+
+    def emit(self, event):
+        self.events.append(event)
+        return "id"
+
+    def count(self, type_name: str) -> int:
+        return sum(1 for e in self.events if type(e).__name__ == type_name)
+
+
+class FakeProvider:
+    def __init__(self, candidates, bars):
+        self._candidates = candidates
+        self._bars = bars
+
+    def get_candidates(self, session_date):
+        return self._candidates
+
+    def get_bars(self, symbol, session_date):
+        return self._bars.get(symbol, pd.DataFrame())
+
+
+def _bars(open_px: float, n: int = 30, volume: int = 50_000) -> pd.DataFrame:
+    """n flat-ish minute bars starting at open_px."""
+    return pd.DataFrame({
+        "timestamp": pd.date_range("2026-09-10 13:30", periods=n, freq="1min"),
+        "open": [open_px] * n,
+        "high": [open_px * 1.01] * n,
+        "low": [open_px * 0.99] * n,
+        "close": [open_px] * n,
+        "volume": [volume] * n,
+    })
+
+
+def _watcher(store, candidates, bars, population=None):
+    return Watcher(
+        store, FakeProvider(candidates, bars),
+        WatcherConfig(session_id="s", mode=EventMode.PAPER, min_bars=5,
+                      population=population),
+    )
+
+
+GAPPER = PopulationConfig(price_min=2.0, price_max=20.0, min_entry_gap_pct=0.10)
+
+
+def test_gate_off_by_default_every_candidate_is_evaluated():
+    """No population passed == the historical behaviour, exactly."""
+    store = FakeStore()
+    w = _watcher(store, [WatchCandidate("NOK", previous_close=4.0)], {"NOK": _bars(4.0)})
+    r = w.tick(SESSION)
+    assert r.skipped == 0
+    assert r.evaluated == 1
+    assert store.count("CriteriaEvaluatedEvent") == 1
+
+
+def test_flat_large_cap_is_skipped_before_any_receipt():
+    """NOK opening 4.00 against a 4.00 close is a 0% gap — inside the price band,
+    permanently outside the population. This is the 80.6%."""
+    store = FakeStore()
+    w = _watcher(store, [WatchCandidate("NOK", previous_close=4.0)],
+                 {"NOK": _bars(4.0)}, population=GAPPER)
+    r = w.tick(SESSION)
+    assert r.skipped == 1
+    assert r.evaluated == 0
+    assert store.count("CriteriaEvaluatedEvent") == 0
+
+
+def test_repeated_ticks_on_a_skipped_name_cost_nothing():
+    """_transition only fires on change, so the receipt is one event per session,
+    not 3,359 per symbol per day."""
+    store = FakeStore()
+    w = _watcher(store, [WatchCandidate("NOK", previous_close=4.0)],
+                 {"NOK": _bars(4.0)}, population=GAPPER)
+    w.tick(SESSION)
+    after_first = len(store.events)
+    for _ in range(10):
+        w.tick(SESSION)
+    assert len(store.events) == after_first
+    assert store.count("CriteriaEvaluatedEvent") == 0
+
+
+def test_real_gapper_is_still_evaluated():
+    """A $2.50 name opening +25% must pass and be scored as before. 25% is inside
+    the tradeable 10-35% band; 35-150% is the deliberately excluded dead band."""
+    store = FakeStore()
+    w = _watcher(store, [WatchCandidate("SKYQ", previous_close=2.0,
+                                        avg_daily_volume=500_000)],
+                 {"SKYQ": _bars(2.5)}, population=GAPPER)
+    r = w.tick(SESSION)
+    assert r.skipped == 0
+    assert r.evaluated == 1
+    assert store.count("CriteriaEvaluatedEvent") == 1
+
+
+def test_price_outside_the_band_is_skipped():
+    store = FakeStore()
+    w = _watcher(store, [WatchCandidate("NVDA", previous_close=100.0)],
+                 {"NVDA": _bars(180.0)}, population=GAPPER)
+    assert w.tick(SESSION).skipped == 1
+
+
+def test_missing_previous_close_fails_open():
+    """A data gap must never silently bench a symbol: gap unknowable, price in
+    band -> still evaluated."""
+    store = FakeStore()
+    w = _watcher(store, [WatchCandidate("ABCD", previous_close=None)],
+                 {"ABCD": _bars(5.0)}, population=GAPPER)
+    assert w.tick(SESSION).evaluated == 1
+
+
+def test_verdict_is_not_cached_so_a_squeeze_can_be_admitted_later():
+    """A >150% gapper reads 'squeeze unverified' on thin volume and is admitted
+    once rvol confirms. Caching the first rejection would bench it all session."""
+    cfg = PopulationConfig(price_min=2.0, price_max=20.0, min_entry_gap_pct=0.10,
+                           gap_dead_band_lo=0.35, gap_dead_band_hi=1.50,
+                           squeeze_rvol_min=5.0)
+    cand = WatchCandidate("WYHG", previous_close=2.0, avg_daily_volume=1_000_000)
+    thin = {"WYHG": _bars(6.0, volume=1_000)}        # +200% gap, rvol 0.03
+    store = FakeStore()
+    w = _watcher(store, [cand], thin, population=cfg)
+    assert w.tick(SESSION).skipped == 1              # unverified -> skipped
+
+    w.provider._bars = {"WYHG": _bars(6.0, volume=500_000)}   # rvol now 15
+    assert w.tick(SESSION).evaluated == 1            # re-admitted, not benched
+
+
+def test_dead_band_gap_is_skipped():
+    """35-150% is excluded by design (+0.02R at year scale), so a +50% gapper is
+    correctly benched even though it clears the 10% floor."""
+    store = FakeStore()
+    w = _watcher(store, [WatchCandidate("MID", previous_close=2.0,
+                                        avg_daily_volume=500_000)],
+                 {"MID": _bars(3.0)}, population=GAPPER)
+    assert w.tick(SESSION).skipped == 1