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

DataFeed-Producer

Browse Source
This commit is contained in:
Michael Schimmel
2025-12-09 11:12:56 +01:00
parent e56bf7ee7d
commit 43afbd6050
2 changed files with 34 additions and 34 deletions
Download Patch File Download Diff File Expand all files Collapse all files
AuraTrader/MainForm.pas
+5 -5
View File
@@ -114,7 +114,7 @@ type
FProcessDone: TState; FProcessDone: TState;
FApplication: IAuraApplication; FApplication: IAuraApplication;
FModulesItem: TTreeViewItem; FModulesItem: TTreeViewItem;
FTickerServer: IScalarBatchServer; FFeed: IScalarRecordFeed;
function SelectedSymbol: String; function SelectedSymbol: String;
procedure ExecuteStrategy(const Symbol: String; Timeframe: TTimeframe; const Consumer: IConsumer<TDataPoint<TOhlcItem>>); procedure ExecuteStrategy(const Symbol: String; Timeframe: TTimeframe; const Consumer: IConsumer<TDataPoint<TOhlcItem>>);
@@ -382,11 +382,11 @@ begin
var terminated := TFlag.CreateObserver(FTerminate.Signal).State; var terminated := TFlag.CreateObserver(FTerminate.Signal).State;
var path := '\\COFFEE\TickData\Pepperstone'; // Or FServer.Path if compatible var path := '\\COFFEE\TickData\Pepperstone'; // Or FServer.Path if compatible
FTickerServer := TScalarBatchServer.Create(path, '.m1', true, def) as IScalarBatchServer; FFeed := TScalarRecordFeed.Create(path, '.m1', true, def);
// Converter: Transforms raw IRecordSeries (Columns) into TArray<TDataPoint> // Converter: Transforms raw IRecordSeries (Columns) into TArray<TDataPoint>
var ticker := var ticker :=
FTickerServer.Ticker.Chain<TDataPoint<Double>>( FFeed.Ticker.Chain<TDataPoint<Double>>(
function(const Tick: IScalarRecord): TDataPoint<Double> function(const Tick: IScalarRecord): TDataPoint<Double>
begin begin
Result.Time := Tick.ItemByIndex(iTime).Value.AsDouble; Result.Time := Tick.ItemByIndex(iTime).Value.AsDouble;
@@ -404,7 +404,7 @@ begin
TAlphaColors.Orange TAlphaColors.Orange
); );
FTickerServer.SetImportOptions( FFeed.SetImportOptions(
SizeOf(TM1FileItem), SizeOf(TM1FileItem),
procedure(const Src: Pointer; Dst: TScalar.PValue) procedure(const Src: Pointer; Dst: TScalar.PValue)
var var
@@ -428,7 +428,7 @@ begin
); );
// Connect pipeline: Server -> Converter // Connect pipeline: Server -> Converter
FProcessDone := FTickerServer.ProcessData(symbol, terminated); FProcessDone := FFeed.ProcessData(symbol, terminated);
end; end;
function TForm1.CreateStrategy2(Timeframe: TTimeframe): IConsumer<TDataPoint<TOhlcItem>>; function TForm1.CreateStrategy2(Timeframe: TTimeframe): IConsumer<TDataPoint<TOhlcItem>>;
Src/Myc.Trade.DataFeed.pas
+29 -29
View File
@@ -14,7 +14,7 @@ uses
Myc.Data.Scalar, Myc.Data.Scalar,
Myc.Data.Keyword, Myc.Data.Keyword,
Myc.Data.Pipeline, Myc.Data.Pipeline,
Myc.Data.Pipeline.Impl, // Added for TMycContainedProducer Myc.Data.Pipeline.Impl,
Myc.Core.FileCache; Myc.Core.FileCache;
type type
@@ -31,7 +31,7 @@ type
property Items[const Key: IKeyword]: TScalar read GetItems; default; property Items[const Key: IKeyword]: TScalar read GetItems; default;
end; end;
IScalarBatchServer = interface IScalarRecordFeed = interface
procedure ClearCache; procedure ClearCache;
procedure SetImportOptions(RecordSize: Integer; const Converter: TImportConverter); procedure SetImportOptions(RecordSize: Integer; const Converter: TImportConverter);
function EnumerateSymbols: TFuture<TArray<String>>; function EnumerateSymbols: TFuture<TArray<String>>;
@@ -43,7 +43,7 @@ type
property Ticker: TProducer<IScalarRecord> read GetTicker; property Ticker: TProducer<IScalarRecord> read GetTicker;
end; end;
TScalarBatchServer = class(TInterfacedObject, IScalarBatchServer) TScalarRecordFeed = class(TInterfacedObject, IScalarRecordFeed)
public public
type type
TBatch = record TBatch = record
@@ -197,9 +197,9 @@ begin
Result.Create(FDef.Fields[Idx].Value, TScalar.PValue(NativeUInt(FData) + NativeUInt(idx * SizeOf(TScalar.TValue)))^); Result.Create(FDef.Fields[Idx].Value, TScalar.PValue(NativeUInt(FData) + NativeUInt(idx * SizeOf(TScalar.TValue)))^);
end; end;
{ TScalarBatchServer.TDataFile } { TScalarRecordFeed.TDataFile }
constructor TScalarBatchServer.TDataFile.Create(const APath, ASymbol, AExtension: String; AYear, AMonth: Integer; const ABasename: String); constructor TScalarRecordFeed.TDataFile.Create(const APath, ASymbol, AExtension: String; AYear, AMonth: Integer; const ABasename: String);
begin begin
FPath := APath; FPath := APath;
FSymbol := ASymbol; FSymbol := ASymbol;
@@ -209,24 +209,24 @@ begin
FBasename := ABasename; FBasename := ABasename;
end; end;
function TScalarBatchServer.TDataFile.GetBaseFileName: string; function TScalarRecordFeed.TDataFile.GetBaseFileName: string;
begin begin
Result := FBasename; Result := FBasename;
end; end;
function TScalarBatchServer.TDataFile.GetFullFileName: string; function TScalarRecordFeed.TDataFile.GetFullFileName: string;
begin begin
Result := TPath.Combine(FPath, GetBaseFileName + FExtension); Result := TPath.Combine(FPath, GetBaseFileName + FExtension);
end; end;
function TScalarBatchServer.TDataFile.GetIsValid: Boolean; function TScalarRecordFeed.TDataFile.GetIsValid: Boolean;
begin begin
Result := (FSymbol <> '') and (FYear > 0); Result := (FSymbol <> '') and (FYear > 0);
end; end;
{ TScalarBatchServer } { TScalarRecordFeed }
constructor TScalarBatchServer.Create( constructor TScalarRecordFeed.Create(
const APath: String; const APath: String;
const AExtension: String; const AExtension: String;
IsCompressed: Boolean; IsCompressed: Boolean;
@@ -249,43 +249,43 @@ begin
.Create(function(const Filename: String): TFuture<TArray<TBatch>> begin Result := DoLoad(Filename, FLoadGate); end); .Create(function(const Filename: String): TFuture<TArray<TBatch>> begin Result := DoLoad(Filename, FLoadGate); end);
end; end;
destructor TScalarBatchServer.Destroy; destructor TScalarRecordFeed.Destroy;
begin begin
ClearCache; ClearCache;
FTicker.Free;
FCachedFiles.Free; FCachedFiles.Free;
FTicker.Free;
inherited Destroy; inherited Destroy;
end; end;
procedure TScalarBatchServer.AfterConstruction; procedure TScalarRecordFeed.AfterConstruction;
begin begin
inherited; inherited;
UpdateSymbols; UpdateSymbols;
end; end;
procedure TScalarBatchServer.ClearCache; procedure TScalarRecordFeed.ClearCache;
begin begin
FCachedFiles.Clear; FCachedFiles.Clear;
end; end;
procedure TScalarBatchServer.SetImportOptions(RecordSize: Integer; const Converter: TImportConverter); procedure TScalarRecordFeed.SetImportOptions(RecordSize: Integer; const Converter: TImportConverter);
begin begin
FImportRecordSize := RecordSize; FImportRecordSize := RecordSize;
FImportConverter := Converter; FImportConverter := Converter;
ClearCache; ClearCache;
end; end;
function TScalarBatchServer.GetPath: String; function TScalarRecordFeed.GetPath: String;
begin begin
Result := FPath; Result := FPath;
end; end;
function TScalarBatchServer.GetTicker: TProducer<IScalarRecord>; function TScalarRecordFeed.GetTicker: TProducer<IScalarRecord>;
begin begin
Result := FTicker; Result := FTicker;
end; end;
procedure TScalarBatchServer.UpdateSymbols; procedure TScalarRecordFeed.UpdateSymbols;
begin begin
FSymbols := FSymbols :=
TFuture<TDictionary<String, TArray<TDataFile>>>.Construct( TFuture<TDictionary<String, TArray<TDataFile>>>.Construct(
@@ -346,7 +346,7 @@ begin
FSymbols.Manage; FSymbols.Manage;
end; end;
function TScalarBatchServer.EnumerateSymbols: TFuture<TArray<String>>; function TScalarRecordFeed.EnumerateSymbols: TFuture<TArray<String>>;
begin begin
Result := Result :=
FSymbols.Chain<TArray<String>>( FSymbols.Chain<TArray<String>>(
@@ -354,7 +354,7 @@ begin
); );
end; end;
function TScalarBatchServer.FindFirstFile(const Symbol: string): TFuture<TDataFile>; function TScalarRecordFeed.FindFirstFile(const Symbol: string): TFuture<TDataFile>;
begin begin
var symName := Symbol; var symName := Symbol;
Result := Result :=
@@ -371,7 +371,7 @@ begin
); );
end; end;
function TScalarBatchServer.GetNextFile(const Curr: TDataFile): TDataFile; function TScalarRecordFeed.GetNextFile(const Curr: TDataFile): TDataFile;
var var
allSymbolFiles: TArray<TDataFile>; allSymbolFiles: TArray<TDataFile>;
foundIndex: Integer; foundIndex: Integer;
@@ -399,7 +399,7 @@ begin
end; end;
end; end;
function TScalarBatchServer.ParseFileName(const FileName: string): TDataFile; function TScalarRecordFeed.ParseFileName(const FileName: string): TDataFile;
var var
path, fileNameNoPath, baseName, ext, symbol: string; path, fileNameNoPath, baseName, ext, symbol: string;
year, month: Integer; year, month: Integer;
@@ -440,7 +440,7 @@ begin
Result := TDataFile.Create(path, symbol, ext, year, month, baseName); Result := TDataFile.Create(path, symbol, ext, year, month, baseName);
end; end;
function TScalarBatchServer.LoadDataFile(const DataFile: TDataFile): TFuture<TArray<TBatch>>; function TScalarRecordFeed.LoadDataFile(const DataFile: TDataFile): TFuture<TArray<TBatch>>;
begin begin
if DataFile.IsValid then if DataFile.IsValid then
Result := FCachedFiles.GetOrAdd(DataFile.GetFullFileName) Result := FCachedFiles.GetOrAdd(DataFile.GetFullFileName)
@@ -448,7 +448,7 @@ begin
Result := TFuture<TArray<TBatch>>.Construct(TArray<TBatch>(nil)); Result := TFuture<TArray<TBatch>>.Construct(TArray<TBatch>(nil));
end; end;
function TScalarBatchServer.DoLoad(const FileName: string; const Gate: TLatch): TFuture<TArray<TBatch>>; function TScalarRecordFeed.DoLoad(const FileName: string; const Gate: TLatch): TFuture<TArray<TBatch>>;
begin begin
Result := TFuture<TArray<TBatch>>.Null; Result := TFuture<TArray<TBatch>>.Null;
if not TFile.Exists(FileName) then if not TFile.Exists(FileName) then
@@ -490,7 +490,7 @@ begin
end; end;
end; end;
function TScalarBatchServer.ReadCompressedData(const InputStream: TStream): TArray<TBatch>; function TScalarRecordFeed.ReadCompressedData(const InputStream: TStream): TArray<TBatch>;
var var
decompressionStream: TStream; decompressionStream: TStream;
localHeader: TZipHeader; localHeader: TZipHeader;
@@ -536,7 +536,7 @@ begin
end; end;
end; end;
function TScalarBatchServer.ReadUncompressedData(const InputStream: TStream): TArray<TBatch>; function TScalarRecordFeed.ReadUncompressedData(const InputStream: TStream): TArray<TBatch>;
var var
fileSize, totalRecords: Int64; fileSize, totalRecords: Int64;
fieldsPerRecord: Integer; fieldsPerRecord: Integer;
@@ -605,7 +605,7 @@ begin
end; end;
end; end;
function TScalarBatchServer.ProcessData(const Symbol: String; const Terminated: TState): TState; function TScalarRecordFeed.ProcessData(const Symbol: String; const Terminated: TState): TState;
begin begin
var cTerminated := Terminated; var cTerminated := Terminated;
@@ -618,7 +618,7 @@ begin
end); end);
end; end;
function TScalarBatchServer.ProcessFile(FileInfo: TDataFile; const DataFile: TFuture<TArray<TBatch>>; const Terminated: TState): TState; function TScalarRecordFeed.ProcessFile(FileInfo: TDataFile; const DataFile: TFuture<TArray<TBatch>>; const Terminated: TState): TState;
begin begin
if not FileInfo.IsValid or Terminated.IsSet then if not FileInfo.IsValid or Terminated.IsSet then
exit(DataFile.Done); exit(DataFile.Done);
@@ -638,7 +638,7 @@ begin
); );
end; end;
function TScalarBatchServer.ProcessChunks(const Batches: TArray<TBatch>; Terminated: TState): TState; function TScalarRecordFeed.ProcessChunks(const Batches: TArray<TBatch>; Terminated: TState): TState;
begin begin
var callProcess := var callProcess :=
function(Idx: Integer): TFunc<TState> function(Idx: Integer): TFunc<TState>