Merge pull request #1179 from c9s/c9s/refactor-indicator

FEATURE: new indicator API design

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/bbgo/standard_indicator_set.go

@@ -29,7 +29,7 @@ type StandardIndicatorSet struct {
 	// interval -> window
 	iwbIndicators  map[types.IntervalWindowBandWidth]*indicator.BOLL
 	iwIndicators   map[indicatorKey]indicator.KLinePusher
-	macdIndicators map[indicator.MACDConfig]*indicator.MACD
+	macdIndicators map[indicator.MACDConfig]*indicator.MACDLegacy
 
 	stream types.Stream
 	store  *MarketDataStore
@@ -47,7 +47,7 @@ func NewStandardIndicatorSet(symbol string, stream types.Stream, store *MarketDa
 		stream:         stream,
 		iwIndicators:   make(map[indicatorKey]indicator.KLinePusher),
 		iwbIndicators:  make(map[types.IntervalWindowBandWidth]*indicator.BOLL),
-		macdIndicators: make(map[indicator.MACDConfig]*indicator.MACD),
+		macdIndicators: make(map[indicator.MACDConfig]*indicator.MACDLegacy),
 	}
 }
 
@@ -154,14 +154,14 @@ func (s *StandardIndicatorSet) BOLL(iw types.IntervalWindow, bandWidth float64)
 	return inc
 }
 
-func (s *StandardIndicatorSet) MACD(iw types.IntervalWindow, shortPeriod, longPeriod int) *indicator.MACD {
+func (s *StandardIndicatorSet) MACD(iw types.IntervalWindow, shortPeriod, longPeriod int) *indicator.MACDLegacy {
 	config := indicator.MACDConfig{IntervalWindow: iw, ShortPeriod: shortPeriod, LongPeriod: longPeriod}
 
 	inc, ok := s.macdIndicators[config]
 	if ok {
 		return inc
 	}
-	inc = &indicator.MACD{MACDConfig: config}
+	inc = &indicator.MACDLegacy{MACDConfig: config}
 	s.macdIndicators[config] = inc
 	s.initAndBind(inc, config.IntervalWindow.Interval)
 	return inc

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

@@ -0,0 +1,6 @@
+package indicator
+
+//go:generate callbackgen -type Float64Updater
+type Float64Updater struct {
+	updateCallbacks []func(v float64)
+}

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

@@ -0,0 +1,39 @@
+package indicator
+
+import "github.com/c9s/bbgo/pkg/types"
+
+const MaxNumOfKLines = 4_000
+
+//go:generate callbackgen -type KLineStream
+type KLineStream struct {
+	updateCallbacks []func(k types.KLine)
+
+	kLines []types.KLine
+}
+
+// AddSubscriber adds the subscriber function and push historical data to the subscriber
+func (s *KLineStream) AddSubscriber(f func(k types.KLine)) {
+	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 {
+	s := &KLineStream{}
+
+	source.OnKLineClosed(func(k types.KLine) {
+		s.kLines = append(s.kLines, k)
+
+		if len(s.kLines) > MaxNumOfKLines {
+			s.kLines = s.kLines[len(s.kLines)-1-MaxNumOfKLines:]
+		}
+		s.EmitUpdate(k)
+	})
+
+	return s
+}

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 (s *KLineStream) OnUpdate(cb func(k types.KLine)) {
+	s.updateCallbacks = append(s.updateCallbacks, cb)
+}
+
+func (s *KLineStream) EmitUpdate(k types.KLine) {
+	for _, cb := range s.updateCallbacks {
+		cb(k)
+	}
+}

pkg/indicator/macd2.go

@@ -0,0 +1,114 @@
+package indicator
+
+import (
+	"github.com/c9s/bbgo/pkg/datatype/floats"
+	"github.com/c9s/bbgo/pkg/types"
+)
+
+/*
+NEW INDICATOR DESIGN:
+
+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)
+*/
+
+type Float64Source interface {
+	types.Series
+	OnUpdate(f func(v float64))
+}
+
+type Float64Subscription interface {
+	types.Series
+	AddSubscriber(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
+
+	if sub, ok := source.(Float64Subscription); ok {
+		sub.AddSubscriber(s.calculateAndPush)
+	} else {
+		source.OnUpdate(s.calculateAndPush)
+	}
+
+	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 {
+	last := s.slice.Last()
+	m := s.multiplier
+	return (1.0-m)*last + m*v
+}
+
+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,30 @@
+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)
+
+	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.InDelta(t, subtract.c[0], subtract.a[0]-subtract.b[0], 0.0001)
+}

pkg/indicator/macd_callbacks.go

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

pkg/indicator/macdlegacy.go

@@ -27,20 +27,19 @@ type MACDConfig struct {
 	LongPeriod int `json:"long"`
 }
 
-//go:generate callbackgen -type MACD
-type MACD struct {
+//go:generate callbackgen -type MACDLegacy
+type MACDLegacy struct {
 	MACDConfig
 
 	Values                         floats.Slice `json:"-"`
 	fastEWMA, slowEWMA, signalLine *EWMA
 	Histogram                      floats.Slice `json:"-"`
 
-	EndTime time.Time
-
 	updateCallbacks []func(macd, signal, histogram float64)
+	EndTime         time.Time
 }
 
-func (inc *MACD) Update(x float64) {
+func (inc *MACDLegacy) Update(x float64) {
 	if len(inc.Values) == 0 {
 		// apply default values
 		inc.fastEWMA = &EWMA{IntervalWindow: types.IntervalWindow{Window: inc.ShortPeriod}}
@@ -76,7 +75,7 @@ func (inc *MACD) Update(x float64) {
 	inc.EmitUpdate(macd, signal, histogram)
 }
 
-func (inc *MACD) Last() float64 {
+func (inc *MACDLegacy) Last() float64 {
 	if len(inc.Values) == 0 {
 		return 0.0
 	}
@@ -84,27 +83,27 @@ func (inc *MACD) Last() float64 {
 	return inc.Values[len(inc.Values)-1]
 }
 
-func (inc *MACD) Length() int {
+func (inc *MACDLegacy) Length() int {
 	return len(inc.Values)
 }
 
-func (inc *MACD) PushK(k types.KLine) {
+func (inc *MACDLegacy) PushK(k types.KLine) {
 	inc.Update(k.Close.Float64())
 }
 
-func (inc *MACD) MACD() types.SeriesExtend {
-	out := &MACDValues{MACD: inc}
+func (inc *MACDLegacy) MACD() types.SeriesExtend {
+	out := &MACDValues{MACDLegacy: inc}
 	out.SeriesBase.Series = out
 	return out
 }
 
-func (inc *MACD) Singals() types.SeriesExtend {
+func (inc *MACDLegacy) Singals() types.SeriesExtend {
 	return inc.signalLine
 }
 
 type MACDValues struct {
 	types.SeriesBase
-	*MACD
+	*MACDLegacy
 }
 
 func (inc *MACDValues) Last() float64 {

pkg/indicator/macdlegacy_callbacks.go

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

pkg/indicator/macdlegacy_test.go

@@ -40,7 +40,7 @@ func Test_calculateMACD(t *testing.T) {
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
 			iw := types.IntervalWindow{Window: 9}
-			macd := MACD{MACDConfig: MACDConfig{IntervalWindow: iw, ShortPeriod: 12, LongPeriod: 26}}
+			macd := MACDLegacy{MACDConfig: MACDConfig{IntervalWindow: iw, ShortPeriod: 12, LongPeriod: 26}}
 			for _, k := range tt.kLines {
 				macd.PushK(k)
 			}

pkg/indicator/price.go

@@ -0,0 +1,49 @@
+package indicator
+
+import (
+	"github.com/c9s/bbgo/pkg/datatype/floats"
+	"github.com/c9s/bbgo/pkg/types"
+)
+
+type KLineSubscription interface {
+	AddSubscriber(f func(k types.KLine))
+}
+
+type PriceStream struct {
+	types.SeriesBase
+	Float64Updater
+
+	slice  floats.Slice
+	mapper KLineValueMapper
+}
+
+func Price(source KLineSubscription, mapper KLineValueMapper) *PriceStream {
+	s := &PriceStream{
+		mapper: mapper,
+	}
+
+	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 ClosePrices(source KLineSubscription) *PriceStream {
+	return Price(source, KLineClosePriceMapper)
+}
+
+func LowPrices(source KLineSubscription) *PriceStream {
+	return Price(source, KLineLowPriceMapper)
+}
+
+func HighPrices(source KLineSubscription) *PriceStream {
+	return Price(source, KLineHighPriceMapper)
+}
+
+func OpenPrices(source KLineSubscription) *PriceStream {
+	return Price(source, KLineOpenPriceMapper)
+}

pkg/strategy/drift/draw.go

@@ -5,10 +5,11 @@ import (
 	"fmt"
 	"os"
 
+	"github.com/wcharczuk/go-chart/v2"
+
 	"github.com/c9s/bbgo/pkg/bbgo"
 	"github.com/c9s/bbgo/pkg/interact"
 	"github.com/c9s/bbgo/pkg/types"
-	"github.com/wcharczuk/go-chart/v2"
 )
 
 func (s *Strategy) InitDrawCommands(profit, cumProfit types.Series) {
@@ -76,7 +77,7 @@ func (s *Strategy) DrawIndicators(time types.Time) *types.Canvas {
 
 	// canvas.Plot("upband", s.ma.Add(s.stdevHigh), time, length)
 	canvas.Plot("ma", s.ma, time, length)
-	// canvas.Plot("downband", s.ma.Minus(s.stdevLow), time, length)
+	// canvas.Plot("downband", s.ma.Sub(s.stdevLow), time, length)
 	fmt.Printf("%f %f\n", highestPrice, hi)
 
 	canvas.Plot("trend", s.trendLine, time, length)

pkg/strategy/pivotshort/failedbreakhigh.go

@@ -47,7 +47,7 @@ type FailedBreakHigh struct {
 
 	MACDDivergence *MACDDivergence `json:"macdDivergence"`
 
-	macd *indicator.MACD
+	macd *indicator.MACDLegacy
 
 	macdTopDivergence bool
 

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,
@@ -412,8 +408,8 @@ type MinusSeriesResult struct {
 	b Series
 }
 
-// Minus two series, result[i] = a[i] - b[i]
-func Minus(a interface{}, b interface{}) SeriesExtend {
+// Sub two series, result[i] = a[i] - b[i]
+func Sub(a interface{}, b interface{}) SeriesExtend {
 	aa := switchIface(a)
 	bb := switchIface(b)
 	return NewSeries(&MinusSeriesResult{aa, bb})
@@ -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/indicator_test.go

@@ -1,7 +1,7 @@
 package types
 
 import (
-	//"os"
+	// "os"
 	"math"
 	"testing"
 	"time"
@@ -22,7 +22,7 @@ func TestQueue(t *testing.T) {
 }
 
 func TestFloat(t *testing.T) {
-	var a Series = Minus(3., 2.)
+	var a Series = Sub(3., 2.)
 	assert.Equal(t, a.Last(), 1.)
 	assert.Equal(t, a.Index(100), 1.)
 }
@@ -44,7 +44,7 @@ func TestNextCross(t *testing.T) {
 func TestFloat64Slice(t *testing.T) {
 	var a = floats.Slice{1.0, 2.0, 3.0}
 	var b = floats.Slice{1.0, 2.0, 3.0}
-	var c Series = Minus(&a, &b)
+	var c Series = Sub(&a, &b)
 	a = append(a, 4.0)
 	b = append(b, 3.0)
 	assert.Equal(t, c.Last(), 1.)
@@ -233,9 +233,9 @@ func TestPlot(t *testing.T) {
 	ct.Plot("test", &a, Time(time.Now()), 4)
 	assert.Equal(t, ct.Interval, Interval5m)
 	assert.Equal(t, ct.Series[0].(chart.TimeSeries).Len(), 4)
-	//f, _ := os.Create("output.png")
-	//defer f.Close()
-	//ct.Render(chart.PNG, f)
+	// f, _ := os.Create("output.png")
+	// defer f.Close()
+	// ct.Render(chart.PNG, f)
 }
 
 func TestFilter(t *testing.T) {

pkg/types/seriesbase_imp.go

@@ -2,6 +2,13 @@ package types
 
 import "github.com/c9s/bbgo/pkg/datatype/floats"
 
+// SeriesBase is a wrapper of the Series interface
+// You can assign a data container that implements the Series interface
+// And this SeriesBase struct provides the implemented methods for manipulating your data
+type SeriesBase struct {
+	Series
+}
+
 func (s *SeriesBase) Index(i int) float64 {
 	if s.Series == nil {
 		return 0
@@ -64,7 +71,7 @@ func (s *SeriesBase) Add(b interface{}) SeriesExtend {
 }
 
 func (s *SeriesBase) Minus(b interface{}) SeriesExtend {
-	return Minus(s, b)
+	return Sub(s, b)
 }
 
 func (s *SeriesBase) Div(b interface{}) SeriesExtend {

pkg/types/trade_stats.go

@@ -126,7 +126,7 @@ func (s *IntervalProfitCollector) GetSharpe() float64 {
 	if s.Profits == nil {
 		panic("profits array empty. Did you create IntervalProfitCollector instance using NewIntervalProfitCollector?")
 	}
-	return Sharpe(Minus(s.Profits, 1.), s.Profits.Length(), true, false)
+	return Sharpe(Sub(s.Profits, 1.), s.Profits.Length(), true, false)
 }
 
 // Get sortino value with the interval of profit collected.
@@ -138,11 +138,11 @@ func (s *IntervalProfitCollector) GetSortino() float64 {
 	if s.Profits == nil {
 		panic("profits array empty. Did you create IntervalProfitCollector instance using NewIntervalProfitCollector?")
 	}
-	return Sortino(Minus(s.Profits, 1.), 0., s.Profits.Length(), true, false)
+	return Sortino(Sub(s.Profits, 1.), 0., s.Profits.Length(), true, false)
 }
 
 func (s *IntervalProfitCollector) GetOmega() float64 {
-	return Omega(Minus(s.Profits, 1.))
+	return Omega(Sub(s.Profits, 1.))
 }
 
 func (s IntervalProfitCollector) MarshalYAML() (interface{}, error) {