MycLib/Src/Myc.Trade.DataStream.pas

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

type
    // Match C# M1Record (Pack = 1, 48 Bytes)
    TM1Record = packed record
        Time: Double;
        Open: Double;
        High: Double;
        Low: Double;
        Close: Double;
        Spread: Single;
        Volume: Integer;
    end;

    // Match C# TickRecord (Pack = 1, 24 Bytes)
    TTickRecord = packed record
        Time: Double;
        Ask: Double;
        Bid: Double;
    end;

    // Interface for an instantiable data server.
    IDataServer<T: record> = interface
        procedure ClearCache;
        function ProcessData(const Symbol: String; const Terminated: TState; const Processor: IConsumer<TArray<TDataPoint<T>>>): TState;
        function EnumerateSymbols: TFuture<TArray<String>>;
    end;

    // Metadata for a monthly data file (SYMBOL_YYYY_MM.EXT)
    TDataFile = record
    private
        FExtension: String;
        FPath: String;
        FSymbol: String;
        FYear: Integer;
        FMonth: Integer;
        FBasename: String;
        function GetIsValid: Boolean;
    public
        constructor Create(const APath, ASymbol, AExtension: String; AYear, AMonth: Integer; const ABasename: String);
        function GetBaseFileName: string;
        function GetFullFileName: string;
        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 base handling the pipeline and preloading.
    TDataServer<T: record; R: record> = class(TInterfacedObject, IDataServer<T>)
    private
        FCachedFiles: TDataFileCache<TArray<TArray<TDataPoint<T>>>>;
        FPath: String;
        FSymbols: TFuture<TDictionary<String, TArray<TDataFile>>>;
        function GetPath: String;

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

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

    strict private
        class var
            FLoadGate: TLatch;

    protected
        // Maps the raw file record to the internal TDataPoint<T>
        function MapRecord(const RecordBuffer: R): TDataPoint<T>; virtual; abstract;
        function DoLoad(const FileName: string; const Gate: TLatch): TFuture<TArray<TArray<TDataPoint<T>>>>;
        function GetNextFile(const Curr: TDataFile): TDataFile; virtual;
        function ReadCompressedData(const InputStream: TStream): TArray<TArray<TDataPoint<T>>>;
        function ReadUncompressedData(const InputStream: TStream): TArray<TArray<TDataPoint<T>>>;

    public
        constructor Create(const APath: String);
        destructor Destroy; override;
        procedure AfterConstruction; override;
        procedure ClearCache;
        procedure UpdateSymbols;
        function FindFirstFile(const Symbol: string): TFuture<TDataFile>;
        function ParseFileName(const FileName: string): TDataFile;
        function EnumerateSymbols: TFuture<TArray<String>>;
        function LoadDataFile(const DataFile: TDataFile): TFuture<TArray<TArray<TDataPoint<T>>>>;
        function ProcessData(const Symbol: String; const Terminated: TState; const Processor: IConsumer<TArray<TDataPoint<T>>>): TState;

        property Path: String read GetPath;
    end;

    TTickFileServer = class(TDataServer<TTickRecord, TTickRecord>)
    protected
        function MapRecord(const RecordBuffer: TTickRecord): TDataPoint<TTickRecord>; override;
    end;

    TM1FileServer = class(TDataServer<TOhlcItem, TM1Record>)
    protected
        function MapRecord(const RecordBuffer: TM1Record): TDataPoint<TOhlcItem>; override;
    end;

implementation

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

const
    ChunkSize = 1024;

{ TDataFile }

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

function TDataFile.GetBaseFileName: string;
begin
    Result := FBasename;
end;

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

function TDataFile.GetIsValid: Boolean;
begin
    Result := (FSymbol <> '') and (FYear > 0) and (FMonth >= 1) and (FMonth <= 12);
end;

{ TDataServer<T, R> }

constructor TDataServer<T, R>.Create(const APath: String);
begin
    inherited Create;
    FPath := APath;
    FCachedFiles := TDataFileCache<TArray<TArray<TDataPoint<T>>>>.Create;
end;

destructor TDataServer<T, R>.Destroy;
begin
    ClearCache;
    FCachedFiles.Free;
    inherited Destroy;
end;

procedure TDataServer<T, R>.AfterConstruction;
begin
    inherited;
    UpdateSymbols;
end;

procedure TDataServer<T, R>.ClearCache;
begin
    FCachedFiles.Clear;
end;

procedure TDataServer<T, R>.UpdateSymbols;
begin
    FSymbols :=
        TFuture<TDictionary<String, TArray<TDataFile>>>.Construct(
            FSymbols.Done,
            function: TDictionary<String, TArray<TDataFile>>
            var
                fileNames: TArray<string>;
                tempSymbolMap: TDictionary<String, TList<TDataFile>>;
                dataFile: TDataFile;
                symbolList: TList<TDataFile>;
            begin
                Result := TDictionary<String, TArray<TDataFile>>.Create;
                tempSymbolMap := TDictionary<String, TList<TDataFile>>.Create;
                try
                    if not TDirectory.Exists(FPath) then
                        exit;

                    fileNames := TDirectory.GetFiles(FPath);
                    for var currentFile in fileNames do
                    begin
                        dataFile := ParseFileName(currentFile);
                        if dataFile.IsValid then
                        begin
                            if not tempSymbolMap.TryGetValue(dataFile.Symbol, symbolList) then
                            begin
                                symbolList := TList<TDataFile>.Create;
                                tempSymbolMap.Add(dataFile.Symbol, symbolList);
                            end;
                            symbolList.Add(dataFile);
                        end;
                    end;

                    for var kvp in tempSymbolMap do
                    begin
                        symbolList := kvp.Value;
                        symbolList.Sort(
                            TComparer<TDataFile>.Construct(
                                function(const Left, Right: TDataFile): Integer
                                begin
                                    if Left.Year <> Right.Year then
                                        Result := Left.Year - Right.Year
                                    else
                                        Result := Left.Month - Right.Month;
                                end
                            )
                        );
                        Result.Add(kvp.Key, symbolList.ToArray);
                    end;
                finally
                    for var kvp in tempSymbolMap do
                        kvp.Value.Free;
                    tempSymbolMap.Free;
                end;
            end
        );

    FSymbols.Manage;
end;

function TDataServer<T, R>.EnumerateSymbols: TFuture<TArray<String>>;
begin
    Result :=
        FSymbols.Chain<TArray<String>>(
            function(const Symbols: TDictionary<String, TArray<TDataFile>>): TArray<String> begin Result := Symbols.Keys.ToArray; end
        );
end;

function TDataServer<T, R>.FindFirstFile(const Symbol: string): TFuture<TDataFile>;
begin
    var symName := Symbol;
    Result :=
        FSymbols.Chain<TDataFile>(
            function(const Symbols: TDictionary<String, TArray<TDataFile>>): TDataFile
            var
                symbolFiles: TArray<TDataFile>;
            begin
                if Symbols.TryGetValue(symName, symbolFiles) and (Length(symbolFiles) > 0) then
                    Result := symbolFiles[0]
                else
                    Result := Default(TDataFile);
            end
        );
end;

function TDataServer<T, R>.GetNextFile(const Curr: TDataFile): TDataFile;
var
    allSymbolFiles: TArray<TDataFile>;
    foundIndex: Integer;
begin
    Result := Default(TDataFile);
    if (not Curr.IsValid) or (not FSymbols.Done.IsSet) then
        exit;

    if not FSymbols.Value.TryGetValue(Curr.Symbol, allSymbolFiles) then
        exit;

    foundIndex := -1;
    for var i := 0 to High(allSymbolFiles) do
    begin
        if allSymbolFiles[i].GetBaseFileName = Curr.GetBaseFileName then
        begin
            foundIndex := i;
            break;
        end;
    end;

    if (foundIndex <> -1) and (foundIndex < High(allSymbolFiles)) then
        Result := allSymbolFiles[foundIndex + 1];
end;

function TDataServer<T, R>.GetPath: String;
begin
    Result := FPath;
end;

function TDataServer<T, R>.ParseFileName(const FileName: string): TDataFile;
var
    fileNameNoPath, baseName, symbol, ext: string;
    parts: TArray<string>;
    year, month: Integer;
begin
    Result := Default(TDataFile);
    fileNameNoPath := TPath.GetFileName(FileName);
    ext := TPath.GetExtension(fileNameNoPath);
    baseName := TPath.GetFileNameWithoutExtension(fileNameNoPath);

    // Format: SYMBOL_YYYY_MM
    parts := baseName.Split(['_']);
    if Length(parts) < 3 then
        exit;

    // Use reverse indexing for year and month
    if (not TryStrToInt(parts[High(parts)], month)) or (not TryStrToInt(parts[High(parts) - 1], year)) then
        exit;

    symbol := string.Join('_', Copy(parts, 0, Length(parts) - 2));
    Result := TDataFile.Create(TPath.GetDirectoryName(FileName), symbol, ext, year, month, baseName);
end;

function TDataServer<T, R>.LoadDataFile(const DataFile: TDataFile): TFuture<TArray<TArray<TDataPoint<T>>>>;
begin
    if DataFile.IsValid then
        Result :=
            FCachedFiles.GetOrAdd(
                DataFile.GetFullFileName,
                function(const Filename: String): TFuture<TArray<TArray<TDataPoint<T>>>> begin Result := DoLoad(Filename, FLoadGate); end
            )
    else
        Result := TFuture<TArray<TArray<TDataPoint<T>>>>.Construct(TArray<TArray<TDataPoint<T>>>(nil));
end;

function TDataServer<T, R>.DoLoad(const FileName: string; const Gate: TLatch): TFuture<TArray<TArray<TDataPoint<T>>>>;
begin
    var capFileName := FileName;
    Result :=
        TFuture<TBytes>
            .Construct(Gate.Enqueue, function: TBytes begin Result := TFile.ReadAllBytes(capFileName); end)
            .Chain<TArray<TArray<TDataPoint<T>>>>(
                function(Bytes: TBytes): TArray<TArray<TDataPoint<T>>>
                var
                    compStream: TBytesStream;
                begin
                    compStream := TBytesStream.Create(Bytes);
                    try
                        Result := ReadCompressedData(compStream);
                    finally
                        compStream.Free;
                    end;
                end);
end;

function TDataServer<T, R>.ReadCompressedData(const InputStream: TStream): TArray<TArray<TDataPoint<T>>>;
var
    zip: TZipFile;
    decompStream: TStream;
    header: TZipHeader;
    entryIdx: Integer;
begin
    Result := nil;
    InputStream.Position := 0;
    zip := TZipFile.Create;
    try
        zip.Open(InputStream, TZipMode.zmRead);
        entryIdx := -1;
        // Search for the .bin entry created by C#
        for var i := 0 to zip.FileCount - 1 do
        begin
            if zip.FileNames[i].EndsWith('.bin', true) then
            begin
                entryIdx := i;
                break;
            end;
        end;

        if entryIdx = -1 then
            exit;

        zip.Read(entryIdx, decompStream, header);
        try
            Result := ReadUncompressedData(decompStream);
        finally
            decompStream.Free;
        end;
    finally
        zip.Free;
    end;
end;

function TDataServer<T, R>.ReadUncompressedData(const InputStream: TStream): TArray<TArray<TDataPoint<T>>>;
var
    totalRecords, chunkCount, i, j, recsInChunk: Integer;
    rawBuffer: TArray<R>;
begin
    Result := nil;
    totalRecords := InputStream.Size div sizeof(R);
    if totalRecords = 0 then
        exit;

    // Fast block read into memory
    SetLength(rawBuffer, totalRecords);
    InputStream.Position := 0;
    InputStream.ReadBuffer(rawBuffer[0], InputStream.Size);

    chunkCount := (totalRecords + ChunkSize - 1) div ChunkSize;
    SetLength(Result, chunkCount);

    for i := 0 to chunkCount - 1 do
    begin
        recsInChunk := Min(ChunkSize, totalRecords - (i * ChunkSize));
        SetLength(Result[i], recsInChunk);
        for j := 0 to recsInChunk - 1 do
        begin
            Result[i][j] := MapRecord(rawBuffer[i * ChunkSize + j]);
        end;
    end;
end;

function TDataServer<T, R>.ProcessChunks(
    const DataChunks: TArray<TArray<TDataPoint<T>>>;
    Terminated: TState;
    Processor: IConsumer<TArray<TDataPoint<T>>>
): TState;
begin
    var callProcess :=
        function(Idx: Integer): TFunc<TState>
        begin
            var cChunk := DataChunks[Idx];
            Result :=
                function: TState
                begin
                    if not Terminated.IsSet then
                        Result := Processor.Consume(cChunk);
                end;
        end;

    for var i := 0 to High(DataChunks) do
    begin
        if not Terminated.IsSet then
        begin
            var q := TaskManager.RunTask(Result, callProcess(i));
            Result := q;
        end;
    end;
end;

function TDataServer<T, R>.ProcessFile(
    FileInfo: TDataFile;
    const DataFile: TFuture<TArray<TArray<TDataPoint<T>>>>;
    const Terminated: TState;
    Processor: IConsumer<TArray<TDataPoint<T>>>
): TState;
begin
    if (not FileInfo.IsValid) or (Terminated.IsSet) then
        exit(DataFile.Done);

    var nextFileInfo := GetNextFile(FileInfo);
    var nextFile := LoadDataFile(nextFileInfo);
    var cTerminated := Terminated;

    Result :=
        DataFile.Chain(
            function(const DataChunks: TArray<TArray<TDataPoint<T>>>): TState
            begin
                var done := ProcessChunks(DataChunks, cTerminated, Processor);
                Result :=
                    TaskManager
                        .RunTask(done, function: TState begin Result := ProcessFile(nextFileInfo, nextFile, cTerminated, Processor); end);
            end
        );
end;

function TDataServer<T, R>.ProcessData(
    const Symbol: String;
    const Terminated: TState;
    const Processor: IConsumer<TArray<TDataPoint<T>>>
): TState;
begin
    var cProc := Processor;
    var cTerm := Terminated;
    Result :=
        FindFirstFile(Symbol)
            .Chain(function(const Info: TDataFile): TState begin Result := ProcessFile(Info, LoadDataFile(Info), cTerm, cProc); end);
end;

{ TTickFileServer }

function TTickFileServer.MapRecord(const RecordBuffer: TTickRecord): TDataPoint<TTickRecord>;
begin
    Result := TDataPoint<TTickRecord>.Create(RecordBuffer.Time, RecordBuffer);
end;

{ TM1FileServer }

function TM1FileServer.MapRecord(const RecordBuffer: TM1Record): TDataPoint<TOhlcItem>;
begin
    Result :=
        TDataPoint<TOhlcItem>.Create(
            RecordBuffer.Time,
            TOhlcItem.Create(RecordBuffer.Open, RecordBuffer.High, RecordBuffer.Low, RecordBuffer.Close, RecordBuffer.Volume)
        );
end;

end.