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:
  - TFileRecord: 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;
        function EnumerateSymbols: TFuture<TArray<String>>;
    end;

    // Aura files

    // Represents metadata for a single data file.
    TAuraDataFile = record
    private
        FExtension: String;
        FPath: String;
        FSymbol: String;
        FYear: Integer;
        FMonth: Integer;
        function GetIsValid: Boolean;
    public
        constructor Create(const APath, ASymbol, AExtension: String; AYear, AMonth: Integer);
        function GetBaseFileName: string;
        function GetFullFileName: string;
        // Gets the next consecutive data file, if it exists on disk.
        function GetNextFile: TAuraDataFile;
        property IsValid: Boolean read GetIsValid;
        property Extension: String read FExtension;
        property Path: String read FPath;
        property Symbol: String read FSymbol;
        property Year: Integer read FYear;
        property Month: Integer read FMonth;
    end;

    // Generic server for loading and managing sequential time-series data from files.
    TAuraDataServer<T: record> = class(TInterfacedObject, IDataServer<T>)
    private
        FCachedFiles: TDataFileCache<TArray<TDataPoint<T>>>;
        FPath: String;
        FSymbols: TFuture<TDictionary<String, TAuraDataFile>>;
        function GetPath: String;

        // Used by cache to actually load an uncached file.
        function DoLoad(const FileName: string): TFuture<TArray<TDataPoint<T>>>;

    strict private
        class var
            FLoadGate: TLatch;

    protected
        // Scans a directory to find the oldest file for a specific symbol.
        function FindFirstFile(const Symbol: string): TFuture<TAuraDataFile>;
        function ParseFileName(const FileName: string): TAuraDataFile; virtual; abstract;

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

    public
        constructor Create(const APath: String);
        destructor Destroy; override;
        procedure AfterConstruction; override;
        function CreateStream(const Symbol: String): IDataStream<T>; virtual; abstract;
        procedure ClearCache;
        procedure UpdateSymbols;
        function EnumerateSymbols: TFuture<TArray<String>>;
        function LoadDataFile(const DataFile: TAuraDataFile): 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
            TDataStream = record
                FileInfo: TAuraDataFile;
                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>)
    protected
        function ParseFileName(const FileName: string): TAuraDataFile; override;
        // Scans a directory and returns the oldest file found for each symbol.
        class function ReadCompressedData(const InputStream: TStream): TArray<TDataPoint<TAskBidItem>>; override;
        class function ReadUncompressedData(const InputStream: TStream): TArray<TDataPoint<TAskBidItem>>; override;
    public
        function CreateStream(const Symbol: String): IDataStream<TAskBidItem>; override;
        function LoadDataSeries(const InitialFile: TAuraDataFile): TFuture<TArray<TDataPoint<TAskBidItem>>>;
    end;

implementation

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

{ TAuraDataFile }

constructor TAuraDataFile.Create(const APath, ASymbol, AExtension: String; AYear, AMonth: Integer);
begin
    FPath := APath;
    FSymbol := ASymbol;
    FExtension := AExtension;
    FYear := AYear;
    FMonth := AMonth;
end;

function TAuraDataFile.GetBaseFileName: string;
begin
    Result := Format('%s_%.4d_%.2d', [FSymbol, FYear, FMonth]);
end;

function TAuraDataFile.GetFullFileName: string;
begin
    Result := TPath.Combine(FPath, GetBaseFileName + FExtension);
end;

function TAuraDataFile.GetIsValid: Boolean;
begin
    Result := (FSymbol <> '') and (FYear > 0);
end;

function TAuraDataFile.GetNextFile: TAuraDataFile;
var
    nextMonth, nextYear: Integer;
    nextFile: TAuraDataFile;
begin
    if not IsValid then
        exit(Default(TAuraDataFile));

    nextMonth := FMonth + 1;
    nextYear := FYear;
    if (nextMonth > 12) then
    begin
        nextMonth := 1;
        Inc(nextYear);
    end;

    // Probe for zipped file first
    nextFile := TAuraDataFile.Create(FPath, FSymbol, '.tab_zip', nextYear, nextMonth);
    if TFile.Exists(nextFile.GetFullFileName) then
        exit(nextFile);

    // Probe for uncompressed file
    nextFile := TAuraDataFile.Create(FPath, FSymbol, '.tab', nextYear, nextMonth);
    if TFile.Exists(nextFile.GetFullFileName) then
        exit(nextFile);

    // No next file found
    Result := Default(TAuraDataFile);
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>.AfterConstruction;
begin
    inherited;
    UpdateSymbols;
end;

procedure TAuraDataServer<T>.ClearCache;
begin
    FCachedFiles.Clear;
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(FLoadGate.Enqueue, 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(
                FLoadGate.Enqueue,
                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;

procedure TAuraDataServer<T>.UpdateSymbols;
begin
    FSymbols :=
        TFuture<TDictionary<String, TAuraDataFile>>.Construct(
            FSymbols.Done,
            function: TDictionary<String, TAuraDataFile>
            var
                fileNames: TArray<string>;
                currentFile: string;
                dataFile: TAuraDataFile;
                trackedInfo: TAuraDataFile;
            begin
                Result := TDictionary<string, TAuraDataFile>.Create;
                if not TDirectory.Exists(FPath) then
                    exit;

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

    FSymbols.Manage;
end;

function TAuraDataServer<T>.EnumerateSymbols: TFuture<TArray<String>>;
var
    dataFiles: TArray<TAuraDataFile>;
    i: Integer;
begin
    Result :=
        FSymbols.Chain<TArray<String>>(
            function(Symbols: TDictionary<String, TAuraDataFile>): TArray<String> begin Result := Symbols.Keys.ToArray; end
        );
end;

function TAuraDataServer<T>.FindFirstFile(const Symbol: string): TFuture<TAuraDataFile>;
begin
    var symName := Symbol;
    Result :=
        FSymbols.Chain<TAuraDataFile>(
            function(Symbols: TDictionary<String, TAuraDataFile>): TAuraDataFile
            begin
                if not Symbols.TryGetValue(symName, Result) then
                    Result := Default(TAuraDataFile);
            end
        );
end;

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

function TAuraDataServer<T>.LoadDataFile(const DataFile: TAuraDataFile): TFuture<TArray<TDataPoint<T>>>;
begin
    Result := FCachedFiles.GetOrAdd(DataFile.GetFullFileName);
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<TAuraDataFile>;
begin
    inherited;

    // We need the path from the server instance to find the file.
    firstFile := FDataServer.FindFirstFile(FSymbol);

    FCurrent :=
        firstFile.Chain<TDataStream>(
            function(const FileInfo: TAuraDataFile): TDataStream
            begin
                if FileInfo.IsValid then
                begin
                    Result.FileInfo := FileInfo;
                    Result.Data := FDataServer.LoadDataFile(FileInfo);
                    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;

    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: TAuraDataFile;
            begin
                if Prev.FileInfo.IsValid then
                begin
                    nextFileInfo := Prev.FileInfo.GetNextFile;
                    if nextFileInfo.IsValid then
                    begin
                        Result.FileInfo := nextFileInfo;
                        Result.Data := FDataServer.LoadDataFile(nextFileInfo);
                        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 InitialFile: TAuraDataFile): TFuture<TArray<TDataPoint<TAskBidItem>>>;
var
    loadedState: TState;
    loadedFiles: TArray<TFuture<TArray<TDataPoint<TAskBidItem>>>>;
    liveData: TFuture<TArray<TDataPoint<TAskBidItem>>>;
    tabFiles: TList<TAuraDataFile>;
    liveFile: TAuraDataFile;
    currentFileInfo: TAuraDataFile;
begin
    tabFiles := TList<TAuraDataFile>.Create;
    try
        currentFileInfo := InitialFile;
        while currentFileInfo.IsValid do
        begin
            tabFiles.Add(currentFileInfo);
            currentFileInfo := currentFileInfo.GetNextFile;
        end;

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

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

            liveData := TFuture<TArray<TDataPoint<TAskBidItem>>>.Null;
            if liveFile.IsValid then
            begin
                liveData := LoadDataFile(liveFile);
                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;

function TAuraTABFileServer.ParseFileName(const FileName: string): TAuraDataFile;
var
    fileNameNoPath, nameForParsing, baseName, ext, path, symbol: string;
    year, month: Integer;
    parts: TArray<string>;
begin
    Result := Default(TAuraDataFile);

    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;

    var fileExt := TPath.GetExtension(nameForParsing) + ext;
    baseName := TPath.GetFileNameWithoutExtension(nameForParsing);

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

    if Length(parts) < 3 then
        exit;

    if not TryStrToInt(parts[High(parts)], month) then
        exit;

    if (month < 1) or (month > 12) then
        exit;

    if not TryStrToInt(parts[High(parts) - 1], year) then
        exit;

    if year <= 0 then
        exit;

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

    if symbol = '' then
        exit;

    Result := TAuraDataFile.Create(path, symbol, fileExt, year, month);
end;

class function TAuraTABFileServer.ReadCompressedData(const InputStream: TStream): TArray<TDataPoint<TAskBidItem>>;
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 TAuraTABFileServer.ReadUncompressedData(const InputStream: TStream): TArray<TDataPoint<TAskBidItem>>;
type
    TFileRecord = packed record
        TimeStamp: TDateTime;
        Data: TAskBidItem;
    end;
var
    fileSize: Int64;
    recordCount, bytesRead: Integer;
    rec: TFileRecord;
begin
    SetLength(Result, 0);
    InputStream.Position := 0;
    fileSize := InputStream.Size;
    if (fileSize = 0) or ((fileSize mod SizeOf(TFileRecord)) <> 0) then
        exit;

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

end.