MycLib/Src/Myc.Trade.DataPoint.pas

unit Myc.Trade.DataPoint;

interface

uses
    Myc.Signals,
    Myc.Mutable,
    Myc.Trade.Types,
    Myc.Trade.DataArray,
    Myc.DataRecord;

type
    // A generic interface for components that process data of type T.
    IMycProcessor<T> = interface
        function ProcessData(const Value: T): TState;
    end;

    // Interface helper for IDataProvider providing the null object pattern.
    TDataProvider<T> = record
    public
        type
            TTag = Pointer;
            IDataProvider = interface
                function Link(const Receiver: IMycProcessor<T>): TTag;
                procedure Unlink(Tag: TTag);
            end;

    strict private
        class var
            FNull: IDataProvider;
        class constructor CreateClass;

    private
        FDataProvider: IDataProvider;

    public
        constructor Create(const ADataProvider: IDataProvider);

        // Managed record operators
        class operator Initialize(out Dest: TDataProvider<T>);
        class operator Implicit(const A: IDataProvider): TDataProvider<T>; overload;
        class operator Implicit(const A: TDataProvider<T>): IDataProvider; overload;

        // Wrapper for IMycDataProvider methods
        function Link(const Receiver: IMycProcessor<T>): TTag; inline;
        procedure Unlink(Tag: TTag); inline;

        // Provides access to the null object instance.
        class property Null: IDataProvider read FNull;
    end;

    IMycDataSequence<T> = interface(IMycProcessor<T>)
        function GetCount: Integer;
        function GetDataProvider(Idx: Integer): TDataProvider<T>;
        property Count: Integer read GetCount;
        property DataProvider[Idx: Integer]: TDataProvider<T> read GetDataProvider; default;
    end;

    // Interface helper for IConverter<S,T> providing the null object pattern.
    TConverter<S, T> = record
    public
        type
            IConverter = interface(IMycProcessor<S>)
                {$region 'private'}
                function GetSender: TDataProvider<T>.IDataProvider;
                {$endregion}
                property Sender: TDataProvider<T>.IDataProvider read GetSender;
            end;

            TBroadcastProc = reference to function(const Value: T): TState;

    strict private
        class var
            FNull: IConverter;
        class constructor CreateClass;
    private
        FConverter: IConverter;
        function GetSender: TDataProvider<T>; inline;
    public
        constructor Create(const AConverter: IConverter);

        // Managed record operators
        class operator Initialize(out Dest: TConverter<S, T>);
        class operator Implicit(const A: IConverter): TConverter<S, T>; overload;
        class operator Implicit(const A: TConverter<S, T>): IConverter; overload;

        class function Construct(const Processor: IMycProcessor<S>; const DataProvider: TDataProvider<T>): TConverter<S, T>; static;
        class function CreateGeneric(const Func: TConstFunc<S, T>): TConverter<S, T>; static;
        class function CreateAggregation(const Func: TConstFunc<S, TBroadcastProc, TState>): TConverter<S, T>; static;
        class function CreateParallel(const Func: TConstFunc<S, T>): TConverter<S, T>; static;

        function Chain<R>(const Next: TConverter<T, R>): TConverter<T, R>; overload; inline;
        function Chain<R>(const Func: TConstFunc<T, R>): TConverter<T, R>; overload; inline;
        function ChainParallel<R>(const Func: TConstFunc<T, R>): TConverter<T, R>; overload; inline;

        function MakeParallel: TConverter<T, T>; overload; inline;

        // Extracts the field of a record by it's name (using RTTI).
        function Field<R>(const FieldName: String): TConverter<T, R>; overload; inline;

        function Sequence(Count: Integer): IMycDataSequence<T>; overload;
        function Sequence(const Items: TArray<IConverter>): IMycDataSequence<S>; overload;

        // Provides access to the null object instance.
        class property Null: IConverter read FNull;
        // Wrapper for IConverter.Sender
        property Sender: TDataProvider<T> read GetSender;
    end;

    // Factory for creating specific converter instances.
    TConverter = record
        class function CreateEndpoint<T>(const DataProvider: TDataProvider<T>; Lookback: Int64): TLazy<TSeries<T>>; static;

        class function CreateCounter<T>: TConverter<T, Int64>; static;
        class function CreateTicker<T>: TConverter<TArray<T>, T>; static;
        class function CreateRecordField<S, T>(const FieldName: String): TConverter<S, T>; static;
        class function CreateIdentity<T>: TConverter<T, T>; static;

        class function CreateTickAggregation(Timeframe: TTimeframe): TConverter<TDataPoint<Double>, TDataPoint<TOhlcItem>>; static;
        class function CreateOhlcAggregation(Timeframe: TTimeframe): TConverter<TDataPoint<TOhlcItem>, TDataPoint<TOhlcItem>>; static;

        class function CreateDataPointConverter<S, T>(const Func: TConstFunc<S, T>): TConverter<TDataPoint<S>, TDataPoint<T>>; static;

        class function CreateSequence<T>(Count: Integer; const Parent: TDataProvider<T>): TArray<TConverter<T, T>>; overload; static;

        class function Parallel<T>(Parent: TDataProvider<T>): TConverter<T, T>; static;

        class function FieldToRecord<T>(const Layout: TDataRecord.TLayout; const Name: String): TConverter<T, TDataRecord>; static;
        class function FieldOfRecord<T>(const Layout: TDataRecord.TLayout; const Name: String): TConverter<TDataRecord, T>; static;

        class function Join<T>(const DataProviders: TArray<TDataProvider<T>>): TDataProvider<TArray<T>>; static;

        type
            TRecordMapping = record
                Layout: TDataRecord.TLayout;
                Fields: array of record
                FromField, ToField: String
                end;
            end;

        class function DataMapping(
            const Inputs: TArray<TRecordMapping>;
            const Output: TDataRecord.TLayout
        ): TConverter<TArray<TDataRecord>, TDataRecord>; static;

        class function JoinRecords(
            const TargetLayout: TDataRecord.TLayout;
            const Mapping: TArray<TRecordMapping>;
            const DataProviders: TArray<TDataProvider<TDataRecord>>
        ): TConverter<TArray<TDataRecord>, TDataRecord>; static;
    end;

implementation

uses
    Myc.Trade.DataPoint.Impl;

{ TDataProvider<T> }

class constructor TDataProvider<T>.CreateClass;
begin
    // Create the singleton null object instance.
    FNull := TNullDataProvider<T>.Create;
end;

constructor TDataProvider<T>.Create(const ADataProvider: IDataProvider);
begin
    FDataProvider := ADataProvider;
    if not Assigned(FDataProvider) then
        FDataProvider := FNull;
end;

class operator TDataProvider<T>.Initialize(out Dest: TDataProvider<T>);
begin
    Dest.FDataProvider := FNull;
end;

class operator TDataProvider<T>.Implicit(const A: IDataProvider): TDataProvider<T>;
begin
    Result.Create(A);
end;

class operator TDataProvider<T>.Implicit(const A: TDataProvider<T>): IDataProvider;
begin
    Result := A.FDataProvider;
end;

function TDataProvider<T>.Link(const Receiver: IMycProcessor<T>): TTag;
begin
    Result := FDataProvider.Link(Receiver);
end;

procedure TDataProvider<T>.Unlink(Tag: TTag);
begin
    FDataProvider.Unlink(Tag);
end;

{ TConverter<S, T> }

class constructor TConverter<S, T>.CreateClass;
begin
    FNull := TNullConverter<S, T>.Create;
end;

constructor TConverter<S, T>.Create(const AConverter: IConverter);
begin
    FConverter := AConverter;
    if not Assigned(FConverter) then
        FConverter := FNull;
end;

function TConverter<S, T>.Chain<R>(const Next: TConverter<T, R>): TConverter<T, R>;
begin
    FConverter.Sender.Link(Next);
    Result := Next;
end;

function TConverter<S, T>.Chain<R>(const Func: TConstFunc<T, R>): TConverter<T, R>;
begin
    Result := Chain<R>(TMycGenericConverter<T, R>.Create(Func));
end;

function TConverter<S, T>.ChainParallel<R>(const Func: TConstFunc<T, R>): TConverter<T, R>;
begin
    Result := Chain<R>(TMycGenericParallelConverter<T, R>.Create(Func));
end;

class function TConverter<S, T>.Construct(const Processor: IMycProcessor<S>; const DataProvider: TDataProvider<T>): TConverter<S, T>;
begin
    Result := TMycComposedConverter<S, T>.Create(Processor, DataProvider);
end;

class function TConverter<S, T>.CreateAggregation(const Func: TConstFunc<S, TBroadcastProc, TState>): TConverter<S, T>;
begin
    Result := TMycGenericAggregator<S, T>.Create(Func);
end;

class function TConverter<S, T>.CreateGeneric(const Func: TConstFunc<S, T>): TConverter<S, T>;
begin
    Result := TMycGenericConverter<S, T>.Create(Func);
end;

class function TConverter<S, T>.CreateParallel(const Func: TConstFunc<S, T>): TConverter<S, T>;
begin
    Result := TMycGenericParallelConverter<S, T>.Create(Func);
end;

function TConverter<S, T>.Field<R>(const FieldName: String): TConverter<T, R>;
begin
    Result := Chain<R>(TConverter.CreateRecordField<T, R>(FieldName));
end;

function TConverter<S, T>.GetSender: TDataProvider<T>;
begin
    Result := FConverter.Sender;
end;

function TConverter<S, T>.MakeParallel: TConverter<T, T>;
begin
    Result := Chain<T>(TMycGenericParallelConverter<T, T>.Create(function(const Value: T): T begin Result := Value; end));
end;

function TConverter<S, T>.Sequence(Count: Integer): IMycDataSequence<T>;
begin
    Result := TMycSequence<T>.Create(Count);
    FConverter.Sender.Link(Result);
end;

function TConverter<S, T>.Sequence(const Items: TArray<IConverter>): IMycDataSequence<S>;
begin
    var seq: IMycDataSequence<S> := TMycSequence<S>.Create(Length(Items));

    for var i := 0 to High(Items) do
        seq[i].Link(Items[i]);

    Result := seq;
end;

class operator TConverter<S, T>.Initialize(out Dest: TConverter<S, T>);
begin
    Dest.FConverter := FNull;
end;

class operator TConverter<S, T>.Implicit(const A: IConverter): TConverter<S, T>;
begin
    Result.Create(A);
end;

class operator TConverter<S, T>.Implicit(const A: TConverter<S, T>): IConverter;
begin
    Result := A.FConverter;
end;

class function TConverter.CreateTickAggregation(Timeframe: TTimeframe): TConverter<TDataPoint<Double>, TDataPoint<TOhlcItem>>;
begin
    Result := TTickAggregation.Create(Timeframe);
end;

{ TConverter }

class function TConverter.CreateCounter<T>: TConverter<T, Int64>;
begin
    Result := TMycDataCounter<T>.Create;
end;

class function TConverter.CreateDataPointConverter<S, T>(const Func: TConstFunc<S, T>): TConverter<TDataPoint<S>, TDataPoint<T>>;
begin
    var cFunc: TConstFunc<S, T> := Func;
    Result :=
        TMycGenericConverter<TDataPoint<S>, TDataPoint<T>>.Create(
            function(const Value: TDataPoint<S>): TDataPoint<T>
            begin
                Result.Time := Value.Time;
                Result.Data := cFunc(Value.Data);
            end
        );
end;

class function TConverter.CreateEndpoint<T>(const DataProvider: TDataProvider<T>; Lookback: Int64): TLazy<TSeries<T>>;
begin
    Result := TMycDataEndpoint<T>.Create(DataProvider, Lookback);
end;

class function TConverter.CreateIdentity<T>: TConverter<T, T>;
begin
    Result := TMycIdentityConverter<T>.Create;
end;

class function TConverter.CreateRecordField<S, T>(const FieldName: String): TConverter<S, T>;
begin
    Result := TMycRecordFieldReader<S, T>.Create(FieldName);
end;

class function TConverter.CreateTicker<T>: TConverter<TArray<T>, T>;
begin
    Result := TMycTicker<T>.Create;
end;

class function TConverter.CreateSequence<T>(Count: Integer; const Parent: TDataProvider<T>): TArray<TConverter<T, T>>;
begin
    var seq: IMycDataSequence<T> := TMycSequence<T>.Create(Count);

    SetLength(Result, Count);
    for var i := 0 to High(Result) do
    begin
        Result[i] := TMycIdentityConverter<T>.Create;
        seq[i].Link(Result[i]);
    end;
    Parent.Link(seq);
end;

class function TConverter.CreateOhlcAggregation(Timeframe: TTimeframe): TConverter<TDataPoint<TOhlcItem>, TDataPoint<TOhlcItem>>;
begin
    Result := TOhlcAggregation.Create(Timeframe);
end;

class function TConverter.DataMapping(
    const Inputs: TArray<TRecordMapping>;
    const Output: TDataRecord.TLayout
): TConverter<TArray<TDataRecord>, TDataRecord>;
var
    Idxs: array of array of record
    FromIdx, ToIdx: Integer
    end;
begin
    SetLength(Idxs, Length(Inputs));
    for var i := 0 to High(Idxs) do
    begin
        SetLength(Idxs[i], Length(Inputs[i].Fields));
        for var j := 0 to High(Idxs[i]) do
        begin
            Idxs[i][j].FromIdx := Inputs[i].Layout.IndexOf(Inputs[i].Fields[j].FromField);
            Idxs[i][j].ToIdx := Output.IndexOf(Inputs[i].Fields[j].ToField);
        end;
    end;

    Result :=
        TConverter<TArray<TDataRecord>, TDataRecord>.CreateGeneric(
            function(const Inputs: TArray<TDataRecord>): TDataRecord
            begin
                Result := TDataRecord.Create(Output);
                for var i := 0 to High(Idxs) do
                    for var j := 0 to High(Idxs[i]) do
                        Inputs[i].CopyField(Idxs[i][j].FromIdx, Result, Idxs[i][j].ToIdx);
            end
        );
end;

class function TConverter.FieldOfRecord<T>(const Layout: TDataRecord.TLayout; const Name: String): TConverter<TDataRecord, T>;
begin
    var idx := Layout.IndexOf(Name);
    Result := TConverter<TDataRecord, T>.CreateGeneric(function(const Value: TDataRecord): T begin Value.GetValue(idx, Result); end);
end;

class function TConverter.FieldToRecord<T>(const Layout: TDataRecord.TLayout; const Name: String): TConverter<T, TDataRecord>;
begin
    var idx := Layout.IndexOf(Name);
    Result :=
        TConverter<T, TDataRecord>.CreateGeneric(
            function(const Value: T): TDataRecord
            begin
                Result := TDataRecord.Create(Layout);
                Result.SetValue(idx, Value);
            end
        );
end;

class function TConverter.Join<T>(const DataProviders: TArray<TDataProvider<T>>): TDataProvider<TArray<T>>;
begin
    Result := TMycDataJoin<T>.Create(DataProviders);
end;

class function TConverter.JoinRecords(
    const TargetLayout: TDataRecord.TLayout;
    const Mapping: TArray<TRecordMapping>;
    const DataProviders: TArray<TDataProvider<TDataRecord>>
): TConverter<TArray<TDataRecord>, TDataRecord>;
begin
    var RecProvider := Join<TDataRecord>(DataProviders);

    Result := DataMapping(Mapping, TargetLayout);

    RecProvider.Link(Result);
end;

class function TConverter.Parallel<T>(Parent: TDataProvider<T>): TConverter<T, T>;
begin
    Result := TMycParallelConverter<T>.Create;
    Parent.Link(Result);
end;

end.