feat: trade-flow microstructure indicators (part 2 of 4) (#113)

* feat(core): add 3 trade-flow microstructure indicators

SignedVolume (per-trade size signed by aggressor), CumulativeVolumeDelta
(running signed-volume total), and TradeImbalance (rolling buy/sell volume
imbalance over a trade window). All consume the Trade type, with full unit
coverage. Extends the Microstructure family.

* feat(bindings): expose trade-flow microstructure indicators

Python, Node and WASM bindings for SignedVolume, CumulativeVolumeDelta and
TradeImbalance. Each takes a trade via update(price, size, is_buy); Python and
Node expose a batch over three parallel arrays, WASM exposes per-trade update.
Regenerates node index.d.ts/.js.

* test(bindings,fuzz,bench): cover trade-flow microstructure indicators

Python and Node: reference values, streaming-vs-batch, lifecycle/repr and input
validation (zero window, negative size, non-positive price, mismatched batch
lengths). New indicator_update_trade fuzz target. Synthetic trade-tape benches
(signed_volume cheapest, trade_imbalance windowed/expensive).

* docs: add trade-flow indicators + bump counter to 227

README Microstructure family row gains signed volume / CVD / trade imbalance and
the counter goes 224 -> 227; CHANGELOG records the trade-flow indicators.

bindings/python/python/wickra/__init__.py

@@ -246,6 +246,10 @@ from ._wickra import (
     OrderBookImbalanceFull,
     Microprice,
     QuotedSpread,
+    # Microstructure: trade flow
+    SignedVolume,
+    CumulativeVolumeDelta,
+    TradeImbalance,
     # Risk / Performance
     SharpeRatio,
     SortinoRatio,
@@ -489,6 +493,10 @@ __all__ = [
     "OrderBookImbalanceFull",
     "Microprice",
     "QuotedSpread",
+    # Microstructure: trade flow
+    "SignedVolume",
+    "CumulativeVolumeDelta",
+    "TradeImbalance",
     # Risk / Performance
     "SharpeRatio",
     "SortinoRatio",

bindings/python/src/lib.rs

@@ -11771,6 +11771,137 @@ impl PyOrderBookImbalanceTopN {
     }
 }
 
+// ============================== Microstructure: Trade Flow ==============================
+//
+// Trade-flow indicators consume a trade tape rather than OHLCV. Streaming
+// `update(price, size, is_buy)` takes one trade (`is_buy=True` for a
+// buyer-initiated trade); `batch` takes three equal-length arrays.
+
+fn build_trade(price: f64, size: f64, is_buy: bool) -> PyResult<wc::Trade> {
+    let side = if is_buy {
+        wc::Side::Buy
+    } else {
+        wc::Side::Sell
+    };
+    wc::Trade::new(price, size, side, 0).map_err(map_err)
+}
+
+macro_rules! py_trade_indicator {
+    ($name:ident, $inner:ty, $repr:expr) => {
+        #[pyclass(name = $repr, module = "wickra._wickra", skip_from_py_object)]
+        #[derive(Clone)]
+        struct $name {
+            inner: $inner,
+        }
+
+        #[pymethods]
+        impl $name {
+            #[new]
+            fn new() -> Self {
+                Self {
+                    inner: <$inner>::new(),
+                }
+            }
+            fn update(&mut self, price: f64, size: f64, is_buy: bool) -> PyResult<Option<f64>> {
+                Ok(self.inner.update(build_trade(price, size, is_buy)?))
+            }
+            fn batch<'py>(
+                &mut self,
+                py: Python<'py>,
+                price: Vec<f64>,
+                size: Vec<f64>,
+                is_buy: Vec<bool>,
+            ) -> PyResult<Bound<'py, PyArray1<f64>>> {
+                if price.len() != size.len() || size.len() != is_buy.len() {
+                    return Err(PyValueError::new_err(
+                        "price, size, is_buy must be equal length",
+                    ));
+                }
+                let mut out = Vec::with_capacity(price.len());
+                for i in 0..price.len() {
+                    let trade = build_trade(price[i], size[i], is_buy[i])?;
+                    out.push(self.inner.update(trade).unwrap_or(f64::NAN));
+                }
+                Ok(out.into_pyarray(py))
+            }
+            fn reset(&mut self) {
+                self.inner.reset();
+            }
+            fn is_ready(&self) -> bool {
+                self.inner.is_ready()
+            }
+            fn warmup_period(&self) -> usize {
+                self.inner.warmup_period()
+            }
+            fn __repr__(&self) -> String {
+                format!("{}()", $repr)
+            }
+        }
+    };
+}
+
+py_trade_indicator!(PySignedVolume, wc::SignedVolume, "SignedVolume");
+py_trade_indicator!(
+    PyCumulativeVolumeDelta,
+    wc::CumulativeVolumeDelta,
+    "CumulativeVolumeDelta"
+);
+
+// Trade imbalance carries a `window` parameter, so it is hand-written.
+#[pyclass(
+    name = "TradeImbalance",
+    module = "wickra._wickra",
+    skip_from_py_object
+)]
+#[derive(Clone)]
+struct PyTradeImbalance {
+    inner: wc::TradeImbalance,
+}
+
+#[pymethods]
+impl PyTradeImbalance {
+    #[new]
+    fn new(window: usize) -> PyResult<Self> {
+        Ok(Self {
+            inner: wc::TradeImbalance::new(window).map_err(map_err)?,
+        })
+    }
+    fn update(&mut self, price: f64, size: f64, is_buy: bool) -> PyResult<Option<f64>> {
+        Ok(self.inner.update(build_trade(price, size, is_buy)?))
+    }
+    fn batch<'py>(
+        &mut self,
+        py: Python<'py>,
+        price: Vec<f64>,
+        size: Vec<f64>,
+        is_buy: Vec<bool>,
+    ) -> PyResult<Bound<'py, PyArray1<f64>>> {
+        if price.len() != size.len() || size.len() != is_buy.len() {
+            return Err(PyValueError::new_err(
+                "price, size, is_buy must be equal length",
+            ));
+        }
+        let mut out = Vec::with_capacity(price.len());
+        for i in 0..price.len() {
+            let trade = build_trade(price[i], size[i], is_buy[i])?;
+            out.push(self.inner.update(trade).unwrap_or(f64::NAN));
+        }
+        Ok(out.into_pyarray(py))
+    }
+    fn reset(&mut self) {
+        self.inner.reset();
+    }
+    fn is_ready(&self) -> bool {
+        self.inner.is_ready()
+    }
+    fn warmup_period(&self) -> usize {
+        self.inner.warmup_period()
+    }
+    fn __repr__(&self) -> String {
+        format!("TradeImbalance(window={})", self.inner.window())
+    }
+}
+
 // ============================== Family 15: Risk / Performance ==============================
 
 #[pyclass(name = "SharpeRatio", module = "wickra._wickra", skip_from_py_object)]
@@ -12882,6 +13013,10 @@ fn _wickra(_py: Python<'_>, m: &Bound<'_, PyModule>) -> PyResult<()> {
     m.add_class::<PyOrderBookImbalanceFull>()?;
     m.add_class::<PyMicroprice>()?;
     m.add_class::<PyQuotedSpread>()?;
+    // Microstructure: trade flow.
+    m.add_class::<PySignedVolume>()?;
+    m.add_class::<PyCumulativeVolumeDelta>()?;
+    m.add_class::<PyTradeImbalance>()?;
     // Family 15: Risk / Performance metrics.
     m.add_class::<PySharpeRatio>()?;
     m.add_class::<PySortinoRatio>()?;

bindings/python/tests/test_input_validation.py

@@ -191,3 +191,23 @@ def test_orderbook_misordered_levels_raise():
     # Bids must be strictly descending in price.
     with pytest.raises(ValueError):
         ta.OrderBookImbalanceFull().update([99.0, 100.0], [1.0, 1.0], [101.0], [1.0])
+
+
+def test_trade_imbalance_zero_window_raises():
+    with pytest.raises(ValueError):
+        ta.TradeImbalance(0)
+
+
+def test_trade_negative_size_raises():
+    with pytest.raises(ValueError):
+        ta.SignedVolume().update(100.0, -1.0, True)
+
+
+def test_trade_non_positive_price_raises():
+    with pytest.raises(ValueError):
+        ta.CumulativeVolumeDelta().update(0.0, 1.0, True)
+
+
+def test_trade_batch_unequal_lengths_raise():
+    with pytest.raises(ValueError):
+        ta.SignedVolume().batch([100.0, 100.0], [1.0], [True, False])

bindings/python/tests/test_known_values.py

@@ -870,3 +870,22 @@ def test_quoted_spread_reference_value():
     # spread 1.0, mid 100.5 -> 1 / 100.5 * 10_000 ≈ 99.5025 bps.
     qs = ta.QuotedSpread()
     assert qs.update([100.0], [1.0], [101.0], [1.0]) == pytest.approx(99.50248756, abs=1e-6)
+
+
+def test_signed_volume_reference_values():
+    assert ta.SignedVolume().update(100.0, 2.0, True) == pytest.approx(2.0)
+    assert ta.SignedVolume().update(100.0, 3.0, False) == pytest.approx(-3.0)
+
+
+def test_cumulative_volume_delta_reference_values():
+    cvd = ta.CumulativeVolumeDelta()
+    assert cvd.update(100.0, 5.0, True) == pytest.approx(5.0)
+    assert cvd.update(100.0, 2.0, False) == pytest.approx(3.0)
+    assert cvd.update(100.0, 4.0, False) == pytest.approx(-1.0)
+
+
+def test_trade_imbalance_reference_value():
+    ti = ta.TradeImbalance(2)
+    assert ti.update(100.0, 3.0, True) is None  # warming up
+    # Window full: buyVol 3, sellVol 1 -> (3 - 1) / 4 = 0.5.
+    assert ti.update(100.0, 1.0, False) == pytest.approx(0.5)

bindings/python/tests/test_lifecycle.py

@@ -150,3 +150,25 @@ def test_orderbook_lifecycle():
 
 def test_orderbook_topn_repr():
     assert repr(ta.OrderBookImbalanceTopN(5)) == "OrderBookImbalanceTopN(levels=5)"
+
+
+def test_tradeflow_lifecycle():
+    for ind in [ta.SignedVolume(), ta.CumulativeVolumeDelta()]:
+        assert ind.warmup_period() == 1
+        assert not ind.is_ready()
+        ind.update(100.0, 1.0, True)
+        assert ind.is_ready()
+        ind.reset()
+        assert not ind.is_ready()
+
+
+def test_trade_imbalance_lifecycle_and_repr():
+    ti = ta.TradeImbalance(3)
+    assert ti.warmup_period() == 3
+    assert not ti.is_ready()
+    for _ in range(3):
+        ti.update(100.0, 1.0, True)
+    assert ti.is_ready()
+    ti.reset()
+    assert not ti.is_ready()
+    assert repr(ta.TradeImbalance(4)) == "TradeImbalance(window=4)"

bindings/python/tests/test_new_indicators.py

@@ -1898,3 +1898,23 @@ def test_orderbook_indicators_streaming_equals_batch():
         )
         assert batch.shape == (len(snaps),)
         assert _eq_nan(batch, streamed)
+
+
+def test_tradeflow_indicators_streaming_equals_batch():
+    n = 40
+    price = np.full(n, 100.0)
+    size = np.array([1.0 + (i % 5) for i in range(n)], dtype=np.float64)
+    is_buy = [i % 2 == 0 for i in range(n)]
+    for make in (
+        ta.SignedVolume,
+        ta.CumulativeVolumeDelta,
+        lambda: ta.TradeImbalance(5),
+    ):
+        batch = make().batch(price, size, is_buy)
+        streamer = make()
+        streamed = np.array(
+            [streamer.update(price[i], size[i], is_buy[i]) for i in range(n)],
+            dtype=np.float64,
+        )
+        assert batch.shape == (n,)
+        assert _eq_nan(batch, streamed)

bindings/python/tests/test_smoke.py

@@ -119,3 +119,19 @@ def test_orderbook_batch_returns_one_value_per_snapshot():
     out = ta.OrderBookImbalanceTop1().batch(snapshots)
     assert out.shape == (5,)
     assert out.dtype == np.float64
+
+
+def test_tradeflow_indicators_construct_and_emit():
+    # SignedVolume and CVD emit from the first trade; TradeImbalance(1) too.
+    assert isinstance(ta.SignedVolume().update(100.0, 2.0, True), float)
+    assert isinstance(ta.CumulativeVolumeDelta().update(100.0, 2.0, True), float)
+    assert isinstance(ta.TradeImbalance(1).update(100.0, 2.0, True), float)
+
+
+def test_tradeflow_batch_returns_one_value_per_trade():
+    price = np.full(6, 100.0)
+    size = np.array([1.0, 2.0, 3.0, 1.0, 2.0, 3.0])
+    is_buy = [True, False, True, False, True, False]
+    out = ta.CumulativeVolumeDelta().batch(price, size, is_buy)
+    assert out.shape == (6,)
+    assert out.dtype == np.float64

bindings/python/tests/test_streaming_vs_batch.py

@@ -217,3 +217,17 @@ def test_orderbook_streaming_matches_batch():
     streamer = ta.Microprice()
     streamed = np.array([streamer.update(*snap) for snap in snaps], dtype=np.float64)
     assert _equal_with_nan(batch, streamed)
+
+
+def test_tradeflow_streaming_matches_batch():
+    n = 30
+    price = np.full(n, 100.0)
+    size = np.array([1.0 + (i % 4) for i in range(n)], dtype=np.float64)
+    is_buy = [i % 3 != 0 for i in range(n)]
+    batch = ta.CumulativeVolumeDelta().batch(price, size, is_buy)
+    streamer = ta.CumulativeVolumeDelta()
+    streamed = np.array(
+        [streamer.update(price[i], size[i], is_buy[i]) for i in range(n)],
+        dtype=np.float64,
+    )
+    assert _equal_with_nan(batch, streamed)