gitadmin/foxhunt
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(cuda): bidirectional HER kernels — track, inject, forward

Browse Source 
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 <noreply@anthropic.com>
This commit is contained in:
jgrusewski
2026-05-26 08:41:05 +02:00
parent 95af0db54f
commit 029f2956f1
3 changed files with 390 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

161
crates/ml-alpha/cuda/rl_hindsight_forward.cu Normal file
View File

@@ -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;
}

153
crates/ml-alpha/cuda/rl_hindsight_inject.cu Normal file
View File

@@ -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];
}
}

76
crates/ml-alpha/cuda/rl_hindsight_track.cu Normal file
View File

@@ -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;
}