Michael Schimmel
2579e2b1fd
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.
48 lines
4.0 KiB
Markdown
48 lines
4.0 KiB
Markdown
# 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.
|