From fc858de59c5556071d142372b862bef337e2e714 Mon Sep 17 00:00:00 2001 From: Michael Schimmel Date: Sun, 1 Mar 2026 21:19:30 +0100 Subject: [PATCH] Add Delphi Stream Architecture Analysis This commit introduces a detailed analysis of the Delphi Reactive Stream Pipeline architecture, contrasting it with the current Rust implementation. It outlines the core concepts of Delphi's stateless streams, `TStreamSignal`, `TRootStream`, `TPipeSource`, `TScalarSeries`, and `TPipeStream`, emphasizing their roles in event propagation, state management, and synchronization. The document also details the integration of the `pipe` statement within the Delphi AST and compiler pipeline, covering the parser, AST structure, type checker, and evaluator. Finally, it proposes a "Dual Series Architecture" for Rust, differentiating between script-local `RecordSeries` (for accumulators) and reactive `SharedSeries` (for the pipeline), and outlines a concrete implementation plan. --- docs/Delphi_Stream_Architecture.md | 128 +++++++++++++++++++++++++++++ 1 file changed, 128 insertions(+) create mode 100644 docs/Delphi_Stream_Architecture.md diff --git a/docs/Delphi_Stream_Architecture.md b/docs/Delphi_Stream_Architecture.md new file mode 100644 index 0000000..f00e8c2 --- /dev/null +++ b/docs/Delphi_Stream_Architecture.md @@ -0,0 +1,128 @@ +# 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. \ No newline at end of file