unit Myc.DataFlow.Impl;

interface

uses
    System.SysUtils,
    System.SyncObjs,
    System.Generics.Collections,
    Myc.Signals,
    Myc.Mutable,
    Myc.Core.Notifier,
    Myc.DataFlow,
    Myc.Trade.Types,
    Myc.Trade.DataArray;

type
    // Abstract base class for data consumers.
    TMycConsumer<T> = class abstract(TContainedObject, IConsumer<T>)
    strict private
        type
            TNull = class(TInterfacedObject, IConsumer<T>)
            protected
                function Consume(const Value: T): TState;
            end;
        class var
            FNull: IConsumer<T>;
        class constructor CreateClass;
    protected
        function Consume(const Value: T): TState; virtual; abstract;
    public
        class property Null: IConsumer<T> read FNull;
    end;

    // A consumer implementation that is owned by a controller.
    TMycGenericConsumer<T> = class(TMycConsumer<T>)
    private
        FProc: TConstFunc<T, TState>;
    protected
        function Consume(const Value: T): TState; override; final;
    public
        constructor Create(const Controller: IInterface; const AProc: TConstFunc<T, TState>);
    end;

    TMycProducer<T> = class(TInterfacedObject, IProducer<T>)
    strict private
        type
            TNull = class(TInterfacedObject, IProducer<T>)
            public
                function Link(const Consumer: IConsumer<T>): TTag;
                procedure Unlink(Tag: TTag);
            end;
        class var
            FNull: IProducer<T>;
        class constructor CreateClass;
    private
        FListeners: TMycNotifyList<IConsumer<T>>;
    public
        constructor Create;
        destructor Destroy; override;
        function Broadcast(const Value: T): TState;
        function Link(const Consumer: IConsumer<T>): TTag;
        procedure Unlink(Tag: TTag);
        class property Null: IProducer<T> read FNull;
    end;

    // Concrete producer that manages a list of consumers (listeners).
    TMycContainedProducer<T> = class(TContainedObject, IProducer<T>)
    private
        FListeners: TMycNotifyList<IConsumer<T>>;
    public
        constructor Create(const Controller: IInterface);
        destructor Destroy; override;
        function Broadcast(const Value: T): TState;
        function Link(const Consumer: IConsumer<T>): TTag;
        procedure Unlink(Tag: TTag);
    end;

    // Abstract base class for components that now act as a producer and contain a consumer.
    TMycConverter<S, T> = class abstract(TMycProducer<T>, IConverter<S, T>)
    strict private
        type
            TNull = class(TInterfacedObject, IConverter<S, T>)
            private
                function GetConsumer: IConsumer<S>;
                function Link(const Consumer: IConsumer<T>): TTag;
                procedure Unlink(Tag: TTag);
            end;
        class var
            FNull: IConverter<S, T>;
        class constructor CreateClass;
    private
        FConsumer: TMycGenericConsumer<S>;
        function GetConsumer: IConsumer<S>;
    protected
        // To be implemented by descendants to perform the actual conversion.
        function Consume(const Value: S): TState; virtual; abstract;
    public
        constructor Create;
        destructor Destroy; override;

        class property Null: IConverter<S, T> read FNull;
    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 Consume(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 Consume(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 Consume(const Value: S): TState; override;
    public
        constructor Create(const AFunc: TConstFunc<S, T>);
    end;

    TMycIdentityConverter<T> = class(TMycConverter<T, T>)
    protected
        function Consume(const Value: T): TState; override; final;
    end;

    // A converter that counts incoming data points and outputs the current count.
    TMycDataCounter<T> = class(TMycConverter<T, Int64>)
    private
        FCount: Int64;
    protected
        function Consume(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>)
    protected
        function Consume(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;
    protected
        function Consume(const Values: S): TState; override;
    public
        constructor Create(const AFieldName: String);
    end;

    // Endpoint that collects data into a series.
    TMycDataEndpoint<T> = class(TInterfacedObject, TLazy<TSeries<T>>.ILazy)
    private
        type
            PItem = ^TItem;
            TItem = record
                Next: PItem;
                Value: T;
            end;
    private
        FConsumer: TMycGenericConsumer<T>;
        FLookback: Integer;
        FChanged: TFlag;
        FLock: TLightweightMREW;
        FFirst: PItem;
        FCount: Integer;
        function GetChanged: TState;
        function Consume(const Value: T): TState;
    public
        constructor Create(ALookback: Integer);
        destructor Destroy; override;
        function Update(var Value: TSeries<T>): Boolean;

        class function CreateDataEndpoint(Lookback: Integer; out Series: TLazy<TSeries<T>>): IConsumer<T>;
    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;
    protected
        function Consume(const Value: TDataPoint<Double>): TState; override;
    public
        constructor Create(const ATimeframe: TTimeframe);
        property CurrentBar: TDataPoint<TOhlcItem> read GetCurrentBar;
        property Timeframe: TTimeframe read GetTimeframe;
    end;

    TMycParallelConverter<T> = class(TMycConverter<T, T>)
    private
        FQueue: TState;
    protected
        function Consume(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;
    protected
        function Consume(const Value: TDataPoint<TOhlcItem>): TState; override;
    public
        constructor Create(const ATimeframe: TTimeframe);
        property CurrentBar: TDataPoint<TOhlcItem> read GetCurrentBar;
        property Timeframe: TTimeframe read GetTimeframe;
    end;

    // Endpoint that collects data into a series.
    TMycDataJoin<T> = class(TInterfacedObject, IProducer<TArray<T>>)
    private
        FConsumers: array of record
            Consumer: TMycGenericConsumer<T>;
            Queue: TQueue<T>;
        end;

        FContainedProvider: TMycContainedProducer<TArray<T>>;
        FLock: TSpinLock;
        FCount: Integer;
        function Consume(Idx: Integer; const Value: T): TState;
        function GetConsumers(Idx: Integer): IConsumer<T>;
        function Link(const Consumer: IConsumer<TArray<T>>): TTag;
        procedure Unlink(Tag: TTag);
    public
        constructor Create(ACount: Integer);
        destructor Destroy; override;
        property Consumers[Idx: Integer]: IConsumer<T> read GetConsumers;
    end;

    TMycComposedConverter<S, T> = class(TInterfacedObject, IConverter<S, T>)
    private
        FConsumer: IConsumer<S>;
        FProducer: TProducer<T>;
    protected
        function GetConsumer: IConsumer<S>;
        function Link(const Consumer: IConsumer<T>): TTag;
        procedure Unlink(Tag: TTag);
    public
        constructor Create(const AConsumer: IConsumer<S>; const AProducer: TProducer<T>);
    end;

    TMycSequence<T> = class(TInterfacedObject, IConsumer<T>)
    private
        FProducers: TArray<TMycContainedProducer<T>>;
    protected
        function Consume(const Value: T): TState;
        function ProcessProducer(Idx: Integer; const Value: T): TState;
    public
        constructor Create(ACount: Integer);
        destructor Destroy; override;

        class function CreateSequence(Count: Integer; out Producers: TArray<IProducer<T>>): IConsumer<T>; static;
    end;

implementation

uses
    System.TypInfo,
    System.RTTI,
    System.DateUtils,
    System.Math,
    Winapi.Windows,
    Myc.TaskManager;

class constructor TMycConsumer<T>.CreateClass;
begin
    FNull := TNull.Create;
end;

{ TMycContainedProducer<T> }

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

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

function TMycContainedProducer<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.Consume(Value).Signal.Subscribe(done);
            item := item.Prev;
        end;

        Result := done.State;
    finally
        FListeners.Release;
    end;
end;

function TMycContainedProducer<T>.Link(const Consumer: IConsumer<T>): TTag;
begin
    // Add the Consumer to the notification list
    FListeners.Lock;
    try
        Result := FListeners.Advise(Consumer);
    finally
        FListeners.Release;
    end;
end;

procedure TMycContainedProducer<T>.Unlink(Tag: TTag);
begin
    FListeners.Lock;
    try
        FListeners.Unadvise(Tag);
    finally
        FListeners.Release;
    end;
end;

{ TMycConverter<S, T> }

constructor TMycConverter<S, T>.Create;
begin
    inherited Create;
    FConsumer := TMycGenericConsumer<S>.Create(Self, Consume);
end;

{ TMycConverter<S, T> }

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

destructor TMycConverter<S, T>.Destroy;
begin
    FConsumer.Free;
    inherited Destroy;
end;

function TMycConverter<S, T>.GetConsumer: IConsumer<S>;
begin
    Result := FConsumer;
end;

{ TMycConverter<S, T>.TNull }

function TMycConverter<S, T>.TNull.GetConsumer: IConsumer<S>;
begin
    Result := TMycConsumer<S>.Null;
end;

function TMycConverter<S, T>.TNull.Link(const Consumer: IConsumer<T>): TTag;
begin
    Result := nil;
end;

procedure TMycConverter<S, T>.TNull.Unlink(Tag: TTag);
begin
    // NOP
end;

{ TMycGenericConverter<S, T> }

constructor TMycGenericConverter<S, T>.Create(const AFunc: TConstFunc<S, T>);
begin
    inherited Create;
    FFunc := AFunc;
end;

function TMycGenericConverter<S, T>.Consume(const Value: S): TState;
begin
    Result := Broadcast(FFunc(Value));
end;

{ TMycDataCounter<T> }

constructor TMycDataCounter<T>.Create;
begin
    inherited Create;
    FCount := 0;
end;

function TMycDataCounter<T>.Consume(const Value: T): TState;
begin
    Result := Broadcast(FCount);
    inc(FCount);
end;

{ TMycTicker<T> }

function TMycTicker<T>.Consume(const Values: TArray<T>): TState;
begin
    var done := TLatch.CreateLatch(Length(Values));

    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));

    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>.Consume(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;

{ TMycGenericConsumer<T> }

constructor TMycGenericConsumer<T>.Create(const Controller: IInterface; const AProc: TConstFunc<T, TState>);
begin
    inherited Create(Controller);
    FProc := AProc;
end;

function TMycGenericConsumer<T>.Consume(const Value: T): TState;
begin
    Result := FProc(Value);
end;

{ TMycDataEndpoint<T> }

constructor TMycDataEndpoint<T>.Create(ALookback: Integer);
begin
    inherited Create;
    FLookback := ALookback;
    FConsumer := TMycGenericConsumer<T>.Create(Self, Consume);
end;

destructor TMycDataEndpoint<T>.Destroy;
begin
    FConsumer.Free;
    inherited;
end;

function TMycDataEndpoint<T>.GetChanged: TState;
begin
    Result := FChanged.State
end;

function TMycDataEndpoint<T>.Consume(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;

class function TMycDataEndpoint<T>.CreateDataEndpoint(Lookback: Integer; out Series: TLazy<TSeries<T>>): IConsumer<T>;
var
    endPoint: TMycDataEndpoint<T>;
begin
    endPoint := TMycDataEndpoint<T>.Create(Lookback);
    Result := endPoint.FConsumer;
    Series := endPoint;
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
    // Implementation is unchanged
    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: Result := Floor(TimeStamp) - (Floor(TimeStamp) mod 2);
        D3: Result := Floor(TimeStamp) - (Floor(TimeStamp) mod 3);
        W: Result := TimeStamp.StartOfTheWeek;
        MN: Result := TimeStamp.StartOfTheMonth;
        MN3: Result := RecodeMonth(TimeStamp.StartOfTheMonth, (MonthOf(TimeStamp) - 1) div 3 * 3 + 1);
        MN6: Result := RecodeMonth(TimeStamp.StartOfTheMonth, (MonthOf(TimeStamp) - 1) div 6 * 6 + 1);
        Y: Result := TimeStamp.StartOfTheYear;
    else
        Result := 0;
    end;
end;

function TTickAggregation.GetCurrentBar: TDataPoint<TOhlcItem>;
begin
    Result := FCurrentBar;
end;

function TTickAggregation.GetTimeframe: TTimeframe;
begin
    Result := FTimeframe;
end;

function TTickAggregation.Consume(const Value: TDataPoint<Double>): TState;
var
    barStartTime: TDateTime;
    lastBarTime: TDateTime;
begin
    barStartTime := GetBarStartTime(Value.Time, FTimeframe);
    lastBarTime := FCurrentBar.Time;

    if (barStartTime > lastBarTime) then
    begin
        if (lastBarTime > 0) then
        begin
            Result := Broadcast(FCurrentBar);
        end;

        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
        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;
        FCurrentBar.Data.Volume := FCurrentBar.Data.Volume + 1;
    end;
end;

{ TMycSequence<T> }

constructor TMycSequence<T>.Create(ACount: Integer);
begin
    inherited Create;

    SetLength(FProducers, ACount);
    for var i := 0 to High(FProducers) do
        FProducers[i] := TMycContainedProducer<T>.Create(Self);
end;

destructor TMycSequence<T>.Destroy;
begin
    for var i := High(FProducers) downto 0 do
        FProducers[i].Free;
    inherited;
end;

class function TMycSequence<T>.CreateSequence(Count: Integer; out Producers: TArray<IProducer<T>>): IConsumer<T>;
var
    Seq: TMycSequence<T>;
begin
    Seq := TMycSequence<T>.Create(Count);
    Result := Seq;

    SetLength(Producers, Length(Seq.FProducers));
    for var i := 0 to High(Producers) do
        Producers[i] := Seq.FProducers[i];
end;

function TMycSequence<T>.Consume(const Value: T): TState;
begin
    Result := ProcessProducer(0, Value);
end;

function TMycSequence<T>.ProcessProducer(Idx: Integer; const Value: T): TState;
begin
    if Idx >= Length(FProducers) then
        exit;

    Result := TaskManager.RunTask(FProducers[idx].Broadcast(Value), function: TState begin Result := ProcessProducer(1 + idx, Value); end);
end;

function TMycIdentityConverter<T>.Consume(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>.Consume(const Value: S): TState;
begin
    var cValue := Value;
    Result := TaskManager.RunTask(FQueue, function: TState begin Result := Broadcast(FFunc(cValue)); end);

    FQueue := Result;
end;

{ TMycParallelConverter<T> }

function TMycParallelConverter<T>.Consume(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;
begin
    // Same implementation as TTickAggregation
    var 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: Result := Floor(TimeStamp) - (Floor(TimeStamp) mod 2);
        D3: Result := Floor(TimeStamp) - (Floor(TimeStamp) mod 3);
        W: Result := TimeStamp.StartOfTheWeek;
        MN: Result := TimeStamp.StartOfTheMonth;
        MN3: Result := RecodeMonth(TimeStamp.StartOfTheMonth, (MonthOf(TimeStamp) - 1) div 3 * 3 + 1);
        MN6: Result := RecodeMonth(TimeStamp.StartOfTheMonth, (MonthOf(TimeStamp) - 1) div 6 * 6 + 1);
        Y: Result := TimeStamp.StartOfTheYear;
    else
        Result := 0;
    end;
end;

function TOhlcAggregation.GetCurrentBar: TDataPoint<TOhlcItem>;
begin
    Result := FCurrentBar;
end;

function TOhlcAggregation.GetTimeframe: TTimeframe;
begin
    Result := FTimeframe;
end;

function TOhlcAggregation.Consume(const Value: TDataPoint<TOhlcItem>): TState;
var
    barStartTime: TDateTime;
    lastBarTime: TDateTime;
begin
    barStartTime := GetBarStartTime(Value.Time, FTimeframe);
    lastBarTime := FCurrentBar.Time;

    if (barStartTime > lastBarTime) then
    begin
        if (lastBarTime > 0) then
        begin
            Result := Broadcast(FCurrentBar);
        end;

        FCurrentBar.Data := Value.Data;
        FCurrentBar.Time := barStartTime;
    end
    else
    begin
        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;
        FCurrentBar.Data.Volume := FCurrentBar.Data.Volume + Value.Data.Volume;
    end;
end;

{ TMycDataJoin<T> }

constructor TMycDataJoin<T>.Create(ACount: Integer);
begin
    inherited Create;

    SetLength(FConsumers, ACount);

    FLock := TSpinLock.Create(false);
    FCount := Length(FConsumers);

    FContainedProvider := TMycContainedProducer<TArray<T>>.Create(Self);

    var cFunc :=
        function(Idx: Integer): TConstFunc<T, TState>
        begin
            Result := function(const Value: T): TState begin Result := Consume(Idx, Value); end
        end;

    for var i := 0 to High(FConsumers) do
        with FConsumers[i] do
        begin
            Queue := TQueue<T>.Create;
            Consumer := TMycGenericConsumer<T>.Create(Self, cFunc(i));
        end;
end;

destructor TMycDataJoin<T>.Destroy;
begin
    for var i := High(FConsumers) downto 0 do
        with FConsumers[i] do
        begin
            Consumer.Free;
            Queue.Free;
        end;

    FContainedProvider.Free;
    inherited;
end;

function TMycDataJoin<T>.Link(const Consumer: IConsumer<TArray<T>>): TTag;
begin
    Result := FContainedProvider.Link(Consumer);
end;

procedure TMycDataJoin<T>.Unlink(Tag: TTag);
begin
    FContainedProvider.Unlink(Tag);
end;

function TMycDataJoin<T>.Consume(Idx: Integer; const Value: T): TState;
begin
    var Arr: TArray<T>;

    FLock.Enter;
    try
        if FConsumers[Idx].Queue.Count = 0 then
            dec(FCount);

        if FCount = 0 then
        begin
            SetLength(Arr, Length(FConsumers));
            Arr[Idx] := Value;
            FCount := Length(FConsumers);
            for var i := 0 to High(FConsumers) do
                if i <> Idx then
                begin
                    Arr[i] := FConsumers[i].Queue.Dequeue;
                    if FConsumers[i].Queue.Count > 0 then
                        dec(FCount);
                end;
        end
        else
            FConsumers[Idx].Queue.Enqueue(Value);
    finally
        FLock.Exit;
    end;

    if Arr <> nil then
        FContainedProvider.Broadcast(Arr);
end;

function TMycDataJoin<T>.GetConsumers(Idx: Integer): IConsumer<T>;
begin
    Result := FConsumers[Idx].Consumer;
end;

{ TMycGenericAggregator<S, T> }

constructor TMycGenericAggregator<S, T>.Create(const AFunc: TConstFunc<S, TConverter<S, T>.TBroadcastProc, TState>);
begin
    inherited Create;
    FFunc := AFunc;
end;

function TMycGenericAggregator<S, T>.Consume(const Value: S): TState;
begin
    Result := FFunc(Value, Broadcast);
end;

{ TMycComposedConverter<S, T> }

constructor TMycComposedConverter<S, T>.Create(const AConsumer: IConsumer<S>; const AProducer: TProducer<T>);
begin
    inherited Create;
    FConsumer := AConsumer;
    FProducer := TProducer<T>(AProducer);
end;

function TMycComposedConverter<S, T>.GetConsumer: IConsumer<S>;
begin
    Result := FConsumer;
end;

function TMycComposedConverter<S, T>.Link(const Consumer: IConsumer<T>): TTag;
begin
    // Delegate to the contained producer
    Result := IProducer<T>(FProducer).Link(Consumer);
end;

procedure TMycComposedConverter<S, T>.Unlink(Tag: TTag);
begin
    // Delegate to the contained producer
    IProducer<T>(FProducer).Unlink(Tag);
end;

{ TMycProducer<T> }

constructor TMycProducer<T>.Create;
begin
    inherited Create;
end;

class constructor TMycProducer<T>.CreateClass;
begin
    FNull := TNull.Create;
end;

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

function TMycProducer<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.Consume(Value).Signal.Subscribe(done);
            item := item.Prev;
        end;

        Result := done.State;
    finally
        FListeners.Release;
    end;
end;

function TMycProducer<T>.Link(const Consumer: IConsumer<T>): TTag;
begin
    // Add the Consumer to the notification list
    FListeners.Lock;
    try
        Result := FListeners.Advise(Consumer);
    finally
        FListeners.Release;
    end;
end;

procedure TMycProducer<T>.Unlink(Tag: TTag);
begin
    FListeners.Lock;
    try
        FListeners.Unadvise(Tag);
    finally
        FListeners.Release;
    end;
end;

function TMycConsumer<T>.TNull.Consume(const Value: T): TState;
begin
    Result := TState.Null;
end;

function TMycProducer<T>.TNull.Link(const Consumer: IConsumer<T>): TTag;
begin
    Result := nil;
end;

procedure TMycProducer<T>.TNull.Unlink(Tag: TTag);
begin

end;

end.