indicator: refactor/add float64 series

pkg/indicator/atr2.go

@@ -0,0 +1,37 @@
+package indicator
+
+import (
+	"github.com/c9s/bbgo/pkg/types"
+)
+
+type ATRStream struct {
+	Float64Updater
+
+	types.SeriesBase
+
+	window     int
+	multiplier float64
+}
+
+func ATR2(source KLineSubscription, window int) *ATRStream {
+	s := &ATRStream{
+		window:     window,
+		multiplier: 2.0 / float64(1+window),
+	}
+
+	s.SeriesBase.Series = s.slice
+
+	source.AddSubscriber(func(k types.KLine) {
+		// v := s.mapper(k)
+		// s.slice.Push(v)
+		// s.EmitUpdate(v)
+	})
+
+	return s
+}
+
+func (s *ATRStream) calculateAndPush(k types.KLine) {
+	// v2 := s.calculate(v)
+	// s.slice.Push(v2)
+	// s.EmitUpdate(v2)
+}

pkg/indicator/ewma2.go

@@ -1,16 +1,7 @@
 package indicator
 
-import (
-	"github.com/c9s/bbgo/pkg/datatype/floats"
-	"github.com/c9s/bbgo/pkg/types"
-)
-
-//go:generate callbackgen -type EWMAStream
 type EWMAStream struct {
-	Float64Updater
-	types.SeriesBase
-
-	slice floats.Slice
+	Float64Series
 
 	window     int
 	multiplier float64

pkg/indicator/ewmastream_callbacks.go

@@ -1,5 +0,0 @@
-// Code generated by "callbackgen -type EWMAStream"; DO NOT EDIT.
-
-package indicator
-
-import ()

pkg/indicator/float64updater.go

@@ -1,6 +1,34 @@
 package indicator
 
+import (
+	"github.com/c9s/bbgo/pkg/datatype/floats"
+	"github.com/c9s/bbgo/pkg/types"
+)
+
 //go:generate callbackgen -type Float64Updater
 type Float64Updater struct {
 	updateCallbacks []func(v float64)
+
+	slice floats.Slice
+}
+
+type Float64Series struct {
+	types.SeriesBase
+	Float64Updater
+}
+
+func (f *Float64Series) Last() float64 {
+	return f.slice.Last()
+}
+
+func (f *Float64Series) Index(i int) float64 {
+	length := len(f.slice)
+	if length == 0 || length-i-1 < 0 {
+		return 0
+	}
+	return f.slice[length-i-1]
+}
+
+func (f *Float64Series) Length() int {
+	return len(f.slice)
 }

pkg/indicator/float64updater_callbacks.go

@@ -4,12 +4,12 @@ package indicator
 
 import ()
 
-func (F *Float64Updater) OnUpdate(cb func(v float64)) {
-	F.updateCallbacks = append(F.updateCallbacks, cb)
+func (f *Float64Updater) OnUpdate(cb func(v float64)) {
+	f.updateCallbacks = append(f.updateCallbacks, cb)
 }
 
-func (F *Float64Updater) EmitUpdate(v float64) {
-	for _, cb := range F.updateCallbacks {
+func (f *Float64Updater) EmitUpdate(v float64) {
+	for _, cb := range f.updateCallbacks {
 		cb(v)
 	}
 }

pkg/indicator/price.go

@@ -1,7 +1,6 @@
 package indicator
 
 import (
-	"github.com/c9s/bbgo/pkg/datatype/floats"
 	"github.com/c9s/bbgo/pkg/types"
 )
 
@@ -10,10 +9,8 @@ type KLineSubscription interface {
 }
 
 type PriceStream struct {
-	types.SeriesBase
-	Float64Updater
+	Float64Series
 
-	slice  floats.Slice
 	mapper KLineValueMapper
 }
 

pkg/indicator/rma2.go

@@ -0,0 +1,68 @@
+package indicator
+
+import (
+	"github.com/c9s/bbgo/pkg/types"
+)
+
+type RMAStream struct {
+	// embedded structs
+	Float64Updater
+	types.SeriesBase
+
+	// config fields
+	types.IntervalWindow
+	Adjust bool
+
+	counter  int
+	sum, tmp float64
+}
+
+func RMA2(source Float64Source, iw types.IntervalWindow) *RMAStream {
+	s := &RMAStream{
+		IntervalWindow: iw,
+	}
+
+	s.SeriesBase.Series = s.slice
+
+	if sub, ok := source.(Float64Subscription); ok {
+		sub.AddSubscriber(s.calculateAndPush)
+	} else {
+		source.OnUpdate(s.calculateAndPush)
+	}
+
+	return s
+}
+
+func (s *RMAStream) calculateAndPush(v float64) {
+	v2 := s.calculate(v)
+	s.slice.Push(v2)
+	s.EmitUpdate(v2)
+}
+
+func (s *RMAStream) calculate(x float64) float64 {
+	lambda := 1 / float64(s.Window)
+	if s.counter == 0 {
+		s.sum = 1
+		s.tmp = x
+	} else {
+		if s.Adjust {
+			s.sum = s.sum*(1-lambda) + 1
+			s.tmp = s.tmp + (x-s.tmp)/s.sum
+		} else {
+			s.tmp = s.tmp*(1-lambda) + x*lambda
+		}
+	}
+	s.counter++
+
+	if s.counter < s.Window {
+		s.slice.Push(0)
+	}
+
+	s.slice.Push(s.tmp)
+
+	if len(s.slice) > MaxNumOfRMA {
+		s.slice = s.slice[MaxNumOfRMATruncateSize-1:]
+	}
+
+	return s.tmp
+}

pkg/indicator/subtract.go

@@ -10,15 +10,15 @@ type SubtractStream struct {
 	Float64Updater
 	types.SeriesBase
 
-	a, b, c floats.Slice
-	i       int
+	a, b floats.Slice
+	i    int
 }
 
 // Subtract creates the SubtractStream object
 // subtract := Subtract(longEWMA, shortEWMA)
 func Subtract(a, b Float64Source) *SubtractStream {
 	s := &SubtractStream{}
-	s.SeriesBase.Series = s.c
+	s.SeriesBase.Series = s.slice
 
 	a.OnUpdate(func(v float64) {
 		s.a.Push(v)
@@ -36,13 +36,13 @@ func (s *SubtractStream) calculate() {
 		return
 	}
 
-	if s.a.Length() > s.c.Length() {
-		var numNewElems = s.a.Length() - s.c.Length()
+	if s.a.Length() > s.slice.Length() {
+		var numNewElems = s.a.Length() - s.slice.Length()
 		var tailA = s.a.Tail(numNewElems)
 		var tailB = s.b.Tail(numNewElems)
 		var tailC = tailA.Sub(tailB)
 		for _, f := range tailC {
-			s.c.Push(f)
+			s.slice.Push(f)
 			s.EmitUpdate(f)
 		}
 	}

pkg/indicator/v2_test.go

@@ -25,6 +25,6 @@ func Test_v2_Subtract(t *testing.T) {
 	t.Logf("slowEMA: %+v", slowEMA.slice)
 
 	assert.Equal(t, len(subtract.a), len(subtract.b))
-	assert.Equal(t, len(subtract.a), len(subtract.c))
-	assert.InDelta(t, subtract.c[0], subtract.a[0]-subtract.b[0], 0.0001)
+	assert.Equal(t, len(subtract.a), len(subtract.slice))
+	assert.InDelta(t, subtract.slice[0], subtract.a[0]-subtract.b[0], 0.0001)
 }