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

diag(policy-quality): finish Tasks 0.3 + 0.6 partial wirings

Browse Source 
Task 0.3 (Track 1 magnitude):
  * q_mag_full/half/quarter: plumbed via GpuDqnTrainer::q_magnitude_bucket_means()
    → FusedTrainingCtx::q_magnitude_bucket_means() → HEALTH_DIAG. Phase-0
    returns [0.0; 3]; per-mag Q reduction kernel deferred per plan §0.3 step 2
    (goal here is the accessor chain, swap-in without touching emit formatting).
  * var_scale_mean: STUB 0.0 — kernel computes `1/(1+sqrt(Var[Q]))` per-sample
    inside experience_env_step but does not persist to a device buffer; it is
    consumed in-place to shrink effective_max_pos. Exposing requires a
    dedicated per-sample output buffer + launch arg + kernel write; deferred
    to Phase 1+. Documented in code.
  * kelly_f_mean / avg_win_ratio: REAL — computed host-side from
    self.trade_stats_history.last() using the SAME formula the kernel uses
    (b = avg_win/avg_loss, kelly = (b·p − (1−p))/b clamped [0,1] × 0.5
    for half-Kelly; avg_win_ratio = (sum_wins/win_count) /
    max(sum_losses/loss_count, 0.001) per plan formula line 239).
    One-epoch lag (trade_stats_history is appended later in the same
    epoch body); zero until first epoch's stats land. Documented in code.

Task 0.6 (Track 1 noisy):
  * vsn_mag / vsn_dir: REAL per-branch — mean |w| across each branch's VSN
    projection weight pair (tensors 26..34 grouped by branch:
    26/27=dir, 28/29=mag, 30/31=order, 32/33=urg). The live VSN gate mask is
    computed per-sample inside variable_select_bottleneck and consumed
    immediately, not materialized into a device buffer we can cheaply read;
    projection-weight mean is the closest stable H7 surrogate. Documented
    in accessor rustdoc. Higher-fidelity measurement would require a
    per-sample mask output buffer following the Task 0.5 pattern — deferred.
  * drift_mag / drift_dir: REAL per-branch — RMS ‖target_w − online_w‖ across
    each branch's 4 weight tensors (indices 8..12 dir, 12..16 mag, 16..20
    order, 20..24 urg). Host-computed cold path via two memcpy_dtoh + normal
    Vec<f32> reduce loops; stream-sync'd before read. NO atomics (project rule).
    ~2.7 MB × 2 DtoH per epoch, negligible diagnostic cost.

NoisyNets sigma_mag/sigma_dir already wired in 999bb2fa0 — preserved untouched.

Accessor chain (mirrors Task 0.4 grad_ratio_mag_dir pattern):
  * GpuDqnTrainer::per_branch_target_drift() -> Result<[f32; 4], MLError>
  * GpuDqnTrainer::per_branch_vsn_mean()     -> Result<[f32; 4], MLError>
  * GpuDqnTrainer::q_magnitude_bucket_means()-> [f32; 3]  (plumbing stub)
  * FusedTrainingCtx wrappers → delegate, fallback to zeros on MLError.
  * training_loop.rs HEALTH_DIAG block: populate slots via
    self.fused_ctx.as_ref().map(|f| f.…).unwrap_or(…) — same pattern
    used for grad_ratio_mag_dir.

HEALTH_DIAG `mag` and `noisy` groups now carry real signal (or plan-sanctioned
plumbing stubs for the q_mag_* / var_scale slots documented in commit + code).
This commit is contained in:
jgrusewski
2026-04-21 23:33:35 +02:00
parent e1ac2c7578
commit aa16ca31f2
3 changed files with 259 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

140
crates/ml/src/cuda_pipeline/gpu_dqn_trainer.rs
View File

@@ -2167,6 +2167,146 @@ impl GpuDqnTrainer {
Ok(if dir > 1e-9 { mag / dir } else { 0.0 })
}
/// Task 0.6 — per-branch target/online parameter drift (H8 detection signal).
///
/// Returns `[direction, magnitude, order, urgency]` RMS deviation between
/// `target_params` and online `params` over each branch's 4 weight tensors
/// (fc weights + bias, out weights + bias — tensor indices 8..12 / 12..16 /
/// 16..20 / 20..24).
///
/// Host-side pageable DtoH of the relevant byte ranges (cold path, once per
/// epoch) — no atomics, matches project rule. Per-branch reduce is a tiny
/// sum-of-squares loop on normal Vec<f32>.
///
/// RMS = `sqrt(Σ(target - online)² / N)` so the value is scale-free w.r.t.
/// branch parameter count (direction branch is smaller than magnitude).
pub fn per_branch_target_drift(&self) -> Result<[f32; 4], MLError> {
let param_sizes = compute_param_sizes(&self.config);
// Full byte range covering branch tensors 8..24 (inclusive of branch 3's
// last tensor = index 23, so stop byte = padded_byte_offset(24)).
let start_byte = padded_byte_offset(&param_sizes, 8) as usize;
let end_byte = padded_byte_offset(&param_sizes, 24) as usize;
let start_idx = start_byte / std::mem::size_of::<f32>();
let end_idx = end_byte / std::mem::size_of::<f32>();
let len = end_idx.saturating_sub(start_idx);
if len == 0 {
return Ok([0.0_f32; 4]);
}
// Sync stream — ensure target EMA update + any online param writes landed.
self.stream.synchronize().map_err(|e| {
MLError::ModelError(format!("per_branch_target_drift sync: {e}"))
})?;
let mut h_online = vec![0.0_f32; len];
let mut h_target = vec![0.0_f32; len];
self.stream
.memcpy_dtoh(&self.params_buf.slice(start_idx..end_idx), &mut h_online)
.map_err(|e| MLError::ModelError(format!("per_branch_target_drift dtoh online: {e}")))?;
self.stream
.memcpy_dtoh(&self.target_params_buf.slice(start_idx..end_idx), &mut h_target)
.map_err(|e| MLError::ModelError(format!("per_branch_target_drift dtoh target: {e}")))?;
let mut drifts = [0.0_f32; 4];
for branch in 0..4_usize {
let first_tensor = 8 + branch * 4; // 8, 12, 16, 20
let last_tensor = first_tensor + 4; // exclusive
let mut sum_sq = 0.0_f64;
let mut n_params: usize = 0;
for tensor_idx in first_tensor..last_tensor {
let t_start_byte = padded_byte_offset(&param_sizes, tensor_idx) as usize;
let t_start_idx = t_start_byte / std::mem::size_of::<f32>();
let t_len = param_sizes[tensor_idx];
// Offset into our local slice (start_byte anchor).
let local_start = t_start_idx.saturating_sub(start_idx);
let local_end = (local_start + t_len).min(h_online.len());
for k in local_start..local_end {
let d = (h_target[k] - h_online[k]) as f64;
sum_sq += d * d;
n_params += 1;
}
}
drifts[branch] = if n_params > 0 {
(sum_sq / n_params as f64).sqrt() as f32
} else {
0.0
};
}
Ok(drifts)
}
/// Task 0.6 — per-branch Variable Selection Network activation magnitude
/// (H7 surrogate signal).
///
/// Returns `[direction, magnitude, order, urgency]` mean |w| across each
/// branch's VSN pair (tensors 26..34 grouped as `[w_vsn1_b, w_vsn2_b]` for
/// b in 0..4). The VSN gate mask is computed per-sample inside the
/// `variable_select_bottleneck` kernel and consumed immediately; it is
/// NOT materialized into a device buffer we can cheaply read. The
/// projection weights are the closest stable proxy for "how much VSN
/// bandwidth is allocated to this branch" — a non-trivial mean indicates
/// the bottleneck is actively shaping that branch's features.
///
/// If higher-fidelity measurement is needed, a dedicated per-sample mask
/// output buffer + host reduce would follow the Task 0.5 pattern — but
/// ~3 KB of VSN weights per branch is cheap and plan-sanctioned for the
/// Phase-0 plumbing pass.
pub fn per_branch_vsn_mean(&self) -> Result<[f32; 4], MLError> {
let param_sizes = compute_param_sizes(&self.config);
// Tensors 26..34 = VSN projection pairs (26/27=dir, 28/29=mag, 30/31=order, 32/33=urg).
let start_byte = padded_byte_offset(&param_sizes, 26) as usize;
let end_byte = padded_byte_offset(&param_sizes, 34) as usize;
let start_idx = start_byte / std::mem::size_of::<f32>();
let end_idx = end_byte / std::mem::size_of::<f32>();
let len = end_idx.saturating_sub(start_idx);
if len == 0 {
return Ok([0.0_f32; 4]);
}
self.stream.synchronize().map_err(|e| {
MLError::ModelError(format!("per_branch_vsn_mean sync: {e}"))
})?;
let mut h_w = vec![0.0_f32; len];
self.stream
.memcpy_dtoh(&self.params_buf.slice(start_idx..end_idx), &mut h_w)
.map_err(|e| MLError::ModelError(format!("per_branch_vsn_mean dtoh: {e}")))?;
let mut means = [0.0_f32; 4];
for branch in 0..4_usize {
let first_tensor = 26 + branch * 2; // 26, 28, 30, 32
let last_tensor = first_tensor + 2; // exclusive
let mut sum_abs = 0.0_f64;
let mut n_params: usize = 0;
for tensor_idx in first_tensor..last_tensor {
let t_start_byte = padded_byte_offset(&param_sizes, tensor_idx) as usize;
let t_start_idx = t_start_byte / std::mem::size_of::<f32>();
let t_len = param_sizes[tensor_idx];
let local_start = t_start_idx.saturating_sub(start_idx);
let local_end = (local_start + t_len).min(h_w.len());
for k in local_start..local_end {
sum_abs += (h_w[k] as f64).abs();
n_params += 1;
}
}
means[branch] = if n_params > 0 {
(sum_abs / n_params as f64) as f32
} else {
0.0
};
}
Ok(means)
}
/// Task 0.3 — per-magnitude-bucket Q-value mean (Full / Half / Quarter).
///
/// Plumbed for HEALTH_DIAG; per-magnitude Q-bucket reduction kernel is
/// deferred to Phase 1+ if measurement signals need (plan §0.3 step 2 —
/// "for Phase 0, this returns zeros. The actual kernel wiring is a later
/// task if measurement confirms it's needed. Goal here is plumbing.").
///
/// Returns `[q_full, q_half, q_quarter]` = the mean Q-value predicted for
/// the Full / Half / Quarter magnitude buckets respectively.
pub fn q_magnitude_bucket_means(&self) -> [f32; 3] {
[0.0_f32; 3]
}
/// Update per-branch liquid tau modulation — GPU kernel (RK4/Euler adaptive ODE).
///
/// Reads per_branch_q_gaps (pinned device-mapped) and qlstm_context from device.

33
crates/ml/src/trainers/dqn/fused_training.rs
View File

@@ -891,6 +891,39 @@ impl FusedTrainingCtx {
self.trainer.grad_ratio_mag_dir().unwrap_or(0.0)
}
/// Task 0.6 — per-branch target/online param drift (H8 detection signal).
/// Returns `[direction, magnitude, order, urgency]` RMS ‖target − online‖.
/// Zeros on accessor failure (non-fatal — diag only).
pub(crate) fn per_branch_target_drift(&self) -> [f32; 4] {
match self.trainer.per_branch_target_drift() {
Ok(v) => v,
Err(e) => {
tracing::warn!("per_branch_target_drift readback failed (zeroing): {e}");
[0.0_f32; 4]
}
}
}
/// Task 0.6 — per-branch VSN projection weight mean (H7 surrogate signal).
/// Mean |w| across each branch's VSN bottleneck pair (tensors 26..34
/// grouped by branch). Zeros on accessor failure (non-fatal — diag only).
pub(crate) fn per_branch_vsn_mean(&self) -> [f32; 4] {
match self.trainer.per_branch_vsn_mean() {
Ok(v) => v,
Err(e) => {
tracing::warn!("per_branch_vsn_mean readback failed (zeroing): {e}");
[0.0_f32; 4]
}
}
}
/// Task 0.3 — per-magnitude-bucket Q-value mean plumbing (Phase-0 stub).
/// Returns `[q_full, q_half, q_quarter] = [0.0; 3]`. Kernel-level
/// per-magnitude Q reduction deferred to Phase 1+ per plan §0.3 step 2.
pub(crate) fn q_magnitude_bucket_means(&self) -> [f32; 3] {
self.trainer.q_magnitude_bucket_means()
}
/// Flush the last in-flight async readback from the training step.
/// Called once at epoch end to retrieve the final step's scalars.
pub(crate) fn flush_readback(&mut self) -> Result<crate::cuda_pipeline::gpu_dqn_trainer::FusedTrainScalars> {

92
crates/ml/src/trainers/dqn/trainer/training_loop.rs
View File

@@ -2194,6 +2194,75 @@ impl DQNTrainer {
// Track 4 sigma_mean = overall NoisyNets σ across mag + dir branches.
let sigma_mean_overall = (sigma_mag + sigma_dir) * 0.5;
// Task 0.6 — per-branch VSN projection-weight mean (H7 surrogate)
// and per-branch target/online drift (H8). Host-computed cold path
// via dtoh of the flat params_buf + target_params_buf. 0 = direction,
// 1 = magnitude in the returned arrays (branches 2 = order, 3 = urg
// not surfaced in HEALTH_DIAG's noisy group — plan covers mag+dir).
let (vsn_mag, vsn_dir, drift_mag, drift_dir) = {
let vsn = self.fused_ctx.as_ref()
.map(|f| f.per_branch_vsn_mean())
.unwrap_or([0.0_f32; 4]);
let drift = self.fused_ctx.as_ref()
.map(|f| f.per_branch_target_drift())
.unwrap_or([0.0_f32; 4]);
(vsn[1], vsn[0], drift[1], drift[0])
};
// Task 0.3 — per-magnitude Q-value buckets (q_full / q_half /
// q_quarter). Phase-0 plumbing: returns [0.0; 3] per plan §0.3
// step 2 (per-mag Q reduction kernel deferred to Phase 1+ if
// measurement signals need). Call the accessor anyway so the
// call site is wired — swap to a real reducer without touching
// HEALTH_DIAG emit formatting.
let q_buckets = self.fused_ctx.as_ref()
.map(|f| f.q_magnitude_bucket_means())
.unwrap_or([0.0_f32; 3]);
let q_full = q_buckets[0];
let q_half = q_buckets[1];
let q_quarter = q_buckets[2];
// Task 0.3 — var_scale_mean / kelly_f_mean / avg_win_ratio.
//
// * `var_scale_mean`: the kernel computes `1 / (1 + sqrt(Var[Q]))`
// per-sample inside experience_env_step (~line 1422) but does
// NOT persist it to a device buffer — it is consumed immediately
// to scale `effective_max_pos` and then falls out of scope.
// Exposing it would require a dedicated per-sample output
// buffer + launch arg + kernel write; deferred to Phase 1+ if
// measurement signals need. Phase-0 stub = 0.0 (documented).
//
// * `kelly_f_mean` + `avg_win_ratio`: computed from the most
// recent `TradeStats` (aggregated ps[14..18] across episodes).
// `trade_stats_history` is appended to later in THIS same
// epoch — so on epoch `e` the last entry holds epoch `e-1`'s
// trade stats (empty on epoch 0 — both fields = 0). Same
// Kelly formula as the kernel: `b = avg_win / avg_loss`,
// `kelly_f = (b·p − (1−p))/b`, clamped [0,1], then half-Kelly.
// `avg_win_ratio = (sum_wins/win_count) / max(sum_losses/loss_count, 0.001)`
// per plan formula (line 239).
let var_scale_mean: f32 = 0.0;
let (kelly_f_mean, avg_win_ratio) = if let Some(ts) = self.trade_stats_history.last() {
let w = ts.winning_trades as f64;
let l = ts.losing_trades as f64;
let total = w + l;
if total > 0.0 && w > 0.0 && l > 0.0 {
let avg_win = ts.sum_wins / w;
let avg_loss = ts.sum_losses / l;
let win_rate = w / total;
let b = avg_win / avg_loss.max(0.001);
let kelly_raw = (b * win_rate - (1.0 - win_rate)) / b.max(0.001);
let kelly_clamped = kelly_raw.clamp(0.0, 1.0);
let kelly_half = 0.5 * kelly_clamped;
let ratio = avg_win / avg_loss.max(0.001);
(kelly_half as f32, ratio as f32)
} else {
(0.0_f32, 0.0_f32)
}
} else {
(0.0_f32, 0.0_f32)
};
// Track 1 action distribution: per-magnitude usage across the epoch.
// action_counts[9] layout is dir*3 + mag — Quarter = mag 0 (indices 0,3,6),
// Half = mag 1 (indices 1,4,7), Full = mag 2 (indices 2,5,8). Computed
@@ -2376,9 +2445,16 @@ impl DQNTrainer {
g12_loss,
// Track 1 — magnitude (10 f32): q_full, q_half, q_quarter, var_scale,
// kelly_f, avg_win_ratio, grad_ratio_mag_dir, dist_q, dist_h, dist_f.
// Q-values, var_scale, Kelly remain to be wired by future Task 0.3 work.
// grad_ratio_mag_dir now real (Task 0.4); action_dist wired.
0.0_f32, 0.0_f32, 0.0_f32, 0.0_f32, 0.0_f32, 0.0_f32,
// * q_full/half/quarter: plumbed via q_magnitude_bucket_means()
// but return [0.0; 3] in Phase 0 (kernel reduction deferred).
// * var_scale: kernel-internal per-sample value not persisted
// to a device buffer — Phase-0 honest stub, exposure deferred.
// * kelly_f, avg_win_ratio: real (from trade_stats_history last()).
// * grad_ratio_mag_dir: real (Task 0.4); action_dist wired.
q_full, q_half, q_quarter,
var_scale_mean,
kelly_f_mean,
avg_win_ratio,
grad_ratio_mag_dir,
dist_q, dist_h, dist_f,
// Track 1 — trail (6 f32): fire_q/h/f, hold_q/h/f (H6 — Task 0.5).
@@ -2386,9 +2462,13 @@ impl DQNTrainer {
trail_rates[0], trail_rates[1], trail_rates[2],
hold_means[0], hold_means[1], hold_means[2],
// Track 1 — noisy/VSN/drift (6 f32): vsn_mag, vsn_dir, sigma_mag,
// sigma_dir, drift_mag, drift_dir. NoisyNets σ wired (Task 0.6);
// VSN mask + target drift remain to be wired by future Task 0.6 work.
0.0_f32, 0.0_f32, sigma_mag, sigma_dir, 0.0_f32, 0.0_f32,
// sigma_dir, drift_mag, drift_dir. All real (Task 0.6):
// * vsn_mag/dir: mean |w| across each branch's VSN projection
// weights (H7 surrogate — live mask not materialized).
// * sigma_mag/dir: NoisyNets σ mean across fc+out layers.
// * drift_mag/dir: RMS ‖target − online‖ across each branch's
// 4 weight tensors (host-computed cold path; no atomics).
vsn_mag, vsn_dir, sigma_mag, sigma_dir, drift_mag, drift_dir,
// Track 1 — eval dist (3 f32): per-magnitude action distribution
// observed in the most recent validation backtest. H10 signal.
self.last_eval_magnitude_dist[0],