MycLib/Src/Myc.Trade.DataPoint.Impl.pas

unit Myc.Trade.DataPoint.Impl;

interface

uses
    System.SysUtils,
    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 ProcessData(const Value: T): TState; virtual; abstract;
    end;

    // Concrete data provider that manages a list of processors (listeners).
    TMycDataProvider<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;

    // 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: TMycDataProvider<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;

    // 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>)
        type
            TProc = reference to function(const Value: T): TState;
    private
        FProc: TProc;
    protected
        function ProcessData(const Value: T): TState; override; final;
    public
        constructor Create(const AProc: TProc);
    end;

    // A processor implementation that is owned by a controller.
    TMycContainedProcessor<T> = class(TContainedObject, IMycProcessor<T>)
        type
            TProc = function(const Value: T): TState of object;
    private
        FProc: TProc;
        function ProcessData(const Value: T): TState;
    public
        constructor Create(const Controller: IInterface; const AProc: TProc);
    end;

    // Endpoint that collects data into a series.
    TMycDataEndpoint<T> = class(TInterfacedObject, TMutable<TSeries<T>>.IMutable)
    private
        FProcessor: TMycContainedProcessor<T>;
        FTag: TDataProvider<T>.TTag;
        FDataProvider: TDataProvider<T>;
        FLookback: Int64;
        FData: TSeries<T>;
        FChanged: TEvent;
        function GetChanged: TSignal;
        function GetValue: TSeries<T>;
        function ProcessData(const Value: T): TState;
    public
        constructor Create(const ADataProvider: TDataProvider<T>; ALookback: Int64);
        destructor Destroy; override;
    end;

implementation

uses
    System.TypInfo,
    System.RTTI;

{ TMycDataProvider<T> }

constructor TMycDataProvider<T>.Create(const Controller: IInterface);
begin
    inherited Create(Controller);
end;

destructor TMycDataProvider<T>.Destroy;
begin
    FListeners.Finalize;
    inherited Destroy;
end;

function TMycDataProvider<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 TMycDataProvider<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 TMycDataProvider<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 := TMycDataProvider<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 := Broadcast(Calculate(Value));
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;
    var name := '';
    if AFieldName = 'Time' then
        for var i := 0 to High(Fields) do
        begin
            name := name + ' ' + Fields[i].Name;
        end;

    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: TProc);
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: TProc);
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: TSignal;
begin
    Result := FChanged.Signal;
end;

function TMycDataEndpoint<T>.GetValue: TSeries<T>;
begin
    Result := FData;
end;

function TMycDataEndpoint<T>.ProcessData(const Value: T): TState;
begin
    Result := TState.Null;

    // Add new data point, respecting the lookback period.
    FData := FData.Add(Value, FLookback);
    FChanged.Notify;
end;

end.