From 029f2956f1e90a010ef4432a43de705e595521a3 Mon Sep 17 00:00:00 2001 From: jgrusewski Date: Tue, 26 May 2026 08:41:05 +0200 Subject: [PATCH] =?UTF-8?q?feat(cuda):=20bidirectional=20HER=20kernels=20?= =?UTF-8?q?=E2=80=94=20track,=20inject,=20forward?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit rl_hindsight_track: per-step mid accumulation + peak tracking. rl_hindsight_inject: backward HER on done, prefix-sum replay write. rl_hindsight_forward: forward HER after lookahead, prefix-sum write. No atomicAdd. Co-Authored-By: Claude Opus 4.7 --- crates/ml-alpha/cuda/rl_hindsight_forward.cu | 161 +++++++++++++++++++ crates/ml-alpha/cuda/rl_hindsight_inject.cu | 153 ++++++++++++++++++ crates/ml-alpha/cuda/rl_hindsight_track.cu | 76 +++++++++ 3 files changed, 390 insertions(+) create mode 100644 crates/ml-alpha/cuda/rl_hindsight_forward.cu create mode 100644 crates/ml-alpha/cuda/rl_hindsight_inject.cu create mode 100644 crates/ml-alpha/cuda/rl_hindsight_track.cu diff --git a/crates/ml-alpha/cuda/rl_hindsight_forward.cu b/crates/ml-alpha/cuda/rl_hindsight_forward.cu new file mode 100644 index 000000000..b839f3aa6 --- /dev/null +++ b/crates/ml-alpha/cuda/rl_hindsight_forward.cu @@ -0,0 +1,161 @@ +/** + * rl_hindsight_forward.cu — Forward HER: post-exit continuation evaluation + * + * Bidirectional HER Phase 3: after a trade closes, monitors whether the + * market continued in the position's direction. If forward_pnl >> actual_pnl + * after a lookahead window, injects a synthetic "should-have-held-longer" + * transition into the replay buffer. + * + * Operates on the closed-trades ring buffer (CLOSED_RING_SIZE entries), + * populated by the host/experience kernel when trades close. + * + * Coordination: Grid=(1), Block=(CLOSED_RING_SIZE=256). Single-block + * prefix-sum assigns contiguous replay write slots. + * + * ISV slots read: + * 549 — HINDSIGHT_THRESHOLD (forward_pnl / actual_pnl threshold) + * 550 — HINDSIGHT_PRIORITY_BOOST (multiplier on max_priority) + * 552 — HINDSIGHT_FORWARD_LOOKAHEAD (steps to wait before evaluating) + * 548 — STEP_COUNTER (current training step) + * 406 — REWARD_SCALE (for scaled reward computation) + * + * No atomicAdd — block prefix-sum only. + */ + +#define CLOSED_RING_SIZE 256 +#define CLOSED_RING_FIELDS 5 /* entry_price, exit_step, direction, actual_pnl, batch_idx */ +#define HIDDEN_DIM 128 +#define SCALARS_PER_TRANSITION 7 + +/* ISV slot indices. */ +#define RL_HINDSIGHT_THRESHOLD_INDEX 549 +#define RL_HINDSIGHT_PRIORITY_BOOST_INDEX 550 +#define RL_HINDSIGHT_FORWARD_LOOKAHEAD_INDEX 552 +#define RL_STEP_COUNTER_ISV_INDEX 548 +#define RL_REWARD_SCALE_INDEX 406 + +extern "C" __global__ void rl_hindsight_forward( + const float* __restrict__ bid_px, /* [BOOK_LEVELS] */ + const float* __restrict__ ask_px, /* [BOOK_LEVELS] */ + const float* __restrict__ isv, + /* Closed trades ring */ + float* __restrict__ closed_ring, /* [CLOSED_RING_SIZE * 5] */ + const float* __restrict__ closed_h_t, /* [CLOSED_RING_SIZE * 128] */ + const unsigned int* __restrict__ closed_step, /* [CLOSED_RING_SIZE] */ + /* Replay write targets */ + float* __restrict__ replay_h_t, + float* __restrict__ replay_h_tp1, + float* __restrict__ replay_scalars, + float* __restrict__ priority_tree, + unsigned int* __restrict__ write_head, + unsigned int* __restrict__ replay_len, + float* __restrict__ max_priority, + int capacity +) +{ + int i = threadIdx.x; + if (i >= CLOSED_RING_SIZE) return; + + /* ── Phase 1: Evaluate each closed-trade entry ─────────────── */ + int wants_inject = 0; + float forward_pnl = 0.0f; + float actual_pnl = 0.0f; + float direction = 0.0f; + float entry_price = 0.0f; + int should_clear = 0; + + entry_price = closed_ring[i * CLOSED_RING_FIELDS + 0]; + + if (entry_price != 0.0f) { + int current_step = (int)isv[RL_STEP_COUNTER_ISV_INDEX]; + int age = current_step - (int)closed_step[i]; + int lookahead = (int)isv[RL_HINDSIGHT_FORWARD_LOOKAHEAD_INDEX]; + + if (age < lookahead) { + /* Too early to evaluate — skip. */ + } else if (age > 2 * lookahead) { + /* Expired — clear the slot. */ + should_clear = 1; + } else { + /* Evaluate forward PnL. */ + direction = closed_ring[i * CLOSED_RING_FIELDS + 2]; + actual_pnl = closed_ring[i * CLOSED_RING_FIELDS + 3]; + + float mid = 0.5f * (bid_px[0] + ask_px[0]); + forward_pnl = direction * (mid - entry_price); + + float threshold = isv[RL_HINDSIGHT_THRESHOLD_INDEX]; + + if (forward_pnl > 0.0f && forward_pnl > actual_pnl * threshold) { + wants_inject = 1; + } + + /* Mark for clearing after evaluation (whether or not we inject). */ + should_clear = 1; + } + } + + /* ── Phase 2: Prefix-sum for contiguous write slots ──────── */ + __shared__ unsigned int smem[CLOSED_RING_SIZE + 2]; + + smem[i] = (unsigned int)wants_inject; + __syncthreads(); + + if (i == 0) { + unsigned int running = 0; + for (int j = 0; j < CLOSED_RING_SIZE; j++) { + unsigned int val = smem[j]; + smem[j] = running; + running += val; + } + smem[CLOSED_RING_SIZE] = running; /* total count */ + smem[CLOSED_RING_SIZE + 1] = write_head[0]; /* base offset */ + + /* Advance write head and replay length. */ + if (running > 0) { + write_head[0] = (write_head[0] + running) % (unsigned int)capacity; + unsigned int new_len = replay_len[0] + running; + if (new_len > (unsigned int)capacity) new_len = (unsigned int)capacity; + replay_len[0] = new_len; + } + } + __syncthreads(); + + /* ── Phase 3: Clear expired / evaluated slots ────────────── */ + if (should_clear) { + closed_ring[i * CLOSED_RING_FIELDS + 0] = 0.0f; + } + + /* ── Phase 4: Write synthetic transitions ────────────────── */ + if (!wants_inject) return; + + unsigned int my_offset = smem[i]; + unsigned int base_head = smem[CLOSED_RING_SIZE + 1]; + unsigned int slot = (base_head + my_offset) % (unsigned int)capacity; + + float reward_scale = isv[RL_REWARD_SCALE_INDEX]; + + /* h_t: copy the hidden state captured at trade close. */ + for (int d = 0; d < HIDDEN_DIM; d++) { + float h = closed_h_t[(long long)i * HIDDEN_DIM + d]; + replay_h_t[(long long)slot * HIDDEN_DIM + d] = h; + /* h_tp1 = self-loop (synthetic terminal). */ + replay_h_tp1[(long long)slot * HIDDEN_DIM + d] = h; + } + + /* Scalars: [action, reward, raw_reward, done, log_pi, n_step_gamma, return] */ + long long sbase = (long long)slot * SCALARS_PER_TRANSITION; + replay_scalars[sbase + 0] = 2.0f; /* Hold */ + replay_scalars[sbase + 1] = forward_pnl * reward_scale; /* reward (scaled) */ + replay_scalars[sbase + 2] = forward_pnl; /* raw_reward */ + replay_scalars[sbase + 3] = 1.0f; /* done */ + replay_scalars[sbase + 4] = -1.0f; /* log_pi (synthetic) */ + replay_scalars[sbase + 5] = 0.0f; /* n_step_gamma (terminal) */ + replay_scalars[sbase + 6] = forward_pnl; /* return */ + + /* Priority: boosted. Tree leaves at [cap..2*cap). */ + float boost = isv[RL_HINDSIGHT_PRIORITY_BOOST_INDEX]; + float prio = boost * max_priority[0]; + if (prio < 1e-6f) prio = 1e-6f; + priority_tree[capacity + (int)slot] = prio; +} diff --git a/crates/ml-alpha/cuda/rl_hindsight_inject.cu b/crates/ml-alpha/cuda/rl_hindsight_inject.cu new file mode 100644 index 000000000..03904a9d7 --- /dev/null +++ b/crates/ml-alpha/cuda/rl_hindsight_inject.cu @@ -0,0 +1,153 @@ +/** + * rl_hindsight_inject.cu — Backward HER: synthetic replay push on trade close + * + * Bidirectional HER Phase 2: when a trade closes (dones[b] > 0.5), checks + * whether peak_pnl >> actual_pnl (controlled by ISV threshold). If the trade + * left significant money on the table, injects a synthetic "what-if-I-held- + * to-peak" transition into the replay buffer at boosted priority. + * + * Coordination: Grid=(1), Block=(B). Single-block prefix-sum assigns + * contiguous write slots — same pattern as rl_per_push_flush. + * + * ISV slots read: + * 549 — HINDSIGHT_THRESHOLD (peak_pnl / actual_pnl threshold) + * 550 — HINDSIGHT_PRIORITY_BOOST (multiplier on max_priority for synthetic) + * 551 — HINDSIGHT_INJECT_COUNT (written: how many synthetics injected) + * 406 — REWARD_SCALE (to recover raw PnL from scaled reward) + * + * No atomicAdd — block tree-reduce / prefix-sum only. + */ + +#define HIDDEN_DIM 128 +#define SCALARS_PER_TRANSITION 7 + +/* ISV slot indices (literals — canonical names in Rust ISV slots module). */ +#define RL_HINDSIGHT_THRESHOLD_INDEX 549 +#define RL_HINDSIGHT_PRIORITY_BOOST_INDEX 550 +#define RL_HINDSIGHT_INJECT_COUNT_INDEX 551 +#define RL_REWARD_SCALE_INDEX 406 + +extern "C" __global__ void rl_hindsight_inject( + const float* __restrict__ dones, /* [B] */ + const float* __restrict__ rewards_raw, /* [B] */ + const float* __restrict__ peak_mid, /* [B] */ + const float* __restrict__ entry_mid, /* [B] */ + const int* __restrict__ position_dir, /* [B] */ + const float* __restrict__ h_t_current, /* [B * 128] */ + const float* __restrict__ isv, + /* Replay write targets */ + float* __restrict__ replay_h_t, /* [cap * 128] */ + float* __restrict__ replay_h_tp1, /* [cap * 128] */ + float* __restrict__ replay_scalars, /* [cap * 7] */ + float* __restrict__ priority_tree, /* [2 * cap] */ + unsigned int* __restrict__ write_head, /* [1] */ + unsigned int* __restrict__ replay_len, /* [1] */ + float* __restrict__ max_priority, /* [1] */ + /* Reset outputs */ + unsigned int* __restrict__ ring_write_idx, /* [B] — reset on done */ + float* __restrict__ peak_mid_out, /* [B] — reset on done */ + int b_size, + int capacity +) +{ + int b = threadIdx.x; + if (b >= b_size) return; + + /* ── Phase 1: Determine which threads want to inject ────────── */ + int wants_inject = 0; + + if (dones[b] > 0.5f) { + float reward_scale = isv[RL_REWARD_SCALE_INDEX]; + float actual_pnl = fabsf(rewards_raw[b]) / fmaxf(reward_scale, 1e-9f); + + int dir = position_dir[b]; + float peak_pnl = (float)dir * (peak_mid[b] - entry_mid[b]); + + float threshold = isv[RL_HINDSIGHT_THRESHOLD_INDEX]; + + /* Peak PnL must exceed actual by the threshold factor AND be positive. */ + if (peak_pnl > 0.0f && peak_pnl > actual_pnl * threshold) { + wants_inject = 1; + } + + /* Reset tracking state for this env regardless of injection. */ + ring_write_idx[b] = 0; + peak_mid_out[b] = 0.0f; + } + + /* ── Phase 2: Prefix-sum to assign contiguous write slots ──── */ + /* Shared memory for the prefix-sum scan (single block, B threads). */ + extern __shared__ unsigned int smem[]; + /* smem[0..b_size): prefix-sum workspace + * smem[b_size]: total inject count + * smem[b_size+1]: base write offset (old write_head value) */ + + smem[b] = (unsigned int)wants_inject; + __syncthreads(); + + /* Thread 0 performs exclusive prefix-sum and advances write_head. */ + if (b == 0) { + unsigned int running = 0; + for (int i = 0; i < b_size; i++) { + unsigned int val = smem[i]; + smem[i] = running; /* exclusive prefix */ + running += val; + } + smem[b_size] = running; /* total inject count */ + + /* Advance write_head (circular) */ + unsigned int old_head = write_head[0]; + smem[b_size + 1] = old_head; + write_head[0] = (old_head + running) % (unsigned int)capacity; + + /* Update replay_len (capped at capacity). */ + unsigned int new_len = replay_len[0] + running; + if (new_len > (unsigned int)capacity) new_len = (unsigned int)capacity; + replay_len[0] = new_len; + } + __syncthreads(); + + /* ── Phase 3: Injecting threads write their synthetic transitions ── */ + if (!wants_inject) return; + + unsigned int my_offset = smem[b]; /* exclusive prefix for this thread */ + unsigned int base_head = smem[b_size + 1]; /* old write_head */ + unsigned int slot = (base_head + my_offset) % (unsigned int)capacity; + + /* Recover raw peak PnL for the synthetic reward. */ + int dir = position_dir[b]; + float peak_pnl = (float)dir * (peak_mid[b] - entry_mid[b]); + float reward_scale = isv[RL_REWARD_SCALE_INDEX]; + + /* h_t: copy current hidden state (the "what if" starts from same state). */ + for (int d = 0; d < HIDDEN_DIM; d++) { + float h = h_t_current[(long long)b * HIDDEN_DIM + d]; + replay_h_t[(long long)slot * HIDDEN_DIM + d] = h; + /* h_tp1 = self-loop (synthetic terminal transition). */ + replay_h_tp1[(long long)slot * HIDDEN_DIM + d] = h; + } + + /* Scalars layout: [action, reward, raw_reward, done, log_pi, n_step_gamma, return] + * [ 0 , 1 , 2 , 3 , 4 , 5 , 6 ] */ + long long sbase = (long long)slot * SCALARS_PER_TRANSITION; + replay_scalars[sbase + 0] = 2.0f; /* action = Hold (2) */ + replay_scalars[sbase + 1] = peak_pnl * reward_scale; /* reward (scaled) */ + replay_scalars[sbase + 2] = peak_pnl; /* raw_reward */ + replay_scalars[sbase + 3] = 1.0f; /* done = terminal */ + replay_scalars[sbase + 4] = -1.0f; /* log_pi (synthetic) */ + replay_scalars[sbase + 5] = 0.0f; /* n_step_gamma = 0 (terminal) */ + replay_scalars[sbase + 6] = peak_pnl; /* return = peak PnL */ + + /* Priority: boosted fraction of current max_priority. + * Tree layout: leaves at [cap..2*cap), internal nodes at [0..cap). */ + float boost = isv[RL_HINDSIGHT_PRIORITY_BOOST_INDEX]; + float prio = boost * max_priority[0]; + if (prio < 1e-6f) prio = 1e-6f; + priority_tree[capacity + (int)slot] = prio; + + /* Thread 0 writes the inject count to ISV for diagnostics. */ + if (b == 0) { + /* Cast away const for the diagnostic write — ISV is writable from GPU. */ + ((float*)isv)[RL_HINDSIGHT_INJECT_COUNT_INDEX] = (float)smem[b_size]; + } +} diff --git a/crates/ml-alpha/cuda/rl_hindsight_track.cu b/crates/ml-alpha/cuda/rl_hindsight_track.cu new file mode 100644 index 000000000..a2be56c06 --- /dev/null +++ b/crates/ml-alpha/cuda/rl_hindsight_track.cu @@ -0,0 +1,76 @@ +/** + * rl_hindsight_track.cu — Per-step mid-price accumulation + peak tracking + * + * Bidirectional HER Phase 1: tracks mid-price during open positions for + * backward hindsight (peak >> actual detection on trade close). + * + * Per thread b (one per environment): + * - If flat (lots == 0): reset ring write index and peak, early return + * - Otherwise: record mid-price in ring buffer, update peak (max for + * long, min for short), advance write index + * + * Launch: Grid=(ceil(B/32)), Block=(32). + */ + +#define RING_LEN 512 + +extern "C" __global__ void rl_hindsight_track( + const unsigned char* __restrict__ pos_d, /* [B * pos_bytes] */ + const float* __restrict__ bid_px, /* [BOOK_LEVELS] */ + const float* __restrict__ ask_px, /* [BOOK_LEVELS] */ + float* __restrict__ mid_ring, /* [B * RING_LEN] */ + unsigned int* __restrict__ ring_write_idx, /* [B] */ + float* __restrict__ peak_mid, /* [B] */ + float* __restrict__ entry_mid, /* [B] */ + int* __restrict__ position_dir, /* [B] */ + int b_size, + int pos_bytes +) +{ + int b = blockIdx.x * blockDim.x + threadIdx.x; + if (b >= b_size) return; + + /* Read position lots from pos_d (offset 0, i32 little-endian). */ + int lots; + { + const unsigned char* base = pos_d + (long long)b * pos_bytes; + lots = *(const int*)base; + } + + /* Flat position: reset tracking state and return. */ + if (lots == 0) { + ring_write_idx[b] = 0; + peak_mid[b] = 0.0f; + return; + } + + /* Compute mid-price from top-of-book. */ + float mid = 0.5f * (bid_px[0] + ask_px[0]); + + /* First observation after position open: bootstrap entry and direction. */ + unsigned int widx = ring_write_idx[b]; + if (widx == 0) { + entry_mid[b] = mid; + position_dir[b] = (lots > 0) ? 1 : -1; + } + + /* Store mid in ring buffer (saturate at RING_LEN-1 to prevent OOB). */ + unsigned int store_idx = (widx < RING_LEN) ? widx : (RING_LEN - 1); + mid_ring[(long long)b * RING_LEN + store_idx] = mid; + + /* Update peak: max for long, min for short. */ + int dir = position_dir[b]; + if (dir > 0) { + peak_mid[b] = fmaxf(peak_mid[b], mid); + } else { + /* For short: on first step (widx==0) init peak to mid so fminf works. */ + if (widx == 0) { + peak_mid[b] = mid; + } else { + peak_mid[b] = fminf(peak_mid[b], mid); + } + } + + /* Advance write index. */ + ring_write_idx[b] = widx + 1; +}