Files

T

Michael Schimmel 2579e2b1fd 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.

2026-03-01 22:18:52 +01:00

4.0 KiB

Raw Permalink Blame History

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:

Erzeugt einen neuen RootStream.
Instanziiert einen deterministischen Zufallsgenerator (via Seed).
Registriert ein Generator-Objekt im Environment, welches das Limit (Anzahl der Ticks) kennt.
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.

4.0 KiB Raw Permalink Blame History