Michael Schimmel
944 lines
30 KiB
ObjectPascal
944 lines
30 KiB
ObjectPascal
unit Myc.Trade.DataPoint.Impl;
|
|
|
|
interface
|
|
|
|
uses
|
|
System.SysUtils,
|
|
System.SyncObjs,
|
|
System.Generics.Collections,
|
|
Myc.Signals,
|
|
Myc.Mutable,
|
|
Myc.Core.Notifier,
|
|
Myc.Trade.Types,
|
|
Myc.Trade.DataArray,
|
|
Myc.Trade.DataPoint;
|
|
|
|
type
|
|
// Abstract base class for data consumers.
|
|
TMycProcessor<T> = class abstract(TInterfacedObject, IMycProcessor<T>)
|
|
protected
|
|
function Execute(const Value: T; const Proc: TConstFunc<T, TState>): TState;
|
|
function ProcessData(const Value: T): TState; virtual; abstract;
|
|
end;
|
|
|
|
// Concrete data provider that manages a list of processors (listeners).
|
|
TMycContainedDataProvider<T> = class abstract(TContainedObject, TDataProvider<T>.IDataProvider)
|
|
private
|
|
FListeners: TMycNotifyList<IMycProcessor<T>>;
|
|
public
|
|
constructor Create(const Controller: IInterface);
|
|
destructor Destroy; override;
|
|
// Notifies all linked processors.
|
|
function Broadcast(const Value: T): TState;
|
|
// Link a Processor
|
|
function Link(const Processor: IMycProcessor<T>): TDataProvider<T>.TTag;
|
|
// Unlink a linked strategy
|
|
procedure Unlink(Tag: TDataProvider<T>.TTag);
|
|
end;
|
|
|
|
TMycSequence<T> = class(TMycProcessor<T>, IMycDataSequence<T>)
|
|
private
|
|
FDataProviders: TArray<TMycContainedDataProvider<T>>;
|
|
function GetCount: Integer;
|
|
function GetDataProvider(Idx: Integer): TDataProvider<T>;
|
|
protected
|
|
function ProcessData(const Value: T): TState; override;
|
|
function ProcessDataProvider(Idx: Integer; const Value: T): TState;
|
|
public
|
|
constructor Create(ACount: Integer);
|
|
destructor Destroy; override;
|
|
end;
|
|
|
|
// Null object implementation for IDataProvider.
|
|
TNullDataProvider<T> = class(TInterfacedObject, TDataProvider<T>.IDataProvider)
|
|
public
|
|
function Link(const Receiver: IMycProcessor<T>): TDataProvider<T>.TTag;
|
|
procedure Unlink(Tag: TDataProvider<T>.TTag);
|
|
end;
|
|
|
|
// Abstract base class for components that process data of type S and provide data of type T.
|
|
TMycConverter<S, T> = class abstract(TMycProcessor<S>, TConverter<S, T>.IConverter)
|
|
private
|
|
FSender: TMycContainedDataProvider<T>;
|
|
function GetSender: TDataProvider<T>.IDataProvider;
|
|
protected
|
|
function ProcessData(const Value: S): TState; override; abstract;
|
|
// Broadcasts the given data to all linked processors.
|
|
function Broadcast(const Value: T): TState;
|
|
public
|
|
constructor Create;
|
|
destructor Destroy; override;
|
|
property Sender: TDataProvider<T>.IDataProvider read GetSender;
|
|
end;
|
|
|
|
// Null object implementation for IConverter.
|
|
TNullConverter<S, T> = class(TInterfacedObject, TConverter<S, T>.IConverter)
|
|
private
|
|
function GetSender: TDataProvider<T>.IDataProvider;
|
|
public
|
|
function ProcessData(const Value: S): TState;
|
|
end;
|
|
|
|
// A generic converter that uses a function reference for the conversion logic.
|
|
TMycGenericConverter<S, T> = class(TMycConverter<S, T>)
|
|
private
|
|
FFunc: TConstFunc<S, T>;
|
|
protected
|
|
function ProcessData(const Value: S): TState; override;
|
|
public
|
|
constructor Create(const AFunc: TConstFunc<S, T>);
|
|
end;
|
|
|
|
TMycGenericAggregator<S, T> = class(TMycConverter<S, T>)
|
|
private
|
|
FFunc: TConstFunc<S, TConverter<S, T>.TBroadcastProc, TState>;
|
|
protected
|
|
function ProcessData(const Value: S): TState; override;
|
|
public
|
|
constructor Create(const AFunc: TConstFunc<S, TConverter<S, T>.TBroadcastProc, TState>);
|
|
end;
|
|
|
|
TMycGenericParallelConverter<S, T> = class(TMycConverter<S, T>)
|
|
private
|
|
FFunc: TConstFunc<S, T>;
|
|
FQueue: TState;
|
|
protected
|
|
function ProcessData(const Value: S): TState; override;
|
|
public
|
|
constructor Create(const AFunc: TConstFunc<S, T>);
|
|
end;
|
|
|
|
TMycIdentityConverter<T> = class(TMycConverter<T, T>)
|
|
protected
|
|
function ProcessData(const Value: T): TState; override; final;
|
|
end;
|
|
|
|
// A converter specialized for calculating indicators.
|
|
TMycIndicator<S, T> = class(TMycConverter<S, T>)
|
|
protected
|
|
function ProcessData(const Value: S): TState; override; final;
|
|
function Calculate(const Value: S): T; virtual; abstract;
|
|
end;
|
|
|
|
// A converter that counts incoming data points and outputs the current count.
|
|
TMycDataCounter<T> = class(TMycConverter<T, Int64>)
|
|
private
|
|
FCount: Int64;
|
|
protected
|
|
function ProcessData(const Value: T): TState; override;
|
|
public
|
|
constructor Create;
|
|
end;
|
|
|
|
// A converter that takes an array and broadcasts each element individually.
|
|
TMycTicker<T> = class(TMycConverter<TArray<T>, T>)
|
|
public
|
|
function ProcessData(const Values: TArray<T>): TState; override;
|
|
end;
|
|
|
|
// A converter that reads a specific field from a record using RTTI.
|
|
TMycRecordFieldReader<S, T> = class(TMycConverter<S, T>)
|
|
private
|
|
FOffset: Integer;
|
|
public
|
|
constructor Create(const AFieldName: String);
|
|
function ProcessData(const Values: S): TState; override;
|
|
end;
|
|
|
|
// A generic processor implementation that uses a function reference.
|
|
TMycGenericProcessor<T> = class(TMycProcessor<T>)
|
|
private
|
|
FProc: TConstFunc<T, TState>;
|
|
protected
|
|
function ProcessData(const Value: T): TState; override; final;
|
|
public
|
|
constructor Create(const AProc: TConstFunc<T, TState>);
|
|
end;
|
|
|
|
// A processor implementation that is owned by a controller.
|
|
TMycContainedProcessor<T> = class(TContainedObject, IMycProcessor<T>)
|
|
private
|
|
FProc: TConstFunc<T, TState>;
|
|
function ProcessData(const Value: T): TState;
|
|
public
|
|
constructor Create(const Controller: IInterface; const AProc: TConstFunc<T, TState>);
|
|
end;
|
|
|
|
// Endpoint that collects data into a series.
|
|
TMycDataEndpoint<T> = class(TInterfacedObject, TLazy<TSeries<T>>.ILazy)
|
|
type
|
|
PItem = ^TItem;
|
|
TItem = record
|
|
Next: PItem;
|
|
Value: T;
|
|
end;
|
|
private
|
|
FProcessor: TMycContainedProcessor<T>;
|
|
FTag: TDataProvider<T>.TTag;
|
|
FDataProvider: TDataProvider<T>;
|
|
FLookback: Int64;
|
|
FChanged: TFlag;
|
|
FLock: TLightweightMREW;
|
|
FFirst: PItem;
|
|
FCount: Integer;
|
|
function GetChanged: TState;
|
|
function ProcessData(const Value: T): TState;
|
|
public
|
|
constructor Create(const ADataProvider: TDataProvider<T>; ALookback: Int64);
|
|
destructor Destroy; override;
|
|
function Update(var Value: TSeries<T>): Boolean;
|
|
end;
|
|
|
|
TTickAggregation = class(TMycConverter<TDataPoint<Double>, 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<Double>): TState; override;
|
|
property CurrentBar: TDataPoint<TOhlcItem> read GetCurrentBar;
|
|
property Timeframe: TTimeframe read GetTimeframe;
|
|
end;
|
|
|
|
TMycParallelConverter<T> = class(TMycConverter<T, T>)
|
|
private
|
|
FQueue: TState;
|
|
protected
|
|
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;
|
|
|
|
// Endpoint that collects data into a series.
|
|
TMycDataJoin<T> = class(TInterfacedObject, TDataProvider<TArray<T>>.IDataProvider)
|
|
private
|
|
FReceivers: array of record
|
|
DataProvider: TDataProvider<T>;
|
|
Processor: TMycContainedProcessor<T>;
|
|
Tag: TDataProvider<T>.TTag;
|
|
Queue: TQueue<T>;
|
|
end;
|
|
|
|
FSender: TMycContainedDataProvider<TArray<T>>;
|
|
FLock: TSpinLock;
|
|
FCount: Integer;
|
|
function ProcessData(Idx: Integer; const Value: T): TState;
|
|
public
|
|
constructor Create(const ADataProviders: TArray<TDataProvider<T>>);
|
|
destructor Destroy; override;
|
|
property Sender: TMycContainedDataProvider<TArray<T>> read FSender implements TDataProvider<TArray<T>>.IDataProvider;
|
|
end;
|
|
|
|
TMycComposedConverter<S, T> = class(TMycProcessor<S>, TConverter<S, T>.IConverter)
|
|
private
|
|
FProcessor: IMycProcessor<S>;
|
|
FDataProvider: TDataProvider<T>;
|
|
protected
|
|
function ProcessData(const Value: S): TState; override;
|
|
function GetSender: TDataProvider<T>.IDataProvider;
|
|
public
|
|
constructor Create(const AProcessor: IMycProcessor<S>; const ADataProvider: TDataProvider<T>);
|
|
end;
|
|
|
|
implementation
|
|
|
|
uses
|
|
System.TypInfo,
|
|
System.RTTI,
|
|
System.DateUtils,
|
|
System.Math,
|
|
Winapi.Windows,
|
|
Myc.TaskManager;
|
|
|
|
{ TMycProcessor<T> }
|
|
|
|
function TMycProcessor<T>.Execute(const Value: T; const Proc: TConstFunc<T, TState>): TState;
|
|
begin
|
|
var cValue := Value;
|
|
Result := TaskManager.RunTask(function: TState begin Result := Proc(cValue); end);
|
|
end;
|
|
|
|
{ TMycContainedDataProvider<T> }
|
|
|
|
constructor TMycContainedDataProvider<T>.Create(const Controller: IInterface);
|
|
begin
|
|
inherited Create(Controller);
|
|
end;
|
|
|
|
destructor TMycContainedDataProvider<T>.Destroy;
|
|
begin
|
|
FListeners.Finalize;
|
|
inherited Destroy;
|
|
end;
|
|
|
|
function TMycContainedDataProvider<T>.Broadcast(const Value: T): TState;
|
|
begin
|
|
FListeners.Lock;
|
|
try
|
|
var item := FListeners.First;
|
|
if item = nil then
|
|
exit(TState.Null);
|
|
|
|
var i := 1;
|
|
while item.Next <> nil do
|
|
begin
|
|
inc(i);
|
|
item := item.Next;
|
|
end;
|
|
|
|
var done := TLatch.CreateLatch(i);
|
|
while item <> nil do
|
|
begin
|
|
item.Receiver.ProcessData(Value).Signal.Subscribe(done);
|
|
item := item.Prev;
|
|
end;
|
|
|
|
Result := done.State;
|
|
finally
|
|
FListeners.Release;
|
|
end;
|
|
end;
|
|
|
|
function TMycContainedDataProvider<T>.Link(const Processor: IMycProcessor<T>): TDataProvider<T>.TTag;
|
|
begin
|
|
// Add the Processor to the notification list
|
|
FListeners.Lock;
|
|
try
|
|
Result := FListeners.Advise(Processor);
|
|
finally
|
|
FListeners.Release;
|
|
end;
|
|
end;
|
|
|
|
procedure TMycContainedDataProvider<T>.Unlink(Tag: TDataProvider<T>.TTag);
|
|
begin
|
|
FListeners.Lock;
|
|
try
|
|
FListeners.Unadvise(Tag);
|
|
finally
|
|
FListeners.Release;
|
|
end;
|
|
end;
|
|
|
|
{ TNullDataProvider<T> }
|
|
|
|
function TNullDataProvider<T>.Link(const Receiver: IMycProcessor<T>): TDataProvider<T>.TTag;
|
|
begin
|
|
Result := nil;
|
|
end;
|
|
|
|
procedure TNullDataProvider<T>.Unlink(Tag: TDataProvider<T>.TTag);
|
|
begin
|
|
// Do nothing in the null implementation.
|
|
end;
|
|
|
|
{ TMycConverter<S, T> }
|
|
|
|
constructor TMycConverter<S, T>.Create;
|
|
begin
|
|
inherited Create;
|
|
FSender := TMycContainedDataProvider<T>.Create(Self);
|
|
end;
|
|
|
|
destructor TMycConverter<S, T>.Destroy;
|
|
begin
|
|
FSender.Free;
|
|
inherited Destroy;
|
|
end;
|
|
|
|
function TMycConverter<S, T>.Broadcast(const Value: T): TState;
|
|
begin
|
|
Result := FSender.Broadcast(Value);
|
|
end;
|
|
|
|
function TMycConverter<S, T>.GetSender: TDataProvider<T>.IDataProvider;
|
|
begin
|
|
Result := FSender;
|
|
end;
|
|
|
|
{ TNullConverter<S, T> }
|
|
|
|
function TNullConverter<S, T>.GetSender: TDataProvider<T>.IDataProvider;
|
|
begin
|
|
Result := TDataProvider<T>.Null;
|
|
end;
|
|
|
|
function TNullConverter<S, T>.ProcessData(const Value: S): TState;
|
|
begin
|
|
Result := TState.Null;
|
|
end;
|
|
|
|
{ TMycGenericConverter<S, T> }
|
|
|
|
constructor TMycGenericConverter<S, T>.Create(const AFunc: TConstFunc<S, T>);
|
|
begin
|
|
inherited Create;
|
|
FFunc := AFunc;
|
|
end;
|
|
|
|
function TMycGenericConverter<S, T>.ProcessData(const Value: S): TState;
|
|
begin
|
|
Result := Broadcast(FFunc(Value));
|
|
end;
|
|
|
|
{ TMycIndicator<S,T> }
|
|
|
|
function TMycIndicator<S, T>.ProcessData(const Value: S): TState;
|
|
begin
|
|
Result := Execute(Value, function(const Value: S): TState begin Result := Broadcast(Calculate(Value)); end);
|
|
end;
|
|
|
|
{ TMycDataCounter<T> }
|
|
|
|
constructor TMycDataCounter<T>.Create;
|
|
begin
|
|
inherited Create;
|
|
FCount := 0;
|
|
end;
|
|
|
|
function TMycDataCounter<T>.ProcessData(const Value: T): TState;
|
|
begin
|
|
Result := Broadcast(FCount);
|
|
inc(FCount);
|
|
end;
|
|
|
|
{ TMycTicker<T> }
|
|
|
|
function TMycTicker<T>.ProcessData(const Values: TArray<T>): TState;
|
|
begin
|
|
var done := TLatch.CreateLatch(Length(Values));
|
|
|
|
// Process each incoming data point
|
|
for var i := 0 to High(Values) do
|
|
Broadcast(Values[i]).Signal.Subscribe(done);
|
|
|
|
Result := done.State;
|
|
end;
|
|
|
|
{ TMycRecordFieldReader<S, T> }
|
|
|
|
constructor TMycRecordFieldReader<S, T>.Create(const AFieldName: String);
|
|
begin
|
|
inherited Create;
|
|
|
|
var Context := TRttiContext.Create;
|
|
var Field := Context.GetType(TypeInfo(S)).GetField(AFieldName);
|
|
var TypeT := Context.GetType(TypeInfo(T));
|
|
|
|
var Fields := Context.GetType(TypeInfo(S)).GetFields;
|
|
|
|
Assert(Assigned(Field), 'Field ' + AFieldName + ' not found');
|
|
Assert(Field.FieldType.TypeKind = TypeT.TypeKind, 'Incorrect type');
|
|
|
|
if Assigned(Field) and (Field.FieldType.TypeKind = TypeT.TypeKind) then
|
|
FOffset := Field.Offset
|
|
else
|
|
FOffset := -1;
|
|
end;
|
|
|
|
function TMycRecordFieldReader<S, T>.ProcessData(const Values: S): TState;
|
|
type
|
|
PT = ^T;
|
|
begin
|
|
if FOffset < 0 then
|
|
exit(TState.Null);
|
|
|
|
var fieldPtr := PByte(@Values);
|
|
inc(fieldPtr, FOffset);
|
|
Result := Broadcast(PT(fieldPtr)^);
|
|
end;
|
|
|
|
{ TMycGenericProcessor<T> }
|
|
|
|
constructor TMycGenericProcessor<T>.Create(const AProc: TConstFunc<T, TState>);
|
|
begin
|
|
inherited Create;
|
|
FProc := AProc;
|
|
end;
|
|
|
|
function TMycGenericProcessor<T>.ProcessData(const Value: T): TState;
|
|
begin
|
|
Result := FProc(Value);
|
|
end;
|
|
|
|
{ TMycContainedProcessor<T> }
|
|
|
|
constructor TMycContainedProcessor<T>.Create(const Controller: IInterface; const AProc: TConstFunc<T, TState>);
|
|
begin
|
|
inherited Create(Controller);
|
|
FProc := AProc;
|
|
end;
|
|
|
|
function TMycContainedProcessor<T>.ProcessData(const Value: T): TState;
|
|
begin
|
|
Result := FProc(Value);
|
|
end;
|
|
|
|
{ TMycDataEndpoint<T> }
|
|
|
|
constructor TMycDataEndpoint<T>.Create(const ADataProvider: TDataProvider<T>; ALookback: Int64);
|
|
begin
|
|
inherited Create;
|
|
FDataProvider := ADataProvider;
|
|
FLookback := ALookback;
|
|
|
|
FProcessor := TMycContainedProcessor<T>.Create(Self, ProcessData);
|
|
FTag := FDataProvider.Link(FProcessor);
|
|
|
|
end;
|
|
|
|
destructor TMycDataEndpoint<T>.Destroy;
|
|
begin
|
|
FDataProvider.Unlink(FTag);
|
|
FProcessor.Free;
|
|
inherited;
|
|
end;
|
|
|
|
function TMycDataEndpoint<T>.GetChanged: TState;
|
|
begin
|
|
Result := FChanged.State
|
|
end;
|
|
|
|
function TMycDataEndpoint<T>.ProcessData(const Value: T): TState;
|
|
begin
|
|
FLock.BeginWrite;
|
|
try
|
|
var P: PItem;
|
|
New(P);
|
|
P.Next := FFirst;
|
|
FFirst := P;
|
|
P.Value := Value;
|
|
inc(FCount);
|
|
FChanged.Notify;
|
|
finally
|
|
FLock.EndWrite;
|
|
end;
|
|
end;
|
|
|
|
function TMycDataEndpoint<T>.Update(var Value: TSeries<T>): Boolean;
|
|
begin
|
|
FLock.BeginWrite;
|
|
try
|
|
Result := FChanged.Reset;
|
|
if Result then
|
|
begin
|
|
if FCount > 0 then
|
|
begin
|
|
var item: PItem;
|
|
var Arr: TArray<T>;
|
|
|
|
SetLength(Arr, FCount);
|
|
while FFirst <> nil do
|
|
begin
|
|
item := FFirst;
|
|
FFirst := item.Next;
|
|
dec(FCount);
|
|
Assert(FCount >= 0);
|
|
Arr[FCount] := item.Value;
|
|
Dispose(item);
|
|
end;
|
|
Assert(FCount = 0);
|
|
|
|
Value.Add(Arr, FLookback);
|
|
end;
|
|
end;
|
|
finally
|
|
FLock.EndWrite;
|
|
end;
|
|
end;
|
|
|
|
{ TTickAggregation }
|
|
|
|
constructor TTickAggregation.Create(const ATimeframe: TTimeframe);
|
|
begin
|
|
inherited Create;
|
|
FTimeframe := ATimeframe;
|
|
end;
|
|
|
|
function TTickAggregation.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 TTickAggregation.GetCurrentBar: TDataPoint<TOhlcItem>;
|
|
begin
|
|
Result := FCurrentBar;
|
|
end;
|
|
|
|
function TTickAggregation.GetTimeframe: TTimeframe;
|
|
begin
|
|
Result := FTimeframe;
|
|
end;
|
|
|
|
function TTickAggregation.ProcessData(const Value: TDataPoint<Double>): 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.Open := Value.Data;
|
|
FCurrentBar.Data.High := Value.Data;
|
|
FCurrentBar.Data.Low := Value.Data;
|
|
FCurrentBar.Data.Close := Value.Data;
|
|
FCurrentBar.Data.Volume := 1;
|
|
FCurrentBar.Time := barStartTime;
|
|
end
|
|
else
|
|
begin
|
|
// Update the currently aggregating bar
|
|
if Value.Data > FCurrentBar.Data.High then
|
|
FCurrentBar.Data.High := Value.Data;
|
|
if Value.Data < FCurrentBar.Data.Low then
|
|
FCurrentBar.Data.Low := Value.Data;
|
|
FCurrentBar.Data.Close := Value.Data;
|
|
// Volume is the number of ticks needed to build the complete bar.
|
|
FCurrentBar.Data.Volume := FCurrentBar.Data.Volume + 1;
|
|
end;
|
|
end;
|
|
|
|
{ TMycSequence<T> }
|
|
|
|
constructor TMycSequence<T>.Create(ACount: Integer);
|
|
begin
|
|
inherited Create;
|
|
|
|
SetLength(FDataProviders, ACount);
|
|
for var i := 0 to High(FDataProviders) do
|
|
FDataProviders[i] := TMycContainedDataProvider<T>.Create(Self);
|
|
end;
|
|
|
|
destructor TMycSequence<T>.Destroy;
|
|
begin
|
|
for var i := High(FDataProviders) downto 0 do
|
|
FDataProviders[i].Free;
|
|
inherited;
|
|
end;
|
|
|
|
function TMycSequence<T>.GetCount: Integer;
|
|
begin
|
|
Result := Length(FDataProviders);
|
|
end;
|
|
|
|
function TMycSequence<T>.GetDataProvider(Idx: Integer): TDataProvider<T>;
|
|
begin
|
|
Result := FDataProviders[Idx];
|
|
end;
|
|
|
|
function TMycSequence<T>.ProcessData(const Value: T): TState;
|
|
begin
|
|
Result := ProcessDataProvider(0, Value);
|
|
end;
|
|
|
|
function TMycSequence<T>.ProcessDataProvider(Idx: Integer; const Value: T): TState;
|
|
begin
|
|
if Idx >= Length(FDataProviders) then
|
|
exit;
|
|
|
|
Result :=
|
|
TaskManager
|
|
.RunTask(FDataProviders[idx].Broadcast(Value), function: TState begin Result := ProcessDataProvider(1 + idx, Value); end);
|
|
end;
|
|
|
|
function TMycIdentityConverter<T>.ProcessData(const Value: T): TState;
|
|
begin
|
|
Result := Broadcast(Value);
|
|
end;
|
|
|
|
{ TMycGenericParallelConverter<S, T> }
|
|
|
|
constructor TMycGenericParallelConverter<S, T>.Create(const AFunc: TConstFunc<S, T>);
|
|
begin
|
|
inherited Create;
|
|
FFunc := AFunc;
|
|
end;
|
|
|
|
function TMycGenericParallelConverter<S, T>.ProcessData(const Value: S): TState;
|
|
begin
|
|
var cValue := Value;
|
|
Result := TaskManager.RunTask(FQueue, function: TState begin Result := Broadcast(FFunc(cValue)); end);
|
|
|
|
FQueue := Result;
|
|
end;
|
|
|
|
function TMycParallelConverter<T>.ProcessData(const Value: T): TState;
|
|
begin
|
|
var cValue := Value;
|
|
Result := TaskManager.RunTask(FQueue, function: TState begin Result := Broadcast(cValue); end);
|
|
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;
|
|
|
|
{ TMycDataJoin<T> }
|
|
|
|
constructor TMycDataJoin<T>.Create(const ADataProviders: TArray<TDataProvider<T>>);
|
|
begin
|
|
inherited Create;
|
|
|
|
SetLength(FReceivers, Length(ADataProviders));
|
|
|
|
FLock := TSpinLock.Create(false);
|
|
FCount := Length(FReceivers);
|
|
|
|
FSender := TMycContainedDataProvider<TArray<T>>.Create(Self);
|
|
|
|
var cFunc :=
|
|
function(Idx: Integer): TConstFunc<T, TState>
|
|
begin
|
|
Result := function(const Value: T): TState begin ProcessData(Idx, Value); end
|
|
end;
|
|
|
|
for var i := 0 to High(FReceivers) do
|
|
with FReceivers[i] do
|
|
begin
|
|
Queue := TQueue<T>.Create;
|
|
DataProvider := ADataProviders[i];
|
|
Processor := TMycContainedProcessor<T>.Create(Self, cFunc(i));
|
|
Tag := DataProvider.Link(Processor);
|
|
end;
|
|
end;
|
|
|
|
destructor TMycDataJoin<T>.Destroy;
|
|
begin
|
|
for var i := High(FReceivers) downto 0 do
|
|
with FReceivers[i] do
|
|
begin
|
|
DataProvider.Unlink(FReceivers[i].Tag);
|
|
Processor.Free;
|
|
Queue.Free;
|
|
end;
|
|
|
|
FSender.Free;
|
|
inherited;
|
|
end;
|
|
|
|
function TMycDataJoin<T>.ProcessData(Idx: Integer; const Value: T): TState;
|
|
begin
|
|
var Arr: TArray<T>;
|
|
|
|
FLock.Enter;
|
|
try
|
|
if FReceivers[Idx].Queue.Count = 0 then
|
|
dec(FCount);
|
|
|
|
if FCount = 0 then
|
|
begin
|
|
SetLength(Arr, Length(FReceivers));
|
|
Arr[Idx] := Value;
|
|
FCount := Length(FReceivers);
|
|
for var i := 0 to High(FReceivers) do
|
|
if i <> Idx then
|
|
begin
|
|
Arr[i] := FReceivers[i].Queue.Dequeue;
|
|
if FReceivers[i].Queue.Count > 0 then
|
|
dec(FCount);
|
|
end;
|
|
end
|
|
else
|
|
FReceivers[Idx].Queue.Enqueue(Value);
|
|
finally
|
|
FLock.Exit;
|
|
end;
|
|
|
|
if Arr <> nil then
|
|
FSender.Broadcast(Arr);
|
|
end;
|
|
|
|
constructor TMycGenericAggregator<S, T>.Create(const AFunc: TConstFunc<S, TConverter<S, T>.TBroadcastProc, TState>);
|
|
begin
|
|
inherited Create;
|
|
FFunc := AFunc;
|
|
end;
|
|
|
|
function TMycGenericAggregator<S, T>.ProcessData(const Value: S): TState;
|
|
begin
|
|
Result := FFunc(Value, function(const Value: T): TState begin Result := FSender.Broadcast(Value); end);
|
|
end;
|
|
|
|
constructor TMycComposedConverter<S, T>.Create(const AProcessor: IMycProcessor<S>; const ADataProvider: TDataProvider<T>);
|
|
begin
|
|
inherited Create;
|
|
FProcessor := AProcessor;
|
|
FDataProvider := ADataProvider;
|
|
end;
|
|
|
|
function TMycComposedConverter<S, T>.GetSender: TDataProvider<T>.IDataProvider;
|
|
begin
|
|
Result := FDataProvider;
|
|
end;
|
|
|
|
function TMycComposedConverter<S, T>.ProcessData(const Value: S): TState;
|
|
begin
|
|
Result := FProcessor.ProcessData(Value);
|
|
end;
|
|
|
|
end.
|