Refactoring DataStream

2025-06-07 16:41:46 +02:00
parent 590e98d614
commit 776067d0a8
9 changed files with 708 additions and 1028 deletions
@@ -39,7 +39,7 @@ Code-Generierung
  - Einrückung mit 4 Leerzeichen anstelle von 2. Auch bei Kommentaren.
  - Das Code-Format ist UTF-8.
  - Folgende Schlüsselwörter müssen klein geschrieben werden:
-	and, or, not, mod, div, in, as, is, array of, sizeof
+	and, or, not, mod, div, in, as, is, array of, sizeof(), inc(), dec()
 * Ändere niemals vorhandene Bezeichner im vom Benutzer bereitgestellten Code, es sei denn du wirst dazu aufgefordert.
@@ -1,172 +0,0 @@
 unit Myc.Core.DataCache;
 interface
 uses
    System.SysUtils,
    System.Generics.Collections,
    System.Generics.Defaults,
    Myc.Futures, // Required for TFuture
    Myc.Signals; // Required for TLatch
 type
    // Represents a single cached file entry with its metadata and loaded data future.
    // TArrayItemType is the type of a single item within the TArray that is loaded by the future.
    TCachedFile<TArrayItemType> = class
        Name: String; // The full path and name of the cached file.
        Age: TDateTime; // The modification timestamp of the file at the time of caching.
        LastUsed: TDateTime; // The last time this cached entry was accessed.
        Data: TFuture<TArray<TArrayItemType>>; // A future representing the asynchronously loaded data.
    end;
    // Generic in-memory cache for data files, supporting asynchronous loading
    // and eviction based on memory pressure and last usage.
    // TArrayItemType is the type of a single item within the TArray that is loaded by the future,
    // e.g., TDataPoint<TAskBidItem>.
    TDataCache<TArrayItemType> = class
    strict private
        FCachedFiles: TDictionary<String, TCachedFile<TArrayItemType>>;
        FLoadGate: TLatch;
        FIsMemoryLow: TFunc<Boolean>;
    private
        // Manages cache size by evicting least recently used items if memory is low.
        procedure ManageCacheSize(const CurrentFileNameBeingLoaded: string);
    public
        constructor Create(const ALoadGate: TLatch; const AIsMemoryLow: TFunc<Boolean>);
        destructor Destroy; override;
        // Clears all entries from the cache.
        procedure Clear;
        // Attempts to retrieve data for a given file from the cache.
        // If found and not stale, updates LastUsed and returns true.
        // Otherwise, returns false.
        function TryGetValue(const FileName: string; out ACachedFile: TCachedFile<TArrayItemType>): Boolean;
        // Adds a new or updates an existing entry in the cache.
        procedure AddOrUpdate(const FileName: string; const AAge: TDateTime; const AData: TFuture<TArray<TArrayItemType>>);
        // Returns the TLoadGate associated with this cache, for coordination of load operations.
        function GetLoadGate: TLatch; inline;
        property LoadGate: TLatch read GetLoadGate;
    end;
 implementation
 uses
    System.Math;
 { TDataCache<TArrayItemType> }
 constructor TDataCache<TArrayItemType>.Create(const ALoadGate: TLatch; const AIsMemoryLow: TFunc<Boolean>);
 begin
    inherited Create;
    FCachedFiles := TObjectDictionary<String, TCachedFile<TArrayItemType>>.Create([doOwnsValues]);
    FLoadGate := ALoadGate;
    FIsMemoryLow := AIsMemoryLow;
 end;
 destructor TDataCache<TArrayItemType>.Destroy;
 begin
    FCachedFiles.Free;
    inherited;
 end;
 procedure TDataCache<TArrayItemType>.Clear;
 begin
    FCachedFiles.Clear;
 end;
 procedure TDataCache<TArrayItemType>.AddOrUpdate(
    const FileName: string;
    const AAge: TDateTime;
    const AData: TFuture<TArray<TArrayItemType>>
 );
 var
    cachedFile: TCachedFile<TArrayItemType>;
 begin
    if FCachedFiles.TryGetValue(FileName, cachedFile) then
    begin
        // Update existing entry
        cachedFile.Age := AAge;
        cachedFile.Data := AData; // This will replace the old future, potentially freeing old data
        cachedFile.LastUsed := Now;
    end
    else
    begin
        // Add new entry
        cachedFile := TCachedFile<TArrayItemType>.Create;
        cachedFile.Name := FileName;
        cachedFile.Age := AAge;
        cachedFile.Data := AData;
        cachedFile.LastUsed := Now;
        FCachedFiles.Add(FileName, cachedFile);
    end;
 end;
 function TDataCache<TArrayItemType>.GetLoadGate: TLatch;
 begin
    Result := FLoadGate;
 end;
 procedure TDataCache<TArrayItemType>.ManageCacheSize(const CurrentFileNameBeingLoaded: string);
 begin
    if Assigned(FIsMemoryLow) and FIsMemoryLow() then
    begin
        var fL := FCachedFiles.Values.ToArray;
        // Sorts the cached files by LastUsed in ascending order (least recently used first).
        TArray.Sort<TCachedFile<TArrayItemType>>(
            fL,
            TComparer<TCachedFile<TArrayItemType>>.Construct(
                function(const Left, Right: TCachedFile<TArrayItemType>): Integer begin Result := Sign(Left.LastUsed - Right.LastUsed) end
            )
        );
        for var i := 0 to High(fL) do
        begin
            if not FIsMemoryLow() then // Stop evicting if memory pressure is relieved
                break;
            // Do not remove the file that is currently being loaded.
            if fL[i].Name <> CurrentFileNameBeingLoaded then
                FCachedFiles.Remove(fL[i].Name);
        end;
    end;
 end;
 function TDataCache<TArrayItemType>.TryGetValue(const FileName: string; out ACachedFile: TCachedFile<TArrayItemType>): Boolean;
 var
    age: TDateTime;
 begin
    ManageCacheSize( FileName );
    Result := FCachedFiles.TryGetValue(FileName, ACachedFile);
    if Result then
    begin
        // Check if the file on disk has changed since it was cached.
        if FileAge(FileName, age, true) then
        begin
            if age = ACachedFile.Age then
            begin
                // Cache hit and not stale, update LastUsed.
                ACachedFile.LastUsed := Now;
            end
            else
            begin
                // File is stale, remove from cache.
                FCachedFiles.Remove(FileName);
                Result := False;
            end;
        end
        else
        begin
            // File no longer exists, remove from cache.
            FCachedFiles.Remove(FileName);
            Result := False;
        end;
    end;
 end;
 end.
@@ -22,7 +22,7 @@ type
                property Done: TState read GetDone;
            end;
-            TFuncConst<T, TResult> = reference to function(const Arg1: T): TResult;
+            TFuncConst<S, TResult> = reference to function(const Arg1: S): TResult;
    {$REGION 'private'}
    strict private
@@ -10,8 +10,7 @@ uses
    Myc.Lazy,
    Myc.Futures,
    Myc.Trade.DataPoint,
-    Myc.Trade.DataSeries,
+    Myc.Trade.DataStream;
    Myc.Trade.DataServer;
 type
    // Test fixture for TDataServer and TAuraTABFileServer, focusing on data equivalence
@@ -20,7 +19,8 @@ type
    [IgnoreMemoryLeaks(true)]
    TTest_TABFileServer_Equivalence = class(TObject)
    private
-        FServer: IDataServer<TDataPoint<TAskBidItem>>;
+        FServer: IDataServer<TAskBidItem>;
        FStream: IDataStream<TDataPoint<TAskBidItem>>;
    public
        [SetupFixture]
        procedure SetupFixture;
@@ -50,12 +50,15 @@ const
 procedure TTest_TABFileServer_Equivalence.SetupFixture;
 begin
    FServer := TAskBidServer.Create;
    // TAskBid.ClearCache ensures a clean state for data series loading.
-    TAskBid.ClearCache;
+    FServer.ClearCache;
 end;
 procedure TTest_TABFileServer_Equivalence.TearDownFixture;
 begin
    FServer := nil;
    // Resets the global IsMemoryLow function pointer, if it was set for testing.
    IsMemoryLow := nil;
 end;
@@ -63,15 +66,15 @@ end;
 procedure TTest_TABFileServer_Equivalence.Setup;
 begin
    // Initializes the data server with a specific test file.
-    FServer := TAuraTABFileServer.Create(C_TEST_FILENAME);
+    FStream := TAuraTABFileStream.Create(C_TEST_FILENAME, FServer);
 end;
 procedure TTest_TABFileServer_Equivalence.TearDown;
 begin
    // Releases the data server instance.
-    FServer := nil;
+    FStream := nil;
    // Clears any cached data to prevent interference with subsequent tests.
-    TAskBid.ClearCache;
+    FServer.ClearCache;
 end;
 procedure TTest_TABFileServer_Equivalence.Test_ServerReturnsSameDataAs_LoadDataSeries;
@@ -81,15 +84,15 @@ var
    i: Int64;
 begin
    // Loads the expected data directly using TAskBid.LoadDataSeries.WaitFor.
-    var ExpectedData := TAskBid.LoadDataSeries(C_TEST_FILENAME).WaitFor;
+    var ExpectedData := FServer.LoadDataSeries(C_TEST_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 FServer.IsLiveData.Value and (cnt < Length(Dst)) do
+    while not FStream.IsLiveData.Value and (cnt < Length(Dst)) do
    begin
-        n := FServer.GetChunk(chunk);
+        n := FStream.GetChunk(chunk);
        if n > 0 then
            Move(chunk[0], dst[cnt], n * sizeof(TDataPoint<TAskBidItem>));
        inc(cnt, n);
@@ -1,757 +0,0 @@
 unit Myc.Trade.DataSeries;
 (*
  @unit Myc.Trade.DataSeries
  @brief Provides functionality to load time-series data from binary files.
  This unit offers a generic class `TDataSeries<T>` and helper functions to handle
  loading and processing of sequential data files, which are often used for
  historical market data.
  **File Naming Convention**
  The functions in this unit expect files to follow a specific naming pattern:
  - Historical data: `Symbol_YYYY_MM.tab` (uncompressed) or `Symbol_YYYY_MM.tab_zip` (compressed)
  - Live/update data: `Symbol_YYYY_MM.tab-live` (never compressed)
  `YYYY` represents the four-digit year and `MM` the two-digit month.
  If both a compressed (`_zip`) and an uncompressed version of a `.tab` file exist,
  the compressed version is always preferred.
  **Core Functionality**
  - `TDataSeries<T>.LoadDataSeries`: The main entry point to load a complete series.
  Given a starting file, it discovers all subsequent chronological `.tab` files by
  calling `FindNextDataFile`. It then looks for a corresponding `.tab-live` file
  for the last month in the series. All discovered files are loaded asynchronously
  and merged into a single, chronologically sorted array of unique data points.
  - `TDataSeries<T>.LoadDataFile`: Loads a single data file asynchronously. This
  function includes a caching mechanism to avoid rereading unchanged files.
  **Helper Functions**
  - `FindNextDataFile`: Given a filename, it finds and returns the path to the
  next file in the chronological sequence, if it exists.
  - `FindOldestFilesPerSymbol`: Scans a directory and finds the oldest (first)
  data file for each symbol based on year and month.
  - `TryParseFileName`: A utility function to parse filenames that adhere to the
  specified naming convention.
 *)
 interface
 uses
    System.SysUtils,
    System.Classes,
    System.Generics.Collections,
    System.Generics.Defaults,
    System.IOUtils,
    Myc.Futures,
    Myc.Signals,
    Myc.Lazy,
    Myc.Trade.DataPoint,
    Myc.Trade.DataServer;
 type
    // Generic class for loading and managing sequential time-series data.
    TDataSeries<T: record> = class
    public
        type
            // Represents a single data point with a timestamp and generic data.
            TDataPoint = packed record
                TimeStamp: TDateTime;
                Data: T;
            end;
    strict private
        type
            TCachedFile = class
                Name: String;
                Age: TDateTime;
                LastUsed: TDateTime;
                Data: TFuture<TArray<TDataPoint>>;
            end;
        class var
            FCachedFiles: TDictionary<String, TCachedFile>;
            FLoadGate: TLatch;
        class constructor CreateClass;
        class destructor DestroyClass;
    protected
        class function ReadCompressedData(const InputStream: TStream): TArray<TDataPoint>; static;
        class function ReadUncompressedData(const InputStream: TStream): TArray<TDataPoint>; static;
    public
        class procedure ClearCache;
        // Asynchronously loads a single data file with caching support.
        class function LoadDataFile(const FileName: string): TFuture<TArray<TDataPoint>>; static;
        // Asynchronously loads a complete chronological series of data files.
        class function LoadDataSeries(const FileName: string): TFuture<TArray<TDataPoint>>; static;
    end;
    // A specialized TDataSeries for handling Ask/Bid tick data.
    TAskBid = class(TDataSeries<TAskBidItem>)
    public
        type
            // Represents a single tick with Ask/Bid data.
            TTick = TDataSeries<TAskBidItem>.TDataPoint;
    end;
    // Implements a data server that reads data from Aura-specific historical data files.
    // It handles sequential loading of data files and transitions to "live data" mode
    // when all historical files are exhausted.
    TAuraFileServer<T: record> = class(TDataServer<TDataPoint<T>>, IDataServer<TDataPoint<T>>)
    private
        FIsLiveData: TMutable<Boolean>.IWriteable;
        FCurrentFileName: string;
        FCurrentData: TFuture<TArray<TDataSeries<T>.TDataPoint>>;
        FNextFileName: string;
        FNextData: TFuture<TArray<TDataSeries<T>.TDataPoint>>;
        FCurrPosInFile: Int64;
        FCurrentIdx: Int64;
        FLastTimeStamp: TDateTime;
    protected
        function GetChunk(var Data: array of TDataPoint<T>): Integer; override;
        function GetIsLiveData: TMutable<Boolean>; override;
    public
        constructor Create(const AFilename: String);
        procedure AfterConstruction; override;
    end;
    // Specialized TAuraFileServer for TAskBidItem data.
    TAuraTABFileServer = TAuraFileServer<TAskBidItem>;
 // Parses a filename to extract symbol, year, month, and path.
 function TryParseFileName(const FileName: string; out PathValue, SymbolValue: string; out YearValue, MonthValue: Integer): Boolean;
 // Scans a directory and finds the oldest data file for each symbol.
 function FindOldestFilesPerSymbol(const DirectoryPath: string): TArray<String>;
 // Finds the next chronological file in a data series, if it exists.
 function FindNextDataFile(const FileName: string): string;
 var
    // A function pointer to a callback that indicates if the system is low on memory.
    IsMemoryLow: TFunc<Boolean>;
 implementation
 uses
    System.Zip,
    System.Math,
    Myc.TaskManager;
 // TryParseFileName parses filenames like "Symbol_Year_MM.Extension" or "Symbol_Year_MM.Extension_zip"
 // to extract Path, Symbol, Year, and Month.
 function TryParseFileName(const FileName: string; out PathValue, SymbolValue: string; out YearValue, MonthValue: Integer): Boolean;
 var
    fileNameNoPath: string;
    nameForParsing: string;
    parts: TArray<string>;
    baseName: string;
 begin
    Result := False; // Default to failure
    PathValue := ''; // Initialize all out parameters
    SymbolValue := '';
    YearValue := 0;
    MonthValue := 0;
    PathValue := TPath.GetDirectoryName(FileName);
    fileNameNoPath := TPath.GetFileName(FileName);
    nameForParsing := fileNameNoPath;
    // Handle cases like "Symbol_Year_MM.Extension_zip" by removing the trailing "_zip" before extension removal
    if nameForParsing.EndsWith('_zip', True) then
    begin
        // Get the part before "_zip"
        baseName := nameForParsing.Substring(0, nameForParsing.Length - 4);
        // Get the extension of the original name (e.g. ".csv_zip")
        // and check if baseName still contains an extension that needs to be removed by GetFileNameWithoutExtension
        if TPath.GetExtension(baseName) <> '' then
        begin
            nameForParsing := baseName;
        end
        else
        begin
            // If baseName (e.g. "EURUSD_2023_01") has no extension, it is the correct parsing base
            // However, TPath.GetFileNameWithoutExtension would also work correctly here,
            // but this handles a potential case where a file might be named "Symbol_Year_MM_zip" (no dot before zip)
        end;
    end;
    // Remove the actual extension (e.g., ".csv", ".dat") from "Symbol_Year_MM.Extension" or the modified "Symbol_Year_MM.Extension" part of "_zip"
    nameForParsing := TPath.GetFileNameWithoutExtension(nameForParsing); // e.g., "EURUSD_2023_01"
    parts := nameForParsing.Split(['_']);
    if Length(parts) < 3 then // Need at least Symbol_Year_Month
        exit;
    // Month is the last part.
    if not TryStrToInt(parts[High(parts)], MonthValue) then
    begin
        exit;
    end;
    if (MonthValue < 1) or (MonthValue > 12) then // Validate month range
    begin
        MonthValue := 0; // Set to invalid if out of range
        exit;
    end;
    // Year is the second to last part.
    if not TryStrToInt(parts[High(parts) - 1], YearValue) then
    begin
        MonthValue := 0; // Also invalidate month if year parsing fails
        exit;
    end;
    // Ensure YearValue is plausible, e.g. not 0 or negative, adjust as needed for your data's year range
    if YearValue <= 0 then
    begin
        YearValue := 0;
        MonthValue := 0;
        exit;
    end;
    // Symbol consists of all parts before Year and Month.
    SymbolValue := string.Join('_', Copy(parts, 0, Length(parts) - 2));
    if SymbolValue = '' then // Ensure symbol is not empty
    begin
        // Reset all out parameters on failure before exit
        PathValue := '';
        YearValue := 0;
        MonthValue := 0;
        exit;
    end;
    Result := True;
 end;
 // Finds the oldest file for each symbol in the given directory.
 // The file extension is ignored for the purpose of identifying the symbol, year, and month.
 function FindOldestFilesPerSymbol(const DirectoryPath: string): TArray<String>;
 type
    TTrackedFileInfo = record
        FullFileName: string;
        Year: Integer;
        Month: Integer;
    end;
 var
    oldestFilesPerSymbol: TDictionary<string, TTrackedFileInfo>;
    fileNames: TArray<string>;
    currentFile: string;
    parsedPath: string;
    parsedSymbol: string;
    parsedYear: Integer;
    parsedMonth: Integer;
    trackedInfo: TTrackedFileInfo;
 begin
    oldestFilesPerSymbol := TDictionary<string, TTrackedFileInfo>.Create; // Dictionary to track the oldest file per symbol
    try
        if not TDirectory.Exists(DirectoryPath) then
        begin
            // Or raise an exception, depending on desired error handling
            exit;
        end;
        fileNames := TDirectory.GetFiles(DirectoryPath); // Get all files in the directory
        for currentFile in fileNames do
        begin
            // Attempt to parse the file name
            if TryParseFileName(currentFile, parsedPath, parsedSymbol, parsedYear, parsedMonth) then
            begin
                // Check if this symbol is already tracked or if this file is older
                if oldestFilesPerSymbol.TryGetValue(parsedSymbol, trackedInfo) then
                begin
                    // If current file's year is less, or year is same and month is less, it's older
                    if (parsedYear < trackedInfo.Year) or ((parsedYear = trackedInfo.Year) and (parsedMonth < trackedInfo.Month)) then
                    begin
                        trackedInfo.FullFileName := currentFile; // Store full path to the file
                        trackedInfo.Year := parsedYear;
                        trackedInfo.Month := parsedMonth;
                        oldestFilesPerSymbol.AddOrSetValue(parsedSymbol, trackedInfo); // Update the entry for this symbol
                    end;
                end
                else
                begin
                    // First time we see this symbol, so it's the oldest so far
                    trackedInfo.FullFileName := currentFile; // Store full path to the file
                    trackedInfo.Year := parsedYear;
                    trackedInfo.Month := parsedMonth;
                    oldestFilesPerSymbol.Add(parsedSymbol, trackedInfo); // Add new entry for this symbol
                end;
            end;
        end;
        // Populate the result array with the full names of the oldest files
        SetLength(Result, oldestFilesPerSymbol.Count);
        var n := 0;
        for trackedInfo in oldestFilesPerSymbol.Values do
        begin
            Result[n] := trackedInfo.FullFileName;
            inc(n);
        end;
    finally
        oldestFilesPerSymbol.Free; // Free the dictionary
    end;
 end;
 function FindNextDataFile(const FileName: string): string;
 var
    pathValue, symbolValue: string;
    yearValue, monthValue: Integer;
    nextBaseName, compressedPath, uncompressedPath: string;
 begin
    // Parse the input filename to extract its components.
    if not TryParseFileName(FileName, pathValue, symbolValue, yearValue, monthValue) then
    begin
        exit(''); // If parsing fails, there is no 'next' file.
    end;
    // Calculate the next month and year.
    Inc(monthValue);
    if monthValue > 12 then
    begin
        monthValue := 1;
        Inc(yearValue);
    end;
    // Construct the base name for the next file.
    nextBaseName := Format('%s_%.4d_%.2d.tab', [symbolValue, yearValue, monthValue]);
    // Check for the existence of the next file, preferring the compressed version.
    compressedPath := TPath.Combine(pathValue, nextBaseName + '_zip');
    if TFile.Exists(compressedPath) then
    begin
        exit(compressedPath);
    end;
    uncompressedPath := TPath.Combine(pathValue, nextBaseName);
    if TFile.Exists(uncompressedPath) then
    begin
        exit(uncompressedPath);
    end;
    // If no next file is found, return an empty string.
    Result := '';
 end;
 class destructor TDataSeries<T>.CreateClass;
 begin
    FCachedFiles := TObjectDictionary<String, TCachedFile>.Create([doOwnsValues]);
 end;
 class destructor TDataSeries<T>.DestroyClass;
 begin
    FCachedFiles.Free;
 end;
 class procedure TDataSeries<T>.ClearCache;
 begin
    FCachedFiles.Clear;
 end;
 // Asynchronously reads tick data from a single specified file.
 // Returns a TFuture that will provide an array of TAskBidSeries.
 class function TDataSeries<T>.LoadDataFile(const FileName: string): TFuture<TArray<TDataPoint>>;
 var
    parsedPath, parsedSymbol: string;
    parsedYear, parsedMonth: Integer;
 begin
    Result := TFuture<TArray<TDataPoint>>.Null;
    // Attempt to parse year and month from the filename
    if not TryParseFileName(FileName, parsedPath, parsedSymbol, parsedYear, parsedMonth) then
        exit;
    if Assigned(IsMemoryLow) then
    begin
        if IsMemoryLow then
        begin
            var fL := FCachedFiles.Values.ToArray;
            TArray.Sort<TCachedFile>(
                fL,
                TComparer<TCachedFile>
                    .Construct(function(const Left, Right: TCachedFile): Integer begin Result := Sign(Left.LastUsed - Right.LastUsed); end)
            );
            for var i := 0 to High(fL) do
            begin
                if not IsMemoryLow then
                    break;
                if fL[i].Name <> FileName then
                    FCachedFiles.Remove(fL[i].Name);
            end;
        end;
        var cachedFile: TCachedFile;
        if FCachedFiles.TryGetValue(Filename, cachedFile) then
        begin
            var age: TDateTime;
            if FileAge(Filename, age, true) then
            begin
                if age = cachedFile.Age then
                begin
                    cachedFile.LastUsed := Now;
                    exit(cachedFile.Data);
                end
                else
                    FCachedFiles.Remove(FileName);
            end;
        end;
    end;
    if FileName.EndsWith('_zip', True) then
    begin
        var capFileName := FileName;
        Result :=
            TFuture<TBytes>
                .Construct(
                    TLatch.Enqueue(FLoadGate),
                    function: TBytes
                    begin
                        Result := TFile.ReadAllBytes(capFileName); // Read all bytes of the .zip file
                    end)
                .Chain<TArray<TDataPoint>>(
                    function(bytes: TBytes): TArray<TDataPoint>
                    begin
                        var zipMemoryStream := TBytesStream.Create(bytes); // Create a memory stream to hold the zip file content
                        try
                            zipMemoryStream.Position := 0;
                            Result := ReadCompressedData(zipMemoryStream);
                        finally
                            zipMemoryStream.Free;
                        end;
                    end);
    end
    else if TFile.Exists(FileName) then
    begin
        var capFileName := FileName;
        Result :=
            TFuture<TArray<TDataPoint>>.Construct(
                TLatch.Enqueue(FLoadGate),
                function: TArray<TDataPoint>
                begin
                    if TFile.Exists(capFileName) then
                    begin
                        var stream := TFileStream.Create(capFileName, fmOpenRead or fmShareDenyWrite);
                        try
                            try
                                Result := ReadUncompressedData(stream);
                            except
                                on E: EReadError do
                                begin
                                    raise EReadError.CreateFmt('File %s: %s', [capFileName, E.Message]);
                                end;
                            end;
                        finally
                            stream.Free;
                        end;
                    end;
                end
            );
    end;
    if Assigned(IsMemoryLow) then
    begin
        var cachedFile := TCachedFile.Create;
        cachedFile.Name := Filename;
        if FileAge(Filename, cachedFile.Age, true) then
        begin
            cachedFile.Data := Result;
            cachedFile.LastUsed := Now;
            FCachedFiles.Add(Filename, cachedFile);
        end;
    end;
 end;
 class function TDataSeries<T>.LoadDataSeries(const FileName: string): TFuture<TArray<TDataPoint>>;
 var
    loadedState: TState;
    loadedFiles: TArray<TFuture<TArray<TDataPoint>>>;
    liveData: TFuture<TArray<TDataPoint>>;
    tabFiles: TList<string>;
    currentFile: string;
    liveFilePath: string;
    pathName, symbolName: string;
    yearValue, monthValue: Integer;
    LoadGate: TLatch;
 begin
    // 1. Discover all sequential .tab/.tab_zip files using the new helper function.
    tabFiles := TList<string>.Create;
    try
        currentFile := FileName;
        if TFile.Exists(currentFile) then
        begin
            // Start with the given file and find all subsequent files in the series.
            while currentFile <> '' do
            begin
                tabFiles.Add(currentFile);
                currentFile := FindNextDataFile(currentFile);
            end;
        end;
        // 2. Determine the corresponding .tab-live file based on the last found .tab file.
        liveFilePath := '';
        if tabFiles.Count > 0 then
        begin
            var lastTabFile := tabFiles.Last;
            if TryParseFileName(lastTabFile, pathName, symbolName, yearValue, monthValue) then
            begin
                var liveFileBaseName := Format('%s_%d_%02d.tab-live', [symbolName, yearValue, monthValue]);
                var potentialLivePath := TPath.Combine(pathName, liveFileBaseName);
                if TFile.Exists(potentialLivePath) then
                begin
                    liveFilePath := potentialLivePath;
                end;
            end;
        end;
        // 3. Asynchronously load all discovered files.
        var loadedFileList := TList<TFuture<TArray<TDataPoint>>>.Create;
        var loadStates := TList<TState>.Create;
        try
            // Create load tasks for the historical .tab files.
            for currentFile in tabFiles do
            begin
                var data := LoadDataFile(currentFile);
                loadedFileList.Add(data);
                loadStates.Add(data.Done);
            end;
            // Create a load task for the .tab-live file, if it exists.
            liveData := TFuture<TArray<TDataPoint>>.Null;
            if liveFilePath <> '' then
            begin
                liveData := LoadDataFile(liveFilePath);
                loadedFileList.Add(liveData);
                loadStates.Add(liveData.Done);
            end;
            loadedState := TState.All(loadStates.ToArray);
            loadedFiles := loadedFileList.ToArray;
        finally
            loadStates.Free;
            loadedFileList.Free;
        end;
    finally
        tabFiles.Free;
    end;
    // 4. Construct the final future that merges the results. This logic remains unchanged.
    Result :=
        TFuture<TArray<TDataPoint>>.Construct(
            loadedState,
            function: TArray<TDataPoint>
            begin
                var tickList := TList<TDataPoint>.Create;
                try
                    var totalTicks := Length(liveData.Value);
                    var overallLastTabTickTime: TDateTime := 0;
                    for var P in loadedFiles do
                        Inc(totalTicks, Length(P.Value));
                    tickList.Capacity := totalTicks;
                    for var P in loadedFiles do
                    begin
                        for var iterTickData in P.Value do
                            if overallLastTabTickTime < iterTickData.TimeStamp then
                            begin
                                tickList.Add(iterTickData);
                                overallLastTabTickTime := iterTickData.TimeStamp;
                            end;
                    end;
                    Result := tickList.ToArray;
                finally
                    tickList.Free;
                end;
            end
        );
 end;
 class function TDataSeries<T>.ReadCompressedData(const InputStream: TStream): TArray<TDataPoint>;
 var
    decompressionStream: TStream; // Holds uncompressed data of a single zip entry
    localHeader: TZipHeader;
    entryIndex: Integer;
    i: Integer;
    zipFileInstance: TZipFile;
 begin
    // Initialize Result record
    SetLength(Result, 0);
    decompressionStream := nil;
    zipFileInstance := nil;
    try
        InputStream.Position := 0; // Reset stream position to the beginning for TZipFile
        // Open TZipFile from this memory stream
        zipFileInstance := TZipFile.Create;
        zipFileInstance.Open(InputStream, TZipMode.zmRead); // Open the zip from the memory stream
        if zipFileInstance.FileCount = 0 then
        begin
            // If TZipFile opens an empty or invalid stream successfully but finds no files
            exit; // FileSuccessfullyLoaded remains False
        end;
        entryIndex := -1;
        for i := 0 to zipFileInstance.FileCount - 1 do
        begin
            if zipFileInstance.FileNames[i].EndsWith('.tab', True) then
            begin
                entryIndex := i;
                break;
            end;
        end;
        if entryIndex = -1 then // If specific name not found, default to the first entry
        begin
            if zipFileInstance.FileCount > 0 then
                entryIndex := 0
            else
                exit; // Should not happen if FileCount > 0 check passed
        end;
        // Decompress the specific entry from the in-memory zip file.
        // TZipFile.Read creates and populates decompressionStream with uncompressed data.
        zipFileInstance.Read(entryIndex, decompressionStream, localHeader);
        if not Assigned(decompressionStream) then
            exit; // Exit if no valid stream could be prepared
        Result := ReadUncompressedData(decompressionStream);
    finally
        // Ensure all potentially created objects are freed
        if Assigned(decompressionStream) then // Holds uncompressed data from a zip entry
            decompressionStream.Free;
        if Assigned(zipFileInstance) then
            zipFileInstance.Free;
    end;
 end;
 class function TDataSeries<T>.ReadUncompressedData(const InputStream: TStream): TArray<TDataPoint>;
 var
    fileSize: Int64;
    recordCount: Integer;
    bytesRead: Integer;
 begin
    // Initialize Result record
    SetLength(Result, 0);
    InputStream.Position := 0;
    fileSize := InputStream.Size;
    if fileSize = 0 then
        exit;
    if (fileSize mod SizeOf(TDataPoint)) <> 0 then
        raise EReadError.CreateFmt('File corrupted. Size %d is not a multiple of record size %d.', [fileSize, SizeOf(TDataPoint)]);
    recordCount := fileSize div SizeOf(TDataPoint);
    if recordCount > 0 then
    begin
        SetLength(Result, recordCount);
        bytesRead := InputStream.Read(Result[0], fileSize);
        if bytesRead <> fileSize then
            raise EReadError.CreateFmt('Read error. Expected to read %d bytes, but actually read %d bytes.', [fileSize, bytesRead]);
    end;
 end;
 { TAuraFileServer<T> }
 constructor TAuraFileServer<T>.Create(const AFilename: String);
 begin
    inherited Create;
    FCurrentFileName := AFilename;
    FIsLiveData := TMutable<Boolean>.CreateWriteable(false);
 end;
 procedure TAuraFileServer<T>.AfterConstruction;
 begin
    inherited;
    // Asynchronously loads the initial data file.
    FCurrentData := TDataSeries<T>.LoadDataFile(FCurrentFileName);
    FCurrPosInFile := 0;
    FCurrentIdx := 0;
    // Initially sets the server to "not live data".
    FIsLiveData.SetValue(false);
    FLastTimeStamp := 0;
    // Determines and asynchronously loads the next chronological file, if available.
    FNextFileName := FindNextDataFile(FCurrentFileName);
    if FNextFileName <> '' then
        FNextData := TDataSeries<T>.LoadDataFile(FNextFileName);
 end;
 function TAuraFileServer<T>.GetChunk(var Data: array of TDataPoint<T>): Integer;
 begin
    Result := 0;
    // Checks if the current data file is ready.
    if FCurrentData.Done.IsSet then
    begin
        var maxLen := Length(Data);
        var currData := FCurrentData.Value;
        while Result < MaxLen do
        begin
            if FCurrPosInFile >= Length(currData) then
            begin
                // Transition to the next file if the current one is exhausted.
                FCurrentData := FNextData;
                FCurrentFileName := FNextFileName;
                FCurrPosInFile := 0;
                FNextFileName := '';
                FNextData := TFuture<TArray<TDataSeries<T>.TDataPoint>>.Null;
                if FCurrentFileName <> '' then
                begin
                    // Loads the next file in the series.
                    FNextFileName := FindNextDataFile(FCurrentFileName);
                    if FNextFileName <> '' then
                        FNextData := TDataSeries<T>.LoadDataFile(FNextFileName);
                    // If the newly assigned current data is already loaded, continue processing.
                    if FCurrentData.Done.IsSet then
                    begin
                        currData := FCurrentData.Value;
                        continue;
                    end;
                end
                else
                begin
                    // All historical files processed; transition to "live data" mode.
                    FIsLiveData.SetValue(true);
                end;
                break; // Exit loop if no more data or transition occurred.
            end;
            // Processes data point if its timestamp is strictly greater than the last processed one.
            // This ensures chronological order and handles potential duplicate timestamps.
            var item := currData[FCurrPosInFile];
            if FLastTimeStamp < item.TimeStamp then
            begin
                FLastTimeStamp := item.TimeStamp;
                Data[Result].Create(FCurrentIdx, FLastTimeStamp, item.Data);
                inc(Result);
                inc(FCurrentIdx);
            end;
            inc(FCurrPosInFile);
        end;
    end;
 end;
 function TAuraFileServer<T>.GetIsLiveData: TMutable<Boolean>;
 begin
    Result := FIsLiveData;
 end;
 end.
@@ -1,52 +0,0 @@
 unit Myc.Trade.DataServer;
 interface
 uses
    System.SysUtils,
    Myc.Lazy,
    Myc.Signals,
    Myc.Futures;
 type
    // Represents a generic data server capable of providing sequential data chunks.
    // The type parameter T specifies the type of data points served.
    IDataServer<T> = interface
        ['{A6E246A2-E84E-49AB-A63E-333E561E488C}']
        // Provides a TMutable<Boolean> that indicates if the server is currently
        // serving live data (true) or historical data (false).
        function GetIsLiveData: TMutable<Boolean>;
        // Retrieves a chunk of data into the provided dynamic array.
        // Returns the number of items successfully read into the array.
        function GetChunk(var Data: array of T): Integer;
        property IsLiveData: TMutable<Boolean> read GetIsLiveData;
    end;
    // Abstract base class for IDataServer implementations.
    TDataServer<T> = class(TInterfacedObject, IDataServer<T>)
    protected
        function GetIsLiveData: TMutable<Boolean>; virtual; abstract;
        function GetChunk(var Data: array of T): Integer; virtual; abstract;
    end;
    // Represents a factory for creating IDataServer instances.
    // The type parameter T specifies the type of data points the created server will provide.
    IDataServerNode<T> = interface
        ['{DA3531F1-158E-4A63-8E5A-34089C537B1B}']
        // Creates and returns a new instance of an IDataServer.
        function CreateDataServer: IDataServer<T>;
    end;
    // Abstract base class for IDataServerNode implementations.
    TDataServerNode<T> = class(TInterfacedObject, IDataServerNode<T>)
    public
        // Factory method to create the actual data server instance.
        function CreateDataServer: IDataServer<T>; virtual; abstract;
    end;
 implementation
 uses
    System.IOUtils;
 end.
@@ -0,0 +1,654 @@
 unit Myc.Trade.DataStream;
 (*
  Myc.Trade.DataStream
  Provides a data server for loading time-series data and streaming it.
  This unit contains the core components for handling historical data:
  - TDataRecord: A generic record for a single time-stamped data entry.
  - IDataServer/TDataServer: A server component responsible for loading and caching
    series of data files from disk. Each server instance manages its own cache,
    but all instances share a central load gate.
  - IDataStream/TDataStream: An interface representing a stream of data points that
    can be consumed sequentially.
  - TAuraFileStream: A concrete implementation of IDataStream that is fed by an
    IDataServer instance.
 *)
 interface
 uses
    System.SysUtils,
    System.Classes,
    System.Generics.Collections,
    System.Generics.Defaults,
    System.IOUtils,
    Myc.Futures,
    Myc.Signals,
    Myc.Lazy,
    Myc.Trade.DataPoint;
 type
    // A generic record for a single time-stamped data entry.
    TDataRecord<T: record> = packed record
        TimeStamp: TDateTime;
        Data: T;
    end;
    // Represents a generic data stream capable of providing sequential data chunks.
    IDataStream<T> = interface
        ['{A6E246A2-E84E-49AB-A63E-333E561E488C}']
        function GetIsLiveData: TMutable<Boolean>;
        function GetChunk(var Data: array of T): Integer;
        property IsLiveData: TMutable<Boolean> read GetIsLiveData;
    end;
    // Abstract base class for IDataStream implementations.
    TDataStream<T> = class(TInterfacedObject, IDataStream<T>)
    protected
        function GetIsLiveData: TMutable<Boolean>; virtual; abstract;
        function GetChunk(var Data: array of T): Integer; virtual; abstract;
    end;
    // Represents a factory for creating IDataStream instances.
    IDataStreamNode<T> = interface
        ['{DA3531F1-158E-4A63-8E5A-34089C537B1B}']
        function CreateDataStream: IDataStream<T>;
    end;
    // Abstract base class for IDataStreamNode implementations.
    TDataStreamNode<T> = class(TInterfacedObject, IDataStreamNode<T>)
    public
        function CreateDataStream: IDataStream<T>; virtual; abstract;
    end;
    // Interface for an instantiable data server.
    IDataServer<T: record> = interface
        ['{1F8E5A9D-E92A-44C1-9F3F-C4B82A6E94B3}']
        function LoadDataFile(const FileName: string): TFuture<TArray<TDataRecord<T>>>;
        function LoadDataSeries(const FileName: string): TFuture<TArray<TDataRecord<T>>>;
        procedure ClearCache;
    end;
    // Generic server for loading and managing sequential time-series data from files.
    TDataServer<T: record> = class(TInterfacedObject, IDataServer<T>)
    strict private
        type
            TCachedFile = class
                Name: String;
                Age: TDateTime;
                LastUsed: TDateTime;
                Data: TFuture<TArray<TDataRecord<T>>>;
            end;
    private
        // The cache is per-instance.
        FCachedFiles: TDictionary<String, TCachedFile>;
    strict private
        // The load gate is shared across all instances of TDataServer.
        class var FLoadGate: TLatch;
    protected
        class function ReadCompressedData(const InputStream: TStream): TArray<TDataRecord<T>>; static;
        class function ReadUncompressedData(const InputStream: TStream): TArray<TDataRecord<T>>; static;
    public
        constructor Create;
        destructor Destroy; override;
        // IDataServer<T> implementation
        procedure ClearCache;
        function LoadDataFile(const FileName: string): TFuture<TArray<TDataRecord<T>>>;
        function LoadDataSeries(const FileName: string): TFuture<TArray<TDataRecord<T>>>;
    end;
    // A specialized TDataServer for handling Ask/Bid tick data.
    TAskBidServer = class(TDataServer<TAskBidItem>)
    end;
    // Implements a data stream that reads data from Aura-specific historical data files.
    TAuraFileStream<T: record> = class(TDataStream<TDataPoint<T>>, IDataStream<TDataPoint<T>>)
    private
        FDataServer: IDataServer<T>;
        FIsLiveData: TMutable<Boolean>.IWriteable;
        FCurrentFileName: string;
        FCurrentData: TFuture<TArray<TDataRecord<T>>>;
        FNextFileName: string;
        FNextData: TFuture<TArray<TDataRecord<T>>>;
        FCurrPosInFile: Int64;
        FCurrentIdx: Int64;
        FLastTimeStamp: TDateTime;
    protected
        function GetChunk(var Data: array of TDataPoint<T>): Integer; override;
        function GetIsLiveData: TMutable<Boolean>; override;
    public
        constructor Create(const AFilename: String; const ADataServer: IDataServer<T>);
        procedure AfterConstruction; override;
    end;
    // Specialized TAuraFileStream for TAskBidItem data.
    TAuraTABFileStream = TAuraFileStream<TAskBidItem>;
 function TryParseFileName(const FileName: string; out PathValue, SymbolValue: string; out YearValue, MonthValue: Integer): Boolean;
 function FindOldestFilesPerSymbol(const DirectoryPath: string): TArray<String>;
 function FindNextDataFile(const FileName: string): string;
 var
    IsMemoryLow: TFunc<Boolean>;
 implementation
 uses
    System.Zip,
    System.Math,
    Myc.TaskManager;
 constructor TDataServer<T>.Create;
 begin
    inherited Create;
    // Initialize the per-instance cache.
    FCachedFiles := TObjectDictionary<String, TCachedFile>.Create([doOwnsValues]);
 end;
 destructor TDataServer<T>.Destroy;
 begin
    FCachedFiles.Free;
    inherited Destroy;
 end;
 procedure TDataServer<T>.ClearCache;
 begin
    FCachedFiles.Clear;
 end;
 function TDataServer<T>.LoadDataFile(const FileName: string): TFuture<TArray<TDataRecord<T>>>;
 var
    parsedPath, parsedSymbol: string;
    parsedYear, parsedMonth: Integer;
 begin
    Result := TFuture<TArray<TDataRecord<T>>>.Null;
    if not TryParseFileName(FileName, parsedPath, parsedSymbol, parsedYear, parsedMonth) then
        exit;
    if Assigned(IsMemoryLow) then
    begin
        if IsMemoryLow() then
        begin
            var fL := FCachedFiles.Values.ToArray;
            TArray.Sort<TCachedFile>(fL,
                TComparer<TCachedFile>.Construct(function(const Left, Right: TCachedFile): Integer
                    begin Result := Sign(Left.LastUsed - Right.LastUsed); end)
            );
            for var i := 0 to High(fL) do
            begin
                if not IsMemoryLow() then
                    break;
                if fL[i].Name <> FileName then
                    FCachedFiles.Remove(fL[i].Name);
            end;
        end;
        var cachedFile: TCachedFile;
        if FCachedFiles.TryGetValue(Filename, cachedFile) then
        begin
            var age: TDateTime;
            if FileAge(Filename, age, true) then
            begin
                if age = cachedFile.Age then
                begin
                    cachedFile.LastUsed := Now;
                    exit(cachedFile.Data);
                end
                else
                    FCachedFiles.Remove(FileName);
            end;
        end;
    end;
    var capFileName := FileName;
    if FileName.EndsWith('_zip', True) then
    begin
        Result := TFuture<TBytes>.Construct(
            TLatch.Enqueue(FLoadGate), // Use shared class var FLoadGate
            function: TBytes
            begin
                Result := TFile.ReadAllBytes(capFileName);
            end
        ).Chain<TArray<TDataRecord<T>>>(
            function(bytes: TBytes): TArray<TDataRecord<T>>
            begin
                var zipMemoryStream := TBytesStream.Create(bytes);
                try
                    zipMemoryStream.Position := 0;
                    Result := ReadCompressedData(zipMemoryStream);
                finally
                    zipMemoryStream.Free;
                end;
            end
        );
    end
    else if TFile.Exists(FileName) then
    begin
        Result := TFuture<TArray<TDataRecord<T>>>.Construct(
            TLatch.Enqueue(FLoadGate), // Use shared class var FLoadGate
            function: TArray<TDataRecord<T>>
            begin
                if TFile.Exists(capFileName) then
                begin
                    var stream := TFileStream.Create(capFileName, fmOpenRead or fmShareDenyWrite);
                    try
                        try
                            Result := ReadUncompressedData(stream);
                        except
                            on E: EReadError do
                                raise EReadError.CreateFmt('File %s: %s', [capFileName, E.Message]);
                        end;
                    finally
                        stream.Free;
                    end;
                end;
            end
        );
    end;
    if Assigned(IsMemoryLow) then
    begin
        var cachedFile := TCachedFile.Create;
        cachedFile.Name := Filename;
        if FileAge(Filename, cachedFile.Age, true) then
        begin
            cachedFile.Data := Result;
            cachedFile.LastUsed := Now;
            FCachedFiles.Add(Filename, cachedFile);
        end;
    end;
 end;
 function TDataServer<T>.LoadDataSeries(const FileName: string): TFuture<TArray<TDataRecord<T>>>;
 var
    loadedState: TState;
    loadedFiles: TArray<TFuture<TArray<TDataRecord<T>>>>;
    liveData: TFuture<TArray<TDataRecord<T>>>;
    tabFiles: TList<string>;
    currentFile: string;
    liveFilePath: string;
    pathName, symbolName: string;
    yearValue, monthValue: Integer;
 begin
    tabFiles := TList<string>.Create;
    try
        currentFile := FileName;
        if TFile.Exists(currentFile) then
            while currentFile <> '' do
            begin
                tabFiles.Add(currentFile);
                currentFile := FindNextDataFile(currentFile);
            end;
        liveFilePath := '';
        if tabFiles.Count > 0 then
        begin
            var lastTabFile := tabFiles.Last;
            if TryParseFileName(lastTabFile, pathName, symbolName, yearValue, monthValue) then
            begin
                var liveFileBaseName := Format('%s_%d_%02d.tab-live', [symbolName, yearValue, monthValue]);
                var potentialLivePath := TPath.Combine(pathName, liveFileBaseName);
                if TFile.Exists(potentialLivePath) then
                    liveFilePath := potentialLivePath;
            end;
        end;
        var loadedFileList := TList<TFuture<TArray<TDataRecord<T>>>>.Create;
        var loadStates := TList<TState>.Create;
        try
            for currentFile in tabFiles do
            begin
                var data := LoadDataFile(currentFile);
                loadedFileList.Add(data);
                loadStates.Add(data.Done);
            end;
            liveData := TFuture<TArray<TDataRecord<T>>>.Null;
            if liveFilePath <> '' then
            begin
                liveData := LoadDataFile(liveFilePath);
                loadedFileList.Add(liveData);
                loadStates.Add(liveData.Done);
            end;
            loadedState := TState.All(loadStates.ToArray);
            loadedFiles := loadedFileList.ToArray;
        finally
            loadStates.Free;
            loadedFileList.Free;
        end;
    finally
        tabFiles.Free;
    end;
    Result := TFuture<TArray<TDataRecord<T>>>.Construct(
        loadedState,
        function: TArray<TDataRecord<T>>
        begin
            var tickList := TList<TDataRecord<T>>.Create;
            try
                var overallLastTabTickTime: TDateTime := 0;
                var cnt := 0;
                for var P in loadedFiles do
                    inc(cnt, Length(P.Value));
                tickList.Capacity := cnt;
                for var P in loadedFiles do
                    for var iterTickData in P.Value do
                        if overallLastTabTickTime < iterTickData.TimeStamp then
                        begin
                            tickList.Add(iterTickData);
                            overallLastTabTickTime := iterTickData.TimeStamp;
                        end;
                Result := tickList.ToArray;
            finally
                tickList.Free;
            end;
        end
    );
 end;
 class function TDataServer<T>.ReadCompressedData(const InputStream: TStream): TArray<TDataRecord<T>>;
 var
    decompressionStream: TStream;
    localHeader: TZipHeader;
    entryIndex, i: Integer;
    zipFileInstance: TZipFile;
 begin
    SetLength(Result, 0);
    decompressionStream := nil;
    zipFileInstance := nil;
    try
        InputStream.Position := 0;
        zipFileInstance := TZipFile.Create;
        zipFileInstance.Open(InputStream, TZipMode.zmRead);
        if zipFileInstance.FileCount = 0 then
            exit;
        entryIndex := -1;
        for i := 0 to zipFileInstance.FileCount - 1 do
            if zipFileInstance.FileNames[i].EndsWith('.tab', True) then
            begin
                entryIndex := i;
                break;
            end;
        if entryIndex = -1 then
            entryIndex := 0;
        zipFileInstance.Read(entryIndex, decompressionStream, localHeader);
        if not Assigned(decompressionStream) then
            exit;
        Result := ReadUncompressedData(decompressionStream);
    finally
        decompressionStream.Free;
        zipFileInstance.Free;
    end;
 end;
 class function TDataServer<T>.ReadUncompressedData(const InputStream: TStream): TArray<TDataRecord<T>>;
 var
    fileSize: Int64;
    recordCount, bytesRead: Integer;
 begin
    SetLength(Result, 0);
    InputStream.Position := 0;
    fileSize := InputStream.Size;
    if (fileSize = 0) or ((fileSize mod SizeOf(TDataRecord<T>)) <> 0) then
        exit;
    recordCount := fileSize div SizeOf(TDataRecord<T>);
    if recordCount > 0 then
    begin
        SetLength(Result, recordCount);
        bytesRead := InputStream.Read(Result[0], fileSize);
        if bytesRead <> fileSize then
            raise EReadError.CreateFmt('Read error. Expected %d bytes, read %d.', [fileSize, bytesRead]);
    end;
 end;
 { TAuraFileStream<T> }
 constructor TAuraFileStream<T>.Create(const AFilename: String; const ADataServer: IDataServer<T>);
 begin
    inherited Create;
    Assert(Assigned(ADataServer));
    FDataServer := ADataServer;
    FCurrentFileName := AFilename;
    FIsLiveData := TMutable<Boolean>.CreateWriteable(false);
 end;
 procedure TAuraFileStream<T>.AfterConstruction;
 begin
    inherited;
    FCurrentData := FDataServer.LoadDataFile(FCurrentFileName);
    FCurrPosInFile := 0;
    FCurrentIdx := 0;
    FIsLiveData.SetValue(false);
    FLastTimeStamp := 0;
    FNextFileName := FindNextDataFile(FCurrentFileName);
    if FNextFileName <> '' then
        FNextData := FDataServer.LoadDataFile(FNextFileName)
    else
        FNextData := nil;
 end;
 function TAuraFileStream<T>.GetChunk(var Data: array of TDataPoint<T>): Integer;
 var
    item: TDataRecord<T>;
    currData: TArray<TDataRecord<T>>;
 begin
    Result := 0;
    if not FCurrentData.Done.IsSet then
        exit;
    var maxLen := Length(Data);
    currData := FCurrentData.Value;
    while Result < maxLen do
    begin
        if FCurrPosInFile >= Length(currData) then
        begin
            FCurrentData := FNextData;
            FCurrentFileName := FNextFileName;
            FCurrPosInFile := 0;
            if FCurrentFileName <> '' then
            begin
                FNextFileName := FindNextDataFile(FCurrentFileName);
                if FNextFileName <> '' then
                    FNextData := FDataServer.LoadDataFile(FNextFileName)
                else
                    FNextData := nil;
                if FCurrentData.Done.IsSet then
                begin
                    currData := FCurrentData.Value;
                    continue;
                end;
            end
            else
            begin
                FIsLiveData.SetValue(true);
            end;
            break;
        end;
        item := currData[FCurrPosInFile];
        if FLastTimeStamp < item.TimeStamp then
        begin
            FLastTimeStamp := item.TimeStamp;
            Data[Result].Create(FCurrentIdx, FLastTimeStamp, item.Data);
            Inc(Result);
            Inc(FCurrentIdx);
        end;
        Inc(FCurrPosInFile);
    end;
 end;
 function TAuraFileStream<T>.GetIsLiveData: TMutable<Boolean>;
 begin
    Result := FIsLiveData;
 end;
 { Helper Functions }
 function TryParseFileName(const FileName: string; out PathValue, SymbolValue: string; out YearValue, MonthValue: Integer): Boolean;
 var
    fileNameNoPath: string;
    nameForParsing: string;
    parts: TArray<string>;
    baseName: string;
 begin
    Result := False;
    PathValue := '';
    SymbolValue := '';
    YearValue := 0;
    MonthValue := 0;
    PathValue := TPath.GetDirectoryName(FileName);
    fileNameNoPath := TPath.GetFileName(FileName);
    nameForParsing := fileNameNoPath;
    if nameForParsing.EndsWith('_zip', True) then
    begin
        baseName := nameForParsing.Substring(0, nameForParsing.Length - 4);
        if TPath.GetExtension(baseName) <> '' then
        begin
            nameForParsing := baseName;
        end;
    end;
    nameForParsing := TPath.GetFileNameWithoutExtension(nameForParsing);
    parts := nameForParsing.Split(['_']);
    if Length(parts) < 3 then
        exit;
    if not TryStrToInt(parts[High(parts)], MonthValue) then
        exit;
    if (MonthValue < 1) or (MonthValue > 12) then
    begin
        MonthValue := 0;
        exit;
    end;
    if not TryStrToInt(parts[High(parts) - 1], YearValue) then
    begin
        MonthValue := 0;
        exit;
    end;
    if YearValue <= 0 then
    begin
        YearValue := 0;
        MonthValue := 0;
        exit;
    end;
    SymbolValue := string.Join('_', Copy(parts, 0, Length(parts) - 2));
    if SymbolValue = '' then
    begin
        PathValue := '';
        YearValue := 0;
        MonthValue := 0;
        exit;
    end;
    Result := True;
 end;
 function FindOldestFilesPerSymbol(const DirectoryPath: string): TArray<String>;
 type
    TTrackedFileInfo = record
        FullFileName: string;
        Year: Integer;
        Month: Integer;
    end;
 var
    oldestFilesPerSymbol: TDictionary<string, TTrackedFileInfo>;
    fileNames: TArray<string>;
    currentFile: string;
    parsedPath: string;
    parsedSymbol: string;
    parsedYear: Integer;
    parsedMonth: Integer;
    trackedInfo: TTrackedFileInfo;
 begin
    oldestFilesPerSymbol := TDictionary<string, TTrackedFileInfo>.Create;
    try
        if not TDirectory.Exists(DirectoryPath) then
            exit(nil);
        fileNames := TDirectory.GetFiles(DirectoryPath);
        for currentFile in fileNames do
        begin
            if TryParseFileName(currentFile, parsedPath, parsedSymbol, parsedYear, parsedMonth) then
            begin
                if oldestFilesPerSymbol.TryGetValue(parsedSymbol, trackedInfo) then
                begin
                    if (parsedYear < trackedInfo.Year) or ((parsedYear = trackedInfo.Year) and (parsedMonth < trackedInfo.Month)) then
                    begin
                        trackedInfo.FullFileName := currentFile;
                        trackedInfo.Year := parsedYear;
                        trackedInfo.Month := parsedMonth;
                        oldestFilesPerSymbol.AddOrSetValue(parsedSymbol, trackedInfo);
                    end;
                end
                else
                begin
                    trackedInfo.FullFileName := currentFile;
                    trackedInfo.Year := parsedYear;
                    trackedInfo.Month := parsedMonth;
                    oldestFilesPerSymbol.Add(parsedSymbol, trackedInfo);
                end;
            end;
        end;
        SetLength(Result, oldestFilesPerSymbol.Count);
        var n := 0;
        for trackedInfo in oldestFilesPerSymbol.Values do
        begin
            Result[n] := trackedInfo.FullFileName;
            inc(n);
        end;
    finally
        oldestFilesPerSymbol.Free;
    end;
 end;
 function FindNextDataFile(const FileName: string): string;
 var
    pathValue, symbolValue: string;
    yearValue, monthValue: Integer;
    nextBaseName, compressedPath, uncompressedPath: string;
 begin
    if not TryParseFileName(FileName, pathValue, symbolValue, yearValue, monthValue) then
        exit('');
    Inc(monthValue);
    if monthValue > 12 then
    begin
        monthValue := 1;
        Inc(yearValue);
    end;
    nextBaseName := Format('%s_%.4d_%.2d.tab', [symbolValue, yearValue, monthValue]);
    compressedPath := TPath.Combine(pathValue, nextBaseName + '_zip');
    if TFile.Exists(compressedPath) then
        exit(compressedPath);
    uncompressedPath := TPath.Combine(pathValue, nextBaseName);
    if TFile.Exists(uncompressedPath) then
        exit(uncompressedPath);
    Result := '';
 end;
 end.
@@ -5,38 +5,37 @@ program MycTests;
 {$ENDIF}
 {$STRONGLINKTYPES ON}
 uses
-    FastMM5,
+  FastMM5,
-    DUnitX.MemoryLeakMonitor.FastMM5,
+  DUnitX.MemoryLeakMonitor.FastMM5,
-    System.SysUtils,
+  System.SysUtils,
-{$IFDEF TESTINSIGHT}
+  {$IFDEF TESTINSIGHT}
-    TestInsight.DUnitX,
+  TestInsight.DUnitX,
-{$ELSE}
+  {$ELSE}
-    DUnitX.Loggers.Console,
+  DUnitX.Loggers.Console,
-{$ENDIF }
+  {$ENDIF }
-    DUnitX.TestFramework,
+  DUnitX.TestFramework,
-    TestNotifier in 'TestNotifier.pas',
+  TestNotifier in 'TestNotifier.pas',
-    TestNotifier_Threading in 'TestNotifier_Threading.pas' {/TestNotifier_ChaosStress in 'TestNotifier_ChaosStress.pas',},
+  TestNotifier_Threading in 'TestNotifier_Threading.pas' {/TestNotifier_ChaosStress in 'TestNotifier_ChaosStress.pas',},
-    TestNotifier_ChaosStress in 'TestNotifier_ChaosStress.pas',
+  TestNotifier_ChaosStress in 'TestNotifier_ChaosStress.pas',
-    Myc.Core.Tasks in '..\Src\Myc.Core.Tasks.pas',
+  Myc.Core.Tasks in '..\Src\Myc.Core.Tasks.pas',
-    TestTasks in 'TestTasks.pas',
+  TestTasks in 'TestTasks.pas',
-    Myc.Futures in '..\Src\Myc.Futures.pas',
+  Myc.Futures in '..\Src\Myc.Futures.pas',
-    TestCoreFutures in 'TestCoreFutures.pas',
+  TestCoreFutures in 'TestCoreFutures.pas',
-    Myc.TaskManager in '..\Src\Myc.TaskManager.pas',
+  Myc.TaskManager in '..\Src\Myc.TaskManager.pas',
-    Myc.Core.Futures in '..\Src\Myc.Core.Futures.pas',
+  Myc.Core.Futures in '..\Src\Myc.Core.Futures.pas',
-    Myc.Core.Signals in '..\Src\Myc.Core.Signals.pas',
+  Myc.Core.Signals in '..\Src\Myc.Core.Signals.pas',
-    TestFutures in 'TestFutures.pas',
+  TestFutures in 'TestFutures.pas',
-    Myc.Lazy in '..\Src\Myc.Lazy.pas',
+  Myc.Lazy in '..\Src\Myc.Lazy.pas',
-    Myc.Core.Lazy in '..\Src\Myc.Core.Lazy.pas',
+  Myc.Core.Lazy in '..\Src\Myc.Core.Lazy.pas',
-    Myc.Test.Core.Lazy in '..\Src\Myc.Test.Core.Lazy.pas',
+  Myc.Test.Core.Lazy in '..\Src\Myc.Test.Core.Lazy.pas',
-    Myc.Test.Lazy in '..\Src\Myc.Test.Lazy.pas',
+  Myc.Test.Lazy in '..\Src\Myc.Test.Lazy.pas',
-    Myc.Test.Core.Atomic in '..\Src\Myc.Test.Core.Atomic.pas',
+  Myc.Test.Core.Atomic in '..\Src\Myc.Test.Core.Atomic.pas',
-    Myc.Trade.Ticker in '..\Src\Myc.Trade.Ticker.pas',
+  Myc.Trade.Ticker in '..\Src\Myc.Trade.Ticker.pas',
-    Myc.Test.Signals.Latch in '..\Src\Myc.Test.Signals.Latch.pas',
+  Myc.Test.Signals.Latch in '..\Src\Myc.Test.Signals.Latch.pas',
-    Myc.Test.Signals.Dirty in '..\Src\Myc.Test.Signals.Dirty.pas',
+  Myc.Test.Signals.Dirty in '..\Src\Myc.Test.Signals.Dirty.pas',
-    Myc.Trade.DataPoint in '..\Src\Myc.Trade.DataPoint.pas',
+  Myc.Trade.DataPoint in '..\Src\Myc.Trade.DataPoint.pas',
-    Myc.Trade.DataServer in '..\Src\Myc.Trade.DataServer.pas',
+  Myc.Trade.Node in '..\Src\Myc.Trade.Node.pas',
-    Myc.Trade.Node in '..\Src\Myc.Trade.Node.pas',
+  Myc.Trade.DataStream in '..\Src\Myc.Trade.DataStream.pas';
    Myc.Test.Trade.DataServer in '..\Src\Myc.Test.Trade.DataServer.pas';
 { keep comment here to protect the following conditional from being removed by the IDE when adding a unit }
 {$IFNDEF TESTINSIGHT}
@@ -4,7 +4,7 @@
        <ProjectVersion>20.3</ProjectVersion>
        <FrameworkType>None</FrameworkType>
        <Base>True</Base>
-        <Config Condition="'$(Config)'==''">Release</Config>
+        <Config Condition="'$(Config)'==''">Debug</Config>
        <Platform Condition="'$(Platform)'==''">Win64</Platform>
        <ProjectName Condition="'$(ProjectName)'==''">MycTests</ProjectName>
        <TargetedPlatforms>3</TargetedPlatforms>
@@ -134,9 +134,8 @@ $(PreBuildEvent)]]></PreBuildEvent>
        <DCCReference Include="..\Src\Myc.Test.Signals.Latch.pas"/>
        <DCCReference Include="..\Src\Myc.Test.Signals.Dirty.pas"/>
        <DCCReference Include="..\Src\Myc.Trade.DataPoint.pas"/>
        <DCCReference Include="..\Src\Myc.Trade.DataServer.pas"/>
        <DCCReference Include="..\Src\Myc.Trade.Node.pas"/>
-        <DCCReference Include="..\Src\Myc.Test.Trade.DataServer.pas"/>
+        <DCCReference Include="..\Src\Myc.Trade.DataStream.pas"/>
        <BuildConfiguration Include="Base">
            <Key>Base</Key>
        </BuildConfiguration>
@@ -192,6 +191,12 @@ $(PreBuildEvent)]]></PreBuildEvent>
                        <Overwrite>true</Overwrite>
                    </Platform>
                </DeployFile>
                <DeployFile LocalName="Win64\Release\MycTests.exe" Configuration="Release" Class="ProjectOutput">
                    <Platform Name="Win64">
                        <RemoteName>MycTests.exe</RemoteName>
                        <Overwrite>true</Overwrite>
                    </Platform>
                </DeployFile>
                <DeployClass Name="AdditionalDebugSymbols">
                    <Platform Name="iOSSimulator">
                        <Operation>1</Operation>