unit Myc.Trade.DataStream;

interface

uses
    System.SysUtils,
    System.Classes,
    System.Generics.Collections,
    System.Generics.Defaults,
    System.IOUtils,
    Myc.Futures,
    Myc.Signals,
    Myc.Mutable,
    Myc.Trade.Types,
    Myc.Trade.DataPoint,
    Myc.Core.FileCache;

type
    // Interface for an instantiable data server.
    IDataServer<T: record> = interface
        ['{1F8E5A9D-E92A-44C1-9F3F-C4B82A6E94B3}']
        procedure ClearCache;
        function ProcessData(const Symbol: String; const Terminated: TState; const Processor: IMycProcessor<TArray<TDataPoint<T>>>): TState;
        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<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<TArray<TDataPoint<T>>>>;

        function ProcessChunks(
            const DataChunks: TArray<TArray<TDataPoint<T>>>;
            const Terminated: TState;
            Processor: IMycProcessor<TArray<TDataPoint<T>>>
        ): TState;

        function ProcessFile(
            FileInfo: TAuraDataFile;
            const DataFile: TFuture<TArray<TArray<TDataPoint<T>>>>;
            const Terminated: TState;
            Processor: IMycProcessor<TArray<TDataPoint<T>>>
        ): TState;

    strict private
        class var
            FLoadGate: TLatch;

    protected
        // Read data from stream and return it chunked.
        class function ReadCompressedData(const InputStream: TStream): TArray<TArray<TDataPoint<T>>>; virtual; abstract;
        // Read data from stream and return it chunked.
        class function ReadUncompressedData(const InputStream: TStream): TArray<TArray<TDataPoint<T>>>; virtual; abstract;

    public
        constructor Create(const APath: String);
        destructor Destroy; override;
        procedure AfterConstruction; override;
        procedure ClearCache;
        procedure UpdateSymbols;
        // 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;
        function EnumerateSymbols: TFuture<TArray<String>>;

        // Load data file and split content into chunks.
        function LoadDataFile(const DataFile: TAuraDataFile): TFuture<TArray<TArray<TDataPoint<T>>>>;

        function ProcessData(const Symbol: String; const Terminated: TState; const Processor: IMycProcessor<TArray<TDataPoint<T>>>): TState;

        property Path: String read GetPath;
    end;

    // Aura tick data file Ask-Bid

    TAuraAskBidFileItem = packed record
        Ask: Single;
        Bid: Single;
    end;

    TAuraTABFileServer = class(TAuraDataServer<TAuraAskBidFileItem>)
    protected
        // Scans a directory and returns the oldest file found for each symbol.
        class function ReadCompressedData(const InputStream: TStream): TArray<TArray<TDataPoint<TAuraAskBidFileItem>>>; override;
        class function ReadUncompressedData(const InputStream: TStream): TArray<TArray<TDataPoint<TAuraAskBidFileItem>>>; override;
    public
        function LoadDataSeries(const InitialFile: TAuraDataFile): TFuture<TArray<TDataPoint<TAuraAskBidFileItem>>>;
        function ParseFileName(const FileName: string): TAuraDataFile; override;
    end;

implementation

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

const
    // The number of records per data chunk when loading files.
    ChunkSize = 512;

{ 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<TArray<TDataPoint<T>>>>
            .Create(function(const Filename: String): TFuture<TArray<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<TArray<TDataPoint<T>>>>;
begin
    Result := TFuture<TArray<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<TArray<TDataPoint<T>>>>(
                    function(bytes: TBytes): TArray<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<TArray<TDataPoint<T>>>>.Construct(
                FLoadGate.Enqueue,
                function: TArray<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>>;
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>.ProcessData(
    const Symbol: String;
    const Terminated: TState;
    const Processor: IMycProcessor<TArray<TDataPoint<T>>>
): TState;
begin
    var cProc := Processor;
    var cTerminated := Terminated;

    Result :=
        FindFirstFile(Symbol)
            .Chain(
                function(const FirstFileInfo: TAuraDataFile): TState
                begin
                    Result := ProcessFile(FirstFileInfo, LoadDataFile(FirstFileInfo), cTerminated, cProc);
                end);
end;

function TAuraDataServer<T>.LoadDataFile(const DataFile: TAuraDataFile): TFuture<TArray<TArray<TDataPoint<T>>>>;
begin
    if DataFile.IsValid then
        Result := FCachedFiles.GetOrAdd(DataFile.GetFullFileName);
end;

function TAuraDataServer<T>.ProcessChunks(
    const DataChunks: TArray<TArray<TDataPoint<T>>>;
    const Terminated: TState;
    Processor: IMycProcessor<TArray<TDataPoint<T>>>
): TState;
begin
    var done := TLatch.CreateLatch(Length(DataChunks));

    for var i := 0 to High(DataChunks) do
        if not Terminated.IsSet then
            Processor.ProcessData(DataChunks[i]).Signal.Subscribe(done)
        else
            done.Notify;

    Result := done.State;
end;

function TAuraDataServer<T>.ProcessFile(
    FileInfo: TAuraDataFile;
    const DataFile: TFuture<TArray<TArray<TDataPoint<T>>>>;
    const Terminated: TState;
    Processor: IMycProcessor<TArray<TDataPoint<T>>>
): TState;
begin
    if not FileInfo.IsValid or Terminated.IsSet then
        exit(DataFile.Done);

    // Read ahead the next file, while processing the current one
    var nextFileInfo := FileInfo.GetNextFile;
    var nextFile := LoadDataFile(nextFileInfo);

    var cTerminated := Terminated;
    Result :=
        DataFile.Chain(
            function(const DataChunks: TArray<TArray<TDataPoint<T>>>): TState
            begin
                // Process each chunk, checking for termination between chunks.
                var done := ProcessChunks(DataChunks, Terminated, Processor);

                // Move to next file (which is currently preloading) once all Processors are done
                Result :=
                    TaskManager
                        .RunTask(done, function: TState begin Result := ProcessFile(nextFileInfo, nextFile, cTerminated, Processor); end);
            end
        );
end;

function TAuraTABFileServer.LoadDataSeries(const InitialFile: TAuraDataFile): TFuture<TArray<TDataPoint<TAuraAskBidFileItem>>>;
var
    loadedState: TState;
    loadedFiles: TArray<TFuture<TArray<TArray<TDataPoint<TAuraAskBidFileItem>>>>>;
    liveData: TFuture<TArray<TArray<TDataPoint<TAuraAskBidFileItem>>>>;
    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<TArray<TDataPoint<TAuraAskBidFileItem>>>>>.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<TArray<TDataPoint<TAuraAskBidFileItem>>>>.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<TAuraAskBidFileItem>>>.Construct(
            loadedState,
            function: TArray<TDataPoint<TAuraAskBidFileItem>>
            begin
                var tickList := TList<TDataPoint<TAuraAskBidFileItem>>.Create;
                try
                    var overallLastTabTickTime: TDateTime := 0;
                    var cnt := 0;
                    for var P in loadedFiles do
                        for var chunk in P.Value do
                            inc(cnt, Length(chunk));
                    tickList.Capacity := cnt;

                    for var P in loadedFiles do
                        for var chunk in P.Value do
                            for var iterTickData in chunk 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<TArray<TDataPoint<TAuraAskBidFileItem>>>;
var
    decompressionStream: TStream;
    localHeader: TZipHeader;
    entryIndex, i: Integer;
    zipFileInstance: TZipFile;
begin
    Result := nil;
    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<TArray<TDataPoint<TAuraAskBidFileItem>>>;
type
    TFileRecord = packed record
        TimeStamp: TDateTime;
        Data: TAuraAskBidFileItem;
    end;
var
    fileSize: Int64;
    totalRecordCount, bytesRead, chunkIndex, indexInChunk, recordIndex: Integer;
    rec: TFileRecord;
begin
    Result := nil;
    InputStream.Position := 0;
    fileSize := InputStream.Size;
    if (fileSize = 0) or ((fileSize mod SizeOf(TFileRecord)) <> 0) then
        exit;

    totalRecordCount := fileSize div SizeOf(TFileRecord);
    if totalRecordCount > 0 then
    begin
        var numChunks := (totalRecordCount + ChunkSize - 1) div ChunkSize;
        SetLength(Result, numChunks);
        chunkIndex := 0;
        indexInChunk := 0;
        SetLength(Result[chunkIndex], Min(ChunkSize, totalRecordCount));

        for recordIndex := 0 to totalRecordCount - 1 do
        begin
            if indexInChunk >= ChunkSize then
            begin
                Inc(chunkIndex);
                indexInChunk := 0;
                var remainingRecords := totalRecordCount - (chunkIndex * ChunkSize);
                SetLength(Result[chunkIndex], Min(ChunkSize, remainingRecords));
            end;

            bytesRead := InputStream.Read(rec, SizeOf(TFileRecord));
            if bytesRead <> SizeOf(TFileRecord) then
                raise EReadError.CreateFmt('Read error. Expected %d bytes, read %d.', [SizeOf(TFileRecord), bytesRead]);

            Result[chunkIndex][indexInChunk].Create(rec.TimeStamp, rec.Data);
            Inc(indexInChunk);
        end;
    end;
end;

end.