indicator: add v2 sma

pkg/indicator/sma.go

@@ -1,7 +1,6 @@
 package indicator
 
 import (
-	"fmt"
 	"time"
 
 	"github.com/c9s/bbgo/pkg/datatype/floats"
@@ -82,21 +81,3 @@ func (inc *SMA) LoadK(allKLines []types.KLine) {
 		inc.PushK(k)
 	}
 }
-
-func calculateSMA(kLines []types.KLine, window int, priceF KLineValueMapper) (float64, error) {
-	length := len(kLines)
-	if length == 0 || length < window {
-		return 0.0, fmt.Errorf("insufficient elements for calculating SMA with window = %d", window)
-	}
-	if length != window {
-		return 0.0, fmt.Errorf("too much klines passed in, requires only %d klines", window)
-	}
-
-	sum := 0.0
-	for _, k := range kLines {
-		sum += priceF(k)
-	}
-
-	avg := sum / float64(window)
-	return avg, nil
-}

pkg/indicator/v2_rsi.go

@@ -17,17 +17,11 @@ func RSI2(source Float64Source, window int) *RSIStream {
 		Float64Series: NewFloat64Series(),
 		window:        window,
 	}
-
-	if sub, ok := source.(Float64Subscription); ok {
-		sub.AddSubscriber(s.calculateAndPush)
-	} else {
-		source.OnUpdate(s.calculateAndPush)
-	}
-
+	s.Bind(source, s)
 	return s
 }
 
-func (s *RSIStream) calculate(_ float64) float64 {
+func (s *RSIStream) Calculate(_ float64) float64 {
 	var gainSum, lossSum float64
 	var sourceLen = s.source.Length()
 	var limit = min(s.window, sourceLen)
@@ -48,9 +42,3 @@ func (s *RSIStream) calculate(_ float64) float64 {
 	rsi := 100.0 - (100.0 / (1.0 + rs))
 	return rsi
 }
-
-func (s *RSIStream) calculateAndPush(x float64) {
-	rsi := s.calculate(x)
-	s.slice.Push(rsi)
-	s.EmitUpdate(rsi)
-}

pkg/indicator/v2_sma.go

@@ -0,0 +1,29 @@
+package indicator
+
+import "github.com/c9s/bbgo/pkg/types"
+
+type SMAStream struct {
+	Float64Series
+	window    int
+	rawValues *types.Queue
+}
+
+func SMA2(source Float64Source, window int) *SMAStream {
+	s := &SMAStream{
+		Float64Series: NewFloat64Series(),
+		window:        window,
+		rawValues:     types.NewQueue(window),
+	}
+	s.Bind(source, s)
+	return s
+}
+
+func (s *SMAStream) Calculate(v float64) float64 {
+	s.rawValues.Update(v)
+	sma := s.rawValues.Mean(s.window)
+	return sma
+}
+
+func (s *SMAStream) Truncate() {
+	s.slice.Truncate(MaxNumOfSMA)
+}