unit Myc.Trade.DataFeed;

interface

uses
    System.SysUtils,
    System.Classes,
    System.Generics.Collections,
    System.Generics.Defaults,
    System.IOUtils,
    Myc.Futures,
    Myc.Signals,
    Myc.Mutable,
    Myc.Data.Scalar,
    Myc.Data.Keyword,
    Myc.Data.Pipeline,
    Myc.Data.Pipeline.Impl,
    Myc.Core.FileCache;

type
    TImportConverter = reference to procedure(const Src: Pointer; Dst: TScalar.PValue);

    IScalarRecordFeed = interface
        {$region 'private'}
        function GetTicker: TProducer<IScalarRecord>;
        {$endregion}

        procedure SetImportOptions(RecordSize: Integer; const Converter: TImportConverter);
        function EnumerateSymbols: TFuture<TArray<String>>;

        // Streaming processing: Reads data from disk and broadcasts it via Ticker
        function ProcessData(const Symbol: String; const Terminated: TState): TState;

        property Ticker: TProducer<IScalarRecord> read GetTicker;
    end;

    TScalarRecordFeed = class(TInterfacedObject, IScalarRecordFeed)
    public
        type
            TBatch = record
                Data: TArray<TScalar.TValue>;
                RecCount: Integer;
            end;

            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;

        var
            FPath: String;
            FDef: IScalarRecordDefinition;
            FLookback: Int64;
            FFileExtension: String;
            FIsCompressed: Boolean;

            FCachedFiles: IDataFileCache<TArray<TBatch>>;
            FSymbols: TFuture<TDictionary<String, TArray<TDataFile>>>;

            // Producer implementation
            FTicker: TMycContainedProducer<IScalarRecord>;

        class var
            FLoadGate: TLatch;

        function GetPath: String;

        // Loading Logic
        function DoLoad(const FileName: string; const Gate: TLatch): TFuture<TArray<TBatch>>;
        function LoadDataFile(const DataFile: TDataFile): TFuture<TArray<TBatch>>;
        function GetNextFile(const Curr: TDataFile): TDataFile;

        // Processing Logic
        function ProcessFile(FileInfo: TDataFile; const DataFile: TFuture<TArray<TBatch>>; const Terminated: TState): TState;

        function ProcessChunks(const Batches: TArray<TBatch>; Terminated: TState): TState;

        // Binary I/O
        function ReadCompressedData(const InputStream: TStream): TArray<TBatch>;
        function ReadUncompressedData(const InputStream: TStream): TArray<TBatch>;

        // Helpers
        function ParseFileName(const FileName: string): TDataFile;
        function FindFirstFile(const Symbol: string): TFuture<TDataFile>;
        procedure UpdateSymbols;

    private
        FImportRecordSize: Integer;
        FImportConverter: TImportConverter;

        function GetTicker: TProducer<IScalarRecord>;

    public
        constructor Create(
            const ACachedFiles: IDataFileCache<TArray<TBatch>>;
            const APath, AExtension: String;
            IsCompressed: Boolean;
            const ADef: IScalarRecordDefinition
        );
        destructor Destroy; override;
        procedure AfterConstruction; override;

        procedure SetImportOptions(RecordSize: Integer; const Converter: TImportConverter);
        function EnumerateSymbols: TFuture<TArray<String>>;

        // Streaming processing: Reads data from disk and broadcasts it via Ticker
        function ProcessData(const Symbol: String; const Terminated: TState): TState;

        property Path: String read GetPath;
        property FileExtension: String read FFileExtension;
        property IsCompressed: Boolean read FIsCompressed;
        property Lookback: Int64 read FLookback write FLookback;
        property Ticker: TProducer<IScalarRecord> read GetTicker;
    end;

implementation

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

const
    BatchSize = 1024;

type
    // Private implementation of the view logic.
    // This object is reused (cursor pattern) to avoid allocations per tick.
    TScalarRecordView = class(TInterfacedObject, IScalarRecord)
    private
        FDef: IScalarRecordDefinition;
        FData: TScalar.PValue;

        // IKeywordMapping<TScalar.TValue> implementation
        function IndexOf(const Key: IKeyword): Integer;
        function GetCount: Integer;
        function GetItems(Idx: Integer): TPair<IKeyword, TScalar>;
        function GetFields(const Key: IKeyword): TScalar;
    public
        constructor Create(const ADef: IScalarRecordDefinition);
        procedure SetData(const AData: TScalar.PValue); inline;
    end;

{ TScalarRecordView }

constructor TScalarRecordView.Create(const ADef: IScalarRecordDefinition);
begin
    inherited Create;
    FDef := ADef;
end;

procedure TScalarRecordView.SetData(const AData: TScalar.PValue);
begin
    FData := AData;
end;

function TScalarRecordView.GetCount: Integer;
begin
    Result := FDef.Count;
end;

function TScalarRecordView.GetFields(const Key: IKeyword): TScalar;
var
    idx: Integer;
begin
    idx := FDef.IndexOf(Key);
    if idx < 0 then
        exit(Default(TScalar));

    // Return the raw TValue at the offset
    Result.Kind := FDef[idx].Value;
    Result.Value := TScalar.PValue(NativeUInt(FData) + NativeUInt(idx * SizeOf(TScalar.TValue)))^;
end;

function TScalarRecordView.GetItems(Idx: Integer): TPair<IKeyword, TScalar>;
begin
    Result.Key := FDef[Idx].Key;
    Result.Value.Kind := FDef[idx].Value;
    Result.Value.Value := TScalar.PValue(NativeUInt(FData) + NativeUInt(Idx * SizeOf(TScalar.TValue)))^;
end;

function TScalarRecordView.IndexOf(const Key: IKeyword): Integer;
begin
    Result := FDef.IndexOf(Key);
end;

{ TScalarRecordFeed.TDataFile }

constructor TScalarRecordFeed.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 TScalarRecordFeed.TDataFile.GetBaseFileName: string;
begin
    Result := FBasename;
end;

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

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

{ TScalarRecordFeed }

constructor TScalarRecordFeed.Create(
    const ACachedFiles: IDataFileCache<TArray<TBatch>>;
    const APath, AExtension: String;
    IsCompressed: Boolean;
    const ADef: IScalarRecordDefinition
);
begin
    inherited Create;
    FPath := APath;
    FDef := ADef;
    FCachedFiles := ACachedFiles;
    FFileExtension := AExtension;
    FIsCompressed := IsCompressed;
    FLookback := -1;
    FImportRecordSize := 0;
    FImportConverter := nil;

    FTicker := TMycContainedProducer<IScalarRecord>.Create(Self);

    FCachedFiles := TDataFileCache<TArray<TBatch>>.Create;
end;

destructor TScalarRecordFeed.Destroy;
begin
    FCachedFiles.Clear;
    FTicker.Free;
    inherited Destroy;
end;

procedure TScalarRecordFeed.AfterConstruction;
begin
    inherited;
    UpdateSymbols;
end;

procedure TScalarRecordFeed.SetImportOptions(RecordSize: Integer; const Converter: TImportConverter);
begin
    FImportRecordSize := RecordSize;
    FImportConverter := Converter;
end;

function TScalarRecordFeed.GetPath: String;
begin
    Result := FPath;
end;

function TScalarRecordFeed.GetTicker: TProducer<IScalarRecord>;
begin
    Result := FTicker;
end;

procedure TScalarRecordFeed.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 := -1
                                    else if Left.Year > Right.Year then
                                        Result := 1
                                    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 TScalarRecordFeed.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 TScalarRecordFeed.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 TScalarRecordFeed.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
    begin
        Result := allSymbolFiles[foundIndex + 1];
    end;
end;

function TScalarRecordFeed.ParseFileName(const FileName: string): TDataFile;
var
    path, fileNameNoPath, baseName, ext, symbol: string;
    year, month: Integer;
    parts, dateParts: TArray<string>;
begin
    Result := Default(TDataFile);
    path := TPath.GetDirectoryName(FileName);
    fileNameNoPath := TPath.GetFileName(FileName);

    ext := TPath.GetExtension(fileNameNoPath);

    if not fileNameNoPath.EndsWith(FFileExtension, True) then
        exit;

    baseName := TPath.GetFileNameWithoutExtension(fileNameNoPath);

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

    var startDatePart := parts[High(parts) - 1];
    dateParts := startDatePart.Split(['-']);
    if Length(dateParts) <> 2 then
        exit;

    if not TryStrToInt(dateParts[0], year) then
        exit;
    if not TryStrToInt(dateParts[1], month) then
        exit;

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

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

    Result := TDataFile.Create(path, symbol, ext, year, month, baseName);
end;

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

function TScalarRecordFeed.DoLoad(const FileName: string; const Gate: TLatch): TFuture<TArray<TBatch>>;
begin
    Result := TFuture<TArray<TBatch>>.Null;
    if not TFile.Exists(FileName) then
        exit;

    var capFileName := FileName;

    if FIsCompressed then
    begin
        Result :=
            TFuture<TBytes>
                .Construct(Gate.Enqueue, function: TBytes begin Result := TFile.ReadAllBytes(capFileName); end)
                .Chain<TArray<TBatch>>(
                    function(bytes: TBytes): TArray<TBatch>
                    begin
                        var ms := TBytesStream.Create(bytes);
                        try
                            Result := ReadCompressedData(ms);
                        finally
                            ms.Free;
                        end;
                    end);
    end
    else
    begin
        Result :=
            TFuture<TArray<TBatch>>.Construct(
                Gate.Enqueue,
                function: TArray<TBatch>
                begin
                    var fs := TFileStream.Create(capFileName, fmOpenRead or fmShareDenyWrite);
                    try
                        Result := ReadUncompressedData(fs);
                    finally
                        fs.Free;
                    end;
                end
            );
    end;
end;

function TScalarRecordFeed.ReadCompressedData(const InputStream: TStream): TArray<TBatch>;
var
    decompressionStream: TStream;
    localHeader: TZipHeader;
    entryIndex, i: Integer;
    zip: TZipFile;
begin
    Result := nil;
    decompressionStream := nil;
    zip := nil;

    if not Assigned(InputStream) or (InputStream.Size = 0) then
        exit;

    try
        InputStream.Position := 0;
        zip := TZipFile.Create;
        zip.Open(InputStream, TZipMode.zmRead);

        if zip.FileCount = 0 then
            exit;

        entryIndex := -1;
        for i := 0 to zip.FileCount - 1 do
        begin
            if zip.FileNames[i].EndsWith(FFileExtension, True) then
            begin
                entryIndex := i;
                break;
            end;
        end;

        if entryIndex = -1 then
            entryIndex := 0;

        zip.Read(entryIndex, decompressionStream, localHeader);
        if not Assigned(decompressionStream) then
            exit;

        Result := ReadUncompressedData(decompressionStream);
    finally
        decompressionStream.Free;
        zip.Free;
    end;
end;

function TScalarRecordFeed.ReadUncompressedData(const InputStream: TStream): TArray<TBatch>;
var
    fileSize, totalRecords: Int64;
    fieldsPerRecord: Integer;
    internalRecordSize, sourceRecordSize: Integer;
    numBatches, batchIdx, k: Integer;
    bytesToRead, valuesToWrite: Integer;
    tempBuffer: TBytes;
    pSrc: PByte;
    pDst: TScalar.PValue;
begin
    Result := nil;
    InputStream.Position := 0;
    fileSize := InputStream.Size;

    fieldsPerRecord := FDef.Count;
    internalRecordSize := fieldsPerRecord * sizeof(TScalar.TValue);

    if (FImportRecordSize > 0) and Assigned(FImportConverter) then
        sourceRecordSize := FImportRecordSize
    else
        sourceRecordSize := internalRecordSize;

    if (fileSize = 0) or ((fileSize mod sourceRecordSize) <> 0) then
        exit;

    totalRecords := fileSize div sourceRecordSize;
    if totalRecords = 0 then
        exit;

    numBatches := (totalRecords + BatchSize - 1) div BatchSize;
    SetLength(Result, numBatches);

    if sourceRecordSize <> internalRecordSize then
        SetLength(tempBuffer, BatchSize * sourceRecordSize);

    for batchIdx := 0 to numBatches - 1 do
    begin
        var recordsInBatch := Min(BatchSize, totalRecords - (batchIdx * BatchSize));
        Result[batchIdx].RecCount := recordsInBatch;
        valuesToWrite := recordsInBatch * fieldsPerRecord;

        SetLength(Result[batchIdx].Data, valuesToWrite);

        if sourceRecordSize = internalRecordSize then
        begin
            bytesToRead := valuesToWrite * sizeof(TScalar.TValue);
            if InputStream.Read(Result[batchIdx].Data[0], bytesToRead) <> bytesToRead then
                raise EReadError.Create('Unexpected end of stream');
        end
        else
        begin
            bytesToRead := recordsInBatch * sourceRecordSize;
            if InputStream.Read(tempBuffer, 0, bytesToRead) <> bytesToRead then
                raise EReadError.Create('Unexpected end of stream');

            pSrc := @tempBuffer[0];
            pDst := @Result[batchIdx].Data[0];

            for k := 0 to recordsInBatch - 1 do
            begin
                FImportConverter(pSrc, pDst);
                Inc(pSrc, sourceRecordSize);
                Inc(pDst, fieldsPerRecord);
            end;
        end;
    end;
end;

function TScalarRecordFeed.ProcessData(const Symbol: String; const Terminated: TState): TState;
begin
    var cTerminated := Terminated;

    Result :=
        FindFirstFile(Symbol)
            .Chain(
                function(const FirstInfo: TDataFile): TState
                begin
                    Result := ProcessFile(FirstInfo, LoadDataFile(FirstInfo), cTerminated);
                end);
end;

function TScalarRecordFeed.ProcessFile(FileInfo: TDataFile; const DataFile: TFuture<TArray<TBatch>>; const Terminated: TState): TState;
begin
    if not FileInfo.IsValid or Terminated.IsSet then
        exit(DataFile.Done);

    var nextInfo := GetNextFile(FileInfo);
    var nextFuture := LoadDataFile(nextInfo);

    var cTerminated := Terminated;
    Result :=
        DataFile.Chain(
            function(const Batches: TArray<TBatch>): TState
            begin
                var done := ProcessChunks(Batches, Terminated);

                Result := TaskManager.RunTask(done, function: TState begin Result := ProcessFile(nextInfo, nextFuture, cTerminated); end);
            end
        );
end;

function TScalarRecordFeed.ProcessChunks(const Batches: TArray<TBatch>; Terminated: TState): TState;
begin
    var callProcess :=
        function(Idx: Integer): TFunc<TState>
        begin
            var cData := Batches[Idx].Data;
            var cCount := Batches[Idx].RecCount;

            Result :=
                function: TState
                begin
                    if not Terminated.IsSet then
                    begin
                        if (Length(cData) > 0) and (cCount > 0) then
                        begin
                            var Stride := Length(cData) div cCount;
                            var pCursor: TScalar.PValue := @cData[0];

                            // Initialize reused View object (Cursor Pattern)
                            // This object implements IScalarRecord and is updated for each tick.
                            var ViewObj := TScalarRecordView.Create(FDef);
                            var View := ViewObj as IScalarRecord;

                            for var i := 0 to cCount - 1 do
                            begin
                                ViewObj.SetData(pCursor);

                                // Broadcast to all listeners
                                FTicker.Broadcast(View);

                                Inc(pCursor, Stride);
                            end;
                        end;
                        Result := TState.Null;
                    end;
                end;
        end;

    Result := TState.Null;

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

end.