From abad66ae5215f306554189bb0742bc0f2dca96b8 Mon Sep 17 00:00:00 2001 From: Michael Schimmel Date: Wed, 16 Jul 2025 22:27:53 +0200 Subject: [PATCH] Removed bottleneck in DataStream --- AuraTrader/MainForm.pas | 88 +++-- Src/Myc.Trade.DataPoint.Impl.pas | 164 ++++++++- Src/Myc.Trade.DataPoint.pas | 10 +- Src/Myc.Trade.DataStream.pas | 563 +++++++++++++++++++------------ 4 files changed, 578 insertions(+), 247 deletions(-) diff --git a/AuraTrader/MainForm.pas b/AuraTrader/MainForm.pas index 4dd5d29..d75cc4b 100644 --- a/AuraTrader/MainForm.pas +++ b/AuraTrader/MainForm.pas @@ -2,6 +2,8 @@ unit MainForm; interface +{.$define TICKDATA} + uses System.SysUtils, System.Types, @@ -88,14 +90,22 @@ type private FOnEvent: TNotifyEvent; { Private declarations } +{$ifdef TICKDATA} FServer: IDataServer; +{$else} + FServer: IDataServer; +{$endif} FSymbols: TFuture>; FTerminate: TEvent; FProcessDone: TState; FApplication: IAuraApplication; FModulesItem: TTreeViewItem; function SelectedSymbol: String; - function ExecuteStrategy(const Symbol: String; const Processor: IMycProcessor>>): TState; + function ExecuteStrategy( + const Symbol: String; + Timeframe: TTimeframe; + const Processor: IMycProcessor> + ): TState; public procedure NewWorkspace; @@ -215,7 +225,11 @@ begin FTerminate := TEvent.CreateEvent; // Create an instance of the TAuraTABFileServer. The server can be reused for multiple stream creations. [364] +{$ifdef TICKDATA} FServer := TAuraTABFileServer.Create('\\COFFEE\TickData\Pepperstone'); +{$else} + FServer := TAuraM1FileServer.Create('\\COFFEE\TickData\Pepperstone'); +{$endif} SymbolsComboBox.Enabled := false; ChartButton.Enabled := false; @@ -299,13 +313,27 @@ begin Result := Res; end; -function TForm1.ExecuteStrategy(const Symbol: String; const Processor: IMycProcessor>>): TState; -var - dataProvider: TConverter>, TArray>>; +function TForm1.ExecuteStrategy(const Symbol: String; Timeframe: TTimeframe; const Processor: IMycProcessor>): TState; begin var terminated := TFlag.CreateObserver(FTerminate.Signal).State; - dataProvider := +{$ifdef TICKDATA} + var ticker := TConverter.CreateTicker>; + + var lastPrice := + ticker.Chain>( + function(const Tick: TDataPoint): TDataPoint + begin + Result.Time := Tick.Time; + Result.Data := 0.5 * (Tick.Data.Ask + Tick.Data.Bid); + end + ); + + var OhlcPoint := lastPrice.Chain>(TConverter.CreateTickAggregation(Timeframe)); + + OhlcPoint.Sender.Link(Processor); + + var dataProvider := TConverter>, TArray>>.CreateGeneric( function(const Values: TArray>): TArray> begin @@ -319,9 +347,20 @@ begin end ); - dataProvider.Sender.Link(Processor); + dataProvider.Sender.Link(ticker); Result := FServer.ProcessData(Symbol, terminated, dataProvider); +{$else} + + var ticker := TConverter.CreateTicker>; + + var OhlcPoint := ticker.Chain>(TConverter.CreateOhlcAggregation(Timeframe)); + + OhlcPoint.Sender.Link(Processor); + + Result := FServer.ProcessData(Symbol, terminated, ticker); + +{$endif} end; function TForm1.SelectedSymbol: String; @@ -350,20 +389,9 @@ begin ///// - var timeframe := TTimeframe.M15; + var OhlcPoint := TConverter.CreateIdentity>; - var ticker := TConverter.CreateTicker>; - - var lastPrice := - ticker.Chain>( - function(const Tick: TDataPoint): TDataPoint - begin - Result.Time := Tick.Time; - Result.Data := 0.5 * (Tick.Data.Ask + Tick.Data.Bid); - end - ); - - var OhlcPoint := lastPrice.Chain>(TConverter.CreateAggregation(timeframe)); + var timeframe := TTimeframe.H; var Timestamps := OhlcPoint.Field('Time'); var Ohlc := OhlcPoint.Field('Data'); @@ -425,7 +453,7 @@ begin } ///// - var done := ExecuteStrategy(Symbol, ticker); + var done := ExecuteStrategy(Symbol, timeframe, OhlcPoint); FProcessDone := TState.All([FProcessDone, done]); end; @@ -437,19 +465,12 @@ type Entry: Double; pnl: Double; end; +var + panel: TMycChart.TPanel; begin var timeframe := TTimeframe.M15; - var ticker := TConverter.CreateTicker>; - - var lastPrice := - ticker.Chain>( - TConverter.CreateDataPointConverter( - function(const Tick: TAskBidItem): Double begin Result := 0.5 * (Tick.Ask + Tick.Bid); end - ) - ); - - var OhlcPoint := lastPrice.Chain>(TConverter.CreateAggregation(timeframe)).MakeParallel; + var OhlcPoint := TConverter.CreateIdentity>; var Ohlc := TConverter.CreateSequence(2, OhlcPoint.Field('Data').Sender); @@ -575,16 +596,13 @@ begin chart.SetXAxisSeries(M15, OhlcPoint.Field('Time').Sender); - var panel := chart.AddPanel; + panel := chart.AddPanel; panel.AddOhlcSeries(Ohlc[0].Sender); panel.AddDoubleSeries(Hull.Sender, TAlphaColors.Cornflowerblue, 2); panel.AddDoubleSeries(Sma.Sender, TAlphaColors.Brown, 1.5); panel.AddDoubleSeries(Signal.Field('SL').Sender, TAlphaColors.Red, 2); panel.AddDoubleSeries(Signal.Field('Entry').Sender, TAlphaColors.Green, 1); - // panel := chart.AddPanel; - // panel.AddDoubleSeries( equity.Sender, TAlphaColors.Blue, 3 ); - var pnlChart := TMycChart.Create(Self); AlignControl(pnlChart); pnlChart.Height := Layout.ChildrenRect.Width * 9 / 24; @@ -597,7 +615,7 @@ begin ///// - var done := ExecuteStrategy(Symbol, ticker); + var done := ExecuteStrategy(Symbol, timeframe, OhlcPoint); FProcessDone := TState.All([FProcessDone, done]); end; diff --git a/Src/Myc.Trade.DataPoint.Impl.pas b/Src/Myc.Trade.DataPoint.Impl.pas index c1107e4..d79f58d 100644 --- a/Src/Myc.Trade.DataPoint.Impl.pas +++ b/Src/Myc.Trade.DataPoint.Impl.pas @@ -158,6 +158,12 @@ type // Endpoint that collects data into a series. TMycDataEndpoint = class(TInterfacedObject, TMutable>.IMutable) + type + PItem = ^TItem; + TItem = record + Next: PItem; + Value: T; + end; private FProcessor: TMycContainedProcessor; FTag: TDataProvider.TTag; @@ -166,6 +172,7 @@ type FData: TSeries; FChanged: TEvent; FLock: TLightweightMREW; + FFirst: PItem; function GetChanged: TSignal; function GetValue: TSeries; function ProcessData(const Value: T): TState; @@ -195,6 +202,20 @@ type function ProcessData(const Value: T): TState; override; final; end; + TOhlcAggregation = class(TMycConverter, TDataPoint>) + private + FTimeframe: TTimeframe; + FCurrentBar: TDataPoint; + function GetBarStartTime(const TimeStamp: TDateTime; const Timeframe: TTimeframe): TDateTime; + function GetCurrentBar: TDataPoint; + function GetTimeframe: TTimeframe; + public + constructor Create(const ATimeframe: TTimeframe); + function ProcessData(const Value: TDataPoint): TState; override; + property CurrentBar: TDataPoint read GetCurrentBar; + property Timeframe: TTimeframe read GetTimeframe; + end; + implementation uses @@ -461,11 +482,42 @@ end; function TMycDataEndpoint.GetValue: TSeries; begin - FLock.BeginRead; + FLock.BeginWrite; try + var cnt := 0; + var tmp: PItem := nil; + var item: PItem; + + while FFirst <> nil do + begin + item := FFirst; + FFirst := item.Next; + item.Next := tmp; + tmp := item; + inc(cnt); + end; + + if cnt > 0 then + begin + var Arr: TArray; + SetLength(Arr, cnt); + + cnt := 0; + while tmp <> nil do + begin + item := tmp; + tmp := item.Next; + Arr[cnt] := item.Value; + inc(cnt); + Dispose(item); + end; + + FData := FData.Add(Arr, 0, cnt, FLookback); + end; + Result := FData; finally - FLock.EndRead; + FLock.EndWrite; end; end; @@ -474,8 +526,14 @@ begin Result := TState.Null; FLock.BeginWrite; try + var P: PItem; + New(P); + P.Next := FFirst; + FFirst := P; + P.Value := Value; + // Add new data point, respecting the lookback period. - FData := FData.Add(Value, FLookback); + // FData := FData.Add(Value, FLookback); finally FLock.EndWrite; end; @@ -657,4 +715,104 @@ begin FQueue := Result; end; +{ TOhlcAggregation } + +constructor TOhlcAggregation.Create(const ATimeframe: TTimeframe); +begin + inherited Create; + FTimeframe := ATimeframe; +end; + +function TOhlcAggregation.GetBarStartTime(const TimeStamp: TDateTime; const Timeframe: TTimeframe): TDateTime; +var + baseTime: TDateTime; +begin + // Align the time grid to UTC 0:00 using functions from System.DateUtils + baseTime := RecodeMilliSecond(TimeStamp, 0); + + case Timeframe of + S: Result := baseTime; + S5: Result := RecodeSecond(baseTime, SecondOf(TimeStamp) - (SecondOf(TimeStamp) mod 5)); + S15: Result := RecodeSecond(baseTime, SecondOf(TimeStamp) - (SecondOf(TimeStamp) mod 15)); + S30: Result := RecodeSecond(baseTime, SecondOf(TimeStamp) - (SecondOf(TimeStamp) mod 30)); + + M: Result := RecodeSecond(baseTime, 0); + M2: Result := RecodeMinute(RecodeSecond(baseTime, 0), MinuteOf(TimeStamp) - (MinuteOf(TimeStamp) mod 2)); + M3: Result := RecodeMinute(RecodeSecond(baseTime, 0), MinuteOf(TimeStamp) - (MinuteOf(TimeStamp) mod 3)); + M5: Result := RecodeMinute(RecodeSecond(baseTime, 0), MinuteOf(TimeStamp) - (MinuteOf(TimeStamp) mod 5)); + M10: Result := RecodeMinute(RecodeSecond(baseTime, 0), MinuteOf(TimeStamp) - (MinuteOf(TimeStamp) mod 10)); + M15: Result := RecodeMinute(RecodeSecond(baseTime, 0), MinuteOf(TimeStamp) - (MinuteOf(TimeStamp) mod 15)); + M30: Result := RecodeMinute(RecodeSecond(baseTime, 0), MinuteOf(TimeStamp) - (MinuteOf(TimeStamp) mod 30)); + + H: Result := RecodeMinute(RecodeSecond(baseTime, 0), 0); + H2: Result := RecodeHour(RecodeMinute(RecodeSecond(baseTime, 0), 0), HourOf(TimeStamp) - (HourOf(TimeStamp) mod 2)); + H3: Result := RecodeHour(RecodeMinute(RecodeSecond(baseTime, 0), 0), HourOf(TimeStamp) - (HourOf(TimeStamp) mod 3)); + H4: Result := RecodeHour(RecodeMinute(RecodeSecond(baseTime, 0), 0), HourOf(TimeStamp) - (HourOf(TimeStamp) mod 4)); + H8: Result := RecodeHour(RecodeMinute(RecodeSecond(baseTime, 0), 0), HourOf(TimeStamp) - (HourOf(TimeStamp) mod 8)); + H12: Result := RecodeHour(RecodeMinute(RecodeSecond(baseTime, 0), 0), HourOf(TimeStamp) - (HourOf(TimeStamp) mod 12)); + + D: Result := StartOfTheDay(TimeStamp); + // D2, D3 are uncommon; this is a simple modulo-based approach relative to TDateTime's epoch. + D2: Result := Floor(TimeStamp) - (Floor(TimeStamp) mod 2); + D3: Result := Floor(TimeStamp) - (Floor(TimeStamp) mod 3); + + W: Result := TimeStamp.StartOfTheWeek; + + MN: Result := TimeStamp.StartOfTheMonth; + // Quarter alignment + MN3: Result := RecodeMonth(TimeStamp.StartOfTheMonth, (MonthOf(TimeStamp) - 1) div 3 * 3 + 1); + // Half-year alignment + MN6: Result := RecodeMonth(TimeStamp.StartOfTheMonth, (MonthOf(TimeStamp) - 1) div 6 * 6 + 1); + + Y: Result := TimeStamp.StartOfTheYear; + else + // Fallback for any undefined timeframe + Result := 0; + end; +end; + +function TOhlcAggregation.GetCurrentBar: TDataPoint; +begin + Result := FCurrentBar; +end; + +function TOhlcAggregation.GetTimeframe: TTimeframe; +begin + Result := FTimeframe; +end; + +function TOhlcAggregation.ProcessData(const Value: TDataPoint): TState; +var + barStartTime: TDateTime; + lastBarTime: TDateTime; +begin + // Update bar for the strategy's timeframe + barStartTime := GetBarStartTime(Value.Time, FTimeframe); + lastBarTime := FCurrentBar.Time; + + if (barStartTime > lastBarTime) then + begin + // A new bar starts, so the previous one is now complete. + if (lastBarTime > 0) then + begin + Result := Broadcast(FCurrentBar); + end; + + // Start a new bar, Volume is 1 because this is the first tick. + FCurrentBar.Data := Value.Data; + FCurrentBar.Time := barStartTime; + end + else + begin + // Update the currently aggregating bar + if Value.Data.High > FCurrentBar.Data.High then + FCurrentBar.Data.High := Value.Data.High; + if Value.Data.Low < FCurrentBar.Data.Low then + FCurrentBar.Data.Low := Value.Data.Low; + FCurrentBar.Data.Close := Value.Data.Close; + // Volume is the number of ticks needed to build the complete bar. + FCurrentBar.Data.Volume := FCurrentBar.Data.Volume + Value.Data.Volume; + end; +end; + end. diff --git a/Src/Myc.Trade.DataPoint.pas b/Src/Myc.Trade.DataPoint.pas index 3e0c07b..7342343 100644 --- a/Src/Myc.Trade.DataPoint.pas +++ b/Src/Myc.Trade.DataPoint.pas @@ -111,7 +111,8 @@ type class function CreateRecordField(const FieldName: String): TConverter; static; class function CreateIdentity: TConverter; static; - class function CreateAggregation(Timeframe: TTimeframe): TConverter, TDataPoint>; static; + class function CreateTickAggregation(Timeframe: TTimeframe): TConverter, TDataPoint>; static; + class function CreateOhlcAggregation(Timeframe: TTimeframe): TConverter, TDataPoint>; static; class function CreateDataPointConverter(const Func: TConstFunc): TConverter, TDataPoint>; static; @@ -251,7 +252,7 @@ begin Result := A.FConverter; end; -class function TConverter.CreateAggregation(Timeframe: TTimeframe): TConverter, TDataPoint>; +class function TConverter.CreateTickAggregation(Timeframe: TTimeframe): TConverter, TDataPoint>; begin Result := TTickAggregation.Create(Timeframe); end; @@ -309,6 +310,11 @@ begin Parent.Link(seq); end; +class function TConverter.CreateOhlcAggregation(Timeframe: TTimeframe): TConverter, TDataPoint>; +begin + Result := TOhlcAggregation.Create(Timeframe); +end; + class function TConverter.Parallel(Parent: TDataProvider): TConverter; begin Result := TMycParallelConverter.Create; diff --git a/Src/Myc.Trade.DataStream.pas b/Src/Myc.Trade.DataStream.pas index 370120e..291f9d0 100644 --- a/Src/Myc.Trade.DataStream.pas +++ b/Src/Myc.Trade.DataStream.pas @@ -34,13 +34,12 @@ type FSymbol: String; FYear: Integer; FMonth: Integer; + FBasename: String; function GetIsValid: Boolean; public - constructor Create(const APath, ASymbol, AExtension: String; AYear, AMonth: Integer); + constructor Create(const APath, ASymbol, AExtension: String; AYear, AMonth: Integer; const ABasename: String); function GetBaseFileName: string; function GetFullFileName: string; - // Gets the next consecutive data file, if it exists on disk. - function GetNextFile: TAuraDataFile; property IsValid: Boolean read GetIsValid; property Extension: String read FExtension; property Path: String read FPath; @@ -54,12 +53,10 @@ type private FCachedFiles: TDataFileCache>>>; FPath: String; - FSymbols: TFuture>; + // The cache now holds a sorted array of all files for each symbol. + FSymbols: TFuture>>; function GetPath: String; - // Used by cache to actually load an uncached file. - function DoLoad(const FileName: string): TFuture>>>; - function ProcessChunks( const DataChunks: TArray>>; Terminated: TState; @@ -78,10 +75,10 @@ type FLoadGate: TLatch; protected - // Read data from stream and return it chunked. - class function ReadCompressedData(const InputStream: TStream): TArray>>; virtual; abstract; - // Read data from stream and return it chunked. - class function ReadUncompressedData(const InputStream: TStream): TArray>>; virtual; abstract; + // Used by cache to actually load an uncached file. Now virtual and abstract. + function DoLoad(const FileName: string; const Gate: TLatch): TFuture>>>; virtual; abstract; + // Gets the next consecutive data file from the cached, sorted list. + function GetNextFile(const Curr: TAuraDataFile): TAuraDataFile; virtual; public constructor Create(const APath: String); @@ -110,12 +107,36 @@ type end; TAuraTABFileServer = class(TAuraDataServer) + private + class function ReadCompressedData(const InputStream: TStream): TArray>>; + class function ReadUncompressedData(const InputStream: TStream): TArray>>; protected - // Scans a directory and returns the oldest file found for each symbol. - class function ReadCompressedData(const InputStream: TStream): TArray>>; override; - class function ReadUncompressedData(const InputStream: TStream): TArray>>; override; + function DoLoad(const FileName: string; const Gate: TLatch): TFuture>>>; override; + public + function ParseFileName(const FileName: string): TAuraDataFile; override; + end; + + // File record for cTrader M1 data + TAuraM1FileItem = packed record + OADateTime: Double; + Open: Int64; + High: Int64; + Low: Int64; + Close: Int64; + Spread: Int64; + TickVolume: Integer; + Digits: Integer; + end; + + // File server for cTrader M1 data. Note that the generic type is TOhlcItem, + // as the server converts the file data into standard OHLC items. + TAuraM1FileServer = class(TAuraDataServer) + private + class function ReadCompressedData(const InputStream: TStream): TArray>>; + class function ReadUncompressedData(const InputStream: TStream): TArray>>; + protected + function DoLoad(const FileName: string; const Gate: TLatch): TFuture>>>; override; public - function LoadDataSeries(const InitialFile: TAuraDataFile): TFuture>>; function ParseFileName(const FileName: string): TAuraDataFile; override; end; @@ -123,6 +144,7 @@ implementation uses System.Zip, + System.ZLib, System.Math, System.StrUtils, Myc.TaskManager; @@ -133,18 +155,19 @@ const { TAuraDataFile } -constructor TAuraDataFile.Create(const APath, ASymbol, AExtension: String; AYear, AMonth: Integer); +constructor TAuraDataFile.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 TAuraDataFile.GetBaseFileName: string; begin - Result := Format('%s_%.4d_%.2d', [FSymbol, FYear, FMonth]); + Result := FBasename; end; function TAuraDataFile.GetFullFileName: string; @@ -157,36 +180,6 @@ begin Result := (FSymbol <> '') and (FYear > 0); end; -function TAuraDataFile.GetNextFile: TAuraDataFile; -var - nextMonth, nextYear: Integer; - nextFile: TAuraDataFile; -begin - if not IsValid then - exit(Default(TAuraDataFile)); - - nextMonth := FMonth + 1; - nextYear := FYear; - if (nextMonth > 12) then - begin - nextMonth := 1; - Inc(nextYear); - end; - - // Probe for zipped file first - nextFile := TAuraDataFile.Create(FPath, FSymbol, '.tab_zip', nextYear, nextMonth); - if TFile.Exists(nextFile.GetFullFileName) then - exit(nextFile); - - // Probe for uncompressed file - nextFile := TAuraDataFile.Create(FPath, FSymbol, '.tab', nextYear, nextMonth); - if TFile.Exists(nextFile.GetFullFileName) then - exit(nextFile); - - // No next file found - Result := Default(TAuraDataFile); -end; - { TAuraDataServer } constructor TAuraDataServer.Create(const APath: String); @@ -195,8 +188,13 @@ begin FPath := APath; FCachedFiles := - TDataFileCache>>> - .Create(function(const Filename: String): TFuture>>> begin Result := DoLoad(Filename); end); + TDataFileCache>>>.Create( + function(const Filename: String): TFuture>>> + begin + // Pass the private load gate to the virtual DoLoad method. + Result := DoLoad(Filename, FLoadGate); + end + ); end; destructor TAuraDataServer.Destroy; @@ -217,92 +215,61 @@ begin FCachedFiles.Clear; end; -function TAuraDataServer.DoLoad(const FileName: string): TFuture>>>; -begin - Result := TFuture>>>.Null; - - if not TFile.Exists(FileName) then - exit; - - var capFileName := FileName; - if FileName.EndsWith('_zip', True) then - begin - Result := - TFuture - .Construct(FLoadGate.Enqueue, function: TBytes begin Result := TFile.ReadAllBytes(capFileName); end) - .Chain>>>( - function(bytes: TBytes): TArray>> - begin - var zipMemoryStream := TBytesStream.Create(bytes); - try - zipMemoryStream.Position := 0; - Result := ReadCompressedData(zipMemoryStream); - finally - zipMemoryStream.Free; - end; - end); - end - else - begin - Result := - TFuture>>>.Construct( - FLoadGate.Enqueue, - function: TArray>> - begin - if TFile.Exists(capFileName) then - begin - var stream := TFileStream.Create(capFileName, fmOpenRead or fmShareDenyWrite); - try - try - Result := ReadUncompressedData(stream); - except - on E: EReadError do - raise EReadError.CreateFmt('File %s: %s', [capFileName, E.Message]); - end; - finally - stream.Free; - end; - end; - end - ); - end; -end; - procedure TAuraDataServer.UpdateSymbols; begin FSymbols := - TFuture>.Construct( + TFuture>>.Construct( FSymbols.Done, - function: TDictionary + function: TDictionary> var fileNames: TArray; - currentFile: string; + tempSymbolMap: TDictionary>; dataFile: TAuraDataFile; - trackedInfo: TAuraDataFile; + symbolList: TList; begin - Result := TDictionary.Create; - if not TDirectory.Exists(FPath) then - exit; + Result := TDictionary>.Create; + tempSymbolMap := TDictionary>.Create; + try + if not TDirectory.Exists(FPath) then + exit; - fileNames := TDirectory.GetFiles(FPath); - for currentFile in fileNames do - begin - dataFile := ParseFileName(currentFile); - if dataFile.IsValid then + fileNames := TDirectory.GetFiles(FPath); + for var currentFile in fileNames do begin - if Result.TryGetValue(dataFile.Symbol, trackedInfo) then + dataFile := ParseFileName(currentFile); + if dataFile.IsValid then begin - if (dataFile.Year < trackedInfo.Year) - or ((dataFile.Year = trackedInfo.Year) and (dataFile.Month < trackedInfo.Month)) then + if not tempSymbolMap.TryGetValue(dataFile.Symbol, symbolList) then begin - Result.AddOrSetValue(dataFile.Symbol, dataFile); + symbolList := TList.Create; + tempSymbolMap.Add(dataFile.Symbol, symbolList); end; - end - else - begin - Result.Add(dataFile.Symbol, dataFile); + symbolList.Add(dataFile); end; end; + + for var kvp in tempSymbolMap do + begin + symbolList := kvp.Value; + symbolList.Sort( + TComparer.Construct( + function(const Left, Right: TAuraDataFile): 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 ); @@ -314,7 +281,7 @@ function TAuraDataServer.EnumerateSymbols: TFuture>; begin Result := FSymbols.Chain>( - function(Symbols: TDictionary): TArray begin Result := Symbols.Keys.ToArray; end + function(const Symbols: TDictionary>): TArray begin Result := Symbols.Keys.ToArray; end ); end; @@ -323,14 +290,46 @@ begin var symName := Symbol; Result := FSymbols.Chain( - function(Symbols: TDictionary): TAuraDataFile + function(const Symbols: TDictionary>): TAuraDataFile + var + symbolFiles: TArray; begin - if not Symbols.TryGetValue(symName, Result) then + if Symbols.TryGetValue(symName, symbolFiles) and (Length(symbolFiles) > 0) then + Result := symbolFiles[0] + else Result := Default(TAuraDataFile); end ); end; +function TAuraDataServer.GetNextFile(const Curr: TAuraDataFile): TAuraDataFile; +var + allSymbolFiles: TArray; + foundIndex: Integer; +begin + Result := Default(TAuraDataFile); + 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 TAuraDataServer.GetPath: String; begin Result := FPath; @@ -357,7 +356,9 @@ end; function TAuraDataServer.LoadDataFile(const DataFile: TAuraDataFile): TFuture>>>; begin if DataFile.IsValid then - Result := FCachedFiles.GetOrAdd(DataFile.GetFullFileName); + Result := FCachedFiles.GetOrAdd(DataFile.GetFullFileName) + else + Result := TFuture>>>.Construct(TArray>>(nil)); end; function TAuraDataServer.ProcessChunks( @@ -397,7 +398,7 @@ begin exit(DataFile.Done); // Read ahead the next file, while processing the current one - var nextFileInfo := FileInfo.GetNextFile; + var nextFileInfo := GetNextFile(FileInfo); var nextFile := LoadDataFile(nextFileInfo); var cTerminated := Terminated; @@ -416,90 +417,52 @@ begin ); end; -function TAuraTABFileServer.LoadDataSeries(const InitialFile: TAuraDataFile): TFuture>>; -var - loadedState: TState; - loadedFiles: TArray>>>>; - liveData: TFuture>>>; - tabFiles: TList; - liveFile: TAuraDataFile; - currentFileInfo: TAuraDataFile; +{ TAuraTABFileServer } + +function TAuraTABFileServer.DoLoad(const FileName: string; const Gate: TLatch): TFuture>>>; begin - tabFiles := TList.Create; - try - currentFileInfo := InitialFile; - while currentFileInfo.IsValid do - begin - tabFiles.Add(currentFileInfo); - currentFileInfo := currentFileInfo.GetNextFile; - end; + Result := TFuture>>>.Null; - liveFile := Default(TAuraDataFile); - if tabFiles.Count > 0 then - begin - var lastTabFile := tabFiles.Last; - var liveFileBaseName := Format('%s_%d_%02d.tab-live', [lastTabFile.Symbol, lastTabFile.Year, lastTabFile.Month]); - var potentialLivePath := TPath.Combine(lastTabFile.Path, liveFileBaseName); - if TFile.Exists(potentialLivePath) then - liveFile := ParseFileName(potentialLivePath); - end; + if not TFile.Exists(FileName) then + exit; - var loadedFileList := TList>>>>.Create; - var loadStates := TList.Create; - try - for var fileInfo in tabFiles do - begin - var data := LoadDataFile(fileInfo); - loadedFileList.Add(data); - loadStates.Add(data.Done); - end; - - liveData := TFuture>>>.Null; - if liveFile.IsValid then - begin - liveData := LoadDataFile(liveFile); - loadedFileList.Add(liveData); - loadStates.Add(liveData.Done); - end; - - loadedState := TState.All(loadStates.ToArray); - loadedFiles := loadedFileList.ToArray; - finally - loadStates.Free; - loadedFileList.Free; - end; - finally - tabFiles.Free; + var capFileName := FileName; + if FileName.EndsWith('_zip', True) then + begin + Result := + TFuture + .Construct(Gate.Enqueue, function: TBytes begin Result := TFile.ReadAllBytes(capFileName); end) + .Chain>>>( + function(bytes: TBytes): TArray>> + begin + var compMemoryStream := TBytesStream.Create(bytes); + try + compMemoryStream.Position := 0; + Result := ReadCompressedData(compMemoryStream); + finally + compMemoryStream.Free; + end; + end); + end + else + begin + Result := + TFuture>>>.Construct( + Gate.Enqueue, + function: TArray>> + begin + if TFile.Exists(capFileName) then + begin + var stream := TFileStream.Create(capFileName, fmOpenRead or fmShareDenyWrite); + try + Result := ReadUncompressedData(stream); + finally + stream.Free; + end; + end; + end + ); end; - - Result := - TFuture>>.Construct( - loadedState, - function: TArray> - begin - var tickList := TList>.Create; - try - var overallLastTabTickTime: TDateTime := 0; - var cnt := 0; - for var P in loadedFiles do - for var chunk in P.Value do - inc(cnt, Length(chunk)); - tickList.Capacity := cnt; - - for var P in loadedFiles do - for var chunk in P.Value do - for var iterTickData in chunk do - if overallLastTabTickTime < iterTickData.Time then - begin - tickList.Add(iterTickData); - overallLastTabTickTime := iterTickData.Time; - end; - Result := tickList.ToArray; - finally - tickList.Free; - end; - end - ); end; function TAuraTABFileServer.ParseFileName(const FileName: string): TAuraDataFile; @@ -546,7 +509,7 @@ begin if symbol = '' then exit; - Result := TAuraDataFile.Create(path, symbol, fileExt, year, month); + Result := TAuraDataFile.Create(path, symbol, fileExt, year, month, baseName); end; class function TAuraTABFileServer.ReadCompressedData(const InputStream: TStream): TArray>>; @@ -628,7 +591,193 @@ begin if bytesRead <> SizeOf(TFileRecord) then raise EReadError.CreateFmt('Read error. Expected %d bytes, read %d.', [SizeOf(TFileRecord), bytesRead]); - Result[chunkIndex][indexInChunk].Create(rec.TimeStamp, rec.Data); + // Corrected syntax: Assign the newly created record instance. + Result[chunkIndex][indexInChunk] := TDataPoint.Create(rec.TimeStamp, rec.Data); + Inc(indexInChunk); + end; + end; +end; + +{ TAuraM1FileServer } + +function TAuraM1FileServer.DoLoad(const FileName: string; const Gate: TLatch): TFuture>>>; +begin + Result := TFuture>>>.Null; + + if not TFile.Exists(FileName) then + exit; + + var capFileName := FileName; + // M1 files are always compressed with the .m1 extension + if FileName.EndsWith('.m1', True) then + begin + Result := + TFuture + .Construct(Gate.Enqueue, function: TBytes begin Result := TFile.ReadAllBytes(capFileName); end) + .Chain>>>( + function(bytes: TBytes): TArray>> + begin + var compMemoryStream := TBytesStream.Create(bytes); + try + compMemoryStream.Position := 0; + Result := ReadCompressedData(compMemoryStream); + finally + compMemoryStream.Free; + end; + end); + end; +end; + +function TAuraM1FileServer.ParseFileName(const FileName: string): TAuraDataFile; +var + path, fileNameNoPath, baseName, ext, symbol: string; + year, month: Integer; + parts, dateParts: TArray; +begin + Result := Default(TAuraDataFile); + path := TPath.GetDirectoryName(FileName); + fileNameNoPath := TPath.GetFileName(FileName); + + // M1 files are compressed and have a .m1 extension + if not fileNameNoPath.EndsWith('.m1', True) then + exit; + + ext := TPath.GetExtension(fileNameNoPath); + baseName := TPath.GetFileNameWithoutExtension(fileNameNoPath); + + // Expected format: SYMBOL_YYYY-MM_YYYY-MM + parts := baseName.Split(['_']); + if Length(parts) < 3 then + exit; // Not enough parts for SYMBOL_START_END + + // The symbol can contain underscores, so we take everything before the last two parts. + var startDatePart := parts[High(parts) - 1]; + + dateParts := startDatePart.Split(['-']); + if Length(dateParts) <> 2 then + exit; // Start date is not YYYY-MM + + 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; + + // The TAuraDataFile only represents the start date of the file batch. + Result := TAuraDataFile.Create(path, symbol, ext, year, month, baseName); +end; + +class function TAuraM1FileServer.ReadCompressedData(const InputStream: TStream): TArray>>; +var + decompressionStream: TStream; + localHeader: TZipHeader; + entryIndex, i: Integer; + zipFileInstance: TZipFile; +begin + Result := nil; + decompressionStream := nil; + zipFileInstance := nil; + + // M1 files are now zip archives containing the actual data file. + if not Assigned(InputStream) or (InputStream.Size = 0) then + exit; + + try + InputStream.Position := 0; + zipFileInstance := TZipFile.Create; + zipFileInstance.Open(InputStream, TZipMode.zmRead); + + if zipFileInstance.FileCount = 0 then + exit; + + // Find the data file entry within the archive. + // The C# code creates an entry with the same name as the base filename, ending in .m1. + entryIndex := -1; + for i := 0 to zipFileInstance.FileCount - 1 do + begin + if zipFileInstance.FileNames[i].EndsWith('.m1', True) then + begin + entryIndex := i; + break; + end; + end; + + // Fallback to the first entry if no specific .m1 file is found inside. + if entryIndex = -1 then + entryIndex := 0; + + zipFileInstance.Read(entryIndex, decompressionStream, localHeader); + if not Assigned(decompressionStream) then + exit; + + // The decompressed stream now contains the raw binary data. + Result := ReadUncompressedData(decompressionStream); + finally + decompressionStream.Free; + zipFileInstance.Free; + end; +end; + +class function TAuraM1FileServer.ReadUncompressedData(const InputStream: TStream): TArray>>; +type + TFileRecord = TAuraM1FileItem; +var + fileSize: Int64; + totalRecordCount, bytesRead, chunkIndex, indexInChunk, recordIndex: Integer; + rec: TFileRecord; + ohlc: TOhlcItem; + factor: Double; +begin + Result := nil; + InputStream.Position := 0; + fileSize := InputStream.Size; + if (fileSize = 0) or ((fileSize mod SizeOf(TFileRecord)) <> 0) then + exit; + + totalRecordCount := fileSize div SizeOf(TFileRecord); + if totalRecordCount > 0 then + begin + var numChunks := (totalRecordCount + ChunkSize - 1) div ChunkSize; + SetLength(Result, numChunks); + chunkIndex := 0; + indexInChunk := 0; + SetLength(Result[chunkIndex], Min(ChunkSize, totalRecordCount)); + + for recordIndex := 0 to totalRecordCount - 1 do + begin + if indexInChunk >= ChunkSize then + begin + Inc(chunkIndex); + indexInChunk := 0; + var remainingRecords := totalRecordCount - (chunkIndex * ChunkSize); + SetLength(Result[chunkIndex], Min(ChunkSize, remainingRecords)); + end; + + bytesRead := InputStream.Read(rec, SizeOf(TFileRecord)); + if bytesRead <> SizeOf(TFileRecord) then + raise EReadError.CreateFmt('Read error. Expected %d bytes, read %d.', [SizeOf(TFileRecord), bytesRead]); + + // Convert from M1 file record to a standard OHLC item + if rec.Digits > 0 then + factor := Power(10, rec.Digits) + else + factor := 1.0; + + ohlc.Open := rec.Open / factor; + ohlc.High := rec.High / factor; + ohlc.Low := rec.Low / factor; + ohlc.Close := rec.Close / factor; + ohlc.Volume := rec.TickVolume; + + // Corrected syntax: Assign the newly created record instance. + Result[chunkIndex][indexInChunk] := TDataPoint.Create(rec.OADateTime, ohlc); Inc(indexInChunk); end; end;