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

Refacoring data pipeline

Browse Source
This commit is contained in:
Michael Schimmel
2025-12-14 12:03:16 +01:00
parent 8beb5d95b2
commit f88fe9f5ef
5 changed files with 119 additions and 116 deletions
Download Patch File Download Diff File Expand all files Collapse all files
ASTPlayground/ASTPlayground.delphilsp.json
+1 -1
View File
File diff suppressed because one or more lines are too long
AuraTrader/MainForm.pas
+1 -1
View File
@@ -416,7 +416,7 @@ begin
// Converter: Transforms raw IRecordSeries (Columns) into TArray<TDataPoint> // Converter: Transforms raw IRecordSeries (Columns) into TArray<TDataPoint>
var unpacker: TConverter<TScalarBatch, IScalarRecord> := TScalarBatchUnpacker.Create(def); var unpacker := TScalarBatchLoader.CreateTicker(def);
var ticker := var ticker :=
unpacker.Producer.Chain<TDataPoint<Double>>( unpacker.Producer.Chain<TDataPoint<Double>>(
Src/Myc.Data.Pipeline.pas
+2 -111
View File
@@ -7,8 +7,7 @@ interface
uses uses
Myc.Signals, Myc.Signals,
Myc.Mutable, Myc.Mutable,
Myc.Data.Series, Myc.Data.Series;
Myc.Data.Records;
type type
TTag = Pointer; TTag = Pointer;
@@ -105,12 +104,7 @@ type
class property Null: IConverter<S, T> read GetNull; class property Null: IConverter<S, T> read GetNull;
end; end;
TDataRecordLayoutHelper = record helper for TDataRecord.TLayout // Tools for creating specific converter instances.
function FieldAsRecord<T>(const Name: String): TConverter<T, TDataRecord>;
function FieldOfRecord<T>(const Name: String): TConverter<TDataRecord, T>;
end;
// Factory for creating specific converter instances.
TConverter = record TConverter = record
class function CreateCounter<T>: TConverter<T, Int64>; static; class function CreateCounter<T>: TConverter<T, Int64>; static;
class function CreateTicker<T>: TConverter<TArray<T>, T>; static; class function CreateTicker<T>: TConverter<TArray<T>, T>; static;
@@ -122,32 +116,6 @@ type
TJoinMode = (jmAll, jmAny); TJoinMode = (jmAll, jmAny);
class function Join<T>(Mode: TJoinMode; const Producers: TArray<TProducer<T>>): TProducer<TArray<T>>; static; class function Join<T>(Mode: TJoinMode; const Producers: TArray<TProducer<T>>): TProducer<TArray<T>>; static;
type
TRecordMapping = record
Layout: TDataRecord.TLayout;
Fields: array of record
FromField, ToField: String
end;
end;
class function DataMapping(
const Inputs: TArray<TRecordMapping>;
const Output: TDataRecord.TLayout
): TConverter<TArray<TDataRecord>, TDataRecord>; static;
class function JoinRecords(
const TargetLayout: TDataRecord.TLayout;
const Mapping: TArray<TRecordMapping>;
const Producers: TArray<TProducer<TDataRecord>>
): TConverter<TArray<TDataRecord>, TDataRecord>; static;
end;
TDataRecordHelper = record helper for TArray<TProducer<TDataRecord>>
function JoinRecords(
const TargetLayout: TDataRecord.TLayout;
const Mapping: TArray<TConverter.TRecordMapping>
): TProducer<TDataRecord>; experimental;
end; end;
implementation implementation
@@ -315,38 +283,6 @@ begin
Result := TMycTicker<T>.Create; Result := TMycTicker<T>.Create;
end; end;
class function TConverter.DataMapping(
const Inputs: TArray<TRecordMapping>;
const Output: TDataRecord.TLayout
): TConverter<TArray<TDataRecord>, TDataRecord>;
var
Idxs: array of array of record
FromIdx, ToIdx: Integer
end;
begin
SetLength(Idxs, Length(Inputs));
for var i := 0 to High(Idxs) do
begin
SetLength(Idxs[i], Length(Inputs[i].Fields));
for var j := 0 to High(Idxs[i]) do
begin
Idxs[i][j].FromIdx := Inputs[i].Layout.IndexOf(Inputs[i].Fields[j].FromField);
Idxs[i][j].ToIdx := Output.IndexOf(Inputs[i].Fields[j].ToField);
end;
end;
Result :=
TConverter<TArray<TDataRecord>, TDataRecord>.CreateConverter(
function(const Inputs: TArray<TDataRecord>): TDataRecord
begin
Result := TDataRecord.Create(Output);
for var i := 0 to High(Idxs) do
for var j := 0 to High(Idxs[i]) do
Result.CopyValue(Inputs[i], Idxs[i][j].FromIdx, Idxs[i][j].ToIdx);
end
);
end;
class function TConverter.Join<T>(Mode: TJoinMode; const Producers: TArray<TProducer<T>>): TProducer<TArray<T>>; class function TConverter.Join<T>(Mode: TJoinMode; const Producers: TArray<TProducer<T>>): TProducer<TArray<T>>;
var var
joiner: IDataJoin<T>; joiner: IDataJoin<T>;
@@ -363,49 +299,4 @@ begin
Producers[i].Chain(joiner.Consumers[i]); Producers[i].Chain(joiner.Consumers[i]);
end; end;
class function TConverter.JoinRecords(
const TargetLayout: TDataRecord.TLayout;
const Mapping: TArray<TRecordMapping>;
const Producers: TArray<TProducer<TDataRecord>>
): TConverter<TArray<TDataRecord>, TDataRecord>;
begin
var RecProvider := Join<TDataRecord>(jmAll, Producers);
Result := DataMapping(Mapping, TargetLayout);
RecProvider.Chain(Result.Consumer);
end;
function TDataRecordLayoutHelper.FieldAsRecord<T>(const Name: String): TConverter<T, TDataRecord>;
begin
var layout := Self;
var idx := IndexOf(Name);
Result :=
TConverter<T, TDataRecord>.CreateConverter(
function(const Value: T): TDataRecord
begin
Result := TDataRecord.Create(layout);
Result.SetValue(idx, Value);
end
);
end;
function TDataRecordLayoutHelper.FieldOfRecord<T>(const Name: String): TConverter<TDataRecord, T>;
begin
var idx := IndexOf(Name);
Result := TConverter<TDataRecord, T>.CreateConverter(function(const Value: TDataRecord): T begin Value.GetValue(idx, Result); end);
end;
function TDataRecordHelper.JoinRecords(
const TargetLayout: TDataRecord.TLayout;
const Mapping: TArray<TConverter.TRecordMapping>
): TProducer<TDataRecord>;
begin
var map := TConverter.DataMapping(Mapping, TargetLayout);
TConverter.Join<TDataRecord>(jmAll, Self).Chain(map.Consumer);
Result := map.Producer;
end;
end. end.
Src/Myc.Data.Records.pas
+113 -1
View File
@@ -4,7 +4,8 @@ interface
uses uses
System.SysUtils, System.SysUtils,
System.TypInfo; System.TypInfo,
Myc.Data.Pipeline;
type type
TDataRecord = record TDataRecord = record
@@ -110,6 +111,40 @@ type
property Layout: TLayout read FLayout; property Layout: TLayout read FLayout;
end; end;
TDataRecordLayoutHelper = record helper for TDataRecord.TLayout
function FieldAsRecord<T>(const Name: String): TConverter<T, TDataRecord>;
function FieldOfRecord<T>(const Name: String): TConverter<TDataRecord, T>;
end;
type
TRecordMapping = record
Layout: TDataRecord.TLayout;
Fields: array of record
FromField, ToField: String
end;
end;
// Factory for creating specific converter instances.
TRecordConverter = record
class function DataMapping(
const Inputs: TArray<TRecordMapping>;
const Output: TDataRecord.TLayout
): TConverter<TArray<TDataRecord>, TDataRecord>; static;
class function JoinRecords(
const TargetLayout: TDataRecord.TLayout;
const Mapping: TArray<TRecordMapping>;
const Producers: TArray<TProducer<TDataRecord>>
): TConverter<TArray<TDataRecord>, TDataRecord>; static;
end;
TDataRecordHelper = record helper for TArray<TProducer<TDataRecord>>
function JoinRecords(
const TargetLayout: TDataRecord.TLayout;
const Mapping: TArray<TRecordMapping>
): TProducer<TDataRecord>; experimental;
end;
implementation implementation
uses uses
@@ -569,4 +604,81 @@ begin
Result := not (A = B); Result := not (A = B);
end; end;
function TDataRecordLayoutHelper.FieldAsRecord<T>(const Name: String): TConverter<T, TDataRecord>;
begin
var layout := Self;
var idx := IndexOf(Name);
Result :=
TConverter<T, TDataRecord>.CreateConverter(
function(const Value: T): TDataRecord
begin
Result := TDataRecord.Create(layout);
Result.SetValue(idx, Value);
end
);
end;
function TDataRecordLayoutHelper.FieldOfRecord<T>(const Name: String): TConverter<TDataRecord, T>;
begin
var idx := IndexOf(Name);
Result := TConverter<TDataRecord, T>.CreateConverter(function(const Value: TDataRecord): T begin Value.GetValue(idx, Result); end);
end;
class function TRecordConverter.DataMapping(
const Inputs: TArray<TRecordMapping>;
const Output: TDataRecord.TLayout
): TConverter<TArray<TDataRecord>, TDataRecord>;
var
Idxs: array of array of record
FromIdx, ToIdx: Integer
end;
begin
SetLength(Idxs, Length(Inputs));
for var i := 0 to High(Idxs) do
begin
SetLength(Idxs[i], Length(Inputs[i].Fields));
for var j := 0 to High(Idxs[i]) do
begin
Idxs[i][j].FromIdx := Inputs[i].Layout.IndexOf(Inputs[i].Fields[j].FromField);
Idxs[i][j].ToIdx := Output.IndexOf(Inputs[i].Fields[j].ToField);
end;
end;
Result :=
TConverter<TArray<TDataRecord>, TDataRecord>.CreateConverter(
function(const Inputs: TArray<TDataRecord>): TDataRecord
begin
Result := TDataRecord.Create(Output);
for var i := 0 to High(Idxs) do
for var j := 0 to High(Idxs[i]) do
Result.CopyValue(Inputs[i], Idxs[i][j].FromIdx, Idxs[i][j].ToIdx);
end
);
end;
class function TRecordConverter.JoinRecords(
const TargetLayout: TDataRecord.TLayout;
const Mapping: TArray<TRecordMapping>;
const Producers: TArray<TProducer<TDataRecord>>
): TConverter<TArray<TDataRecord>, TDataRecord>;
begin
var RecProvider := TConverter.Join<TDataRecord>(jmAll, Producers);
Result := TRecordConverter.DataMapping(Mapping, TargetLayout);
RecProvider.Chain(Result.Consumer);
end;
function TDataRecordHelper.JoinRecords(
const TargetLayout: TDataRecord.TLayout;
const Mapping: TArray<TRecordMapping>
): TProducer<TDataRecord>;
begin
var map := TRecordConverter.DataMapping(Mapping, TargetLayout);
TConverter.Join<TDataRecord>(jmAll, Self).Chain(map.Consumer);
Result := map.Producer;
end;
end. end.
Src/Myc.Trade.DataFeed.pas
+2 -2
View File
@@ -135,7 +135,7 @@ type
// Streaming processing: Reads data from disk and broadcasts it via BatchTicker // Streaming processing: Reads data from disk and broadcasts it via BatchTicker
function ProcessData(const Symbol: String; const Terminated: TState): TState; function ProcessData(const Symbol: String; const Terminated: TState): TState;
function CreateTicker(const Def: IScalarRecordDefinition): TConverter<TScalarBatch, IScalarRecord>; class function CreateTicker(const Def: IScalarRecordDefinition): TConverter<TScalarBatch, IScalarRecord>; static;
property Path: String read GetPath; property Path: String read GetPath;
property FileExtension: String read FFileExtension; property FileExtension: String read FFileExtension;
@@ -279,7 +279,7 @@ begin
UpdateSymbols; UpdateSymbols;
end; end;
function TScalarBatchLoader.CreateTicker(const Def: IScalarRecordDefinition): TConverter<TScalarBatch, IScalarRecord>; class function TScalarBatchLoader.CreateTicker(const Def: IScalarRecordDefinition): TConverter<TScalarBatch, IScalarRecord>;
var var
viewObj: TScalarRecordView; viewObj: TScalarRecordView;
view: IScalarRecord; view: IScalarRecord;