unit Myc.Trade.DataFeed;

interface

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

type
    // Callback to convert raw bytes from a file into a TScalar.TValue sequence
    TImportConverter = reference to procedure(const Src: Pointer; Dst: TScalar.PValue);

    // Represents a raw block of data read from disk
    TScalarBatch = record
        Data: TArray<TScalar.TValue>;
        RecCount: Integer;
    end;

    IBroker = interface
        function GetEquity: Double;
        procedure Buy(Count: Integer);
        procedure Sell(Count: Integer);
        property Equity: Double read GetEquity;
    end;

    IScalarBatchLoader = interface
        {$region 'private'}
        function GetBatchProducer: TProducer<TScalarBatch>;
        {$endregion}

        function EnumerateSymbols: TFuture<TArray<String>>;

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

        // Provides raw TScalarBatch events
        property BatchProducer: TProducer<TScalarBatch> read GetBatchProducer;
    end;

    TScalarBatchLoader = class(TInterfacedObject, IScalarBatchLoader)
    public
        type
            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 GetFullFileName: string;
                property BaseFileName: string read FBasename;
                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;
            FLookback: Int64;
            FFileExtension: String;
            FIsCompressed: Boolean;

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

            // Internal producer for raw batches
            FBatchProducer: TMycContainedProducer<TScalarBatch>;

        class var
            FLoadGate: TLatch;

        function GetPath: String;

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

        // Processing Logic
        function ProcessFile(
            const Files: TArray<TDataFile>;
            Index: Integer;
            const DataFile: TFuture<TArray<TScalarBatch>>;
            const Terminated: TState
        ): TState;

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

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

        // Helpers
        function ParseFileName(const FileName: string): TDataFile;
        procedure UpdateSymbols;

    private
        FRecordSize: Integer;
        FConverter: TImportConverter;

        function GetBatchProducer: TProducer<TScalarBatch>;

    public
        constructor Create(
            const ACachedFiles: IDataFileCache<TArray<TScalarBatch>>;
            const APath, AExtension: String;
            IsCompressed: Boolean;
            // RecordSize defines both the input bytes per record AND the output TValues per record
            ARecordSize: Integer;
            const AConverter: TImportConverter
        );
        destructor Destroy; override;
        procedure AfterConstruction; override;

        function EnumerateSymbols: TFuture<TArray<String>>;

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

        class function CreateTicker(const Def: IScalarRecordDefinition): TConverter<TScalarBatch, IScalarRecord>; static;

        property Path: String read GetPath;
        property FileExtension: String read FFileExtension;
        property IsCompressed: Boolean read FIsCompressed;
        property Lookback: Int64 read FLookback write FLookback;
        property BatchProducer: TProducer<TScalarBatch> read GetBatchProducer;
    end;

implementation

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

const
    BatchSize = 1024;

type
    // Private implementation of the view logic (Cursor Pattern).
    // This object is reused 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;
        function GetDef: IScalarRecordDefinition;
    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.GetDef: IScalarRecordDefinition;
begin
    Result := FDef;
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;

{ TScalarBatchLoader.TDataFile }

constructor TScalarBatchLoader.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 TScalarBatchLoader.TDataFile.GetFullFileName: string;
begin
    Result := TPath.Combine(FPath, FBasename + FExtension);
end;

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

{ TScalarBatchLoader }

constructor TScalarBatchLoader.Create(
    const ACachedFiles: IDataFileCache<TArray<TScalarBatch>>;
    const APath, AExtension: String;
    IsCompressed: Boolean;
    ARecordSize: Integer;
    const AConverter: TImportConverter
);
begin
    inherited Create;
    FPath := APath;
    FCachedFiles := ACachedFiles;
    FFileExtension := AExtension;
    FIsCompressed := IsCompressed;
    FLookback := -1;

    // RecordSize is used for both Input Bytes Stride AND Output TValues Count
    FRecordSize := ARecordSize;
    FConverter := AConverter;

    Assert(FRecordSize > 0);
    Assert(Assigned(FConverter));

    FBatchProducer := TMycContainedProducer<TScalarBatch>.Create(Self);
    FCachedFiles := TDataFileCache<TArray<TScalarBatch>>.Create;
end;

destructor TScalarBatchLoader.Destroy;
begin
    FCachedFiles.Clear;
    FBatchProducer.Free;
    inherited Destroy;
end;

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

class function TScalarBatchLoader.CreateTicker(const Def: IScalarRecordDefinition): TConverter<TScalarBatch, IScalarRecord>;
var
    viewObj: TScalarRecordView;
    view: IScalarRecord;
begin
    Result :=
        TConverter<TScalarBatch, IScalarRecord>.CreateAggregation(
            function(const Value: TScalarBatch; const Broadcast: TBroadcastFunc<IScalarRecord>): TState
            begin
                if (Length(Value.Data) > 0) and (Value.RecCount > 0) then
                begin
                    var Stride := Length(Value.Data) div Value.RecCount;
                    var pCursor: TScalar.PValue := @Value.Data[0];

                    for var i := 0 to Value.RecCount - 1 do
                    begin
                        if view = nil then
                        begin
                            // Initialize reused View object (Cursor Pattern)
                            // This object implements IScalarRecord and is updated for each tick.
                            viewObj := TScalarRecordView.Create(Def);
                            view := viewObj as IScalarRecord;
                        end;

                        var rc := viewObj._AddRef;
                        try
                            // Broadcast tick
                            viewObj.SetData(pCursor);
                            Broadcast(view);
                        finally
                            // If client picked it up, we release it
                            if viewObj._Release >= rc then
                                view := nil;
                        end;

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

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

function TScalarBatchLoader.GetBatchProducer: TProducer<TScalarBatch>;
begin
    Result := FBatchProducer;
end;

procedure TScalarBatchLoader.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 TScalarBatchLoader.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 TScalarBatchLoader.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 TScalarBatchLoader.LoadDataFile(const DataFile: TDataFile): TFuture<TArray<TScalarBatch>>;
begin
    if DataFile.IsValid then
        Result :=
            FCachedFiles.GetOrAdd(
                DataFile.GetFullFileName,
                function(const Filename: String): TFuture<TArray<TScalarBatch>> begin Result := DoLoad(Filename, FLoadGate); end
            )
    else
        Result := TFuture<TArray<TScalarBatch>>.Construct(TArray<TScalarBatch>(nil));
end;

function TScalarBatchLoader.DoLoad(const FileName: string; const Gate: TLatch): TFuture<TArray<TScalarBatch>>;
begin
    Result := TFuture<TArray<TScalarBatch>>.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<TScalarBatch>>(
                    function(bytes: TBytes): TArray<TScalarBatch>
                    begin
                        var ms := TBytesStream.Create(bytes);
                        try
                            Result := ReadCompressedData(ms);
                        finally
                            ms.Free;
                        end;
                    end);
    end
    else
    begin
        Result :=
            TFuture<TArray<TScalarBatch>>.Construct(
                Gate.Enqueue,
                function: TArray<TScalarBatch>
                begin
                    var fs := TFileStream.Create(capFileName, fmOpenRead or fmShareDenyWrite);
                    try
                        Result := ReadUncompressedData(fs);
                    finally
                        fs.Free;
                    end;
                end
            );
    end;
end;

function TScalarBatchLoader.ReadCompressedData(const InputStream: TStream): TArray<TScalarBatch>;
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 TScalarBatchLoader.ReadUncompressedData(const InputStream: TStream): TArray<TScalarBatch>;
var
    fileSize, totalRecords: Int64;
    numBatches, batchIdx, k: Integer;
    bytesToRead, valuesToWrite: Integer;
    tempBuffer: TBytes;
    pSrc: PByte;
    pDst: TScalar.PValue;
begin
    Result := nil;
    InputStream.Position := 0;
    fileSize := InputStream.Size;

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

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

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

    // Buffer for reading raw data chunks before conversion
    SetLength(tempBuffer, BatchSize * FRecordSize);

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

        // FRecordSize determines both allocation count and input byte stride
        valuesToWrite := recordsInBatch * FRecordSize;
        SetLength(Result[batchIdx].Data, valuesToWrite);

        bytesToRead := recordsInBatch * FRecordSize;
        if InputStream.Read(tempBuffer, 0, bytesToRead) <> bytesToRead then
            raise EReadError.Create('Unexpected end of stream');

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

        // Always use the converter to transform raw bytes to TValues
        for k := 0 to recordsInBatch - 1 do
        begin
            FConverter(pSrc, pDst);
            Inc(pSrc, FRecordSize);
            Inc(pDst, FRecordSize);
        end;
    end;
end;

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

    // We fetch the list once, then iterate by index.
    Result :=
        FSymbols.Chain(
            function(const AllSymbols: TDictionary<String, TArray<TDataFile>>): TState
            var
                files: TArray<TDataFile>;
            begin
                if AllSymbols.TryGetValue(cSymbol, files) and (Length(files) > 0) then
                begin
                    // Start with Index 0
                    Result := ProcessFile(files, 0, LoadDataFile(files[0]), cTerminated);
                end
                else
                begin
                    Result := TState.Null;
                end;
            end
        );
end;

function TScalarBatchLoader.ProcessFile(
    const Files: TArray<TDataFile>;
    Index: Integer;
    const DataFile: TFuture<TArray<TScalarBatch>>;
    const Terminated: TState
): TState;
begin
    // Check bounds and termination
    if (Index < 0) or (Index > High(Files)) or Terminated.IsSet then
        exit(DataFile.Done);

    var currentInfo := Files[Index];
    if not currentInfo.IsValid then
        exit(DataFile.Done);

    // Prefetch logic: Prepare the next file immediately (if available)
    var nextIndex := Index + 1;
    var nextFuture: TFuture<TArray<TScalarBatch>>;

    if nextIndex <= High(Files) then
        nextFuture := LoadDataFile(Files[nextIndex])
    else
        nextFuture := TFuture<TArray<TScalarBatch>>.Construct(TArray<TScalarBatch>(nil)); // End of stream

    var cTerminated := Terminated;

    // Process current data, then recurse to next index
    Result :=
        DataFile.Chain(
            function(const Batches: TArray<TScalarBatch>): TState
            begin
                var done := ProcessChunks(Batches, Terminated);

                // Recurse with Index + 1
                Result :=
                    TaskManager.RunTask(done, function: TState begin Result := ProcessFile(Files, nextIndex, nextFuture, cTerminated); end);
            end
        );
end;

function TScalarBatchLoader.ProcessChunks(const Batches: TArray<TScalarBatch>; Terminated: TState): TState;
begin
    var callProcess :=
        function(Idx: Integer): TFunc<TState>
        begin
            var cBatch := Batches[Idx];

            Result :=
                function: TState
                begin
                    if not Terminated.IsSet then
                    begin
                        // Broadcast the raw batch.
                        FBatchProducer.Broadcast(cBatch);
                        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.