all: re-design and refactor indicator api

go.mod

@@ -2,7 +2,7 @@
 
 module github.com/c9s/bbgo
 
-go 1.17
+go 1.18
 
 require (
 	github.com/DATA-DOG/go-sqlmock v1.5.0

pkg/datatype/floats/slice.go

@@ -16,6 +16,12 @@ func (s *Slice) Push(v float64) {
 	*s = append(*s, v)
 }
 
+func (s *Slice) Append(vs ...float64) {
+	*s = append(*s, vs...)
+}
+
+// Update equals to Push()
+// which push an element into the slice
 func (s *Slice) Update(v float64) {
 	*s = append(*s, v)
 }
@@ -34,6 +40,38 @@ func (s Slice) Min() float64 {
 	return floats.Min(s)
 }
 
+func (s Slice) Sub(b Slice) (c Slice) {
+	if len(s) != len(b) {
+		return c
+	}
+
+	c = make(Slice, len(s))
+	for i := 0; i < len(s); i++ {
+		ai := s[i]
+		bi := b[i]
+		ci := ai - bi
+		c[i] = ci
+	}
+
+	return c
+}
+
+func (s Slice) Add(b Slice) (c Slice) {
+	if len(s) != len(b) {
+		return c
+	}
+
+	c = make(Slice, len(s))
+	for i := 0; i < len(s); i++ {
+		ai := s[i]
+		bi := b[i]
+		ci := ai + bi
+		c[i] = ci
+	}
+
+	return c
+}
+
 func (s Slice) Sum() (sum float64) {
 	return floats.Sum(s)
 }
@@ -125,27 +163,27 @@ func (s Slice) Normalize() Slice {
 	return s.DivScalar(s.Sum())
 }
 
-func (s *Slice) Last() float64 {
-	length := len(*s)
-	if length > 0 {
-		return (*s)[length-1]
-	}
-	return 0.0
-}
-
-func (s *Slice) Index(i int) float64 {
-	length := len(*s)
-	if length-i <= 0 || i < 0 {
-		return 0.0
-	}
-	return (*s)[length-i-1]
-}
-
-func (s *Slice) Length() int {
-	return len(*s)
-}
-
 func (s Slice) Addr() *Slice {
 	return &s
 }
 
+// Last, Index, Length implements the types.Series interface
+func (s Slice) Last() float64 {
+	length := len(s)
+	if length > 0 {
+		return (s)[length-1]
+	}
+	return 0.0
+}
+
+func (s Slice) Index(i int) float64 {
+	length := len(s)
+	if length-i <= 0 || i < 0 {
+		return 0.0
+	}
+	return (s)[length-i-1]
+}
+
+func (s Slice) Length() int {
+	return len(s)
+}

pkg/datatype/floats/slice_test.go

@@ -0,0 +1,25 @@
+package floats
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestSub(t *testing.T) {
+	a := New(1, 2, 3, 4, 5)
+	b := New(1, 2, 3, 4, 5)
+	c := a.Sub(b)
+	assert.Equal(t, Slice{.0, .0, .0, .0, .0}, c)
+	assert.Equal(t, 5, len(c))
+	assert.Equal(t, 5, c.Length())
+}
+
+func TestAdd(t *testing.T) {
+	a := New(1, 2, 3, 4, 5)
+	b := New(1, 2, 3, 4, 5)
+	c := a.Add(b)
+	assert.Equal(t, Slice{2.0, 4.0, 6.0, 8.0, 10.0}, c)
+	assert.Equal(t, 5, len(c))
+	assert.Equal(t, 5, c.Length())
+}

pkg/indicator/ewmastream_callbacks.go

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

pkg/indicator/float64updater_callbacks.go

@@ -0,0 +1,15 @@
+// Code generated by "callbackgen -type Float64Updater"; DO NOT EDIT.
+
+package indicator
+
+import ()
+
+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 {
+		cb(v)
+	}
+}

pkg/indicator/klinestream_callbacks.go

@@ -0,0 +1,17 @@
+// Code generated by "callbackgen -type KLineStream"; DO NOT EDIT.
+
+package indicator
+
+import (
+	"github.com/c9s/bbgo/pkg/types"
+)
+
+func (K *KLineStream) OnUpdate(cb func(k types.KLine)) {
+	K.updateCallbacks = append(K.updateCallbacks, cb)
+}
+
+func (K *KLineStream) EmitUpdate(k types.KLine) {
+	for _, cb := range K.updateCallbacks {
+		cb(k)
+	}
+}

pkg/indicator/macd.go

@@ -27,6 +27,22 @@ type MACDConfig struct {
 	LongPeriod int `json:"long"`
 }
 
+/*
+klines := kLines(marketDataStream)
+closePrices := closePrices(klines)
+macd := MACD(klines, {Fast: 12, Slow: 10})
+
+equals to:
+
+klines := KLines(marketDataStream)
+closePrices := ClosePrice(klines)
+fastEMA := EMA(closePrices, 7)
+slowEMA := EMA(closePrices, 25)
+macd := Subtract(fastEMA, slowEMA)
+signal := EMA(macd, 16)
+histogram := Subtract(macd, signal)
+*/
+
 //go:generate callbackgen -type MACD
 type MACD struct {
 	MACDConfig

pkg/indicator/macd2.go

@@ -0,0 +1,132 @@
+package indicator
+
+import (
+	"github.com/c9s/bbgo/pkg/datatype/floats"
+	"github.com/c9s/bbgo/pkg/types"
+)
+
+//go:generate callbackgen -type KLineStream
+type KLineStream struct {
+	updateCallbacks []func(k types.KLine)
+}
+
+func KLines(source types.Stream) *KLineStream {
+	stream := &KLineStream{}
+	source.OnKLineClosed(stream.EmitUpdate)
+	return stream
+}
+
+type KLineSource interface {
+	OnUpdate(f func(k types.KLine))
+}
+
+type PriceStream struct {
+	types.SeriesBase
+	Float64Updater
+
+	slice  floats.Slice
+	mapper KLineValueMapper
+}
+
+func Price(source KLineSource, mapper KLineValueMapper) *PriceStream {
+	s := &PriceStream{
+		mapper: mapper,
+	}
+	s.SeriesBase.Series = s.slice
+	source.OnUpdate(func(k types.KLine) {
+		v := s.mapper(k)
+		s.slice.Push(v)
+		s.EmitUpdate(v)
+	})
+	return s
+}
+
+func ClosePrices(source KLineSource) *PriceStream {
+	return Price(source, KLineClosePriceMapper)
+}
+
+func OpenPrices(source KLineSource) *PriceStream {
+	return Price(source, KLineOpenPriceMapper)
+}
+
+type Float64Source interface {
+	types.Series
+	OnUpdate(f func(v float64))
+}
+
+//go:generate callbackgen -type EWMAStream
+type EWMAStream struct {
+	Float64Updater
+	types.SeriesBase
+
+	slice floats.Slice
+
+	window     int
+	multiplier float64
+}
+
+func EWMA2(source Float64Source, window int) *EWMAStream {
+	s := &EWMAStream{
+		window:     window,
+		multiplier: 2.0 / float64(1+window),
+	}
+
+	s.SeriesBase.Series = s.slice
+	source.OnUpdate(func(v float64) {
+		v2 := s.calculate(v)
+		s.slice.Push(v2)
+		s.EmitUpdate(v2)
+	})
+	return s
+}
+
+func (s *EWMAStream) calculate(v float64) float64 {
+	last := s.slice.Last()
+	m := s.multiplier
+	return (1.0-m)*last + m*v
+}
+
+//go:generate callbackgen -type Float64Updater
+type Float64Updater struct {
+	updateCallbacks []func(v float64)
+}
+
+type SubtractStream struct {
+	Float64Updater
+	types.SeriesBase
+
+	a, b, c floats.Slice
+	i       int
+}
+
+func Subtract(a, b Float64Source) *SubtractStream {
+	s := &SubtractStream{}
+	s.SeriesBase.Series = s.c
+
+	a.OnUpdate(func(v float64) {
+		s.a.Push(v)
+		s.calculate()
+	})
+	b.OnUpdate(func(v float64) {
+		s.b.Push(v)
+		s.calculate()
+	})
+	return s
+}
+
+func (s *SubtractStream) calculate() {
+	if s.a.Length() != s.b.Length() {
+		return
+	}
+
+	if s.a.Length() > s.c.Length() {
+		var numNewElems = s.a.Length() - s.c.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.EmitUpdate(f)
+		}
+	}
+}

pkg/indicator/macd2_test.go

@@ -0,0 +1,35 @@
+package indicator
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+
+	"github.com/c9s/bbgo/pkg/fixedpoint"
+	"github.com/c9s/bbgo/pkg/types"
+)
+
+func TestSubtract(t *testing.T) {
+	stream := &types.StandardStream{}
+	kLines := KLines(stream)
+	closePrices := ClosePrices(kLines)
+	fastEMA := EWMA2(closePrices, 10)
+	slowEMA := EWMA2(closePrices, 25)
+	subtract := Subtract(fastEMA, slowEMA)
+	_ = subtract
+
+	for i := .0; i < 50.0; i++ {
+		stream.EmitKLineClosed(types.KLine{Close: fixedpoint.NewFromFloat(19_000.0 + i)})
+	}
+
+	t.Logf("fastEMA: %+v", fastEMA.slice)
+	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.Equal(t, subtract.c[0], subtract.a[0]-subtract.b[0])
+
+	t.Logf("subtract.a: %+v", subtract.a)
+	t.Logf("subtract.b: %+v", subtract.b)
+	t.Logf("subtract.c: %+v", subtract.c)
+}

pkg/types/indicator.go

@@ -115,10 +115,6 @@ type SeriesExtend interface {
 	Filter(b func(i int, value float64) bool, length int) SeriesExtend
 }
 
-type SeriesBase struct {
-	Series
-}
-
 func NewSeries(a Series) SeriesExtend {
 	return &SeriesBase{
 		Series: a,
@@ -618,7 +614,7 @@ func Dot(a interface{}, b interface{}, limit ...int) float64 {
 	}
 }
 
-// Extract elements from the Series to a float64 array, following the order of Index(0..limit)
+// Array extracts elements from the Series to a float64 array, following the order of Index(0..limit)
 // if limit is given, will only take the first limit numbers (a.Index[0..limit])
 // otherwise will operate on all elements
 func Array(a Series, limit ...int) (result []float64) {

pkg/types/seriesbase_imp.go

@@ -2,6 +2,10 @@ package types
 
 import "github.com/c9s/bbgo/pkg/datatype/floats"
 
+type SeriesBase struct {
+	Series
+}
+
 func (s *SeriesBase) Index(i int) float64 {
 	if s.Series == nil {
 		return 0