indicator: add v2 pivot low indicator

pkg/indicator/float64updater.go

@@ -40,6 +40,14 @@ func (f *Float64Series) PushAndEmit(x float64) {
 	f.EmitUpdate(x)
 }
 
+func (f *Float64Series) Subscribe(source Float64Source, c func(x float64)) {
+	if sub, ok := source.(Float64Subscription); ok {
+		sub.AddSubscriber(c)
+	} else {
+		source.OnUpdate(c)
+	}
+}
+
 // Bind binds the source event to the target (Float64Calculator)
 // A Float64Calculator should be able to calculate the float64 result from a single float64 argument input
 func (f *Float64Series) Bind(source Float64Source, target Float64Calculator) {
@@ -60,9 +68,5 @@ func (f *Float64Series) Bind(source Float64Source, target Float64Calculator) {
 		}
 	}
 
-	if sub, ok := source.(Float64Subscription); ok {
+	f.Subscribe(source, c)
-		sub.AddSubscriber(c)
-	} else {
-		source.OnUpdate(c)
-	}
 }

pkg/indicator/v2_pivotlow.go

@@ -0,0 +1,29 @@
+package indicator
+
+import (
+	"github.com/c9s/bbgo/pkg/datatype/floats"
+)
+
+type PivotLowStream struct {
+	Float64Series
+
+	rawValues floats.Slice
+
+	window, rightWindow int
+}
+
+func PivotLow2(source Float64Source, window, rightWindow int) *PivotLowStream {
+	s := &PivotLowStream{
+		Float64Series: NewFloat64Series(),
+		window:        window,
+		rightWindow:   rightWindow,
+	}
+
+	s.Subscribe(source, func(x float64) {
+		s.rawValues.Push(x)
+		if low, ok := calculatePivotLow(s.rawValues, s.window, s.rightWindow); ok {
+			s.PushAndEmit(low)
+		}
+	})
+	return s
+}