# Projektplan: Meilenstein Daten-Infrastruktur

*Datum: 13. Juni 2025*

## Status: Design-Phase abgeschlossen, Basis-Implementierung erfolgt

---

## Stufe 1: Asynchrone Datenstrom-Schnittstelle (`IDataStream<T>`) - **ABGESCHLOSSEN**

* **Motivation**: Das ursprüngliche, zustandsbehaftete Design war für die Verarbeitung asynchron eintreffender Daten (z.B. von `TFuture`-Objekten) zu komplex und fehleranfällig.

* **Ziel**: Die Schaffung eines robusten, ereignisgesteuerten und non-blocking Modells, das den Datenproduzenten sauber vom Konsumenten entkoppelt.

* **Ergebnis**:
    * Die `IDataStream<T>`-Schnittstelle wurde überarbeitet. Die `HasData`-Eigenschaft liefert nun ein `TSignal`, das Konsumenten aktiv über potenziell neue Daten informiert.
    * Die Referenzimplementierung `TAuraFileStream<T>` wurde erfolgreich angepasst und nutzt eine saubere Ereignis-Kopplung (`Subscribe`) für eine robuste und wartungsarme Logik.
    * Die korrekte Funktionalität wurde durch eine angepasste DUnitX-Test-Suite verifiziert.

---

## Stufe 2: Abstraktion für Handelssysteme (`IDataSeriesProvider`) - **ENTWORFEN**

* **Motivation**: Ein Handelssystem benötigt einen stets validen und kontinuierlichen Daten-Lookback. Ein roher `IDataStream<T>` kann dies nicht garantieren, da Lücken in den Daten auftreten können (z.B. beim Übergang von Historie zu Live).

* **Ziel**: Die Konzeption einer übergeordneten Abstraktionsschicht, die diese komplexe Anforderung kapselt, die Datenintegrität sicherstellt und dem Handelssystem eine einfache, sichere Schnittstelle bietet.

* **Ergebnis**:
    Entworfen wurde der `IDataSeriesProvider`, der als "Black Box" für das Handelssystem fungiert und die Komplexität der Datenbeschaffung vollständig verbirgt. Er wurde mit zwei unterschiedlichen, vom Anwender wählbaren Betriebsmodi konzipiert:

    ### Modus 1: "Live-Handel"
    * **Motivation**: Um einen echten **"Sofort-Start"** im Live-Handel zu ermöglichen, muss die Lücke zwischen den statischen, lokalen Historiendaten und dem aktuellen Zeitpunkt geschlossen werden.
    * **Ziel**: Ein lückenloser, tagesaktueller Start des Handelssystems ohne manuelles Eingreifen oder lange Wartezeiten für den Nutzer.
    * **Ergebnis**: Das Design einer **Drei-Phasen-Synchronisation**: (1) Lokale History laden, (2) "Catch-up"-Daten vom Broker holen, (3) auf den Live-Stream umschalten.

    ### Modus 2: "Simulation & Backtest"
    * **Motivation**: Für Entwicklung, Test und Analyse muss die Software **völlig autonom** und ohne Abhängigkeit von einer externen, potenziell nicht verfügbaren Broker-API lauffähig sein.
    * **Ziel**: Einen "Sofort-Start" für Backtests zu jedem beliebigen Zeitpunkt in der Vergangenheit zu ermöglichen, der rein auf lokalen Dateien basiert.
    * **Ergebnis**: Ein Design, bei dem der relevante Datenkontext in den Speicher geladen wird, um von dort aus einen schnellen Start und ein "Playback" der Daten zu ermöglichen.

Src/Myc.Test.Trade.DataPoint.pas

@@ -170,7 +170,11 @@ begin
     ATimeStamp := EncodeDate(2020, 7, 7); // Oldest item
     expectedIndex := 9;
 
-    Assert.AreEqual(expectedIndex, FSeries.IndexOf(ATimeStamp), 'IndexOf for the oldest existing item timestamp should return its correct index');
+    Assert.AreEqual(
+        expectedIndex,
+        FSeries.IndexOf(ATimeStamp),
+        'IndexOf for the oldest existing item timestamp should return its correct index'
+    );
 end;
 
 // Tests finding the index for a timestamp that falls between two existing items.
@@ -183,7 +187,11 @@ begin
     ATimeStamp := EncodeDate(2020, 7, 7) + 0.5; // 12:00 on the day of the oldest item
     expectedIndex := Int64(9); // Should find the item from 00:00
 
-    Assert.AreEqual(expectedIndex, FSeries.IndexOf(ATimeStamp), 'IndexOf for a non-existing timestamp should return the index of the immediately preceding item');
+    Assert.AreEqual(
+        expectedIndex,
+        FSeries.IndexOf(ATimeStamp),
+        'IndexOf for a non-existing timestamp should return the index of the immediately preceding item'
+    );
 end;
 
 // Tests finding a timestamp that is older than any item in the series.
@@ -209,7 +217,11 @@ begin
     baseTime := EncodeDate(2020, 7, 7);
     ATimeStamp := baseTime + 10; // A day after the newest item
 
-    Assert.AreEqual(Int64(0), FSeries.IndexOf(ATimeStamp), 'IndexOf for a timestamp after the newest item should return the index of the newest item');
+    Assert.AreEqual(
+        Int64(0),
+        FSeries.IndexOf(ATimeStamp),
+        'IndexOf for a timestamp after the newest item should return the index of the newest item'
+    );
 end;
 
 // Specifically tests finding the exact oldest item.

Src/Myc.Test.Trade.DataStream.pas

@@ -81,42 +81,62 @@ end;
 procedure TTest_TABFileServer_Equivalence.Test_ServerReturnsSameDataAs_LoadDataSeries;
 var
     chunk: array[0..C_MAX_FETCH - 1] of TDataPoint<TAskBidItem>;
-    Dst: TArray<TDataPoint<TAskBidItem>>;
+    dst: TArray<TDataPoint<TAskBidItem>>;
     i: Int64;
+    n: Integer;
+    cnt: Int64;
+    timeout: Integer;
+    filename: string;
+    expectedData: TArray<TDataPoint<TAskBidItem>>;
 begin
-    // Loads the expected data directly using TAskBid.LoadDataSeries.WaitFor.
+    // 1. Expected data is loaded directly using LoadDataSeries.
+    filename := TAuraTABFileServer.FindFirstDataFile(C_TEST_PATH, C_TEST_SYMBOL);
+    Assert.IsNotEmpty(filename, 'Test data file could not be found.');
 
-    var Filename := TAuraTABFileServer.FindFirstDataFile(C_TEST_PATH, C_TEST_SYMBOL);
+    expectedData := FServer.LoadDataSeries(filename).WaitFor;
+    SetLength(dst, Length(expectedData));
 
-    Assert.IsNotEmpty(Filename);
-
-    var ExpectedData := FServer.LoadDataSeries(Filename).WaitFor;
-    SetLength(Dst, Length(ExpectedData));
-
-    var cnt: Int64 := 0;
-    var n: Integer;
-    // Fetches data chunks from the server until all data is retrieved or server indicates live data.
-    while not FStream.IsLiveData.Value and (cnt < Length(Dst)) do
+    // 2. Fetch data chunks from the stream until all data is retrieved.
+    cnt := 0;
+    timeout := 0;
+    while (cnt < Length(expectedData)) do
     begin
         n := FStream.GetChunk(chunk);
         if n > 0 then
-            Move(chunk[0], dst[cnt], n * sizeof(TDataPoint<TAskBidItem>));
-        inc(cnt, n);
+        begin
+            Move(chunk[0], dst[cnt], n * SizeOf(TDataPoint<TAskBidItem>));
+            Inc(cnt, n);
+            timeout := 0; // Reset timeout on progress
+        end
+        else
+        begin
+            // If GetChunk returns 0, the stream might be waiting for async I/O.
+            // A small sleep prevents a tight loop from consuming 100% CPU.
+            Sleep(1);
+            Inc(timeout);
+            Assert.IsTrue(timeout < 5000, 'Test timed out waiting for data from stream.');
+        end;
     end;
 
+    // 3. A final call should return 0, as the stream must be depleted.
+    n := FStream.GetChunk(chunk);
+    Assert.AreEqual(0, n, 'Stream returned data after it should have been depleted.');
+    n := FStream.GetChunk(chunk);
+    Assert.AreEqual(0, n, 'Stream returned data after it should have been depleted.');
+
     // --- Assertions ---
 
-    // 1. Verify that the total number of records fetched from the server matches the expected count.
-    Assert.AreEqual(Length(ExpectedData), cnt, 'Data record count mismatch');
+    // 4. Verify that the total number of records fetched matches the expected count.
+    Assert.AreEqual(Length(expectedData), cnt, 'Data record count mismatch.');
 
-    // 2. Verify that the content of each record is identical.
+    // 5. Verify that the content of each record is identical.
     // The loop checks every 1000th element for efficiency.
-    for n := 0 to High(ExpectedData) div 1000 do
+    for n := 0 to High(expectedData) div 1000 do
     begin
         i := n * 1000;
-        Assert.AreEqual(ExpectedData[i].Time, Dst[i].Time, 'Timestamp mismatch');
-        Assert.AreEqual(ExpectedData[i].Data.Ask, Dst[i].Data.Ask, 'Ask price mismatch');
-        Assert.AreEqual(ExpectedData[i].Data.Bid, Dst[i].Data.Bid, 'Bid price mismatch');
+        Assert.AreEqual(expectedData[i].Time, dst[i].Time, 'Timestamp mismatch at index ' + i.ToString);
+        Assert.AreEqual(expectedData[i].Data.Ask, dst[i].Data.Ask, 'Ask price mismatch at index ' + i.ToString);
+        Assert.AreEqual(expectedData[i].Data.Bid, dst[i].Data.Bid, 'Bid price mismatch at index ' + i.ToString);
     end;
 end;
 

Src/Myc.Trade.DataPoint.pas

@@ -22,7 +22,7 @@ type
 
     // A time-ordered series of data points, optimized for chronological additions.
     // The most recently added element has the logical index 0.
-    TDataSeries<T> = record
+    TDataSeries<T: record> = record
     private
         const
             ChunkSize = 1024;
@@ -31,12 +31,18 @@ type
     private
         FChunks: TArray<TChunk>;
         FCount: Int64;
+        // Converts a logical index (0=newest) to a physical storage index (0=oldest).
+        function LogicalToPhysicalIndex(LogicalIndex: Int64): Int64; inline;
         function GetCount: Int64;
+        function GetData(Idx: Int64): T;
         function GetItems(Idx: Int64): TDataPoint<T>;
+        function GetTime(Idx: Int64): TDateTime;
+        procedure SetData(Idx: Int64; const Value: T);
     public
         // Adds a new data point to the series.
         procedure Add(const Data: TDataPoint<T>);
-
+        // Clears all data from the series.
+        procedure Clear;
         // Searches for a data point by its timestamp.
         // Returns the logical index of the matching item.
         // If no exact match, returns the index of the item immediately preceding the timestamp.
@@ -47,10 +53,11 @@ type
         class operator Finalize(var Dest: TDataSeries<T>);
 
         property Count: Int64 read GetCount;
-
         // Accesses data points by their logical index.
         // Index 0 is the newest element, Index (Count - 1) is the oldest.
         property Items[Idx: Int64]: TDataPoint<T> read GetItems; default;
+        property Time[Idx: Int64]: TDateTime read GetTime;
+        property Data[Idx: Int64]: T read GetData write SetData;
     end;
 
 implementation
@@ -75,17 +82,14 @@ end;
 
 procedure TDataSeries<T>.Add(const Data: TDataPoint<T>);
 begin
-    // Enforce chronological order: new items cannot be older than the newest existing item.
-    // This is a prerequisite for the binary search in IndexOf to work correctly.
-    Assert((Length(FChunks) = 0) or (Data.Time >= GetItems(0).Time), 'Time stamp older than last item');
+    // Enforce chronological order for new items.
+    Assert((FCount = 0) or (Data.Time >= GetTime(0)), 'Time stamp older than last item');
 
     var ci := FCount div ChunkSize;
     var di := FCount mod ChunkSize;
 
-    if di = 0 then
+    if (di = 0) then
     begin
-        Assert(ci = Length(FChunks));
-        Assert(di = 0);
         SetLength(FChunks, ci + 1);
         SetLength(FChunks[ci], ChunkSize);
     end;
@@ -94,10 +98,15 @@ begin
     Inc(FCount);
 end;
 
+procedure TDataSeries<T>.Clear;
+begin
+    FChunks := nil;
+    FCount := 0;
+end;
+
 class operator TDataSeries<T>.Finalize(var Dest: TDataSeries<T>);
 begin
-    Dest.FChunks := nil;
-    Dest.FCount := 0;
+    Dest.Clear;
 end;
 
 function TDataSeries<T>.GetCount: Int64;
@@ -105,12 +114,28 @@ begin
     Result := FCount;
 end;
 
-function TDataSeries<T>.GetItems(Idx: Int64): TDataPoint<T>;
+function TDataSeries<T>.GetData(Idx: Int64): T;
+var
+    physicalIndex: Int64;
 begin
-    Assert((Idx >= 0) and (Idx < FCount));
-    // Convert logical index (0 = newest) to physical index (0 = oldest).
-    Idx := FCount - Idx - 1;
-    Result := FChunks[Idx div ChunkSize][Idx mod ChunkSize];
+    physicalIndex := LogicalToPhysicalIndex(Idx);
+    Result := FChunks[physicalIndex div ChunkSize][physicalIndex mod ChunkSize].Data;
+end;
+
+function TDataSeries<T>.GetItems(Idx: Int64): TDataPoint<T>;
+var
+    physicalIndex: Int64;
+begin
+    physicalIndex := LogicalToPhysicalIndex(Idx);
+    Result := FChunks[physicalIndex div ChunkSize][physicalIndex mod ChunkSize];
+end;
+
+function TDataSeries<T>.GetTime(Idx: Int64): TDateTime;
+var
+    physicalIndex: Int64;
+begin
+    physicalIndex := LogicalToPhysicalIndex(Idx);
+    Result := FChunks[physicalIndex div ChunkSize][physicalIndex mod ChunkSize].Time;
 end;
 
 class operator TDataSeries<T>.Initialize(out Dest: TDataSeries<T>);
@@ -134,7 +159,7 @@ begin
     while (low <= high) do
     begin
         mid := low + (high - low) div 2;
-        dataPointTime := GetItems(mid).Time;
+        dataPointTime := GetTime(mid);
 
         if (dataPointTime = TimeStamp) then
         begin
@@ -153,4 +178,18 @@ begin
     end;
 end;
 
+function TDataSeries<T>.LogicalToPhysicalIndex(LogicalIndex: Int64): Int64;
+begin
+    Assert((LogicalIndex >= 0) and (LogicalIndex < FCount), 'Logical index is out of bounds.');
+    Result := FCount - LogicalIndex - 1;
+end;
+
+procedure TDataSeries<T>.SetData(Idx: Int64; const Value: T);
+var
+    physicalIndex: Int64;
+begin
+    physicalIndex := LogicalToPhysicalIndex(Idx);
+    FChunks[physicalIndex div ChunkSize][physicalIndex mod ChunkSize].Data := Value;
+end;
+
 end.

Src/Myc.Trade.DataStream.pas

@@ -31,18 +31,24 @@ uses
 type
 
     // Represents a generic data stream capable of providing sequential data chunks.
+    // IsHistory:
+    //   - true, if this stream is a history stream. Once HasData becomes false, it reached it's end and will not provide more data.
+    //   - false, we expect more Data to come. This stream has no end.
+    // HasData: set, if a call to GetChunk will return new data.
     IDataStream<T> = interface
         ['{A6E246A2-E84E-49AB-A63E-333E561E488C}']
-        function GetIsLiveData: TMutable<Boolean>;
+        function GetHasData: TSignal;
         function GetChunk(var Data: array of TDataPoint<T>): Integer;
-        property IsLiveData: TMutable<Boolean> read GetIsLiveData;
+        function IsHistory: Boolean;
+        property HasData: TSignal read GetHasData;
     end;
 
     // Abstract base class for IDataStream implementations.
     TDataStream<T> = class(TInterfacedObject, IDataStream<T>)
     protected
-        function GetIsLiveData: TMutable<Boolean>; virtual; abstract;
+        function GetHasData: TSignal; virtual; abstract;
         function GetChunk(var Data: array of TDataPoint<T>): Integer; virtual; abstract;
+        function IsHistory: Boolean; virtual; abstract;
     end;
 
     // Represents a factory for creating IDataStream instances.
@@ -131,10 +137,10 @@ type
     end;
 
     // Implements a data stream that reads from Aura-specific historical data files.
-    TAuraFileStream<T: record> = class(TDataStream<T>, IDataStream<T>)
+    TAuraFileStream<T: record> = class(TDataStream<T>)
     private
         FDataServer: TAuraDataServer<T>;
-        FIsLiveData: TMutable<Boolean>.IWriteable;
+        FHasData: TEvent;
         FCurrentFileName: string;
         FCurrentData: TFuture<TArray<TDataPoint<T>>>;
         FNextFileName: string;
@@ -142,10 +148,12 @@ type
         FCurrPosInFile: Int64;
         FLastTimeStamp: TDateTime;
     protected
+        function GetHasData: TSignal; override;
         function GetChunk(var Data: array of TDataPoint<T>): Integer; override;
-        function GetIsLiveData: TMutable<Boolean>; override;
+        function IsHistory: Boolean; override;
     public
         constructor Create(ADataServer: TAuraDataServer<T>; const AFilename: String);
+        destructor Destroy; override;
         procedure AfterConstruction; override;
     end;
 
@@ -581,21 +589,30 @@ begin
     Assert(Assigned(ADataServer));
     FDataServer := ADataServer;
     FCurrentFileName := AFilename;
-    FIsLiveData := TMutable<Boolean>.CreateWriteable(false);
+    // Create an event
+    FHasData := TEvent.CreateEvent; // interface helper benutzen!
+end;
+
+destructor TAuraFileStream<T>.Destroy;
+begin
+    inherited;
 end;
 
 procedure TAuraFileStream<T>.AfterConstruction;
 begin
     inherited;
-    FCurrentData := FDataServer.LoadDataFile(FCurrentFileName);
-    FCurrPosInFile := 0;
-    FIsLiveData.SetValue(false);
     FLastTimeStamp := 0;
+    FCurrPosInFile := 0;
+    FCurrentData := FDataServer.LoadDataFile(FCurrentFileName);
+    // Forward all future done events, because this means there is new data.
+    FCurrentData.Done.Subscribe( FHasData );
+
     FNextFileName := FDataServer.FindNextDataFile(FCurrentFileName);
     if FNextFileName <> '' then
-        FNextData := FDataServer.LoadDataFile(FNextFileName)
-    else
-        FNextData := nil;
+    begin
+        FNextData := FDataServer.LoadDataFile(FNextFileName);
+        FNextData.Done.Subscribe( FHasData );
+    end;
 end;
 
 function TAuraFileStream<T>.GetChunk(var Data: array of TDataPoint<T>): Integer;
@@ -604,39 +621,37 @@ var
     currData: TArray<TDataPoint<T>>;
 begin
     Result := 0;
+
+    // This is an asynchronous operation! We don't wait for data. It's totally valid to result nothing, if there is nothing.
     if not FCurrentData.Done.IsSet then
         exit;
 
-    var maxLen := Length(Data);
-    currData := FCurrentData.Value;
-    while Result < maxLen do
+    if FCurrPosInFile >= Length(FCurrentData.Value) then
     begin
-        if FCurrPosInFile >= Length(currData) then
-        begin
-            FCurrentData := FNextData;
-            FCurrentFileName := FNextFileName;
-            FCurrPosInFile := 0;
-            if FCurrentFileName <> '' then
-            begin
-                FNextFileName := FDataServer.FindNextDataFile(FCurrentFileName);
-                if FNextFileName <> '' then
-                    FNextData := FDataServer.LoadDataFile(FNextFileName)
-                else
-                    FNextData := nil;
+        FCurrentData := FNextData;
+        FCurrentFileName := FNextFileName;
+        FCurrPosInFile := 0;
+        FNextData := TFuture<TArray<TDataPoint<T>>>.Null;
 
-                if FCurrentData.Done.IsSet then
-                begin
-                    currData := FCurrentData.Value;
-                    continue;
-                end;
-            end
-            else
-            begin
-                FIsLiveData.SetValue(true);
-            end;
-            break;
+        if FCurrentFileName = '' then
+            exit;
+
+        FNextFileName := FDataServer.FindNextDataFile(FCurrentFileName);
+        if FNextFileName <> '' then
+        begin
+            FNextData := FDataServer.LoadDataFile(FNextFileName);
+            FNextData.Done.Subscribe( FHasData );
         end;
 
+        if not FCurrentData.Done.IsSet then
+            exit;
+    end;
+
+    currData := FCurrentData.Value;
+
+    var maxLen := Length(Data);
+    while (Result < maxLen) and (FCurrPosInFile < Length(currData)) do
+    begin
         item := currData[FCurrPosInFile];
         if FLastTimeStamp < item.Time then
         begin
@@ -646,11 +661,23 @@ begin
         end;
         Inc(FCurrPosInFile);
     end;
+
+    // If there is data left in the current file, signal new data. If it is finished, we do nothing, because
+    // the next signal will come from the next future being done.
+    if FCurrPosInFile < Length(currData) then
+    begin
+        FHasData.Notify;
+    end;
 end;
 
-function TAuraFileStream<T>.GetIsLiveData: TMutable<Boolean>;
+function TAuraFileStream<T>.GetHasData: TSignal;
 begin
-    Result := FIsLiveData;
+    Result := FHasData.Signal;
+end;
+
+function TAuraFileStream<T>.IsHistory: Boolean;
+begin
+    Result := True;
 end;
 
 { TAuraTABFileServer }