DataFeed-Producer

Src/Data/Myc.Data.Scalar.pas

@@ -3,11 +3,11 @@ unit Myc.Data.Scalar;
 interface
 
 uses
-    system.sysutils,
-    system.generics.collections,
-    system.generics.defaults,
-    system.math,
-    system.dateutils,
+    System.SysUtils,
+    System.Generics.Collections,
+    System.Generics.Defaults,
+    System.Math,
+    System.DateUtils,
     Myc.Utils,
     Myc.Data.Decimal,
     Myc.Data.Series,
@@ -24,6 +24,7 @@ type
             TKind = (Ordinal, Float, Keyword, Boolean, DateTime);
 
             // The 8-byte storage for the scalar value
+            PValue = ^TValue;
             TValue = record
             case TKind of
                 TKind.Ordinal, TKind.Keyword, TKind.Boolean: (AsInt64: Int64);
@@ -157,7 +158,7 @@ type
     strict private
         FDef: IScalarRecordDefinition;
         FFields: TArray<TScalar.TValue>;
-    end;
+    end deprecated;
 
     ISeries = interface
         {$region 'private'}
@@ -174,7 +175,7 @@ type
         procedure Add(const Item: TScalar.TValue; Lookback: Int64 = -1);
     end;
 
-    // A time series of scalar records, optimized for memory and access speed.
+    // A series of scalar records, optimized for memory and access speed.
     TScalarSeries = class(TInterfacedObject, ISeries, IWriteableSeries)
     private
         FKind: TScalar.TKind;
@@ -196,13 +197,14 @@ type
         function GetFields(const Key: IKeyword): ISeries;
         {$endregion}
         procedure Add(const Item: TScalarRecord; Lookback: Int64 = -1);
+        procedure AddRaw(const Data; Lookback: Int64 = -1);
         property Count: Int64 read GetCount;
         property Def: IScalarRecordDefinition read GetDef;
         property Fields[const Key: IKeyword]: ISeries read GetFields; default;
         property TotalCount: Int64 read GetTotalCount;
     end;
 
-    // A time series of scalar records, optimized for memory and access speed.
+    // A series of scalar records, optimized for memory and access speed.
     TScalarRecordSeries = class(TInterfacedObject, IRecordSeries)
         type
             TMemberSeries = class(TGenericContainedObject<TScalarRecordSeries>, ISeries)
@@ -231,6 +233,10 @@ type
         constructor Create(const ADef: IScalarRecordDefinition);
         destructor Destroy; override;
         procedure Add(const Item: TScalarRecord; Lookback: Int64 = -1);
+
+        // Copies data directly from the Data reference (zero-copy wrapper)
+        procedure AddRaw(const Data; Lookback: Int64 = -1);
+
         property Count: Int64 read GetCount;
         property Fields[const Key: IKeyword]: ISeries read GetFields; default;
         property Def: IScalarRecordDefinition read GetDef;
@@ -849,6 +855,19 @@ begin
     inc(FTotalCount);
 end;
 
+procedure TScalarRecordSeries.AddRaw(const Data; Lookback: Int64 = -1);
+type
+    TValueArr = array[0..MaxInt div SizeOf(TScalar.TValue) - 1] of TScalar.TValue;
+    PValueArr = ^TValueArr;
+var
+    len: Integer;
+begin
+    len := Length(FDef.Fields);
+    // Use Slice to pass the raw data as an open array without copying.
+    FArray.Add(Slice(PValueArr(@Data)^, len), len * Integer(Lookback));
+    inc(FTotalCount);
+end;
+
 function TScalarRecordSeries.GetFields(const Key: IKeyword): ISeries;
 begin
     var elem := FDef.IndexOf(Key);

Src/Myc.Trade.DataFeed.pas

@@ -0,0 +1,688 @@
+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, // Added for TMycContainedProducer
+    Myc.Core.FileCache;
+
+type
+    TImportConverter = reference to procedure(const Src: Pointer; Dst: TScalar.PValue);
+
+    // Interface representing a read-only view of a specific data row.
+    IScalarRecord = interface
+        function GetDef: IScalarRecordDefinition;
+        function GetItems(const Key: IKeyword): TScalar;
+
+        function ItemByIndex(Idx: Integer): TScalar;
+
+        property Def: IScalarRecordDefinition read GetDef;
+        property Items[const Key: IKeyword]: TScalar read GetItems; default;
+    end;
+
+    IScalarBatchServer = interface
+        procedure ClearCache;
+        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;
+
+        function GetTicker: TProducer<IScalarRecord>;
+        property Ticker: TProducer<IScalarRecord> read GetTicker;
+    end;
+
+    TScalarBatchServer = class(TInterfacedObject, IScalarBatchServer)
+    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: TDataFileCache<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 APath: String; const AExtension: String; IsCompressed: Boolean; const ADef: IScalarRecordDefinition);
+        destructor Destroy; override;
+        procedure AfterConstruction; override;
+        procedure ClearCache;
+
+        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;
+    public
+        constructor Create(const ADef: IScalarRecordDefinition);
+        procedure SetData(const AData: TScalar.PValue); inline;
+
+        function ItemByIndex(Idx: Integer): TScalar; inline;
+
+        function GetDef: IScalarRecordDefinition;
+        function GetItems(const Key: IKeyword): TScalar;
+    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.GetDef: IScalarRecordDefinition;
+begin
+    Result := FDef;
+end;
+
+function TScalarRecordView.GetItems(const Key: IKeyword): TScalar;
+begin
+    // Optimization: Depending on hot-path usage, one could cache the Fields array
+    // or use direct index access if the definition allows.
+    var idx := FDef.IndexOf(Key);
+    if idx < 0 then
+        exit(TScalar.Create(TScalar.TKind.Ordinal, Default(TScalar.TValue)));
+
+    Result := ItemByIndex(idx);
+end;
+
+function TScalarRecordView.ItemByIndex(Idx: Integer): TScalar;
+begin
+    // Calculate pointer offset
+    Result.Create(FDef.Fields[Idx].Value, TScalar.PValue(NativeUInt(FData) + NativeUInt(idx * SizeOf(TScalar.TValue)))^);
+end;
+
+{ TScalarBatchServer.TDataFile }
+
+constructor TScalarBatchServer.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 TScalarBatchServer.TDataFile.GetBaseFileName: string;
+begin
+    Result := FBasename;
+end;
+
+function TScalarBatchServer.TDataFile.GetFullFileName: string;
+begin
+    Result := TPath.Combine(FPath, GetBaseFileName + FExtension);
+end;
+
+function TScalarBatchServer.TDataFile.GetIsValid: Boolean;
+begin
+    Result := (FSymbol <> '') and (FYear > 0);
+end;
+
+{ TScalarBatchServer }
+
+constructor TScalarBatchServer.Create(
+    const APath: String;
+    const AExtension: String;
+    IsCompressed: Boolean;
+    const ADef: IScalarRecordDefinition
+);
+begin
+    inherited Create;
+    FPath := APath;
+    FDef := ADef;
+    FFileExtension := AExtension;
+    FIsCompressed := IsCompressed;
+    FLookback := -1;
+    FImportRecordSize := 0;
+    FImportConverter := nil;
+
+    FTicker := TMycContainedProducer<IScalarRecord>.Create(Self);
+
+    FCachedFiles :=
+        TDataFileCache<TArray<TBatch>>
+            .Create(function(const Filename: String): TFuture<TArray<TBatch>> begin Result := DoLoad(Filename, FLoadGate); end);
+end;
+
+destructor TScalarBatchServer.Destroy;
+begin
+    ClearCache;
+    FTicker.Free;
+    FCachedFiles.Free;
+    inherited Destroy;
+end;
+
+procedure TScalarBatchServer.AfterConstruction;
+begin
+    inherited;
+    UpdateSymbols;
+end;
+
+procedure TScalarBatchServer.ClearCache;
+begin
+    FCachedFiles.Clear;
+end;
+
+procedure TScalarBatchServer.SetImportOptions(RecordSize: Integer; const Converter: TImportConverter);
+begin
+    FImportRecordSize := RecordSize;
+    FImportConverter := Converter;
+    ClearCache;
+end;
+
+function TScalarBatchServer.GetPath: String;
+begin
+    Result := FPath;
+end;
+
+function TScalarBatchServer.GetTicker: TProducer<IScalarRecord>;
+begin
+    Result := FTicker;
+end;
+
+procedure TScalarBatchServer.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 TScalarBatchServer.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 TScalarBatchServer.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 TScalarBatchServer.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 TScalarBatchServer.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 TScalarBatchServer.LoadDataFile(const DataFile: TDataFile): TFuture<TArray<TBatch>>;
+begin
+    if DataFile.IsValid then
+        Result := FCachedFiles.GetOrAdd(DataFile.GetFullFileName)
+    else
+        Result := TFuture<TArray<TBatch>>.Construct(TArray<TBatch>(nil));
+end;
+
+function TScalarBatchServer.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 TScalarBatchServer.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 TScalarBatchServer.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 := Length(FDef.Fields);
+    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 TScalarBatchServer.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 TScalarBatchServer.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 TScalarBatchServer.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 View := TScalarRecordView.Create(FDef);
+
+                            for var i := 0 to cCount - 1 do
+                            begin
+                                View.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.

Src/Myc.Trade.DataStream.pas

@@ -96,8 +96,6 @@ type
         property Path: String read GetPath;
     end;
 
-    // Aura tick data file Ask-Bid
-
     TAskBidFileItem = packed record
         Ask: Single;
         Bid: Single;
@@ -113,7 +111,6 @@ type
         function ParseFileName(const FileName: string): TDataFile; override;
     end;
 
-    // File record for cTrader M1 data
     TM1FileItem = packed record
         OADateTime: Double;
         Open: Int64;
@@ -138,7 +135,6 @@ type
     end;
 
 implementation
-
 uses
     System.Zip,
     System.ZLib,