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

//! `SimBroker` — the sim-optimal broker (class (a) of C10): deterministic,
//! frictionless, perfect-fill. Consumes an exposure stream (slot 0) + a price
//! stream (slot 1) and integrates the return earned by the exposure held INTO
//! each cycle (decided at t-1) into a cumulative synthetic pip-equity output.
//! Measures signal quality, not execution-modelled P&L.

use aura_core::{Ctx, FieldSpec, Firing, InputSpec, Node, NodeSchema, Scalar, ScalarKind};

/// Integrates `exposure * price-return` into cumulative pips. `pip_size` is
/// per-instrument reference metadata (beside the hot path, C7/C15), held here,
/// never streamed.
///
/// # Input slots
///
/// Two `f64` inputs, **order is load-bearing** (both are `f64`, so a swapped
/// wiring is *not* caught at bootstrap — it would silently produce a wrong
/// equity curve):
///
/// - **slot 0 — exposure** ∈ [-1, +1] (the strategy's intent, e.g. from
///   [`Exposure`](crate::Exposure)).
/// - **slot 1 — price** (the instrument price the exposure is marked against).
///
/// # Firing and warm-up
///
/// Both inputs are [`Firing::Any`](aura_core::Firing::Any), so the broker fires
/// on **every price-fresh cycle** (price is typically the source tick). A
/// not-yet-warmed exposure leg is read as flat `0.0`, so the broker emits an
/// equity row from the **first** price tick, reading `0.0` until the exposure
/// leg produces its first value — the recorded curve therefore has one row per
/// price cycle, with leading `0.0`s during warm-up, not one row per exposure.
///
/// # Output
///
/// One `f64` column, the cumulative pip equity (a producer; tap it with a
/// recording sink to persist the curve).
pub struct SimBroker {
    pip_size: f64,
    prev_price: Option<f64>,
    prev_exposure: f64, // exposure held into this cycle (decided at t-1); 0.0 = flat
    cum: f64,           // cumulative pips
    out: [Scalar; 1],
}

impl SimBroker {
    /// Build a sim-optimal broker for an instrument whose pip is `pip_size`
    /// (price units per pip; must be > 0).
    pub fn new(pip_size: f64) -> Self {
        assert!(pip_size > 0.0, "SimBroker pip_size must be > 0");
        Self {
            pip_size,
            prev_price: None,
            prev_exposure: 0.0,
            cum: 0.0,
            out: [Scalar::F64(0.0)],
        }
    }
}

impl Node for SimBroker {
    fn schema(&self) -> NodeSchema {
        NodeSchema {
            inputs: vec![
                InputSpec { kind: ScalarKind::F64, lookback: 1, firing: Firing::Any }, // 0 exposure
                InputSpec { kind: ScalarKind::F64, lookback: 1, firing: Firing::Any }, // 1 price
            ],
            output: vec![FieldSpec { name: "equity", kind: ScalarKind::F64 }],
        }
    }

    fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Scalar]> {
        let price = ctx.f64_in(1);
        if price.is_empty() {
            return None; // no price yet — nothing to mark
        }
        let price = price[0];
        let expo = ctx.f64_in(0).get(0).unwrap_or(0.0); // flat until exposure warms up
        if let Some(pp) = self.prev_price {
            self.cum += self.prev_exposure * (price - pp) / self.pip_size;
        }
        self.prev_price = Some(price);
        self.prev_exposure = expo; // update AFTER taking PnL — no look-ahead (C2)
        self.out[0] = Scalar::F64(self.cum);
        Some(&self.out)
    }

    fn label(&self) -> String {
        format!("SimBroker({})", self.pip_size)
    }
}

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

    fn two_f64_inputs() -> Vec<AnyColumn> {
        vec![
            AnyColumn::with_capacity(ScalarKind::F64, 1),
            AnyColumn::with_capacity(ScalarKind::F64, 1),
        ]
    }

    // Drive one broker cycle by hand: optionally push an exposure into slot 0
    // (None models a cycle where the exposure chain has not warmed up — slot 0
    // stays empty) and a price into slot 1, then eval and return the equity.
    fn step(b: &mut SimBroker, inputs: &mut [AnyColumn], expo: Option<f64>, price: f64) -> f64 {
        if let Some(e) = expo {
            inputs[0].push(Scalar::F64(e)).unwrap();
        }
        inputs[1].push(Scalar::F64(price)).unwrap();
        match b.eval(Ctx::new(inputs, Timestamp(0))) {
            Some([Scalar::F64(v)]) => *v,
            other => panic!("expected Some([F64]), got {other:?}"),
        }
    }

    #[test]
    fn sim_broker_integrates_lagged_exposure_times_return() {
        let mut b = SimBroker::new(1.0);
        let mut inputs = two_f64_inputs();
        assert_eq!(step(&mut b, &mut inputs, Some(0.5), 100.0), 0.0); // no prev price
        assert_eq!(step(&mut b, &mut inputs, Some(0.5), 110.0), 5.0); // 0.5*(110-100)
        assert_eq!(step(&mut b, &mut inputs, Some(-1.0), 108.0), 4.0); // +0.5*(108-110) = -1
        assert_eq!(step(&mut b, &mut inputs, Some(-1.0), 100.0), 12.0); // +(-1)*(100-108) = +8
    }

    #[test]
    fn sim_broker_no_lookahead() {
        let mut b = SimBroker::new(1.0);
        let mut inputs = two_f64_inputs();
        step(&mut b, &mut inputs, Some(1.0), 100.0);
        // exposure flips to 0.0 THIS cycle, but the PnL must use the 1.0 held
        // into it: 1.0*(110-100) = 10. Using the fresh 0.0 would give 0.
        assert_eq!(step(&mut b, &mut inputs, Some(0.0), 110.0), 10.0);
    }

    #[test]
    fn sim_broker_is_flat_during_warmup() {
        let mut b = SimBroker::new(1.0);
        let mut inputs = two_f64_inputs();
        // exposure never pushed (slot 0 empty) -> treated as flat; equity stays 0
        assert_eq!(step(&mut b, &mut inputs, None, 100.0), 0.0);
        assert_eq!(step(&mut b, &mut inputs, None, 110.0), 0.0);
        assert_eq!(step(&mut b, &mut inputs, None, 90.0), 0.0);
    }

    #[test]
    fn sim_broker_first_cycle_has_no_pnl() {
        let mut b = SimBroker::new(1.0);
        let mut inputs = two_f64_inputs();
        assert_eq!(step(&mut b, &mut inputs, Some(1.0), 100.0), 0.0); // no prev price to mark
    }
}