diff --git a/AuraTrader/FirstStrategy.pas b/AuraTrader/FirstStrategy.pas index 079fcbd..a9a546f 100644 --- a/AuraTrader/FirstStrategy.pas +++ b/AuraTrader/FirstStrategy.pas @@ -24,7 +24,7 @@ type protected // Broadcasts the given data to all linked processors. procedure Broadcast(const Value: T); - procedure Update; override; + public constructor Create; destructor Destroy; override; @@ -42,33 +42,21 @@ type constructor Create(const AFunc: TConvertFunc); end; - ITicksToTimeframe = interface(IMycConverter>, TDataPoint>) - function GetCurrentBar: TDataPoint; - function GetStateText: TWriteable; - function GetTimeframe: TTimeframe; - property CurrentBar: TDataPoint read GetCurrentBar; - property StateText: TWriteable read GetStateText; - property Timeframe: TTimeframe read GetTimeframe; - end; - - TTicksToTimeframe = class(TMycConverter>, TDataPoint>, ITicksToTimeframe) + TTicksToTimeframe = class(TMycConverter>, TDataPoint>) private FTimeframe: TTimeframe; - FStateText: TWriteable; // Stores the currently aggregating OHLC data. FCurrentBar: TDataPoint; function GetBarStartTime(const TimeStamp: TDateTime; const Timeframe: TTimeframe): TDateTime; function GetCurrentBar: TDataPoint; - function GetStateText: TWriteable; function GetTimeframe: TTimeframe; public - constructor Create(const ATimeframe: TTimeframe; const AStateText: TWriteable); + constructor Create(const ATimeframe: TTimeframe); // Process new data. This is called concurrently and must not have side effects out of the scope of this class! function ProcessData(const Values: TArray>): Boolean; override; property CurrentBar: TDataPoint read GetCurrentBar; - property StateText: TWriteable read GetStateText; property Timeframe: TTimeframe read GetTimeframe; end; @@ -99,10 +87,9 @@ uses { TTicksToTimeframe } -constructor TTicksToTimeframe.Create(const ATimeframe: TTimeframe; const AStateText: TWriteable); +constructor TTicksToTimeframe.Create(const ATimeframe: TTimeframe); begin inherited Create; - FStateText := AStateText; FTimeframe := ATimeframe; end; @@ -124,11 +111,6 @@ begin Result := FCurrentBar; end; -function TTicksToTimeframe.GetStateText: TWriteable; -begin - Result := FStateText; -end; - function TTicksToTimeframe.GetTimeframe: TTimeframe; begin Result := FTimeframe; @@ -178,10 +160,6 @@ begin FCurrentBar.Data := currentBar; end; end; - - with FCurrentBar do - FStateText.Value := - Format('Cuur Bar: O:%.5f H:%.5f L:%.5f C:%.5f V:%.0f', [Data.Open, Data.High, Data.Low, Data.Close, Data.Volume]); end; { THullMovingAverage } @@ -302,15 +280,4 @@ begin Result := FSender; end; -procedure TMycConverter.Update; -begin - FSender.Notify( - function(const Processor: IMycProcessor): Boolean - begin - Processor.Update; - Result := true - end - ); -end; - end. diff --git a/AuraTrader/MainForm.pas b/AuraTrader/MainForm.pas index 7131e9d..e20327c 100644 --- a/AuraTrader/MainForm.pas +++ b/AuraTrader/MainForm.pas @@ -156,12 +156,6 @@ begin if Symbol = '' then exit; - var stateText := TWriteable.CreateWriteable.Protect; - var stateLabel := TLabel.Create(Self); - AlignControl(stateLabel); - - stateLabel.Text := 'Initializing...'; - var chart := TMycChart.Create(Self); AlignControl(chart); chart.Height := Layout.ChildrenRect.Width * 9 / 16; @@ -169,7 +163,7 @@ begin ///// - var OhlcPoint: ITicksToTimeframe := TTicksToTimeframe.Create(H1, stateText); + var OhlcPoint: IMycConverter>, TDataPoint> := TTicksToTimeframe.Create(M1); var Ohlc: IMycConverter, TOhlcItem> := TMycGenericConverter, TOhlcItem> @@ -182,18 +176,23 @@ begin Ohlc.Sender.Link(Closes); - var Hull: IMycConverter := THullMovingAverage.Create(250); + for var i := 0 to 200 do + begin + var Hull: IMycConverter := THullMovingAverage.Create(20 + (20 * i)); + Closes.Sender.Link(Hull); + chart.AddDoubleSeries(Hull.Sender, Random(Cardinal.MaxValue)); + end; - Closes.Sender.Link(Hull); - - chart.AddDoubleSeries(Hull.Sender); + // var Hull: IMycConverter := THullMovingAverage.Create(250); + // + // Closes.Sender.Link(Hull); + // + // chart.AddDoubleSeries(Hull.Sender); var done := ExecuteStrategy(Symbol, OhlcPoint); ///// - stateLabel.ProcessSignal(stateText.Changed, done, procedure begin stateLabel.Text := stateText.Value; end); - FProcessDone := TState.All([FProcessDone, done]); chart.AddOhlcSeries(OhlcPoint.Sender); diff --git a/AuraTrader/Myc.Fmx.Chart.pas b/AuraTrader/Myc.Fmx.Chart.pas index 96f06b9..eb56980 100644 --- a/AuraTrader/Myc.Fmx.Chart.pas +++ b/AuraTrader/Myc.Fmx.Chart.pas @@ -31,9 +31,10 @@ type function GetMainSeries: TSeries; protected function GetCount: Int64; virtual; abstract; + function GetTotalCount: Int64; virtual; abstract; function GetValueRange(StartIndex, Count: Int64; out Min, Max: Double): Boolean; virtual; abstract; function Update: Boolean; virtual; abstract; - // The signature is changed to support viewport painting + // The signature is changed to support viewport painting with an index offset procedure Paint( const ACanvas: TCanvas; StartIndex, Count: Int64; @@ -44,6 +45,7 @@ type constructor Create(AOwner: TMycChart); property Count: Int64 read GetCount; property Owner: TMycChart read FOwner; + property TotalCount: Int64 read GetTotalCount; end; private @@ -68,20 +70,20 @@ type constructor Create(AOwner: TComponent); override; destructor Destroy; override; - // Creates an OHLC candlestick/bar series - procedure AddOhlcSeries( + // Creates an OHLC candlestick/bar series and returns it + function AddOhlcSeries( const DataProvider: IMycDataProvider>; const AUpColor: TAlphaColor = TAlphaColors.Green; const ADownColor: TAlphaColor = TAlphaColors.Red; const AStyle: TCandleStyle = csCandleStick - ); + ): TSeries; - // Creates a simple line series for double values - procedure AddDoubleSeries( + // Creates a simple line series for double values and returns it + function AddDoubleSeries( const DataProvider: IMycDataProvider; const ALineColor: TAlphaColor = TAlphaColors.Cornflowerblue; const ALineWidth: Single = 1.5 - ); + ): TSeries; property Lookback: TWriteable read FLookback write FLookback; end; @@ -103,7 +105,6 @@ type FData: TWriteable>; protected function ProcessData(const Value: T): Boolean; override; - procedure Update; override; public constructor Create(const ALookback: TWriteable); function GetData(var Data: TMycDataArray): Boolean; @@ -121,6 +122,7 @@ type FReceiverTag: TTag; protected function GetCount: Int64; override; + function GetTotalCount: Int64; override; function Update: Boolean; override; public @@ -205,29 +207,25 @@ begin inherited; end; -procedure TMycChart.AddDoubleSeries( +function TMycChart.AddDoubleSeries( const DataProvider: IMycDataProvider; const ALineColor: TAlphaColor = TAlphaColors.Cornflowerblue; const ALineWidth: Single = 1.5 -); -var - series: TChartLineSeries; +): TSeries; begin - series := TChartLineSeries.Create(Self, DataProvider, ALineColor, ALineWidth); - FSeriesList.Add(series); + Result := TChartLineSeries.Create(Self, DataProvider, ALineColor, ALineWidth); + FSeriesList.Add(Result); end; -procedure TMycChart.AddOhlcSeries( +function TMycChart.AddOhlcSeries( const DataProvider: IMycDataProvider>; const AUpColor: TAlphaColor = TAlphaColors.Green; const ADownColor: TAlphaColor = TAlphaColors.Red; const AStyle: TCandleStyle = csCandleStick -); -var - series: TChartOhlcSeries; +): TSeries; begin - series := TChartOhlcSeries.Create(Self, DataProvider, AUpColor, ADownColor, AStyle); - FSeriesList.Add(series); + Result := TChartOhlcSeries.Create(Self, DataProvider, AUpColor, ADownColor, AStyle); + FSeriesList.Add(Result); end; procedure TMycChart.DoIdle; @@ -235,22 +233,18 @@ var doRepaint: Boolean; mainSeries: TSeries; isLiveView: Boolean; - prevCount, newCount, newCandles: Int64; + prevTotalCount, newTotalCount, countDelta: Int64; begin if (FSeriesList.Count = 0) then - begin exit; - end; mainSeries := FSeriesList[0]; if (mainSeries = nil) then - begin exit; - end; // Remember state before update isLiveView := (FViewStartIndex = 0); - prevCount := mainSeries.Count; + prevTotalCount := mainSeries.TotalCount; // Check all series for updates doRepaint := false; @@ -262,20 +256,20 @@ begin end; end; - if doRepaint then + if doRepaint and (not isLiveView) then begin - newCount := mainSeries.Count; - newCandles := newCount - prevCount; + newTotalCount := mainSeries.TotalCount; + countDelta := newTotalCount - prevTotalCount; - // Adjust viewport only if user was not watching the live data - if (not isLiveView) and (newCandles > 0) then + if (countDelta > 0) then begin - FViewStartIndex := FViewStartIndex + newCandles; + // Adjust the start index based on the absolute change in total data points + FViewStartIndex := FViewStartIndex + countDelta; end; - // If the view was live, it implicitly stays at StartIndex = 0, showing the new data. - - Repaint; end; + + if doRepaint then + Repaint; end; procedure TMycChart.MouseDown(Button: TMouseButton; Shift: TShiftState; X, Y: Single); @@ -385,7 +379,9 @@ begin if (FViewStartIndex > maxIndex) then FViewStartIndex := maxIndex; - FDragStartPoint := TPointF.Create(X, Y); + // Optimized point update + FDragStartPoint.X := X; + FDragStartPoint.Y := Y; Repaint; end; end; @@ -452,6 +448,24 @@ begin end; procedure TMycChart.Paint; + + function CalcView(Series: TSeries; MasterCount: Int64; out First: Int64; out Count: Int64): Boolean; + begin + var seriesTotalCount := series.TotalCount; + var offset := MasterCount - seriesTotalCount; + var startIdx := FViewStartIndex - offset; + + var effectiveStart := Max(0, startIdx); + var effectiveEnd := Min(series.Count, startIdx + FViewCount) - 1; + + Result := effectiveStart <= effectiveEnd - 1; + if Result then + begin + First := effectiveStart; + Count := effectiveEnd - effectiveStart + 1; + end; + end; + var rect: TRectF; series: TSeries; @@ -462,6 +476,8 @@ var isButtonVisible: Boolean; buttonColor: TAlphaColor; path: TPathData; + mainSeries: TSeries; + masterTotalCount, seriesStart, seriesCount: Int64; begin inherited; rect := Self.LocalRect; @@ -474,22 +490,26 @@ begin end; rangeInitialized := false; + mainSeries := FSeriesList[0]; - // Determine value range for the visible viewport only + masterTotalCount := mainSeries.TotalCount; for series in FSeriesList do begin - if series.GetValueRange(FViewStartIndex, FViewCount, seriesMin, seriesMax) then + if CalcView(series, masterTotalCount, seriesStart, seriesCount) then begin - if not rangeInitialized then + if series.GetValueRange(seriesStart, seriesCount, seriesMin, seriesMax) then begin - globalMin := seriesMin; - globalMax := seriesMax; - rangeInitialized := true; - end - else - begin - globalMin := Min(globalMin, seriesMin); - globalMax := Max(globalMax, seriesMax); + if not rangeInitialized then + begin + globalMin := seriesMin; + globalMax := seriesMax; + rangeInitialized := true; + end + else + begin + globalMin := Min(globalMin, seriesMin); + globalMax := Max(globalMax, seriesMax); + end; end; end; end; @@ -504,29 +524,29 @@ begin if (globalMax - globalMin) = 0 then exit; - // Transform maps an index from the viewport to a screen coordinate - xTransform := - function(index: Double): Single - begin - if (FViewCount <= 1) then - begin - Result := rect.Left + rect.Width / 2; - exit; - end; - Result := rect.Right - (((index - FViewStartIndex) / (FViewCount - 1)) * rect.Width); - end; - + // Base transform maps an index from the viewport to a screen coordinate yTransform := function(value: Double): Single begin Result := rect.Top + (1 - (value - globalMin) / (globalMax - globalMin)) * rect.Height; end; + // --- Draw Series with Offset Compensation --- + // The masterTotalCount is already calculated from the range finding loop for series in FSeriesList do begin - series.Paint(Self.Canvas, FViewStartIndex, FViewCount, xTransform, yTransform); + if CalcView(series, masterTotalCount, seriesStart, seriesCount) then + begin + var offset := masterTotalCount - series.TotalCount; + xTransform := + function(index: Double): Single + begin + Result := rect.Right - (((index + offset - FViewStartIndex) / (FViewCount - 1)) * rect.Width); + end; + + series.Paint(Self.Canvas, seriesStart, seriesCount, xTransform, yTransform); + end; end; // --- Draw the "jump to latest" button --- isButtonVisible := (FViewStartIndex > 0); - FJumpButtonHot := FJumpButtonHot and isButtonVisible; if isButtonVisible then begin @@ -545,7 +565,7 @@ begin Canvas.Fill.Color := buttonColor; Canvas.FillRect(FJumpButtonRect, 4, 4, AllCorners, 0.7); - // Draw an icon (e.g., a "fast forward" double arrow) + // Draw an icon Canvas.Stroke.Color := TAlphaColors.White; Canvas.Stroke.Thickness := 1.5; @@ -553,11 +573,9 @@ begin try var cx := FJumpButtonRect.CenterPoint.X; var cy := FJumpButtonRect.CenterPoint.Y; - // First arrow path.MoveTo(PointF(cx - 5, cy - 6)); path.LineTo(PointF(cx, cy)); path.LineTo(PointF(cx - 5, cy + 6)); - // Second arrow path.MoveTo(PointF(cx + 2, cy - 6)); path.LineTo(PointF(cx + 7, cy)); path.LineTo(PointF(cx + 2, cy + 6)); @@ -574,6 +592,30 @@ begin end; end; +{ TChartSeriesReceiver } + +constructor TChartSeriesReceiver.Create(const ALookback: TWriteable); +begin + inherited Create; + FLookback := ALookback; + FCurrData := TMycDataArray.CreateEmpty; + FData := TWriteable>.CreateWriteable(FCurrData).Protect; +end; + +function TChartSeriesReceiver.GetData(var Data: TMycDataArray): Boolean; +begin + var prevTotalCount := Data.TotalCount; + Data := FData.Value; + Result := prevTotalCount <> Data.TotalCount; +end; + +function TChartSeriesReceiver.ProcessData(const Value: T): Boolean; +begin + Result := true; + FCurrData := FCurrData.Add(Value, FLookback.Value); + FData.Value := FCurrData; +end; + { TChartSeriesProcessor } constructor TChartSeriesProcessor.Create(AOwner: TMycChart; const ADataProvider: IMycDataProvider); @@ -598,6 +640,11 @@ begin Result := FData.Count; end; +function TChartSeriesProcessor.GetTotalCount: Int64; +begin + Result := FData.TotalCount; +end; + function TChartSeriesProcessor.Update: Boolean; begin Result := FReceiver.GetData(FData); @@ -621,17 +668,30 @@ end; function TChartOhlcSeries.GetValueRange(StartIndex, Count: Int64; out MinValue, MaxValue: Double): Boolean; var i: Int64; - lastIndex: Int64; + seriesCount: Int64; + effectiveStart, effectiveEnd: Int64; begin - Result := GetCount > 0; + seriesCount := GetCount; + // Ensure the request is valid and there's data to check + Result := (seriesCount > 0) and (Count > 0); if not Result then Exit; + // Calculate the effective range that is actually available within this series's data + effectiveStart := Max(0, StartIndex); + effectiveEnd := Min(seriesCount - 1, StartIndex + Count - 1); + + // If the entire requested range is outside our available data, there's nothing to do. + if (effectiveStart > effectiveEnd) then + begin + Result := false; + Exit; + end; + MinValue := MaxDouble; MaxValue := -MaxDouble; - lastIndex := Min(GetCount - 1, StartIndex + Count - 1); - for i := StartIndex to lastIndex do + for i := effectiveStart to effectiveEnd do begin var dp := Data.Items[i]; MinValue := Min(MinValue, dp.Data.Low); @@ -704,16 +764,29 @@ end; function TChartLineSeries.GetValueRange(StartIndex, Count: Int64; out MinValue, MaxValue: Double): Boolean; var i: Int64; - lastIndex: Int64; + seriesCount: Int64; + effectiveStart, effectiveEnd: Int64; begin - Result := GetCount > 0; + seriesCount := GetCount; + // Ensure the request is valid and there's data to check + Result := (seriesCount > 0) and (Count > 0); if not Result then Exit; + // Calculate the effective range that is actually available within this series's data + effectiveStart := Max(0, StartIndex); + effectiveEnd := Min(seriesCount - 1, StartIndex + Count - 1); + + // If the entire requested range is outside our available data, there's nothing to do. + if (effectiveStart > effectiveEnd) then + begin + Result := false; + Exit; + end; + MinValue := MaxDouble; MaxValue := -MaxDouble; - lastIndex := Min(Data.Count - 1, StartIndex + Count - 1); - for i := StartIndex to lastIndex do + for i := effectiveStart to effectiveEnd do begin var v := Data.Items[i]; if not IsNaN(v) then @@ -764,32 +837,4 @@ begin end; end; -{ TChartSeriesReceiver } - -constructor TChartSeriesReceiver.Create(const ALookback: TWriteable); -begin - inherited Create; - FLookback := ALookback; - FCurrData := TMycDataArray.CreateEmpty; - FData := TWriteable>.CreateWriteable(FCurrData).Protect; -end; - -function TChartSeriesReceiver.GetData(var Data: TMycDataArray): Boolean; -begin - var prevTotalCount := Data.TotalCount; - Data := FData.Value; - Result := prevTotalCount <> Data.TotalCount; -end; - -function TChartSeriesReceiver.ProcessData(const Value: T): Boolean; -begin - Result := true; - FCurrData := FCurrData.Add(Value, FLookback.Value); -end; - -procedure TChartSeriesReceiver.Update; -begin - FData.Value := FCurrData; -end; - end. diff --git a/Src/Myc.Trade.DataPoint.pas b/Src/Myc.Trade.DataPoint.pas index f7464db..5673b76 100644 --- a/Src/Myc.Trade.DataPoint.pas +++ b/Src/Myc.Trade.DataPoint.pas @@ -32,13 +32,11 @@ type IMycProcessor = interface function ProcessData(const Value: T): Boolean; - procedure Update; end; TMycProcessor = class abstract(TInterfacedObject, IMycProcessor) protected function ProcessData(const Value: T): Boolean; virtual; abstract; - procedure Update; virtual; end; TTag = Pointer; @@ -79,7 +77,6 @@ type private FProc: TProc; function ProcessData(const Value: T): Boolean; - procedure Update; public constructor Create(const Controller: IInterface; const AProc: TProc); end; @@ -467,11 +464,6 @@ begin Result := FProc(Value); end; -procedure TMycContainedProcessor.Update; -begin - -end; - { TMycDataProvider } constructor TMycDataProvider.Create(const Controller: IInterface); @@ -516,9 +508,4 @@ begin end; end; -procedure TMycProcessor.Update; -begin - -end; - end. diff --git a/Src/Myc.Trade.DataStream.pas b/Src/Myc.Trade.DataStream.pas index f9cbff2..8d53ee8 100644 --- a/Src/Myc.Trade.DataStream.pas +++ b/Src/Myc.Trade.DataStream.pas @@ -91,17 +91,17 @@ type // Generic server for loading and managing sequential time-series data from files. TAuraDataServer = class(TInterfacedObject, IDataServer) private - FCachedFiles: TDataFileCache>>; + FCachedFiles: TDataFileCache>>>; FPath: String; FSymbols: TFuture>; function GetPath: String; // Used by cache to actually load an uncached file. - function DoLoad(const FileName: string): TFuture>>; + function DoLoad(const FileName: string): TFuture>>>; function ProcessFile( FileInfo: TAuraDataFile; - const DataFile: TFuture>>; + const DataFile: TFuture>>>; const Terminated: TState; Processor: IMycProcessor>> ): TState; @@ -111,8 +111,10 @@ type FLoadGate: TLatch; protected - class function ReadCompressedData(const InputStream: TStream): TArray>; virtual; abstract; - class function ReadUncompressedData(const InputStream: TStream): TArray>; virtual; abstract; + // 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; public constructor Create(const APath: String); @@ -125,7 +127,9 @@ type function FindFirstFile(const Symbol: string): TFuture; function ParseFileName(const FileName: string): TAuraDataFile; virtual; abstract; function EnumerateSymbols: TFuture>; - function LoadDataFile(const DataFile: TAuraDataFile): TFuture>>; + + // Load data file and split content into chunks. + function LoadDataFile(const DataFile: TAuraDataFile): TFuture>>>; function ProcessData(const Symbol: String; const Terminated: TState; const Processor: IMycProcessor>>): TState; @@ -138,7 +142,7 @@ type type TDataStream = record FileInfo: TAuraDataFile; - Data: TFuture>>; + Data: TFuture>>>; end; private FDataServer: TAuraDataServer; @@ -147,6 +151,7 @@ type FNextReady: TFlag; FCurrent: TFuture; FNext: TFuture; + FCurrChunkInFile: Int64; FCurrPosInFile: Int64; FLastTimeStamp: TDateTime; procedure PreloadNext; @@ -172,8 +177,8 @@ type TAuraTABFileServer = class(TAuraDataServer) 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; + class function ReadCompressedData(const InputStream: TStream): TArray>>; override; + class function ReadUncompressedData(const InputStream: TStream): TArray>>; override; public function CreateStream(const Symbol: String): IDataStream; override; function LoadDataSeries(const InitialFile: TAuraDataFile): TFuture>>; @@ -188,6 +193,10 @@ uses System.StrUtils, Myc.TaskManager; +const + // The number of records per data chunk when loading files. + ChunkSize = 8192; + { TAuraDataFile } constructor TAuraDataFile.Create(const APath, ASymbol, AExtension: String; AYear, AMonth: Integer); @@ -252,8 +261,8 @@ begin FPath := APath; FCachedFiles := - TDataFileCache>> - .Create(function(const Filename: String): TFuture>> begin Result := DoLoad(Filename); end); + TDataFileCache>>> + .Create(function(const Filename: String): TFuture>>> begin Result := DoLoad(Filename); end); end; destructor TAuraDataServer.Destroy; @@ -274,9 +283,9 @@ begin FCachedFiles.Clear; end; -function TAuraDataServer.DoLoad(const FileName: string): TFuture>>; +function TAuraDataServer.DoLoad(const FileName: string): TFuture>>>; begin - Result := TFuture>>.Null; + Result := TFuture>>>.Null; if not TFile.Exists(FileName) then exit; @@ -287,8 +296,8 @@ begin Result := TFuture .Construct(FLoadGate.Enqueue, function: TBytes begin Result := TFile.ReadAllBytes(capFileName); end) - .Chain>>( - function(bytes: TBytes): TArray> + .Chain>>>( + function(bytes: TBytes): TArray>> begin var zipMemoryStream := TBytesStream.Create(bytes); try @@ -302,9 +311,9 @@ begin else begin Result := - TFuture>>.Construct( + TFuture>>>.Construct( FLoadGate.Enqueue, - function: TArray> + function: TArray>> begin if TFile.Exists(capFileName) then begin @@ -411,7 +420,7 @@ begin end); end; -function TAuraDataServer.LoadDataFile(const DataFile: TAuraDataFile): TFuture>>; +function TAuraDataServer.LoadDataFile(const DataFile: TAuraDataFile): TFuture>>>; begin if DataFile.IsValid then Result := FCachedFiles.GetOrAdd(DataFile.GetFullFileName); @@ -419,7 +428,7 @@ end; function TAuraDataServer.ProcessFile( FileInfo: TAuraDataFile; - const DataFile: TFuture>>; + const DataFile: TFuture>>>; const Terminated: TState; Processor: IMycProcessor>> ): TState; @@ -434,11 +443,20 @@ begin var cTerminated := Terminated; Result := DataFile.Chain( - function(const Data: TArray>): TState + function(const DataChunks: TArray>>): TState begin - if not Processor.ProcessData(Data) then - exit(TState.Null); - Processor.Update; + if Assigned(DataChunks) then + begin + // Process each chunk, checking for termination between chunks. + for var chunk in DataChunks do + begin + if cTerminated.IsSet then + exit(TState.Null); + + if not Processor.ProcessData(chunk) then + exit(TState.Null); + end; + end; Result := ProcessFile(nextFileInfo, nextFile, cTerminated, Processor); end @@ -485,6 +503,7 @@ begin ); FLastTimeStamp := 0; + FCurrChunkInFile := 0; FCurrPosInFile := 0; PreloadNext; @@ -502,6 +521,8 @@ end; function TAuraFileStream.GetChunk(var Data: array of TDataPoint): Integer; var item: TDataPoint; + fileChunks: TArray>>; + currentChunk: TArray>; label loop; begin @@ -516,19 +537,33 @@ begin if not FCurrent.Value.FileInfo.IsValid then exit; - if FCurrPosInFile >= Length(FCurrent.Value.Data.Value) then + fileChunks := FCurrent.Value.Data.Value; + if (fileChunks = nil) or (FCurrChunkInFile >= Length(fileChunks)) then begin + // Finished with current file, move to next + FCurrChunkInFile := 0; FCurrPosInFile := 0; FCurrent := FNext; PreloadNext; goto loop; end; - var currData := FCurrent.Value.Data.Value; var maxLen := Length(Data); - while (Result < maxLen) and (FCurrPosInFile < Length(currData)) do + while (Result < maxLen) do begin - item := currData[FCurrPosInFile]; + // Check if we need to advance to the next chunk + if (FCurrChunkInFile < Length(fileChunks)) and (FCurrPosInFile >= Length(fileChunks[FCurrChunkInFile])) then + begin + Inc(FCurrChunkInFile); + FCurrPosInFile := 0; + end; + + // Exit loop if all chunks are processed for the current file + if FCurrChunkInFile >= Length(fileChunks) then + break; + + currentChunk := fileChunks[FCurrChunkInFile]; + item := currentChunk[FCurrPosInFile]; if FLastTimeStamp < item.Time then begin FLastTimeStamp := item.Time; @@ -538,7 +573,8 @@ begin Inc(FCurrPosInFile); end; - if (FCurrPosInFile < Length(currData)) or FNextReady.Reset then + // Notify if there is potentially more data + if (FCurrChunkInFile < Length(fileChunks)) or FNextReady.Reset then begin FHasData.Notify; end; @@ -593,8 +629,8 @@ end; function TAuraTABFileServer.LoadDataSeries(const InitialFile: TAuraDataFile): TFuture>>; var loadedState: TState; - loadedFiles: TArray>>>; - liveData: TFuture>>; + loadedFiles: TArray>>>>; + liveData: TFuture>>>; tabFiles: TList; liveFile: TAuraDataFile; currentFileInfo: TAuraDataFile; @@ -618,7 +654,7 @@ begin liveFile := ParseFileName(potentialLivePath); end; - var loadedFileList := TList>>>.Create; + var loadedFileList := TList>>>>.Create; var loadStates := TList.Create; try for var fileInfo in tabFiles do @@ -628,7 +664,7 @@ begin loadStates.Add(data.Done); end; - liveData := TFuture>>.Null; + liveData := TFuture>>>.Null; if liveFile.IsValid then begin liveData := LoadDataFile(liveFile); @@ -656,16 +692,18 @@ begin var overallLastTabTickTime: TDateTime := 0; var cnt := 0; for var P in loadedFiles do - inc(cnt, Length(P.Value)); + for var chunk in P.Value do + inc(cnt, Length(chunk)); tickList.Capacity := cnt; for var P in loadedFiles do - for var iterTickData in P.Value do - if overallLastTabTickTime < iterTickData.Time then - begin - tickList.Add(iterTickData); - overallLastTabTickTime := iterTickData.Time; - end; + 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; @@ -721,14 +759,14 @@ begin Result := TAuraDataFile.Create(path, symbol, fileExt, year, month); end; -class function TAuraTABFileServer.ReadCompressedData(const InputStream: TStream): TArray>; +class function TAuraTABFileServer.ReadCompressedData(const InputStream: TStream): TArray>>; var decompressionStream: TStream; localHeader: TZipHeader; entryIndex, i: Integer; zipFileInstance: TZipFile; begin - SetLength(Result, 0); + Result := nil; decompressionStream := nil; zipFileInstance := nil; try @@ -760,7 +798,7 @@ begin end; end; -class function TAuraTABFileServer.ReadUncompressedData(const InputStream: TStream): TArray>; +class function TAuraTABFileServer.ReadUncompressedData(const InputStream: TStream): TArray>>; type TFileRecord = packed record TimeStamp: TDateTime; @@ -768,25 +806,40 @@ type end; var fileSize: Int64; - recordCount, bytesRead: Integer; + totalRecordCount, bytesRead, chunkIndex, indexInChunk, recordIndex: Integer; rec: TFileRecord; begin - SetLength(Result, 0); + Result := nil; InputStream.Position := 0; fileSize := InputStream.Size; if (fileSize = 0) or ((fileSize mod SizeOf(TFileRecord)) <> 0) then exit; - recordCount := fileSize div SizeOf(TFileRecord); - if recordCount > 0 then + totalRecordCount := fileSize div SizeOf(TFileRecord); + if totalRecordCount > 0 then begin - SetLength(Result, recordCount); - for var i := 0 to High(Result) do + 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.', [fileSize, bytesRead]); - Result[i].Create(rec.TimeStamp, rec.Data); + raise EReadError.CreateFmt('Read error. Expected %d bytes, read %d.', [SizeOf(TFileRecord), bytesRead]); + + Result[chunkIndex][indexInChunk].Create(rec.TimeStamp, rec.Data); + Inc(indexInChunk); end; end; end;