Brummel/Aura
1
0
Fork 0
Code Issues 33 Pull Requests Actions Packages Projects Releases Wiki Activity

refactor(cli): wrap_r's optional cost leg — behaviour-preserving rebuild (#234 task 3)

Browse Source 
wrap_r accepts an optional CostLeg (cost node builders + the tx_cost/
tx_net channels); non-empty rebuilds the #221-deleted wiring — the
cost_graph fed by the executor's four geometry outputs, the vol-proxy
RollingMax/RollingMin/Sub (back from cfg(test) to production) feeding
each vol_slippage component, the interned cost[k].* names, a gated
4-col cost recorder in reduce mode (the summarize_r join input), and
the LinComb(4) net_r_equity curve in !reduce. All callers pass None —
behaviour-preserving; the entire suite green unchanged is the gate.
Co-temporality of cost rows with R rows unit-pinned.

Verified: full workspace suite green unchanged, clippy -D warnings
clean; in-loop spec compliant + quality approved (one held Minor: the
CostLeg doc's forward reference to cost_nodes_for, which task 4 lands).

refs #234
This commit is contained in:
Brummel
2026-07-11 08:33:22 +02:00
parent bdcf0452f4
commit a1afc2fd02
2 changed files with 284 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/aura-cli/src/campaign_run.rs
+1 -1
View File
@@ -862,7 +862,7 @@ pub(crate) fn persist_campaign_traces(
let (tx_ex, rx_ex) = mpsc::channel();
let (tx_r, rx_r) = mpsc::channel();
let (tx_req, rx_req) = mpsc::channel();
let mut h = crate::wrap_r(signal, tx_eq, tx_ex, tx_r, tx_req, stop, false, geo.pip_size, &binding)
let mut h = crate::wrap_r(signal, tx_eq, tx_ex, tx_r, tx_req, stop, false, geo.pip_size, &binding, None)
.bootstrap_with_cells(&point)
.expect("the member's point re-bootstraps (it already ran this realization)");
h.run(sources);
crates/aura-cli/src/main.rs
+283 -13
View File
@@ -13,8 +13,8 @@ mod scaffold;
mod verb_sugar;
use render::{ChartData, ChartMeta, ChartMode, ReduceKind, Series};
use aura_core::{zip_params, Cell, Firing, ParamSpec, Scalar, ScalarKind, Timestamp};
use aura_composites::{risk_executor, StopRule};
use aura_core::{zip_params, Cell, Firing, ParamSpec, PrimitiveBuilder, Scalar, ScalarKind, Timestamp};
use aura_composites::{cost_graph, risk_executor, StopRule};
use aura_engine::{
blueprint_from_json, blueprint_to_json, f64_field, join_on_ts, monte_carlo, param_stability,
r_metrics_from_rs, summarize, summarize_r, walk_forward, window_of, BindError, BlueprintNode,
@@ -30,8 +30,8 @@ use aura_registry::{
DEFLATION_BLOCK_LEN, DEFLATION_N_RESAMPLES,
};
use aura_std::{
GatedRecorder, LinComb, Recorder,
SeriesReducer, SimBroker, PM_FIELD_NAMES,
cost_port, GatedRecorder, LinComb, Recorder, RollingMax, RollingMin,
SeriesReducer, SimBroker, Sub, GEOMETRY_WIDTH, PM_FIELD_NAMES,
PM_RECORD_KINDS,
};
// `std_vocabulary` is now only reached through `project::Env::resolve` in production
@@ -60,12 +60,8 @@ use aura_std::{Add, Gt, Latch, Mul, Sqrt};
use aura_engine::ColumnarTrace;
#[cfg(test)]
use aura_std::{Bias, Ema, Sma};
// `RollingMax`/`RollingMin`/`Sub` are now only reached from the test-only
// `r_breakout_signal`/`r_meanrev_signal` carves: #221's `wrap_r` cost-graph
// removal dropped the last production caller (the vol-slippage proxy), mirroring
// `Delay`/`Scale` above.
#[cfg(test)]
use aura_std::{RollingMax, RollingMin, Sub};
use aura_std::{ConstantCost, VolSlippageCost};
use std::sync::mpsc;
use std::sync::LazyLock;
use std::collections::{BTreeMap, HashSet};
@@ -1130,7 +1126,7 @@ fn reproduce_family_in(
eprintln!("aura: {}", binding::synthetic_refusal(&hash, &binding));
std::process::exit(1);
}
let space = wrap_r(reload(), tx_eq, tx_ex, tx_r, tx_req, stop, true, SYNTHETIC_PIP_SIZE, &binding).param_space();
let space = wrap_r(reload(), tx_eq, tx_ex, tx_r, tx_req, stop, true, SYNTHETIC_PIP_SIZE, &binding, None).param_space();
let point = point_from_params(&space, &stored.manifest.params);
// A MonteCarlo member carries no tuning params (the params-join is empty), so its
// reproduce line would print a BLANK member label; the seed IS its realization
@@ -1241,6 +1237,24 @@ pub(crate) const R_SMA_STOP_LENGTH: i64 = 3;
/// `StopRule::Vol` and its manifest record, like [`R_SMA_STOP_LENGTH`].
pub(crate) const R_SMA_STOP_K: f64 = 2.0;
/// Short-horizon realized-range window for vol-scaled slippage. Deliberately
/// distinct from `R_SMA_STOP_LENGTH` (3): scaling slippage by the stop's own
/// vol would collapse cost-in-R to a constant (spec 0082). Short enough to warm
/// within the synthetic smoke fixture so the run path exercises non-zero slippage.
const SLIP_VOL_LENGTH: i64 = 5;
/// The optional cost leg of the R scaffolding (#234): the resolved cost-node
/// builders (from `campaign_run::cost_nodes_for`, fully bound — they add no
/// open param, so the wrapped `param_space` is cost-invariant) plus the two
/// recording channels — the aggregate 3-field cost record (`tx_cost`, the
/// `summarize_r` join input) and the `net_r_equity` curve (`tx_net`, recorded
/// only in `!reduce` trace mode).
struct CostLeg {
nodes: Vec<PrimitiveBuilder>,
tx_cost: mpsc::Sender<(Timestamp, Vec<Scalar>)>,
tx_net: mpsc::Sender<(Timestamp, Vec<Scalar>)>,
}
/// Which data a `run` drives a harness on: the built-in synthetic stream, or real M1
/// close bars for a vetted symbol over an optional window.
#[derive(Debug)]
@@ -1304,6 +1318,7 @@ fn wrap_r(
reduce: bool,
pip_size: f64,
binding: &binding::ResolvedBinding,
cost: Option<CostLeg>,
) -> Composite {
let mut g = GraphBuilder::new("r_sma");
// The strategy signal → Bias, nested as a serializable roles→`bias` leg.
@@ -1382,6 +1397,95 @@ fn wrap_r(
g.connect(exec.output("unrealized_r"), r_equity.input("term[1]"));
g.connect(r_equity.output("value"), req.input("col[0]"));
}
// The optional cost leg (#234 — the #221-deleted wiring rebuilt as a
// campaign-documented feature): a `cost_graph` fed by the executor's four
// geometry outputs, its aggregate record recorded co-temporally with the R
// record (the `summarize_r` positional-join input), and — in `!reduce`
// trace mode — the LinComb(4) net-equity curve into `tx_net`. Absent leg
// (`None`) == today's graph, byte-identical.
if let Some(leg) = cost {
// Components taking a `volatility` extra input (the vol_slippage
// shape), discovered from each builder's own schema past the geometry
// prefix — no second vocabulary of "which node needs the proxy".
let vol_slots: Vec<usize> = leg
.nodes
.iter()
.enumerate()
.filter(|(_, n)| {
n.schema().inputs[GEOMETRY_WIDTH..].iter().any(|p| p.name == "volatility")
})
.map(|(k, _)| k)
.collect();
// The short-horizon realized-range vol proxy, fed from the close role,
// shared by every vol-scaled component (the #221-deleted arm, verbatim
// wiring; built in BOTH modes — the reduce-mode member run charges
// slippage too).
let vol_range = if vol_slots.is_empty() {
None
} else {
let vhi = g.add(RollingMax::builder().named("slip_vol_hi").bind("length", Scalar::i64(SLIP_VOL_LENGTH)));
let vlo = g.add(RollingMin::builder().named("slip_vol_lo").bind("length", Scalar::i64(SLIP_VOL_LENGTH)));
let vrange = g.add(Sub::builder().named("slip_vol_range"));
g.connect(vhi.output("value"), vrange.input("lhs"));
g.connect(vlo.output("value"), vrange.input("rhs"));
g.feed(close, [vhi.input("series"), vlo.input("series")]);
Some(vrange)
};
// One cost_graph composite fans the shared PM-geometry into the
// components and sums their per-field charges (C10).
let cg = g.add(cost_graph(leg.nodes));
g.connect(exec.output("closed_this_cycle"), cg.input("closed"));
g.connect(exec.output("open"), cg.input("open"));
g.connect(exec.output("entry_price"), cg.input("entry_price"));
g.connect(exec.output("stop_price"), cg.input("stop_price"));
for k in vol_slots {
let vrange = vol_range.as_ref().expect("proxy is built whenever a vol slot exists");
g.connect(vrange.output("value"), cg.input(cost_port(k, "volatility")));
}
if reduce {
// The aggregate cost record, gated on the SAME closed flag as the R
// GatedRecorder and flushed once at finalize — so the cost rows stay
// positionally 1:1 with the gated R rows (`summarize_r`'s join). The
// gate rides as an APPENDED col 3: cols 0..2 keep the aura-analysis
// `cost_col` contract (cost_in_r = 0, open_cost_in_r = 2).
let crec = g.add(GatedRecorder::builder(
vec![ScalarKind::F64, ScalarKind::F64, ScalarKind::F64, ScalarKind::Bool],
3,
Firing::Any,
leg.tx_cost,
));
g.connect(cg.output("cost_in_r"), crec.input("col[0]"));
g.connect(cg.output("cum_cost_in_r"), crec.input("col[1]"));
g.connect(cg.output("open_cost_in_r"), crec.input("col[2]"));
g.connect(exec.output("closed_this_cycle"), crec.input("col[3]"));
} else {
// The full per-cycle aggregate cost record (col 0 per-close, col 2
// window-end — what summarize_r folds on the trace path).
let crec = g.add(Recorder::builder(
vec![ScalarKind::F64, ScalarKind::F64, ScalarKind::F64],
Firing::Any,
leg.tx_cost,
));
g.connect(cg.output("cost_in_r"), crec.input("col[0]"));
g.connect(cg.output("cum_cost_in_r"), crec.input("col[1]"));
g.connect(cg.output("open_cost_in_r"), crec.input("col[2]"));
// net_r_equity = cum_realized_r + unrealized_r − Σcum_cost_in_r
// − Σopen_cost_in_r (the #221-deleted LinComb(4), weights 1,1,-1,-1).
let net_eq = g.add(
LinComb::builder(4)
.bind("weights[0]", Scalar::f64(1.0))
.bind("weights[1]", Scalar::f64(1.0))
.bind("weights[2]", Scalar::f64(-1.0))
.bind("weights[3]", Scalar::f64(-1.0)),
);
g.connect(exec.output("cum_realized_r"), net_eq.input("term[0]"));
g.connect(exec.output("unrealized_r"), net_eq.input("term[1]"));
g.connect(cg.output("cum_cost_in_r"), net_eq.input("term[2]"));
g.connect(cg.output("open_cost_in_r"), net_eq.input("term[3]"));
let net_rec = g.add(Recorder::builder(vec![ScalarKind::F64], Firing::Any, leg.tx_net));
g.connect(net_eq.output("value"), net_rec.input("col[0]"));
}
}
g.build().expect("r_sma wiring resolves")
}
@@ -1445,7 +1549,7 @@ fn run_signal_r(
// receiver alive so the sink's sends do not fail, but do not drain it.
let (tx_req, _rx_req) = mpsc::channel();
let (sources, window, pip_size) = resolve_run_data(&data, env, &binding);
let wrapped = wrap_r(signal, tx_eq, tx_ex, tx_r, tx_req, StopRule::Vol { length: R_SMA_STOP_LENGTH, k: R_SMA_STOP_K }, false, pip_size, &binding);
let wrapped = wrap_r(signal, tx_eq, tx_ex, tx_r, tx_req, StopRule::Vol { length: R_SMA_STOP_LENGTH, k: R_SMA_STOP_K }, false, pip_size, &binding, None);
let flat = wrapped
.compile_with_params(params)
.expect("signal binds + wraps to a valid harness");
@@ -1489,7 +1593,7 @@ fn run_blueprint_member(
let (tx_ex, rx_ex) = mpsc::channel();
let (tx_r, rx_r) = mpsc::channel();
let (tx_req, _rx_req) = mpsc::channel();
let mut h = wrap_r(signal, tx_eq, tx_ex, tx_r, tx_req, stop, true, pip, binding)
let mut h = wrap_r(signal, tx_eq, tx_ex, tx_r, tx_req, stop, true, pip, binding, None)
.bootstrap_with_cells(point)
.expect("member bootstraps (point kind-checked against param_space)");
h.run(sources);
@@ -1540,7 +1644,7 @@ fn blueprint_axis_probe(doc: &str, env: &project::Env) -> Composite {
let (tx_ex, _) = mpsc::channel();
let (tx_r, _) = mpsc::channel();
let (tx_req, _) = mpsc::channel();
wrap_r(signal, tx_eq, tx_ex, tx_r, tx_req, StopRule::Vol { length: R_SMA_STOP_LENGTH, k: R_SMA_STOP_K }, true, SYNTHETIC_PIP_SIZE, &probe)
wrap_r(signal, tx_eq, tx_ex, tx_r, tx_req, StopRule::Vol { length: R_SMA_STOP_LENGTH, k: R_SMA_STOP_K }, true, SYNTHETIC_PIP_SIZE, &probe, None)
}
/// `aura sweep <blueprint.json> --list-axes`: one `<name>:<kind>` line per open
@@ -3582,6 +3686,7 @@ mod tests {
false,
SYNTHETIC_PIP_SIZE,
&binding,
None,
)
.compile_with_params(&[])
.expect("closed hl_channel wraps to a valid harness");
@@ -3671,6 +3776,7 @@ mod tests {
false,
SYNTHETIC_PIP_SIZE,
&binding,
None,
)
.compile_with_params(&[])
.expect("r-meanrev signal wraps to a valid harness");
@@ -3694,6 +3800,170 @@ mod tests {
assert_eq!(*bias.last().unwrap(), 1.0, "the down-fade long must hold to the end: {bias:?}");
}
/// #234: the optional cost leg records an aggregate cost stream positionally
/// 1:1 with the R record (`summarize_r`'s positional-join contract) plus a
/// per-cycle net_r_equity curve, in non-reduce trace mode. The cost-less
/// side (`cost: None` == today's graph, byte-identical) is pinned by the
/// suite's untouched equivalence anchors.
#[test]
fn wrap_r_cost_leg_records_cost_rows_co_temporal_with_the_r_record() {
let signal = load_closed_r_sma();
let binding = binding::resolve_binding(signal.name(), signal.input_roles(), &BTreeMap::new())
.expect("the price role resolves");
let (tx_eq, _rx_eq) = mpsc::channel();
let (tx_ex, _rx_ex) = mpsc::channel();
let (tx_r, rx_r) = mpsc::channel();
let (tx_req, _rx_req) = mpsc::channel();
let (tx_cost, rx_cost) = mpsc::channel();
let (tx_net, rx_net) = mpsc::channel();
let leg = CostLeg {
nodes: vec![ConstantCost::builder().bind("cost_per_trade", Scalar::f64(0.0005))],
tx_cost,
tx_net,
};
let flat = wrap_r(
signal,
tx_eq,
tx_ex,
tx_r,
tx_req,
StopRule::Vol { length: R_SMA_STOP_LENGTH, k: R_SMA_STOP_K },
false,
SYNTHETIC_PIP_SIZE,
&binding,
Some(leg),
)
.compile_with_params(&[])
.expect("costed wrap builds");
let mut h = Harness::bootstrap(flat).expect("costed harness bootstraps");
let src: Vec<Box<dyn aura_engine::Source>> = vec![Box::new(VecSource::new(r_sma_prices()))];
h.run(src);
let r_rows: Vec<(Timestamp, Vec<Scalar>)> = rx_r.try_iter().collect();
let cost_rows: Vec<(Timestamp, Vec<Scalar>)> = rx_cost.try_iter().collect();
let net_rows: Vec<(Timestamp, Vec<Scalar>)> = rx_net.try_iter().collect();
assert!(!r_rows.is_empty(), "the fixture must warm the executor");
assert_eq!(
cost_rows.len(),
r_rows.len(),
"the cost stream is positionally 1:1 with the R record (co-temporality)"
);
assert_eq!(net_rows.len(), cost_rows.len(), "the net curve emits per cost row");
assert!(
cost_rows.iter().any(|(_, row)| row[0].as_f64() != 0.0),
"at least one close must charge a non-zero cost (non-vacuous)"
);
}
/// #234: the reduce-mode cost branch (a `GatedRecorder` with the appended
/// `closed_this_cycle` gate column) stays co-temporal with the gated R
/// record too — the same 1:1 join contract as the trace-mode leg above,
/// but over the member/sweep run path `summarize_r` actually consumes.
#[test]
fn wrap_r_cost_leg_in_reduce_mode_stays_co_temporal_with_the_gated_r_record() {
let signal = load_closed_r_sma();
let binding = binding::resolve_binding(signal.name(), signal.input_roles(), &BTreeMap::new())
.expect("the price role resolves");
let (tx_eq, _rx_eq) = mpsc::channel();
let (tx_ex, _rx_ex) = mpsc::channel();
let (tx_r, rx_r) = mpsc::channel();
let (tx_req, _rx_req) = mpsc::channel();
let (tx_cost, rx_cost) = mpsc::channel();
let (tx_net, _rx_net) = mpsc::channel();
let leg = CostLeg {
nodes: vec![ConstantCost::builder().bind("cost_per_trade", Scalar::f64(0.0005))],
tx_cost,
tx_net,
};
let flat = wrap_r(
signal,
tx_eq,
tx_ex,
tx_r,
tx_req,
StopRule::Vol { length: R_SMA_STOP_LENGTH, k: R_SMA_STOP_K },
true,
SYNTHETIC_PIP_SIZE,
&binding,
Some(leg),
)
.compile_with_params(&[])
.expect("costed reduce-mode wrap builds");
let mut h = Harness::bootstrap(flat).expect("costed reduce-mode harness bootstraps");
let src: Vec<Box<dyn aura_engine::Source>> = vec![Box::new(VecSource::new(r_sma_prices()))];
h.run(src);
let r_rows: Vec<(Timestamp, Vec<Scalar>)> = rx_r.try_iter().collect();
let cost_rows: Vec<(Timestamp, Vec<Scalar>)> = rx_cost.try_iter().collect();
assert!(!r_rows.is_empty(), "the fixture must warm and close at least one trade");
assert_eq!(
cost_rows.len(),
r_rows.len(),
"the reduce-mode gated cost stream is positionally 1:1 with the gated R record"
);
assert!(
cost_rows.iter().any(|(_, row)| row[0].as_f64() != 0.0),
"at least one gated close must charge a non-zero cost (non-vacuous)"
);
}
/// #234: a cost component that declares an extra `volatility` port (the
/// `VolSlippageCost` shape) is wired to a live realized-range proxy built
/// from the close role (`RollingMax`/`RollingMin`/`Sub` over
/// `SLIP_VOL_LENGTH`), discovered purely from the component's own schema
/// past the geometry prefix (`vol_slots`) — never a disconnected or
/// hardwired-zero input. If that wiring regresses, `ctx.f64_in(GEOMETRY_WIDTH)`
/// inside the component stays permanently empty and every charge is exactly
/// 0.0 (`vol_not_yet_warm_emits_zero_cost_co_temporally`'s withhold case), so a
/// non-zero charge over a genuinely moving price series is a direct witness
/// that the proxy reached the component. The two tests above use
/// `ConstantCost`, which never touches this wiring at all.
#[test]
fn wrap_r_cost_leg_wires_the_vol_slippage_proxy_from_the_close_role() {
let signal = load_closed_r_sma();
let binding = binding::resolve_binding(signal.name(), signal.input_roles(), &BTreeMap::new())
.expect("the price role resolves");
let (tx_eq, _rx_eq) = mpsc::channel();
let (tx_ex, _rx_ex) = mpsc::channel();
let (tx_r, rx_r) = mpsc::channel();
let (tx_req, _rx_req) = mpsc::channel();
let (tx_cost, rx_cost) = mpsc::channel();
let (tx_net, _rx_net) = mpsc::channel();
let leg = CostLeg {
nodes: vec![VolSlippageCost::builder().bind("slip_vol_mult", Scalar::f64(0.5))],
tx_cost,
tx_net,
};
let flat = wrap_r(
signal,
tx_eq,
tx_ex,
tx_r,
tx_req,
StopRule::Vol { length: R_SMA_STOP_LENGTH, k: R_SMA_STOP_K },
false,
SYNTHETIC_PIP_SIZE,
&binding,
Some(leg),
)
.compile_with_params(&[])
.expect("vol-slippage-costed wrap builds");
let mut h = Harness::bootstrap(flat).expect("vol-slippage-costed harness bootstraps");
let src: Vec<Box<dyn aura_engine::Source>> = vec![Box::new(VecSource::new(r_sma_prices()))];
h.run(src);
let r_rows: Vec<(Timestamp, Vec<Scalar>)> = rx_r.try_iter().collect();
let cost_rows: Vec<(Timestamp, Vec<Scalar>)> = rx_cost.try_iter().collect();
assert!(!r_rows.is_empty(), "the fixture must warm and close at least one trade");
assert_eq!(
cost_rows.len(),
r_rows.len(),
"the vol-slippage cost stream stays positionally 1:1 with the R record"
);
assert!(
cost_rows.iter().any(|(_, row)| row[0].as_f64() != 0.0),
"the volatility proxy must actually feed the component — a disconnected \
proxy leaves cost_in_r at exactly 0.0 on every close: {cost_rows:?}"
);
}
/// Property: the pip a run resolves and stamps into its manifest must be the
/// pip the in-graph `SimBroker` divides by — the resolved (real, per-instrument)
/// pip has to reach the graph, not just the manifest label. The `SimBroker`