#![allow(unsafe_code)] // Raw pointer borrow splits for deterministic eval
//! DQN Trainer — Training metrics, Q-value diagnostics, and validation

use anyhow::Result;
use super::DQNTrainer;
use crate::TrainingMetrics;
use crate::fxcache::TARGET_RAW_CLOSE;
use super::super::config::DQNAgentType;

/// Rolling EMA values for LearningHealth inputs.
#[derive(Debug, Clone, Default)]
pub(crate) struct HealthEmaTrackers {
    pub q_gap_ema: f32,
    pub q_var_ema: f32,
    pub grad_norm_ema: f32,
    prev_grad_norm: Option<f32>,        // legacy scalar fallback
    prev_grad_vec: Option<Vec<f32>>,    // F2: true vector for cosine
    ema_alpha: f32, // initialized to 0.1 on first update
}

impl HealthEmaTrackers {
    pub(crate) fn update(&mut self, q_gap: f32, q_var: f32, grad_norm: f32) {
        if self.ema_alpha == 0.0 { self.ema_alpha = 0.1; }
        self.q_gap_ema = (1.0 - self.ema_alpha) * self.q_gap_ema + self.ema_alpha * q_gap;
        self.q_var_ema = (1.0 - self.ema_alpha) * self.q_var_ema + self.ema_alpha * q_var;
        self.grad_norm_ema = (1.0 - self.ema_alpha) * self.grad_norm_ema + self.ema_alpha * grad_norm;
    }

    /// Simple scalar proxy for gradient direction consistency: 1 - |delta|/max.
    /// Returns 0.0 on first call; +1.0 means no change, -1.0 means opposite sign.
    /// Kept as fallback for when fused_ctx is None or readback fails.
    pub(crate) fn update_grad_consistency(&mut self, current_grad_norm: f32) -> f32 {
        let cos = match self.prev_grad_norm {
            None => 0.0,
            Some(prev) => {
                let denom = prev.abs().max(current_grad_norm.abs()).max(1e-6);
                let delta = (current_grad_norm - prev).abs() / denom;
                (1.0 - 2.0 * delta).clamp(-1.0, 1.0)
            }
        };
        self.prev_grad_norm = Some(current_grad_norm);
        cos
    }

    /// F2: True vector cosine similarity between current and previous aggregated
    /// Adam m-state sample. Returns [-1, 1], with 0 on first call (no prior).
    /// Falls back to 0.0 on dimension mismatch (which shouldn't happen in practice,
    /// but would indicate a recompile-time layout change).
    pub(crate) fn update_grad_consistency_vec(&mut self, current: &[f32]) -> f32 {
        let cos = match &self.prev_grad_vec {
            None => 0.0,
            Some(prev) if prev.len() != current.len() => 0.0,
            Some(prev) => {
                let n = current.len();
                let dot: f32 = (0..n).map(|i| current[i] * prev[i]).sum();
                let norm_c: f32 = (0..n).map(|i| current[i] * current[i]).sum::<f32>().sqrt();
                let norm_p: f32 = (0..n).map(|i| prev[i] * prev[i]).sum::<f32>().sqrt();
                if norm_c < 1e-8 || norm_p < 1e-8 { 0.0 } else { (dot / (norm_c * norm_p)).clamp(-1.0, 1.0) }
            }
        };
        self.prev_grad_vec = Some(current.to_vec());
        cos
    }
}

impl DQNTrainer {
    /// Full training loop with existing logic (Wave 12 Group 3)
    /// Calculate average metrics for an epoch
    pub(crate) fn calculate_epoch_metrics(
        epoch_loss: f64,
        epoch_q_value: f64,
        epoch_gradient_norm: f64,
        samples_processed: usize,
    ) -> (f64, f64, f64) {
        if samples_processed > 0 {
            let count = samples_processed as f64;
            (
                epoch_loss / count,
                epoch_q_value / count,
                epoch_gradient_norm / count,
            )
        } else {
            (0.0, 0.0, 0.0)
        }
    }

    /// Compute validation loss on held-out data
    /// WAVE 10.6: Batched validation for 5-10x speedup
    
    // collect_qvalue_statistics removed: zero callers, was a Candle forward path.
    // Q-value stats are now computed exclusively via fused_ctx.compute_q_stats() (cuBLAS).

    /// Create final training metrics
    pub(crate) async fn create_final_metrics(
        &self,
        total_loss: f64,
        total_q_value: f64,
        total_gradient_norm: f64,
        total_reward: f64,
        num_epochs: usize,
        training_duration: std::time::Duration,
        early_stopped: bool,
        total_action_counts: [usize; 12], // 12 exposure bins (dir×mag = 4×3)
        total_factored_action_counts: [usize; 108], // 108 factored actions (12 exp * 3 ord * 3 urg)
    ) -> Result<TrainingMetrics> {
        let final_loss = total_loss / num_epochs as f64;
        let avg_q_value_final = total_q_value / num_epochs as f64;
        let avg_grad_norm_final = total_gradient_norm / num_epochs as f64;
        let avg_episode_reward = total_reward / num_epochs as f64;

        let mut metrics = TrainingMetrics {
            loss: final_loss,
            accuracy: 0.0,
            precision: 0.0,
            recall: 0.0,
            f1_score: 0.0,
            training_time_seconds: training_duration.as_secs_f64(),
            epochs_trained: num_epochs as u32,
            convergence_achieved: final_loss < 1.0,
            additional_metrics: std::collections::HashMap::new(),
        };

        metrics.add_metric("avg_q_value", avg_q_value_final);
        metrics.add_metric("avg_gradient_norm", avg_grad_norm_final);
        metrics.add_metric("final_epsilon", self.get_epsilon().await.unwrap_or(0.1));
        metrics.add_metric("avg_episode_reward", avg_episode_reward);
        metrics.add_metric("best_sharpe", self.best_sharpe);
        metrics.add_metric("best_val_loss", self.best_val_loss);
        metrics.add_metric("best_epoch", self.best_epoch as f64);
        metrics.add_metric("epoch_sharpe", self.best_sharpe);
        metrics.add_metric("val_loss", self.best_val_loss);

        // Per-epoch history for overfitting detection (last fold only — history resets per fold)
        let first_epoch_loss = self.loss_history.first().copied().unwrap_or(f64::NAN);
        let last_epoch_loss = self.loss_history.last().copied().unwrap_or(f64::NAN);
        let final_sharpe = self.sharpe_history.last().copied().unwrap_or(f64::NAN);
        let first_sharpe = self.sharpe_history.first().copied().unwrap_or(f64::NAN);
        let final_val_loss = self.val_loss_history.last().copied().unwrap_or(f64::NAN);
        metrics.add_metric("first_epoch_loss", first_epoch_loss);
        metrics.add_metric("last_epoch_loss", last_epoch_loss);
        metrics.add_metric("final_sharpe", final_sharpe);
        metrics.add_metric("first_sharpe", first_sharpe);
        metrics.add_metric("final_val_loss", final_val_loss);

        // Action metrics: use factored 108-action space if branching, else 12 exposure bins
        let total_factored: usize = total_factored_action_counts.iter().sum();
        let total_exposure: usize = total_action_counts.iter().sum();
        let total_actions = total_factored.max(total_exposure);
        if total_actions > 0 {
            // Factored 108-action diversity (primary metric when branching)
            if total_factored > 0 {
                let unique_factored = total_factored_action_counts.iter()
                    .filter(|&&count| count > 0).count();
                let factored_diversity = (unique_factored as f64 / 108.0) * 100.0;
                metrics.add_metric("action_diversity", factored_diversity);
                metrics.add_metric("factored_unique_actions", unique_factored as f64);
                metrics.add_metric("action_space_size", 108.0);

                // Active factored actions (used >0.5% of the time)
                let active_threshold = (total_factored as f64 * 0.005).max(1.0);
                let active_count = total_factored_action_counts.iter()
                    .filter(|&&count| count as f64 >= active_threshold)
                    .count();
                let active_diversity_pct = (active_count as f64 / 108.0) * 100.0;
                metrics.add_metric("active_actions_count", active_count as f64);
                metrics.add_metric("active_diversity_pct", active_diversity_pct);

                // Top factored actions
                let mut sorted_actions: Vec<(usize, usize)> = total_factored_action_counts.iter()
                    .enumerate()
                    .map(|(idx, &count)| (idx, count))
                    .collect();
                sorted_actions.sort_by(|a, b| b.1.cmp(&a.1));

                if let Some((top1_idx, top1_count)) = sorted_actions.first() {
                    let top1_pct = (*top1_count as f64 / total_factored as f64) * 100.0;
                    metrics.add_metric("top1_action_idx", *top1_idx as f64);
                    metrics.add_metric("top1_action_count", *top1_count as f64);
                    metrics.add_metric("top1_action_pct", top1_pct);
                }

                let top5_count: usize = sorted_actions.iter().take(5).map(|(_, c)| c).sum();
                let top5_coverage_pct = (top5_count as f64 / total_factored as f64) * 100.0;
                metrics.add_metric("top5_coverage_pct", top5_coverage_pct);
            } else {
                // Fallback: exposure-only (12 bins: dir×mag = 4×3)
                let unique_actions = total_action_counts.iter()
                    .filter(|&&count| count > 0).count();
                let action_diversity = (unique_actions as f64 / 12.0) * 100.0;
                metrics.add_metric("action_diversity", action_diversity);
                metrics.add_metric("action_space_size", 12.0);

                let active_threshold = (total_exposure as f64 * 0.005).max(1.0);
                let active_count = total_action_counts.iter()
                    .filter(|&&count| count as f64 >= active_threshold)
                    .count();
                let active_diversity_pct = (active_count as f64 / 12.0) * 100.0;
                metrics.add_metric("active_actions_count", active_count as f64);
                metrics.add_metric("active_diversity_pct", active_diversity_pct);

                let mut sorted_actions: Vec<(usize, usize)> = total_action_counts.iter()
                    .enumerate()
                    .map(|(idx, &count)| (idx, count))
                    .collect();
                sorted_actions.sort_by(|a, b| b.1.cmp(&a.1));

                if let Some((top1_idx, top1_count)) = sorted_actions.first() {
                    let top1_pct = (*top1_count as f64 / total_exposure as f64) * 100.0;
                    metrics.add_metric("top1_action_idx", *top1_idx as f64);
                    metrics.add_metric("top1_action_count", *top1_count as f64);
                    metrics.add_metric("top1_action_pct", top1_pct);
                }

                let top5_count: usize = sorted_actions.iter().take(5).map(|(_, c)| c).sum();
                let top5_coverage_pct = (top5_count as f64 / total_exposure as f64) * 100.0;
                metrics.add_metric("top5_coverage_pct", top5_coverage_pct);
            }

            metrics.add_metric("total_actions", total_actions as f64);

            // Buy/sell/hold from 12-bin exposure space (meaningful for P&L).
            // Layout: exp_idx = dir*3 + mag, dir ∈ {Short=0, Hold=1, Long=2, Flat=3}.
            //   Sell  = Short directions    = indices 0..3
            //   Hold  = Hold + Flat bins    = indices 3..6 ∪ 9..12
            //   Buy   = Long directions     = indices 6..9
            // Flat is "close to zero" which from a trading-flow perspective is neither
            // a long nor a short — we classify it as Hold (no new directional bet).
            let sell_count: usize = total_action_counts[0..3].iter().sum();
            let hold_count: usize = total_action_counts[3..6].iter().sum::<usize>()
                + total_action_counts[9..12].iter().sum::<usize>();
            let buy_count: usize = total_action_counts[6..9].iter().sum();
            metrics.add_metric("buy_count", buy_count as f64);
            metrics.add_metric("sell_count", sell_count as f64);
            metrics.add_metric("hold_count", hold_count as f64);
        }

        // Compute Q-value standard deviation across epochs for hyperopt stability penalty.
        // self.q_value_history stores per-epoch average Q-values; their std measures
        // how much Q-values fluctuate during training (volatility indicator).
        if self.q_value_history.len() >= 2 {
            let n = self.q_value_history.len() as f64;
            let mean = self.q_value_history.iter().sum::<f64>() / n;
            let variance = self
                .q_value_history
                .iter()
                .map(|&q| (q - mean).powi(2))
                .sum::<f64>()
                / n;
            let std_dev = variance.sqrt();
            metrics.add_metric("q_value_std", std_dev);
        } else {
            // Not enough data points to compute std; default to 0.0 (no volatility signal)
            metrics.add_metric("q_value_std", 0.0);
        }

        if early_stopped {
            metrics.add_metric("early_stopped", 1.0);
        }

        // Aggregate trade stats across all epochs
        let agg_total_trades: usize = self.trade_stats_history.iter()
            .map(|ts| ts.total_trades)
            .sum();
        let agg_winning: usize = self.trade_stats_history.iter()
            .map(|ts| ts.winning_trades)
            .sum();
        let overall_win_rate = if agg_total_trades > 0 {
            agg_winning as f64 / agg_total_trades as f64
        } else {
            0.0
        };
        let agg_sum_wins: f64 = self.trade_stats_history.iter()
            .map(|ts| ts.sum_wins)
            .sum();
        let agg_sum_losses: f64 = self.trade_stats_history.iter()
            .map(|ts| ts.sum_losses)
            .sum();
        let overall_pf = if agg_sum_losses > 1e-10 {
            agg_sum_wins / agg_sum_losses
        } else if agg_sum_wins > 0.0 {
            f64::INFINITY
        } else {
            0.0
        };

        metrics.add_metric("total_trades", agg_total_trades as f64);
        metrics.add_metric("overall_win_rate", overall_win_rate);
        metrics.add_metric("overall_profit_factor", overall_pf);

        // Also add last-epoch Sharpe for hyperopt consumption
        if let Some(last_ts) = self.trade_stats_history.last() {
            let financials = super::super::financials::compute_epoch_financials(
                last_ts,
                &total_action_counts,
                100_000.0,
                f64::from(self.hyperparams.bars_per_day),
            );
            metrics.add_metric("sharpe_ratio", financials.sharpe);
            metrics.add_metric("max_drawdown", financials.max_drawdown);
        }

        Ok(metrics)
    }

    /// Epoch-end Q-value diagnostics: gap analysis + per-action averages.
    ///
    /// Uses cuBLAS forward (`fused_ctx.compute_q_values`) — zero Candle involvement.
    /// Downloads the Q-value matrix to host for gap + per-action computation
    /// (cold path, epoch boundary only, ~8KB for 200 samples).
    ///
    /// Returns (gap_stats, per_action_avgs) where:
    /// - gap_stats: (mean_gap, min_gap, max_gap) of Q_best - Q_second_best
    /// - per_action_avgs: `[f64; 12]` averages indexed by `dir*3 + mag` (4 dirs × 3 mags)
    pub(crate) async fn compute_epoch_q_diagnostics(&mut self, agent: &mut DQNAgentType) -> Option<(
        (f64, f64, f64),
        [f64; 12],
    )> {
        let buffer = agent.memory_mut();

        if buffer.len() < 10 {
            return None;
        }

        // Cap sample_size at fused_ctx batch_size to avoid exceeding GPU buffer
        let fused_batch = self.fused_ctx.as_ref().map(|f| f.batch_size()).unwrap_or(0);
        if fused_batch == 0 {
            return None;
        }
        let sample_size = buffer.len().min(200).min(fused_batch);
        let gpu_batch = match buffer.sample(sample_size) {
            Ok(s) => s,
            Err(_) => return None,
        };

        let fused = self.fused_ctx.as_mut()?;
        let total_actions = fused.total_actions();
        // Pad states + graphed forward for deterministic Q-values
        let dst = fused.trainer_states_buf_ptr();
        if let Err(e) = fused.launch_pad_states(dst, gpu_batch.states_ptr, sample_size) {
            tracing::warn!("Q-stats pad_states failed: {e}");
            return None;
        }
        let q_out = match fused.replay_forward_for_q_values(sample_size) {
            Ok(q) => q,
            Err(_) => return None,
        };

        // Download Q-values to host (cold path, epoch boundary only).
        // q_out is [config.batch_size, total_actions] but we only need [sample_size, total_actions].
        // Allocate host buffer to match GPU buffer size, then truncate.
        let stream = self.cuda_stream.as_ref()?;
        let mut host_q = vec![0.0_f32; q_out.len()];
        crate::cuda_pipeline::dtoh_f32(stream, q_out, &mut host_q).ok()?;
        host_q.truncate(sample_size * total_actions);

        // Compute gap (best - second_best) WITHIN DIRECTION BRANCH ONLY
        // and per-action averages on host.
        // Q-values layout: [dir_0..dir_2, mag_0..mag_2, ord_0..ord_2, urg_0..urg_2]
        // The gap measures direction conviction (best_dir - second_best_dir).
        let cols = total_actions;
        let rows = sample_size;
        let (b0, b1, _, _) = {
            let ag = self.agent.read().await;
            ag.branch_sizes()
        };
        let mut gaps = Vec::with_capacity(rows);
        // 12 bins = b0 * b1 for the 4-branch DQN (dir×mag = 4×3). Aligns with
        // the monitoring kernel's exposure layout and per_action_avgs consumer.
        let mut col_sums = [0.0_f64; 12];
        for r in 0..rows {
            let row_offset = r * cols;
            // Direction gap: best vs second_best within first b0 columns only
            let mut best_val = f32::NEG_INFINITY;
            let mut second_best = f32::NEG_INFINITY;
            for c in 0..b0.min(cols) {
                let v = host_q.get(row_offset + c).copied().unwrap_or(0.0);
                if v > best_val {
                    second_best = best_val;
                    best_val = v;
                } else if v > second_best {
                    second_best = v;
                }
            }
            gaps.push(best_val - second_best);
            // Per-action sums: map branch outputs to dir×mag combos
            // dir_d × mag_m → index d*3 + m (matches monitoring kernel layout).
            // Cap at 4 directions × 3 magnitudes = 12 slots.
            let b0_cap = b0.min(4);
            let b1_cap = b1.min(3);
            for d in 0..b0_cap {
                let q_dir = host_q.get(row_offset + d).copied().unwrap_or(0.0) as f64;
                for m in 0..b1_cap {
                    let q_mag = host_q.get(row_offset + b0 + m).copied().unwrap_or(0.0) as f64;
                    let combo_idx = d * 3 + m; // kernel layout: always multiply by 3 (full mag dim)
                    if let Some(s) = col_sums.get_mut(combo_idx) {
                        // Combined Q ≈ Q_dir + Q_mag (additive advantage)
                        *s += q_dir + q_mag;
                    }
                }
            }
        }

        let rows_f64 = rows as f64;
        let gap_mean = gaps.iter().sum::<f32>() / rows as f32;
        let gap_min = gaps.iter().copied().fold(f32::INFINITY, f32::min);
        let gap_max = gaps.iter().copied().fold(f32::NEG_INFINITY, f32::max);
        let mut per_action_avgs = [0.0_f64; 12];
        for (i, avg) in per_action_avgs.iter_mut().enumerate() {
            *avg = col_sums.get(i).copied().unwrap_or(0.0) / rows_f64;
        }

        Some((
            (gap_mean as f64, gap_min as f64, gap_max as f64),
            per_action_avgs,
        ))
    }

    /// Get current training metrics
    pub async fn get_metrics(&self) -> TrainingMetrics {
        self.metrics.read().await.clone()
    }

    /// Check if early stopping criteria are met
    pub(crate) fn check_early_stopping(&self, avg_q_value: f64, epoch: usize) -> Option<String> {
        if !self.hyperparams.early_stopping_enabled
            || epoch + 1 < self.hyperparams.min_epochs_before_stopping
        {
            return None;
        }

        // Criterion 1: Q-value floor check
        if avg_q_value < self.hyperparams.q_value_floor {
            return Some(format!(
                "Q-value {:.4} below floor threshold {:.4}",
                avg_q_value, self.hyperparams.q_value_floor
            ));
        }

        // C4 FIX: Criterion 2 — Sharpe plateau check (was val-loss plateau).
        // Sharpe directly measures trading quality. A plateau means the model
        // has stopped improving its trading strategy, even if TD-loss still moves.
        if self.sharpe_history.len() >= self.hyperparams.plateau_window {
            let window = self.hyperparams.plateau_window;
            let recent_sharpes: Vec<f64> = self
                .sharpe_history
                .iter()
                .rev()
                .take(window)
                .copied()
                .collect();

            if let (Some(&newest), Some(&oldest)) = (recent_sharpes.first(), recent_sharpes.last()) {
                // For Sharpe, improvement = newest - oldest (higher is better)
                let improvement = newest - oldest;

                if improvement < 0.01 {
                    let msg = if improvement < -0.01 {
                        format!(
                            "Sharpe worsening detected (delta: {:.4}, window: {})",
                            improvement,
                            window
                        )
                    } else {
                        format!(
                            "Sharpe plateau detected (improvement: {:.4}, window: {})",
                            improvement,
                            window
                        )
                    };
                    return Some(msg);
                }
            }
        }

        None
    }

    /// Launch validation backtest on the dedicated `validation_stream` —
    /// submits all eval kernels, queues the metrics readback into mapped-pinned
    /// memory, records `eval_done_event`, then returns **without** any host
    /// wait. The companion [`consume_validation_loss`] performs the single
    /// host `cuEventSynchronize` and parses the metrics for HEALTH_DIAG emit.
    ///
    /// Splitting launch from consume restores the cross-stream pipelining the
    /// dedicated `validation_stream` is supposed to provide: the host CPU
    /// thread immediately returns to submit the NEXT epoch's training kernels
    /// to `cuda_stream` while eval drains on `validation_stream`. The previous
    /// monolithic `compute_validation_loss` blocked inside `synchronize()`
    /// after the metrics launch, defeating the pipelining.
    ///
    /// Runs the full backtest step loop on GPU: gather kernel -> cuBLAS forward ->
    /// env kernel -> metrics reduction. The evaluator is lazy-initialized on the
    /// first call and cached across epochs. Weight pointers change each epoch
    /// (training updates weights), so we call `invalidate_dqn_graph()` to force
    /// CUDA Graph re-capture.
    pub(crate) async fn launch_validation_loss(&mut self) -> Result<()> {
        use crate::cuda_pipeline::gpu_backtest_evaluator::{
            DqnBacktestConfig, GpuBacktestConfig, GpuBacktestEvaluator,
        };

        if self.val_data.is_empty() {
            return Ok(());
        }

        // Run validation on the dedicated `validation_stream` so eval kernels
        // overlap with the next epoch's training kernels on `cuda_stream`.
        //
        // Determinism is preserved by the per-stream cuBLAS handle architecture:
        // each stream owns its own `PerStreamCublasHandles` (training has one,
        // validation has another — see the val TLOB cuBLAS init below at the
        // `PerStreamCublasHandles::new(&eval_stream)` call, and the evaluator
        // forks its own internal stream + cuBLAS handle from
        // `parent_stream` in `GpuBacktestEvaluator::new`). cuBLAS bit-wise
        // reproducibility caveats apply per-handle, not across the process —
        // so as long as a single stream uses a single handle for any given
        // GEMM, both sides remain deterministic. Prior code routed eval back
        // through the training stream under a stale interpretation of that
        // caveat, costing ~30-40s/epoch in serialised wall time.
        //
        // GPU-side ordering: record a fresh event on `cuda_stream` (capturing
        // all training kernels submitted so far this epoch — Q-stat readback,
        // weight writes, ISV updates) and have `validation_stream` wait on
        // it before launching eval kernels. The evaluator's internal forked
        // stream inherits the dependency through the parent_stream chain.
        // No CPU sync.
        let val_stream = self.validation_stream.as_ref()
            .or(self.cuda_stream.as_ref());
        let stream = match val_stream {
            Some(s) => s,
            None => return Ok(()), // CPU device — skip GPU validation
        };
        let train_done_event = if let (Some(ref vs), Some(ref ts)) =
            (self.validation_stream.as_ref(), self.cuda_stream.as_ref())
        {
            // Cross-stream barrier — only effective when val and train streams
            // differ. Skipped on the legacy single-stream path where
            // `validation_stream == None` and the fallback is `cuda_stream`.
            let train_done = ts.record_event(None).map_err(|e| anyhow::anyhow!(
                "training_done event record: {e}"
            ))?;
            vs.wait(&train_done).map_err(|e| anyhow::anyhow!(
                "validation_stream wait training_done: {e}"
            ))?;
            Some(train_done)
        } else {
            None
        };

        // ── Lazy-init the GPU evaluator (once per fold, reused across epochs) ──
        if self.gpu_evaluator.is_none() {
            let market_dim: usize = ml_core::state_layout::MARKET_DIM;
            let ofi_dim: usize = ml_core::state_layout::OFI_DIM;
            let feature_dim: usize = market_dim + ofi_dim; // 42 market + OFI (portfolio+MTF in state_dim, not feature_dim)

            // Per-step subsample stride for validation — every Nth bar is fed to
            // the GPU backtest, scaling val cost by 1/N. The stride is paired
            // with `GpuBacktestConfig::val_subsample_stride` so the
            // annualization factor reflects the effective sampling cadence and
            // not the raw bar cadence (audit
            // `docs/lookahead-bias-audit-2026-04-28.md` §5: prior code left
            // `bars_per_day` unscaled, inflating reported Sharpe by sqrt(N)).
            // Single source of truth — change here propagates to both the
            // sampling loop AND the annualization scalar.
            const VAL_SUBSAMPLE_STRIDE: u32 = 4;

            // Build a single window from all val_data
            let mut prices: Vec<[f32; 4]> = Vec::with_capacity(self.val_data.len());
            let mut features: Vec<Vec<f32>> = Vec::with_capacity(self.val_data.len());

            for (i, (fv, target)) in self.val_data.iter().enumerate() {
                // v8 perf: subsample every Nth bar — N× faster validation.
                // Annualization compensated via val_subsample_stride below so
                // sharpe_annualised reflects the effective sampling cadence.
                if (i as u32) % VAL_SUBSAMPLE_STRIDE != 0 { continue; }

                // Prices: [close, close, close, close] — same convention as hyperopt adapter.
                // Fix 30 Stale-A: read raw_close at the contract-owner column
                // (`TARGET_RAW_CLOSE = 2`), not `target[0]` which is `preproc_close`
                // (z-normed log-return) post Bug-1 (commit `5a5dd0fed`). The pre-Fix-30
                // fallback `fv[3]` was a feature-vector log-return slot — also not a
                // dollar price. With `TARGET_DIM = 6` post commit `063fd2716`,
                // `set_val_data_from_slices` always yields `target.len() == 6`, so
                // every production path hits the constant-indexed read; no fallback
                // is reachable. Indexing via the named constant ensures any future
                // column rename moves the call site with the writer (Fix 27 pattern).
                let close = target[TARGET_RAW_CLOSE] as f32;
                prices.push([close, close, close, close]);

                // Features: 42 market features + OFI_DIM OFI
                let fv_slice = &fv[..market_dim.min(fv.len())];
                let mut fv_f32: Vec<f32> = fv_slice.iter().map(|&v| v as f32).collect();
                let ofi_idx = self.ofi_val_offset + i;
                let ofi_row = self.ofi_features.as_ref()
                    .and_then(|o| o.get(ofi_idx))
                    .expect("OFI features missing or index out of bounds — model requires OFI data");
                for &v in ofi_row.iter() {
                    fv_f32.push(v as f32);
                }
                features.push(fv_f32);
            }

            let window_prices = vec![prices];
            let window_features = vec![features];

            let hp = &self.hyperparams;
            #[allow(clippy::cast_possible_truncation)]
            let config = GpuBacktestConfig {
                max_position: hp.max_position_absolute as f32,
                tx_cost_bps: hp.transaction_cost_multiplier as f32,
                spread_cost: (hp.tick_size * hp.fill_spread_cost_frac) as f32, // points only, no contract_multiplier
                initial_capital: hp.initial_capital as f32,
                contract_multiplier: hp.contract_multiplier as f32,
                margin_pct: hp.margin_pct as f32,
                max_leverage: hp.max_leverage as f32,
                ofi_dim,
                holding_cost_rate: hp.holding_cost_rate as f32,
                churn_threshold: hp.churn_threshold_bars as f32,
                churn_penalty_scale: hp.churn_penalty_scale as f32,
                opportunity_cost_scale: 0.0,  // disabled during backtest evaluation
                bars_per_day: hp.bars_per_day as f32,
                trading_days_per_year: hp.trading_days_per_year as f32,
                val_subsample_stride: VAL_SUBSAMPLE_STRIDE,
            };

            let evaluator = GpuBacktestEvaluator::new(
                &window_prices,
                &window_features,
                feature_dim,
                config,
                stream,
            ).map_err(|e| anyhow::anyhow!("GpuBacktestEvaluator init: {e}"))?;

            self.gpu_evaluator = Some(evaluator);
        }

        // Wire ISV signals to evaluator for adaptive hold in validation
        if let (Some(ref mut eval), Some(ref fused)) = (&mut self.gpu_evaluator, &self.fused_ctx) {
            eval.set_isv_signals_ptr(fused.isv_signals_dev_ptr());
        }

        // D.8 TLOB val-path: sync training weights into the evaluator's TLOB instance.
        // Sizes the eval TLOB to `val_tlob_batch_size()` = DQN_BACKTEST_CHUNK_SIZE *
        // n_windows so `submit_dqn_step_loop_cublas` can run TLOB.forward once per
        // chunk on the chunked states buffer, replacing the prior per-step call
        // on states_buf at batch=n_windows. Cuts TLOB launches by chunk_len (~512x
        // fewer cuBLAS SGEMMs per chunk; matches the chunked-gather refactor).
        // Called every epoch so val TLOB tracks online weights after each Adam step.
        if let (Some(ref mut eval), Some(ref fused)) = (&mut self.gpu_evaluator, &self.fused_ctx) {
            use crate::cuda_pipeline::gpu_tlob::GpuTlob;
            use crate::cuda_pipeline::shared_cublas_handle::PerStreamCublasHandles;
            use std::sync::Arc;
            let train_tlob = &fused.gpu_tlob;
            let eval_stream = Arc::clone(eval.stream());
            let val_tlob_batch = eval.val_tlob_batch_size();
            // No graceful-degrade: TLOB is on the val production path. If
            // init or weight sync fails we surface the error so the
            // train/val state-distribution parity is preserved. Per
            // `feedback_no_hiding.md`: wire up or surface, never suppress.
            let h = PerStreamCublasHandles::new(&eval_stream)
                .map_err(|e| anyhow::anyhow!("Val TLOB shared_cublas: {e:?}"))?;
            let eval_tlob = GpuTlob::new(Arc::new(h), val_tlob_batch)
                .map_err(|e| anyhow::anyhow!("Val TLOB init: {e:?}"))?;
            eval.set_tlob_from_training(eval_tlob, train_tlob)
                .map_err(|e| anyhow::anyhow!("Val TLOB weight sync: {e:?}"))?;
        }

        // ── Extract weight pointers + v_range, then split borrow for fused_ctx ──
        // DuelingWeightSet / BranchingWeightSet contain raw u64 device pointers.
        // We extract stable pointers to them, then take &mut fused_ctx for QValueProvider.
        let fused_ptr = self.fused_ctx.as_mut()
            .ok_or_else(|| anyhow::anyhow!("fused_ctx required for GPU validation"))?
            as *mut super::super::fused_training::FusedTrainingCtx;
        // Safety: fused_ctx pointer is stable (no reallocation during evaluation).
        let (online_weights_ptr, branching_weights_ptr) = unsafe {
            let f = &*fused_ptr;
            (f.online_dueling_ref() as *const _, f.online_branching_ref() as *const _)
        };
        // Eval v_range from EMA-smoothed Q-stats — adaptive to actual Q-value scale.
        // Fixed [-240, 240] was 4800x too wide for Q-values near 0, making backtest
        // action selection essentially random (all Q-values map to same center atoms).
        let fused_ref = unsafe { &*fused_ptr };
        let vr = fused_ref.eval_v_range();

        let agent = self.agent.read().await;
        let network_dims = agent.network_dims();
        let (b0, b1, b2, b3) = agent.branch_sizes();
        let is_branching = agent.is_using_branching();
        drop(agent);

        let evaluator = self.gpu_evaluator.as_mut()
            .ok_or_else(|| anyhow::anyhow!("gpu_evaluator must be initialized"))?;
        evaluator.invalidate_dqn_graph();

        // The evaluator's internal stream forks parent_stream once at
        // construction; subsequent training-stream operations don't auto-flow
        // into the evaluator stream. Have the evaluator's stream wait on the
        // training_done event captured above so DQN weights and ISV slots
        // written by the just-finished training step are globally visible
        // before the eval kernels read them. No-op when val and train share
        // a stream (legacy single-stream path).
        if let Some(ref ev) = train_done_event {
            evaluator.stream().wait(ev).map_err(|e| anyhow::anyhow!(
                "evaluator stream wait training_done: {e}"
            ))?;
        }

        let hp = &self.hyperparams;
        #[allow(clippy::cast_possible_truncation)]
        let dqn_cfg = DqnBacktestConfig {
            shared_h1: network_dims.0,
            shared_h2: network_dims.1,
            value_h: network_dims.2,
            adv_h: network_dims.3,
            num_atoms: hp.num_atoms,
            branch_0_size: b0,
            branch_1_size: if is_branching { b1 } else { b0 },
            branch_2_size: if is_branching { b2 } else { b0 },
            branch_3_size: if is_branching { b3 } else { b0 },
            v_min: vr[0],
            v_max: vr[1],
        };

        // Safety: fused_ptr and evaluator are independent fields — no aliasing.
        // `_async` submits eval kernels + records `eval_done_event` then returns
        // immediately — no host wait. The matching `consume_validation_loss`
        // (called at the start of the NEXT epoch) does the single
        // `cuEventSynchronize` and parses the metrics for HEALTH_DIAG emit.
        unsafe {
            evaluator.evaluate_dqn_graphed_async(
                &*online_weights_ptr,
                Some(&*branching_weights_ptr),
                &dqn_cfg,
                Some(&mut *fused_ptr),
            )
        }.map_err(|e| anyhow::anyhow!("GPU backtest evaluate_dqn_graphed_async: {e}"))?;

        Ok(())
    }

    /// Wait on `eval_done_event` (host-side `cuEventSynchronize`), parse the
    /// per-window metrics from mapped-pinned memory, emit HEALTH_DIAG `val[...]`
    /// + per-direction / intent-magnitude / picked-direction distribution lines,
    /// and return the val_loss (= -sharpe) for downstream early-stopping /
    /// best-checkpoint logic.
    ///
    /// **MUST** be called exactly once after each [`launch_validation_loss`].
    /// Calling without a prior launch returns `Ok(0.0)` (the val_data-empty
    /// fall-through).
    pub(crate) async fn consume_validation_loss(&mut self) -> Result<f64> {
        if self.val_data.is_empty() {
            return Ok(0.0);
        }

        let evaluator = match self.gpu_evaluator.as_mut() {
            Some(ev) => ev,
            // Lazy-init only happens inside `launch_validation_loss`. If we got
            // here without a launch (e.g., CPU device path), there's nothing to
            // consume — return 0 to match the empty-val_data fall-through.
            None => return Ok(0.0),
        };

        let metrics = evaluator.consume_metrics_after_event()
            .map_err(|e| anyhow::anyhow!("GPU backtest consume_metrics: {e}"))?;

        // Single window — log full validation metrics
        let val_sharpe = if let Some(m) = metrics.first() {
            tracing::info!(
                "Validation backtest: Sharpe={:.2} Sortino={:.2} WinRate={:.1}% MaxDD={:.3}% Trades={:.0} Calmar={:.2}",
                m.sharpe, m.sortino, m.win_rate * 100.0, m.max_drawdown * 100.0, m.total_trades, m.calmar,
            );

            // Plan 5 Task 5 Phase A — flat-emit per-epoch validation metrics
            // in HEALTH_DIAG `val [...]` block format so the
            // `aggregate-multi-seed-metrics.py` parser picks them up as
            // top-level `val_*` aggregate keys (joined with single underscore
            // by the aggregator). Tier-2/tier-3 check scripts read these keys
            // directly. Derivations (no new GPU work):
            //   * `trade_count`         = m.total_trades (kernel's boundary-
            //     stitched count of position-sign cycles)
            //   * `window_bars`         = buy + sell + hold (every bar
            //     contributes exactly one direction tally per kernel)
            //   * `trades_per_bar`      = trade_count / window_bars
            //   * `active_frac`         = (buy + sell) / window_bars (kernel
            //     counts Hold AND Flat into hold_count, so "active" = bars
            //     where the policy chose Short or Long — directly meets the
            //     Tier-2 spec "model is not always Hold" intent)
            //   * `dir_entropy`         = Shannon entropy in nats over the
            //     3-bucket {short, hold-or-flat, long} distribution from the
            //     kernel; max log(3) ≈ 1.0986. Note: the spec compares
            //     against 0.8·log(4) ≈ 1.109 which is unreachable here
            //     because the kernel collapses Hold+Flat into one bucket.
            //     Tier-2's check_tier2.py needs to use the 3-bucket ceiling
            //     (or fall back to per-direction `val_dir_dist_*`); see audit
            //     doc row for follow-up.
            //   * `profit_factor`       = m.omega_ratio (per-step gain_sum /
            //     loss_sum from the kernel's bitonic-sorted return array;
            //     mathematically equivalent to profit-factor when threshold
            //     = 0, which is the standard definition. NOT trade-level PF
            //     — that would require boundary-aware per-trade P&L which
            //     the kernel doesn't yet emit; flagged as future enhancement
            //     in audit row).
            //   * `sharpe_annualised`   = m.sharpe (kernel already multiplies
            //     by `sqrt(bars_per_day * 252)` — see backtest_metrics_kernel
            //     line 266). Aliased so tier3's primary key is present.
            let buy = m.buy_count;
            let sell = m.sell_count;
            let hold = m.hold_count;
            let window_bars = buy + sell + hold;
            let trades_per_bar = if window_bars > 0.0 { m.total_trades / window_bars } else { 0.0 };
            let active_frac = if window_bars > 0.0 { (buy + sell) / window_bars } else { 0.0 };

            // Shannon entropy in nats over 3-bucket direction distribution
            // (short / hold-or-flat / long). Uses single-pass log of the
            // probability; non-positive buckets contribute nothing.
            let dir_entropy = if window_bars > 0.0 {
                let p_short = (sell / window_bars).max(0.0);
                let p_hold  = (hold / window_bars).max(0.0);
                let p_long  = (buy  / window_bars).max(0.0);
                let mut h = 0.0_f32;
                for &p in &[p_short, p_hold, p_long] {
                    if p > 0.0 { h -= p * p.ln(); }
                }
                h
            } else { 0.0 };

            // Snapshot for downstream consumers (smoke tests, future telemetry).
            self.last_val_metrics = Some([
                m.sharpe,
                m.sortino,
                m.win_rate,
                m.max_drawdown,
                m.total_trades,
                m.calmar,
                m.omega_ratio,
                m.total_pnl,
                m.var_95,
                m.cvar_95,
                m.buy_count,
                m.sell_count,
                m.hold_count,
                window_bars,
            ]);

            // HEALTH_DIAG `val [...]` block. Block name `val` + single-`_`
            // aggregator join → top-level keys `val_sharpe`, `val_sortino`,
            // ..., `val_profit_factor`. Additive line — does not perturb the
            // existing giant HEALTH_DIAG line.
            let epoch = self.current_epoch;
            tracing::info!(
                "HEALTH_DIAG[{}]: val [sharpe={:.4} sortino={:.4} win_rate={:.4} max_drawdown={:.4} trade_count={:.0} calmar={:.4} omega_ratio={:.4} total_pnl={:.4} var_95={:.4e} cvar_95={:.4e} trades_per_bar={:.6} active_frac={:.4} dir_entropy={:.4} sharpe_annualised={:.4} profit_factor={:.4} window_bars={:.0}]",
                epoch,
                m.sharpe,
                m.sortino,
                m.win_rate,
                m.max_drawdown,
                m.total_trades,
                m.calmar,
                m.omega_ratio,
                m.total_pnl,
                m.var_95,
                m.cvar_95,
                trades_per_bar,
                active_frac,
                dir_entropy,
                m.sharpe,        // sharpe_annualised alias (kernel already annualises)
                m.omega_ratio,   // profit_factor alias (per-step PF == omega at threshold 0)
                window_bars,
            );

            // Plan C Task 5 — training-rollout counterpart to val_active_frac.
            // SP8 (Fix 36, 2026-05-03): source migrated from host-side
            // computation to ISV[TRAIN_ACTIVE_FRAC_INDEX] (Pearls A+D smoothed,
            // produced by `train_active_frac_compute_kernel` after the
            // monitoring reduce). The accessor `last_train_active_frac()`
            // reads from ISV via `read_isv_signal_at` instead of the prior
            // host-cached field — eliminates the host-side reduction over
            // `monitor.action_counts[12]` per `feedback_no_cpu_compute_strict.md`.
            // `val_active_frac` above measures direction commitment during
            // the EVAL backtest; this measures it during TRAINING rollout —
            // divergence localises whether Thompson is generating directional
            // exploration during experience collection (Plan D L3 verification
            // gate, target > 0.40 = at least 40% of training-rollout actions
            // are Long or Short, not Hold/Flat). Emitted next to
            // val_active_frac so the two lines are adjacent in HEALTH_DIAG
            // output and easy to compare.
            tracing::info!(
                "HEALTH_DIAG[{}]: train_active_frac={:.4} (Long+Short during training rollout)",
                epoch, self.last_train_active_frac(),
            );

            m.sharpe as f64
        } else {
            0.0
        };

        // Task 0.7 — capture eval-mode per-magnitude action distribution from
        // the actions_history buffer for H10 detection. Failure is non-fatal:
        // distribution stays at last value if readback errors (won't break val).
        if let Some(ref ev) = self.gpu_evaluator {
            match ev.read_eval_action_distribution_per_magnitude() {
                Ok(dist) => self.last_eval_magnitude_dist = dist,
                Err(e) => tracing::warn!("eval magnitude dist readback failed (non-fatal): {e}"),
            }
            // SP9 (Fix 37, 2026-05-03): GPU-resident magnitude intent
            // distribution. Replaces the prior host-side DtoH reduction
            // `read_eval_intent_magnitude_distribution()` per
            // `feedback_no_cpu_compute_strict.md` and
            // `feedback_no_htod_htoh_only_mapped_pinned.md`. The SP9
            // producer kernel reduces `intent_mag_buf` directly on the GPU,
            // chains apply_pearls_ad, and writes ISV[EVAL_DIST_Q/H/F_INDEX].
            // We launch the producer here (after `consume_metrics_after_event`
            // has already synchronised the eval stream) and read back via
            // mapped-pinned `read_isv_signal_at` — the only allowed
            // CPU↔GPU path for ISV slots.
            //
            // The producer also drives the SP9 `intent_eval_divergence`
            // controller via ISV[EVAL_DIST_F_INDEX] — that producer fires
            // each training step, picking up whatever value this validation
            // window most recently wrote.
            let (intent_mag_dev_ptr, intent_mag_n_total) = ev.intent_mag_dev_ptr_and_n();
            if let Some(ref fused) = self.fused_ctx {
                if let Err(e) = fused.trainer().launch_eval_intent_dist_compute(
                    intent_mag_dev_ptr, intent_mag_n_total,
                ) {
                    tracing::warn!(
                        "SP9 eval_intent_dist_compute launch failed (non-fatal): {e}"
                    );
                }
                use crate::cuda_pipeline::sp5_isv_slots::{
                    EVAL_DIST_Q_INDEX, EVAL_DIST_H_INDEX, EVAL_DIST_F_INDEX,
                };
                let trainer = fused.trainer();
                self.last_eval_intent_magnitude_dist = [
                    trainer.read_isv_signal_at(EVAL_DIST_Q_INDEX),
                    trainer.read_isv_signal_at(EVAL_DIST_H_INDEX),
                    trainer.read_isv_signal_at(EVAL_DIST_F_INDEX),
                ];
            }
            // Per-direction eval distribution — needed to disambiguate eval
            // dir_entropy near zero. The kernel-side dir_entropy collapses
            // Hold+Flat into one bucket (3 buckets), so a healthy 25/25/25/25
            // direction split would still register as ~ln(3) entropy. A flat
            // dir_dist confirms the eval policy actually picks one direction;
            // a varied dir_dist with low entropy in the 3-bucket metric points
            // at the Hold/Flat collapse instead of mono-direction.
            let epoch = self.current_epoch;
            match ev.read_eval_action_distribution_per_direction() {
                Ok(dd) => tracing::info!(
                    "HEALTH_DIAG[{}]: val_dir_dist [short={:.4} hold={:.4} long={:.4} flat={:.4}]",
                    epoch, dd[0], dd[1], dd[2], dd[3],
                ),
                Err(e) => tracing::warn!(
                    "eval direction dist readback failed (non-fatal): {e}"
                ),
            }
            // Raw Boltzmann pick histogram from the most recent chunk's
            // chunked_actions buffer — BEFORE env_step touches `actual_dir`.
            // A divergence from `val_dir_dist` (post-physics) localises whether
            // the eval collapse is upstream (kernel produces collapsed picks
            // → both lines flat) or downstream (kernel diverse, env_step /
            // Kelly cap drains active picks to Flat → only val_dir_dist flat).
            // Sample is a single chunk (~64 bars on 1-window val), so noise
            // is higher than the full-window val_dir_dist; useful for
            // qualitative comparison against Boltzmann theory bounds.
            match ev.read_chunked_actions_direction_distribution() {
                Ok(rd) => tracing::info!(
                    "HEALTH_DIAG[{}]: val_picked_dir_dist [short={:.4} hold={:.4} long={:.4} flat={:.4}]",
                    epoch, rd[0], rd[1], rd[2], rd[3],
                ),
                Err(e) => tracing::warn!(
                    "raw Boltzmann pick dist readback failed (non-fatal): {e}"
                ),
            }
        }

        Ok(-val_sharpe)
    }

    /// Estimate average Q-value from replay buffer samples for monitoring.
    ///
    /// Uses the cuBLAS forward path (`fused_ctx.compute_q_values`) — zero Candle involvement.
    /// Eliminates the 27ms/epoch overhead from `agent.forward()` (Candle dispatch chain).
    ///
    /// WAVE 23 P0: Includes Q-value divergence check (early stopping) via
    /// `agent.log_q_values_from_stats()` — computed from the host readback.
    pub(crate) async fn estimate_avg_q_value_with_early_stopping(&mut self, agent: &mut DQNAgentType) -> Result<f64> {
        let buffer = agent.memory_mut();

        if buffer.len() == 0 {
            return Ok(0.0);
        }

        // Sample up to 10 experiences for Q-value estimation
        let sample_size = buffer.len().min(10);
        let gpu_batch = buffer
            .sample(sample_size)
            .map_err(|e| anyhow::anyhow!("Failed to sample experiences: {}", e))?;

        // cuBLAS forward + GPU reduction → 5 scalars, 20-byte readback
        let fused = self.fused_ctx.as_mut()
            .ok_or_else(|| anyhow::anyhow!("fused_ctx required for Q-value estimation"))?;

        let stats = fused.compute_q_stats_from_ptr(gpu_batch.states_ptr, sample_size)
            .map_err(|e| anyhow::anyhow!("cuBLAS Q-stats: {}", e))?;

        // Q-value divergence check (early stopping)
        let num_actions = fused.total_actions();
        agent.log_q_values_from_stats(stats.q_min, stats.q_max, stats.q_mean, stats.q_variance, num_actions)
            .map_err(|e| {
                tracing::info!("Early stopping triggered (Q-value divergence): {}", e);
                anyhow::anyhow!("Early stopping: {}", e)
            })?;

        Ok(stats.avg_max_q)
    }

}
