Brummel/MycLib
1
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects 1 Releases Wiki Activity

DataFeed Refactoring

Browse Source
This commit is contained in:
Michael Schimmel
2025-12-10 22:30:53 +01:00
parent a3f6f4af26
commit 3ed5a4011f
4 changed files with 227 additions and 205 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Src/Data/Myc.Data.Scalar.pas
-18
View File
@@ -130,16 +130,6 @@ type
class operator Trunc(const A: TScalar): TScalar; inline;
end;
// An array of scalar values of the same kind.
TScalarArray = record
public
Kind: TScalar.TKind;
Items: TArray<TScalar.TValue>;
constructor Create(AKind: TScalar.TKind; const AItems: TArray<TScalar.TValue>);
end;
TScalarTuple = TArray<TScalar>;
// A field definition for a scalar record.
TScalarRecordField = TPair<IKeyword, TScalar.TKind>;
IScalarRecordDefinition = IKeywordMapping<TScalar.TKind>;
@@ -792,14 +782,6 @@ begin
Result := FromInt64(System.Trunc(val));
end;
{ TScalarArray }
constructor TScalarArray.Create(AKind: TScalar.TKind; const AItems: TArray<TScalar.TValue>);
begin
Kind := AKind;
Items := AItems;
end;
{ TScalarRecordSeries }
constructor TScalarRecordSeries.Create(const ADef: IScalarRecordDefinition);
Src/Myc.Trade.DataFeed.pas
+191 -137
View File
@@ -8,6 +8,7 @@ uses
System.Generics.Collections,
System.Generics.Defaults,
System.IOUtils,
System.TimeSpan,
Myc.Futures,
Myc.Signals,
Myc.Mutable,
@@ -18,30 +19,43 @@ uses
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);
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;
// Represents a raw block of data read from disk
TScalarBatch = record
Data: TArray<TScalar.TValue>;
RecCount: Integer;
end;
TScalarRecordFeed = class(TInterfacedObject, IScalarRecordFeed)
// Converts a TScalarBatch into a stream of IScalarRecord events using the cursor pattern.
// NOTE: This uses a zero-copy approach. Consumers must process data synchronously or copy it.
TScalarBatchUnpacker = class(TMycConverter<TScalarBatch, IScalarRecord>)
private
FDef: IScalarRecordDefinition;
protected
function Consume(const Value: TScalarBatch): TState; override;
public
constructor Create(const ADef: IScalarRecordDefinition);
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
TBatch = record
Data: TArray<TScalar.TValue>;
RecCount: Integer;
end;
TDataFile = record
private
FExtension: String;
@@ -65,16 +79,15 @@ type
var
FPath: String;
FDef: IScalarRecordDefinition;
FLookback: Int64;
FFileExtension: String;
FIsCompressed: Boolean;
FCachedFiles: IDataFileCache<TArray<TBatch>>;
FCachedFiles: IDataFileCache<TArray<TScalarBatch>>;
FSymbols: TFuture<TDictionary<String, TArray<TDataFile>>>;
// Producer implementation
FTicker: TMycContainedProducer<IScalarRecord>;
// Internal producer for raw batches
FBatchProducer: TMycContainedProducer<TScalarBatch>;
class var
FLoadGate: TLatch;
@@ -82,54 +95,57 @@ type
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 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<TBatch>>;
const DataFile: TFuture<TArray<TScalarBatch>>;
const Terminated: TState
): TState;
function ProcessChunks(const Batches: TArray<TBatch>; Terminated: TState): TState;
function ProcessChunks(const Batches: TArray<TScalarBatch>; Terminated: TState): TState;
// Binary I/O
function ReadCompressedData(const InputStream: TStream): TArray<TBatch>;
function ReadUncompressedData(const InputStream: TStream): TArray<TBatch>;
function ReadCompressedData(const InputStream: TStream): TArray<TScalarBatch>;
function ReadUncompressedData(const InputStream: TStream): TArray<TScalarBatch>;
// Helpers
function ParseFileName(const FileName: string): TDataFile;
procedure UpdateSymbols;
private
FImportRecordSize: Integer;
FImportConverter: TImportConverter;
FRecordSize: Integer;
FConverter: TImportConverter;
function GetTicker: TProducer<IScalarRecord>;
function GetBatchProducer: TProducer<TScalarBatch>;
public
constructor Create(
const ACachedFiles: IDataFileCache<TArray<TBatch>>;
const ACachedFiles: IDataFileCache<TArray<TScalarBatch>>;
const APath, AExtension: String;
IsCompressed: Boolean;
const ADef: IScalarRecordDefinition
// 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;
procedure SetImportOptions(RecordSize: Integer; const Converter: TImportConverter);
function EnumerateSymbols: TFuture<TArray<String>>;
// Streaming processing: Reads data from disk and broadcasts it via Ticker
// Streaming processing: Reads data from disk and broadcasts it via BatchTicker
function ProcessData(const Symbol: String; const Terminated: TState): TState;
function CreateTicker(const Def: IScalarRecordDefinition): TConverter<TScalarBatch, IScalarRecord>;
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;
property BatchProducer: TProducer<TScalarBatch> read GetBatchProducer;
end;
implementation
@@ -144,8 +160,8 @@ const
BatchSize = 1024;
type
// Private implementation of the view logic.
// This object is reused (cursor pattern) to avoid allocations per tick.
// Private implementation of the view logic (Cursor Pattern).
// This object is reused to avoid allocations per tick.
TScalarRecordView = class(TInterfacedObject, IScalarRecord)
private
FDef: IScalarRecordDefinition;
@@ -204,9 +220,42 @@ begin
Result := FDef.IndexOf(Key);
end;
{ TScalarRecordFeed.TDataFile }
{ TScalarBatchUnpacker }
constructor TScalarRecordFeed.TDataFile.Create(const APath, ASymbol, AExtension: String; AYear, AMonth: Integer; const ABasename: String);
constructor TScalarBatchUnpacker.Create(const ADef: IScalarRecordDefinition);
begin
inherited Create;
FDef := ADef;
end;
function TScalarBatchUnpacker.Consume(const Value: TScalarBatch): 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];
// 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 Value.RecCount - 1 do
begin
ViewObj.SetData(pCursor);
// Broadcast to all listeners
Broadcast(View);
Inc(pCursor, Stride);
end;
end;
Result := TState.Null;
end;
{ TScalarBatchLoader.TDataFile }
constructor TScalarBatchLoader.TDataFile.Create(const APath, ASymbol, AExtension: String; AYear, AMonth: Integer; const ABasename: String);
begin
FPath := APath;
FSymbol := ASymbol;
@@ -216,70 +265,111 @@ begin
FBasename := ABasename;
end;
function TScalarRecordFeed.TDataFile.GetFullFileName: string;
function TScalarBatchLoader.TDataFile.GetFullFileName: string;
begin
Result := TPath.Combine(FPath, FBasename + FExtension);
end;
function TScalarRecordFeed.TDataFile.GetIsValid: Boolean;
function TScalarBatchLoader.TDataFile.GetIsValid: Boolean;
begin
Result := (FSymbol <> '') and (FYear > 0);
end;
{ TScalarRecordFeed }
{ TScalarBatchLoader }
constructor TScalarRecordFeed.Create(
const ACachedFiles: IDataFileCache<TArray<TBatch>>;
constructor TScalarBatchLoader.Create(
const ACachedFiles: IDataFileCache<TArray<TScalarBatch>>;
const APath, AExtension: String;
IsCompressed: Boolean;
const ADef: IScalarRecordDefinition
ARecordSize: Integer;
const AConverter: TImportConverter
);
begin
inherited Create;
FPath := APath;
FDef := ADef;
FCachedFiles := ACachedFiles;
FFileExtension := AExtension;
FIsCompressed := IsCompressed;
FLookback := -1;
FImportRecordSize := 0;
FImportConverter := nil;
FTicker := TMycContainedProducer<IScalarRecord>.Create(Self);
// RecordSize is used for both Input Bytes Stride AND Output TValues Count
FRecordSize := ARecordSize;
FConverter := AConverter;
FCachedFiles := TDataFileCache<TArray<TBatch>>.Create;
Assert(FRecordSize > 0);
Assert(Assigned(FConverter));
FBatchProducer := TMycContainedProducer<TScalarBatch>.Create(Self);
FCachedFiles := TDataFileCache<TArray<TScalarBatch>>.Create;
end;
destructor TScalarRecordFeed.Destroy;
destructor TScalarBatchLoader.Destroy;
begin
FCachedFiles.Clear;
FTicker.Free;
FBatchProducer.Free;
inherited Destroy;
end;
procedure TScalarRecordFeed.AfterConstruction;
procedure TScalarBatchLoader.AfterConstruction;
begin
inherited;
UpdateSymbols;
end;
procedure TScalarRecordFeed.SetImportOptions(RecordSize: Integer; const Converter: TImportConverter);
function TScalarBatchLoader.CreateTicker(const Def: IScalarRecordDefinition): TConverter<TScalarBatch, IScalarRecord>;
var
viewObj: TScalarRecordView;
view: IScalarRecord;
begin
FImportRecordSize := RecordSize;
FImportConverter := Converter;
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 TScalarRecordFeed.GetPath: String;
function TScalarBatchLoader.GetPath: String;
begin
Result := FPath;
end;
function TScalarRecordFeed.GetTicker: TProducer<IScalarRecord>;
function TScalarBatchLoader.GetBatchProducer: TProducer<TScalarBatch>;
begin
Result := FTicker;
Result := FBatchProducer;
end;
procedure TScalarRecordFeed.UpdateSymbols;
procedure TScalarBatchLoader.UpdateSymbols;
begin
FSymbols :=
TFuture<TDictionary<String, TArray<TDataFile>>>.Construct(
@@ -340,7 +430,7 @@ begin
FSymbols.Manage;
end;
function TScalarRecordFeed.EnumerateSymbols: TFuture<TArray<String>>;
function TScalarBatchLoader.EnumerateSymbols: TFuture<TArray<String>>;
begin
Result :=
FSymbols.Chain<TArray<String>>(
@@ -348,7 +438,7 @@ begin
);
end;
function TScalarRecordFeed.ParseFileName(const FileName: string): TDataFile;
function TScalarBatchLoader.ParseFileName(const FileName: string): TDataFile;
var
path, fileNameNoPath, baseName, ext, symbol: string;
year, month: Integer;
@@ -389,21 +479,21 @@ begin
Result := TDataFile.Create(path, symbol, ext, year, month, baseName);
end;
function TScalarRecordFeed.LoadDataFile(const DataFile: TDataFile): TFuture<TArray<TBatch>>;
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<TBatch>> begin Result := DoLoad(Filename, FLoadGate); end
function(const Filename: String): TFuture<TArray<TScalarBatch>> begin Result := DoLoad(Filename, FLoadGate); end
)
else
Result := TFuture<TArray<TBatch>>.Construct(TArray<TBatch>(nil));
Result := TFuture<TArray<TScalarBatch>>.Construct(TArray<TScalarBatch>(nil));
end;
function TScalarRecordFeed.DoLoad(const FileName: string; const Gate: TLatch): TFuture<TArray<TBatch>>;
function TScalarBatchLoader.DoLoad(const FileName: string; const Gate: TLatch): TFuture<TArray<TScalarBatch>>;
begin
Result := TFuture<TArray<TBatch>>.Null;
Result := TFuture<TArray<TScalarBatch>>.Null;
if not TFile.Exists(FileName) then
exit;
@@ -414,8 +504,8 @@ begin
Result :=
TFuture<TBytes>
.Construct(Gate.Enqueue, function: TBytes begin Result := TFile.ReadAllBytes(capFileName); end)
.Chain<TArray<TBatch>>(
function(bytes: TBytes): TArray<TBatch>
.Chain<TArray<TScalarBatch>>(
function(bytes: TBytes): TArray<TScalarBatch>
begin
var ms := TBytesStream.Create(bytes);
try
@@ -428,9 +518,9 @@ begin
else
begin
Result :=
TFuture<TArray<TBatch>>.Construct(
TFuture<TArray<TScalarBatch>>.Construct(
Gate.Enqueue,
function: TArray<TBatch>
function: TArray<TScalarBatch>
begin
var fs := TFileStream.Create(capFileName, fmOpenRead or fmShareDenyWrite);
try
@@ -443,7 +533,7 @@ begin
end;
end;
function TScalarRecordFeed.ReadCompressedData(const InputStream: TStream): TArray<TBatch>;
function TScalarBatchLoader.ReadCompressedData(const InputStream: TStream): TArray<TScalarBatch>;
var
decompressionStream: TStream;
localHeader: TZipHeader;
@@ -489,11 +579,9 @@ begin
end;
end;
function TScalarRecordFeed.ReadUncompressedData(const InputStream: TStream): TArray<TBatch>;
function TScalarBatchLoader.ReadUncompressedData(const InputStream: TStream): TArray<TScalarBatch>;
var
fileSize, totalRecords: Int64;
fieldsPerRecord: Integer;
internalRecordSize, sourceRecordSize: Integer;
numBatches, batchIdx, k: Integer;
bytesToRead, valuesToWrite: Integer;
tempBuffer: TBytes;
@@ -504,61 +592,46 @@ begin
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
if (fileSize = 0) or ((fileSize mod FRecordSize) <> 0) then
exit;
totalRecords := fileSize div sourceRecordSize;
totalRecords := fileSize div FRecordSize;
if totalRecords = 0 then
exit;
numBatches := (totalRecords + BatchSize - 1) div BatchSize;
SetLength(Result, numBatches);
if sourceRecordSize <> internalRecordSize then
SetLength(tempBuffer, BatchSize * sourceRecordSize);
// 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;
valuesToWrite := recordsInBatch * fieldsPerRecord;
// FRecordSize determines both allocation count and input byte stride
valuesToWrite := recordsInBatch * FRecordSize;
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');
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];
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;
// 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 TScalarRecordFeed.ProcessData(const Symbol: String; const Terminated: TState): TState;
function TScalarBatchLoader.ProcessData(const Symbol: String; const Terminated: TState): TState;
begin
var cTerminated := Terminated;
var cSymbol := Symbol;
@@ -583,10 +656,10 @@ begin
);
end;
function TScalarRecordFeed.ProcessFile(
function TScalarBatchLoader.ProcessFile(
const Files: TArray<TDataFile>;
Index: Integer;
const DataFile: TFuture<TArray<TBatch>>;
const DataFile: TFuture<TArray<TScalarBatch>>;
const Terminated: TState
): TState;
begin
@@ -600,19 +673,19 @@ begin
// Prefetch logic: Prepare the next file immediately (if available)
var nextIndex := Index + 1;
var nextFuture: TFuture<TArray<TBatch>>;
var nextFuture: TFuture<TArray<TScalarBatch>>;
if nextIndex <= High(Files) then
nextFuture := LoadDataFile(Files[nextIndex])
else
nextFuture := TFuture<TArray<TBatch>>.Construct(TArray<TBatch>(nil)); // End of stream
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<TBatch>): TState
function(const Batches: TArray<TScalarBatch>): TState
begin
var done := ProcessChunks(Batches, Terminated);
@@ -623,39 +696,20 @@ begin
);
end;
function TScalarRecordFeed.ProcessChunks(const Batches: TArray<TBatch>; Terminated: TState): TState;
function TScalarBatchLoader.ProcessChunks(const Batches: TArray<TScalarBatch>; Terminated: TState): TState;
begin
var callProcess :=
function(Idx: Integer): TFunc<TState>
begin
var cData := Batches[Idx].Data;
var cCount := Batches[Idx].RecCount;
var cBatch := Batches[Idx];
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;
// Broadcast the raw batch.
FBatchProducer.Broadcast(cBatch);
Result := TState.Null;
end;
end;