diff --git a/crates/ml-alpha/cuda/actions_to_market_targets.cu b/crates/ml-alpha/cuda/actions_to_market_targets.cu index 1d56f49df..bfdb3c625 100644 --- a/crates/ml-alpha/cuda/actions_to_market_targets.cu +++ b/crates/ml-alpha/cuda/actions_to_market_targets.cu @@ -1,12 +1,5 @@ -// actions_to_market_targets.cu — translate the 9-action discrete grid -// to LobSim's `market_targets[B*2] = {side, size}` device buffer -// (Phase R6 of the integrated RL trainer rebuild; -// see docs/superpowers/plans/2026-05-23-integrated-rl-trainer-rebuild.md). -// -// Replaces the host-side `submit_market(batch_idx, side, size, ts_ns)` -// loop the flawed Phase F shipped in LobSimEnvAdapter (which iterated -// the batch on CPU and called submit_market per batch — host -// orchestration of GPU work per `feedback_cpu_is_read_only`). +// actions_to_market_targets.cu — translate the 11-action discrete grid +// to LobSim's `market_targets[B*2] = {side, size}` device buffer. // // Action mapping (must match `crate::rl::common::Action` and // `LobSimCuda::submit_market_immediate`'s expected `{side, size}` pair): @@ -20,23 +13,56 @@ // 6 LongLarge side=0 (buy), size=2 // 7 TrailTighten side=2 (no-op),size=0 — handled by rl_trail_mutate kernel // 8 TrailLoosen side=2 (no-op),size=0 — handled by rl_trail_mutate kernel -// 9 HalfFlatLong side=1 (sell), size=max(1, position_lots/2) if pos>0 (SP20 P4) -// 10 HalfFlatShort side=0 (buy), size=max(1, |position_lots|/2) if pos<0 (SP20 P4) +// 9 HalfFlatLong side=1 (sell), size=oldest_unit_lots (if pyramid>1) +// or ceil(|pos|/2) (if pyramid<=1) +// 10 HalfFlatShort side=0 (buy), size=oldest_unit_lots (if pyramid>1) +// or ceil(|pos|/2) (if pyramid<=1) // -// Actions 3/4 require reading the current `position_lots` from the -// Pos struct (`offset 0`, i32). Other actions are pure constant -// translations. +// Pyramid semantics for actions 5/6 when already long (and 0/1 when +// already short): +// - If (mid - last_entry) >= threshold AND count < MAX_UNITS → ADD +// - Else if count < MAX_UNITS → REPLACE (fill replaces oldest unit) +// - Else (at MAX_UNITS, threshold not met) → no-op // -// `Pos` struct layout (from crates/ml-backtesting/src/lob/mod.rs::PosFlat): -// offset 0: position_lots: i32 -// (followed by other fields — see extract_realized_pnl_delta.cu header) +// HalfFlat semantics (a9/a10) when pyramid_units_count > 1: +// - Close the oldest active unit (lowest active index, OR +// close_unit_index[b] if trail-stop overrode it to a specific slot). +// - Emit size = |lots| of that unit. +// - When pyramid_units_count <= 1, fall back to ceil(|pos|/2). // // Element-wise, one thread per batch entry. No atomics. +#include + +#define MAX_UNITS 4 +#define RL_PYRAMID_THRESHOLD_INDEX 506 +#define RL_OUTCOME_EMA_INDEX 508 +#define RL_ANTIMARTINGALE_KAPPA_INDEX 509 +#define RL_ANTIMARTINGALE_MIN_INDEX 510 +#define RL_ANTIMARTINGALE_MAX_INDEX 511 + +__device__ __forceinline__ int antimartingale_size(int base_size, const float* isv) { + const float ema = isv[RL_OUTCOME_EMA_INDEX]; + const float kappa = isv[RL_ANTIMARTINGALE_KAPPA_INDEX]; + const float lo = isv[RL_ANTIMARTINGALE_MIN_INDEX]; + const float hi = isv[RL_ANTIMARTINGALE_MAX_INDEX]; + const float scale = fmaxf(lo, fminf(1.0f + kappa * ema, hi)); + const int sized = (int)(((float)base_size) * scale + 0.5f); + return (sized > 0) ? sized : 1; +} + extern "C" __global__ void actions_to_market_targets( - const int* __restrict__ actions, // [b_size] - const unsigned char* __restrict__ pos_state, // [b_size * pos_bytes] - int* __restrict__ market_targets, // [b_size * 2] OUT + const int* __restrict__ actions, // [b_size] + const unsigned char* __restrict__ pos_state, // [b_size * pos_bytes] + int* __restrict__ market_targets, // [b_size * 2] OUT + const float* __restrict__ isv, + const float* __restrict__ bid_px, // [BOOK_LEVELS] + const float* __restrict__ ask_px, // [BOOK_LEVELS] + const float* __restrict__ unit_entry_price, // [b_size * MAX_UNITS] + const unsigned char* __restrict__ unit_active, // [b_size * MAX_UNITS] + const int* __restrict__ unit_lots, // [b_size * MAX_UNITS] + const int* __restrict__ pyramid_units_count, // [b_size] + const int* __restrict__ close_unit_index, // [b_size] (-1 = auto oldest) int b_size, int pos_bytes ) { @@ -44,54 +70,173 @@ extern "C" __global__ void actions_to_market_targets( if (b >= b_size) return; const int action = actions[b]; + const int position_lots = *(const int*)(pos_state + b * pos_bytes); int side = 2; // default no-op int size = 0; if (action == 0) { - side = 1; size = 2; + // ShortLarge — when already short, apply pyramid logic. + if (position_lots < 0) { + const int count = pyramid_units_count[b]; + const float threshold = isv[RL_PYRAMID_THRESHOLD_INDEX]; + const float mid = 0.5f * (bid_px[0] + ask_px[0]); + float last_entry = 0.0f; + const int base = b * MAX_UNITS; + for (int u = MAX_UNITS - 1; u >= 0; --u) { + if (unit_active[base + u]) { + last_entry = unit_entry_price[base + u]; + break; + } + } + const float dist = last_entry - mid; + if (dist >= threshold && count < MAX_UNITS) { + side = 1; size = 2; + } else if (count < MAX_UNITS) { + side = 1; size = 2; + } + } else { + side = 1; size = antimartingale_size(2, isv); + } } else if (action == 1) { - side = 1; size = 1; + // ShortSmall — when already short, apply pyramid logic. + if (position_lots < 0) { + const int count = pyramid_units_count[b]; + const float threshold = isv[RL_PYRAMID_THRESHOLD_INDEX]; + const float mid = 0.5f * (bid_px[0] + ask_px[0]); + float last_entry = 0.0f; + const int base = b * MAX_UNITS; + for (int u = MAX_UNITS - 1; u >= 0; --u) { + if (unit_active[base + u]) { + last_entry = unit_entry_price[base + u]; + break; + } + } + const float dist = last_entry - mid; + if (dist >= threshold && count < MAX_UNITS) { + side = 1; size = 1; + } else if (count < MAX_UNITS) { + side = 1; size = 1; + } + } else { + side = 1; size = antimartingale_size(1, isv); + } } else if (action == 2) { // Hold — no-op. } else if (action == 3) { - const int position_lots = *(const int*)(pos_state + b * pos_bytes); if (position_lots > 0) { side = 1; size = position_lots; } - // else no-op (no long to flatten) } else if (action == 4) { - const int position_lots = *(const int*)(pos_state + b * pos_bytes); if (position_lots < 0) { side = 0; size = -position_lots; } - // else no-op (no short to flatten) } else if (action == 5) { - side = 0; size = 1; + // LongSmall — when already long, apply pyramid logic. + if (position_lots > 0) { + const int count = pyramid_units_count[b]; + const float threshold = isv[RL_PYRAMID_THRESHOLD_INDEX]; + const float mid = 0.5f * (bid_px[0] + ask_px[0]); + float last_entry = 0.0f; + const int base = b * MAX_UNITS; + for (int u = MAX_UNITS - 1; u >= 0; --u) { + if (unit_active[base + u]) { + last_entry = unit_entry_price[base + u]; + break; + } + } + const float dist = mid - last_entry; + if (dist >= threshold && count < MAX_UNITS) { + side = 0; size = 1; + } else if (count < MAX_UNITS) { + side = 0; size = 1; + } + } else { + side = 0; size = antimartingale_size(1, isv); + } } else if (action == 6) { - side = 0; size = 2; + // LongLarge — when already long, apply pyramid logic. + if (position_lots > 0) { + const int count = pyramid_units_count[b]; + const float threshold = isv[RL_PYRAMID_THRESHOLD_INDEX]; + const float mid = 0.5f * (bid_px[0] + ask_px[0]); + float last_entry = 0.0f; + const int base = b * MAX_UNITS; + for (int u = MAX_UNITS - 1; u >= 0; --u) { + if (unit_active[base + u]) { + last_entry = unit_entry_price[base + u]; + break; + } + } + const float dist = mid - last_entry; + if (dist >= threshold && count < MAX_UNITS) { + side = 0; size = 2; + } else if (count < MAX_UNITS) { + side = 0; size = 2; + } + } else { + side = 0; size = antimartingale_size(2, isv); + } } else if (action == 9) { - // SP20 P4 HalfFlatLong: close ⌈|pos|/2⌉ lots of long position. - // No-op if not long. - const int position_lots = *(const int*)(pos_state + b * pos_bytes); + // HalfFlatLong: close oldest unit when pyramided, else ceil(|pos|/2). if (position_lots > 0) { side = 1; - const int half = (position_lots + 1) / 2; // round-up so min 1 - size = (half > 0) ? half : 1; + const int count = pyramid_units_count[b]; + if (count > 1) { + const int base = b * MAX_UNITS; + const int target_u = close_unit_index[b]; + int close_slot = -1; + if (target_u >= 0 && target_u < MAX_UNITS && unit_active[base + target_u]) { + close_slot = target_u; + } else { + for (int u = 0; u < MAX_UNITS; ++u) { + if (unit_active[base + u]) { close_slot = u; break; } + } + } + if (close_slot >= 0) { + const int slot_lots = unit_lots[base + close_slot]; + size = (slot_lots > 0) ? slot_lots : -slot_lots; + } else { + const int half = (position_lots + 1) / 2; + size = (half > 0) ? half : 1; + } + } else { + const int half = (position_lots + 1) / 2; + size = (half > 0) ? half : 1; + } } } else if (action == 10) { - // SP20 P4 HalfFlatShort: close ⌈|pos|/2⌉ lots of short position. - const int position_lots = *(const int*)(pos_state + b * pos_bytes); + // HalfFlatShort: close oldest unit when pyramided, else ceil(|pos|/2). if (position_lots < 0) { side = 0; + const int count = pyramid_units_count[b]; const int abs_pos = -position_lots; - const int half = (abs_pos + 1) / 2; - size = (half > 0) ? half : 1; + if (count > 1) { + const int base = b * MAX_UNITS; + const int target_u = close_unit_index[b]; + int close_slot = -1; + if (target_u >= 0 && target_u < MAX_UNITS && unit_active[base + target_u]) { + close_slot = target_u; + } else { + for (int u = 0; u < MAX_UNITS; ++u) { + if (unit_active[base + u]) { close_slot = u; break; } + } + } + if (close_slot >= 0) { + const int slot_lots = unit_lots[base + close_slot]; + size = (slot_lots > 0) ? slot_lots : -slot_lots; + } else { + const int half = (abs_pos + 1) / 2; + size = (half > 0) ? half : 1; + } + } else { + const int half = (abs_pos + 1) / 2; + size = (half > 0) ? half : 1; + } } } // actions 7, 8 (TrailTighten/Loosen): no fill — handled by - // rl_trail_mutate kernel (SP20 P5). The side=2 no-op default is - // correct here. + // rl_trail_mutate kernel. The side=2 no-op default is correct here. market_targets[b * 2 + 0] = side; market_targets[b * 2 + 1] = size; diff --git a/crates/ml-alpha/cuda/rl_trail_stop_check.cu b/crates/ml-alpha/cuda/rl_trail_stop_check.cu index ad38f59f5..581fc5ce0 100644 --- a/crates/ml-alpha/cuda/rl_trail_stop_check.cu +++ b/crates/ml-alpha/cuda/rl_trail_stop_check.cu @@ -1,31 +1,24 @@ -// rl_trail_stop_check.cu — SP20 P5 trail-stop breach check. +// rl_trail_stop_check.cu — trail-stop breach check with per-unit +// partial-flat routing. // -// For each batch with an active unit (slot 0 currently used; slots -// 1-3 reserved for SP20 P7 pyramid expansion), check if the current -// market mid moved adversely beyond the unit's trail_distance from -// its entry_price. On breach, OVERRIDE actions[b] to FlatFromLong -// (a3) or FlatFromShort (a4) so actions_to_market_targets force- -// closes the position via existing flat-from-X plumbing. -// -// Per `pearl_stop_checks_run_at_deadline_cadence`: stop check fires -// every step (event rate). Force-close routes through -// actions_to_market_targets → step_fill_from_market_targets → -// apply_fill_to_pos so realized PnL is captured correctly. -// -// SP20 v3 spec §3 P5 calls for routing stop-close through the -// per-unit partial-flat path (a9/a10). That requires P4 (N_ACTIONS=11) -// which is NOT in this commit's scope. For now, ANY unit's trail -// breach closes the ENTIRE position via full FlatFromLong/Short. -// When SP20 P4 ships, the close_unit_index buffer + a9/a10 routing -// per spec §3 P5 supersedes this full-flat fallback. +// For each batch with active units, checks if the current market mid +// moved adversely beyond the unit's trail_distance from its +// entry_price. On breach: +// - If pyramid_units_count > 1: override actions[b] to HalfFlatLong +// (a9) or HalfFlatShort (a10) and write close_unit_index[b] = u +// so actions_to_market_targets closes THAT specific unit. +// - If pyramid_units_count <= 1: override actions[b] to FlatFromLong +// (a3) or FlatFromShort (a4) for full position close. // // Market mid sourced from lobsim's bid_px_d[0] + ask_px_d[0] (shared // across batches — single market). One thread per (batch, unit). // -// Per `feedback_no_atomicadd`: writes are per-batch (action override -// uses last-thread-wins by writing same value; trail breach by ANY -// unit triggers the same action override regardless of which thread). -// Per `feedback_cpu_is_read_only`: pure device-side. +// Thread 0 of each block resets close_unit_index[b] = -1 before any +// breach logic runs (device-side per-step reset, no host traffic). +// +// Per `feedback_no_atomicadd`: writes are per-batch (same-value +// last-writer-wins for action override). Per +// `feedback_cpu_is_read_only`: pure device-side. #include @@ -33,22 +26,33 @@ #define BOOK_LEVELS 10 #define ACTION_FLAT_FROM_LONG 3 #define ACTION_FLAT_FROM_SHORT 4 +#define ACTION_HALF_FLAT_LONG 9 +#define ACTION_HALF_FLAT_SHORT 10 extern "C" __global__ void rl_trail_stop_check( - int* __restrict__ actions, // [B] IN/OUT (may be overridden) - const float* __restrict__ bid_px, // [BOOK_LEVELS] shared best book + int* __restrict__ actions, // [B] IN/OUT + const float* __restrict__ bid_px, // [BOOK_LEVELS] const float* __restrict__ ask_px, // [BOOK_LEVELS] const unsigned char* __restrict__ unit_active, // [B * MAX_UNITS] const float* __restrict__ unit_entry_price, // [B * MAX_UNITS] const int* __restrict__ unit_lots, // [B * MAX_UNITS] const float* __restrict__ unit_trail_distance, // [B * MAX_UNITS] + const int* __restrict__ pyramid_units_count, // [B] + int* __restrict__ close_unit_index, // [B] OUT (-1 = auto) int b_size ) { const int b = blockIdx.x; const int u = threadIdx.x; if (b >= b_size || u >= MAX_UNITS) return; - // Compute current mid from shared book (all threads read same). + // Thread 0 resets close_unit_index for this batch (device-side + // per-step clear — avoids host memset per + // feedback_no_htod_htoh_only_mapped_pinned). + if (u == 0) { + close_unit_index[b] = -1; + } + __syncthreads(); + const float best_bid = bid_px[0]; const float best_ask = ask_px[0]; const float mid = 0.5f * (best_bid + best_ask); @@ -57,31 +61,27 @@ extern "C" __global__ void rl_trail_stop_check( if (unit_active[idx] == 0) return; const int lots = unit_lots[idx]; - if (lots == 0) return; // defensive — active flag should imply lots != 0 + if (lots == 0) return; const float entry = unit_entry_price[idx]; const float trail = unit_trail_distance[idx]; bool breach = false; - int override_action = -1; if (lots > 0) { - // Long: stop fires when mid drops below entry - trail - if (mid < entry - trail) { - breach = true; - override_action = ACTION_FLAT_FROM_LONG; - } + if (mid < entry - trail) breach = true; } else { - // Short: stop fires when mid rises above entry + trail - if (mid > entry + trail) { - breach = true; - override_action = ACTION_FLAT_FROM_SHORT; - } + if (mid > entry + trail) breach = true; } if (breach) { - // Multiple unit threads may write the same value — that's fine - // since the action is identical (close the whole position via - // full flat per the spec-noted P4 dependency). Last write wins. - actions[b] = override_action; + const int count = pyramid_units_count[b]; + if (count > 1) { + // Partial close — route through HalfFlat with unit override. + close_unit_index[b] = u; + actions[b] = (lots > 0) ? ACTION_HALF_FLAT_LONG : ACTION_HALF_FLAT_SHORT; + } else { + // Single unit — full position close. + actions[b] = (lots > 0) ? ACTION_FLAT_FROM_LONG : ACTION_FLAT_FROM_SHORT; + } } } diff --git a/crates/ml-alpha/cuda/rl_unit_state_update.cu b/crates/ml-alpha/cuda/rl_unit_state_update.cu index c5df0c438..4a188db12 100644 --- a/crates/ml-alpha/cuda/rl_unit_state_update.cu +++ b/crates/ml-alpha/cuda/rl_unit_state_update.cu @@ -1,4 +1,5 @@ -// rl_unit_state_update.cu — SP20 P1 per-unit trade state machine. +// rl_unit_state_update.cu — per-unit trade state machine with pyramid +// expansion. // // Runs AFTER each step's fill+extract sequence. Detects per-batch // position transitions vs the prior step and updates per-unit @@ -7,15 +8,13 @@ // prev_pos == 0 AND curr_pos != 0 → OPEN: write unit slot 0 // prev_pos != 0 AND curr_pos == 0 → CLOSE: deactivate all units // sign(prev) != sign(curr) AND both != 0 → REVERSE: close + open -// sign(prev) == sign(curr) AND both != 0 → CONTINUE: refresh lots -// -// This commit uses unit slot 0 only — slots 1-3 are allocated for -// SP20 P7 pyramid expansion but not populated yet. Pyramid add / -// partial-flat logic lives in actions_to_market_targets (P7). +// sign(prev) == sign(curr) AND both != 0 → CONTINUE: +// |curr| > |prev| → pyramid ADD: allocate next free slot +// |curr| < |prev| → partial FLAT: deactivate oldest active slot +// |curr| == |prev| → no structural change // // Per `pearl_first_observation_bootstrap`: prev_pos starts at sentinel // 0; first non-zero curr_pos triggers OPEN via the standard branch. -// No special bootstrap path needed. // // Per `feedback_no_atomicadd`: per-batch independent writes only. // Per `feedback_cpu_is_read_only`: pure device-side update. @@ -26,22 +25,7 @@ #define MAX_UNITS 4 #define RL_TRAIL_K_INIT_INDEX 496 -// realized_return_var_ema lives at a known SP19-era slot — reuse -// rather than allocate a new one. Currently RL_ADV_VAR_STREAM_M2_INDEX -// holds the streaming variance running M2; bootstrap derives σ from -// √(M2 / count). For SP20 P1+P5 simplicity, use a fallback constant -// if streaming-var hasn't bootstrapped (avoids div-by-zero collapse). -// -// Practical compromise: use `mean_abs_pnl_ema` (slot 417) as a proxy -// for typical PnL magnitude. trail_init = k_init × mean_abs_pnl_ema / -// position_lots. This is in dollar units which matches vwap_entry's -// price units after the position-lots normalization. -// -// Even simpler: use a fixed bootstrap fraction of vwap_entry as the -// trail distance — k_init × vwap_entry / 1000 = ~0.2% of price for -// k_init=2. This is rough but tradable and avoids dependency on -// streaming-var bootstrap. SP20 P5 follow-up will integrate proper -// vol-derived initialization. +#define RL_PYRAMID_ADD_COUNT_INDEX 507 #define BOOTSTRAP_TRAIL_FRACTION 1.0e-3f // PosFlat layout (matches crates/ml-backtesting/src/lob/mod.rs::PosFlat): @@ -52,7 +36,7 @@ extern "C" __global__ void rl_unit_state_update( const unsigned char* __restrict__ pos_state, // [B * pos_bytes] - const float* __restrict__ isv, + float* __restrict__ isv, int* __restrict__ prev_pos_lots, // [B] tracking float* __restrict__ unit_entry_price, // [B * MAX_UNITS] int* __restrict__ unit_entry_step, // [B * MAX_UNITS] @@ -73,7 +57,6 @@ extern "C" __global__ void rl_unit_state_update( const int prev_pos = prev_pos_lots[b]; const float k_init = isv[RL_TRAIL_K_INIT_INDEX]; - // Bootstrap trail uses fraction of price (see header comment). const float bootstrap_d = fabsf(vwap) * BOOTSTRAP_TRAIL_FRACTION * k_init; const int base = b * MAX_UNITS; @@ -82,6 +65,8 @@ extern "C" __global__ void rl_unit_state_update( const bool is_flat = (curr_pos == 0); const bool reversed = (!was_flat && !is_flat && ((prev_pos > 0) != (curr_pos > 0))); + int add_count = 0; + if (is_flat && !was_flat) { // CLOSE: deactivate all units, zero lots. #pragma unroll @@ -104,15 +89,52 @@ extern "C" __global__ void rl_unit_state_update( unit_trail_distance[base + 0] = bootstrap_d; unit_active[base + 0] = 1; pyramid_units_count[b] = 1; + add_count = 1; } else if (!is_flat && !was_flat && !reversed) { - // CONTINUE: same direction, update unit_lots[0] to current - // position (covers pyramid adds done by actions_to_market_targets - // even though SP20 P7 isn't shipped yet — unit_lots stays in - // sync with aggregate position). - unit_lots[base + 0] = curr_pos; + // CONTINUE: same direction. + const int abs_curr = (curr_pos > 0) ? curr_pos : -curr_pos; + const int abs_prev = (prev_pos > 0) ? prev_pos : -prev_pos; + + if (abs_curr > abs_prev) { + // Pyramid ADD: lots grew. Allocate next free unit slot. + int count = pyramid_units_count[b]; + if (count < MAX_UNITS) { + const int slot = count; + unit_entry_price[base + slot] = vwap; + unit_entry_step[base + slot] = current_step; + const int delta_lots = abs_curr - abs_prev; + unit_lots[base + slot] = (curr_pos > 0) ? delta_lots : -delta_lots; + unit_initial_r[base + slot] = bootstrap_d; + unit_trail_distance[base + slot] = bootstrap_d; + unit_active[base + slot] = 1; + pyramid_units_count[b] = count + 1; + add_count = 1; + } + } else if (abs_curr < abs_prev) { + // Partial FLAT: lots shrank. Deactivate the oldest active + // unit (lowest index among active slots). + for (int u = 0; u < MAX_UNITS; ++u) { + if (unit_active[base + u]) { + unit_active[base + u] = 0; + unit_lots[base + u] = 0; + int count = pyramid_units_count[b]; + if (count > 0) { + pyramid_units_count[b] = count - 1; + } + break; + } + } + } + // abs_curr == abs_prev: no structural change. } // (is_flat && was_flat): no-op, position remains flat. // Always advance prev_pos tracker. prev_pos_lots[b] = curr_pos; + + // Write add_count to ISV diag slot. b_size=1 in practice; for + // multi-batch, only thread 0 writes (avoids atomicAdd). + if (b == 0) { + isv[RL_PYRAMID_ADD_COUNT_INDEX] = (float)add_count; + } } diff --git a/crates/ml-alpha/examples/alpha_rl_train.rs b/crates/ml-alpha/examples/alpha_rl_train.rs index e04d608b4..cde02f902 100644 --- a/crates/ml-alpha/examples/alpha_rl_train.rs +++ b/crates/ml-alpha/examples/alpha_rl_train.rs @@ -572,6 +572,12 @@ fn main() -> Result<()> { cli.n_backtests, ) .context("diag: read prev_position_lots_d")?; + let pyramid_count_host = read_slice_i32_d_pub( + dev_stream, + &trainer.pyramid_units_count_d, + cli.n_backtests, + ) + .context("diag: read pyramid_units_count_d")?; let mut act_hist = [0u32; N_ACTIONS]; for &a in &actions_host { @@ -930,13 +936,18 @@ fn main() -> Result<()> { "position": { "lots": position_lots_host, }, - // SP20 P6 heat cap diag — how many batch entries fired the - // position-size guard this step. 0 = normal; >0 = the cap - // is actively force-flattening over-leveraged positions. + // Heat cap diag — how many batch entries fired the position-size + // guard this step. 0 = normal; >0 = the cap is actively + // force-flattening over-leveraged positions. "heat_cap": { "fired_count": isv[ml_alpha::rl::isv_slots::RL_HEAT_CAP_FIRED_COUNT_INDEX], "max_lots": isv[ml_alpha::rl::isv_slots::RL_HEAT_CAP_MAX_LOTS_INDEX], }, + "pyramid": { + "units_count": pyramid_count_host, + "add_count": isv[ml_alpha::rl::isv_slots::RL_PYRAMID_ADD_COUNT_INDEX], + "outcome_ema": isv[ml_alpha::rl::isv_slots::RL_OUTCOME_EMA_INDEX], + }, // SP20 P3 FRD head diag — per-horizon softmax entropy + argmax-mode // bucket index (averaged across the batch). At init, Xavier × 0.1 // weights → logits ≈ 0 → entropy ≈ ln(21) = 3.044 and the argmax diff --git a/crates/ml-alpha/src/rl/isv_slots.rs b/crates/ml-alpha/src/rl/isv_slots.rs index 96ffaeaac..cf65d1680 100644 --- a/crates/ml-alpha/src/rl/isv_slots.rs +++ b/crates/ml-alpha/src/rl/isv_slots.rs @@ -792,19 +792,51 @@ pub const RL_FRD_HORIZON_3_TICKS_INDEX: usize = 502; /// realised σ drifts; only the 21-atom count is structural compile-time. pub const RL_FRD_BUCKET_RANGE_SIGMA_INDEX: usize = 503; -/// SP20 P6 — maximum absolute position (lots) before the heat-cap -/// kernel force-flats the entire position. Default 8 = +/// Maximum absolute position (lots) before the heat-cap kernel +/// force-flats the entire position. Default 8 = /// `MAX_UNITS(4) × max_single_order_size(2)`. The kernel compares /// `|position_lots|` from `pos_state` against this slot and overrides /// `actions[b]` to FlatFromLong/Short when exceeded. Last-defense /// guard against runaway pyramid accumulation. pub const RL_HEAT_CAP_MAX_LOTS_INDEX: usize = 504; -/// SP20 P6 — diagnostic: number of batch entries where the heat cap -/// fired this step. Written by `rl_position_heat_check`; read by -/// diag JSONL. Zero when no batch entry exceeds the cap. +/// Diagnostic: number of batch entries where the heat cap fired this +/// step. Written by `rl_position_heat_check`; read by diag JSONL. +/// Zero when no batch entry exceeds the cap. pub const RL_HEAT_CAP_FIRED_COUNT_INDEX: usize = 505; +/// Minimum mid-price distance from the last active unit's entry price +/// before allowing a pyramid add. When position already exists in the +/// same direction and mid hasn't moved at least this distance from the +/// newest unit's entry, actions 5/6 (long) or 0/1 (short) emit REPLACE +/// (close oldest + re-open) instead of ADD. +/// Default: 0.50 (2× ES tick size = 2 × 0.25). +pub const RL_PYRAMID_THRESHOLD_INDEX: usize = 506; + +/// Diagnostic: number of pyramid-add events (new unit slot populated) +/// this step across the batch. Written by `rl_unit_state_update`; read +/// by diag JSONL. +pub const RL_PYRAMID_ADD_COUNT_INDEX: usize = 507; + +/// Signed trade-outcome EMA (updated on done steps only). Positive = +/// recent trades profitable, negative = recent losses. Used by the +/// anti-martingale sizing logic in `actions_to_market_targets` to scale +/// opening order sizes. Bootstrap: first-observation replace (sentinel 0). +pub const RL_OUTCOME_EMA_INDEX: usize = 508; + +/// Anti-martingale κ: scaling coefficient in +/// `size_eff = base × clamp(1 + κ × outcome_ema, MIN, MAX)`. +/// Higher κ = more aggressive scaling. Default: 1.0. +pub const RL_ANTIMARTINGALE_KAPPA_INDEX: usize = 509; + +/// Anti-martingale size floor (MIN in the clamp). Default: 0.5 (never +/// go below half the base order size). +pub const RL_ANTIMARTINGALE_MIN_INDEX: usize = 510; + +/// Anti-martingale size ceiling (MAX in the clamp). Default: 2.0 +/// (never exceed 2× base order size). +pub const RL_ANTIMARTINGALE_MAX_INDEX: usize = 511; + /// Last RL-allocated slot index (exclusive). The integrated trainer /// extends `ISV_TOTAL_DIM` to at least this value at trainer init time. -pub const RL_SLOTS_END: usize = 506; +pub const RL_SLOTS_END: usize = 512; diff --git a/crates/ml-alpha/src/rl/reward.rs b/crates/ml-alpha/src/rl/reward.rs index 880482e44..5cb9c20eb 100644 --- a/crates/ml-alpha/src/rl/reward.rs +++ b/crates/ml-alpha/src/rl/reward.rs @@ -92,13 +92,20 @@ pub trait RlLobBackend { /// Number of parallel backtests (batch dimension). fn n_backtests(&self) -> usize; - /// SP20 P5: read-only accessor for the shared best-bid price - /// array (`BOOK_LEVELS` floats, lobsim-wide — single book shared - /// across batches). Used by `rl_trail_stop_check` to compute - /// current mid for per-batch trail breach checks. + /// Read-only accessor for the shared best-bid price array + /// (`BOOK_LEVELS` floats, lobsim-wide — single book shared across + /// batches). Used by trail-stop and pyramid-threshold checks. fn bid_px_d(&self) -> &CudaSlice; - /// SP20 P5: read-only accessor for the shared best-ask price - /// array. Pair to `bid_px_d`. + /// Read-only accessor for the shared best-ask price array. Pair to + /// `bid_px_d`. fn ask_px_d(&self) -> &CudaSlice; + + /// Combined disjoint-field accessor for the actions_to_market_targets + /// kernel which needs pos (read), market_targets (write), AND book + /// prices (read) simultaneously. Implementations split via Rust's + /// disjoint-field rule across four independent struct fields. + fn pos_book_and_market_targets_mut( + &mut self, + ) -> (&CudaSlice, &CudaSlice, &CudaSlice, &mut CudaSlice); } diff --git a/crates/ml-alpha/src/trainer/integrated.rs b/crates/ml-alpha/src/trainer/integrated.rs index 864cbffe5..26e7edcbb 100644 --- a/crates/ml-alpha/src/trainer/integrated.rs +++ b/crates/ml-alpha/src/trainer/integrated.rs @@ -456,6 +456,9 @@ pub struct IntegratedTrainer { pub unit_trail_distance_d: CudaSlice, // [B * 4] pub unit_active_d: CudaSlice, // [B * 4] pub pyramid_units_count_d: CudaSlice, // [B] + pub close_unit_index_d: CudaSlice, // [B] — trail-stop override + // (-1 = auto oldest). Reset to + // -1 each step before trail check. pub unit_prev_pos_lots_d: CudaSlice, // [B] — separate from // extract_realized_pnl_delta's // prev_position_lots_d so unit @@ -1119,6 +1122,9 @@ impl IntegratedTrainer { let pyramid_units_count_d = stream .alloc_zeros::(b_size) .context("alloc pyramid_units_count_d")?; + let close_unit_index_d = stream + .alloc_zeros::(b_size) + .context("alloc close_unit_index_d")?; let unit_prev_pos_lots_d = stream .alloc_zeros::(b_size) .context("alloc unit_prev_pos_lots_d")?; @@ -1312,6 +1318,7 @@ impl IntegratedTrainer { unit_trail_distance_d, unit_active_d, pyramid_units_count_d, + close_unit_index_d, unit_prev_pos_lots_d, _actions_to_market_targets_module: actions_to_market_targets_module, actions_to_market_targets_fn, @@ -1438,7 +1445,7 @@ impl IntegratedTrainer { // (slot, value) pair — pure device write, no HtoD per // `feedback_no_htod_htoh_only_mapped_pinned`. { - let isv_constants: [(usize, f32); 47] = [ + let isv_constants: [(usize, f32); 51] = [ // Static seeds for the adaptive reward-clamp controller — // these are the initial values that // `rl_reward_clamp_controller` will replace once it @@ -1494,9 +1501,11 @@ impl IntegratedTrainer { crate::rl::common::FRD_HORIZON_TICKS[2] as f32), (crate::rl::isv_slots::RL_FRD_BUCKET_RANGE_SIGMA_INDEX, crate::rl::common::FRD_BUCKET_RANGE_SIGMA), - // SP20 P6 heat-cap: max absolute position (lots). - // Default 8 = MAX_UNITS(4) × max_single_order_size(2). (crate::rl::isv_slots::RL_HEAT_CAP_MAX_LOTS_INDEX, 8.0), + (crate::rl::isv_slots::RL_PYRAMID_THRESHOLD_INDEX, 0.50), + (crate::rl::isv_slots::RL_ANTIMARTINGALE_KAPPA_INDEX, 1.0), + (crate::rl::isv_slots::RL_ANTIMARTINGALE_MIN_INDEX, 0.5), + (crate::rl::isv_slots::RL_ANTIMARTINGALE_MAX_INDEX, 2.0), (crate::rl::isv_slots::RL_KL_TARGET_INDEX, 0.01), (crate::rl::isv_slots::RL_IMPROVEMENT_THRESHOLD_INDEX, 0.99), (crate::rl::isv_slots::RL_PLATEAU_PATIENCE_INDEX, 1000.0), @@ -1945,17 +1954,30 @@ impl IntegratedTrainer { /// as `[b_size * 2]` row-major. Reads `pos_state_d` for actions /// that depend on current position (a3/a4 flat, a9/a10 half-flat). /// Public for `sp20_trade_management_kernels` GPU-oracle tests. + #[allow(clippy::too_many_arguments)] pub fn launch_actions_to_market_targets( &self, actions_d: &CudaSlice, pos_state_d: &CudaSlice, market_targets_d: &mut CudaSlice, + bid_px_d: &CudaSlice, + ask_px_d: &CudaSlice, + unit_entry_price_d: &CudaSlice, + unit_active_d: &CudaSlice, + unit_lots_d: &CudaSlice, + pyramid_units_count_d: &CudaSlice, + close_unit_index_d: &CudaSlice, b_size: usize, pos_bytes: usize, ) -> Result<()> { debug_assert_eq!(actions_d.len(), b_size); debug_assert_eq!(market_targets_d.len(), b_size * 2); debug_assert_eq!(pos_state_d.len(), b_size * pos_bytes); + debug_assert_eq!(unit_entry_price_d.len(), b_size * 4); + debug_assert_eq!(unit_active_d.len(), b_size * 4); + debug_assert_eq!(unit_lots_d.len(), b_size * 4); + debug_assert_eq!(pyramid_units_count_d.len(), b_size); + debug_assert_eq!(close_unit_index_d.len(), b_size); let cfg = LaunchConfig { grid_dim: (((b_size as u32) + 31) / 32, 1, 1), block_dim: (32, 1, 1), @@ -1968,6 +1990,14 @@ impl IntegratedTrainer { .arg(actions_d) .arg(pos_state_d) .arg(market_targets_d) + .arg(&self.isv_d) + .arg(bid_px_d) + .arg(ask_px_d) + .arg(unit_entry_price_d) + .arg(unit_active_d) + .arg(unit_lots_d) + .arg(pyramid_units_count_d) + .arg(close_unit_index_d) .arg(&b_size_i) .arg(&pos_bytes_i); unsafe { @@ -2012,11 +2042,13 @@ impl IntegratedTrainer { Ok(()) } - /// SP20 P5: launch `rl_trail_stop_check` — for each active unit, - /// check if mid (`0.5*(bid_px[0]+ask_px[0])`) breached the unit's - /// stop (entry ± trail_distance). On breach, OVERRIDES `actions_d[b]` - /// to FlatFromLong (a3) or FlatFromShort (a4). Public for SP20 P14 - /// GPU-oracle tests. + /// Launch `rl_trail_stop_check` — for each active unit, check if + /// mid breached the unit's stop (entry ± trail_distance). On breach + /// with multiple units: overrides to HalfFlat + writes + /// close_unit_index. With single unit: overrides to full flat. + /// Resets close_unit_index_d to -1 device-side each invocation. + /// Public for GPU-oracle tests. + #[allow(clippy::too_many_arguments)] pub fn launch_rl_trail_stop_check( &self, actions_d: &mut CudaSlice, @@ -2026,6 +2058,8 @@ impl IntegratedTrainer { unit_entry_price_d: &CudaSlice, unit_lots_d: &CudaSlice, unit_trail_distance_d: &CudaSlice, + pyramid_units_count_d: &CudaSlice, + close_unit_index_d: &mut CudaSlice, b_size: usize, ) -> Result<()> { debug_assert_eq!(actions_d.len(), b_size); @@ -2033,6 +2067,8 @@ impl IntegratedTrainer { debug_assert_eq!(unit_entry_price_d.len(), b_size * 4); debug_assert_eq!(unit_lots_d.len(), b_size * 4); debug_assert_eq!(unit_trail_distance_d.len(), b_size * 4); + debug_assert_eq!(pyramid_units_count_d.len(), b_size); + debug_assert_eq!(close_unit_index_d.len(), b_size); let cfg = LaunchConfig { grid_dim: (b_size as u32, 1, 1), block_dim: (4, 1, 1), @@ -2048,6 +2084,8 @@ impl IntegratedTrainer { .arg(unit_entry_price_d) .arg(unit_lots_d) .arg(unit_trail_distance_d) + .arg(pyramid_units_count_d) + .arg(close_unit_index_d) .arg(&b_size_i); unsafe { launch @@ -2096,12 +2134,13 @@ impl IntegratedTrainer { Ok(()) } - /// SP20 P1: launch `rl_unit_state_update` — detects per-batch - /// position transitions vs `prev_pos_lots_d` and updates per-unit - /// state (entry_price, entry_step, lots, initial_r, trail_distance, - /// active, pyramid_units_count). Uses ISV slot 496 (K_INIT) from - /// `self.isv_d` to scale the bootstrap trail. Public for SP20 P14 - /// GPU-oracle tests. + /// Launch `rl_unit_state_update` — detects per-batch position + /// transitions vs `prev_pos_lots_d` and updates per-unit state + /// (entry_price, entry_step, lots, initial_r, trail_distance, + /// active, pyramid_units_count). Handles pyramid add (allocate next + /// slot) and partial flat (deactivate oldest slot). Uses ISV slot + /// 496 (K_INIT) from `self.isv_d` to scale the bootstrap trail. + /// Public for GPU-oracle tests. #[allow(clippy::too_many_arguments)] pub fn launch_rl_unit_state_update( &self, @@ -3589,6 +3628,8 @@ impl IntegratedTrainer { .arg(&self.unit_entry_price_d) .arg(&self.unit_lots_d) .arg(&self.unit_trail_distance_d) + .arg(&self.pyramid_units_count_d) + .arg(&mut self.close_unit_index_d) .arg(&b_size_i); unsafe { launch @@ -3627,7 +3668,8 @@ impl IntegratedTrainer { } { - let (pos_d_ref, market_targets_d) = lobsim.pos_and_market_targets_mut(); + let (pos_d_ref, bid_px_d, ask_px_d, market_targets_d) = + lobsim.pos_book_and_market_targets_mut(); let cfg = LaunchConfig { grid_dim: (((b_size as u32) + 31) / 32, 1, 1), block_dim: (32, 1, 1), @@ -3639,6 +3681,14 @@ impl IntegratedTrainer { .arg(&self.actions_d) .arg(pos_d_ref) .arg(market_targets_d) + .arg(&self.isv_d) + .arg(bid_px_d) + .arg(ask_px_d) + .arg(&self.unit_entry_price_d) + .arg(&self.unit_active_d) + .arg(&self.unit_lots_d) + .arg(&self.pyramid_units_count_d) + .arg(&self.close_unit_index_d) .arg(&b_size_i) .arg(&pos_bytes_i); unsafe { @@ -3838,6 +3888,31 @@ impl IntegratedTrainer { } } + // Signed outcome EMA (ISV[508]) — updated on trade close only. + // Feeds anti-martingale sizing in actions_to_market_targets. + { + let cfg = LaunchConfig { + grid_dim: (1, 1, 1), + block_dim: (b_size as u32, 1, 1), + shared_mem_bytes: (2 * b_size * std::mem::size_of::()) as u32, + }; + let slot_i = crate::rl::isv_slots::RL_OUTCOME_EMA_INDEX as i32; + let alpha = RL_LR_CONTROLLER_ALPHA; + let mut launch = self.stream.launch_builder(&self.ema_update_on_done_fn); + launch + .arg(&self.isv_d) + .arg(&slot_i) + .arg(&alpha) + .arg(&self.rewards_d) + .arg(&self.dones_d) + .arg(&b_size_i); + unsafe { + launch + .launch(cfg) + .context("ema_update_on_done(OUTCOME_EMA) launch")?; + } + } + // Fire all 7 RL controllers — each reads its ISV EMA-input // slot and Wiener-blends its output. Critical: this happens // BEFORE apply_reward_scale so the scale ISV[406] reflects diff --git a/crates/ml-alpha/tests/trade_management_kernels.rs b/crates/ml-alpha/tests/trade_management_kernels.rs index 4726eb354..abdae6048 100644 --- a/crates/ml-alpha/tests/trade_management_kernels.rs +++ b/crates/ml-alpha/tests/trade_management_kernels.rs @@ -40,7 +40,8 @@ use anyhow::Result; use cudarc::driver::{CudaSlice, CudaStream}; use ml_alpha::rl::isv_slots::{ - RL_TRAIL_ADJUST_RATE_INDEX, RL_TRAIL_K_INIT_INDEX, RL_TRAIL_MAX_INDEX, RL_TRAIL_MIN_INDEX, + RL_PYRAMID_THRESHOLD_INDEX, RL_TRAIL_ADJUST_RATE_INDEX, RL_TRAIL_K_INIT_INDEX, + RL_TRAIL_MAX_INDEX, RL_TRAIL_MIN_INDEX, }; use ml_alpha::trainer::integrated::{ read_slice_d_pub, read_slice_i32_d_pub, read_slice_u8_d_pub, write_slice_f32_d_pub, @@ -110,6 +111,37 @@ fn pos_buf(position_lots: i32, vwap_entry: f32) -> Vec { buf } +const BOOK_LEVELS: usize = 10; + +struct PyramidContext { + bid_px_d: CudaSlice, + ask_px_d: CudaSlice, + unit_entry_price_d: CudaSlice, + unit_active_d: CudaSlice, + unit_lots_d: CudaSlice, + pyramid_units_count_d: CudaSlice, + close_unit_index_d: CudaSlice, +} + +fn default_pyramid_ctx(stream: &Arc, b_size: usize) -> Result { + let bid_px_d = upload_f32(stream, &[100.0; BOOK_LEVELS])?; + let ask_px_d = upload_f32(stream, &[100.25; BOOK_LEVELS])?; + let unit_entry_price_d = stream.alloc_zeros::(b_size * MAX_UNITS)?; + let unit_active_d = stream.alloc_zeros::(b_size * MAX_UNITS)?; + let unit_lots_d = stream.alloc_zeros::(b_size * MAX_UNITS)?; + let pyramid_units_count_d = stream.alloc_zeros::(b_size)?; + let close_unit_index_d = upload_i32(stream, &vec![-1i32; b_size])?; + Ok(PyramidContext { + bid_px_d, + ask_px_d, + unit_entry_price_d, + unit_active_d, + unit_lots_d, + pyramid_units_count_d, + close_unit_index_d, + }) +} + /// Overwrite a single ISV slot host-side then push the whole isv_d /// to the device. Cheap because the trainer's `isv_d` is small /// (~500 floats). @@ -130,6 +162,7 @@ fn half_flat_long_emits_half_position_size() -> Result<()> { let stream = dev.cuda_stream()?.clone(); let pos_bytes = POS_BYTES; let b_size = 1; + let ctx = default_pyramid_ctx(&stream, b_size)?; // Long 4 lots, action a9 (HalfFlatLong): ceil(4/2) = 2 → sell 2. let actions_d = upload_i32(&stream, &[9])?; @@ -140,6 +173,13 @@ fn half_flat_long_emits_half_position_size() -> Result<()> { &actions_d, &pos_state_d, &mut market_targets_d, + &ctx.bid_px_d, + &ctx.ask_px_d, + &ctx.unit_entry_price_d, + &ctx.unit_active_d, + &ctx.unit_lots_d, + &ctx.pyramid_units_count_d, + &ctx.close_unit_index_d, b_size, pos_bytes, )?; @@ -154,6 +194,13 @@ fn half_flat_long_emits_half_position_size() -> Result<()> { &actions_d, &pos_short_d, &mut market_targets_d, + &ctx.bid_px_d, + &ctx.ask_px_d, + &ctx.unit_entry_price_d, + &ctx.unit_active_d, + &ctx.unit_lots_d, + &ctx.pyramid_units_count_d, + &ctx.close_unit_index_d, b_size, pos_bytes, )?; @@ -167,6 +214,13 @@ fn half_flat_long_emits_half_position_size() -> Result<()> { &actions_d, &pos_odd_d, &mut market_targets_d, + &ctx.bid_px_d, + &ctx.ask_px_d, + &ctx.unit_entry_price_d, + &ctx.unit_active_d, + &ctx.unit_lots_d, + &ctx.pyramid_units_count_d, + &ctx.close_unit_index_d, b_size, pos_bytes, )?; @@ -184,6 +238,7 @@ fn half_flat_short_emits_half_position_size() -> Result<()> { let stream = dev.cuda_stream()?.clone(); let pos_bytes = POS_BYTES; let b_size = 1; + let ctx = default_pyramid_ctx(&stream, b_size)?; // Short 4 lots, action a10 (HalfFlatShort): ceil(4/2) = 2 → buy 2. let actions_d = upload_i32(&stream, &[10])?; @@ -194,6 +249,13 @@ fn half_flat_short_emits_half_position_size() -> Result<()> { &actions_d, &pos_state_d, &mut market_targets_d, + &ctx.bid_px_d, + &ctx.ask_px_d, + &ctx.unit_entry_price_d, + &ctx.unit_active_d, + &ctx.unit_lots_d, + &ctx.pyramid_units_count_d, + &ctx.close_unit_index_d, b_size, pos_bytes, )?; @@ -207,6 +269,13 @@ fn half_flat_short_emits_half_position_size() -> Result<()> { &actions_d, &pos_long_d, &mut market_targets_d, + &ctx.bid_px_d, + &ctx.ask_px_d, + &ctx.unit_entry_price_d, + &ctx.unit_active_d, + &ctx.unit_lots_d, + &ctx.pyramid_units_count_d, + &ctx.close_unit_index_d, b_size, pos_bytes, )?; @@ -397,6 +466,8 @@ fn trail_stop_check_overrides_action_on_breach() -> Result<()> { let unit_entry_d = upload_f32(&stream, &[100.0, 0.0, 0.0, 0.0])?; let unit_lots_d = upload_i32(&stream, &[2, 0, 0, 0])?; let unit_trail_d = upload_f32(&stream, &[0.5, 0.0, 0.0, 0.0])?; + let pyramid_count_d = upload_i32(&stream, &[1])?; + let mut close_idx_d = upload_i32(&stream, &[-1])?; // Shared book (single market). bid_px[0]=98.99, ask_px[0]=99.01 → // mid = 99.0. Both arrays are BOOK_LEVELS long; only slot 0 read. @@ -415,6 +486,8 @@ fn trail_stop_check_overrides_action_on_breach() -> Result<()> { &unit_entry_d, &unit_lots_d, &unit_trail_d, + &pyramid_count_d, + &mut close_idx_d, b_size, )?; let actions = read_slice_i32_d_pub(&stream, &actions_d, b_size)?; @@ -442,6 +515,8 @@ fn trail_stop_check_overrides_action_on_breach() -> Result<()> { &unit_entry_d, &unit_lots_d, &unit_trail_d, + &pyramid_count_d, + &mut close_idx_d, b_size, )?; let actions = read_slice_i32_d_pub(&stream, &actions_d2, b_size)?; @@ -470,6 +545,8 @@ fn trail_stop_check_overrides_action_on_breach() -> Result<()> { &unit_entry_d, &unit_lots_short_d, &unit_trail_d, + &pyramid_count_d, + &mut close_idx_d, b_size, )?; let actions = read_slice_i32_d_pub(&stream, &actions_d3, b_size)?; @@ -545,3 +622,249 @@ fn position_heat_cap_overrides_on_breach() -> Result<()> { eprintln!("PASS — heat cap overrides on breach + leaves alone within cap"); Ok(()) } + +#[test] +#[ignore = "requires CUDA (MlDevice::cuda(0))"] +fn pyramid_add_populates_next_unit_slot() -> Result<()> { + let Some((dev, mut trainer)) = build_trainer() else { return Ok(()) }; + let stream = dev.cuda_stream()?.clone(); + let pos_bytes = POS_BYTES; + let b_size = 1; + + set_isv_slot(&mut trainer, &stream, RL_TRAIL_K_INIT_INDEX, 2.0)?; + set_isv_slot(&mut trainer, &stream, RL_PYRAMID_THRESHOLD_INDEX, 0.50)?; + + let mut prev_pos_d = upload_i32(&stream, &[0])?; + let mut unit_entry_price_d = stream.alloc_zeros::(b_size * MAX_UNITS)?; + let mut unit_entry_step_d = stream.alloc_zeros::(b_size * MAX_UNITS)?; + let mut unit_lots_d = stream.alloc_zeros::(b_size * MAX_UNITS)?; + let mut unit_initial_r_d = stream.alloc_zeros::(b_size * MAX_UNITS)?; + let mut unit_trail_d = stream.alloc_zeros::(b_size * MAX_UNITS)?; + let mut unit_active_d = stream.alloc_zeros::(b_size * MAX_UNITS)?; + let mut pyramid_count_d = stream.alloc_zeros::(b_size)?; + + // ── Step 0: OPEN — flat → long 1 @ 100.0. Slot 0 active. + let pos0_d = upload_u8(&stream, &pos_buf(1, 100.0))?; + trainer.launch_rl_unit_state_update( + &pos0_d, + &mut prev_pos_d, + &mut unit_entry_price_d, + &mut unit_entry_step_d, + &mut unit_lots_d, + &mut unit_initial_r_d, + &mut unit_trail_d, + &mut unit_active_d, + &mut pyramid_count_d, + b_size, + pos_bytes, + 0, + )?; + let pyramid = read_slice_i32_d_pub(&stream, &pyramid_count_d, b_size)?; + assert_eq!(pyramid[0], 1, "after OPEN: pyramid_count == 1"); + + // ── Step 1: Pyramid ADD — long 1 → long 2 @ 101.0 (price moved). + // |curr|=2 > |prev|=1 → allocates slot 1. + let pos1_d = upload_u8(&stream, &pos_buf(2, 101.0))?; + trainer.launch_rl_unit_state_update( + &pos1_d, + &mut prev_pos_d, + &mut unit_entry_price_d, + &mut unit_entry_step_d, + &mut unit_lots_d, + &mut unit_initial_r_d, + &mut unit_trail_d, + &mut unit_active_d, + &mut pyramid_count_d, + b_size, + pos_bytes, + 1, + )?; + + let active = read_slice_u8_d_pub(&stream, &unit_active_d, b_size * MAX_UNITS)?; + let lots = read_slice_i32_d_pub(&stream, &unit_lots_d, b_size * MAX_UNITS)?; + let entry_px = read_slice_d_pub(&stream, &unit_entry_price_d, b_size * MAX_UNITS)?; + let pyramid = read_slice_i32_d_pub(&stream, &pyramid_count_d, b_size)?; + + assert_eq!(active[0], 1, "slot 0 still active"); + assert_eq!(active[1], 1, "slot 1 activated by pyramid add"); + assert_eq!(lots[0], 1, "slot 0 retains original lots"); + assert_eq!(lots[1], 1, "slot 1 gets delta lots (2-1=1)"); + assert!((entry_px[0] - 100.0).abs() < 1e-6, "slot 0 entry unchanged"); + assert!((entry_px[1] - 101.0).abs() < 1e-6, "slot 1 entry = new vwap"); + assert_eq!(pyramid[0], 2, "pyramid_count == 2 after add"); + + // ── Step 2: Partial FLAT — long 2 → long 1 @ 101.0. + // |curr|=1 < |prev|=2 → deactivates oldest active (slot 0). + let pos2_d = upload_u8(&stream, &pos_buf(1, 101.0))?; + trainer.launch_rl_unit_state_update( + &pos2_d, + &mut prev_pos_d, + &mut unit_entry_price_d, + &mut unit_entry_step_d, + &mut unit_lots_d, + &mut unit_initial_r_d, + &mut unit_trail_d, + &mut unit_active_d, + &mut pyramid_count_d, + b_size, + pos_bytes, + 2, + )?; + + let active = read_slice_u8_d_pub(&stream, &unit_active_d, b_size * MAX_UNITS)?; + let lots = read_slice_i32_d_pub(&stream, &unit_lots_d, b_size * MAX_UNITS)?; + let pyramid = read_slice_i32_d_pub(&stream, &pyramid_count_d, b_size)?; + + assert_eq!(active[0], 0, "partial flat deactivated oldest (slot 0)"); + assert_eq!(lots[0], 0, "slot 0 lots zeroed"); + assert_eq!(active[1], 1, "slot 1 remains active"); + assert_eq!(pyramid[0], 1, "pyramid_count == 1 after partial flat"); + + // ── Step 3: Full CLOSE — long 1 → flat 0. + let pos3_d = upload_u8(&stream, &pos_buf(0, 0.0))?; + trainer.launch_rl_unit_state_update( + &pos3_d, + &mut prev_pos_d, + &mut unit_entry_price_d, + &mut unit_entry_step_d, + &mut unit_lots_d, + &mut unit_initial_r_d, + &mut unit_trail_d, + &mut unit_active_d, + &mut pyramid_count_d, + b_size, + pos_bytes, + 3, + )?; + + let active = read_slice_u8_d_pub(&stream, &unit_active_d, b_size * MAX_UNITS)?; + let pyramid = read_slice_i32_d_pub(&stream, &pyramid_count_d, b_size)?; + for u in 0..MAX_UNITS { + assert_eq!(active[u], 0, "CLOSE deactivates all; slot {u} still active"); + } + assert_eq!(pyramid[0], 0, "pyramid_count == 0 after full close"); + + eprintln!("PASS — pyramid add/partial-flat/close unit slot lifecycle"); + Ok(()) +} + +#[test] +#[ignore = "requires CUDA (MlDevice::cuda(0))"] +fn pyramid_threshold_gates_add_in_actions_kernel() -> Result<()> { + let Some((dev, mut trainer)) = build_trainer() else { return Ok(()) }; + let stream = dev.cuda_stream()?.clone(); + let pos_bytes = POS_BYTES; + let b_size = 1; + + set_isv_slot(&mut trainer, &stream, RL_PYRAMID_THRESHOLD_INDEX, 0.50)?; + + // Setup: already long 1 lot, 1 unit active at entry 100.0. + let unit_entry_price_d = upload_f32(&stream, &[100.0, 0.0, 0.0, 0.0])?; + let unit_active_d = upload_u8(&stream, &[1, 0, 0, 0])?; + let unit_lots_d = upload_i32(&stream, &[1, 0, 0, 0])?; + let pyramid_count_d = upload_i32(&stream, &[1])?; + let close_idx_d = upload_i32(&stream, &[-1])?; + let pos_long_d = upload_u8(&stream, &pos_buf(1, 100.0))?; + + // Case 1: mid = 100.625 (threshold 0.50 met: 100.625 - 100.0 = 0.625). + // Action a5 (LongSmall) should emit an order (side=0, size=1). + let mut bid = vec![0.0_f32; BOOK_LEVELS]; + let mut ask = vec![0.0_f32; BOOK_LEVELS]; + bid[0] = 100.50; + ask[0] = 100.75; + let bid_d = upload_f32(&stream, &bid)?; + let ask_d = upload_f32(&stream, &ask)?; + + let actions_d = upload_i32(&stream, &[5])?; + let mut market_targets_d = stream.alloc_zeros::(b_size * 2)?; + + trainer.launch_actions_to_market_targets( + &actions_d, + &pos_long_d, + &mut market_targets_d, + &bid_d, + &ask_d, + &unit_entry_price_d, + &unit_active_d, + &unit_lots_d, + &pyramid_count_d, + &close_idx_d, + b_size, + pos_bytes, + )?; + let mt = read_slice_i32_d_pub(&stream, &market_targets_d, b_size * 2)?; + assert_eq!(mt[0], 0, "threshold met, count &CudaSlice { self.ask_px_d() } + + fn pos_book_and_market_targets_mut( + &mut self, + ) -> (&CudaSlice, &CudaSlice, &CudaSlice, &mut CudaSlice) { + (&self.pos_d, &self.bid_px_d, &self.ask_px_d, &mut self.market_targets_d) + } } // Re-open the impl block for `LobSimCuda` so subsequent methods (if any