From 342eb07c42e11cab6d54332c32cc2df8d10b7614 Mon Sep 17 00:00:00 2001
From: Michael Schimmel <schimmel@mycelium.de>
Date: Wed, 16 Jul 2025 14:12:07 +0200
Subject: [PATCH] Fixed concurrent processing

---
 AuraTrader/MainForm.pas          | 44 ++++++++++--------
 Src/Myc.Fmx.Chart.Series.pas     | 38 ++-------------
 Src/Myc.Fmx.Chart.pas            | 51 +++++++++-----------
 Src/Myc.TaskManager.pas          | 20 +++++++-
 Src/Myc.Trade.DataPoint.Impl.pas | 80 +++++++++++++++++++++++++++-----
 Src/Myc.Trade.DataPoint.pas      | 27 +++++++++++
 Src/Myc.Trade.DataStream.pas     | 36 ++++++++++----
 Src/Myc.Trade.Types.pas          |  1 +
 8 files changed, 192 insertions(+), 105 deletions(-)

diff --git a/AuraTrader/MainForm.pas b/AuraTrader/MainForm.pas
index 5997f3d..069099a 100644
--- a/AuraTrader/MainForm.pas
+++ b/AuraTrader/MainForm.pas
@@ -108,6 +108,7 @@ type
     TEquitySum = class(TMycConverter<Double, Double>)
     private
         FEquity: Double;
+        FInit: Boolean;
     protected
         function ProcessData(const Value: Double): TState; override;
     public
@@ -360,12 +361,12 @@ begin
     var Ohlc := OhlcPoint.Field<TOhlcItem>('Data');
     var Closes := Ohlc.Field<Double>('Close');
 
-    var Hull := Closes.Chain<Double>(TIndicators.CreateHMA(150));
+    var Hull := Closes.MakeParallel.Chain<Double>(TIndicators.CreateHMA(150));
     var Sma := Closes.Chain<Double>(TIndicators.CreateSMA(50));
     var Ema := Closes.Chain<Double>(TIndicators.CreateEMA(21));
-    var Boli := Closes.Chain<TBollingerBandsResult>(TIndicators.CreateBollingerBands(20, 2.0));
-    var Rsi := Closes.Chain<Double>(TIndicators.CreateRSI(14));
-    var Macd := Closes.Chain<TMacdResult>(TIndicators.CreateMACD(12, 26, 9));
+    var Boli := Closes.MakeParallel.Chain<TBollingerBandsResult>(TIndicators.CreateBollingerBands(20, 2.0));
+    var Rsi := Closes.MakeParallel.Chain<Double>(TIndicators.CreateRSI(14));
+    var Macd := Closes.MakeParallel.Chain<TMacdResult>(TIndicators.CreateMACD(12, 26, 9));
     var Stoch := Ohlc.Chain<TStochasticResult>(TIndicators.CreateStochastic(14, 3));
 
     chart.SetXAxisSeries(timeframe, Timestamps.Sender);
@@ -429,7 +430,7 @@ type
         pnl: Double;
     end;
 begin
-    var timeframe := TTimeframe.M15;
+    var timeframe := TTimeframe.D;
 
     var ticker := TConverter.CreateTicker<TDataPoint<TAskBidItem>>;
 
@@ -440,37 +441,35 @@ begin
             )
         );
 
-    var OhlcPoint := lastPrice.Chain<TDataPoint<TOhlcItem>>(TConverter.CreateAggregation(timeframe));
+    var OhlcPoint := lastPrice.Chain<TDataPoint<TOhlcItem>>(TConverter.CreateAggregation(timeframe)).MakeParallel;
 
     var Ohlc := TConverter.CreateSequence<TOhlcItem>(2, OhlcPoint.Field<TOhlcItem>('Data').Sender);
 
     var Closes := Ohlc[0].Field<Double>('Close');
 
-    var Hull := Closes.Chain<Double>(TIndicators.CreateHMA(250));
-    var Sma := Closes.Chain<Double>(TIndicators.CreateSMA(200));
-
-    var HullSeries := TConverter.CreateEndpoint<Double>(Hull.Sender, 5);
-    var SmaSeries := TConverter.CreateEndpoint<Double>(Sma.Sender, 5);
+    var Hull := Closes.Chain<Double>(TIndicators.CreateHMA(250)).MakeParallel;
+    var Sma := Closes.Chain<Double>(TIndicators.CreateSMA(200)).MakeParallel;
 
     var Lowest: Double := Double.MaxValue;
     var Highest: Double := Double.MinValue;
 
+    var ATR := Ohlc[0].Chain<Double>(TIndicators.CreateATR(15));
+    var ATRSeries := TConverter.CreateEndpoint<Double>(ATR.Sender, 5);
+
+    var HullSeries := TConverter.CreateEndpoint<Double>(Hull.Sender, 5);
+    var SmaSeries := TConverter.CreateEndpoint<Double>(Sma.Sender, 5);
+
+    // next stage
+
     var curr: TSignal;
     curr.SL := Double.NaN;
     curr.Entry := Double.NaN;
 
-    var ATR := Ohlc[0].Chain<Double>(TIndicators.CreateATR(50));
-    var ATRSeries := TConverter.CreateEndpoint<Double>(ATR.Sender, 5);
-
-    // next stage
-
     var Signal :=
         Ohlc[1]
             .Chain<TSignal>(
                 function(const Ohlc: TOhlcItem): TSignal
                 begin
-                    var pnl: Double := 0;
-
                     if Ohlc.Low < Lowest then
                         Lowest := Ohlc.Low;
                     if Ohlc.High > Highest then
@@ -478,7 +477,7 @@ begin
 
                     Result := curr;
                     Result.Sig := 0;
-                    pnl := NaN;
+                    var pnl: double := NaN;
 
                     if (HullSeries.Value[0] < SmaSeries.Value[0]) and (HullSeries.Value[1] >= SmaSeries.Value[1]) then
                     begin
@@ -600,10 +599,17 @@ constructor TEquitySum.Create(AEquity: Double);
 begin
     inherited Create;
     FEquity := AEquity;
+    FInit := false;
 end;
 
 function TEquitySum.ProcessData(const Value: Double): TState;
 begin
+    if not FInit then
+    begin
+        FInit := true;
+        Broadcast(FEquity);
+    end;
+
     if not IsNan(Value) then
     begin
         FEquity := FEquity + Value;
diff --git a/Src/Myc.Fmx.Chart.Series.pas b/Src/Myc.Fmx.Chart.Series.pas
index cd428a3..798e690 100644
--- a/Src/Myc.Fmx.Chart.Series.pas
+++ b/Src/Myc.Fmx.Chart.Series.pas
@@ -15,27 +15,13 @@ uses
     Myc.Fmx.Chart;
 
 type
-    TChartSeriesReceiver<T> = class(TMycProcessor<T>)
-    strict private
-        FCurrData: TSeries<T>;
-        FLookback: TMutable<Int64>;
-    private
-        FData: TWriteable<TSeries<T>>;
-    protected
-        function ProcessData(const Value: T): TState; override;
-    public
-        constructor Create(const ALookback: TMutable<Int64>);
-        property Data: TWriteable<TSeries<T>> read FData;
-    end;
-
     TChartSeriesProcessor<T> = class(TMycChart.TSeries)
     strict private
         FDataSeries: TSeries<T>;
     private
         FData: TSeries<T>;
         FDataProvider: TDataProvider<T>;
-        FReceiver: TChartSeriesReceiver<T>;
-        FReceiverTag: TDataProvider<T>.TTag;
+        FReceiver: TMutable<TSeries<T>>;
     protected
         function GetCount: Int64; override;
         function GetTotalCount: Int64; override;
@@ -140,22 +126,6 @@ uses
     System.Math,
     FMX.Types;
 
-{ TChartSeriesReceiver<T> }
-
-constructor TChartSeriesReceiver<T>.Create(const ALookback: TMutable<Int64>);
-begin
-    inherited Create;
-    FLookback := ALookback;
-    FData := TWriteable<TSeries<T>>.CreateWriteable(FCurrData).Protect;
-end;
-
-function TChartSeriesReceiver<T>.ProcessData(const Value: T): TState;
-begin
-    Result := TState.Null;
-    FCurrData := FCurrData.Add(Value, FLookback.Value);
-    FData.Value := FCurrData;
-end;
-
 { TChartSeriesProcessor<T> }
 
 constructor TChartSeriesProcessor<T>.Create(const ADataProvider: TDataProvider<T>; const ALookback: TMutable<Int64>);
@@ -164,13 +134,11 @@ begin
     FDataProvider := ADataProvider;
     FData := FDataSeries;
 
-    FReceiver := TChartSeriesReceiver<T>.Create(ALookback);
-    FReceiverTag := FDataProvider.Link(FReceiver);
+    FReceiver := TConverter.CreateEndpoint<T>(FDataProvider, ALookback.Value);
 end;
 
 destructor TChartSeriesProcessor<T>.Destroy;
 begin
-    FDataProvider.Unlink(FReceiverTag);
     inherited;
 end;
 
@@ -186,7 +154,7 @@ end;
 
 procedure TChartSeriesProcessor<T>.Update;
 begin
-    FData := FReceiver.Data.Value;
+    FData := FReceiver.Value;
 end;
 
 { TChartLineLayer }
diff --git a/Src/Myc.Fmx.Chart.pas b/Src/Myc.Fmx.Chart.pas
index eb69e03..b0a27bd 100644
--- a/Src/Myc.Fmx.Chart.pas
+++ b/Src/Myc.Fmx.Chart.pas
@@ -64,18 +64,19 @@ type
                 constructor Create(AParent: TPanel);
             end;
 
+            TDragPoint = record
+                Point: TPointF;
+                Idx: Int64;
+                BarX: Single;
+            end;
+
             TAxisLayer = class abstract(TLayer)
             private
                 FOwner: TMycChart;
             protected
                 function GetOwner: TMycChart; override; final;
                 // Paint crosshair and other axis related overlays
-                procedure Paint(
-                    const Canvas: TCanvas;
-                    const Viewport: TRectF;
-                    ViewStartIndex, ViewCount: Int64;
-                    IdxAtMousePos: Integer
-                ); virtual; abstract;
+                procedure Paint(const Canvas: TCanvas; const Viewport: TRectF; const MousePos: TDragPoint); virtual; abstract;
             public
                 constructor Create(AOwner: TMycChart);
             end;
@@ -83,12 +84,7 @@ type
             TXAxisLayer = class(TAxisLayer)
             protected
                 // Paint crosshair and time caption
-                procedure Paint(
-                    const Canvas: TCanvas;
-                    const Viewport: TRectF;
-                    ViewStartIndex, ViewCount: Int64;
-                    IdxAtMousePos: Integer
-                ); override;
+                procedure Paint(const Canvas: TCanvas; const Viewport: TRectF; const MousePos: TDragPoint); override;
                 // This function delivers the text for the caption.
                 function GetCaption(Idx: Int64): String; virtual; abstract;
             end;
@@ -124,11 +120,6 @@ type
                 property Weight: Single read FWeight write SetWeight;
             end;
 
-            TDragPoint = record
-                Point: TPointF;
-                Idx: Int64;
-            end;
-
     private
         FPanelList: TObjectList<TPanel>;
         FXAxisSeries: TMycChart.TXAxisLayer;
@@ -175,6 +166,8 @@ type
         property Panels[Index: Integer]: TPanel read GetPanel; default;
         // The rectangle occupied by the jump-to-latest button, used for hit testing.
         property JumpButtonRect: TRectF read FJumpButtonRect;
+        property ViewCount: Int64 read FViewCount;
+        property ViewStartIndex: Int64 read FViewStartIndex;
     end;
 
 implementation
@@ -232,11 +225,16 @@ end;
 
 function TMycChart.CreateDragPoint(X, Y: Single): TDragPoint;
 begin
+    var rect := LocalRect;
+
     Result.Point.X := X;
     Result.Point.Y := Y;
-    Result.Idx := Round((LocalRect.Right - X) * (FViewCount - 1) / LocalRect.Width);
+    Result.Idx := Round((rect.Right - X) * (FViewCount - 1) / rect.Width);
     if (Result.Idx < 0) or (Result.Idx >= FXAxisSeries.Series.Count) then
         Result.Idx := -1;
+    Result.BarX := NaN;
+    if Result.Idx >= 0 then
+        Result.BarX := rect.Right - (Result.Idx * (FViewCount - 1)) / (FViewCount - 1) * (rect.Width / (FViewCount - 1));
 end;
 
 function TMycChart.GetPanel(Index: Integer): TPanel;
@@ -356,7 +354,7 @@ begin
     begin
         // Do not draw crosshair if mouse is over the jump button
         if not FJumpButtonRect.Contains(FMousePos.Point) then
-            FXAxisSeries.Paint(Self.Canvas, rect, FViewStartIndex, FViewCount, FViewStartIndex + FMousePos.Idx);
+            FXAxisSeries.Paint(Self.Canvas, rect, FMousePos);
     end;
 
     // --- Draw the "jump to latest" button (code unchanged) ---
@@ -839,12 +837,7 @@ end;
 
 { TMycChart.TXAxisLayer }
 
-procedure TMycChart.TXAxisLayer.Paint(
-    const Canvas: TCanvas;
-    const Viewport: TRectF;
-    ViewStartIndex, ViewCount: Int64;
-    IdxAtMousePos: Integer
-);
+procedure TMycChart.TXAxisLayer.Paint(const Canvas: TCanvas; const Viewport: TRectF; const MousePos: TDragPoint);
 var
     caption: string;
     captionRect: TRectF;
@@ -853,11 +846,11 @@ var
     textSize: TRectF;
 begin
     // Do not draw if no data or view is invalid
-    if (not Assigned(Series)) or (Series.Count = 0) or (ViewCount <= 1) then
+    if (not Assigned(Series)) or (Series.Count = 0) then
         exit;
 
-    // Snap X to the candle's center
-    candleX := Viewport.Right - ((IdxAtMousePos - ViewStartIndex) * (ViewCount - 1)) / (ViewCount - 1) * (Viewport.Width / (ViewCount - 1));
+    // 1. Snap X to the candle's center
+    candleX := MousePos.BarX;
 
     // 2. Draw the vertical line at the snapped position
     Canvas.Stroke.Kind := TBrushKind.Solid;
@@ -866,7 +859,7 @@ begin
     Canvas.DrawLine(PointF(candleX, Viewport.Top), PointF(candleX, Viewport.Bottom), 1.0);
 
     // 3. Get the caption for the index
-    caption := GetCaption(IdxAtMousePos);
+    caption := GetCaption(FOwner.ViewStartIndex + MousePos.Idx);
     if caption.IsEmpty then
         exit;
 
diff --git a/Src/Myc.TaskManager.pas b/Src/Myc.TaskManager.pas
index b8dc021..cb05cec 100644
--- a/Src/Myc.TaskManager.pas
+++ b/Src/Myc.TaskManager.pas
@@ -29,7 +29,10 @@ type
 
         // Run a task when the gate is opened.
         // Returns a State to await thread completion.
-        function RunTask(const Gate: TState; const Proc: TFunc<TState>): TState;
+        function RunTask(const Proc: TFunc<TState>): TState; overload;
+        function RunTask(const Gate: TState; const Proc: TFunc<TState>): TState; overload;
+        procedure RunTask(const Gate: TState; const Proc: TProc); overload;
+        procedure RunTask(const Proc: TProc); overload;
 
         class function RunSequence(const Gate: TState; First, Count: Integer; const Proc: TFunc<Integer, TState>): TState; static;
 
@@ -181,6 +184,21 @@ begin
     );
 end;
 
+function TTaskManager.RunTask(const Proc: TFunc<TState>): TState;
+begin
+    Result := RunTask(TState.Null, Proc);
+end;
+
+procedure TTaskManager.RunTask(const Proc: TProc);
+begin
+    RunTask(TState.Null, Proc);
+end;
+
+procedure TTaskManager.RunTask(const Gate: TState; const Proc: TProc);
+begin
+    FTaskManager.RunTask(Gate, Proc);
+end;
+
 procedure TTaskManager.WaitFor(const State: TState);
 begin
     FTaskManager.WaitFor(State);
diff --git a/Src/Myc.Trade.DataPoint.Impl.pas b/Src/Myc.Trade.DataPoint.Impl.pas
index c567d84..c1107e4 100644
--- a/Src/Myc.Trade.DataPoint.Impl.pas
+++ b/Src/Myc.Trade.DataPoint.Impl.pas
@@ -4,6 +4,7 @@ interface
 
 uses
     System.SysUtils,
+    System.SyncObjs,
     Myc.Signals,
     Myc.Mutable,
     Myc.Core.Notifier,
@@ -15,6 +16,7 @@ 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;
 
@@ -86,7 +88,16 @@ type
         constructor Create(const AFunc: TConstFunc<S, T>);
     end;
 
-    // A converter specialized for calculating indicators.
+    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;
@@ -126,14 +137,12 @@ type
 
     // A generic processor implementation that uses a function reference.
     TMycGenericProcessor<T> = class(TMycProcessor<T>)
-        type
-            TProc = reference to function(const Value: T): TState;
     private
-        FProc: TProc;
+        FProc: TConstFunc<T, TState>;
     protected
         function ProcessData(const Value: T): TState; override; final;
     public
-        constructor Create(const AProc: TProc);
+        constructor Create(const AProc: TConstFunc<T, TState>);
     end;
 
     // A processor implementation that is owned by a controller.
@@ -156,6 +165,7 @@ type
         FLookback: Int64;
         FData: TSeries<T>;
         FChanged: TEvent;
+        FLock: TLightweightMREW;
         function GetChanged: TSignal;
         function GetValue: TSeries<T>;
         function ProcessData(const Value: T): TState;
@@ -178,6 +188,13 @@ type
         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;
+
 implementation
 
 uses
@@ -187,6 +204,14 @@ uses
     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);
@@ -314,7 +339,7 @@ end;
 
 function TMycIndicator<S, T>.ProcessData(const Value: S): TState;
 begin
-    Result := Broadcast(Calculate(Value));
+    Result := Execute(Value, function(const Value: S): TState begin Result := Broadcast(Calculate(Value)); end);
 end;
 
 { TMycDataCounter<T> }
@@ -385,7 +410,7 @@ end;
 
 { TMycGenericProcessor<T> }
 
-constructor TMycGenericProcessor<T>.Create(const AProc: TProc);
+constructor TMycGenericProcessor<T>.Create(const AProc: TConstFunc<T, TState>);
 begin
     inherited Create;
     FProc := AProc;
@@ -419,6 +444,7 @@ begin
 
     FProcessor := TMycContainedProcessor<T>.Create(Self, ProcessData);
     FTag := FDataProvider.Link(FProcessor);
+
 end;
 
 destructor TMycDataEndpoint<T>.Destroy;
@@ -435,15 +461,24 @@ end;
 
 function TMycDataEndpoint<T>.GetValue: TSeries<T>;
 begin
-    Result := FData;
+    FLock.BeginRead;
+    try
+        Result := FData;
+    finally
+        FLock.EndRead;
+    end;
 end;
 
 function TMycDataEndpoint<T>.ProcessData(const Value: T): TState;
 begin
     Result := TState.Null;
-
-    // Add new data point, respecting the lookback period.
-    FData := FData.Add(Value, FLookback);
+    FLock.BeginWrite;
+    try
+        // Add new data point, respecting the lookback period.
+        FData := FData.Add(Value, FLookback);
+    finally
+        FLock.EndWrite;
+    end;
     FChanged.Notify;
 end;
 
@@ -599,4 +634,27 @@ 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;
+
 end.
diff --git a/Src/Myc.Trade.DataPoint.pas b/Src/Myc.Trade.DataPoint.pas
index 8725cba..3e0c07b 100644
--- a/Src/Myc.Trade.DataPoint.pas
+++ b/Src/Myc.Trade.DataPoint.pas
@@ -82,9 +82,13 @@ type
         class operator Implicit(const A: TConverter<S, T>): IConverter; overload;
 
         class function CreateGeneric(const Func: TConstFunc<S, T>): TConverter<S, T>; static;
+        class function CreateParallel(const Func: TConstFunc<S, T>): TConverter<S, T>; static;
 
         function Chain<R>(const Next: TConverter<T, R>): TConverter<T, R>; overload; inline;
         function Chain<R>(const Func: TConstFunc<T, R>): TConverter<T, R>; overload; inline;
+        function ChainParallel<R>(const Func: TConstFunc<T, R>): TConverter<T, R>; overload; inline;
+
+        function MakeParallel: TConverter<T, T>; overload; inline;
 
         // Extracts the field of a record by it's name (using RTTI).
         function Field<R>(const FieldName: String): TConverter<T, R>; overload; inline;
@@ -112,6 +116,8 @@ type
         class function CreateDataPointConverter<S, T>(const Func: TConstFunc<S, T>): TConverter<TDataPoint<S>, TDataPoint<T>>; static;
 
         class function CreateSequence<T>(Count: Integer; const Parent: TDataProvider<T>): TArray<TConverter<T, T>>; overload; static;
+
+        class function Parallel<T>(Parent: TDataProvider<T>): TConverter<T, T>; static;
     end;
 
 implementation
@@ -184,11 +190,21 @@ begin
     Result := Chain<R>(TMycGenericConverter<T, R>.Create(Func));
 end;
 
+function TConverter<S, T>.ChainParallel<R>(const Func: TConstFunc<T, R>): TConverter<T, R>;
+begin
+    Result := Chain<R>(TMycGenericParallelConverter<T, R>.Create(Func));
+end;
+
 class function TConverter<S, T>.CreateGeneric(const Func: TConstFunc<S, T>): TConverter<S, T>;
 begin
     Result := TMycGenericConverter<S, T>.Create(Func);
 end;
 
+class function TConverter<S, T>.CreateParallel(const Func: TConstFunc<S, T>): TConverter<S, T>;
+begin
+    Result := TMycGenericParallelConverter<S, T>.Create(Func);
+end;
+
 function TConverter<S, T>.Field<R>(const FieldName: String): TConverter<T, R>;
 begin
     Result := Chain<R>(TConverter.CreateRecordField<T, R>(FieldName));
@@ -199,6 +215,11 @@ begin
     Result := FConverter.Sender;
 end;
 
+function TConverter<S, T>.MakeParallel: TConverter<T, T>;
+begin
+    Result := Chain<T>(TMycGenericParallelConverter<T, T>.Create(function(const Value: T): T begin Result := Value; end));
+end;
+
 function TConverter<S, T>.Sequence(Count: Integer): IMycDataSequence<T>;
 begin
     Result := TMycSequence<T>.Create(Count);
@@ -288,4 +309,10 @@ begin
     Parent.Link(seq);
 end;
 
+class function TConverter.Parallel<T>(Parent: TDataProvider<T>): TConverter<T, T>;
+begin
+    Result := TMycParallelConverter<T>.Create;
+    Parent.Link(Result);
+end;
+
 end.
diff --git a/Src/Myc.Trade.DataStream.pas b/Src/Myc.Trade.DataStream.pas
index 635361b..d053857 100644
--- a/Src/Myc.Trade.DataStream.pas
+++ b/Src/Myc.Trade.DataStream.pas
@@ -60,6 +60,12 @@ type
         // Used by cache to actually load an uncached file.
         function DoLoad(const FileName: string): TFuture<TArray<TArray<TDataPoint<T>>>>;
 
+        function ProcessChunks(
+            const DataChunks: TArray<TArray<TDataPoint<T>>>;
+            const Terminated: TState;
+            Processor: IMycProcessor<TArray<TDataPoint<T>>>
+        ): TState;
+
         function ProcessFile(
             FileInfo: TAuraDataFile;
             const DataFile: TFuture<TArray<TArray<TDataPoint<T>>>>;
@@ -354,6 +360,23 @@ begin
         Result := FCachedFiles.GetOrAdd(DataFile.GetFullFileName);
 end;
 
+function TAuraDataServer<T>.ProcessChunks(
+    const DataChunks: TArray<TArray<TDataPoint<T>>>;
+    const Terminated: TState;
+    Processor: IMycProcessor<TArray<TDataPoint<T>>>
+): TState;
+begin
+    var done := TLatch.CreateLatch(Length(DataChunks));
+
+    for var i := 0 to High(DataChunks) do
+        if not Terminated.IsSet then
+            Processor.ProcessData(DataChunks[i]).Signal.Subscribe(done)
+        else
+            done.Notify;
+
+    Result := done.State;
+end;
+
 function TAuraDataServer<T>.ProcessFile(
     FileInfo: TAuraDataFile;
     const DataFile: TFuture<TArray<TArray<TDataPoint<T>>>>;
@@ -374,19 +397,12 @@ begin
             function(const DataChunks: TArray<TArray<TDataPoint<T>>>): TState
             begin
                 // Process each chunk, checking for termination between chunks.
-                var done := TLatch.CreateLatch(Length(DataChunks));
-                for var i := 0 to High(DataChunks) do
-                    if not cTerminated.IsSet then
-                        Processor.ProcessData(DataChunks[i]).Signal.Subscribe(done)
-                    else
-                        done.Notify;
+                var done := ProcessChunks(DataChunks, Terminated, Processor);
 
                 // Move to next file (which is currently preloading) once all Processors are done
                 Result :=
-                    TaskManager.RunTask(
-                        done.State,
-                        function: TState begin Result := ProcessFile(nextFileInfo, nextFile, cTerminated, Processor); end
-                    )
+                    TaskManager
+                        .RunTask(done, function: TState begin Result := ProcessFile(nextFileInfo, nextFile, cTerminated, Processor); end);
             end
         );
 end;
diff --git a/Src/Myc.Trade.Types.pas b/Src/Myc.Trade.Types.pas
index 551f2d4..d7012b0 100644
--- a/Src/Myc.Trade.Types.pas
+++ b/Src/Myc.Trade.Types.pas
@@ -29,6 +29,7 @@ type
     end;
 
     TConstFunc<S, T> = reference to function(const Value: S): T;
+    TConstProc<T> = reference to procedure(const Value: T);
     TConstFuncPredicate<S, T> = reference to function(const Value: S; out Res: T): Boolean;
 
 implementation