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

Data file handling

Browse Source
This commit is contained in:
Michael Schimmel
2025-06-15 20:07:18 +02:00
parent 2c56f6e750
commit f81337c98d
3 changed files with 268 additions and 246 deletions
Download Patch File Download Diff File Expand all files Collapse all files
KI/Prompt-Delphi-Entwicklung.md
+1 -1
View File
@@ -44,7 +44,7 @@
* Befolge die gängigen Delphi-Formatierungsstandards mit folgenden Ausnahmen:
- Einrückung mit 4 Leerzeichen anstelle von 2. Auch bei Kommentaren.
- Das Code-Format ist UTF-8.
- Das Code-Format ist UTF-8. Nur ASCII, keine Sonderzeichen erlaubt (insb. kein No-Break-Space!)
- Folgende Schlüsselwörter müssen klein geschrieben werden:
and, or, not, mod, div, in, as, is, array of, sizeof(), inc(), dec()
Src/Myc.Core.FileCache.pas
+9
View File
@@ -100,6 +100,8 @@ begin
FLock.Enter;
try
if IsMemoryLow() then
begin
// Caching is active. First, try to free memory if needed.
fileList := FCachedFiles.Values.ToArray;
TArray.Sort<TCachedFile>(
@@ -119,7 +121,14 @@ begin
break; // Stop evicting if memory pressure is relieved.
if fileList[i].Name <> AFileName then
begin
if fileList[i].Data.Done.IsSet then
begin
FCachedFiles.Remove(fileList[i].Name);
fileList[i].Data := TFuture<T>.Null;
end;
end;
end;
end;
// Check if a valid entry exists in the cache.
Src/Myc.Trade.DataStream.pas
+238 -225
View File
@@ -5,7 +5,7 @@ unit 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:
- TAuraFileDataRecord: A generic record for a single time-stamped data entry.
- 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.
@@ -30,6 +30,34 @@ uses
Myc.Core.FileCache;
type
// Represents metadata for a single data file.
TDataFile = record
private
FExtension: String;
FPath: String;
FSymbol: String;
FYear: Integer;
FMonth: Integer;
function GetIsValid: Boolean;
public
constructor Create(const APath, ASymbol, AExtension: String; AYear, AMonth: Integer);
function GetBaseFileName: string;
function GetFullFileName: string;
// Gets the next consecutive data file, if it exists on disk.
function GetNextFile: TDataFile;
class function ParseFileName(const FileName: string): TDataFile; static;
// Scans a directory to find the oldest file for a specific symbol.
class function FindFirst(const DirectoryPath, Symbol: string): TDataFile; static;
// Scans a directory and returns the oldest file found for each symbol.
class function FindOldestPerSymbol(const DirectoryPath: string): TArray<TDataFile>; static;
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;
property Month: Integer read FMonth;
end;
// 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.
@@ -71,35 +99,13 @@ type
['{1F8E5A9D-E92A-44C1-9F3F-C4B82A6E94B3}']
function CreateStream(const Symbol: String): IDataStream<T>;
procedure ClearCache;
end;
// A generic record for a single time-stamped data entry.
TAuraFileDataRecord<T: record> = packed record
TimeStamp: TDateTime;
Data: T;
end;
IAuraDataServer<T: record> = interface(IDataServer<T>)
['{481E27DE-DC58-49AF-B8A8-B6316980C423}']
function LoadDataFile(const FileName: string): TFuture<TArray<TDataPoint<T>>>;
function EnumerateAssets: TArray<String>;
function EnumerateSymbols: TArray<String>;
end;
// Aura files
// Generic server for loading and managing sequential time-series data from files.
TAuraDataServer<T: record> = class(TInterfacedObject, IDataServer<T>, IAuraDataServer<T>)
public type
TDataFile = record
Extension: String;
Path, Symbol: String;
Year, Month: Integer;
function GetBaseFileName: string;
function GetFullFileName: string;
function GetIsValid: Boolean;
property IsValid: Boolean read GetIsValid;
end;
TAuraDataServer<T: record> = class(TInterfacedObject, IDataServer<T>)
private
FCachedFiles: TDataFileCache<TArray<TDataPoint<T>>>;
FPath: String;
@@ -117,17 +123,10 @@ type
public
constructor Create(const APath: String);
destructor Destroy; override;
class function TryParseFileName(const FileName: string; out DataFile: TDataFile): Boolean; virtual; abstract;
function FindFirstDataFile(const Symbol: string): TDataFile;
class function FindNextDataFile(const CurrentFile: TDataFile): TDataFile;
class function FindOldestFilesPerSymbol(const DirectoryPath: string): TArray<TDataFile>;
function CreateStream(const Symbol: String): IDataStream<T>; virtual; abstract;
procedure ClearCache;
function EnumerateAssets: TArray<String>;
function LoadDataFile(const FileName: string): TFuture<TArray<TDataPoint<T>>>;
function EnumerateSymbols: TArray<String>;
function LoadDataFile(const DataFile: TDataFile): TFuture<TArray<TDataPoint<T>>>;
property Path: String read GetPath;
end;
@@ -135,9 +134,8 @@ type
TAuraFileStream<T: record> = class(TDataStream<T>, IDataStream<T>)
private
type
// Changed: Renamed "File" to "FileInfo" for clarity.
TDataStream = record
FileInfo: TAuraDataServer<T>.TDataFile;
FileInfo: TDataFile;
Data: TFuture<TArray<TDataPoint<T>>>;
end;
private
@@ -166,8 +164,7 @@ type
TAuraTABFileServer = class(TAuraDataServer<TAskBidItem>)
public
function CreateStream(const Symbol: String): IDataStream<TAskBidItem>; override;
function LoadDataSeries(const FileName: string): TFuture<TArray<TDataPoint<TAskBidItem>>>;
class function TryParseFileName(const FileName: string; out DataFile: TAuraDataServer<TAskBidItem>.TDataFile): Boolean; override;
function LoadDataSeries(const InitialFile: TDataFile): TFuture<TArray<TDataPoint<TAskBidItem>>>;
end;
implementation
@@ -178,23 +175,180 @@ uses
System.StrUtils,
Myc.TaskManager;
{ TAuraDataServer<T>.TDataFile }
{ TDataFile }
function TAuraDataServer<T>.TDataFile.GetBaseFileName: string;
constructor TDataFile.Create(const APath, ASymbol, AExtension: String; AYear, AMonth: Integer);
begin
Result := Format('%s_%.4d_%.2d', [Symbol, Year, Month]);
FPath := APath;
FSymbol := ASymbol;
FExtension := AExtension;
FYear := AYear;
FMonth := AMonth;
end;
function TAuraDataServer<T>.TDataFile.GetFullFileName: string;
function TDataFile.GetBaseFileName: string;
begin
Result := TPath.Combine(Path, GetBaseFileName + Extension);
Result := Format('%s_%.4d_%.2d', [FSymbol, FYear, FMonth]);
end;
function TAuraDataServer<T>.TDataFile.GetIsValid: Boolean;
function TDataFile.GetFullFileName: string;
begin
Result := (Symbol <> '') and (Year > 0);
Result := TPath.Combine(FPath, GetBaseFileName + FExtension);
end;
function TDataFile.GetIsValid: Boolean;
begin
Result := (FSymbol <> '') and (FYear > 0);
end;
function TDataFile.GetNextFile: TDataFile;
var
nextMonth, nextYear: Integer;
nextFile: TDataFile;
begin
if not IsValid then
exit(Default(TDataFile));
nextMonth := FMonth + 1;
nextYear := FYear;
if (nextMonth > 12) then
begin
nextMonth := 1;
Inc(nextYear);
end;
// Probe for zipped file first
nextFile := TDataFile.Create(FPath, FSymbol, '.tab_zip', nextYear, nextMonth);
if TFile.Exists(nextFile.GetFullFileName) then
exit(nextFile);
// Probe for uncompressed file
nextFile := TDataFile.Create(FPath, FSymbol, '.tab', nextYear, nextMonth);
if TFile.Exists(nextFile.GetFullFileName) then
exit(nextFile);
// No next file found
Result := Default(TDataFile);
end;
class function TDataFile.ParseFileName(const FileName: string): TDataFile;
var
fileNameNoPath, nameForParsing, baseName, ext, path, symbol: string;
year, month: Integer;
parts: TArray<string>;
begin
Result := Default(TDataFile);
path := TPath.GetDirectoryName(FileName);
fileNameNoPath := TPath.GetFileName(FileName);
nameForParsing := fileNameNoPath;
ext := '';
if nameForParsing.EndsWith('_zip', True) then
begin
ext := '_zip';
nameForParsing := nameForParsing.Substring(0, nameForParsing.Length - 4);
end;
var fileExt := TPath.GetExtension(nameForParsing) + ext;
baseName := TPath.GetFileNameWithoutExtension(nameForParsing);
parts := baseName.Split(['_']);
if Length(parts) < 3 then
exit;
if not TryStrToInt(parts[High(parts)], month) then
exit;
if (month < 1) or (month > 12) then
exit;
if not TryStrToInt(parts[High(parts) - 1], year) then
exit;
if year <= 0 then
exit;
symbol := string.Join('_', Copy(parts, 0, Length(parts) - 2));
if symbol = '' then
exit;
Result := TDataFile.Create(path, symbol, fileExt, year, month);
end;
class function TDataFile.FindFirst(const DirectoryPath, Symbol: string): TDataFile;
var
fileNames: TArray<string>;
currentFileName: string;
parsedFile: TDataFile;
begin
Result := Default(TDataFile);
if not TDirectory.Exists(DirectoryPath) then
exit;
fileNames := TDirectory.GetFiles(DirectoryPath);
for currentFileName in fileNames do
begin
parsedFile := TDataFile.ParseFileName(currentFileName);
if parsedFile.IsValid then
begin
if SameText(parsedFile.Symbol, Symbol) then
begin
// If it's the first match or older than the current result, update.
if (not Result.IsValid) or
(parsedFile.Year < Result.Year) or
((parsedFile.Year = Result.Year) and (parsedFile.Month < Result.Month)) then
begin
Result := parsedFile;
end;
end;
end;
end;
end;
class function TDataFile.FindOldestPerSymbol(const DirectoryPath: string): TArray<TDataFile>;
var
oldestFilesPerSymbol: TDictionary<string, TDataFile>;
fileNames: TArray<string>;
currentFile: string;
dataFile: TDataFile;
trackedInfo: TDataFile;
begin
oldestFilesPerSymbol := TDictionary<string, TDataFile>.Create;
try
if not TDirectory.Exists(DirectoryPath) then
exit(nil);
fileNames := TDirectory.GetFiles(DirectoryPath);
for currentFile in fileNames do
begin
dataFile := TDataFile.ParseFileName(currentFile);
if dataFile.IsValid then
begin
if oldestFilesPerSymbol.TryGetValue(dataFile.Symbol, trackedInfo) then
begin
if (dataFile.Year < trackedInfo.Year) or ((dataFile.Year = trackedInfo.Year) and (dataFile.Month < trackedInfo.Month)) then
begin
oldestFilesPerSymbol.AddOrSetValue(dataFile.Symbol, dataFile);
end;
end
else
begin
oldestFilesPerSymbol.Add(dataFile.Symbol, dataFile);
end;
end;
end;
Result := oldestFilesPerSymbol.Values.ToArray;
finally
oldestFilesPerSymbol.Free;
end;
end;
{ TAuraDataServer<T> }
constructor TAuraDataServer<T>.Create(const APath: String);
@@ -219,91 +373,12 @@ begin
FCachedFiles.Clear;
end;
function TAuraDataServer<T>.FindFirstDataFile(const Symbol: string): TDataFile;
var
oldestFiles: TArray<TDataFile>;
fileInfo: TDataFile;
begin
oldestFiles := FindOldestFilesPerSymbol(FPath);
for fileInfo in oldestFiles do
begin
if SameText(fileInfo.Symbol, Symbol) then
exit(fileInfo);
end;
Result := Default(TDataFile);
end;
class function TAuraDataServer<T>.FindNextDataFile(const CurrentFile: TDataFile): TDataFile;
var
dataFileToProbe: TDataFile;
begin
Result := Default(TDataFile);
if not CurrentFile.IsValid then
exit;
// Changed: Simplified logic, no reparsing needed.
dataFileToProbe := CurrentFile;
Inc(dataFileToProbe.Month);
if (dataFileToProbe.Month > 12) then
begin
dataFileToProbe.Month := 1;
Inc(dataFileToProbe.Year);
end;
dataFileToProbe.Extension := '.tab_zip';
if TFile.Exists(dataFileToProbe.GetFullFileName) then
exit(dataFileToProbe);
dataFileToProbe.Extension := '.tab';
if TFile.Exists(dataFileToProbe.GetFullFileName) then
exit(dataFileToProbe);
end;
class function TAuraDataServer<T>.FindOldestFilesPerSymbol(const DirectoryPath: string): TArray<TDataFile>;
var
oldestFilesPerSymbol: TDictionary<string, TDataFile>;
fileNames: TArray<string>;
currentFile: string;
dataFile: TDataFile;
trackedInfo: TDataFile;
begin
oldestFilesPerSymbol := TDictionary<string, TDataFile>.Create;
try
if not TDirectory.Exists(DirectoryPath) then
exit(nil);
fileNames := TDirectory.GetFiles(DirectoryPath);
for currentFile in fileNames do
begin
if TryParseFileName(currentFile, dataFile) then
begin
if oldestFilesPerSymbol.TryGetValue(dataFile.Symbol, trackedInfo) then
begin
if (dataFile.Year < trackedInfo.Year) or ((dataFile.Year = trackedInfo.Year) and (dataFile.Month < trackedInfo.Month)) then
begin
oldestFilesPerSymbol.AddOrSetValue(dataFile.Symbol, dataFile);
end;
end
else
begin
oldestFilesPerSymbol.Add(dataFile.Symbol, dataFile);
end;
end;
end;
Result := oldestFilesPerSymbol.Values.ToArray;
finally
oldestFilesPerSymbol.Free;
end;
end;
function TAuraDataServer<T>.EnumerateAssets: TArray<String>;
function TAuraDataServer<T>.EnumerateSymbols: TArray<String>;
var
dataFiles: TArray<TDataFile>;
i: Integer;
begin
dataFiles := FindOldestFilesPerSymbol(FPath);
dataFiles := TDataFile.FindOldestPerSymbol(FPath);
SetLength(Result, Length(dataFiles));
for i := 0 to High(dataFiles) do
Result[i] := dataFiles[i].Symbol;
@@ -367,9 +442,9 @@ begin
end;
end;
function TAuraDataServer<T>.LoadDataFile(const FileName: string): TFuture<TArray<TDataPoint<T>>>;
function TAuraDataServer<T>.LoadDataFile(const DataFile: TDataFile): TFuture<TArray<TDataPoint<T>>>;
begin
Result := FCachedFiles.GetOrAdd(Filename);
Result := FCachedFiles.GetOrAdd(DataFile.GetFullFileName);
end;
class function TAuraDataServer<T>.ReadCompressedData(const InputStream: TStream): TArray<TDataPoint<T>>;
@@ -412,25 +487,30 @@ begin
end;
class function TAuraDataServer<T>.ReadUncompressedData(const InputStream: TStream): TArray<TDataPoint<T>>;
type
TFileRecord = packed record
TimeStamp: TDateTime;
Data: T;
end;
var
fileSize: Int64;
recordCount, bytesRead: Integer;
rec: TAuraFileDataRecord<T>;
rec: TFileRecord;
begin
SetLength(Result, 0);
InputStream.Position := 0;
fileSize := InputStream.Size;
if (fileSize = 0) or ((fileSize mod SizeOf(TAuraFileDataRecord<T>)) <> 0) then
if (fileSize = 0) or ((fileSize mod SizeOf(TFileRecord)) <> 0) then
exit;
recordCount := fileSize div SizeOf(TAuraFileDataRecord<T>);
recordCount := fileSize div SizeOf(TFileRecord);
if recordCount > 0 then
begin
SetLength(Result, recordCount);
for var i := 0 to High(Result) do
begin
bytesRead := InputStream.Read(rec, SizeOf(TAuraFileDataRecord<T>));
if bytesRead <> SizeOf(TAuraFileDataRecord<T>) then
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);
end;
@@ -455,21 +535,23 @@ end;
procedure TAuraFileStream<T>.AfterConstruction;
var
firstFile: TFuture<TAuraDataServer<T>.TDataFile>;
firstFile: TFuture<TDataFile>;
serverPath: string;
begin
inherited;
firstFile := TFuture<TAuraDataServer<T>.TDataFile>.Construct(function: TAuraDataServer<T>.TDataFile begin Result := FDataServer.FindFirstDataFile(FSymbol); end);
// We need the path from the server instance to find the file.
serverPath := (FDataServer as TAuraDataServer<T>).Path;
firstFile := TFuture<TDataFile>.Construct(function: TDataFile begin Result := TDataFile.FindFirst(serverPath, FSymbol); end);
FCurrent :=
firstFile.Chain<TDataStream>(
function(const FileInfo: TAuraDataServer<T>.TDataFile): TDataStream
function(const FileInfo: TDataFile): TDataStream
begin
if FileInfo.IsValid then
begin
// Changed: Use renamed field "FileInfo".
Result.FileInfo := FileInfo;
Result.Data := FDataServer.LoadDataFile(FileInfo.GetFullFileName);
Result.Data := FDataServer.LoadDataFile(FileInfo);
Result.Data.Done.Subscribe(FHasData);
end;
end
@@ -504,7 +586,6 @@ begin
if not FCurrent.Value.Data.Done.IsSet then
exit;
// Changed: Use renamed field "FileInfo".
if not FCurrent.Value.FileInfo.IsValid then
exit;
@@ -557,16 +638,15 @@ begin
FCurrent.Chain<TDataStream>(
function(const Prev: TDataStream): TDataStream
var
nextFileInfo: TAuraDataServer<T>.TDataFile;
nextFileInfo: TDataFile;
begin
// Changed: Use renamed field "FileInfo".
if Prev.FileInfo.IsValid then
begin
nextFileInfo := TAuraDataServer<T>.FindNextDataFile(Prev.FileInfo);
nextFileInfo := Prev.FileInfo.GetNextFile;
if nextFileInfo.IsValid then
begin
Result.FileInfo := nextFileInfo;
Result.Data := FDataServer.LoadDataFile(nextFileInfo.GetFullFileName);
Result.Data := FDataServer.LoadDataFile(nextFileInfo);
Result.Data.Done.Subscribe(FHasData);
end;
end;
@@ -581,53 +661,48 @@ begin
Result := TAuraFileStream<TAskBidItem>.Create(Self, Symbol);
end;
function TAuraTABFileServer.LoadDataSeries(const FileName: string): TFuture<TArray<TDataPoint<TAskBidItem>>>;
function TAuraTABFileServer.LoadDataSeries(const InitialFile: TDataFile): TFuture<TArray<TDataPoint<TAskBidItem>>>;
var
loadedState: TState;
loadedFiles: TArray<TFuture<TArray<TDataPoint<TAskBidItem>>>>;
liveData: TFuture<TArray<TDataPoint<TAskBidItem>>>;
tabFiles: TList<string>;
liveFilePath: string;
tabFiles: TList<TDataFile>;
liveFile: TDataFile;
currentFileInfo: TDataFile;
begin
tabFiles := TList<string>.Create;
tabFiles := TList<TDataFile>.Create;
try
if TryParseFileName(FileName, currentFileInfo) then
begin
currentFileInfo := InitialFile;
while currentFileInfo.IsValid do
begin
tabFiles.Add(currentFileInfo.GetFullFileName);
currentFileInfo := FindNextDataFile(currentFileInfo);
end;
tabFiles.Add(currentFileInfo);
currentFileInfo := currentFileInfo.GetNextFile;
end;
liveFilePath := '';
liveFile := Default(TDataFile);
if tabFiles.Count > 0 then
begin
var lastTabFile := tabFiles.Last;
if TryParseFileName(lastTabFile, currentFileInfo) then
begin
var liveFileBaseName := Format('%s_%d_%02d.tab-live', [currentFileInfo.Symbol, currentFileInfo.Year, currentFileInfo.Month]);
var potentialLivePath := TPath.Combine(currentFileInfo.Path, liveFileBaseName);
var liveFileBaseName := Format('%s_%d_%02d.tab-live', [lastTabFile.Symbol, lastTabFile.Year, lastTabFile.Month]);
var potentialLivePath := TPath.Combine(lastTabFile.Path, liveFileBaseName);
if TFile.Exists(potentialLivePath) then
liveFilePath := potentialLivePath;
end;
liveFile := TDataFile.ParseFileName(potentialLivePath);
end;
var loadedFileList := TList<TFuture<TArray<TDataPoint<TAskBidItem>>>>.Create;
var loadStates := TList<TState>.Create;
try
for var fileStr in tabFiles do
for var fileInfo in tabFiles do
begin
var data := LoadDataFile(fileStr);
var data := LoadDataFile(fileInfo);
loadedFileList.Add(data);
loadStates.Add(data.Done);
end;
liveData := TFuture<TArray<TDataPoint<TAskBidItem>>>.Null;
if liveFilePath <> '' then
if liveFile.IsValid then
begin
liveData := LoadDataFile(liveFilePath);
liveData := LoadDataFile(liveFile);
loadedFileList.Add(liveData);
loadStates.Add(liveData.Done);
end;
@@ -670,66 +745,4 @@ begin
);
end;
class function TAuraTABFileServer.TryParseFileName(const FileName: string; out DataFile: TAuraDataServer<TAskBidItem>.TDataFile): Boolean;
var
fileNameNoPath, nameForParsing, baseName, ext: string;
parts: TArray<string>;
begin
Result := False;
DataFile := Default(TDataFile);
DataFile.Path := TPath.GetDirectoryName(FileName);
fileNameNoPath := TPath.GetFileName(FileName);
nameForParsing := fileNameNoPath;
ext := '';
if nameForParsing.EndsWith('_zip', True) then
begin
ext := '_zip';
nameForParsing := nameForParsing.Substring(0, nameForParsing.Length - 4);
end;
DataFile.Extension := TPath.GetExtension(nameForParsing) + ext;
baseName := TPath.GetFileNameWithoutExtension(nameForParsing);
parts := baseName.Split(['_']);
if Length(parts) < 3 then
exit;
if not TryStrToInt(parts[High(parts)], DataFile.Month) then
exit;
if (DataFile.Month < 1) or (DataFile.Month > 12) then
begin
DataFile.Month := 0;
exit;
end;
if not TryStrToInt(parts[High(parts) - 1], DataFile.Year) then
begin
DataFile.Month := 0;
exit;
end;
if DataFile.Year <= 0 then
begin
DataFile.Year := 0;
DataFile.Month := 0;
exit;
end;
DataFile.Symbol := string.Join('_', Copy(parts, 0, Length(parts) - 2));
if DataFile.Symbol = '' then
begin
DataFile.Path := '';
DataFile.Year := 0;
DataFile.Month := 0;
exit;
end;
Result := True;
end;
end.