DataStream updated

AuraTrader/MainForm.pas

@@ -102,11 +102,8 @@ type
     private
         FOnEvent: TNotifyEvent;
         { Private declarations }
-{$ifdef TICKDATA}
+        FFileCache: IDataFileCache<TArray<TBytes>>;
-        FServer: IDataServer<TAuraAskBidFileItem>;
+        FServer: IDataServer;
-{$else}
-        FServer: IDataServer<TOhlcItem>;
-{$endif}
         FSymbols: TFuture<TArray<String>>;
         FTerminate: TEvent;
         FProcessDone: TState;
@@ -228,10 +225,11 @@ begin
     FTerminate := TEvent.CreateEvent;
 
     // Create an instance of the TAuraTABFileServer. The server can be reused for multiple stream creations. [364]
+    FFileCache := TDataFileCache<TArray<TBytes>>.Create;
 {$ifdef TICKDATA}
-    FServer := TAskBidFileServer.Create('\\COFFEE\TickData\Pepperstone');
+    FServer := TTickFileServer.Create(FFileCache, '\\COFFEE\TickData\Pepperstone');
 {$else}
-    FServer := TM1FileServer.Create('\\COFFEE\TickData\Pepperstone');
+    FServer := TM1FileServer.Create(FFileCache, '\\COFFEE\TickData\Pepperstone');
 {$endif}
 
     SymbolsComboBox.Enabled := false;
@@ -339,7 +337,7 @@ begin
     var panel := pnlChart.AddPanel;
     panel.AddDoubleSeries(equity, TAlphaColors.Blue, 3);
 
-    FProcessDone := FProcessDone + FServer.ProcessData(Symbol, terminated, ticker.Consumer);
+    FProcessDone := FProcessDone + (FServer as IM1FileServer).ProcessData(Symbol, terminated, ticker.Consumer);
 
 end;
 
@@ -587,44 +585,28 @@ begin
     var terminated := TFlag.CreateObserver(FTerminate.Signal).State;
 
 {$ifdef TICKDATA}
-    var ticker := TConverter.CreateTicker<TDataPoint<TAskBidItem>>;
+    var ticker := TConverter.CreateTicker<TDataPoint<TTickRecord>>;
 
     var lastPrice :=
-        ticker.Chain<TDataPoint<Double>>(
+        ticker.Producer.Chain<TDataPoint<Double>>(
-            function(const Tick: TDataPoint<TAskBidItem>): TDataPoint<Double>
+            function(const Tick: TDataPoint<TTickRecord>): TDataPoint<Double>
             begin
                 Result.Time := Tick.Time;
                 Result.Data := 0.5 * (Tick.Data.Ask + Tick.Data.Bid);
             end
         );
 
-    var OhlcPoint := lastPrice.Chain<TDataPoint<TOhlcItem>>(TConverter.CreateTickAggregation(Timeframe));
+    var OhlcPoint := lastPrice.Chain<TDataPoint<TOhlcItem>>(TTradeConverter.CreateTickAggregation(Timeframe));
-    OhlcPoint.Producer.Link(Consumer);
+    OhlcPoint.CreateLink(Consumer);
 
-    var Producer :=
+    FProcessDone := FProcessDone + (FServer as ITickFileServer).ProcessData(Symbol, terminated, ticker.Consumer);
-        TConverter<TArray<TDataPoint<TAuraAskBidFileItem>>, TArray<TDataPoint<TAskBidItem>>>.CreateGeneric(
-            function(const Values: TArray<TDataPoint<TAuraAskBidFileItem>>): TArray<TDataPoint<TAskBidItem>>
-            begin
-                SetLength(Result, Length(Values));
-                for var i := 0 to High(Result) do
-                begin
-                    Result[i].Time := Values[i].Time;
-                    Result[i].Data.Ask := Values[i].Data.Ask;
-                    Result[i].Data.Bid := Values[i].Data.Bid;
-                end;
-            end
-        );
-
-    Producer.Producer.Link(ticker);
-
-    FProcessDone := FProcessDone + FServer.ProcessData(Symbol, terminated, Producer);
 {$else}
 
     var ticker := TConverter.CreateTicker<TDataPoint<TOhlcItem>>;
 
     ticker.Producer.Chain(Consumer);
 
-    FProcessDone := FProcessDone + FServer.ProcessData(Symbol, terminated, ticker.Consumer);
+    FProcessDone := FProcessDone + (FServer as IM1FileServer).ProcessData(Symbol, terminated, ticker.Consumer);
 {$endif}
 end;
 

Src/Myc.Core.FileCache.pas

@@ -77,11 +77,12 @@ end;
 
 procedure TDataFileCache<T>.Clear;
 begin
-    FLock.Enter;
-    try
     // Wait for any pending load operations to finish before clearing.
     for var cachedFile in FCachedFiles.Values do
         cachedFile.Data.WaitFor;
+
+    FLock.Enter;
+    try
         FCachedFiles.Clear;
     finally
         FLock.Leave;

Src/Myc.Trade.DataStream.pas

@@ -34,29 +34,30 @@ type
         Bid: Double;
     end;
 
-    // Interface for an instantiable data server.
+    IDataServer = interface
-    IDataServer<T: record> = interface
         procedure ClearCache;
-        function ProcessData(const Symbol: String; const Terminated: TState; const Processor: IConsumer<TArray<TDataPoint<T>>>): TState;
         function EnumerateSymbols: TFuture<TArray<String>>;
     end;
 
+    // Interface for an instantiable data server.
+    IDataServer<T: record> = interface(IDataServer)
+        function ProcessData(const Symbol: String; const Terminated: TState; const Processor: IConsumer<TArray<TDataPoint<T>>>): TState;
+    end;
+
     // Metadata for a monthly data file (SYMBOL_YYYY_MM.EXT)
     TDataFile = record
     private
-        FExtension: String;
         FPath: String;
-        FSymbol: String;
+        FSymbol: String; // Includes extension, e.g. "EURUSD.M1"
         FYear: Integer;
         FMonth: Integer;
         FBasename: String;
         function GetIsValid: Boolean;
     public
-        constructor Create(const APath, ASymbol, AExtension: String; AYear, AMonth: Integer; const ABasename: String);
+        constructor Create(const APath, ASymbol: 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;
@@ -64,22 +65,19 @@ type
     end;
 
     // Generic server base handling the pipeline and preloading.
-    TDataServer<T: record; R: record> = class(TInterfacedObject, IDataServer<T>)
+    TDataServer<T: record; R: record> = class(TInterfacedObject, IDataServer, IDataServer<T>)
     private
-        FCachedFiles: TDataFileCache<TArray<TArray<TDataPoint<T>>>>;
+        // Cache stores binary blobs of the MAPPED data (TDataPoint<T>)
+        FCachedFiles: IDataFileCache<TArray<TBytes>>;
         FPath: String;
         FSymbols: TFuture<TDictionary<String, TArray<TDataFile>>>;
         function GetPath: String;
 
-        function ProcessChunks(
+        function ProcessChunks(const MappedBlobs: TArray<TBytes>; Terminated: TState; Processor: IConsumer<TArray<TDataPoint<T>>>): TState;
-            const DataChunks: TArray<TArray<TDataPoint<T>>>;
-            Terminated: TState;
-            Processor: IConsumer<TArray<TDataPoint<T>>>
-        ): TState;
 
         function ProcessFile(
             FileInfo: TDataFile;
-            const DataFile: TFuture<TArray<TArray<TDataPoint<T>>>>;
+            const DataFile: TFuture<TArray<TBytes>>;
             const Terminated: TState;
             Processor: IConsumer<TArray<TDataPoint<T>>>
         ): TState;
@@ -87,17 +85,22 @@ type
     strict private
         class var
             FLoadGate: TLatch;
+        class constructor CreateClass;
 
     protected
         // Maps the raw file record to the internal TDataPoint<T>
         function MapRecord(const RecordBuffer: R): TDataPoint<T>; virtual; abstract;
-        function DoLoad(const FileName: string; const Gate: TLatch): TFuture<TArray<TArray<TDataPoint<T>>>>;
+
+        // Returns Future of Processed Blobs (Serialized TDataPoint<T>)
+        function DoLoad(const FileName: string; const Gate: TLatch): TFuture<TArray<TBytes>>;
+
         function GetNextFile(const Curr: TDataFile): TDataFile; virtual;
-        function ReadCompressedData(const InputStream: TStream): TArray<TArray<TDataPoint<T>>>;
+        function ReadCompressedData(const InputStream: TStream): TArray<TBytes>;
-        function ReadUncompressedData(const InputStream: TStream): TArray<TArray<TDataPoint<T>>>;
+        function ReadUncompressedData(const InputStream: TStream): TArray<TBytes>;
 
     public
-        constructor Create(const APath: String);
+        // Cache is injected. It stores TArray<TBytes>.
+        constructor Create(const ACache: IDataFileCache<TArray<TBytes>>; const APath: String);
         destructor Destroy; override;
         procedure AfterConstruction; override;
         procedure ClearCache;
@@ -105,18 +108,27 @@ type
         function FindFirstFile(const Symbol: string): TFuture<TDataFile>;
         function ParseFileName(const FileName: string): TDataFile;
         function EnumerateSymbols: TFuture<TArray<String>>;
-        function LoadDataFile(const DataFile: TDataFile): TFuture<TArray<TArray<TDataPoint<T>>>>;
+
+        function LoadDataFile(const DataFile: TDataFile): TFuture<TArray<TBytes>>;
         function ProcessData(const Symbol: String; const Terminated: TState; const Processor: IConsumer<TArray<TDataPoint<T>>>): TState;
 
         property Path: String read GetPath;
     end;
 
-    TTickFileServer = class(TDataServer<TTickRecord, TTickRecord>)
+    ITickFileServer = interface(IDataServer<TTickRecord>)
+        ['{A6DE8B82-B0FB-48DB-A39A-CE91D861C5D2}']
+    end;
+
+    TTickFileServer = class(TDataServer<TTickRecord, TTickRecord>, ITickFileServer)
     protected
         function MapRecord(const RecordBuffer: TTickRecord): TDataPoint<TTickRecord>; override;
     end;
 
-    TM1FileServer = class(TDataServer<TOhlcItem, TM1Record>)
+    IM1FileServer = interface(IDataServer<TOhlcItem>)
+        ['{7BB980D9-1969-49A4-991F-908F2DDAC1A0}']
+    end;
+
+    TM1FileServer = class(TDataServer<TOhlcItem, TM1Record>, IM1FileServer)
     protected
         function MapRecord(const RecordBuffer: TM1Record): TDataPoint<TOhlcItem>; override;
     end;
@@ -135,11 +147,10 @@ const
 
 { TDataFile }
 
-constructor TDataFile.Create(const APath, ASymbol, AExtension: String; AYear, AMonth: Integer; const ABasename: String);
+constructor TDataFile.Create(const APath, ASymbol: String; AYear, AMonth: Integer; const ABasename: String);
 begin
     FPath := APath;
     FSymbol := ASymbol;
-    FExtension := AExtension;
     FYear := AYear;
     FMonth := AMonth;
     FBasename := ABasename;
@@ -151,8 +162,11 @@ begin
 end;
 
 function TDataFile.GetFullFileName: string;
+var
+    ext: string;
 begin
-    Result := TPath.Combine(FPath, FBasename + FExtension);
+    ext := TPath.GetExtension(FSymbol);
+    Result := TPath.Combine(FPath, FBasename + ext);
 end;
 
 function TDataFile.GetIsValid: Boolean;
@@ -162,17 +176,20 @@ end;
 
 { TDataServer<T, R> }
 
-constructor TDataServer<T, R>.Create(const APath: String);
+class constructor TDataServer<T, R>.CreateClass;
+begin
+    FLoadGate := TLatch.CreateLatch(0);
+end;
+
+constructor TDataServer<T, R>.Create(const ACache: IDataFileCache<TArray<TBytes>>; const APath: String);
 begin
     inherited Create;
     FPath := APath;
-    FCachedFiles := TDataFileCache<TArray<TArray<TDataPoint<T>>>>.Create;
+    FCachedFiles := ACache;
 end;
 
 destructor TDataServer<T, R>.Destroy;
 begin
-    ClearCache;
-    FCachedFiles.Free;
     inherited Destroy;
 end;
 
@@ -184,6 +201,7 @@ end;
 
 procedure TDataServer<T, R>.ClearCache;
 begin
+    if Assigned(FCachedFiles) then
         FCachedFiles.Clear;
 end;
 
@@ -227,7 +245,7 @@ begin
                             TComparer<TDataFile>.Construct(
                                 function(const Left, Right: TDataFile): Integer
                                 begin
-                                    if Left.Year <> Right.Year then
+                                    if (Left.Year <> Right.Year) then
                                         Result := Left.Year - Right.Year
                                     else
                                         Result := Left.Month - Right.Month;
@@ -264,7 +282,7 @@ begin
             var
                 symbolFiles: TArray<TDataFile>;
             begin
-                if Symbols.TryGetValue(symName, symbolFiles) and (Length(symbolFiles) > 0) then
+                if (Symbols.TryGetValue(symName, symbolFiles)) and (Length(symbolFiles) > 0) then
                     Result := symbolFiles[0]
                 else
                     Result := Default(TDataFile);
@@ -287,7 +305,7 @@ begin
     foundIndex := -1;
     for var i := 0 to High(allSymbolFiles) do
     begin
-        if allSymbolFiles[i].GetBaseFileName = Curr.GetBaseFileName then
+        if (allSymbolFiles[i].GetBaseFileName = Curr.GetBaseFileName) then
         begin
             foundIndex := i;
             break;
@@ -314,39 +332,37 @@ begin
     ext := TPath.GetExtension(fileNameNoPath);
     baseName := TPath.GetFileNameWithoutExtension(fileNameNoPath);
 
-    // Format: SYMBOL_YYYY_MM
     parts := baseName.Split(['_']);
-    if Length(parts) < 3 then
+    if (Length(parts) < 3) then
         exit;
 
-    // Use reverse indexing for year and month
     if (not TryStrToInt(parts[High(parts)], month)) or (not TryStrToInt(parts[High(parts) - 1], year)) then
         exit;
 
-    symbol := string.Join('_', Copy(parts, 0, Length(parts) - 2));
+    symbol := string.Join('_', Copy(parts, 0, Length(parts) - 2)) + ext;
-    Result := TDataFile.Create(TPath.GetDirectoryName(FileName), symbol, ext, year, month, baseName);
+    Result := TDataFile.Create(TPath.GetDirectoryName(FileName), symbol, year, month, baseName);
 end;
 
-function TDataServer<T, R>.LoadDataFile(const DataFile: TDataFile): TFuture<TArray<TArray<TDataPoint<T>>>>;
+function TDataServer<T, R>.LoadDataFile(const DataFile: TDataFile): TFuture<TArray<TBytes>>;
 begin
-    if DataFile.IsValid then
+    if (DataFile.IsValid) then
         Result :=
             FCachedFiles.GetOrAdd(
                 DataFile.GetFullFileName,
-                function(const Filename: String): TFuture<TArray<TArray<TDataPoint<T>>>> begin Result := DoLoad(Filename, FLoadGate); end
+                function(const Filename: String): TFuture<TArray<TBytes>> begin Result := DoLoad(Filename, FLoadGate); end
             )
     else
-        Result := TFuture<TArray<TArray<TDataPoint<T>>>>.Construct(TArray<TArray<TDataPoint<T>>>(nil));
+        Result := TFuture<TArray<TBytes>>.Construct(TArray<TBytes>(nil));
 end;
 
-function TDataServer<T, R>.DoLoad(const FileName: string; const Gate: TLatch): TFuture<TArray<TArray<TDataPoint<T>>>>;
+function TDataServer<T, R>.DoLoad(const FileName: string; const Gate: TLatch): TFuture<TArray<TBytes>>;
 begin
     var capFileName := FileName;
     Result :=
         TFuture<TBytes>
             .Construct(Gate.Enqueue, function: TBytes begin Result := TFile.ReadAllBytes(capFileName); end)
-            .Chain<TArray<TArray<TDataPoint<T>>>>(
+            .Chain<TArray<TBytes>>(
-                function(Bytes: TBytes): TArray<TArray<TDataPoint<T>>>
+                function(Bytes: TBytes): TArray<TBytes>
                 var
                     compStream: TBytesStream;
                 begin
@@ -359,7 +375,7 @@ begin
                 end);
 end;
 
-function TDataServer<T, R>.ReadCompressedData(const InputStream: TStream): TArray<TArray<TDataPoint<T>>>;
+function TDataServer<T, R>.ReadCompressedData(const InputStream: TStream): TArray<TBytes>;
 var
     zip: TZipFile;
     decompStream: TStream;
@@ -375,14 +391,14 @@ begin
         // Search for the .bin entry created by C#
         for var i := 0 to zip.FileCount - 1 do
         begin
-            if zip.FileNames[i].EndsWith('.bin', true) then
+            if (zip.FileNames[i].EndsWith('.bin', true)) then
             begin
                 entryIdx := i;
                 break;
             end;
         end;
 
-        if entryIdx = -1 then
+        if (entryIdx = -1) then
             exit;
 
         zip.Read(entryIdx, decompStream, header);
@@ -396,37 +412,54 @@ begin
     end;
 end;
 
-function TDataServer<T, R>.ReadUncompressedData(const InputStream: TStream): TArray<TArray<TDataPoint<T>>>;
+function TDataServer<T, R>.ReadUncompressedData(const InputStream: TStream): TArray<TBytes>;
 var
     totalRecords, chunkCount, i, j, recsInChunk: Integer;
-    rawBuffer: TArray<R>;
+    inputRecordSize, outputPointSize: Integer;
+    chunkBuffer: TArray<TDataPoint<T>>;
+    rawBufferChunk: TArray<R>; // Only buffer one chunk at a time
 begin
     Result := nil;
-    totalRecords := InputStream.Size div sizeof(R);
+    inputRecordSize := SizeOf(R);
-    if totalRecords = 0 then
+    outputPointSize := SizeOf(TDataPoint<T>);
-        exit;
 
-    // Fast block read into memory
+    // Ensure we are working with complete records
-    SetLength(rawBuffer, totalRecords);
+    if (InputStream.Size mod inputRecordSize <> 0) then
-    InputStream.Position := 0;
+        raise Exception.Create('Stream size is not a multiple of record size.');
-    InputStream.ReadBuffer(rawBuffer[0], InputStream.Size);
+
+    totalRecords := InputStream.Size div inputRecordSize;
+    if (totalRecords = 0) then
+        exit;
 
     chunkCount := (totalRecords + ChunkSize - 1) div ChunkSize;
     SetLength(Result, chunkCount);
+    SetLength(rawBufferChunk, ChunkSize); // Reuse buffer
+
+    InputStream.Position := 0;
 
     for i := 0 to chunkCount - 1 do
     begin
         recsInChunk := Min(ChunkSize, totalRecords - (i * ChunkSize));
-        SetLength(Result[i], recsInChunk);
+
+        // Read only needed bytes for this chunk
+        InputStream.ReadBuffer(rawBufferChunk[0], recsInChunk * inputRecordSize);
+
+        // Temp buffer for mapped objects
+        SetLength(chunkBuffer, recsInChunk);
+
         for j := 0 to recsInChunk - 1 do
         begin
-            Result[i][j] := MapRecord(rawBuffer[i * ChunkSize + j]);
+            chunkBuffer[j] := MapRecord(rawBufferChunk[j]);
         end;
+
+        // Serialize mapped data to TBytes
+        SetLength(Result[i], recsInChunk * outputPointSize);
+        Move(chunkBuffer[0], Result[i][0], recsInChunk * outputPointSize);
     end;
 end;
 
 function TDataServer<T, R>.ProcessChunks(
-    const DataChunks: TArray<TArray<TDataPoint<T>>>;
+    const MappedBlobs: TArray<TBytes>;
     Terminated: TState;
     Processor: IConsumer<TArray<TDataPoint<T>>>
 ): TState;
@@ -434,16 +467,28 @@ begin
     var callProcess :=
         function(Idx: Integer): TFunc<TState>
         begin
-            var cChunk := DataChunks[Idx];
+            // Capture the blob
+            var cBlob := MappedBlobs[Idx];
             Result :=
                 function: TState
+                var
+                    dataChunk: TArray<TDataPoint<T>>;
+                    count: Integer;
                 begin
                     if not Terminated.IsSet then
-                        Result := Processor.Consume(cChunk);
+                    begin
+                        // Quick deserialize (Mem Copy)
+                        count := Length(cBlob) div SizeOf(TDataPoint<T>);
+                        SetLength(dataChunk, count);
+                        if count > 0 then
+                            Move(cBlob[0], dataChunk[0], Length(cBlob));
+
+                        Result := Processor.Consume(dataChunk);
+                    end;
                 end;
         end;
 
-    for var i := 0 to High(DataChunks) do
+    for var i := 0 to High(MappedBlobs) do
     begin
         if not Terminated.IsSet then
         begin
@@ -455,7 +500,7 @@ end;
 
 function TDataServer<T, R>.ProcessFile(
     FileInfo: TDataFile;
-    const DataFile: TFuture<TArray<TArray<TDataPoint<T>>>>;
+    const DataFile: TFuture<TArray<TBytes>>;
     const Terminated: TState;
     Processor: IConsumer<TArray<TDataPoint<T>>>
 ): TState;
@@ -469,9 +514,9 @@ begin
 
     Result :=
         DataFile.Chain(
-            function(const DataChunks: TArray<TArray<TDataPoint<T>>>): TState
+            function(const MappedBlobs: TArray<TBytes>): TState
             begin
-                var done := ProcessChunks(DataChunks, cTerminated, Processor);
+                var done := ProcessChunks(MappedBlobs, cTerminated, Processor);
                 Result :=
                     TaskManager
                         .RunTask(done, function: TState begin Result := ProcessFile(nextFileInfo, nextFile, cTerminated, Processor); end);