MycLib/Src/Myc.Trade.DataStream.pas

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:
  - TAuraFileDataRecord: A generic record for a single time-stamped data entry.
  - IDataServer/TAuraDataServer: 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

    // 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 GetHasData: TSignal;
        function GetChunk(var Data: array of TDataPoint<T>): Integer;
        function IsHistory: Boolean;
        property HasData: TSignal read GetHasData;
    end;

    // Abstract base class for IDataStream implementations.
    TDataStream<T> = class(TInterfacedObject, IDataStream<T>)
    protected
        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.
    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 CreateStream(const Symbol: String): IDataStream<T>;
        procedure ClearCache;
    end;

    // A generic record for a single time-stamped data entry.
    TAuraFileDataRecord<T: record> = packed record
        TimeStamp: TDateTime;
        Data: T;
    end;

    IAuraDataServer<T: record> = interface(IDataServer<T>)
        ['{481E27DE-DC58-49AF-B8A8-B6316980C423}']
        function LoadDataFile(const FileName: string): TFuture<TArray<TDataPoint<T>>>;
        function LoadDataSeries(const FileName: string): TFuture<TArray<TDataPoint<T>>>;
        function EnumerateAssetFiles: TArray<String>;
    end;

    // Aura files

    // Generic server for loading and managing sequential time-series data from files.
    TAuraDataServer<T: record> = class(TInterfacedObject, IDataServer<T>, IAuraDataServer<T>)
    private
        type
            TCachedFile = class
            public
                Name: String;
                Age: TDateTime;
                LastUsed: TDateTime;
                Data: TFuture<TArray<TDataPoint<T>>>;
            end;
    private
        // The cache is per-instance.
        FCachedFiles: TDictionary<String, TCachedFile>;
        FPath: String;
        function GetPath: String;

    strict private
        // The load gate is shared across all instances of TAuraDataServer.
        class var
            FLoadGate: TLatch;

    protected
        class function ReadCompressedData(const InputStream: TStream): TArray<TDataPoint<T>>; static;
        class function ReadUncompressedData(const InputStream: TStream): TArray<TDataPoint<T>>; static;

    public
        constructor Create(const APath: String);
        destructor Destroy; override;

        class function TryParseFileName(
            const FileName: string;
            out PathValue, SymbolValue: string;
            out YearValue, MonthValue: Integer
        ): Boolean; virtual; abstract;

        class function FindFirstDataFile(const Path, Symbol: string): string;
        class function FindNextDataFile(const FileName: string): string;
        class function FindOldestFilesPerSymbol(const DirectoryPath: string): TArray<String>;

        // IDataServer<T> implementation
        function CreateStream(const Symbol: String): IDataStream<T>; virtual; abstract;
        procedure ClearCache;
        function EnumerateAssetFiles: TArray<String>;

        function LoadDataFile(const FileName: string): TFuture<TArray<TDataPoint<T>>>;
        function LoadDataSeries(const FileName: string): TFuture<TArray<TDataPoint<T>>>;

        property Path: String read GetPath;
    end;

    // Implements a data stream that reads from Aura-specific historical data files.
    TAuraFileStream<T: record> = class(TDataStream<T>)
    private
        FDataServer: TAuraDataServer<T>;
        FHasData: TEvent;
        FCurrentFileName: string;
        FCurrentData: TFuture<TArray<TDataPoint<T>>>;
        FNextFileName: string;
        FNextData: TFuture<TArray<TDataPoint<T>>>;
        FCurrPosInFile: Int64;
        FLastTimeStamp: TDateTime;
    protected
        function GetHasData: TSignal; override;
        function GetChunk(var Data: array of TDataPoint<T>): Integer; override;
        function IsHistory: Boolean; override;
    public
        constructor Create(ADataServer: TAuraDataServer<T>; const AFilename: String);
        destructor Destroy; override;
        procedure AfterConstruction; override;
    end;

    // Aura tick data file Ask-Bid

    TAuraTABFileServer = class(TAuraDataServer<TAskBidItem>)
    public
        // IDataServer<T> implementation
        function CreateStream(const Symbol: String): IDataStream<TAskBidItem>; override;

        class function TryParseFileName(
            const FileName: string;
            out PathValue, SymbolValue: string;
            out YearValue, MonthValue: Integer
        ): Boolean; override;
    end;

var
    IsMemoryLow: TFunc<Boolean>;

implementation

uses
    System.Zip,
    System.Math,
    System.StrUtils,
    Myc.TaskManager;

constructor TAuraDataServer<T>.Create(const APath: String);
begin
    inherited Create;
    // Initialize the per-instance cache.
    FCachedFiles := TObjectDictionary<String, TCachedFile>.Create([doOwnsValues]);
    FPath := APath;
end;

destructor TAuraDataServer<T>.Destroy;
begin
    ClearCache;
    FCachedFiles.Free;
    inherited Destroy;
end;

procedure TAuraDataServer<T>.ClearCache;
begin
    for var cF in FCachedFiles.Values do
        cF.Data.WaitFor;

    FCachedFiles.Clear;
end;

class function TAuraDataServer<T>.FindFirstDataFile(const Path, Symbol: string): string;
var
    oldestFiles: TArray<string>;
    fileName: string;
    pathValue, symbolValue: string;
    yearValue, monthValue: Integer;
begin
    oldestFiles := FindOldestFilesPerSymbol(Path);
    for fileName in oldestFiles do
    begin
        if TryParseFileName(fileName, pathValue, symbolValue, yearValue, monthValue) then
        begin
            if SameText(symbolValue, Symbol) then
                exit(fileName);
        end;
    end;
    Result := '';
end;

class function TAuraDataServer<T>.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;

class function TAuraDataServer<T>.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 TAuraDataServer<T>.EnumerateAssetFiles: TArray<String>;
begin
    Result := FindOldestFilesPerSymbol(FPath);
end;

function TAuraDataServer<T>.GetPath: String;
begin
    Result := FPath;
end;

function TAuraDataServer<T>.LoadDataFile(const FileName: string): TFuture<TArray<TDataPoint<T>>>;
var
    parsedPath, parsedSymbol: string;
    parsedYear, parsedMonth: Integer;
begin
    Result := TFuture<TArray<TDataPoint<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<TDataPoint<T>>>(
                    function(bytes: TBytes): TArray<TDataPoint<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<TDataPoint<T>>>.Construct(
                TLatch.Enqueue(FLoadGate), // Use shared class var FLoadGate
                function: TArray<TDataPoint<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 TAuraDataServer<T>.LoadDataSeries(const FileName: string): TFuture<TArray<TDataPoint<T>>>;
var
    loadedState: TState;
    loadedFiles: TArray<TFuture<TArray<TDataPoint<T>>>>;
    liveData: TFuture<TArray<TDataPoint<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<TDataPoint<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<TDataPoint<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<TDataPoint<T>>>.Construct(
            loadedState,
            function: TArray<TDataPoint<T>>
            begin
                var tickList := TList<TDataPoint<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.Time then
                            begin
                                tickList.Add(iterTickData);
                                overallLastTabTickTime := iterTickData.Time;
                            end;
                    Result := tickList.ToArray;
                finally
                    tickList.Free;
                end;
            end
        );
end;

class function TAuraDataServer<T>.ReadCompressedData(const InputStream: TStream): TArray<TDataPoint<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 TAuraDataServer<T>.ReadUncompressedData(const InputStream: TStream): TArray<TDataPoint<T>>;
var
    fileSize: Int64;
    recordCount, bytesRead: Integer;
    rec: TAuraFileDataRecord<T>;
begin
    SetLength(Result, 0);
    InputStream.Position := 0;
    fileSize := InputStream.Size;
    if (fileSize = 0) or ((fileSize mod SizeOf(TAuraFileDataRecord<T>)) <> 0) then
        exit;

    recordCount := fileSize div SizeOf(TAuraFileDataRecord<T>);
    if recordCount > 0 then
    begin
        SetLength(Result, recordCount);
        for var i := 0 to High(Result) do
        begin
            bytesRead := InputStream.Read(rec, SizeOf(TAuraFileDataRecord<T>));
            if bytesRead <> SizeOf(TAuraFileDataRecord<T>) then
                raise EReadError.CreateFmt('Read error. Expected %d bytes, read %d.', [fileSize, bytesRead]);
            Result[i].Create(rec.TimeStamp, rec.Data);
        end;
    end;
end;

{ TAuraFileStream<T> }

constructor TAuraFileStream<T>.Create(ADataServer: TAuraDataServer<T>; const AFilename: String);
begin
    inherited Create;
    Assert(Assigned(ADataServer));
    FDataServer := ADataServer;
    FCurrentFileName := AFilename;
    // Create an event
    FHasData := TEvent.CreateEvent; // interface helper benutzen!
end;

destructor TAuraFileStream<T>.Destroy;
begin
    inherited;
end;

procedure TAuraFileStream<T>.AfterConstruction;
begin
    inherited;
    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
    begin
        FNextData := FDataServer.LoadDataFile(FNextFileName);
        FNextData.Done.Subscribe(FHasData);
    end;
end;

function TAuraFileStream<T>.GetChunk(var Data: array of TDataPoint<T>): Integer;
var
    item: TDataPoint<T>;
    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;

    if FCurrPosInFile >= Length(FCurrentData.Value) then
    begin
        FCurrentData := FNextData;
        FCurrentFileName := FNextFileName;
        FCurrPosInFile := 0;
        FNextData := TFuture<TArray<TDataPoint<T>>>.Null;

        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
            FLastTimeStamp := item.Time;
            Data[Result].Create(FLastTimeStamp, item.Data);
            Inc(Result);
        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>.GetHasData: TSignal;
begin
    Result := FHasData.Signal;
end;

function TAuraFileStream<T>.IsHistory: Boolean;
begin
    Result := True;
end;

{ TAuraTABFileServer }

function TAuraTABFileServer.CreateStream(const Symbol: String): IDataStream<TAskBidItem>;
var
    fileName: string;
begin
    // FindFirstDataFile uses the server's path and the given symbol to locate the initial data file.
    fileName := FindFirstDataFile(FPath, Symbol);

    // Exit if no file is found or if it is not a .tab file, preventing stream creation for invalid sources.
    if (fileName = '') or (not TPath.GetExtension(fileName).StartsWith('.tab')) then
        exit(nil);

    // Creates a new stream instance, passing itself as the data server and the found filename.
    Result := TAuraFileStream<TAskBidItem>.Create(Self, fileName);
end;

class function TAuraTABFileServer.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;

end.