RustAst/docs/Delphi_Stream_Architecture.md

# Architektur-Analyse: Delphi Reactive Stream Pipeline

Die Delphi-Implementierung der Streaming- und Pipeline-Architektur (insbesondere in `Myc.Data.Stream.pas`) unterscheidet sich fundamental von der aktuellen Rust-Implementierung der `RecordSeries`. Während die Rust-Version auf einem zentralen, zustandsbehafteten Ringspeicher ("Struct of Arrays") aufbaut, implementiert Delphi ein hochgradig entkoppeltes, reaktives Event-System.

Hier ist eine detaillierte Aufschlüsselung der Kernkomponenten und Konzepte in Delphi:

## 1. Trennung von Zustand und Event-Propagation (Stateless Streams)

In Delphi ist der Stream selbst (`IStream` / `TCustomDataStream`) **zustandslos (stateless)**. Er speichert keine Historie.
* Ein Stream hat lediglich eine Definition (`IScalarRecordDefinition`), die das Schema der Daten beschreibt.
* Er verwaltet eine Liste von Abonnenten (`TSignalProc`).
* Wenn neue Daten eintreffen, ruft der Stream die Abonnenten über `Emit()` auf und reicht das `TStreamSignal` durch.

## 2. Das Datenpaket: `TStreamSignal`

Das `TStreamSignal` ist das "Blut" der Pipeline. Es handelt sich um ein leichtgewichtiges Datenpaket, das durch den Graphen fließt.
* **CycleID:** Ein integraler Bestandteil für die Synchronisation (siehe Barrieren). Jeder Tick/Datensatz erhält eine eindeutige ID.
* **Data:** Ein Pointer (`TScalar.PValue`) auf die flachen Skalarwerte des Records. Der Stream baut keine komplexen Objekte auf, sondern schickt nur Referenzen auf den Speicherbereich herum.

## 3. Die Taktung: `TRootStream`

Der `TRootStream` ist der Eintrittspunkt für Marktdaten (z.B. Ticks). 
* Er generiert die globale `CycleID`.
* Beim Aufruf von `Push(RowData)` inkrementiert er die `CycleID` und ruft `Emit()` auf. Dieser Root-Stream ist der Taktgeber (Clock) für alle nachgelagerten Berechnungen.

## 4. Zustand und Historie: `TPipeSource` & `TScalarSeries`

Da der Stream selbst keine Historie speichert, muss jemand anders diese Aufgabe übernehmen. Das ist die Rolle von `TPipeSource`.
* **Rolle:** Ein Observer, der sich an einen `IStream` ankoppelt.
* **Filterung:** Eine `TPipeSource` abonniert nicht den gesamten Record, sondern konzentriert sich immer auf **einzelnes Feld** (z.B. den Preis).
* **Historie:** Die geparsten Werte dieses einen Feldes werden lokal in einer **1-dimensionalen Serie** (`TScalarSeries`) akkumuliert.
* **Shared Source:** Delphi optimiert den Speicherbedarf über eine `TSharedSource`. Wenn 10 verschiedene Indikatoren (Pipes) auf das Feld "Close" eines Root-Streams lauschen, wird intern nur eine einzige `TScalarSeries` aufgebaut und zwischen ihnen geteilt.

*(Anmerkung: Die in Delphi existierende Klasse `TScalarRecordSeries` ist nicht das Rückgrat dieses reaktiven Systems, sondern existiert primär als Container oder für spezielle Evaluator-Zugriffe.)*

## 5. Synchronisation & Transformation: `TPipeStream`

Dies ist das Herzstück für komplexe Berechnungen (z.B. Indikatoren, die auf Preis und Volumen basieren).
Ein `TPipeStream` ist gleichzeitig Observer und wieder ein Stream.

* **Barriere-Synchronisation (`CheckBarrierAndFire`):** Ein `TPipeStream` hat mehrere Inputs (Quellen). Da Signale in einem komplexen Graphen asynchron eintreffen können, wartet der `TPipeStream`, bis **alle** seine Quellen Signale mit **derselben** `CycleID` empfangen haben.
* **Warm-up / Lookback:** Er feuert erst, wenn genügend Historie (`FRequiredLookback`) aufgebaut wurde.
* **Transformation (`TPipeLambda`):** Sobald die Barriere durchbrochen ist, wird die Transformations-Funktion ausgeführt. Diese liest die synchronisierten 1D-Serien und generiert einen neuen skalaren Output.
* **Emission:** Der generierte Output wird wieder als neues Signal (`Emit()`) mit der aktuellen `CycleID` an nachgelagerte Streams verschickt.

---

## 6. AST & Compiler-Integration des `pipe`-Statements

Der Aufruf und die Definition von Pipes sind tief in die Delphi-AST-Pipeline integriert. Der Weg vom Quellcode zur Ausführung verläuft in folgenden Schritten:

### a) Parser (`Myc.Ast.Script.pas`)
Die Grammatik für Pipes sieht im Skript wie folgt aus: `(pipe [inputs] lambda)`.
Die Methode `ParsePipe` erwartet exakt zwei Argumente:
1. Einen **Vektor** (`akTuple`), der die Inputs spezifiziert.
2. Einen **Lambda-Ausdruck** (`akLambdaExpression`), der die Berechnungslogik enthält.
Daraus wird im AST ein `TPipeNode` (Kind = `akPipe`) erzeugt.

### b) AST-Struktur (`Myc.Ast.Nodes.pas`)
Der `TPipeNode` speichert die Komponenten strukturiert ab:
* `FInputs`: Ein `ITupleNode`, das wiederum Tupel enthält, die die Ziel-Streams und die gewünschten Felder (Selektoren) referenzieren (z.B. `[my_ticks [:price :volume]]`).
* `FTransformation`: Ein `ILambdaExpressionNode`, das auf die Inputs angewandt wird.

### c) TypeChecker (`Myc.Ast.Compiler.TypeChecker.pas`)
Der TypeChecker analysiert den `TPipeNode` detailliert (`VisitPipe`):
* Er verifiziert die Struktur der `Inputs` (muss ein Array von Streams und deren Keyword-Selektoren sein).
* Er validiert, dass die referenzierten Felder (z.B. `:price`) in der `RecordDefinition` des jeweiligen Streams auch tatsächlich existieren.
* Er passt die Signatur (die Parameter) des Lambdas automatisch an die statischen Typen der ausgewählten Stream-Felder an.
* Er leitet den statischen Rückgabetyp ab. Je nach Lambda-Rückgabe ist das Ergebnis entweder wieder eine `stRecordSeries` (die Pipe generiert einen neuen Stream) oder `stVoid` (ein reiner Endpunkt ohne Ausgabe).

### d) Evaluator (`Myc.Ast.Evaluator.pas`)
Zur Laufzeit (`VisitPipe`) wird die Pipe materialisiert:
1. Die Inputs werden über den aktuellen Scope ("Environment") als `IStream`-Instanzen aufgelöst.
2. Die Feld-Selektoren werden als Konfiguration (`TPipeConfig`) aufgebaut.
3. Der Lambda-Ausdruck wird zu einer ausführbaren Funktion (`TDataValue.TFunc`) evaluiert.
4. Es wird ein neuer `TPipeStream` instanziiert, der die Source-Streams und die Lambda-Funktion als `TPipeLambda`-Adapter übergeben bekommt.
5. Der `TPipeStream` meldet sich bei den Quell-Streams als Observer (Subscriber) an und das reaktive Netzwerk ist bereit.

---

## Fazit & Zielarchitektur für Rust: Die Duale Serien-Architektur

Die ursprüngliche Annahme, die aktuelle Rust-Architektur (`RecordSeries`) sei grundlegend fehlerhaft und müsse abgelöst werden, greift zu kurz. Stattdessen haben wir es mit **zwei völlig unterschiedlichen Konzepten und Lebenszyklen** zu tun, die im Skript zwar unter dem gemeinsamen Begriff "Serie" zusammengefasst werden, intern in der Rust-Implementierung jedoch strikt getrennt sein müssen:

### 1. Die Skript-lokale Serie (Die aktuelle `RecordSeries` / SoA)
* **Zweck:** Dient als lokaler, zustandsbehafteter Puffer oder Akkumulator. Ein typischer Anwendungsfall ist die Definition innerhalb einer Indikator-Berechnung (z.B. ein lokaler Ringpuffer für einen SMA).
* **Verhalten:** Vom Skript-Autor explizit deklariert und mutiert (read/write). Die Serie existiert nur im lokalen Scope der Ausführung.
* **Aktualisierung:** Erfolgt manuell durch das Skript (z.B. mittels eines expliziten `push`-Befehls in einer Schleife oder Berechnungslogik).
* **Bewertung:** Die aktuelle Rust-Implementierung als Struct-of-Arrays (SoA) ist für diesen Anwendungsfall **perfekt geeignet**, da sie maximale Performance für schnelle, massenhafte lokale Operationen bietet. **Sie bleibt als Datenstruktur erhalten, darf aber nicht in das reaktive Event-System integriert werden.**

### 2. Die reaktive Pipeline & Publizierte Serien (`SharedSeries`)
* **Zweck:** Dies ist das systemgesteuerte, reaktive Rückgrat der Engine. Hier fließen kontinuierlich Marktdaten (Ticks) als getaktete Signale durch den von Pipes aufgespannten Berechnungsgraphen.
* **Verhalten:** Aus Sicht des Skripts sind diese Serien **streng Read-Only**. Das Skript kann diese Serien nur als Input konsumieren (z.B. für Lookbacks: `close[0]`), aber keine Werte hineinpushen.
* **Speicher-Architektur:** Die Datenhaltung erfolgt *dedupliziert* im Hintergrund. Wenn zehn verschiedene Indikator-Pipes denselben Input-Datenstrom (z.B. den Preis) konsumieren, teilen sie sich exakt denselben Speicherbereich im RAM. Dies ist das Rust-Äquivalent zur Delphi `TSharedSeries`.

### 3. Der Klebstoff: Polymorphismus in der VM (`Series` Trait)
Damit der Skript-Autor keinen syntaktischen Unterschied zwischen der lokalen `RecordSeries` und einer reaktiven `SharedSeries` bemerkt (z.B. bei einem Lookback-Zugriff `my_series[0]`), müssen beide Konzepte hinter einer gemeinsamen Schnittstelle verborgen werden. Aus der Perspektive des AST und der VM (Evaluator) implementieren beide Typen dasselbe Verhalten für Lesezugriffe.

---

## Konkreter Implementierungsplan für Rust

Um diese duale Architektur sauber und idiomatisch in Rust abzubilden, sind folgende konkrete Implementierungsschritte erforderlich:

### Schritt 1: Die polymorphe Serien-Abstraktion
* **`Series`-Trait definieren:** Einführung eines Traits (oder eines entsprechenden Enums in der VM-Darstellung wie `BoundSeries`), das die gemeinsamen Leseoperationen definiert. Zwingend erforderlich ist hier die Index-Operation für Lookbacks (z.B. `fn get_lookback(&self, index: usize) -> ScalarValue`).
* **Anpassung der VM:** Der TypeChecker und Evaluator arbeiten zukünftig gegen diese Abstraktion, nicht mehr direkt gegen die konkrete `RecordSeries`, wenn es um Lesezugriffe geht.

### Schritt 2: Implementierung der `SharedSeries`
* **Struktur:** Eine neue Datenstruktur `SharedSeries`, die das polymorphe `Series`-Trait implementiert.
* **Shared State:** Intern hält sie eine Referenz (`Rc<RefCell<...>>` oder ähnlich, unter Einhaltung der Vorgabe, dass innerhalb des Root-Scopes alles Single-Threaded abläuft) auf den deduplizierten Datenpuffer der Pipeline.
* **Read-Only Enforcement:** Implementierungsseitig werden alle Mutationsversuche (wie `push` oder `clear`) mit einem Panic oder Runtime-Error blockiert.

### Schritt 3: Das reaktive Event-Routing (Stateless Streams)
* **Publisher (`Stream`-Trait):** Definition eines zustandslosen Traits für Komponenten, die Events aussenden können.
* **Signal & Taktung:** Einführung des `Signal`-Structs (analog zu `TStreamSignal`), welches die global generierte `CycleID` sowie einen Verweis auf die zu verarbeitenden Datensätze trägt.
* **`RootStream`:** Implementierung des Taktgebers. Er dient als Eintrittspunkt für neue externe Daten, generiert inkrementell die `CycleID` und ruft die nachgelagerten Observer auf.

### Schritt 4: Die Pipe-Logik (`PipeStream`)
* **Subscriber (`Observer`-Trait):** Definition für Komponenten, die sich an Streams anhängen (`Subscribe`).
* **Barrier-Synchronisation:** Der `PipeStream` sammelt eintreffende Signale seiner deklarierten Quell-Streams. Er puffert diese und wartet, bis alle Signale mit exakt derselben `CycleID` eingetroffen sind (Barrier-Wait).
* **Ausführung:** Sobald die Barriere durchbrochen ist (alle Inputs synchronisiert), wird das transformierende Lambda (das in der Pipe definierte Closure) ausgeführt.
* **Kaskadierung:** Das Ergebnis des Lambdas wird sofort wieder als neues Signal mit der aktuellen `CycleID` an weitere nachgelagerte Pipes (via `Emit()`) weitergereicht.

### Schritt 5: AST & Compiler-Integration des `pipe`-Statements
* **Lexer/Parser:** Erweiterung der Grammatik für das Konstrukt `(pipe [inputs] lambda)` und Erzeugung eines entsprechenden `akPipe` AST-Knotens.
* **TypeChecker:** Statische Analyse der Pipe. Überprüfung der Input-Selektoren gegen die `RecordDefinition` der referenzierten Streams. Automatische Inferenz der Typ-Signaturen, um die Parameter des übergebenen Lambdas zur Compilezeit korrekt an die statischen Datentypen der Stream-Felder zu binden.
* **Evaluator:** Implementierung der Laufzeit-Logik (`VisitPipe`), die das Lambda zu einer ausführbaren Funktion kompiliert, den `PipeStream` als reaktiven Knoten instanziiert, im Graphen einhängt und mit den entsprechenden Quell-Streams verdrahtet.