# 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>` 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.