Brummel/MycLib
1
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects 1 Releases Wiki Activity

Writeables revised

Browse Source 
Read-ahead for DataStream Processing
This commit is contained in:
Michael Schimmel
2025-06-25 10:07:00 +02:00
parent f791667264
commit e1159e883b
9 changed files with 272 additions and 227 deletions
Download Patch File Download Diff File Expand all files Collapse all files
AuraTrader/AuraTrader.dproj
+1 -1
View File
@@ -4,7 +4,7 @@
<ProjectVersion>20.3</ProjectVersion>
<FrameworkType>FMX</FrameworkType>
<Base>True</Base>
<Config Condition="'$(Config)'==''">Debug</Config>
<Config Condition="'$(Config)'==''">Release</Config>
<Platform Condition="'$(Platform)'==''">Win64</Platform>
<ProjectName Condition="'$(ProjectName)'==''">AuraTrader</ProjectName>
<TargetedPlatforms>3</TargetedPlatforms>
AuraTrader/MainForm.fmx
+6 -6
View File
@@ -64,21 +64,21 @@ object Form1: TForm1
JustifyLastLine = Left
FlowDirection = LeftToRight
end
object Button1: TButton
object ChartButton: TButton
Position.X = 712.00000000000000000
Position.Y = 72.00000000000000000
TabOrder = 6
Text = 'Button1'
Text = 'ChartButton'
TextSettings.Trimming = None
OnClick = Button1Click
OnClick = ChartButtonClick
end
object Button2: TButton
object StopButton: TButton
Position.X = 720.00000000000000000
Position.Y = 96.00000000000000000
TabOrder = 8
Text = 'Button2'
Text = 'StopButton'
TextSettings.Trimming = None
OnClick = Button2Click
OnClick = StopButtonClick
end
end
object LogMemo: TMemo
AuraTrader/MainForm.pas
+89 -82
View File
@@ -45,14 +45,14 @@ type
SymbolsComboBox: TComboBox;
LoadButton: TButton;
FlowLayout: TFlowLayout;
Button1: TButton;
Button2: TButton;
ChartButton: TButton;
StopButton: TButton;
procedure RandomButtonClick(Sender: TObject);
procedure FormCreate(Sender: TObject);
procedure FormDestroy(Sender: TObject);
procedure LoadButtonClick(Sender: TObject);
procedure Button1Click(Sender: TObject);
procedure Button2Click(Sender: TObject);
procedure ChartButtonClick(Sender: TObject);
procedure StopButtonClick(Sender: TObject);
private
const
cnt = 20;
@@ -83,84 +83,7 @@ implementation
{$R *.fmx}
procedure TForm1.Button1Click(Sender: TObject);
begin
if SymbolsComboBox.ItemIndex < 0 then
exit;
var curr := TWriteable<TDataSeries<TAskBidItem>>.CreateProtected;
var Symbol := FSymbols.WaitFor[SymbolsComboBox.ItemIndex];
var currPathData := TWriteable<IObjectRef<TPathData>>.CreateProtected;
LogMemo.AddIdleHandler(curr.Changed, procedure begin LogMemo.Lines.Add(Symbol + ': ' + curr.Value.TotalCount.ToString); end);
var arr := TDataSeries<TAskBidItem>.CreateArray(3000);
var done :=
FServer.ProcessData(
Symbol,
FTerminate.Signal,
procedure(const Values: TArray<TDataPoint<TAskBidItem>>; const Terminated: TState)
begin
arr.Add(Values);
curr.Value := arr.Copy;
var PathData := TPathData.Create;
var Prices := arr;
if Prices.Count > 0 then
begin
PathData.MoveTo(PointF(Prices.Count - 1, Prices[0].Data.Ask));
for var i := 1 to Prices.Count - 1 do
PathData.LineTo(PointF(Prices.Count - i - 1, Prices[i].Data.Ask));
end;
currPathData.Value := TObjectRef<TPathData>.Create(PathData);
end
);
FLoadDone := TState.All([FLoadDone, done]);
Path1.AddIdleHandler(
currPathData.Changed,
procedure
begin
if currPathData.Value<>nil then
begin
Path1.Data.Assign( currPathData.Value.Obj );
currPathData.Value := nil;
end;
end );
(*
FLoadDone := TState.All([FLoadDone, done]);
Path1.AddIdleHandler(
curr.Changed,
procedure
begin
var Prices := curr.Value;
if Prices.Count > 0 then
begin
Path1.BeginUpdate;
try
Path1.Data.Clear;
Path1.Data.MoveTo(PointF(Path1.Width - 1, Prices[0].Data.Ask));
for var i := 1 to Prices.Count - 1 do
begin
Path1.Data.LineTo(PointF(Path1.Width - i - 1, Prices[i].Data.Ask));
end;
finally
Path1.EndUpdate;
end;
end;
end
);
*)
end;
procedure TForm1.Button2Click(Sender: TObject);
procedure TForm1.StopButtonClick(Sender: TObject);
begin
FTerminate.Notify;
end;
@@ -174,6 +97,8 @@ begin
FServer := TAuraTABFileServer.Create('\\COFFEE\TickData\Pepperstone');
SymbolsComboBox.Enabled := false;
ChartButton.Enabled := false;
LoadButton.Enabled := false;
FSymbols := FServer.EnumerateSymbols;
@@ -190,6 +115,8 @@ begin
if SymbolsComboBox.ItemIndex < 0 then
SymbolsComboBox.ItemIndex := SymbolsComboBox.Items.IndexOf('GER40');
SymbolsComboBox.Enabled := true;
ChartButton.Enabled := true;
LoadButton.Enabled := true;
end;
finally
SymbolsComboBox.EndUpdate;
@@ -200,6 +127,7 @@ end;
procedure TForm1.FormDestroy(Sender: TObject);
begin
FSymbols.WaitFor;
FTerminate.Notify;
TaskManager.WaitFor(FLoadDone);
Application.OnIdle := nil;
@@ -279,4 +207,83 @@ begin
Proc(FRandom.Count - 1);
end;
procedure TForm1.ChartButtonClick(Sender: TObject);
begin
if SymbolsComboBox.ItemIndex < 0 then
exit;
var Symbol := FSymbols.WaitFor[SymbolsComboBox.ItemIndex];
var currPathData := TMutable<IObjectRef<TPathData>>.CreateProtected;
var currLog := TMutable<String>.CreateProtected;
const width = 1000;
const incsize = 100;
begin
var arr := TDataSeries<TAskBidItem>.CreateWriteable(width);
var done :=
FServer.ProcessData(
Symbol,
FTerminate.Signal,
procedure(const Values: TArray<TDataPoint<TAskBidItem>>; const Terminated: TState)
begin
arr.Add(Values);
currLog.Value := arr.TotalCount.ToString;
var PathData := TPathData.Create;
var Prices := arr;
if Prices.Count > 0 then
begin
PathData.MoveTo(PointF(Prices.Count - 1, Prices[0].Data.Ask));
for var i := 1 to Prices.Count - 1 do
PathData.LineTo(PointF(Prices.Count - i - 1, Prices[i].Data.Ask));
end;
currPathData.Value := TObjectRef<TPathData>.Create(PathData);
end
);
FLoadDone := TState.All([FLoadDone, done]);
end;
LogMemo.AddIdleHandler(currLog.Changed, procedure begin LogMemo.Lines.Add(Symbol + ': ' + currLog.Value); end);
Path1.AddIdleHandler(
currPathData.Changed,
procedure
begin
if currPathData.Value<>nil then
begin
Path1.Data.Assign( currPathData.Value.Obj );
end;
end );
(*
FLoadDone := TState.All([FLoadDone, done]);
Path1.AddIdleHandler(
curr.Changed,
procedure
begin
var Prices := curr.Value;
if Prices.Count > 0 then
begin
Path1.BeginUpdate;
try
Path1.Data.Clear;
Path1.Data.MoveTo(PointF(Path1.Width - 1, Prices[0].Data.Ask));
for var i := 1 to Prices.Count - 1 do
begin
Path1.Data.LineTo(PointF(Path1.Width - i - 1, Prices[i].Data.Ask));
end;
finally
Path1.EndUpdate;
end;
end;
end
);
*)
end;
end.
Src/Myc.Core.Lazy.pas
+27 -3
View File
@@ -10,9 +10,10 @@ uses
type
TMycNullMutable<T> = class(TInterfacedObject, TMutable<T>.IMutable)
protected
private
function GetChanged: TSignal;
function GetValue: T;
function GetWriter: TMutable<T>.IWriter;
end;
TMycMutableBase<T> = class(TInterfacedObject, TMutable<T>.IMutable)
@@ -20,6 +21,7 @@ type
FChanged: TEvent;
FChangeState: TSignal.TSubscription;
function GetChanged: TSignal;
function GetWriter: TMutable<T>.IWriter;
protected
function GetValue: T; virtual; abstract;
public
@@ -36,13 +38,14 @@ type
constructor Create(const AChanged: TSignal; const AProc: TFunc<T>);
end;
TMycWriteableMutable<T> = class(TInterfacedObject, TMutable<T>.IMutable, TWriteable<T>.IWriteable)
TMycWriteableMutable<T> = class(TInterfacedObject, TMutable<T>.IMutable, TMutable<T>.IWriter)
private
FValue: T;
FChanged: TEvent;
protected
function GetChanged: TSignal;
function GetValue: T;
function GetWriter: TMutable<T>.IWriter;
public
constructor Create(const AValue: T);
procedure SetValue(const Value: T);
@@ -77,7 +80,7 @@ type
constructor Create(const AChanged: TSignal.ISignal; const AProc: TFunc<T>);
end;
TMycProtectedWriteableMutable<T> = class(TInterfacedObject, TMutable<T>.IMutable, TWriteable<T>.IWriteable)
TMycProtectedWriteableMutable<T> = class(TInterfacedObject, TMutable<T>.IMutable, TMutable<T>.IWriter)
private
FValue: T;
FChanged: TEvent;
@@ -85,6 +88,7 @@ type
protected
function GetChanged: TSignal;
function GetValue: T;
function GetWriter: TMutable<T>.IWriter;
procedure Lock; inline;
procedure Release; inline;
public
@@ -106,6 +110,11 @@ begin
Result := Default(T);
end;
function TMycNullMutable<T>.GetWriter: TMutable<T>.IWriter;
begin
Result := nil;
end;
{ TMycMutableBase<T> }
constructor TMycMutableBase<T>.Create(const AChanged: TSignal);
@@ -126,6 +135,11 @@ begin
Result := FChanged.Signal;
end;
function TMycMutableBase<T>.GetWriter: TMutable<T>.IWriter;
begin
Result := nil;
end;
{ TMycWriteableMutable<T> }
constructor TMycWriteableMutable<T>.Create(const AValue: T);
@@ -145,6 +159,11 @@ begin
Result := FValue;
end;
function TMycWriteableMutable<T>.GetWriter: TMutable<T>.IWriter;
begin
Result := Self;
end;
procedure TMycWriteableMutable<T>.SetValue(const Value: T);
begin
FValue := Value;
@@ -244,6 +263,11 @@ begin
end;
end;
function TMycProtectedWriteableMutable<T>.GetWriter: TMutable<T>.IWriter;
begin
Result := Self;
end;
procedure TMycProtectedWriteableMutable<T>.Lock;
begin
while AtomicExchange(FLock, 1) = 1 do
Src/Myc.Lazy.pas
+32 -68
View File
@@ -9,13 +9,19 @@ uses
type
TMutable<T> = record
type
IWriter = interface
procedure SetValue(const Value: T);
end;
IMutable = interface
{$REGION 'property access'}
function GetChanged: TSignal;
function GetValue: T;
function GetWriter: IWriter;
{$ENDREGION}
property Changed: TSignal read GetChanged;
property Value: T read GetValue;
property Writer: IWriter read GetWriter;
end;
{$REGION 'private'}
@@ -27,7 +33,9 @@ type
private
FMutable: IMutable;
function GetChanged: TSignal; inline;
function GetIsWriteable: Boolean;
function GetValue: T; inline;
procedure SetValue(const Value: T);
{$ENDREGION}
public
constructor Create(const AMutable: IMutable);
@@ -38,35 +46,12 @@ type
class property Null: IMutable read FNull;
class function Construct(const Changing: TSignal; const Proc: TFunc<T>): TMutable<T>; overload; static;
property Value: T read GetValue;
property Changed: TSignal read GetChanged;
end;
TWriteable<T> = record
type
IWriteable = interface(TMutable<T>.IMutable)
procedure SetValue(const Value: T);
property Value: T read GetValue write SetValue;
end;
private
FWriteable: IWriteable;
function GetChanged: TSignal; inline;
function GetValue: T; inline;
procedure SetValue(const Value: T); inline;
public
constructor Create(const AWriteable: IWriteable);
class function CreateProtected: TWriteable<T>; overload; static;
class function CreateWriteable: TWriteable<T>; overload; static;
class operator Implicit(const A: IWriteable): TWriteable<T>; overload;
class operator Implicit(const A: TWriteable<T>): IWriteable; overload;
class function CreateProtected: TMutable<T>; overload; static;
class function CreateWriteable: TMutable<T>; overload; static;
property Value: T read GetValue write SetValue;
property Changed: TSignal read GetChanged;
property IsWriteable: Boolean read GetIsWriteable;
end;
TLazy<T> = record
@@ -127,16 +112,37 @@ begin
Result := TMycFuncMutable<T>.Create(Changing, Proc);
end;
class function TMutable<T>.CreateProtected: TMutable<T>;
begin
Result := TMycProtectedWriteableMutable<T>.Create(Default(T));
end;
class function TMutable<T>.CreateWriteable: TMutable<T>;
begin
Result := TMycWriteableMutable<T>.Create(Default(T));
end;
function TMutable<T>.GetChanged: TSignal;
begin
Result := FMutable.Changed;
end;
function TMutable<T>.GetIsWriteable: Boolean;
begin
Result := Assigned( FMutable.Writer );
end;
function TMutable<T>.GetValue: T;
begin
Result := FMutable.Value;
end;
procedure TMutable<T>.SetValue(const Value: T);
begin
Assert( IsWriteable );
FMutable.Writer.SetValue( Value );
end;
class operator TMutable<T>.Implicit(const A: TMutable<T>): IMutable;
begin
Result := A.FMutable;
@@ -201,46 +207,4 @@ begin
Dest.FLazy := FNull;
end;
{ TWriteable<T> }
constructor TWriteable<T>.Create(const AWriteable: IWriteable);
begin
FWriteable := AWriteable;
end;
class function TWriteable<T>.CreateProtected: TWriteable<T>;
begin
Result := TMycProtectedWriteableMutable<T>.Create(Default(T));
end;
class function TWriteable<T>.CreateWriteable: TWriteable<T>;
begin
Result := TMycWriteableMutable<T>.Create(Default(T));
end;
function TWriteable<T>.GetChanged: TSignal;
begin
Result := FWriteable.Changed;
end;
function TWriteable<T>.GetValue: T;
begin
Result := FWriteable.Value;
end;
procedure TWriteable<T>.SetValue(const Value: T);
begin
FWriteable.Value := Value;
end;
class operator TWriteable<T>.Implicit(const A: IWriteable): TWriteable<T>;
begin
Result.Create(A);
end;
class operator TWriteable<T>.Implicit(const A: TWriteable<T>): IWriteable;
begin
Result := A.FWriteable;
end;
end.
Src/Myc.Trade.Core.DataPoint.pas
+52 -34
View File
@@ -7,7 +7,7 @@ uses
type
// The implementation class for IDataSeries<T>.
TDataArray<T> = class(TInterfacedObject, IDataSeries<T>, IDataArray<T>)
TDataArray<T> = class(TInterfacedObject, IDataSeries<T>, IDataSeriesWriter<T>)
const
ChunkSize = 1024;
type
@@ -15,7 +15,7 @@ type
private
FChunks: TArray<TChunk>;
FCount: Int64;
FMaxLookback: Int64;
FLookback: Int64;
FTotalCount: Int64;
function LogicalToPhysicalIndex(LogicalIndex: Int64): Int64; inline;
@@ -25,15 +25,16 @@ type
// IDataSeries<T> implementation
function GetCount: Int64;
function GetItems(Idx: Int64): TDataPoint<T>;
function GetMaxLookback: Int64;
function GetLookback: Int64;
function GetTotalCount: Int64;
function GetWriter: IDataSeriesWriter<T>;
public
constructor Create(AMaxLookback: Int64);
constructor Create(ALookback: Int64);
procedure Add(const Data: TDataPoint<T>); overload;
procedure Add(const Data: array of TDataPoint<T>; NumToAdd: Integer = -1); overload;
procedure Clear;
function Copy: IDataSeries<T>;
property MaxLookback: Int64 read GetMaxLookback;
function Immutable: IDataSeries<T>;
property Lookback: Int64 read GetLookback;
end;
// Null object implementation for IDataSeries<T>
@@ -43,12 +44,13 @@ type
FNull: IDataSeries<T>;
private
class constructor CreateClass;
function GetWriter: IDataSeriesWriter<T>;
public
function GetCount: Int64;
function GetItems(Idx: Int64): TDataPoint<T>;
function GetTotalCount: Int64;
function IndexOf(TimeStamp: TDateTime): Int64;
function Copy: IDataSeries<T>;
function Immutable: IDataSeries<T>;
class property Null: IDataSeries<T> read FNull;
end;
@@ -57,7 +59,7 @@ type
private
FChunks: TArray<TDataArray<T>.TChunk>;
FCount: Int64;
FMaxLookback: Int64;
FLookback: Int64;
FTotalCount: Int64;
function LogicalToPhysicalIndex(LogicalIndex: Int64): Int64; inline;
@@ -66,23 +68,24 @@ type
// IDataSeries<T> implementation
function GetCount: Int64;
function GetItems(Idx: Int64): TDataPoint<T>;
function GetMaxLookback: Int64;
function GetLookback: Int64;
function GetTotalCount: Int64;
function GetWriter: IDataSeriesWriter<T>;
public
constructor Create(const AChunks: TArray<TDataArray<T>.TChunk>; ACount, AMaxLookback, ATotalCount: Int64);
function Copy: IDataSeries<T>;
property MaxLookback: Int64 read GetMaxLookback;
constructor Create(const AChunks: TArray<TDataArray<T>.TChunk>; ACount, ALookback, ATotalCount: Int64);
function Immutable: IDataSeries<T>;
property Lookback: Int64 read GetLookback;
end;
implementation
{ TDataArray<T> }
constructor TDataArray<T>.Create(AMaxLookback: Int64);
constructor TDataArray<T>.Create(ALookback: Int64);
begin
inherited Create;
FMaxLookback := AMaxLookback;
FLookback := ALookback;
FCount := 0;
FTotalCount := 0;
FChunks := nil;
@@ -170,16 +173,16 @@ begin
FTotalCount := 0;
end;
function TDataArray<T>.Copy: IDataSeries<T>;
function TDataArray<T>.Immutable: IDataSeries<T>;
begin
Result := TStaticDataSeries<T>.Create(FChunks, FCount, FMaxLookback, FTotalCount);
Result := TStaticDataSeries<T>.Create(FChunks, FCount, FLookback, FTotalCount);
end;
function TDataArray<T>.GetCount: Int64;
begin
Result := FCount;
if (FMaxLookback > 0) and (Result > FMaxLookback) then
Result := FMaxLookback;
if (FLookback > 0) and (Result > FLookback) then
Result := FLookback;
end;
function TDataArray<T>.GetItems(Idx: Int64): TDataPoint<T>;
@@ -191,9 +194,9 @@ begin
Result := FChunks[physicalIndex div ChunkSize][physicalIndex mod ChunkSize];
end;
function TDataArray<T>.GetMaxLookback: Int64;
function TDataArray<T>.GetLookback: Int64;
begin
Result := FMaxLookback;
Result := FLookback;
end;
function TDataArray<T>.GetTotalCount: Int64;
@@ -201,9 +204,14 @@ begin
Result := FTotalCount;
end;
function TDataArray<T>.GetWriter: IDataSeriesWriter<T>;
begin
Result := Self;
end;
function TDataArray<T>.IsIndexInLimits(LogicalIndex: Int64): Boolean;
begin
Result := (FMaxLookback <= 0) or (LogicalIndex < FMaxLookback);
Result := (FLookback <= 0) or (LogicalIndex < FLookback);
end;
function TDataArray<T>.LogicalToPhysicalIndex(LogicalIndex: Int64): Int64;
@@ -216,12 +224,12 @@ procedure TDataArray<T>.Trim;
var
itemsToRemove, chunksToRemove: Int64;
begin
if (FCount = 0) or (FMaxLookback <= 0) then
if (FCount = 0) or (FLookback <= 0) then
Exit;
itemsToRemove := 0;
if FCount > FMaxLookback then
itemsToRemove := FCount - FMaxLookback;
if FCount > FLookback then
itemsToRemove := FCount - FLookback;
if itemsToRemove <= 0 then
Exit;
@@ -231,7 +239,7 @@ begin
if chunksToRemove > 0 then
begin
var itemsInFreedChunks := chunksToRemove * ChunkSize;
FChunks := System.Copy(FChunks, chunksToRemove, Length(FChunks) - chunksToRemove);
FChunks := Copy(FChunks, chunksToRemove, Length(FChunks) - chunksToRemove);
FCount := FCount - itemsInFreedChunks;
end;
end;
@@ -243,7 +251,7 @@ begin
FNull := TNullDataSeries<T>.Create;
end;
function TNullDataSeries<T>.Copy: IDataSeries<T>;
function TNullDataSeries<T>.Immutable: IDataSeries<T>;
begin
Result := FNull;
end;
@@ -264,6 +272,11 @@ begin
Result := 0;
end;
function TNullDataSeries<T>.GetWriter: IDataSeriesWriter<T>;
begin
Result := nil;
end;
function TNullDataSeries<T>.IndexOf(TimeStamp: TDateTime): Int64;
begin
Result := -1;
@@ -271,16 +284,16 @@ end;
{ TStaticDataSeries<T> }
constructor TStaticDataSeries<T>.Create(const AChunks: TArray<TDataArray<T>.TChunk>; ACount, AMaxLookback, ATotalCount: Int64);
constructor TStaticDataSeries<T>.Create(const AChunks: TArray<TDataArray<T>.TChunk>; ACount, ALookback, ATotalCount: Int64);
begin
inherited Create;
FChunks := AChunks;
FCount := ACount;
FMaxLookback := AMaxLookback;
FLookback := ALookback;
FTotalCount := ATotalCount;
end;
function TStaticDataSeries<T>.Copy: IDataSeries<T>;
function TStaticDataSeries<T>.Immutable: IDataSeries<T>;
begin
Result := Self;
end;
@@ -288,8 +301,8 @@ end;
function TStaticDataSeries<T>.GetCount: Int64;
begin
Result := FCount;
if (FMaxLookback > 0) and (Result > FMaxLookback) then
Result := FMaxLookback;
if (FLookback > 0) and (Result > FLookback) then
Result := FLookback;
end;
function TStaticDataSeries<T>.GetItems(Idx: Int64): TDataPoint<T>;
@@ -301,9 +314,9 @@ begin
Result := FChunks[physicalIndex div TDataArray<T>.ChunkSize][physicalIndex mod TDataArray<T>.ChunkSize];
end;
function TStaticDataSeries<T>.GetMaxLookback: Int64;
function TStaticDataSeries<T>.GetLookback: Int64;
begin
Result := FMaxLookback;
Result := FLookback;
end;
function TStaticDataSeries<T>.GetTotalCount: Int64;
@@ -311,9 +324,14 @@ begin
Result := FTotalCount;
end;
function TStaticDataSeries<T>.GetWriter: IDataSeriesWriter<T>;
begin
Result := nil;
end;
function TStaticDataSeries<T>.IsIndexInLimits(LogicalIndex: Int64): Boolean;
begin
Result := (FMaxLookback <= 0) or (LogicalIndex < FMaxLookback);
Result := (FLookback <= 0) or (LogicalIndex < FLookback);
end;
function TStaticDataSeries<T>.LogicalToPhysicalIndex(LogicalIndex: Int64): Int64;
Src/Myc.Trade.DataPoint.pas
+38 -21
View File
@@ -17,25 +17,27 @@ type
constructor Create(ATime: TDateTime; const AData: T);
end;
IDataSeriesWriter<T> = interface
procedure Add(const Data: TDataPoint<T>); overload;
procedure Add(const Data: array of TDataPoint<T>; NumToAdd: Integer = -1); overload;
procedure Clear;
end;
// A time-ordered series of data points, optimized for chronological additions.
// The most recently added element has the logical index 0.
IDataSeries<T> = interface
function GetCount: Int64;
function GetItems(Idx: Int64): TDataPoint<T>;
function GetTotalCount: Int64;
function Copy: IDataSeries<T>;
function Immutable: IDataSeries<T>;
function GetWriter: IDataSeriesWriter<T>;
property Count: Int64 read GetCount;
// Accesses data points by their logical index.
// Index 0 is the newest element, Index (Count - 1) is the oldest.
property Items[Idx: Int64]: TDataPoint<T> read GetItems; default;
// The total number of items ever added to the series.
property TotalCount: Int64 read GetTotalCount;
end;
IDataArray<T> = interface(IDataSeries<T>)
procedure Add(const Data: TDataPoint<T>); overload;
procedure Add(const Data: array of TDataPoint<T>; NumToAdd: Integer = -1); overload;
procedure Clear;
property Writer: IDataSeriesWriter<T> read GetWriter;
end;
// Interface Helper for IDataSeries<T>.
@@ -43,12 +45,11 @@ type
TDataSeries<T> = record
private
FDataSeries: IDataSeries<T>;
FMaxLookback: Int64;
function GetCount: Int64;
function GetItems(Idx: Int64): TDataPoint<T>;
function GetData(Idx: Int64): T;
function GetIsWriteable: Boolean;
function GetTime(Idx: Int64): TDateTime;
function GetMaxLookback: Int64;
class function GetNull: IDataSeries<T>; static;
function GetTotalCount: Int64;
public
@@ -60,10 +61,14 @@ type
class operator Implicit(const A: TDataSeries<T>): IDataSeries<T>;
class operator Implicit(const A: IDataSeries<T>): TDataSeries<T>;
class function CreateArray(MaxLookback: Int64): IDataArray<T>; static;
class function CreateWriteable(MaxLookback: Int64): TDataSeries<T>; static;
// Create a guaranteed immutable version of the given series.
function Copy: TDataSeries<T>;
// Create an immutable version of the given series.
function Immutable: TDataSeries<T>;
// Writing (only if IsWriteable=true)
procedure Add(const Data: array of TDataPoint<T>; NumToAdd: Integer = -1);
procedure Clear;
// Searches for a data point by its timestamp.
// Returns the logical index of the matching item.
@@ -75,8 +80,8 @@ type
property Count: Int64 read GetCount;
property Items[Idx: Int64]: TDataPoint<T> read GetItems; default;
property Data[Idx: Int64]: T read GetData;
property IsWriteable: Boolean read GetIsWriteable;
property Time[Idx: Int64]: TDateTime read GetTime;
property MaxLookback: Int64 read GetMaxLookback;
property TotalCount: Int64 read GetTotalCount;
end;
@@ -109,12 +114,24 @@ begin
FDataSeries := Null;
end;
function TDataSeries<T>.Copy: TDataSeries<T>;
procedure TDataSeries<T>.Add(const Data: array of TDataPoint<T>; NumToAdd: Integer = -1);
begin
Result := FDataSeries.Copy;
Assert( IsWriteable );
FDataSeries.Writer.Add( Data, NumToAdd );
end;
class function TDataSeries<T>.CreateArray(MaxLookback: Int64): IDataArray<T>;
procedure TDataSeries<T>.Clear;
begin
Assert( IsWriteable );
FDataSeries.Writer.Clear;
end;
function TDataSeries<T>.Immutable: TDataSeries<T>;
begin
Result := FDataSeries.Immutable;
end;
class function TDataSeries<T>.CreateWriteable(MaxLookback: Int64): TDataSeries<T>;
begin
Result := TDataArray<T>.Create(MaxLookback);
end;
@@ -154,16 +171,16 @@ begin
Result := FDataSeries[Idx].Data;
end;
function TDataSeries<T>.GetIsWriteable: Boolean;
begin
Result := Assigned( FDataSeries.Writer );
end;
function TDataSeries<T>.GetTime(Idx: Int64): TDateTime;
begin
Result := FDataSeries[Idx].Time;
end;
function TDataSeries<T>.GetMaxLookback: Int64;
begin
Result := FMaxLookback;
end;
class function TDataSeries<T>.GetNull: IDataSeries<T>;
begin
Result := TNullDataSeries<T>.Null;
Src/Myc.Trade.DataProvider.pas
+10 -4
View File
@@ -12,12 +12,13 @@ type
TDataStreamProvider<T> = class(TInterfacedObject, TMutable<TDataSeries<T>>.IMutable)
private
FStream: IDataStream<T>;
FDataArray: IDataArray<T>;
FDataSeries: TDataSeries<T>;
FNewDataAvailable: TFlag;
FChunk: TArray<TDataPoint<T>>;
FLookback: Int64;
function GetChanged: TSignal;
function GetValue: TDataSeries<T>;
function GetWriter: TMutable<TDataSeries<T>>.IWriter;
public
constructor Create(ALookback, AMaxChunkSize: Int64; const AStream: IDataStream<T>);
destructor Destroy; override;
@@ -38,7 +39,7 @@ begin
SetLength(FChunk, AMaxChunkSize);
FStream := AStream;
FNewDataAvailable := TFlag.CreateObserver(FStream.HasData);
FDataArray := TDataSeries<T>.CreateArray(ALookback);
FDataSeries := TDataSeries<T>.CreateWriteable(ALookback);
end;
destructor TDataStreamProvider<T>.Destroy;
@@ -57,9 +58,14 @@ begin
begin
var cnt := FStream.GetChunk(FChunk);
if cnt > 0 then
FDataArray.Add(FChunk, cnt);
FDataSeries.Add(FChunk, cnt);
end;
Result := FDataArray;
Result := FDataSeries;
end;
function TDataStreamProvider<T>.GetWriter: TMutable<TDataSeries<T>>.IWriter;
begin
Result := nil;
end;
{ TDataStreamProvider }
Src/Myc.Trade.DataStream.pas
+17 -8
View File
@@ -101,7 +101,8 @@ type
// Used by cache to actually load an uncached file.
function DoLoad(const FileName: string): TFuture<TArray<TDataPoint<T>>>;
function LoadFile(DataFile: TAuraDataFile; Terminated: TState; Proc: TDataProc<T>): TState;
function ProcessFile(FileInfo: TAuraDataFile; const DataFile: TFuture<TArray<TDataPoint<T>>>; Terminated: TState; Proc: TDataProc<T>):
TState;
strict private
class var
@@ -392,27 +393,35 @@ begin
Result :=
FindFirstFile(Symbol)
.Chain(function(const FirstFile: TAuraDataFile): TState begin Result := LoadFile(FirstFile, terminated, capProc); end);
.Chain(
function(const FirstFileInfo: TAuraDataFile): TState
begin
Result := ProcessFile(FirstFileInfo, LoadDataFile(FirstFileInfo), terminated, capProc);
end);
end;
function TAuraDataServer<T>.LoadDataFile(const DataFile: TAuraDataFile): TFuture<TArray<TDataPoint<T>>>;
begin
Result := FCachedFiles.GetOrAdd(DataFile.GetFullFileName);
if DataFile.IsValid then
Result := FCachedFiles.GetOrAdd(DataFile.GetFullFileName);
end;
function TAuraDataServer<T>.LoadFile(DataFile: TAuraDataFile; Terminated: TState; Proc: TDataProc<T>): TState;
function TAuraDataServer<T>.ProcessFile(FileInfo: TAuraDataFile; const DataFile: TFuture<TArray<TDataPoint<T>>>; Terminated: TState; Proc:
TDataProc<T>): TState;
begin
if not DataFile.IsValid or Terminated.IsSet then
if not FileInfo.IsValid or Terminated.IsSet then
exit(TState.Null);
var data := FCachedFiles.GetOrAdd(DataFile.GetFullFileName);
// Read ahead the next file, while processing the current one
var nextFileInfo := FileInfo.GetNextFile;
var nextFile := LoadDataFile( nextFileInfo );
Result :=
data.Chain(
DataFile.Chain(
function(const Data: TArray<TDataPoint<T>>): TState
begin
Proc(Data, Terminated);
Result := LoadFile(DataFile.GetNextFile, Terminated, Proc);
Result := ProcessFile(nextFileInfo, nextFile, Terminated, Proc);
end
);
end;