diff --git a/crates/ml/src/env/execution_env.rs b/crates/ml/src/env/execution_env.rs new file mode 100644 index 000000000..9f9280fc0 --- /dev/null +++ b/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, + cursor: usize, + rng: ReplayRng, +} + +impl ExecutionEnv { + pub fn new( + config: ExecutionEnvConfig, + fill_model: FillModel, + snapshots: Vec, + 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 { + // 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); + } + } +} diff --git a/crates/ml/src/env/mod.rs b/crates/ml/src/env/mod.rs index 619876026..1e77e2bf9 100644 --- a/crates/ml/src/env/mod.rs +++ b/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 -//! - `execution_env` (Task 6) — 10-dim observation, step/reset loop +//! - `fill_model` (Task 4-5) — Poisson regression fill simulator + cloglog fitter +//! - `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};