Implementing first "strategy" for proof of concept

Src/Myc.Core.Notifier.pas

@@ -23,6 +23,9 @@ type
                 Receiver: T; // The registered interface instance (the event sink).
             end;
 
+            // Notify function. If this results false, it will be removed from the notify list
+            TNotifyProc = reference to function(const Obj: T): Boolean;
+
     strict private
         // Head of the linked list. The pointer value itself is repurposed for a spinlock.
         // Bit 0 of the address stores the lock state (0 = locked, 1 = unlocked).
@@ -36,9 +39,9 @@ type
 
     public
         // Initializes the list in an unlocked state.
-        procedure Create;
+        class operator Initialize(out Dest: TMycNotifyList<T>);
         // Safely clears all registered receivers and cleans up resources.
-        procedure Destroy;
+        procedure Finalize;
         // Registers a receiver interface and returns an opaque tag for later unsubscription.
         function Advise(const Receiver: T): TTag;
         // Unregisters a single receiver using its subscription tag.
@@ -54,17 +57,17 @@ type
         // Invokes a predicate for each registered receiver.
         // If the predicate returns false, the receiver is detached from further notifications
         // by setting its interface reference to nil. The list item itself is not freed here.
-        procedure Notify(Func: TPredicate<T>);
+        procedure Notify(const Func: TNotifyProc);
     end;
 
 implementation
 
-procedure TMycNotifyList<T>.Create;
+class operator TMycNotifyList<T>.Initialize(out Dest: TMycNotifyList<T>);
 begin
-    NativeUInt(FList) := 1;
+    NativeUInt(Dest.FList) := 1;
 end;
 
-procedure TMycNotifyList<T>.Destroy;
+procedure TMycNotifyList<T>.Finalize;
 begin
     Lock;
     UnadviseAll;
@@ -114,7 +117,7 @@ begin
     Result := NativeUInt(FList) and 1 = 0;
 end;
 
-procedure TMycNotifyList<T>.Notify(Func: TPredicate<T>);
+procedure TMycNotifyList<T>.Notify(const Func: TNotifyProc);
 var
     Item: PItem;
     Last: PItem;

Src/Myc.Core.Signals.pas

@@ -184,12 +184,11 @@ end;
 constructor TMycEvent.Create;
 begin
     inherited Create;
-    FSubscribers.Create;
 end;
 
 destructor TMycEvent.Destroy;
 begin
-    FSubscribers.Destroy;
+    FSubscribers.Finalize;
     inherited;
 end;
 
@@ -202,7 +201,7 @@ function TMycEvent.Notify: Boolean;
 begin
     FSubscribers.Lock;
     try
-        FSubscribers.Notify(function(Subscriber: TSignal.ISubscriber): Boolean begin Result := Subscriber.Notify; end);
+        FSubscribers.Notify(function(const Subscriber: TSignal.ISubscriber): Boolean begin Result := Subscriber.Notify; end);
     finally
         FSubscribers.Release;
     end;
@@ -275,12 +274,11 @@ begin
     inherited Create;
     Assert(ACount >= 0);
     FCount := ACount;
-    FSubscribers.Create;
 end;
 
 destructor TMycLatch.Destroy;
 begin
-    FSubscribers.Destroy;
+    FSubscribers.Finalize;
     inherited;
 end;
 
@@ -300,7 +298,7 @@ begin
 
         if shouldNotifySubscribers then
         begin
-            FSubscribers.Notify(function(Subscriber: TSignal.ISubscriber): Boolean begin Result := Subscriber.Notify; end);
+            FSubscribers.Notify(function(const Subscriber: TSignal.ISubscriber): Boolean begin Result := Subscriber.Notify; end);
         end;
 
         // Returns true if the latch has not yet been set by this Notify call (count > 0).
@@ -395,12 +393,11 @@ constructor TMycFlag.Create(AInit: Boolean);
 begin
     inherited Create;
     FFlag := AInit;
-    FSubscribers.Create;
 end;
 
 destructor TMycFlag.Destroy;
 begin
-    FSubscribers.Destroy; // Clean up the subscriber list.
+    FSubscribers.Finalize;
     inherited;
 end;
 
@@ -438,7 +435,7 @@ begin
 
         if wasPreviouslyClean then // Only notify subscribers if state changed from clean to dirty.
         begin
-            FSubscribers.Notify(function(Subscriber: TSignal.ISubscriber): Boolean begin Result := Subscriber.Notify; end);
+            FSubscribers.Notify(function(const Subscriber: TSignal.ISubscriber): Boolean begin Result := Subscriber.Notify; end);
         end;
     finally
         FSubscribers.Release;

Src/Myc.Signals.FMX.pas

@@ -14,6 +14,7 @@ type
     public
         function ProcessSignal(const Signal: TSignal; const Proc: TMsgProc): TComponent; overload;
         function ProcessSignal(const Signal: TSignal; const Proc: TProc): TComponent; overload;
+        function ProcessSignal(const Signal: TSignal; const Done: TState; const Proc: TProc): TComponent; overload;
     end;
 
     TSignalSyncHelper = record helper for TSignal
@@ -198,6 +199,21 @@ begin
     Result := TSignalSubscriber.Create(Self, Signal, Proc);
 end;
 
+function TSignalComponentHelper.ProcessSignal(const Signal: TSignal; const Done: TState; const Proc: TProc): TComponent;
+begin
+    var cProc: TProc := Proc;
+    var cDone: TState := Done;
+    Result :=
+        ProcessSignal(
+            Signal,
+            procedure(out IsDone: Boolean)
+            begin
+                cProc();
+                IsDone := cDone.IsSet;
+            end
+        );
+end;
+
 { TSignalSyncHelper }
 
 function TSignalSyncHelper.Queue(const Proc: TProc; Delay: Integer = 0): TSignal.TSubscription;

Src/Myc.Signals.pas

@@ -153,7 +153,7 @@ type
 
         class function CreateLatch(Count: Integer): ILatch; static;
 
-        class function Enqueue1(var Gate: TLatch): TState; overload; static;
+        class function Enqueue1(var Gate: TLatch): TState; overload; static; deprecated; experimental;
         function Enqueue: TState; overload;
 
         class property Null: ILatch read FNull;

Src/Myc.Test.Trade.DataPoint.pas

@@ -134,7 +134,7 @@ procedure TTestDataSeries.TestSetupSeriesWithDataVerification;
 var
     i: Integer;
     baseTime, expectedTime: TDateTime;
-    expectedAsk: Single;
+    expectedAsk: Double;
 begin
     SetupSeriesWithData;
     baseTime := EncodeDate(2020, 7, 7);
@@ -200,7 +200,7 @@ begin
     begin
         expectedTime := baseTime + (9 - i);
         Assert.AreEqual(expectedTime, FSeries.Items[i].Time, 'Item at logical index should have reversed chronological time');
-        Assert.AreEqual(Single(9 - i), FSeries.Items[i].Data.Ask, 'Item data at logical index should match reversed insertion order');
+        Assert.AreEqual(Double(9 - i), FSeries.Items[i].Data.Ask, 'Item data at logical index should match reversed insertion order');
     end;
 end;
 
@@ -648,9 +648,9 @@ procedure TTestDataSeries.TestDataAndTimePropertiesRespectLookback;
 var
     baseTime: TDateTime;
     expectedOldestTime: TDateTime;
-    expectedOldestAsk: Single;
+    expectedOldestAsk: Double;
     expectedNewestTime: TDateTime;
-    expectedNewestAsk: Single;
+    expectedNewestAsk: Double;
 begin
     SetupSeriesWithData(20, 15); // Add 20 items, lookback 15.
     // Visible items are those originally at index 5 through 19.

Src/Myc.Test.Trade.DataStream.pas

@@ -19,8 +19,8 @@ type
     [IgnoreMemoryLeaks(true)]
     TTest_TABFileServer_Equivalence = class(TObject)
     private
-        FServer: IDataServer<TAskBidItem>;
-        FStream: IDataStream<TAskBidItem>;
+        FServer: IDataServer<TAuraAskBidFileItem>;
+        FStream: IDataStream<TAuraAskBidFileItem>;
     public
         [SetupFixture]
         procedure SetupFixture;
@@ -77,17 +77,17 @@ end;
 
 procedure TTest_TABFileServer_Equivalence.Test_ServerReturnsSameDataAs_LoadDataSeries;
 var
-    chunk: array[0..C_MAX_FETCH - 1] of TDataPoint<TAskBidItem>;
-    dst: TArray<TDataPoint<TAskBidItem>>;
+    chunk: array[0..C_MAX_FETCH - 1] of TDataPoint<TAuraAskBidFileItem>;
+    dst: TArray<TDataPoint<TAuraAskBidFileItem>>;
     i: Int64;
     n: Integer;
     cnt: Int64;
     timeout: Integer;
     filename: TAuraDataFile;
-    expectedData: TArray<TDataPoint<TAskBidItem>>;
+    expectedData: TArray<TDataPoint<TAuraAskBidFileItem>>;
 begin
     // 1. Expected data is loaded directly using LoadDataSeries.
-    filename := (FServer as TAuraDataServer<TAskBidItem>).FindFirstFile(C_TEST_SYMBOL).WaitFor;
+    filename := (FServer as TAuraDataServer<TAuraAskBidFileItem>).FindFirstFile(C_TEST_SYMBOL).WaitFor;
     Assert.IsTrue(filename.IsValid, 'Test data file could not be found.');
 
     expectedData := (FServer as TAuraTABFileServer).LoadDataSeries(filename).WaitFor;

Src/Myc.Trade.Core.DataPoint.pas

@@ -5,35 +5,14 @@ interface
 uses
     System.Generics.Collections,
     System.TimeSpan,
-    Myc.Trade.DataPoint;
+    Myc.Trade.DataPoint,
+    Myc.Trade.DataArray;
 
 type
-    TMycDataArray<T> = record
-    private
-        const
-            ChunkSize = 1024;
-        type
-            TChunk = TArray<TDataPoint<T>>;
-    private
-        FChunks: TArray<TChunk>;
-        FCount: Int64;
-
-        function LogicalToPhysicalIndex(LogicalIndex: Int64): Int64; inline;
-        function GetItems(Idx: Int64): TDataPoint<T>; inline;
-    public
-        constructor Create(const AChunks: TArray<TChunk>; ACount: Int64);
-        function Add(const Data: array of TDataPoint<T>; First, Count, Lookback: Int64): TMycDataArray<T>;
-        class function CreateEmpty: TMycDataArray<T>; static;
-        // Helper to create a data array from a raw TArray.
-        class function CreateFromArray(const AData: TArray<TDataPoint<T>>; First, Count: Integer): TMycDataArray<T>; static;
-        property Count: Int64 read FCount;
-        property Items[Idx: Int64]: TDataPoint<T> read GetItems; default;
-    end;
-
     // The implementation class for IDataSeries<T>.
     TMycDataSeries<T> = class(TInterfacedObject, IDataSeries<T>)
     private
-        FData: TMycDataArray<T>;
+        FData: TMycDataArray<TDataPoint<T>>;
         FLookback: Int64;
         FTotalCount: Int64;
         function GetCount: Int64;
@@ -41,10 +20,14 @@ type
         function GetLookback: Int64;
         function GetTotalCount: Int64;
     public
-        constructor Create(ALookback: Int64; const AData: TMycDataArray<T>; ATotalCount: Int64);
+        constructor Create(ALookback: Int64; const AData: TMycDataArray<TDataPoint<T>>; ATotalCount: Int64);
         destructor Destroy; override;
         function Add(const Data: TArray<TDataPoint<T>>; First, Count: Integer): IDataSeries<T>;
-        class function CreateDataSeries(Lookback: Int64; const AData: TMycDataArray<T>; ATotalCount: Int64): IDataSeries<T>; static;
+        class function CreateDataSeries(
+            Lookback: Int64;
+            const AData: TMycDataArray<TDataPoint<T>>;
+            ATotalCount: Int64
+        ): IDataSeries<T>; static;
     end;
 
     // Null object implementation for IDataSeries<T>
@@ -69,7 +52,7 @@ type
     TCompositeDataSeries<T> = class(TInterfacedObject, IDataSeries<T>)
     private
         FBaseSeries: IDataSeries<T>;
-        FAddedData: TMycDataArray<T>;
+        FAddedData: TMycDataArray<TDataPoint<T>>;
         FLookback: Int64;
         FCount: Int64;
         function GetCount: Int64;
@@ -77,7 +60,7 @@ type
         function GetLookback: Int64;
         function GetTotalCount: Int64;
     public
-        constructor Create(const ABaseSeries: IDataSeries<T>; const AAddedData: TMycDataArray<T>);
+        constructor Create(const ABaseSeries: IDataSeries<T>; const AAddedData: TMycDataArray<TDataPoint<T>>);
         function Add(const Data: TArray<TDataPoint<T>>; First, Count: Integer): IDataSeries<T>;
         class function CreateComposite(
             const BaseSeries: IDataSeries<T>;
@@ -128,97 +111,9 @@ uses
     System.SysUtils,
     System.Math;
 
-{ TMycDataArray<T> }
-
-constructor TMycDataArray<T>.Create(const AChunks: TArray<TChunk>; ACount: Int64);
-begin
-    FChunks := AChunks;
-    FCount := ACount;
-end;
-
-class function TMycDataArray<T>.CreateEmpty: TMycDataArray<T>;
-begin
-    Result.FChunks := nil;
-    Result.FCount := 0;
-end;
-
-class function TMycDataArray<T>.CreateFromArray(const AData: TArray<TDataPoint<T>>; First, Count: Integer): TMycDataArray<T>;
-begin
-    // Use the Add method on an empty array to perform the chunking logic.
-    Result := CreateEmpty.Add(AData, First, Count, Count);
-end;
-
-function TMycDataArray<T>.Add(const Data: array of TDataPoint<T>; First, Count, Lookback: Int64): TMycDataArray<T>;
-var
-    destPhysicalIdx, sourcePhysicalIdx: Int64;
-    itemsToSkip: Int64;
-    numNewChunks: Integer;
-    newChunks: TArray<TChunk>;
-    destChunkIdx, destSubIdx: Integer;
-    sumCount, newCount: Int64;
-begin
-    if Count < 0 then
-        Count := Length(Data) - First;
-    if (Lookback <= 0) or (Count = 0) then
-        exit(Self);
-
-    Assert(Count <= (Length(Data) - First), 'Count cannot be larger than the source array');
-    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 FCount > 0 then
-        Assert(Data[First].Time >= Self.Items[0].Time, 'First new item is older than last existing item');
-
-    sumCount := FCount + Count;
-    newCount := sumCount;
-    if (Lookback > 0) and (newCount > Lookback) then
-        newCount := Lookback;
-    itemsToSkip := sumCount - newCount;
-
-    numNewChunks := 0;
-    if newCount > 0 then
-        numNewChunks := (newCount - 1) div ChunkSize + 1;
-    SetLength(newChunks, numNewChunks);
-
-    for destPhysicalIdx := 0 to newCount - 1 do
-    begin
-        destChunkIdx := destPhysicalIdx div ChunkSize;
-        destSubIdx := destPhysicalIdx mod ChunkSize;
-
-        if destSubIdx = 0 then
-            SetLength(newChunks[destChunkIdx], ChunkSize);
-
-        sourcePhysicalIdx := itemsToSkip + destPhysicalIdx;
-
-        if sourcePhysicalIdx < FCount then
-        begin
-            newChunks[destChunkIdx][destSubIdx] := FChunks[sourcePhysicalIdx div ChunkSize][sourcePhysicalIdx mod ChunkSize];
-        end
-        else
-        begin
-            newChunks[destChunkIdx][destSubIdx] := Data[First + (sourcePhysicalIdx - FCount)];
-        end;
-    end;
-
-    Result := TMycDataArray<T>.Create(newChunks, newCount);
-end;
-
-function TMycDataArray<T>.GetItems(Idx: Int64): TDataPoint<T>;
-var
-    physicalIndex: Int64;
-begin
-    Assert((Idx >= 0) and (Idx < FCount), 'Logical index is out of bounds.');
-    physicalIndex := LogicalToPhysicalIndex(Idx);
-    Result := FChunks[physicalIndex div ChunkSize][physicalIndex mod ChunkSize];
-end;
-
-function TMycDataArray<T>.LogicalToPhysicalIndex(LogicalIndex: Int64): Int64;
-begin
-    Result := FCount - LogicalIndex - 1;
-end;
-
 { TMycDataSeries<T> }
 
-constructor TMycDataSeries<T>.Create(ALookback: Int64; const AData: TMycDataArray<T>; ATotalCount: Int64);
+constructor TMycDataSeries<T>.Create(ALookback: Int64; const AData: TMycDataArray<TDataPoint<T>>; ATotalCount: Int64);
 begin
     inherited Create;
     FLookback := ALookback;
@@ -231,7 +126,11 @@ begin
     inherited;
 end;
 
-class function TMycDataSeries<T>.CreateDataSeries(Lookback: Int64; const AData: TMycDataArray<T>; ATotalCount: Int64): IDataSeries<T>;
+class function TMycDataSeries<T>.CreateDataSeries(
+    Lookback: Int64;
+    const AData: TMycDataArray<TDataPoint<T>>;
+    ATotalCount: Int64
+): IDataSeries<T>;
 begin
     if Lookback > 0 then
         Result := TMycDataSeries<T>.Create(Lookback, AData, ATotalCount)
@@ -241,7 +140,7 @@ end;
 
 function TMycDataSeries<T>.Add(const Data: TArray<TDataPoint<T>>; First, Count: Integer): IDataSeries<T>;
 var
-    newData: TMycDataArray<T>;
+    newData: TMycDataArray<TDataPoint<T>>;
     newTotalCount: Int64;
 begin
     if Count < 0 then
@@ -319,7 +218,7 @@ end;
 
 { TCompositeDataSeries<T> }
 
-constructor TCompositeDataSeries<T>.Create(const ABaseSeries: IDataSeries<T>; const AAddedData: TMycDataArray<T>);
+constructor TCompositeDataSeries<T>.Create(const ABaseSeries: IDataSeries<T>; const AAddedData: TMycDataArray<TDataPoint<T>>);
 begin
     inherited Create;
     FBaseSeries := ABaseSeries;
@@ -334,7 +233,7 @@ end;
 
 function TCompositeDataSeries<T>.Add(const Data: TArray<TDataPoint<T>>; First, Count: Integer): IDataSeries<T>;
 var
-    newAddedData: TMycDataArray<T>;
+    newAddedData: TMycDataArray<TDataPoint<T>>;
     itemsInBase: Int64;
     lookbackForAdd: Int64;
 begin
@@ -373,7 +272,7 @@ begin
     if Count = 0 then
         exit(BaseSeries);
 
-    Result := TCompositeDataSeries<T>.Create(BaseSeries, TMycDataArray<T>.CreateFromArray(Data, First, Count));
+    Result := TCompositeDataSeries<T>.Create(BaseSeries, TMycDataArray<TDataPoint<T>>.CreateFromArray(Data, First, Count));
 end;
 
 function TCompositeDataSeries<T>.GetCount: Int64;

Src/Myc.Trade.DataArray.pas

@@ -0,0 +1,123 @@
+unit Myc.Trade.DataArray;
+
+interface
+
+uses
+    Myc.Trade.DataPoint;
+
+type
+    TMycDataArray<T> = record
+    private
+        const
+            ChunkSize = 1024;
+        type
+            TChunk = TArray<T>;
+    private
+        FChunks: TArray<TChunk>;
+        FCount: Int64;
+
+        function LogicalToPhysicalIndex(LogicalIndex: Int64): Int64; inline;
+        function GetItems(Idx: Int64): T; inline;
+    public
+        constructor Create(const AChunks: TArray<TChunk>; ACount: Int64);
+        function Add(const Data: T; Lookback: Int64): TMycDataArray<T>; overload;
+        function Add(const Data: array of T; First, Count, Lookback: Int64): TMycDataArray<T>; overload;
+        class function CreateEmpty: TMycDataArray<T>; static;
+        // Helper to create a data array from a raw TArray.
+        class function CreateFromArray(const AData: TArray<T>; First, Count: Integer): TMycDataArray<T>; static;
+        property Count: Int64 read FCount;
+        property Items[Idx: Int64]: T read GetItems; default;
+    end;
+
+implementation
+
+{ TMycDataArray<T> }
+
+constructor TMycDataArray<T>.Create(const AChunks: TArray<TChunk>; ACount: Int64);
+begin
+    FChunks := AChunks;
+    FCount := ACount;
+end;
+
+class function TMycDataArray<T>.CreateEmpty: TMycDataArray<T>;
+begin
+    Result.FChunks := nil;
+    Result.FCount := 0;
+end;
+
+class function TMycDataArray<T>.CreateFromArray(const AData: TArray<T>; First, Count: Integer): TMycDataArray<T>;
+begin
+    // Use the Add method on an empty array to perform the chunking logic.
+    Result := CreateEmpty.Add(AData, First, Count, Count);
+end;
+
+function TMycDataArray<T>.Add(const Data: array of T; First, Count, Lookback: Int64): TMycDataArray<T>;
+var
+    destPhysicalIdx, sourcePhysicalIdx: Int64;
+    itemsToSkip: Int64;
+    numNewChunks: Integer;
+    newChunks: TArray<TChunk>;
+    destChunkIdx, destSubIdx: Integer;
+    sumCount, newCount: Int64;
+begin
+    if Count < 0 then
+        Count := Length(Data) - First;
+    if (Lookback <= 0) or (Count = 0) then
+        exit(Self);
+
+    Assert(Count <= (Length(Data) - First), 'Count cannot be larger than the source array');
+
+    sumCount := FCount + Count;
+    newCount := sumCount;
+    if (Lookback > 0) and (newCount > Lookback) then
+        newCount := Lookback;
+    itemsToSkip := sumCount - newCount;
+
+    numNewChunks := 0;
+    if newCount > 0 then
+        numNewChunks := (newCount - 1) div ChunkSize + 1;
+    SetLength(newChunks, numNewChunks);
+
+    for destPhysicalIdx := 0 to newCount - 1 do
+    begin
+        destChunkIdx := destPhysicalIdx div ChunkSize;
+        destSubIdx := destPhysicalIdx mod ChunkSize;
+
+        if destSubIdx = 0 then
+            SetLength(newChunks[destChunkIdx], ChunkSize);
+
+        sourcePhysicalIdx := itemsToSkip + destPhysicalIdx;
+
+        if sourcePhysicalIdx < FCount then
+        begin
+            newChunks[destChunkIdx][destSubIdx] := FChunks[sourcePhysicalIdx div ChunkSize][sourcePhysicalIdx mod ChunkSize];
+        end
+        else
+        begin
+            newChunks[destChunkIdx][destSubIdx] := Data[First + (sourcePhysicalIdx - FCount)];
+        end;
+    end;
+
+    Result := TMycDataArray<T>.Create(newChunks, newCount);
+end;
+
+function TMycDataArray<T>.Add(const Data: T; Lookback: Int64): TMycDataArray<T>;
+begin
+    Result := Add([Data], 0, 1, Lookback);
+end;
+
+function TMycDataArray<T>.GetItems(Idx: Int64): T;
+var
+    physicalIndex: Int64;
+begin
+    Assert((Idx >= 0) and (Idx < FCount), 'Logical index is out of bounds.');
+    physicalIndex := LogicalToPhysicalIndex(Idx);
+    Result := FChunks[physicalIndex div ChunkSize][physicalIndex mod ChunkSize];
+end;
+
+function TMycDataArray<T>.LogicalToPhysicalIndex(LogicalIndex: Int64): Int64;
+begin
+    Result := FCount - LogicalIndex - 1;
+end;
+
+end.

Src/Myc.Trade.DataPoint.pas

@@ -8,18 +8,18 @@ uses
 type
     // A data record for an Ask/Bid price pair.
     TAskBidItem = packed record
-        Ask: Single;
-        Bid: Single;
-        constructor Create(AAsk, ABid: Single);
+        Ask: Double;
+        Bid: Double;
+        constructor Create(AAsk, ABid: Double);
     end;
 
     TOhlcItem = record
-        Open: Single;
-        High: Single;
-        Low: Single;
-        Close: Single;
-        Volume: Single;
-        constructor Create(AOpen, AHigh, ALow, AClose, AVolume: Single);
+        Open: Double;
+        High: Double;
+        Low: Double;
+        Close: Double;
+        Volume: Double;
+        constructor Create(AOpen, AHigh, ALow, AClose, AVolume: Double);
     end;
 
     // Represents a time-stamped data point in a series.
@@ -29,13 +29,38 @@ type
         constructor Create(ATime: TDateTime; const AData: T);
     end;
 
-    // A time-ordered series of data points, optimized for chronological additions.
-    // The most recently added element has the logical index 0.
+    IMycProcessor<T> = interface
+        procedure ProcessData(const Value: T);
+    end;
+
+    TMycGenericProcessor<T> = class(TInterfacedObject, IMycProcessor<T>)
+        type
+            TProc = reference to procedure(const Value: T);
+    private
+        FProc: TProc;
+        procedure ProcessData(const Value: T);
+    public
+        constructor Create(const AProc: TProc);
+    end;
+
+    TMycContainedProcessor<T> = class(TContainedObject, IMycProcessor<T>)
+        type
+            TProc = procedure(const Value: T) of object;
+    private
+        FProc: TProc;
+        procedure ProcessData(const Value: T);
+    public
+        constructor Create(const Controller: IInterface; const AProc: TProc);
+    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.
@@ -74,7 +99,6 @@ type
 
         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.
@@ -99,12 +123,21 @@ type
         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.
@@ -180,17 +213,40 @@ begin
         if not firstPointInBar then
             ohlcPoints.Add(TDataPoint<TOhlcItem>.Create(windowEndTime, currentBar));
 
-        var dataArray := TMycDataArray<TOhlcItem>.CreateFromArray(ohlcPoints.ToArray, 0, ohlcPoints.Count);
-        var seriesImpl := TMycDataSeries<TOhlcItem>.Create(ohlcPoints.Count, dataArray, ohlcPoints.Count);
-        Result := TDataSeries<TOhlcItem>.Create(seriesImpl);
+        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;
+
 { TAskBidItem }
 
-constructor TAskBidItem.Create(AAsk, ABid: Single);
+constructor TAskBidItem.Create(AAsk, ABid: Double);
 begin
     Ask := AAsk;
     Bid := ABid;
@@ -198,7 +254,7 @@ end;
 
 { TOhlcItem }
 
-constructor TOhlcItem.Create(AOpen, AHigh, ALow, AClose, AVolume: Single);
+constructor TOhlcItem.Create(AOpen, AHigh, ALow, AClose, AVolume: Double);
 begin
     Open := AOpen;
     High := AHigh;
@@ -225,12 +281,19 @@ 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 := FDataSeries.Add(Data, 0, Length(Data));
+    Result := Add(Data, 0, Length(Data));
 end;
 
 function TDataSeries<T>.Convert<S>(const Func: TConvertFunc<S>): TDataSeries<S>;
@@ -240,7 +303,8 @@ end;
 
 class function TDataSeries<T>.CreateDataSeries(Lookback: Int64; const Data: TArray<TDataPoint<T>> = nil): TDataSeries<T>;
 begin
-    Result := TMycDataSeries<T>.CreateDataSeries(Lookback, TMycDataArray<T>.CreateFromArray(Data, 0, Length(Data)), Length(Data));
+    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>);
@@ -350,4 +414,26 @@ begin
         Result[i] := FDataSeries[n - i].Data;
 end;
 
+constructor TMycGenericProcessor<T>.Create(const AProc: TProc);
+begin
+    inherited Create;
+    FProc := AProc;
+end;
+
+procedure TMycGenericProcessor<T>.ProcessData(const Value: T);
+begin
+    FProc(Value);
+end;
+
+constructor TMycContainedProcessor<T>.Create(const Controller: IInterface; const AProc: TProc);
+begin
+    inherited Create(Controller);
+    FProc := AProc;
+end;
+
+procedure TMycContainedProcessor<T>.ProcessData(const Value: T);
+begin
+    FProc(Value);
+end;
+
 end.

Src/Myc.Trade.DataStream.pas

@@ -54,14 +54,12 @@ type
         function IsHistory: Boolean; virtual; abstract;
     end;
 
-    TDataProc<T> = reference to procedure(const Values: TArray<TDataPoint<T>>; const Terminated: TState);
-
     // Interface for an instantiable data server.
     IDataServer<T: record> = interface
         ['{1F8E5A9D-E92A-44C1-9F3F-C4B82A6E94B3}']
         function CreateStream(const Symbol: String): IDataStream<T>;
         procedure ClearCache;
-        function ProcessData(const Symbol: String; const Terminate: TSignal; const Proc: TDataProc<T>): TState;
+        function ProcessData(const Symbol: String; const Terminated: TState; const Processor: IMycProcessor<TArray<TDataPoint<T>>>): TState;
         function EnumerateSymbols: TFuture<TArray<String>>;
     end;
 
@@ -104,8 +102,8 @@ type
         function ProcessFile(
             FileInfo: TAuraDataFile;
             const DataFile: TFuture<TArray<TDataPoint<T>>>;
-            Terminated: TState;
-            Proc: TDataProc<T>
+            const Terminated: TState;
+            Processor: IMycProcessor<TArray<TDataPoint<T>>>
         ): TState;
 
     strict private
@@ -129,7 +127,7 @@ type
         function EnumerateSymbols: TFuture<TArray<String>>;
         function LoadDataFile(const DataFile: TAuraDataFile): TFuture<TArray<TDataPoint<T>>>;
 
-        function ProcessData(const Symbol: String; const Terminate: TSignal; const Proc: TDataProc<T>): TState;
+        function ProcessData(const Symbol: String; const Terminated: TState; const Processor: IMycProcessor<TArray<TDataPoint<T>>>): TState;
 
         property Path: String read GetPath;
     end;
@@ -166,14 +164,19 @@ type
 
     // Aura tick data file Ask-Bid
 
-    TAuraTABFileServer = class(TAuraDataServer<TAskBidItem>)
+    TAuraAskBidFileItem = packed record
+        Ask: Single;
+        Bid: Single;
+    end;
+
+    TAuraTABFileServer = class(TAuraDataServer<TAuraAskBidFileItem>)
     protected
         // Scans a directory and returns the oldest file found for each symbol.
-        class function ReadCompressedData(const InputStream: TStream): TArray<TDataPoint<TAskBidItem>>; override;
-        class function ReadUncompressedData(const InputStream: TStream): TArray<TDataPoint<TAskBidItem>>; override;
+        class function ReadCompressedData(const InputStream: TStream): TArray<TDataPoint<TAuraAskBidFileItem>>; override;
+        class function ReadUncompressedData(const InputStream: TStream): TArray<TDataPoint<TAuraAskBidFileItem>>; override;
     public
-        function CreateStream(const Symbol: String): IDataStream<TAskBidItem>; override;
-        function LoadDataSeries(const InitialFile: TAuraDataFile): TFuture<TArray<TDataPoint<TAskBidItem>>>;
+        function CreateStream(const Symbol: String): IDataStream<TAuraAskBidFileItem>; override;
+        function LoadDataSeries(const InitialFile: TAuraDataFile): TFuture<TArray<TDataPoint<TAuraAskBidFileItem>>>;
         function ParseFileName(const FileName: string): TAuraDataFile; override;
     end;
 
@@ -390,17 +393,21 @@ begin
     Result := FPath;
 end;
 
-function TAuraDataServer<T>.ProcessData(const Symbol: String; const Terminate: TSignal; const Proc: TDataProc<T>): TState;
+function TAuraDataServer<T>.ProcessData(
+    const Symbol: String;
+    const Terminated: TState;
+    const Processor: IMycProcessor<TArray<TDataPoint<T>>>
+): TState;
 begin
-    var capProc := Proc;
-    var terminated := TFlag.CreateObserver(Terminate).State;
+    var cProc := Processor;
+    var cTerminated := Terminated;
 
     Result :=
         FindFirstFile(Symbol)
             .Chain(
                 function(const FirstFileInfo: TAuraDataFile): TState
                 begin
-                    Result := ProcessFile(FirstFileInfo, LoadDataFile(FirstFileInfo), terminated, capProc);
+                    Result := ProcessFile(FirstFileInfo, LoadDataFile(FirstFileInfo), cTerminated, cProc);
                 end);
 end;
 
@@ -413,23 +420,24 @@ end;
 function TAuraDataServer<T>.ProcessFile(
     FileInfo: TAuraDataFile;
     const DataFile: TFuture<TArray<TDataPoint<T>>>;
-    Terminated: TState;
-    Proc: TDataProc<T>
+    const Terminated: TState;
+    Processor: IMycProcessor<TArray<TDataPoint<T>>>
 ): TState;
 begin
     if not FileInfo.IsValid or Terminated.IsSet then
-        exit(TState.Null);
+        exit(DataFile.Done);
 
     // Read ahead the next file, while processing the current one
     var nextFileInfo := FileInfo.GetNextFile;
     var nextFile := LoadDataFile(nextFileInfo);
 
+    var cTerminated := Terminated;
     Result :=
         DataFile.Chain(
             function(const Data: TArray<TDataPoint<T>>): TState
             begin
-                Proc(Data, Terminated);
-                Result := ProcessFile(nextFileInfo, nextFile, Terminated, Proc);
+                Processor.ProcessData(Data);
+                Result := ProcessFile(nextFileInfo, nextFile, cTerminated, Processor);
             end
         );
 end;
@@ -574,16 +582,16 @@ end;
 
 { TAuraTABFileServer }
 
-function TAuraTABFileServer.CreateStream(const Symbol: String): IDataStream<TAskBidItem>;
+function TAuraTABFileServer.CreateStream(const Symbol: String): IDataStream<TAuraAskBidFileItem>;
 begin
-    Result := TAuraFileStream<TAskBidItem>.Create(Self, Symbol);
+    Result := TAuraFileStream<TAuraAskBidFileItem>.Create(Self, Symbol);
 end;
 
-function TAuraTABFileServer.LoadDataSeries(const InitialFile: TAuraDataFile): TFuture<TArray<TDataPoint<TAskBidItem>>>;
+function TAuraTABFileServer.LoadDataSeries(const InitialFile: TAuraDataFile): TFuture<TArray<TDataPoint<TAuraAskBidFileItem>>>;
 var
     loadedState: TState;
-    loadedFiles: TArray<TFuture<TArray<TDataPoint<TAskBidItem>>>>;
-    liveData: TFuture<TArray<TDataPoint<TAskBidItem>>>;
+    loadedFiles: TArray<TFuture<TArray<TDataPoint<TAuraAskBidFileItem>>>>;
+    liveData: TFuture<TArray<TDataPoint<TAuraAskBidFileItem>>>;
     tabFiles: TList<TAuraDataFile>;
     liveFile: TAuraDataFile;
     currentFileInfo: TAuraDataFile;
@@ -607,7 +615,7 @@ begin
                 liveFile := ParseFileName(potentialLivePath);
         end;
 
-        var loadedFileList := TList<TFuture<TArray<TDataPoint<TAskBidItem>>>>.Create;
+        var loadedFileList := TList<TFuture<TArray<TDataPoint<TAuraAskBidFileItem>>>>.Create;
         var loadStates := TList<TState>.Create;
         try
             for var fileInfo in tabFiles do
@@ -617,7 +625,7 @@ begin
                 loadStates.Add(data.Done);
             end;
 
-            liveData := TFuture<TArray<TDataPoint<TAskBidItem>>>.Null;
+            liveData := TFuture<TArray<TDataPoint<TAuraAskBidFileItem>>>.Null;
             if liveFile.IsValid then
             begin
                 liveData := LoadDataFile(liveFile);
@@ -636,11 +644,11 @@ begin
     end;
 
     Result :=
-        TFuture<TArray<TDataPoint<TAskBidItem>>>.Construct(
+        TFuture<TArray<TDataPoint<TAuraAskBidFileItem>>>.Construct(
             loadedState,
-            function: TArray<TDataPoint<TAskBidItem>>
+            function: TArray<TDataPoint<TAuraAskBidFileItem>>
             begin
-                var tickList := TList<TDataPoint<TAskBidItem>>.Create;
+                var tickList := TList<TDataPoint<TAuraAskBidFileItem>>.Create;
                 try
                     var overallLastTabTickTime: TDateTime := 0;
                     var cnt := 0;
@@ -710,7 +718,7 @@ begin
     Result := TAuraDataFile.Create(path, symbol, fileExt, year, month);
 end;
 
-class function TAuraTABFileServer.ReadCompressedData(const InputStream: TStream): TArray<TDataPoint<TAskBidItem>>;
+class function TAuraTABFileServer.ReadCompressedData(const InputStream: TStream): TArray<TDataPoint<TAuraAskBidFileItem>>;
 var
     decompressionStream: TStream;
     localHeader: TZipHeader;
@@ -749,11 +757,11 @@ begin
     end;
 end;
 
-class function TAuraTABFileServer.ReadUncompressedData(const InputStream: TStream): TArray<TDataPoint<TAskBidItem>>;
+class function TAuraTABFileServer.ReadUncompressedData(const InputStream: TStream): TArray<TDataPoint<TAuraAskBidFileItem>>;
 type
     TFileRecord = packed record
         TimeStamp: TDateTime;
-        Data: TAskBidItem;
+        Data: TAuraAskBidFileItem;
     end;
 var
     fileSize: Int64;