feat: footprint microstructure indicator (part 4 of 4) (#123)

bindings/python/python/wickra/__init__.py

@@ -255,6 +255,8 @@ from ._wickra import (
     EffectiveSpread,
     RealizedSpread,
     KylesLambda,
+    # Microstructure: footprint
+    Footprint,
     # Risk / Performance
     SharpeRatio,
     SortinoRatio,
@@ -507,6 +509,8 @@ __all__ = [
     "EffectiveSpread",
     "RealizedSpread",
     "KylesLambda",
+    # Microstructure: footprint
+    "Footprint",
     # Risk / Performance
     "SharpeRatio",
     "SortinoRatio",

bindings/python/src/lib.rs

@@ -12095,6 +12095,93 @@ impl PyKylesLambda {
     }
 }
 
+// ============================== Microstructure: Footprint ==============================
+//
+// Footprint is a multi-output, variable-length indicator: each `update(price,
+// size, is_buy)` returns the full bar footprint accumulated since the last
+// `reset()` as a `(k, 3)` array with columns `[price, bid_vol, ask_vol]`, one
+// row per touched price bucket (sorted ascending by price). `batch` returns a
+// list of such arrays, one per trade.
+
+fn footprint_to_array<'py>(
+    py: Python<'py>,
+    out: &wc::FootprintOutput,
+) -> Bound<'py, PyArray2<f64>> {
+    let rows = out.levels.len();
+    let mut data = Vec::with_capacity(rows * 3);
+    for level in &out.levels {
+        data.push(level.price);
+        data.push(level.bid_vol);
+        data.push(level.ask_vol);
+    }
+    numpy::ndarray::Array2::from_shape_vec((rows, 3), data)
+        .expect("shape consistent")
+        .into_pyarray(py)
+}
+
+#[pyclass(name = "Footprint", module = "wickra._wickra", skip_from_py_object)]
+#[derive(Clone)]
+struct PyFootprint {
+    inner: wc::Footprint,
+}
+
+#[pymethods]
+impl PyFootprint {
+    #[new]
+    fn new(tick_size: f64) -> PyResult<Self> {
+        Ok(Self {
+            inner: wc::Footprint::new(tick_size).map_err(map_err)?,
+        })
+    }
+    fn update<'py>(
+        &mut self,
+        py: Python<'py>,
+        price: f64,
+        size: f64,
+        is_buy: bool,
+    ) -> PyResult<Bound<'py, PyArray2<f64>>> {
+        let out = self
+            .inner
+            .update(build_trade(price, size, is_buy)?)
+            .expect("footprint emits on every trade");
+        Ok(footprint_to_array(py, &out))
+    }
+    fn batch<'py>(
+        &mut self,
+        py: Python<'py>,
+        price: Vec<f64>,
+        size: Vec<f64>,
+        is_buy: Vec<bool>,
+    ) -> PyResult<Vec<Bound<'py, PyArray2<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 snapshot = self
+                .inner
+                .update(build_trade(price[i], size[i], is_buy[i])?)
+                .expect("footprint emits on every trade");
+            out.push(footprint_to_array(py, &snapshot));
+        }
+        Ok(out)
+    }
+    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!("Footprint(tick_size={})", self.inner.tick_size())
+    }
+}
+
 // ============================== Family 15: Risk / Performance ==============================
 
 #[pyclass(name = "SharpeRatio", module = "wickra._wickra", skip_from_py_object)]
@@ -13215,6 +13302,8 @@ fn _wickra(_py: Python<'_>, m: &Bound<'_, PyModule>) -> PyResult<()> {
     m.add_class::<PyEffectiveSpread>()?;
     m.add_class::<PyRealizedSpread>()?;
     m.add_class::<PyKylesLambda>()?;
+    // Microstructure: footprint.
+    m.add_class::<PyFootprint>()?;
     // Family 15: Risk / Performance metrics.
     m.add_class::<PySharpeRatio>()?;
     m.add_class::<PySortinoRatio>()?;

bindings/python/tests/test_input_validation.py

@@ -231,3 +231,10 @@ def test_realized_spread_zero_horizon_raises():
 def test_kyles_lambda_window_below_two_raises():
     with pytest.raises(ValueError):
         ta.KylesLambda(1)
+
+
+def test_footprint_non_positive_tick_raises():
+    with pytest.raises(ValueError):
+        ta.Footprint(0.0)
+    with pytest.raises(ValueError):
+        ta.Footprint(-1.0)

bindings/python/tests/test_known_values.py

@@ -882,6 +882,17 @@ def test_depth_slope_reference_value():
     assert ta.DepthSlope().update([100.0], [1.0], [101.0], [1.0]) == pytest.approx(0.0)
 
 
+def test_footprint_buckets_buy_and_sell_volume():
+    fp = ta.Footprint(1.0)
+    fp.update(100.2, 2.0, True)  # bucket 100 -> ask 2
+    fp.update(100.7, 3.0, False)  # bucket 101 -> bid 3
+    out = fp.update(100.1, 1.0, True)  # bucket 100 -> ask 3
+    # Columns are [price, bid_vol, ask_vol], rows sorted ascending by price.
+    assert out.shape == (2, 3)
+    assert list(out[0]) == [100.0, 0.0, 3.0]
+    assert list(out[1]) == [101.0, 3.0, 0.0]
+
+
 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)

bindings/python/tests/test_lifecycle.py

@@ -207,3 +207,14 @@ def test_kyles_lambda_lifecycle_and_repr():
     kl.reset()
     assert not kl.is_ready()
     assert repr(ta.KylesLambda(7)) == "KylesLambda(window=7)"
+
+
+def test_footprint_lifecycle_and_repr():
+    fp = ta.Footprint(0.5)
+    assert fp.warmup_period() == 1
+    assert not fp.is_ready()
+    fp.update(100.0, 1.0, True)
+    assert fp.is_ready()
+    fp.reset()
+    assert not fp.is_ready()
+    assert repr(ta.Footprint(0.25)) == "Footprint(tick_size=0.25)"

bindings/python/tests/test_new_indicators.py

@@ -1939,3 +1939,16 @@ def test_price_impact_indicators_streaming_equals_batch():
         )
         assert batch.shape == (n,)
         assert _eq_nan(batch, streamed)
+
+
+def test_footprint_streaming_equals_batch():
+    n = 20
+    price = [100.0 + (i % 5) * 0.3 for i in range(n)]
+    size = [1.0 + (i % 3) for i in range(n)]
+    is_buy = [i % 2 == 0 for i in range(n)]
+    batch = ta.Footprint(1.0).batch(price, size, is_buy)
+    streamer = ta.Footprint(1.0)
+    assert len(batch) == n
+    for i in range(n):
+        streamed = streamer.update(price[i], size[i], is_buy[i])
+        assert np.array_equal(streamed, batch[i])

bindings/python/tests/test_smoke.py

@@ -154,3 +154,16 @@ def test_price_impact_batch_returns_one_value_per_trade():
         out = ind.batch(price, size, is_buy, mid)
         assert out.shape == (4,)
         assert out.dtype == np.float64
+
+
+def test_footprint_constructs_and_emits():
+    out = ta.Footprint(1.0).update(100.2, 2.0, True)
+    assert out.shape == (1, 3)
+    assert out.dtype == np.float64
+
+
+def test_footprint_batch_returns_list_of_arrays():
+    res = ta.Footprint(1.0).batch([100.2, 100.7], [2.0, 3.0], [True, False])
+    assert isinstance(res, list)
+    assert len(res) == 2
+    assert res[-1].shape[1] == 3