Michael Schimmel
585 lines
21 KiB
ObjectPascal
585 lines
21 KiB
ObjectPascal
unit Myc.Trade.DataSeries;
|
|
|
|
interface
|
|
|
|
uses
|
|
System.SysUtils,
|
|
System.Classes,
|
|
System.Generics.Collections,
|
|
System.Generics.Defaults,
|
|
System.IOUtils,
|
|
Myc.Futures,
|
|
Myc.Signals;
|
|
|
|
type
|
|
TDataSeries<T: record> = class
|
|
type
|
|
TDataPoint = packed record
|
|
OADateTime: Double;
|
|
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
|
|
class function LoadDataFile(var LoadGate: TLatch; const FileName: string): TFuture<TArray<TDataPoint>>; static;
|
|
class function LoadDataSeries(const FileName: string): TFuture<TArray<TDataPoint>>; static;
|
|
end;
|
|
|
|
TAskBidItem = packed record
|
|
Ask: Single;
|
|
Bid: Single;
|
|
end;
|
|
|
|
TAskBid = class(TDataSeries<TAskBidItem>)
|
|
type
|
|
TTick = TDataSeries<TAskBidItem>.TDataPoint;
|
|
end;
|
|
|
|
function TryParseFileName(const FileName: string; out PathValue, SymbolValue: string; out YearValue, MonthValue: Integer): Boolean;
|
|
|
|
// Finds the full filename of the oldest (=first) file for each symbol in the path.
|
|
function FindOldestFilesPerSymbol(const DirectoryPath: string): TArray<String>;
|
|
|
|
var
|
|
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;
|
|
|
|
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(var LoadGate: TLatch; 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(
|
|
TLatch.Enqueue(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(
|
|
TLatch.Enqueue(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>>;
|
|
// Local type declaration for storing file information for phase 1 processing
|
|
type
|
|
TPhase1FileEntry = record
|
|
Path: string;
|
|
Year: Integer;
|
|
Month: Integer;
|
|
end;
|
|
var
|
|
initialYear, initialMonth: Integer;
|
|
currentTabYear, currentTabMonth: Integer;
|
|
pathName, symbolName: string;
|
|
tabFileEntryRecord: TPhase1FileEntry; // Variable to construct record instances
|
|
loadedState: TState;
|
|
loadedFiles: TArray<TFuture<TArray<TDataPoint>>>;
|
|
liveData: TFuture<TArray<TDataPoint>>;
|
|
|
|
begin // Start of LoadDataSeries
|
|
if not TryParseFileName(FileName, pathName, symbolName, initialYear, initialMonth) then
|
|
begin
|
|
raise EArgumentException.CreateFmt(
|
|
'Invalid initial file name format: %s. Expected Symbol_Year_MM.Extension or Symbol_Year_MM.Extension_zip',
|
|
[FileName]);
|
|
end;
|
|
|
|
currentTabYear := initialYear;
|
|
currentTabMonth := initialMonth;
|
|
var tabBaseNameFormat: string;
|
|
var compressedTabPath, uncompressedTabPath: string;
|
|
var actualTabFileToLoad: string;
|
|
var maxTabYearFound := -1;
|
|
var maxTabMonthFound := -1;
|
|
|
|
var filesToLoad := TList<TPhase1FileEntry>.Create; // Create list with the named record type
|
|
try
|
|
while True do
|
|
begin
|
|
tabBaseNameFormat := Format('%s_%.4d_%.2d.tab', [symbolName, currentTabYear, currentTabMonth]);
|
|
compressedTabPath := TPath.Combine(pathName, tabBaseNameFormat + '_zip');
|
|
uncompressedTabPath := TPath.Combine(pathName, tabBaseNameFormat);
|
|
|
|
actualTabFileToLoad := '';
|
|
if TFile.Exists(compressedTabPath) then
|
|
begin
|
|
actualTabFileToLoad := compressedTabPath;
|
|
end
|
|
else if TFile.Exists(uncompressedTabPath) then
|
|
begin
|
|
actualTabFileToLoad := uncompressedTabPath;
|
|
end;
|
|
|
|
if actualTabFileToLoad = '' then
|
|
break; // No more sequential .tab files
|
|
|
|
// Construct the record and add it to the list
|
|
tabFileEntryRecord.Path := actualTabFileToLoad;
|
|
tabFileEntryRecord.Year := currentTabYear;
|
|
tabFileEntryRecord.Month := currentTabMonth;
|
|
filesToLoad.Add(tabFileEntryRecord);
|
|
|
|
// Advance to the next month
|
|
Inc(currentTabMonth);
|
|
if currentTabMonth > 12 then
|
|
begin
|
|
currentTabMonth := 1;
|
|
Inc(currentTabYear);
|
|
end;
|
|
end;
|
|
|
|
// Now load and process identified .tab files
|
|
var loadedFileList := TList<TFuture<TArray<TDataPoint>>>.Create;
|
|
var loadStates := TList<TState>.Create;
|
|
try
|
|
var LoadGate: TLatch;
|
|
|
|
for var tabFileEntry: TPhase1FileEntry in filesToLoad do // Iterate with the correct type
|
|
begin
|
|
var data := LoadDataFile(LoadGate, tabFileEntry.Path);
|
|
|
|
loadedFileList.Add(TFuture<TArray<TDataPoint>>.Create(data));
|
|
loadStates.Add(data.Done);
|
|
|
|
if tabFileEntry.Year > maxTabYearFound then
|
|
begin
|
|
maxTabYearFound := tabFileEntry.Year;
|
|
maxTabMonthFound := tabFileEntry.Month;
|
|
end
|
|
else if (tabFileEntry.Year = maxTabYearFound) and (tabFileEntry.Month > maxTabMonthFound) then
|
|
begin
|
|
maxTabMonthFound := tabFileEntry.Month;
|
|
end;
|
|
end;
|
|
|
|
var liveFileBaseName := Format('%s_%d_%02d.tab-live', [symbolName, maxTabYearFound, maxTabMonthFound]);
|
|
liveData := LoadDataFile(LoadGate, TPath.Combine(pathName, liveFileBaseName));
|
|
loadStates.Add(liveData.Done);
|
|
loadedFileList.Add(liveData);
|
|
|
|
loadedState := TState.All(loadStates.ToArray);
|
|
loadedFiles := loadedFileList.ToArray;
|
|
finally
|
|
loadStates.Free;
|
|
loadedFileList.Free;
|
|
end;
|
|
finally
|
|
filesToLoad.Free;
|
|
end;
|
|
|
|
Result :=
|
|
TFuture<TArray<TDataPoint>>.Construct(
|
|
loadedState,
|
|
function: TArray<TDataPoint>
|
|
begin
|
|
var tickList := TList<TDataPoint>.Create;
|
|
try
|
|
var totalTicks := Length(liveData.Value);
|
|
var overallLastTabTickTime := 0.0;
|
|
|
|
for var P in loadedFiles do
|
|
Inc(totalTicks, Length(P.Value));
|
|
|
|
tickList.Capacity := totalTicks;
|
|
|
|
for var P in loadedFiles do
|
|
begin
|
|
if Length(P.Value) > 0 then
|
|
begin
|
|
for var iterTickData in P.Value do
|
|
if iterTickData.OADateTime > overallLastTabTickTime then
|
|
begin
|
|
tickList.Add(iterTickData);
|
|
overallLastTabTickTime := iterTickData.OADateTime;
|
|
end;
|
|
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.
|