diff --git a/crates/aura-cli/src/main.rs b/crates/aura-cli/src/main.rs index 35d6d32..3f16f26 100644 --- a/crates/aura-cli/src/main.rs +++ b/crates/aura-cli/src/main.rs @@ -31,7 +31,7 @@ use aura_registry::{ use aura_std::{ Add, Bias, ConstantCost, CostSum, Delay, Ema, GatedRecorder, Gt, Latch, LinComb, LongOnly, Mul, Recorder, RollingMax, RollingMin, SeriesReducer, SimBroker, Sma, Sqrt, Sub, VolSlippageCost, - PM_FIELD_NAMES, PM_RECORD_KINDS, + COST_FIELD_NAMES, COST_WIDTH, PM_FIELD_NAMES, PM_RECORD_KINDS, }; use std::sync::mpsc::{self, Receiver}; use std::sync::LazyLock; @@ -2575,18 +2575,14 @@ const SLIP_VOL_LENGTH: i64 = 5; /// vol slippage). Sizes the interned `CostSum` input-port names below. const MAX_RUN_COST_NODES: usize = 2; -/// The 3-field cost-in-R record order — a lockstep contract with each cost node's -/// output schema and `CostSum`'s inputs (and `summarize_r`'s `cost_col`). -const COST_FIELDS: [&str; 3] = ["cost_in_r", "cum_cost_in_r", "open_cost_in_r"]; - /// Interned `cost[k].` `CostSum` input-port names, built once. Same /// `&'static str`-from-a-static rationale as `COL_PORTS`: `GraphBuilder::input` /// wants `&'static str`, so the names live in a `static` rather than being /// `format!(...).leak()`ed per build. static COST_SUM_PORTS: LazyLock> = LazyLock::new(|| { - let mut v = Vec::with_capacity(MAX_RUN_COST_NODES * 3); + let mut v = Vec::with_capacity(MAX_RUN_COST_NODES * COST_WIDTH); for k in 0..MAX_RUN_COST_NODES { - for field in COST_FIELDS { + for field in COST_FIELD_NAMES { v.push(format!("cost[{k}].{field}")); } } @@ -2762,8 +2758,8 @@ fn stage1_r_graph( g.connect(exec.output("open"), cc.input("open")); g.connect(exec.output("entry_price"), cc.input("entry_price")); g.connect(exec.output("stop_price"), cc.input("stop_price")); - for (f, field) in COST_FIELDS.iter().copied().enumerate() { - g.connect(cc.output(field), agg.input(COST_SUM_PORTS[slot * 3 + f].as_str())); + for (f, field) in COST_FIELD_NAMES.iter().copied().enumerate() { + g.connect(cc.output(field), agg.input(COST_SUM_PORTS[slot * COST_WIDTH + f].as_str())); } slot += 1; } @@ -2776,8 +2772,8 @@ fn stage1_r_graph( g.connect(exec.output("entry_price"), vs.input("entry_price")); g.connect(exec.output("stop_price"), vs.input("stop_price")); g.connect(vrange.output("value"), vs.input("volatility")); - for (f, field) in COST_FIELDS.iter().copied().enumerate() { - g.connect(vs.output(field), agg.input(COST_SUM_PORTS[slot * 3 + f].as_str())); + for (f, field) in COST_FIELD_NAMES.iter().copied().enumerate() { + g.connect(vs.output(field), agg.input(COST_SUM_PORTS[slot * COST_WIDTH + f].as_str())); } slot += 1; } diff --git a/crates/aura-cli/tests/cli_run.rs b/crates/aura-cli/tests/cli_run.rs index d967280..16e2864 100644 --- a/crates/aura-cli/tests/cli_run.rs +++ b/crates/aura-cli/tests/cli_run.rs @@ -1671,6 +1671,63 @@ fn stage1_r_both_costs_compose_net_below_each_alone() { let _ = std::fs::remove_dir_all(&dir); } +/// Golden characterization (cycle 0083, the CostNode-trait migration): pins the +/// EXACT `net_expectancy_r` of `aura run --harness stage1-r --cost-per-trade 2`, so +/// the ConstantCost-through-CostRunner migration (Task 2) is VERIFIED byte-preserving +/// at the observable boundary, not assumed. The sibling +/// `stage1_r_cost_run_persists_net_r_equity_and_charges_cost` only asserts the +/// RELATION `net < gross`; a regression that drifted the cost numerator or the +/// `cost_per_trade / |entry - stop|` R-normalization (the token form the cycle +/// promises to keep verbatim) would keep `net < gross` true and pass that test +/// silently while shifting this value. Pins only the cost-affected float — gross +/// `expectancy_r` is already pinned by `stage1_r_single_run_output_golden` and is +/// unchanged by cost. Deterministic over the fixed synthetic stream (C1). +#[test] +fn stage1_r_flat_cost_net_expectancy_r_golden() { + let out = Command::new(BIN) + .args(["run", "--harness", "stage1-r", "--cost-per-trade", "2"]) + .output() + .unwrap(); + assert!(out.status.success(), "exit: {:?}; stderr: {}", out.status, + String::from_utf8_lossy(&out.stderr)); + let s = String::from_utf8(out.stdout).expect("utf-8 stdout"); + let v: serde_json::Value = serde_json::from_str(s.trim()).unwrap(); + let net = v["metrics"]["r"]["net_expectancy_r"].as_f64().unwrap(); + assert_eq!( + net, -614.3134020253314, + "flat-cost net_expectancy_r drifted from the golden — the ConstantCost \ + factor migration is not byte-preserving: {s}" + ); +} + +/// Golden characterization (cycle 0083, the composed cost path): pins the EXACT +/// `net_expectancy_r` of `aura run --harness stage1-r --cost-per-trade 2 +/// --slip-vol-mult 0.5`, so the VolSlippageCost-through-CostRunner migration (Task 3) +/// AND the CostSum per-field aggregation reading the shared `COST_FIELD_NAMES` +/// contract (Task 4) are VERIFIED byte-preserving end to end. The sibling +/// `stage1_r_both_costs_compose_net_below_each_alone` only asserts the RELATION +/// `net_both < net_flat`; a drift in the vol-scaled numerator, the CostSum field +/// order, or the geometry-prefix slot offsets (`slot * COST_WIDTH + f`) would keep +/// that relation true and pass silently while shifting this value. Deterministic +/// over the fixed synthetic stream (C1). +#[test] +fn stage1_r_composed_cost_net_expectancy_r_golden() { + let out = Command::new(BIN) + .args(["run", "--harness", "stage1-r", "--cost-per-trade", "2", "--slip-vol-mult", "0.5"]) + .output() + .unwrap(); + assert!(out.status.success(), "exit: {:?}; stderr: {}", out.status, + String::from_utf8_lossy(&out.stderr)); + let s = String::from_utf8(out.stdout).expect("utf-8 stdout"); + let v: serde_json::Value = serde_json::from_str(s.trim()).unwrap(); + let net = v["metrics"]["r"]["net_expectancy_r"].as_f64().unwrap(); + assert_eq!( + net, -615.0304388396047, + "composed-cost net_expectancy_r drifted from the golden — the VolSlippageCost \ + factor + CostSum aggregation is not byte-preserving: {s}" + ); +} + /// Property (the spec's dual-yardstick headline): `aura run --harness stage1-r` scores /// ONE signal in BOTH yardsticks at once — the single emitted `RunReport` carries the pip /// `metrics.total_pips` (off the SimBroker branch) AND the nested `r` block (off the diff --git a/crates/aura-std/src/constant_cost.rs b/crates/aura-std/src/constant_cost.rs index fea7ff3..dbd887b 100644 --- a/crates/aura-std/src/constant_cost.rs +++ b/crates/aura-std/src/constant_cost.rs @@ -1,97 +1,51 @@ //! `ConstantCost` — a flat round-trip cost charged once per closed trade, in R. -//! The first cost node of the C10 cost-model graph: an ordinary downstream -//! node (C9) that reads the executor's trade-geometry and emits a cost-in-R record -//! isomorphic to `PositionManagement`'s R-triple — `cost_in_r` (charged on a close) -//! / `cum_cost_in_r` (its running sum) / `open_cost_in_r` (the open trade's would-be -//! cost, for the window-end trade `summarize_r` synthesises). R-pure: with flat-1R -//! sizing the cost is `cost_per_trade / |entry - stop|`; notional cancels (C10). +//! The simplest cost node of the C10 cost-model graph: a stateless [`CostNode`] +//! factor whose price-unit numerator is a flat `cost_per_trade`. The shared +//! [`CostRunner`] supplies the co-temporality skeleton (geometry gating, the +//! `cost_per_trade / |entry - stop|` R-normalization, the closed/open charge, the +//! running `cum`, the 3-field emit). R-pure: notional cancels (C10). -use aura_core::{ - Cell, Ctx, FieldSpec, Firing, Node, NodeSchema, ParamSpec, PortSpec, PrimitiveBuilder, - ScalarKind, -}; +use aura_core::{Ctx, ParamSpec, PrimitiveBuilder, ScalarKind}; -/// A flat per-trade cost in price units (`cost_per_trade`), emitted in R. Inputs are -/// four executor-exposed fields: `closed` (closed_this_cycle), `open`, `entry_price`, -/// `stop_price`. Emits `None` until the executor has produced a record this cycle. +use crate::cost::{cost_node_builder, CostNode, CostRunner}; + +/// A flat per-trade cost in price units (`cost_per_trade`), emitted in R via the +/// shared [`CostRunner`]. pub struct ConstantCost { cost_per_trade: f64, - cum: f64, - out: [Cell; 3], } impl ConstantCost { - pub fn new(cost_per_trade: f64) -> Self { + /// A flat per-trade cost node: the factor wrapped in the shared [`CostRunner`]. + pub fn new(cost_per_trade: f64) -> CostRunner { assert!(cost_per_trade >= 0.0, "ConstantCost cost_per_trade must be >= 0"); - Self { cost_per_trade, cum: 0.0, out: [Cell::from_f64(0.0); 3] } + CostRunner::new(ConstantCost { cost_per_trade }) } - /// The param-generic recipe: one `cost_per_trade` F64 knob. + /// The param-generic recipe: one `cost_per_trade` F64 knob, no extra inputs. pub fn builder() -> PrimitiveBuilder { - PrimitiveBuilder::new( + cost_node_builder( "ConstantCost", - NodeSchema { - inputs: vec![ - PortSpec { kind: ScalarKind::Bool, firing: Firing::Any, name: "closed".into() }, - PortSpec { kind: ScalarKind::Bool, firing: Firing::Any, name: "open".into() }, - PortSpec { kind: ScalarKind::F64, firing: Firing::Any, name: "entry_price".into() }, - PortSpec { kind: ScalarKind::F64, firing: Firing::Any, name: "stop_price".into() }, - ], - output: vec![ - FieldSpec { name: "cost_in_r".into(), kind: ScalarKind::F64 }, - FieldSpec { name: "cum_cost_in_r".into(), kind: ScalarKind::F64 }, - FieldSpec { name: "open_cost_in_r".into(), kind: ScalarKind::F64 }, - ], - params: vec![ParamSpec { name: "cost_per_trade".into(), kind: ScalarKind::F64 }], - }, + Vec::new(), + vec![ParamSpec { name: "cost_per_trade".into(), kind: ScalarKind::F64 }], |p| Box::new(ConstantCost::new(p[0].f64())), ) } } -impl Node for ConstantCost { - fn lookbacks(&self) -> Vec { - vec![1, 1, 1, 1] - } - - fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Cell]> { - // Withhold until the executor's full geometry record is present this cycle - // (all four fields fire together; before warm-up none of them do). - let closed_w = ctx.bool_in(0); - let open_w = ctx.bool_in(1); - let entry_w = ctx.f64_in(2); - let stop_w = ctx.f64_in(3); - if closed_w.is_empty() || open_w.is_empty() || entry_w.is_empty() || stop_w.is_empty() { - return None; - } - let closed = closed_w[0]; - let open = open_w[0]; - let entry = entry_w[0]; - let stop = stop_w[0]; - let latched = (entry - stop).abs(); - // A zero latched distance is a flat/degenerate cycle (no valid 1R denominator): - // it contributes no cost rather than dividing by zero — matching summarize_r's guard. - let per = if latched > 0.0 { self.cost_per_trade / latched } else { 0.0 }; - let cost_in_r = if closed { per } else { 0.0 }; - let open_cost_in_r = if open { per } else { 0.0 }; - self.cum += cost_in_r; - self.out = [ - Cell::from_f64(cost_in_r), - Cell::from_f64(self.cum), - Cell::from_f64(open_cost_in_r), - ]; - Some(&self.out) - } - +impl CostNode for ConstantCost { fn label(&self) -> String { format!("ConstantCost({})", self.cost_per_trade) } + fn cost_numerator(&mut self, _ctx: &Ctx<'_>) -> f64 { + self.cost_per_trade + } } #[cfg(test)] mod tests { use super::*; - use aura_core::{AnyColumn, Scalar, Timestamp}; + use aura_core::{AnyColumn, Cell, Node, Scalar, Timestamp}; fn cols() -> Vec { vec![ diff --git a/crates/aura-std/src/cost.rs b/crates/aura-std/src/cost.rs new file mode 100644 index 0000000..b859737 --- /dev/null +++ b/crates/aura-std/src/cost.rs @@ -0,0 +1,368 @@ +//! The cost-model-graph node contract (C10): the `CostNode` factor trait and the +//! `CostRunner` adapter that wraps a factor into an engine `Node`. +//! +//! A cost node's only per-node difference is the **price-unit cost numerator** the +//! runner divides by the latched 1R distance. Everything else — gating on the PM +//! geometry (the co-temporality contract: the cost stream stays 1:1 with the +//! executor's record), the closed/open charge, the running `cum`, the 3-field +//! emit — is the runner's, written once. The 3-field cost record +//! (`COST_FIELD_NAMES`) is one source of truth, read by both the producer side +//! (`cost_node_builder`) and the `CostSum` aggregator; this mirrors the +//! `position_management::{FIELD_NAMES, WIDTH}` precedent and replaces what was a +//! by-convention triple lockstep. + +use aura_core::{ + Cell, Ctx, FieldSpec, Firing, Node, NodeSchema, ParamSpec, PortSpec, PrimitiveBuilder, + ScalarKind, +}; + +/// The 3-field cost-in-R record every cost node emits, in slot order — one source +/// of truth for the producer schema and the `CostSum` aggregator. +pub const COST_WIDTH: usize = 3; +pub const COST_FIELD_NAMES: [&str; COST_WIDTH] = + ["cost_in_r", "cum_cost_in_r", "open_cost_in_r"]; + +/// The PM-geometry input prefix every cost node gates on (`closed`, `open`, +/// `entry_price`, `stop_price`). A factor's own extra inputs are appended after +/// these, beginning at slot `GEOMETRY_WIDTH`. +pub const GEOMETRY_WIDTH: usize = 4; + +fn geometry_input_ports() -> Vec { + vec![ + PortSpec { kind: ScalarKind::Bool, firing: Firing::Any, name: "closed".into() }, + PortSpec { kind: ScalarKind::Bool, firing: Firing::Any, name: "open".into() }, + PortSpec { kind: ScalarKind::F64, firing: Firing::Any, name: "entry_price".into() }, + PortSpec { kind: ScalarKind::F64, firing: Firing::Any, name: "stop_price".into() }, + ] +} + +fn cost_output_fields() -> Vec { + COST_FIELD_NAMES + .iter() + .map(|n| FieldSpec { name: (*n).into(), kind: ScalarKind::F64 }) + .collect() +} + +/// A cost factor: the per-round-trip cost in *price units* (the numerator the +/// runner divides by the latched 1R distance). The only thing a cost node differs +/// in; the co-temporality skeleton is [`CostRunner`]'s, shared. +/// +/// # Authoring a cost node +/// +/// ``` +/// use aura_core::Ctx; +/// use aura_std::{CostNode, CostRunner}; +/// +/// pub struct HalfSpreadCost { +/// half_spread: f64, +/// } +/// +/// impl HalfSpreadCost { +/// pub fn new(half_spread: f64) -> CostRunner { +/// assert!(half_spread >= 0.0, "HalfSpreadCost half_spread must be >= 0"); +/// CostRunner::new(HalfSpreadCost { half_spread }) +/// } +/// } +/// +/// impl CostNode for HalfSpreadCost { +/// fn label(&self) -> String { +/// format!("HalfSpreadCost({})", self.half_spread) +/// } +/// fn cost_numerator(&mut self, _ctx: &Ctx<'_>) -> f64 { +/// self.half_spread // price units; the runner divides by the latched 1R distance +/// } +/// } +/// +/// let _node = HalfSpreadCost::new(0.5); // a ready-to-wire cost node +/// ``` +pub trait CostNode: 'static { + /// One-line render label carrying the identifying param (C23). + fn label(&self) -> String; + /// Extra input ports beyond the 4 geometry inputs, appended at slot + /// `GEOMETRY_WIDTH`. Default: none. + fn extra_inputs(&self) -> Vec { + Vec::new() + } + /// The round-trip cost in price units this cycle, BEFORE R-normalization and + /// BEFORE the closed/open gate. Reads its extra inputs from `ctx` at + /// `GEOMETRY_WIDTH + i`; an empty window during warm-up means 0 (the runner + /// still emits the row — co-temporality). + fn cost_numerator(&mut self, ctx: &Ctx<'_>) -> f64; +} + +/// The shared co-temporality skeleton wrapping any [`CostNode`] factor into a +/// `Node`. Holds the only running state a cost node needs — `cum` and the output +/// buffer — so a factor impl stays pure. +pub struct CostRunner { + factor: F, + cum: f64, + out: [Cell; COST_WIDTH], +} + +impl CostRunner { + pub fn new(factor: F) -> Self { + Self { factor, cum: 0.0, out: [Cell::from_f64(0.0); COST_WIDTH] } + } +} + +impl Node for CostRunner { + fn lookbacks(&self) -> Vec { + vec![1; GEOMETRY_WIDTH + self.factor.extra_inputs().len()] + } + + fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Cell]> { + // Gate ONLY on the PM geometry (co-temporality): the cost stream stays 1:1 + // with the executor's record. A factor's not-yet-warm input contributes 0 + // (handled in `cost_numerator`), it does not withhold the row. + let closed_w = ctx.bool_in(0); + let open_w = ctx.bool_in(1); + let entry_w = ctx.f64_in(2); + let stop_w = ctx.f64_in(3); + if closed_w.is_empty() || open_w.is_empty() || entry_w.is_empty() || stop_w.is_empty() { + return None; + } + let closed = closed_w[0]; + let open = open_w[0]; + let latched = (entry_w[0] - stop_w[0]).abs(); + let numerator = self.factor.cost_numerator(&ctx); + // Zero latched distance = no valid 1R denominator -> no cost. The + // `numerator / latched` token form is preserved verbatim from the + // pre-migration nodes for byte-identity (IEEE-754). + let per = if latched > 0.0 { numerator / latched } else { 0.0 }; + let cost_in_r = if closed { per } else { 0.0 }; + let open_cost_in_r = if open { per } else { 0.0 }; + self.cum += cost_in_r; + self.out = [ + Cell::from_f64(cost_in_r), + Cell::from_f64(self.cum), + Cell::from_f64(open_cost_in_r), + ]; + Some(&self.out) + } + + fn label(&self) -> String { + self.factor.label() + } +} + +/// Assemble a cost-node `PrimitiveBuilder`: the 4 geometry inputs ++ the factor's +/// extra inputs, the standard 3-field cost output, the given params, and a build +/// closure. The single home for the cost-node schema shape. +pub fn cost_node_builder( + name: &'static str, + extra_inputs: Vec, + params: Vec, + build: impl Fn(&[Cell]) -> Box + 'static, +) -> PrimitiveBuilder { + let mut inputs = geometry_input_ports(); + inputs.extend(extra_inputs); + PrimitiveBuilder::new(name, NodeSchema { inputs, output: cost_output_fields(), params }, build) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::{ConstantCost, CostSum}; + use aura_core::{AnyColumn, Scalar, Timestamp}; + + /// A test-only factor: a constant numerator, no extra inputs. + struct StubCost(f64); + impl CostNode for StubCost { + fn label(&self) -> String { + format!("StubCost({})", self.0) + } + fn cost_numerator(&mut self, _ctx: &Ctx<'_>) -> f64 { + self.0 + } + } + + /// A test-only factor with one extra f64 input, read at `GEOMETRY_WIDTH`. + struct StubExtra(f64); + impl CostNode for StubExtra { + fn label(&self) -> String { + "StubExtra".into() + } + fn extra_inputs(&self) -> Vec { + vec![PortSpec { kind: ScalarKind::F64, firing: Firing::Any, name: "x".into() }] + } + fn cost_numerator(&mut self, ctx: &Ctx<'_>) -> f64 { + let w = ctx.f64_in(GEOMETRY_WIDTH); + let x = if w.is_empty() { 0.0 } else { w[0] }; + self.0 * x + } + } + + fn geom_cols() -> Vec { + 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 + ] + } + + #[test] + fn withholds_until_geometry_present() { + let mut r = CostRunner::new(StubCost(2.0)); + let inputs = geom_cols(); // empty + assert_eq!(r.eval(Ctx::new(&inputs, Timestamp(0))), None); + } + + #[test] + fn charges_numerator_over_latched_on_close() { + let mut r = CostRunner::new(StubCost(2.0)); + let mut inputs = geom_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(96.0)).unwrap(); // latched 4 -> 2/4 = 0.5 + assert_eq!( + r.eval(Ctx::new(&inputs, Timestamp(0))), + Some([Cell::from_f64(0.5), Cell::from_f64(0.5), Cell::from_f64(0.0)].as_slice()) + ); + } + + #[test] + fn open_emits_would_be_cost_not_in_cum() { + let mut r = CostRunner::new(StubCost(2.0)); + let mut inputs = geom_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(); + assert_eq!( + r.eval(Ctx::new(&inputs, Timestamp(0))), + Some([Cell::from_f64(0.0), Cell::from_f64(0.0), Cell::from_f64(0.5)].as_slice()) + ); + } + + #[test] + fn zero_latched_no_cost() { + let mut r = CostRunner::new(StubCost(2.0)); + let mut inputs = geom_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 + assert_eq!( + r.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() { + let mut r = CostRunner::new(StubCost(2.0)); + let mut a = geom_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(); // 0.5 + let _ = r.eval(Ctx::new(&a, Timestamp(0))); + let mut b = geom_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 -> 1.0; cum 1.5 + assert_eq!( + r.eval(Ctx::new(&b, Timestamp(1))), + Some([Cell::from_f64(1.0), Cell::from_f64(1.5), Cell::from_f64(0.0)].as_slice()) + ); + } + + #[test] + fn extra_input_cold_contributes_zero_but_row_emits() { + // Co-temporality: a not-yet-warm factor input -> 0 cost, but a row IS emitted. + let mut r = CostRunner::new(StubExtra(0.5)); + let mut inputs = geom_cols(); + inputs.push(AnyColumn::with_capacity(ScalarKind::F64, 1)); // x, empty + 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(96.0)).unwrap(); + assert_eq!( + r.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 extra_input_warm_scales_numerator() { + let mut r = CostRunner::new(StubExtra(0.5)); + let mut inputs = geom_cols(); + inputs.push(AnyColumn::with_capacity(ScalarKind::F64, 1)); + 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(96.0)).unwrap(); // latched 4 + inputs[4].push(Scalar::f64(3.0)).unwrap(); // x=3 -> 0.5*3=1.5 -> 1.5/4 = 0.375 + assert_eq!( + r.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 lookbacks_count_geometry_plus_extra() { + assert_eq!(CostRunner::new(StubCost(1.0)).lookbacks(), vec![1; GEOMETRY_WIDTH]); + assert_eq!(CostRunner::new(StubExtra(1.0)).lookbacks(), vec![1; GEOMETRY_WIDTH + 1]); + } + + #[test] + fn runner_label_delegates_to_factor() { + assert_eq!(CostRunner::new(StubCost(2.0)).label(), "StubCost(2)"); + } + + #[test] + fn geometry_width_matches_port_count() { + assert_eq!(GEOMETRY_WIDTH, geometry_input_ports().len()); + } + + #[test] + fn cost_output_fields_are_the_triple() { + let names: Vec = cost_output_fields().into_iter().map(|f| f.name).collect(); + assert_eq!(names, COST_FIELD_NAMES.to_vec()); + } + + #[test] + fn cost_node_builder_assembles_geometry_prefix_then_extras() { + // The builder's input-assembly contract, at the schema level: the 4 geometry + // ports first (in slot order), the factor's extra inputs appended beginning + // at GEOMETRY_WIDTH, the shared 3-field cost output, and the given params + // passed through verbatim. + let extra = vec![PortSpec { kind: ScalarKind::F64, firing: Firing::Any, name: "vol".into() }]; + let params = vec![ParamSpec { name: "k".into(), kind: ScalarKind::F64 }]; + let builder = cost_node_builder( + "Probe", + extra.clone(), + params.clone(), + |_p| -> Box { Box::new(CostRunner::new(StubCost(0.0))) }, + ); + let schema = builder.schema(); + // The first GEOMETRY_WIDTH inputs are the geometry prefix verbatim... + assert_eq!(schema.inputs[..GEOMETRY_WIDTH], geometry_input_ports()[..]); + // ...and the factor's extras begin exactly at slot GEOMETRY_WIDTH. + assert_eq!(schema.inputs[GEOMETRY_WIDTH..], extra[..]); + assert_eq!(schema.inputs.len(), GEOMETRY_WIDTH + extra.len()); + // The output is the shared 3-field cost triple; params pass through. + assert_eq!(schema.output, cost_output_fields()); + assert_eq!(schema.params, params); + } + + #[test] + fn producer_and_aggregator_share_the_triple() { + // The structural lockstep: producer output and aggregator output/inputs all + // read COST_FIELD_NAMES (one source), replacing the by-convention triple. + let prod: Vec = + ConstantCost::builder().schema().output.iter().map(|f| f.name.clone()).collect(); + assert_eq!(prod, COST_FIELD_NAMES.to_vec()); + let agg_out: Vec = + CostSum::builder(1).schema().output.iter().map(|f| f.name.clone()).collect(); + assert_eq!(agg_out, COST_FIELD_NAMES.to_vec()); + let agg_in: Vec = + CostSum::builder(1).schema().inputs.iter().map(|p| p.name.clone()).collect(); + let expected: Vec = + COST_FIELD_NAMES.iter().map(|f| format!("cost[0].{f}")).collect(); + assert_eq!(agg_in, expected); + } +} diff --git a/crates/aura-std/src/cost_sum.rs b/crates/aura-std/src/cost_sum.rs index af0ceaa..645a1b2 100644 --- a/crates/aura-std/src/cost_sum.rs +++ b/crates/aura-std/src/cost_sum.rs @@ -10,14 +10,12 @@ use aura_core::{ Cell, Ctx, FieldSpec, Firing, Node, NodeSchema, PortSpec, PrimitiveBuilder, ScalarKind, }; -/// The cost-record field triple every cost node emits, in slot order, and its -/// width. Interned once (mirroring `position_management::FIELD_NAMES`/`WIDTH`) so -/// the input-name loop, the output schema, the lookback vector, and the eval -/// accumulator all read one source of truth instead of restating the triple. The -/// producer nodes (`ConstantCost`, `VolSlippageCost`) emit this same triple; that -/// cross-node match remains a by-convention lockstep contract. -const COST_FIELDS: [&str; 3] = ["cost_in_r", "cum_cost_in_r", "open_cost_in_r"]; -const COST_WIDTH: usize = COST_FIELDS.len(); +/// The cost-record field triple and its width come from the shared cost contract +/// (`crate::cost::{COST_FIELD_NAMES, COST_WIDTH}`) — one source of truth, read by +/// both the producer side (`cost_node_builder`) and this aggregator. The input-name +/// loop, the output schema, the lookback vector, and the eval accumulator all read +/// it, so the producer↔aggregator field match is structural, not by-convention. +use crate::cost::{COST_FIELD_NAMES, COST_WIDTH}; /// Per-field sum of `n_costs` cost-in-R records. Inputs are /// `cost[k].{cost_in_r,cum_cost_in_r,open_cost_in_r}` for `k in 0..n_costs`, in @@ -40,7 +38,7 @@ impl CostSum { pub fn builder(n_costs: usize) -> PrimitiveBuilder { let mut inputs = Vec::with_capacity(n_costs * COST_WIDTH); for k in 0..n_costs { - for field in COST_FIELDS { + for field in COST_FIELD_NAMES { inputs.push(PortSpec { kind: ScalarKind::F64, firing: Firing::Any, @@ -52,7 +50,7 @@ impl CostSum { "CostSum", NodeSchema { inputs, - output: COST_FIELDS + output: COST_FIELD_NAMES .iter() .map(|n| FieldSpec { name: (*n).into(), kind: ScalarKind::F64 }) .collect(), @@ -69,7 +67,7 @@ impl Node for CostSum { } fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Cell]> { - let mut acc = [0.0_f64; COST_WIDTH]; // per-field accumulator, COST_FIELDS order + let mut acc = [0.0_f64; COST_WIDTH]; // per-field accumulator, COST_FIELD_NAMES order for k in 0..self.n_costs { for (f, slot) in acc.iter_mut().enumerate() { let w = ctx.f64_in(k * COST_WIDTH + f); diff --git a/crates/aura-std/src/lib.rs b/crates/aura-std/src/lib.rs index 48f9a60..2c51089 100644 --- a/crates/aura-std/src/lib.rs +++ b/crates/aura-std/src/lib.rs @@ -19,6 +19,7 @@ mod add; mod and; mod bias; mod constant_cost; +mod cost; mod cost_sum; mod delay; mod ema; @@ -47,6 +48,7 @@ pub use add::Add; pub use and::And; pub use bias::Bias; pub use constant_cost::ConstantCost; +pub use cost::{cost_node_builder, CostNode, CostRunner, COST_FIELD_NAMES, COST_WIDTH}; pub use cost_sum::CostSum; pub use delay::Delay; pub use ema::Ema; diff --git a/crates/aura-std/src/vol_slippage_cost.rs b/crates/aura-std/src/vol_slippage_cost.rs index 99518fa..695a074 100644 --- a/crates/aura-std/src/vol_slippage_cost.rs +++ b/crates/aura-std/src/vol_slippage_cost.rs @@ -1,108 +1,71 @@ //! `VolSlippageCost` — a slippage cost that scales with a measured volatility -//! input, charged once per closed trade, in R. The second cost node of the C10 -//! cost-model graph and the first *state-dependent* one: identical in shape to -//! [`crate::ConstantCost`] but its per-trade charge numerator is -//! `slip_vol_mult · volatility` instead of a flat constant, so the cost-in-R -//! varies trade-to-trade. R-pure: `slip_vol_mult · vol / |entry - stop|`; -//! notional cancels (C10). The vol is supplied as an 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. +//! 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 contract: a cost node's stream must stay 1:1 with the -//! executor's PM record (the positional join `summarize_r` relies on). So the -//! node is gated ONLY by the PM geometry; a not-yet-warm `volatility` input (the -//! realized-range proxy warms later than PM) contributes 0 cost that cycle rather -//! than withholding and desyncing the stream — honest, since there is no slippage -//! estimate yet. This generalizes to any state-dependent cost factor. +//! 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::{ - Cell, Ctx, FieldSpec, Firing, Node, NodeSchema, ParamSpec, PortSpec, PrimitiveBuilder, - ScalarKind, -}; +use aura_core::{Ctx, Firing, ParamSpec, PortSpec, PrimitiveBuilder, ScalarKind}; -/// A volatility-scaled per-trade slippage, emitted in R. Inputs are the four -/// executor-exposed geometry fields `closed`/`open`/`entry_price`/`stop_price` -/// plus a `volatility` stream (price units). Withholds (`None`) only until the -/// four geometry inputs are present; a not-yet-warm `volatility` contributes 0 -/// cost, so the stream stays co-temporal with the PM record. +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 { + 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, - cum: f64, - out: [Cell; 3], } impl VolSlippageCost { - pub fn new(slip_vol_mult: f64) -> Self { + /// A volatility-scaled slippage cost node (the factor wrapped in the runner). + pub fn new(slip_vol_mult: f64) -> CostRunner { assert!(slip_vol_mult >= 0.0, "VolSlippageCost slip_vol_mult must be >= 0"); - Self { slip_vol_mult, cum: 0.0, out: [Cell::from_f64(0.0); 3] } + CostRunner::new(VolSlippageCost { slip_vol_mult }) } - /// The param-generic recipe: one `slip_vol_mult` F64 knob; five inputs. + /// The param-generic recipe: one `slip_vol_mult` F64 knob; one extra + /// `volatility` input appended after the geometry. pub fn builder() -> PrimitiveBuilder { - PrimitiveBuilder::new( + cost_node_builder( "VolSlippageCost", - NodeSchema { - inputs: vec![ - PortSpec { kind: ScalarKind::Bool, firing: Firing::Any, name: "closed".into() }, - PortSpec { kind: ScalarKind::Bool, firing: Firing::Any, name: "open".into() }, - PortSpec { kind: ScalarKind::F64, firing: Firing::Any, name: "entry_price".into() }, - PortSpec { kind: ScalarKind::F64, firing: Firing::Any, name: "stop_price".into() }, - PortSpec { kind: ScalarKind::F64, firing: Firing::Any, name: "volatility".into() }, - ], - output: vec![ - FieldSpec { name: "cost_in_r".into(), kind: ScalarKind::F64 }, - FieldSpec { name: "cum_cost_in_r".into(), kind: ScalarKind::F64 }, - FieldSpec { name: "open_cost_in_r".into(), kind: ScalarKind::F64 }, - ], - params: vec![ParamSpec { name: "slip_vol_mult".into(), kind: ScalarKind::F64 }], - }, + volatility_input_ports(), + vec![ParamSpec { name: "slip_vol_mult".into(), kind: ScalarKind::F64 }], |p| Box::new(VolSlippageCost::new(p[0].f64())), ) } } -impl Node for VolSlippageCost { - fn lookbacks(&self) -> Vec { - vec![1, 1, 1, 1, 1] - } - - fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Cell]> { - let closed_w = ctx.bool_in(0); - let open_w = ctx.bool_in(1); - let entry_w = ctx.f64_in(2); - let stop_w = ctx.f64_in(3); - let vol_w = ctx.f64_in(4); - // Gate only on the PM geometry (co-temporality contract); a not-yet-warm - // volatility proxy contributes 0 cost rather than withholding the row. - if closed_w.is_empty() || open_w.is_empty() || entry_w.is_empty() || stop_w.is_empty() { - return None; - } - let closed = closed_w[0]; - let open = open_w[0]; - let latched = (entry_w[0] - stop_w[0]).abs(); - let vol = if vol_w.is_empty() { 0.0 } else { vol_w[0] }; // 0 during proxy warm-up - // Same zero-latched guard as ConstantCost: no valid 1R denominator -> no cost. - let per = if latched > 0.0 { self.slip_vol_mult * vol / latched } else { 0.0 }; - let cost_in_r = if closed { per } else { 0.0 }; - let open_cost_in_r = if open { per } else { 0.0 }; - self.cum += cost_in_r; - self.out = [ - Cell::from_f64(cost_in_r), - Cell::from_f64(self.cum), - Cell::from_f64(open_cost_in_r), - ]; - Some(&self.out) - } - +impl CostNode for VolSlippageCost { fn label(&self) -> String { format!("VolSlippageCost({})", self.slip_vol_mult) } + fn extra_inputs(&self) -> Vec { + 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, Scalar, Timestamp}; + use aura_core::{AnyColumn, Cell, Node, Scalar, Timestamp}; fn cols() -> Vec { vec![ @@ -219,4 +182,15 @@ mod tests { 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); + } }