population: extract strategy/population.py; live gate delegates to it

The tradeable population is now defined ONCE. trading_execution._population_reject
becomes a thin wrapper building a PopulationConfig from its settings and calling
strategy.population.population_reject — logic moved verbatim (dead-band 0.35/1.50
and squeeze rvol 5.0 are module CONSTANTS, gap floor / price band / volume floor
from config). Behaviour byte-identical: the pre-extraction characterization suite
stays green through the new public seam.

from_env mirrors ExecutionSettings.from_env (same 4 keys + the [0,1] _qfloat clamp)
so a live-shaped shell reconstructs the identical gate. gap_pct FRACTION contract
and the as_of open-knowable contract are documented for offline consumers.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
momentum devandClaude Fable 5 committed 2026-07-13 22:36:59 -04:00
1 parent 256481910f
commit 66e81a48f3
3 files changed
+190 -31

No files matched your search

+115
View File
@@ -0,0 +1,115 @@
"""strategy/population.py — the ONE definition of the tradeable population.
Every consumer (the live gate in trading_execution, the nightly tuner via
learn_params, the whole-market eod_replay, and future experiment harnesses)
screens sessions through ``population_reject`` so a backtest, a tune, and a live
entry all agree on which names are inside the +EV universe.
UNIT CONTRACT: ``gap_pct`` is a FRACTION (0.12 == 12%), matching the live call
site (``signal.get("gap_pct")``). ``as_of`` documents that every input must be
knowable at evaluation time — ``"open"`` means the static overnight gap
(session open vs prev close) and premarket/at-entry volume. Day-end fields
(full-day volume, close price) are FORBIDDEN: feeding them in re-creates the
outcome-time look-ahead this module exists to kill. The convention is enforced
by review, not at runtime.
Import-clean: no env reads at import time (env parsing lives in ``from_env``),
matching the ``ExitConfig.from_env()`` convention in strategy/exits.py.
"""
from __future__ import annotations
import os
from dataclasses import dataclass
@dataclass(frozen=True)
class PopulationConfig:
price_min: float = 2.0
price_max: float = 20.0
# FRACTION (0.10 == 10%); env TRADING_MIN_ENTRY_GAP_PCT. Code default 0.0 —
# the whole gap gate is disabled at 0, matching the live default. Production
# sets 0.10.
min_entry_gap_pct: float = 0.0
min_day_volume: float = 0.0 # shares; env TRADING_MIN_DAY_VOLUME
# Hardcoded in the live gate — module CONSTANTS, no env read. (The
# TRIGGER_GAP_MAX / TRIGGER_GAP_SQUEEZE_* env keys configure the SEPARATE
# ArmedTriggerBook watcher gate in runtime/triggers.py, NOT this function.)
gap_dead_band_lo: float = 0.35
gap_dead_band_hi: float = 1.50
squeeze_rvol_min: float = 5.0
@classmethod
def from_env(cls, env: dict | None = None) -> "PopulationConfig":
"""Build from environment — SAME keys + defaults as trading_execution's
ExecutionSettings.from_env today (incl. the [0,1] clamp on the gap floor).
Reads ONLY the four population keys; the dead-band / squeeze constants are
never env-driven."""
v = dict(os.environ)
if env:
v.update(env)
def _qfloat(val: str) -> float: # mirrors trading_execution._qfloat
try:
return max(0.0, min(1.0, float(val)))
except (TypeError, ValueError):
return 0.0
return cls(
price_min=float(v.get("TRADING_PRICE_MIN", "2.0")),
price_max=float(v.get("TRADING_PRICE_MAX", "20.0")),
min_entry_gap_pct=_qfloat(v.get("TRADING_MIN_ENTRY_GAP_PCT", "0.0")),
min_day_volume=float(v.get("TRADING_MIN_DAY_VOLUME", "0.0")),
)
def population_reject(
cfg: PopulationConfig,
*,
gap_pct,
price,
volume,
rvol,
as_of: str = "open",
) -> str | None:
"""Return None if inside the tradeable population, else the reject reason.
UNIT CONTRACT: ``gap_pct`` is a FRACTION (0.12 == 12%), matching the live
call site (``signal.get("gap_pct")``). CONTRACT: every input must be
knowable at evaluation time (``as_of="open"`` = static open gap,
premarket/at-entry volume). Day-end fields are forbidden (see module doc).
Fail-open: missing values never block (except the dead band, which requires
a VERIFIED squeeze to re-admit). Logic moved verbatim from the live gate
``TradingExecutionService._population_reject``.
"""
if cfg.min_entry_gap_pct > 0 and gap_pct is not None:
try:
g = float(gap_pct)
if g < cfg.min_entry_gap_pct:
return f"gap {g:.1%} < {cfg.min_entry_gap_pct:.0%}"
# dead band (35-150%): +0.02R at year scale — excluded. Above 150%
# only VOLUME-VERIFIED squeezes re-enter (real ones ran +0.63R;
# reverse-split fakes show rvol ~0.01).
if cfg.gap_dead_band_lo < g < cfg.gap_dead_band_hi:
return f"gap {g:.0%} in dead band 35-150%"
if g >= cfg.gap_dead_band_hi and (rvol is None or float(rvol) < cfg.squeeze_rvol_min):
return f"gap {g:.0%} squeeze unverified (rvol {rvol})"
except (TypeError, ValueError):
pass
if price is not None:
try:
px = float(price)
lo, hi = cfg.price_min, cfg.price_max
if px > 0 and not (lo <= px <= hi):
return f"price {px:.2f} outside ${lo:g}-{hi:g} band"
except (TypeError, ValueError):
pass
if cfg.min_day_volume > 0 and volume is not None:
try:
if float(volume) < cfg.min_day_volume:
return (f"day volume {float(volume):,.0f} < "
f"{cfg.min_day_volume:,.0f}")
except (TypeError, ValueError):
pass
return None
@@ -0,0 +1,62 @@
"""Module-level tests for strategy/population.py — the shared population gate.
Complements tests/unit/test_population_gate.py (which pins the LIVE seam through
the service). Here we test the module directly: from_env key parity with
production, the as_of contract docstring, and fail-open semantics.
"""
import inspect
from strategy.population import PopulationConfig, population_reject
def test_from_env_defaults_match_production_keys():
cfg = PopulationConfig.from_env(env={
"TRADING_PRICE_MIN": "2.0",
"TRADING_PRICE_MAX": "20.0",
"TRADING_MIN_ENTRY_GAP_PCT": "0.10",
"TRADING_MIN_DAY_VOLUME": "300000",
})
assert cfg.price_min == 2.0
assert cfg.price_max == 20.0
assert cfg.min_entry_gap_pct == 0.10
assert cfg.min_day_volume == 300_000
# constants are NOT env-driven
assert cfg.gap_dead_band_lo == 0.35
assert cfg.gap_dead_band_hi == 1.50
assert cfg.squeeze_rvol_min == 5.0
def test_from_env_bare_shell_disables_gap_gate():
# No production env => min_entry_gap_pct clamps to 0.0 (gap gate off), the
# exact hazard the plan warns offline consumers about.
cfg = PopulationConfig.from_env(env={})
assert cfg.min_entry_gap_pct == 0.0
assert population_reject(cfg, gap_pct=0.05, price=10.0, volume=None, rvol=None) is None
def test_from_env_clamps_gap_floor_to_unit_interval():
# mirrors trading_execution._qfloat: percent-typed 40 must NOT survive as 40.
assert PopulationConfig.from_env(env={"TRADING_MIN_ENTRY_GAP_PCT": "40"}).min_entry_gap_pct == 1.0
assert PopulationConfig.from_env(env={"TRADING_MIN_ENTRY_GAP_PCT": "junk"}).min_entry_gap_pct == 0.0
def test_as_of_contract_documented():
# day-end rejection is by-convention (review-enforced), so the contract MUST
# be spelled out in the signature + docstring.
assert "as_of" in inspect.signature(population_reject).parameters
doc = population_reject.__doc__ or ""
assert "FRACTION" in doc and "forbidden" in doc.lower()
def test_fail_open_on_all_missing():
cfg = PopulationConfig(min_entry_gap_pct=0.10, min_day_volume=300_000)
assert population_reject(cfg, gap_pct=None, price=None, volume=None, rvol=None) is None
def test_offline_consumer_must_pass_prod_thresholds():
# bare config admits everything; explicit prod thresholds gate correctly.
bare = PopulationConfig()
assert population_reject(bare, gap_pct=0.02, price=3.0, volume=1, rvol=None) is None
prod = PopulationConfig(min_entry_gap_pct=0.10, min_day_volume=300_000)
assert population_reject(prod, gap_pct=0.02, price=3.0, volume=1, rvol=None) is not None
+13 -31
View File
@@ -41,6 +41,7 @@ from storage.projections import (
query_approval_queue,
query_ready_signals_snapshot,
)
from strategy.population import PopulationConfig, population_reject
from strategy.risk.position_sizing import (
PositionSizingConfig,
calculate_position_size,
@@ -830,37 +831,18 @@ class TradingExecutionService:
paths passing dynamic/intraday gaps, letting non-gappers (TVRD 6%, SKIN 2.2%,
JLHL 4.3%) and dead-band names (SEER 38.6%, IOTR 40.7%) into the book.
Fail-open: missing values never block (except the dead band, which requires
a VERIFIED squeeze to re-admit)."""
if self.settings.min_entry_gap_pct > 0 and gap_pct is not None:
try:
g = float(gap_pct)
if g < self.settings.min_entry_gap_pct:
return f"gap {g:.1%} < {self.settings.min_entry_gap_pct:.0%}"
# dead band (35-150%): +0.02R at year scale — excluded. Above 150%
# only VOLUME-VERIFIED squeezes re-enter (real ones ran +0.63R;
# reverse-split fakes show rvol ~0.01).
if 0.35 < g < 1.50:
return f"gap {g:.0%} in dead band 35-150%"
if g >= 1.50 and (rvol is None or float(rvol) < 5.0):
return f"gap {g:.0%} squeeze unverified (rvol {rvol})"
except (TypeError, ValueError):
pass
if price is not None:
try:
px = float(price)
lo, hi = self.settings.price_band_min, self.settings.price_band_max
if px > 0 and not (lo <= px <= hi):
return f"price {px:.2f} outside ${lo:g}-{hi:g} band"
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
a VERIFIED squeeze to re-admit).
Thin wrapper: the population is defined ONCE in strategy/population.py so
the tuner, replay, and experiments screen through the identical logic."""
cfg = PopulationConfig(
price_min=self.settings.price_band_min,
price_max=self.settings.price_band_max,
min_entry_gap_pct=self.settings.min_entry_gap_pct,
min_day_volume=self.settings.min_day_volume,
)
return population_reject(
cfg, gap_pct=gap_pct, price=price, volume=cum_volume, rvol=rvol)
def request_approvals_for_ready_signals(self) -> list[str]:
"""Turn fresh ready signals into approval requests. Returns order ids."""