DataSeries refactoring
This commit is contained in:
Michael Schimmel
+136
-80
@@ -1,5 +1,41 @@
|
||||
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
|
||||
@@ -12,8 +48,11 @@ uses
|
||||
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: Double;
|
||||
Data: T;
|
||||
@@ -37,26 +76,37 @@ type
|
||||
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(var LoadGate: TLatch; 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;
|
||||
|
||||
// Finds the full filename of the oldest (=first) file for each symbol in the path.
|
||||
// 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
|
||||
@@ -223,6 +273,46 @@ begin
|
||||
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;
|
||||
@@ -348,102 +438,68 @@ begin
|
||||
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
|
||||
tabFiles: TList<string>;
|
||||
currentFile: string;
|
||||
liveFilePath: string;
|
||||
pathName, symbolName: string;
|
||||
yearValue, monthValue: Integer;
|
||||
begin
|
||||
// 1. Discover all sequential .tab/.tab_zip files using the new helper function.
|
||||
tabFiles := TList<string>.Create;
|
||||
try
|
||||
while True do
|
||||
currentFile := FileName;
|
||||
if TFile.Exists(currentFile) then
|
||||
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
|
||||
// Start with the given file and find all subsequent files in the series.
|
||||
while currentFile <> '' do
|
||||
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);
|
||||
tabFiles.Add(currentFile);
|
||||
currentFile := FindNextDataFile(currentFile);
|
||||
end;
|
||||
end;
|
||||
|
||||
// Now load and process identified .tab files
|
||||
// 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
|
||||
var LoadGate: TLatch;
|
||||
|
||||
for var tabFileEntry: TPhase1FileEntry in filesToLoad do // Iterate with the correct type
|
||||
// Create load tasks for the historical .tab files.
|
||||
for currentFile in tabFiles do
|
||||
begin
|
||||
var data := LoadDataFile(LoadGate, tabFileEntry.Path);
|
||||
|
||||
loadedFileList.Add(TFuture<TArray<TDataPoint>>.Create(data));
|
||||
var data := LoadDataFile(LoadGate, currentFile);
|
||||
loadedFileList.Add(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);
|
||||
// Create a load task for the .tab-live file, if it exists.
|
||||
liveData := TFuture<TArray<TDataPoint>>.Null;
|
||||
if liveFilePath <> '' then
|
||||
begin
|
||||
liveData := LoadDataFile(LoadGate, liveFilePath);
|
||||
loadedFileList.Add(liveData);
|
||||
loadStates.Add(liveData.Done);
|
||||
end;
|
||||
|
||||
loadedState := TState.All(loadStates.ToArray);
|
||||
loadedFiles := loadedFileList.ToArray;
|
||||
@@ -452,9 +508,10 @@ begin // Start of LoadDataSeries
|
||||
loadedFileList.Free;
|
||||
end;
|
||||
finally
|
||||
filesToLoad.Free;
|
||||
tabFiles.Free;
|
||||
end;
|
||||
|
||||
// 4. Construct the final future that merges the results. This logic remains unchanged.
|
||||
Result :=
|
||||
TFuture<TArray<TDataPoint>>.Construct(
|
||||
loadedState,
|
||||
@@ -482,7 +539,6 @@ begin // Start of LoadDataSeries
|
||||
end;
|
||||
end;
|
||||
end;
|
||||
|
||||
Result := tickList.ToArray;
|
||||
finally
|
||||
tickList.Free;
|
||||
|
||||
Binary file not shown.
Reference in New Issue
Block a user