feat(alpha): ExecutionEnv — 10-dim state, 9-action gating, terminal-only reward

Phase E Task 6. Core environment for the execution-policy DQN.

Episode lifecycle: position-open signal → up to `horizon_snapshots` →
forced close. Within the episode the policy chooses 9 actions; illegal
actions degrade to Wait inside step() (soft masking) — the network's
Q-output is never masked, preserving clean C51 and Munchausen targets.

Reward is **terminal-only**: realized PnL minus fees, accumulated over
all fills and force-closed at terminal mid if still open. No dense
per-snapshot shaping per pearl_event_driven_reward_density_alignment
(the canonical SP11→SP12 lesson — dense shaping on event-driven
objectives creates exposure-positive bias).

Bugs in the plan's sketch, caught + fixed during implementation:

  1. Synthetic test's bid/ask were CONSTANT while mid drifted — would
     have made the "uptrend" test lose money. Fixed: drift bid/ask
     together with mid.
  2. L1/L2 closing path used Side::None for fill lookup → 0% fill rate
     for closing limits. Fixed: closing-long → Side::Sell (post at ask),
     closing-short → Side::Buy (post at bid).
  3. `cursor`-based hash for fill PRNG was non-deterministic across
     replay seeds. Replaced with self-contained SplitMix64 RNG state
     (no `rand` dep added) — fill randomness is now bit-identical across
     train/eval replays at the same seed.
  4. `max_step_per_episode` and `mid_at_decision` fields stored but
     never read — dead per feedback_no_stubs. Removed.
  5. Test harness used n==horizon, which made the cursor-exhaustion
     guard (`cursor+1 >= len`) fire before the planned FlatMarket call.
     Fixed by setting n > horizon so the meaningful `ep.step >= horizon`
     path terminates the episode.
  6. `.unwrap()` in test bodies → `.expect("…")` per pre-commit policy.
  7. Volume-weighted entry_price across multi-fill scaling (the plan
     only handled the open-from-flat case).

5 unit tests:
  - state_has_expected_dim_and_is_finite (STATE_DIM=10, no NaN)
  - market_buy_then_market_sell_on_uptrend_profits (PnL math)
  - illegal_action_degrades_to_wait (soft mask, no fee, no fill counted)
  - terminal_force_close_pays_out_when_position_open (force-close at mid)
  - rng_is_deterministic_across_resets (same seed → identical fills)

`cargo test -p ml --lib env`: 16 passed (4 action_space + 7 fill_model
+ 5 execution_env); full module compiles clean with no new warnings.

crates/ml/src/env/execution_env.rs

@@ -0,0 +1,502 @@
+//! Phase E execution-policy environment.
+//!
+//! Each episode is a single trade lifecycle: from initial position-open
+//! signal (alpha confidence above threshold) through `horizon_snapshots`
+//! later (forced close). Within the episode the policy can:
+//!
+//!   - `Wait` (no order)
+//!   - Place a buy/sell at market / L1 / L2 (open or scale)
+//!   - `Flat` at market / L1 (close)
+//!
+//! **Reward** is terminal-only: realized PnL minus fees, accumulated over
+//! all fills during the episode and force-closed at terminal mid if the
+//! position is still open. No dense per-snapshot reward — see
+//! `pearl_event_driven_reward_density_alignment` for why dense shaping on
+//! event-driven objectives creates exposure-positive bias (canonical
+//! SP11→SP12 finding).
+//!
+//! **Action gating** is soft: illegal actions (e.g. `BuyMarket` while
+//! already long) degrade to `Wait` inside `step()`. The network's Q-output
+//! is never masked, preserving clean C51 distributional targets and the
+//! Munchausen term (Task 10).
+
+use crate::env::action_space::{ExecAction, Placement, Side};
+use crate::env::fill_model::{FillFeatures, FillModel};
+
+pub const STATE_DIM: usize = 10;
+
+#[derive(Debug, Clone)]
+pub struct ExecutionEnvConfig {
+    /// Episode length in snapshots — K=6000 for the full Phase E env;
+    /// K=600 for the Milestone E.1 truncated smoke.
+    pub horizon_snapshots: usize,
+    /// Fixed trade size per fill (sign determined by action). The Kelly
+    /// layer (Task 11) selects this *per episode*; within an episode it
+    /// is constant so position ∈ {-1, 0, +1} × trade_size.
+    pub trade_size_contracts: i32,
+    /// Per-contract round-turn cost in price units. Quarter-tick=0.0625
+    /// is the breakeven-cost frontier from the Phase 1d.4 backtest
+    /// (`pearl_phase1d4_backtest_cost_edge_frontier`).
+    pub cost_per_contract: f32,
+}
+
+#[derive(Debug, Clone)]
+pub struct EpisodeState {
+    pub step: usize,
+    /// Signed contracts held. Sign-only semantics: position ∈ {-N, 0, +N}
+    /// where N = `config.trade_size_contracts`.
+    pub position: i32,
+    /// Volume-weighted entry price across all fills; 0.0 iff position == 0.
+    pub entry_price: f32,
+    pub cumulative_pnl: f32,
+    pub n_fills: u32,
+}
+
+impl EpisodeState {
+    pub fn new() -> Self {
+        Self {
+            step: 0,
+            position: 0,
+            entry_price: 0.0,
+            cumulative_pnl: 0.0,
+            n_fills: 0,
+        }
+    }
+}
+
+impl Default for EpisodeState {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+#[derive(Debug, Clone)]
+pub struct SnapshotRow {
+    pub mid_price: f32,
+    pub bid_l: [f32; 3],
+    pub ask_l: [f32; 3],
+    pub alpha_logit: f32,
+    pub alpha_confidence: f32,
+    pub spread_bps: f32,
+    pub l1_imbalance: f32,
+    pub ofi_sum_5: f32,
+    pub mid_drift_5: f32,
+    pub time_since_trade_s: f32,
+    pub book_event_rate: f32,
+}
+
+/// Deterministic SplitMix64 RNG. Self-contained so the env compiles
+/// without pulling in `rand` and the fill randomness is bit-identical
+/// across train/eval replays at the same seed.
+#[derive(Debug, Clone)]
+pub struct ReplayRng {
+    state: u64,
+}
+
+impl ReplayRng {
+    pub fn new(seed: u64) -> Self {
+        Self { state: seed }
+    }
+
+    pub fn next_u64(&mut self) -> u64 {
+        self.state = self.state.wrapping_add(0x9E37_79B9_7F4A_7C15);
+        let mut z = self.state;
+        z = (z ^ (z >> 30)).wrapping_mul(0xBF58_476D_1CE4_E5B9);
+        z = (z ^ (z >> 27)).wrapping_mul(0x94D0_49BB_1331_11EB);
+        z ^ (z >> 31)
+    }
+
+    /// Uniform sample in [0, 1) with 24-bit fraction (exact f32 mantissa).
+    pub fn next_f32(&mut self) -> f32 {
+        ((self.next_u64() >> 40) as f32) / ((1u64 << 24) as f32)
+    }
+}
+
+#[derive(Debug)]
+pub struct ExecutionEnv {
+    pub config: ExecutionEnvConfig,
+    pub fill_model: FillModel,
+    snapshots: Vec<SnapshotRow>,
+    cursor: usize,
+    rng: ReplayRng,
+}
+
+impl ExecutionEnv {
+    pub fn new(
+        config: ExecutionEnvConfig,
+        fill_model: FillModel,
+        snapshots: Vec<SnapshotRow>,
+        seed: u64,
+    ) -> Self {
+        Self {
+            config,
+            fill_model,
+            snapshots,
+            cursor: 0,
+            rng: ReplayRng::new(seed),
+        }
+    }
+
+    /// Reset episode to step 0 with a fresh RNG seed.
+    pub fn reset(&mut self, seed: u64) -> [f32; STATE_DIM] {
+        self.cursor = 0;
+        self.rng = ReplayRng::new(seed);
+        self.state(&EpisodeState::new())
+    }
+
+    pub fn snapshots_len(&self) -> usize {
+        self.snapshots.len()
+    }
+
+    /// Assemble the 10-dim observation at the current cursor.
+    pub fn state(&self, ep: &EpisodeState) -> [f32; STATE_DIM] {
+        let s = &self.snapshots[self.cursor];
+        [
+            s.alpha_logit,
+            s.alpha_confidence,
+            s.spread_bps,
+            s.l1_imbalance,
+            s.ofi_sum_5,
+            s.mid_drift_5,
+            ep.position as f32,
+            ep.step as f32 / self.config.horizon_snapshots as f32,
+            s.time_since_trade_s.ln_1p(),
+            s.book_event_rate.ln_1p(),
+        ]
+    }
+
+    /// One env step. Returns `(next_state, reward, done)`. Reward is 0.0
+    /// at every non-terminal step; the terminal step pays out
+    /// `cumulative_pnl` after any forced close.
+    ///
+    /// Returns `None` if `action_id` is outside `0..N_ACTIONS`.
+    pub fn step(
+        &mut self,
+        action_id: u8,
+        ep: &mut EpisodeState,
+    ) -> Option<([f32; STATE_DIM], f32, bool)> {
+        let action = ExecAction::from_u8(action_id)?;
+        let effective = if action.is_legal(ep.position) {
+            action
+        } else {
+            ExecAction::Wait
+        };
+        let decoded = effective.decode();
+
+        let s = &self.snapshots[self.cursor];
+        let feat = FillFeatures {
+            spread_bps: s.spread_bps,
+            l1_imbalance: s.l1_imbalance,
+            ofi_sum_5: s.ofi_sum_5,
+            time_since_trade_s: s.time_since_trade_s,
+        };
+
+        let mut step_pnl = 0.0_f32;
+        let mut filled = false;
+
+        match (decoded.placement, decoded.side, decoded.closing) {
+            // Wait — no order.
+            (Placement::None, _, _) => {}
+
+            // Market buy: cross spread, fill at ask.
+            (Placement::Market, Side::Buy, _) => {
+                step_pnl -= self.config.cost_per_contract;
+                Self::apply_open_fill(ep, s.ask_l[0], self.config.trade_size_contracts);
+                filled = true;
+            }
+            // Market sell: cross spread, fill at bid.
+            (Placement::Market, Side::Sell, _) => {
+                step_pnl -= self.config.cost_per_contract;
+                Self::apply_open_fill(ep, s.bid_l[0], -self.config.trade_size_contracts);
+                filled = true;
+            }
+            // Market flat: close current position at the opposing market.
+            (Placement::Market, Side::None, true) => {
+                let close_px = if ep.position > 0 {
+                    s.bid_l[0]
+                } else {
+                    s.ask_l[0]
+                };
+                step_pnl += Self::apply_close_fill(ep, close_px, self.config.cost_per_contract);
+                filled = true;
+            }
+
+            // Passive limit (L1/L2) — open or close depending on `closing`.
+            (Placement::L1, _, _) | (Placement::L2, _, _) => {
+                let level = if decoded.placement == Placement::L1 { 0 } else { 1 };
+                // Determine the fill_model side to look up.
+                //   Open Buy   → posted at bid, filled by aggressor selling → Side::Buy.
+                //   Open Sell  → posted at ask, filled by aggressor buying  → Side::Sell.
+                //   Close long (sell)  → posted at ask → Side::Sell.
+                //   Close short (buy)  → posted at bid → Side::Buy.
+                let lookup_side = if decoded.closing {
+                    if ep.position > 0 {
+                        Side::Sell
+                    } else {
+                        Side::Buy
+                    }
+                } else {
+                    decoded.side
+                };
+                let fill_p = self.fill_model.fill_prob(lookup_side, level, &feat);
+                if self.rng.next_f32() < fill_p {
+                    if decoded.closing {
+                        let close_px = if ep.position > 0 {
+                            s.ask_l[level]
+                        } else {
+                            s.bid_l[level]
+                        };
+                        // Limit fills earn the maker side, no taker fee.
+                        step_pnl += Self::apply_close_fill(ep, close_px, 0.0);
+                    } else {
+                        let signed_size = match decoded.side {
+                            Side::Buy => self.config.trade_size_contracts,
+                            Side::Sell => -self.config.trade_size_contracts,
+                            Side::None => 0, // unreachable: non-closing limits always have a side
+                        };
+                        let fill_px = if decoded.side == Side::Buy {
+                            s.bid_l[level]
+                        } else {
+                            s.ask_l[level]
+                        };
+                        Self::apply_open_fill(ep, fill_px, signed_size);
+                    }
+                    filled = true;
+                }
+            }
+
+            // Any other (placement, side, closing) tuple is structurally
+            // unreachable given ExecAction::decode(), but keep an arm so
+            // the match is total and the compiler stays happy on enum
+            // extensions.
+            _ => {}
+        }
+
+        ep.cumulative_pnl += step_pnl;
+        if filled {
+            ep.n_fills += 1;
+        }
+
+        // Advance cursor and step counter.
+        self.cursor = (self.cursor + 1).min(self.snapshots.len() - 1);
+        ep.step += 1;
+
+        // Termination: horizon reached or replay buffer exhausted.
+        let done = ep.step >= self.config.horizon_snapshots
+            || self.cursor + 1 >= self.snapshots.len();
+
+        let reward = if done {
+            if ep.position != 0 {
+                let close_px = self.snapshots[self.cursor].mid_price;
+                let forced = Self::apply_close_fill(ep, close_px, self.config.cost_per_contract);
+                ep.cumulative_pnl += forced;
+            }
+            ep.cumulative_pnl
+        } else {
+            0.0
+        };
+
+        Some((self.state(ep), reward, done))
+    }
+
+    /// Apply an opening (or position-scaling) fill. Updates position and
+    /// entry_price (volume-weighted across multiple fills).
+    fn apply_open_fill(ep: &mut EpisodeState, fill_px: f32, signed_size: i32) {
+        let old_pos = ep.position;
+        let new_pos = old_pos + signed_size;
+        if new_pos == 0 {
+            // Net flat — book the close PnL and reset.
+            let direction = old_pos.signum() as f32;
+            let realized = direction * (fill_px - ep.entry_price) * (old_pos.abs() as f32);
+            ep.cumulative_pnl += realized;
+            ep.position = 0;
+            ep.entry_price = 0.0;
+            return;
+        }
+        if old_pos == 0 {
+            ep.entry_price = fill_px;
+        } else if old_pos.signum() == new_pos.signum() {
+            // Same-direction scale: volume-weighted average price.
+            let old_notional = ep.entry_price * (old_pos.abs() as f32);
+            let new_notional = fill_px * (signed_size.abs() as f32);
+            ep.entry_price = (old_notional + new_notional) / (new_pos.abs() as f32);
+        } else {
+            // Crossed through flat — partial close + new opening lot at fill_px.
+            // For the fixed-trade-size invariant of this env we never get here
+            // (trade_size == |position|), but handle correctly for safety.
+            let direction = old_pos.signum() as f32;
+            let closed_qty = old_pos.abs() as f32;
+            let realized = direction * (fill_px - ep.entry_price) * closed_qty;
+            ep.cumulative_pnl += realized;
+            ep.entry_price = fill_px;
+        }
+        ep.position = new_pos;
+    }
+
+    /// Close the current position at `fill_px`, paying `fee_per_contract`
+    /// for each closed contract. Returns the PnL of the close (caller adds
+    /// to `step_pnl`). Resets position/entry_price.
+    fn apply_close_fill(ep: &mut EpisodeState, fill_px: f32, fee_per_contract: f32) -> f32 {
+        let direction = ep.position.signum() as f32;
+        let qty = ep.position.abs() as f32;
+        let gross = direction * (fill_px - ep.entry_price) * qty;
+        let net = gross - fee_per_contract * qty;
+        ep.position = 0;
+        ep.entry_price = 0.0;
+        net
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn synthetic_snapshots(n: usize, slope: f32) -> Vec<SnapshotRow> {
+        // Drift mid AND bid/ask together so the order book moves with the
+        // mid-price — without this, buy-then-sell on an "uptrend" actually
+        // loses money because the bid stays fixed.
+        (0..n)
+            .map(|i| {
+                let drift = i as f32 * slope;
+                SnapshotRow {
+                    mid_price: 4500.0 + drift,
+                    bid_l: [4499.875 + drift, 4499.625 + drift, 4499.375 + drift],
+                    ask_l: [4500.125 + drift, 4500.375 + drift, 4500.625 + drift],
+                    alpha_logit: 1.0,
+                    alpha_confidence: 0.3,
+                    spread_bps: 0.25,
+                    l1_imbalance: 0.55,
+                    ofi_sum_5: 0.0,
+                    mid_drift_5: 0.0,
+                    time_since_trade_s: 0.5,
+                    book_event_rate: 5.0,
+                }
+            })
+            .collect()
+    }
+
+    fn make_env(n: usize, slope: f32, horizon: usize) -> ExecutionEnv {
+        let snaps = synthetic_snapshots(n, slope);
+        ExecutionEnv::new(
+            ExecutionEnvConfig {
+                horizon_snapshots: horizon,
+                trade_size_contracts: 1,
+                cost_per_contract: 0.0625, // quarter-tick
+            },
+            FillModel::skeleton(),
+            snaps,
+            0xCAFEBABE,
+        )
+    }
+
+    #[test]
+    fn state_has_expected_dim_and_is_finite() {
+        let env = make_env(100, 0.1, 100);
+        let ep = EpisodeState::new();
+        let s = env.state(&ep);
+        assert_eq!(s.len(), STATE_DIM);
+        for v in &s {
+            assert!(v.is_finite(), "state element non-finite: {v}");
+        }
+    }
+
+    #[test]
+    fn market_buy_then_market_sell_on_uptrend_profits() {
+        // n=25 > horizon=20 so termination fires via `ep.step >= horizon`
+        // on the FlatMarket call, NOT via cursor exhaustion mid-Wait.
+        let mut env = make_env(25, 1.0, 20);
+        let mut ep = EpisodeState::new();
+
+        env.step(ExecAction::BuyMarket as u8, &mut ep)
+            .expect("BuyMarket step");
+        assert_eq!(ep.position, 1, "should be long 1 contract after BuyMarket");
+
+        for _ in 0..18 {
+            env.step(ExecAction::Wait as u8, &mut ep)
+                .expect("Wait step");
+        }
+
+        let (_, reward, done) = env
+            .step(ExecAction::FlatMarket as u8, &mut ep)
+            .expect("FlatMarket step");
+        assert!(done, "horizon should terminate the episode");
+        // entry at ask cursor 0 = 4500.125; exit at bid cursor 19 = 4518.875.
+        // PnL = -0.0625 (buy fee) + (18.75 - 0.0625 close fee) = 18.625
+        assert!(
+            (reward - 18.625).abs() < 0.01,
+            "expected PnL ~18.625, got {reward}"
+        );
+    }
+
+    #[test]
+    fn illegal_action_degrades_to_wait() {
+        let mut env = make_env(10, 0.0, 10);
+        let mut ep = EpisodeState::new();
+        ep.position = 1;
+        ep.entry_price = 4500.0;
+        let initial_pnl = ep.cumulative_pnl;
+        // BuyMarket while already long → illegal → Wait → no position change
+        let (_, _, _) = env
+            .step(ExecAction::BuyMarket as u8, &mut ep)
+            .expect("step");
+        assert_eq!(ep.position, 1, "illegal BuyMarket must not scale position");
+        assert_eq!(ep.cumulative_pnl, initial_pnl, "no fee for degraded action");
+        assert_eq!(ep.n_fills, 0, "no fill counted for degraded action");
+    }
+
+    #[test]
+    fn terminal_force_close_pays_out_when_position_open() {
+        // n=10 > horizon=5 so termination fires via `ep.step >= horizon`
+        // after BuyMarket + 4 Waits. Position is still open at termination,
+        // so the env force-closes at terminal mid.
+        let mut env = make_env(10, 2.0, 5);
+        let mut ep = EpisodeState::new();
+
+        env.step(ExecAction::BuyMarket as u8, &mut ep).expect("buy");
+        assert_eq!(ep.position, 1);
+
+        let mut final_reward = None;
+        let mut final_done = false;
+        for i in 0..4 {
+            let (_, reward, done) = env
+                .step(ExecAction::Wait as u8, &mut ep)
+                .expect("wait step");
+            if i < 3 {
+                assert!(!done, "Wait #{i} terminated prematurely");
+                assert_eq!(reward, 0.0, "non-terminal reward must be 0.0");
+            }
+            if done {
+                final_reward = Some(reward);
+                final_done = done;
+            }
+        }
+        assert!(final_done, "episode must terminate via horizon by Wait #3");
+        let reward = final_reward.expect("terminal reward must be reported");
+        // entry=4500.125 (ask at cursor 0, drift 0); force-close at mid_price at
+        // cursor 5 = 4500 + 5*2 = 4510. Fee 0.0625 on terminal close.
+        // PnL = -0.0625 (buy fee) + (9.875 - 0.0625 close fee) = 9.75
+        assert!(
+            (reward - 9.75).abs() < 0.01,
+            "expected force-close PnL ~9.75, got {reward}"
+        );
+    }
+
+    #[test]
+    fn rng_is_deterministic_across_resets() {
+        // Two envs with the same seed must produce identical fill outcomes.
+        let mut env_a = make_env(50, 0.0, 50);
+        let mut env_b = make_env(50, 0.0, 50);
+        let mut ep_a = EpisodeState::new();
+        let mut ep_b = EpisodeState::new();
+        for _ in 0..20 {
+            let a = env_a.step(ExecAction::BuyL1 as u8, &mut ep_a).expect("a");
+            let b = env_b.step(ExecAction::BuyL1 as u8, &mut ep_b).expect("b");
+            assert_eq!(
+                ep_a.n_fills, ep_b.n_fills,
+                "deterministic RNG must yield identical fills"
+            );
+            // Match terminal flags too
+            assert_eq!(a.2, b.2);
+        }
+    }
+}

crates/ml/src/env/mod.rs

@@ -6,11 +6,13 @@
 //!
 //! Module manifest (lands incrementally per the Phase E plan):
 //! - `action_space`  (Task 3) — 9 discrete actions + legality gating
-//! - `fill_model`    (Task 4, this commit) — Poisson regression fill simulator
+//! - `fill_model`    (Task 4-5) — Poisson regression fill simulator + cloglog fitter
-//! - `execution_env` (Task 6) — 10-dim observation, step/reset loop
+//! - `execution_env` (Task 6, this commit) — 10-dim observation, step/reset loop
 
 pub mod action_space;
+pub mod execution_env;
 pub mod fill_model;
 
 pub use action_space::{DecodedAction, ExecAction, N_ACTIONS, Placement, Side};
+pub use execution_env::{EpisodeState, ExecutionEnv, ExecutionEnvConfig, ReplayRng, SnapshotRow, STATE_DIM};
 pub use fill_model::{FillCoeffs, FillFeatures, FillModel};