TTicker

AuraTrader/AuraTrader.dproj

@@ -4,7 +4,7 @@
         <ProjectVersion>20.3</ProjectVersion>
         <FrameworkType>FMX</FrameworkType>
         <Base>True</Base>
-        <Config Condition="'$(Config)'==''">Debug</Config>
+        <Config Condition="'$(Config)'==''">Release</Config>
         <Platform Condition="'$(Platform)'==''">Win64</Platform>
         <ProjectName Condition="'$(ProjectName)'==''">AuraTrader</ProjectName>
         <TargetedPlatforms>3</TargetedPlatforms>

AuraTrader/FirstStrategy.pas

@@ -24,24 +24,6 @@ type
         constructor Create(const AFunc: TConvertFunc);
     end;
 
-    TTicksToTimeframe = class(TMycConverter<TArray<TDataPoint<TAskBidItem>>, TDataPoint<TOhlcItem>>)
-    private
-        FTimeframe: TTimeframe;
-        // Stores the currently aggregating OHLC data.
-        FCurrentBar: TDataPoint<TOhlcItem>;
-        function GetBarStartTime(const TimeStamp: TDateTime; const Timeframe: TTimeframe): TDateTime;
-        function GetCurrentBar: TDataPoint<TOhlcItem>;
-        function  GetTimeframe: TTimeframe;
-    public
-        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>>): TState; override;
-
-        property CurrentBar: TDataPoint<TOhlcItem> read GetCurrentBar;
-        property Timeframe: TTimeframe read GetTimeframe;
-    end;
-
     TIndicator<S, T> = class(TMycConverter<S, T>)
     protected
         function ProcessData(const Value: S): TState; override; final;
@@ -57,21 +39,77 @@ type
         constructor Create(const AFunc: TIndicatorFunc<S, T>);
     end;
 
+    TTicksToBars = class(TMycConverter<TDataPoint<TAskBidItem>, TDataPoint<TOhlcItem>>)
+    private
+        FTimeframe: TTimeframe;
+        // Stores the currently aggregating OHLC data.
+        FCurrentBar: TDataPoint<TOhlcItem>;
+        function GetBarStartTime(const TimeStamp: TDateTime; const Timeframe: TTimeframe): TDateTime;
+        function GetCurrentBar: TDataPoint<TOhlcItem>;
+        function GetTimeframe: TTimeframe;
+    public
+        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 Value: TDataPoint<TAskBidItem>): TState; override;
+
+        property CurrentBar: TDataPoint<TOhlcItem> read GetCurrentBar;
+        property Timeframe: TTimeframe read GetTimeframe;
+    end;
+
+    TTicker<T> = class(TMycConverter<TArray<T>, T>)
+    public
+        function ProcessData(const Values: TArray<T>): TState; override;
+    end;
+
 implementation
 
 uses
     System.DateUtils,
     System.Math;
 
-{ TTicksToTimeframe }
+{ TMycGenericConverter<S, T> }
 
-constructor TTicksToTimeframe.Create(const ATimeframe: TTimeframe);
+constructor TMycGenericConverter<S, T>.Create(const AFunc: TConvertFunc);
+begin
+    inherited Create;
+    FFunc := AFunc;
+end;
+
+function TMycGenericConverter<S, T>.ProcessData(const Value: S): TState;
+begin
+    Result := Broadcast(FFunc(Value));
+end;
+
+{ TIndicator<S,T> }
+
+function TIndicator<S, T>.ProcessData(const Value: S): TState;
+begin
+    Result := Broadcast(Calculate(Value));
+end;
+
+{ TGenericIndicator<S,T> }
+
+constructor TGenericIndicator<S, T>.Create(const AFunc: TIndicatorFunc<S, T>);
+begin
+    inherited Create;
+    FFunc := AFunc;
+end;
+
+function TGenericIndicator<S, T>.Calculate(const Value: S): T;
+begin
+    Result := FFunc(Value);
+end;
+
+{ TTicksToBars }
+
+constructor TTicksToBars.Create(const ATimeframe: TTimeframe);
 begin
     inherited Create;
     FTimeframe := ATimeframe;
 end;
 
-function TTicksToTimeframe.GetBarStartTime(const TimeStamp: TDateTime; const Timeframe: TTimeframe): TDateTime;
+function TTicksToBars.GetBarStartTime(const TimeStamp: TDateTime; const Timeframe: TTimeframe): TDateTime;
 var
     baseTime: TDateTime;
 begin
@@ -119,99 +157,64 @@ begin
     end;
 end;
 
-function TTicksToTimeframe.GetCurrentBar: TDataPoint<TOhlcItem>;
+function TTicksToBars.GetCurrentBar: TDataPoint<TOhlcItem>;
 begin
     Result := FCurrentBar;
 end;
 
-function TTicksToTimeframe.GetTimeframe: TTimeframe;
+function TTicksToBars.GetTimeframe: TTimeframe;
 begin
     Result := FTimeframe;
 end;
 
-function TTicksToTimeframe.ProcessData(const Values: TArray<TDataPoint<TAskBidItem>>): TState;
+function TTicksToBars.ProcessData(const Value: TDataPoint<TAskBidItem>): TState;
 var
-    point: TDataPoint<TAskBidItem>;
     midPrice: Single;
     barStartTime: TDateTime;
     lastBarTime: TDateTime;
     currentBar: TOhlcItem;
-    states: TList<TState>;
 begin
-    states := TList<TState>.Create;
-    try
-        // Process each incoming data point
-        for point in Values do
+    midPrice := (Value.Data.Ask + Value.Data.Bid) / 2;
+
+    // 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
-            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
-                // A new bar starts, so the previous one is now complete.
-                if (lastBarTime > 0) then
-                begin
-                    states.Add(Broadcast(FCurrentBar));
-                end;
-
-                // 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;
+            Result := Broadcast(FCurrentBar);
         end;
 
-        Result := TState.All(states.ToArray);
-    finally
-        states.Free;
+        // 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;
 
-{ TMycGenericConverter<S, T> }
-
-constructor TMycGenericConverter<S, T>.Create(const AFunc: TConvertFunc);
+function TTicker<T>.ProcessData(const Values: TArray<T>): TState;
 begin
-    inherited Create;
-    FFunc := AFunc;
-end;
+    var done := TLatch.CreateLatch( Length(Values) );
 
-function TMycGenericConverter<S, T>.ProcessData(const Value: S): TState;
-begin
-    Result := Broadcast(FFunc(Value));
-end;
+    // Process each incoming data point
+    for var i:=0 to High(Values) do
+        Broadcast(Values[i]).Signal.Subscribe(done);
 
-{ TIndicator<S,T> }
-
-function TIndicator<S, T>.ProcessData(const Value: S): TState;
-begin
-    Result := Broadcast(Calculate(Value));
-end;
-
-{ TGenericIndicator<S,T> }
-
-constructor TGenericIndicator<S, T>.Create(const AFunc: TIndicatorFunc<S, T>);
-begin
-    inherited Create;
-    FFunc := AFunc;
-end;
-
-function TGenericIndicator<S, T>.Calculate(const Value: S): T;
-begin
-    Result := FFunc(Value);
+    Result := done.State;
 end;
 
 end.

AuraTrader/MainForm.pas

@@ -167,7 +167,10 @@ begin
 
     var timeframe := TTimeframe.S15;
 
-    var OhlcPoint: IMycConverter<TArray<TDataPoint<TAskBidItem>>, TDataPoint<TOhlcItem>> := TTicksToTimeframe.Create(timeframe);
+    var ticker := TTicker<TDataPoint<TAskBidItem>>.Create;
+
+    var OhlcPoint := TTicksToBars.Create(timeframe);
+    ticker.Sender.Link(OhlcPoint);
 
     var Timestamps: IMycConverter<TDataPoint<TOhlcItem>, TDateTime> :=
         TMycGenericConverter<TDataPoint<TOhlcItem>, TDateTime>
@@ -272,7 +275,7 @@ begin
     Stoch.Sender.Link(StochD);
     Panel.AddDoubleSeries(StochD.Sender, TAlphaColors.Red);
 
-    var done := ExecuteStrategy(Symbol, OhlcPoint);
+    var done := ExecuteStrategy(Symbol, ticker);
 
     /////
 

Src/Myc.Core.Tasks.pas

@@ -112,7 +112,11 @@ begin
 
     FWorkGate := TSemaphore.Create(nil, 0, MaxInt, ''); // Initial count is 0, workers will wait
 
-    SetLength(FWorkThreads, TThread.ProcessorCount); // Create one thread per processor core
+    var threadCnt := TThread.ProcessorCount - 1;
+    if threadCnt < 2 then
+        threadCnt := 2;
+
+    SetLength(FWorkThreads, threadCnt);
     for var i := 0 to High(FWorkThreads) do
     begin
         FWorkThreads[i] := CreateAnonymousThread('Thread pool [' + IntToStr(i) + ']', WorkerThread);
@@ -158,7 +162,7 @@ begin
     if TThread.CurrentThread.ThreadID = MainThreadID then
     begin
         prio := TThread.CurrentThread.Priority;
-        if prio > Low(TThreadPriority) then
+        if prio > tpLowest then
             prio := TThreadPriority(Ord(prio) - 1); // Decrease priority by one step
 
         Result.Priority := prio;

Src/Myc.Fmx.Chart.pas

@@ -65,8 +65,12 @@ type
             protected
                 function GetOwner: TMycChart; override; final;
                 // Paint crosshair and other axis related overlays
-                procedure Paint(const Canvas: TCanvas; const Viewport: TRectF; ViewStartIndex, ViewCount: Int64; IdxAtMousePos: Integer); virtual;
-                    abstract;
+                procedure Paint(
+                    const Canvas: TCanvas;
+                    const Viewport: TRectF;
+                    ViewStartIndex, ViewCount: Int64;
+                    IdxAtMousePos: Integer
+                ); virtual; abstract;
             public
                 constructor Create(AOwner: TMycChart);
             end;
@@ -74,7 +78,12 @@ type
             TXAxisLayer = class(TAxisLayer)
             protected
                 // Paint crosshair and time caption
-                procedure Paint(const Canvas: TCanvas; const Viewport: TRectF; ViewStartIndex, ViewCount: Int64; IdxAtMousePos: Integer); override;
+                procedure Paint(
+                    const Canvas: TCanvas;
+                    const Viewport: TRectF;
+                    ViewStartIndex, ViewCount: Int64;
+                    IdxAtMousePos: Integer
+                ); override;
                 // This function delivers the text for the caption.
                 function GetCaption(Idx: Int64): String; virtual; abstract;
             end;
@@ -339,11 +348,11 @@ begin
     end;
 
     // Draw crosshair if mouse is in control
-    if FIsMouseInControl and Assigned(FXAxisSeries) and (FMousePos.Idx >= 0 ) then
+    if FIsMouseInControl and Assigned(FXAxisSeries) and (FMousePos.Idx >= 0) then
     begin
         // Do not draw crosshair if mouse is over the jump button
         if not FJumpButtonRect.Contains(FMousePos.Point) then
-            FXAxisSeries.Paint(Self.Canvas, rect, FViewStartIndex, FViewCount, FViewStartIndex + FMousePos.Idx );
+            FXAxisSeries.Paint(Self.Canvas, rect, FViewStartIndex, FViewCount, FViewStartIndex + FMousePos.Idx);
     end;
 
     // --- Draw the "jump to latest" button (code unchanged) ---
@@ -512,7 +521,7 @@ begin
             panel2.Weight := (newH2 / (newH1 + newH2)) * twoPanelWeight;
         end;
 
-        FDragStartPoint := CreateDragPoint( FDragStartPoint.Point.X, Y ); // Update start point for next move
+        FDragStartPoint := CreateDragPoint(FDragStartPoint.Point.X, Y); // Update start point for next move
         Repaint;
         exit; // Do not continue with other mouse move logic
     end;
@@ -615,7 +624,7 @@ begin
     FViewCount := newViewCount;
 
     var oldIdx := FViewStartIndex + FMousePos.Idx;
-    FMousePos := CreateDragPoint( FMousePos.Point.X, FMousePos.Point.Y );
+    FMousePos := CreateDragPoint(FMousePos.Point.X, FMousePos.Point.Y);
     FViewStartIndex := oldIdx - FMousePos.Idx;
 
     // Clamp start index
@@ -819,8 +828,12 @@ end;
 
 { TMycChart.TXAxisLayer }
 
-procedure TMycChart.TXAxisLayer.Paint(const Canvas: TCanvas; const Viewport: TRectF; ViewStartIndex, ViewCount: Int64; IdxAtMousePos:
-    Integer);
+procedure TMycChart.TXAxisLayer.Paint(
+    const Canvas: TCanvas;
+    const Viewport: TRectF;
+    ViewStartIndex, ViewCount: Int64;
+    IdxAtMousePos: Integer
+);
 var
     caption: string;
     captionRect: TRectF;

Test/MycTests.dpr

@@ -5,34 +5,34 @@ program MycTests;
 {$ENDIF}
 {$STRONGLINKTYPES ON}
 uses
-  FastMM5,
-  DUnitX.MemoryLeakMonitor.FastMM5,
-  System.SysUtils,
-  {$IFDEF TESTINSIGHT}
-  TestInsight.DUnitX,
-  {$ELSE}
-  DUnitX.Loggers.Console,
-  {$ENDIF }
-  DUnitX.TestFramework,
-  TestNotifier in 'TestNotifier.pas',
-  TestNotifier_Threading in 'TestNotifier_Threading.pas' {/TestNotifier_ChaosStress in 'TestNotifier_ChaosStress.pas',},
-  TestNotifier_ChaosStress in 'TestNotifier_ChaosStress.pas',
-  TestTasks in 'TestTasks.pas',
-  Myc.Futures in '..\Src\Myc.Futures.pas',
-  TestCoreFutures in 'TestCoreFutures.pas',
-  Myc.TaskManager in '..\Src\Myc.TaskManager.pas',
-  TestFutures in 'TestFutures.pas',
-  Myc.Test.Core.Atomic in '..\Src\Myc.Test.Core.Atomic.pas',
-  Myc.Test.Signals.Latch in '..\Src\Myc.Test.Signals.Latch.pas',
-  Myc.Test.Signals.Dirty in '..\Src\Myc.Test.Signals.Dirty.pas',
-  Myc.Trade.DataPoint in '..\Src\Myc.Trade.DataPoint.pas',
-  Myc.Trade.Node in '..\Src\Myc.Trade.Node.pas',
-  Myc.Trade.DataStream in '..\Src\Myc.Trade.DataStream.pas',
-  Myc.Test.Trade.DataStream in '..\Src\Myc.Test.Trade.DataStream.pas',
-  Myc.Test.Trade.DataPoint in '..\Src\Myc.Test.Trade.DataPoint.pas',
-  Myc.Trade.DataProvider in '..\Src\Myc.Trade.DataProvider.pas',
-  Myc.Mutable in '..\Src\Myc.Mutable.pas',
-  Test.Core.Mutable in 'Test.Core.Mutable.pas';
+    FastMM5,
+    DUnitX.MemoryLeakMonitor.FastMM5,
+    System.SysUtils,
+{$IFDEF TESTINSIGHT}
+    TestInsight.DUnitX,
+{$ELSE}
+    DUnitX.Loggers.Console,
+{$ENDIF }
+    DUnitX.TestFramework,
+    TestNotifier in 'TestNotifier.pas',
+    TestNotifier_Threading in 'TestNotifier_Threading.pas' {/TestNotifier_ChaosStress in 'TestNotifier_ChaosStress.pas',},
+    TestNotifier_ChaosStress in 'TestNotifier_ChaosStress.pas',
+    TestTasks in 'TestTasks.pas',
+    Myc.Futures in '..\Src\Myc.Futures.pas',
+    TestCoreFutures in 'TestCoreFutures.pas',
+    Myc.TaskManager in '..\Src\Myc.TaskManager.pas',
+    TestFutures in 'TestFutures.pas',
+    Myc.Test.Core.Atomic in '..\Src\Myc.Test.Core.Atomic.pas',
+    Myc.Test.Signals.Latch in '..\Src\Myc.Test.Signals.Latch.pas',
+    Myc.Test.Signals.Dirty in '..\Src\Myc.Test.Signals.Dirty.pas',
+    Myc.Trade.DataPoint in '..\Src\Myc.Trade.DataPoint.pas',
+    Myc.Trade.Node in '..\Src\Myc.Trade.Node.pas',
+    Myc.Trade.DataStream in '..\Src\Myc.Trade.DataStream.pas',
+    Myc.Test.Trade.DataStream in '..\Src\Myc.Test.Trade.DataStream.pas',
+    Myc.Test.Trade.DataPoint in '..\Src\Myc.Test.Trade.DataPoint.pas',
+    Myc.Trade.DataProvider in '..\Src\Myc.Trade.DataProvider.pas',
+    Myc.Mutable in '..\Src\Myc.Mutable.pas',
+    Test.Core.Mutable in 'Test.Core.Mutable.pas';
 
 { keep comment here to protect the following conditional from being removed by the IDE when adding a unit }
 {$IFNDEF TESTINSIGHT}

Test/Test.Core.Mutable.pas

@@ -34,9 +34,6 @@ type
         [TestCase('Writeable', '10,20')]
         procedure TestWriteableMutable(InitialValue, NewValue: Integer);
 
-        [Test]
-        procedure TestWriteableMutable_SignalOnSameValue;
-
         [Test]
         [TestCase('Protected', '100,200')]
         procedure TestProtectedMutable_Delegation(InitialValue, NewValue: Integer);
@@ -106,7 +103,8 @@ var
     func: TFunc<Integer>;
 begin
     callCount := 0;
-    func := function: Integer
+    func :=
+        function: Integer
         begin
             Inc(callCount);
             Result := 42;
@@ -125,10 +123,7 @@ var
     func: TFunc<Integer>;
 begin
     source := TEvent.CreateEvent;
-    func := function: Integer
-        begin
-            Result := 123;
-        end;
+    func := function: Integer begin Result := 123; end;
     mutable := TMutable<Integer>.Construct(source.Signal, func);
 
     lazy := TLazy<Integer>.Create(mutable);
@@ -156,19 +151,6 @@ begin
     subscription.Unsubscribe;
 end;
 
-procedure TTestCoreMutable.TestWriteableMutable_SignalOnSameValue;
-var
-    writeable: TWriteable<Integer>;
-    subscriber: TSignal.ISubscriber;
-    subscription: TSignal.TSubscription;
-begin
-    writeable := TWriteable<Integer>.CreateWriteable(100);
-    subscriber := TTestSubscriber.Create(SubscriberCallback);
-    subscription := writeable.Changed.Subscribe(subscriber);
-
-    subscription.Unsubscribe;
-end;
-
 procedure TTestCoreMutable.TestProtectedMutable_Delegation(InitialValue, NewValue: Integer);
 var
     writeable: TWriteable<Integer>;
@@ -203,21 +185,24 @@ begin
 
     for i := 0 to ThreadCount - 1 do
     begin
-        threads := threads + [TThread.CreateAnonymousThread(
-            procedure
-            var
-                j: Integer;
-            begin
-                for j := 1 to Iterations do
-                begin
-                    // This is an intentional race condition to test lock stability.
-                    // The lock protects the GetValue calls individually,
-                    // preventing memory corruption during concurrent access.
-                    var curr := writeable.Value;
-                    writeable.Value := curr + 1;
-                end;
-            end
-        )];
+        threads :=
+            threads
+                + [
+                    TThread.CreateAnonymousThread(
+                        procedure
+                        var
+                            j: Integer;
+                        begin
+                            for j := 1 to Iterations do
+                            begin
+                                // This is an intentional race condition to test lock stability.
+                                // The lock protects the GetValue calls individually,
+                                // preventing memory corruption during concurrent access.
+                                var curr := writeable.Value;
+                                writeable.Value := curr + 1;
+                            end;
+                        end
+                    )];
     end;
 
     for var thread in threads do