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

Chart now solid

Browse Source
This commit is contained in:
Michael Schimmel
2025-07-12 21:19:11 +02:00
parent 45ff69fd92
commit a06640665a
5 changed files with 264 additions and 213 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Src/Myc.Trade.DataStream.pas
+103 -50
View File
@@ -91,17 +91,17 @@ type
// Generic server for loading and managing sequential time-series data from files.
TAuraDataServer<T: record> = class(TInterfacedObject, IDataServer<T>)
private
FCachedFiles: TDataFileCache<TArray<TDataPoint<T>>>;
FCachedFiles: TDataFileCache<TArray<TArray<TDataPoint<T>>>>;
FPath: String;
FSymbols: TFuture<TDictionary<String, TAuraDataFile>>;
function GetPath: String;
// Used by cache to actually load an uncached file.
function DoLoad(const FileName: string): TFuture<TArray<TDataPoint<T>>>;
function DoLoad(const FileName: string): TFuture<TArray<TArray<TDataPoint<T>>>>;
function ProcessFile(
FileInfo: TAuraDataFile;
const DataFile: TFuture<TArray<TDataPoint<T>>>;
const DataFile: TFuture<TArray<TArray<TDataPoint<T>>>>;
const Terminated: TState;
Processor: IMycProcessor<TArray<TDataPoint<T>>>
): TState;
@@ -111,8 +111,10 @@ type
FLoadGate: TLatch;
protected
class function ReadCompressedData(const InputStream: TStream): TArray<TDataPoint<T>>; virtual; abstract;
class function ReadUncompressedData(const InputStream: TStream): TArray<TDataPoint<T>>; virtual; abstract;
// Read data from stream and return it chunked.
class function ReadCompressedData(const InputStream: TStream): TArray<TArray<TDataPoint<T>>>; virtual; abstract;
// Read data from stream and return it chunked.
class function ReadUncompressedData(const InputStream: TStream): TArray<TArray<TDataPoint<T>>>; virtual; abstract;
public
constructor Create(const APath: String);
@@ -125,7 +127,9 @@ type
function FindFirstFile(const Symbol: string): TFuture<TAuraDataFile>;
function ParseFileName(const FileName: string): TAuraDataFile; virtual; abstract;
function EnumerateSymbols: TFuture<TArray<String>>;
function LoadDataFile(const DataFile: TAuraDataFile): TFuture<TArray<TDataPoint<T>>>;
// Load data file and split content into chunks.
function LoadDataFile(const DataFile: TAuraDataFile): TFuture<TArray<TArray<TDataPoint<T>>>>;
function ProcessData(const Symbol: String; const Terminated: TState; const Processor: IMycProcessor<TArray<TDataPoint<T>>>): TState;
@@ -138,7 +142,7 @@ type
type
TDataStream = record
FileInfo: TAuraDataFile;
Data: TFuture<TArray<TDataPoint<T>>>;
Data: TFuture<TArray<TArray<TDataPoint<T>>>>;
end;
private
FDataServer: TAuraDataServer<T>;
@@ -147,6 +151,7 @@ type
FNextReady: TFlag;
FCurrent: TFuture<TDataStream>;
FNext: TFuture<TDataStream>;
FCurrChunkInFile: Int64;
FCurrPosInFile: Int64;
FLastTimeStamp: TDateTime;
procedure PreloadNext;
@@ -172,8 +177,8 @@ type
TAuraTABFileServer = class(TAuraDataServer<TAuraAskBidFileItem>)
protected
// Scans a directory and returns the oldest file found for each symbol.
class function ReadCompressedData(const InputStream: TStream): TArray<TDataPoint<TAuraAskBidFileItem>>; override;
class function ReadUncompressedData(const InputStream: TStream): TArray<TDataPoint<TAuraAskBidFileItem>>; override;
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>>>;
@@ -188,6 +193,10 @@ uses
System.StrUtils,
Myc.TaskManager;
const
// The number of records per data chunk when loading files.
ChunkSize = 8192;
{ TAuraDataFile }
constructor TAuraDataFile.Create(const APath, ASymbol, AExtension: String; AYear, AMonth: Integer);
@@ -252,8 +261,8 @@ begin
FPath := APath;
FCachedFiles :=
TDataFileCache<TArray<TDataPoint<T>>>
.Create(function(const Filename: String): TFuture<TArray<TDataPoint<T>>> begin Result := DoLoad(Filename); end);
TDataFileCache<TArray<TArray<TDataPoint<T>>>>
.Create(function(const Filename: String): TFuture<TArray<TArray<TDataPoint<T>>>> begin Result := DoLoad(Filename); end);
end;
destructor TAuraDataServer<T>.Destroy;
@@ -274,9 +283,9 @@ begin
FCachedFiles.Clear;
end;
function TAuraDataServer<T>.DoLoad(const FileName: string): TFuture<TArray<TDataPoint<T>>>;
function TAuraDataServer<T>.DoLoad(const FileName: string): TFuture<TArray<TArray<TDataPoint<T>>>>;
begin
Result := TFuture<TArray<TDataPoint<T>>>.Null;
Result := TFuture<TArray<TArray<TDataPoint<T>>>>.Null;
if not TFile.Exists(FileName) then
exit;
@@ -287,8 +296,8 @@ begin
Result :=
TFuture<TBytes>
.Construct(FLoadGate.Enqueue, function: TBytes begin Result := TFile.ReadAllBytes(capFileName); end)
.Chain<TArray<TDataPoint<T>>>(
function(bytes: TBytes): TArray<TDataPoint<T>>
.Chain<TArray<TArray<TDataPoint<T>>>>(
function(bytes: TBytes): TArray<TArray<TDataPoint<T>>>
begin
var zipMemoryStream := TBytesStream.Create(bytes);
try
@@ -302,9 +311,9 @@ begin
else
begin
Result :=
TFuture<TArray<TDataPoint<T>>>.Construct(
TFuture<TArray<TArray<TDataPoint<T>>>>.Construct(
FLoadGate.Enqueue,
function: TArray<TDataPoint<T>>
function: TArray<TArray<TDataPoint<T>>>
begin
if TFile.Exists(capFileName) then
begin
@@ -411,7 +420,7 @@ begin
end);
end;
function TAuraDataServer<T>.LoadDataFile(const DataFile: TAuraDataFile): TFuture<TArray<TDataPoint<T>>>;
function TAuraDataServer<T>.LoadDataFile(const DataFile: TAuraDataFile): TFuture<TArray<TArray<TDataPoint<T>>>>;
begin
if DataFile.IsValid then
Result := FCachedFiles.GetOrAdd(DataFile.GetFullFileName);
@@ -419,7 +428,7 @@ end;
function TAuraDataServer<T>.ProcessFile(
FileInfo: TAuraDataFile;
const DataFile: TFuture<TArray<TDataPoint<T>>>;
const DataFile: TFuture<TArray<TArray<TDataPoint<T>>>>;
const Terminated: TState;
Processor: IMycProcessor<TArray<TDataPoint<T>>>
): TState;
@@ -434,11 +443,20 @@ begin
var cTerminated := Terminated;
Result :=
DataFile.Chain(
function(const Data: TArray<TDataPoint<T>>): TState
function(const DataChunks: TArray<TArray<TDataPoint<T>>>): TState
begin
if not Processor.ProcessData(Data) then
exit(TState.Null);
Processor.Update;
if Assigned(DataChunks) then
begin
// Process each chunk, checking for termination between chunks.
for var chunk in DataChunks do
begin
if cTerminated.IsSet then
exit(TState.Null);
if not Processor.ProcessData(chunk) then
exit(TState.Null);
end;
end;
Result := ProcessFile(nextFileInfo, nextFile, cTerminated, Processor);
end
@@ -485,6 +503,7 @@ begin
);
FLastTimeStamp := 0;
FCurrChunkInFile := 0;
FCurrPosInFile := 0;
PreloadNext;
@@ -502,6 +521,8 @@ 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
@@ -516,19 +537,33 @@ begin
if not FCurrent.Value.FileInfo.IsValid then
exit;
if FCurrPosInFile >= Length(FCurrent.Value.Data.Value) then
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 currData := FCurrent.Value.Data.Value;
var maxLen := Length(Data);
while (Result < maxLen) and (FCurrPosInFile < Length(currData)) do
while (Result < maxLen) do
begin
item := currData[FCurrPosInFile];
// 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;
@@ -538,7 +573,8 @@ begin
Inc(FCurrPosInFile);
end;
if (FCurrPosInFile < Length(currData)) or FNextReady.Reset then
// Notify if there is potentially more data
if (FCurrChunkInFile < Length(fileChunks)) or FNextReady.Reset then
begin
FHasData.Notify;
end;
@@ -593,8 +629,8 @@ end;
function TAuraTABFileServer.LoadDataSeries(const InitialFile: TAuraDataFile): TFuture<TArray<TDataPoint<TAuraAskBidFileItem>>>;
var
loadedState: TState;
loadedFiles: TArray<TFuture<TArray<TDataPoint<TAuraAskBidFileItem>>>>;
liveData: TFuture<TArray<TDataPoint<TAuraAskBidFileItem>>>;
loadedFiles: TArray<TFuture<TArray<TArray<TDataPoint<TAuraAskBidFileItem>>>>>;
liveData: TFuture<TArray<TArray<TDataPoint<TAuraAskBidFileItem>>>>;
tabFiles: TList<TAuraDataFile>;
liveFile: TAuraDataFile;
currentFileInfo: TAuraDataFile;
@@ -618,7 +654,7 @@ begin
liveFile := ParseFileName(potentialLivePath);
end;
var loadedFileList := TList<TFuture<TArray<TDataPoint<TAuraAskBidFileItem>>>>.Create;
var loadedFileList := TList<TFuture<TArray<TArray<TDataPoint<TAuraAskBidFileItem>>>>>.Create;
var loadStates := TList<TState>.Create;
try
for var fileInfo in tabFiles do
@@ -628,7 +664,7 @@ begin
loadStates.Add(data.Done);
end;
liveData := TFuture<TArray<TDataPoint<TAuraAskBidFileItem>>>.Null;
liveData := TFuture<TArray<TArray<TDataPoint<TAuraAskBidFileItem>>>>.Null;
if liveFile.IsValid then
begin
liveData := LoadDataFile(liveFile);
@@ -656,16 +692,18 @@ begin
var overallLastTabTickTime: TDateTime := 0;
var cnt := 0;
for var P in loadedFiles do
inc(cnt, Length(P.Value));
for var chunk in P.Value do
inc(cnt, Length(chunk));
tickList.Capacity := cnt;
for var P in loadedFiles do
for var iterTickData in P.Value do
if overallLastTabTickTime < iterTickData.Time then
begin
tickList.Add(iterTickData);
overallLastTabTickTime := iterTickData.Time;
end;
for var chunk in P.Value do
for var iterTickData in chunk do
if overallLastTabTickTime < iterTickData.Time then
begin
tickList.Add(iterTickData);
overallLastTabTickTime := iterTickData.Time;
end;
Result := tickList.ToArray;
finally
tickList.Free;
@@ -721,14 +759,14 @@ begin
Result := TAuraDataFile.Create(path, symbol, fileExt, year, month);
end;
class function TAuraTABFileServer.ReadCompressedData(const InputStream: TStream): TArray<TDataPoint<TAuraAskBidFileItem>>;
class function TAuraTABFileServer.ReadCompressedData(const InputStream: TStream): TArray<TArray<TDataPoint<TAuraAskBidFileItem>>>;
var
decompressionStream: TStream;
localHeader: TZipHeader;
entryIndex, i: Integer;
zipFileInstance: TZipFile;
begin
SetLength(Result, 0);
Result := nil;
decompressionStream := nil;
zipFileInstance := nil;
try
@@ -760,7 +798,7 @@ begin
end;
end;
class function TAuraTABFileServer.ReadUncompressedData(const InputStream: TStream): TArray<TDataPoint<TAuraAskBidFileItem>>;
class function TAuraTABFileServer.ReadUncompressedData(const InputStream: TStream): TArray<TArray<TDataPoint<TAuraAskBidFileItem>>>;
type
TFileRecord = packed record
TimeStamp: TDateTime;
@@ -768,25 +806,40 @@ type
end;
var
fileSize: Int64;
recordCount, bytesRead: Integer;
totalRecordCount, bytesRead, chunkIndex, indexInChunk, recordIndex: Integer;
rec: TFileRecord;
begin
SetLength(Result, 0);
Result := nil;
InputStream.Position := 0;
fileSize := InputStream.Size;
if (fileSize = 0) or ((fileSize mod SizeOf(TFileRecord)) <> 0) then
exit;
recordCount := fileSize div SizeOf(TFileRecord);
if recordCount > 0 then
totalRecordCount := fileSize div SizeOf(TFileRecord);
if totalRecordCount > 0 then
begin
SetLength(Result, recordCount);
for var i := 0 to High(Result) do
var numChunks := (totalRecordCount + ChunkSize - 1) div ChunkSize;
SetLength(Result, numChunks);
chunkIndex := 0;
indexInChunk := 0;
SetLength(Result[chunkIndex], Min(ChunkSize, totalRecordCount));
for recordIndex := 0 to totalRecordCount - 1 do
begin
if indexInChunk >= ChunkSize then
begin
Inc(chunkIndex);
indexInChunk := 0;
var remainingRecords := totalRecordCount - (chunkIndex * ChunkSize);
SetLength(Result[chunkIndex], Min(ChunkSize, remainingRecords));
end;
bytesRead := InputStream.Read(rec, SizeOf(TFileRecord));
if bytesRead <> SizeOf(TFileRecord) then
raise EReadError.CreateFmt('Read error. Expected %d bytes, read %d.', [fileSize, bytesRead]);
Result[i].Create(rec.TimeStamp, rec.Data);
raise EReadError.CreateFmt('Read error. Expected %d bytes, read %d.', [SizeOf(TFileRecord), bytesRead]);
Result[chunkIndex][indexInChunk].Create(rec.TimeStamp, rec.Data);
Inc(indexInChunk);
end;
end;
end;