Unit refactoring

Src/Myc.Trade.DataPoint.Impl.pas

@@ -0,0 +1,196 @@
+unit Myc.Trade.DataPoint.Impl;
+
+interface
+
+uses
+    Myc.Signals,
+    Myc.Trade.Types,
+    Myc.Trade.DataPoint;
+
+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.