Brummel/MycLib
1
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects 1 Releases Wiki Activity

Streamlining DataServer

Browse Source
This commit is contained in:
Michael Schimmel
2025-06-07 10:12:41 +02:00
parent 031b99acc8
commit 2cdae3d3f6
5 changed files with 148 additions and 320 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Src/Myc.Futures.pas
+10 -1
View File
@@ -22,6 +22,8 @@ type
property Done: TState read GetDone;
end;
TFuncConst<T, TResult> = reference to function(const Arg1: T): TResult;
{$REGION 'private'}
strict private
class var
@@ -47,7 +49,8 @@ type
class property Null: IFuture read FNull;
function Chain<S>(const Proc: TFunc<T, S>): TFuture<S>;
function Chain<S>(const Proc: TFunc<T, S>): TFuture<S>; overload;
function Chain<S>(const Proc: TFuncConst<T, S>): TFuture<S>; overload;
function WaitFor: T;
property Done: TState read GetDone;
@@ -84,6 +87,12 @@ begin
Result := TFuture<S>.Construct(FFuture.Done, function: S begin Result := Proc(Cap.Value); end);
end;
function TFuture<T>.Chain<S>(const Proc: TFuncConst<T, S>): TFuture<S>;
begin
var Cap := FFuture;
Result := TFuture<S>.Construct(FFuture.Done, function: S begin Result := Proc(Cap.Value); end);
end;
class function TFuture<T>.Construct(const Proc: TFunc<T>): TFuture<T>;
begin
Result := TMycGateFuncFuture<T>.Create(TaskManager, nil, Proc);
Src/Myc.Test.Trade.DataServer.pas
+50 -58
View File
@@ -14,19 +14,20 @@ uses
Myc.Trade.DataServer;
type
// Define ITickData as an alias for the specific IDataPoint interface.
ITickData = IDataPoint<ITick>;
[TestFixture]
[IgnoreMemoryLeaks(true)]
TTest_TABFileServer_Equivalence = class(TObject)
private
FServer: IDataServer<ITickData>;
FExpectedData: TArray<TAskBid.TTick>;
FServer: IDataServer<TDataPoint<TAskBidItem>>;
public
[SetupFixture]
procedure SetupFixture;
[Setup]
procedure Setup;
[TearDown]
procedure TearDown;
[TearDownFixture]
procedure TearDownFixture;
[Test]
procedure Test_ServerReturnsSameDataAs_LoadDataSeries;
@@ -39,79 +40,70 @@ uses
const
// The reference data file for this test.
C_TEST_FILENAME = '\\COFFEE\TickData\Pepperstone\AUDNZD_2024_05.tab_zip';
C_TEST_FILENAME = '\\COFFEE\TickData\Pepperstone\GER40_2025_03.tab_zip';
// The chunk size for fetching data from the server.
C_MAX_FETCH = 20;
C_MAX_FETCH = 200;
{ TTest_TABFileServer_Equivalence }
procedure TTest_TABFileServer_Equivalence.SetupFixture;
begin
// IsMemoryLow := function: Boolean begin exit(false); end;
// Load the ground truth data using the original LoadDataSeries function.
// SetupTaskManagerMock;
TAskBid.ClearCache;
end;
procedure TTest_TABFileServer_Equivalence.TearDownFixture;
begin
IsMemoryLow := nil;
end;
procedure TTest_TABFileServer_Equivalence.Setup;
begin
// SetupTaskManagerMock;
// IsMemoryLow := function: Boolean begin exit(false); end;
// 1. Load the ground truth data using the original LoadDataSeries function.
FExpectedData := TAskBid.LoadDataSeries(C_TEST_FILENAME).WaitFor;
// 2. Set up the filename and the TABFileServer instance.
FServer := TTABFileServer.Create(C_TEST_FILENAME, C_MAX_FETCH);
FServer := TTABFileServer.Create(C_TEST_FILENAME);
end;
procedure TTest_TABFileServer_Equivalence.TearDown;
begin
FServer := nil;
SetLength(FExpectedData, 0);
IsMemoryLow := nil;
TAskBid.ClearCache;
end;
procedure TTest_TABFileServer_Equivalence.Test_ServerReturnsSameDataAs_LoadDataSeries;
var
actualDataList: TList<ITickData>;
chunk: TArray<ITickData>;
newChunk: TLazy<TArray<ITickData>>;
chunk: array[0..C_MAX_FETCH-1] of TDataPoint<TAskBidItem>;
Dst: TArray<TDataPoint<TAskBidItem>>;
i: Int64;
begin
actualDataList := TList<ITickData>.Create;
try
newChunk := TLazy<TArray<ITickData>>.Construct(FServer.Data);
var ExpectedData := TAskBid.LoadDataSeries(C_TEST_FILENAME).WaitFor;
SetLength( Dst, Length( ExpectedData ) );
// Loop until the server stops providing new data chunks.
var cnt: Int64 := 0;
while not FServer.IsLiveData.Value do
begin
FServer.Fetch;
var cnt: Int64 := 0;
var n: Integer;
while not FServer.IsLiveData.Value and (cnt < Length(Dst)) do
begin
n := FServer.GetChunk(chunk);
if n > 0 then
Move( chunk[0], dst[cnt], n * sizeof( TDataPoint<TAskBidItem> ) );
inc( cnt, n );
end;
// Get the data chunk produced by the last Fetch call.
if newChunk.Pop(chunk) then
begin
for var i := 0 to High(chunk) do
begin
actualDataList.Add(chunk[i]);
Assert.AreEqual(chunk[i].Idx, cnt, 'Index mismatch');
inc(cnt);
end;
end;
end;
// --- Assertions ---
// --- Assertions ---
// 1. Verify that the total number of records matches.
Assert.AreEqual(Length(ExpectedData), cnt, 'Data record count mismatch');
// 1. Verify that the total number of records matches.
Assert.AreEqual(Length(FExpectedData), actualDataList.Count, 'Data record count mismatch');
// 2. Verify that the content of each record is identical.
for var i := 0 to High(FExpectedData) do
begin
var expectedTick := FExpectedData[i];
var actualTick := actualDataList[i];
// E2003: Removed assertion for 'Index', as it's not part of the IDataPoint interface.
// The index is an implementation detail of the faulty ConvertTicks and cannot be tested here.
Assert.AreEqual(expectedTick.OADateTime, actualTick.Time, 'Timestamp mismatch ');
Assert.IsTrue(Abs(expectedTick.Data.Ask - actualTick.Data.Ask) < 0.0001, 'Ask price mismatch');
Assert.IsTrue(Abs(expectedTick.Data.Bid - actualTick.Data.Bid) < 0.0001, 'Bid price mismatch');
end;
finally
actualDataList.Free;
// 2. Verify that the content of each record is identical.
for n := 0 to High(ExpectedData) div 1000 do
begin
i := n * 1000;
// E2003: Removed assertion for 'Index', as it's not part of the IDataPoint interface.
// The index is an implementation detail of the faulty ConvertTicks and cannot be tested here.
Assert.AreEqual(Dst[i].Idx, i, 'Index mismatch ');
Assert.AreEqual(ExpectedData[i].TimeStamp, Dst[i].Time, 'Timestamp mismatch ');
Assert.AreEqual(ExpectedData[i].Data.Ask, Dst[i].Data.Ask, 'Ask price mismatch');
Assert.AreEqual(ExpectedData[i].Data.Bid, Dst[i].Data.Bid, 'Bid price mismatch');
end;
end;
Src/Myc.Trade.DataPoint.pas
+13 -148
View File
@@ -6,163 +6,28 @@ uses
System.SysUtils;
type
IDataPoint<T> = interface
['{6A52697A-2868-42A9-982D-993542B7B360}']
function GetData: T;
function GetIdx: Int64;
function GetTime: TDateTime;
function GetVolume: Double;
property Time: TDateTime read GetTime;
property Volume: Double read GetVolume;
property Data: T read GetData;
property Idx: Int64 read GetIdx;
// A specific data record structure for Ask and Bid prices.
TAskBidItem = packed record
Ask: Single;
Bid: Single;
end;
ITick = interface
['{B342223B-D299-4A1C-82F3-569F54417A66}']
function GetAsk: Double;
function GetBid: Double;
property Ask: Double read GetAsk;
property Bid: Double read GetBid;
end;
ITickData = IDataPoint<ITick>;
IOHLC = interface
['{05374E3C-731B-44FA-A23E-051F1461B78D}']
function GetOpen: Double;
function GetHigh: Double;
function GetLow: Double;
function GetClose: Double;
property Open: Double read GetOpen;
property High: Double read GetHigh;
property Low: Double read GetLow;
property Close: Double read GetClose;
end;
IOHLCData = IDataPoint<IOHLC>;
// Abstract base class for all data points.
TDataPoint<T> = class abstract(TInterfacedObject, IDataPoint<T>)
protected
fTime: TDateTime;
fVolume: Double;
FData: T;
FIdx: Int64;
function GetTime: TDateTime;
function GetVolume: Double;
function GetData: T;
function GetIdx: Int64;
public
constructor Create(Idx: Int64; Time: TDateTime; Volume: Double; const AData: T);
end;
TTick = class(TInterfacedObject, ITick)
private
fAsk: Double;
fBid: Double;
function GetAsk: Double;
function GetBid: Double;
public
constructor Create(Ask: Double; Bid: Double);
end;
TOHLC = class(TInterfacedObject, IOHLC)
private
fOpen: Double;
fHigh: Double;
fLow: Double;
fClose: Double;
function GetClose: Double;
function GetHigh: Double;
function GetLow: Double;
function GetOpen: Double;
public
constructor Create(Open, High, Low, Close: Double);
TDataPoint<T> = record
Idx: Int64;
Time: TDateTime;
Data: T;
constructor Create(AIdx: Int64; ATime: TDateTime; const AData: T);
end;
implementation
{ TDataPoint }
constructor TDataPoint<T>.Create(Idx: Int64; Time: TDateTime; Volume: Double; const AData: T);
constructor TDataPoint<T>.Create(AIdx: Int64; ATime: TDateTime; const AData: T);
begin
inherited Create;
FIdx := Idx;
fTime := Time;
fVolume := Volume;
FData := AData;
end;
function TDataPoint<T>.GetData: T;
begin
Result := FData;
end;
function TDataPoint<T>.GetIdx: Int64;
begin
Result := FIdx;
end;
function TDataPoint<T>.GetTime: TDateTime;
begin
result := fTime;
end;
function TDataPoint<T>.GetVolume: Double;
begin
result := fVolume;
end;
{ TTick }
constructor TTick.Create(Ask: Double; Bid: Double);
begin
inherited Create;
fAsk := Ask;
fBid := Bid;
end;
function TTick.GetAsk: Double;
begin
result := fAsk;
end;
function TTick.GetBid: Double;
begin
result := fBid;
end;
{ TOHLC }
constructor TOHLC.Create(Open, High, Low, Close: Double);
begin
inherited Create;
fOpen := Open;
fHigh := High;
fLow := Low;
fClose := Close;
end;
function TOHLC.GetClose: Double;
begin
result := fClose;
end;
function TOHLC.GetHigh: Double;
begin
result := fHigh;
end;
function TOHLC.GetLow: Double;
begin
result := fLow;
end;
function TOHLC.GetOpen: Double;
begin
result := fOpen;
Idx := AIdx;
Time := ATime;
Data := AData;
end;
end.
Src/Myc.Trade.DataSeries.pas
+26 -23
View File
@@ -45,7 +45,8 @@ uses
System.Generics.Defaults,
System.IOUtils,
Myc.Futures,
Myc.Signals;
Myc.Signals,
Myc.Trade.DataPoint;
type
// Generic class for loading and managing sequential time-series data.
@@ -54,13 +55,13 @@ type
type
// Represents a single data point with a timestamp and generic data.
TDataPoint = packed record
OADateTime: TDateTime;
TimeStamp: TDateTime;
Data: T;
end;
strict private
type
TCachedFile = record
TCachedFile = class
Name: String;
Age: TDateTime;
LastUsed: TDateTime;
@@ -68,6 +69,7 @@ type
end;
class var
FCachedFiles: TDictionary<String, TCachedFile>;
FLoadGate: TLatch;
class constructor CreateClass;
class destructor DestroyClass;
@@ -76,16 +78,12 @@ type
class function ReadCompressedData(const InputStream: TStream): TArray<TDataPoint>; static;
class function ReadUncompressedData(const InputStream: TStream): TArray<TDataPoint>; static;
public
class procedure ClearCache;
// Asynchronously loads a single data file with caching support.
class function LoadDataFile(const LoadGate: TState; const FileName: string): TFuture<TArray<TDataPoint>>; static;
class function LoadDataFile(const FileName: string): TFuture<TArray<TDataPoint>>; static;
// Asynchronously loads a complete chronological series of data files.
class function LoadDataSeries(const FileName: string): TFuture<TArray<TDataPoint>>; static;
end;
// A specific data record structure for Ask and Bid prices.
TAskBidItem = packed record
Ask: Single;
Bid: Single;
end;
// A specialized TDataSeries for handling Ask/Bid tick data.
@@ -315,7 +313,7 @@ end;
class destructor TDataSeries<T>.CreateClass;
begin
FCachedFiles := TDictionary<String, TCachedFile>.Create;
FCachedFiles := TObjectDictionary<String, TCachedFile>.Create([doOwnsValues]);
end;
class destructor TDataSeries<T>.DestroyClass;
@@ -323,9 +321,14 @@ begin
FCachedFiles.Free;
end;
class procedure TDataSeries<T>.ClearCache;
begin
FCachedFiles.Clear;
end;
// Asynchronously reads tick data from a single specified file.
// Returns a TFuture that will provide an array of TAskBidSeries.
class function TDataSeries<T>.LoadDataFile(const LoadGate: TState; const FileName: string): TFuture<TArray<TDataPoint>>;
class function TDataSeries<T>.LoadDataFile(const FileName: string): TFuture<TArray<TDataPoint>>;
var
parsedPath, parsedSymbol: string;
parsedYear, parsedMonth: Integer;
@@ -351,24 +354,24 @@ begin
begin
if not IsMemoryLow then
break;
FCachedFiles.Remove(fL[i].Name);
if fL[i].Name <> FileName then
FCachedFiles.Remove(fL[i].Name);
end;
end;
var cachedFile: TCachedFile;
if FCachedFiles.TryGetValue(Filename, cachedFile) then
begin
FCachedFiles.Remove(FileName);
var age: TDateTime;
if FileAge(Filename, age, true) then
begin
if age = cachedFile.Age then
begin
cachedFile.LastUsed := Now;
FCachedFiles.Add(Filename, cachedFile);
exit(cachedFile.Data);
end;
end
else
FCachedFiles.Remove(FileName);
end;
end;
end;
@@ -379,7 +382,7 @@ begin
Result :=
TFuture<TBytes>
.Construct(
LoadGate,
TLatch.Enqueue(FLoadGate),
function: TBytes
begin
Result := TFile.ReadAllBytes(capFileName); // Read all bytes of the .zip file
@@ -401,7 +404,7 @@ begin
var capFileName := FileName;
Result :=
TFuture<TArray<TDataPoint>>.Construct(
LoadGate,
TLatch.Enqueue(FLoadGate),
function: TArray<TDataPoint>
begin
if TFile.Exists(capFileName) then
@@ -426,7 +429,7 @@ begin
if Assigned(IsMemoryLow) then
begin
var cachedFile: TCachedFile;
var cachedFile := TCachedFile.Create;
cachedFile.Name := Filename;
if FileAge(Filename, cachedFile.Age, true) then
begin
@@ -486,7 +489,7 @@ begin
// Create load tasks for the historical .tab files.
for currentFile in tabFiles do
begin
var data := LoadDataFile(TLatch.Enqueue(LoadGate), currentFile);
var data := LoadDataFile(currentFile);
loadedFileList.Add(data);
loadStates.Add(data.Done);
end;
@@ -495,7 +498,7 @@ begin
liveData := TFuture<TArray<TDataPoint>>.Null;
if liveFilePath <> '' then
begin
liveData := LoadDataFile(TLatch.Enqueue(LoadGate), liveFilePath);
liveData := LoadDataFile(liveFilePath);
loadedFileList.Add(liveData);
loadStates.Add(liveData.Done);
end;
@@ -529,10 +532,10 @@ begin
for var P in loadedFiles do
begin
for var iterTickData in P.Value do
if overallLastTabTickTime < iterTickData.OADateTime then
if overallLastTabTickTime < iterTickData.TimeStamp then
begin
tickList.Add(iterTickData);
overallLastTabTickTime := iterTickData.OADateTime;
overallLastTabTickTime := iterTickData.TimeStamp;
end;
end;
Result := tickList.ToArray;
Src/Myc.Trade.DataServer.pas
+49 -90
View File
@@ -12,24 +12,17 @@ uses
Myc.Futures; // Added for TFuture
type
IDataProvider = interface
['{B2A9996C-724B-4416-A882-595B7E58066D}']
procedure Fetch;
end;
IDataServer<T> = interface(IDataProvider)
IDataServer<T> = interface
['{A6E246A2-E84E-49AB-A63E-333E561E488C}']
function GetData: TMutable<TArray<T>>;
function GetIsLiveData: TMutable<Boolean>;
property Data: TMutable<TArray<T>> read GetData;
function GetChunk(var Data: array of T): Integer;
property IsLiveData: TMutable<Boolean> read GetIsLiveData;
end;
TDataServer<T> = class(TInterfacedObject, IDataServer<T>, IDataProvider)
TDataServer<T> = class(TInterfacedObject, IDataServer<T>)
protected
function GetData: TMutable<TArray<T>>; virtual; abstract;
function GetIsLiveData: TMutable<Boolean>; virtual; abstract;
procedure Fetch; virtual; abstract;
function GetChunk(var Data: array of T): Integer; virtual; abstract;
end;
IDataServerNode<T> = interface
@@ -43,13 +36,10 @@ type
function CreateDataServer: IDataServer<T>; virtual; abstract;
end;
TTABFileServer = class(TDataServer<ITickData>)
TTABFileServer = class(TDataServer<TDataPoint<TAskBidItem>>)
private
FFilename: String;
// The maximum number of data points to be produced per Fetch call.
FMaxFetch: Integer;
FHasNewData: TEvent;
FNeedNewChunk: Boolean;
FIsLiveData: TMutable<Boolean>.IWriteable;
// Internal state
@@ -57,21 +47,16 @@ type
FCurrentData: TFuture<TArray<TAskBid.TTick>>;
FNextFileName: string;
FNextData: TFuture<TArray<TAskBid.TTick>>;
FData: TMutable<TArray<ITickData>>;
FCurrPosInFile: Int64;
FCurrChunk: TArray<ITickData>;
FLoadGate: TLatch;
FCurrentIdx: Int64;
FLastTimeStamp: TDateTime;
function ConvertTicks(var Idx: Int64; const Ticks: TArray<TAskBid.TTick>): TArray<ITickData>;
function UpdateChunk: TArray<ITickData>;
protected
procedure Fetch; override;
function GetData: TMutable<TArray<ITickData>>; override;
function GetChunk(var Data: array of TDataPoint<TAskBidItem>): Integer; override;
function GetIsLiveData: TMutable<Boolean>; override;
public
constructor Create(const AFilename: String; AMaxFetch: Integer);
constructor Create(const AFilename: String);
procedure AfterConstruction; override;
end;
implementation
@@ -81,78 +66,45 @@ uses
{ TTABFileServer<T> }
constructor TTABFileServer.Create(const AFilename: String; AMaxFetch: Integer);
constructor TTABFileServer.Create(const AFilename: String);
begin
inherited Create;
FFilename := AFilename;
FMaxFetch := AMaxFetch;
FNextData := TFuture<TArray<TAskBid.TTick>>.Null;
FNeedNewChunk := true;
FIsLiveData := TMutable<Boolean>.CreateWriteable(false);
FHasNewData := TEvent.CreateEvent;
FData := TMutable<TArray<ITickData>>.Construct(FHasNewData.Signal, function: TArray<ITickData> begin Result := UpdateChunk; end);
end;
function TTABFileServer.ConvertTicks(var Idx: Int64; const Ticks: TArray<TAskBid.TTick>): TArray<ITickData>;
procedure TTABFileServer.AfterConstruction;
begin
SetLength(Result, Length(Ticks));
for var i := 0 to High(Ticks) do
begin
Result[i] := TDataPoint<ITick>.Create(Idx, Ticks[i].OADateTime, 1, TTick.Create(Ticks[i].Data.Ask, Ticks[i].Data.Bid));
inc(Idx);
end;
inherited;
FCurrentFileName := FFilename;
FCurrentData := TAskBid.LoadDataFile(FCurrentFileName);
FCurrPosInFile := 0;
FCurrentIdx := 0;
FIsLiveData.SetValue(false);
FLastTimeStamp := 0;
FFilename := '';
FNextFileName := FindNextDataFile(FCurrentFileName);
if FNextFileName <> '' then
FNextData := TAskBid.LoadDataFile(FNextFileName);
exit;
end;
procedure TTABFileServer.Fetch;
function TTABFileServer.GetChunk(var Data: array of TDataPoint<TAskBidItem>): Integer;
begin
// Check if the source filename has changed.
if (FFilename <> FCurrentFileName) and (FFilename <> '') then
begin
// Filename has changed, initialize by loading the new file.
FCurrentFileName := FFilename;
FCurrentData := TAskBid.LoadDataFile(TLatch.Enqueue(FLoadGate), FCurrentFileName);
FCurrPosInFile := 0;
FCurrentIdx := 0;
FIsLiveData.SetValue(false);
FLastTimeStamp := 0;
FFilename := '';
FNextFileName := FindNextDataFile(FCurrentFileName);
if FNextFileName <> '' then
FNextData := TAskBid.LoadDataFile(TLatch.Enqueue(FLoadGate), FNextFileName);
end;
FHasNewData.Notify;
FNeedNewChunk := true;
FCurrChunk := nil;
end;
function TTABFileServer.GetData: TMutable<TArray<ITickData>>;
begin
Result := FData;
end;
function TTABFileServer.GetIsLiveData: TMutable<Boolean>;
begin
Result := FIsLiveData;
end;
function TTABFileServer.UpdateChunk: TArray<ITickData>;
begin
if not FNeedNewChunk then
exit(FCurrChunk);
FNeedNewChunk := false;
Result := 0;
if FCurrentData.Done.IsSet then
begin
SetLength(FCurrChunk, FMaxFetch);
var n := 0;
while n < FMaxFetch do
var maxLen := Length(Data);
var currData := FCurrentData.Value;
while Result < MaxLen do
begin
if FCurrPosInFile >= Length(FCurrentData.Value) then
if FCurrPosInFile >= Length(currData) then
begin
// Next File!
FCurrentData := FNextData;
@@ -166,31 +118,38 @@ begin
begin
FNextFileName := FindNextDataFile(FCurrentFileName);
if FNextFileName <> '' then
FNextData := TAskBid.LoadDataFile(TLatch.Enqueue(FLoadGate), FNextFileName);
FNextData := TAskBid.LoadDataFile(FNextFileName);
if FCurrentData.Done.IsSet then
begin
currData := FCurrentData.Value;
continue;
end;
end
else
begin
// We're done with this series
FIsLiveData.SetValue(true);
end;
SetLength(FCurrChunk, n);
break;
end;
var currData := FCurrentData.Value[FCurrPosInFile];
if FLastTimeStamp < currData.OADateTime then
var item := currData[FCurrPosInFile];
if FLastTimeStamp < item.TimeStamp then
begin
FLastTimeStamp := currData.OADateTime;
FCurrChunk[n] :=
TDataPoint<ITick>.Create(FCurrentIdx, FLastTimeStamp, 1, TTick.Create(currData.Data.Ask, currData.Data.Bid));
FLastTimeStamp := item.TimeStamp;
Data[Result].Create(FCurrentIdx, FLastTimeStamp, item.Data);
inc(Result);
inc(FCurrentIdx);
inc(n);
end;
inc(FCurrPosInFile);
end;
end;
Result := FCurrChunk;
end;
function TTABFileServer.GetIsLiveData: TMutable<Boolean>;
begin
Result := FIsLiveData;
end;
end.