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.
    IDataStream<T> = interface
        ['{A6E246A2-E84E-49AB-A63E-333E561E488C}']
        function GetIsLiveData: TMutable<Boolean>;
        function GetChunk(var Data: array of TDataPoint<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 TDataPoint<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 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<TAuraFileDataRecord<T>>>;
        function LoadDataSeries(const FileName: string): TFuture<TArray<TAuraFileDataRecord<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<TAuraFileDataRecord<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<TAuraFileDataRecord<T>>; static;
        class function ReadUncompressedData(const InputStream: TStream): TArray<TAuraFileDataRecord<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<TAuraFileDataRecord<T>>>;
        function LoadDataSeries(const FileName: string): TFuture<TArray<TAuraFileDataRecord<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>, IDataStream<T>)
    private
        FDataServer: TAuraDataServer<T>;
        FIsLiveData: TMutable<Boolean>.IWriteable;
        FCurrentFileName: string;
        FCurrentData: TFuture<TArray<TAuraFileDataRecord<T>>>;
        FNextFileName: string;
        FNextData: TFuture<TArray<TAuraFileDataRecord<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(ADataServer: TAuraDataServer<T>; const AFilename: String);
        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<TAuraFileDataRecord<T>>>;
var
    parsedPath, parsedSymbol: string;
    parsedYear, parsedMonth: Integer;
begin
    Result := TFuture<TArray<TAuraFileDataRecord<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<TAuraFileDataRecord<T>>>(
                    function(bytes: TBytes): TArray<TAuraFileDataRecord<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<TAuraFileDataRecord<T>>>.Construct(
                TLatch.Enqueue(FLoadGate), // Use shared class var FLoadGate
                function: TArray<TAuraFileDataRecord<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<TAuraFileDataRecord<T>>>;
var
    loadedState: TState;
    loadedFiles: TArray<TFuture<TArray<TAuraFileDataRecord<T>>>>;
    liveData: TFuture<TArray<TAuraFileDataRecord<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<TAuraFileDataRecord<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<TAuraFileDataRecord<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<TAuraFileDataRecord<T>>>.Construct(
            loadedState,
            function: TArray<TAuraFileDataRecord<T>>
            begin
                var tickList := TList<TAuraFileDataRecord<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 TAuraDataServer<T>.ReadCompressedData(const InputStream: TStream): TArray<TAuraFileDataRecord<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<TAuraFileDataRecord<T>>;
var
    fileSize: Int64;
    recordCount, bytesRead: Integer;
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);
        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(ADataServer: TAuraDataServer<T>; const AFilename: String);
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 := FDataServer.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: TAuraFileDataRecord<T>;
    currData: TArray<TAuraFileDataRecord<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 := FDataServer.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;

{ 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.