Aura/crates/aura-std/src/vol_slippage_cost.rs

//! `VolSlippageCost` — a slippage cost that scales with a measured volatility
//! input, charged once per closed trade, in R. The first *state-dependent*
//! [`CostNode`] factor: its price-unit numerator is `slip_vol_mult · volatility`
//! instead of a flat constant, so the cost-in-R varies trade-to-trade. The vol is
//! supplied as an extra input (an upstream realized-range estimator), kept
//! independent of the stop's own vol — scaling by the stop's vol would collapse
//! cost-in-R to a constant. R-pure: `slip_vol_mult · vol / |entry - stop|` (C10).
//!
//! Co-temporality is the shared [`CostRunner`]'s contract: it gates only on the PM
//! geometry, so a not-yet-warm `volatility` input makes this factor's numerator 0
//! that cycle (handled below) rather than withholding and desyncing the stream.

use aura_core::{Ctx, Firing, ParamSpec, PortSpec, PrimitiveBuilder, ScalarKind};

use crate::cost::{cost_node_builder, CostNode, CostRunner, GEOMETRY_WIDTH};

/// This node's one extra input beyond the geometry: a `volatility` stream (price
/// units), read at slot `GEOMETRY_WIDTH`. One source of truth, so the schema-input
/// list (`builder()`, → graph wiring) and the runner's lookbacks count
/// (`extra_inputs()`) cannot desync — mirroring `cost::geometry_input_ports`.
fn volatility_input_ports() -> Vec<PortSpec> {
    vec![PortSpec { kind: ScalarKind::F64, firing: Firing::Any, name: "volatility".into() }]
}

/// A volatility-scaled per-trade slippage, emitted in R via the shared
/// [`CostRunner`]. One extra input beyond the geometry: a `volatility` stream
/// (price units), read at slot `GEOMETRY_WIDTH`.
pub struct VolSlippageCost {
    slip_vol_mult: f64,
}

impl VolSlippageCost {
    /// A volatility-scaled slippage cost node (the factor wrapped in the runner).
    pub fn new(slip_vol_mult: f64) -> CostRunner<VolSlippageCost> {
        assert!(slip_vol_mult >= 0.0, "VolSlippageCost slip_vol_mult must be >= 0");
        CostRunner::new(VolSlippageCost { slip_vol_mult })
    }

    /// The param-generic recipe: one `slip_vol_mult` F64 knob; one extra
    /// `volatility` input appended after the geometry.
    pub fn builder() -> PrimitiveBuilder {
        cost_node_builder(
            "VolSlippageCost",
            volatility_input_ports(),
            vec![ParamSpec { name: "slip_vol_mult".into(), kind: ScalarKind::F64 }],
            |p| Box::new(VolSlippageCost::new(p[0].f64())),
        )
    }
}

impl CostNode for VolSlippageCost {
    fn label(&self) -> String {
        format!("VolSlippageCost({})", self.slip_vol_mult)
    }
    fn extra_inputs(&self) -> Vec<PortSpec> {
        volatility_input_ports()
    }
    fn cost_numerator(&mut self, ctx: &Ctx<'_>) -> f64 {
        let vol_w = ctx.f64_in(GEOMETRY_WIDTH);
        let vol = if vol_w.is_empty() { 0.0 } else { vol_w[0] }; // 0 during proxy warm-up
        self.slip_vol_mult * vol
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use aura_core::{AnyColumn, Cell, Node, Scalar, Timestamp};

    fn cols() -> Vec<AnyColumn> {
        vec![
            AnyColumn::with_capacity(ScalarKind::Bool, 1), // closed
            AnyColumn::with_capacity(ScalarKind::Bool, 1), // open
            AnyColumn::with_capacity(ScalarKind::F64, 1),  // entry
            AnyColumn::with_capacity(ScalarKind::F64, 1),  // stop
            AnyColumn::with_capacity(ScalarKind::F64, 1),  // volatility
        ]
    }

    #[test]
    fn no_geometry_yet_withholds() {
        let mut c = VolSlippageCost::new(0.5);
        let inputs = cols(); // all columns empty
        assert_eq!(c.eval(Ctx::new(&inputs, Timestamp(0))), None);
    }

    #[test]
    fn vol_not_yet_warm_emits_zero_cost_co_temporally() {
        // The realized-range proxy warms after the PM geometry; during warm-up the
        // node must still EMIT (a 0-cost row) so the cost stream stays 1:1 with the
        // PM record — withholding here would desync the positional join.
        let mut c = VolSlippageCost::new(0.5);
        let mut inputs = cols();
        inputs[0].push(Scalar::bool(true)).unwrap(); // closed
        inputs[1].push(Scalar::bool(false)).unwrap();
        inputs[2].push(Scalar::f64(100.0)).unwrap();
        inputs[3].push(Scalar::f64(96.0)).unwrap(); // latched 4.0
        // volatility column empty (proxy not warm) -> vol = 0 -> 0 cost, but a row IS emitted
        assert_eq!(
            c.eval(Ctx::new(&inputs, Timestamp(0))),
            Some([Cell::from_f64(0.0), Cell::from_f64(0.0), Cell::from_f64(0.0)].as_slice())
        );
    }

    #[test]
    fn closed_charges_mult_times_vol_over_latched() {
        let mut c = VolSlippageCost::new(0.5);
        let mut inputs = cols();
        inputs[0].push(Scalar::bool(true)).unwrap(); // closed
        inputs[1].push(Scalar::bool(false)).unwrap();
        inputs[2].push(Scalar::f64(100.0)).unwrap();
        inputs[3].push(Scalar::f64(96.0)).unwrap(); // latched 4.0
        inputs[4].push(Scalar::f64(3.0)).unwrap(); // vol 3.0
        // per = 0.5 * 3.0 / 4.0 = 0.375; cum = 0.375; open = 0.0
        assert_eq!(
            c.eval(Ctx::new(&inputs, Timestamp(0))),
            Some([Cell::from_f64(0.375), Cell::from_f64(0.375), Cell::from_f64(0.0)].as_slice())
        );
    }

    #[test]
    fn open_emits_would_be_cost_not_charged_to_cum() {
        let mut c = VolSlippageCost::new(0.5);
        let mut inputs = cols();
        inputs[0].push(Scalar::bool(false)).unwrap();
        inputs[1].push(Scalar::bool(true)).unwrap(); // open
        inputs[2].push(Scalar::f64(100.0)).unwrap();
        inputs[3].push(Scalar::f64(96.0)).unwrap(); // latched 4.0
        inputs[4].push(Scalar::f64(3.0)).unwrap(); // vol 3.0
        // cost_in_r = 0; cum 0; open_cost_in_r = 0.375
        assert_eq!(
            c.eval(Ctx::new(&inputs, Timestamp(0))),
            Some([Cell::from_f64(0.0), Cell::from_f64(0.0), Cell::from_f64(0.375)].as_slice())
        );
    }

    #[test]
    fn zero_latched_contributes_no_cost() {
        let mut c = VolSlippageCost::new(0.5);
        let mut inputs = cols();
        inputs[0].push(Scalar::bool(true)).unwrap();
        inputs[1].push(Scalar::bool(false)).unwrap();
        inputs[2].push(Scalar::f64(100.0)).unwrap();
        inputs[3].push(Scalar::f64(100.0)).unwrap(); // latched 0 -> no divide
        inputs[4].push(Scalar::f64(3.0)).unwrap();
        assert_eq!(
            c.eval(Ctx::new(&inputs, Timestamp(0))),
            Some([Cell::from_f64(0.0), Cell::from_f64(0.0), Cell::from_f64(0.0)].as_slice())
        );
    }

    #[test]
    fn cum_accumulates_across_closes() {
        let mut c = VolSlippageCost::new(0.5);
        let mut a = cols();
        a[0].push(Scalar::bool(true)).unwrap();
        a[1].push(Scalar::bool(false)).unwrap();
        a[2].push(Scalar::f64(100.0)).unwrap();
        a[3].push(Scalar::f64(96.0)).unwrap(); // latched 4
        a[4].push(Scalar::f64(3.0)).unwrap(); // 0.5*3/4 = 0.375
        let _ = c.eval(Ctx::new(&a, Timestamp(0)));
        let mut b = cols();
        b[0].push(Scalar::bool(true)).unwrap();
        b[1].push(Scalar::bool(false)).unwrap();
        b[2].push(Scalar::f64(100.0)).unwrap();
        b[3].push(Scalar::f64(98.0)).unwrap(); // latched 2
        b[4].push(Scalar::f64(4.0)).unwrap(); // 0.5*4/2 = 1.0; cum 1.375
        assert_eq!(
            c.eval(Ctx::new(&b, Timestamp(1))),
            Some([Cell::from_f64(1.0), Cell::from_f64(1.375), Cell::from_f64(0.0)].as_slice())
        );
    }

    #[test]
    fn label_carries_the_mult() {
        assert_eq!(VolSlippageCost::new(0.5).label(), "VolSlippageCost(0.5)");
        assert_eq!(VolSlippageCost::new(2.0).label(), "VolSlippageCost(2)");
    }

    #[test]
    #[should_panic(expected = "slip_vol_mult must be >= 0")]
    fn new_panics_on_negative_mult() {
        let _ = VolSlippageCost::new(-1.0);
    }

    #[test]
    fn builder_schema_extras_match_extra_inputs() {
        // The two consumers of the extra-input list must agree: the schema beyond
        // the geometry prefix (builder() -> graph wiring) and the factor's
        // extra_inputs() (CostRunner::lookbacks()'s count). One helper feeds both, so
        // they cannot desync; this pins that invariant against a future re-inlining.
        let schema_extras = VolSlippageCost::builder().schema().inputs[GEOMETRY_WIDTH..].to_vec();
        let factor_extras = VolSlippageCost { slip_vol_mult: 0.5 }.extra_inputs();
        assert_eq!(schema_extras, factor_extras);
    }
}