From ff5b379fde1e6a4aaaf857002c397648d2dcb089 Mon Sep 17 00:00:00 2001 From: Michael Schimmel Date: Thu, 24 Jul 2025 13:07:29 +0200 Subject: [PATCH] Memory hole in Data-Join fixed --- AuraTrader/MainForm.pas | 19 ++-- AuraTrader/StrategyTest.pas | 2 +- Src/Myc.Fmx.Chart.Series.pas | 27 ++++-- Src/Myc.Trade.DataPoint.Impl.pas | 87 ++++++++----------- Src/Myc.Trade.DataPoint.pas | 143 +++++++++++++++++-------------- 5 files changed, 141 insertions(+), 137 deletions(-) diff --git a/AuraTrader/MainForm.pas b/AuraTrader/MainForm.pas index 303241f..37c3241 100644 --- a/AuraTrader/MainForm.pas +++ b/AuraTrader/MainForm.pas @@ -315,15 +315,6 @@ begin // next stage - var ATREndPoint := ATR.CreateEndpoint(5); - var ATRSeries: TSeries; - - var HullEndPoint := Hull.CreateEndpoint(5); - var HullSeries: TSeries; - - var SmaEndPoint := Sma.CreateEndpoint(5); - var SmaSeries: TSeries; - var curr: TSignal; curr.SL := Double.NaN; curr.Entry := Double.NaN; @@ -333,7 +324,7 @@ begin var conv := TConverter.Join([Ohlc.Field('Low'), Ohlc.Field('High'), Closes, ATR, Hull, Sma]); var Signal := - TConverter, TSignal>.CreateGeneric( + TConverter, TSignal>.CreateConverter( function(const Values: TArray): TSignal begin var low := Values[0]; @@ -473,7 +464,7 @@ begin panel.AddDoubleSeries(Signal.Producer.Field('Entry'), TAlphaColors.Green, 1); panel.AddDoubleSeries(Signal.Producer.Field('SL'), TAlphaColors.Red, 2); - var mean := TConverter, Double>.CreateGeneric(TIndicators.CreateMean()); + var mean := TConverter, Double>.CreateConverter(TIndicators.CreateMean()); TConverter.Join([Hull, Sma]).Chain(mean); @@ -494,14 +485,14 @@ begin var indi := EMAFActory.CreateIndicator(Params); - var EMAConv := TConverter.CreateGeneric(indi); + var EMAConv := TConverter.CreateConverter(indi); var equityEMA := equity .Producer - .Chain(TConverter.FieldToRecord(EMAFactory.Input, 'Price')) + .Chain(EMAFactory.Input.FieldAsRecord('Price')) .Chain(EMAConv) - .Chain(TConverter.FieldOfRecord(EMAFactory.Output, 'MA')); + .Chain(EMAFactory.Output.FieldOfRecord('MA')); ////////////// diff --git a/AuraTrader/StrategyTest.pas b/AuraTrader/StrategyTest.pas index cff102b..6d08fd5 100644 --- a/AuraTrader/StrategyTest.pas +++ b/AuraTrader/StrategyTest.pas @@ -50,7 +50,7 @@ begin var signalGenerator := conv.Chain( - TConverter, TSignalEvent>.CreateGeneric( + TConverter, TSignalEvent>.CreateConverter( function(const Values: TArray): TSignalEvent begin Result.Low := Values[0]; diff --git a/Src/Myc.Fmx.Chart.Series.pas b/Src/Myc.Fmx.Chart.Series.pas index cb527ae..4444493 100644 --- a/Src/Myc.Fmx.Chart.Series.pas +++ b/Src/Myc.Fmx.Chart.Series.pas @@ -18,8 +18,8 @@ type { TChartCustomLayer } TChartCustomLayer = class abstract(TMycChart.TDataLayer) private - FProducer: TProducer; - FReceiver: TLazy>; + FProducer: TProducer.TSubscription; + FSeries: TLazy>; FData: TSeries; protected function GetCount: Int64; override; @@ -34,6 +34,7 @@ type ); override; abstract; public constructor Create(AParent: TMycChart.TPanel; const AProducer: TProducer); + destructor Destroy; override; property Data: TSeries read FData; end; @@ -83,7 +84,7 @@ type { TChartXAxisLayer } TChartXAxisLayer = class(TMycChart.TXAxisLayer) private - FProducer: TProducer; + FProducer: TProducer.TSubscription; FReceiver: TLazy>; FData: TSeries; protected @@ -93,6 +94,7 @@ type function GetTotalCount: Int64; override; public constructor Create(AOwner: TMycChart; const AProducer: TProducer); + destructor Destroy; override; property Data: TSeries read FData; end; @@ -270,9 +272,13 @@ end; constructor TChartCustomLayer.Create(AParent: TMycChart.TPanel; const AProducer: TProducer); begin inherited Create(AParent); - FProducer := AProducer; + FProducer := AProducer.CreateEndpoint(Owner.Lookback.Value, FSeries); +end; - FReceiver := FProducer.CreateEndpoint(Owner.Lookback.Value); +destructor TChartCustomLayer.Destroy; +begin + FProducer.Unlink; + inherited; end; function TChartCustomLayer.GetCount: Int64; @@ -287,7 +293,7 @@ end; function TChartCustomLayer.Update: Boolean; begin - Result := FReceiver.Update(FData); + Result := FSeries.Update(FData); end; { TChartXAxisLayer } @@ -295,8 +301,13 @@ end; constructor TChartXAxisLayer.Create(AOwner: TMycChart; const AProducer: TProducer); begin inherited Create(AOwner); - FProducer := AProducer; - FReceiver := FProducer.CreateEndpoint(Owner.Lookback.Value); + FProducer := AProducer.CreateEndpoint(Owner.Lookback.Value, FReceiver); +end; + +destructor TChartXAxisLayer.Destroy; +begin + FProducer.Unlink; + inherited; end; function TChartXAxisLayer.GetCaption(Idx: Int64): String; diff --git a/Src/Myc.Trade.DataPoint.Impl.pas b/Src/Myc.Trade.DataPoint.Impl.pas index 93a1533..26ff5e1 100644 --- a/Src/Myc.Trade.DataPoint.Impl.pas +++ b/Src/Myc.Trade.DataPoint.Impl.pas @@ -31,6 +31,16 @@ type class property Null: IConsumer read FNull; end; + // A consumer implementation that is owned by a controller. + TMycGenericConsumer = class(TMycConsumer) + private + FProc: TConstFunc; + protected + function Consume(const Value: T): TState; override; final; + public + constructor Create(const Controller: IInterface; const AProc: TConstFunc); + end; + TMycProducer = class(TInterfacedObject, IProducer) strict private type @@ -79,18 +89,7 @@ type FNull: IConverter; class constructor CreateClass; private - type - TConsumer = class(TMycConsumer) - private - FOwner: TMycConverter; - protected - function Consume(const Value: S): TState; override; final; - public - constructor Create(AOwner: TMycConverter); - end; - - private - FConsumer: TConsumer; + FConsumer: TMycGenericConsumer; function GetConsumer: IConsumer; protected // To be implemented by descendants to perform the actual conversion. @@ -162,16 +161,6 @@ type constructor Create(const AFieldName: String); end; - // A consumer implementation that is owned by a controller. - TMycGenericConsumer = class(TMycConsumer) - private - FProc: TConstFunc; - protected - function Consume(const Value: T): TState; override; - public - constructor Create(const Controller: IInterface; const AProc: TConstFunc); - end; - // Endpoint that collects data into a series. TMycDataEndpoint = class(TInterfacedObject, TLazy>.ILazy) private @@ -183,9 +172,7 @@ type end; private FConsumer: TMycGenericConsumer; - FTag: TTag; - FProducer: IProducer; - FLookback: Int64; + FLookback: Integer; FChanged: TFlag; FLock: TLightweightMREW; FFirst: PItem; @@ -193,9 +180,11 @@ type function GetChanged: TState; function Consume(const Value: T): TState; public - constructor Create(const AProducer: TProducer; ALookback: Int64); + constructor Create(ALookback: Integer); destructor Destroy; override; function Update(var Value: TSeries): Boolean; + + class function CreateDataEndpoint(Lookback: Integer; out Series: TLazy>): IConsumer; end; TTickAggregation = class(TMycConverter, TDataPoint>) @@ -239,9 +228,7 @@ type TMycDataJoin = class(TInterfacedObject, IProducer>) private FConsumers: array of record - Producer: IProducer; Consumer: TMycGenericConsumer; - Tag: TTag; Queue: TQueue; end; @@ -249,11 +236,13 @@ type FLock: TSpinLock; FCount: Integer; function Consume(Idx: Integer; const Value: T): TState; + function GetConsumers(Idx: Integer): IConsumer; function Link(const Consumer: IConsumer>): TTag; procedure Unlink(Tag: TTag); public - constructor Create(const AProducers: TArray>); + constructor Create(ACount: Integer); destructor Destroy; override; + property Consumers[Idx: Integer]: IConsumer read GetConsumers; end; TMycComposedConverter = class(TInterfacedObject, IConverter) @@ -363,7 +352,7 @@ end; constructor TMycConverter.Create; begin inherited Create; - FConsumer := TConsumer.Create(Self); + FConsumer := TMycGenericConsumer.Create(Self, Consume); end; { TMycConverter } @@ -486,19 +475,15 @@ end; { TMycDataEndpoint } -constructor TMycDataEndpoint.Create(const AProducer: TProducer; ALookback: Int64); +constructor TMycDataEndpoint.Create(ALookback: Integer); begin inherited Create; - FProducer := AProducer; FLookback := ALookback; - FConsumer := TMycGenericConsumer.Create(Self, Consume); - FTag := FProducer.Link(FConsumer); end; destructor TMycDataEndpoint.Destroy; begin - FProducer.Unlink(FTag); FConsumer.Free; inherited; end; @@ -524,6 +509,15 @@ begin end; end; +class function TMycDataEndpoint.CreateDataEndpoint(Lookback: Integer; out Series: TLazy>): IConsumer; +var + endPoint: TMycDataEndpoint; +begin + endPoint := TMycDataEndpoint.Create(Lookback); + Result := endPoint.FConsumer; + Series := endPoint; +end; + function TMycDataEndpoint.Update(var Value: TSeries): Boolean; begin FLock.BeginWrite; @@ -801,11 +795,11 @@ end; { TMycDataJoin } -constructor TMycDataJoin.Create(const AProducers: TArray>); +constructor TMycDataJoin.Create(ACount: Integer); begin inherited Create; - SetLength(FConsumers, Length(AProducers)); + SetLength(FConsumers, ACount); FLock := TSpinLock.Create(false); FCount := Length(FConsumers); @@ -822,9 +816,7 @@ begin with FConsumers[i] do begin Queue := TQueue.Create; - Producer := AProducers[i]; Consumer := TMycGenericConsumer.Create(Self, cFunc(i)); - Tag := Producer.Link(Consumer); end; end; @@ -833,7 +825,6 @@ begin for var i := High(FConsumers) downto 0 do with FConsumers[i] do begin - Producer.Unlink(FConsumers[i].Tag); Consumer.Free; Queue.Free; end; @@ -884,6 +875,11 @@ begin FContainedProvider.Broadcast(Arr); end; +function TMycDataJoin.GetConsumers(Idx: Integer): IConsumer; +begin + Result := FConsumers[Idx].Consumer; +end; + { TMycGenericAggregator } constructor TMycGenericAggregator.Create(const AFunc: TConstFunc.TBroadcastProc, TState>); @@ -1005,15 +1001,4 @@ begin end; -constructor TMycConverter.TConsumer.Create(AOwner: TMycConverter); -begin - inherited Create(AOwner); - FOwner := AOwner; -end; - -function TMycConverter.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 91358ac..b43dae4 100644 --- a/Src/Myc.Trade.DataPoint.pas +++ b/Src/Myc.Trade.DataPoint.pas @@ -17,6 +17,7 @@ type function Consume(const Value: T): TState; end; + // A producer generates data and distributes it to linked consumers. IProducer = interface function Link(const Consumer: IConsumer): TTag; procedure Unlink(Tag: TTag); @@ -30,17 +31,18 @@ type property Consumer: IConsumer read GetConsumer; end; - // Interface helper for IProducer providing the null object pattern. + // Interface helper for IProducer providing the null object pattern nad subscriptions for linked consumers. TProducer = record public type TSubscription = record private - FProducer: IProducer; + FOwner: IProducer; FTag: TTag; public procedure Unlink; - property Producer: IProducer read FProducer; + class operator Implicit(A: TSubscription): TProducer; overload; + property Owner: IProducer read FOwner; end; private FProducer: IProducer; @@ -54,17 +56,14 @@ type class operator Implicit(const A: IProducer): TProducer; overload; class operator Implicit(const A: TProducer): IProducer; overload; - function Link(const Consumer: IConsumer): TSubscription; inline; + function CreateLink(const Consumer: IConsumer): TSubscription; inline; + function CreateEndpoint(Lookback: Int64; out Series: TLazy>): TSubscription; + function CreateSequence(Count: Integer; out Seq: TArray>): TSubscription; overload; experimental; + // Chain consumers function Chain(const Next: IConsumer): IConsumer; overload; inline; - function Chain(const Next: IConverter): TProducer; overload; inline; function Chain(const Func: TConstFunc): TProducer; overload; inline; - - function CreateEndpoint(Lookback: Int64): TLazy>; - - function CreateSequence(Count: Integer): TArray>; overload; experimental; - // Extracts the field of a record by it's name (using RTTI). function Field(const FieldName: String): TProducer; inline; @@ -93,7 +92,8 @@ type class operator Implicit(const A: TConverter): IConverter; overload; class function Construct(const Consumer: IConsumer; const Producer: TProducer): TConverter; static; - class function CreateGeneric(const Func: TConstFunc): TConverter; static; + + class function CreateConverter(const Func: TConstFunc): TConverter; static; class function CreateAggregation(const Func: TConstFunc): TConverter; static; property Consumer: IConsumer read GetConsumer; @@ -102,20 +102,21 @@ type class property Null: IConverter read GetNull; end; + TDataRecordLayoutHelper = record helper for TDataRecord.TLayout + function FieldAsRecord(const Name: String): TConverter; + function FieldOfRecord(const Name: String): TConverter; + end; + // Factory for creating specific converter instances. TConverter = record class function CreateCounter: TConverter; static; class function CreateTicker: TConverter, T>; static; - class function CreateRecordField(const FieldName: String): TConverter; static; class function CreateIdentity: TConverter; static; class function CreateTickAggregation(Timeframe: TTimeframe): TConverter, TDataPoint>; static; class function CreateOhlcAggregation(Timeframe: TTimeframe): TConverter, TDataPoint>; static; - class function CreateDataPointConverter(const Func: TConstFunc): TConverter, TDataPoint>; static; - - class function FieldToRecord(const Layout: TDataRecord.TLayout; const Name: String): TConverter; static; - class function FieldOfRecord(const Layout: TDataRecord.TLayout; const Name: String): TConverter; static; + class function CreateEndpoint(Lookback: Int64; out Series: TLazy>): IConsumer; static; class function Join(const Producers: TArray>): TProducer>; static; @@ -139,6 +140,13 @@ type ): TConverter, TDataRecord>; static; end; + TDataRecordHelper = record helper for TArray> + function JoinRecords( + const TargetLayout: TDataRecord.TLayout; + const Mapping: TArray + ): TProducer; experimental; + end; + implementation uses @@ -150,11 +158,16 @@ procedure TProducer.TSubscription.Unlink; begin if FTag <> nil then begin - FProducer.Unlink(FTag); + FOwner.Unlink(FTag); FTag := nil; end; end; +class operator TProducer.TSubscription.Implicit(A: TSubscription): TProducer; +begin + Result := A.FOwner; +end; + constructor TProducer.Create(const AProducer: IProducer); begin FProducer := AProducer; @@ -170,7 +183,6 @@ end; function TProducer.Chain(const Next: IConverter): TProducer; begin - // REFACTOR: Link to the consumer part, return the converter as the new producer FProducer.Link(Next.Consumer); Result := Next; end; @@ -180,14 +192,14 @@ begin Result := Chain(TMycGenericConverter.Create(Func)); end; -function TProducer.CreateEndpoint(Lookback: Int64): TLazy>; +function TProducer.CreateEndpoint(Lookback: Int64; out Series: TLazy>): TSubscription; begin - Result := TMycDataEndpoint.Create(FProducer, Lookback); + Result := CreateLink(TMycDataEndpoint.CreateDataEndpoint(Lookback, Series)); end; -function TProducer.CreateSequence(Count: Integer): TArray>; +function TProducer.CreateSequence(Count: Integer; out Seq: TArray>): TSubscription; begin - FProducer.Link(TMycSequence.CreateSequence(Count, Result)); + Result := CreateLink(TMycSequence.CreateSequence(Count, Seq)); end; function TProducer.Field(const FieldName: String): TProducer; @@ -215,9 +227,9 @@ begin Result := A.FProducer; end; -function TProducer.Link(const Consumer: IConsumer): TSubscription; +function TProducer.CreateLink(const Consumer: IConsumer): TSubscription; begin - Result.FProducer := FProducer; + Result.FOwner := FProducer; Result.FTag := FProducer.Link(Consumer); end; @@ -243,7 +255,7 @@ begin Result := TMycGenericAggregator.Create(Func); end; -class function TConverter.CreateGeneric(const Func: TConstFunc): TConverter; +class function TConverter.CreateConverter(const Func: TConstFunc): TConverter; begin Result := TMycGenericConverter.Create(Func); end; @@ -290,17 +302,9 @@ begin Result := TMycDataCounter.Create; end; -class function TConverter.CreateDataPointConverter(const Func: TConstFunc): TConverter, TDataPoint>; +class function TConverter.CreateEndpoint(Lookback: Int64; out Series: TLazy>): IConsumer; begin - var cFunc: TConstFunc := Func; - Result := - TMycGenericConverter, TDataPoint>.Create( - function(const Value: TDataPoint): TDataPoint - begin - Result.Time := Value.Time; - Result.Data := cFunc(Value.Data); - end - ); + Result := TMycDataEndpoint.CreateDataEndpoint(Lookback, Series); end; class function TConverter.CreateIdentity: TConverter; @@ -308,11 +312,6 @@ begin Result := TMycIdentityConverter.Create; end; -class function TConverter.CreateRecordField(const FieldName: String): TConverter; -begin - Result := TMycRecordFieldReader.Create(FieldName); -end; - class function TConverter.CreateTicker: TConverter, T>; begin Result := TMycTicker.Create; @@ -344,7 +343,7 @@ begin end; Result := - TConverter, TDataRecord>.CreateGeneric( + TConverter, TDataRecord>.CreateConverter( function(const Inputs: TArray): TDataRecord begin Result := TDataRecord.Create(Output); @@ -355,28 +354,15 @@ begin ); end; -class function TConverter.FieldOfRecord(const Layout: TDataRecord.TLayout; const Name: String): TConverter; -begin - var idx := Layout.IndexOf(Name); - Result := TConverter.CreateGeneric(function(const Value: TDataRecord): T begin Value.GetValue(idx, Result); end); -end; - -class function TConverter.FieldToRecord(const Layout: TDataRecord.TLayout; const Name: String): TConverter; -begin - var idx := Layout.IndexOf(Name); - Result := - TConverter.CreateGeneric( - function(const Value: T): TDataRecord - begin - Result := TDataRecord.Create(Layout); - Result.SetValue(idx, Value); - end - ); -end; - class function TConverter.Join(const Producers: TArray>): TProducer>; +var + joiner: TMycDataJoin; begin - Result := TMycDataJoin.Create(Producers); + joiner := TMycDataJoin.Create(Length(Producers)); + Result := joiner; + + for var i := 0 to High(Producers) do + Producers[i].Chain(joiner.Consumers[i]); end; class function TConverter.JoinRecords( @@ -389,8 +375,39 @@ begin Result := DataMapping(Mapping, TargetLayout); - // REFACTOR: Link to the consumer part of the mapping converter. - RecProvider.Link(Result.Consumer); + RecProvider.Chain(Result.Consumer); +end; + +function TDataRecordLayoutHelper.FieldAsRecord(const Name: String): TConverter; +begin + var layout := Self; + var idx := IndexOf(Name); + Result := + TConverter.CreateConverter( + function(const Value: T): TDataRecord + begin + Result := TDataRecord.Create(layout); + Result.SetValue(idx, Value); + end + ); +end; + +function TDataRecordLayoutHelper.FieldOfRecord(const Name: String): TConverter; +begin + var idx := IndexOf(Name); + Result := TConverter.CreateConverter(function(const Value: TDataRecord): T begin Value.GetValue(idx, Result); end); +end; + +function TDataRecordHelper.JoinRecords( + const TargetLayout: TDataRecord.TLayout; + const Mapping: TArray +): TProducer; +begin + var map := TConverter.DataMapping(Mapping, TargetLayout); + + TConverter.Join(Self).Chain(map.Consumer); + + Result := map.Producer; end; end.