Michael Schimmel
781 lines
26 KiB
ObjectPascal
781 lines
26 KiB
ObjectPascal
unit Myc.Trade.DataStream;
|
|
|
|
interface
|
|
|
|
uses
|
|
System.SysUtils,
|
|
System.Classes,
|
|
System.Generics.Collections,
|
|
System.Generics.Defaults,
|
|
System.IOUtils,
|
|
Myc.Futures,
|
|
Myc.Signals,
|
|
Myc.Mutable,
|
|
Myc.Trade.Types,
|
|
Myc.Data.Pipeline,
|
|
Myc.Core.FileCache;
|
|
|
|
type
|
|
// Interface for an instantiable data server.
|
|
IDataServer<T: record> = interface
|
|
procedure ClearCache;
|
|
function ProcessData(const Symbol: String; const Terminated: TState; const Processor: IConsumer<TArray<TDataPoint<T>>>): TState;
|
|
function EnumerateSymbols: TFuture<TArray<String>>;
|
|
end;
|
|
|
|
// Represents metadata for a single data file.
|
|
TDataFile = record
|
|
private
|
|
FExtension: String;
|
|
FPath: String;
|
|
FSymbol: String;
|
|
FYear: Integer;
|
|
FMonth: Integer;
|
|
FBasename: String;
|
|
function GetIsValid: Boolean;
|
|
public
|
|
constructor Create(const APath, ASymbol, AExtension: String; AYear, AMonth: Integer; const ABasename: String);
|
|
function GetBaseFileName: string;
|
|
function GetFullFileName: string;
|
|
property IsValid: Boolean read GetIsValid;
|
|
property Extension: String read FExtension;
|
|
property Path: String read FPath;
|
|
property Symbol: String read FSymbol;
|
|
property Year: Integer read FYear;
|
|
property Month: Integer read FMonth;
|
|
end;
|
|
|
|
// Generic server for loading and managing sequential time-series data from files.
|
|
TDataServer<T: record> = class(TInterfacedObject, IDataServer<T>)
|
|
private
|
|
FCachedFiles: TDataFileCache<TArray<TArray<TDataPoint<T>>>>;
|
|
FPath: String;
|
|
// The cache now holds a sorted array of all files for each symbol.
|
|
FSymbols: TFuture<TDictionary<String, TArray<TDataFile>>>;
|
|
function GetPath: String;
|
|
|
|
function ProcessChunks(
|
|
const DataChunks: TArray<TArray<TDataPoint<T>>>;
|
|
Terminated: TState;
|
|
Processor: IConsumer<TArray<TDataPoint<T>>>
|
|
): TState;
|
|
|
|
function ProcessFile(
|
|
FileInfo: TDataFile;
|
|
const DataFile: TFuture<TArray<TArray<TDataPoint<T>>>>;
|
|
const Terminated: TState;
|
|
Processor: IConsumer<TArray<TDataPoint<T>>>
|
|
): TState;
|
|
|
|
strict private
|
|
class var
|
|
FLoadGate: TLatch;
|
|
|
|
protected
|
|
// Used by cache to actually load an uncached file. Now virtual and abstract.
|
|
function DoLoad(const FileName: string; const Gate: TLatch): TFuture<TArray<TArray<TDataPoint<T>>>>; virtual; abstract;
|
|
// Gets the next consecutive data file from the cached, sorted list.
|
|
function GetNextFile(const Curr: TDataFile): TDataFile; virtual;
|
|
|
|
public
|
|
constructor Create(const APath: String);
|
|
destructor Destroy; override;
|
|
procedure AfterConstruction; override;
|
|
procedure ClearCache;
|
|
procedure UpdateSymbols;
|
|
// Scans a directory to find the oldest file for a specific symbol.
|
|
function FindFirstFile(const Symbol: string): TFuture<TDataFile>;
|
|
function ParseFileName(const FileName: string): TDataFile; virtual; abstract;
|
|
function EnumerateSymbols: TFuture<TArray<String>>;
|
|
|
|
// Load data file and split content into chunks.
|
|
function LoadDataFile(const DataFile: TDataFile): TFuture<TArray<TArray<TDataPoint<T>>>>;
|
|
|
|
function ProcessData(const Symbol: String; const Terminated: TState; const Processor: IConsumer<TArray<TDataPoint<T>>>): TState;
|
|
|
|
property Path: String read GetPath;
|
|
end;
|
|
|
|
TAskBidFileItem = packed record
|
|
Ask: Single;
|
|
Bid: Single;
|
|
end;
|
|
|
|
TAskBidFileServer = class(TDataServer<TAskBidFileItem>)
|
|
private
|
|
class function ReadCompressedData(const InputStream: TStream): TArray<TArray<TDataPoint<TAskBidFileItem>>>;
|
|
class function ReadUncompressedData(const InputStream: TStream): TArray<TArray<TDataPoint<TAskBidFileItem>>>;
|
|
protected
|
|
function DoLoad(const FileName: string; const Gate: TLatch): TFuture<TArray<TArray<TDataPoint<TAskBidFileItem>>>>; override;
|
|
public
|
|
function ParseFileName(const FileName: string): TDataFile; override;
|
|
end;
|
|
|
|
TM1FileItem = packed record
|
|
OADateTime: Double;
|
|
Open: Int64;
|
|
High: Int64;
|
|
Low: Int64;
|
|
Close: Int64;
|
|
Spread: Int64;
|
|
TickVolume: Integer;
|
|
Digits: Integer;
|
|
end;
|
|
|
|
// File server for cTrader M1 data. Note that the generic type is TOhlcItem,
|
|
// as the server converts the file data into standard OHLC items.
|
|
TM1FileServer = class(TDataServer<TOhlcItem>)
|
|
private
|
|
class function ReadCompressedData(const InputStream: TStream): TArray<TArray<TDataPoint<TOhlcItem>>>;
|
|
class function ReadUncompressedData(const InputStream: TStream): TArray<TArray<TDataPoint<TOhlcItem>>>;
|
|
protected
|
|
function DoLoad(const FileName: string; const Gate: TLatch): TFuture<TArray<TArray<TDataPoint<TOhlcItem>>>>; override;
|
|
public
|
|
function ParseFileName(const FileName: string): TDataFile; override;
|
|
end;
|
|
|
|
implementation
|
|
uses
|
|
System.Zip,
|
|
System.ZLib,
|
|
System.Math,
|
|
System.StrUtils,
|
|
Myc.TaskManager;
|
|
|
|
const
|
|
// The number of records per data chunk when loading files.
|
|
ChunkSize = 512;
|
|
|
|
{ TDataFile }
|
|
|
|
constructor TDataFile.Create(const APath, ASymbol, AExtension: String; AYear, AMonth: Integer; const ABasename: String);
|
|
begin
|
|
FPath := APath;
|
|
FSymbol := ASymbol;
|
|
FExtension := AExtension;
|
|
FYear := AYear;
|
|
FMonth := AMonth;
|
|
FBasename := ABasename;
|
|
end;
|
|
|
|
function TDataFile.GetBaseFileName: string;
|
|
begin
|
|
Result := FBasename;
|
|
end;
|
|
|
|
function TDataFile.GetFullFileName: string;
|
|
begin
|
|
Result := TPath.Combine(FPath, GetBaseFileName + FExtension);
|
|
end;
|
|
|
|
function TDataFile.GetIsValid: Boolean;
|
|
begin
|
|
Result := (FSymbol <> '') and (FYear > 0);
|
|
end;
|
|
|
|
{ TDataServer<T> }
|
|
|
|
constructor TDataServer<T>.Create(const APath: String);
|
|
begin
|
|
inherited Create;
|
|
FPath := APath;
|
|
|
|
FCachedFiles := TDataFileCache<TArray<TArray<TDataPoint<T>>>>.Create;
|
|
end;
|
|
|
|
destructor TDataServer<T>.Destroy;
|
|
begin
|
|
ClearCache;
|
|
FCachedFiles.Free;
|
|
inherited Destroy;
|
|
end;
|
|
|
|
procedure TDataServer<T>.AfterConstruction;
|
|
begin
|
|
inherited;
|
|
UpdateSymbols;
|
|
end;
|
|
|
|
procedure TDataServer<T>.ClearCache;
|
|
begin
|
|
FCachedFiles.Clear;
|
|
end;
|
|
|
|
procedure TDataServer<T>.UpdateSymbols;
|
|
begin
|
|
FSymbols :=
|
|
TFuture<TDictionary<String, TArray<TDataFile>>>.Construct(
|
|
FSymbols.Done,
|
|
function: TDictionary<String, TArray<TDataFile>>
|
|
var
|
|
fileNames: TArray<string>;
|
|
tempSymbolMap: TDictionary<String, TList<TDataFile>>;
|
|
dataFile: TDataFile;
|
|
symbolList: TList<TDataFile>;
|
|
begin
|
|
Result := TDictionary<String, TArray<TDataFile>>.Create;
|
|
tempSymbolMap := TDictionary<String, TList<TDataFile>>.Create;
|
|
try
|
|
if not TDirectory.Exists(FPath) then
|
|
exit;
|
|
|
|
fileNames := TDirectory.GetFiles(FPath);
|
|
for var currentFile in fileNames do
|
|
begin
|
|
dataFile := ParseFileName(currentFile);
|
|
if dataFile.IsValid then
|
|
begin
|
|
if not tempSymbolMap.TryGetValue(dataFile.Symbol, symbolList) then
|
|
begin
|
|
symbolList := TList<TDataFile>.Create;
|
|
tempSymbolMap.Add(dataFile.Symbol, symbolList);
|
|
end;
|
|
symbolList.Add(dataFile);
|
|
end;
|
|
end;
|
|
|
|
for var kvp in tempSymbolMap do
|
|
begin
|
|
symbolList := kvp.Value;
|
|
symbolList.Sort(
|
|
TComparer<TDataFile>.Construct(
|
|
function(const Left, Right: TDataFile): Integer
|
|
begin
|
|
if Left.Year < Right.Year then
|
|
Result := -1
|
|
else if Left.Year > Right.Year then
|
|
Result := 1
|
|
else
|
|
Result := Left.Month - Right.Month;
|
|
end
|
|
)
|
|
);
|
|
Result.Add(kvp.Key, symbolList.ToArray);
|
|
end;
|
|
finally
|
|
for var kvp in tempSymbolMap do
|
|
kvp.Value.Free;
|
|
tempSymbolMap.Free;
|
|
end;
|
|
end
|
|
);
|
|
|
|
FSymbols.Manage;
|
|
end;
|
|
|
|
function TDataServer<T>.EnumerateSymbols: TFuture<TArray<String>>;
|
|
begin
|
|
Result :=
|
|
FSymbols.Chain<TArray<String>>(
|
|
function(const Symbols: TDictionary<String, TArray<TDataFile>>): TArray<String> begin Result := Symbols.Keys.ToArray; end
|
|
);
|
|
end;
|
|
|
|
function TDataServer<T>.FindFirstFile(const Symbol: string): TFuture<TDataFile>;
|
|
begin
|
|
var symName := Symbol;
|
|
Result :=
|
|
FSymbols.Chain<TDataFile>(
|
|
function(const Symbols: TDictionary<String, TArray<TDataFile>>): TDataFile
|
|
var
|
|
symbolFiles: TArray<TDataFile>;
|
|
begin
|
|
if Symbols.TryGetValue(symName, symbolFiles) and (Length(symbolFiles) > 0) then
|
|
Result := symbolFiles[0]
|
|
else
|
|
Result := Default(TDataFile);
|
|
end
|
|
);
|
|
end;
|
|
|
|
function TDataServer<T>.GetNextFile(const Curr: TDataFile): TDataFile;
|
|
var
|
|
allSymbolFiles: TArray<TDataFile>;
|
|
foundIndex: Integer;
|
|
begin
|
|
Result := Default(TDataFile);
|
|
if not Curr.IsValid or not FSymbols.Done.IsSet then
|
|
exit;
|
|
|
|
if not FSymbols.Value.TryGetValue(Curr.Symbol, allSymbolFiles) then
|
|
exit;
|
|
|
|
foundIndex := -1;
|
|
for var i := 0 to High(allSymbolFiles) do
|
|
begin
|
|
if allSymbolFiles[i].GetBaseFileName() = Curr.GetBaseFileName() then
|
|
begin
|
|
foundIndex := i;
|
|
break;
|
|
end;
|
|
end;
|
|
|
|
if (foundIndex <> -1) and (foundIndex < High(allSymbolFiles)) then
|
|
begin
|
|
Result := allSymbolFiles[foundIndex + 1];
|
|
end;
|
|
end;
|
|
|
|
function TDataServer<T>.GetPath: String;
|
|
begin
|
|
Result := FPath;
|
|
end;
|
|
|
|
function TDataServer<T>.ProcessData(
|
|
const Symbol: String;
|
|
const Terminated: TState;
|
|
const Processor: IConsumer<TArray<TDataPoint<T>>>
|
|
): TState;
|
|
begin
|
|
var cProc := Processor;
|
|
var cTerminated := Terminated;
|
|
|
|
Result :=
|
|
FindFirstFile(Symbol)
|
|
.Chain(
|
|
function(const FirstFileInfo: TDataFile): TState
|
|
begin
|
|
Result := ProcessFile(FirstFileInfo, LoadDataFile(FirstFileInfo), cTerminated, cProc);
|
|
end);
|
|
end;
|
|
|
|
function TDataServer<T>.LoadDataFile(const DataFile: TDataFile): TFuture<TArray<TArray<TDataPoint<T>>>>;
|
|
begin
|
|
if DataFile.IsValid then
|
|
Result :=
|
|
FCachedFiles.GetOrAdd(
|
|
DataFile.GetFullFileName,
|
|
function(const Filename: String): TFuture<TArray<TArray<TDataPoint<T>>>>
|
|
begin
|
|
// Pass the private load gate to the virtual DoLoad method.
|
|
Result := DoLoad(Filename, FLoadGate);
|
|
end
|
|
)
|
|
else
|
|
Result := TFuture<TArray<TArray<TDataPoint<T>>>>.Construct(TArray<TArray<TDataPoint<T>>>(nil));
|
|
end;
|
|
|
|
function TDataServer<T>.ProcessChunks(
|
|
const DataChunks: TArray<TArray<TDataPoint<T>>>;
|
|
Terminated: TState;
|
|
Processor: IConsumer<TArray<TDataPoint<T>>>
|
|
): TState;
|
|
begin
|
|
var callProcess :=
|
|
function(Idx: Integer): TFunc<TState>
|
|
begin
|
|
var cChunk := DataChunks[Idx];
|
|
Result :=
|
|
function: TState
|
|
begin
|
|
if not Terminated.IsSet then
|
|
Result := Processor.Consume(cChunk);
|
|
end;
|
|
end;
|
|
|
|
for var i := 0 to High(DataChunks) do
|
|
if not Terminated.IsSet then
|
|
begin
|
|
var q := TaskManager.RunTask(Result, callProcess(i));
|
|
Result := q;
|
|
end
|
|
end;
|
|
|
|
function TDataServer<T>.ProcessFile(
|
|
FileInfo: TDataFile;
|
|
const DataFile: TFuture<TArray<TArray<TDataPoint<T>>>>;
|
|
const Terminated: TState;
|
|
Processor: IConsumer<TArray<TDataPoint<T>>>
|
|
): TState;
|
|
begin
|
|
if not FileInfo.IsValid or Terminated.IsSet then
|
|
exit(DataFile.Done);
|
|
|
|
// Read ahead the next file, while processing the current one
|
|
var nextFileInfo := GetNextFile(FileInfo);
|
|
var nextFile := LoadDataFile(nextFileInfo);
|
|
|
|
var cTerminated := Terminated;
|
|
Result :=
|
|
DataFile.Chain(
|
|
function(const DataChunks: TArray<TArray<TDataPoint<T>>>): TState
|
|
begin
|
|
// Process each chunk, checking for termination between chunks.
|
|
var done := ProcessChunks(DataChunks, Terminated, Processor);
|
|
|
|
// Move to next file (which is currently preloading) once all Processors are done
|
|
Result :=
|
|
TaskManager
|
|
.RunTask(done, function: TState begin Result := ProcessFile(nextFileInfo, nextFile, cTerminated, Processor); end);
|
|
end
|
|
);
|
|
end;
|
|
|
|
{ TAskBidFileServer }
|
|
|
|
function TAskBidFileServer.DoLoad(const FileName: string; const Gate: TLatch): TFuture<TArray<TArray<TDataPoint<TAskBidFileItem>>>>;
|
|
begin
|
|
Result := TFuture<TArray<TArray<TDataPoint<TAskBidFileItem>>>>.Null;
|
|
|
|
if not TFile.Exists(FileName) then
|
|
exit;
|
|
|
|
var capFileName := FileName;
|
|
if FileName.EndsWith('_zip', True) then
|
|
begin
|
|
Result :=
|
|
TFuture<TBytes>
|
|
.Construct(Gate.Enqueue, function: TBytes begin Result := TFile.ReadAllBytes(capFileName); end)
|
|
.Chain<TArray<TArray<TDataPoint<TAskBidFileItem>>>>(
|
|
function(bytes: TBytes): TArray<TArray<TDataPoint<TAskBidFileItem>>>
|
|
begin
|
|
var compMemoryStream := TBytesStream.Create(bytes);
|
|
try
|
|
compMemoryStream.Position := 0;
|
|
Result := ReadCompressedData(compMemoryStream);
|
|
finally
|
|
compMemoryStream.Free;
|
|
end;
|
|
end);
|
|
end
|
|
else
|
|
begin
|
|
Result :=
|
|
TFuture<TArray<TArray<TDataPoint<TAskBidFileItem>>>>.Construct(
|
|
Gate.Enqueue,
|
|
function: TArray<TArray<TDataPoint<TAskBidFileItem>>>
|
|
begin
|
|
if TFile.Exists(capFileName) then
|
|
begin
|
|
var stream := TFileStream.Create(capFileName, fmOpenRead or fmShareDenyWrite);
|
|
try
|
|
Result := ReadUncompressedData(stream);
|
|
finally
|
|
stream.Free;
|
|
end;
|
|
end;
|
|
end
|
|
);
|
|
end;
|
|
end;
|
|
|
|
function TAskBidFileServer.ParseFileName(const FileName: string): TDataFile;
|
|
var
|
|
fileNameNoPath, nameForParsing, baseName, ext, path, symbol: string;
|
|
year, month: Integer;
|
|
parts: TArray<string>;
|
|
begin
|
|
Result := Default(TDataFile);
|
|
|
|
path := TPath.GetDirectoryName(FileName);
|
|
fileNameNoPath := TPath.GetFileName(FileName);
|
|
nameForParsing := fileNameNoPath;
|
|
|
|
ext := '';
|
|
if nameForParsing.EndsWith('_zip', True) then
|
|
begin
|
|
ext := '_zip';
|
|
nameForParsing := nameForParsing.Substring(0, nameForParsing.Length - 4);
|
|
end;
|
|
|
|
var fileExt := TPath.GetExtension(nameForParsing) + ext;
|
|
baseName := TPath.GetFileNameWithoutExtension(nameForParsing);
|
|
|
|
parts := baseName.Split(['_']);
|
|
|
|
if Length(parts) < 3 then
|
|
exit;
|
|
|
|
if not TryStrToInt(parts[High(parts)], month) then
|
|
exit;
|
|
|
|
if (month < 1) or (month > 12) then
|
|
exit;
|
|
|
|
if not TryStrToInt(parts[High(parts) - 1], year) then
|
|
exit;
|
|
|
|
if year <= 0 then
|
|
exit;
|
|
|
|
symbol := string.Join('_', Copy(parts, 0, Length(parts) - 2));
|
|
|
|
if symbol = '' then
|
|
exit;
|
|
|
|
Result := TDataFile.Create(path, symbol, fileExt, year, month, baseName);
|
|
end;
|
|
|
|
class function TAskBidFileServer.ReadCompressedData(const InputStream: TStream): TArray<TArray<TDataPoint<TAskBidFileItem>>>;
|
|
var
|
|
decompressionStream: TStream;
|
|
localHeader: TZipHeader;
|
|
entryIndex, i: Integer;
|
|
zipFileInstance: TZipFile;
|
|
begin
|
|
Result := nil;
|
|
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 TAskBidFileServer.ReadUncompressedData(const InputStream: TStream): TArray<TArray<TDataPoint<TAskBidFileItem>>>;
|
|
type
|
|
TFileRecord = packed record
|
|
TimeStamp: TDateTime;
|
|
Data: TAskBidFileItem;
|
|
end;
|
|
var
|
|
fileSize: Int64;
|
|
totalRecordCount, bytesRead, chunkIndex, indexInChunk, recordIndex: Integer;
|
|
rec: TFileRecord;
|
|
begin
|
|
Result := nil;
|
|
InputStream.Position := 0;
|
|
fileSize := InputStream.Size;
|
|
if (fileSize = 0) or ((fileSize mod SizeOf(TFileRecord)) <> 0) then
|
|
exit;
|
|
|
|
totalRecordCount := fileSize div SizeOf(TFileRecord);
|
|
if totalRecordCount > 0 then
|
|
begin
|
|
var numChunks := (totalRecordCount + ChunkSize - 1) div ChunkSize;
|
|
SetLength(Result, numChunks);
|
|
chunkIndex := 0;
|
|
indexInChunk := 0;
|
|
SetLength(Result[chunkIndex], Min(ChunkSize, totalRecordCount));
|
|
|
|
for recordIndex := 0 to totalRecordCount - 1 do
|
|
begin
|
|
if indexInChunk >= ChunkSize then
|
|
begin
|
|
Inc(chunkIndex);
|
|
indexInChunk := 0;
|
|
var remainingRecords := totalRecordCount - (chunkIndex * ChunkSize);
|
|
SetLength(Result[chunkIndex], Min(ChunkSize, remainingRecords));
|
|
end;
|
|
|
|
bytesRead := InputStream.Read(rec, SizeOf(TFileRecord));
|
|
if bytesRead <> SizeOf(TFileRecord) then
|
|
raise EReadError.CreateFmt('Read error. Expected %d bytes, read %d.', [SizeOf(TFileRecord), bytesRead]);
|
|
|
|
// Corrected syntax: Assign the newly created record instance.
|
|
Result[chunkIndex][indexInChunk] := TDataPoint<TAskBidFileItem>.Create(rec.TimeStamp, rec.Data);
|
|
Inc(indexInChunk);
|
|
end;
|
|
end;
|
|
end;
|
|
|
|
{ TM1FileServer }
|
|
|
|
function TM1FileServer.DoLoad(const FileName: string; const Gate: TLatch): TFuture<TArray<TArray<TDataPoint<TOhlcItem>>>>;
|
|
begin
|
|
Result := TFuture<TArray<TArray<TDataPoint<TOhlcItem>>>>.Null;
|
|
|
|
if not TFile.Exists(FileName) then
|
|
exit;
|
|
|
|
var capFileName := FileName;
|
|
// M1 files are always compressed with the .m1 extension
|
|
if FileName.EndsWith('.m1', True) then
|
|
begin
|
|
Result :=
|
|
TFuture<TBytes>
|
|
.Construct(Gate.Enqueue, function: TBytes begin Result := TFile.ReadAllBytes(capFileName); end)
|
|
.Chain<TArray<TArray<TDataPoint<TOhlcItem>>>>(
|
|
function(bytes: TBytes): TArray<TArray<TDataPoint<TOhlcItem>>>
|
|
begin
|
|
var compMemoryStream := TBytesStream.Create(bytes);
|
|
try
|
|
compMemoryStream.Position := 0;
|
|
Result := ReadCompressedData(compMemoryStream);
|
|
finally
|
|
compMemoryStream.Free;
|
|
end;
|
|
end);
|
|
end;
|
|
end;
|
|
|
|
function TM1FileServer.ParseFileName(const FileName: string): TDataFile;
|
|
var
|
|
path, fileNameNoPath, baseName, ext, symbol: string;
|
|
year, month: Integer;
|
|
parts, dateParts: TArray<string>;
|
|
begin
|
|
Result := Default(TDataFile);
|
|
path := TPath.GetDirectoryName(FileName);
|
|
fileNameNoPath := TPath.GetFileName(FileName);
|
|
|
|
// M1 files are compressed and have a .m1 extension
|
|
if not fileNameNoPath.EndsWith('.m1', True) then
|
|
exit;
|
|
|
|
ext := TPath.GetExtension(fileNameNoPath);
|
|
baseName := TPath.GetFileNameWithoutExtension(fileNameNoPath);
|
|
|
|
// Expected format: SYMBOL_YYYY-MM_YYYY-MM
|
|
parts := baseName.Split(['_']);
|
|
if Length(parts) < 3 then
|
|
exit; // Not enough parts for SYMBOL_START_END
|
|
|
|
// The symbol can contain underscores, so we take everything before the last two parts.
|
|
var startDatePart := parts[High(parts) - 1];
|
|
|
|
dateParts := startDatePart.Split(['-']);
|
|
if Length(dateParts) <> 2 then
|
|
exit; // Start date is not YYYY-MM
|
|
|
|
if not TryStrToInt(dateParts[0], year) then
|
|
exit;
|
|
|
|
if not TryStrToInt(dateParts[1], month) then
|
|
exit;
|
|
|
|
if (month < 1) or (month > 12) or (year <= 0) then
|
|
exit;
|
|
|
|
symbol := string.Join('_', Copy(parts, 0, Length(parts) - 2));
|
|
if symbol = '' then
|
|
exit;
|
|
|
|
// The TDataFile only represents the start date of the file batch.
|
|
Result := TDataFile.Create(path, symbol, ext, year, month, baseName);
|
|
end;
|
|
|
|
class function TM1FileServer.ReadCompressedData(const InputStream: TStream): TArray<TArray<TDataPoint<TOhlcItem>>>;
|
|
var
|
|
decompressionStream: TStream;
|
|
localHeader: TZipHeader;
|
|
entryIndex, i: Integer;
|
|
zipFileInstance: TZipFile;
|
|
begin
|
|
Result := nil;
|
|
decompressionStream := nil;
|
|
zipFileInstance := nil;
|
|
|
|
// M1 files are now zip archives containing the actual data file.
|
|
if not Assigned(InputStream) or (InputStream.Size = 0) then
|
|
exit;
|
|
|
|
try
|
|
InputStream.Position := 0;
|
|
zipFileInstance := TZipFile.Create;
|
|
zipFileInstance.Open(InputStream, TZipMode.zmRead);
|
|
|
|
if zipFileInstance.FileCount = 0 then
|
|
exit;
|
|
|
|
// Find the data file entry within the archive.
|
|
// The C# code creates an entry with the same name as the base filename, ending in .m1.
|
|
entryIndex := -1;
|
|
for i := 0 to zipFileInstance.FileCount - 1 do
|
|
begin
|
|
if zipFileInstance.FileNames[i].EndsWith('.m1', True) then
|
|
begin
|
|
entryIndex := i;
|
|
break;
|
|
end;
|
|
end;
|
|
|
|
// Fallback to the first entry if no specific .m1 file is found inside.
|
|
if entryIndex = -1 then
|
|
entryIndex := 0;
|
|
|
|
zipFileInstance.Read(entryIndex, decompressionStream, localHeader);
|
|
if not Assigned(decompressionStream) then
|
|
exit;
|
|
|
|
// The decompressed stream now contains the raw binary data.
|
|
Result := ReadUncompressedData(decompressionStream);
|
|
finally
|
|
decompressionStream.Free;
|
|
zipFileInstance.Free;
|
|
end;
|
|
end;
|
|
|
|
class function TM1FileServer.ReadUncompressedData(const InputStream: TStream): TArray<TArray<TDataPoint<TOhlcItem>>>;
|
|
type
|
|
TFileRecord = TM1FileItem;
|
|
var
|
|
fileSize: Int64;
|
|
totalRecordCount, bytesRead, chunkIndex, indexInChunk, recordIndex: Integer;
|
|
rec: TFileRecord;
|
|
ohlc: TOhlcItem;
|
|
factor: Double;
|
|
begin
|
|
Result := nil;
|
|
InputStream.Position := 0;
|
|
fileSize := InputStream.Size;
|
|
if (fileSize = 0) or ((fileSize mod SizeOf(TFileRecord)) <> 0) then
|
|
exit;
|
|
|
|
totalRecordCount := fileSize div SizeOf(TFileRecord);
|
|
if totalRecordCount > 0 then
|
|
begin
|
|
var numChunks := (totalRecordCount + ChunkSize - 1) div ChunkSize;
|
|
SetLength(Result, numChunks);
|
|
chunkIndex := 0;
|
|
indexInChunk := 0;
|
|
SetLength(Result[chunkIndex], Min(ChunkSize, totalRecordCount));
|
|
|
|
for recordIndex := 0 to totalRecordCount - 1 do
|
|
begin
|
|
if indexInChunk >= ChunkSize then
|
|
begin
|
|
Inc(chunkIndex);
|
|
indexInChunk := 0;
|
|
var remainingRecords := totalRecordCount - (chunkIndex * ChunkSize);
|
|
SetLength(Result[chunkIndex], Min(ChunkSize, remainingRecords));
|
|
end;
|
|
|
|
bytesRead := InputStream.Read(rec, SizeOf(TFileRecord));
|
|
if bytesRead <> SizeOf(TFileRecord) then
|
|
raise EReadError.CreateFmt('Read error. Expected %d bytes, read %d.', [SizeOf(TFileRecord), bytesRead]);
|
|
|
|
// Convert from M1 file record to a standard OHLC item
|
|
if rec.Digits > 0 then
|
|
factor := Power(10, rec.Digits)
|
|
else
|
|
factor := 1.0;
|
|
|
|
ohlc.Open := rec.Open / factor;
|
|
ohlc.High := rec.High / factor;
|
|
ohlc.Low := rec.Low / factor;
|
|
ohlc.Close := rec.Close / factor;
|
|
ohlc.Volume := rec.TickVolume;
|
|
|
|
// Corrected syntax: Assign the newly created record instance.
|
|
Result[chunkIndex][indexInChunk] := TDataPoint<TOhlcItem>.Create(rec.OADateTime, ohlc);
|
|
Inc(indexInChunk);
|
|
end;
|
|
end;
|
|
end;
|
|
|
|
end.
|