unit Myc.Trade.DataPoint.Impl;

interface

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

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

    TMycSequence<T> = class(TMycProcessor<T>, IMycDataSequence<T>)
    private
        FDataProviders: TArray<TMycContainedDataProvider<T>>;
        function GetCount: Integer;
        function GetDataProvider(Idx: Integer): TDataProvider<T>;
    protected
        function ProcessData(const Value: T): TState; override;
        function ProcessDataProvider(Idx: Integer; const Value: T): TState;
    public
        constructor Create(ACount: Integer);
        destructor Destroy; override;
    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: TMycContainedDataProvider<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;

    TMycGenericParallelConverter<S, T> = class(TMycConverter<S, T>)
    private
        FFunc: TConstFunc<S, T>;
        FQueue: TState;
    protected
        function ProcessData(const Value: S): TState; override;
    public
        constructor Create(const AFunc: TConstFunc<S, T>);
    end;

    TMycIdentityConverter<T> = class(TMycConverter<T, T>)
    protected
        function ProcessData(const Value: T): TState; override; final;
    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>)
    private
        FProc: TConstFunc<T, TState>;
    protected
        function ProcessData(const Value: T): TState; override; final;
    public
        constructor Create(const AProc: TConstFunc<T, TState>);
    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, TLazy<TSeries<T>>.ILazy)
        type
            PItem = ^TItem;
            TItem = record
                Next: PItem;
                Value: T;
            end;
    private
        FProcessor: TMycContainedProcessor<T>;
        FTag: TDataProvider<T>.TTag;
        FDataProvider: TDataProvider<T>;
        FLookback: Int64;
        FChanged: TFlag;
        FLock: TLightweightMREW;
        FFirst: PItem;
        FCount: Integer;
        function GetChanged: TState;
        function ProcessData(const Value: T): TState;
    public
        constructor Create(const ADataProvider: TDataProvider<T>; ALookback: Int64);
        destructor Destroy; override;
        function Update(var Value: TSeries<T>): Boolean;
    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;
    public
        constructor Create(const ATimeframe: TTimeframe);
        function ProcessData(const Value: TDataPoint<Double>): TState; override;
        property CurrentBar: TDataPoint<TOhlcItem> read GetCurrentBar;
        property Timeframe: TTimeframe read GetTimeframe;
    end;

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

implementation

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

{ TMycProcessor<T> }

function TMycProcessor<T>.Execute(const Value: T; const Proc: TConstFunc<T, TState>): TState;
begin
    var cValue := Value;
    Result := TaskManager.RunTask(function: TState begin Result := Proc(cValue); end);
end;

{ TMycContainedDataProvider<T> }

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

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

function TMycContainedDataProvider<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 TMycContainedDataProvider<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 TMycContainedDataProvider<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 := TMycContainedDataProvider<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 := Execute(Value, function(const Value: S): TState begin Result := Broadcast(Calculate(Value)); end);
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: TConstFunc<T, TState>);
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: TState;
begin
    Result := FChanged.State
end;

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

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 := Value.Add(Arr, 0, Length(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
    // Align the time grid to UTC 0:00 using functions from System.DateUtils
    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, D3 are uncommon; this is a simple modulo-based approach relative to TDateTime's epoch.
        D2: Result := Floor(TimeStamp) - (Floor(TimeStamp) mod 2);
        D3: Result := Floor(TimeStamp) - (Floor(TimeStamp) mod 3);

        W: Result := TimeStamp.StartOfTheWeek;

        MN: Result := TimeStamp.StartOfTheMonth;
        // Quarter alignment
        MN3: Result := RecodeMonth(TimeStamp.StartOfTheMonth, (MonthOf(TimeStamp) - 1) div 3 * 3 + 1);
        // Half-year alignment
        MN6: Result := RecodeMonth(TimeStamp.StartOfTheMonth, (MonthOf(TimeStamp) - 1) div 6 * 6 + 1);

        Y: Result := TimeStamp.StartOfTheYear;
    else
        // Fallback for any undefined timeframe
        Result := 0;
    end;
end;

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

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

function TTickAggregation.ProcessData(const Value: TDataPoint<Double>): TState;
var
    barStartTime: TDateTime;
    lastBarTime: TDateTime;
begin
    // Update bar for the strategy's timeframe
    barStartTime := GetBarStartTime(Value.Time, FTimeframe);
    lastBarTime := FCurrentBar.Time;

    if (barStartTime > lastBarTime) then
    begin
        // A new bar starts, so the previous one is now complete.
        if (lastBarTime > 0) then
        begin
            Result := Broadcast(FCurrentBar);
        end;

        // Start a new bar, Volume is 1 because this is the first tick.
        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
        // Update the currently aggregating bar
        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;
        // Volume is the number of ticks needed to build the complete bar.
        FCurrentBar.Data.Volume := FCurrentBar.Data.Volume + 1;
    end;
end;

{ TMycSequence<T> }

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

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

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

function TMycSequence<T>.GetCount: Integer;
begin
    Result := Length(FDataProviders);
end;

function TMycSequence<T>.GetDataProvider(Idx: Integer): TDataProvider<T>;
begin
    Result := FDataProviders[Idx];
end;

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

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

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

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

    FQueue := Result;
end;

function TMycParallelConverter<T>.ProcessData(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;
var
    baseTime: TDateTime;
begin
    // Align the time grid to UTC 0:00 using functions from System.DateUtils
    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, D3 are uncommon; this is a simple modulo-based approach relative to TDateTime's epoch.
        D2: Result := Floor(TimeStamp) - (Floor(TimeStamp) mod 2);
        D3: Result := Floor(TimeStamp) - (Floor(TimeStamp) mod 3);

        W: Result := TimeStamp.StartOfTheWeek;

        MN: Result := TimeStamp.StartOfTheMonth;
        // Quarter alignment
        MN3: Result := RecodeMonth(TimeStamp.StartOfTheMonth, (MonthOf(TimeStamp) - 1) div 3 * 3 + 1);
        // Half-year alignment
        MN6: Result := RecodeMonth(TimeStamp.StartOfTheMonth, (MonthOf(TimeStamp) - 1) div 6 * 6 + 1);

        Y: Result := TimeStamp.StartOfTheYear;
    else
        // Fallback for any undefined timeframe
        Result := 0;
    end;
end;

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

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

function TOhlcAggregation.ProcessData(const Value: TDataPoint<TOhlcItem>): TState;
var
    barStartTime: TDateTime;
    lastBarTime: TDateTime;
begin
    // Update bar for the strategy's timeframe
    barStartTime := GetBarStartTime(Value.Time, FTimeframe);
    lastBarTime := FCurrentBar.Time;

    if (barStartTime > lastBarTime) then
    begin
        // A new bar starts, so the previous one is now complete.
        if (lastBarTime > 0) then
        begin
            Result := Broadcast(FCurrentBar);
        end;

        // Start a new bar, Volume is 1 because this is the first tick.
        FCurrentBar.Data := Value.Data;
        FCurrentBar.Time := barStartTime;
    end
    else
    begin
        // Update the currently aggregating bar
        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;
        // Volume is the number of ticks needed to build the complete bar.
        FCurrentBar.Data.Volume := FCurrentBar.Data.Volume + Value.Data.Volume;
    end;
end;

end.