diff --git a/AuraTrader/AuraTrader.dproj b/AuraTrader/AuraTrader.dproj
index 18a82b0..0ded5a4 100644
--- a/AuraTrader/AuraTrader.dproj
+++ b/AuraTrader/AuraTrader.dproj
@@ -4,7 +4,7 @@
         <ProjectVersion>20.3</ProjectVersion>
         <FrameworkType>FMX</FrameworkType>
         <Base>True</Base>
-        <Config Condition="'$(Config)'==''">Release</Config>
+        <Config Condition="'$(Config)'==''">Debug</Config>
         <Platform Condition="'$(Platform)'==''">Win64</Platform>
         <ProjectName Condition="'$(ProjectName)'==''">AuraTrader</ProjectName>
         <TargetedPlatforms>3</TargetedPlatforms>
diff --git a/AuraTrader/MainForm.pas b/AuraTrader/MainForm.pas
index 6d8de84..303241f 100644
--- a/AuraTrader/MainForm.pas
+++ b/AuraTrader/MainForm.pas
@@ -105,13 +105,13 @@ type
         FApplication: IAuraApplication;
         FModulesItem: TTreeViewItem;
         function SelectedSymbol: String;
-        procedure ExecuteStrategy(const Symbol: String; Timeframe: TTimeframe; const Processor: IMycProcessor<TDataPoint<TOhlcItem>>);
+        procedure ExecuteStrategy(const Symbol: String; Timeframe: TTimeframe; const Consumer: IConsumer<TDataPoint<TOhlcItem>>);
 
     public
         procedure NewWorkspace;
         function CurrLayout<T: TControl>: T;
         procedure AlignControl(Control: TControl);
-        function CreateStrategy2(Timeframe: TTimeframe): IMycProcessor<TDataPoint<TOhlcItem>>;
+        function CreateStrategy2(Timeframe: TTimeframe): IConsumer<TDataPoint<TOhlcItem>>;
 
     published
         property OnEvent: TNotifyEvent read FOnEvent write FOnEvent;
@@ -285,7 +285,7 @@ begin
     Control.Align := TAlignLayout.Top;
 end;
 
-function TForm1.CreateStrategy2(Timeframe: TTimeframe): IMycProcessor<TDataPoint<TOhlcItem>>;
+function TForm1.CreateStrategy2(Timeframe: TTimeframe): IConsumer<TDataPoint<TOhlcItem>>;
 type
     TSignal = record
         Sig: Double;
@@ -297,9 +297,9 @@ var
     panel: TMycChart.TPanel;
 begin
     var ticker := TConverter.CreateIdentity<TDataPoint<TOhlcItem>>;
-    Result := ticker;
+    Result := ticker.Consumer;
 
-    var OhlcPoint := ticker.DataProvider.Chain<TDataPoint<TOhlcItem>>(TConverter.CreateOhlcAggregation(Timeframe));
+    var OhlcPoint := ticker.Producer.Chain<TDataPoint<TOhlcItem>>(TConverter.CreateOhlcAggregation(Timeframe));
 
     var Ohlc := OhlcPoint.Field<TOhlcItem>('Data');
 
@@ -426,7 +426,7 @@ begin
 
     conv.Chain<TSignal>(Signal);
 
-    var pnl := Signal.DataProvider.Field<Double>('pnl');
+    var pnl := Signal.Producer.Field<Double>('pnl');
 
     var FEquity: Double := 10000;
     var FInit: Boolean := false;
@@ -470,21 +470,21 @@ begin
     panel.AddOhlcSeries(Ohlc);
     panel.AddDoubleSeries(Hull, TAlphaColors.Cornflowerblue, 2);
     panel.AddDoubleSeries(Sma, TAlphaColors.Brown, 1.5);
-    panel.AddDoubleSeries(Signal.DataProvider.Field<Double>('Entry'), TAlphaColors.Green, 1);
-    panel.AddDoubleSeries(Signal.DataProvider.Field<Double>('SL'), TAlphaColors.Red, 2);
+    panel.AddDoubleSeries(Signal.Producer.Field<Double>('Entry'), TAlphaColors.Green, 1);
+    panel.AddDoubleSeries(Signal.Producer.Field<Double>('SL'), TAlphaColors.Red, 2);
 
     var mean := TConverter<TArray<Double>, Double>.CreateGeneric(TIndicators.CreateMean());
 
     TConverter.Join<Double>([Hull, Sma]).Chain<Double>(mean);
 
-    panel.AddDoubleSeries(mean.DataProvider, TAlphaColors.Blue, 5);
+    panel.AddDoubleSeries(mean.Producer, TAlphaColors.Blue, 5);
 
     var pnlChart := TMycChart.Create(Self);
     AlignControl(pnlChart);
     pnlChart.Height := Layout.ChildrenRect.Width * 9 / 24;
     pnlChart.Lookback.Value := 50000;
 
-    pnlChart.SetXAxisCounter<Double>(equity.DataProvider);
+    pnlChart.SetXAxisCounter<Double>(equity.Producer);
 
     ////////////
     var EMAFactory := TEMA.Create;
@@ -498,7 +498,7 @@ begin
 
     var equityEMA :=
         equity
-            .DataProvider
+            .Producer
             .Chain<TDataRecord>(TConverter.FieldToRecord<Double>(EMAFactory.Input, 'Price'))
             .Chain<TDataRecord>(EMAConv)
             .Chain<Double>(TConverter.FieldOfRecord<Double>(EMAFactory.Output, 'MA'));
@@ -506,7 +506,7 @@ begin
     //////////////
 
     panel := pnlChart.AddPanel;
-    panel.AddDoubleSeries(equity.DataProvider, TAlphaColors.Blue, 3);
+    panel.AddDoubleSeries(equity.Producer, TAlphaColors.Blue, 3);
     panel.AddDoubleSeries(equityEMA, TAlphaColors.Gray, 2);
 
     /////
@@ -533,7 +533,7 @@ begin
     Result := Res;
 end;
 
-procedure TForm1.ExecuteStrategy(const Symbol: String; Timeframe: TTimeframe; const Processor: IMycProcessor<TDataPoint<TOhlcItem>>);
+procedure TForm1.ExecuteStrategy(const Symbol: String; Timeframe: TTimeframe; const Consumer: IConsumer<TDataPoint<TOhlcItem>>);
 begin
     var terminated := TFlag.CreateObserver(FTerminate.Signal).State;
 
@@ -550,9 +550,9 @@ begin
         );
 
     var OhlcPoint := lastPrice.Chain<TDataPoint<TOhlcItem>>(TConverter.CreateTickAggregation(Timeframe));
-    OhlcPoint.DataProvider.Link(Processor);
+    OhlcPoint.Producer.Link(Consumer);
 
-    var dataProvider :=
+    var Producer :=
         TConverter<TArray<TDataPoint<TAuraAskBidFileItem>>, TArray<TDataPoint<TAskBidItem>>>.CreateGeneric(
             function(const Values: TArray<TDataPoint<TAuraAskBidFileItem>>): TArray<TDataPoint<TAskBidItem>>
             begin
@@ -566,16 +566,16 @@ begin
             end
         );
 
-    dataProvider.DataProvider.Link(ticker);
+    Producer.Producer.Link(ticker);
 
-    FProcessDone := FProcessDone + FServer.ProcessData(Symbol, terminated, dataProvider);
+    FProcessDone := FProcessDone + FServer.ProcessData(Symbol, terminated, Producer);
 {$else}
 
     var ticker := TConverter.CreateTicker<TDataPoint<TOhlcItem>>;
 
-    ticker.DataProvider.Chain(Processor);
+    ticker.Producer.Chain(Consumer);
 
-    FProcessDone := FProcessDone + FServer.ProcessData(Symbol, terminated, ticker);
+    FProcessDone := FProcessDone + FServer.ProcessData(Symbol, terminated, ticker.Consumer);
 {$endif}
 end;
 
@@ -609,7 +609,7 @@ begin
 
     var ticker := TConverter.CreateIdentity<TDataPoint<TOhlcItem>>;
 
-    var OhlcPoint := ticker.DataProvider.Chain<TDataPoint<TOhlcItem>>(TConverter.CreateOhlcAggregation(timeframe));
+    var OhlcPoint := ticker.Producer.Chain<TDataPoint<TOhlcItem>>(TConverter.CreateOhlcAggregation(timeframe));
 
     //    var OhlcTicker := TConverter.CreateIdentity<TDataPoint<TOhlcItem>>;
     //    var OhlcPoint := OhlcTicker.Sender;
@@ -674,7 +674,7 @@ begin
 }
     /////
 
-    ExecuteStrategy(Symbol, timeframe, ticker);
+    ExecuteStrategy(Symbol, timeframe, ticker.Consumer);
 end;
 
 procedure TForm1.Strat2ButtonClick(Sender: TObject);
@@ -687,7 +687,7 @@ begin
     //ExecuteStrategy(Symbol, timeframe, CreateStrategy2(timeframe));
 
     var tstStrat := StrategyTest.CreateStrategy1(timeframe);
-    ExecuteStrategy(Symbol, timeframe, tstStrat);
+    ExecuteStrategy(Symbol, timeframe, tstStrat.Consumer);
 
     var Layout := CurrLayout<TVertScrollBox>;
     if Layout = nil then
@@ -698,10 +698,10 @@ begin
     pnlChart.Height := Layout.ChildrenRect.Width * 9 / 24;
     pnlChart.Lookback.Value := 50000;
 
-    pnlChart.SetXAxisCounter<Double>(tstStrat.DataProvider);
+    pnlChart.SetXAxisCounter<Double>(tstStrat.Producer);
 
     var panel := pnlChart.AddPanel;
-    panel.AddDoubleSeries(tstStrat.DataProvider, TAlphaColors.Blue, 3);
+    panel.AddDoubleSeries(tstStrat.Producer, TAlphaColors.Blue, 3);
 end;
 
 end.
diff --git a/AuraTrader/StrategyTest.pas b/AuraTrader/StrategyTest.pas
index d82bc80..cff102b 100644
--- a/AuraTrader/StrategyTest.pas
+++ b/AuraTrader/StrategyTest.pas
@@ -30,7 +30,7 @@ type
 begin
     var ticker := TConverter.CreateIdentity<TDataPoint<TOhlcItem>>;
 
-    var OhlcPoint := ticker.DataProvider.Chain<TDataPoint<TOhlcItem>>(TConverter.CreateOhlcAggregation(Timeframe));
+    var OhlcPoint := ticker.Producer.Chain<TDataPoint<TOhlcItem>>(TConverter.CreateOhlcAggregation(Timeframe));
 
     var Ohlc := OhlcPoint.Field<TOhlcItem>('Data');
 
@@ -183,7 +183,7 @@ begin
             )
         );
 
-    Result := TConverter<TDataPoint<TOhlcItem>, Double>.Construct(ticker, equity);
+    Result := TConverter<TDataPoint<TOhlcItem>, Double>.Construct(ticker.Consumer, equity);
 end;
 
 end.
diff --git a/Src/Myc.Fmx.Chart.Series.pas b/Src/Myc.Fmx.Chart.Series.pas
index 4182e57..cb527ae 100644
--- a/Src/Myc.Fmx.Chart.Series.pas
+++ b/Src/Myc.Fmx.Chart.Series.pas
@@ -18,7 +18,7 @@ type
     { TChartCustomLayer }
     TChartCustomLayer<T> = class abstract(TMycChart.TDataLayer)
     private
-        FDataProvider: TDataProvider<T>;
+        FProducer: TProducer<T>;
         FReceiver: TLazy<TSeries<T>>;
         FData: TSeries<T>;
     protected
@@ -33,7 +33,7 @@ type
             const YForm: TFunc<Double, Single>
         ); override; abstract;
     public
-        constructor Create(AParent: TMycChart.TPanel; const ADataProvider: TDataProvider<T>);
+        constructor Create(AParent: TMycChart.TPanel; const AProducer: TProducer<T>);
         property Data: TSeries<T> read FData;
     end;
 
@@ -53,7 +53,7 @@ type
     public
         constructor Create(
             AParent: TMycChart.TPanel;
-            const ADataProvider: TDataProvider<TOhlcItem>;
+            const AProducer: TProducer<TOhlcItem>;
             const AUpColor, ADownColor: TMutable<TAlphaColor>
         );
     end;
@@ -74,7 +74,7 @@ type
     public
         constructor Create(
             AParent: TMycChart.TPanel;
-            const ADataProvider: TDataProvider<Double>;
+            const AProducer: TProducer<Double>;
             const ALineColor: TAlphaColor;
             ALineWidth: Single
         );
@@ -83,7 +83,7 @@ type
     { TChartXAxisLayer }
     TChartXAxisLayer<T> = class(TMycChart.TXAxisLayer)
     private
-        FDataProvider: TDataProvider<T>;
+        FProducer: TProducer<T>;
         FReceiver: TLazy<TSeries<T>>;
         FData: TSeries<T>;
     protected
@@ -92,7 +92,7 @@ type
         function GetCount: Int64; override;
         function GetTotalCount: Int64; override;
     public
-        constructor Create(AOwner: TMycChart; const ADataProvider: TDataProvider<T>);
+        constructor Create(AOwner: TMycChart; const AProducer: TProducer<T>);
         property Data: TSeries<T> read FData;
     end;
 
@@ -104,7 +104,7 @@ type
         // Provide a formatted timestamp string for a given data index.
         function GetCaption(Idx: Int64): String; override; final;
     public
-        constructor Create(AOwner: TMycChart; ATimeframe: TTimeframe; const ADataProvider: TDataProvider<TDateTime>);
+        constructor Create(AOwner: TMycChart; ATimeframe: TTimeframe; const AProducer: TProducer<TDateTime>);
     end;
 
 implementation
@@ -118,12 +118,12 @@ uses
 
 constructor TChartLineLayer.Create(
     AParent: TMycChart.TPanel;
-    const ADataProvider: TDataProvider<Double>;
+    const AProducer: TProducer<Double>;
     const ALineColor: TAlphaColor;
     ALineWidth: Single
 );
 begin
-    inherited Create(AParent, ADataProvider);
+    inherited Create(AParent, AProducer);
     FLineColor := ALineColor;
     FLineWidth := ALineWidth;
 end;
@@ -194,11 +194,11 @@ end;
 
 constructor TChartOhlcLayer.Create(
     AParent: TMycChart.TPanel;
-    const ADataProvider: TDataProvider<TOhlcItem>;
+    const AProducer: TProducer<TOhlcItem>;
     const AUpColor, ADownColor: TMutable<TAlphaColor>
 );
 begin
-    inherited Create(AParent, ADataProvider);
+    inherited Create(AParent, AProducer);
     FUpColor := AUpColor;
     FDownColor := ADownColor;
 
@@ -267,12 +267,12 @@ end;
 
 { TChartCustomLayer }
 
-constructor TChartCustomLayer<T>.Create(AParent: TMycChart.TPanel; const ADataProvider: TDataProvider<T>);
+constructor TChartCustomLayer<T>.Create(AParent: TMycChart.TPanel; const AProducer: TProducer<T>);
 begin
     inherited Create(AParent);
-    FDataProvider := ADataProvider;
+    FProducer := AProducer;
 
-    FReceiver := FDataProvider.CreateEndpoint(Owner.Lookback.Value);
+    FReceiver := FProducer.CreateEndpoint(Owner.Lookback.Value);
 end;
 
 function TChartCustomLayer<T>.GetCount: Int64;
@@ -292,11 +292,11 @@ end;
 
 { TChartXAxisLayer }
 
-constructor TChartXAxisLayer<T>.Create(AOwner: TMycChart; const ADataProvider: TDataProvider<T>);
+constructor TChartXAxisLayer<T>.Create(AOwner: TMycChart; const AProducer: TProducer<T>);
 begin
     inherited Create(AOwner);
-    FDataProvider := ADataProvider;
-    FReceiver := FDataProvider.CreateEndpoint(Owner.Lookback.Value);
+    FProducer := AProducer;
+    FReceiver := FProducer.CreateEndpoint(Owner.Lookback.Value);
 end;
 
 function TChartXAxisLayer<T>.GetCaption(Idx: Int64): String;
@@ -321,9 +321,9 @@ end;
 
 { TChartXAxisTimestampLayer }
 
-constructor TChartXAxisTimestampLayer.Create(AOwner: TMycChart; ATimeframe: TTimeframe; const ADataProvider: TDataProvider<TDateTime>);
+constructor TChartXAxisTimestampLayer.Create(AOwner: TMycChart; ATimeframe: TTimeframe; const AProducer: TProducer<TDateTime>);
 begin
-    inherited Create(AOwner, ADataProvider);
+    inherited Create(AOwner, AProducer);
     FTimeframe := ATimeframe;
 end;
 
diff --git a/Src/Myc.Fmx.Chart.pas b/Src/Myc.Fmx.Chart.pas
index ab334cf..8484cbb 100644
--- a/Src/Myc.Fmx.Chart.pas
+++ b/Src/Myc.Fmx.Chart.pas
@@ -97,13 +97,13 @@ type
                 destructor Destroy; override;
 
                 function AddOhlcSeries(
-                    const DataProvider: TDataProvider<TOhlcItem>;
+                    const Producer: TProducer<TOhlcItem>;
                     const AUpColor: TAlphaColor = TAlphaColors.Green;
                     const ADownColor: TAlphaColor = TAlphaColors.Red
                 ): TMycChart.TDataLayer;
 
                 function AddDoubleSeries(
-                    const DataProvider: TDataProvider<Double>;
+                    const Producer: TProducer<Double>;
                     const ALineColor: TAlphaColor = TAlphaColors.Cornflowerblue;
                     const ALineWidth: Single = 1.5
                 ): TMycChart.TDataLayer;
@@ -150,8 +150,8 @@ type
         function AddPanel: TPanel;
 
         // Sets the master series that defines the time scale (X-axis).
-        function SetXAxisSeries(Timeframe: TTimeframe; const DataProvider: TDataProvider<TDateTime>): TMycChart.TXAxisLayer; overload;
-        function SetXAxisCounter<T>(const DataProvider: TDataProvider<T>): TMycChart.TXAxisLayer; overload;
+        function SetXAxisSeries(Timeframe: TTimeframe; const Producer: TProducer<TDateTime>): TMycChart.TXAxisLayer; overload;
+        function SetXAxisCounter<T>(const Producer: TProducer<T>): TMycChart.TXAxisLayer; overload;
 
         property Lookback: TWriteable<Int64> read FLookback write FLookback;
         property NeedRepaint: TFlag read FNeedRepaint;
@@ -632,19 +632,19 @@ begin
     Repaint;
 end;
 
-function TMycChart.SetXAxisCounter<T>(const DataProvider: TDataProvider<T>): TMycChart.TXAxisLayer;
+function TMycChart.SetXAxisCounter<T>(const Producer: TProducer<T>): TMycChart.TXAxisLayer;
 begin
     FXAxisSeries.Free;
 
-    FXAxisSeries := TChartXAxisLayer<Int64>.Create(Self, DataProvider.Chain<Int64>(TConverter.CreateCounter<T>));
+    FXAxisSeries := TChartXAxisLayer<Int64>.Create(Self, Producer.Chain<Int64>(TConverter.CreateCounter<T>));
     Result := FXAxisSeries;
 end;
 
-function TMycChart.SetXAxisSeries(Timeframe: TTimeframe; const DataProvider: TDataProvider<TDateTime>): TMycChart.TXAxisLayer;
+function TMycChart.SetXAxisSeries(Timeframe: TTimeframe; const Producer: TProducer<TDateTime>): TMycChart.TXAxisLayer;
 begin
     FXAxisSeries.Free;
 
-    FXAxisSeries := TChartXAxisTimestampLayer.Create(Self, Timeframe, DataProvider);
+    FXAxisSeries := TChartXAxisTimestampLayer.Create(Self, Timeframe, Producer);
     Result := FXAxisSeries;
 end;
 
@@ -670,23 +670,22 @@ begin
 end;
 
 function TMycChart.TPanel.AddDoubleSeries(
-    const DataProvider: TDataProvider<Double>;
+    const Producer: TProducer<Double>;
     const ALineColor: TAlphaColor = TAlphaColors.Cornflowerblue;
     const ALineWidth: Single = 1.5
 ): TMycChart.TDataLayer;
 begin
-    Result := TChartLineLayer.Create(Self, DataProvider, ALineColor, ALineWidth);
+    Result := TChartLineLayer.Create(Self, Producer, ALineColor, ALineWidth);
     FLayers.Add(Result);
 end;
 
 function TMycChart.TPanel.AddOhlcSeries(
-    const DataProvider: TDataProvider<TOhlcItem>;
+    const Producer: TProducer<TOhlcItem>;
     const AUpColor: TAlphaColor = TAlphaColors.Green;
     const ADownColor: TAlphaColor = TAlphaColors.Red
 ): TMycChart.TDataLayer;
 begin
-    Result :=
-        TChartOhlcLayer.Create(Self, DataProvider, TMutable<TAlphaColor>.Constant(AUpColor), TMutable<TAlphaColor>.Constant(ADownColor));
+    Result := TChartOhlcLayer.Create(Self, Producer, TMutable<TAlphaColor>.Constant(AUpColor), TMutable<TAlphaColor>.Constant(ADownColor));
     FLayers.Add(Result);
 end;
 
diff --git a/Src/Myc.Trade.DataPoint.Impl.pas b/Src/Myc.Trade.DataPoint.Impl.pas
index 085af95..93a1533 100644
--- a/Src/Myc.Trade.DataPoint.Impl.pas
+++ b/Src/Myc.Trade.DataPoint.Impl.pas
@@ -15,68 +15,91 @@ uses
 
 type
     // Abstract base class for data consumers.
-    TMycProcessor<T> = class abstract(TInterfacedObject, IMycProcessor<T>)
+    TMycConsumer<T> = class abstract(TContainedObject, IConsumer<T>)
+    strict private
+        type
+            TNull = class(TInterfacedObject, IConsumer<T>)
+            protected
+                function Consume(const Value: T): TState;
+            end;
+        class var
+            FNull: IConsumer<T>;
+        class constructor CreateClass;
     protected
-        function Execute(const Value: T; const Proc: TConstFunc<T, TState>): TState;
-        function ProcessData(const Value: T): TState; virtual; abstract;
-    end;
-
-    // Concrete data provider that manages a list of processors (listeners).
-    TMycContainedDataProvider<T> = class abstract(TContainedObject, IDataProvider<T>)
-    private
-        FListeners: TMycNotifyList<IMycProcessor<T>>;
+        function Consume(const Value: T): TState; virtual; abstract;
     public
-        constructor Create(const Controller: IInterface);
-        destructor Destroy; override;
-        // Notifies all linked processors.
-        function Broadcast(const Value: T): TState;
-        // Link a Processor
-        function Link(const Processor: IMycProcessor<T>): TTag;
-        // Unlink a linked strategy
-        procedure Unlink(Tag: TTag);
+        class property Null: IConsumer<T> read FNull;
     end;
 
-    TMycSequence<T> = class(TMycProcessor<T>, IMycDataSequence<T>)
+    TMycProducer<T> = class(TInterfacedObject, IProducer<T>)
+    strict private
+        type
+            TNull = class(TInterfacedObject, IProducer<T>)
+            public
+                function Link(const Consumer: IConsumer<T>): TTag;
+                procedure Unlink(Tag: TTag);
+            end;
+        class var
+            FNull: IProducer<T>;
+        class constructor CreateClass;
     private
-        FDataProviders: TArray<TMycContainedDataProvider<T>>;
-        function GetCount: Integer;
-        function GetDataProvider(Idx: Integer): TDataProvider<T>;
-    protected
-        function ProcessData(const Value: T): TState; override;
-        function ProcessDataProvider(Idx: Integer; const Value: T): TState;
-    public
-        constructor Create(ACount: Integer);
-        destructor Destroy; override;
-    end;
-
-    // Null object implementation for IDataProvider.
-    TNullDataProvider<T> = class(TInterfacedObject, IDataProvider<T>)
-    public
-        function Link(const Receiver: IMycProcessor<T>): TTag;
-        procedure Unlink(Tag: TTag);
-    end;
-
-    // Abstract base class for components that process data of type S and provide data of type T.
-    TMycConverter<S, T> = class abstract(TMycProcessor<S>, IConverter<S, T>)
-    private
-        FDataProvider: TMycContainedDataProvider<T>;
-        function GetDataProvider: IDataProvider<T>;
-    protected
-        function ProcessData(const Value: S): TState; override; abstract;
-        // Broadcasts the given data to all linked processors.
-        function Broadcast(const Value: T): TState;
+        FListeners: TMycNotifyList<IConsumer<T>>;
     public
         constructor Create;
         destructor Destroy; override;
-        property DataProvider: IDataProvider<T> read GetDataProvider;
+        function Broadcast(const Value: T): TState;
+        function Link(const Consumer: IConsumer<T>): TTag;
+        procedure Unlink(Tag: TTag);
+        class property Null: IProducer<T> read FNull;
     end;
 
-    // Null object implementation for IConverter.
-    TNullConverter<S, T> = class(TInterfacedObject, IConverter<S, T>)
+    // Concrete producer that manages a list of consumers (listeners).
+    TMycContainedProducer<T> = class(TContainedObject, IProducer<T>)
     private
-        function GetDataProvider: IDataProvider<T>;
+        FListeners: TMycNotifyList<IConsumer<T>>;
     public
-        function ProcessData(const Value: S): TState;
+        constructor Create(const Controller: IInterface);
+        destructor Destroy; override;
+        function Broadcast(const Value: T): TState;
+        function Link(const Consumer: IConsumer<T>): TTag;
+        procedure Unlink(Tag: TTag);
+    end;
+
+    // Abstract base class for components that now act as a producer and contain a consumer.
+    TMycConverter<S, T> = class abstract(TMycProducer<T>, IConverter<S, T>)
+    strict private
+        type
+            TNull = class(TInterfacedObject, IConverter<S, T>)
+            private
+                function GetConsumer: IConsumer<S>;
+                function Link(const Consumer: IConsumer<T>): TTag;
+                procedure Unlink(Tag: TTag);
+            end;
+        class var
+            FNull: IConverter<S, T>;
+        class constructor CreateClass;
+    private
+        type
+            TConsumer = class(TMycConsumer<S>)
+            private
+                FOwner: TMycConverter<S, T>;
+            protected
+                function Consume(const Value: S): TState; override; final;
+            public
+                constructor Create(AOwner: TMycConverter<S, T>);
+            end;
+
+    private
+        FConsumer: TConsumer;
+        function GetConsumer: IConsumer<S>;
+    protected
+        // To be implemented by descendants to perform the actual conversion.
+        function Consume(const Value: S): TState; virtual; abstract;
+    public
+        constructor Create;
+        destructor Destroy; override;
+
+        class property Null: IConverter<S, T> read FNull;
     end;
 
     // A generic converter that uses a function reference for the conversion logic.
@@ -84,7 +107,7 @@ type
     private
         FFunc: TConstFunc<S, T>;
     protected
-        function ProcessData(const Value: S): TState; override;
+        function Consume(const Value: S): TState; override;
     public
         constructor Create(const AFunc: TConstFunc<S, T>);
     end;
@@ -93,7 +116,7 @@ type
     private
         FFunc: TConstFunc<S, TConverter<S, T>.TBroadcastProc, TState>;
     protected
-        function ProcessData(const Value: S): TState; override;
+        function Consume(const Value: S): TState; override;
     public
         constructor Create(const AFunc: TConstFunc<S, TConverter<S, T>.TBroadcastProc, TState>);
     end;
@@ -103,21 +126,14 @@ type
         FFunc: TConstFunc<S, T>;
         FQueue: TState;
     protected
-        function ProcessData(const Value: S): TState; override;
+        function Consume(const Value: S): TState; override;
     public
         constructor Create(const AFunc: TConstFunc<S, T>);
     end;
 
     TMycIdentityConverter<T> = class(TMycConverter<T, T>)
     protected
-        function ProcessData(const Value: T): TState; override; final;
-    end;
-
-    // A converter specialized for calculating indicators.
-    TMycIndicator<S, T> = class(TMycConverter<S, T>)
-    protected
-        function ProcessData(const Value: S): TState; override; final;
-        function Calculate(const Value: S): T; virtual; abstract;
+        function Consume(const Value: T): TState; override; final;
     end;
 
     // A converter that counts incoming data points and outputs the current count.
@@ -125,47 +141,40 @@ type
     private
         FCount: Int64;
     protected
-        function ProcessData(const Value: T): TState; override;
+        function Consume(const Value: T): TState; override;
     public
         constructor Create;
     end;
 
     // A converter that takes an array and broadcasts each element individually.
     TMycTicker<T> = class(TMycConverter<TArray<T>, T>)
-    public
-        function ProcessData(const Values: TArray<T>): TState; override;
+    protected
+        function Consume(const Values: TArray<T>): TState; override;
     end;
 
     // A converter that reads a specific field from a record using RTTI.
     TMycRecordFieldReader<S, T> = class(TMycConverter<S, T>)
     private
         FOffset: Integer;
+    protected
+        function Consume(const Values: S): TState; override;
     public
         constructor Create(const AFieldName: String);
-        function ProcessData(const Values: S): TState; override;
     end;
 
-    // A generic processor implementation that uses a function reference.
-    TMycGenericProcessor<T> = class(TMycProcessor<T>)
+    // A consumer implementation that is owned by a controller.
+    TMycGenericConsumer<T> = class(TMycConsumer<T>)
     private
         FProc: TConstFunc<T, TState>;
     protected
-        function ProcessData(const Value: T): TState; override; final;
-    public
-        constructor Create(const AProc: TConstFunc<T, TState>);
-    end;
-
-    // A processor implementation that is owned by a controller.
-    TMycContainedProcessor<T> = class(TContainedObject, IMycProcessor<T>)
-    private
-        FProc: TConstFunc<T, TState>;
-        function ProcessData(const Value: T): TState;
+        function Consume(const Value: T): TState; override;
     public
         constructor Create(const Controller: IInterface; const AProc: TConstFunc<T, TState>);
     end;
 
     // Endpoint that collects data into a series.
     TMycDataEndpoint<T> = class(TInterfacedObject, TLazy<TSeries<T>>.ILazy)
+    private
         type
             PItem = ^TItem;
             TItem = record
@@ -173,18 +182,18 @@ type
                 Value: T;
             end;
     private
-        FProcessor: TMycContainedProcessor<T>;
+        FConsumer: TMycGenericConsumer<T>;
         FTag: TTag;
-        FDataProvider: IDataProvider<T>;
+        FProducer: IProducer<T>;
         FLookback: Int64;
         FChanged: TFlag;
         FLock: TLightweightMREW;
         FFirst: PItem;
         FCount: Integer;
         function GetChanged: TState;
-        function ProcessData(const Value: T): TState;
+        function Consume(const Value: T): TState;
     public
-        constructor Create(const ADataProvider: TDataProvider<T>; ALookback: Int64);
+        constructor Create(const AProducer: TProducer<T>; ALookback: Int64);
         destructor Destroy; override;
         function Update(var Value: TSeries<T>): Boolean;
     end;
@@ -196,9 +205,10 @@ type
         function GetBarStartTime(const TimeStamp: TDateTime; const Timeframe: TTimeframe): TDateTime;
         function GetCurrentBar: TDataPoint<TOhlcItem>;
         function GetTimeframe: TTimeframe;
+    protected
+        function Consume(const Value: TDataPoint<Double>): TState; override;
     public
         constructor Create(const ATimeframe: TTimeframe);
-        function ProcessData(const Value: TDataPoint<Double>): TState; override;
         property CurrentBar: TDataPoint<TOhlcItem> read GetCurrentBar;
         property Timeframe: TTimeframe read GetTimeframe;
     end;
@@ -207,7 +217,7 @@ type
     private
         FQueue: TState;
     protected
-        function ProcessData(const Value: T): TState; override; final;
+        function Consume(const Value: T): TState; override; final;
     end;
 
     TOhlcAggregation = class(TMycConverter<TDataPoint<TOhlcItem>, TDataPoint<TOhlcItem>>)
@@ -217,42 +227,58 @@ type
         function GetBarStartTime(const TimeStamp: TDateTime; const Timeframe: TTimeframe): TDateTime;
         function GetCurrentBar: TDataPoint<TOhlcItem>;
         function GetTimeframe: TTimeframe;
+    protected
+        function Consume(const Value: TDataPoint<TOhlcItem>): TState; override;
     public
         constructor Create(const ATimeframe: TTimeframe);
-        function ProcessData(const Value: TDataPoint<TOhlcItem>): TState; override;
         property CurrentBar: TDataPoint<TOhlcItem> read GetCurrentBar;
         property Timeframe: TTimeframe read GetTimeframe;
     end;
 
     // Endpoint that collects data into a series.
-    TMycDataJoin<T> = class(TInterfacedObject, IDataProvider<TArray<T>>)
+    TMycDataJoin<T> = class(TInterfacedObject, IProducer<TArray<T>>)
     private
-        FReceivers: array of record
-            DataProvider: IDataProvider<T>;
-            Processor: TMycContainedProcessor<T>;
+        FConsumers: array of record
+            Producer: IProducer<T>;
+            Consumer: TMycGenericConsumer<T>;
             Tag: TTag;
             Queue: TQueue<T>;
         end;
 
-        FDataProvider: TMycContainedDataProvider<TArray<T>>;
+        FContainedProvider: TMycContainedProducer<TArray<T>>;
         FLock: TSpinLock;
         FCount: Integer;
-        function ProcessData(Idx: Integer; const Value: T): TState;
+        function Consume(Idx: Integer; const Value: T): TState;
+        function Link(const Consumer: IConsumer<TArray<T>>): TTag;
+        procedure Unlink(Tag: TTag);
     public
-        constructor Create(const ADataProviders: TArray<TDataProvider<T>>);
+        constructor Create(const AProducers: TArray<TProducer<T>>);
         destructor Destroy; override;
-        property DataProvider: TMycContainedDataProvider<TArray<T>> read FDataProvider implements IDataProvider<TArray<T>>;
     end;
 
-    TMycComposedConverter<S, T> = class(TMycProcessor<S>, IConverter<S, T>)
+    TMycComposedConverter<S, T> = class(TInterfacedObject, IConverter<S, T>)
     private
-        FProcessor: IMycProcessor<S>;
-        FDataProvider: TDataProvider<T>;
+        FConsumer: IConsumer<S>;
+        FProducer: TProducer<T>;
     protected
-        function ProcessData(const Value: S): TState; override;
-        function GetDataProvider: IDataProvider<T>;
+        function GetConsumer: IConsumer<S>;
+        function Link(const Consumer: IConsumer<T>): TTag;
+        procedure Unlink(Tag: TTag);
     public
-        constructor Create(const AProcessor: IMycProcessor<S>; const ADataProvider: TDataProvider<T>);
+        constructor Create(const AConsumer: IConsumer<S>; const AProducer: TProducer<T>);
+    end;
+
+    TMycSequence<T> = class(TInterfacedObject, IConsumer<T>)
+    private
+        FProducers: TArray<TMycContainedProducer<T>>;
+    protected
+        function Consume(const Value: T): TState;
+        function ProcessProducer(Idx: Integer; const Value: T): TState;
+    public
+        constructor Create(ACount: Integer);
+        destructor Destroy; override;
+
+        class function CreateSequence(Count: Integer; out Producers: TArray<IProducer<T>>): IConsumer<T>; static;
     end;
 
 implementation
@@ -265,28 +291,25 @@ uses
     Winapi.Windows,
     Myc.TaskManager;
 
-{ TMycProcessor<T> }
-
-function TMycProcessor<T>.Execute(const Value: T; const Proc: TConstFunc<T, TState>): TState;
+class constructor TMycConsumer<T>.CreateClass;
 begin
-    var cValue := Value;
-    Result := TaskManager.RunTask(function: TState begin Result := Proc(cValue); end);
+    FNull := TNull.Create;
 end;
 
-{ TMycContainedDataProvider<T> }
+{ TMycContainedProducer<T> }
 
-constructor TMycContainedDataProvider<T>.Create(const Controller: IInterface);
+constructor TMycContainedProducer<T>.Create(const Controller: IInterface);
 begin
     inherited Create(Controller);
 end;
 
-destructor TMycContainedDataProvider<T>.Destroy;
+destructor TMycContainedProducer<T>.Destroy;
 begin
     FListeners.Finalize;
     inherited Destroy;
 end;
 
-function TMycContainedDataProvider<T>.Broadcast(const Value: T): TState;
+function TMycContainedProducer<T>.Broadcast(const Value: T): TState;
 begin
     FListeners.Lock;
     try
@@ -304,7 +327,7 @@ begin
         var done := TLatch.CreateLatch(i);
         while item <> nil do
         begin
-            item.Receiver.ProcessData(Value).Signal.Subscribe(done);
+            item.Receiver.Consume(Value).Signal.Subscribe(done);
             item := item.Prev;
         end;
 
@@ -314,18 +337,18 @@ begin
     end;
 end;
 
-function TMycContainedDataProvider<T>.Link(const Processor: IMycProcessor<T>): TTag;
+function TMycContainedProducer<T>.Link(const Consumer: IConsumer<T>): TTag;
 begin
-    // Add the Processor to the notification list
+    // Add the Consumer to the notification list
     FListeners.Lock;
     try
-        Result := FListeners.Advise(Processor);
+        Result := FListeners.Advise(Consumer);
     finally
         FListeners.Release;
     end;
 end;
 
-procedure TMycContainedDataProvider<T>.Unlink(Tag: TTag);
+procedure TMycContainedProducer<T>.Unlink(Tag: TTag);
 begin
     FListeners.Lock;
     try
@@ -335,52 +358,47 @@ begin
     end;
 end;
 
-{ TNullDataProvider<T> }
-
-function TNullDataProvider<T>.Link(const Receiver: IMycProcessor<T>): TTag;
-begin
-    Result := nil;
-end;
-
-procedure TNullDataProvider<T>.Unlink(Tag: TTag);
-begin
-    // Do nothing in the null implementation.
-end;
-
 { TMycConverter<S, T> }
 
 constructor TMycConverter<S, T>.Create;
 begin
     inherited Create;
-    FDataProvider := TMycContainedDataProvider<T>.Create(Self);
+    FConsumer := TConsumer.Create(Self);
+end;
+
+{ TMycConverter<S, T> }
+
+class constructor TMycConverter<S, T>.CreateClass;
+begin
+    FNull := TNull.Create;
 end;
 
 destructor TMycConverter<S, T>.Destroy;
 begin
-    FDataProvider.Free;
+    FConsumer.Free;
     inherited Destroy;
 end;
 
-function TMycConverter<S, T>.Broadcast(const Value: T): TState;
+function TMycConverter<S, T>.GetConsumer: IConsumer<S>;
 begin
-    Result := FDataProvider.Broadcast(Value);
+    Result := FConsumer;
 end;
 
-function TMycConverter<S, T>.GetDataProvider: IDataProvider<T>;
+{ TMycConverter<S, T>.TNull }
+
+function TMycConverter<S, T>.TNull.GetConsumer: IConsumer<S>;
 begin
-    Result := FDataProvider;
+    Result := TMycConsumer<S>.Null;
 end;
 
-{ TNullConverter<S, T> }
-
-function TNullConverter<S, T>.GetDataProvider: IDataProvider<T>;
+function TMycConverter<S, T>.TNull.Link(const Consumer: IConsumer<T>): TTag;
 begin
-    Result := TDataProvider<T>.Null;
+    Result := nil;
 end;
 
-function TNullConverter<S, T>.ProcessData(const Value: S): TState;
+procedure TMycConverter<S, T>.TNull.Unlink(Tag: TTag);
 begin
-    Result := TState.Null;
+    // NOP
 end;
 
 { TMycGenericConverter<S, T> }
@@ -391,18 +409,11 @@ begin
     FFunc := AFunc;
 end;
 
-function TMycGenericConverter<S, T>.ProcessData(const Value: S): TState;
+function TMycGenericConverter<S, T>.Consume(const Value: S): TState;
 begin
     Result := Broadcast(FFunc(Value));
 end;
 
-{ TMycIndicator<S,T> }
-
-function TMycIndicator<S, T>.ProcessData(const Value: S): TState;
-begin
-    Result := Execute(Value, function(const Value: S): TState begin Result := Broadcast(Calculate(Value)); end);
-end;
-
 { TMycDataCounter<T> }
 
 constructor TMycDataCounter<T>.Create;
@@ -411,7 +422,7 @@ begin
     FCount := 0;
 end;
 
-function TMycDataCounter<T>.ProcessData(const Value: T): TState;
+function TMycDataCounter<T>.Consume(const Value: T): TState;
 begin
     Result := Broadcast(FCount);
     inc(FCount);
@@ -419,11 +430,10 @@ end;
 
 { TMycTicker<T> }
 
-function TMycTicker<T>.ProcessData(const Values: TArray<T>): TState;
+function TMycTicker<T>.Consume(const Values: TArray<T>): TState;
 begin
     var done := TLatch.CreateLatch(Length(Values));
 
-    // Process each incoming data point
     for var i := 0 to High(Values) do
         Broadcast(Values[i]).Signal.Subscribe(done);
 
@@ -440,8 +450,6 @@ begin
     var Field := Context.GetType(TypeInfo(S)).GetField(AFieldName);
     var TypeT := Context.GetType(TypeInfo(T));
 
-    var Fields := Context.GetType(TypeInfo(S)).GetFields;
-
     Assert(Assigned(Field), 'Field ' + AFieldName + ' not found');
     Assert(Field.FieldType.TypeKind = TypeT.TypeKind, 'Incorrect type');
 
@@ -451,7 +459,7 @@ begin
         FOffset := -1;
 end;
 
-function TMycRecordFieldReader<S, T>.ProcessData(const Values: S): TState;
+function TMycRecordFieldReader<S, T>.Consume(const Values: S): TState;
 type
     PT = ^T;
 begin
@@ -463,48 +471,35 @@ begin
     Result := Broadcast(PT(fieldPtr)^);
 end;
 
-{ TMycGenericProcessor<T> }
+{ TMycGenericConsumer<T> }
 
-constructor TMycGenericProcessor<T>.Create(const AProc: TConstFunc<T, TState>);
-begin
-    inherited Create;
-    FProc := AProc;
-end;
-
-function TMycGenericProcessor<T>.ProcessData(const Value: T): TState;
-begin
-    Result := FProc(Value);
-end;
-
-{ TMycContainedProcessor<T> }
-
-constructor TMycContainedProcessor<T>.Create(const Controller: IInterface; const AProc: TConstFunc<T, TState>);
+constructor TMycGenericConsumer<T>.Create(const Controller: IInterface; const AProc: TConstFunc<T, TState>);
 begin
     inherited Create(Controller);
     FProc := AProc;
 end;
 
-function TMycContainedProcessor<T>.ProcessData(const Value: T): TState;
+function TMycGenericConsumer<T>.Consume(const Value: T): TState;
 begin
     Result := FProc(Value);
 end;
 
 { TMycDataEndpoint<T> }
 
-constructor TMycDataEndpoint<T>.Create(const ADataProvider: TDataProvider<T>; ALookback: Int64);
+constructor TMycDataEndpoint<T>.Create(const AProducer: TProducer<T>; ALookback: Int64);
 begin
     inherited Create;
-    FDataProvider := ADataProvider;
+    FProducer := AProducer;
     FLookback := ALookback;
 
-    FProcessor := TMycContainedProcessor<T>.Create(Self, ProcessData);
-    FTag := FDataProvider.Link(FProcessor);
+    FConsumer := TMycGenericConsumer<T>.Create(Self, Consume);
+    FTag := FProducer.Link(FConsumer);
 end;
 
 destructor TMycDataEndpoint<T>.Destroy;
 begin
-    FDataProvider.Unlink(FTag);
-    FProcessor.Free;
+    FProducer.Unlink(FTag);
+    FConsumer.Free;
     inherited;
 end;
 
@@ -513,7 +508,7 @@ begin
     Result := FChanged.State
 end;
 
-function TMycDataEndpoint<T>.ProcessData(const Value: T): TState;
+function TMycDataEndpoint<T>.Consume(const Value: T): TState;
 begin
     FLock.BeginWrite;
     try
@@ -573,15 +568,13 @@ function TTickAggregation.GetBarStartTime(const TimeStamp: TDateTime; const Time
 var
     baseTime: TDateTime;
 begin
-    // Align the time grid to UTC 0:00 using functions from System.DateUtils
+    // Implementation is unchanged
     baseTime := RecodeMilliSecond(TimeStamp, 0);
-
     case Timeframe of
         S: Result := baseTime;
         S5: Result := RecodeSecond(baseTime, SecondOf(TimeStamp) - (SecondOf(TimeStamp) mod 5));
         S15: Result := RecodeSecond(baseTime, SecondOf(TimeStamp) - (SecondOf(TimeStamp) mod 15));
         S30: Result := RecodeSecond(baseTime, SecondOf(TimeStamp) - (SecondOf(TimeStamp) mod 30));
-
         M: Result := RecodeSecond(baseTime, 0);
         M2: Result := RecodeMinute(RecodeSecond(baseTime, 0), MinuteOf(TimeStamp) - (MinuteOf(TimeStamp) mod 2));
         M3: Result := RecodeMinute(RecodeSecond(baseTime, 0), MinuteOf(TimeStamp) - (MinuteOf(TimeStamp) mod 3));
@@ -589,30 +582,21 @@ begin
         M10: Result := RecodeMinute(RecodeSecond(baseTime, 0), MinuteOf(TimeStamp) - (MinuteOf(TimeStamp) mod 10));
         M15: Result := RecodeMinute(RecodeSecond(baseTime, 0), MinuteOf(TimeStamp) - (MinuteOf(TimeStamp) mod 15));
         M30: Result := RecodeMinute(RecodeSecond(baseTime, 0), MinuteOf(TimeStamp) - (MinuteOf(TimeStamp) mod 30));
-
         H: Result := RecodeMinute(RecodeSecond(baseTime, 0), 0);
         H2: Result := RecodeHour(RecodeMinute(RecodeSecond(baseTime, 0), 0), HourOf(TimeStamp) - (HourOf(TimeStamp) mod 2));
         H3: Result := RecodeHour(RecodeMinute(RecodeSecond(baseTime, 0), 0), HourOf(TimeStamp) - (HourOf(TimeStamp) mod 3));
         H4: Result := RecodeHour(RecodeMinute(RecodeSecond(baseTime, 0), 0), HourOf(TimeStamp) - (HourOf(TimeStamp) mod 4));
         H8: Result := RecodeHour(RecodeMinute(RecodeSecond(baseTime, 0), 0), HourOf(TimeStamp) - (HourOf(TimeStamp) mod 8));
         H12: Result := RecodeHour(RecodeMinute(RecodeSecond(baseTime, 0), 0), HourOf(TimeStamp) - (HourOf(TimeStamp) mod 12));
-
         D: Result := StartOfTheDay(TimeStamp);
-        // D2, D3 are uncommon; this is a simple modulo-based approach relative to TDateTime's epoch.
         D2: Result := Floor(TimeStamp) - (Floor(TimeStamp) mod 2);
         D3: Result := Floor(TimeStamp) - (Floor(TimeStamp) mod 3);
-
         W: Result := TimeStamp.StartOfTheWeek;
-
         MN: Result := TimeStamp.StartOfTheMonth;
-        // Quarter alignment
         MN3: Result := RecodeMonth(TimeStamp.StartOfTheMonth, (MonthOf(TimeStamp) - 1) div 3 * 3 + 1);
-        // Half-year alignment
         MN6: Result := RecodeMonth(TimeStamp.StartOfTheMonth, (MonthOf(TimeStamp) - 1) div 6 * 6 + 1);
-
         Y: Result := TimeStamp.StartOfTheYear;
     else
-        // Fallback for any undefined timeframe
         Result := 0;
     end;
 end;
@@ -627,24 +611,21 @@ begin
     Result := FTimeframe;
 end;
 
-function TTickAggregation.ProcessData(const Value: TDataPoint<Double>): TState;
+function TTickAggregation.Consume(const Value: TDataPoint<Double>): TState;
 var
     barStartTime: TDateTime;
     lastBarTime: TDateTime;
 begin
-    // Update bar for the strategy's timeframe
     barStartTime := GetBarStartTime(Value.Time, FTimeframe);
     lastBarTime := FCurrentBar.Time;
 
     if (barStartTime > lastBarTime) then
     begin
-        // A new bar starts, so the previous one is now complete.
         if (lastBarTime > 0) then
         begin
             Result := Broadcast(FCurrentBar);
         end;
 
-        // Start a new bar, Volume is 1 because this is the first tick.
         FCurrentBar.Data.Open := Value.Data;
         FCurrentBar.Data.High := Value.Data;
         FCurrentBar.Data.Low := Value.Data;
@@ -654,13 +635,11 @@ begin
     end
     else
     begin
-        // Update the currently aggregating bar
         if Value.Data > FCurrentBar.Data.High then
             FCurrentBar.Data.High := Value.Data;
         if Value.Data < FCurrentBar.Data.Low then
             FCurrentBar.Data.Low := Value.Data;
         FCurrentBar.Data.Close := Value.Data;
-        // Volume is the number of ticks needed to build the complete bar.
         FCurrentBar.Data.Volume := FCurrentBar.Data.Volume + 1;
     end;
 end;
@@ -671,44 +650,44 @@ constructor TMycSequence<T>.Create(ACount: Integer);
 begin
     inherited Create;
 
-    SetLength(FDataProviders, ACount);
-    for var i := 0 to High(FDataProviders) do
-        FDataProviders[i] := TMycContainedDataProvider<T>.Create(Self);
+    SetLength(FProducers, ACount);
+    for var i := 0 to High(FProducers) do
+        FProducers[i] := TMycContainedProducer<T>.Create(Self);
 end;
 
 destructor TMycSequence<T>.Destroy;
 begin
-    for var i := High(FDataProviders) downto 0 do
-        FDataProviders[i].Free;
+    for var i := High(FProducers) downto 0 do
+        FProducers[i].Free;
     inherited;
 end;
 
-function TMycSequence<T>.GetCount: Integer;
+class function TMycSequence<T>.CreateSequence(Count: Integer; out Producers: TArray<IProducer<T>>): IConsumer<T>;
+var
+    Seq: TMycSequence<T>;
 begin
-    Result := Length(FDataProviders);
+    Seq := TMycSequence<T>.Create(Count);
+    Result := Seq;
+
+    SetLength(Producers, Length(Seq.FProducers));
+    for var i := 0 to High(Producers) do
+        Producers[i] := Seq.FProducers[i];
 end;
 
-function TMycSequence<T>.GetDataProvider(Idx: Integer): TDataProvider<T>;
+function TMycSequence<T>.Consume(const Value: T): TState;
 begin
-    Result := FDataProviders[Idx];
+    Result := ProcessProducer(0, Value);
 end;
 
-function TMycSequence<T>.ProcessData(const Value: T): TState;
+function TMycSequence<T>.ProcessProducer(Idx: Integer; const Value: T): TState;
 begin
-    Result := ProcessDataProvider(0, Value);
-end;
-
-function TMycSequence<T>.ProcessDataProvider(Idx: Integer; const Value: T): TState;
-begin
-    if Idx >= Length(FDataProviders) then
+    if Idx >= Length(FProducers) then
         exit;
 
-    Result :=
-        TaskManager
-            .RunTask(FDataProviders[idx].Broadcast(Value), function: TState begin Result := ProcessDataProvider(1 + idx, Value); end);
+    Result := TaskManager.RunTask(FProducers[idx].Broadcast(Value), function: TState begin Result := ProcessProducer(1 + idx, Value); end);
 end;
 
-function TMycIdentityConverter<T>.ProcessData(const Value: T): TState;
+function TMycIdentityConverter<T>.Consume(const Value: T): TState;
 begin
     Result := Broadcast(Value);
 end;
@@ -721,7 +700,7 @@ begin
     FFunc := AFunc;
 end;
 
-function TMycGenericParallelConverter<S, T>.ProcessData(const Value: S): TState;
+function TMycGenericParallelConverter<S, T>.Consume(const Value: S): TState;
 begin
     var cValue := Value;
     Result := TaskManager.RunTask(FQueue, function: TState begin Result := Broadcast(FFunc(cValue)); end);
@@ -729,7 +708,9 @@ begin
     FQueue := Result;
 end;
 
-function TMycParallelConverter<T>.ProcessData(const Value: T): TState;
+{ TMycParallelConverter<T> }
+
+function TMycParallelConverter<T>.Consume(const Value: T): TState;
 begin
     var cValue := Value;
     Result := TaskManager.RunTask(FQueue, function: TState begin Result := Broadcast(cValue); end);
@@ -745,18 +726,14 @@ begin
 end;
 
 function TOhlcAggregation.GetBarStartTime(const TimeStamp: TDateTime; const Timeframe: TTimeframe): TDateTime;
-var
-    baseTime: TDateTime;
 begin
-    // Align the time grid to UTC 0:00 using functions from System.DateUtils
-    baseTime := RecodeMilliSecond(TimeStamp, 0);
-
+    // Same implementation as TTickAggregation
+    var baseTime := RecodeMilliSecond(TimeStamp, 0);
     case Timeframe of
         S: Result := baseTime;
         S5: Result := RecodeSecond(baseTime, SecondOf(TimeStamp) - (SecondOf(TimeStamp) mod 5));
         S15: Result := RecodeSecond(baseTime, SecondOf(TimeStamp) - (SecondOf(TimeStamp) mod 15));
         S30: Result := RecodeSecond(baseTime, SecondOf(TimeStamp) - (SecondOf(TimeStamp) mod 30));
-
         M: Result := RecodeSecond(baseTime, 0);
         M2: Result := RecodeMinute(RecodeSecond(baseTime, 0), MinuteOf(TimeStamp) - (MinuteOf(TimeStamp) mod 2));
         M3: Result := RecodeMinute(RecodeSecond(baseTime, 0), MinuteOf(TimeStamp) - (MinuteOf(TimeStamp) mod 3));
@@ -764,30 +741,21 @@ begin
         M10: Result := RecodeMinute(RecodeSecond(baseTime, 0), MinuteOf(TimeStamp) - (MinuteOf(TimeStamp) mod 10));
         M15: Result := RecodeMinute(RecodeSecond(baseTime, 0), MinuteOf(TimeStamp) - (MinuteOf(TimeStamp) mod 15));
         M30: Result := RecodeMinute(RecodeSecond(baseTime, 0), MinuteOf(TimeStamp) - (MinuteOf(TimeStamp) mod 30));
-
         H: Result := RecodeMinute(RecodeSecond(baseTime, 0), 0);
         H2: Result := RecodeHour(RecodeMinute(RecodeSecond(baseTime, 0), 0), HourOf(TimeStamp) - (HourOf(TimeStamp) mod 2));
         H3: Result := RecodeHour(RecodeMinute(RecodeSecond(baseTime, 0), 0), HourOf(TimeStamp) - (HourOf(TimeStamp) mod 3));
         H4: Result := RecodeHour(RecodeMinute(RecodeSecond(baseTime, 0), 0), HourOf(TimeStamp) - (HourOf(TimeStamp) mod 4));
         H8: Result := RecodeHour(RecodeMinute(RecodeSecond(baseTime, 0), 0), HourOf(TimeStamp) - (HourOf(TimeStamp) mod 8));
         H12: Result := RecodeHour(RecodeMinute(RecodeSecond(baseTime, 0), 0), HourOf(TimeStamp) - (HourOf(TimeStamp) mod 12));
-
         D: Result := StartOfTheDay(TimeStamp);
-        // D2, D3 are uncommon; this is a simple modulo-based approach relative to TDateTime's epoch.
         D2: Result := Floor(TimeStamp) - (Floor(TimeStamp) mod 2);
         D3: Result := Floor(TimeStamp) - (Floor(TimeStamp) mod 3);
-
         W: Result := TimeStamp.StartOfTheWeek;
-
         MN: Result := TimeStamp.StartOfTheMonth;
-        // Quarter alignment
         MN3: Result := RecodeMonth(TimeStamp.StartOfTheMonth, (MonthOf(TimeStamp) - 1) div 3 * 3 + 1);
-        // Half-year alignment
         MN6: Result := RecodeMonth(TimeStamp.StartOfTheMonth, (MonthOf(TimeStamp) - 1) div 6 * 6 + 1);
-
         Y: Result := TimeStamp.StartOfTheYear;
     else
-        // Fallback for any undefined timeframe
         Result := 0;
     end;
 end;
@@ -802,141 +770,250 @@ begin
     Result := FTimeframe;
 end;
 
-function TOhlcAggregation.ProcessData(const Value: TDataPoint<TOhlcItem>): TState;
+function TOhlcAggregation.Consume(const Value: TDataPoint<TOhlcItem>): TState;
 var
     barStartTime: TDateTime;
     lastBarTime: TDateTime;
 begin
-    // Update bar for the strategy's timeframe
     barStartTime := GetBarStartTime(Value.Time, FTimeframe);
     lastBarTime := FCurrentBar.Time;
 
     if (barStartTime > lastBarTime) then
     begin
-        // A new bar starts, so the previous one is now complete.
         if (lastBarTime > 0) then
         begin
             Result := Broadcast(FCurrentBar);
         end;
 
-        // Start a new bar, Volume is 1 because this is the first tick.
         FCurrentBar.Data := Value.Data;
         FCurrentBar.Time := barStartTime;
     end
     else
     begin
-        // Update the currently aggregating bar
         if Value.Data.High > FCurrentBar.Data.High then
             FCurrentBar.Data.High := Value.Data.High;
         if Value.Data.Low < FCurrentBar.Data.Low then
             FCurrentBar.Data.Low := Value.Data.Low;
         FCurrentBar.Data.Close := Value.Data.Close;
-        // Volume is the number of ticks needed to build the complete bar.
         FCurrentBar.Data.Volume := FCurrentBar.Data.Volume + Value.Data.Volume;
     end;
 end;
 
 { TMycDataJoin<T> }
 
-constructor TMycDataJoin<T>.Create(const ADataProviders: TArray<TDataProvider<T>>);
+constructor TMycDataJoin<T>.Create(const AProducers: TArray<TProducer<T>>);
 begin
     inherited Create;
 
-    SetLength(FReceivers, Length(ADataProviders));
+    SetLength(FConsumers, Length(AProducers));
 
     FLock := TSpinLock.Create(false);
-    FCount := Length(FReceivers);
+    FCount := Length(FConsumers);
 
-    FDataProvider := TMycContainedDataProvider<TArray<T>>.Create(Self);
+    FContainedProvider := TMycContainedProducer<TArray<T>>.Create(Self);
 
     var cFunc :=
         function(Idx: Integer): TConstFunc<T, TState>
         begin
-            Result := function(const Value: T): TState begin ProcessData(Idx, Value); end
+            Result := function(const Value: T): TState begin Result := Consume(Idx, Value); end
         end;
 
-    for var i := 0 to High(FReceivers) do
-        with FReceivers[i] do
+    for var i := 0 to High(FConsumers) do
+        with FConsumers[i] do
         begin
             Queue := TQueue<T>.Create;
-            DataProvider := ADataProviders[i];
-            Processor := TMycContainedProcessor<T>.Create(Self, cFunc(i));
-            Tag := DataProvider.Link(Processor);
+            Producer := AProducers[i];
+            Consumer := TMycGenericConsumer<T>.Create(Self, cFunc(i));
+            Tag := Producer.Link(Consumer);
         end;
 end;
 
 destructor TMycDataJoin<T>.Destroy;
 begin
-    for var i := High(FReceivers) downto 0 do
-        with FReceivers[i] do
+    for var i := High(FConsumers) downto 0 do
+        with FConsumers[i] do
         begin
-            DataProvider.Unlink(FReceivers[i].Tag);
-            Processor.Free;
+            Producer.Unlink(FConsumers[i].Tag);
+            Consumer.Free;
             Queue.Free;
         end;
 
-    FDataProvider.Free;
+    FContainedProvider.Free;
     inherited;
 end;
 
-function TMycDataJoin<T>.ProcessData(Idx: Integer; const Value: T): TState;
+function TMycDataJoin<T>.Link(const Consumer: IConsumer<TArray<T>>): TTag;
+begin
+    Result := FContainedProvider.Link(Consumer);
+end;
+
+procedure TMycDataJoin<T>.Unlink(Tag: TTag);
+begin
+    FContainedProvider.Unlink(Tag);
+end;
+
+function TMycDataJoin<T>.Consume(Idx: Integer; const Value: T): TState;
 begin
     var Arr: TArray<T>;
 
     FLock.Enter;
     try
-        if FReceivers[Idx].Queue.Count = 0 then
+        if FConsumers[Idx].Queue.Count = 0 then
             dec(FCount);
 
         if FCount = 0 then
         begin
-            SetLength(Arr, Length(FReceivers));
+            SetLength(Arr, Length(FConsumers));
             Arr[Idx] := Value;
-            FCount := Length(FReceivers);
-            for var i := 0 to High(FReceivers) do
+            FCount := Length(FConsumers);
+            for var i := 0 to High(FConsumers) do
                 if i <> Idx then
                 begin
-                    Arr[i] := FReceivers[i].Queue.Dequeue;
-                    if FReceivers[i].Queue.Count > 0 then
+                    Arr[i] := FConsumers[i].Queue.Dequeue;
+                    if FConsumers[i].Queue.Count > 0 then
                         dec(FCount);
                 end;
         end
         else
-            FReceivers[Idx].Queue.Enqueue(Value);
+            FConsumers[Idx].Queue.Enqueue(Value);
     finally
         FLock.Exit;
     end;
 
     if Arr <> nil then
-        FDataProvider.Broadcast(Arr);
+        FContainedProvider.Broadcast(Arr);
 end;
 
+{ TMycGenericAggregator<S, T> }
+
 constructor TMycGenericAggregator<S, T>.Create(const AFunc: TConstFunc<S, TConverter<S, T>.TBroadcastProc, TState>);
 begin
     inherited Create;
     FFunc := AFunc;
 end;
 
-function TMycGenericAggregator<S, T>.ProcessData(const Value: S): TState;
+function TMycGenericAggregator<S, T>.Consume(const Value: S): TState;
 begin
-    Result := FFunc(Value, function(const Value: T): TState begin Result := FDataProvider.Broadcast(Value); end);
+    Result := FFunc(Value, Broadcast);
 end;
 
-constructor TMycComposedConverter<S, T>.Create(const AProcessor: IMycProcessor<S>; const ADataProvider: TDataProvider<T>);
+{ TMycComposedConverter<S, T> }
+
+constructor TMycComposedConverter<S, T>.Create(const AConsumer: IConsumer<S>; const AProducer: TProducer<T>);
 begin
     inherited Create;
-    FProcessor := AProcessor;
-    FDataProvider := ADataProvider;
+    FConsumer := AConsumer;
+    FProducer := TProducer<T>(AProducer);
 end;
 
-function TMycComposedConverter<S, T>.GetDataProvider: IDataProvider<T>;
+function TMycComposedConverter<S, T>.GetConsumer: IConsumer<S>;
 begin
-    Result := FDataProvider;
+    Result := FConsumer;
 end;
 
-function TMycComposedConverter<S, T>.ProcessData(const Value: S): TState;
+function TMycComposedConverter<S, T>.Link(const Consumer: IConsumer<T>): TTag;
 begin
-    Result := FProcessor.ProcessData(Value);
+    // Delegate to the contained producer
+    Result := IProducer<T>(FProducer).Link(Consumer);
+end;
+
+procedure TMycComposedConverter<S, T>.Unlink(Tag: TTag);
+begin
+    // Delegate to the contained producer
+    IProducer<T>(FProducer).Unlink(Tag);
+end;
+
+{ TMycProducer<T> }
+
+constructor TMycProducer<T>.Create;
+begin
+    inherited Create;
+end;
+
+class constructor TMycProducer<T>.CreateClass;
+begin
+    FNull := TNull.Create;
+end;
+
+destructor TMycProducer<T>.Destroy;
+begin
+    FListeners.Finalize;
+    inherited Destroy;
+end;
+
+function TMycProducer<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.Consume(Value).Signal.Subscribe(done);
+            item := item.Prev;
+        end;
+
+        Result := done.State;
+    finally
+        FListeners.Release;
+    end;
+end;
+
+function TMycProducer<T>.Link(const Consumer: IConsumer<T>): TTag;
+begin
+    // Add the Consumer to the notification list
+    FListeners.Lock;
+    try
+        Result := FListeners.Advise(Consumer);
+    finally
+        FListeners.Release;
+    end;
+end;
+
+procedure TMycProducer<T>.Unlink(Tag: TTag);
+begin
+    FListeners.Lock;
+    try
+        FListeners.Unadvise(Tag);
+    finally
+        FListeners.Release;
+    end;
+end;
+
+function TMycConsumer<T>.TNull.Consume(const Value: T): TState;
+begin
+    Result := TState.Null;
+end;
+
+function TMycProducer<T>.TNull.Link(const Consumer: IConsumer<T>): TTag;
+begin
+    Result := nil;
+end;
+
+procedure TMycProducer<T>.TNull.Unlink(Tag: TTag);
+begin
+
+end;
+
+constructor TMycConverter<S, T>.TConsumer.Create(AOwner: TMycConverter<S, T>);
+begin
+    inherited Create(AOwner);
+    FOwner := AOwner;
+end;
+
+function TMycConverter<S, T>.TConsumer.Consume(const Value: S): TState;
+begin
+    Result := FOwner.Consume(Value);
 end;
 
 end.
diff --git a/Src/Myc.Trade.DataPoint.pas b/Src/Myc.Trade.DataPoint.pas
index 23f4a82..91358ac 100644
--- a/Src/Myc.Trade.DataPoint.pas
+++ b/Src/Myc.Trade.DataPoint.pas
@@ -10,74 +10,68 @@ uses
     Myc.DataRecord;
 
 type
-    // A generic interface for components that process data of type T.
-    IMycProcessor<T> = interface
-        function ProcessData(const Value: T): TState;
+    TTag = Pointer;
+
+    // A generic interface for components that consume data of type T.
+    IConsumer<T> = interface
+        function Consume(const Value: T): TState;
     end;
 
-    TTag = Pointer;
-    IDataProvider<T> = interface
-        function Link(const Receiver: IMycProcessor<T>): TTag;
+    IProducer<T> = interface
+        function Link(const Consumer: IConsumer<T>): TTag;
         procedure Unlink(Tag: TTag);
     end;
 
-    IConverter<S, T> = interface(IMycProcessor<S>)
+    // A converter is a producer that internally uses a consumer to transform data.
+    IConverter<S, T> = interface(IProducer<T>)
         {$region 'private'}
-        function GetDataProvider: IDataProvider<T>;
+        function GetConsumer: IConsumer<S>;
         {$endregion}
-        property DataProvider: IDataProvider<T> read GetDataProvider;
+        property Consumer: IConsumer<S> read GetConsumer;
     end;
 
-    // Interface helper for IDataProvider providing the null object pattern.
-    TDataProvider<T> = record
+    // Interface helper for IProducer providing the null object pattern.
+    TProducer<T> = record
+    public
         type
             TSubscription = record
             private
-                FDataProvider: IDataProvider<T>;
+                FProducer: IProducer<T>;
                 FTag: TTag;
             public
                 procedure Unlink;
-                property DataProvider: IDataProvider<T> read FDataProvider;
+                property Producer: IProducer<T> read FProducer;
             end;
-
-    strict private
-        class var
-            FNull: IDataProvider<T>;
-        class constructor CreateClass;
-
     private
-        FDataProvider: IDataProvider<T>;
+        FProducer: IProducer<T>;
+        class function GetNull: IProducer<T>; static; inline;
 
     public
-        constructor Create(const ADataProvider: IDataProvider<T>);
+        constructor Create(const AProducer: IProducer<T>);
 
         // Managed record operators
-        class operator Initialize(out Dest: TDataProvider<T>);
-        class operator Implicit(const A: IDataProvider<T>): TDataProvider<T>; overload;
-        class operator Implicit(const A: TDataProvider<T>): IDataProvider<T>; overload;
+        class operator Initialize(out Dest: TProducer<T>);
+        class operator Implicit(const A: IProducer<T>): TProducer<T>; overload;
+        class operator Implicit(const A: TProducer<T>): IProducer<T>; overload;
 
-        function Link(const Receiver: IMycProcessor<T>): TSubscription; inline;
+        function Link(const Consumer: IConsumer<T>): TSubscription; inline;
 
-        function Chain(const Next: IMycProcessor<T>): IMycProcessor<T>; overload; inline;
-        function Chain<R>(const Next: IConverter<T, R>): TDataProvider<R>; overload; inline;
-        function Chain<R>(const Func: TConstFunc<T, R>): TDataProvider<R>; overload; inline;
+        function Chain(const Next: IConsumer<T>): IConsumer<T>; overload; inline;
+
+        function Chain<R>(const Next: IConverter<T, R>): TProducer<R>; overload; inline;
+        function Chain<R>(const Func: TConstFunc<T, R>): TProducer<R>; overload; inline;
 
         function CreateEndpoint(Lookback: Int64): TLazy<TSeries<T>>;
 
-        // Extracts the field of a record by it's name (using RTTI).
-        function Field<R>(const FieldName: String): TDataProvider<R>; inline;
+        function CreateSequence(Count: Integer): TArray<IProducer<T>>; overload; experimental;
 
-        function MakeParallel: TDataProvider<T>; inline;
+        // Extracts the field of a record by it's name (using RTTI).
+        function Field<R>(const FieldName: String): TProducer<R>; inline;
+
+        function MakeParallel: TProducer<T>; inline;
 
         // Provides access to the null object instance.
-        class property Null: IDataProvider<T> read FNull;
-    end;
-
-    IMycDataSequence<T> = interface(IMycProcessor<T>)
-        function GetCount: Integer;
-        function GetDataProvider(Idx: Integer): TDataProvider<T>;
-        property Count: Integer read GetCount;
-        property DataProvider[Idx: Integer]: TDataProvider<T> read GetDataProvider; default;
+        class property Null: IProducer<T> read GetNull;
     end;
 
     // Interface helper for IConverter<S,T> providing the null object pattern.
@@ -85,14 +79,11 @@ type
     public
         type
             TBroadcastProc = reference to function(const Value: T): TState;
-
-    strict private
-        class var
-            FNull: IConverter<S, T>;
-        class constructor CreateClass;
     private
         FConverter: IConverter<S, T>;
-        function GetDataProvider: TDataProvider<T>; inline;
+        function GetConsumer: IConsumer<S>; inline;
+        function GetProducer: TProducer<T>; inline;
+        class function GetNull: IConverter<S, T>; static; inline;
     public
         constructor Create(const AConverter: IConverter<S, T>);
 
@@ -101,17 +92,14 @@ type
         class operator Implicit(const A: IConverter<S, T>): TConverter<S, T>; overload;
         class operator Implicit(const A: TConverter<S, T>): IConverter<S, T>; overload;
 
-        class function Construct(const Processor: IMycProcessor<S>; const DataProvider: TDataProvider<T>): TConverter<S, T>; static;
+        class function Construct(const Consumer: IConsumer<S>; const Producer: TProducer<T>): TConverter<S, T>; static;
         class function CreateGeneric(const Func: TConstFunc<S, T>): TConverter<S, T>; static;
         class function CreateAggregation(const Func: TConstFunc<S, TBroadcastProc, TState>): TConverter<S, T>; static;
 
-        function CreateSequence(Count: Integer): IMycDataSequence<T>; overload;
-        function Sequence(const Items: TArray<IConverter<S, T>>): IMycDataSequence<S>; overload;
+        property Consumer: IConsumer<S> read GetConsumer;
+        property Producer: TProducer<T> read GetProducer;
 
-        // Provides access to the null object instance.
-        class property Null: IConverter<S, T> read FNull;
-        // Wrapper for IConverter.DataProvider
-        property DataProvider: TDataProvider<T> read GetDataProvider;
+        class property Null: IConverter<S, T> read GetNull;
     end;
 
     // Factory for creating specific converter instances.
@@ -126,12 +114,10 @@ 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 FieldToRecord<T>(const Layout: TDataRecord.TLayout; const Name: String): TConverter<T, TDataRecord>; static;
         class function FieldOfRecord<T>(const Layout: TDataRecord.TLayout; const Name: String): TConverter<TDataRecord, T>; static;
 
-        class function Join<T>(const DataProviders: TArray<TDataProvider<T>>): TDataProvider<TArray<T>>; static;
+        class function Join<T>(const Producers: TArray<TProducer<T>>): TProducer<TArray<T>>; static;
 
         type
             TRecordMapping = record
@@ -149,7 +135,7 @@ type
         class function JoinRecords(
             const TargetLayout: TDataRecord.TLayout;
             const Mapping: TArray<TRecordMapping>;
-            const DataProviders: TArray<TDataProvider<TDataRecord>>
+            const Producers: TArray<TProducer<TDataRecord>>
         ): TConverter<TArray<TDataRecord>, TDataRecord>; static;
     end;
 
@@ -158,100 +144,98 @@ implementation
 uses
     Myc.Trade.DataPoint.Impl;
 
-{ TDataProvider<T> }
+{ TProducer<T> }
 
-procedure TDataProvider<T>.TSubscription.Unlink;
+procedure TProducer<T>.TSubscription.Unlink;
 begin
     if FTag <> nil then
     begin
-        FDataProvider.Unlink(FTag);
+        FProducer.Unlink(FTag);
         FTag := nil;
     end;
 end;
 
-class constructor TDataProvider<T>.CreateClass;
+constructor TProducer<T>.Create(const AProducer: IProducer<T>);
 begin
-    // Create the singleton null object instance.
-    FNull := TNullDataProvider<T>.Create;
+    FProducer := AProducer;
+    if not Assigned(FProducer) then
+        FProducer := Null;
 end;
 
-constructor TDataProvider<T>.Create(const ADataProvider: IDataProvider<T>);
+function TProducer<T>.Chain(const Next: IConsumer<T>): IConsumer<T>;
 begin
-    FDataProvider := ADataProvider;
-    if not Assigned(FDataProvider) then
-        FDataProvider := FNull;
-end;
-
-function TDataProvider<T>.Chain(const Next: IMycProcessor<T>): IMycProcessor<T>;
-begin
-    FDataProvider.Link(Next);
+    FProducer.Link(Next);
     Result := Next;
 end;
 
-function TDataProvider<T>.Chain<R>(const Next: IConverter<T, R>): TDataProvider<R>;
+function TProducer<T>.Chain<R>(const Next: IConverter<T, R>): TProducer<R>;
 begin
-    FDataProvider.Link(Next);
-    Result := Next.DataProvider;
+    // REFACTOR: Link to the consumer part, return the converter as the new producer
+    FProducer.Link(Next.Consumer);
+    Result := Next;
 end;
 
-function TDataProvider<T>.Chain<R>(const Func: TConstFunc<T, R>): TDataProvider<R>;
+function TProducer<T>.Chain<R>(const Func: TConstFunc<T, R>): TProducer<R>;
 begin
     Result := Chain<R>(TMycGenericConverter<T, R>.Create(Func));
 end;
 
-function TDataProvider<T>.CreateEndpoint(Lookback: Int64): TLazy<TSeries<T>>;
+function TProducer<T>.CreateEndpoint(Lookback: Int64): TLazy<TSeries<T>>;
 begin
-    Result := TMycDataEndpoint<T>.Create(FDataProvider, Lookback);
+    Result := TMycDataEndpoint<T>.Create(FProducer, Lookback);
 end;
 
-function TDataProvider<T>.Field<R>(const FieldName: String): TDataProvider<R>;
+function TProducer<T>.CreateSequence(Count: Integer): TArray<IProducer<T>>;
+begin
+    FProducer.Link(TMycSequence<T>.CreateSequence(Count, Result));
+end;
+
+function TProducer<T>.Field<R>(const FieldName: String): TProducer<R>;
 begin
     Result := Chain<R>(TMycRecordFieldReader<T, R>.Create(FieldName));
 end;
 
-class operator TDataProvider<T>.Initialize(out Dest: TDataProvider<T>);
+class function TProducer<T>.GetNull: IProducer<T>;
 begin
-    Dest.FDataProvider := FNull;
+    Result := TMycProducer<T>.Null;
 end;
 
-class operator TDataProvider<T>.Implicit(const A: IDataProvider<T>): TDataProvider<T>;
+class operator TProducer<T>.Initialize(out Dest: TProducer<T>);
+begin
+    Dest.FProducer := Null;
+end;
+
+class operator TProducer<T>.Implicit(const A: IProducer<T>): TProducer<T>;
 begin
     Result.Create(A);
 end;
 
-class operator TDataProvider<T>.Implicit(const A: TDataProvider<T>): IDataProvider<T>;
+class operator TProducer<T>.Implicit(const A: TProducer<T>): IProducer<T>;
 begin
-    Result := A.FDataProvider;
+    Result := A.FProducer;
 end;
 
-function TDataProvider<T>.Link(const Receiver: IMycProcessor<T>): TSubscription;
+function TProducer<T>.Link(const Consumer: IConsumer<T>): TSubscription;
 begin
-    Result.FDataProvider := FDataProvider;
-    Result.FTag := FDataProvider.Link(Receiver);
+    Result.FProducer := FProducer;
+    Result.FTag := FProducer.Link(Consumer);
 end;
 
-function TDataProvider<T>.MakeParallel: TDataProvider<T>;
+function TProducer<T>.MakeParallel: TProducer<T>;
 begin
     Result := Chain<T>(TMycParallelConverter<T>.Create as IConverter<T, T>);
 end;
 
-{ TConverter<S, T> }
-
-class constructor TConverter<S, T>.CreateClass;
-begin
-    FNull := TNullConverter<S, T>.Create;
-end;
-
 constructor TConverter<S, T>.Create(const AConverter: IConverter<S, T>);
 begin
     FConverter := AConverter;
     if not Assigned(FConverter) then
-        FConverter := FNull;
+        FConverter := Null;
 end;
 
-class function TConverter<S, T>.Construct(const Processor: IMycProcessor<S>; const DataProvider: TDataProvider<T>): TConverter<S, T>;
+class function TConverter<S, T>.Construct(const Consumer: IConsumer<S>; const Producer: TProducer<T>): TConverter<S, T>;
 begin
-    Result := TMycComposedConverter<S, T>.Create(Processor, DataProvider);
+    Result := TMycComposedConverter<S, T>.Create(Consumer, Producer);
 end;
 
 class function TConverter<S, T>.CreateAggregation(const Func: TConstFunc<S, TBroadcastProc, TState>): TConverter<S, T>;
@@ -264,30 +248,24 @@ begin
     Result := TMycGenericConverter<S, T>.Create(Func);
 end;
 
-function TConverter<S, T>.GetDataProvider: TDataProvider<T>;
+function TConverter<S, T>.GetConsumer: IConsumer<S>;
 begin
-    Result := FConverter.DataProvider;
+    Result := FConverter.Consumer;
 end;
 
-function TConverter<S, T>.CreateSequence(Count: Integer): IMycDataSequence<T>;
+function TConverter<S, T>.GetProducer: TProducer<T>;
 begin
-    Result := TMycSequence<T>.Create(Count);
-    FConverter.DataProvider.Link(Result);
+    Result := FConverter;
 end;
 
-function TConverter<S, T>.Sequence(const Items: TArray<IConverter<S, T>>): IMycDataSequence<S>;
+class function TConverter<S, T>.GetNull: IConverter<S, T>;
 begin
-    var seq: IMycDataSequence<S> := TMycSequence<S>.Create(Length(Items));
-
-    for var i := 0 to High(Items) do
-        seq[i].Link(Items[i]);
-
-    Result := seq;
+    Result := TMycConverter<S, T>.Null;
 end;
 
 class operator TConverter<S, T>.Initialize(out Dest: TConverter<S, T>);
 begin
-    Dest.FConverter := FNull;
+    Dest.FConverter := Null;
 end;
 
 class operator TConverter<S, T>.Implicit(const A: IConverter<S, T>): TConverter<S, T>;
@@ -340,19 +318,6 @@ begin
     Result := TMycTicker<T>.Create;
 end;
 
-class function TConverter.CreateSequence<T>(Count: Integer; const Parent: TDataProvider<T>): TArray<TConverter<T, T>>;
-begin
-    var seq: IMycDataSequence<T> := TMycSequence<T>.Create(Count);
-
-    SetLength(Result, Count);
-    for var i := 0 to High(Result) do
-    begin
-        Result[i] := TMycIdentityConverter<T>.Create;
-        seq[i].Link(Result[i]);
-    end;
-    Parent.Link(seq);
-end;
-
 class function TConverter.CreateOhlcAggregation(Timeframe: TTimeframe): TConverter<TDataPoint<TOhlcItem>, TDataPoint<TOhlcItem>>;
 begin
     Result := TOhlcAggregation.Create(Timeframe);
@@ -409,22 +374,23 @@ begin
         );
 end;
 
-class function TConverter.Join<T>(const DataProviders: TArray<TDataProvider<T>>): TDataProvider<TArray<T>>;
+class function TConverter.Join<T>(const Producers: TArray<TProducer<T>>): TProducer<TArray<T>>;
 begin
-    Result := TMycDataJoin<T>.Create(DataProviders);
+    Result := TMycDataJoin<T>.Create(Producers);
 end;
 
 class function TConverter.JoinRecords(
     const TargetLayout: TDataRecord.TLayout;
     const Mapping: TArray<TRecordMapping>;
-    const DataProviders: TArray<TDataProvider<TDataRecord>>
+    const Producers: TArray<TProducer<TDataRecord>>
 ): TConverter<TArray<TDataRecord>, TDataRecord>;
 begin
-    var RecProvider := Join<TDataRecord>(DataProviders);
+    var RecProvider := Join<TDataRecord>(Producers);
 
     Result := DataMapping(Mapping, TargetLayout);
 
-    RecProvider.Link(Result);
+    // REFACTOR: Link to the consumer part of the mapping converter.
+    RecProvider.Link(Result.Consumer);
 end;
 
 end.
diff --git a/Src/Myc.Trade.DataStream.pas b/Src/Myc.Trade.DataStream.pas
index 291f9d0..0ed1661 100644
--- a/Src/Myc.Trade.DataStream.pas
+++ b/Src/Myc.Trade.DataStream.pas
@@ -20,7 +20,7 @@ type
     IDataServer<T: record> = interface
         ['{1F8E5A9D-E92A-44C1-9F3F-C4B82A6E94B3}']
         procedure ClearCache;
-        function ProcessData(const Symbol: String; const Terminated: TState; const Processor: IMycProcessor<TArray<TDataPoint<T>>>): TState;
+        function ProcessData(const Symbol: String; const Terminated: TState; const Processor: IConsumer<TArray<TDataPoint<T>>>): TState;
         function EnumerateSymbols: TFuture<TArray<String>>;
     end;
 
@@ -60,14 +60,14 @@ type
         function ProcessChunks(
             const DataChunks: TArray<TArray<TDataPoint<T>>>;
             Terminated: TState;
-            Processor: IMycProcessor<TArray<TDataPoint<T>>>
+            Processor: IConsumer<TArray<TDataPoint<T>>>
         ): TState;
 
         function ProcessFile(
             FileInfo: TAuraDataFile;
             const DataFile: TFuture<TArray<TArray<TDataPoint<T>>>>;
             const Terminated: TState;
-            Processor: IMycProcessor<TArray<TDataPoint<T>>>
+            Processor: IConsumer<TArray<TDataPoint<T>>>
         ): TState;
 
     strict private
@@ -94,7 +94,7 @@ type
         // Load data file and split content into chunks.
         function LoadDataFile(const DataFile: TAuraDataFile): TFuture<TArray<TArray<TDataPoint<T>>>>;
 
-        function ProcessData(const Symbol: String; const Terminated: TState; const Processor: IMycProcessor<TArray<TDataPoint<T>>>): TState;
+        function ProcessData(const Symbol: String; const Terminated: TState; const Processor: IConsumer<TArray<TDataPoint<T>>>): TState;
 
         property Path: String read GetPath;
     end;
@@ -338,7 +338,7 @@ end;
 function TAuraDataServer<T>.ProcessData(
     const Symbol: String;
     const Terminated: TState;
-    const Processor: IMycProcessor<TArray<TDataPoint<T>>>
+    const Processor: IConsumer<TArray<TDataPoint<T>>>
 ): TState;
 begin
     var cProc := Processor;
@@ -364,7 +364,7 @@ end;
 function TAuraDataServer<T>.ProcessChunks(
     const DataChunks: TArray<TArray<TDataPoint<T>>>;
     Terminated: TState;
-    Processor: IMycProcessor<TArray<TDataPoint<T>>>
+    Processor: IConsumer<TArray<TDataPoint<T>>>
 ): TState;
 begin
     var callProcess :=
@@ -375,7 +375,7 @@ begin
                 function: TState
                 begin
                     if not Terminated.IsSet then
-                        Result := Processor.ProcessData(cChunk);
+                        Result := Processor.Consume(cChunk);
                 end;
         end;
 
@@ -391,7 +391,7 @@ function TAuraDataServer<T>.ProcessFile(
     FileInfo: TAuraDataFile;
     const DataFile: TFuture<TArray<TArray<TDataPoint<T>>>>;
     const Terminated: TState;
-    Processor: IMycProcessor<TArray<TDataPoint<T>>>
+    Processor: IConsumer<TArray<TDataPoint<T>>>
 ): TState;
 begin
     if not FileInfo.IsValid or Terminated.IsSet then