Brummel/MycLib
1
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects 1 Releases Wiki Activity

Identifiers refactored to Producer-Consumer-Pattern

Browse Source
This commit is contained in:
Michael Schimmel
2025-07-24 10:15:18 +02:00
parent 6a114f77c5
commit 4859c49738
8 changed files with 504 additions and 462 deletions
Download Patch File Download Diff File Expand all files Collapse all files
AuraTrader/AuraTrader.dproj
+1 -1
View File
@@ -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>
AuraTrader/MainForm.pas
+24 -24
View File
@@ -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.
AuraTrader/StrategyTest.pas
+2 -2
View File
@@ -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.
Src/Myc.Fmx.Chart.Series.pas
+19 -19
View File
@@ -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;
Src/Myc.Fmx.Chart.pas
+12 -13
View File
@@ -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;
Src/Myc.Trade.DataPoint.Impl.pas
+344 -267
View File
File diff suppressed because it is too large Load Diff
Src/Myc.Trade.DataPoint.pas
+94 -128
View File
@@ -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.
Src/Myc.Trade.DataStream.pas
+8 -8
View File
@@ -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