MycLib/Src/Data/Myc.Data.Stream.Pipes.pas

(*TODO
Ich möchte ein neues Sprachkonzept einbauen. Folgendes Skript skizziert, was möglich sein soll:

; Record-Streams sind Record-Series, an die "Pipes" andocken können (Observer) um neue Daten zu bekommen
; Es handelt sich um reaktives Producer-Consumer-Pattern

;; Beispiel: SMA auf Close
(def btc (create-ticker "BTCUSD")) ; ein Open-High-Low-Close-Record-Stream
(def sma (create-sma 20)) ; erzeugt eine Lambda (sma(price)), die den sma20 berechnet


; Das Ergebnis eines pipe-Befehls ist wieder ein Record-Stream.
; btc-sma ist also ein Record-Stream, der {:ma ..} records enthält:
(def btc-sma 
  (pipe [btc [:Close]]
    (fn [price] 
      ; pipe mappt die Elemente des Record-Streams (hier :Close) auf die Parameter der fn. Price ist als eine ScalarSeries.
      ; (Index=0 ist das neueste Element)
      ; emit ist eine von pipe in den lambda-scope injizierte Funktion, die das neue Element in den Ergebnis-Stream 
      ; pusht (und damit die Observer des Ergebnis-Streams benachrichtigt).
      (emit {:ma (sma (get price 0))})
    )
  )
)

; Diese pipe hat zwei Streams als Input: den Closing-Price und den SMA. Jede pipe kann also N Input Stream haben
(def btc-signal
  (pipe [btc [:Close] btc-sma [:ma]]
    (fn [price ma]
	(emit {:signal (? (> (get price 0) (get ma 0)) :buy : sell)})
    )
  )
)
*)

unit Myc.Data.Stream.Pipes;

interface

uses
    System.SysUtils,
    System.Classes,
    System.Generics.Collections,
    System.SyncObjs,
    Myc.Data.Scalar,
    Myc.Data.Keyword,
    Myc.Data.Stream,
    Myc.Core.Notifier;

type
    // Forward declaration
    TPipeStream = class;

    // mapped fields per input stream
    TPipeConfig = TArray<TArray<TScalarRecordField>>;

    TPipeSource = class(TInterfacedObject, IStreamObserver)
    private
        [Weak]
        FOwner: TPipeStream;
        FSource: IStream;
        FTag: TSubscriptionTag;

        // State for barrier synchronization
        FLastSeenCycle: Int64;

        procedure OnSignal(const Signal: TStreamSignal);
        procedure Destroying;

    public
        constructor Create(AOwner: TPipeStream; ASource: IStream; const AInputs: TArray<TScalarRecordField>);
        property LastSeenCycle: Int64 read FLastSeenCycle;
    end;

    // --- Runtime: The Pipe ---
    // The executable instance of a pipe.
    // Acts as IStream (Producer) for downstream consumers and manages internal inputs.
    TPipeStream = class(TInterfacedObject, IStream)
        type
            TProc = reference to procedure(const Series: TArray<ISeries>);
    private
        FConfig: TPipeConfig;
        FSources: TArray<TPipeSource>;
        FObservers: TMycNotifyList<IStreamObserver>;
        FSeries: IWriteableScalarRecordSeries;

        // Barrier State
        FLastFiredCycleID: Int64;

        // Execution Logic
        // The compiled lambda function representing (fn [inputs...] ...)
        FLambda: TProc;
        FLambdaArgs: TArray<ISeries>;

        procedure CheckBarrierAndFire(CurrentCycle: Int64);
        function GetSeries: IScalarRecordSeries;

    public
        constructor Create(const AConfig: TPipeConfig; const ASources: TArray<IStream>; const ALambda: TProc);
        destructor Destroy; override;

        // Injected into lambda scope as "emit"
        procedure Emit(const Value: IScalarRecord);

        // IStream Implementation
        function Subscribe(const Observer: IStreamObserver): TSubscriptionTag;
        procedure Unsubscribe(Tag: TSubscriptionTag);
    end;

implementation

{ TPipeSource }

constructor TPipeSource.Create(AOwner: TPipeStream; ASource: IStream; const AInputs: TArray<TScalarRecordField>);
begin
    inherited Create;

    FOwner := AOwner;
    FSource := ASource;
    FLastSeenCycle := -1;

    FTag := FSource.Subscribe(Self);
end;

procedure TPipeSource.Destroying;
begin
    FSource.Unsubscribe(FTag);
    FOwner := nil;
end;

procedure TPipeSource.OnSignal(const Signal: TStreamSignal);
begin
    FLastSeenCycle := Signal.CycleID;

    if Assigned(FOwner) then
        FOwner.CheckBarrierAndFire(FLastSeenCycle);
end;

{ TPipeStream }

constructor TPipeStream.Create(const AConfig: TPipeConfig; const ASources: TArray<IStream>; const ALambda: TProc);
var
    fields: TArray<TScalarRecordField>;
begin
    inherited Create;
    FConfig := AConfig;
    FLambda := ALambda;
    FLastFiredCycleID := -1;

    Assert(Length(AConfig) = Length(ASources));

    SetLength(FSources, Length(ASources));

    // Construct record series definition from source config
    var n := 0;
    for var i := 0 to High(AConfig) do
    begin
        FSources[i] := TPipeSource.Create(Self, ASources[i], AConfig[i]);
        FSources[i]._AddRef;
        inc(n, Length(AConfig[i]));
    end;

    SetLength(FLambdaArgs, n);

    n := 0;
    for var i := 0 to High(AConfig) do
    begin
        for var j := 0 to High(AConfig[i]) do
        begin
            fields[n] := AConfig[i][j];

            FLambdaArgs[n] := ASources[i].Series[j].Value;

            inc(n);
        end;
    end;

    FSeries := TScalarRecordSeries.Create(TScalarRecord.TRegistry.Intern(fields));
end;

destructor TPipeStream.Destroy;
begin
    for var src in FSources do
    begin
        src.Destroying;
        src._Release;
    end;
    FObservers.Finalize;
    inherited;
end;

function TPipeStream.Subscribe(const Observer: IStreamObserver): TSubscriptionTag;
begin
    FObservers.Lock;
    try
        Result := FObservers.Advise(Observer);
    finally
        FObservers.Release;
    end;
end;

procedure TPipeStream.Unsubscribe(Tag: TSubscriptionTag);
begin
    FObservers.Lock;
    try
        FObservers.Unadvise(Tag);
    finally
        FObservers.Release;
    end;
end;

procedure TPipeStream.CheckBarrierAndFire(CurrentCycle: Int64);
begin
    // Barrier Check
    for var src in FSources do
    begin
        if src.LastSeenCycle < CurrentCycle then
            exit;
    end;

    // --- Barrier Open ---

    if FLastFiredCycleID >= CurrentCycle then
        exit;

    FLastFiredCycleID := CurrentCycle;

    // Execute Lambda
    if Assigned(FLambda) then
    begin
        try
            FLambda(FLambdaArgs);
        except
            // Error handling strategy: Propagate/Crash for now.
            raise;
        end;
    end;
end;

procedure TPipeStream.Emit(const Value: IScalarRecord);
begin
    FObservers.Lock;
    try
        FSeries.Add(Value);
        var signal := TStreamSignal.Create(skData, FLastFiredCycleID);

        FObservers.Notify(
            function(const Obs: IStreamObserver): Boolean
            begin
                Obs.OnSignal(signal);
                Result := True;
            end
        );
    finally
        FObservers.Release;
    end;
end;

function TPipeStream.GetSeries: IScalarRecordSeries;
begin
    Result := FSeries;
end;

end.