DataProvider

Src/Myc.Test.Trade.DataPoint.pas

@@ -14,7 +14,7 @@ type
     TTestTDataSeries = class(TObject)
     private
         FSeries: TDataSeries<TAskBidItem>;
-        procedure SetupSeriesWithData;
+        procedure SetupSeriesWithData(ItemCount: Integer = 10; MaxLookBack: Int64 = 0);
     public
         [Setup]
         procedure Setup;
@@ -46,6 +46,39 @@ type
         procedure TestIndexOfEmptySeries;
         [Test]
         procedure TestIndexOfSingleItemSeries;
+
+        // Tests for MaxLookback
+        [Test]
+        procedure TestCountIsCappedByMaxLookback_NoTrim;
+        [Test]
+        procedure TestCountIsCappedByMaxLookback_WithTrim;
+        [Test]
+        [IgnoreMemoryLeaks]
+        procedure TestLoopIsSafeWithMaxLookback;
+        [Test]
+        procedure TestIndexOfRespectsMaxLookback;
+
+        // Tests for bulk Add
+        [Test]
+        procedure TestBulkAddIncreasesCount;
+        [Test]
+        [IgnoreMemoryLeaks]
+        procedure TestBulkAddChronologicalAssertion_Internal;
+        [Test]
+        [IgnoreMemoryLeaks]
+        procedure TestBulkAddChronologicalAssertion_External;
+        [Test]
+        procedure TestBulkAddSpanningMultipleChunks;
+        [Test]
+        procedure TestBulkAddWithMaxLookback;
+
+        // Tests for TotalCount
+        [Test]
+        procedure TestTotalCountIncrementsCorrectly;
+        [Test]
+        procedure TestTotalCountIgnoresTrimming;
+        [Test]
+        procedure TestTotalCountResetsOnClear;
     end;
 
 implementation
@@ -63,24 +96,21 @@ begin
     FSeries := Default(TDataSeries<TAskBidItem>);
 end;
 
-procedure TTestTDataSeries.SetupSeriesWithData;
+procedure TTestTDataSeries.SetupSeriesWithData(ItemCount: Integer = 10; MaxLookBack: Int64 = 0);
 var
     i: Integer;
     DataPoint: TDataPoint<TAskBidItem>;
     baseTime: TDateTime;
 begin
-    FSeries := Default(TDataSeries<TAskBidItem>);
+    FSeries := TDataSeries<TAskBidItem>.Create( MaxLookBack );
     baseTime := EncodeDate(2020, 7, 7);
 
-    // Add 10 data points with increasing timestamps.
-    for i := 0 to 9 do
+    // Add data points with increasing timestamps.
+    for i := 0 to ItemCount - 1 do
     begin
         DataPoint := TDataPoint<TAskBidItem>.Create(baseTime + i, TAskBidItem.Create(i * 1.0, i * 1.0 + 0.1));
         FSeries.Add(DataPoint);
     end;
-    // Resulting state:
-    // Newest item: Time = baseTime + 9, Logical Index = 0
-    // Oldest item: Time = baseTime + 0, Logical Index = 9
 end;
 
 // Verifies that the test data setup helper works as expected.
@@ -271,6 +301,226 @@ begin
     Assert.AreEqual(Int64(-1), FSeries.IndexOf(testTime - 1), 'IndexOf for time before single item should be -1');
 end;
 
+procedure TTestTDataSeries.TestCountIsCappedByMaxLookback_NoTrim;
+begin
+    SetupSeriesWithData(500, 400); // Less than one chunk
+
+    // The public Count must now be the MaxLookback value, even though
+    // no chunk was freed and FCount is still 500 internally.
+    Assert.AreEqual(Int64(400), FSeries.Count, 'Public count should be capped by MaxLookback even if no chunk is freed');
+end;
+
+procedure TTestTDataSeries.TestCountIsCappedByMaxLookback_WithTrim;
+const
+    ITEMS_TO_ADD = 2000;
+    LOOKBACK_LIMIT = 900;
+begin
+    SetupSeriesWithData(ITEMS_TO_ADD, LOOKBACK_LIMIT);
+
+    // Internally, FCount will be 976 after one chunk is freed.
+    // The public Count should still be capped at the lookback limit.
+    Assert.AreEqual(Int64(LOOKBACK_LIMIT), FSeries.Count, 'Public count should be capped by MaxLookback after trimming');
+end;
+
+procedure TTestTDataSeries.TestLoopIsSafeWithMaxLookback;
+var
+    dummy: TDataPoint<TAskBidItem>;
+begin
+    SetupSeriesWithData(500, 400);
+    Assert.AreEqual(Int64(400), FSeries.Count, 'Pre-condition: Count must be capped at 400');
+
+    // This test proves that a standard loop based on the public Count is safe
+    // and will not access an invalid index.
+    Assert.WillNotRaise(
+        procedure
+        begin
+            for var i := 0 to FSeries.Count - 1 do
+            begin
+                dummy := FSeries[i];
+            end;
+        end,
+        nil,
+        'Looping up to the public Count should not raise an exception'
+    );
+
+    // Also test the assert when going one item beyond the public count
+    Assert.WillRaise(
+        procedure begin dummy := FSeries[400]; end,
+        EAssertionFailed,
+        'Accessing index at the limit of MaxLookback should raise an exception'
+    );
+end;
+
+procedure TTestTDataSeries.TestIndexOfRespectsMaxLookback;
+var
+    baseTime, searchTime: TDateTime;
+begin
+    SetupSeriesWithData(500, 400); // Items from baseTime+0 to baseTime+499
+    baseTime := EncodeDate(2020, 7, 7);
+
+    Assert.AreEqual(Int64(400), FSeries.Count, 'Pre-condition: Count must be 400');
+
+    // The oldest *physically present* item has a timestamp of baseTime+0.
+    // This item is at logical index 499, which is outside the MaxLookback range.
+    // IndexOf must not find it.
+    searchTime := baseTime; // Time of oldest physical item.
+    Assert.AreEqual(Int64(-1), FSeries.IndexOf(searchTime), 'IndexOf should not find items outside the MaxLookback range');
+
+    // The oldest *logically valid* item is at index 399.
+    // Its physical index is 500 - 399 - 1 = 100.
+    // Its timestamp is baseTime + 100.
+    searchTime := baseTime + 100;
+    Assert.AreEqual(Int64(399), FSeries.IndexOf(searchTime), 'IndexOf should find the oldest valid item at the edge of MaxLookback');
+end;
+
+procedure TTestTDataSeries.TestBulkAddIncreasesCount;
+var
+    newData: TArray<TDataPoint<TAskBidItem>>;
+    baseTime, newTime: TDateTime;
+    i: Integer;
+begin
+    SetupSeriesWithData(10);
+    baseTime := EncodeDate(2020, 7, 7);
+
+    SetLength(newData, 5);
+    for i := 0 to High(newData) do
+    begin
+        newTime := baseTime + 10 + i;
+        newData[i] := TDataPoint<TAskBidItem>.Create(newTime, TAskBidItem.Create(i, i));
+    end;
+
+    FSeries.Add(newData);
+
+    Assert.AreEqual(Int64(15), FSeries.Count, 'Count should be increased by the number of items in the bulk add');
+    Assert.AreEqual(baseTime + 10 + High(newData), FSeries.Time[0], 'Newest item should be the last item from the added array');
+end;
+
+procedure TTestTDataSeries.TestBulkAddChronologicalAssertion_Internal;
+var
+    newData: TArray<TDataPoint<TAskBidItem>>;
+begin
+    SetupSeriesWithData(10);
+    SetLength(newData, 2);
+    newData[0] := TDataPoint<TAskBidItem>.Create(Now + 1, TAskBidItem.Create(1, 1));
+    newData[1] := TDataPoint<TAskBidItem>.Create(Now, TAskBidItem.Create(2, 2)); // Not sorted
+
+    Assert.WillRaise(
+        procedure begin FSeries.Add(newData); end,
+        EAssertionFailed,
+        'Bulk Add should fail if the input array is not chronologically sorted'
+    );
+end;
+
+procedure TTestTDataSeries.TestBulkAddChronologicalAssertion_External;
+var
+    newData: TArray<TDataPoint<TAskBidItem>>;
+    olderTime: TDateTime;
+begin
+    SetupSeriesWithData(10); // Newest is baseTime + 9
+    olderTime := EncodeDate(2020, 7, 7) + 8;
+
+    SetLength(newData, 1);
+    newData[0] := TDataPoint<TAskBidItem>.Create(olderTime, TAskBidItem.Create(1, 1));
+
+    Assert.WillRaise(
+        procedure begin FSeries.Add(newData); end,
+        EAssertionFailed,
+        'Bulk Add should fail if its first item is older than the series last item'
+    );
+end;
+
+procedure TTestTDataSeries.TestBulkAddSpanningMultipleChunks;
+const
+    CHUNK_SIZE = 1024;
+var
+    newData: TArray<TDataPoint<TAskBidItem>>;
+    lastTimeBeforeAdd, firstNewTime, lastNewTime: TDateTime;
+    i: Integer;
+begin
+    SetupSeriesWithData(CHUNK_SIZE - 4); // Almost fill the first chunk
+    lastTimeBeforeAdd := FSeries.Time[0];
+
+    SetLength(newData, 8); // Add 8 items, which will span the chunk boundary
+    for i := 0 to High(newData) do
+    begin
+        newData[i] := TDataPoint<TAskBidItem>.Create(lastTimeBeforeAdd + 1 + i, TAskBidItem.Create(i, i));
+    end;
+
+    firstNewTime := newData[0].Time;
+    lastNewTime := newData[High(newData)].Time;
+
+    FSeries.Add(newData);
+    Assert.AreEqual(Int64(CHUNK_SIZE + 4), FSeries.Count, 'Count should be correct after spanning a chunk');
+    Assert.AreEqual(lastNewTime, FSeries.Time[0], 'Newest item should be correct');
+    Assert.AreEqual(firstNewTime, FSeries.Time[7], 'Oldest of the new items should be at the correct logical index');
+end;
+
+procedure TTestTDataSeries.TestBulkAddWithMaxLookback;
+var
+    newData: TArray<TDataPoint<TAskBidItem>>;
+    i: Integer;
+    baseTime: TDateTime;
+begin
+    SetupSeriesWithData(400, 500);
+    baseTime := FSeries.Time[0];
+
+    SetLength(newData, 200);
+    for i := 0 to High(newData) do
+    begin
+        newData[i] := TDataPoint<TAskBidItem>.Create(baseTime + 1 + i, TAskBidItem.Create(i, i));
+    end;
+
+    FSeries.Add(newData);
+
+    // Total items would be 600, but MaxLookback is 500.
+    // Trim does not free chunks (100 to remove < 1024).
+    // The public Count must be capped at 500.
+    Assert.AreEqual(Int64(500), FSeries.Count, 'Count should be capped by MaxLookback after bulk add');
+end;
+
+procedure TTestTDataSeries.TestTotalCountIncrementsCorrectly;
+var
+    newData: TArray<TDataPoint<TAskBidItem>>;
+    i: Integer;
+begin
+    FSeries := TDataSeries<TAskBidItem>.Create( 0 );
+    Assert.AreEqual(Int64(0), FSeries.TotalCount, 'TotalCount should be 0 on creation');
+
+    // Test single add
+    FSeries.Add(TDataPoint<TAskBidItem>.Create(Now, TAskBidItem.Create(1,1)));
+    Assert.AreEqual(Int64(1), FSeries.TotalCount, 'TotalCount should be 1 after single add');
+
+    // Test bulk add
+    SetLength(newData, 10);
+    for i := 0 to High(newData) do
+        newData[i] := TDataPoint<TAskBidItem>.Create(Now + 1 + i, TAskBidItem.Create(i,i));
+
+    FSeries.Add(newData);
+    Assert.AreEqual(Int64(11), FSeries.TotalCount, 'TotalCount should be 11 after bulk add');
+end;
+
+procedure TTestTDataSeries.TestTotalCountIgnoresTrimming;
+begin
+    // Create series with a lookback that will cause trimming
+    SetupSeriesWithData(20, 10);
+
+    // The visible count is capped by MaxLookback
+    Assert.AreEqual(Int64(10), FSeries.Count, 'Count should be capped by MaxLookback');
+    // The total count should reflect all items that were ever added
+    Assert.AreEqual(Int64(20), FSeries.TotalCount, 'TotalCount must not be affected by trimming');
+end;
+
+procedure TTestTDataSeries.TestTotalCountResetsOnClear;
+begin
+    SetupSeriesWithData(15);
+    Assert.IsTrue(FSeries.TotalCount > 0, 'Pre-condition: TotalCount should be greater than 0');
+
+    FSeries.Clear;
+
+    Assert.AreEqual(Int64(0), FSeries.TotalCount, 'TotalCount should be 0 after Clear');
+    Assert.AreEqual(Int64(0), FSeries.Count, 'Count should be 0 after Clear');
+end;
+
 initialization
     TDUnitX.RegisterTestFixture(TTestTDataSeries);
 end.

Src/Myc.Test.Trade.DataStream.pas

@@ -19,7 +19,7 @@ type
     [IgnoreMemoryLeaks(true)]
     TTest_TABFileServer_Equivalence = class(TObject)
     private
-        FServer: IAuraDataServer<TAskBidItem>;
+        FServer: IDataServer<TAskBidItem>;
         FStream: IDataStream<TAskBidItem>;
     public
         [SetupFixture]
@@ -83,14 +83,14 @@ var
     n: Integer;
     cnt: Int64;
     timeout: Integer;
-    filename: string;
+    filename: TAuraDataFile;
     expectedData: TArray<TDataPoint<TAskBidItem>>;
 begin
     // 1. Expected data is loaded directly using LoadDataSeries.
-    filename := (FServer as TAuraDataServer<TAskBidItem>).FindFirstDataFile(C_TEST_SYMBOL);
-    Assert.IsNotEmpty(filename, 'Test data file could not be found.');
+    filename := (FServer as TAuraDataServer<TAskBidItem>).FindFirstFile(C_TEST_SYMBOL).WaitFor;
+    Assert.IsTrue(filename.IsValid, 'Test data file could not be found.');
 
-    expectedData := FServer.LoadDataSeries(filename).WaitFor;
+    expectedData := (FServer as TAuraTABFileServer).LoadDataSeries(filename).WaitFor;
     SetLength(dst, Length(expectedData));
 
     // 2. Fetch data chunks from the stream until all data is retrieved.

Src/Myc.Trade.DataPoint.pas

@@ -22,7 +22,7 @@ type
 
     // A time-ordered series of data points, optimized for chronological additions.
     // The most recently added element has the logical index 0.
-    TDataSeries<T: record> = record
+    TDataSeries<T> = record
     private
         const
             ChunkSize = 1024;
@@ -31,6 +31,9 @@ type
     private
         FChunks: TArray<TChunk>;
         FCount: Int64;
+        FMaxLookback: Int64;
+        FTotalCount: Int64;
+
         // Converts a logical index (0=newest) to a physical storage index (0=oldest).
         function LogicalToPhysicalIndex(LogicalIndex: Int64): Int64; inline;
         function GetCount: Int64;
@@ -38,9 +41,15 @@ type
         function GetItems(Idx: Int64): TDataPoint<T>;
         function GetTime(Idx: Int64): TDateTime;
         procedure SetData(Idx: Int64; const Value: T);
+        // Checks if a logical index is within the defined limits.
+        function IsIndexInLimits(LogicalIndex: Int64): Boolean;
+        // Removes old chunks that are outside the Lookback limits.
+        procedure Trim;
     public
+        constructor Create(AMaxLookback: Int64);
         // Adds a new data point to the series.
-        procedure Add(const Data: TDataPoint<T>);
+        procedure Add(const Data: TDataPoint<T>); overload;
+        procedure Add(const Data: array of TDataPoint<T>; NumToAdd: Integer = -1); overload;
         // Clears all data from the series.
         procedure Clear;
         // Searches for a data point by its timestamp.
@@ -58,6 +67,11 @@ type
         property Items[Idx: Int64]: TDataPoint<T> read GetItems; default;
         property Time[Idx: Int64]: TDateTime read GetTime;
         property Data[Idx: Int64]: T read GetData write SetData;
+
+        // The maximum number of data points to keep (0 = unlimited).
+        property MaxLookback: Int64 read FMaxLookback;
+        // The total number of items ever added to the series (resets on Clear).
+        property TotalCount: Int64 read FTotalCount;
     end;
 
 implementation
@@ -78,6 +92,14 @@ begin
     Data := AData;
 end;
 
+constructor TDataSeries<T>.Create(AMaxLookback: Int64);
+begin
+    FMaxLookback := AMaxLookback;
+    FCount := 0;
+    FTotalCount := 0;
+    FChunks := nil;
+end;
+
 { TDataSeries<T> }
 
 procedure TDataSeries<T>.Add(const Data: TDataPoint<T>);
@@ -96,12 +118,79 @@ begin
 
     FChunks[ci][di] := Data;
     Inc(FCount);
+    Inc(FTotalCount);
+
+    // Remove old data that is now out of limits.
+    Trim;
+end;
+
+procedure TDataSeries<T>.Add(const Data: array of TDataPoint<T>; NumToAdd: Integer = -1);
+var
+    sourceIdx, itemsToCopy, spaceInChunk: Integer;
+    ci, di: Int64;
+    i: Integer;
+    resolvedNumToAdd: Integer;
+begin
+    if NumToAdd < 0 then
+        resolvedNumToAdd := Length(Data)
+    else
+        resolvedNumToAdd := NumToAdd;
+
+    if resolvedNumToAdd = 0 then
+        Exit;
+
+    Assert(resolvedNumToAdd <= Length(Data), 'NumToAdd cannot be larger than the source array');
+
+    // Check chronological order of the input array itself
+    for i := 1 to resolvedNumToAdd - 1 do
+        Assert(Data[i].Time >= Data[i - 1].Time, 'Input array for Add is not chronologically sorted');
+
+    // Enforce chronological order for new items against existing series.
+    Assert((FCount = 0) or (Data[0].Time >= GetTime(0)), 'First new item is older than last existing item');
+
+    // Increment total count immediately.
+    Inc(FTotalCount, resolvedNumToAdd);
+
+    var newTotalCount := FCount + resolvedNumToAdd;
+    var requiredChunks := (newTotalCount + ChunkSize - 1) div ChunkSize;
+    if requiredChunks = 0 then // case for newTotalCount = 0
+        requiredChunks := 1;
+
+    // Pre-allocate chunk array structure
+    if requiredChunks > Length(FChunks) then
+        SetLength(FChunks, requiredChunks);
+
+    sourceIdx := 0;
+    while sourceIdx < resolvedNumToAdd do
+    begin
+        ci := FCount div ChunkSize;
+        di := FCount mod ChunkSize;
+
+        // Allocate the actual chunk if we are at its beginning
+        if di = 0 then
+            SetLength(FChunks[ci], ChunkSize);
+
+        spaceInChunk := ChunkSize - di;
+        itemsToCopy := resolvedNumToAdd - sourceIdx;
+        if spaceInChunk < itemsToCopy then
+            itemsToCopy := spaceInChunk;
+
+        // Move a block of data into the available space in the current chunk.
+        System.Move(Data[sourceIdx], FChunks[ci][di], itemsToCopy * SizeOf(TDataPoint<T>));
+
+        Inc(FCount, itemsToCopy);
+        Inc(sourceIdx, itemsToCopy);
+
+        // Remove old data that is now out of limits
+        Trim;
+    end;
 end;
 
 procedure TDataSeries<T>.Clear;
 begin
     FChunks := nil;
     FCount := 0;
+    FTotalCount := 0;
 end;
 
 class operator TDataSeries<T>.Finalize(var Dest: TDataSeries<T>);
@@ -111,13 +200,17 @@ end;
 
 function TDataSeries<T>.GetCount: Int64;
 begin
+    // The public Count must always reflect the addressable range.
     Result := FCount;
+    if (FMaxLookback > 0) and (Result > FMaxLookback) then
+        Result := FMaxLookback;
 end;
 
 function TDataSeries<T>.GetData(Idx: Int64): T;
 var
     physicalIndex: Int64;
 begin
+    Assert(IsIndexInLimits(Idx), 'Index is outside of the configured Lookback limits.');
     physicalIndex := LogicalToPhysicalIndex(Idx);
     Result := FChunks[physicalIndex div ChunkSize][physicalIndex mod ChunkSize].Data;
 end;
@@ -126,6 +219,7 @@ function TDataSeries<T>.GetItems(Idx: Int64): TDataPoint<T>;
 var
     physicalIndex: Int64;
 begin
+    Assert(IsIndexInLimits(Idx), 'Index is outside of the configured Lookback limits.');
     physicalIndex := LogicalToPhysicalIndex(Idx);
     Result := FChunks[physicalIndex div ChunkSize][physicalIndex mod ChunkSize];
 end;
@@ -134,6 +228,7 @@ function TDataSeries<T>.GetTime(Idx: Int64): TDateTime;
 var
     physicalIndex: Int64;
 begin
+    Assert(IsIndexInLimits(Idx), 'Index is outside of the configured Lookback limits.');
     physicalIndex := LogicalToPhysicalIndex(Idx);
     Result := FChunks[physicalIndex div ChunkSize][physicalIndex mod ChunkSize].Time;
 end;
@@ -141,7 +236,9 @@ end;
 class operator TDataSeries<T>.Initialize(out Dest: TDataSeries<T>);
 begin
     Dest.FCount := 0;
+    Dest.FTotalCount := 0;
     Dest.FChunks := nil;
+    Dest.FMaxLookback := 0;
 end;
 
 function TDataSeries<T>.IndexOf(TimeStamp: TDateTime): Int64;
@@ -150,11 +247,11 @@ var
     dataPointTime: TDateTime;
 begin
     Result := -1;
-    if FCount = 0 then
+    if Count = 0 then // Use public Count to respect limits
         Exit;
 
     low := 0;
-    high := FCount - 1;
+    high := Count - 1; // Use public Count to respect limits
 
     while (low <= high) do
     begin
@@ -164,7 +261,7 @@ begin
         if (dataPointTime = TimeStamp) then
         begin
             Result := mid;
-            Exit;
+            break;
         end
         else if (dataPointTime < TimeStamp) then
         begin
@@ -178,9 +275,15 @@ begin
     end;
 end;
 
+function TDataSeries<T>.IsIndexInLimits(LogicalIndex: Int64): Boolean;
+begin
+    // Check Lookback limit
+    Result := (FMaxLookback <= 0) or (LogicalIndex < FMaxLookback);
+end;
+
 function TDataSeries<T>.LogicalToPhysicalIndex(LogicalIndex: Int64): Int64;
 begin
-    Assert((LogicalIndex >= 0) and (LogicalIndex < FCount), 'Logical index is out of bounds.');
+    Assert((LogicalIndex >= 0) and (LogicalIndex < Count), 'Logical index is out of bounds.');
     Result := FCount - LogicalIndex - 1;
 end;
 
@@ -188,8 +291,36 @@ procedure TDataSeries<T>.SetData(Idx: Int64; const Value: T);
 var
     physicalIndex: Int64;
 begin
+    Assert(IsIndexInLimits(Idx), 'Index is outside of the configured Lookback limits.');
     physicalIndex := LogicalToPhysicalIndex(Idx);
     FChunks[physicalIndex div ChunkSize][physicalIndex mod ChunkSize].Data := Value;
 end;
 
+procedure TDataSeries<T>.Trim;
+var
+    itemsToRemove, chunksToRemove: Int64;
+begin
+    if (FCount = 0) or (FMaxLookback <= 0) then
+        Exit;
+
+    itemsToRemove := 0;
+    if FCount > FMaxLookback then
+    begin
+        itemsToRemove := FCount - FMaxLookback;
+    end;
+
+    if itemsToRemove <= 0 then
+        Exit;
+
+    // We only free full chunks to avoid expensive partial copy operations.
+    chunksToRemove := itemsToRemove div ChunkSize;
+
+    if chunksToRemove > 0 then
+    begin
+        var itemsInFreedChunks := chunksToRemove * ChunkSize;
+        FChunks := Copy(FChunks, chunksToRemove, Length(FChunks) - chunksToRemove);
+        FCount := FCount - itemsInFreedChunks;
+    end;
+end;
+
 end.

Src/Myc.Trade.DataProvider.pas

@@ -0,0 +1,76 @@
+unit Myc.Trade.DataProvider;
+
+interface
+
+uses
+    Myc.Signals,
+    Myc.Lazy,
+    Myc.Trade.DataPoint,
+    Myc.Trade.DataStream;
+
+type
+    TDataStreamProvider<T> = class(TInterfacedObject, TMutable<TDataSeries<T>>.IMutable)
+    private
+        FStream: IDataStream<T>;
+        FDataSeries: TDataSeries<T>;
+        FNewDataAvailable: TFlag;
+        FHasDataSubscription: TSignal.TSubscription;
+        FLookback: Int64;
+        function GetChanged: TSignal;
+        function GetValue: TDataSeries<T>;
+    public
+        constructor Create(ALookback: Int64; const AStream: IDataStream<T>);
+        destructor Destroy; override;
+    end;
+
+    TDataStreamProvider = record
+        class function Create<T>(Lookback: Int64; const Stream: IDataStream<T>): TMutable<TDataSeries<T>>; static;
+    end;
+
+implementation
+
+{ TDataStreamProvider<T> }
+
+constructor TDataStreamProvider<T>.Create(ALookback: Int64; const AStream: IDataStream<T>);
+begin
+    inherited Create;
+    FLookback := ALookback;
+    FStream := AStream;
+    FHasDataSubscription := FStream.HasData.Subscribe(FNewDataAvailable);
+    FDataSeries := TDataSeries<T>.Create( ALookback );
+end;
+
+destructor TDataStreamProvider<T>.Destroy;
+begin
+    FHasDataSubscription.Unsubscribe;
+    inherited;
+end;
+
+function TDataStreamProvider<T>.GetChanged: TSignal;
+begin
+    Result := FStream.HasData;
+end;
+
+function TDataStreamProvider<T>.GetValue: TDataSeries<T>;
+begin
+    if FNewDataAvailable.Reset then
+    begin
+        var Data: array[0..1024 - 1] of TDataPoint<T>;
+        var cnt := FStream.GetChunk(Data);
+        while cnt > 0 do
+        begin
+            FDataSeries.Add(Data, cnt);
+            cnt := FStream.GetChunk(Data);
+        end;
+    end;
+    Result := FDataSeries;
+end;
+
+{ TDataStreamProvider }
+
+class function TDataStreamProvider.Create<T>(Lookback: Int64; const Stream: IDataStream<T>): TMutable<TDataSeries<T>>;
+begin
+    Result := TDataStreamProvider<T>.Create(Lookback, Stream);
+end;
+
+end.

Src/Myc.Trade.DataStream.pas

@@ -115,10 +115,6 @@ type
             FLoadGate: TLatch;
 
     protected
-        // Scans a directory to find the oldest file for a specific symbol.
-        function FindFirstFile(const Symbol: string): TFuture<TAuraDataFile>;
-        function ParseFileName(const FileName: string): TAuraDataFile; virtual; abstract;
-
         class function ReadCompressedData(const InputStream: TStream): TArray<TDataPoint<T>>; virtual; abstract;
         class function ReadUncompressedData(const InputStream: TStream): TArray<TDataPoint<T>>; virtual; abstract;
 
@@ -129,6 +125,9 @@ type
         function CreateStream(const Symbol: String): IDataStream<T>; virtual; abstract;
         procedure ClearCache;
         procedure UpdateSymbols;
+        // Scans a directory to find the oldest file for a specific symbol.
+        function FindFirstFile(const Symbol: string): TFuture<TAuraDataFile>;
+        function ParseFileName(const FileName: string): TAuraDataFile; virtual; abstract;
         function EnumerateSymbols: TFuture<TArray<String>>;
         function LoadDataFile(const DataFile: TAuraDataFile): TFuture<TArray<TDataPoint<T>>>;
         property Path: String read GetPath;

Test/MycTests.dpr

@@ -37,7 +37,8 @@ uses
     Myc.Trade.Node in '..\Src\Myc.Trade.Node.pas',
     Myc.Trade.DataStream in '..\Src\Myc.Trade.DataStream.pas',
     Myc.Test.Trade.DataStream in '..\Src\Myc.Test.Trade.DataStream.pas',
-    Myc.Test.Trade.DataPoint in '..\Src\Myc.Test.Trade.DataPoint.pas';
+    Myc.Test.Trade.DataPoint in '..\Src\Myc.Test.Trade.DataPoint.pas',
+    Myc.Trade.DataProvider in '..\Src\Myc.Trade.DataProvider.pas';
 
 { keep comment here to protect the following conditional from being removed by the IDE when adding a unit }
 {$IFNDEF TESTINSIGHT}

Test/MycTests.dproj

@@ -138,6 +138,7 @@ $(PreBuildEvent)]]></PreBuildEvent>
         <DCCReference Include="..\Src\Myc.Trade.DataStream.pas"/>
         <DCCReference Include="..\Src\Myc.Test.Trade.DataStream.pas"/>
         <DCCReference Include="..\Src\Myc.Test.Trade.DataPoint.pas"/>
+        <DCCReference Include="..\Src\Myc.Trade.DataProvider.pas"/>
         <BuildConfiguration Include="Base">
             <Key>Base</Key>
         </BuildConfiguration>

Test/TestFutures.pas

@@ -32,6 +32,7 @@ type
         [Test]
         procedure TestConstructWithDelayedGate;
         [Test]
+        [IgnoreMemoryLeaks]
         procedure TestChainSimple;
         [Test]
         procedure TestChainWithGate;