diff --git a/AuraTrader/AuraTrader.dpr b/AuraTrader/AuraTrader.dpr index 0c6aa76..479440b 100644 --- a/AuraTrader/AuraTrader.dpr +++ b/AuraTrader/AuraTrader.dpr @@ -9,7 +9,8 @@ uses DynamicFMXControl in 'DynamicFMXControl.pas', Myc.Trade.Pipeline.Impl in '..\Src\Myc.Trade.Pipeline.Impl.pas', Myc.Trade.Indicators_v2 in '..\Src\Myc.Trade.Indicators_v2.pas', - Strategy2 in 'Strategy2.pas'; + Strategy2 in 'Strategy2.pas', + Myc.Trade.DataFeed in '..\Src\Myc.Trade.DataFeed.pas'; {$R *.res} diff --git a/AuraTrader/AuraTrader.dproj b/AuraTrader/AuraTrader.dproj index 0383d45..db13ae8 100644 --- a/AuraTrader/AuraTrader.dproj +++ b/AuraTrader/AuraTrader.dproj @@ -4,7 +4,7 @@ 20.3 FMX True - Debug + Release Win64 AuraTrader 3 @@ -140,6 +140,7 @@ + Base diff --git a/AuraTrader/MainForm.fmx b/AuraTrader/MainForm.fmx index 2b5b9f6..b6df951 100644 --- a/AuraTrader/MainForm.fmx +++ b/AuraTrader/MainForm.fmx @@ -204,6 +204,13 @@ object Form1: TForm1 Text = 'Button2' OnClick = Button2Click end + object Button3: TButton + Position.X = 64.000000000000000000 + Position.Y = 152.000000000000000000 + TabOrder = 2 + Text = 'Button3' + OnClick = Button3Click + end end end end diff --git a/AuraTrader/MainForm.pas b/AuraTrader/MainForm.pas index a3b017a..1514a2b 100644 --- a/AuraTrader/MainForm.pas +++ b/AuraTrader/MainForm.pas @@ -40,6 +40,8 @@ uses Myc.Signals.FMX, Myc.TaskManager, Myc.Aura.Module, + Myc.Data.Scalar, + Myc.Data.Keyword, FMX.ListBox, FMX.Layouts, FMX.TreeView, @@ -53,6 +55,7 @@ uses Myc.FMX.Chart, Myc.Trade.Indicators, Myc.Trade.Indicators.Common, + Myc.Trade.DataFeed, StrategyTest, TestMethodCallFromRecordParams; @@ -85,6 +88,7 @@ type Strat2Button: TSpeedButton; Button1: TButton; Button2: TButton; + Button3: TButton; procedure FormCreate(Sender: TObject); procedure FormDestroy(Sender: TObject); procedure StopButtonClick(Sender: TObject); @@ -92,6 +96,7 @@ type procedure AddWorkspaceActionExecute(Sender: TObject); procedure Button1Click(Sender: TObject); procedure Button2Click(Sender: TObject); + procedure Button3Click(Sender: TObject); procedure LogMemoChange(Sender: TObject); procedure Strat2ButtonClick(Sender: TObject); procedure TestActionExecute(Sender: TObject); @@ -109,6 +114,7 @@ type FProcessDone: TState; FApplication: IAuraApplication; FModulesItem: TTreeViewItem; + FTickerServer: IScalarBatchServer; function SelectedSymbol: String; procedure ExecuteStrategy(const Symbol: String; Timeframe: TTimeframe; const Consumer: IConsumer>); @@ -340,6 +346,91 @@ begin end; +procedure TForm1.Button3Click(Sender: TObject); +begin + var symbol := SelectedSymbol; + if symbol = '' then + exit; + + var layout := CurrLayout; + if layout = nil then + exit; + + // 1. Setup Chart + var chart := TMycChart.Create(Self); + AlignControl(chart); + chart.Height := layout.ChildrenRect.Width * 9 / 16; + chart.Lookback.Value := 500000; + + // 2. Define Schema (Keywords & Layout) + var kTime := TKeywordRegistry.Intern('Time'); + var kClose := TKeywordRegistry.Intern('Close'); + + var def := + TScalarRecordRegistry.Intern( + [ + TPair.Create(kTime, TScalar.TKind.DateTime), + TPair.Create(kClose, TScalar.TKind.Float) + ] + ); + + // 3. Build Pipeline + + var iTime := def.IndexOf(kTime); + var iClose := def.IndexOf(kClose); + + var terminated := TFlag.CreateObserver(FTerminate.Signal).State; + var path := '\\COFFEE\TickData\Pepperstone'; // Or FServer.Path if compatible + + FTickerServer := TScalarBatchServer.Create(path, '.m1', true, def) as IScalarBatchServer; + + // Converter: Transforms raw IRecordSeries (Columns) into TArray + var ticker := + FTickerServer.Ticker.Chain>( + function(const Tick: IScalarRecord): TDataPoint + begin + Result.Time := Tick.ItemByIndex(iTime).Value.AsDouble; + Result.Data := Tick.ItemByIndex(iClose).Value.AsDouble; + end + ); + + // Link Chart to Ticker + chart + .SetXAxisSeries(TTimeframe.M, ticker.Chain(function(const p: TDataPoint): TDateTime begin Result := p.Time end)); + + var panel := chart.AddPanel; + panel.AddDoubleSeries( + ticker.Chain(function(const p: TDataPoint): Double begin Result := p.Data end), + TAlphaColors.Orange + ); + + FTickerServer.SetImportOptions( + SizeOf(TM1FileItem), + procedure(const Src: Pointer; Dst: TScalar.PValue) + var + Item: ^TM1FileItem; + begin + Item := Src; + + // Field 0: Time (DateTime is stored as Double/OADate in TScalar) + Dst.AsDouble := Item.OADateTime; + + // Move to next field in destination (TScalar values are contiguous) + Inc(Dst); + + // Field 1: Close (Float) + // Convert Int64 price to Double using Digits: Price / 10^Digits + if Item.Digits > 0 then + Dst.AsDouble := Item.Close / Power(10, Item.Digits) + else + Dst.AsDouble := Item.Close; + end + ); + + // Connect pipeline: Server -> Converter + FProcessDone := FTickerServer.ProcessData(symbol, terminated); +end; + function TForm1.CreateStrategy2(Timeframe: TTimeframe): IConsumer>; type TSignal = record diff --git a/Src/Data/Myc.Data.Scalar.pas b/Src/Data/Myc.Data.Scalar.pas index 2dbfd82..0679c6f 100644 --- a/Src/Data/Myc.Data.Scalar.pas +++ b/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; - 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, 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); diff --git a/Src/Myc.Trade.DataFeed.pas b/Src/Myc.Trade.DataFeed.pas new file mode 100644 index 0000000..c050ee3 --- /dev/null +++ b/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>; + + // Streaming processing: Reads data from disk and broadcasts it via Ticker + function ProcessData(const Symbol: String; const Terminated: TState): TState; + + function GetTicker: TProducer; + property Ticker: TProducer read GetTicker; + end; + + TScalarBatchServer = class(TInterfacedObject, IScalarBatchServer) + public + type + TBatch = record + Data: TArray; + 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>; + FSymbols: TFuture>>; + + // Producer implementation + FTicker: TMycContainedProducer; + + class var + FLoadGate: TLatch; + + function GetPath: String; + + // Loading Logic + function DoLoad(const FileName: string; const Gate: TLatch): TFuture>; + function LoadDataFile(const DataFile: TDataFile): TFuture>; + function GetNextFile(const Curr: TDataFile): TDataFile; + + // Processing Logic + function ProcessFile(FileInfo: TDataFile; const DataFile: TFuture>; const Terminated: TState): TState; + + function ProcessChunks(const Batches: TArray; Terminated: TState): TState; + + // Binary I/O + function ReadCompressedData(const InputStream: TStream): TArray; + function ReadUncompressedData(const InputStream: TStream): TArray; + + // Helpers + function ParseFileName(const FileName: string): TDataFile; + function FindFirstFile(const Symbol: string): TFuture; + procedure UpdateSymbols; + + private + FImportRecordSize: Integer; + FImportConverter: TImportConverter; + + function GetTicker: TProducer; + + 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>; + + // 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 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.Create(Self); + + FCachedFiles := + TDataFileCache> + .Create(function(const Filename: String): TFuture> 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; +begin + Result := FTicker; +end; + +procedure TScalarBatchServer.UpdateSymbols; +begin + FSymbols := + TFuture>>.Construct( + FSymbols.Done, + function: TDictionary> + var + fileNames: TArray; + tempSymbolMap: TDictionary>; + dataFile: TDataFile; + symbolList: TList; + begin + Result := TDictionary>.Create; + tempSymbolMap := TDictionary>.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.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.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>; +begin + Result := + FSymbols.Chain>( + function(const Symbols: TDictionary>): TArray begin Result := Symbols.Keys.ToArray; end + ); +end; + +function TScalarBatchServer.FindFirstFile(const Symbol: string): TFuture; +begin + var symName := Symbol; + Result := + FSymbols.Chain( + function(const Symbols: TDictionary>): TDataFile + var + symbolFiles: TArray; + 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; + 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; +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>; +begin + if DataFile.IsValid then + Result := FCachedFiles.GetOrAdd(DataFile.GetFullFileName) + else + Result := TFuture>.Construct(TArray(nil)); +end; + +function TScalarBatchServer.DoLoad(const FileName: string; const Gate: TLatch): TFuture>; +begin + Result := TFuture>.Null; + if not TFile.Exists(FileName) then + exit; + + var capFileName := FileName; + + if FIsCompressed then + begin + Result := + TFuture + .Construct(Gate.Enqueue, function: TBytes begin Result := TFile.ReadAllBytes(capFileName); end) + .Chain>( + function(bytes: TBytes): TArray + begin + var ms := TBytesStream.Create(bytes); + try + Result := ReadCompressedData(ms); + finally + ms.Free; + end; + end); + end + else + begin + Result := + TFuture>.Construct( + Gate.Enqueue, + function: TArray + 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; +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; +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>; 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): 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; Terminated: TState): TState; +begin + var callProcess := + function(Idx: Integer): TFunc + 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. diff --git a/Src/Myc.Trade.DataStream.pas b/Src/Myc.Trade.DataStream.pas index 90f4a86..bd9aa8f 100644 --- a/Src/Myc.Trade.DataStream.pas +++ b/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,