Files
MycLib/Src/Myc.Trade.DataSeries.pas
T
Michael Schimmel 031b99acc8 DataServer
Bugfix in DataSeries
2025-06-06 23:24:08 +02:00

637 lines
23 KiB
ObjectPascal

unit Myc.Trade.DataSeries;
(*
@unit Myc.Trade.DataSeries
@brief Provides functionality to load time-series data from binary files.
This unit offers a generic class `TDataSeries<T>` and helper functions to handle
loading and processing of sequential data files, which are often used for
historical market data.
**File Naming Convention**
The functions in this unit expect files to follow a specific naming pattern:
- Historical data: `Symbol_YYYY_MM.tab` (uncompressed) or `Symbol_YYYY_MM.tab_zip` (compressed)
- Live/update data: `Symbol_YYYY_MM.tab-live` (never compressed)
`YYYY` represents the four-digit year and `MM` the two-digit month.
If both a compressed (`_zip`) and an uncompressed version of a `.tab` file exist,
the compressed version is always preferred.
**Core Functionality**
- `TDataSeries<T>.LoadDataSeries`: The main entry point to load a complete series.
Given a starting file, it discovers all subsequent chronological `.tab` files by
calling `FindNextDataFile`. It then looks for a corresponding `.tab-live` file
for the last month in the series. All discovered files are loaded asynchronously
and merged into a single, chronologically sorted array of unique data points.
- `TDataSeries<T>.LoadDataFile`: Loads a single data file asynchronously. This
function includes a caching mechanism to avoid rereading unchanged files.
**Helper Functions**
- `FindNextDataFile`: Given a filename, it finds and returns the path to the
next file in the chronological sequence, if it exists.
- `FindOldestFilesPerSymbol`: Scans a directory and finds the oldest (first)
data file for each symbol based on year and month.
- `TryParseFileName`: A utility function to parse filenames that adhere to the
specified naming convention.
*)
interface
uses
System.SysUtils,
System.Classes,
System.Generics.Collections,
System.Generics.Defaults,
System.IOUtils,
Myc.Futures,
Myc.Signals;
type
// Generic class for loading and managing sequential time-series data.
TDataSeries<T: record> = class
public
type
// Represents a single data point with a timestamp and generic data.
TDataPoint = packed record
OADateTime: TDateTime;
Data: T;
end;
strict private
type
TCachedFile = record
Name: String;
Age: TDateTime;
LastUsed: TDateTime;
Data: TFuture<TArray<TDataPoint>>;
end;
class var
FCachedFiles: TDictionary<String, TCachedFile>;
class constructor CreateClass;
class destructor DestroyClass;
protected
class function ReadCompressedData(const InputStream: TStream): TArray<TDataPoint>; static;
class function ReadUncompressedData(const InputStream: TStream): TArray<TDataPoint>; static;
public
// Asynchronously loads a single data file with caching support.
class function LoadDataFile(const LoadGate: TState; 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.
TAskBid = class(TDataSeries<TAskBidItem>)
public
type
// Represents a single tick with Ask/Bid data.
TTick = TDataSeries<TAskBidItem>.TDataPoint;
end;
// Parses a filename to extract symbol, year, month, and path.
function TryParseFileName(const FileName: string; out PathValue, SymbolValue: string; out YearValue, MonthValue: Integer): Boolean;
// Scans a directory and finds the oldest data file for each symbol.
function FindOldestFilesPerSymbol(const DirectoryPath: string): TArray<String>;
// Finds the next chronological file in a data series, if it exists.
function FindNextDataFile(const FileName: string): string;
var
// A function pointer to a callback that indicates if the system is low on memory.
IsMemoryLow: TFunc<Boolean>;
implementation
uses
System.Zip,
System.Math,
Myc.TaskManager;
// TryParseFileName parses filenames like "Symbol_Year_MM.Extension" or "Symbol_Year_MM.Extension_zip"
// to extract Path, Symbol, Year, and Month.
function TryParseFileName(const FileName: string; out PathValue, SymbolValue: string; out YearValue, MonthValue: Integer): Boolean;
var
fileNameNoPath: string;
nameForParsing: string;
parts: TArray<string>;
baseName: string;
begin
Result := False; // Default to failure
PathValue := ''; // Initialize all out parameters
SymbolValue := '';
YearValue := 0;
MonthValue := 0;
PathValue := TPath.GetDirectoryName(FileName);
fileNameNoPath := TPath.GetFileName(FileName);
nameForParsing := fileNameNoPath;
// Handle cases like "Symbol_Year_MM.Extension_zip" by removing the trailing "_zip" before extension removal
if nameForParsing.EndsWith('_zip', True) then
begin
// Get the part before "_zip"
baseName := nameForParsing.Substring(0, nameForParsing.Length - 4);
// Get the extension of the original name (e.g. ".csv_zip")
// and check if baseName still contains an extension that needs to be removed by GetFileNameWithoutExtension
if TPath.GetExtension(baseName) <> '' then
begin
nameForParsing := baseName;
end
else
begin
// If baseName (e.g. "EURUSD_2023_01") has no extension, it is the correct parsing base
// However, TPath.GetFileNameWithoutExtension would also work correctly here,
// but this handles a potential case where a file might be named "Symbol_Year_MM_zip" (no dot before zip)
end;
end;
// Remove the actual extension (e.g., ".csv", ".dat") from "Symbol_Year_MM.Extension" or the modified "Symbol_Year_MM.Extension" part of "_zip"
nameForParsing := TPath.GetFileNameWithoutExtension(nameForParsing); // e.g., "EURUSD_2023_01"
parts := nameForParsing.Split(['_']);
if Length(parts) < 3 then // Need at least Symbol_Year_Month
exit;
// Month is the last part.
if not TryStrToInt(parts[High(parts)], MonthValue) then
begin
exit;
end;
if (MonthValue < 1) or (MonthValue > 12) then // Validate month range
begin
MonthValue := 0; // Set to invalid if out of range
exit;
end;
// Year is the second to last part.
if not TryStrToInt(parts[High(parts) - 1], YearValue) then
begin
MonthValue := 0; // Also invalidate month if year parsing fails
exit;
end;
// Ensure YearValue is plausible, e.g. not 0 or negative, adjust as needed for your data's year range
if YearValue <= 0 then
begin
YearValue := 0;
MonthValue := 0;
exit;
end;
// Symbol consists of all parts before Year and Month.
SymbolValue := string.Join('_', Copy(parts, 0, Length(parts) - 2));
if SymbolValue = '' then // Ensure symbol is not empty
begin
// Reset all out parameters on failure before exit
PathValue := '';
YearValue := 0;
MonthValue := 0;
exit;
end;
Result := True;
end;
// Finds the oldest file for each symbol in the given directory.
// The file extension is ignored for the purpose of identifying the symbol, year, and month.
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; // Dictionary to track the oldest file per symbol
try
if not TDirectory.Exists(DirectoryPath) then
begin
// Or raise an exception, depending on desired error handling
exit;
end;
fileNames := TDirectory.GetFiles(DirectoryPath); // Get all files in the directory
for currentFile in fileNames do
begin
// Attempt to parse the file name
if TryParseFileName(currentFile, parsedPath, parsedSymbol, parsedYear, parsedMonth) then
begin
// Check if this symbol is already tracked or if this file is older
if oldestFilesPerSymbol.TryGetValue(parsedSymbol, trackedInfo) then
begin
// If current file's year is less, or year is same and month is less, it's older
if (parsedYear < trackedInfo.Year) or ((parsedYear = trackedInfo.Year) and (parsedMonth < trackedInfo.Month)) then
begin
trackedInfo.FullFileName := currentFile; // Store full path to the file
trackedInfo.Year := parsedYear;
trackedInfo.Month := parsedMonth;
oldestFilesPerSymbol.AddOrSetValue(parsedSymbol, trackedInfo); // Update the entry for this symbol
end;
end
else
begin
// First time we see this symbol, so it's the oldest so far
trackedInfo.FullFileName := currentFile; // Store full path to the file
trackedInfo.Year := parsedYear;
trackedInfo.Month := parsedMonth;
oldestFilesPerSymbol.Add(parsedSymbol, trackedInfo); // Add new entry for this symbol
end;
end;
end;
// Populate the result array with the full names of the oldest files
SetLength(Result, oldestFilesPerSymbol.Count);
var n := 0;
for trackedInfo in oldestFilesPerSymbol.Values do
begin
Result[n] := trackedInfo.FullFileName;
inc(n);
end;
finally
oldestFilesPerSymbol.Free; // Free the dictionary
end;
end;
function FindNextDataFile(const FileName: string): string;
var
pathValue, symbolValue: string;
yearValue, monthValue: Integer;
nextBaseName, compressedPath, uncompressedPath: string;
begin
// Parse the input filename to extract its components.
if not TryParseFileName(FileName, pathValue, symbolValue, yearValue, monthValue) then
begin
exit(''); // If parsing fails, there is no 'next' file.
end;
// Calculate the next month and year.
Inc(monthValue);
if monthValue > 12 then
begin
monthValue := 1;
Inc(yearValue);
end;
// Construct the base name for the next file.
nextBaseName := Format('%s_%.4d_%.2d.tab', [symbolValue, yearValue, monthValue]);
// Check for the existence of the next file, preferring the compressed version.
compressedPath := TPath.Combine(pathValue, nextBaseName + '_zip');
if TFile.Exists(compressedPath) then
begin
exit(compressedPath);
end;
uncompressedPath := TPath.Combine(pathValue, nextBaseName);
if TFile.Exists(uncompressedPath) then
begin
exit(uncompressedPath);
end;
// If no next file is found, return an empty string.
Result := '';
end;
class destructor TDataSeries<T>.CreateClass;
begin
FCachedFiles := TDictionary<String, TCachedFile>.Create;
end;
class destructor TDataSeries<T>.DestroyClass;
begin
FCachedFiles.Free;
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>>;
var
parsedPath, parsedSymbol: string;
parsedYear, parsedMonth: Integer;
begin
Result := TFuture<TArray<TDataPoint>>.Null;
// Attempt to parse year and month from the filename
if not TryParseFileName(FileName, parsedPath, parsedSymbol, parsedYear, parsedMonth) then
exit;
if Assigned(IsMemoryLow) then
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)
);
for var i := 0 to High(fL) do
begin
if not IsMemoryLow then
break;
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;
end;
end;
if FileName.EndsWith('_zip', True) then
begin
var capFileName := FileName;
Result :=
TFuture<TBytes>
.Construct(
LoadGate,
function: TBytes
begin
Result := TFile.ReadAllBytes(capFileName); // Read all bytes of the .zip file
end)
.Chain<TArray<TDataPoint>>(
function(bytes: TBytes): TArray<TDataPoint>
begin
var zipMemoryStream := TBytesStream.Create(bytes); // Create a memory stream to hold the zip file content
try
zipMemoryStream.Position := 0;
Result := ReadCompressedData(zipMemoryStream);
finally
zipMemoryStream.Free;
end;
end);
end
else if TFile.Exists(FileName) then
begin
var capFileName := FileName;
Result :=
TFuture<TArray<TDataPoint>>.Construct(
LoadGate,
function: TArray<TDataPoint>
begin
if TFile.Exists(capFileName) then
begin
var stream := TFileStream.Create(capFileName, fmOpenRead or fmShareDenyWrite);
try
try
Result := ReadUncompressedData(stream);
except
on E: EReadError do
begin
raise EReadError.CreateFmt('File %s: %s', [capFileName, E.Message]);
end;
end;
finally
stream.Free;
end;
end;
end
);
end;
if Assigned(IsMemoryLow) then
begin
var cachedFile: TCachedFile;
cachedFile.Name := Filename;
if FileAge(Filename, cachedFile.Age, true) then
begin
cachedFile.Data := Result;
cachedFile.LastUsed := Now;
FCachedFiles.Add(Filename, cachedFile);
end;
end;
end;
class function TDataSeries<T>.LoadDataSeries(const FileName: string): TFuture<TArray<TDataPoint>>;
var
loadedState: TState;
loadedFiles: TArray<TFuture<TArray<TDataPoint>>>;
liveData: TFuture<TArray<TDataPoint>>;
tabFiles: TList<string>;
currentFile: string;
liveFilePath: string;
pathName, symbolName: string;
yearValue, monthValue: Integer;
LoadGate: TLatch;
begin
// 1. Discover all sequential .tab/.tab_zip files using the new helper function.
tabFiles := TList<string>.Create;
try
currentFile := FileName;
if TFile.Exists(currentFile) then
begin
// Start with the given file and find all subsequent files in the series.
while currentFile <> '' do
begin
tabFiles.Add(currentFile);
currentFile := FindNextDataFile(currentFile);
end;
end;
// 2. Determine the corresponding .tab-live file based on the last found .tab file.
liveFilePath := '';
if tabFiles.Count > 0 then
begin
var lastTabFile := tabFiles.Last;
if TryParseFileName(lastTabFile, pathName, symbolName, yearValue, monthValue) then
begin
var liveFileBaseName := Format('%s_%d_%02d.tab-live', [symbolName, yearValue, monthValue]);
var potentialLivePath := TPath.Combine(pathName, liveFileBaseName);
if TFile.Exists(potentialLivePath) then
begin
liveFilePath := potentialLivePath;
end;
end;
end;
// 3. Asynchronously load all discovered files.
var loadedFileList := TList<TFuture<TArray<TDataPoint>>>.Create;
var loadStates := TList<TState>.Create;
try
// Create load tasks for the historical .tab files.
for currentFile in tabFiles do
begin
var data := LoadDataFile(TLatch.Enqueue(LoadGate), currentFile);
loadedFileList.Add(data);
loadStates.Add(data.Done);
end;
// Create a load task for the .tab-live file, if it exists.
liveData := TFuture<TArray<TDataPoint>>.Null;
if liveFilePath <> '' then
begin
liveData := LoadDataFile(TLatch.Enqueue(LoadGate), liveFilePath);
loadedFileList.Add(liveData);
loadStates.Add(liveData.Done);
end;
loadedState := TState.All(loadStates.ToArray);
loadedFiles := loadedFileList.ToArray;
finally
loadStates.Free;
loadedFileList.Free;
end;
finally
tabFiles.Free;
end;
// 4. Construct the final future that merges the results. This logic remains unchanged.
Result :=
TFuture<TArray<TDataPoint>>.Construct(
loadedState,
function: TArray<TDataPoint>
begin
var tickList := TList<TDataPoint>.Create;
try
var totalTicks := Length(liveData.Value);
var overallLastTabTickTime: TDateTime := 0;
for var P in loadedFiles do
Inc(totalTicks, Length(P.Value));
tickList.Capacity := totalTicks;
for var P in loadedFiles do
begin
for var iterTickData in P.Value do
if overallLastTabTickTime < iterTickData.OADateTime then
begin
tickList.Add(iterTickData);
overallLastTabTickTime := iterTickData.OADateTime;
end;
end;
Result := tickList.ToArray;
finally
tickList.Free;
end;
end
);
end;
class function TDataSeries<T>.ReadCompressedData(const InputStream: TStream): TArray<TDataPoint>;
var
decompressionStream: TStream; // Holds uncompressed data of a single zip entry
localHeader: TZipHeader;
entryIndex: Integer;
i: Integer;
zipFileInstance: TZipFile;
begin
// Initialize Result record
SetLength(Result, 0);
decompressionStream := nil;
zipFileInstance := nil;
try
InputStream.Position := 0; // Reset stream position to the beginning for TZipFile
// Open TZipFile from this memory stream
zipFileInstance := TZipFile.Create;
zipFileInstance.Open(InputStream, TZipMode.zmRead); // Open the zip from the memory stream
if zipFileInstance.FileCount = 0 then
begin
// If TZipFile opens an empty or invalid stream successfully but finds no files
exit; // FileSuccessfullyLoaded remains False
end;
entryIndex := -1;
for i := 0 to zipFileInstance.FileCount - 1 do
begin
if zipFileInstance.FileNames[i].EndsWith('.tab', True) then
begin
entryIndex := i;
break;
end;
end;
if entryIndex = -1 then // If specific name not found, default to the first entry
begin
if zipFileInstance.FileCount > 0 then
entryIndex := 0
else
exit; // Should not happen if FileCount > 0 check passed
end;
// Decompress the specific entry from the in-memory zip file.
// TZipFile.Read creates and populates decompressionStream with uncompressed data.
zipFileInstance.Read(entryIndex, decompressionStream, localHeader);
if not Assigned(decompressionStream) then
exit; // Exit if no valid stream could be prepared
Result := ReadUncompressedData(decompressionStream);
finally
// Ensure all potentially created objects are freed
if Assigned(decompressionStream) then // Holds uncompressed data from a zip entry
decompressionStream.Free;
if Assigned(zipFileInstance) then
zipFileInstance.Free;
end;
end;
class function TDataSeries<T>.ReadUncompressedData(const InputStream: TStream): TArray<TDataPoint>;
var
fileSize: Int64;
recordCount: Integer;
bytesRead: Integer;
begin
// Initialize Result record
SetLength(Result, 0);
InputStream.Position := 0;
fileSize := InputStream.Size;
if fileSize = 0 then
exit;
if (fileSize mod SizeOf(TDataPoint)) <> 0 then
raise EReadError.CreateFmt('File corrupted. Size %d is not a multiple of record size %d.', [fileSize, SizeOf(TDataPoint)]);
recordCount := fileSize div SizeOf(TDataPoint);
if recordCount > 0 then
begin
SetLength(Result, recordCount);
bytesRead := InputStream.Read(Result[0], fileSize);
if bytesRead <> fileSize then
raise EReadError.CreateFmt('Read error. Expected to read %d bytes, but actually read %d bytes.', [fileSize, bytesRead]);
end;
end;
end.