Removed bottleneck in DataStream
This commit is contained in:
Michael Schimmel
@@ -158,6 +158,12 @@ type
|
||||
|
||||
// Endpoint that collects data into a series.
|
||||
TMycDataEndpoint<T> = class(TInterfacedObject, TMutable<TSeries<T>>.IMutable)
|
||||
type
|
||||
PItem = ^TItem;
|
||||
TItem = record
|
||||
Next: PItem;
|
||||
Value: T;
|
||||
end;
|
||||
private
|
||||
FProcessor: TMycContainedProcessor<T>;
|
||||
FTag: TDataProvider<T>.TTag;
|
||||
@@ -166,6 +172,7 @@ type
|
||||
FData: TSeries<T>;
|
||||
FChanged: TEvent;
|
||||
FLock: TLightweightMREW;
|
||||
FFirst: PItem;
|
||||
function GetChanged: TSignal;
|
||||
function GetValue: TSeries<T>;
|
||||
function ProcessData(const Value: T): TState;
|
||||
@@ -195,6 +202,20 @@ type
|
||||
function ProcessData(const Value: T): TState; override; final;
|
||||
end;
|
||||
|
||||
TOhlcAggregation = class(TMycConverter<TDataPoint<TOhlcItem>, TDataPoint<TOhlcItem>>)
|
||||
private
|
||||
FTimeframe: TTimeframe;
|
||||
FCurrentBar: TDataPoint<TOhlcItem>;
|
||||
function GetBarStartTime(const TimeStamp: TDateTime; const Timeframe: TTimeframe): TDateTime;
|
||||
function GetCurrentBar: TDataPoint<TOhlcItem>;
|
||||
function GetTimeframe: TTimeframe;
|
||||
public
|
||||
constructor Create(const ATimeframe: TTimeframe);
|
||||
function ProcessData(const Value: TDataPoint<TOhlcItem>): TState; override;
|
||||
property CurrentBar: TDataPoint<TOhlcItem> read GetCurrentBar;
|
||||
property Timeframe: TTimeframe read GetTimeframe;
|
||||
end;
|
||||
|
||||
implementation
|
||||
|
||||
uses
|
||||
@@ -461,11 +482,42 @@ end;
|
||||
|
||||
function TMycDataEndpoint<T>.GetValue: TSeries<T>;
|
||||
begin
|
||||
FLock.BeginRead;
|
||||
FLock.BeginWrite;
|
||||
try
|
||||
var cnt := 0;
|
||||
var tmp: PItem := nil;
|
||||
var item: PItem;
|
||||
|
||||
while FFirst <> nil do
|
||||
begin
|
||||
item := FFirst;
|
||||
FFirst := item.Next;
|
||||
item.Next := tmp;
|
||||
tmp := item;
|
||||
inc(cnt);
|
||||
end;
|
||||
|
||||
if cnt > 0 then
|
||||
begin
|
||||
var Arr: TArray<T>;
|
||||
SetLength(Arr, cnt);
|
||||
|
||||
cnt := 0;
|
||||
while tmp <> nil do
|
||||
begin
|
||||
item := tmp;
|
||||
tmp := item.Next;
|
||||
Arr[cnt] := item.Value;
|
||||
inc(cnt);
|
||||
Dispose(item);
|
||||
end;
|
||||
|
||||
FData := FData.Add(Arr, 0, cnt, FLookback);
|
||||
end;
|
||||
|
||||
Result := FData;
|
||||
finally
|
||||
FLock.EndRead;
|
||||
FLock.EndWrite;
|
||||
end;
|
||||
end;
|
||||
|
||||
@@ -474,8 +526,14 @@ begin
|
||||
Result := TState.Null;
|
||||
FLock.BeginWrite;
|
||||
try
|
||||
var P: PItem;
|
||||
New(P);
|
||||
P.Next := FFirst;
|
||||
FFirst := P;
|
||||
P.Value := Value;
|
||||
|
||||
// Add new data point, respecting the lookback period.
|
||||
FData := FData.Add(Value, FLookback);
|
||||
// FData := FData.Add(Value, FLookback);
|
||||
finally
|
||||
FLock.EndWrite;
|
||||
end;
|
||||
@@ -657,4 +715,104 @@ begin
|
||||
FQueue := Result;
|
||||
end;
|
||||
|
||||
{ TOhlcAggregation }
|
||||
|
||||
constructor TOhlcAggregation.Create(const ATimeframe: TTimeframe);
|
||||
begin
|
||||
inherited Create;
|
||||
FTimeframe := ATimeframe;
|
||||
end;
|
||||
|
||||
function TOhlcAggregation.GetBarStartTime(const TimeStamp: TDateTime; const Timeframe: TTimeframe): TDateTime;
|
||||
var
|
||||
baseTime: TDateTime;
|
||||
begin
|
||||
// Align the time grid to UTC 0:00 using functions from System.DateUtils
|
||||
baseTime := RecodeMilliSecond(TimeStamp, 0);
|
||||
|
||||
case Timeframe of
|
||||
S: Result := baseTime;
|
||||
S5: Result := RecodeSecond(baseTime, SecondOf(TimeStamp) - (SecondOf(TimeStamp) mod 5));
|
||||
S15: Result := RecodeSecond(baseTime, SecondOf(TimeStamp) - (SecondOf(TimeStamp) mod 15));
|
||||
S30: Result := RecodeSecond(baseTime, SecondOf(TimeStamp) - (SecondOf(TimeStamp) mod 30));
|
||||
|
||||
M: Result := RecodeSecond(baseTime, 0);
|
||||
M2: Result := RecodeMinute(RecodeSecond(baseTime, 0), MinuteOf(TimeStamp) - (MinuteOf(TimeStamp) mod 2));
|
||||
M3: Result := RecodeMinute(RecodeSecond(baseTime, 0), MinuteOf(TimeStamp) - (MinuteOf(TimeStamp) mod 3));
|
||||
M5: Result := RecodeMinute(RecodeSecond(baseTime, 0), MinuteOf(TimeStamp) - (MinuteOf(TimeStamp) mod 5));
|
||||
M10: Result := RecodeMinute(RecodeSecond(baseTime, 0), MinuteOf(TimeStamp) - (MinuteOf(TimeStamp) mod 10));
|
||||
M15: Result := RecodeMinute(RecodeSecond(baseTime, 0), MinuteOf(TimeStamp) - (MinuteOf(TimeStamp) mod 15));
|
||||
M30: Result := RecodeMinute(RecodeSecond(baseTime, 0), MinuteOf(TimeStamp) - (MinuteOf(TimeStamp) mod 30));
|
||||
|
||||
H: Result := RecodeMinute(RecodeSecond(baseTime, 0), 0);
|
||||
H2: Result := RecodeHour(RecodeMinute(RecodeSecond(baseTime, 0), 0), HourOf(TimeStamp) - (HourOf(TimeStamp) mod 2));
|
||||
H3: Result := RecodeHour(RecodeMinute(RecodeSecond(baseTime, 0), 0), HourOf(TimeStamp) - (HourOf(TimeStamp) mod 3));
|
||||
H4: Result := RecodeHour(RecodeMinute(RecodeSecond(baseTime, 0), 0), HourOf(TimeStamp) - (HourOf(TimeStamp) mod 4));
|
||||
H8: Result := RecodeHour(RecodeMinute(RecodeSecond(baseTime, 0), 0), HourOf(TimeStamp) - (HourOf(TimeStamp) mod 8));
|
||||
H12: Result := RecodeHour(RecodeMinute(RecodeSecond(baseTime, 0), 0), HourOf(TimeStamp) - (HourOf(TimeStamp) mod 12));
|
||||
|
||||
D: Result := StartOfTheDay(TimeStamp);
|
||||
// D2, D3 are uncommon; this is a simple modulo-based approach relative to TDateTime's epoch.
|
||||
D2: Result := Floor(TimeStamp) - (Floor(TimeStamp) mod 2);
|
||||
D3: Result := Floor(TimeStamp) - (Floor(TimeStamp) mod 3);
|
||||
|
||||
W: Result := TimeStamp.StartOfTheWeek;
|
||||
|
||||
MN: Result := TimeStamp.StartOfTheMonth;
|
||||
// Quarter alignment
|
||||
MN3: Result := RecodeMonth(TimeStamp.StartOfTheMonth, (MonthOf(TimeStamp) - 1) div 3 * 3 + 1);
|
||||
// Half-year alignment
|
||||
MN6: Result := RecodeMonth(TimeStamp.StartOfTheMonth, (MonthOf(TimeStamp) - 1) div 6 * 6 + 1);
|
||||
|
||||
Y: Result := TimeStamp.StartOfTheYear;
|
||||
else
|
||||
// Fallback for any undefined timeframe
|
||||
Result := 0;
|
||||
end;
|
||||
end;
|
||||
|
||||
function TOhlcAggregation.GetCurrentBar: TDataPoint<TOhlcItem>;
|
||||
begin
|
||||
Result := FCurrentBar;
|
||||
end;
|
||||
|
||||
function TOhlcAggregation.GetTimeframe: TTimeframe;
|
||||
begin
|
||||
Result := FTimeframe;
|
||||
end;
|
||||
|
||||
function TOhlcAggregation.ProcessData(const Value: TDataPoint<TOhlcItem>): TState;
|
||||
var
|
||||
barStartTime: TDateTime;
|
||||
lastBarTime: TDateTime;
|
||||
begin
|
||||
// Update bar for the strategy's timeframe
|
||||
barStartTime := GetBarStartTime(Value.Time, FTimeframe);
|
||||
lastBarTime := FCurrentBar.Time;
|
||||
|
||||
if (barStartTime > lastBarTime) then
|
||||
begin
|
||||
// A new bar starts, so the previous one is now complete.
|
||||
if (lastBarTime > 0) then
|
||||
begin
|
||||
Result := Broadcast(FCurrentBar);
|
||||
end;
|
||||
|
||||
// Start a new bar, Volume is 1 because this is the first tick.
|
||||
FCurrentBar.Data := Value.Data;
|
||||
FCurrentBar.Time := barStartTime;
|
||||
end
|
||||
else
|
||||
begin
|
||||
// Update the currently aggregating bar
|
||||
if Value.Data.High > FCurrentBar.Data.High then
|
||||
FCurrentBar.Data.High := Value.Data.High;
|
||||
if Value.Data.Low < FCurrentBar.Data.Low then
|
||||
FCurrentBar.Data.Low := Value.Data.Low;
|
||||
FCurrentBar.Data.Close := Value.Data.Close;
|
||||
// Volume is the number of ticks needed to build the complete bar.
|
||||
FCurrentBar.Data.Volume := FCurrentBar.Data.Volume + Value.Data.Volume;
|
||||
end;
|
||||
end;
|
||||
|
||||
end.
|
||||
|
||||
Reference in New Issue
Block a user