From 2579e2b1fd98daa561b9739e0373e750e0961b03 Mon Sep 17 00:00:00 2001 From: Michael Schimmel Date: Sun, 1 Mar 2026 22:18:52 +0100 Subject: [PATCH] Add pipeline generator registration Introduce a new field to `Environment` to store a list of pipeline generator functions. Add a `run_pipeline` method to `Environment` that iterates through and executes all registered generators until they are exhausted. Implement `ObservableStream` trait for `RootStream` and `PipeStream`. Add a `StreamNode` wrapper for `ObservableStream` to be used as a script object. Register `create-random-ohlc` as a native function that creates a `RootStream`, sets up a OHLC record layout, and registers a stateful generator closure in the environment's pipeline generators. --- ...Pipeline_Architecture_and_Bootstrapping.md | 47 +++++++ src/ast/environment.rs | 16 +++ src/ast/rtl/streams.rs | 119 +++++++++++++++++- 3 files changed, 181 insertions(+), 1 deletion(-) create mode 100644 docs/Pipeline_Architecture_and_Bootstrapping.md diff --git a/docs/Pipeline_Architecture_and_Bootstrapping.md b/docs/Pipeline_Architecture_and_Bootstrapping.md new file mode 100644 index 0000000..7dfdde0 --- /dev/null +++ b/docs/Pipeline_Architecture_and_Bootstrapping.md @@ -0,0 +1,47 @@ +# Reactive Pipeline Architecture & Bootstrapping + +Dieses Dokument beschreibt die Architektur der reaktiven Datenverarbeitung innerhalb des MYC AST Compilers in Rust. Es konkretisiert das "Dual Series Architecture"-Konzept um die Aspekte der Initialisierung, Ausführung und Synchronisation. + +## 1. Das Phasen-Modell (Bootstrapping vs. Pumping) + +Das wichtigste Designprinzip der Pipeline ist die strikte zeitliche und logische Trennung zwischen dem **Aufbau des Datenflussgraphen** und der **tatsächlichen Datenverarbeitung**. + +### Phase 1: Bootstrapping (Skript-Evaluation) +Wenn das MYC-Skript von der VM ausgeführt wird, fließen **keine echten Zeitreihen-Daten**. +Das Skript fungiert als reiner *Graph-Builder*: +* Aufrufe wie `(create-random-ohlc ...)` erzeugen Quell-Knoten (`RootStream` + Daten-Generator) und registrieren diese im Host-System (z.B. dem `Environment`). +* Das `(pipe ...)` Statement nimmt existierende Streams, erzeugt eine Transformation (`PipeStream` mit einer Lambda-Funktion) und hängt sich als `Observer` an die Eingangs-Streams an. +* Als Rückgabewert liefert `pipe` eine leere `SharedSeries` (bzw. `SharedRecordSeries`), die als Referenz für weitere Pipes genutzt werden kann. + +Das Resultat der Skriptausführung ist ein gerichteter azyklischer Graph (DAG) aus Observern, der bereit für die Datenaufnahme ist. + +### Phase 2: Pumping (Host-Execution) +Nachdem das Skript vollständig durchlaufen ist, übernimmt die Rust-Host-Applikation (z.B. eine Backtesting-Engine oder ein Live-Trading-Loop). +* Der Host iteriert über die registrierten Daten-Generatoren. +* Er ruft deren `tick()` Methoden auf. +* Dies löst Kettenreaktionen aus: Jeder `tick` treibt die Daten durch die angemeldeten `PipeStreams`, führt dort die Lambdas aus und speichert die Ergebnisse in den `SharedSeries`. + +## 2. Mehrere Root-Streams + +Die Architektur erlaubt ausdrücklich die Existenz mehrerer unabhängiger `RootStreams` (z.B. `BTCUSD-M1`, `ETHUSD-M5`). Ein `RootStream` ist lediglich die Quelle von Signalen (`cycle_id` + `Value`). + +### Synchronisations-Szenarien & Barrieren + +* **Szenario A (Unabhängige Graphen):** Mehrere Streams fließen durch völlig getrennte Pipe-Äste. Dies funktioniert out-of-the-box ohne Konflikte. +* **Szenario B (Synchrone Zusammenführung):** Mehrere Streams (z.B. zwei Indikatoren basierend auf dem gleichen `RootStream`) fließen in eine gemeinsame `pipe`. + * *Lösung:* Die `PipeStream` besitzt eine integrierte **Barrier-Synchronisation**. Sie wartet, bis alle konfigurierten Eingänge dieselbe `cycle_id` geliefert haben, bevor sie ihr Lambda ausführt und das kombinierte Ergebnis weitergibt. +* **Szenario C (Asynchrone Zusammenführung / Mixed Timeframes):** Zwei unterschiedliche `RootStreams` (mit unterschiedlichen Taktfrequenzen) sollen in einer Pipe kombiniert werden. + * *Achtung:* Die aktuelle `cycle_id`-basierte Barriere würde hier blockieren. + * *Lösung (Zukünftig):* Für dieses Szenario wird eine spezialisierte Form der Pipe oder ein dedizierter `JoinStream` benötigt, der Signale anhand ihres `DateTime`-Stempels synchronisiert statt über den monotonen Index, oder ein "Hold"-Verhalten für den langsameren Stream implementiert. + +## 3. Test-Infrastruktur: Pseudo-Random Streams + +Da echte Datenserver als `RootStreams` oft extern angebunden werden müssen, stellt die RTL (Runtime Library) Mechanismen für reproduzierbare Tests bereit. + +Die Funktion `(create-random-ohlc seed limit)` ist der Blueprint für solche Test-Quellen: +1. Erzeugt einen neuen `RootStream`. +2. Instanziiert einen deterministischen Zufallsgenerator (via Seed). +3. Registriert ein Generator-Objekt im Environment, welches das Limit (Anzahl der Ticks) kennt. +4. Gibt den Stream-Knoten an das Skript zurück, damit dieser an eine `pipe` angeschlossen werden kann. + +Die Methode `Environment::run_pipeline()` kann später von Rust aus aufgerufen werden, um alle generierten Test-Streams bis zu ihrem jeweiligen Limit durchlaufen zu lassen. diff --git a/src/ast/environment.rs b/src/ast/environment.rs index 45d2e7f..46c361a 100644 --- a/src/ast/environment.rs +++ b/src/ast/environment.rs @@ -33,6 +33,7 @@ pub struct Environment { pub debug_mode: bool, pub optimization: bool, pub macro_registry: Rc>, + pub pipeline_generators: Rc bool>>>>, } struct EnvFunctionRegistry { @@ -106,6 +107,7 @@ impl Environment { debug_mode: false, optimization: true, macro_registry: Rc::new(RefCell::new(MacroRegistry::new())), + pipeline_generators: Rc::new(RefCell::new(Vec::new())), }; env.register_stdlib(); env @@ -115,6 +117,20 @@ impl Environment { self.debug_mode = enabled; } + /// Pumps data through all registered pipeline generators until they are exhausted. + pub fn run_pipeline(&self) { + let mut generators = self.pipeline_generators.borrow_mut(); + let mut any_active = true; + while any_active { + any_active = false; + for generator in generators.iter_mut() { + if generator() { + any_active = true; + } + } + } + } + fn get_expander(&self) -> MacroExpander { let evaluator = RuntimeMacroEvaluator { global_names: self.global_names.clone(), diff --git a/src/ast/rtl/streams.rs b/src/ast/rtl/streams.rs index 592f7cb..d0bb831 100644 --- a/src/ast/rtl/streams.rs +++ b/src/ast/rtl/streams.rs @@ -24,6 +24,32 @@ pub trait Observer { fn notify(&mut self, source_index: usize, cycle_id: u64, value: Value); } +/// Polymorphic Interface for any stream that can accept observers (like Delphi's IStream). +pub trait ObservableStream { + fn add_observer(&self, observer: Rc>); +} + +/// A generic wrapper to pass ANY ObservableStream (Root, Pipe, etc.) as an Object to the VM. +#[derive(Clone)] +pub struct StreamNode { + pub inner: Rc, +} + +impl std::fmt::Debug for StreamNode { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!(f, "StreamNode") + } +} + +impl crate::ast::types::Object for StreamNode { + fn type_name(&self) -> &'static str { + "StreamNode" + } + fn as_any(&self) -> &dyn std::any::Any { + self + } +} + /// The RootStream is the "Clock" and data source of the entire pipeline. /// It generates the monotonic `cycle_id` and triggers the observers. pub struct RootStream { @@ -31,6 +57,12 @@ pub struct RootStream { observers: RefCell>>>, } +impl ObservableStream for RootStream { + fn add_observer(&self, observer: Rc>) { + self.observers.borrow_mut().push(observer); + } +} + impl RootStream { pub fn new() -> Self { Self { @@ -98,8 +130,10 @@ impl PipeStream { let cycles = self.last_cycle_per_input.borrow(); cycles.iter().all(|&c| c == cycle_id) } +} - pub fn add_observer(&self, observer: Rc>) { +impl ObservableStream for PipeStream { + fn add_observer(&self, observer: Rc>) { self.observers.borrow_mut().push(observer); } } @@ -153,6 +187,89 @@ impl Observer for SeriesPusher { } } +// ============================================================================ +// Script Integration (RTL Registration) +// ============================================================================ + +use crate::ast::environment::Environment; +use crate::ast::types::{Purity, Signature, StaticType, Keyword, RecordLayout}; + +pub fn register(env: &Environment) { + // (create-random-ohlc seed limit) -> StreamNode + let generators = env.pipeline_generators.clone(); + + env.register_native_fn( + "create-random-ohlc", + StaticType::Function(Box::new(Signature { + params: StaticType::Tuple(vec![StaticType::Int, StaticType::Int]), + ret: StaticType::Any, // Returns a StreamNode + })), + Purity::Impure, // Modifies global generator registry + move |args: std::vec::Vec| { + if args.len() != 2 { + panic!("create-random-ohlc expects exactly 2 arguments (seed, limit)"); + } + let seed = if let Value::Int(s) = args[0] { s as u64 } else { 0 }; + let limit = if let Value::Int(l) = args[1] { l as usize } else { 0 }; + + // 1. Create the RootStream + let root_stream = Rc::new(RootStream::new()); + let stream_node = StreamNode { inner: root_stream.clone() }; + + // 2. Setup the Layout for OHLC records + let layout = RecordLayout::get_or_create(vec![ + (Keyword::intern("open"), StaticType::Float), + (Keyword::intern("high"), StaticType::Float), + (Keyword::intern("low"), StaticType::Float), + (Keyword::intern("close"), StaticType::Float), + ]); + + // 3. Create the stateful generator closure + let mut current_tick = 0; + let mut last_close = 100.0; + + // We use a local PRNG instance for reproducibility based on the seed + let mut rng = fastrand::Rng::with_seed(seed); + + let generator = move || -> bool { + if current_tick >= limit { + return false; // Exhausted + } + + // Generate random OHLC (Random Walk) + let change = (rng.f64() - 0.5) * 2.0; + let open = last_close; + let high = open + (rng.f64() * 2.0).abs(); + let low = open - (rng.f64() * 2.0).abs(); + let close = open + change; + last_close = close; + + let record = Value::Record( + layout.clone(), + Rc::new(vec![ + Value::Float(open), + Value::Float(high), + Value::Float(low), + Value::Float(close), + ]) + ); + + // Pump the signal into the RootStream + root_stream.tick(record); + + current_tick += 1; + true // Still active + }; + + // 4. Register the generator in the Environment + generators.borrow_mut().push(Box::new(generator)); + + // 5. Return the stream reference to the script + Value::Object(Rc::new(stream_node)) + }, + ); +} + #[cfg(test)] mod tests { use super::*;