Merge pull request #1184 from c9s/feature/v2indicator-macd

FEATURE: [indicator] add v2 MACD, SMA

pkg/datatype/floats/slice.go

@@ -191,14 +191,18 @@ func (s Slice) Last(i int) float64 {
 	return s[length-1-i]
 }
 
+func (s Slice) Truncate(size int) Slice {
+	if size < 0 || len(s) <= size {
+		return s
+	}
+
+	return s[len(s)-size:]
+}
+
 // Index fetches the element from the end of the slice
 // WARNING: it does not start from 0!!!
 func (s Slice) Index(i int) float64 {
-	length := len(s)
-	if i < 0 || length-1-i < 0 {
-		return 0.0
-	}
-	return s[length-1-i]
+	return s.Last(i)
 }
 
 func (s Slice) Length() int {

pkg/datatype/floats/slice_test.go

@@ -15,6 +15,14 @@ func TestSub(t *testing.T) {
 	assert.Equal(t, 5, c.Length())
 }
 
+func TestTruncate(t *testing.T) {
+	a := New(1, 2, 3, 4, 5)
+	for i := 5; i > 0; i-- {
+		a = a.Truncate(i)
+		assert.Equal(t, i, a.Length())
+	}
+}
+
 func TestAdd(t *testing.T) {
 	a := New(1, 2, 3, 4, 5)
 	b := New(1, 2, 3, 4, 5)

pkg/indicator/float64updater.go

@@ -28,9 +28,41 @@ func (f *Float64Series) Last(i int) float64 {
 }
 
 func (f *Float64Series) Index(i int) float64 {
-	return f.slice.Last(i)
+	return f.Last(i)
 }
 
 func (f *Float64Series) Length() int {
 	return len(f.slice)
 }
+
+func (f *Float64Series) PushAndEmit(x float64) {
+	f.slice.Push(x)
+	f.EmitUpdate(x)
+}
+
+// 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) {
+	var c func(x float64)
+
+	// optimize the truncation check
+	trc, canTruncate := target.(Float64Truncator)
+	if canTruncate {
+		c = func(x float64) {
+			y := target.Calculate(x)
+			target.PushAndEmit(y)
+			trc.Truncate()
+		}
+	} else {
+		c = func(x float64) {
+			y := target.Calculate(x)
+			target.PushAndEmit(y)
+		}
+	}
+
+	if sub, ok := source.(Float64Subscription); ok {
+		sub.AddSubscriber(c)
+	} else {
+		source.OnUpdate(c)
+	}
+}

pkg/indicator/klinestream.go

@@ -11,29 +11,43 @@ type KLineStream struct {
 	kLines []types.KLine
 }
 
+func (s *KLineStream) Length() int {
+	return len(s.kLines)
+}
+
+func (s *KLineStream) Last(i int) *types.KLine {
+	l := len(s.kLines)
+	if i < 0 || l-1-i < 0 {
+		return nil
+	}
+
+	return &s.kLines[l-1-i]
+}
+
 // AddSubscriber adds the subscriber function and push historical data to the subscriber
 func (s *KLineStream) AddSubscriber(f func(k types.KLine)) {
+	s.OnUpdate(f)
+
 	if len(s.kLines) > 0 {
 		// push historical klines to the subscriber
 		for _, k := range s.kLines {
 			f(k)
 		}
 	}
-	s.OnUpdate(f)
 }
 
 // KLines creates a KLine stream that pushes the klines to the subscribers
-func KLines(source types.Stream) *KLineStream {
+func KLines(source types.Stream, symbol string, interval types.Interval) *KLineStream {
 	s := &KLineStream{}
 
-	source.OnKLineClosed(func(k types.KLine) {
+	source.OnKLineClosed(types.KLineWith(symbol, interval, func(k types.KLine) {
 		s.kLines = append(s.kLines, k)
+		s.EmitUpdate(k)
 
 		if len(s.kLines) > MaxNumOfKLines {
 			s.kLines = s.kLines[len(s.kLines)-1-MaxNumOfKLines:]
 		}
-		s.EmitUpdate(k)
-	})
+	}))
 
 	return s
 }

pkg/indicator/low.go

@@ -1,37 +0,0 @@
-package indicator
-
-import (
-	"time"
-
-	"github.com/c9s/bbgo/pkg/datatype/floats"
-	"github.com/c9s/bbgo/pkg/types"
-)
-
-//go:generate callbackgen -type Low
-type Low struct {
-	types.IntervalWindow
-	types.SeriesBase
-
-	Values  floats.Slice
-	EndTime time.Time
-
-	updateCallbacks []func(value float64)
-}
-
-func (inc *Low) Update(value float64) {
-	if len(inc.Values) == 0 {
-		inc.SeriesBase.Series = inc
-	}
-
-	inc.Values.Push(value)
-}
-
-func (inc *Low) PushK(k types.KLine) {
-	if k.EndTime.Before(inc.EndTime) {
-		return
-	}
-
-	inc.Update(k.Low.Float64())
-	inc.EndTime = k.EndTime.Time()
-	inc.EmitUpdate(inc.Last(0))
-}

pkg/indicator/low_callbacks.go

@@ -1,15 +0,0 @@
-// Code generated by "callbackgen -type Low"; DO NOT EDIT.
-
-package indicator
-
-import ()
-
-func (inc *Low) OnUpdate(cb func(value float64)) {
-	inc.updateCallbacks = append(inc.updateCallbacks, cb)
-}
-
-func (inc *Low) EmitUpdate(value float64) {
-	for _, cb := range inc.updateCallbacks {
-		cb(value)
-	}
-}

pkg/indicator/macd.go

@@ -75,12 +75,8 @@ func (inc *MACDLegacy) Update(x float64) {
 	inc.EmitUpdate(macd, signal, histogram)
 }
 
-func (inc *MACDLegacy) Last(int) float64 {
-	if len(inc.Values) == 0 {
-		return 0.0
-	}
-
-	return inc.Values[len(inc.Values)-1]
+func (inc *MACDLegacy) Last(i int) float64 {
+	return inc.Values.Last(i)
 }
 
 func (inc *MACDLegacy) Length() int {
@@ -111,7 +107,7 @@ func (inc *MACDValues) Last(i int) float64 {
 }
 
 func (inc *MACDValues) Index(i int) float64 {
-	return inc.Values.Last(i)
+	return inc.Last(i)
 }
 
 func (inc *MACDValues) Length() int {

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/types.go

@@ -4,6 +4,7 @@ import "github.com/c9s/bbgo/pkg/types"
 
 type Float64Calculator interface {
 	Calculate(x float64) float64
+	PushAndEmit(x float64)
 }
 
 type Float64Source interface {
@@ -15,3 +16,7 @@ type Float64Subscription interface {
 	types.Series
 	AddSubscriber(f func(v float64))
 }
+
+type Float64Truncator interface {
+	Truncate()
+}

pkg/indicator/v2_atr_test.go

@@ -64,7 +64,7 @@ func Test_ATR2(t *testing.T) {
 		t.Run(tt.name, func(t *testing.T) {
 			stream := &types.StandardStream{}
 
-			kLines := KLines(stream)
+			kLines := KLines(stream, "", "")
 			atr := ATR2(kLines, tt.window)
 
 			for _, k := range tt.kLines {

pkg/indicator/v2_ewma.go

@@ -13,23 +13,11 @@ func EWMA2(source Float64Source, window int) *EWMAStream {
 		window:        window,
 		multiplier:    2.0 / float64(1+window),
 	}
-
-	if sub, ok := source.(Float64Subscription); ok {
-		sub.AddSubscriber(s.calculateAndPush)
-	} else {
-		source.OnUpdate(s.calculateAndPush)
-	}
-
+	s.Bind(source, s)
 	return s
 }
 
-func (s *EWMAStream) calculateAndPush(v float64) {
-	v2 := s.calculate(v)
-	s.slice.Push(v2)
-	s.EmitUpdate(v2)
-}
-
-func (s *EWMAStream) calculate(v float64) float64 {
+func (s *EWMAStream) Calculate(v float64) float64 {
 	last := s.slice.Last(0)
 	m := s.multiplier
 	return (1.0-m)*last + m*v

pkg/indicator/v2_macd.go

@@ -0,0 +1,29 @@
+package indicator
+
+type MACDStream struct {
+	*SubtractStream
+
+	shortWindow, longWindow, signalWindow int
+
+	fastEWMA, slowEWMA, signal *EWMAStream
+	histogram                  *SubtractStream
+}
+
+func MACD2(source Float64Source, shortWindow, longWindow, signalWindow int) *MACDStream {
+	// bind and calculate these first
+	fastEWMA := EWMA2(source, shortWindow)
+	slowEWMA := EWMA2(source, longWindow)
+	macd := Subtract(fastEWMA, slowEWMA)
+	signal := EWMA2(macd, signalWindow)
+	histogram := Subtract(macd, signal)
+	return &MACDStream{
+		SubtractStream: macd,
+		shortWindow:    shortWindow,
+		longWindow:     longWindow,
+		signalWindow:   signalWindow,
+		fastEWMA:       fastEWMA,
+		slowEWMA:       slowEWMA,
+		signal:         signal,
+		histogram:      histogram,
+	}
+}

pkg/indicator/v2_macd_test.go

@@ -0,0 +1,57 @@
+package indicator
+
+import (
+	"encoding/json"
+	"math"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+
+	"github.com/c9s/bbgo/pkg/fixedpoint"
+	"github.com/c9s/bbgo/pkg/types"
+)
+
+/*
+python:
+
+import pandas as pd
+s = pd.Series([0,1,2,3,4,5,6,7,8,9,0,1,2,3,4,5,6,7,8,9,0,1,2,3,4,5,6,7,8,9,0,1,2,3,4,5,6,7,8,9,0,1,2,3,4,5,6,7,8,9])
+slow = s.ewm(span=26, adjust=False).mean()
+fast = s.ewm(span=12, adjust=False).mean()
+print(fast - slow)
+*/
+
+func Test_MACD2(t *testing.T) {
+	var randomPrices = []byte(`[0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 0, 1, 2, 3, 4, 5, 6, 7, 8, 9]`)
+	var input []fixedpoint.Value
+	err := json.Unmarshal(randomPrices, &input)
+	assert.NoError(t, err)
+
+	tests := []struct {
+		name   string
+		kLines []types.KLine
+		want   float64
+	}{
+		{
+			name:   "random_case",
+			kLines: buildKLines(input),
+			want:   0.7967670223776384,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			prices := &PriceStream{}
+			macd := MACD2(prices, 12, 26, 9)
+			for _, k := range tt.kLines {
+				prices.EmitUpdate(k.Close.Float64())
+			}
+
+			got := macd.Last(0)
+			diff := math.Trunc((got-tt.want)*100) / 100
+			if diff != 0 {
+				t.Errorf("MACD2() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}

pkg/indicator/v2_rma.go

@@ -19,23 +19,11 @@ func RMA2(source Float64Source, window int, adjust bool) *RMAStream {
 		Adjust:        adjust,
 	}
 
-	if sub, ok := source.(Float64Subscription); ok {
-		sub.AddSubscriber(s.calculateAndPush)
-	} else {
-		source.OnUpdate(s.calculateAndPush)
-	}
-
+	s.Bind(source, s)
 	return s
 }
 
-func (s *RMAStream) calculateAndPush(v float64) {
-	v2 := s.calculate(v)
-	s.slice.Push(v2)
-	s.EmitUpdate(v2)
-	s.truncate()
-}
-
-func (s *RMAStream) calculate(x float64) float64 {
+func (s *RMAStream) Calculate(x float64) float64 {
 	lambda := 1 / float64(s.window)
 	tmp := 0.0
 	if s.counter == 0 {
@@ -62,7 +50,7 @@ func (s *RMAStream) calculate(x float64) float64 {
 	return tmp
 }
 
-func (s *RMAStream) truncate() {
+func (s *RMAStream) Truncate() {
 	if len(s.slice) > MaxNumOfRMA {
 		s.slice = s.slice[MaxNumOfRMATruncateSize-1:]
 	}

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 = s.slice.Truncate(MaxNumOfSMA)
+}

pkg/indicator/v2_test.go

@@ -11,7 +11,7 @@ import (
 
 func Test_v2_Subtract(t *testing.T) {
 	stream := &types.StandardStream{}
-	kLines := KLines(stream)
+	kLines := KLines(stream, "", "")
 	closePrices := ClosePrices(kLines)
 	fastEMA := EWMA2(closePrices, 10)
 	slowEMA := EWMA2(closePrices, 25)

pkg/indicator/v2_tr_test.go

@@ -64,7 +64,7 @@ func Test_TR_and_RMA(t *testing.T) {
 		t.Run(tt.name, func(t *testing.T) {
 			stream := &types.StandardStream{}
 
-			kLines := KLines(stream)
+			kLines := KLines(stream, "", "")
 			atr := TR2(kLines)
 			rma := RMA2(atr, tt.window, true)