MycLib/Src/Myc.Trade.DataPoint.pas

unit Myc.Trade.DataPoint;

interface

uses
    System.TimeSpan,
    Myc.Signals,
    Myc.Core.Notifier,
    Myc.Trade.Types;

type
    // Represents a time-stamped data point in a series.
    TDataPoint<T> = record
        Time: TDateTime;
        Data: T;
        constructor Create(ATime: TDateTime; const AData: T);
    end;

    IMycProcessor<T> = interface
        function ProcessData(const Value: T): TState;
    end;

    TMycProcessor<T> = class abstract(TInterfacedObject, IMycProcessor<T>)
    protected
        function ProcessData(const Value: T): TState; virtual; abstract;
    end;

    TTag = Pointer;

    IMycDataProvider<T> = interface
        function Link(const Receiver: IMycProcessor<T>): TTag;
        procedure Unlink(Tag: TTag);
    end;

    TMycDataProvider<T> = class abstract(TContainedObject, IMycDataProvider<T>)
    private
        FListeners: TMycNotifyList<IMycProcessor<T>>;
    public
        constructor Create(const Controller: IInterface);
        destructor Destroy; override;
        // Notifies all linked processors.
        function Broadcast(const Value: T): TState;
        procedure Notify(const Func: TMycNotifyList<IMycProcessor<T>>.TNotifyProc);
        // Link a Processor
        function Link(const Processor: IMycProcessor<T>): TTag;
        // Unlink a linked strategy
        procedure Unlink(Tag: TTag);
    end;

    TMycGenericProcessor<T> = class(TMycProcessor<T>)
        type
            TProc = reference to function(const Value: T): TState;
    private
        FProc: TProc;
    protected
        function ProcessData(const Value: T): TState; override; final;
    public
        constructor Create(const AProc: TProc);
    end;

    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;

    IMycConverter<S, T> = interface(IMycProcessor<S>)
        function GetSender: IMycDataProvider<T>;
        property Sender: IMycDataProvider<T> read GetSender;
    end;

    TMycConverter<S, T> = class abstract(TMycProcessor<S>, IMycConverter<S, T>)
    private
        FSender: TMycDataProvider<T>;
        function GetSender: IMycDataProvider<T>;
    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: IMycDataProvider<T> read GetSender;
    end;

    // An immutable time-ordered series of data points.
    // The most recent element has the logical index 0.
    IDataSeries<T> = interface
        function GetCount: Int64;
        function GetItems(Idx: Int64): TDataPoint<T>;
        function GetTotalCount: Int64;
        function GetLookback: Int64;
        // Add data and result the new series.
        function Add(const Data: TArray<TDataPoint<T>>; First, Count: Integer): IDataSeries<T>;
        property Count: Int64 read GetCount;
        // Accesses data points by their logical index.
        // Index 0 is the newest element, Index (Count - 1) is the oldest.
        property Items[Idx: Int64]: TDataPoint<T> read GetItems; default;
        // The maximum number of adressable items. Count will never be bigger than the Lookback.
        property Lookback: Int64 read GetLookback;
        // The total number of items ever added to the series.
        property TotalCount: Int64 read GetTotalCount;
    end;

    // Interface Helper for IDataSeries<T>.
    // Provides a safe, value-type-like wrapper around the interface.
    TDataSeries<T> = record
        type
            TConvertFunc<S> = reference to function(const Val: TDataPoint<T>): S;
    private
        FDataSeries: IDataSeries<T>;
        function GetCount: Int64;
        function GetItems(Idx: Int64): TDataPoint<T>;
        function GetData(Idx: Int64): T;
        function GetTime(Idx: Int64): TDateTime;
        function GetTotalCount: Int64;
        function GetLookback: Int64;
        class function GetNull: IDataSeries<T>; static;
    public
        constructor Create(ADataSeries: IDataSeries<T>);

        class operator Initialize(out Dest: TDataSeries<T>);
        class operator Finalize(var Dest: TDataSeries<T>);

        class operator Implicit(const A: TDataSeries<T>): IDataSeries<T>;
        class operator Implicit(const A: IDataSeries<T>): TDataSeries<T>;

        class function CreateDataSeries(Lookback: Int64; const Data: TArray<TDataPoint<T>> = nil): TDataSeries<T>; static;

        function Add(const Data: TArray<TDataPoint<T>>): TDataSeries<T>; overload;
        function Add(const Data: TArray<TDataPoint<T>>; First, Count: Integer): TDataSeries<T>; overload;
        function Convert<S>(const Func: TConvertFunc<S>): TDataSeries<S>;

        // Searches for a data point by its timestamp.
        // Returns the logical index of the matching item.
        // If no exact match, returns the index of the item immediately preceding the timestamp.
        // Returns -1 if the timestamp is before the oldest item in the series.
        function IndexOf(TimeStamp: TDateTime): Int64;

        function ToArray: TArray<TDataPoint<T>>;
        function ToDataArray: TArray<T>;

        class property Null: IDataSeries<T> read GetNull;
        property Count: Int64 read GetCount;
        property TotalCount: Int64 read GetTotalCount;
        property Lookback: Int64 read GetLookback;
        property Items[Idx: Int64]: TDataPoint<T> read GetItems; default;
        property Data[Idx: Int64]: T read GetData;
        property Time[Idx: Int64]: TDateTime read GetTime;
    end;

    TDataSeriesDoubleHelper = record helper for TDataSeries<Double>
        function ToOhlc(TimeFrame: TTimeSpan): TDataSeries<TOhlcItem>;
    end;

    TDataSeriesOhlcHelper = record helper for TDataSeries<TOhlcItem>
        function ToOpen: TDataSeries<Single>;
        function ToClose: TDataSeries<Single>;
        function ToHigh: TDataSeries<Single>;
        function ToLow: TDataSeries<Single>;
        function ToVolume: TDataSeries<Single>;
    end;

implementation

uses
    System.SysUtils,
    System.Math,
    System.Generics.Collections,
    Myc.Trade.DataArray,
    Myc.Trade.Core.DataPoint;

// Optimized helper function using direct TTimeSpan features and integer arithmetic.
function CeilToTimeSpan(const ATime: TDateTime; const ATimeSpan: TTimeSpan): TDateTime;
var
    timeSinceMidnight: TTimeSpan;
    timeSpanTicks: Int64;
    numIntervals, ceiledTicks: Int64;
begin
    timeSpanTicks := ATimeSpan.Ticks;
    Assert(timeSpanTicks > 0, 'TimeSpan must be positive.');

    // Get the time portion of ATime directly as a TTimeSpan.
    timeSinceMidnight := TTimeSpan.Subtract(ATime, Trunc(ATime));

    // Using integer arithmetic to find the ceiling is robust.
    // This is a standard formula for integer ceiling division: (numerator + denominator - 1) / denominator
    numIntervals := (timeSinceMidnight.Ticks + timeSpanTicks - 1) div timeSpanTicks;

    ceiledTicks := numIntervals * timeSpanTicks;

    // Construct the final DateTime from the date part and the new, aligned time part.
    Result := Trunc(ATime) + TTimeSpan.FromTicks(ceiledTicks);
end;

function TDataSeriesDoubleHelper.ToOhlc(TimeFrame: TTimeSpan): TDataSeries<TOhlcItem>;
var
    ohlcPoints: TList<TDataPoint<TOhlcItem>>;
    currentBar: TOhlcItem;
    windowEndTime: TDateTime;
    sourceIdx: Int64;
    firstPointInBar: Boolean;
begin
    if Self.Count = 0 then
        exit(TDataSeries<TOhlcItem>.Create(TNullDataSeries<TOhlcItem>.Null));

    ohlcPoints := TList<TDataPoint<TOhlcItem>>.Create;
    try
        if TimeFrame.Ticks <= 0 then
            raise EArgumentException.Create('Invalid TimeFrame for OHLC aggregation.');

        sourceIdx := Self.Count - 1; // Start with the oldest data point
        firstPointInBar := True;
        windowEndTime := 0;

        // Iterate through all source points chronologically (oldest to newest)
        while sourceIdx >= 0 do
        begin
            var P := Self.Items[sourceIdx];

            if firstPointInBar then
            begin
                windowEndTime := CeilToTimeSpan(P.Time, TimeFrame);
                currentBar.Create(P.Data, P.Data, P.Data, P.Data, 0);
                firstPointInBar := False;
            end;

            if P.Time >= windowEndTime then
            begin
                ohlcPoints.Add(TDataPoint<TOhlcItem>.Create(windowEndTime, currentBar));
                firstPointInBar := True;
                Continue; // Re-evaluate the same point for the next bar
            end;

            currentBar.High := Max(currentBar.High, P.Data);
            currentBar.Low := Min(currentBar.Low, P.Data);
            currentBar.Close := P.Data;
            currentBar.Volume := currentBar.Volume + 1;

            Dec(sourceIdx);
        end;

        if not firstPointInBar then
            ohlcPoints.Add(TDataPoint<TOhlcItem>.Create(windowEndTime, currentBar));

        Result := TDataSeries<TOhlcItem>.CreateDataSeries(ohlcPoints.Count, ohlcPoints.ToArray);
    finally
        ohlcPoints.Free;
    end;
end;

function TDataSeriesOhlcHelper.ToClose: TDataSeries<Single>;
begin
    Result := Self.Convert<Single>(function(const Ohlc: TDataPoint<TOhlcItem>): Single begin Result := Ohlc.Data.Close; end);
end;

function TDataSeriesOhlcHelper.ToHigh: TDataSeries<Single>;
begin
    Result := Self.Convert<Single>(function(const Ohlc: TDataPoint<TOhlcItem>): Single begin Result := Ohlc.Data.High; end);
end;

function TDataSeriesOhlcHelper.ToLow: TDataSeries<Single>;
begin
    Result := Self.Convert<Single>(function(const Ohlc: TDataPoint<TOhlcItem>): Single begin Result := Ohlc.Data.Low; end);
end;

function TDataSeriesOhlcHelper.ToOpen: TDataSeries<Single>;
begin
    Result := Self.Convert<Single>(function(const Ohlc: TDataPoint<TOhlcItem>): Single begin Result := Ohlc.Data.Open; end);
end;

function TDataSeriesOhlcHelper.ToVolume: TDataSeries<Single>;
begin
    Result := Self.Convert<Single>(function(const Ohlc: TDataPoint<TOhlcItem>): Single begin Result := Ohlc.Data.Volume; end);
end;

{ TDataPoint<T> }

constructor TDataPoint<T>.Create(ATime: TDateTime; const AData: T);
begin
    Time := ATime;
    Data := AData;
end;

constructor TDataSeries<T>.Create(ADataSeries: IDataSeries<T>);
begin
    if Assigned(ADataSeries) then
        FDataSeries := ADataSeries
    else
        FDataSeries := Null;
end;

function TDataSeries<T>.Add(const Data: TArray<TDataPoint<T>>; First, Count: Integer): TDataSeries<T>;
begin
{$ifdef DEBUG}
    for var i := First + 1 to First + Count - 1 do
        Assert(Data[i].Time >= Data[i - 1].Time, 'Input array for Add is not chronologically sorted');
    if FDataSeries.Count > 0 then
        Assert(Data[First].Time >= FDataSeries[0].Time, 'First new item is older than last existing item');
{$endif}

    Result := FDataSeries.Add(Data, First, Count);
end;

function TDataSeries<T>.Add(const Data: TArray<TDataPoint<T>>): TDataSeries<T>;
begin
    Result := Add(Data, 0, Length(Data));
end;

function TDataSeries<T>.Convert<S>(const Func: TConvertFunc<S>): TDataSeries<S>;
begin
    Result := TConvertSeries<T, S>.Create(FDataSeries, Func);
end;

class function TDataSeries<T>.CreateDataSeries(Lookback: Int64; const Data: TArray<TDataPoint<T>> = nil): TDataSeries<T>;
begin
    Result :=
        TMycDataSeries<T>.CreateDataSeries(Lookback, TMycDataArray<TDataPoint<T>>.CreateFromArray(Data, 0, Length(Data)), Length(Data));
end;

class operator TDataSeries<T>.Finalize(var Dest: TDataSeries<T>);
begin
    Dest.FDataSeries := nil;
end;

class operator TDataSeries<T>.Initialize(out Dest: TDataSeries<T>);
begin
    Dest.FDataSeries := Null;
end;

class operator TDataSeries<T>.Implicit(const A: TDataSeries<T>): IDataSeries<T>;
begin
    Result := A.FDataSeries;
end;

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

function TDataSeries<T>.GetCount: Int64;
begin
    Result := FDataSeries.GetCount;
end;

function TDataSeries<T>.GetItems(Idx: Int64): TDataPoint<T>;
begin
    Result := FDataSeries[Idx];
end;

function TDataSeries<T>.GetData(Idx: Int64): T;
begin
    Result := FDataSeries[Idx].Data;
end;

function TDataSeries<T>.GetLookback: Int64;
begin
    Result := FDataSeries.Lookback;
end;

function TDataSeries<T>.GetTime(Idx: Int64): TDateTime;
begin
    Result := FDataSeries[Idx].Time;
end;

class function TDataSeries<T>.GetNull: IDataSeries<T>;
begin
    Result := TNullDataSeries<T>.Null;
end;

function TDataSeries<T>.GetTotalCount: Int64;
begin
    Result := FDataSeries.GetTotalCount;
end;

function TDataSeries<T>.IndexOf(TimeStamp: TDateTime): Int64;
var
    low, high, mid: Int64;
    dataPointTime: TDateTime;
begin
    Result := -1;
    if Count = 0 then
        Exit;

    low := 0;
    high := Count - 1;

    while (low <= high) do
    begin
        mid := low + (high - low) div 2;
        dataPointTime := FDataSeries[mid].Time;

        if (dataPointTime = TimeStamp) then
        begin
            Result := mid;
            break;
        end
        else if (dataPointTime < TimeStamp) then
        begin
            Result := mid;
            high := mid - 1;
        end
        else
        begin
            low := mid + 1;
        end;
    end;
end;

function TDataSeries<T>.ToArray: TArray<TDataPoint<T>>;
begin
    var n := FDataSeries.Count;
    SetLength(Result, n);
    dec(n);
    for var i := 0 to n do
        Result[i] := FDataSeries[n - i];
end;

function TDataSeries<T>.ToDataArray: TArray<T>;
begin
    var n := FDataSeries.Count;
    SetLength(Result, n);
    dec(n);
    for var i := 0 to n do
        Result[i] := FDataSeries[n - i].Data;
end;

constructor TMycGenericProcessor<T>.Create(const AProc: TProc);
begin
    inherited Create;
    FProc := AProc;
end;

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

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;

{ TMycDataProvider<S, T> }

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

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

function TMycDataProvider<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;

procedure TMycDataProvider<T>.Notify(const Func: TMycNotifyList<IMycProcessor<T>>.TNotifyProc);
begin
    FListeners.Lock;
    try
        FListeners.Notify(Func);
    finally
        FListeners.Release;
    end;
end;

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

procedure TMycDataProvider<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;
    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: IMycDataProvider<T>;
begin
    Result := FSender;
end;

end.