feat(0071): stage1 breakout candidate — Donchian channel signal (#137)
A second Stage-1 R strategy candidate for the edge-research milestone, a structurally
different trend mechanic than the refuted SMA-momentum: a Donchian channel breakout.
Swaps only the signal leg of the stage1-r graph —
close -> Delay(1) -> {RollingMax,RollingMin}(N) -> {Gt,Gt} -> {Latch,Latch} ->
Sub = bias in {-1,0,+1} — keeping the vol-stop R definition unchanged, so it screens
under the identical R yardstick (clean A/B vs momentum).
- aura-std: new RollingMax / RollingMin nodes (sliding-window extremum via a monotonic
deque, O(1) amortized; mirror the Sma/Delay ring-buffer shape, warm-up skip-emit,
C7/C8). The +/-1 direction latch composes from two existing Latch nodes + Sub (no new
latch type).
- aura-cli: stage1_breakout_graph + a manual-grid stage1_breakout_sweep_family
(fully-bound graph per point, sidestepping parameter-ganging since one channel length
drives both rolling nodes); --strategy stage1-breakout + --channel grid flag; usage
strings updated.
- Causality (C2): one Delay(1) on close feeds both rolling nodes, so each channel covers
close[t-N..t-1] — the current bar is excluded. Pinned by a contrastive e2e test (a
strictly rising series up-breaks every warmed bar; a window including the current bar
never could) plus a +/-1-latch-hold test.
All existing goldens byte-identical (stage1-r/sma/momentum untouched); folded-no-trace
== raw-trace metrics; workspace tests + clippy green.
refs #137
This commit is contained in:
+203
-7
@@ -20,7 +20,8 @@ use aura_engine::{
|
||||
f64_field, join_on_ts, monte_carlo, param_stability, summarize, summarize_r, walk_forward,
|
||||
window_of, ColumnarTrace, Composite, Edge, FlatGraph, GraphBuilder, Harness, JoinedRow,
|
||||
McAggregate, McFamily, RollMode, RunManifest, RunMetrics, RunReport, SourceSpec, SweepFamily,
|
||||
SyntheticSpec, Target, VecSource, WalkForwardResult, WindowBounds, WindowRoller, WindowRun,
|
||||
SweepPoint, SyntheticSpec, Target, VecSource, WalkForwardResult, WindowBounds, WindowRoller,
|
||||
WindowRun,
|
||||
};
|
||||
use aura_registry::{
|
||||
group_families, mc_member_reports, optimize, rank_by, sweep_member_reports,
|
||||
@@ -28,8 +29,8 @@ use aura_registry::{
|
||||
WriteKind,
|
||||
};
|
||||
use aura_std::{
|
||||
Bias, Ema, GatedRecorder, LinComb, LongOnly, Recorder, SeriesReducer, SimBroker, Sma, Sub,
|
||||
PM_FIELD_NAMES, PM_RECORD_KINDS,
|
||||
Bias, Delay, Ema, GatedRecorder, Gt, Latch, LinComb, LongOnly, Recorder, RollingMax,
|
||||
RollingMin, SeriesReducer, SimBroker, Sma, Sub, PM_FIELD_NAMES, PM_RECORD_KINDS,
|
||||
};
|
||||
use std::sync::mpsc::{self, Receiver};
|
||||
use std::sync::LazyLock;
|
||||
@@ -1126,6 +1127,7 @@ struct Stage1RGrid {
|
||||
slow: Vec<i64>,
|
||||
stop_length: Vec<i64>,
|
||||
stop_k: Vec<f64>,
|
||||
channel: Vec<i64>,
|
||||
}
|
||||
|
||||
impl Default for Stage1RGrid {
|
||||
@@ -1135,6 +1137,7 @@ impl Default for Stage1RGrid {
|
||||
slow: vec![6, 12],
|
||||
stop_length: vec![STAGE1_R_STOP_LENGTH],
|
||||
stop_k: vec![STAGE1_R_STOP_K],
|
||||
channel: vec![1920],
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -1261,6 +1264,79 @@ fn stage1_r_sweep_family(trace: Option<&str>, data: &DataSource, grid: &Stage1RG
|
||||
.expect("the stage1-r named grid matches the stage1-r param-space")
|
||||
}
|
||||
|
||||
/// `aura sweep --strategy stage1-breakout`: sweep the breakout harness over a channel ×
|
||||
/// stop grid. One `channel` length drives BOTH rolling nodes (parameter-ganging, #61),
|
||||
/// so the family iterates the cartesian product MANUALLY with a fully-bound graph per
|
||||
/// point (compile_with_params(&[]) + Harness::bootstrap, like run_stage1_r) rather than
|
||||
/// the open .axis/bootstrap_with_cells path. Each member folds the dense R-record via
|
||||
/// summarize_r, so the family is rankable by sqn / expectancy_r / ... (parity with
|
||||
/// stage1-r). With --trace, raw recorders persist the per-cycle streams.
|
||||
fn stage1_breakout_sweep_family(trace: Option<&str>, data: &DataSource, grid: &Stage1RGrid) -> SweepFamily {
|
||||
let pip = data.pip_size();
|
||||
let window = data.full_window();
|
||||
let mut varying: HashSet<String> = HashSet::new();
|
||||
if grid.channel.len() > 1 {
|
||||
varying.insert("channel".to_string());
|
||||
}
|
||||
if grid.stop_length.len() > 1 {
|
||||
varying.insert("stop_length".to_string());
|
||||
}
|
||||
if grid.stop_k.len() > 1 {
|
||||
varying.insert("stop_k".to_string());
|
||||
}
|
||||
let mut points = Vec::new();
|
||||
for &c in &grid.channel {
|
||||
for &sl in &grid.stop_length {
|
||||
for &sk in &grid.stop_k {
|
||||
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 reduce = trace.is_none();
|
||||
let flat = stage1_breakout_graph(tx_eq, tx_ex, tx_r, tx_req, Some(c), sl, sk, reduce)
|
||||
.compile_with_params(&[])
|
||||
.expect("valid stage1-breakout blueprint");
|
||||
let mut h = Harness::bootstrap(flat).expect("valid stage1-breakout harness");
|
||||
h.run(data.run_sources());
|
||||
let named: Vec<(String, Scalar)> = vec![
|
||||
("channel".to_string(), Scalar::i64(c)),
|
||||
("stop_length".to_string(), Scalar::i64(sl)),
|
||||
("stop_k".to_string(), Scalar::f64(sk)),
|
||||
];
|
||||
let key = member_key(&named, &varying);
|
||||
let mut manifest = sim_optimal_manifest(named, window, 0, pip);
|
||||
manifest.broker = stage1_r_broker_label(pip);
|
||||
let metrics = if reduce {
|
||||
let r_rows: Vec<(Timestamp, Vec<Scalar>)> = rx_r.try_iter().collect();
|
||||
let (total_pips, max_drawdown) = rx_eq
|
||||
.try_iter()
|
||||
.next()
|
||||
.map(|(_, row)| (row[0].as_f64(), row[1].as_f64()))
|
||||
.unwrap_or((0.0, 0.0));
|
||||
let bias_sign_flips =
|
||||
rx_ex.try_iter().next().map(|(_, row)| row[2].as_i64() as u64).unwrap_or(0);
|
||||
let mut m = RunMetrics { total_pips, max_drawdown, bias_sign_flips, r: None };
|
||||
m.r = Some(summarize_r(&r_rows, 0.0));
|
||||
m
|
||||
} else {
|
||||
let eq_rows: Vec<(Timestamp, Vec<Scalar>)> = rx_eq.try_iter().collect();
|
||||
let ex_rows: Vec<(Timestamp, Vec<Scalar>)> = rx_ex.try_iter().collect();
|
||||
let r_rows: Vec<(Timestamp, Vec<Scalar>)> = rx_r.try_iter().collect();
|
||||
let req_rows: Vec<(Timestamp, Vec<Scalar>)> = rx_req.try_iter().collect();
|
||||
if let Some(name) = trace {
|
||||
persist_traces_r(&format!("{name}/{key}"), &manifest, &eq_rows, &ex_rows, &req_rows);
|
||||
}
|
||||
let mut m = summarize(&f64_field(&eq_rows, 0), &f64_field(&ex_rows, 0));
|
||||
m.r = Some(summarize_r(&r_rows, 0.0));
|
||||
m
|
||||
};
|
||||
points.push(SweepPoint { params: vec![], report: RunReport { manifest, metrics } });
|
||||
}
|
||||
}
|
||||
}
|
||||
SweepFamily { space: vec![], points }
|
||||
}
|
||||
|
||||
/// Render a sweep family as one `RunReport` JSON line per point. Test helper:
|
||||
/// production (`run_sweep`) renders *and* persists per point.
|
||||
#[cfg(test)]
|
||||
@@ -1287,6 +1363,7 @@ enum Strategy {
|
||||
SmaCross,
|
||||
Momentum,
|
||||
Stage1R,
|
||||
Stage1Breakout,
|
||||
}
|
||||
|
||||
/// Parse a comma-separated list of `T` (each item parsed via `FromStr`), rejecting
|
||||
@@ -1306,7 +1383,7 @@ fn parse_csv_list<T: std::str::FromStr>(s: &str) -> Result<Vec<T>, ()> {
|
||||
}
|
||||
|
||||
/// Parse the `sweep` tail:
|
||||
/// `[--strategy <sma|momentum|stage1-r>] [--real <SYMBOL> [--from <ms>] [--to <ms>]] [--name <n> | --trace <n>] [--fast <csv>] [--slow <csv>] [--stop-length <csv>] [--stop-k <csv>]`.
|
||||
/// `[--strategy <sma|momentum|stage1-r|stage1-breakout>] [--real <SYMBOL> [--from <ms>] [--to <ms>]] [--name <n> | --trace <n>] [--fast <csv>] [--slow <csv>] [--stop-length <csv>] [--stop-k <csv>] [--channel <csv>]`.
|
||||
/// Defaults: SMA-cross, synthetic, name "sweep", no persist (today's bare `aura
|
||||
/// sweep`). `--name` and `--trace` are mutually exclusive; `--from`/`--to` (ms,
|
||||
/// `i64`) require `--real` (there is no synthetic window knob). The four optional grid
|
||||
@@ -1319,7 +1396,7 @@ fn parse_csv_list<T: std::str::FromStr>(s: &str) -> Result<Vec<T>, ()> {
|
||||
fn parse_sweep_args(
|
||||
rest: &[&str],
|
||||
) -> Result<(Strategy, String, bool, DataChoice, Stage1RGrid), String> {
|
||||
let usage = || "sweep [--strategy <sma|momentum|stage1-r>] [--real <SYMBOL> [--from <ms>] [--to <ms>]] [--name <n> | --trace <n>] [--fast <csv>] [--slow <csv>] [--stop-length <csv>] [--stop-k <csv>]".to_string();
|
||||
let usage = || "sweep [--strategy <sma|momentum|stage1-r|stage1-breakout>] [--real <SYMBOL> [--from <ms>] [--to <ms>]] [--name <n> | --trace <n>] [--fast <csv>] [--slow <csv>] [--stop-length <csv>] [--stop-k <csv>] [--channel <csv>]".to_string();
|
||||
let mut strategy = Strategy::SmaCross;
|
||||
let mut name: Option<(String, bool)> = None; // (name, persist)
|
||||
let mut real = RealWindowGrammar::default();
|
||||
@@ -1337,6 +1414,7 @@ fn parse_sweep_args(
|
||||
"sma" => Strategy::SmaCross,
|
||||
"momentum" => Strategy::Momentum,
|
||||
"stage1-r" => Strategy::Stage1R,
|
||||
"stage1-breakout" => Strategy::Stage1Breakout,
|
||||
_ => return Err(usage()),
|
||||
};
|
||||
}
|
||||
@@ -1346,6 +1424,7 @@ fn parse_sweep_args(
|
||||
"--slow" => grid.slow = parse_csv_list(value).map_err(|()| usage())?,
|
||||
"--stop-length" => grid.stop_length = parse_csv_list(value).map_err(|()| usage())?,
|
||||
"--stop-k" => grid.stop_k = parse_csv_list(value).map_err(|()| usage())?,
|
||||
"--channel" => grid.channel = parse_csv_list(value).map_err(|()| usage())?,
|
||||
_ => return Err(usage()),
|
||||
}
|
||||
tail = t;
|
||||
@@ -1404,7 +1483,7 @@ fn mc_member_line(id: &str, seed: u64, report: &RunReport) -> String {
|
||||
)
|
||||
}
|
||||
|
||||
/// `aura sweep [--strategy <sma|momentum|stage1-r>] [--name <n>|--trace <n>]`: run the
|
||||
/// `aura sweep [--strategy <sma|momentum|stage1-r|stage1-breakout>] [--name <n>|--trace <n>]`: run the
|
||||
/// selected built-in sweep, persist it as a *family* (related records sharing one
|
||||
/// `family_id`, C18/C21) via `append_family`, and print each point's record line
|
||||
/// carrying the assigned id. With `--trace`, every strategy
|
||||
@@ -1423,6 +1502,7 @@ fn run_sweep(strategy: Strategy, name: &str, persist: bool, data: DataSource, gr
|
||||
Strategy::SmaCross => sweep_family(persist.then_some(name), &data),
|
||||
Strategy::Momentum => momentum_sweep_family(persist.then_some(name), &data),
|
||||
Strategy::Stage1R => stage1_r_sweep_family(persist.then_some(name), &data, grid),
|
||||
Strategy::Stage1Breakout => stage1_breakout_sweep_family(persist.then_some(name), &data, grid),
|
||||
};
|
||||
let id = match reg.append_family(name, FamilyKind::Sweep, &sweep_member_reports(&family)) {
|
||||
Ok(id) => id,
|
||||
@@ -2027,6 +2107,103 @@ fn stage1_r_graph(
|
||||
g.build().expect("stage1_r wiring resolves")
|
||||
}
|
||||
|
||||
/// The stage1-r harness with its signal leg swapped for a Donchian channel breakout:
|
||||
/// `close -> Delay(1) -> {RollingMax,RollingMin}(channel) -> {Gt,Gt} -> {Latch,Latch}
|
||||
/// -> Sub = bias in {-1,0,+1}`. The one Delay(1) on close feeds both rolling nodes, so
|
||||
/// each channel covers `close[t-N..t-1]` (the C2 guard: the current bar is excluded).
|
||||
/// `channel = None` leaves the lengths open (unused today; the family binds them); the
|
||||
/// stop is bound (defines R), identical downstream to stage1_r_graph.
|
||||
#[allow(clippy::type_complexity, clippy::too_many_arguments)]
|
||||
fn stage1_breakout_graph(
|
||||
tx_eq: mpsc::Sender<(Timestamp, Vec<Scalar>)>,
|
||||
tx_ex: mpsc::Sender<(Timestamp, Vec<Scalar>)>,
|
||||
tx_r: mpsc::Sender<(Timestamp, Vec<Scalar>)>,
|
||||
tx_req: mpsc::Sender<(Timestamp, Vec<Scalar>)>,
|
||||
channel: Option<i64>,
|
||||
stop_length: i64,
|
||||
stop_k: f64,
|
||||
reduce: bool,
|
||||
) -> Composite {
|
||||
let mut g = GraphBuilder::new("stage1_breakout");
|
||||
// Donchian breakout signal leg.
|
||||
let delay = g.add(Delay::builder().bind("lag", Scalar::i64(1)));
|
||||
let mut mx_b = RollingMax::builder().named("channel_hi");
|
||||
let mut mn_b = RollingMin::builder().named("channel_lo");
|
||||
if let Some(n) = channel {
|
||||
mx_b = mx_b.bind("length", Scalar::i64(n));
|
||||
mn_b = mn_b.bind("length", Scalar::i64(n));
|
||||
}
|
||||
let mx = g.add(mx_b);
|
||||
let mn = g.add(mn_b);
|
||||
let gt_up = g.add(Gt::builder());
|
||||
let gt_down = g.add(Gt::builder());
|
||||
let up_latch = g.add(Latch::builder());
|
||||
let down_latch = g.add(Latch::builder());
|
||||
let exposure = g.add(Sub::builder()); // up_latch - down_latch -> bias in {-1,0,+1}
|
||||
// pip branch (verbatim from stage1_r_graph).
|
||||
let broker = g.add(SimBroker::builder(SYNTHETIC_PIP_SIZE));
|
||||
let gate_col = PM_FIELD_NAMES
|
||||
.iter()
|
||||
.position(|&n| n == "closed_this_cycle")
|
||||
.expect("PM record has a closed_this_cycle column");
|
||||
let eq = if reduce {
|
||||
g.add(SeriesReducer::builder(Firing::Any, tx_eq))
|
||||
} else {
|
||||
g.add(Recorder::builder(vec![ScalarKind::F64], Firing::Any, tx_eq))
|
||||
};
|
||||
let ex = if reduce {
|
||||
g.add(SeriesReducer::builder(Firing::Any, tx_ex))
|
||||
} else {
|
||||
g.add(Recorder::builder(vec![ScalarKind::F64], Firing::Any, tx_ex))
|
||||
};
|
||||
let exec = g.add(risk_executor(StopRule::Vol { length: stop_length, k: stop_k }, 1.0));
|
||||
let rrec = if reduce {
|
||||
g.add(GatedRecorder::builder(PM_RECORD_KINDS.to_vec(), gate_col, Firing::Any, tx_r))
|
||||
} else {
|
||||
g.add(Recorder::builder(PM_RECORD_KINDS.to_vec(), Firing::Any, tx_r))
|
||||
};
|
||||
let price = g.source_role("price", ScalarKind::F64);
|
||||
g.feed(
|
||||
price,
|
||||
[
|
||||
delay.input("series"),
|
||||
gt_up.input("a"),
|
||||
gt_down.input("b"),
|
||||
broker.input("price"),
|
||||
exec.input("price"),
|
||||
],
|
||||
);
|
||||
g.connect(delay.output("value"), mx.input("series"));
|
||||
g.connect(delay.output("value"), mn.input("series"));
|
||||
g.connect(mx.output("value"), gt_up.input("b"));
|
||||
g.connect(mn.output("value"), gt_down.input("a"));
|
||||
g.connect(gt_up.output("value"), up_latch.input("set"));
|
||||
g.connect(gt_down.output("value"), up_latch.input("reset"));
|
||||
g.connect(gt_down.output("value"), down_latch.input("set"));
|
||||
g.connect(gt_up.output("value"), down_latch.input("reset"));
|
||||
g.connect(up_latch.output("value"), exposure.input("lhs"));
|
||||
g.connect(down_latch.output("value"), exposure.input("rhs"));
|
||||
g.connect(exposure.output("value"), broker.input("exposure"));
|
||||
g.connect(exposure.output("value"), ex.input("col[0]"));
|
||||
g.connect(exposure.output("value"), exec.input("bias"));
|
||||
g.connect(broker.output("equity"), eq.input("col[0]"));
|
||||
for (i, field) in PM_FIELD_NAMES.iter().enumerate() {
|
||||
g.connect(exec.output(field), rrec.input(COL_PORTS[i].as_str()));
|
||||
}
|
||||
if !reduce {
|
||||
let r_equity = g.add(
|
||||
LinComb::builder(2)
|
||||
.bind("weights[0]", Scalar::f64(1.0))
|
||||
.bind("weights[1]", Scalar::f64(1.0)),
|
||||
);
|
||||
let req = g.add(Recorder::builder(vec![ScalarKind::F64], Firing::Any, tx_req));
|
||||
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]"));
|
||||
}
|
||||
g.build().expect("stage1_breakout wiring resolves")
|
||||
}
|
||||
|
||||
/// `aura run --harness stage1-r [--real <SYM> [--from][--to]] [--trace <n>]`: build the
|
||||
/// dual-tap stage1-r harness, run it on synthetic or real M1 data, fold the pip taps via
|
||||
/// `summarize` and the dense R-record via `summarize_r`, and attach the R block as
|
||||
@@ -2209,7 +2386,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>] | aura chart <name> [--tap <t>] [--panels] | aura graph | aura sweep [--strategy <sma|momentum|stage1-r>] [--real <SYMBOL> [--from <ms>] [--to <ms>]] [--name <n>|--trace <n>] | aura mc [--name <n>|--trace <n>] | aura walkforward [--real <SYMBOL> [--from <ms>] [--to <ms>]] [--name <n>|--trace <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>] | aura chart <name> [--tap <t>] [--panels] | aura graph | aura sweep [--strategy <sma|momentum|stage1-r|stage1-breakout>] [--real <SYMBOL> [--from <ms>] [--to <ms>]] [--name <n>|--trace <n>] | aura mc [--name <n>|--trace <n>] | aura walkforward [--real <SYMBOL> [--from <ms>] [--to <ms>]] [--name <n>|--trace <n>] | aura runs families | aura runs family <id> [rank <metric>]";
|
||||
|
||||
fn main() {
|
||||
// Collect argv and match the whole vector: every accepted form is exhaustive,
|
||||
@@ -3158,6 +3335,25 @@ mod tests {
|
||||
assert!(parse_sweep_args(&["--stop-k", "abc"]).is_err()); // non-float
|
||||
}
|
||||
|
||||
/// Property: the breakout `--channel` flag parses a comma-separated list onto
|
||||
/// `Stage1RGrid.channel` (the breakout family's sole signal axis), with the same
|
||||
/// strictness as the stage1-r grid flags — an absent flag keeps the historical
|
||||
/// default `[1920]`, a malformed list is the usage error (not a downstream panic).
|
||||
/// Guards the new flag added alongside the breakout family; without it the
|
||||
/// `--channel` arm could silently drift (e.g. last-write-wins, lenient parse).
|
||||
#[test]
|
||||
fn parse_sweep_args_parses_the_channel_grid_flag() {
|
||||
let parsed = parse_sweep_args(&["--strategy", "stage1-breakout", "--channel", "20,55,100"])
|
||||
.expect("the breakout channel flag parses");
|
||||
assert_eq!(parsed.0, Strategy::Stage1Breakout);
|
||||
assert_eq!(parsed.4.channel, vec![20, 55, 100]);
|
||||
// absent --channel keeps the historical default list.
|
||||
assert_eq!(parse_sweep_args(&[]).unwrap().4.channel, vec![1920]);
|
||||
// a malformed list is the strict usage error.
|
||||
assert!(parse_sweep_args(&["--channel", "20,x"]).is_err()); // non-numeric item
|
||||
assert!(parse_sweep_args(&["--channel", ""]).is_err()); // empty item
|
||||
}
|
||||
|
||||
/// `parse_sweep_args` admits a real symbol with an optional `--from`/`--to`
|
||||
/// window (yielding `DataChoice::Real`), still honouring `--strategy`/`--trace`;
|
||||
/// a `--real` without its symbol, or a window flag without `--real`, is rejected.
|
||||
|
||||
@@ -2053,3 +2053,98 @@ fn sweep_strategy_stage1_r_folded_no_trace_metrics_equal_raw_trace_metrics() {
|
||||
}
|
||||
let _ = std::fs::remove_dir_all(&cwd);
|
||||
}
|
||||
|
||||
/// The breakout strategy reaches the CLI seam: `aura sweep --strategy stage1-breakout`
|
||||
/// emits an R-bearing member, and the folded no-trace path equals the raw --trace path
|
||||
/// byte-for-byte (parity with the stage1-r fold-vs-raw guard). channel 3 warms up on
|
||||
/// the synthetic stream; one channel value × the single default stop = one member.
|
||||
#[test]
|
||||
fn sweep_strategy_stage1_breakout_folded_no_trace_metrics_equal_raw_trace_metrics() {
|
||||
let cwd = temp_cwd("sweep-stage1-breakout-fold-vs-raw");
|
||||
|
||||
let folded = Command::new(BIN)
|
||||
.args(["sweep", "--strategy", "stage1-breakout", "--channel", "3"])
|
||||
.current_dir(&cwd)
|
||||
.output()
|
||||
.expect("spawn folded (no-trace) stage1-breakout sweep");
|
||||
assert!(
|
||||
folded.status.success(),
|
||||
"folded no-trace breakout sweep exit: {:?}; stderr: {}",
|
||||
folded.status,
|
||||
String::from_utf8_lossy(&folded.stderr)
|
||||
);
|
||||
let folded_out = String::from_utf8(folded.stdout).expect("utf-8");
|
||||
|
||||
let raw = Command::new(BIN)
|
||||
.args(["sweep", "--strategy", "stage1-breakout", "--channel", "3", "--trace", "b1"])
|
||||
.current_dir(&cwd)
|
||||
.output()
|
||||
.expect("spawn raw (--trace) stage1-breakout sweep");
|
||||
assert!(
|
||||
raw.status.success(),
|
||||
"raw --trace breakout sweep exit: {:?}; stderr: {}",
|
||||
raw.status,
|
||||
String::from_utf8_lossy(&raw.stderr)
|
||||
);
|
||||
let raw_out = String::from_utf8(raw.stdout).expect("utf-8");
|
||||
|
||||
let folded_lines: Vec<&str> = folded_out.lines().collect();
|
||||
let raw_lines: Vec<&str> = raw_out.lines().collect();
|
||||
assert_eq!(folded_lines.len(), 1, "folded breakout sweep must print 1 member: {folded_out:?}");
|
||||
assert_eq!(
|
||||
raw_lines.len(),
|
||||
folded_lines.len(),
|
||||
"member count must match: folded {} vs raw {}",
|
||||
folded_lines.len(),
|
||||
raw_lines.len()
|
||||
);
|
||||
|
||||
for (i, (f, r)) in folded_lines.iter().zip(raw_lines.iter()).enumerate() {
|
||||
let fm = metrics_object(f);
|
||||
let rm = metrics_object(r);
|
||||
assert!(fm.contains("\"r\":{"), "folded breakout member {i} must carry an r block: {fm}");
|
||||
assert_eq!(
|
||||
fm, rm,
|
||||
"folded vs raw breakout metrics diverge at member {i}\n folded: {fm}\n raw: {rm}"
|
||||
);
|
||||
}
|
||||
let _ = std::fs::remove_dir_all(&cwd);
|
||||
}
|
||||
|
||||
/// Property: the breakout family is the cartesian product of its grid axes — a
|
||||
/// multi-value `--channel` list yields exactly one member per channel value, each
|
||||
/// member carrying its own bound channel length in the manifest params. Guards the
|
||||
/// manual cartesian-product loop in `stage1_breakout_sweep_family`: a regression that
|
||||
/// collapsed the loop (sweeping only the last value, or ganging the two channels)
|
||||
/// would silently drop members here. Channels 2 and 3 both warm up on the 18-bar
|
||||
/// synthetic stream, so two members must appear, in grid order, distinctly.
|
||||
#[test]
|
||||
fn sweep_strategy_stage1_breakout_grids_one_member_per_channel() {
|
||||
let cwd = temp_cwd("sweep-stage1-breakout-channel-grid");
|
||||
let out = Command::new(BIN)
|
||||
.args(["sweep", "--strategy", "stage1-breakout", "--channel", "2,3"])
|
||||
.current_dir(&cwd)
|
||||
.output()
|
||||
.expect("spawn 2-channel stage1-breakout sweep");
|
||||
assert!(
|
||||
out.status.success(),
|
||||
"channel-grid breakout sweep exit: {:?}; stderr: {}",
|
||||
out.status,
|
||||
String::from_utf8_lossy(&out.stderr)
|
||||
);
|
||||
let stdout = String::from_utf8(out.stdout).expect("utf-8");
|
||||
let lines: Vec<&str> = stdout.lines().collect();
|
||||
assert_eq!(lines.len(), 2, "2-channel grid must print 2 members: {stdout:?}");
|
||||
// each member carries its own bound channel length (grid order: 2 then 3).
|
||||
assert!(
|
||||
lines[0].contains("[\"channel\",{\"I64\":2}]"),
|
||||
"first member must bind channel=2: {}",
|
||||
lines[0]
|
||||
);
|
||||
assert!(
|
||||
lines[1].contains("[\"channel\",{\"I64\":3}]"),
|
||||
"second member must bind channel=3: {}",
|
||||
lines[1]
|
||||
);
|
||||
let _ = std::fs::remove_dir_all(&cwd);
|
||||
}
|
||||
|
||||
@@ -0,0 +1,82 @@
|
||||
//! Breakout signal composition: close -> Delay(1) -> {RollingMax,RollingMin}(N) ->
|
||||
//! {Gt,Gt} -> {Latch,Latch} -> Sub = bias in {-1,0,+1}. Tests the C2 causality guard
|
||||
//! (the channel excludes the current bar via Delay(1)) and the held ±1 direction,
|
||||
//! built straight from aura-std nodes (no CLI dependency).
|
||||
|
||||
use aura_core::{Scalar, Timestamp};
|
||||
use aura_engine::{GraphBuilder, Harness, Source, VecSource};
|
||||
use aura_std::{Delay, Gt, Latch, Recorder, RollingMax, RollingMin, Sub};
|
||||
use std::sync::mpsc;
|
||||
|
||||
// A minimal single-f64 source role "price" fed into the breakout signal subgraph,
|
||||
// tapping the Sub (bias) output through a Recorder. channel N = 3.
|
||||
fn run_breakout_bias(closes: &[f64]) -> Vec<(Timestamp, f64)> {
|
||||
let (tx, rx) = mpsc::channel();
|
||||
let mut g = GraphBuilder::new("breakout_sig");
|
||||
let delay = g.add(Delay::builder().bind("lag", Scalar::i64(1)));
|
||||
let mx = g.add(RollingMax::builder().bind("length", Scalar::i64(3)));
|
||||
let mn = g.add(RollingMin::builder().bind("length", Scalar::i64(3)));
|
||||
let gt_up = g.add(Gt::builder());
|
||||
let gt_down = g.add(Gt::builder());
|
||||
let up_latch = g.add(Latch::builder());
|
||||
let down_latch = g.add(Latch::builder());
|
||||
let bias = g.add(Sub::builder());
|
||||
let rec = g.add(Recorder::builder(vec![aura_core::ScalarKind::F64], aura_core::Firing::Any, tx));
|
||||
let price = g.source_role("price", aura_core::ScalarKind::F64);
|
||||
g.feed(price, [delay.input("series"), gt_up.input("a"), gt_down.input("b")]);
|
||||
g.connect(delay.output("value"), mx.input("series"));
|
||||
g.connect(delay.output("value"), mn.input("series"));
|
||||
g.connect(mx.output("value"), gt_up.input("b")); // close > max(prior N)
|
||||
g.connect(mn.output("value"), gt_down.input("a")); // min(prior N) > close
|
||||
g.connect(gt_up.output("value"), up_latch.input("set"));
|
||||
g.connect(gt_down.output("value"), up_latch.input("reset"));
|
||||
g.connect(gt_down.output("value"), down_latch.input("set"));
|
||||
g.connect(gt_up.output("value"), down_latch.input("reset"));
|
||||
g.connect(up_latch.output("value"), bias.input("lhs"));
|
||||
g.connect(down_latch.output("value"), bias.input("rhs"));
|
||||
g.connect(bias.output("value"), rec.input("col[0]"));
|
||||
let flat = g.build().expect("breakout signal wiring resolves").compile_with_params(&[]).expect("compiles");
|
||||
let mut h = Harness::bootstrap(flat).expect("bootstraps");
|
||||
let prices: Vec<(Timestamp, Scalar)> =
|
||||
closes.iter().enumerate().map(|(i, &c)| (Timestamp(i as i64), Scalar::f64(c))).collect();
|
||||
let src: Vec<Box<dyn Source>> = vec![Box::new(VecSource::new(prices))];
|
||||
h.run(src);
|
||||
rx.try_iter().map(|(ts, row): (Timestamp, Vec<Scalar>)| (ts, row[0].as_f64())).collect()
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn breakout_channel_excludes_the_current_bar_c2() {
|
||||
// C2 (no look-ahead), contrastive form: on a strictly monotonically rising series
|
||||
// every warmed bar is a fresh all-time high. The channel is max(close[t-N..t-1])
|
||||
// (Delay(1) excludes the current bar), so close[t] > that prior max on EVERY warmed
|
||||
// bar -> a continuous UP-break -> bias pinned at +1.
|
||||
//
|
||||
// If the channel WRONGLY included the current bar (the look-ahead bug), the test
|
||||
// would be `close[t] > max(close[t-N..t])`, and a value is never strictly greater
|
||||
// than the max of a window that contains itself -> the up-break could NEVER fire ->
|
||||
// bias would never reach +1. So a sustained +1 here is only reachable when the
|
||||
// current bar is structurally excluded: this asserts the C2 guard directly.
|
||||
let closes = [100.0, 101.0, 102.0, 103.0, 104.0, 105.0];
|
||||
let vals: Vec<f64> = run_breakout_bias(&closes).iter().map(|(_, v)| *v).collect();
|
||||
assert!(
|
||||
!vals.is_empty() && vals.iter().all(|&v| v == 1.0),
|
||||
"a strictly rising series must up-break every warmed bar (+1); got {vals:?}"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn breakout_is_causal_and_holds_pm1_direction() {
|
||||
// closes: an up-break at bar 3, quiet hold, a down-break at bar 6.
|
||||
// idx: 0 1 2 3 4 5 6 7
|
||||
let closes = [100.0, 101.0, 99.0, 102.0, 100.0, 100.0, 95.0, 100.0];
|
||||
// bar3: max(close[0..2])=101, close[3]=102 > 101 -> UP-break. (C2: a window INCLUDING
|
||||
// bar3 would be max=102, and 102>102 strict is false -> no break; the break
|
||||
// firing here proves the current bar is excluded.)
|
||||
// bars4,5: no break -> bias HOLDS +1.
|
||||
// bar6: min(close[3..5])=100, close[6]=95 < 100 -> DOWN-break -> bias flips to -1.
|
||||
// bar7: no break -> HOLDS -1.
|
||||
let got = run_breakout_bias(&closes);
|
||||
let vals: Vec<f64> = got.iter().map(|(_, v)| *v).collect();
|
||||
// bias is emitted only once the channel warms up (Delay(1)+RollingMax(3) -> first at bar 3).
|
||||
assert_eq!(vals, vec![1.0, 1.0, 1.0, -1.0, -1.0], "expected +1 held then -1 held; got {vals:?}");
|
||||
}
|
||||
@@ -30,6 +30,8 @@ mod mul;
|
||||
mod position_management;
|
||||
mod recorder;
|
||||
mod resample;
|
||||
mod rolling_max;
|
||||
mod rolling_min;
|
||||
mod series_reducer;
|
||||
mod session;
|
||||
mod sim_broker;
|
||||
@@ -56,6 +58,8 @@ pub use position_management::{
|
||||
};
|
||||
pub use recorder::Recorder;
|
||||
pub use resample::Resample;
|
||||
pub use rolling_max::RollingMax;
|
||||
pub use rolling_min::RollingMin;
|
||||
pub use series_reducer::SeriesReducer;
|
||||
pub use session::Session;
|
||||
pub use sim_broker::SimBroker;
|
||||
|
||||
@@ -0,0 +1,222 @@
|
||||
//! `RollingMax` — maximum over the last `length` values of one f64 input.
|
||||
//!
|
||||
//! Sliding-window maximum maintained by a **descending monotonic deque** (the
|
||||
//! textbook O(1)-amortized sliding-window-max): each cycle pops from the back every
|
||||
//! element <= the new sample (they can never again be the window max), pushes the new
|
||||
//! one, and evicts the front once its stream index leaves the window. The front is
|
||||
//! always the current window max. A per-cycle O(N) re-scan would be too slow at the
|
||||
//! large `N` (channel lengths in the thousands of bars) the breakout screen sweeps.
|
||||
//!
|
||||
//! Like `Sma`/`Delay`, the window lives in node state, so `eval` reads only the newest
|
||||
//! sample and `lookbacks()` is `1`. Warm-up is **skip-emit**: `None` until `length`
|
||||
//! samples have passed (the window is not yet full). The deque is pre-sized to
|
||||
//! `length` so the hot path is allocation-free (C7). Per the "operator is topology"
|
||||
//! convention (see `gt.rs`), `RollingMax` and `RollingMin` are two node types, not one
|
||||
//! node with a max/min param.
|
||||
|
||||
use aura_core::{
|
||||
Cell, Ctx, FieldSpec, Firing, Node, NodeSchema, ParamSpec, PortSpec, PrimitiveBuilder,
|
||||
ScalarKind,
|
||||
};
|
||||
use std::collections::VecDeque;
|
||||
|
||||
/// Maximum over the last `length` values of one f64 input, maintained by a descending
|
||||
/// monotonic deque (front = current window max). `None` during warm-up.
|
||||
pub struct RollingMax {
|
||||
length: usize,
|
||||
// descending monotonic deque of (value, stream_index); front is the window max.
|
||||
// node-owned, bounded by `length` (C7), pre-sized so push_back does not allocate.
|
||||
deque: VecDeque<(f64, u64)>,
|
||||
seen: u64, // stream position of the next sample (drives front eviction)
|
||||
count: usize, // samples seen so far, capped at `length` — the warm-up gate
|
||||
out: [Cell; 1],
|
||||
}
|
||||
|
||||
impl RollingMax {
|
||||
/// Build a rolling max of window `length` (must be >= 1; mirror `Sma::new`).
|
||||
pub fn new(length: usize) -> Self {
|
||||
assert!(length >= 1, "RollingMax length must be >= 1");
|
||||
Self {
|
||||
length,
|
||||
deque: VecDeque::with_capacity(length),
|
||||
seen: 0,
|
||||
count: 0,
|
||||
out: [Cell::from_f64(0.0)],
|
||||
}
|
||||
}
|
||||
|
||||
/// The param-generic recipe for a blueprint primitive: declares `length` and builds
|
||||
/// through `RollingMax::new`.
|
||||
pub fn builder() -> PrimitiveBuilder {
|
||||
PrimitiveBuilder::new(
|
||||
"RollingMax",
|
||||
NodeSchema {
|
||||
inputs: vec![PortSpec { kind: ScalarKind::F64, firing: Firing::Any, name: "series".into() }],
|
||||
output: vec![FieldSpec { name: "value".into(), kind: ScalarKind::F64 }],
|
||||
params: vec![ParamSpec { name: "length".into(), kind: ScalarKind::I64 }],
|
||||
},
|
||||
|p| Box::new(RollingMax::new(p[0].i64() as usize)),
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
impl Node for RollingMax {
|
||||
// The window lives in node state, so only the newest sample is read each cycle.
|
||||
fn lookbacks(&self) -> Vec<usize> {
|
||||
vec![1]
|
||||
}
|
||||
|
||||
fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Cell]> {
|
||||
let w = ctx.f64_in(0);
|
||||
if w.is_empty() {
|
||||
return None; // no sample yet
|
||||
}
|
||||
let x = w[0]; // index 0 = newest (financial indexing)
|
||||
let idx = self.seen;
|
||||
self.seen += 1;
|
||||
// descending invariant: a back element <= x can never again be the max — drop it.
|
||||
while let Some(&(v, _)) = self.deque.back() {
|
||||
if v <= x {
|
||||
self.deque.pop_back();
|
||||
} else {
|
||||
break;
|
||||
}
|
||||
}
|
||||
self.deque.push_back((x, idx));
|
||||
// evict the front once it leaves the window [idx-length+1, idx] (no subtraction:
|
||||
// front index `i` is out when `i + length <= idx`).
|
||||
while let Some(&(_, i)) = self.deque.front() {
|
||||
if i + self.length as u64 <= idx {
|
||||
self.deque.pop_front();
|
||||
} else {
|
||||
break;
|
||||
}
|
||||
}
|
||||
if self.count < self.length {
|
||||
self.count += 1;
|
||||
}
|
||||
if self.count < self.length {
|
||||
return None; // not yet warmed up
|
||||
}
|
||||
self.out[0] = Cell::from_f64(self.deque.front().expect("deque non-empty after push").0);
|
||||
Some(&self.out)
|
||||
}
|
||||
|
||||
fn label(&self) -> String {
|
||||
format!("RollingMax({})", self.length)
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use aura_core::{AnyColumn, Scalar, Timestamp};
|
||||
|
||||
fn drive(node: &mut RollingMax, feed: &[f64]) -> Vec<Option<f64>> {
|
||||
let mut inputs = vec![AnyColumn::with_capacity(ScalarKind::F64, 1)];
|
||||
let mut out = Vec::new();
|
||||
for &v in feed {
|
||||
inputs[0].push(Scalar::f64(v)).unwrap();
|
||||
out.push(node.eval(Ctx::new(&inputs, Timestamp(0))).map(|r| r[0].f64()));
|
||||
}
|
||||
out
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn rolling_max_warms_up_then_tracks_the_window_max() {
|
||||
// max of [1,3,2], [3,2,5], [2,5,4] once warmed up; silent for the first two.
|
||||
let mut node = RollingMax::new(3);
|
||||
let got = drive(&mut node, &[1.0, 3.0, 2.0, 5.0, 4.0]);
|
||||
assert_eq!(got, vec![None, None, Some(3.0), Some(5.0), Some(5.0)]);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn rolling_max_evicts_the_expiring_window_maximum() {
|
||||
// the front-eviction path: the early big value must leave the window. length 2.
|
||||
// windows: [9,1]->9, [1,2]->2, [2,3]->3 (the 9 has expired, not stuck as max).
|
||||
let mut node = RollingMax::new(2);
|
||||
let got = drive(&mut node, &[9.0, 1.0, 2.0, 3.0]);
|
||||
assert_eq!(got, vec![None, Some(9.0), Some(2.0), Some(3.0)]);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn rolling_max_length_one_is_identity() {
|
||||
let mut node = RollingMax::new(1);
|
||||
assert_eq!(drive(&mut node, &[7.0, 9.0, 4.0]), vec![Some(7.0), Some(9.0), Some(4.0)]);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn deque_matches_naive_window_max() {
|
||||
// the monotonic deque must equal a naive O(N) per-window max at every warmed
|
||||
// cycle, across a long deterministic series (mirrors Sma's re-sum cross-check).
|
||||
let length = 37;
|
||||
let mut node = RollingMax::new(length);
|
||||
let series: Vec<f64> = (0..3_000u64)
|
||||
.map(|i| (i.wrapping_mul(1103515245).wrapping_add(12345) % 1000) as f64 * 0.5 - 250.0)
|
||||
.collect();
|
||||
let mut inputs = vec![AnyColumn::with_capacity(ScalarKind::F64, 1)];
|
||||
for (t, &x) in series.iter().enumerate() {
|
||||
inputs[0].push(Scalar::f64(x)).unwrap();
|
||||
let got = node.eval(Ctx::new(&inputs, Timestamp(0))).map(|r| r[0].f64());
|
||||
if t + 1 >= length {
|
||||
let want = series[t + 1 - length..=t].iter().cloned().fold(f64::NEG_INFINITY, f64::max);
|
||||
assert_eq!(got, Some(want), "at t={t}");
|
||||
} else {
|
||||
assert_eq!(got, None, "silent until warmed up at t={t}");
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn lookback_is_one_window_lives_in_node_state() {
|
||||
assert_eq!(RollingMax::new(20).lookbacks(), vec![1]);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn length_must_be_at_least_one() {
|
||||
let r = std::panic::catch_unwind(|| RollingMax::new(0));
|
||||
assert!(r.is_err(), "RollingMax::new(0) must panic");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn label_carries_the_window() {
|
||||
assert_eq!(RollingMax::new(20).label(), "RollingMax(20)");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn builder_declares_length_param_and_value_output() {
|
||||
// the blueprint seam later tasks bootstrap through: the param-generic recipe
|
||||
// declares a single I64 `length` knob and an F64 `value` output (mirrors Sma's
|
||||
// nodes_declare_expected_params).
|
||||
let schema = RollingMax::builder().schema().clone();
|
||||
assert_eq!(schema.params, vec![ParamSpec { name: "length".into(), kind: ScalarKind::I64 }]);
|
||||
assert_eq!(schema.output.len(), 1);
|
||||
assert_eq!(schema.output[0].name, "value");
|
||||
assert_eq!(schema.output[0].kind, ScalarKind::F64);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn builder_input_slot_is_named_series() {
|
||||
// the named f64 input the breakout graph wires close into (mirrors Sma's
|
||||
// input_slot_is_named_series).
|
||||
assert_eq!(RollingMax::builder().schema().inputs[0].name, "series");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn builder_bind_removes_length_from_param_space() {
|
||||
// a bound length reports an empty param surface; the open form keeps it (mirrors
|
||||
// Sma's bind_removes_slot_from_param_space).
|
||||
let bound = RollingMax::builder().named("channel").bind("length", Scalar::i64(20));
|
||||
assert!(bound.schema().params.is_empty());
|
||||
assert_eq!(RollingMax::builder().named("channel").params().len(), 1);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn builder_bound_node_builds_with_injected_length() {
|
||||
// built from an empty open slice, the bound builder yields a RollingMax(20) —
|
||||
// the `p[0].i64() as usize` build closure is exercised end-to-end (mirrors Sma's
|
||||
// bound_node_builds_with_injected_value).
|
||||
let node = RollingMax::builder().bind("length", Scalar::i64(20)).build(&[]);
|
||||
assert_eq!(node.label(), "RollingMax(20)");
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,208 @@
|
||||
//! `RollingMin` — minimum over the last `length` values of one f64 input.
|
||||
//!
|
||||
//! The mirror of `RollingMax`: an **ascending** monotonic deque (front = window min).
|
||||
//! Each cycle pops from the back every element >= the new sample, pushes the new one,
|
||||
//! and evicts the front once it leaves the window. See `rolling_max.rs` for the shared
|
||||
//! rationale (O(1) amortized, window in node state, warm-up skip-emit, C7).
|
||||
|
||||
use aura_core::{
|
||||
Cell, Ctx, FieldSpec, Firing, Node, NodeSchema, ParamSpec, PortSpec, PrimitiveBuilder,
|
||||
ScalarKind,
|
||||
};
|
||||
use std::collections::VecDeque;
|
||||
|
||||
/// Minimum over the last `length` values of one f64 input, maintained by an ascending
|
||||
/// monotonic deque (front = current window min). `None` during warm-up.
|
||||
pub struct RollingMin {
|
||||
length: usize,
|
||||
deque: VecDeque<(f64, u64)>, // ascending; front = window min
|
||||
seen: u64,
|
||||
count: usize,
|
||||
out: [Cell; 1],
|
||||
}
|
||||
|
||||
impl RollingMin {
|
||||
/// Build a rolling min of window `length` (must be >= 1).
|
||||
pub fn new(length: usize) -> Self {
|
||||
assert!(length >= 1, "RollingMin length must be >= 1");
|
||||
Self {
|
||||
length,
|
||||
deque: VecDeque::with_capacity(length),
|
||||
seen: 0,
|
||||
count: 0,
|
||||
out: [Cell::from_f64(0.0)],
|
||||
}
|
||||
}
|
||||
|
||||
/// The param-generic recipe for a blueprint primitive: declares `length` and builds
|
||||
/// through `RollingMin::new`.
|
||||
pub fn builder() -> PrimitiveBuilder {
|
||||
PrimitiveBuilder::new(
|
||||
"RollingMin",
|
||||
NodeSchema {
|
||||
inputs: vec![PortSpec { kind: ScalarKind::F64, firing: Firing::Any, name: "series".into() }],
|
||||
output: vec![FieldSpec { name: "value".into(), kind: ScalarKind::F64 }],
|
||||
params: vec![ParamSpec { name: "length".into(), kind: ScalarKind::I64 }],
|
||||
},
|
||||
|p| Box::new(RollingMin::new(p[0].i64() as usize)),
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
impl Node for RollingMin {
|
||||
fn lookbacks(&self) -> Vec<usize> {
|
||||
vec![1]
|
||||
}
|
||||
|
||||
fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Cell]> {
|
||||
let w = ctx.f64_in(0);
|
||||
if w.is_empty() {
|
||||
return None;
|
||||
}
|
||||
let x = w[0];
|
||||
let idx = self.seen;
|
||||
self.seen += 1;
|
||||
// ascending invariant: a back element >= x can never again be the min — drop it.
|
||||
while let Some(&(v, _)) = self.deque.back() {
|
||||
if v >= x {
|
||||
self.deque.pop_back();
|
||||
} else {
|
||||
break;
|
||||
}
|
||||
}
|
||||
self.deque.push_back((x, idx));
|
||||
while let Some(&(_, i)) = self.deque.front() {
|
||||
if i + self.length as u64 <= idx {
|
||||
self.deque.pop_front();
|
||||
} else {
|
||||
break;
|
||||
}
|
||||
}
|
||||
if self.count < self.length {
|
||||
self.count += 1;
|
||||
}
|
||||
if self.count < self.length {
|
||||
return None;
|
||||
}
|
||||
self.out[0] = Cell::from_f64(self.deque.front().expect("deque non-empty after push").0);
|
||||
Some(&self.out)
|
||||
}
|
||||
|
||||
fn label(&self) -> String {
|
||||
format!("RollingMin({})", self.length)
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use aura_core::{AnyColumn, Scalar, Timestamp};
|
||||
|
||||
fn drive(node: &mut RollingMin, feed: &[f64]) -> Vec<Option<f64>> {
|
||||
let mut inputs = vec![AnyColumn::with_capacity(ScalarKind::F64, 1)];
|
||||
let mut out = Vec::new();
|
||||
for &v in feed {
|
||||
inputs[0].push(Scalar::f64(v)).unwrap();
|
||||
out.push(node.eval(Ctx::new(&inputs, Timestamp(0))).map(|r| r[0].f64()));
|
||||
}
|
||||
out
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn rolling_min_warms_up_then_tracks_the_window_min() {
|
||||
// min of [9,7,8], [7,8,5], [8,5,6]; silent for the first two.
|
||||
let mut node = RollingMin::new(3);
|
||||
let got = drive(&mut node, &[9.0, 7.0, 8.0, 5.0, 6.0]);
|
||||
assert_eq!(got, vec![None, None, Some(7.0), Some(5.0), Some(5.0)]);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn rolling_min_evicts_the_expiring_window_minimum() {
|
||||
// length 2; the early small value must leave the window: [1,9]->1, [9,8]->8, [8,7]->7.
|
||||
let mut node = RollingMin::new(2);
|
||||
let got = drive(&mut node, &[1.0, 9.0, 8.0, 7.0]);
|
||||
assert_eq!(got, vec![None, Some(1.0), Some(8.0), Some(7.0)]);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn rolling_min_length_one_is_identity() {
|
||||
let mut node = RollingMin::new(1);
|
||||
assert_eq!(drive(&mut node, &[7.0, 9.0, 4.0]), vec![Some(7.0), Some(9.0), Some(4.0)]);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn deque_matches_naive_window_min() {
|
||||
// the monotonic deque must equal a naive O(N) per-window min at every warmed
|
||||
// cycle, across a long deterministic series (mirrors RollingMax's cross-check).
|
||||
// u64 + wrapping arithmetic matches rolling_max.rs and avoids i32 overflow panics.
|
||||
let length = 37;
|
||||
let mut node = RollingMin::new(length);
|
||||
let series: Vec<f64> = (0..3_000u64)
|
||||
.map(|i| (i.wrapping_mul(1103515245).wrapping_add(12345) % 1000) as f64 * 0.5 - 250.0)
|
||||
.collect();
|
||||
let mut inputs = vec![AnyColumn::with_capacity(ScalarKind::F64, 1)];
|
||||
for (t, &x) in series.iter().enumerate() {
|
||||
inputs[0].push(Scalar::f64(x)).unwrap();
|
||||
let got = node.eval(Ctx::new(&inputs, Timestamp(0))).map(|r| r[0].f64());
|
||||
if t + 1 >= length {
|
||||
let want = series[t + 1 - length..=t].iter().cloned().fold(f64::INFINITY, f64::min);
|
||||
assert_eq!(got, Some(want), "at t={t}");
|
||||
} else {
|
||||
assert_eq!(got, None, "silent until warmed up at t={t}");
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn lookback_is_one_window_lives_in_node_state() {
|
||||
assert_eq!(RollingMin::new(20).lookbacks(), vec![1]);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn length_must_be_at_least_one() {
|
||||
let r = std::panic::catch_unwind(|| RollingMin::new(0));
|
||||
assert!(r.is_err(), "RollingMin::new(0) must panic");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn label_carries_the_window() {
|
||||
assert_eq!(RollingMin::new(20).label(), "RollingMin(20)");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn builder_declares_length_param_and_value_output() {
|
||||
// the blueprint seam later tasks bootstrap through: the param-generic recipe
|
||||
// declares a single I64 `length` knob and an F64 `value` output (mirrors
|
||||
// RollingMax's builder_declares_length_param_and_value_output).
|
||||
let schema = RollingMin::builder().schema().clone();
|
||||
assert_eq!(schema.params, vec![ParamSpec { name: "length".into(), kind: ScalarKind::I64 }]);
|
||||
assert_eq!(schema.output.len(), 1);
|
||||
assert_eq!(schema.output[0].name, "value");
|
||||
assert_eq!(schema.output[0].kind, ScalarKind::F64);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn builder_input_slot_is_named_series() {
|
||||
// the named f64 input the breakout graph wires close into (mirrors RollingMax's
|
||||
// builder_input_slot_is_named_series).
|
||||
assert_eq!(RollingMin::builder().schema().inputs[0].name, "series");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn builder_bind_removes_length_from_param_space() {
|
||||
// a bound length reports an empty param surface; the open form keeps it (mirrors
|
||||
// RollingMax's builder_bind_removes_length_from_param_space).
|
||||
let bound = RollingMin::builder().named("channel").bind("length", Scalar::i64(20));
|
||||
assert!(bound.schema().params.is_empty());
|
||||
assert_eq!(RollingMin::builder().named("channel").params().len(), 1);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn builder_bound_node_builds_with_injected_length() {
|
||||
// built from an empty open slice, the bound builder yields a RollingMin(20) —
|
||||
// the `p[0].i64() as usize` build closure is exercised end-to-end (mirrors
|
||||
// RollingMax's builder_bound_node_builds_with_injected_length).
|
||||
let node = RollingMin::builder().bind("length", Scalar::i64(20)).build(&[]);
|
||||
assert_eq!(node.label(), "RollingMin(20)");
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user