RxSwift observable creation which takes the last value of the stream itself

First let's cleanup your sma operator. You are creating dispose bags inside the function which is inappropriate. The subscribe returns a disposable and the create's closure needs to return a disposable. Just return the subscribe's disposable...

func sma(_ source: Observable<Double>, _ length: Int) -> Observable<Double?> { Observable<Double?>.create { observer -> Disposable in source .scan([]) { Array($0 + [$1].suffix(length)) } .subscribe(onNext: { value in if value.count < length { observer.onNext(nil) } else { observer.onNext(value.reduce(0.0, { $0 + $1 / Double(length) })) } }) } } 

But since you are outputting the same number of events as you input, we can simplify even more. Whenever you are outputting the same number of events as you input, think map.

func sma(_ source: Observable<Double>, _ length: Int) -> Observable<Double?> { source .scan([]) { Array($0 + [$1].suffix(length)) } .map { value in if value.count < length { return nil } else { return value.reduce(0.0, { $0 + $1 / Double(length) }) } } } 

And whenever you only have one Observable as input, consider making it an extension on the Observable type so it can be easily inserted into a chain...

extension ObservableType where Element == Double { func sma(_ length: Int) -> Observable<Double?> { scan([]) { Array($0 + [$1].suffix(length)) } .map { $0.count < length ? nil : $0.reduce(0.0, { $0 + $1 / Double(length) }) } } } 

Now that we have done all that, let's tackle your actual question. First express your formula as a function...

func getEMA(prices: [Double], k: Double? = nil) -> Double { guard !prices.isEmpty else { return 0 } let k = k ?? Double(2 / (prices.count + 1)) return prices[0] * k + getEMA(prices: prices.suffix(prices.count - 1), k: k) * (1 - k) } 

The above should be easy to test using some sample values. I'll leave you to do that. Once we have the above, we can use the same pattern you used for your sma to create the operator:

extension ObservableType where Element == Double { func ema(_ length: Int) -> Observable<Double?> { scan([]) { Array([$1] + $0).suffix(length) } // put the most recent price in front to correctly handle the formula .map { $0.count < length ? nil : getEMA(prices: $0) } } } 

I found a solution (not sure if its the cleanest way)

which is to create a BehaviourRelay inside the function declaration while outside of the Observable.create block, which keeps a local copy of the latest EMA calculated.

in this way, it does not need function's consumer to dependency inject an Observable nor having a complex stream transformation.

Below is the EMA(exponentially weighted moving average) implementation using RxSwift

func ema(_ source: Observable<Double>, _ length: Int) -> Observable<Double?> { let bag = DisposeBag() let lastEMA: BehaviorRelay<Double?> = BehaviorRelay.init(value: nil) return Observable<Double?>.create { observer -> Disposable in source.scan([]) { Array($0 + [$1].suffix(length)) } .subscribe(onNext: { value in let alpha: Double = Double(2) / Double(length + 1) let src = value var sum: Double? = 0.0 sum = na(lastEMA.value) ? sma(src, length) : alpha * src.last! + (1 - alpha) * nz(lastEMA.value) observer.onNext(sum) lastEMA.accept(sum) }).disposed(by: bag) return Disposables.create() } } 

Remarks:

na and nz are method copied from TradingView pinescript

https://www.tradingview.com/pine-script-reference/#var_na

https://www.tradingview.com/pine-script-reference/#fun_nz

https://www.tradingview.com/pine-script-reference/#fun_ema