feat(aura-std): signal-quality loop — Exposure node + sim-optimal broker

Realizes the C10 reframe (cycle 0007): the strategy DAG's output is an
intent/exposure stream, and a sim-optimal broker integrates it into a synthetic
pip-equity curve that measures SIGNAL QUALITY — the project's first trading
result, and its primary research loop.

Two new aura-std nodes (plain structs; the engine stays domain-free):

  - Exposure { scale }: the decision/sizing node — clamp(signal/scale, -1, +1),
    one f64 exposure per fired cycle, None until warmed up. Sizing/risk live in
    `scale`.
  - SimBroker { pip_size }: class (a) of C10 — consumes exposure (slot 0) + price
    (slot 1), integrates the return earned by the exposure HELD INTO each cycle
    (prev_exposure, decided at t-1) so it is causal / look-ahead-free (C2), and
    emits cumulative pips. pip_size is held reference metadata (C7/C15), never
    streamed.

End-to-end proof in the aura-engine harness test module (it dev-depends on
aura-std): a SMA(2)-SMA(4) cross -> Exposure -> SimBroker -> Recorder sink wires
a full signal-quality backtest; the recorded equity matches a hand-computed pip
curve ([0,0,0,0, 0.035]) and is bit-identical across two runs (C1). The engine
itself is unchanged — pure node authoring on the frozen substrate.

Verified: cargo test --workspace green (20 core + 30 engine + 9 std; 8 new);
clippy --all-targets -D warnings clean; engine/core surface-purity grep (no
dyn Any/Rc/RefCell) clean.

Note: the plan inlined `Window::first()`, which aura-core's Window does not
expose; the shipped code uses the semantically identical `Window::get(0)`
(newest value, or 0.0 when the exposure leg is cold).

closes #4
closes #5

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

crates/aura-engine/src/harness.rs

@@ -337,7 +337,7 @@ mod tests {
     // only reads `nd.schema()` fields, never naming the types), so they are not
     // brought in by `use super::*` — the fixtures construct them, so import here.
     use aura_core::{FieldSpec, InputSpec, NodeSchema};
-    use aura_std::{Sma, Sub};
+    use aura_std::{Exposure, Sma, SimBroker, Sub};
     use std::sync::mpsc;
 
     /// Build an f64 source stream from (timestamp, value) points.
@@ -1696,4 +1696,103 @@ mod tests {
         assert_eq!(rs2.try_iter().collect::<Vec<_>>(), sub_expected);
         assert_eq!(ri2.try_iter().collect::<Vec<_>>(), i64_expected);
     }
+
+    #[test]
+    fn signal_quality_loop_records_pip_equity() {
+        let (tx_eq, rx_eq) = mpsc::channel();
+        let mut h = Harness::bootstrap(
+            vec![
+                Box::new(Sma::new(2)),                              // 0 fast
+                Box::new(Sma::new(4)),                              // 1 slow
+                Box::new(Sub::new()),                               // 2 raw signal
+                Box::new(Exposure::new(0.5)),                       // 3 exposure
+                Box::new(SimBroker::new(0.0001)),                   // 4 pip equity
+                Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx_eq)), // 5 sink
+            ],
+            vec![SourceSpec {
+                kind: ScalarKind::F64,
+                targets: vec![
+                    Target { node: 0, slot: 0 }, // price -> SMA fast
+                    Target { node: 1, slot: 0 }, // price -> SMA slow
+                    Target { node: 4, slot: 1 }, // price -> broker price input
+                ],
+            }],
+            vec![
+                Edge { from: 0, to: 2, slot: 0, from_field: 0 }, // fast     -> Sub.0
+                Edge { from: 1, to: 2, slot: 1, from_field: 0 }, // slow     -> Sub.1
+                Edge { from: 2, to: 3, slot: 0, from_field: 0 }, // signal   -> Exposure
+                Edge { from: 3, to: 4, slot: 0, from_field: 0 }, // exposure -> broker.0
+                Edge { from: 4, to: 5, slot: 0, from_field: 0 }, // equity   -> recorder
+            ],
+        )
+        .expect("valid signal-quality harness");
+
+        h.run(vec![f64_stream(&[
+            (1, 1.0000),
+            (2, 1.0010),
+            (3, 1.0025),
+            (4, 1.0020),
+            (5, 1.0040),
+        ])]);
+
+        let equity: Vec<(Timestamp, Vec<Scalar>)> = rx_eq.try_iter().collect();
+        // broker fires every cycle (price fresh each tick): five records, ts 1..=5
+        assert_eq!(equity.len(), 5);
+        assert_eq!(
+            equity.iter().map(|(t, _)| t.0).collect::<Vec<_>>(),
+            vec![1, 2, 3, 4, 5]
+        );
+        // flat until SMA(4) warms (cycle 4) and the held exposure meets the next
+        // price move (cycle 5): the first four equities are exactly 0.
+        for (_, row) in &equity[0..4] {
+            assert_eq!(row, &vec![Scalar::F64(0.0)]);
+        }
+        // cycle 5: prev_exposure 0.00175 * (1.0040 - 1.0020) / 0.0001 = 0.035 pips
+        let Scalar::F64(last) = equity[4].1[0] else {
+            panic!("equity is f64");
+        };
+        assert!((last - 0.035).abs() < 1e-9, "final equity = {last}, want ~0.035");
+    }
+
+    #[test]
+    fn signal_quality_loop_is_deterministic() {
+        let build = || {
+            let (tx, rx) = mpsc::channel();
+            let mut h = Harness::bootstrap(
+                vec![
+                    Box::new(Sma::new(2)),
+                    Box::new(Sma::new(4)),
+                    Box::new(Sub::new()),
+                    Box::new(Exposure::new(0.5)),
+                    Box::new(SimBroker::new(0.0001)),
+                    Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx)),
+                ],
+                vec![SourceSpec {
+                    kind: ScalarKind::F64,
+                    targets: vec![
+                        Target { node: 0, slot: 0 },
+                        Target { node: 1, slot: 0 },
+                        Target { node: 4, slot: 1 },
+                    ],
+                }],
+                vec![
+                    Edge { from: 0, to: 2, slot: 0, from_field: 0 },
+                    Edge { from: 1, to: 2, slot: 1, from_field: 0 },
+                    Edge { from: 2, to: 3, slot: 0, from_field: 0 },
+                    Edge { from: 3, to: 4, slot: 0, from_field: 0 },
+                    Edge { from: 4, to: 5, slot: 0, from_field: 0 },
+                ],
+            )
+            .expect("valid harness");
+            h.run(vec![f64_stream(&[
+                (1, 1.0000),
+                (2, 1.0010),
+                (3, 1.0025),
+                (4, 1.0020),
+                (5, 1.0040),
+            ])]);
+            rx.try_iter().collect::<Vec<(Timestamp, Vec<Scalar>)>>()
+        };
+        assert_eq!(build(), build());
+    }
 }

crates/aura-std/src/exposure.rs

@@ -0,0 +1,73 @@
+//! `Exposure` — shapes a raw signal score into a bounded exposure (intent).
+//! The decision/sizing node of C10's chain `signals -> decision/sizing node ->
+//! exposure stream`: one f64 input, one f64 output `clamp(signal / scale, -1, +1)`.
+//! `scale` sets which signal magnitude maps to full exposure (sizing lives here).
+
+use aura_core::{Ctx, FieldSpec, Firing, InputSpec, Node, NodeSchema, Scalar, ScalarKind};
+
+/// Bounded exposure from a raw signal score: `clamp(signal / scale, -1.0, +1.0)`.
+/// Emits `None` until its input is present (warm-up filter, C8).
+pub struct Exposure {
+    scale: f64,
+    out: [Scalar; 1],
+}
+
+impl Exposure {
+    /// Build an exposure node with saturation magnitude `scale` (must be > 0).
+    pub fn new(scale: f64) -> Self {
+        assert!(scale > 0.0, "Exposure scale must be > 0");
+        Self { scale, out: [Scalar::F64(0.0)] }
+    }
+}
+
+impl Node for Exposure {
+    fn schema(&self) -> NodeSchema {
+        NodeSchema {
+            inputs: vec![InputSpec { kind: ScalarKind::F64, lookback: 1, firing: Firing::Any }],
+            output: vec![FieldSpec { name: "exposure", kind: ScalarKind::F64 }],
+        }
+    }
+
+    fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Scalar]> {
+        let w = ctx.f64_in(0);
+        if w.is_empty() {
+            return None; // not yet warmed up (C8 filter)
+        }
+        self.out[0] = Scalar::F64((w[0] / self.scale).clamp(-1.0, 1.0));
+        Some(&self.out)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use aura_core::{AnyColumn, Timestamp};
+
+    #[test]
+    fn exposure_clamps_to_unit_band() {
+        let mut e = Exposure::new(0.5);
+        let mut inputs = vec![AnyColumn::with_capacity(ScalarKind::F64, 1)];
+        // (raw signal, expected clamped exposure) for scale 0.5
+        let cases = [
+            (0.1_f64, 0.2_f64), // within band
+            (0.5, 1.0),         // at the high edge
+            (1.0, 1.0),         // saturates high
+            (-0.1, -0.2),       // within band, negative
+            (-1.0, -1.0),       // saturates low
+        ];
+        for (sig, want) in cases {
+            inputs[0].push(Scalar::F64(sig)).unwrap();
+            assert_eq!(
+                e.eval(Ctx::new(&inputs, Timestamp(0))),
+                Some([Scalar::F64(want)].as_slice())
+            );
+        }
+    }
+
+    #[test]
+    fn exposure_is_none_until_input_present() {
+        let mut e = Exposure::new(0.5);
+        let inputs = vec![AnyColumn::with_capacity(ScalarKind::F64, 1)];
+        assert_eq!(e.eval(Ctx::new(&inputs, Timestamp(0))), None);
+    }
+}

crates/aura-std/src/lib.rs

@@ -15,7 +15,11 @@
 //! The first block lands with the walking skeleton: [`Sma`], the simple moving
 //! average — a worked producer node proving the `aura-core` `Node` contract.
 
+mod exposure;
+mod sim_broker;
 mod sma;
 mod sub;
+pub use exposure::Exposure;
+pub use sim_broker::SimBroker;
 pub use sma::Sma;
 pub use sub::Sub;

crates/aura-std/src/sim_broker.rs

@@ -0,0 +1,125 @@
+//! `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.
+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)
+    }
+}
+
+#[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
+    }
+}