MycLib/Src/Myc.Trade.Core.DataConverter.pas

unit Myc.Trade.Core.DataConverter;

interface

uses
    Myc.Signals,
    Myc.Trade.Types,
    Myc.Trade.DataPoint,
    Myc.Trade.DataConverter;

type
    // Null object implementation for IMycConverter
    TNullConverter<S, T> = class(TInterfacedObject, TConverter<S, T>.IConverter)
    private
        function GetSender: TDataProvider<T>.IDataProvider;
    public
        function ProcessData(const Value: S): TState;
    end;

    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;

    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;

    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;

    TMycDataCounter<T> = class(TMycConverter<T, Int64>)
    private
        FCount: Int64;
    protected
        function ProcessData(const Value: T): TState; override;
    public
        constructor Create;
    end;

    TMycTicker<T> = class(TMycConverter<TArray<T>, T>)
    public
        function ProcessData(const Values: TArray<T>): TState; override;
    end;

    TMycRecordFieldReader<S, T> = class(TMycConverter<S, T>)
    private
        FOffset: Integer;

    public
        constructor Create(const AFieldName: String);
        function ProcessData(const Values: S): TState; override;
    end;

implementation

uses
    System.TypInfo,
    System.SysUtils,
    System.RTTI;

{ 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 }

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;

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;

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;

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;

end.