From a3da63ad6a7ea17224198127bdaf2edfc56e8ddc Mon Sep 17 00:00:00 2001 From: Michael Schimmel Date: Tue, 24 Jun 2025 08:11:46 +0200 Subject: [PATCH] DataPoint --- AuraTrader/AuraTrader.dpr | 3 +- AuraTrader/AuraTrader.dproj | 1 + AuraTrader/MainForm.fmx | 13 +- AuraTrader/MainForm.pas | 124 ++++--- Src/Myc.Core.Signals.pas | 115 ++++-- Src/Myc.Futures.pas | 6 +- Src/Myc.Signals.FMX.pas | 36 +- Src/Myc.Signals.pas | 58 +++- Src/Myc.Test.Trade.DataPoint.pas | 579 ++++++++++++++++--------------- Src/Myc.Trade.Core.DataPoint.pas | 226 ++++++++++++ Src/Myc.Trade.DataPoint.pas | 325 ++++++----------- Src/Myc.Trade.DataProvider.pas | 32 +- Src/Myc.Trade.DataStream.pas | 17 +- Test/MycTests.res | Bin 152412 -> 152412 bytes 14 files changed, 902 insertions(+), 633 deletions(-) create mode 100644 Src/Myc.Trade.Core.DataPoint.pas diff --git a/AuraTrader/AuraTrader.dpr b/AuraTrader/AuraTrader.dpr index 3ee392a..03891ae 100644 --- a/AuraTrader/AuraTrader.dpr +++ b/AuraTrader/AuraTrader.dpr @@ -4,7 +4,8 @@ uses FastMM5, System.StartUpCopy, FMX.Forms, - MainForm in 'MainForm.pas' {Form1}; + MainForm in 'MainForm.pas' {Form1}, + Myc.Trade.Core.DataPoint in '..\Src\Myc.Trade.Core.DataPoint.pas'; {$R *.res} diff --git a/AuraTrader/AuraTrader.dproj b/AuraTrader/AuraTrader.dproj index d7d0f0e..db3eec0 100644 --- a/AuraTrader/AuraTrader.dproj +++ b/AuraTrader/AuraTrader.dproj @@ -133,6 +133,7 @@
Form1
 + Base diff --git a/AuraTrader/MainForm.fmx b/AuraTrader/MainForm.fmx index ce121c7..b2f7f61 100644 --- a/AuraTrader/MainForm.fmx +++ b/AuraTrader/MainForm.fmx @@ -28,7 +28,7 @@ object Form1: TForm1 Anchors = [akLeft, akTop, akRight] Position.X = 8.00000000000000000 Position.Y = 8.00000000000000000 - Size.Width = 489.00000000000000000 + Size.Width = 273.00000000000000000 Size.Height = 513.00000000000000000 Size.PlatformDefault = False end @@ -53,6 +53,17 @@ object Form1: TForm1 TextSettings.Trimming = None OnClick = LoadButtonClick end + object FlowLayout: TFlowLayout + Position.X = 304.00000000000000000 + Position.Y = 104.00000000000000000 + Size.Width = 449.00000000000000000 + Size.Height = 361.00000000000000000 + Size.PlatformDefault = False + TabOrder = 5 + Justify = Left + JustifyLastLine = Left + FlowDirection = LeftToRight + end end object LogMemo: TMemo Touch.InteractiveGestures = [Pan, LongTap, DoubleTap] diff --git a/AuraTrader/MainForm.pas b/AuraTrader/MainForm.pas index d2fdf64..8617ac6 100644 --- a/AuraTrader/MainForm.pas +++ b/AuraTrader/MainForm.pas @@ -9,6 +9,7 @@ uses System.Classes, System.Variants, System.DateUtils, + System.Generics.Collections, FMX.Types, FMX.Controls, FMX.Forms, @@ -27,9 +28,12 @@ uses Myc.Futures, Myc.Trade.DataStream, Myc.Trade.DataPoint, + Myc.Trade.DataProvider, Myc.Signals, + Myc.Lazy, Myc.Signals.FMX, - FMX.ListBox; + FMX.ListBox, + FMX.Layouts; type TForm1 = class(TForm) @@ -39,10 +43,20 @@ type Path1: TPath; SymbolsComboBox: TComboBox; LoadButton: TButton; + FlowLayout: TFlowLayout; procedure RandomButtonClick(Sender: TObject); procedure FormCreate(Sender: TObject); procedure FormDestroy(Sender: TObject); procedure LoadButtonClick(Sender: TObject); + private + const + cnt = 20; + type + TRndItem = record + Stream: IDataStream; + Data: TMutable>; + Labl: TLabel; + end; private FOnEvent: TNotifyEvent; { Private declarations } @@ -51,9 +65,7 @@ type FStreamUpdate: TFMXValidationState; FSymbols: TFuture>; FStream: IDataStream; - FRandomStream: array[0..19] of IDataStream; - FStreamIdx: Integer; - procedure DoIdle(Sender: TObject; var Done: Boolean); + FRandom: TList; public { Public declarations } published @@ -69,6 +81,8 @@ implementation procedure TForm1.FormCreate(Sender: TObject); begin + FRandom := TList.Create; + // Create an instance of the TAuraTABFileServer. The server can be reused for multiple stream creations. [364] FServer := TAuraTABFileServer.Create('\\COFFEE\TickData\Pepperstone'); @@ -97,76 +111,80 @@ begin end; end ); - - Application.OnIdle := DoIdle; end; procedure TForm1.FormDestroy(Sender: TObject); begin Application.OnIdle := nil; FSymbolsValid.Free; + FRandom.Free; end; procedure TForm1.LoadButtonClick(Sender: TObject); begin FStream := FServer.CreateStream(FSymbols.WaitFor[SymbolsComboBox.ItemIndex]); - FStreamUpdate := TFMXValidationState.Create(Path1, FStream.HasData, - procedure - var - Data: array [0..5] of TDataPoint; - begin - var cnt := FStream.GetChunk(Data); - if cnt = 0 then exit; + FStreamUpdate := + TFMXValidationState.Create( + Path1, + FStream.HasData, + procedure + var + Data: array[0..200] of TDataPoint; + begin + var cnt := FStream.GetChunk(Data); + if cnt = 0 then + exit; - for var i:=0 to cnt-1 do - Path1.Data.LineTo( PointF(Path1.Data.Count, Data[i].Data.Ask) ); - end); + for var i := 0 to cnt - 1 do + Path1.Data.LineTo(PointF(Path1.Data.Count, Data[i].Data.Ask)); + end + ); end; procedure TForm1.RandomButtonClick(Sender: TObject); begin + var tst: TFlag := TFlag.CreateObserver( TSignal.Null ); + + + var rnd: TRndItem; + var ass := FSymbols.WaitFor; - FRandomStream[FStreamIdx] := FServer.CreateStream(ass[Random(Length(ass))]); + rnd.Stream := FServer.CreateStream(ass[Random(Length(FSymbols.WaitFor))]); + rnd.Data := TDataStreamProvider.Create(3000, 100, rnd.Stream); - for var i := 0 to High(FRandomStream) do - if LogMemo.Lines.Count = i then - LogMemo.Lines.Add(''); - LogMemo.Lines[FStreamIdx] := FRandomStream[FStreamIdx].Symbol + ' loading...'; + rnd.Labl := TLabel.Create(Self); + rnd.Labl.Parent := FlowLayout; + rnd.Labl.Align := TAlignLayout.None; + rnd.Labl.WordWrap := false; + rnd.Labl.Width := 300; + rnd.Labl.Text := rnd.Stream.Symbol + ' loading...'; + FRandom.Add(rnd); - FStreamIdx := (FStreamIdx + 1) mod Length(FRandomStream); -end; - -procedure TForm1.DoIdle(Sender: TObject; var Done: Boolean); -var - Items: TArray>; -begin - SetLength(Items, 200); - - LogMemo.BeginUpdate; - try - for var i := 0 to High(FRandomStream) do + var Proc := + procedure(idx: Integer) begin - var str := ''; - if LogMemo.Lines.Count > i then - str := LogMemo.Lines[i] - else - LogMemo.Lines.Add(''); - - if Assigned(FRandomStream[i]) then - begin - var cnt := FRandomStream[i].GetChunk(Items); - if cnt > 0 then - begin - var d := Items[cnt - 1]; - str := FRandomStream[i].Symbol + ' ' + cnt.ToString + ' ' + d.Time.ToString + ': ' + d.Data.Ask.ToString; - end; - end; - - LogMemo.Lines[i] := str; + TFMXValidationState.Create( + FRandom[idx].Labl, + FRandom[idx].Stream.HasData, + procedure + begin + var data := FRandom[idx].Data.Value; + if data.Count > 0 then + begin + var dp := data[0]; + FRandom[idx].Labl.Text := + FRandom[idx].Stream.Symbol + + ' ' + + dp.Time.ToString + + ' ' + + data.TotalCount.ToString + + ' ' + + dp.Data.Ask.ToString; + end; + end + ); end; - finally - LogMemo.EndUpdate; - end; + Proc(FRandom.Count - 1); end; end. diff --git a/Src/Myc.Core.Signals.pas b/Src/Myc.Core.Signals.pas index 480b9d6..e346181 100644 --- a/Src/Myc.Core.Signals.pas +++ b/Src/Myc.Core.Signals.pas @@ -13,15 +13,18 @@ type TMycNullSignal = class(TInterfacedObject, TSignal.ISignal) public function Subscribe(Subscriber: TSignal.ISubscriber): Pointer; - procedure Unsubscribe(Tag: TMycNotifyList.TTag); + procedure Unsubscribe(Tag: Pointer); end; // TMycNullState implements a "null object" pattern for IMycState. // This state is always considered "set". - TMycNullState = class(TMycNullSignal, TState.IState) + TMycNullState = class(TInterfacedObject, TSignal.ISignal, TState.IState) protected // TState.IState function GetIsSet: Boolean; + public + function Subscribe(Subscriber: TSignal.ISubscriber): Pointer; + procedure Unsubscribe(Tag: Pointer); end; // A state that acts as an event. It's state ist always set and it will always Trigger it's subscribers, if it gets notified by itself. @@ -35,7 +38,7 @@ type constructor Create; destructor Destroy; override; function Subscribe(Subscriber: TSignal.ISubscriber): Pointer; - procedure Unsubscribe(Tag: TMycNotifyList.TTag); + procedure Unsubscribe(Tag: Pointer); function Notify: Boolean; end; @@ -45,6 +48,16 @@ type function Notify: Boolean; end; + TMycRouterEvent = class(TMycEvent) + private + FSignal: TSignal.ISignal; + FSubscriptionTag: TSignal.TSubscriptionTag; + public + constructor Create(const ASignal: TSignal.ISignal); + procedure AfterConstruction; override; + procedure BeforeDestruction; override; + end; + // TMycLatch implements a countdown latch. // It is initialized with a count. Calls to its Notify method (as an TSignal.ISubscriber) // decrement this count. When the count reaches zero, the latch transitions to the "set" @@ -66,7 +79,7 @@ type // ISignal implementation function Subscribe(Subscriber: TSignal.ISubscriber): Pointer; - procedure Unsubscribe(Tag: TMycNotifyList.TTag); + procedure Unsubscribe(Tag: Pointer); // TSignal.ISubscriber implementation for TMycLatch itself. // Decrements the internal count. If the count reaches zero, @@ -101,7 +114,7 @@ type class function CreateDirty: TFlag.IFlag; static; function Subscribe(Subscriber: TSignal.ISubscriber): Pointer; - procedure Unsubscribe(Tag: TMycNotifyList.TTag); + procedure Unsubscribe(Tag: Pointer); // TSignal.ISubscriber implementation for TMycFlag. // Sets this flag to dirty (true). Notifies subscribers if it transitioned from clean to dirty. @@ -118,16 +131,54 @@ type function Reset: Boolean; end; + TMycObserverFlag = class(TMycFlag) + private + FSignal: TSignal.ISignal; + FSubscriptionTag: TSignal.TSubscriptionTag; + public + constructor Create(const ASignal: TSignal.ISignal); + procedure AfterConstruction; override; + procedure BeforeDestruction; override; + end; + implementation uses System.SyncObjs; // For TInterlocked +{ TMycNullSignal } + +function TMycNullSignal.Subscribe(Subscriber: TSignal.ISubscriber): Pointer; +begin + // No ongoing subscription is necessary or meaningful for a null state. + Result := nil; +end; + +procedure TMycNullSignal.Unsubscribe(Tag: Pointer); +begin + // Unsubscribing from a null signal has no effect. +end; + +{ TMycNullState } + function TMycNullState.GetIsSet: Boolean; begin Result := true; // Null state is always set. end; +function TMycNullState.Subscribe(Subscriber: TSignal.ISubscriber): Pointer; +begin + // Since the state is always set, notify the subscriber immediately. + if Assigned(Subscriber) then + Subscriber.Notify; + Result := nil; +end; + +procedure TMycNullState.Unsubscribe(Tag: Pointer); +begin + // Unsubscribing from a null state has no effect. +end; + { TMycEvent } constructor TMycEvent.Create; @@ -178,7 +229,7 @@ begin end; end; -procedure TMycEvent.Unsubscribe(Tag: TMycNotifyList.TTag); +procedure TMycEvent.Unsubscribe(Tag: Pointer); begin if Tag = nil then exit; @@ -191,6 +242,26 @@ begin end; end; +{ TMycRouterEvent } + +constructor TMycRouterEvent.Create(const ASignal: TSignal.ISignal); +begin + inherited Create; + FSignal := ASignal; +end; + +procedure TMycRouterEvent.AfterConstruction; +begin + inherited; + FSubscriptionTag := FSignal.Subscribe(Self); +end; + +procedure TMycRouterEvent.BeforeDestruction; +begin + FSignal.Unsubscribe(FSubscriptionTag); + inherited; +end; + { TMycNullLatch } function TMycNullLatch.GetState: TState; @@ -289,7 +360,7 @@ begin end; end; -procedure TMycLatch.Unsubscribe(Tag: TMycNotifyList.TTag); +procedure TMycLatch.Unsubscribe(Tag: Pointer); begin if Tag = nil then exit; @@ -324,7 +395,7 @@ end; constructor TMycFlag.Create; begin inherited Create; - FFlag := true; // A new dirty flag is initially considered dirty. + FFlag := false; FSubscribers.Create; end; @@ -371,9 +442,9 @@ begin begin FSubscribers.Notify(function(Subscriber: TSignal.ISubscriber): Boolean begin Result := Subscriber.Notify; end); end; - Result := true; finally FSubscribers.Release; + Result := true; end; end; @@ -399,7 +470,7 @@ begin end; end; -procedure TMycFlag.Unsubscribe(Tag: TMycNotifyList.TTag); +procedure TMycFlag.Unsubscribe(Tag: Pointer); begin if Tag = nil then exit; @@ -422,20 +493,24 @@ begin Result := false; end; -function TMycNullSignal.Subscribe(Subscriber: TSignal.ISubscriber): Pointer; +{ TMycObserverFlag } + +constructor TMycObserverFlag.Create(const ASignal: TSignal.ISignal); begin - // Since the state is always set, notify the subscriber immediately. - if Assigned(Subscriber) then - begin - Subscriber.Notify; - end; - // No ongoing subscription is necessary or meaningful for a null state. - Result := nil; + inherited Create; + FSignal := ASignal; end; -procedure TMycNullSignal.Unsubscribe(Tag: TMycNotifyList.TTag); +procedure TMycObserverFlag.AfterConstruction; begin - // Unsubscribing from a null state has no effect. + inherited; + FSubscriptionTag := FSignal.Subscribe(Self); +end; + +procedure TMycObserverFlag.BeforeDestruction; +begin + FSignal.Unsubscribe(FSubscriptionTag); + inherited; end; end. diff --git a/Src/Myc.Futures.pas b/Src/Myc.Futures.pas index 4d0754e..67f5b66 100644 --- a/Src/Myc.Futures.pas +++ b/Src/Myc.Futures.pas @@ -34,7 +34,7 @@ type private FFuture: IFuture; - function GetDone: TState; inline; + function GetDone: TState.IState; inline; function GetValue: T; inline; {$ENDREGION} public @@ -55,7 +55,7 @@ type function Chain(const Proc: TFuncConst): TFuture; overload; function WaitFor: T; - property Done: TState read GetDone; + property Done: TState.IState read GetDone; property Value: T read GetValue; end; @@ -105,7 +105,7 @@ begin Result := TMycGateFuncFuture.Create(TaskManager, Gate, Proc); end; -function TFuture.GetDone: TState; +function TFuture.GetDone: TState.IState; begin Result := FFuture.Done; end; diff --git a/Src/Myc.Signals.FMX.pas b/Src/Myc.Signals.FMX.pas index aa78c16..8f088c0 100644 --- a/Src/Myc.Signals.FMX.pas +++ b/Src/Myc.Signals.FMX.pas @@ -5,23 +5,25 @@ interface uses System.Classes, System.SysUtils, + System.Messaging, FMX.Controls, Myc.Signals; type - TFMXValidationState = class(TComponent, TSignal.ISubscriber) + TFMXValidationState = class(TComponent) private [volatile] - FInvalidated: Integer; - FSignal: TSignal.ISignal; - FSubscription: TSignal.TSubscriptionTag; + FSignal: TSignal; + FInvalidated: TFlag; + FSubscription: TSignal.TSubscription; FProc: TProc; - procedure CallProc; + FIdleSubsription: TMessageSubscriptionId; + procedure DoIdle(const Sender: TObject; const M: TMessage); protected function Notify: Boolean; public - constructor Create(AOwner: TControl; const ASignal: TSignal.ISignal; const AProc: TProc); reintroduce; + constructor Create(AOwner: TControl; const ASignal: TSignal; const AProc: TProc); reintroduce; destructor Destroy; override; procedure AfterConstruction; override; procedure BeforeDestruction; override; @@ -30,47 +32,47 @@ type implementation uses - System.SyncObjs; + System.SyncObjs, + FMX.Types; { TFMXValidationState } -constructor TFMXValidationState.Create(AOwner: TControl; const ASignal: TSignal.ISignal; const AProc: TProc); +constructor TFMXValidationState.Create(AOwner: TControl; const ASignal: TSignal; const AProc: TProc); begin inherited Create(AOwner); - FInvalidated := 0; FSignal := ASignal; + FInvalidated := TFlag.CreateFlag; FProc := AProc; end; destructor TFMXValidationState.Destroy; begin - TThread.RemoveQueuedEvents(CallProc); inherited; end; procedure TFMXValidationState.AfterConstruction; begin inherited; - FSubscription := FSignal.Subscribe(Self); + FSubscription := FSignal.Subscribe(FInvalidated); + FIdleSubsription := TMessageManager.DefaultManager.SubscribeToMessage(TIdleMessage, DoIdle); end; procedure TFMXValidationState.BeforeDestruction; begin - FSignal.Unsubscribe(FSubscription); + TMessageManager.DefaultManager.Unsubscribe(TIdleMessage, FIdleSubsription); + FSubscription.Unsubscribe; inherited; end; -procedure TFMXValidationState.CallProc; +procedure TFMXValidationState.DoIdle(const Sender: TObject; const M: TMessage); begin - if TInterlocked.Exchange(FInvalidated, 0) > 0 then + if FInvalidated.Reset then FProc; end; function TFMXValidationState.Notify: Boolean; begin - if TInterlocked.Exchange(FInvalidated, 1) = 0 then - TThread.Queue(nil, CallProc); - exit(true); + Result := FInvalidated.Notify; end; end. diff --git a/Src/Myc.Signals.pas b/Src/Myc.Signals.pas index dfd9855..18eecbc 100644 --- a/Src/Myc.Signals.pas +++ b/Src/Myc.Signals.pas @@ -69,6 +69,7 @@ type private FState: IState; function GetIsSet: Boolean; inline; + function GetAsSignal: TSignal; inline; {$ENDREGION} public constructor Create(const AState: IState); @@ -84,6 +85,10 @@ type function Subscribe(Subscriber: TSignal.ISubscriber): TSignal.TSubscription; inline; property IsSet: Boolean read GetIsSet; + + // Explicitly cast a state to a signal. This is dangerous, use with care! + // States only notify, when they change. This behaviour is obscured by treating it like a signal. + property AsSignal: TSignal read GetAsSignal; end; // An event is a stateless signal that forwards all notifications to the subscribers. @@ -113,6 +118,7 @@ type class operator Implicit(const A: TEvent): IEvent; overload; class function CreateEvent: TEvent; static; + class function CreateRouter(const Signal: TSignal): TEvent; static; class property Null: IEvent read FNull; @@ -141,7 +147,7 @@ type private FLatch: ILatch; - function GetState: TState; inline; + function GetState: TState.IState; inline; {$ENDREGION} public constructor Create(const ALatch: ILatch); @@ -149,7 +155,7 @@ type class operator Implicit(const A: ILatch): TLatch; overload; class operator Implicit(const A: TLatch): ILatch; overload; - class function CreateLatch(Count: Integer): TLatch; static; + class function CreateLatch(Count: Integer): ILatch; static; class function Enqueue1(var Gate: TLatch): TState; overload; static; function Enqueue: TState; overload; @@ -158,7 +164,7 @@ type function Notify: Boolean; inline; - property State: TState read GetState; + property State: TState.IState read GetState; end; // TFlag represents a resettable flag, typically indicating if a State is "dirty" (requiring attention) or "clean". @@ -179,17 +185,18 @@ type class constructor ClassCreate; private - FDirty: IFlag; + FFlag: IFlag; function GetState: TState; inline; {$ENDREGION} public - constructor Create(const ADirty: IFlag); + constructor Create(const AFlag: IFlag); class operator Initialize(out Dest: TFlag); class operator Implicit(const A: IFlag): TFlag; overload; class operator Implicit(const A: TFlag): IFlag; overload; - class function CreateFlag: IFlag; static; + class function CreateFlag: TFlag; static; + class function CreateObserver(const Signal: TSignal): TFlag; static; class property Null: IFlag read FNull; function Notify: Boolean; inline; @@ -268,6 +275,11 @@ begin end; end; +function TState.GetAsSignal: TSignal; +begin + Result := FState; +end; + function TState.GetIsSet: Boolean; begin Result := FState.IsSet; @@ -312,6 +324,11 @@ begin Result := TMycEvent.Create; end; +class function TEvent.CreateRouter(const Signal: TSignal): TEvent; +begin + Result := TMycRouterEvent.Create(Signal); +end; + function TEvent.GetSignal: TSignal; begin Result := FEvent.Signal; @@ -344,31 +361,36 @@ begin FNull := TMycNullFlag.Create; end; -constructor TFlag.Create(const ADirty: IFlag); +constructor TFlag.Create(const AFlag: IFlag); begin - FDirty := ADirty; - if not Assigned(FDirty) then - FDirty := FNull; + FFlag := AFlag; + if not Assigned(FFlag) then + FFlag := FNull; end; -class function TFlag.CreateFlag: IFlag; +class function TFlag.CreateFlag: TFlag; begin Result := TMycFlag.Create; end; +class function TFlag.CreateObserver(const Signal: TSignal): TFlag; +begin + Result.Create(TMycObserverFlag.Create(Signal)); +end; + function TFlag.GetState: TState; begin - Result := FDirty.State; + Result := FFlag.State; end; function TFlag.Notify: Boolean; begin - Result := FDirty.Notify; + Result := FFlag.Notify; end; function TFlag.Reset: Boolean; begin - Result := FDirty.Reset; + Result := FFlag.Reset; end; class operator TFlag.Implicit(const A: IFlag): TFlag; @@ -378,12 +400,12 @@ end; class operator TFlag.Implicit(const A: TFlag): IFlag; begin - Result := A.FDirty; + Result := A.FFlag; end; class operator TFlag.Initialize(out Dest: TFlag); begin - Dest.FDirty := FNull; + Dest.FFlag := FNull; end; { TLatch } @@ -402,7 +424,7 @@ begin FLatch := FNull; end; -class function TLatch.CreateLatch(Count: Integer): TLatch; +class function TLatch.CreateLatch(Count: Integer): ILatch; begin if Count > 0 then Result := TMycLatch.Create(Count) @@ -452,7 +474,7 @@ begin end; end; -function TLatch.GetState: TState; +function TLatch.GetState: TState.IState; begin Result := FLatch.State; end; diff --git a/Src/Myc.Test.Trade.DataPoint.pas b/Src/Myc.Test.Trade.DataPoint.pas index 0b2fcb8..ba048c3 100644 --- a/Src/Myc.Test.Trade.DataPoint.pas +++ b/Src/Myc.Test.Trade.DataPoint.pas @@ -11,9 +11,9 @@ uses type [TestFixture] - TTestTDataSeries = class(TObject) + TTestDataArray = class(TObject) private - FSeries: TDataSeries; + FArray: IDataArray; procedure SetupSeriesWithData(ItemCount: Integer = 10; MaxLookBack: Int64 = 0); public [Setup] @@ -21,6 +21,7 @@ type [Teardown] procedure Teardown; + // Tests for Setup and basic Add/Count [Test] procedure TestSetupSeriesWithDataVerification; [Test] @@ -30,22 +31,6 @@ type procedure TestAddOrderAssertion; [Test] procedure TestGetItemsIndexing; - [Test] - procedure TestIndexOfExistingTimeStamp; - [Test] - procedure TestIndexOfNonExistingBetween; - [Test] - procedure TestIndexOfBeforeFirstItem; - [Test] - procedure TestIndexOfAfterLastItem; - [Test] - procedure TestIndexOfExactOldestItem; - [Test] - procedure TestIndexOfExactNewestItem; - [Test] - procedure TestIndexOfEmptySeries; - [Test] - procedure TestIndexOfSingleItemSeries; // Tests for MaxLookback [Test] @@ -55,8 +40,6 @@ type [Test] [IgnoreMemoryLeaks] procedure TestLoopIsSafeWithMaxLookback; - [Test] - procedure TestIndexOfRespectsMaxLookback; // Tests for bulk Add [Test] @@ -73,48 +56,79 @@ type procedure TestBulkAddWithMaxLookback; // Tests for TotalCount - [Test] - procedure TestTotalCountIncrementsCorrectly; - [Test] - procedure TestTotalCountIgnoresTrimming; - [Test] + [Test] + procedure TestTotalCountIncrementsCorrectly; + [Test] + procedure TestTotalCountIgnoresTrimming; + [Test] procedure TestTotalCountResetsOnClear; end; + [TestFixture] + TTestDataSeries = class(TObject) + private + FArray: IDataArray; + FSeries: TDataSeries; + procedure SetupSeriesWithData(ItemCount: Integer = 10; MaxLookBack: Int64 = 0); + public + [Setup] + procedure Setup; + [Teardown] + procedure Teardown; + + // Tests for IndexOf + [Test] + procedure TestIndexOfExistingTimeStamp; + [Test] + procedure TestIndexOfNonExistingBetween; + [Test] + procedure TestIndexOfBeforeFirstItem; + [Test] + procedure TestIndexOfAfterLastItem; + [Test] + procedure TestIndexOfExactOldestItem; + [Test] + procedure TestIndexOfExactNewestItem; + [Test] + procedure TestIndexOfEmptySeries; + [Test] + procedure TestIndexOfSingleItemSeries; + [Test] + procedure TestIndexOfRespectsMaxLookback; + end; + implementation -{ TTestTDataSeries } +{ TTestDataArray } -procedure TTestTDataSeries.Setup; +procedure TTestDataArray.Setup; begin - // Managed record is implicitly initialized. + FArray := TDataSeries.CreateArray(0); end; -procedure TTestTDataSeries.Teardown; +procedure TTestDataArray.Teardown; begin - // Ensure managed record is finalized to prevent false leak reports. - FSeries := Default(TDataSeries); + FArray := nil; end; -procedure TTestTDataSeries.SetupSeriesWithData(ItemCount: Integer = 10; MaxLookBack: Int64 = 0); +procedure TTestDataArray.SetupSeriesWithData(ItemCount: Integer = 10; MaxLookBack: Int64 = 0); var i: Integer; DataPoint: TDataPoint; baseTime: TDateTime; begin - FSeries := TDataSeries.Create( MaxLookBack ); + FArray := TDataSeries.CreateArray(MaxLookBack); baseTime := EncodeDate(2020, 7, 7); // Add data points with increasing timestamps. for i := 0 to ItemCount - 1 do begin DataPoint := TDataPoint.Create(baseTime + i, TAskBidItem.Create(i * 1.0, i * 1.0 + 0.1)); - FSeries.Add(DataPoint); + FArray.Add(DataPoint); end; end; -// Verifies that the test data setup helper works as expected. -procedure TTestTDataSeries.TestSetupSeriesWithDataVerification; +procedure TTestDataArray.TestSetupSeriesWithDataVerification; var i: Integer; baseTime, expectedTime: TDateTime; @@ -123,20 +137,19 @@ begin SetupSeriesWithData; baseTime := EncodeDate(2020, 7, 7); - Assert.AreEqual(Int64(10), FSeries.Count, 'Setup should create exactly 10 items'); + Assert.AreEqual(Int64(10), FArray.Count, 'Setup should create exactly 10 items'); // Check all items to ensure correct reverse chronological order. - for i := 0 to FSeries.Count - 1 do + for i := 0 to FArray.Count - 1 do begin expectedTime := baseTime + (9 - i); expectedAsk := Single(9 - i); - Assert.AreEqual(expectedTime, FSeries[i].Time, 'Item at logical index should have correct reversed timestamp'); - Assert.AreEqual(expectedAsk, FSeries[i].Data.Ask, 'Item at logical index should have correct reversed data'); + Assert.AreEqual(expectedTime, FArray[i].Time, 'Item at logical index should have correct reversed timestamp'); + Assert.AreEqual(expectedAsk, FArray[i].Data.Ask, 'Item at logical index should have correct reversed data'); end; end; -// Tests adding new items and the resulting count. -procedure TTestTDataSeries.TestAddAndCount; +procedure TTestDataArray.TestAddAndCount; var DataPoint: TDataPoint; newTime: TDateTime; @@ -144,17 +157,16 @@ begin SetupSeriesWithData; newTime := EncodeDate(2020, 7, 17); - Assert.AreEqual(Int64(10), FSeries.Count, 'Initial count should be 10'); + Assert.AreEqual(Int64(10), FArray.Count, 'Initial count should be 10'); DataPoint := TDataPoint.Create(newTime, TAskBidItem.Create(100.0, 100.1)); - FSeries.Add(DataPoint); + FArray.Add(DataPoint); - Assert.AreEqual(Int64(11), FSeries.Count, 'Count should be 11 after adding one more'); - Assert.AreEqual(newTime, FSeries.Items[0].Time, 'Newest item should be at index 0'); + Assert.AreEqual(Int64(11), FArray.Count, 'Count should be 11 after adding one more'); + Assert.AreEqual(newTime, FArray.Items[0].Time, 'Newest item should be at index 0'); end; -// Ensures that adding an item with an older timestamp raises an assertion. -procedure TTestTDataSeries.TestAddOrderAssertion; +procedure TTestDataArray.TestAddOrderAssertion; var olderTime: TDateTime; begin @@ -165,15 +177,14 @@ begin procedure begin var DataPoint := TDataPoint.Create(olderTime, TAskBidItem.Create(0.0, 0.0)); - FSeries.Add(DataPoint); + FArray.Add(DataPoint); end, EAssertionFailed, 'Adding item with older timestamp should raise an assert error' ); end; -// Verifies that the logical-to-physical index mapping is correct. -procedure TTestTDataSeries.TestGetItemsIndexing; +procedure TTestDataArray.TestGetItemsIndexing; var i: Integer; expectedTime: TDateTime; @@ -185,13 +196,241 @@ begin for i := 0 to 9 do begin expectedTime := baseTime + (9 - i); - Assert.AreEqual(expectedTime, FSeries.Items[i].Time, 'Item at logical index should have reversed chronological time'); - Assert.AreEqual(Single(9 - i), FSeries.Items[i].Data.Ask, 'Item data at logical index should match reversed insertion order'); + Assert.AreEqual(expectedTime, FArray.Items[i].Time, 'Item at logical index should have reversed chronological time'); + Assert.AreEqual(Single(9 - i), FArray.Items[i].Data.Ask, 'Item data at logical index should match reversed insertion order'); end; end; -// Tests finding an existing item (in this case, the oldest). -procedure TTestTDataSeries.TestIndexOfExistingTimeStamp; +procedure TTestDataArray.TestCountIsCappedByMaxLookback_NoTrim; +begin + SetupSeriesWithData(500, 400); // Less than one chunk + + // The public Count must now be the MaxLookback value, even though + // no chunk was freed and FCount is still 500 internally. + Assert.AreEqual(Int64(400), FArray.Count, 'Public count should be capped by MaxLookback even if no chunk is freed'); +end; + +procedure TTestDataArray.TestCountIsCappedByMaxLookback_WithTrim; +const + ITEMS_TO_ADD = 2000; + LOOKBACK_LIMIT = 900; +begin + SetupSeriesWithData(ITEMS_TO_ADD, LOOKBACK_LIMIT); + + // Internally, FCount will be 976 after one chunk is freed. + // The public Count should still be capped at the lookback limit. + Assert.AreEqual(Int64(LOOKBACK_LIMIT), FArray.Count, 'Public count should be capped by MaxLookback after trimming'); +end; + +procedure TTestDataArray.TestLoopIsSafeWithMaxLookback; +var + dummy: TDataPoint; +begin + SetupSeriesWithData(500, 400); + Assert.AreEqual(Int64(400), FArray.Count, 'Pre-condition: Count must be capped at 400'); + + // This test proves that a standard loop based on the public Count is safe + // and will not access an invalid index. + Assert.WillNotRaise( + procedure + begin + for var i := 0 to FArray.Count - 1 do + begin + dummy := FArray[i]; + end; + end, + nil, + 'Looping up to the public Count should not raise an exception' + ); + + // Also test the assert when going one item beyond the public count + Assert.WillRaise( + procedure begin dummy := FArray[400]; end, + EAssertionFailed, + 'Accessing index at the limit of MaxLookback should raise an exception' + ); +end; + +procedure TTestDataArray.TestBulkAddIncreasesCount; +var + newData: TArray>; + baseTime, newTime: TDateTime; + i: Integer; +begin + SetupSeriesWithData(10); + baseTime := EncodeDate(2020, 7, 7); + + SetLength(newData, 5); + for i := 0 to High(newData) do + begin + newTime := baseTime + 10 + i; + newData[i] := TDataPoint.Create(newTime, TAskBidItem.Create(i, i)); + end; + + FArray.Add(newData); + + Assert.AreEqual(Int64(15), FArray.Count, 'Count should be increased by the number of items in the bulk add'); + Assert.AreEqual(baseTime + 10 + High(newData), FArray[0].Time, 'Newest item should be the last item from the added array'); +end; + +procedure TTestDataArray.TestBulkAddChronologicalAssertion_Internal; +var + newData: TArray>; +begin + SetupSeriesWithData(10); + SetLength(newData, 2); + newData[0] := TDataPoint.Create(Now + 1, TAskBidItem.Create(1, 1)); + newData[1] := TDataPoint.Create(Now, TAskBidItem.Create(2, 2)); // Not sorted + + Assert.WillRaise( + procedure begin FArray.Add(newData); end, + EAssertionFailed, + 'Bulk Add should fail if the input array is not chronologically sorted' + ); +end; + +procedure TTestDataArray.TestBulkAddChronologicalAssertion_External; +var + newData: TArray>; + olderTime: TDateTime; +begin + SetupSeriesWithData(10); // Newest is baseTime + 9 + olderTime := EncodeDate(2020, 7, 7) + 8; + + SetLength(newData, 1); + newData[0] := TDataPoint.Create(olderTime, TAskBidItem.Create(1, 1)); + + Assert.WillRaise( + procedure begin FArray.Add(newData); end, + EAssertionFailed, + 'Bulk Add should fail if its first item is older than the series last item' + ); +end; + +procedure TTestDataArray.TestBulkAddSpanningMultipleChunks; +const + CHUNK_SIZE = 1024; +var + newData: TArray>; + lastTimeBeforeAdd, firstNewTime, lastNewTime: TDateTime; + i: Integer; +begin + SetupSeriesWithData(CHUNK_SIZE - 4); // Almost fill the first chunk + lastTimeBeforeAdd := FArray[0].Time; + + SetLength(newData, 8); // Add 8 items, which will span the chunk boundary + for i := 0 to High(newData) do + begin + newData[i] := TDataPoint.Create(lastTimeBeforeAdd + 1 + i, TAskBidItem.Create(i, i)); + end; + + firstNewTime := newData[0].Time; + lastNewTime := newData[High(newData)].Time; + + FArray.Add(newData); + Assert.AreEqual(Int64(CHUNK_SIZE + 4), FArray.Count, 'Count should be correct after spanning a chunk'); + Assert.AreEqual(lastNewTime, FArray[0].Time, 'Newest item should be correct'); + Assert.AreEqual(firstNewTime, FArray[7].Time, 'Oldest of the new items should be at the correct logical index'); +end; + +procedure TTestDataArray.TestBulkAddWithMaxLookback; +var + newData: TArray>; + i: Integer; + baseTime: TDateTime; +begin + SetupSeriesWithData(400, 500); + baseTime := FArray[0].Time; + + SetLength(newData, 200); + for i := 0 to High(newData) do + begin + newData[i] := TDataPoint.Create(baseTime + 1 + i, TAskBidItem.Create(i, i)); + end; + + FArray.Add(newData); + + // Total items would be 600, but MaxLookback is 500. + // Trim does not free chunks (100 to remove < 1024). + // The public Count must be capped at 500. + Assert.AreEqual(Int64(500), FArray.Count, 'Count should be capped by MaxLookback after bulk add'); +end; + +procedure TTestDataArray.TestTotalCountIncrementsCorrectly; +var + newData: TArray>; + i: Integer; +begin + Assert.AreEqual(Int64(0), FArray.TotalCount, 'TotalCount should be 0 on creation'); + + // Test single add + FArray.Add(TDataPoint.Create(Now, TAskBidItem.Create(1, 1))); + Assert.AreEqual(Int64(1), FArray.TotalCount, 'TotalCount should be 1 after single add'); + + // Test bulk add + SetLength(newData, 10); + for i := 0 to High(newData) do + newData[i] := TDataPoint.Create(Now + 1 + i, TAskBidItem.Create(i, i)); + + FArray.Add(newData); + Assert.AreEqual(Int64(11), FArray.TotalCount, 'TotalCount should be 11 after bulk add'); +end; + +procedure TTestDataArray.TestTotalCountIgnoresTrimming; +begin + // Create series with a lookback that will cause trimming + SetupSeriesWithData(20, 10); + + // The visible count is capped by MaxLookback + Assert.AreEqual(Int64(10), FArray.Count, 'Count should be capped by MaxLookback'); + // The total count should reflect all items that were ever added + Assert.AreEqual(Int64(20), FArray.TotalCount, 'TotalCount must not be affected by trimming'); +end; + +procedure TTestDataArray.TestTotalCountResetsOnClear; +begin + SetupSeriesWithData(15); + Assert.IsTrue(FArray.TotalCount > 0, 'Pre-condition: TotalCount should be greater than 0'); + + FArray.Clear; + + Assert.AreEqual(Int64(0), FArray.TotalCount, 'TotalCount should be 0 after Clear'); + Assert.AreEqual(Int64(0), FArray.Count, 'Count should be 0 after Clear'); +end; + +{ TTestDataSeries } + +procedure TTestDataSeries.Setup; +begin + FArray := TDataSeries.CreateArray(0); + FSeries.Create(FArray); +end; + +procedure TTestDataSeries.Teardown; +begin + FSeries := Default(TDataSeries); + FArray := nil; +end; + +procedure TTestDataSeries.SetupSeriesWithData(ItemCount: Integer = 10; MaxLookBack: Int64 = 0); +var + i: Integer; + DataPoint: TDataPoint; + baseTime: TDateTime; +begin + FArray := TDataSeries.CreateArray(MaxLookBack); + FSeries.Create(FArray); + baseTime := EncodeDate(2020, 7, 7); + + // Add data points with increasing timestamps. + for i := 0 to ItemCount - 1 do + begin + DataPoint := TDataPoint.Create(baseTime + i, TAskBidItem.Create(i * 1.0, i * 1.0 + 0.1)); + FArray.Add(DataPoint); + end; +end; + +procedure TTestDataSeries.TestIndexOfExistingTimeStamp; var ATimeStamp: TDateTime; expectedIndex: Int64; @@ -207,8 +446,7 @@ begin ); end; -// Tests finding the index for a timestamp that falls between two existing items. -procedure TTestTDataSeries.TestIndexOfNonExistingBetween; +procedure TTestDataSeries.TestIndexOfNonExistingBetween; var ATimeStamp: TDateTime; expectedIndex: Int64; @@ -224,8 +462,7 @@ begin ); end; -// Tests finding a timestamp that is older than any item in the series. -procedure TTestTDataSeries.TestIndexOfBeforeFirstItem; +procedure TTestDataSeries.TestIndexOfBeforeFirstItem; var ATimeStamp: TDateTime; baseTime: TDateTime; @@ -237,8 +474,7 @@ begin Assert.AreEqual(Int64(-1), FSeries.IndexOf(ATimeStamp), 'IndexOf for a timestamp before the oldest item should return -1'); end; -// Tests finding a timestamp that is newer than any item in the series. -procedure TTestTDataSeries.TestIndexOfAfterLastItem; +procedure TTestDataSeries.TestIndexOfAfterLastItem; var ATimeStamp: TDateTime; baseTime: TDateTime; @@ -254,8 +490,7 @@ begin ); end; -// Specifically tests finding the exact oldest item. -procedure TTestTDataSeries.TestIndexOfExactOldestItem; +procedure TTestDataSeries.TestIndexOfExactOldestItem; var ATimeStamp: TDateTime; begin @@ -265,8 +500,7 @@ begin Assert.AreEqual(Int64(9), FSeries.IndexOf(ATimeStamp), 'IndexOf for the exact oldest timestamp should return the last index'); end; -// Specifically tests finding the exact newest item. -procedure TTestTDataSeries.TestIndexOfExactNewestItem; +procedure TTestDataSeries.TestIndexOfExactNewestItem; var ATimeStamp: TDateTime; begin @@ -276,24 +510,22 @@ begin Assert.AreEqual(Int64(0), FSeries.IndexOf(ATimeStamp), 'IndexOf for the exact newest timestamp should return index 0'); end; -// Tests IndexOf on an empty series. -procedure TTestTDataSeries.TestIndexOfEmptySeries; +procedure TTestDataSeries.TestIndexOfEmptySeries; begin - FSeries := Default(TDataSeries); + // Use default setup with no data Assert.AreEqual(Int64(0), FSeries.Count); Assert.AreEqual(Int64(-1), FSeries.IndexOf(Now), 'IndexOf on empty series should return -1'); end; -// Tests IndexOf on a series with only one item. -procedure TTestTDataSeries.TestIndexOfSingleItemSeries; +procedure TTestDataSeries.TestIndexOfSingleItemSeries; var DataPoint: TDataPoint; testTime: TDateTime; begin - FSeries := Default(TDataSeries); + // Use default setup, but add one item testTime := EncodeDate(2025, 1, 1); DataPoint := TDataPoint.Create(testTime, TAskBidItem.Create(1.0, 2.0)); - FSeries.Add(DataPoint); + FArray.Add(DataPoint); Assert.AreEqual(Int64(1), FSeries.Count); Assert.AreEqual(Int64(0), FSeries.IndexOf(testTime), 'IndexOf for exact single item should be 0'); @@ -301,57 +533,7 @@ begin Assert.AreEqual(Int64(-1), FSeries.IndexOf(testTime - 1), 'IndexOf for time before single item should be -1'); end; -procedure TTestTDataSeries.TestCountIsCappedByMaxLookback_NoTrim; -begin - SetupSeriesWithData(500, 400); // Less than one chunk - - // The public Count must now be the MaxLookback value, even though - // no chunk was freed and FCount is still 500 internally. - Assert.AreEqual(Int64(400), FSeries.Count, 'Public count should be capped by MaxLookback even if no chunk is freed'); -end; - -procedure TTestTDataSeries.TestCountIsCappedByMaxLookback_WithTrim; -const - ITEMS_TO_ADD = 2000; - LOOKBACK_LIMIT = 900; -begin - SetupSeriesWithData(ITEMS_TO_ADD, LOOKBACK_LIMIT); - - // Internally, FCount will be 976 after one chunk is freed. - // The public Count should still be capped at the lookback limit. - Assert.AreEqual(Int64(LOOKBACK_LIMIT), FSeries.Count, 'Public count should be capped by MaxLookback after trimming'); -end; - -procedure TTestTDataSeries.TestLoopIsSafeWithMaxLookback; -var - dummy: TDataPoint; -begin - SetupSeriesWithData(500, 400); - Assert.AreEqual(Int64(400), FSeries.Count, 'Pre-condition: Count must be capped at 400'); - - // This test proves that a standard loop based on the public Count is safe - // and will not access an invalid index. - Assert.WillNotRaise( - procedure - begin - for var i := 0 to FSeries.Count - 1 do - begin - dummy := FSeries[i]; - end; - end, - nil, - 'Looping up to the public Count should not raise an exception' - ); - - // Also test the assert when going one item beyond the public count - Assert.WillRaise( - procedure begin dummy := FSeries[400]; end, - EAssertionFailed, - 'Accessing index at the limit of MaxLookback should raise an exception' - ); -end; - -procedure TTestTDataSeries.TestIndexOfRespectsMaxLookback; +procedure TTestDataSeries.TestIndexOfRespectsMaxLookback; var baseTime, searchTime: TDateTime; begin @@ -373,154 +555,7 @@ begin Assert.AreEqual(Int64(399), FSeries.IndexOf(searchTime), 'IndexOf should find the oldest valid item at the edge of MaxLookback'); end; -procedure TTestTDataSeries.TestBulkAddIncreasesCount; -var - newData: TArray>; - baseTime, newTime: TDateTime; - i: Integer; -begin - SetupSeriesWithData(10); - baseTime := EncodeDate(2020, 7, 7); - - SetLength(newData, 5); - for i := 0 to High(newData) do - begin - newTime := baseTime + 10 + i; - newData[i] := TDataPoint.Create(newTime, TAskBidItem.Create(i, i)); - end; - - FSeries.Add(newData); - - Assert.AreEqual(Int64(15), FSeries.Count, 'Count should be increased by the number of items in the bulk add'); - Assert.AreEqual(baseTime + 10 + High(newData), FSeries.Time[0], 'Newest item should be the last item from the added array'); -end; - -procedure TTestTDataSeries.TestBulkAddChronologicalAssertion_Internal; -var - newData: TArray>; -begin - SetupSeriesWithData(10); - SetLength(newData, 2); - newData[0] := TDataPoint.Create(Now + 1, TAskBidItem.Create(1, 1)); - newData[1] := TDataPoint.Create(Now, TAskBidItem.Create(2, 2)); // Not sorted - - Assert.WillRaise( - procedure begin FSeries.Add(newData); end, - EAssertionFailed, - 'Bulk Add should fail if the input array is not chronologically sorted' - ); -end; - -procedure TTestTDataSeries.TestBulkAddChronologicalAssertion_External; -var - newData: TArray>; - olderTime: TDateTime; -begin - SetupSeriesWithData(10); // Newest is baseTime + 9 - olderTime := EncodeDate(2020, 7, 7) + 8; - - SetLength(newData, 1); - newData[0] := TDataPoint.Create(olderTime, TAskBidItem.Create(1, 1)); - - Assert.WillRaise( - procedure begin FSeries.Add(newData); end, - EAssertionFailed, - 'Bulk Add should fail if its first item is older than the series last item' - ); -end; - -procedure TTestTDataSeries.TestBulkAddSpanningMultipleChunks; -const - CHUNK_SIZE = 1024; -var - newData: TArray>; - lastTimeBeforeAdd, firstNewTime, lastNewTime: TDateTime; - i: Integer; -begin - SetupSeriesWithData(CHUNK_SIZE - 4); // Almost fill the first chunk - lastTimeBeforeAdd := FSeries.Time[0]; - - SetLength(newData, 8); // Add 8 items, which will span the chunk boundary - for i := 0 to High(newData) do - begin - newData[i] := TDataPoint.Create(lastTimeBeforeAdd + 1 + i, TAskBidItem.Create(i, i)); - end; - - firstNewTime := newData[0].Time; - lastNewTime := newData[High(newData)].Time; - - FSeries.Add(newData); - Assert.AreEqual(Int64(CHUNK_SIZE + 4), FSeries.Count, 'Count should be correct after spanning a chunk'); - Assert.AreEqual(lastNewTime, FSeries.Time[0], 'Newest item should be correct'); - Assert.AreEqual(firstNewTime, FSeries.Time[7], 'Oldest of the new items should be at the correct logical index'); -end; - -procedure TTestTDataSeries.TestBulkAddWithMaxLookback; -var - newData: TArray>; - i: Integer; - baseTime: TDateTime; -begin - SetupSeriesWithData(400, 500); - baseTime := FSeries.Time[0]; - - SetLength(newData, 200); - for i := 0 to High(newData) do - begin - newData[i] := TDataPoint.Create(baseTime + 1 + i, TAskBidItem.Create(i, i)); - end; - - FSeries.Add(newData); - - // Total items would be 600, but MaxLookback is 500. - // Trim does not free chunks (100 to remove < 1024). - // The public Count must be capped at 500. - Assert.AreEqual(Int64(500), FSeries.Count, 'Count should be capped by MaxLookback after bulk add'); -end; - -procedure TTestTDataSeries.TestTotalCountIncrementsCorrectly; -var - newData: TArray>; - i: Integer; -begin - FSeries := TDataSeries.Create( 0 ); - Assert.AreEqual(Int64(0), FSeries.TotalCount, 'TotalCount should be 0 on creation'); - - // Test single add - FSeries.Add(TDataPoint.Create(Now, TAskBidItem.Create(1,1))); - Assert.AreEqual(Int64(1), FSeries.TotalCount, 'TotalCount should be 1 after single add'); - - // Test bulk add - SetLength(newData, 10); - for i := 0 to High(newData) do - newData[i] := TDataPoint.Create(Now + 1 + i, TAskBidItem.Create(i,i)); - - FSeries.Add(newData); - Assert.AreEqual(Int64(11), FSeries.TotalCount, 'TotalCount should be 11 after bulk add'); -end; - -procedure TTestTDataSeries.TestTotalCountIgnoresTrimming; -begin - // Create series with a lookback that will cause trimming - SetupSeriesWithData(20, 10); - - // The visible count is capped by MaxLookback - Assert.AreEqual(Int64(10), FSeries.Count, 'Count should be capped by MaxLookback'); - // The total count should reflect all items that were ever added - Assert.AreEqual(Int64(20), FSeries.TotalCount, 'TotalCount must not be affected by trimming'); -end; - -procedure TTestTDataSeries.TestTotalCountResetsOnClear; -begin - SetupSeriesWithData(15); - Assert.IsTrue(FSeries.TotalCount > 0, 'Pre-condition: TotalCount should be greater than 0'); - - FSeries.Clear; - - Assert.AreEqual(Int64(0), FSeries.TotalCount, 'TotalCount should be 0 after Clear'); - Assert.AreEqual(Int64(0), FSeries.Count, 'Count should be 0 after Clear'); -end; - initialization - TDUnitX.RegisterTestFixture(TTestTDataSeries); + TDUnitX.RegisterTestFixture(TTestDataArray); + TDUnitX.RegisterTestFixture(TTestDataSeries); end. diff --git a/Src/Myc.Trade.Core.DataPoint.pas b/Src/Myc.Trade.Core.DataPoint.pas new file mode 100644 index 0000000..69aaab7 --- /dev/null +++ b/Src/Myc.Trade.Core.DataPoint.pas @@ -0,0 +1,226 @@ +unit Myc.Trade.Core.DataPoint; + +interface + +uses + Myc.Trade.DataPoint; + +type + // The implementation class for IDataSeries. + TDataArray = class(TInterfacedObject, IDataSeries, IDataArray) + private + const + ChunkSize = 1024; + type + TChunk = TArray>; + private + FChunks: TArray; + FCount: Int64; + FMaxLookback: Int64; + FTotalCount: Int64; + + function LogicalToPhysicalIndex(LogicalIndex: Int64): Int64; inline; + function IsIndexInLimits(LogicalIndex: Int64): Boolean; + procedure Trim; + + // IDataSeries implementation + function GetCount: Int64; + function GetItems(Idx: Int64): TDataPoint; + function GetMaxLookback: Int64; + function GetTotalCount: Int64; + public + constructor Create(AMaxLookback: Int64); + procedure Add(const Data: TDataPoint); overload; + procedure Add(const Data: array of TDataPoint; NumToAdd: Integer = -1); overload; + procedure Clear; + property MaxLookback: Int64 read GetMaxLookback; + end; + +type + // Null object implementation for IDataSeries + TNullDataSeries = class(TInterfacedObject, IDataSeries) + public + function GetCount: Int64; + function GetItems(Idx: Int64): TDataPoint; + function GetTotalCount: Int64; + function IndexOf(TimeStamp: TDateTime): Int64; + end; + +implementation + +{ TDataArray } + +constructor TDataArray.Create(AMaxLookback: Int64); +begin + inherited Create; + FMaxLookback := AMaxLookback; + FCount := 0; + FTotalCount := 0; + FChunks := nil; +end; + +procedure TDataArray.Add(const Data: TDataPoint); +begin + Assert((FCount = 0) or (Data.Time >= GetItems(0).Time), 'Time stamp older than last item'); + + var ci := FCount div ChunkSize; + var di := FCount mod ChunkSize; + + if (di = 0) then + begin + SetLength(FChunks, ci + 1); + SetLength(FChunks[ci], ChunkSize); + end; + + FChunks[ci][di] := Data; + Inc(FCount); + Inc(FTotalCount); + + Trim; +end; + +procedure TDataArray.Add(const Data: array of TDataPoint; NumToAdd: Integer = -1); +var + sourceIdx, itemsToCopy, spaceInChunk: Integer; + ci, di: Int64; + i: Integer; + resolvedNumToAdd: Integer; +begin + if NumToAdd < 0 then + resolvedNumToAdd := Length(Data) + else + resolvedNumToAdd := NumToAdd; + + if resolvedNumToAdd = 0 then + Exit; + + Assert(resolvedNumToAdd <= Length(Data), 'NumToAdd cannot be larger than the source array'); + + for i := 1 to resolvedNumToAdd - 1 do + Assert(Data[i].Time >= Data[i - 1].Time, 'Input array for Add is not chronologically sorted'); + + Assert((FCount = 0) or (Data[0].Time >= GetItems(0).Time), 'First new item is older than last existing item'); + + Inc(FTotalCount, resolvedNumToAdd); + + var newTotalCount := FCount + resolvedNumToAdd; + var requiredChunks := (newTotalCount + ChunkSize - 1) div ChunkSize; + if requiredChunks = 0 then + requiredChunks := 1; + + if requiredChunks > Length(FChunks) then + SetLength(FChunks, requiredChunks); + + sourceIdx := 0; + while sourceIdx < resolvedNumToAdd do + begin + ci := FCount div ChunkSize; + di := FCount mod ChunkSize; + + if di = 0 then + SetLength(FChunks[ci], ChunkSize); + + spaceInChunk := ChunkSize - di; + itemsToCopy := resolvedNumToAdd - sourceIdx; + if spaceInChunk < itemsToCopy then + itemsToCopy := spaceInChunk; + + System.Move(Data[sourceIdx], FChunks[ci][di], itemsToCopy * SizeOf(TDataPoint)); + + Inc(FCount, itemsToCopy); + Inc(sourceIdx, itemsToCopy); + + Trim; + end; +end; + +procedure TDataArray.Clear; +begin + FChunks := nil; + FCount := 0; + FTotalCount := 0; +end; + +function TDataArray.GetCount: Int64; +begin + Result := FCount; + if (FMaxLookback > 0) and (Result > FMaxLookback) then + Result := FMaxLookback; +end; + +function TDataArray.GetItems(Idx: Int64): TDataPoint; +var + physicalIndex: Int64; +begin + Assert(IsIndexInLimits(Idx), 'Index is outside of the configured Lookback limits.'); + physicalIndex := LogicalToPhysicalIndex(Idx); + Result := FChunks[physicalIndex div ChunkSize][physicalIndex mod ChunkSize]; +end; + +function TDataArray.GetMaxLookback: Int64; +begin + Result := FMaxLookback; +end; + +function TDataArray.GetTotalCount: Int64; +begin + Result := FTotalCount; +end; + +function TDataArray.IsIndexInLimits(LogicalIndex: Int64): Boolean; +begin + Result := (FMaxLookback <= 0) or (LogicalIndex < FMaxLookback); +end; + +function TDataArray.LogicalToPhysicalIndex(LogicalIndex: Int64): Int64; +begin + Assert((LogicalIndex >= 0) and (LogicalIndex < FCount), 'Logical index is out of bounds.'); + Result := FCount - LogicalIndex - 1; +end; + +procedure TDataArray.Trim; +var + itemsToRemove, chunksToRemove: Int64; +begin + if (FCount = 0) or (FMaxLookback <= 0) then + Exit; + + itemsToRemove := 0; + if FCount > FMaxLookback then + itemsToRemove := FCount - FMaxLookback; + + if itemsToRemove <= 0 then + Exit; + + chunksToRemove := itemsToRemove div ChunkSize; + + if chunksToRemove > 0 then + begin + var itemsInFreedChunks := chunksToRemove * ChunkSize; + FChunks := Copy(FChunks, chunksToRemove, Length(FChunks) - chunksToRemove); + FCount := FCount - itemsInFreedChunks; + end; +end; + +function TNullDataSeries.GetCount: Int64; +begin + Result := 0; +end; + +function TNullDataSeries.GetItems(Idx: Int64): TDataPoint; +begin + Assert(false, 'Index out of bounds.'); + Result := Default(TDataPoint); +end; + +function TNullDataSeries.GetTotalCount: Int64; +begin + Result := 0; +end; + +function TNullDataSeries.IndexOf(TimeStamp: TDateTime): Int64; +begin + Result := -1; +end; + +end. diff --git a/Src/Myc.Trade.DataPoint.pas b/Src/Myc.Trade.DataPoint.pas index ddaf284..932a389 100644 --- a/Src/Myc.Trade.DataPoint.pas +++ b/Src/Myc.Trade.DataPoint.pas @@ -2,9 +2,6 @@ unit Myc.Trade.DataPoint; interface -uses - System.SysUtils; - type // A data record for an Ask/Bid price pair. TAskBidItem = packed record @@ -22,60 +19,71 @@ type // A time-ordered series of data points, optimized for chronological additions. // The most recently added element has the logical index 0. - TDataSeries = record - private - const - ChunkSize = 1024; - type - TChunk = TArray>; - private - FChunks: TArray; - FCount: Int64; - FMaxLookback: Int64; - FTotalCount: Int64; - - // Converts a logical index (0=newest) to a physical storage index (0=oldest). - function LogicalToPhysicalIndex(LogicalIndex: Int64): Int64; inline; + IDataSeries = interface function GetCount: Int64; - function GetData(Idx: Int64): T; function GetItems(Idx: Int64): TDataPoint; - function GetTime(Idx: Int64): TDateTime; - procedure SetData(Idx: Int64; const Value: T); - // Checks if a logical index is within the defined limits. - function IsIndexInLimits(LogicalIndex: Int64): Boolean; - // Removes old chunks that are outside the Lookback limits. - procedure Trim; - public - constructor Create(AMaxLookback: Int64); - // Adds a new data point to the series. + function GetTotalCount: Int64; + property Count: Int64 read GetCount; + // Accesses data points by their logical index. + // Index 0 is the newest element, Index (Count - 1) is the oldest. + property Items[Idx: Int64]: TDataPoint read GetItems; default; + // The total number of items ever added to the series. + property TotalCount: Int64 read GetTotalCount; + end; + + IDataArray = interface(IDataSeries) procedure Add(const Data: TDataPoint); overload; procedure Add(const Data: array of TDataPoint; NumToAdd: Integer = -1); overload; - // Clears all data from the series. procedure Clear; + end; + + // Interface Helper for IDataSeries. + // Provides a safe, value-type-like wrapper around the interface. + TDataSeries = record + strict private + class var + FNull: IDataSeries; + private + FDataSeries: IDataSeries; + FMaxLookback: Int64; + class constructor CreateClass; + function GetCount: Int64; + function GetItems(Idx: Int64): TDataPoint; + function GetData(Idx: Int64): T; + function GetTime(Idx: Int64): TDateTime; + function GetMaxLookback: Int64; + function GetTotalCount: Int64; + public + constructor Create(ADataSeries: IDataSeries); + + class operator Initialize(out Dest: TDataSeries); + class operator Finalize(var Dest: TDataSeries); + + class operator Implicit(const A: TDataSeries): IDataSeries; + class operator Implicit(const A: IDataSeries): TDataSeries; + + class function CreateArray(MaxLookback: Int64): IDataArray; static; + // Searches for a data point by its timestamp. // Returns the logical index of the matching item. // If no exact match, returns the index of the item immediately preceding the timestamp. // Returns -1 if the timestamp is before the oldest item in the series. function IndexOf(TimeStamp: TDateTime): Int64; - class operator Initialize(out Dest: TDataSeries); - class operator Finalize(var Dest: TDataSeries); - + class property Null: IDataSeries read FNull; property Count: Int64 read GetCount; - // Accesses data points by their logical index. - // Index 0 is the newest element, Index (Count - 1) is the oldest. property Items[Idx: Int64]: TDataPoint read GetItems; default; + property Data[Idx: Int64]: T read GetData; property Time[Idx: Int64]: TDateTime read GetTime; - property Data[Idx: Int64]: T read GetData write SetData; - - // The maximum number of data points to keep (0 = unlimited). - property MaxLookback: Int64 read FMaxLookback; - // The total number of items ever added to the series (resets on Clear). - property TotalCount: Int64 read FTotalCount; + property MaxLookback: Int64 read GetMaxLookback; + property TotalCount: Int64 read GetTotalCount; end; implementation +uses + Myc.Trade.Core.DataPoint; + { TAskBidItem } constructor TAskBidItem.Create(AAsk, ABid: Single); @@ -92,153 +100,74 @@ begin Data := AData; end; -constructor TDataSeries.Create(AMaxLookback: Int64); +{ TDataSeries Interface Helper } + +class constructor TDataSeries.CreateClass; begin - FMaxLookback := AMaxLookback; - FCount := 0; - FTotalCount := 0; - FChunks := nil; + FNull := TNullDataSeries.Create; end; -{ TDataSeries } - -procedure TDataSeries.Add(const Data: TDataPoint); +constructor TDataSeries.Create(ADataSeries: IDataSeries); begin - // Enforce chronological order for new items. - Assert((FCount = 0) or (Data.Time >= GetTime(0)), 'Time stamp older than last item'); - - var ci := FCount div ChunkSize; - var di := FCount mod ChunkSize; - - if (di = 0) then - begin - SetLength(FChunks, ci + 1); - SetLength(FChunks[ci], ChunkSize); - end; - - FChunks[ci][di] := Data; - Inc(FCount); - Inc(FTotalCount); - - // Remove old data that is now out of limits. - Trim; -end; - -procedure TDataSeries.Add(const Data: array of TDataPoint; NumToAdd: Integer = -1); -var - sourceIdx, itemsToCopy, spaceInChunk: Integer; - ci, di: Int64; - i: Integer; - resolvedNumToAdd: Integer; -begin - if NumToAdd < 0 then - resolvedNumToAdd := Length(Data) + if Assigned(ADataSeries) then + FDataSeries := ADataSeries else - resolvedNumToAdd := NumToAdd; - - if resolvedNumToAdd = 0 then - Exit; - - Assert(resolvedNumToAdd <= Length(Data), 'NumToAdd cannot be larger than the source array'); - - // Check chronological order of the input array itself - for i := 1 to resolvedNumToAdd - 1 do - Assert(Data[i].Time >= Data[i - 1].Time, 'Input array for Add is not chronologically sorted'); - - // Enforce chronological order for new items against existing series. - Assert((FCount = 0) or (Data[0].Time >= GetTime(0)), 'First new item is older than last existing item'); - - // Increment total count immediately. - Inc(FTotalCount, resolvedNumToAdd); - - var newTotalCount := FCount + resolvedNumToAdd; - var requiredChunks := (newTotalCount + ChunkSize - 1) div ChunkSize; - if requiredChunks = 0 then // case for newTotalCount = 0 - requiredChunks := 1; - - // Pre-allocate chunk array structure - if requiredChunks > Length(FChunks) then - SetLength(FChunks, requiredChunks); - - sourceIdx := 0; - while sourceIdx < resolvedNumToAdd do - begin - ci := FCount div ChunkSize; - di := FCount mod ChunkSize; - - // Allocate the actual chunk if we are at its beginning - if di = 0 then - SetLength(FChunks[ci], ChunkSize); - - spaceInChunk := ChunkSize - di; - itemsToCopy := resolvedNumToAdd - sourceIdx; - if spaceInChunk < itemsToCopy then - itemsToCopy := spaceInChunk; - - // Move a block of data into the available space in the current chunk. - System.Move(Data[sourceIdx], FChunks[ci][di], itemsToCopy * SizeOf(TDataPoint)); - - Inc(FCount, itemsToCopy); - Inc(sourceIdx, itemsToCopy); - - // Remove old data that is now out of limits - Trim; - end; + FDataSeries := FNull; end; -procedure TDataSeries.Clear; +class function TDataSeries.CreateArray(MaxLookback: Int64): IDataArray; begin - FChunks := nil; - FCount := 0; - FTotalCount := 0; + Result := TDataArray.Create(MaxLookback); end; class operator TDataSeries.Finalize(var Dest: TDataSeries); begin - Dest.Clear; -end; - -function TDataSeries.GetCount: Int64; -begin - // The public Count must always reflect the addressable range. - Result := FCount; - if (FMaxLookback > 0) and (Result > FMaxLookback) then - Result := FMaxLookback; -end; - -function TDataSeries.GetData(Idx: Int64): T; -var - physicalIndex: Int64; -begin - Assert(IsIndexInLimits(Idx), 'Index is outside of the configured Lookback limits.'); - physicalIndex := LogicalToPhysicalIndex(Idx); - Result := FChunks[physicalIndex div ChunkSize][physicalIndex mod ChunkSize].Data; -end; - -function TDataSeries.GetItems(Idx: Int64): TDataPoint; -var - physicalIndex: Int64; -begin - Assert(IsIndexInLimits(Idx), 'Index is outside of the configured Lookback limits.'); - physicalIndex := LogicalToPhysicalIndex(Idx); - Result := FChunks[physicalIndex div ChunkSize][physicalIndex mod ChunkSize]; -end; - -function TDataSeries.GetTime(Idx: Int64): TDateTime; -var - physicalIndex: Int64; -begin - Assert(IsIndexInLimits(Idx), 'Index is outside of the configured Lookback limits.'); - physicalIndex := LogicalToPhysicalIndex(Idx); - Result := FChunks[physicalIndex div ChunkSize][physicalIndex mod ChunkSize].Time; + Dest.FDataSeries := nil; end; class operator TDataSeries.Initialize(out Dest: TDataSeries); begin - Dest.FCount := 0; - Dest.FTotalCount := 0; - Dest.FChunks := nil; - Dest.FMaxLookback := 0; + Dest.FDataSeries := FNull; +end; + +class operator TDataSeries.Implicit(const A: TDataSeries): IDataSeries; +begin + Result := A.FDataSeries; +end; + +class operator TDataSeries.Implicit(const A: IDataSeries): TDataSeries; +begin + Result.Create(A); +end; + +function TDataSeries.GetCount: Int64; +begin + Result := FDataSeries.GetCount; +end; + +function TDataSeries.GetItems(Idx: Int64): TDataPoint; +begin + Result := FDataSeries[Idx]; +end; + +function TDataSeries.GetData(Idx: Int64): T; +begin + Result := FDataSeries[Idx].Data; +end; + +function TDataSeries.GetTime(Idx: Int64): TDateTime; +begin + Result := FDataSeries[Idx].Time; +end; + +function TDataSeries.GetMaxLookback: Int64; +begin + Result := FMaxLookback; +end; + +function TDataSeries.GetTotalCount: Int64; +begin + Result := FDataSeries.GetTotalCount; end; function TDataSeries.IndexOf(TimeStamp: TDateTime): Int64; @@ -247,16 +176,16 @@ var dataPointTime: TDateTime; begin Result := -1; - if Count = 0 then // Use public Count to respect limits + if Count = 0 then Exit; low := 0; - high := Count - 1; // Use public Count to respect limits + high := Count - 1; while (low <= high) do begin mid := low + (high - low) div 2; - dataPointTime := GetTime(mid); + dataPointTime := GetItems(mid).Time; // Use GetItems to stay within the class if (dataPointTime = TimeStamp) then begin @@ -268,59 +197,11 @@ begin Result := mid; high := mid - 1; end - else // dataPointTime > TimeStamp + else begin low := mid + 1; end; end; end; -function TDataSeries.IsIndexInLimits(LogicalIndex: Int64): Boolean; -begin - // Check Lookback limit - Result := (FMaxLookback <= 0) or (LogicalIndex < FMaxLookback); -end; - -function TDataSeries.LogicalToPhysicalIndex(LogicalIndex: Int64): Int64; -begin - Assert((LogicalIndex >= 0) and (LogicalIndex < Count), 'Logical index is out of bounds.'); - Result := FCount - LogicalIndex - 1; -end; - -procedure TDataSeries.SetData(Idx: Int64; const Value: T); -var - physicalIndex: Int64; -begin - Assert(IsIndexInLimits(Idx), 'Index is outside of the configured Lookback limits.'); - physicalIndex := LogicalToPhysicalIndex(Idx); - FChunks[physicalIndex div ChunkSize][physicalIndex mod ChunkSize].Data := Value; -end; - -procedure TDataSeries.Trim; -var - itemsToRemove, chunksToRemove: Int64; -begin - if (FCount = 0) or (FMaxLookback <= 0) then - Exit; - - itemsToRemove := 0; - if FCount > FMaxLookback then - begin - itemsToRemove := FCount - FMaxLookback; - end; - - if itemsToRemove <= 0 then - Exit; - - // We only free full chunks to avoid expensive partial copy operations. - chunksToRemove := itemsToRemove div ChunkSize; - - if chunksToRemove > 0 then - begin - var itemsInFreedChunks := chunksToRemove * ChunkSize; - FChunks := Copy(FChunks, chunksToRemove, Length(FChunks) - chunksToRemove); - FCount := FCount - itemsInFreedChunks; - end; -end; - end. diff --git a/Src/Myc.Trade.DataProvider.pas b/Src/Myc.Trade.DataProvider.pas index 829ba81..4fe60dc 100644 --- a/Src/Myc.Trade.DataProvider.pas +++ b/Src/Myc.Trade.DataProvider.pas @@ -12,37 +12,37 @@ type TDataStreamProvider = class(TInterfacedObject, TMutable>.IMutable) private FStream: IDataStream; - FDataSeries: TDataSeries; + FDataArray: IDataArray; FNewDataAvailable: TFlag; - FHasDataSubscription: TSignal.TSubscription; + FChunk: TArray>; FLookback: Int64; function GetChanged: TSignal; function GetValue: TDataSeries; public - constructor Create(ALookback: Int64; const AStream: IDataStream); + constructor Create(ALookback, AMaxChunkSize: Int64; const AStream: IDataStream); destructor Destroy; override; end; TDataStreamProvider = record - class function Create(Lookback: Int64; const Stream: IDataStream): TMutable>; static; + class function Create(Lookback, MaxChunkSize: Int64; const Stream: IDataStream): TMutable>; static; end; implementation { TDataStreamProvider } -constructor TDataStreamProvider.Create(ALookback: Int64; const AStream: IDataStream); +constructor TDataStreamProvider.Create(ALookback, AMaxChunkSize: Int64; const AStream: IDataStream); begin inherited Create; FLookback := ALookback; + SetLength(FChunk, AMaxChunkSize); FStream := AStream; - FHasDataSubscription := FStream.HasData.Subscribe(FNewDataAvailable); - FDataSeries := TDataSeries.Create( ALookback ); + FNewDataAvailable := TFlag.CreateObserver(FStream.HasData); + FDataArray := TDataSeries.CreateArray(ALookback); end; destructor TDataStreamProvider.Destroy; begin - FHasDataSubscription.Unsubscribe; inherited; end; @@ -55,22 +55,18 @@ function TDataStreamProvider.GetValue: TDataSeries; begin if FNewDataAvailable.Reset then begin - var Data: array[0..1024 - 1] of TDataPoint; - var cnt := FStream.GetChunk(Data); - while cnt > 0 do - begin - FDataSeries.Add(Data, cnt); - cnt := FStream.GetChunk(Data); - end; + var cnt := FStream.GetChunk(FChunk); + if cnt > 0 then + FDataArray.Add(FChunk, cnt); end; - Result := FDataSeries; + Result := FDataArray; end; { TDataStreamProvider } -class function TDataStreamProvider.Create(Lookback: Int64; const Stream: IDataStream): TMutable>; +class function TDataStreamProvider.Create(Lookback, MaxChunkSize: Int64; const Stream: IDataStream): TMutable>; begin - Result := TDataStreamProvider.Create(Lookback, Stream); + Result := TDataStreamProvider.Create(Lookback, MaxChunkSize, Stream); end; end. diff --git a/Src/Myc.Trade.DataStream.pas b/Src/Myc.Trade.DataStream.pas index 5657a7f..4da9488 100644 --- a/Src/Myc.Trade.DataStream.pas +++ b/Src/Myc.Trade.DataStream.pas @@ -145,6 +145,7 @@ type FDataServer: TAuraDataServer; FSymbol: String; FHasData: TEvent; + FNextReady: TFlag; FCurrent: TFuture; FNext: TFuture; FCurrPosInFile: Int64; @@ -166,13 +167,13 @@ type TAuraTABFileServer = class(TAuraDataServer) protected - function ParseFileName(const FileName: string): TAuraDataFile; override; // Scans a directory and returns the oldest file found for each symbol. class function ReadCompressedData(const InputStream: TStream): TArray>; override; class function ReadUncompressedData(const InputStream: TStream): TArray>; override; public function CreateStream(const Symbol: String): IDataStream; override; function LoadDataSeries(const InitialFile: TAuraDataFile): TFuture>>; + function ParseFileName(const FileName: string): TAuraDataFile; override; end; implementation @@ -363,9 +364,6 @@ begin end; function TAuraDataServer.EnumerateSymbols: TFuture>; -var - dataFiles: TArray; - i: Integer; begin Result := FSymbols.Chain>( @@ -404,7 +402,8 @@ begin Assert(Assigned(ADataServer)); FDataServer := ADataServer; FSymbol := ASymbol; - FHasData := TEvent.CreateEvent; + FNextReady := TFlag.CreateFlag; + FHasData := TEvent.CreateRouter(FNextReady.State.AsSignal); end; destructor TAuraFileStream.Destroy; @@ -429,7 +428,7 @@ begin begin Result.FileInfo := FileInfo; Result.Data := FDataServer.LoadDataFile(FileInfo); - Result.Data.Done.Subscribe(FHasData); + Result.Data.Done.Subscribe(FNextReady); end; end ); @@ -488,7 +487,7 @@ begin Inc(FCurrPosInFile); end; - if FCurrPosInFile < Length(currData) then + if (FCurrPosInFile < Length(currData)) or FNextReady.Reset then begin FHasData.Notify; end; @@ -511,6 +510,8 @@ end; procedure TAuraFileStream.PreloadNext; begin + FNextReady.Reset; + FNext := FCurrent.Chain( function(const Prev: TDataStream): TDataStream @@ -524,7 +525,7 @@ begin begin Result.FileInfo := nextFileInfo; Result.Data := FDataServer.LoadDataFile(nextFileInfo); - Result.Data.Done.Subscribe(FHasData); + Result.Data.Done.Subscribe(FNextReady); end; end; end diff --git a/Test/MycTests.res b/Test/MycTests.res index e30ea808a0625e531bc6e2d44971efe04e66a16e..333684ad278e9b20d4042824f2f96e983b6e3a11 100644 GIT binary patch delta 18 Zcmcb!jPuSi&W0AoEld%?EKCdx3;;+S1%m(p delta 18 Zcmcb!jPuSi&W0AoEld%?EX)iH3;;+W1%v