From 43b0b267368f3d7eab9dc2eb54586ac14cab8283 Mon Sep 17 00:00:00 2001 From: kingchenc Date: Sat, 23 May 2026 00:18:46 +0200 Subject: [PATCH] examples: add parallel-assets demos for Rust and Node MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Python's parallel_assets.py demoed GIL-release multi-core throughput; Rust and Node both lacked a sibling that shows their own native parallelism. Close the gap with two real, runnable examples. * examples/rust/src/bin/parallel_assets.rs — synthesises an (assets, bars) panel with a deterministic per-asset LCG, runs a serial baseline, then `Sma::batch_parallel` / `Rsi::batch_parallel` via rayon, asserts the two outputs are element-wise identical and prints the speedup. Toggle indicator with `--indicator sma|rsi`. * examples/node/parallel_assets.js — same shape, but the parallel run is a `worker_threads` pool that re-loads the native binding in each worker. Each worker computes the last non-null indicator value for its slice; the main thread aggregates and verifies serial == parallel per asset. Both examples report timings and the serial-vs-parallel sanity check passes. Defaults (200 × 5000) keep the example fast on dev hardware; larger `--assets`/`--bars` is where the speedup numbers move (Node's worker spawn cost dominates the smallest sizes, which is honest and educational). examples/README.md gains the two new rows. --- examples/README.md | 2 + examples/node/parallel_assets.js | 169 +++++++++++++++++++++++ examples/rust/src/bin/parallel_assets.rs | 141 +++++++++++++++++++ 3 files changed, 312 insertions(+) create mode 100644 examples/node/parallel_assets.js create mode 100644 examples/rust/src/bin/parallel_assets.rs diff --git a/examples/README.md b/examples/README.md index bc749596..dca644c7 100644 --- a/examples/README.md +++ b/examples/README.md @@ -14,6 +14,7 @@ The Rust examples live in the `wickra-examples` workspace member crate. | `streaming.rs` | Feed a synthetic price series through SMA / EMA / RSI / MACD tick by tick. | `cargo run -p wickra-examples --bin streaming` | | `backtest.rs` | Compute a basket of indicators over an OHLCV CSV and print a summary. | `cargo run -p wickra-examples --bin backtest -- ` | | `multi_timeframe.rs` | Resample a 1-minute CSV via wickra-data and print indicators per timeframe. | `cargo run -p wickra-examples --bin multi_timeframe` | +| `parallel_assets.rs` | Serial vs `BatchExt::batch_parallel` (rayon) over a synthetic panel, with speedup. | `cargo run --release -p wickra-examples --bin parallel_assets -- --assets 200 --bars 5000` | | `fetch_btcusdt.rs` | Download real BTCUSDT klines from the Binance REST API into `examples/data/`. | `cargo run -p wickra-examples --bin fetch_btcusdt` | | `live_binance.rs` | Stream live Binance klines through an indicator over a resilient WebSocket. | `cargo run -p wickra-examples --bin live_binance` | @@ -43,6 +44,7 @@ cd ../../examples/node && npm install # links wickra + installs `ws` | `streaming.js` | Feed a synthetic price series through several indicators tick by tick. | `node streaming.js` | | `backtest.js` | Basket of indicators over an OHLCV CSV; defaults to the bundled BTCUSDT daily dataset. | `node backtest.js [ohlcv.csv]` | | `multi_timeframe.js` | Roll a 1-minute CSV up to 5m / 15m / 1h / 4h / 1d and print indicators per timeframe. | `node multi_timeframe.js [path/to/1m.csv]` | +| `parallel_assets.js` | Serial vs `worker_threads` pool over a synthetic panel, with speedup. | `node parallel_assets.js --assets 200 --bars 5000` | | `live_trading.js` | Live Binance feed → RSI / MACD / Bollinger → signals. | `node live_trading.js --symbol BTCUSDT --interval 1m` | ## WebAssembly — `examples/wasm/` diff --git a/examples/node/parallel_assets.js b/examples/node/parallel_assets.js new file mode 100644 index 00000000..e3525b2a --- /dev/null +++ b/examples/node/parallel_assets.js @@ -0,0 +1,169 @@ +// Parallel multi-asset indicator computation via Node.js worker_threads. +// +// Builds a synthetic (assets, bars) panel, runs a serial baseline on the +// main thread, then dispatches the same workload to a pool of workers +// (each loading its own copy of the native wickra binding) and reports +// the speedup. The Node counterpart of examples/python/parallel_assets.py. +// +// Run with: +// node parallel_assets.js [--assets N] [--bars M] +// [--indicator sma|rsi] [--workers W] + +const { Worker, isMainThread, workerData, parentPort } = require('node:worker_threads'); +const os = require('node:os'); + +// Deterministic LCG matching the Rust sibling so a side-by-side serial +// computation on both languages produces visibly comparable timings. +function makeRng(seed) { + let state = seed >>> 0; + return () => { + state = (Math.imul(state, 1103515245) + 12345) & 0x7fffffff; + return state / 0x7fffffff; + }; +} + +function synthesizePanel(nAssets, nBars) { + const panel = new Array(nAssets); + for (let a = 0; a < nAssets; a++) { + const rng = makeRng((1234567 + Math.imul(a, 2654435761)) >>> 0); + const series = new Float64Array(nBars); + let price = 100.0; + for (let i = 0; i < nBars; i++) { + price += (rng() - 0.5) * 0.4; + series[i] = price; + } + panel[a] = series; + } + return panel; +} + +// Compute the last non-null indicator output for one price series. Kept +// tiny so the same code runs on the main thread (serial baseline) and +// inside each worker (parallel run). +function lastValue(prices, indicator, wickra) { + const ind = indicator === 'sma' ? new wickra.SMA(14) : new wickra.RSI(14); + let last = null; + for (let i = 0; i < prices.length; i++) { + const v = ind.update(prices[i]); + if (v !== null) last = v; + } + return last; +} + +if (!isMainThread) { + // ---------------- worker side ---------------- + const wickra = require('wickra'); + const { panelSlice, indicator } = workerData; + const results = new Array(panelSlice.length); + for (let i = 0; i < panelSlice.length; i++) { + results[i] = lastValue(panelSlice[i], indicator, wickra); + } + parentPort.postMessage(results); +} else { + // ---------------- main side ---------------- + const wickra = require('wickra'); + + function parseArgs(argv) { + const args = { + assets: 200, + bars: 5000, + indicator: 'sma', + workers: Math.max(1, os.cpus().length), + }; + for (let i = 0; i < argv.length; i++) { + const k = argv[i]; + if (k === '--assets') args.assets = Number(argv[++i]); + else if (k === '--bars') args.bars = Number(argv[++i]); + else if (k === '--indicator') args.indicator = argv[++i]; + else if (k === '--workers') args.workers = Number(argv[++i]); + else throw new Error(`unexpected argument: ${k}`); + } + if (!Number.isInteger(args.assets) || args.assets <= 0) { + throw new Error('--assets must be a positive integer'); + } + if (!Number.isInteger(args.bars) || args.bars <= 0) { + throw new Error('--bars must be a positive integer'); + } + if (args.indicator !== 'sma' && args.indicator !== 'rsi') { + throw new Error("--indicator: expected 'sma' or 'rsi'"); + } + if (!Number.isInteger(args.workers) || args.workers <= 0) { + throw new Error('--workers must be a positive integer'); + } + return args; + } + + let args; + try { + args = parseArgs(process.argv.slice(2)); + } catch (err) { + console.error(`error: ${err.message}`); + process.exit(2); + } + + console.log(`Generating ${args.assets}×${args.bars} synthetic panel…`); + const panel = synthesizePanel(args.assets, args.bars); + + // Serial baseline (main thread). + let t0 = process.hrtime.bigint(); + const serial = new Array(args.assets); + for (let a = 0; a < args.assets; a++) { + serial[a] = lastValue(panel[a], args.indicator, wickra); + } + const tSerial = Number(process.hrtime.bigint() - t0) / 1e9; + console.log( + `Serial: ${tSerial.toFixed(3).padStart(8)} s (${args.assets} assets, indicator=${args.indicator})`, + ); + + // Parallel via worker_threads. + const workerCount = Math.max(1, Math.min(args.workers, args.assets)); + const sliceSize = Math.ceil(args.assets / workerCount); + t0 = process.hrtime.bigint(); + const promises = []; + for (let w = 0; w < workerCount; w++) { + const start = w * sliceSize; + const end = Math.min(start + sliceSize, args.assets); + if (start >= end) continue; + const panelSlice = panel.slice(start, end); + promises.push( + new Promise((resolve, reject) => { + const worker = new Worker(__filename, { + workerData: { panelSlice, indicator: args.indicator }, + }); + worker.on('message', (msg) => resolve({ start, results: msg })); + worker.on('error', reject); + worker.on('exit', (code) => { + if (code !== 0) reject(new Error(`worker exited with code ${code}`)); + }); + }), + ); + } + + Promise.all(promises) + .then((chunks) => { + const parallel = new Array(args.assets); + for (const { start, results } of chunks) { + for (let j = 0; j < results.length; j++) parallel[start + j] = results[j]; + } + const tParallel = Number(process.hrtime.bigint() - t0) / 1e9; + const speedup = tSerial / Math.max(tParallel, 1e-9); + console.log( + `Parallel: ${tParallel.toFixed(3).padStart(8)} s ` + + `(workers=${workerCount}, speedup ~${speedup.toFixed(2)}x)`, + ); + + for (let i = 0; i < args.assets; i++) { + const s = serial[i]; + const p = parallel[i]; + if (s !== p && !(s === null && p === null)) { + console.error(`mismatch at asset ${i}: serial=${s} parallel=${p}`); + process.exit(1); + } + } + console.log('Parallel results match serial results — OK.'); + }) + .catch((err) => { + console.error(`parallel run failed: ${err.message}`); + process.exit(1); + }); +} diff --git a/examples/rust/src/bin/parallel_assets.rs b/examples/rust/src/bin/parallel_assets.rs new file mode 100644 index 00000000..3794ec45 --- /dev/null +++ b/examples/rust/src/bin/parallel_assets.rs @@ -0,0 +1,141 @@ +//! Parallel multi-asset indicator computation via rayon. +//! +//! Builds a synthetic `(assets, bars)` panel, runs a serial baseline, then +//! runs the same computation with [`BatchExt::batch_parallel`] (which is +//! gated behind wickra's default `parallel` feature) and reports the +//! speedup. The Rust counterpart of `examples/python/parallel_assets.py`. +//! +//! Run with: +//! ```text +//! cargo run --release -p wickra-examples --bin parallel_assets -- \ +//! --assets 200 --bars 5000 +//! ``` + +use std::env; +use std::time::Instant; + +use wickra::{BatchExt, Rsi, Sma}; + +#[derive(Debug, Clone)] +enum Which { + Sma, + Rsi, +} + +impl Which { + fn parse(s: &str) -> Result> { + match s { + "sma" => Ok(Self::Sma), + "rsi" => Ok(Self::Rsi), + other => Err(format!("--indicator: expected 'sma' or 'rsi', got {other}").into()), + } + } +} + +struct Args { + assets: usize, + bars: usize, + indicator: Which, +} + +fn parse_args() -> Result> { + let mut assets = 200_usize; + let mut bars = 5_000_usize; + let mut indicator = Which::Sma; + let mut it = env::args().skip(1); + while let Some(arg) = it.next() { + match arg.as_str() { + "--assets" => { + assets = it.next().ok_or("--assets needs a value")?.parse()?; + } + "--bars" => { + bars = it.next().ok_or("--bars needs a value")?.parse()?; + } + "--indicator" => { + indicator = Which::parse(&it.next().ok_or("--indicator needs a value")?)?; + } + other => return Err(format!("unexpected argument: {other}").into()), + } + } + if assets == 0 || bars == 0 { + return Err("--assets and --bars must be positive".into()); + } + Ok(Args { + assets, + bars, + indicator, + }) +} + +/// Deterministic synthetic `(assets, bars)` panel. Each asset uses an +/// independent LCG seed so the series are uncorrelated but reproducible. +fn synthesize_panel(n_assets: usize, n_bars: usize) -> Vec> { + let mut series = Vec::with_capacity(n_assets); + for a in 0..n_assets { + let mut s = Vec::with_capacity(n_bars); + let mut price = 100.0_f64; + let mut state: u32 = 1_234_567_u32 + .wrapping_add(a as u32) + .wrapping_mul(2_654_435_761); + for _ in 0..n_bars { + state = state.wrapping_mul(1_103_515_245).wrapping_add(12_345) & 0x7FFF_FFFF; + let r = f64::from(state) / f64::from(0x7FFF_FFFF_u32); + price += (r - 0.5) * 0.4; + s.push(price); + } + series.push(s); + } + series +} + +fn run_serial(panel: &[Vec], indicator: &Which) -> Vec>> { + panel + .iter() + .map(|prices| match indicator { + Which::Sma => Sma::new(14).expect("SMA(14)").batch(prices), + Which::Rsi => Rsi::new(14).expect("RSI(14)").batch(prices), + }) + .collect() +} + +fn run_parallel(panel: &[Vec], indicator: &Which) -> Vec>> { + match indicator { + Which::Sma => Sma::batch_parallel(panel, || Sma::new(14).expect("SMA(14)")), + Which::Rsi => Rsi::batch_parallel(panel, || Rsi::new(14).expect("RSI(14)")), + } +} + +fn main() -> Result<(), Box> { + let args = parse_args()?; + println!("Generating {}×{} synthetic panel…", args.assets, args.bars); + let panel = synthesize_panel(args.assets, args.bars); + + let t0 = Instant::now(); + let serial = run_serial(&panel, &args.indicator); + let t_serial = t0.elapsed(); + println!( + "Serial: {:>8.3} s ({} assets, indicator={:?})", + t_serial.as_secs_f64(), + args.assets, + args.indicator + ); + + let t0 = Instant::now(); + let parallel = run_parallel(&panel, &args.indicator); + let t_parallel = t0.elapsed(); + let speedup = t_serial.as_secs_f64() / t_parallel.as_secs_f64().max(1e-9); + println!( + "Parallel: {:>8.3} s (rayon, speedup ~{:.2}x)", + t_parallel.as_secs_f64(), + speedup + ); + + assert_eq!(serial.len(), parallel.len(), "asset count mismatch"); + for (i, (s, p)) in serial.iter().zip(parallel.iter()).enumerate() { + if s != p { + return Err(format!("asset {i}: serial and parallel results disagree").into()); + } + } + println!("Parallel results match serial results — OK."); + Ok(()) +}