bench: native data-layer throughput benchmark (#318)

Adds a measured benchmark for the native data layer (Task 6).

## What
- New criterion bench `crates/wickra/benches/data_layer.rs` exercising the data layer on **50 000 real BTCUSDT 1m candles** (`examples/data/btcusdt-1m.csv`):
  - **CSV parse** (`CandleReader`)
  - **Tick aggregation** → 1m candles (`TickAggregator`)
  - **Resample** 1m → 5m (`Resampler`)
- New **"4. Data layer"** section in `BENCHMARKS.md` with the measured numbers and the same "one core, FFI boundary" framing as the per-binding table.

## Measured (Rust core, Ryzen 9 9950X, median of 100 samples)
| Operation | Throughput | Per element |
|---|--:|--:|
| CSV parse | 3.0 M candles/s | 329 ns |
| Tick aggregate → 1m | 44 M ticks/s | 22.6 ns |
| Resample 1m → 5m | 234 M candles/s | 4.3 ns |

This is the same native code every binding rides through the FFI boundary characterised in §3 — the data I/O that replaces pandas / csv-parse / manual bucketing / pandas.resample, with zero third-party packages.

Run: `cargo bench -p wickra --bench data_layer`.

Verified locally: `cargo clippy -p wickra --benches --all-features -- -D warnings` clean; bench runs green.

BENCHMARKS.md

@@ -166,3 +166,41 @@ mvn -q -f bindings/java install -DskipTests \
   && mvn -q -f bindings/java/benchmarks exec:exec -Dexec.mainClass=org.wickra.benchmarks.Throughput
 Rscript bindings/r/benchmarks/throughput.R                        # R (.Call)
 ```
+
+## 4. Data layer — native I/O throughput
+
+Wickra ships its own data layer — a CSV candle reader, a tick-to-candle
+aggregator, and a timeframe resampler — so loading and reshaping market data
+needs **no third-party package** (`pandas`, `csv-parse`, manual bucketing,
+`pandas.resample`, …) in any of the ten languages. These run on the same Rust
+core as the indicators, so every binding reaches these speeds minus the FFI
+boundary characterised in section 3 (a `read` / `push` / `flush` call crosses the
+boundary once per batch, like `batch`, so bindings land close to the core).
+
+Rust core, 50 000 real BTCUSDT one-minute candles
+(`examples/data/btcusdt-1m.csv`), median of 100 samples, on the reference machine
+(Windows 11, AMD Ryzen 9 9950X):
+
+| Operation                          | Throughput          | Per element |
+|------------------------------------|--------------------:|------------:|
+| CSV parse (`CandleReader`)         |   3.0 M candles/s   |      329 ns |
+| Tick aggregate → 1m (`TickAggregator`) |  44 M ticks/s   |     22.6 ns |
+| Resample 1m → 5m (`Resampler`)     | 234 M candles/s     |      4.3 ns |
+
+Reading and validating a 50 000-row CSV into typed candles takes ~16 ms;
+aggregating 50 000 ticks into one-minute bars ~1.1 ms; resampling 50 000
+one-minute candles to five-minute bars ~0.2 ms. CSV parsing is the floor because
+it does the most per row (UTF-8 scan, field split, six `f64` parses, finiteness
+checks); aggregation and resampling are pure arithmetic over already-typed
+candles.
+
+The live and historical Binance feeds (`BinanceFeed`, `fetch_binance_klines`) are
+network-bound — their throughput is set by the exchange and the socket, not by
+Wickra — so they are not micro-benchmarked here; the relevant Wickra cost is the
+per-event parse, which is the same arithmetic measured above.
+
+Run it with:
+
+```bash
+cargo bench -p wickra --bench data_layer
+```

crates/wickra/Cargo.toml

@@ -36,3 +36,7 @@ wickra-data = { path = "../wickra-data" }
 [[bench]]
 name = "indicators"
 harness = false
+
+[[bench]]
+name = "data_layer"
+harness = false

crates/wickra/benches/data_layer.rs

@@ -0,0 +1,119 @@
+//! Throughput microbenchmarks for the native data layer: CSV parsing, tick
+//! aggregation, and resampling.
+//!
+//! Run with:
+//! ```text
+//! cargo bench -p wickra --bench data_layer
+//! ```
+//!
+//! These exercise `wickra-data` — the same native code every binding rides
+//! through the FFI boundary characterised in `BENCHMARKS.md` §3. It is what
+//! replaces `pandas` / `csv-parse` / manual tick bucketing / `pandas.resample`
+//! in the nine non-Rust languages, so the numbers here are the upper bound a
+//! binding can reach for "load a CSV, roll ticks into candles, resample a
+//! series" without pulling in a single third-party package.
+//!
+//! The dataset is the checked-in `examples/data/btcusdt-1m.csv` (50 000 real
+//! BTCUSDT one-minute candles). Regenerate it with
+//! `cargo run -p wickra-examples --bin fetch_btcusdt`.
+
+use criterion::{criterion_group, criterion_main, Criterion, Throughput};
+use std::hint::black_box;
+use wickra::{Candle, Tick};
+use wickra_data::{
+    aggregator::{TickAggregator, Timeframe},
+    csv::CandleReader,
+    resample::Resampler,
+};
+
+const DATASET: &str = concat!(
+    env!("CARGO_MANIFEST_DIR"),
+    "/../../examples/data/btcusdt-1m.csv"
+);
+const ONE_MINUTE_MS: i64 = 60_000;
+
+fn dataset_bytes() -> Vec<u8> {
+    std::fs::read(DATASET).unwrap_or_else(|e| {
+        panic!(
+            "could not read the benchmark dataset {DATASET}: {e}\n\
+             generate it with `cargo run -p wickra-examples --bin fetch_btcusdt`"
+        )
+    })
+}
+
+fn load_candles() -> Vec<Candle> {
+    CandleReader::from_reader(dataset_bytes().as_slice())
+        .unwrap()
+        .read_all()
+        .unwrap()
+}
+
+/// CSV bytes -> `Vec<Candle>`. Throughput is candles (rows) parsed per second.
+fn bench_csv_read(c: &mut Criterion) {
+    let bytes = dataset_bytes();
+    let rows = load_candles().len() as u64;
+    let mut group = c.benchmark_group("data_layer/csv_read");
+    group.throughput(Throughput::Elements(rows));
+    group.bench_function("btcusdt_1m", |b| {
+        b.iter(|| {
+            let candles = CandleReader::from_reader(black_box(bytes.as_slice()))
+                .unwrap()
+                .read_all()
+                .unwrap();
+            black_box(candles.len())
+        });
+    });
+    group.finish();
+}
+
+/// Ticks -> one-minute candles. Throughput is ticks aggregated per second.
+fn bench_tick_aggregate(c: &mut Criterion) {
+    let ticks: Vec<Tick> = load_candles()
+        .iter()
+        .map(|candle| Tick::new(candle.close, candle.volume, candle.timestamp).unwrap())
+        .collect();
+    let mut group = c.benchmark_group("data_layer/tick_aggregate");
+    group.throughput(Throughput::Elements(ticks.len() as u64));
+    group.bench_function("1m_buckets", |b| {
+        b.iter(|| {
+            let mut agg = TickAggregator::new(Timeframe::millis(ONE_MINUTE_MS).unwrap());
+            let mut emitted = 0usize;
+            for tick in &ticks {
+                emitted += agg.push(black_box(*tick)).unwrap().len();
+            }
+            black_box(emitted)
+        });
+    });
+    group.finish();
+}
+
+/// One-minute candles -> five-minute candles. Throughput is input candles per second.
+fn bench_resample(c: &mut Criterion) {
+    let candles = load_candles();
+    let mut group = c.benchmark_group("data_layer/resample");
+    group.throughput(Throughput::Elements(candles.len() as u64));
+    group.bench_function("1m_to_5m", |b| {
+        b.iter(|| {
+            let mut resampler = Resampler::new(Timeframe::millis(5 * ONE_MINUTE_MS).unwrap());
+            let mut emitted = 0usize;
+            for candle in &candles {
+                if resampler.push(black_box(*candle)).unwrap().is_some() {
+                    emitted += 1;
+                }
+            }
+            if resampler.flush().unwrap().is_some() {
+                emitted += 1;
+            }
+            black_box(emitted)
+        });
+    });
+    group.finish();
+}
+
+criterion_group!(
+    benches,
+    bench_csv_read,
+    bench_tick_aggregate,
+    bench_resample
+);
+criterion_main!(benches);