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.