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

Streamlining main app

Browse Source 
Bugfix in Chart -> StrokeJoint set to Bevel
This commit is contained in:
Michael Schimmel
2025-07-14 19:52:40 +02:00
parent 6f0b927a05
commit d9a82365d3
8 changed files with 317 additions and 737 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Src/Myc.Trade.DataStream.pas
+1 -237
View File
@@ -1,20 +1,5 @@
unit Myc.Trade.DataStream;
(*
Myc.Trade.DataStream
Provides a data server for loading time-series data and streaming it.
This unit contains the core components for handling historical data:
- TFileRecord: A generic record for a single time-stamped data entry.
- 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
can be consumed sequentially.
- TAuraFileStream: A concrete implementation of IDataStream that is fed by an
IDataServer instance.
*)
interface
uses
@@ -30,34 +15,9 @@ uses
Myc.Core.FileCache;
type
// Represents a generic data stream capable of providing sequential data chunks.
// IsHistory:
// - true, if this stream is a history stream. Once HasData becomes false, it reached it's end and will not provide more data.
// - false, we expect more Data to come. This stream has no end.
// HasData: set, if a call to GetChunk will return new data.
IDataStream<T> = interface
['{A6E246A2-E84E-49AB-A63E-333E561E488C}']
function GetHasData: TSignal;
function GetChunk(var Data: array of TDataPoint<T>): Integer;
function GetSymbol: String;
function IsHistory: Boolean;
property HasData: TSignal read GetHasData;
property Symbol: String read GetSymbol;
end;
// Abstract base class for IDataStream implementations.
TDataStream<T> = class(TInterfacedObject, IDataStream<T>)
protected
function GetSymbol: String; virtual; abstract;
function GetHasData: TSignal; virtual; abstract;
function GetChunk(var Data: array of TDataPoint<T>): Integer; virtual; abstract;
function IsHistory: Boolean; virtual; abstract;
end;
// Interface for an instantiable data server.
IDataServer<T: record> = interface
['{1F8E5A9D-E92A-44C1-9F3F-C4B82A6E94B3}']
function CreateStream(const Symbol: String): IDataStream<T>;
procedure ClearCache;
function ProcessData(const Symbol: String; const Terminated: TState; const Processor: IMycProcessor<TArray<TDataPoint<T>>>): TState;
function EnumerateSymbols: TFuture<TArray<String>>;
@@ -120,7 +80,6 @@ type
constructor Create(const APath: String);
destructor Destroy; override;
procedure AfterConstruction; override;
function CreateStream(const Symbol: String): IDataStream<T>; virtual; abstract;
procedure ClearCache;
procedure UpdateSymbols;
// Scans a directory to find the oldest file for a specific symbol.
@@ -136,37 +95,6 @@ type
property Path: String read GetPath;
end;
// Implements a data stream that reads from Aura-specific historical data files.
TAuraFileStream<T: record> = class(TDataStream<T>, IDataStream<T>)
private
type
TDataStream = record
FileInfo: TAuraDataFile;
Data: TFuture<TArray<TArray<TDataPoint<T>>>>;
end;
private
FDataServer: TAuraDataServer<T>;
FSymbol: String;
FHasData: TEvent;
FNextReady: TFlag;
FCurrent: TFuture<TDataStream>;
FNext: TFuture<TDataStream>;
FCurrChunkInFile: Int64;
FCurrPosInFile: Int64;
FLastTimeStamp: TDateTime;
procedure PreloadNext;
protected
function GetSymbol: String; override;
function GetHasData: TSignal; override;
function GetChunk(var Data: array of TDataPoint<T>): Integer; override;
function IsHistory: Boolean; override;
public
constructor Create(ADataServer: TAuraDataServer<T>; const ASymbol: String);
destructor Destroy; override;
procedure AfterConstruction; override;
procedure BeforeDestruction; override;
end;
// Aura tick data file Ask-Bid
TAuraAskBidFileItem = packed record
@@ -180,7 +108,6 @@ type
class function ReadCompressedData(const InputStream: TStream): TArray<TArray<TDataPoint<TAuraAskBidFileItem>>>; override;
class function ReadUncompressedData(const InputStream: TStream): TArray<TArray<TDataPoint<TAuraAskBidFileItem>>>; override;
public
function CreateStream(const Symbol: String): IDataStream<TAuraAskBidFileItem>; override;
function LoadDataSeries(const InitialFile: TAuraDataFile): TFuture<TArray<TDataPoint<TAuraAskBidFileItem>>>;
function ParseFileName(const FileName: string): TAuraDataFile; override;
end;
@@ -195,7 +122,7 @@ uses
const
// The number of records per data chunk when loading files.
ChunkSize = 8192;
ChunkSize = 512;
{ TAuraDataFile }
@@ -463,169 +390,6 @@ begin
);
end;
{ TAuraFileStream<T> }
constructor TAuraFileStream<T>.Create(ADataServer: TAuraDataServer<T>; const ASymbol: String);
begin
inherited Create;
Assert(Assigned(ADataServer));
FDataServer := ADataServer;
FSymbol := ASymbol;
FNextReady := TFlag.CreateFlag;
FHasData := TEvent.CreateRouter(FNextReady.State.Signal);
end;
destructor TAuraFileStream<T>.Destroy;
begin
inherited;
end;
procedure TAuraFileStream<T>.AfterConstruction;
var
firstFile: TFuture<TAuraDataFile>;
begin
inherited;
// We need the path from the server instance to find the file.
firstFile := FDataServer.FindFirstFile(FSymbol);
FCurrent :=
firstFile.Chain<TDataStream>(
function(const FileInfo: TAuraDataFile): TDataStream
begin
if FileInfo.IsValid then
begin
Result.FileInfo := FileInfo;
Result.Data := FDataServer.LoadDataFile(FileInfo);
Result.Data.Done.Signal.Subscribe(FNextReady);
end;
end
);
FLastTimeStamp := 0;
FCurrChunkInFile := 0;
FCurrPosInFile := 0;
PreloadNext;
end;
procedure TAuraFileStream<T>.BeforeDestruction;
begin
FCurrent.WaitFor;
FCurrent.Value.Data.WaitFor;
FNext.WaitFor;
FNext.Value.Data.WaitFor;
inherited;
end;
function TAuraFileStream<T>.GetChunk(var Data: array of TDataPoint<T>): Integer;
var
item: TDataPoint<T>;
fileChunks: TArray<TArray<TDataPoint<T>>>;
currentChunk: TArray<TDataPoint<T>>;
label
loop;
begin
Result := 0;
loop:
if not FCurrent.Done.IsSet then
exit;
if not FCurrent.Value.Data.Done.IsSet then
exit;
if not FCurrent.Value.FileInfo.IsValid then
exit;
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 maxLen := Length(Data);
while (Result < maxLen) do
begin
// 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;
Data[Result].Create(FLastTimeStamp, item.Data);
Inc(Result);
end;
Inc(FCurrPosInFile);
end;
// Notify if there is potentially more data
if (FCurrChunkInFile < Length(fileChunks)) or FNextReady.Reset then
begin
FHasData.Notify;
end;
end;
function TAuraFileStream<T>.GetHasData: TSignal;
begin
Result := FHasData.Signal;
end;
function TAuraFileStream<T>.GetSymbol: String;
begin
Result := FSymbol;
end;
function TAuraFileStream<T>.IsHistory: Boolean;
begin
Result := True;
end;
procedure TAuraFileStream<T>.PreloadNext;
begin
FNextReady.Reset;
FNext :=
FCurrent.Chain<TDataStream>(
function(const Prev: TDataStream): TDataStream
var
nextFileInfo: TAuraDataFile;
begin
if Prev.FileInfo.IsValid then
begin
nextFileInfo := Prev.FileInfo.GetNextFile;
if nextFileInfo.IsValid then
begin
Result.FileInfo := nextFileInfo;
Result.Data := FDataServer.LoadDataFile(nextFileInfo);
Result.Data.Done.Signal.Subscribe(FNextReady);
end;
end;
end
);
end;
{ TAuraTABFileServer }
function TAuraTABFileServer.CreateStream(const Symbol: String): IDataStream<TAuraAskBidFileItem>;
begin
Result := TAuraFileStream<TAuraAskBidFileItem>.Create(Self, Symbol);
end;
function TAuraTABFileServer.LoadDataSeries(const InitialFile: TAuraDataFile): TFuture<TArray<TDataPoint<TAuraAskBidFileItem>>>;
var
loadedState: TState;