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

Refactor data streams to support time windows

Browse Source 
This commit introduces time windowing functionality to the M1 and tick
data streams.

Key changes:

- **`DataServer::stream_m1_windowed` and
  `DataServer::stream_tick_windowed`**: New methods added to
  `DataServer` to allow streaming data within a specified `from_ms` and
  `to_ms` range.
- **`SymbolChunkIter` modifications**: The `SymbolChunkIter` struct now
  includes `from_ms`, `to_ms`, and `exhausted` fields to manage time
  window filtering.
- **`filter_chunk` helper function**: A new utility function to
  efficiently filter individual data chunks based on the time window.
  This function also handles early termination when `to_ms` is exceeded.
- **`HasTimestamp` trait**: Introduced to provide a common interface for
  accessing timestamps across different record types (`M1Parsed`,
  `TickParsed`), enabling generic filtering.
- **`filter_files` method**: Added to `DataServer` to pre-filter file
  keys based on the time window before creating iterators. This
  optimizes file loading by skipping files that fall entirely outside
  the desired range.
- **`create-m1-stream` and `create-tick-stream`**: The native functions
  in `data_streams.rs` are updated to accept optional `DateTime`
  arguments for `from` and `to` timestamps. They now also handle unknown
  symbols by returning `Value::Void` and return an empty stream if the
  time window results in no data.
- **Documentation**: Updated doc comments for the stream creation
  functions to reflect the new time windowing capabilities and to
  clarify behavior for unknown symbols and empty time windows.

These changes significantly enhance the flexibility of data streaming,
allowing users to query specific time ranges of M1 and tick data more
efficiently.
This commit is contained in:
Brummel
2026-03-29 22:03:33 +02:00
parent 1f68371920
commit e82d48d1cb
3 changed files with 352 additions and 46 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/ast/rtl/data_streams.rs
+48 -17
View File
@@ -36,7 +36,7 @@ fn register_m1_stream(env: &Environment) {
env.register_native_fn(
"create-m1-stream",
StaticType::Function(Box::new(Signature {
params: StaticType::Tuple(vec![StaticType::Text]),
params: StaticType::Any,
ret: ret_type,
})),
Purity::Impure,
@@ -46,12 +46,24 @@ fn register_m1_stream(env: &Environment) {
_ => panic!("create-m1-stream: first argument must be a string"),
};
let from_ms = args.get(1).map(|v| match v {
Value::DateTime(ms) => *ms,
_ => panic!("create-m1-stream: 'from' must be a DateTime"),
});
let to_ms = args.get(2).map(|v| match v {
Value::DateTime(ms) => *ms,
_ => panic!("create-m1-stream: 'to' must be a DateTime"),
});
let server = get_data_server()
.expect("create-m1-stream: DataServer not initialized. Call init_data_server() first.");
let mut iter = server
.stream_m1(symbol)
.unwrap_or_else(|| panic!("create-m1-stream: symbol '{}' not found", symbol));
// Unknown symbol → Void (no stream created)
if !server.has_symbol(symbol) {
return Value::Void;
}
let mut iter = server.stream_m1_windowed(symbol, from_ms, to_ms);
// Create the pipeline entry point
let root_stream = Rc::new(RootStream::new());
@@ -61,8 +73,8 @@ fn register_m1_stream(env: &Environment) {
element_type: StaticType::Record(layout.clone()),
};
// Pre-fetch the first chunk
let mut current_chunk = iter.next_chunk();
// Pre-fetch the first chunk (None if time window is empty)
let mut current_chunk = iter.as_mut().and_then(|it| it.next_chunk());
let mut pos = 0;
let generator = move || -> bool {
@@ -93,7 +105,7 @@ fn register_m1_stream(env: &Environment) {
}
// Advance to next chunk (triggers prefetch internally)
current_chunk = iter.next_chunk();
current_chunk = iter.as_mut().and_then(|it| it.next_chunk());
pos = 0;
}
};
@@ -103,8 +115,11 @@ fn register_m1_stream(env: &Environment) {
},
)
.doc("Creates a stream of M1 (minute) OHLCV bars from tick data files.")
.description("Reads pre-cached binary data from the DataServer. The symbol must match a directory entry (e.g. \"EURUSD\", \"AAPL.US\").")
.examples(&["(def ohlc (create-m1-stream \"EURUSD\"))"]);
.description("Reads pre-cached binary data from the DataServer. Optional from/to DateTime arguments restrict the time window (millisecond precision). Returns Void if the symbol is unknown. Returns an empty (immediately exhausted) stream if the symbol exists but the time window contains no data.")
.examples(&[
"(def ohlc (create-m1-stream \"EURUSD\"))",
"(def ohlc (create-m1-stream \"EURUSD\" (date \"2020-01-01\") (date \"2020-12-31\")))",
]);
}
// -- Tick Stream -------------------------------------------------------------
@@ -123,7 +138,7 @@ fn register_tick_stream(env: &Environment) {
env.register_native_fn(
"create-tick-stream",
StaticType::Function(Box::new(Signature {
params: StaticType::Tuple(vec![StaticType::Text]),
params: StaticType::Any,
ret: ret_type,
})),
Purity::Impure,
@@ -133,12 +148,24 @@ fn register_tick_stream(env: &Environment) {
_ => panic!("create-tick-stream: first argument must be a string"),
};
let from_ms = args.get(1).map(|v| match v {
Value::DateTime(ms) => *ms,
_ => panic!("create-tick-stream: 'from' must be a DateTime"),
});
let to_ms = args.get(2).map(|v| match v {
Value::DateTime(ms) => *ms,
_ => panic!("create-tick-stream: 'to' must be a DateTime"),
});
let server = get_data_server()
.expect("create-tick-stream: DataServer not initialized.");
let mut iter = server
.stream_tick(symbol)
.unwrap_or_else(|| panic!("create-tick-stream: symbol '{}' not found", symbol));
// Unknown symbol → Void (no stream created)
if !server.has_symbol(symbol) {
return Value::Void;
}
let mut iter = server.stream_tick_windowed(symbol, from_ms, to_ms);
let root_stream = Rc::new(RootStream::new());
let layout = tick_layout.clone();
@@ -147,7 +174,8 @@ fn register_tick_stream(env: &Environment) {
element_type: StaticType::Record(layout.clone()),
};
let mut current_chunk = iter.next_chunk();
// Pre-fetch the first chunk (None if time window is empty)
let mut current_chunk = iter.as_mut().and_then(|it| it.next_chunk());
let mut pos = 0;
let generator = move || -> bool {
@@ -173,7 +201,7 @@ fn register_tick_stream(env: &Environment) {
return true;
}
current_chunk = iter.next_chunk();
current_chunk = iter.as_mut().and_then(|it| it.next_chunk());
pos = 0;
}
};
@@ -183,6 +211,9 @@ fn register_tick_stream(env: &Environment) {
},
)
.doc("Creates a stream of tick data (ask/bid) from tick data files.")
.description("Reads pre-cached binary data from the DataServer. The symbol must match a directory entry.")
.examples(&["(def ticks (create-tick-stream \"EURUSD\"))"]);
.description("Reads pre-cached binary data from the DataServer. Optional from/to DateTime arguments restrict the time window (millisecond precision). Returns Void if the symbol is unknown. Returns an empty (immediately exhausted) stream if the symbol exists but the time window contains no data.")
.examples(&[
"(def ticks (create-tick-stream \"EURUSD\"))",
"(def ticks (create-tick-stream \"EURUSD\" (date \"2020-01-01\") (date \"2020-12-31\")))",
]);
}