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

feat(0082): cost-graph composition — VolSlippageCost + CostSum net-R aggregate

Browse Source 
Cycle 2 of the "Cost-model graph (in R)" milestone (#148): the milestone's real
architectural claim — the cost graph composes. Two cost nodes now sum into one
net-R curve while summarize_r and the net_r_equity tap stay structurally
unchanged.

- VolSlippageCost (aura-std): a second, STATE-DEPENDENT cost node; per-trade
  charge = slip_vol_mult * volatility / |entry-stop|, in R. The vol is an
  independent short-horizon realized range (SLIP_VOL_LENGTH=5, distinct from the
  stop's EWMA-3) — scaling slippage by the stop's own vol would collapse cost-in-R
  to a constant (indistinguishable from ConstantCost).
- CostSum (aura-std): the cost-graph OUTPUT node — sums N cost nodes' 3-field
  cost-in-R records per-field into one aggregate. summarize_r and net_r_equity
  read the aggregate (one home for cost; n=1 is the identity, so the cost path is
  uniform). A future CostNode trait (deferred) will unify the cost-triple the two
  producer nodes currently restate by-convention.
- Run path: --cost-per-trade and --slip-vol-mult combine, their costs summing into
  the net-R curve; a hoisted vol proxy (RollingMax-RollingMin) keeps the single
  feed. Run-path-scoped; sweep/walkforward/mc pass None.

Co-temporality contract (the load-bearing design decision; corrected from the
signed spec after the implement-loop correctly BLOCKED Task 3 on it). summarize_r
positional-joins cost[i] <-> record[i], so the cost stream must be co-temporal
1:1 with the PM record. A cost node is therefore gated ONLY by the PM geometry
(closed/open/entry/stop); a not-yet-warm state input (the vol proxy warms later
than PM) contributes 0 cost that cycle rather than withholding and desyncing the
stream. This makes co-temporality structural + warmup-independent, preserves the
C18 golden, and generalizes to any future cost factor. The rejected alternative
(a key-join in summarize_r) would have moved the golden and pushed cost-graph
logic into the post-run fold. Recorded on #148.

Tests: VolSlippageCost + CostSum unit sets (incl. the co-temporal-zero-during-
warmup case); the CLI composition run (both flags -> net_both < net_flat,
net_r_equity persisted); node-level EXACT additive composition (net_both ==
net_flat + net_vol - gross over the real nodes), the aggregate net_r_equity ==
post-run net total, and CostSum(1) identity. Full workspace suite green; clippy
-D warnings clean; the no-cost C18 golden byte-identical (the regression floor).

Deferred (later cycles of this milestone): the general CostNode trait + cost-graph
composite-builder (now justified by two concrete nodes), the conviction-weighting
R-aggregation axis, and cost on the reduce-mode sweep path. Spec + plan amended to
the corrected contract; both are cycle ephemera (git rm at cycle close).

refs #148
This commit is contained in:
Brummel
2026-06-28 16:51:48 +02:00
parent 31ef46314a
commit 7f3756a395
8 changed files with 725 additions and 59 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/aura-cli/src/main.rs
+132 -34
View File
@@ -29,9 +29,9 @@ use aura_registry::{
FamilyMember, Generalization, NameKind, PlateauMode, Registry, RunTraces, TraceStore, WriteKind,
};
use aura_std::{
Add, Bias, ConstantCost, Delay, Ema, GatedRecorder, Gt, Latch, LinComb, LongOnly, Mul, Recorder,
RollingMax, RollingMin, SeriesReducer, SimBroker, Sma, Sqrt, Sub, PM_FIELD_NAMES,
PM_RECORD_KINDS,
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,
};
use std::sync::mpsc::{self, Receiver};
use std::sync::LazyLock;
@@ -2565,6 +2565,41 @@ const STAGE1_R_STOP_LENGTH: i64 = 3;
/// `StopRule::Vol` and its manifest record, like [`STAGE1_R_STOP_LENGTH`].
const STAGE1_R_STOP_K: f64 = 2.0;
/// Short-horizon realized-range window for vol-scaled slippage. Deliberately
/// distinct from `STAGE1_R_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 maximum number of cost nodes the run-path cost graph wires (flat cost +
/// 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].<field>` `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<Vec<String>> = LazyLock::new(|| {
let mut v = Vec::with_capacity(MAX_RUN_COST_NODES * 3);
for k in 0..MAX_RUN_COST_NODES {
for field in COST_FIELDS {
v.push(format!("cost[{k}].{field}"));
}
}
v
});
/// Which cost nodes the run-path cost graph builds. At least one field is `Some`
/// (the carrier `Option` is `None` when no cost flag was given).
struct CostConfig {
const_cost: Option<f64>, // --cost-per-trade
slip_vol_mult: Option<f64>, // --slip-vol-mult
}
/// 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.
enum RunData {
@@ -2621,7 +2656,7 @@ fn stage1_r_graph(
slow_len: Option<i64>,
stop_open: bool,
reduce: bool,
cost: Option<(f64, mpsc::Sender<(Timestamp, Vec<Scalar>)>, mpsc::Sender<(Timestamp, Vec<Scalar>)>)>,
cost: Option<(CostConfig, mpsc::Sender<(Timestamp, Vec<Scalar>)>, mpsc::Sender<(Timestamp, Vec<Scalar>)>)>,
) -> Composite {
let mut g = GraphBuilder::new("stage1_r");
// SMA-cross signal → Bias (the same signal the pip sample uses).
@@ -2669,11 +2704,32 @@ fn stage1_r_graph(
} else {
g.add(Recorder::builder(PM_RECORD_KINDS.to_vec(), Firing::Any, tx_r))
};
// Hoisted above the single main feed: the short-horizon vol proxy iff a
// vol-slippage cost is actually wired (run path, non-reduce), so its `price`
// inputs join the one `price_targets` array (no second feed call).
let vol_proxy = match &cost {
Some((cfg, _, _)) if !reduce && cfg.slip_vol_mult.is_some() => {
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"));
Some((vhi, vlo, vrange))
}
_ => None,
};
let price = g.source_role("price", ScalarKind::F64);
g.feed(
price,
[fast.input("series"), slow.input("series"), broker.input("price"), exec.input("price")],
);
let mut price_targets = vec![
fast.input("series"),
slow.input("series"),
broker.input("price"),
exec.input("price"),
];
if let Some((vhi, vlo, _)) = vol_proxy {
price_targets.push(vhi.input("series"));
price_targets.push(vlo.input("series"));
}
g.feed(price, price_targets);
g.connect(fast.output("value"), spread.input("lhs"));
g.connect(slow.output("value"), spread.input("rhs"));
g.connect(spread.output("value"), exposure.input("signal"));
@@ -2695,15 +2751,39 @@ fn stage1_r_graph(
g.connect(exec.output("cum_realized_r"), r_equity.input("term[0]"));
g.connect(exec.output("unrealized_r"), r_equity.input("term[1]"));
g.connect(r_equity.output("value"), req.input("col[0]"));
if let Some((cost_per_trade, tx_net, tx_cost)) = cost {
let cost_node = g.add(
ConstantCost::builder().bind("cost_per_trade", Scalar::f64(cost_per_trade)),
);
g.connect(exec.output("closed_this_cycle"), cost_node.input("closed"));
g.connect(exec.output("open"), cost_node.input("open"));
g.connect(exec.output("entry_price"), cost_node.input("entry_price"));
g.connect(exec.output("stop_price"), cost_node.input("stop_price"));
// net_r_equity = cum_realized_r + unrealized_r - cum_cost_in_r - open_cost_in_r
if let Some((cfg, tx_net, tx_cost)) = cost {
let n = cfg.const_cost.is_some() as usize + cfg.slip_vol_mult.is_some() as usize;
let agg = g.add(CostSum::builder(n));
let mut slot = 0usize;
if let Some(cpt) = cfg.const_cost {
let cc = g.add(ConstantCost::builder().bind("cost_per_trade", Scalar::f64(cpt)));
g.connect(exec.output("closed_this_cycle"), cc.input("closed"));
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()));
}
slot += 1;
}
if let Some(svm) = cfg.slip_vol_mult {
let (_, _, vrange) = vol_proxy.expect("vol proxy is built whenever slip_vol_mult is set");
let vs = g.add(VolSlippageCost::builder().bind("slip_vol_mult", Scalar::f64(svm)));
g.connect(exec.output("closed_this_cycle"), vs.input("closed"));
g.connect(exec.output("open"), vs.input("open"));
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()));
}
slot += 1;
}
debug_assert_eq!(slot, n);
// net_r_equity = cum_realized_r + unrealized_r - Σcum_cost_in_r - Σopen_cost_in_r
let net_eq = g.add(
LinComb::builder(4)
.bind("weights[0]", Scalar::f64(1.0))
@@ -2713,22 +2793,20 @@ fn stage1_r_graph(
);
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(cost_node.output("cum_cost_in_r"), net_eq.input("term[2]"));
g.connect(cost_node.output("open_cost_in_r"), net_eq.input("term[3]"));
g.connect(agg.output("cum_cost_in_r"), net_eq.input("term[2]"));
g.connect(agg.output("open_cost_in_r"), net_eq.input("term[3]"));
let net_rec = g.add(Recorder::builder(vec![ScalarKind::F64], Firing::Any, tx_net));
g.connect(net_eq.output("value"), net_rec.input("col[0]"));
// The cost stream `summarize_r` folds: the `ConstantCost` node's full
// [cost_in_r, cum_cost_in_r, open_cost_in_r] schema. `summarize_r`'s `cost_col`
// reads col 0 (per-close) and col 2 (window-end open) — the lockstep contract
// with aura-std's ConstantCost output — so the stream is recorded 3-wide.
// The aggregate cost record summarize_r folds (col 0 per-close, col 2 window-end).
let cost_rec = g.add(Recorder::builder(
vec![ScalarKind::F64, ScalarKind::F64, ScalarKind::F64],
Firing::Any,
tx_cost,
));
g.connect(cost_node.output("cost_in_r"), cost_rec.input("col[0]"));
g.connect(cost_node.output("cum_cost_in_r"), cost_rec.input("col[1]"));
g.connect(cost_node.output("open_cost_in_r"), cost_rec.input("col[2]"));
g.connect(agg.output("cost_in_r"), cost_rec.input("col[0]"));
g.connect(agg.output("cum_cost_in_r"), cost_rec.input("col[1]"));
g.connect(agg.output("open_cost_in_r"), cost_rec.input("col[2]"));
}
}
g.build().expect("stage1_r wiring resolves")
@@ -2953,7 +3031,12 @@ fn stage1_meanrev_graph(
/// round-trip cost per trade (price units): `Some(c)` wires the `ConstantCost` node and
/// the `net_r_equity` tap and folds the cost stream into `net_expectancy_r`; `None` is the
/// frictionless gross-R baseline (net == gross, no extra tap, byte-unchanged).
fn run_stage1_r(data: RunData, trace: Option<&str>, cost: Option<f64>) -> RunReport {
fn run_stage1_r(
data: RunData,
trace: Option<&str>,
const_cost: Option<f64>,
slip_vol_mult: Option<f64>,
) -> RunReport {
if let Some(n) = trace
&& let Err(e) = TraceStore::open("runs").ensure_name_free(n, WriteKind::Run)
{
@@ -2966,7 +3049,11 @@ fn run_stage1_r(data: RunData, trace: Option<&str>, cost: Option<f64>) -> RunRep
let (tx_req, rx_req) = mpsc::channel();
let (tx_net, rx_net) = mpsc::channel();
let (tx_cost, rx_cost) = mpsc::channel();
let cost_bundle = cost.map(|c| (c, tx_net, tx_cost));
let cost_bundle = if const_cost.is_some() || slip_vol_mult.is_some() {
Some((CostConfig { const_cost, slip_vol_mult }, tx_net, tx_cost))
} else {
None
};
let flat = stage1_r_graph(tx_eq, tx_ex, tx_r, tx_req, Some(2), Some(4), false, false, cost_bundle)
.compile_with_params(&[])
.expect("valid stage1-r blueprint");
@@ -3061,16 +3148,17 @@ struct RunArgs {
data: RunData,
trace: Option<String>,
cost: Option<f64>,
slip_vol_mult: Option<f64>,
}
/// Parse the `run` tail: `[--harness <name>] [--real <SYM> [--from <ms>] [--to <ms>]]
/// [--trace <name>] [--cost-per-trade <f64>]` in any order, each flag at most once.
/// [--trace <name>] [--cost-per-trade <f64>] [--slip-vol-mult <f64>]` in any order, each flag at most once.
/// `--harness sma` is the default (`--harness macd` / `--harness stage1-r` select the
/// others); `--from`/`--to` are legal only with `--real`. Pure (no I/O, no exit) so the
/// grammar is unit-testable; `main` does the side effects.
fn parse_run_args(rest: &[&str]) -> Result<RunArgs, String> {
let usage = || {
"usage: aura run [--harness <sma|macd|stage1-r>] [--real <SYMBOL> [--from <ms>] [--to <ms>]] [--trace <name>] [--cost-per-trade <f64>]"
"usage: aura run [--harness <sma|macd|stage1-r>] [--real <SYMBOL> [--from <ms>] [--to <ms>]] [--trace <name>] [--cost-per-trade <f64>] [--slip-vol-mult <f64>]"
.to_string()
};
let mut harness: Option<HarnessKind> = None;
@@ -3079,6 +3167,7 @@ fn parse_run_args(rest: &[&str]) -> Result<RunArgs, String> {
let mut to: Option<i64> = None;
let mut trace: Option<String> = None;
let mut cost: Option<f64> = None;
let mut slip_vol_mult: Option<f64> = None;
let mut tail = rest;
while let Some((flag, t)) = tail.split_first() {
match *flag {
@@ -3120,6 +3209,15 @@ fn parse_run_args(rest: &[&str]) -> Result<RunArgs, String> {
cost = Some(value.parse().map_err(|_| usage())?);
tail = t;
}
"--slip-vol-mult" if slip_vol_mult.is_none() => {
let (value, t) = t.split_first().ok_or_else(usage)?;
let v: f64 = value.parse().map_err(|_| usage())?;
if v < 0.0 {
return Err(usage());
}
slip_vol_mult = Some(v);
tail = t;
}
_ => return Err(usage()),
}
}
@@ -3132,7 +3230,7 @@ fn parse_run_args(rest: &[&str]) -> Result<RunArgs, String> {
Some(s) => RunData::Real { symbol: s, from, to },
None => RunData::Synthetic,
};
Ok(RunArgs { harness, data, trace, cost })
Ok(RunArgs { harness, data, trace, cost, slip_vol_mult })
}
/// Route a parsed `run` invocation to its harness handler. The compile-time `match` IS
@@ -3142,7 +3240,7 @@ fn run_dispatch(args: RunArgs) -> Result<RunReport, String> {
Ok(match (args.harness, args.data) {
(HarnessKind::Sma, RunData::Synthetic) => run_sample(trace),
(HarnessKind::Macd, RunData::Synthetic) => run_macd(trace),
(HarnessKind::Stage1R, data) => run_stage1_r(data, trace, args.cost),
(HarnessKind::Stage1R, data) => run_stage1_r(data, trace, args.cost, args.slip_vol_mult),
(HarnessKind::Sma, RunData::Real { symbol, from, to }) => {
run_sample_real(&symbol, from, to, trace)
}
@@ -3153,7 +3251,7 @@ fn run_dispatch(args: RunArgs) -> Result<RunReport, String> {
}
const USAGE: &str =
"usage: aura run [--harness <sma|macd|stage1-r>] [--real <SYMBOL> [--from <ms>] [--to <ms>]] [--trace <name>] [--cost-per-trade <f64>] | aura chart <name> [--tap <t>] [--panels] | aura graph | aura sweep [--strategy <sma|momentum|stage1-r|stage1-breakout|stage1-meanrev>] [--real <SYMBOL> [--from <ms>] [--to <ms>]] [--name <n>|--trace <n>] | aura mc [--name <n>|--trace <n>] | aura mc --strategy stage1-r [--real <SYMBOL> [--from <ms>] [--to <ms>]] [--fast <csv>] [--slow <csv>] [--stop-length <csv>] [--stop-k <csv>] [--block-len <n>] [--resamples <n>] [--seed <n>] | aura walkforward [--strategy <sma|momentum|stage1-r|stage1-breakout|stage1-meanrev>] [--real <SYMBOL> [--from <ms>] [--to <ms>]] [--name <n>|--trace <n>] [--fast <csv>] [--slow <csv>] [--stop-length <csv>] [--stop-k <csv>] [--select <argmax|plateau:mean|plateau:worst>] | aura generalize [--strategy stage1-r] --real <SYM1,SYM2,...> --fast <n> --slow <n> --stop-length <n> --stop-k <f> [--from <ms>] [--to <ms>] [--metric <r-metric>] [--name <n>] | aura runs families | aura runs family <id> [rank <metric>]";
"usage: aura run [--harness <sma|macd|stage1-r>] [--real <SYMBOL> [--from <ms>] [--to <ms>]] [--trace <name>] [--cost-per-trade <f64>] [--slip-vol-mult <f64>] | aura chart <name> [--tap <t>] [--panels] | aura graph | aura sweep [--strategy <sma|momentum|stage1-r|stage1-breakout|stage1-meanrev>] [--real <SYMBOL> [--from <ms>] [--to <ms>]] [--name <n>|--trace <n>] | aura mc [--name <n>|--trace <n>] | aura mc --strategy stage1-r [--real <SYMBOL> [--from <ms>] [--to <ms>]] [--fast <csv>] [--slow <csv>] [--stop-length <csv>] [--stop-k <csv>] [--block-len <n>] [--resamples <n>] [--seed <n>] | aura walkforward [--strategy <sma|momentum|stage1-r|stage1-breakout|stage1-meanrev>] [--real <SYMBOL> [--from <ms>] [--to <ms>]] [--name <n>|--trace <n>] [--fast <csv>] [--slow <csv>] [--stop-length <csv>] [--stop-k <csv>] [--select <argmax|plateau:mean|plateau:worst>] | aura generalize [--strategy stage1-r] --real <SYM1,SYM2,...> --fast <n> --slow <n> --stop-length <n> --stop-k <f> [--from <ms>] [--to <ms>] [--metric <r-metric>] [--name <n>] | aura runs families | aura runs family <id> [rank <metric>]";
fn main() {
// Restore the default SIGPIPE disposition. Rust's runtime sets SIGPIPE to SIG_IGN
@@ -4370,7 +4468,7 @@ mod tests {
fn run_stage1_r_synthetic_folds_an_r_block() {
// the stage1-r harness scores the SMA-cross signal in R: one shell-callable run
// yields a RunReport whose metrics.r is Some with a finite SQN and >= 1 trade.
let report = run_stage1_r(RunData::Synthetic, None, None);
let report = run_stage1_r(RunData::Synthetic, None, None, None);
let r = report.metrics.r.as_ref().expect("stage1-r run must populate metrics.r");
assert!(r.n_trades >= 1, "expected >= 1 trade, got {}", r.n_trades);
assert!(r.sqn.is_finite(), "SQN must be finite, got {}", r.sqn);