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,
    Myc.Core.FileCache;

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 GetSymbol: String;
        function IsHistory: Boolean;
        property HasData: TSignal read GetHasData;
        property Symbol: String read GetSymbol;
    end;

    // Abstract base class for IDataStream implementations.
    TDataStream<T> = class(TInterfacedObject, IDataStream<T>)
    protected
        function GetSymbol: String; virtual; abstract;
        function GetHasData: TSignal; virtual; abstract;
        function GetChunk(var Data: array of TDataPoint<T>): Integer; virtual; abstract;
        function IsHistory: Boolean; virtual; abstract;
    end;

    // Represents a factory for creating IDataStream instances.
    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 EnumerateAssets: 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>)
    public type
        TDataFile = record
            Extension: String;
            Path, Symbol: String;
            Year, Month: Integer;
            function GetBaseFileName: string;
            function GetFullFileName: string;
            function GetIsValid: Boolean;
            property IsValid: Boolean read GetIsValid;
        end;

    private
        FCachedFiles: TDataFileCache<TArray<TDataPoint<T>>>;
        FPath: String;
        function DoLoad(const FileName: string): TFuture<TArray<TDataPoint<T>>>;
        function GetPath: String;

    strict private
        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 DataFile: TDataFile): Boolean; virtual; abstract;

        function FindFirstDataFile(const Symbol: string): TDataFile;
        class function FindNextDataFile(const CurrentFile: TDataFile): TDataFile;
        class function FindOldestFilesPerSymbol(const DirectoryPath: string): TArray<TDataFile>;

        function CreateStream(const Symbol: String): IDataStream<T>; virtual; abstract;
        procedure ClearCache;
        function EnumerateAssets: TArray<String>;
        function LoadDataFile(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>, IDataStream<T>)
    private
        type
            // Changed: Renamed "File" to "FileInfo" for clarity.
            TDataStream = record
                FileInfo: TAuraDataServer<T>.TDataFile;
                Data: TFuture<TArray<TDataPoint<T>>>;
            end;
    private
        FDataServer: TAuraDataServer<T>;
        FSymbol: String;
        FHasData: TEvent;
        FCurrent: TFuture<TDataStream>;
        FNext: TFuture<TDataStream>;
        FCurrPosInFile: Int64;
        FLastTimeStamp: TDateTime;
        procedure PreloadNext;
    protected
        function GetSymbol: String; override;
        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 ASymbol: String);
        destructor Destroy; override;
        procedure AfterConstruction; override;
        procedure BeforeDestruction; override;
    end;

    // Aura tick data file Ask-Bid

    TAuraTABFileServer = class(TAuraDataServer<TAskBidItem>)
    public
        function CreateStream(const Symbol: String): IDataStream<TAskBidItem>; override;
        function LoadDataSeries(const FileName: string): TFuture<TArray<TDataPoint<TAskBidItem>>>;
        class function TryParseFileName(const FileName: string; out DataFile: TAuraDataServer<TAskBidItem>.TDataFile): Boolean; override;
    end;

implementation

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

{ TAuraDataServer<T>.TDataFile }

function TAuraDataServer<T>.TDataFile.GetBaseFileName: string;
begin
    Result := Format('%s_%.4d_%.2d', [Symbol, Year, Month]);
end;

function TAuraDataServer<T>.TDataFile.GetFullFileName: string;
begin
    Result := TPath.Combine(Path, GetBaseFileName + Extension);
end;

function TAuraDataServer<T>.TDataFile.GetIsValid: Boolean;
begin
    Result := (Symbol <> '') and (Year > 0);
end;

{ TAuraDataServer<T> }

constructor TAuraDataServer<T>.Create(const APath: String);
begin
    inherited Create;
    FPath := APath;

    FCachedFiles :=
        TDataFileCache<TArray<TDataPoint<T>>>
            .Create(function(const Filename: String): TFuture<TArray<TDataPoint<T>>> begin Result := DoLoad(Filename); end);
end;

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

procedure TAuraDataServer<T>.ClearCache;
begin
    FCachedFiles.Clear;
end;

function TAuraDataServer<T>.FindFirstDataFile(const Symbol: string): TDataFile;
var
    oldestFiles: TArray<TDataFile>;
    fileInfo: TDataFile;
begin
    oldestFiles := FindOldestFilesPerSymbol(FPath);
    for fileInfo in oldestFiles do
    begin
        if SameText(fileInfo.Symbol, Symbol) then
            exit(fileInfo);
    end;
    Result := Default(TDataFile);
end;

class function TAuraDataServer<T>.FindNextDataFile(const CurrentFile: TDataFile): TDataFile;
var
    dataFileToProbe: TDataFile;
begin
    Result := Default(TDataFile);

    if not CurrentFile.IsValid then
        exit;

    // Changed: Simplified logic, no reparsing needed.
    dataFileToProbe := CurrentFile;
    Inc(dataFileToProbe.Month);
    if (dataFileToProbe.Month > 12) then
    begin
        dataFileToProbe.Month := 1;
        Inc(dataFileToProbe.Year);
    end;

    dataFileToProbe.Extension := '.tab_zip';
    if TFile.Exists(dataFileToProbe.GetFullFileName) then
        exit(dataFileToProbe);

    dataFileToProbe.Extension := '.tab';
    if TFile.Exists(dataFileToProbe.GetFullFileName) then
        exit(dataFileToProbe);
end;

class function TAuraDataServer<T>.FindOldestFilesPerSymbol(const DirectoryPath: string): TArray<TDataFile>;
var
    oldestFilesPerSymbol: TDictionary<string, TDataFile>;
    fileNames: TArray<string>;
    currentFile: string;
    dataFile: TDataFile;
    trackedInfo: TDataFile;
begin
    oldestFilesPerSymbol := TDictionary<string, TDataFile>.Create;
    try
        if not TDirectory.Exists(DirectoryPath) then
            exit(nil);

        fileNames := TDirectory.GetFiles(DirectoryPath);
        for currentFile in fileNames do
        begin
            if TryParseFileName(currentFile, dataFile) then
            begin
                if oldestFilesPerSymbol.TryGetValue(dataFile.Symbol, trackedInfo) then
                begin
                    if (dataFile.Year < trackedInfo.Year) or ((dataFile.Year = trackedInfo.Year) and (dataFile.Month < trackedInfo.Month)) then
                    begin
                        oldestFilesPerSymbol.AddOrSetValue(dataFile.Symbol, dataFile);
                    end;
                end
                else
                begin
                    oldestFilesPerSymbol.Add(dataFile.Symbol, dataFile);
                end;
            end;
        end;

        Result := oldestFilesPerSymbol.Values.ToArray;
    finally
        oldestFilesPerSymbol.Free;
    end;
end;

function TAuraDataServer<T>.EnumerateAssets: TArray<String>;
var
    dataFiles: TArray<TDataFile>;
    i: Integer;
begin
    dataFiles := FindOldestFilesPerSymbol(FPath);
    SetLength(Result, Length(dataFiles));
    for i := 0 to High(dataFiles) do
        Result[i] := dataFiles[i].Symbol;
end;

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

function TAuraDataServer<T>.DoLoad(const FileName: string): TFuture<TArray<TDataPoint<T>>>;
begin
    Result := TFuture<TArray<TDataPoint<T>>>.Null;

    if not TFile.Exists(FileName) then
        exit;

    var capFileName := FileName;
    if FileName.EndsWith('_zip', True) then
    begin
        Result :=
            TFuture<TBytes>
                .Construct(
                    TLatch.Enqueue(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
    begin
        Result :=
            TFuture<TArray<TDataPoint<T>>>.Construct(
                TLatch.Enqueue(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;
end;

function TAuraDataServer<T>.LoadDataFile(const FileName: string): TFuture<TArray<TDataPoint<T>>>;
begin
    Result := FCachedFiles.GetOrAdd(Filename);
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 ASymbol: String);
begin
    inherited Create;
    Assert(Assigned(ADataServer));
    FDataServer := ADataServer;
    FSymbol := ASymbol;
    FHasData := TEvent.CreateEvent;
end;

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

procedure TAuraFileStream<T>.AfterConstruction;
var
    firstFile: TFuture<TAuraDataServer<T>.TDataFile>;
begin
    inherited;

    firstFile := TFuture<TAuraDataServer<T>.TDataFile>.Construct(function: TAuraDataServer<T>.TDataFile begin Result := FDataServer.FindFirstDataFile(FSymbol); end);

    FCurrent :=
        firstFile.Chain<TDataStream>(
            function(const FileInfo: TAuraDataServer<T>.TDataFile): TDataStream
            begin
                if FileInfo.IsValid then
                begin
                    // Changed: Use renamed field "FileInfo".
                    Result.FileInfo := FileInfo;
                    Result.Data := FDataServer.LoadDataFile(FileInfo.GetFullFileName);
                    Result.Data.Done.Subscribe(FHasData);
                end;
            end
        );

    FLastTimeStamp := 0;
    FCurrPosInFile := 0;

    PreloadNext;
end;

procedure TAuraFileStream<T>.BeforeDestruction;
begin
    FCurrent.WaitFor;
    FCurrent.Value.Data.WaitFor;
    FNext.WaitFor;
    FNext.Value.Data.WaitFor;
    inherited;
end;

function TAuraFileStream<T>.GetChunk(var Data: array of TDataPoint<T>): Integer;
var
    item: TDataPoint<T>;
label
    loop;
begin
    Result := 0;

    loop:
    if not FCurrent.Done.IsSet then
        exit;
    if not FCurrent.Value.Data.Done.IsSet then
        exit;

    // Changed: Use renamed field "FileInfo".
    if not FCurrent.Value.FileInfo.IsValid then
        exit;

    if FCurrPosInFile >= Length(FCurrent.Value.Data.Value) then
    begin
        FCurrPosInFile := 0;
        FCurrent := FNext;
        PreloadNext;
        goto loop;
    end;

    var currData := FCurrent.Value.Data.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 FCurrPosInFile < Length(currData) then
    begin
        FHasData.Notify;
    end;
end;

function TAuraFileStream<T>.GetHasData: TSignal;
begin
    Result := FHasData.Signal;
end;

function TAuraFileStream<T>.GetSymbol: String;
begin
    Result := FSymbol;
end;

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

procedure TAuraFileStream<T>.PreloadNext;
begin
    FNext :=
        FCurrent.Chain<TDataStream>(
            function(const Prev: TDataStream): TDataStream
            var
                nextFileInfo: TAuraDataServer<T>.TDataFile;
            begin
                // Changed: Use renamed field "FileInfo".
                if Prev.FileInfo.IsValid then
                begin
                    nextFileInfo := TAuraDataServer<T>.FindNextDataFile(Prev.FileInfo);
                    if nextFileInfo.IsValid then
                    begin
                        Result.FileInfo := nextFileInfo;
                        Result.Data := FDataServer.LoadDataFile(nextFileInfo.GetFullFileName);
                        Result.Data.Done.Subscribe(FHasData);
                    end;
                end;
            end
        );
end;

{ TAuraTABFileServer }

function TAuraTABFileServer.CreateStream(const Symbol: String): IDataStream<TAskBidItem>;
begin
    Result := TAuraFileStream<TAskBidItem>.Create(Self, Symbol);
end;

function TAuraTABFileServer.LoadDataSeries(const FileName: string): TFuture<TArray<TDataPoint<TAskBidItem>>>;
var
    loadedState: TState;
    loadedFiles: TArray<TFuture<TArray<TDataPoint<TAskBidItem>>>>;
    liveData: TFuture<TArray<TDataPoint<TAskBidItem>>>;
    tabFiles: TList<string>;
    liveFilePath: string;
    currentFileInfo: TDataFile;
begin
    tabFiles := TList<string>.Create;
    try
        if TryParseFileName(FileName, currentFileInfo) then
        begin
            while currentFileInfo.IsValid do
            begin
                tabFiles.Add(currentFileInfo.GetFullFileName);
                currentFileInfo := FindNextDataFile(currentFileInfo);
            end;
        end;

        liveFilePath := '';
        if tabFiles.Count > 0 then
        begin
            var lastTabFile := tabFiles.Last;
            if TryParseFileName(lastTabFile, currentFileInfo) then
            begin
                var liveFileBaseName := Format('%s_%d_%02d.tab-live', [currentFileInfo.Symbol, currentFileInfo.Year, currentFileInfo.Month]);
                var potentialLivePath := TPath.Combine(currentFileInfo.Path, liveFileBaseName);
                if TFile.Exists(potentialLivePath) then
                    liveFilePath := potentialLivePath;
            end;
        end;

        var loadedFileList := TList<TFuture<TArray<TDataPoint<TAskBidItem>>>>.Create;
        var loadStates := TList<TState>.Create;
        try
            for var fileStr in tabFiles do
            begin
                var data := LoadDataFile(fileStr);
                loadedFileList.Add(data);
                loadStates.Add(data.Done);
            end;

            liveData := TFuture<TArray<TDataPoint<TAskBidItem>>>.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<TAskBidItem>>>.Construct(
            loadedState,
            function: TArray<TDataPoint<TAskBidItem>>
            begin
                var tickList := TList<TDataPoint<TAskBidItem>>.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 TAuraTABFileServer.TryParseFileName(const FileName: string; out DataFile: TAuraDataServer<TAskBidItem>.TDataFile): Boolean;
var
    fileNameNoPath, nameForParsing, baseName, ext: string;
    parts: TArray<string>;
begin
    Result := False;
    DataFile := Default(TDataFile);

    DataFile.Path := TPath.GetDirectoryName(FileName);
    fileNameNoPath := TPath.GetFileName(FileName);
    nameForParsing := fileNameNoPath;

    ext := '';
    if nameForParsing.EndsWith('_zip', True) then
    begin
        ext := '_zip';
        nameForParsing := nameForParsing.Substring(0, nameForParsing.Length - 4);
    end;

    DataFile.Extension := TPath.GetExtension(nameForParsing) + ext;
    baseName := TPath.GetFileNameWithoutExtension(nameForParsing);

    parts := baseName.Split(['_']);

    if Length(parts) < 3 then
        exit;

    if not TryStrToInt(parts[High(parts)], DataFile.Month) then
        exit;

    if (DataFile.Month < 1) or (DataFile.Month > 12) then
    begin
        DataFile.Month := 0;
        exit;
    end;

    if not TryStrToInt(parts[High(parts) - 1], DataFile.Year) then
    begin
        DataFile.Month := 0;
        exit;
    end;

    if DataFile.Year <= 0 then
    begin
        DataFile.Year := 0;
        DataFile.Month := 0;
        exit;
    end;

    DataFile.Symbol := string.Join('_', Copy(parts, 0, Length(parts) - 2));

    if DataFile.Symbol = '' then
    begin
        DataFile.Path := '';
        DataFile.Year := 0;
        DataFile.Month := 0;
        exit;
    end;

    Result := True;
end;

end.