Refactoring

Src/Myc.Trade.DataSeries.pas

@@ -46,7 +46,9 @@ uses
     System.IOUtils,
     Myc.Futures,
     Myc.Signals,
-    Myc.Trade.DataPoint;
+    Myc.Lazy,
+    Myc.Trade.DataPoint,
+    Myc.Trade.DataServer;
 
 type
     // Generic class for loading and managing sequential time-series data.
@@ -94,6 +96,31 @@ type
             TTick = TDataSeries<TAskBidItem>.TDataPoint;
     end;
 
+    // Implements a data server that reads data from Aura-specific historical data files.
+    // It handles sequential loading of data files and transitions to "live data" mode
+    // when all historical files are exhausted.
+    TAuraFileServer<T: record> = class(TDataServer<TDataPoint<T>>, IDataServer<TDataPoint<T>>)
+    private
+        FIsLiveData: TMutable<Boolean>.IWriteable;
+        FCurrentFileName: string;
+        FCurrentData: TFuture<TArray<TDataSeries<T>.TDataPoint>>;
+        FNextFileName: string;
+        FNextData: TFuture<TArray<TDataSeries<T>.TDataPoint>>;
+        FCurrPosInFile: Int64;
+        FCurrentIdx: Int64;
+        FLastTimeStamp: TDateTime;
+
+    protected
+        function GetChunk(var Data: array of TDataPoint<T>): Integer; override;
+        function GetIsLiveData: TMutable<Boolean>; override;
+    public
+        constructor Create(const AFilename: String);
+        procedure AfterConstruction; override;
+    end;
+
+    // Specialized TAuraFileServer for TAskBidItem data.
+    TAuraTABFileServer = TAuraFileServer<TAskBidItem>;
+
 // Parses a filename to extract symbol, year, month, and path.
 function TryParseFileName(const FileName: string; out PathValue, SymbolValue: string; out YearValue, MonthValue: Integer): Boolean;
 
@@ -636,4 +663,95 @@ begin
     end;
 end;
 
+{ TAuraFileServer<T> }
+
+constructor TAuraFileServer<T>.Create(const AFilename: String);
+begin
+    inherited Create;
+    FCurrentFileName := AFilename;
+    FIsLiveData := TMutable<Boolean>.CreateWriteable(false);
+end;
+
+procedure TAuraFileServer<T>.AfterConstruction;
+begin
+    inherited;
+
+    // Asynchronously loads the initial data file.
+    FCurrentData := TDataSeries<T>.LoadDataFile(FCurrentFileName);
+    FCurrPosInFile := 0;
+    FCurrentIdx := 0;
+    // Initially sets the server to "not live data".
+    FIsLiveData.SetValue(false);
+    FLastTimeStamp := 0;
+
+    // Determines and asynchronously loads the next chronological file, if available.
+    FNextFileName := FindNextDataFile(FCurrentFileName);
+    if FNextFileName <> '' then
+        FNextData := TDataSeries<T>.LoadDataFile(FNextFileName);
+end;
+
+function TAuraFileServer<T>.GetChunk(var Data: array of TDataPoint<T>): Integer;
+begin
+    Result := 0;
+
+    // Checks if the current data file is ready.
+    if FCurrentData.Done.IsSet then
+    begin
+        var maxLen := Length(Data);
+
+        var currData := FCurrentData.Value;
+        while Result < MaxLen do
+        begin
+            if FCurrPosInFile >= Length(currData) then
+            begin
+                // Transition to the next file if the current one is exhausted.
+                FCurrentData := FNextData;
+                FCurrentFileName := FNextFileName;
+                FCurrPosInFile := 0;
+
+                FNextFileName := '';
+                FNextData := TFuture<TArray<TDataSeries<T>.TDataPoint>>.Null;
+
+                if FCurrentFileName <> '' then
+                begin
+                    // Loads the next file in the series.
+                    FNextFileName := FindNextDataFile(FCurrentFileName);
+                    if FNextFileName <> '' then
+                        FNextData := TDataSeries<T>.LoadDataFile(FNextFileName);
+
+                    // If the newly assigned current data is already loaded, continue processing.
+                    if FCurrentData.Done.IsSet then
+                    begin
+                        currData := FCurrentData.Value;
+                        continue;
+                    end;
+                end
+                else
+                begin
+                    // All historical files processed; transition to "live data" mode.
+                    FIsLiveData.SetValue(true);
+                end;
+                break; // Exit loop if no more data or transition occurred.
+            end;
+
+            // Processes data point if its timestamp is strictly greater than the last processed one.
+            // This ensures chronological order and handles potential duplicate timestamps.
+            var item := currData[FCurrPosInFile];
+            if FLastTimeStamp < item.TimeStamp then
+            begin
+                FLastTimeStamp := item.TimeStamp;
+                Data[Result].Create(FCurrentIdx, FLastTimeStamp, item.Data);
+                inc(Result);
+                inc(FCurrentIdx);
+            end;
+            inc(FCurrPosInFile);
+        end;
+    end;
+end;
+
+function TAuraFileServer<T>.GetIsLiveData: TMutable<Boolean>;
+begin
+    Result := FIsLiveData;
+end;
+
 end.