Concurrent data processing

AuraTrader/FirstStrategy.pas

@@ -6,6 +6,7 @@ uses
     System.Generics.Collections,
     Myc.Signals,
     Myc.Lazy,
+    Myc.TaskManager,
     Myc.Trade.DataPoint,
     Myc.Trade.DataArray;
 
@@ -18,7 +19,7 @@ type
     private
         FFunc: TConvertFunc;
     protected
-        function ProcessData(const Value: S): Boolean; override;
+        function ProcessData(const Value: S): TState; override;
     public
         constructor Create(const AFunc: TConvertFunc);
     end;
@@ -35,7 +36,7 @@ type
         constructor Create(const ATimeframe: TTimeframe);
 
         // Process new data. This is called concurrently and must not have side effects out of the scope of this class!
-        function ProcessData(const Values: TArray<TDataPoint<TAskBidItem>>): Boolean; override;
+        function ProcessData(const Values: TArray<TDataPoint<TAskBidItem>>): TState; override;
 
         property CurrentBar: TDataPoint<TOhlcItem> read GetCurrentBar;
         property Timeframe: TTimeframe read GetTimeframe;
@@ -51,10 +52,11 @@ type
         FSourceData: TMycDataArray<Double>;
         // Intermediate data series for HMA calculation (2*WMA(n/2) - WMA(n))
         FDiffSeries: TMycDataArray<Double>;
+        FQueue: TQueue;
         // Calculates the Weighted Moving Average for the most recent data.
         function CalculateWMA(const Series: TMycDataArray<Double>; const Period: Integer): Double;
     protected
-        function ProcessData(const Value: Double): Boolean; override;
+        function ProcessData(const Value: Double): TState; override;
     public
         constructor Create(const APeriod: Integer);
     end;
@@ -97,49 +99,53 @@ begin
     Result := FTimeframe;
 end;
 
-function TTicksToTimeframe.ProcessData(const Values: TArray<TDataPoint<TAskBidItem>>): Boolean;
+function TTicksToTimeframe.ProcessData(const Values: TArray<TDataPoint<TAskBidItem>>): TState;
 var
     point: TDataPoint<TAskBidItem>;
     midPrice: Single;
     barStartTime: TDateTime;
     lastBarTime: TDateTime;
     currentBar: TOhlcItem;
+    states: TList<TState>;
 begin
-    Result := true;
-
-    // Process each incoming data point
-    for point in Values do
-    begin
-        midPrice := (point.Data.Ask + point.Data.Bid) / 2;
-
-        // Update bar for the strategy's timeframe
-        barStartTime := GetBarStartTime(point.Time, FTimeframe);
-        lastBarTime := FCurrentBar.Time;
-
-        if (barStartTime > lastBarTime) then
+    states := TList<TState>.Create;
+    try
+        // Process each incoming data point
+        for point in Values do
         begin
-            // A new bar starts, so the previous one is now complete.
-            if (lastBarTime > 0) then
+            midPrice := (point.Data.Ask + point.Data.Bid) / 2;
+
+            // Update bar for the strategy's timeframe
+            barStartTime := GetBarStartTime(point.Time, FTimeframe);
+            lastBarTime := FCurrentBar.Time;
+
+            if (barStartTime > lastBarTime) then
             begin
-                Broadcast(FCurrentBar);
-            end;
+                // A new bar starts, so the previous one is now complete.
+                if (lastBarTime > 0) then
+                    states.Add(Broadcast(FCurrentBar));
 
-            // Start a new bar, Volume is 1 because this is the first tick.
-            currentBar := TOhlcItem.Create(midPrice, midPrice, midPrice, midPrice, 1);
-            FCurrentBar.Data := currentBar;
-            FCurrentBar.Time := barStartTime;
-        end
-        else
-        begin
-            // Update the currently aggregating bar
-            currentBar := FCurrentBar.Data;
-            currentBar.High := Max(currentBar.High, midPrice);
-            currentBar.Low := Min(currentBar.Low, midPrice);
-            currentBar.Close := midPrice;
-            // Volume is the number of ticks needed to build the complete bar.
-            currentBar.Volume := currentBar.Volume + 1;
-            FCurrentBar.Data := currentBar;
+                // Start a new bar, Volume is 1 because this is the first tick.
+                currentBar := TOhlcItem.Create(midPrice, midPrice, midPrice, midPrice, 1);
+                FCurrentBar.Data := currentBar;
+                FCurrentBar.Time := barStartTime;
+            end
+            else
+            begin
+                // Update the currently aggregating bar
+                currentBar := FCurrentBar.Data;
+                currentBar.High := Max(currentBar.High, midPrice);
+                currentBar.Low := Min(currentBar.Low, midPrice);
+                currentBar.Close := midPrice;
+                // Volume is the number of ticks needed to build the complete bar.
+                currentBar.Volume := currentBar.Volume + 1;
+                FCurrentBar.Data := currentBar;
+            end;
         end;
+
+        Result := TState.All(states.ToArray);
+    finally
+        states.Free;
     end;
 end;
 
@@ -183,43 +189,47 @@ begin
     Result := numerator / denominator;
 end;
 
-function THullMovingAverage.ProcessData(const Value: Double): Boolean;
-var
-    price: Double;
-    wmaHalf, wmaFull, diff: Double;
-    hma: Double;
+function THullMovingAverage.ProcessData(const Value: Double): TState;
 begin
-    Result := true;
+    Result :=
+        FQueue.Enqueue(
+            function: TState
+            var
+                price: Double;
+                wmaHalf, wmaFull, diff: Double;
+                hma: Double;
+            begin
+                price := Value;
 
-    price := Value;
+                // Default HMA to NaN for the warm-up period.
+                hma := Double.NaN;
 
-    // Default HMA to NaN for the warm-up period.
-    hma := Double.NaN;
+                // Add new price to the source data array, respecting the lookback period.
+                FSourceData := FSourceData.Add(price, FPeriod);
 
-    // Add new price to the source data array, respecting the lookback period.
-    FSourceData := FSourceData.Add(price, FPeriod);
+                // Check if there is enough data to start the first stage of calculation.
+                if (FSourceData.Count >= FPeriod) then
+                begin
+                    // Calculate the two WMAs for the first step.
+                    wmaHalf := CalculateWMA(FSourceData, FPeriodHalf);
+                    wmaFull := CalculateWMA(FSourceData, FPeriod);
 
-    // Check if there is enough data to start the first stage of calculation.
-    if (FSourceData.Count >= FPeriod) then
-    begin
-        // Calculate the two WMAs for the first step.
-        wmaHalf := CalculateWMA(FSourceData, FPeriodHalf);
-        wmaFull := CalculateWMA(FSourceData, FPeriod);
+                    // Calculate the difference and add to the intermediate series.
+                    diff := 2 * wmaHalf - wmaFull;
+                    FDiffSeries := FDiffSeries.Add(diff, FPeriodSqrt);
 
-        // Calculate the difference and add to the intermediate series.
-        diff := 2 * wmaHalf - wmaFull;
-        FDiffSeries := FDiffSeries.Add(diff, FPeriodSqrt);
+                    // Check if there is enough intermediate data for the final calculation.
+                    if (FDiffSeries.Count >= FPeriodSqrt) then
+                    begin
+                        // Calculate the final HMA value, overwriting the default 0.0.
+                        hma := CalculateWMA(FDiffSeries, FPeriodSqrt);
+                    end;
+                end;
 
-        // Check if there is enough intermediate data for the final calculation.
-        if (FDiffSeries.Count >= FPeriodSqrt) then
-        begin
-            // Calculate the final HMA value, overwriting the default 0.0.
-            hma := CalculateWMA(FDiffSeries, FPeriodSqrt);
-        end;
-    end;
-
-    // Broadcast the result
-    Broadcast(hma);
+                // Broadcast the result
+                Result := Broadcast(hma);
+            end
+        );
 end;
 
 { TMycGenericConverter<S, T> }
@@ -230,10 +240,9 @@ begin
     FFunc := AFunc;
 end;
 
-function TMycGenericConverter<S, T>.ProcessData(const Value: S): Boolean;
+function TMycGenericConverter<S, T>.ProcessData(const Value: S): TState;
 begin
-    Result := true;
-    Broadcast(FFunc(Value));
+    Result := Broadcast(FFunc(Value));
 end;
 
 end.

AuraTrader/MainForm.pas

@@ -182,9 +182,14 @@ begin
 
     for var i := 0 to 3 do
     begin
-        var Hull: IMycConverter<Double, Double> := THullMovingAverage.Create(50 + (50 * i));
+        var Hull: IMycConverter<Double, Double> := THullMovingAverage.Create(50 + (500 * i));
         Closes.Sender.Link(Hull);
-        chart.AddDoubleSeries(Hull.Sender, TAlphaColor(Random(Integer.MaxValue - 1)));
+        var col: TAlphaColorRec;
+        col.R := 25 * i;
+        col.G := 255 - 25 * i;
+        col.B := 100 + 5 * i;
+        col.A := 255;
+        chart.AddDoubleSeries(Hull.Sender, col.Color);
     end;
 
     //    var Hull: IMycConverter<Double, Double> := THullMovingAverage.Create(250);
@@ -193,7 +198,9 @@ begin
     //
     //    chart.AddDoubleSeries(Hull.Sender);
 
-    chart.SetXAxisSeries<TOhlcItem>(OHLC.Sender);
+    OhlcPoint.Sender.Link(TimeStamps);
+
+    chart.SetXAxisSeries<TDateTime>(Timestamps.Sender);
     chart.AddOhlcSeries(Ohlc.Sender);
 
     var done := ExecuteStrategy(Symbol, OhlcPoint);
@@ -444,9 +451,10 @@ begin
                         Symbol,
                         terminated,
                         TMycGenericProcessor<TArray<TDataPoint<TAuraAskBidFileItem>>>.Create(
-                            function(const Values: TArray<TDataPoint<TAuraAskBidFileItem>>): Boolean
+                            function(const Values: TArray<TDataPoint<TAuraAskBidFileItem>>): TState
                             begin
-                                Result := true;
+                                Result := TState.Null;
+
                                 Prices := Prices.Add(Values);
                                 currLog.Value := Prices.TotalCount.ToString;
 

Src/Myc.Core.Notifier.pas

@@ -37,6 +37,9 @@ type
         // Frees memory previously allocated for a TItem.
         class procedure FreeItem(Item: PItem); static; inline;
 
+    private
+        function GetFirst: PItem;
+
     public
         // Initializes the list in an unlocked state.
         class operator Initialize(out Dest: TMycNotifyList<T>);
@@ -58,6 +61,7 @@ type
         // If the predicate returns false, the receiver is detached from further notifications
         // by setting its interface reference to nil. The list item itself is not freed here.
         procedure Notify(const Func: TNotifyProc);
+        property First: PItem read GetFirst;
     end;
 
 implementation
@@ -112,6 +116,12 @@ begin
     FreeMem(Item, sizeof(TItem));
 end;
 
+function TMycNotifyList<T>.GetFirst: PItem;
+begin
+    Assert(IsLocked);
+    Result := FList;
+end;
+
 function TMycNotifyList<T>.IsLocked: Boolean;
 begin
     Result := NativeUInt(FList) and 1 = 0;

Src/Myc.Fmx.Chart.Series.pas

@@ -18,7 +18,7 @@ type
     private
         FCount: Int64;
     protected
-        function ProcessData(const Value: T): Boolean; override;
+        function ProcessData(const Value: T): TState; override;
     public
         constructor Create;
     end;
@@ -30,7 +30,7 @@ type
     private
         FData: TWriteable<TMycDataArray<T>>;
     protected
-        function ProcessData(const Value: T): Boolean; override;
+        function ProcessData(const Value: T): TState; override;
     public
         constructor Create(const ALookback: TMutable<Int64>);
         property Data: TWriteable<TMycDataArray<T>> read FData;
@@ -116,10 +116,9 @@ begin
     FCount := 0;
 end;
 
-function TChartSeriesCounter<T>.ProcessData(const Value: T): Boolean;
+function TChartSeriesCounter<T>.ProcessData(const Value: T): TState;
 begin
-    Result := true;
-    Broadcast(FCount);
+    Result := Broadcast(FCount);
     inc(FCount);
 end;
 
@@ -133,9 +132,9 @@ begin
     FData := TWriteable<TMycDataArray<T>>.CreateWriteable(FCurrData).Protect;
 end;
 
-function TChartSeriesReceiver<T>.ProcessData(const Value: T): Boolean;
+function TChartSeriesReceiver<T>.ProcessData(const Value: T): TState;
 begin
-    Result := true;
+    Result := TState.Null;
     FCurrData := FCurrData.Add(Value, FLookback.Value);
     FData.Value := FCurrData;
 end;

Src/Myc.Fmx.Chart.pas

@@ -210,7 +210,7 @@ begin
     end;
 
     // finally, check the repaint signal
-    var repaintSignal := FNeedRepaint.Reset;
+    var repaintSignal := false; // FNeedRepaint.Reset;
 
     if repaintSignal or (seriesRepaint and seriesChanged) then
         Repaint;

Src/Myc.Signals.pas

@@ -79,7 +79,7 @@ type
         class operator Implicit(const A: TState): IState; overload;
 
         class function All(const States: TArray<TState>): TState; static;
-        class function Any(const States: TArray<TState>): TState; static;
+        class function Any(const States: TArray<TState>; Count: Integer = 1): TState; static;
 
         class property Null: IState read FNull;
 
@@ -138,6 +138,7 @@ type
     strict private
         class var
             FNull: ILatch;
+            [volatile]
             FQueueLock: Integer;
         class constructor ClassCreate;
 
@@ -243,32 +244,23 @@ begin
 end;
 
 class function TState.All(const States: TArray<TState>): TState;
+begin
+    Result := Any(States, Length(States));
+end;
+
+class function TState.Any(const States: TArray<TState>; Count: Integer = 1): TState;
 var
     Latch: TLatch.ILatch;
 begin
-    Latch := TLatch.CreateLatch(Length(States));
+    if (States = nil) or (Count <= 0) then
+        exit(TState.Null);
+
+    Latch := TLatch.CreateLatch(Count);
     for var state in States do
         state.Signal.Subscribe(Latch);
     Result := Latch.State;
 end;
 
-class function TState.Any(const States: TArray<TState>): TState;
-var
-    Latch: TLatch.ILatch;
-begin
-    if Length(States) = 0 then
-    begin
-        Result := TState.Null;
-    end
-    else
-    begin
-        Latch := TLatch.CreateLatch(1);
-        for var state in States do
-            state.Signal.Subscribe(Latch);
-        Result := Latch.State;
-    end;
-end;
-
 function TState.GetSignal: TSignal;
 begin
     Result := FState.Signal;

Src/Myc.TaskManager.pas

@@ -4,7 +4,8 @@ interface
 
 uses
     System.SysUtils,
-    Myc.Signals;
+    Myc.Signals,
+    System.Contnrs;
 
 type
     TTaskManager = record
@@ -38,6 +39,16 @@ type
         procedure WaitFor(const State: TState); inline;
     end;
 
+    TQueue = record
+    strict private
+        [volatile]
+        FQueueLock: Integer;
+        FState: TState;
+    public
+        function Enqueue(const Func: TFunc<TState>): TState;
+        class operator Initialize(out Dest: TQueue);
+    end;
+
 var
     TaskManager: TTaskManager;
 
@@ -174,4 +185,29 @@ begin
     Result := A.FTaskManager;
 end;
 
+function TQueue.Enqueue(const Func: TFunc<TState>): TState;
+begin
+    var state := TaskManager.RunTask(FState, Func);
+
+    repeat
+        if FQueueLock > 0 then
+        begin
+            YieldProcessor;
+            continue;
+        end;
+    until AtomicCmpExchange(FQueueLock, 1, 0) = 0;
+
+    try
+        Result := FState;
+        FState := state;
+    finally
+        AtomicDecrement(FQueueLock);
+    end;
+end;
+
+class operator TQueue.Initialize(out Dest: TQueue);
+begin
+    Dest.FQueueLock := 0;
+end;
+
 end.

Src/Myc.Trade.DataPoint.pas

@@ -4,6 +4,7 @@ interface
 
 uses
     System.TimeSpan,
+    Myc.Signals,
     Myc.Core.Notifier;
 
 type
@@ -31,12 +32,12 @@ type
     end;
 
     IMycProcessor<T> = interface
-        function ProcessData(const Value: T): Boolean;
+        function ProcessData(const Value: T): TState;
     end;
 
     TMycProcessor<T> = class abstract(TInterfacedObject, IMycProcessor<T>)
     protected
-        function ProcessData(const Value: T): Boolean; virtual; abstract;
+        function ProcessData(const Value: T): TState; virtual; abstract;
     end;
 
     TTag = Pointer;
@@ -53,6 +54,7 @@ type
         constructor Create(const Controller: IInterface);
         destructor Destroy; override;
         // Notifies all linked processors.
+        function Broadcast(const Value: T): TState;
         procedure Notify(const Func: TMycNotifyList<IMycProcessor<T>>.TNotifyProc);
         // Link a Processor
         function Link(const Processor: IMycProcessor<T>): TTag;
@@ -62,21 +64,21 @@ type
 
     TMycGenericProcessor<T> = class(TMycProcessor<T>)
         type
-            TProc = reference to function(const Value: T): Boolean;
+            TProc = reference to function(const Value: T): TState;
     private
         FProc: TProc;
     protected
-        function ProcessData(const Value: T): Boolean; override; final;
+        function ProcessData(const Value: T): TState; override; final;
     public
         constructor Create(const AProc: TProc);
     end;
 
     TMycContainedProcessor<T> = class(TContainedObject, IMycProcessor<T>)
         type
-            TProc = function(const Value: T): Boolean of object;
+            TProc = function(const Value: T): TState of object;
     private
         FProc: TProc;
-        function ProcessData(const Value: T): Boolean;
+        function ProcessData(const Value: T): TState;
     public
         constructor Create(const Controller: IInterface; const AProc: TProc);
     end;
@@ -91,10 +93,10 @@ type
         FSender: TMycDataProvider<T>;
         function GetSender: IMycDataProvider<T>;
     protected
-        function ProcessData(const Value: S): Boolean; override; abstract;
+        function ProcessData(const Value: S): TState; override; abstract;
 
         // Broadcasts the given data to all linked processors.
-        procedure Broadcast(const Value: T);
+        function Broadcast(const Value: T): TState;
 
     public
         constructor Create;
@@ -469,7 +471,7 @@ begin
     FProc := AProc;
 end;
 
-function TMycGenericProcessor<T>.ProcessData(const Value: T): Boolean;
+function TMycGenericProcessor<T>.ProcessData(const Value: T): TState;
 begin
     Result := FProc(Value);
 end;
@@ -480,7 +482,7 @@ begin
     FProc := AProc;
 end;
 
-function TMycContainedProcessor<T>.ProcessData(const Value: T): Boolean;
+function TMycContainedProcessor<T>.ProcessData(const Value: T): TState;
 begin
     Result := FProc(Value);
 end;
@@ -498,6 +500,34 @@ begin
     inherited Destroy;
 end;
 
+function TMycDataProvider<T>.Broadcast(const Value: T): TState;
+begin
+    FListeners.Lock;
+    try
+        var item := FListeners.First;
+        if item = nil then
+            exit(TState.Null);
+
+        var i := 1;
+        while item.Next <> nil do
+        begin
+            inc(i);
+            item := item.Next;
+        end;
+
+        var done := TLatch.CreateLatch(i);
+        while item <> nil do
+        begin
+            item.Receiver.ProcessData(Value).Signal.Subscribe(done);
+            item := item.Prev;
+        end;
+
+        Result := done.State;
+    finally
+        FListeners.Release;
+    end;
+end;
+
 procedure TMycDataProvider<T>.Notify(const Func: TMycNotifyList<IMycProcessor<T>>.TNotifyProc);
 begin
     FListeners.Lock;
@@ -543,10 +573,9 @@ begin
     inherited Destroy;
 end;
 
-procedure TMycConverter<S, T>.Broadcast(const Value: T);
+function TMycConverter<S, T>.Broadcast(const Value: T): TState;
 begin
-    var cValue := Value;
-    FSender.Notify(function(const Processor: IMycProcessor<T>): Boolean begin Result := Processor.ProcessData(cValue) end);
+    Result := FSender.Broadcast(Value);
 end;
 
 function TMycConverter<S, T>.GetSender: IMycDataProvider<T>;

Src/Myc.Trade.DataStream.pas

@@ -445,20 +445,20 @@ begin
         DataFile.Chain(
             function(const DataChunks: TArray<TArray<TDataPoint<T>>>): TState
             begin
-                if Assigned(DataChunks) then
-                begin
-                    // Process each chunk, checking for termination between chunks.
-                    for var chunk in DataChunks do
-                    begin
-                        if cTerminated.IsSet then
-                            exit(TState.Null);
+                // 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;
 
-                        if not Processor.ProcessData(chunk) then
-                            exit(TState.Null);
-                    end;
-                end;
-
-                Result := ProcessFile(nextFileInfo, nextFile, cTerminated, 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
+                    )
             end
         );
 end;