diff --git a/AuraTrader/FirstStrategy.pas b/AuraTrader/FirstStrategy.pas index 405cc0d..3ce4162 100644 --- a/AuraTrader/FirstStrategy.pas +++ b/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>): Boolean; override; + function ProcessData(const Values: TArray>): TState; override; property CurrentBar: TDataPoint read GetCurrentBar; property Timeframe: TTimeframe read GetTimeframe; @@ -51,10 +52,11 @@ type FSourceData: TMycDataArray; // Intermediate data series for HMA calculation (2*WMA(n/2) - WMA(n)) FDiffSeries: TMycDataArray; + FQueue: TQueue; // Calculates the Weighted Moving Average for the most recent data. function CalculateWMA(const Series: TMycDataArray; 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>): Boolean; +function TTicksToTimeframe.ProcessData(const Values: TArray>): TState; var point: TDataPoint; midPrice: Single; barStartTime: TDateTime; lastBarTime: TDateTime; currentBar: TOhlcItem; + states: TList; 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.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 } @@ -230,10 +240,9 @@ begin FFunc := AFunc; end; -function TMycGenericConverter.ProcessData(const Value: S): Boolean; +function TMycGenericConverter.ProcessData(const Value: S): TState; begin - Result := true; - Broadcast(FFunc(Value)); + Result := Broadcast(FFunc(Value)); end; end. diff --git a/AuraTrader/MainForm.pas b/AuraTrader/MainForm.pas index 210c2d5..a1b4059 100644 --- a/AuraTrader/MainForm.pas +++ b/AuraTrader/MainForm.pas @@ -182,9 +182,14 @@ begin for var i := 0 to 3 do begin - var Hull: IMycConverter := THullMovingAverage.Create(50 + (50 * i)); + var Hull: IMycConverter := 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 := THullMovingAverage.Create(250); @@ -193,7 +198,9 @@ begin // // chart.AddDoubleSeries(Hull.Sender); - chart.SetXAxisSeries(OHLC.Sender); + OhlcPoint.Sender.Link(TimeStamps); + + chart.SetXAxisSeries(Timestamps.Sender); chart.AddOhlcSeries(Ohlc.Sender); var done := ExecuteStrategy(Symbol, OhlcPoint); @@ -444,9 +451,10 @@ begin Symbol, terminated, TMycGenericProcessor>>.Create( - function(const Values: TArray>): Boolean + function(const Values: TArray>): TState begin - Result := true; + Result := TState.Null; + Prices := Prices.Add(Values); currLog.Value := Prices.TotalCount.ToString; diff --git a/Src/Myc.Core.Notifier.pas b/Src/Myc.Core.Notifier.pas index 2774de6..4401406 100644 --- a/Src/Myc.Core.Notifier.pas +++ b/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); @@ -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.GetFirst: PItem; +begin + Assert(IsLocked); + Result := FList; +end; + function TMycNotifyList.IsLocked: Boolean; begin Result := NativeUInt(FList) and 1 = 0; diff --git a/Src/Myc.Fmx.Chart.Series.pas b/Src/Myc.Fmx.Chart.Series.pas index 88fb61e..214f982 100644 --- a/Src/Myc.Fmx.Chart.Series.pas +++ b/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>; protected - function ProcessData(const Value: T): Boolean; override; + function ProcessData(const Value: T): TState; override; public constructor Create(const ALookback: TMutable); property Data: TWriteable> read FData; @@ -116,10 +116,9 @@ begin FCount := 0; end; -function TChartSeriesCounter.ProcessData(const Value: T): Boolean; +function TChartSeriesCounter.ProcessData(const Value: T): TState; begin - Result := true; - Broadcast(FCount); + Result := Broadcast(FCount); inc(FCount); end; @@ -133,9 +132,9 @@ begin FData := TWriteable>.CreateWriteable(FCurrData).Protect; end; -function TChartSeriesReceiver.ProcessData(const Value: T): Boolean; +function TChartSeriesReceiver.ProcessData(const Value: T): TState; begin - Result := true; + Result := TState.Null; FCurrData := FCurrData.Add(Value, FLookback.Value); FData.Value := FCurrData; end; diff --git a/Src/Myc.Fmx.Chart.pas b/Src/Myc.Fmx.Chart.pas index f286969..b263d14 100644 --- a/Src/Myc.Fmx.Chart.pas +++ b/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; diff --git a/Src/Myc.Signals.pas b/Src/Myc.Signals.pas index 0cbd751..d2b477b 100644 --- a/Src/Myc.Signals.pas +++ b/Src/Myc.Signals.pas @@ -79,7 +79,7 @@ type class operator Implicit(const A: TState): IState; overload; class function All(const States: TArray): TState; static; - class function Any(const States: TArray): TState; static; + class function Any(const States: TArray; 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; +begin + Result := Any(States, Length(States)); +end; + +class function TState.Any(const States: TArray; 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; -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; diff --git a/Src/Myc.TaskManager.pas b/Src/Myc.TaskManager.pas index 5bef920..c3e8ede 100644 --- a/Src/Myc.TaskManager.pas +++ b/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; + 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; +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. diff --git a/Src/Myc.Trade.DataPoint.pas b/Src/Myc.Trade.DataPoint.pas index 40b1bf4..52702a5 100644 --- a/Src/Myc.Trade.DataPoint.pas +++ b/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 = interface - function ProcessData(const Value: T): Boolean; + function ProcessData(const Value: T): TState; end; TMycProcessor = class abstract(TInterfacedObject, IMycProcessor) 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>.TNotifyProc); // Link a Processor function Link(const Processor: IMycProcessor): TTag; @@ -62,21 +64,21 @@ type TMycGenericProcessor = class(TMycProcessor) 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 = class(TContainedObject, IMycProcessor) 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; function GetSender: IMycDataProvider; 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.ProcessData(const Value: T): Boolean; +function TMycGenericProcessor.ProcessData(const Value: T): TState; begin Result := FProc(Value); end; @@ -480,7 +482,7 @@ begin FProc := AProc; end; -function TMycContainedProcessor.ProcessData(const Value: T): Boolean; +function TMycContainedProcessor.ProcessData(const Value: T): TState; begin Result := FProc(Value); end; @@ -498,6 +500,34 @@ begin inherited Destroy; end; +function TMycDataProvider.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.Notify(const Func: TMycNotifyList>.TNotifyProc); begin FListeners.Lock; @@ -543,10 +573,9 @@ begin inherited Destroy; end; -procedure TMycConverter.Broadcast(const Value: T); +function TMycConverter.Broadcast(const Value: T): TState; begin - var cValue := Value; - FSender.Notify(function(const Processor: IMycProcessor): Boolean begin Result := Processor.ProcessData(cValue) end); + Result := FSender.Broadcast(Value); end; function TMycConverter.GetSender: IMycDataProvider; diff --git a/Src/Myc.Trade.DataStream.pas b/Src/Myc.Trade.DataStream.pas index 8d53ee8..5cacabe 100644 --- a/Src/Myc.Trade.DataStream.pas +++ b/Src/Myc.Trade.DataStream.pas @@ -445,20 +445,20 @@ begin DataFile.Chain( function(const DataChunks: TArray>>): 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;