diff --git a/KI/Prompt-Delphi-Entwicklung.txt b/KI/Prompt-Delphi-Entwicklung.txt index 750d55a..7b343ab 100644 --- a/KI/Prompt-Delphi-Entwicklung.txt +++ b/KI/Prompt-Delphi-Entwicklung.txt @@ -39,7 +39,7 @@ Code-Generierung - Einrückung mit 4 Leerzeichen anstelle von 2. Auch bei Kommentaren. - Das Code-Format ist UTF-8. - Folgende Schlüsselwörter müssen klein geschrieben werden: - and, or, not, mod, div, in, as, is, array of, sizeof + and, or, not, mod, div, in, as, is, array of, sizeof(), inc(), dec() * Ändere niemals vorhandene Bezeichner im vom Benutzer bereitgestellten Code, es sei denn du wirst dazu aufgefordert. diff --git a/Src/Myc.Core.DataCache.pas b/Src/Myc.Core.DataCache.pas deleted file mode 100644 index 0c2b5e4..0000000 --- a/Src/Myc.Core.DataCache.pas +++ /dev/null @@ -1,172 +0,0 @@ -unit Myc.Core.DataCache; - -interface - -uses - System.SysUtils, - System.Generics.Collections, - System.Generics.Defaults, - Myc.Futures, // Required for TFuture - Myc.Signals; // Required for TLatch - -type - // Represents a single cached file entry with its metadata and loaded data future. - // TArrayItemType is the type of a single item within the TArray that is loaded by the future. - TCachedFile = class - Name: String; // The full path and name of the cached file. - Age: TDateTime; // The modification timestamp of the file at the time of caching. - LastUsed: TDateTime; // The last time this cached entry was accessed. - Data: TFuture>; // A future representing the asynchronously loaded data. - end; - - // Generic in-memory cache for data files, supporting asynchronous loading - // and eviction based on memory pressure and last usage. - // TArrayItemType is the type of a single item within the TArray that is loaded by the future, - // e.g., TDataPoint. - TDataCache = class - strict private - FCachedFiles: TDictionary>; - FLoadGate: TLatch; - FIsMemoryLow: TFunc; - - private - // Manages cache size by evicting least recently used items if memory is low. - procedure ManageCacheSize(const CurrentFileNameBeingLoaded: string); - - public - constructor Create(const ALoadGate: TLatch; const AIsMemoryLow: TFunc); - destructor Destroy; override; - - // Clears all entries from the cache. - procedure Clear; - - // Attempts to retrieve data for a given file from the cache. - // If found and not stale, updates LastUsed and returns true. - // Otherwise, returns false. - function TryGetValue(const FileName: string; out ACachedFile: TCachedFile): Boolean; - - // Adds a new or updates an existing entry in the cache. - procedure AddOrUpdate(const FileName: string; const AAge: TDateTime; const AData: TFuture>); - - // Returns the TLoadGate associated with this cache, for coordination of load operations. - function GetLoadGate: TLatch; inline; - - property LoadGate: TLatch read GetLoadGate; - end; - -implementation - -uses - System.Math; - -{ TDataCache } - -constructor TDataCache.Create(const ALoadGate: TLatch; const AIsMemoryLow: TFunc); -begin - inherited Create; - FCachedFiles := TObjectDictionary>.Create([doOwnsValues]); - FLoadGate := ALoadGate; - FIsMemoryLow := AIsMemoryLow; -end; - -destructor TDataCache.Destroy; -begin - FCachedFiles.Free; - inherited; -end; - -procedure TDataCache.Clear; -begin - FCachedFiles.Clear; -end; - -procedure TDataCache.AddOrUpdate( - const FileName: string; - const AAge: TDateTime; - const AData: TFuture> -); -var - cachedFile: TCachedFile; -begin - if FCachedFiles.TryGetValue(FileName, cachedFile) then - begin - // Update existing entry - cachedFile.Age := AAge; - cachedFile.Data := AData; // This will replace the old future, potentially freeing old data - cachedFile.LastUsed := Now; - end - else - begin - // Add new entry - cachedFile := TCachedFile.Create; - cachedFile.Name := FileName; - cachedFile.Age := AAge; - cachedFile.Data := AData; - cachedFile.LastUsed := Now; - FCachedFiles.Add(FileName, cachedFile); - end; -end; - -function TDataCache.GetLoadGate: TLatch; -begin - Result := FLoadGate; -end; - -procedure TDataCache.ManageCacheSize(const CurrentFileNameBeingLoaded: string); -begin - if Assigned(FIsMemoryLow) and FIsMemoryLow() then - begin - var fL := FCachedFiles.Values.ToArray; - // Sorts the cached files by LastUsed in ascending order (least recently used first). - TArray.Sort>( - fL, - TComparer>.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 FIsMemoryLow() then // Stop evicting if memory pressure is relieved - break; - // Do not remove the file that is currently being loaded. - if fL[i].Name <> CurrentFileNameBeingLoaded then - FCachedFiles.Remove(fL[i].Name); - end; - end; -end; - -function TDataCache.TryGetValue(const FileName: string; out ACachedFile: TCachedFile): Boolean; -var - age: TDateTime; -begin - ManageCacheSize( FileName ); - - Result := FCachedFiles.TryGetValue(FileName, ACachedFile); - if Result then - begin - // Check if the file on disk has changed since it was cached. - if FileAge(FileName, age, true) then - begin - if age = ACachedFile.Age then - begin - // Cache hit and not stale, update LastUsed. - ACachedFile.LastUsed := Now; - end - else - begin - // File is stale, remove from cache. - FCachedFiles.Remove(FileName); - Result := False; - end; - end - else - begin - // File no longer exists, remove from cache. - FCachedFiles.Remove(FileName); - Result := False; - end; - end; -end; - -end. diff --git a/Src/Myc.Futures.pas b/Src/Myc.Futures.pas index 0c99d84..67cfe74 100644 --- a/Src/Myc.Futures.pas +++ b/Src/Myc.Futures.pas @@ -22,7 +22,7 @@ type property Done: TState read GetDone; end; - TFuncConst = reference to function(const Arg1: T): TResult; + TFuncConst = reference to function(const Arg1: S): TResult; {$REGION 'private'} strict private diff --git a/Src/Myc.Test.Trade.DataServer.pas b/Src/Myc.Test.Trade.DataServer.pas index 7f61336..0454073 100644 --- a/Src/Myc.Test.Trade.DataServer.pas +++ b/Src/Myc.Test.Trade.DataServer.pas @@ -10,8 +10,7 @@ uses Myc.Lazy, Myc.Futures, Myc.Trade.DataPoint, - Myc.Trade.DataSeries, - Myc.Trade.DataServer; + Myc.Trade.DataStream; type // Test fixture for TDataServer and TAuraTABFileServer, focusing on data equivalence @@ -20,7 +19,8 @@ type [IgnoreMemoryLeaks(true)] TTest_TABFileServer_Equivalence = class(TObject) private - FServer: IDataServer>; + FServer: IDataServer; + FStream: IDataStream>; public [SetupFixture] procedure SetupFixture; @@ -50,12 +50,15 @@ const procedure TTest_TABFileServer_Equivalence.SetupFixture; begin + FServer := TAskBidServer.Create; // TAskBid.ClearCache ensures a clean state for data series loading. - TAskBid.ClearCache; + FServer.ClearCache; end; procedure TTest_TABFileServer_Equivalence.TearDownFixture; begin + FServer := nil; + // Resets the global IsMemoryLow function pointer, if it was set for testing. IsMemoryLow := nil; end; @@ -63,15 +66,15 @@ end; procedure TTest_TABFileServer_Equivalence.Setup; begin // Initializes the data server with a specific test file. - FServer := TAuraTABFileServer.Create(C_TEST_FILENAME); + FStream := TAuraTABFileStream.Create(C_TEST_FILENAME, FServer); end; procedure TTest_TABFileServer_Equivalence.TearDown; begin // Releases the data server instance. - FServer := nil; + FStream := nil; // Clears any cached data to prevent interference with subsequent tests. - TAskBid.ClearCache; + FServer.ClearCache; end; procedure TTest_TABFileServer_Equivalence.Test_ServerReturnsSameDataAs_LoadDataSeries; @@ -81,15 +84,15 @@ var i: Int64; begin // Loads the expected data directly using TAskBid.LoadDataSeries.WaitFor. - var ExpectedData := TAskBid.LoadDataSeries(C_TEST_FILENAME).WaitFor; + var ExpectedData := FServer.LoadDataSeries(C_TEST_FILENAME).WaitFor; SetLength(Dst, Length(ExpectedData)); var cnt: Int64 := 0; var n: Integer; // Fetches data chunks from the server until all data is retrieved or server indicates live data. - while not FServer.IsLiveData.Value and (cnt < Length(Dst)) do + while not FStream.IsLiveData.Value and (cnt < Length(Dst)) do begin - n := FServer.GetChunk(chunk); + n := FStream.GetChunk(chunk); if n > 0 then Move(chunk[0], dst[cnt], n * sizeof(TDataPoint)); inc(cnt, n); diff --git a/Src/Myc.Trade.DataSeries.pas b/Src/Myc.Trade.DataSeries.pas deleted file mode 100644 index 099c964..0000000 --- a/Src/Myc.Trade.DataSeries.pas +++ /dev/null @@ -1,757 +0,0 @@ -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` 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.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.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, - Myc.Lazy, - Myc.Trade.DataPoint, - Myc.Trade.DataServer; - -type - // Generic class for loading and managing sequential time-series data. - TDataSeries = class - public - type - // Represents a single data point with a timestamp and generic data. - TDataPoint = packed record - TimeStamp: TDateTime; - Data: T; - end; - - strict private - type - TCachedFile = class - Name: String; - Age: TDateTime; - LastUsed: TDateTime; - Data: TFuture>; - end; - class var - FCachedFiles: TDictionary; - FLoadGate: TLatch; - - class constructor CreateClass; - class destructor DestroyClass; - - protected - class function ReadCompressedData(const InputStream: TStream): TArray; static; - class function ReadUncompressedData(const InputStream: TStream): TArray; static; - public - class procedure ClearCache; - // Asynchronously loads a single data file with caching support. - class function LoadDataFile(const FileName: string): TFuture>; static; - // Asynchronously loads a complete chronological series of data files. - class function LoadDataSeries(const FileName: string): TFuture>; static; - - end; - - // A specialized TDataSeries for handling Ask/Bid tick data. - TAskBid = class(TDataSeries) - public - type - // Represents a single tick with Ask/Bid data. - TTick = TDataSeries.TDataPoint; - end; - - // Implements a data server that reads data from Aura-specific historical data files. - // It handles sequential loading of data files and transitions to "live data" mode - // when all historical files are exhausted. - TAuraFileServer = class(TDataServer>, IDataServer>) - private - FIsLiveData: TMutable.IWriteable; - FCurrentFileName: string; - FCurrentData: TFuture.TDataPoint>>; - FNextFileName: string; - FNextData: TFuture.TDataPoint>>; - FCurrPosInFile: Int64; - FCurrentIdx: Int64; - FLastTimeStamp: TDateTime; - - protected - function GetChunk(var Data: array of TDataPoint): Integer; override; - function GetIsLiveData: TMutable; override; - public - constructor Create(const AFilename: String); - procedure AfterConstruction; override; - end; - - // Specialized TAuraFileServer for TAskBidItem data. - TAuraTABFileServer = TAuraFileServer; - -// 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; - -// 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; - -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; - 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; -type - TTrackedFileInfo = record - FullFileName: string; - Year: Integer; - Month: Integer; - end; -var - oldestFilesPerSymbol: TDictionary; - fileNames: TArray; - currentFile: string; - parsedPath: string; - parsedSymbol: string; - parsedYear: Integer; - parsedMonth: Integer; - trackedInfo: TTrackedFileInfo; -begin - oldestFilesPerSymbol := TDictionary.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.CreateClass; -begin - FCachedFiles := TObjectDictionary.Create([doOwnsValues]); -end; - -class destructor TDataSeries.DestroyClass; -begin - FCachedFiles.Free; -end; - -class procedure TDataSeries.ClearCache; -begin - FCachedFiles.Clear; -end; - -// Asynchronously reads tick data from a single specified file. -// Returns a TFuture that will provide an array of TAskBidSeries. -class function TDataSeries.LoadDataFile(const FileName: string): TFuture>; -var - parsedPath, parsedSymbol: string; - parsedYear, parsedMonth: Integer; -begin - Result := TFuture>.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( - fL, - TComparer - .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; - if fL[i].Name <> FileName then - FCachedFiles.Remove(fL[i].Name); - end; - end; - - var cachedFile: TCachedFile; - if FCachedFiles.TryGetValue(Filename, cachedFile) then - begin - var age: TDateTime; - if FileAge(Filename, age, true) then - begin - if age = cachedFile.Age then - begin - cachedFile.LastUsed := Now; - exit(cachedFile.Data); - end - else - FCachedFiles.Remove(FileName); - end; - end; - end; - - if FileName.EndsWith('_zip', True) then - begin - var capFileName := FileName; - Result := - TFuture - .Construct( - TLatch.Enqueue(FLoadGate), - function: TBytes - begin - Result := TFile.ReadAllBytes(capFileName); // Read all bytes of the .zip file - end) - .Chain>( - function(bytes: TBytes): TArray - 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>.Construct( - TLatch.Enqueue(FLoadGate), - function: TArray - 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.Create; - 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.LoadDataSeries(const FileName: string): TFuture>; -var - loadedState: TState; - loadedFiles: TArray>>; - liveData: TFuture>; - tabFiles: TList; - 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.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>>.Create; - var loadStates := TList.Create; - try - // Create load tasks for the historical .tab files. - for currentFile in tabFiles do - begin - var data := LoadDataFile(currentFile); - loadedFileList.Add(data); - loadStates.Add(data.Done); - end; - - // Create a load task for the .tab-live file, if it exists. - liveData := TFuture>.Null; - if liveFilePath <> '' then - begin - liveData := LoadDataFile(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>.Construct( - loadedState, - function: TArray - begin - var tickList := TList.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.TimeStamp then - begin - tickList.Add(iterTickData); - overallLastTabTickTime := iterTickData.TimeStamp; - end; - end; - Result := tickList.ToArray; - finally - tickList.Free; - end; - end - ); -end; - -class function TDataSeries.ReadCompressedData(const InputStream: TStream): TArray; -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.ReadUncompressedData(const InputStream: TStream): TArray; -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; - -{ TAuraFileServer } - -constructor TAuraFileServer.Create(const AFilename: String); -begin - inherited Create; - FCurrentFileName := AFilename; - FIsLiveData := TMutable.CreateWriteable(false); -end; - -procedure TAuraFileServer.AfterConstruction; -begin - inherited; - - // Asynchronously loads the initial data file. - FCurrentData := TDataSeries.LoadDataFile(FCurrentFileName); - FCurrPosInFile := 0; - FCurrentIdx := 0; - // Initially sets the server to "not live data". - FIsLiveData.SetValue(false); - FLastTimeStamp := 0; - - // Determines and asynchronously loads the next chronological file, if available. - FNextFileName := FindNextDataFile(FCurrentFileName); - if FNextFileName <> '' then - FNextData := TDataSeries.LoadDataFile(FNextFileName); -end; - -function TAuraFileServer.GetChunk(var Data: array of TDataPoint): Integer; -begin - Result := 0; - - // Checks if the current data file is ready. - if FCurrentData.Done.IsSet then - begin - var maxLen := Length(Data); - - var currData := FCurrentData.Value; - while Result < MaxLen do - begin - if FCurrPosInFile >= Length(currData) then - begin - // Transition to the next file if the current one is exhausted. - FCurrentData := FNextData; - FCurrentFileName := FNextFileName; - FCurrPosInFile := 0; - - FNextFileName := ''; - FNextData := TFuture.TDataPoint>>.Null; - - if FCurrentFileName <> '' then - begin - // Loads the next file in the series. - FNextFileName := FindNextDataFile(FCurrentFileName); - if FNextFileName <> '' then - FNextData := TDataSeries.LoadDataFile(FNextFileName); - - // If the newly assigned current data is already loaded, continue processing. - if FCurrentData.Done.IsSet then - begin - currData := FCurrentData.Value; - continue; - end; - end - else - begin - // All historical files processed; transition to "live data" mode. - FIsLiveData.SetValue(true); - end; - break; // Exit loop if no more data or transition occurred. - end; - - // Processes data point if its timestamp is strictly greater than the last processed one. - // This ensures chronological order and handles potential duplicate timestamps. - var item := currData[FCurrPosInFile]; - if FLastTimeStamp < item.TimeStamp then - begin - FLastTimeStamp := item.TimeStamp; - Data[Result].Create(FCurrentIdx, FLastTimeStamp, item.Data); - inc(Result); - inc(FCurrentIdx); - end; - inc(FCurrPosInFile); - end; - end; -end; - -function TAuraFileServer.GetIsLiveData: TMutable; -begin - Result := FIsLiveData; -end; - -end. diff --git a/Src/Myc.Trade.DataServer.pas b/Src/Myc.Trade.DataServer.pas deleted file mode 100644 index 6443023..0000000 --- a/Src/Myc.Trade.DataServer.pas +++ /dev/null @@ -1,52 +0,0 @@ -unit Myc.Trade.DataServer; - -interface - -uses - System.SysUtils, - Myc.Lazy, - Myc.Signals, - Myc.Futures; - -type - // Represents a generic data server capable of providing sequential data chunks. - // The type parameter T specifies the type of data points served. - IDataServer = interface - ['{A6E246A2-E84E-49AB-A63E-333E561E488C}'] - // Provides a TMutable that indicates if the server is currently - // serving live data (true) or historical data (false). - function GetIsLiveData: TMutable; - // Retrieves a chunk of data into the provided dynamic array. - // Returns the number of items successfully read into the array. - function GetChunk(var Data: array of T): Integer; - property IsLiveData: TMutable read GetIsLiveData; - end; - - // Abstract base class for IDataServer implementations. - TDataServer = class(TInterfacedObject, IDataServer) - protected - function GetIsLiveData: TMutable; virtual; abstract; - function GetChunk(var Data: array of T): Integer; virtual; abstract; - end; - - // Represents a factory for creating IDataServer instances. - // The type parameter T specifies the type of data points the created server will provide. - IDataServerNode = interface - ['{DA3531F1-158E-4A63-8E5A-34089C537B1B}'] - // Creates and returns a new instance of an IDataServer. - function CreateDataServer: IDataServer; - end; - - // Abstract base class for IDataServerNode implementations. - TDataServerNode = class(TInterfacedObject, IDataServerNode) - public - // Factory method to create the actual data server instance. - function CreateDataServer: IDataServer; virtual; abstract; - end; - -implementation - -uses - System.IOUtils; - -end. diff --git a/Src/Myc.Trade.DataStream.pas b/Src/Myc.Trade.DataStream.pas new file mode 100644 index 0000000..6bc5e16 --- /dev/null +++ b/Src/Myc.Trade.DataStream.pas @@ -0,0 +1,654 @@ +unit Myc.Trade.DataStream; + +(* + 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: + - TDataRecord: A generic record for a single time-stamped data entry. + - IDataServer/TDataServer: 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. + - IDataStream/TDataStream: An interface representing a stream of data points that + can be consumed sequentially. + - TAuraFileStream: A concrete implementation of IDataStream that is fed by an + IDataServer instance. +*) + +interface + +uses + System.SysUtils, + System.Classes, + System.Generics.Collections, + System.Generics.Defaults, + System.IOUtils, + Myc.Futures, + Myc.Signals, + Myc.Lazy, + Myc.Trade.DataPoint; + +type + // A generic record for a single time-stamped data entry. + TDataRecord = packed record + TimeStamp: TDateTime; + Data: T; + end; + + // Represents a generic data stream capable of providing sequential data chunks. + IDataStream = interface + ['{A6E246A2-E84E-49AB-A63E-333E561E488C}'] + function GetIsLiveData: TMutable; + function GetChunk(var Data: array of T): Integer; + property IsLiveData: TMutable read GetIsLiveData; + end; + + // Abstract base class for IDataStream implementations. + TDataStream = class(TInterfacedObject, IDataStream) + protected + function GetIsLiveData: TMutable; virtual; abstract; + function GetChunk(var Data: array of T): Integer; virtual; abstract; + end; + + // Represents a factory for creating IDataStream instances. + IDataStreamNode = interface + ['{DA3531F1-158E-4A63-8E5A-34089C537B1B}'] + function CreateDataStream: IDataStream; + end; + + // Abstract base class for IDataStreamNode implementations. + TDataStreamNode = class(TInterfacedObject, IDataStreamNode) + public + function CreateDataStream: IDataStream; virtual; abstract; + end; + + // Interface for an instantiable data server. + IDataServer = interface + ['{1F8E5A9D-E92A-44C1-9F3F-C4B82A6E94B3}'] + function LoadDataFile(const FileName: string): TFuture>>; + function LoadDataSeries(const FileName: string): TFuture>>; + procedure ClearCache; + end; + + // Generic server for loading and managing sequential time-series data from files. + TDataServer = class(TInterfacedObject, IDataServer) + strict private + type + TCachedFile = class + Name: String; + Age: TDateTime; + LastUsed: TDateTime; + Data: TFuture>>; + end; + private + // The cache is per-instance. + FCachedFiles: TDictionary; + strict private + // The load gate is shared across all instances of TDataServer. + class var FLoadGate: TLatch; + + protected + class function ReadCompressedData(const InputStream: TStream): TArray>; static; + class function ReadUncompressedData(const InputStream: TStream): TArray>; static; + + public + constructor Create; + destructor Destroy; override; + + // IDataServer implementation + procedure ClearCache; + function LoadDataFile(const FileName: string): TFuture>>; + function LoadDataSeries(const FileName: string): TFuture>>; + end; + + // A specialized TDataServer for handling Ask/Bid tick data. + TAskBidServer = class(TDataServer) + end; + + // Implements a data stream that reads data from Aura-specific historical data files. + TAuraFileStream = class(TDataStream>, IDataStream>) + private + FDataServer: IDataServer; + FIsLiveData: TMutable.IWriteable; + FCurrentFileName: string; + FCurrentData: TFuture>>; + FNextFileName: string; + FNextData: TFuture>>; + FCurrPosInFile: Int64; + FCurrentIdx: Int64; + FLastTimeStamp: TDateTime; + protected + function GetChunk(var Data: array of TDataPoint): Integer; override; + function GetIsLiveData: TMutable; override; + public + constructor Create(const AFilename: String; const ADataServer: IDataServer); + procedure AfterConstruction; override; + end; + + // Specialized TAuraFileStream for TAskBidItem data. + TAuraTABFileStream = TAuraFileStream; + +function TryParseFileName(const FileName: string; out PathValue, SymbolValue: string; out YearValue, MonthValue: Integer): Boolean; +function FindOldestFilesPerSymbol(const DirectoryPath: string): TArray; +function FindNextDataFile(const FileName: string): string; + +var + IsMemoryLow: TFunc; + +implementation + +uses + System.Zip, + System.Math, + Myc.TaskManager; + +constructor TDataServer.Create; +begin + inherited Create; + // Initialize the per-instance cache. + FCachedFiles := TObjectDictionary.Create([doOwnsValues]); +end; + +destructor TDataServer.Destroy; +begin + FCachedFiles.Free; + inherited Destroy; +end; + +procedure TDataServer.ClearCache; +begin + FCachedFiles.Clear; +end; + +function TDataServer.LoadDataFile(const FileName: string): TFuture>>; +var + parsedPath, parsedSymbol: string; + parsedYear, parsedMonth: Integer; +begin + Result := TFuture>>.Null; + + 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(fL, + TComparer.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; + if fL[i].Name <> FileName then + FCachedFiles.Remove(fL[i].Name); + end; + end; + + var cachedFile: TCachedFile; + if FCachedFiles.TryGetValue(Filename, cachedFile) then + begin + var age: TDateTime; + if FileAge(Filename, age, true) then + begin + if age = cachedFile.Age then + begin + cachedFile.LastUsed := Now; + exit(cachedFile.Data); + end + else + FCachedFiles.Remove(FileName); + end; + end; + end; + + var capFileName := FileName; + if FileName.EndsWith('_zip', True) then + begin + Result := TFuture.Construct( + TLatch.Enqueue(FLoadGate), // Use shared class var FLoadGate + function: TBytes + begin + Result := TFile.ReadAllBytes(capFileName); + end + ).Chain>>( + function(bytes: TBytes): TArray> + begin + var zipMemoryStream := TBytesStream.Create(bytes); + try + zipMemoryStream.Position := 0; + Result := ReadCompressedData(zipMemoryStream); + finally + zipMemoryStream.Free; + end; + end + ); + end + else if TFile.Exists(FileName) then + begin + Result := TFuture>>.Construct( + TLatch.Enqueue(FLoadGate), // Use shared class var FLoadGate + function: TArray> + 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 + raise EReadError.CreateFmt('File %s: %s', [capFileName, E.Message]); + end; + finally + stream.Free; + end; + end; + end + ); + end; + + if Assigned(IsMemoryLow) then + begin + var cachedFile := TCachedFile.Create; + cachedFile.Name := Filename; + if FileAge(Filename, cachedFile.Age, true) then + begin + cachedFile.Data := Result; + cachedFile.LastUsed := Now; + FCachedFiles.Add(Filename, cachedFile); + end; + end; +end; + +function TDataServer.LoadDataSeries(const FileName: string): TFuture>>; +var + loadedState: TState; + loadedFiles: TArray>>>; + liveData: TFuture>>; + tabFiles: TList; + currentFile: string; + liveFilePath: string; + pathName, symbolName: string; + yearValue, monthValue: Integer; +begin + tabFiles := TList.Create; + try + currentFile := FileName; + if TFile.Exists(currentFile) then + while currentFile <> '' do + begin + tabFiles.Add(currentFile); + currentFile := FindNextDataFile(currentFile); + end; + + 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 + liveFilePath := potentialLivePath; + end; + end; + + var loadedFileList := TList>>>.Create; + var loadStates := TList.Create; + try + for currentFile in tabFiles do + begin + var data := LoadDataFile(currentFile); + loadedFileList.Add(data); + loadStates.Add(data.Done); + end; + + liveData := TFuture>>.Null; + if liveFilePath <> '' then + begin + liveData := LoadDataFile(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; + + Result := TFuture>>.Construct( + loadedState, + function: TArray> + begin + var tickList := TList>.Create; + try + var overallLastTabTickTime: TDateTime := 0; + var cnt := 0; + for var P in loadedFiles do + inc(cnt, Length(P.Value)); + tickList.Capacity := cnt; + + for var P in loadedFiles do + for var iterTickData in P.Value do + if overallLastTabTickTime < iterTickData.TimeStamp then + begin + tickList.Add(iterTickData); + overallLastTabTickTime := iterTickData.TimeStamp; + end; + Result := tickList.ToArray; + finally + tickList.Free; + end; + end + ); +end; + +class function TDataServer.ReadCompressedData(const InputStream: TStream): TArray>; +var + decompressionStream: TStream; + localHeader: TZipHeader; + entryIndex, i: Integer; + zipFileInstance: TZipFile; +begin + SetLength(Result, 0); + decompressionStream := nil; + zipFileInstance := nil; + try + InputStream.Position := 0; + zipFileInstance := TZipFile.Create; + zipFileInstance.Open(InputStream, TZipMode.zmRead); + + if zipFileInstance.FileCount = 0 then + exit; + + entryIndex := -1; + for i := 0 to zipFileInstance.FileCount - 1 do + if zipFileInstance.FileNames[i].EndsWith('.tab', True) then + begin + entryIndex := i; + break; + end; + + if entryIndex = -1 then + entryIndex := 0; + + zipFileInstance.Read(entryIndex, decompressionStream, localHeader); + if not Assigned(decompressionStream) then + exit; + Result := ReadUncompressedData(decompressionStream); + finally + decompressionStream.Free; + zipFileInstance.Free; + end; +end; + +class function TDataServer.ReadUncompressedData(const InputStream: TStream): TArray>; +var + fileSize: Int64; + recordCount, bytesRead: Integer; +begin + SetLength(Result, 0); + InputStream.Position := 0; + fileSize := InputStream.Size; + if (fileSize = 0) or ((fileSize mod SizeOf(TDataRecord)) <> 0) then + exit; + + recordCount := fileSize div SizeOf(TDataRecord); + if recordCount > 0 then + begin + SetLength(Result, recordCount); + bytesRead := InputStream.Read(Result[0], fileSize); + if bytesRead <> fileSize then + raise EReadError.CreateFmt('Read error. Expected %d bytes, read %d.', [fileSize, bytesRead]); + end; +end; + +{ TAuraFileStream } + +constructor TAuraFileStream.Create(const AFilename: String; const ADataServer: IDataServer); +begin + inherited Create; + Assert(Assigned(ADataServer)); + FDataServer := ADataServer; + FCurrentFileName := AFilename; + FIsLiveData := TMutable.CreateWriteable(false); +end; + +procedure TAuraFileStream.AfterConstruction; +begin + inherited; + FCurrentData := FDataServer.LoadDataFile(FCurrentFileName); + FCurrPosInFile := 0; + FCurrentIdx := 0; + FIsLiveData.SetValue(false); + FLastTimeStamp := 0; + FNextFileName := FindNextDataFile(FCurrentFileName); + if FNextFileName <> '' then + FNextData := FDataServer.LoadDataFile(FNextFileName) + else + FNextData := nil; +end; + +function TAuraFileStream.GetChunk(var Data: array of TDataPoint): Integer; +var + item: TDataRecord; + currData: TArray>; +begin + Result := 0; + if not FCurrentData.Done.IsSet then + exit; + + var maxLen := Length(Data); + currData := FCurrentData.Value; + while Result < maxLen do + begin + if FCurrPosInFile >= Length(currData) then + begin + FCurrentData := FNextData; + FCurrentFileName := FNextFileName; + FCurrPosInFile := 0; + if FCurrentFileName <> '' then + begin + FNextFileName := FindNextDataFile(FCurrentFileName); + if FNextFileName <> '' then + FNextData := FDataServer.LoadDataFile(FNextFileName) + else + FNextData := nil; + + if FCurrentData.Done.IsSet then + begin + currData := FCurrentData.Value; + continue; + end; + end + else + begin + FIsLiveData.SetValue(true); + end; + break; + end; + + item := currData[FCurrPosInFile]; + if FLastTimeStamp < item.TimeStamp then + begin + FLastTimeStamp := item.TimeStamp; + Data[Result].Create(FCurrentIdx, FLastTimeStamp, item.Data); + Inc(Result); + Inc(FCurrentIdx); + end; + Inc(FCurrPosInFile); + end; +end; + +function TAuraFileStream.GetIsLiveData: TMutable; +begin + Result := FIsLiveData; +end; + +{ Helper Functions } + +function TryParseFileName(const FileName: string; out PathValue, SymbolValue: string; out YearValue, MonthValue: Integer): Boolean; +var + fileNameNoPath: string; + nameForParsing: string; + parts: TArray; + baseName: string; +begin + Result := False; + PathValue := ''; + SymbolValue := ''; + YearValue := 0; + MonthValue := 0; + PathValue := TPath.GetDirectoryName(FileName); + fileNameNoPath := TPath.GetFileName(FileName); + nameForParsing := fileNameNoPath; + + if nameForParsing.EndsWith('_zip', True) then + begin + baseName := nameForParsing.Substring(0, nameForParsing.Length - 4); + if TPath.GetExtension(baseName) <> '' then + begin + nameForParsing := baseName; + end; + end; + + nameForParsing := TPath.GetFileNameWithoutExtension(nameForParsing); + parts := nameForParsing.Split(['_']); + + if Length(parts) < 3 then + exit; + + if not TryStrToInt(parts[High(parts)], MonthValue) then + exit; + + if (MonthValue < 1) or (MonthValue > 12) then + begin + MonthValue := 0; + exit; + end; + + if not TryStrToInt(parts[High(parts) - 1], YearValue) then + begin + MonthValue := 0; + exit; + end; + + if YearValue <= 0 then + begin + YearValue := 0; + MonthValue := 0; + exit; + end; + + SymbolValue := string.Join('_', Copy(parts, 0, Length(parts) - 2)); + + if SymbolValue = '' then + begin + PathValue := ''; + YearValue := 0; + MonthValue := 0; + exit; + end; + + Result := True; +end; + +function FindOldestFilesPerSymbol(const DirectoryPath: string): TArray; +type + TTrackedFileInfo = record + FullFileName: string; + Year: Integer; + Month: Integer; + end; +var + oldestFilesPerSymbol: TDictionary; + fileNames: TArray; + currentFile: string; + parsedPath: string; + parsedSymbol: string; + parsedYear: Integer; + parsedMonth: Integer; + trackedInfo: TTrackedFileInfo; +begin + oldestFilesPerSymbol := TDictionary.Create; + try + if not TDirectory.Exists(DirectoryPath) then + exit(nil); + + fileNames := TDirectory.GetFiles(DirectoryPath); + for currentFile in fileNames do + begin + if TryParseFileName(currentFile, parsedPath, parsedSymbol, parsedYear, parsedMonth) then + begin + if oldestFilesPerSymbol.TryGetValue(parsedSymbol, trackedInfo) then + begin + if (parsedYear < trackedInfo.Year) or ((parsedYear = trackedInfo.Year) and (parsedMonth < trackedInfo.Month)) then + begin + trackedInfo.FullFileName := currentFile; + trackedInfo.Year := parsedYear; + trackedInfo.Month := parsedMonth; + oldestFilesPerSymbol.AddOrSetValue(parsedSymbol, trackedInfo); + end; + end + else + begin + trackedInfo.FullFileName := currentFile; + trackedInfo.Year := parsedYear; + trackedInfo.Month := parsedMonth; + oldestFilesPerSymbol.Add(parsedSymbol, trackedInfo); + end; + end; + end; + + SetLength(Result, oldestFilesPerSymbol.Count); + var n := 0; + for trackedInfo in oldestFilesPerSymbol.Values do + begin + Result[n] := trackedInfo.FullFileName; + inc(n); + end; + finally + oldestFilesPerSymbol.Free; + end; +end; + +function FindNextDataFile(const FileName: string): string; +var + pathValue, symbolValue: string; + yearValue, monthValue: Integer; + nextBaseName, compressedPath, uncompressedPath: string; +begin + if not TryParseFileName(FileName, pathValue, symbolValue, yearValue, monthValue) then + exit(''); + + Inc(monthValue); + if monthValue > 12 then + begin + monthValue := 1; + Inc(yearValue); + end; + + nextBaseName := Format('%s_%.4d_%.2d.tab', [symbolValue, yearValue, monthValue]); + + compressedPath := TPath.Combine(pathValue, nextBaseName + '_zip'); + if TFile.Exists(compressedPath) then + exit(compressedPath); + + uncompressedPath := TPath.Combine(pathValue, nextBaseName); + if TFile.Exists(uncompressedPath) then + exit(uncompressedPath); + + Result := ''; +end; + +end. diff --git a/Test/MycTests.dpr b/Test/MycTests.dpr index 394da68..48876f4 100644 --- a/Test/MycTests.dpr +++ b/Test/MycTests.dpr @@ -5,38 +5,37 @@ program MycTests; {$ENDIF} {$STRONGLINKTYPES ON} uses - FastMM5, - DUnitX.MemoryLeakMonitor.FastMM5, - System.SysUtils, -{$IFDEF TESTINSIGHT} - TestInsight.DUnitX, -{$ELSE} - DUnitX.Loggers.Console, -{$ENDIF } - DUnitX.TestFramework, - TestNotifier in 'TestNotifier.pas', - TestNotifier_Threading in 'TestNotifier_Threading.pas' {/TestNotifier_ChaosStress in 'TestNotifier_ChaosStress.pas',}, - TestNotifier_ChaosStress in 'TestNotifier_ChaosStress.pas', - Myc.Core.Tasks in '..\Src\Myc.Core.Tasks.pas', - TestTasks in 'TestTasks.pas', - Myc.Futures in '..\Src\Myc.Futures.pas', - TestCoreFutures in 'TestCoreFutures.pas', - Myc.TaskManager in '..\Src\Myc.TaskManager.pas', - Myc.Core.Futures in '..\Src\Myc.Core.Futures.pas', - Myc.Core.Signals in '..\Src\Myc.Core.Signals.pas', - TestFutures in 'TestFutures.pas', - Myc.Lazy in '..\Src\Myc.Lazy.pas', - Myc.Core.Lazy in '..\Src\Myc.Core.Lazy.pas', - Myc.Test.Core.Lazy in '..\Src\Myc.Test.Core.Lazy.pas', - Myc.Test.Lazy in '..\Src\Myc.Test.Lazy.pas', - Myc.Test.Core.Atomic in '..\Src\Myc.Test.Core.Atomic.pas', - Myc.Trade.Ticker in '..\Src\Myc.Trade.Ticker.pas', - Myc.Test.Signals.Latch in '..\Src\Myc.Test.Signals.Latch.pas', - Myc.Test.Signals.Dirty in '..\Src\Myc.Test.Signals.Dirty.pas', - Myc.Trade.DataPoint in '..\Src\Myc.Trade.DataPoint.pas', - Myc.Trade.DataServer in '..\Src\Myc.Trade.DataServer.pas', - Myc.Trade.Node in '..\Src\Myc.Trade.Node.pas', - Myc.Test.Trade.DataServer in '..\Src\Myc.Test.Trade.DataServer.pas'; + FastMM5, + DUnitX.MemoryLeakMonitor.FastMM5, + System.SysUtils, + {$IFDEF TESTINSIGHT} + TestInsight.DUnitX, + {$ELSE} + DUnitX.Loggers.Console, + {$ENDIF } + DUnitX.TestFramework, + TestNotifier in 'TestNotifier.pas', + TestNotifier_Threading in 'TestNotifier_Threading.pas' {/TestNotifier_ChaosStress in 'TestNotifier_ChaosStress.pas',}, + TestNotifier_ChaosStress in 'TestNotifier_ChaosStress.pas', + Myc.Core.Tasks in '..\Src\Myc.Core.Tasks.pas', + TestTasks in 'TestTasks.pas', + Myc.Futures in '..\Src\Myc.Futures.pas', + TestCoreFutures in 'TestCoreFutures.pas', + Myc.TaskManager in '..\Src\Myc.TaskManager.pas', + Myc.Core.Futures in '..\Src\Myc.Core.Futures.pas', + Myc.Core.Signals in '..\Src\Myc.Core.Signals.pas', + TestFutures in 'TestFutures.pas', + Myc.Lazy in '..\Src\Myc.Lazy.pas', + Myc.Core.Lazy in '..\Src\Myc.Core.Lazy.pas', + Myc.Test.Core.Lazy in '..\Src\Myc.Test.Core.Lazy.pas', + Myc.Test.Lazy in '..\Src\Myc.Test.Lazy.pas', + Myc.Test.Core.Atomic in '..\Src\Myc.Test.Core.Atomic.pas', + Myc.Trade.Ticker in '..\Src\Myc.Trade.Ticker.pas', + Myc.Test.Signals.Latch in '..\Src\Myc.Test.Signals.Latch.pas', + Myc.Test.Signals.Dirty in '..\Src\Myc.Test.Signals.Dirty.pas', + Myc.Trade.DataPoint in '..\Src\Myc.Trade.DataPoint.pas', + Myc.Trade.Node in '..\Src\Myc.Trade.Node.pas', + Myc.Trade.DataStream in '..\Src\Myc.Trade.DataStream.pas'; { keep comment here to protect the following conditional from being removed by the IDE when adding a unit } {$IFNDEF TESTINSIGHT} diff --git a/Test/MycTests.dproj b/Test/MycTests.dproj index e5462be..8af0f76 100644 --- a/Test/MycTests.dproj +++ b/Test/MycTests.dproj @@ -4,7 +4,7 @@ 20.3 None True - Release + Debug Win64 MycTests 3 @@ -134,9 +134,8 @@ $(PreBuildEvent)]]> - - + Base @@ -192,6 +191,12 @@ $(PreBuildEvent)]]> true + + + MycTests.exe + true + + 1