shape_runtime/simulation/

dense_kernel.rs

//! DenseKernel - High-Performance Simulation Hot Loop
//!
//! This module provides the core simulation kernel for maximum throughput
//! (>10M ticks/sec) using zero-copy data access and avoiding allocations.

use shape_ast::error::{Result, ShapeError};
use shape_value::DataTable;

/// JIT-compiled simulation kernel function type.
///
/// This signature matches the kernel ABI in shape-jit:
/// - cursor_index: Current row in the simulation
/// - series_ptrs: Pointer to column data pointers (*const *const f64)
/// - state_ptr: Pointer to TypedObject state buffer (*mut u8)
/// - Returns: i32 (0 = continue, 1 = done, negative = error)
pub type SimulationKernelFn = unsafe extern "C" fn(
    cursor_index: usize,
    series_ptrs: *const *const f64,
    state_ptr: *mut u8,
) -> i32;

// ============================================================================
// JIT Kernel Compiler Trait (for dependency injection)
// ============================================================================

/// Configuration for compiling a simulation kernel.
///
/// This is passed to the JIT compiler to provide field offset information.
#[derive(Debug, Clone, Default)]
pub struct KernelCompileConfig {
    /// State field name -> byte offset mapping
    pub state_field_offsets: Vec<(String, usize)>,
    /// Schema ID for the state type
    pub state_schema_id: u32,
    /// Column name -> index mapping for series data access
    pub column_map: Vec<(String, usize)>,
    /// Number of columns in the series
    pub column_count: usize,
}

impl KernelCompileConfig {
    /// Create a new empty config.
    pub fn new(schema_id: u32, column_count: usize) -> Self {
        Self {
            state_schema_id: schema_id,
            column_count,
            ..Default::default()
        }
    }

    /// Add a state field offset.
    pub fn with_state_field(mut self, name: &str, offset: usize) -> Self {
        self.state_field_offsets.push((name.to_string(), offset));
        self
    }

    /// Add a column mapping.
    pub fn with_column(mut self, name: &str, index: usize) -> Self {
        self.column_map.push((name.to_string(), index));
        self
    }
}

/// Trait for JIT kernel compilation.
///
/// This trait is implemented by `shape-jit::JITCompiler` and injected into
/// `ExecutionContext` to enable JIT kernel compilation without circular dependencies.
///
/// The bytecode is passed as raw bytes since we can't reference BytecodeProgram here.
pub trait KernelCompiler: Send + Sync {
    /// Compile a strategy function to a JIT kernel.
    ///
    /// # Arguments
    /// * `name` - Name for the compiled kernel (for caching/debugging)
    /// * `function_bytecode` - Serialized bytecode of the strategy function
    /// * `config` - Kernel configuration with field offsets
    ///
    /// # Returns
    /// The compiled kernel function pointer, or an error message.
    fn compile_kernel(
        &self,
        name: &str,
        function_bytecode: &[u8],
        config: &KernelCompileConfig,
    ) -> std::result::Result<SimulationKernelFn, String>;

    /// Check if the compiler supports a given feature.
    fn supports_feature(&self, feature: &str) -> bool {
        match feature {
            "typed_object" => true,
            "closures" => false, // Phase 1: no closure support
            _ => false,
        }
    }
}

/// Configuration for dense kernel execution.
#[derive(Debug, Clone)]
pub struct DenseKernelConfig {
    /// Start tick (inclusive)
    pub start: usize,
    /// End tick (exclusive)
    pub end: usize,
    /// Warmup period (ticks to skip at start for indicator initialization)
    pub warmup: usize,
}

impl DenseKernelConfig {
    /// Create a new config for the full range.
    pub fn full(len: usize) -> Self {
        Self {
            start: 0,
            end: len,
            warmup: 0,
        }
    }

    /// Create a config with warmup period.
    pub fn with_warmup(len: usize, warmup: usize) -> Self {
        Self {
            start: 0,
            end: len,
            warmup,
        }
    }

    /// Create a config for a specific range.
    pub fn range(start: usize, end: usize) -> Self {
        Self {
            start,
            end,
            warmup: 0,
        }
    }
}

/// Result of dense kernel execution.
#[derive(Debug)]
pub struct DenseKernelResult<S> {
    /// Final state after all ticks processed
    pub final_state: S,
    /// Number of ticks processed
    pub ticks_processed: usize,
    /// Whether simulation completed successfully
    pub completed: bool,
}

/// The dense simulation kernel - zero-allocation hot loop.
///
/// This kernel is designed for maximum throughput (>10M ticks/sec) by:
/// - Using zero-copy data access
/// - Operating on raw memory via `AccessPolicy`
/// - Avoiding allocations in the hot loop
pub struct DenseKernel {
    config: DenseKernelConfig,
}

impl DenseKernel {
    /// Create a new dense kernel with the given configuration.
    pub fn new(config: DenseKernelConfig) -> Self {
        Self { config }
    }

    /// Run simulation on a DataTable with a closure-based strategy.
    ///
    /// Extracts f64 column pointers from the DataTable and iterates
    /// `effective_start..config.end`, calling the strategy each tick.
    #[inline(always)]
    pub fn run<S, F>(
        &self,
        data: &DataTable,
        mut initial_state: S,
        mut strategy: F,
    ) -> Result<DenseKernelResult<S>>
    where
        F: FnMut(usize, &[*const f64], &mut S) -> i32,
    {
        // Extract f64 column pointers (stride == 8 means f64)
        let col_ptrs: Vec<*const f64> = data
            .column_ptrs()
            .iter()
            .filter(|cp| cp.stride == 8)
            .map(|cp| cp.values_ptr as *const f64)
            .collect();

        let effective_start = self.config.start + self.config.warmup;

        if effective_start >= self.config.end {
            return Err(ShapeError::RuntimeError {
                message: format!(
                    "Warmup ({}) exceeds available range ({} - {})",
                    self.config.warmup, self.config.start, self.config.end
                ),
                location: None,
            });
        }

        let mut ticks_processed = 0;

        for cursor_index in effective_start..self.config.end {
            let result = strategy(cursor_index, &col_ptrs, &mut initial_state);

            if result != 0 {
                return Ok(DenseKernelResult {
                    final_state: initial_state,
                    ticks_processed,
                    completed: result == 1,
                });
            }

            ticks_processed += 1;
        }

        Ok(DenseKernelResult {
            final_state: initial_state,
            ticks_processed,
            completed: true,
        })
    }

    /// Run simulation with JIT-compiled kernel function.
    ///
    /// This is the highest-performance path - bypasses SeriesCursor and calls
    /// the kernel directly with raw pointers.
    ///
    /// # Arguments
    /// * `column_ptrs` - Slice of column data pointers
    /// * `state_ptr` - Mutable pointer to state buffer (TypedObject)
    /// * `kernel` - JIT-compiled kernel function
    ///
    /// # Returns
    /// Result containing execution statistics (state is modified in-place).
    ///
    /// # Safety
    /// The column_ptrs must point to valid f64 arrays with length >= self.config.end.
    /// The state_ptr must point to a valid TypedObject buffer.
    #[inline(always)]
    pub unsafe fn run_jit(
        &self,
        column_ptrs: &[*const f64],
        state_ptr: *mut u8,
        kernel: SimulationKernelFn,
    ) -> Result<DenseKernelResult<()>> {
        let series_ptrs = column_ptrs.as_ptr();
        let effective_start = self.config.start + self.config.warmup;

        if effective_start >= self.config.end {
            return Err(ShapeError::RuntimeError {
                message: format!(
                    "Warmup ({}) exceeds available range ({} - {})",
                    self.config.warmup, self.config.start, self.config.end
                ),
                location: None,
            });
        }

        let mut ticks_processed = 0;

        // Hot loop - pure pointer arithmetic, no allocations
        for cursor_index in effective_start..self.config.end {
            let result = unsafe { kernel(cursor_index, series_ptrs, state_ptr) };

            if result != 0 {
                // Non-zero means stop or error
                return Ok(DenseKernelResult {
                    final_state: (),
                    ticks_processed,
                    completed: result == 1, // 1 = normal completion
                });
            }

            ticks_processed += 1;
        }

        Ok(DenseKernelResult {
            final_state: (),
            ticks_processed,
            completed: true,
        })
    }
}

/// Run a simulation on a DataTable with a closure-based strategy.
///
/// The strategy closure receives (tick_index, column_pointers, state) and returns
/// an i32 result code: 0 = continue, 1 = done, negative = error.
pub fn simulate<S, F>(
    data: &DataTable,
    initial_state: S,
    strategy: F,
) -> Result<DenseKernelResult<S>>
where
    F: FnMut(usize, &[*const f64], &mut S) -> i32,
{
    let config = DenseKernelConfig::full(data.row_count());
    let kernel = DenseKernel::new(config);
    kernel.run(data, initial_state, strategy)
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_kernel_compile_config() {
        let config = KernelCompileConfig::new(42, 3)
            .with_state_field("cash", 0)
            .with_state_field("position", 8)
            .with_column("open", 0)
            .with_column("close", 1)
            .with_column("volume", 2);

        assert_eq!(config.state_schema_id, 42);
        assert_eq!(config.column_count, 3);
        assert_eq!(config.state_field_offsets.len(), 2);
        assert_eq!(config.column_map.len(), 3);
    }

    #[test]
    fn test_dense_kernel_config() {
        let config = DenseKernelConfig::full(100);
        assert_eq!(config.start, 0);
        assert_eq!(config.end, 100);
        assert_eq!(config.warmup, 0);

        let config = DenseKernelConfig::with_warmup(100, 10);
        assert_eq!(config.warmup, 10);

        let config = DenseKernelConfig::range(5, 50);
        assert_eq!(config.start, 5);
        assert_eq!(config.end, 50);
    }

    /// Helper: build a DataTable with a single f64 "price" column.
    fn make_price_table(prices: Vec<f64>) -> DataTable {
        use arrow_array::{ArrayRef, Float64Array};
        use arrow_schema::{DataType, Field, Schema};
        use std::sync::Arc;

        let schema = Schema::new(vec![Field::new("price", DataType::Float64, false)]);
        let col: ArrayRef = Arc::new(Float64Array::from(prices));
        let batch = arrow_array::RecordBatch::try_new(Arc::new(schema), vec![col]).unwrap();
        DataTable::new(batch)
    }

    #[test]
    fn test_dense_kernel_run_sum() {
        let table = make_price_table(vec![10.0, 20.0, 30.0, 40.0, 50.0]);

        let config = DenseKernelConfig::full(table.row_count());
        let kernel = DenseKernel::new(config);

        // Strategy: sum all prices into state
        let result = kernel
            .run(&table, 0.0_f64, |idx, col_ptrs, state| {
                unsafe {
                    let price = *col_ptrs[0].add(idx);
                    *state += price;
                }
                0 // continue
            })
            .unwrap();

        assert!(result.completed);
        assert_eq!(result.ticks_processed, 5);
        assert_eq!(result.final_state, 150.0); // 10+20+30+40+50
    }

    #[test]
    fn test_dense_kernel_run_early_stop() {
        let table = make_price_table(vec![10.0, 20.0, 100.0, 40.0, 50.0]);

        let config = DenseKernelConfig::full(table.row_count());
        let kernel = DenseKernel::new(config);

        // Strategy: stop when price > 50
        let result = kernel
            .run(&table, 0.0_f64, |idx, col_ptrs, state| {
                let price = unsafe { *col_ptrs[0].add(idx) };
                if price > 50.0 {
                    return 1; // done
                }
                *state += price;
                0 // continue
            })
            .unwrap();

        assert!(result.completed); // result == 1 means normal completion
        assert_eq!(result.ticks_processed, 2); // processed indices 0, 1 then stopped at 2
        assert_eq!(result.final_state, 30.0); // 10+20
    }

    #[test]
    fn test_dense_kernel_with_warmup() {
        let table = make_price_table(vec![1.0, 2.0, 3.0, 4.0, 5.0]);

        let config = DenseKernelConfig::with_warmup(table.row_count(), 2);
        let kernel = DenseKernel::new(config);

        // Collect processed indices
        let mut processed_indices = Vec::new();
        let result = kernel
            .run(&table, 0.0_f64, |idx, col_ptrs, state| {
                unsafe {
                    processed_indices.push(idx);
                    *state += *col_ptrs[0].add(idx);
                }
                0
            })
            .unwrap();

        assert!(result.completed);
        // Should skip first 2 ticks (warmup), process indices 2, 3, 4
        assert_eq!(processed_indices, vec![2, 3, 4]);
        assert_eq!(result.ticks_processed, 3);
        assert_eq!(result.final_state, 12.0); // 3+4+5
    }

    #[test]
    fn test_dense_kernel_range() {
        let table = make_price_table(vec![1.0, 2.0, 3.0, 4.0, 5.0]);

        let config = DenseKernelConfig::range(1, 4);
        let kernel = DenseKernel::new(config);

        let result = kernel
            .run(&table, 0.0_f64, |idx, col_ptrs, state| {
                unsafe { *state += *col_ptrs[0].add(idx) };
                0
            })
            .unwrap();

        assert!(result.completed);
        assert_eq!(result.ticks_processed, 3); // indices 1, 2, 3
        assert_eq!(result.final_state, 9.0); // 2+3+4
    }

    #[test]
    fn test_dense_kernel_warmup_exceeds_range() {
        let table = make_price_table(vec![1.0, 2.0, 3.0]);

        // Warmup of 10 exceeds the 3-element range
        let config = DenseKernelConfig::with_warmup(table.row_count(), 10);
        let kernel = DenseKernel::new(config);

        let result = kernel.run(&table, 0.0_f64, |_idx, _col_ptrs, _state| 0);
        assert!(result.is_err());
    }

    #[test]
    fn test_simulate_convenience_fn() {
        let table = make_price_table(vec![10.0, 20.0, 30.0]);

        let result = simulate(&table, 0.0_f64, |idx, col_ptrs, state| {
            unsafe { *state += *col_ptrs[0].add(idx) };
            0
        })
        .unwrap();

        assert!(result.completed);
        assert_eq!(result.ticks_processed, 3);
        assert_eq!(result.final_state, 60.0);
    }

    #[test]
    fn test_dense_kernel_struct_state() {
        // Test with a more complex state struct
        #[derive(Debug, Default)]
        struct BacktestState {
            cash: f64,
            position: f64,
            trades: u32,
        }

        let table = make_price_table(vec![100.0, 105.0, 110.0]);
        let config = DenseKernelConfig::full(table.row_count());
        let kernel = DenseKernel::new(config);

        let initial = BacktestState {
            cash: 10000.0,
            position: 0.0,
            trades: 0,
        };

        // Buy at 100, hold at 105, sell at 110
        let result = kernel
            .run(&table, initial, |idx, col_ptrs, state| {
                let price = unsafe { *col_ptrs[0].add(idx) };
                match idx {
                    0 => {
                        // Buy 10 shares
                        state.cash -= 10.0 * price;
                        state.position = 10.0;
                        state.trades += 1;
                    }
                    2 => {
                        // Sell all
                        state.cash += state.position * price;
                        state.position = 0.0;
                        state.trades += 1;
                    }
                    _ => {} // hold
                }
                0
            })
            .unwrap();

        assert!(result.completed);
        assert_eq!(result.final_state.trades, 2);
        assert_eq!(result.final_state.position, 0.0);
        // Started 10000, bought 10@100 = -1000, sold 10@110 = +1100, net = 10100
        assert_eq!(result.final_state.cash, 10100.0);
    }

    #[test]
    fn test_dense_kernel_multi_column() {
        // Test with multiple f64 columns (price + volume)
        use arrow_array::{ArrayRef, Float64Array};
        use arrow_schema::{DataType, Field, Schema};
        use std::sync::Arc;

        let schema = Schema::new(vec![
            Field::new("price", DataType::Float64, false),
            Field::new("volume", DataType::Float64, false),
        ]);
        let prices: ArrayRef = Arc::new(Float64Array::from(vec![100.0, 105.0, 98.0]));
        let volumes: ArrayRef = Arc::new(Float64Array::from(vec![1000.0, 2000.0, 1500.0]));
        let batch =
            arrow_array::RecordBatch::try_new(Arc::new(schema), vec![prices, volumes]).unwrap();
        let table = DataTable::new(batch);

        let config = DenseKernelConfig::full(table.row_count());
        let kernel = DenseKernel::new(config);

        // Compute volume-weighted average price
        let result = kernel
            .run(&table, (0.0_f64, 0.0_f64), |idx, col_ptrs, state| {
                unsafe {
                    let price = *col_ptrs[0].add(idx);
                    let volume = *col_ptrs[1].add(idx);
                    state.0 += price * volume; // weighted sum
                    state.1 += volume; // total volume
                }
                0
            })
            .unwrap();

        let (weighted_sum, total_vol) = result.final_state;
        let vwap = weighted_sum / total_vol;
        // (100*1000 + 105*2000 + 98*1500) / (1000+2000+1500)
        // = (100000 + 210000 + 147000) / 4500 = 457000 / 4500 = 101.555...
        assert!((vwap - 101.5556).abs() < 0.001);
    }

    /// Full-loop integration test: OHLCV DataTable -> DenseKernel backtest -> verify metrics.
    ///
    /// This simulates the complete pipeline that a Shape script would execute:
    /// 1. CSV loader produces a DataTable with OHLCV columns (all f64)
    /// 2. DenseKernel runs a momentum strategy over the data
    /// 3. Results are verified: trades, P&L, slippage, commission
    ///
    /// The test uses synthetic data with known outcomes to validate correctness.
    #[test]
    fn test_full_loop_csv_to_backtest() {
        use arrow_array::{ArrayRef, Float64Array};
        use arrow_schema::{DataType, Field, Schema};
        use std::sync::Arc;

        // Step 1: Build a DataTable as a typical external loader would produce.
        // 10 bars of synthetic OHLCV data with a clear uptrend then downtrend.
        let opens = vec![
            100.0, 102.0, 104.0, 106.0, 108.0, 110.0, 108.0, 106.0, 104.0, 102.0,
        ];
        let highs = vec![
            103.0, 105.0, 107.0, 109.0, 111.0, 112.0, 110.0, 108.0, 106.0, 104.0,
        ];
        let lows = vec![
            99.0, 101.0, 103.0, 105.0, 107.0, 108.0, 106.0, 104.0, 102.0, 100.0,
        ];
        let closes = vec![
            102.0, 104.0, 106.0, 108.0, 110.0, 109.0, 107.0, 105.0, 103.0, 101.0,
        ];
        let vols = vec![
            1000.0, 1200.0, 1100.0, 1300.0, 1500.0, 1400.0, 1600.0, 1100.0, 900.0, 800.0,
        ];

        let schema = Schema::new(vec![
            Field::new("open", DataType::Float64, false),
            Field::new("high", DataType::Float64, false),
            Field::new("low", DataType::Float64, false),
            Field::new("close", DataType::Float64, false),
            Field::new("volume", DataType::Float64, false),
        ]);
        let batch = arrow_array::RecordBatch::try_new(
            Arc::new(schema),
            vec![
                Arc::new(Float64Array::from(opens)) as ArrayRef,
                Arc::new(Float64Array::from(highs)) as ArrayRef,
                Arc::new(Float64Array::from(lows)) as ArrayRef,
                Arc::new(Float64Array::from(closes.clone())) as ArrayRef,
                Arc::new(Float64Array::from(vols)) as ArrayRef,
            ],
        )
        .unwrap();
        let table = DataTable::new(batch);

        // Verify DataTable structure matches common loader output.
        assert_eq!(table.row_count(), 10);
        assert_eq!(table.column_count(), 5);
        assert_eq!(
            table.column_names(),
            vec!["open", "high", "low", "close", "volume"]
        );

        // Verify all columns produce f64 column pointers (stride == 8)
        // This is critical: DenseKernel filters col_ptrs by stride == 8
        let f64_col_count = table
            .column_ptrs()
            .iter()
            .filter(|cp| cp.stride == 8)
            .count();
        assert_eq!(f64_col_count, 5, "All OHLCV columns must be f64 (stride 8)");

        // Step 2: Run DenseKernel with a simple momentum strategy.
        // Strategy: buy when close > prev_close (uptrend), sell when close < prev_close.
        // Slippage: 5 bps. Commission: 0.1% of trade value.
        let config = DenseKernelConfig::with_warmup(table.row_count(), 1); // skip first bar
        let kernel = DenseKernel::new(config);

        #[derive(Debug)]
        struct BacktestState {
            cash: f64,
            position: f64,
            entry_price: f64,
            trades: u32,
            wins: u32,
            losses: u32,
            total_pnl: f64,
        }

        let initial = BacktestState {
            cash: 100_000.0,
            position: 0.0,
            entry_price: 0.0,
            trades: 0,
            wins: 0,
            losses: 0,
            total_pnl: 0.0,
        };

        let slippage_bps = 5.0;
        let commission_pct = 0.1;

        let result = kernel
            .run(&table, initial, |idx, col_ptrs, state| {
                // col_ptrs indices: 0=open, 1=high, 2=low, 3=close, 4=volume
                let close = unsafe { *col_ptrs[3].add(idx) };
                let prev_close = unsafe { *col_ptrs[3].add(idx - 1) };

                let signal = if close > prev_close { "buy" } else { "sell" };

                if signal == "buy" && state.position == 0.0 {
                    // Buy: calculate fill price with slippage
                    let slip = close * slippage_bps / 10_000.0;
                    let fill_price = close + slip;
                    let size = (state.cash * 0.1 / fill_price).floor(); // 10% of equity
                    if size > 0.0 {
                        let cost = fill_price * size;
                        let commission = cost * commission_pct / 100.0;
                        state.cash -= cost + commission;
                        state.position = size;
                        state.entry_price = fill_price;
                    }
                } else if signal == "sell" && state.position > 0.0 {
                    // Sell: calculate fill price with slippage
                    let slip = close * slippage_bps / 10_000.0;
                    let fill_price = close - slip;
                    let proceeds = fill_price * state.position;
                    let commission = proceeds * commission_pct / 100.0;
                    let pnl = (fill_price - state.entry_price) * state.position - commission;

                    state.cash += proceeds - commission;
                    state.total_pnl += pnl;
                    state.trades += 1;
                    if pnl > 0.0 {
                        state.wins += 1;
                    } else {
                        state.losses += 1;
                    }
                    state.position = 0.0;
                    state.entry_price = 0.0;
                }
                0 // continue
            })
            .unwrap();

        // Step 3: Verify results
        assert!(result.completed);
        assert_eq!(result.ticks_processed, 9); // 10 bars - 1 warmup

        let s = &result.final_state;

        // Should have made at least 1 trade (uptrend buy at bar 1, sell at bar 5 or 6)
        assert!(s.trades > 0, "Should have completed at least one trade");

        // Total PnL should reflect the price movement minus slippage and commission
        // The exact value depends on strategy execution, but it should be finite
        assert!(s.total_pnl.is_finite(), "P&L should be finite");

        // Cash + position value should be close to initial capital +/- P&L
        let equity = if s.position > 0.0 {
            s.cash + s.position * closes[9]
        } else {
            s.cash
        };
        assert!(equity > 0.0, "Equity should be positive");

        // Wins + losses should equal total trades
        assert_eq!(
            s.wins + s.losses,
            s.trades,
            "wins + losses should equal total trades"
        );

        // Verify slippage impact: entry price should be slightly above close for buys
        // (This is verified by the fact that total_pnl accounts for slippage)
        // If we had bought at exact close and sold at exact close, PnL would differ
    }

    /// Test that CSV-loaded DataTable with mixed column types (Int64 volume)
    /// is compatible with DenseKernel's f64 column filtering.
    ///
    /// Gap: CSV loader infers volume as Int64, but DenseKernel only sees
    /// f64 columns (stride == 8). Int64 also has stride 8, so the raw pointer
    /// access would misinterpret Int64 as f64. This documents the gap.
    #[test]
    fn test_csv_int64_column_compatibility() {
        use arrow_array::{ArrayRef, Float64Array, Int64Array};
        use arrow_schema::{DataType, Field, Schema};
        use std::sync::Arc;

        // CSV-style integer volume inference case.
        let schema = Schema::new(vec![
            Field::new("close", DataType::Float64, false),
            Field::new("volume", DataType::Int64, false), // Int64 like CSV infers
        ]);
        let closes: ArrayRef = Arc::new(Float64Array::from(vec![100.0, 105.0, 110.0]));
        let volumes: ArrayRef = Arc::new(Int64Array::from(vec![1000_i64, 2000, 3000]));
        let batch =
            arrow_array::RecordBatch::try_new(Arc::new(schema), vec![closes, volumes]).unwrap();
        let table = DataTable::new(batch);

        // Both columns have stride 8 (f64 and i64 are both 8 bytes)
        let strides: Vec<usize> = table.column_ptrs().iter().map(|cp| cp.stride).collect();
        assert_eq!(strides, vec![8, 8]);

        // DenseKernel includes both as "f64" columns, but volume is actually i64.
        // Accessing col_ptrs[1] as *const f64 when it's actually i64 data
        // will produce incorrect float values (bit reinterpretation).
        //
        // GAP: DenseKernel filters columns by stride == 8 but doesn't distinguish
        // between Float64 and Int64. Strategies must either:
        // a) Ensure CSV data uses Float64 for all numeric columns, OR
        // b) Cast Int64 columns to Float64 before simulation, OR
        // c) Check column data types and use appropriate pointer casts.
        //
        // For the f64-only close column, DenseKernel works correctly:
        let config = DenseKernelConfig::full(table.row_count());
        let kernel = DenseKernel::new(config);

        let result = kernel
            .run(&table, 0.0_f64, |idx, col_ptrs, state| {
                // Only access col_ptrs[0] (close, which IS f64)
                unsafe { *state += *col_ptrs[0].add(idx) };
                0
            })
            .unwrap();

        assert_eq!(result.final_state, 315.0); // 100+105+110
    }
}