implementing DataStream

This commit is contained in:
Michael Schimmel
2025-06-07 23:53:45 +02:00
parent 776067d0a8
commit 312be15cc8
4 changed files with 316 additions and 222 deletions
@@ -1,4 +1,4 @@
unit Myc.Test.Trade.DataServer;
unit Myc.Test.Trade.DataStream;
interface
@@ -19,8 +19,8 @@ type
[IgnoreMemoryLeaks(true)]
TTest_TABFileServer_Equivalence = class(TObject)
private
FServer: IDataServer<TAskBidItem>;
FStream: IDataStream<TDataPoint<TAskBidItem>>;
FServer: IAuraDataServer<TAskBidItem>;
FStream: IDataStream<TAskBidItem>;
public
[SetupFixture]
procedure SetupFixture;
@@ -42,7 +42,8 @@ uses
const
// The reference data file for this test.
C_TEST_FILENAME = '\\COFFEE\TickData\Pepperstone\GER40_2025_03.tab_zip';
C_TEST_PATH = '\\COFFEE\TickData\Pepperstone';
C_TEST_SYMBOL = 'GER40';
// The chunk size for fetching data from the server.
C_MAX_FETCH = 200;
@@ -50,7 +51,7 @@ const
procedure TTest_TABFileServer_Equivalence.SetupFixture;
begin
FServer := TAskBidServer.Create;
FServer := TAuraTABFileServer.Create(C_TEST_PATH);
// TAskBid.ClearCache ensures a clean state for data series loading.
FServer.ClearCache;
end;
@@ -66,7 +67,7 @@ end;
procedure TTest_TABFileServer_Equivalence.Setup;
begin
// Initializes the data server with a specific test file.
FStream := TAuraTABFileStream.Create(C_TEST_FILENAME, FServer);
FStream := FServer.CreateStream(C_TEST_SYMBOL);
end;
procedure TTest_TABFileServer_Equivalence.TearDown;
@@ -84,7 +85,12 @@ var
i: Int64;
begin
// Loads the expected data directly using TAskBid.LoadDataSeries.WaitFor.
var ExpectedData := FServer.LoadDataSeries(C_TEST_FILENAME).WaitFor;
var Filename := TAuraTABFileServer.FindFirstDataFile(C_TEST_PATH, C_TEST_SYMBOL);
Assert.IsNotEmpty(Filename);
var ExpectedData := FServer.LoadDataSeries(Filename).WaitFor;
SetLength(Dst, Length(ExpectedData));
var cnt: Int64 := 0;
+270 -184
View File
@@ -6,7 +6,7 @@ unit Myc.Trade.DataStream;
This unit contains the core components for handling historical data:
- TDataRecord: A generic record for a single time-stamped data entry.
- IDataServer/TDataServer: A server component responsible for loading and caching
- IDataServer/TAuraDataServer: A server component responsible for loading and caching
series of data files from disk. Each server instance manages its own cache,
but all instances share a central load gate.
- IDataStream/TDataStream: An interface representing a stream of data points that
@@ -39,7 +39,7 @@ type
IDataStream<T> = interface
['{A6E246A2-E84E-49AB-A63E-333E561E488C}']
function GetIsLiveData: TMutable<Boolean>;
function GetChunk(var Data: array of T): Integer;
function GetChunk(var Data: array of TDataPoint<T>): Integer;
property IsLiveData: TMutable<Boolean> read GetIsLiveData;
end;
@@ -47,7 +47,7 @@ type
TDataStream<T> = class(TInterfacedObject, IDataStream<T>)
protected
function GetIsLiveData: TMutable<Boolean>; virtual; abstract;
function GetChunk(var Data: array of T): Integer; virtual; abstract;
function GetChunk(var Data: array of TDataPoint<T>): Integer; virtual; abstract;
end;
// Represents a factory for creating IDataStream instances.
@@ -65,16 +65,25 @@ type
// Interface for an instantiable data server.
IDataServer<T: record> = interface
['{1F8E5A9D-E92A-44C1-9F3F-C4B82A6E94B3}']
function LoadDataFile(const FileName: string): TFuture<TArray<TDataRecord<T>>>;
function LoadDataSeries(const FileName: string): TFuture<TArray<TDataRecord<T>>>;
function CreateStream(const Symbol: String): IDataStream<T>;
procedure ClearCache;
end;
IAuraDataServer<T: record> = interface(IDataServer<T>)
['{481E27DE-DC58-49AF-B8A8-B6316980C423}']
function LoadDataFile(const FileName: string): TFuture<TArray<TDataRecord<T>>>;
function LoadDataSeries(const FileName: string): TFuture<TArray<TDataRecord<T>>>;
function EnumerateAssetFiles: TArray<String>;
end;
// Aura files
// Generic server for loading and managing sequential time-series data from files.
TDataServer<T: record> = class(TInterfacedObject, IDataServer<T>)
strict private
TAuraDataServer<T: record> = class(TInterfacedObject, IDataServer<T>, IAuraDataServer<T>)
private
type
TCachedFile = class
public
Name: String;
Age: TDateTime;
LastUsed: TDateTime;
@@ -83,32 +92,47 @@ type
private
// The cache is per-instance.
FCachedFiles: TDictionary<String, TCachedFile>;
FPath: String;
function GetPath: String;
strict private
// The load gate is shared across all instances of TDataServer.
class var FLoadGate: TLatch;
// The load gate is shared across all instances of TAuraDataServer.
class var
FLoadGate: TLatch;
protected
class function ReadCompressedData(const InputStream: TStream): TArray<TDataRecord<T>>; static;
class function ReadUncompressedData(const InputStream: TStream): TArray<TDataRecord<T>>; static;
public
constructor Create;
constructor Create(const APath: String);
destructor Destroy; override;
class function TryParseFileName(
const FileName: string;
out PathValue, SymbolValue: string;
out YearValue, MonthValue: Integer
): Boolean; virtual; abstract;
class function FindFirstDataFile(const Path, Symbol: string): string;
class function FindNextDataFile(const FileName: string): string;
class function FindOldestFilesPerSymbol(const DirectoryPath: string): TArray<String>;
// IDataServer<T> implementation
function CreateStream(const Symbol: String): IDataStream<T>; virtual; abstract;
procedure ClearCache;
function EnumerateAssetFiles: TArray<String>;
function LoadDataFile(const FileName: string): TFuture<TArray<TDataRecord<T>>>;
function LoadDataSeries(const FileName: string): TFuture<TArray<TDataRecord<T>>>;
property Path: String read GetPath;
end;
// A specialized TDataServer for handling Ask/Bid tick data.
TAskBidServer = class(TDataServer<TAskBidItem>)
end;
// Implements a data stream that reads data from Aura-specific historical data files.
TAuraFileStream<T: record> = class(TDataStream<TDataPoint<T>>, IDataStream<TDataPoint<T>>)
// Implements a data stream that reads from Aura-specific historical data files.
TAuraFileStream<T: record> = class(TDataStream<T>, IDataStream<T>)
private
FDataServer: IDataServer<T>;
FDataServer: TAuraDataServer<T>;
FIsLiveData: TMutable<Boolean>.IWriteable;
FCurrentFileName: string;
FCurrentData: TFuture<TArray<TDataRecord<T>>>;
@@ -121,16 +145,23 @@ type
function GetChunk(var Data: array of TDataPoint<T>): Integer; override;
function GetIsLiveData: TMutable<Boolean>; override;
public
constructor Create(const AFilename: String; const ADataServer: IDataServer<T>);
constructor Create(ADataServer: TAuraDataServer<T>; const AFilename: String);
procedure AfterConstruction; override;
end;
// Specialized TAuraFileStream for TAskBidItem data.
TAuraTABFileStream = TAuraFileStream<TAskBidItem>;
// Aura tick data file Ask-Bid
function TryParseFileName(const FileName: string; out PathValue, SymbolValue: string; out YearValue, MonthValue: Integer): Boolean;
function FindOldestFilesPerSymbol(const DirectoryPath: string): TArray<String>;
function FindNextDataFile(const FileName: string): string;
TAuraTABFileServer = class(TAuraDataServer<TAskBidItem>)
public
// IDataServer<T> implementation
function CreateStream(const Symbol: String): IDataStream<TAskBidItem>; override;
class function TryParseFileName(
const FileName: string;
out PathValue, SymbolValue: string;
out YearValue, MonthValue: Integer
): Boolean; override;
end;
var
IsMemoryLow: TFunc<Boolean>;
@@ -140,27 +171,150 @@ implementation
uses
System.Zip,
System.Math,
System.StrUtils,
Myc.TaskManager;
constructor TDataServer<T>.Create;
constructor TAuraDataServer<T>.Create(const APath: String);
begin
inherited Create;
// Initialize the per-instance cache.
FCachedFiles := TObjectDictionary<String, TCachedFile>.Create([doOwnsValues]);
FPath := APath;
end;
destructor TDataServer<T>.Destroy;
destructor TAuraDataServer<T>.Destroy;
begin
ClearCache;
FCachedFiles.Free;
inherited Destroy;
end;
procedure TDataServer<T>.ClearCache;
procedure TAuraDataServer<T>.ClearCache;
begin
for var cF in FCachedFiles.Values do
cF.Data.WaitFor;
FCachedFiles.Clear;
end;
function TDataServer<T>.LoadDataFile(const FileName: string): TFuture<TArray<TDataRecord<T>>>;
class function TAuraDataServer<T>.FindFirstDataFile(const Path, Symbol: string): string;
var
oldestFiles: TArray<string>;
fileName: string;
pathValue, symbolValue: string;
yearValue, monthValue: Integer;
begin
oldestFiles := FindOldestFilesPerSymbol(Path);
for fileName in oldestFiles do
begin
if TryParseFileName(fileName, pathValue, symbolValue, yearValue, monthValue) then
begin
if SameText(symbolValue, Symbol) then
exit(fileName);
end;
end;
Result := '';
end;
class function TAuraDataServer<T>.FindNextDataFile(const FileName: string): string;
var
pathValue, symbolValue: string;
yearValue, monthValue: Integer;
nextBaseName, compressedPath, uncompressedPath: string;
begin
if not TryParseFileName(FileName, pathValue, symbolValue, yearValue, monthValue) then
exit('');
Inc(monthValue);
if monthValue > 12 then
begin
monthValue := 1;
Inc(yearValue);
end;
nextBaseName := Format('%s_%.4d_%.2d.tab', [symbolValue, yearValue, monthValue]);
compressedPath := TPath.Combine(pathValue, nextBaseName + '_zip');
if TFile.Exists(compressedPath) then
exit(compressedPath);
uncompressedPath := TPath.Combine(pathValue, nextBaseName);
if TFile.Exists(uncompressedPath) then
exit(uncompressedPath);
Result := '';
end;
class function TAuraDataServer<T>.FindOldestFilesPerSymbol(const DirectoryPath: string): TArray<String>;
type
TTrackedFileInfo = record
FullFileName: string;
Year: Integer;
Month: Integer;
end;
var
oldestFilesPerSymbol: TDictionary<string, TTrackedFileInfo>;
fileNames: TArray<string>;
currentFile: string;
parsedPath: string;
parsedSymbol: string;
parsedYear: Integer;
parsedMonth: Integer;
trackedInfo: TTrackedFileInfo;
begin
oldestFilesPerSymbol := TDictionary<string, TTrackedFileInfo>.Create;
try
if not TDirectory.Exists(DirectoryPath) then
exit(nil);
fileNames := TDirectory.GetFiles(DirectoryPath);
for currentFile in fileNames do
begin
if TryParseFileName(currentFile, parsedPath, parsedSymbol, parsedYear, parsedMonth) then
begin
if oldestFilesPerSymbol.TryGetValue(parsedSymbol, trackedInfo) then
begin
if (parsedYear < trackedInfo.Year) or ((parsedYear = trackedInfo.Year) and (parsedMonth < trackedInfo.Month)) then
begin
trackedInfo.FullFileName := currentFile;
trackedInfo.Year := parsedYear;
trackedInfo.Month := parsedMonth;
oldestFilesPerSymbol.AddOrSetValue(parsedSymbol, trackedInfo);
end;
end
else
begin
trackedInfo.FullFileName := currentFile;
trackedInfo.Year := parsedYear;
trackedInfo.Month := parsedMonth;
oldestFilesPerSymbol.Add(parsedSymbol, trackedInfo);
end;
end;
end;
SetLength(Result, oldestFilesPerSymbol.Count);
var n := 0;
for trackedInfo in oldestFilesPerSymbol.Values do
begin
Result[n] := trackedInfo.FullFileName;
inc(n);
end;
finally
oldestFilesPerSymbol.Free;
end;
end;
function TAuraDataServer<T>.EnumerateAssetFiles: TArray<String>;
begin
Result := FindOldestFilesPerSymbol(FPath);
end;
function TAuraDataServer<T>.GetPath: String;
begin
Result := FPath;
end;
function TAuraDataServer<T>.LoadDataFile(const FileName: string): TFuture<TArray<TDataRecord<T>>>;
var
parsedPath, parsedSymbol: string;
parsedYear, parsedMonth: Integer;
@@ -175,9 +329,10 @@ begin
if IsMemoryLow() then
begin
var fL := FCachedFiles.Values.ToArray;
TArray.Sort<TCachedFile>(fL,
TComparer<TCachedFile>.Construct(function(const Left, Right: TCachedFile): Integer
begin Result := Sign(Left.LastUsed - Right.LastUsed); end)
TArray.Sort<TCachedFile>(
fL,
TComparer<TCachedFile>
.Construct(function(const Left, Right: TCachedFile): Integer begin Result := Sign(Left.LastUsed - Right.LastUsed); end)
);
for var i := 0 to High(fL) do
begin
@@ -208,47 +363,46 @@ begin
var capFileName := FileName;
if FileName.EndsWith('_zip', True) then
begin
Result := TFuture<TBytes>.Construct(
TLatch.Enqueue(FLoadGate), // Use shared class var FLoadGate
function: TBytes
begin
Result := TFile.ReadAllBytes(capFileName);
end
).Chain<TArray<TDataRecord<T>>>(
function(bytes: TBytes): TArray<TDataRecord<T>>
begin
var zipMemoryStream := TBytesStream.Create(bytes);
try
zipMemoryStream.Position := 0;
Result := ReadCompressedData(zipMemoryStream);
finally
zipMemoryStream.Free;
end;
end
);
Result :=
TFuture<TBytes>
.Construct(
TLatch.Enqueue(FLoadGate), // Use shared class var FLoadGate
function: TBytes begin Result := TFile.ReadAllBytes(capFileName); end)
.Chain<TArray<TDataRecord<T>>>(
function(bytes: TBytes): TArray<TDataRecord<T>>
begin
var zipMemoryStream := TBytesStream.Create(bytes);
try
zipMemoryStream.Position := 0;
Result := ReadCompressedData(zipMemoryStream);
finally
zipMemoryStream.Free;
end;
end);
end
else if TFile.Exists(FileName) then
begin
Result := TFuture<TArray<TDataRecord<T>>>.Construct(
TLatch.Enqueue(FLoadGate), // Use shared class var FLoadGate
function: TArray<TDataRecord<T>>
begin
if TFile.Exists(capFileName) then
Result :=
TFuture<TArray<TDataRecord<T>>>.Construct(
TLatch.Enqueue(FLoadGate), // Use shared class var FLoadGate
function: TArray<TDataRecord<T>>
begin
var stream := TFileStream.Create(capFileName, fmOpenRead or fmShareDenyWrite);
try
if TFile.Exists(capFileName) then
begin
var stream := TFileStream.Create(capFileName, fmOpenRead or fmShareDenyWrite);
try
Result := ReadUncompressedData(stream);
except
on E: EReadError do
raise EReadError.CreateFmt('File %s: %s', [capFileName, E.Message]);
try
Result := ReadUncompressedData(stream);
except
on E: EReadError do
raise EReadError.CreateFmt('File %s: %s', [capFileName, E.Message]);
end;
finally
stream.Free;
end;
finally
stream.Free;
end;
end;
end
);
end
);
end;
if Assigned(IsMemoryLow) then
@@ -264,7 +418,7 @@ begin
end;
end;
function TDataServer<T>.LoadDataSeries(const FileName: string): TFuture<TArray<TDataRecord<T>>>;
function TAuraDataServer<T>.LoadDataSeries(const FileName: string): TFuture<TArray<TDataRecord<T>>>;
var
loadedState: TState;
loadedFiles: TArray<TFuture<TArray<TDataRecord<T>>>>;
@@ -326,34 +480,35 @@ begin
tabFiles.Free;
end;
Result := TFuture<TArray<TDataRecord<T>>>.Construct(
loadedState,
function: TArray<TDataRecord<T>>
begin
var tickList := TList<TDataRecord<T>>.Create;
try
var overallLastTabTickTime: TDateTime := 0;
var cnt := 0;
for var P in loadedFiles do
inc(cnt, Length(P.Value));
tickList.Capacity := cnt;
Result :=
TFuture<TArray<TDataRecord<T>>>.Construct(
loadedState,
function: TArray<TDataRecord<T>>
begin
var tickList := TList<TDataRecord<T>>.Create;
try
var overallLastTabTickTime: TDateTime := 0;
var cnt := 0;
for var P in loadedFiles do
inc(cnt, Length(P.Value));
tickList.Capacity := cnt;
for var P in loadedFiles do
for var iterTickData in P.Value do
if overallLastTabTickTime < iterTickData.TimeStamp then
begin
tickList.Add(iterTickData);
overallLastTabTickTime := iterTickData.TimeStamp;
end;
Result := tickList.ToArray;
finally
tickList.Free;
end;
end
);
for var P in loadedFiles do
for var iterTickData in P.Value do
if overallLastTabTickTime < iterTickData.TimeStamp then
begin
tickList.Add(iterTickData);
overallLastTabTickTime := iterTickData.TimeStamp;
end;
Result := tickList.ToArray;
finally
tickList.Free;
end;
end
);
end;
class function TDataServer<T>.ReadCompressedData(const InputStream: TStream): TArray<TDataRecord<T>>;
class function TAuraDataServer<T>.ReadCompressedData(const InputStream: TStream): TArray<TDataRecord<T>>;
var
decompressionStream: TStream;
localHeader: TZipHeader;
@@ -392,7 +547,7 @@ begin
end;
end;
class function TDataServer<T>.ReadUncompressedData(const InputStream: TStream): TArray<TDataRecord<T>>;
class function TAuraDataServer<T>.ReadUncompressedData(const InputStream: TStream): TArray<TDataRecord<T>>;
var
fileSize: Int64;
recordCount, bytesRead: Integer;
@@ -415,7 +570,7 @@ end;
{ TAuraFileStream<T> }
constructor TAuraFileStream<T>.Create(const AFilename: String; const ADataServer: IDataServer<T>);
constructor TAuraFileStream<T>.Create(ADataServer: TAuraDataServer<T>; const AFilename: String);
begin
inherited Create;
Assert(Assigned(ADataServer));
@@ -432,7 +587,7 @@ begin
FCurrentIdx := 0;
FIsLiveData.SetValue(false);
FLastTimeStamp := 0;
FNextFileName := FindNextDataFile(FCurrentFileName);
FNextFileName := FDataServer.FindNextDataFile(FCurrentFileName);
if FNextFileName <> '' then
FNextData := FDataServer.LoadDataFile(FNextFileName)
else
@@ -459,7 +614,7 @@ begin
FCurrPosInFile := 0;
if FCurrentFileName <> '' then
begin
FNextFileName := FindNextDataFile(FCurrentFileName);
FNextFileName := FDataServer.FindNextDataFile(FCurrentFileName);
if FNextFileName <> '' then
FNextData := FDataServer.LoadDataFile(FNextFileName)
else
@@ -495,9 +650,28 @@ begin
Result := FIsLiveData;
end;
{ Helper Functions }
{ TAuraTABFileServer }
function TryParseFileName(const FileName: string; out PathValue, SymbolValue: string; out YearValue, MonthValue: Integer): Boolean;
function TAuraTABFileServer.CreateStream(const Symbol: String): IDataStream<TAskBidItem>;
var
fileName: string;
begin
// FindFirstDataFile uses the server's path and the given symbol to locate the initial data file.
fileName := FindFirstDataFile(FPath, Symbol);
// Exit if no file is found or if it is not a .tab file, preventing stream creation for invalid sources.
if (fileName = '') or (not TPath.GetExtension(fileName).StartsWith('.tab')) then
exit(nil);
// Creates a new stream instance, passing itself as the data server and the found filename.
Result := TAuraFileStream<TAskBidItem>.Create(Self, fileName);
end;
class function TAuraTABFileServer.TryParseFileName(
const FileName: string;
out PathValue, SymbolValue: string;
out YearValue, MonthValue: Integer
): Boolean;
var
fileNameNoPath: string;
nameForParsing: string;
@@ -563,92 +737,4 @@ begin
Result := True;
end;
function FindOldestFilesPerSymbol(const DirectoryPath: string): TArray<String>;
type
TTrackedFileInfo = record
FullFileName: string;
Year: Integer;
Month: Integer;
end;
var
oldestFilesPerSymbol: TDictionary<string, TTrackedFileInfo>;
fileNames: TArray<string>;
currentFile: string;
parsedPath: string;
parsedSymbol: string;
parsedYear: Integer;
parsedMonth: Integer;
trackedInfo: TTrackedFileInfo;
begin
oldestFilesPerSymbol := TDictionary<string, TTrackedFileInfo>.Create;
try
if not TDirectory.Exists(DirectoryPath) then
exit(nil);
fileNames := TDirectory.GetFiles(DirectoryPath);
for currentFile in fileNames do
begin
if TryParseFileName(currentFile, parsedPath, parsedSymbol, parsedYear, parsedMonth) then
begin
if oldestFilesPerSymbol.TryGetValue(parsedSymbol, trackedInfo) then
begin
if (parsedYear < trackedInfo.Year) or ((parsedYear = trackedInfo.Year) and (parsedMonth < trackedInfo.Month)) then
begin
trackedInfo.FullFileName := currentFile;
trackedInfo.Year := parsedYear;
trackedInfo.Month := parsedMonth;
oldestFilesPerSymbol.AddOrSetValue(parsedSymbol, trackedInfo);
end;
end
else
begin
trackedInfo.FullFileName := currentFile;
trackedInfo.Year := parsedYear;
trackedInfo.Month := parsedMonth;
oldestFilesPerSymbol.Add(parsedSymbol, trackedInfo);
end;
end;
end;
SetLength(Result, oldestFilesPerSymbol.Count);
var n := 0;
for trackedInfo in oldestFilesPerSymbol.Values do
begin
Result[n] := trackedInfo.FullFileName;
inc(n);
end;
finally
oldestFilesPerSymbol.Free;
end;
end;
function FindNextDataFile(const FileName: string): string;
var
pathValue, symbolValue: string;
yearValue, monthValue: Integer;
nextBaseName, compressedPath, uncompressedPath: string;
begin
if not TryParseFileName(FileName, pathValue, symbolValue, yearValue, monthValue) then
exit('');
Inc(monthValue);
if monthValue > 12 then
begin
monthValue := 1;
Inc(yearValue);
end;
nextBaseName := Format('%s_%.4d_%.2d.tab', [symbolValue, yearValue, monthValue]);
compressedPath := TPath.Combine(pathValue, nextBaseName + '_zip');
if TFile.Exists(compressedPath) then
exit(compressedPath);
uncompressedPath := TPath.Combine(pathValue, nextBaseName);
if TFile.Exists(uncompressedPath) then
exit(uncompressedPath);
Result := '';
end;
end.