Merge pull request #845 from c9s/refactor/standard-indicator

refactor: refactor standard indicator and add simple indicator interface

pkg/bbgo/exit_lower_shadow_take_profit.go

@@ -27,7 +27,7 @@ func (s *LowerShadowTakeProfit) Bind(session *ExchangeSession, orderExecutor *Ge
 	s.session = session
 	s.orderExecutor = orderExecutor
 
-	stdIndicatorSet, _ := session.StandardIndicatorSet(s.Symbol)
+	stdIndicatorSet := session.StandardIndicatorSet(s.Symbol)
 	ewma := stdIndicatorSet.EWMA(s.IntervalWindow)
 
 

pkg/bbgo/session.go

@@ -438,17 +438,16 @@ func (session *ExchangeSession) initSymbol(ctx context.Context, environ *Environ
 	return nil
 }
 
-func (session *ExchangeSession) StandardIndicatorSet(symbol string) (*StandardIndicatorSet, bool) {
+func (session *ExchangeSession) StandardIndicatorSet(symbol string) *StandardIndicatorSet {
 	set, ok := session.standardIndicatorSets[symbol]
-	if !ok {
+	if ok {
-		if store, ok2 := session.MarketDataStore(symbol); ok2 {
+		return set
-			set = NewStandardIndicatorSet(symbol, session.MarketDataStream, store)
-			session.standardIndicatorSets[symbol] = set
-			return set, true
-		}
 	}
 
-	return set, ok
+	store, _ := session.MarketDataStore(symbol)
+	set = NewStandardIndicatorSet(symbol, session.MarketDataStream, store)
+	session.standardIndicatorSets[symbol] = set
+	return set
 }
 
 func (session *ExchangeSession) Position(symbol string) (pos *types.Position, ok bool) {

pkg/bbgo/standard_indicator_set.go

@@ -9,27 +9,23 @@ import (
 )
 
 var (
-	debugEWMA = false
-	debugSMA  = false
 	debugBOLL = false
 )
 
 func init() {
 	// when using --dotenv option, the dotenv is loaded from command.PersistentPreRunE, not init.
 	// hence here the env var won't enable the debug flag
-	util.SetEnvVarBool("DEBUG_EWMA", &debugEWMA)
-	util.SetEnvVarBool("DEBUG_SMA", &debugSMA)
 	util.SetEnvVarBool("DEBUG_BOLL", &debugBOLL)
 }
 
 type StandardIndicatorSet struct {
 	Symbol string
+
 	// Standard indicators
 	// interval -> window
-	sma   map[types.IntervalWindow]*indicator.SMA
-	ewma  map[types.IntervalWindow]*indicator.EWMA
 	boll    map[types.IntervalWindowBandWidth]*indicator.BOLL
 	stoch   map[types.IntervalWindow]*indicator.STOCH
+	simples map[types.IntervalWindow]indicator.Simple
 
 	stream types.Stream
 	store  *MarketDataStore
@@ -38,96 +34,104 @@ type StandardIndicatorSet struct {
 func NewStandardIndicatorSet(symbol string, stream types.Stream, store *MarketDataStore) *StandardIndicatorSet {
 	return &StandardIndicatorSet{
 		Symbol:  symbol,
-		sma:    make(map[types.IntervalWindow]*indicator.SMA),
-		ewma:   make(map[types.IntervalWindow]*indicator.EWMA),
-		boll:   make(map[types.IntervalWindowBandWidth]*indicator.BOLL),
-		stoch:  make(map[types.IntervalWindow]*indicator.STOCH),
 		store:   store,
 		stream:  stream,
+		simples: make(map[types.IntervalWindow]indicator.Simple),
+
+		boll:  make(map[types.IntervalWindowBandWidth]*indicator.BOLL),
+		stoch: make(map[types.IntervalWindow]*indicator.STOCH),
 	}
 }
 
+func (s *StandardIndicatorSet) initAndBind(inc indicator.KLinePusher, iw types.IntervalWindow) {
+	if klines, ok := s.store.KLinesOfInterval(iw.Interval); ok {
+		for _, k := range *klines {
+			inc.PushK(k)
+		}
+	}
+
+	s.stream.OnKLineClosed(types.KLineWith(s.Symbol, iw.Interval, inc.PushK))
+}
+
+func (s *StandardIndicatorSet) allocateSimpleIndicator(t indicator.Simple, iw types.IntervalWindow) indicator.Simple {
+	inc, ok := s.simples[iw]
+	if ok {
+		return inc
+	}
+
+	inc = t
+	s.initAndBind(inc, iw)
+	s.simples[iw] = inc
+	return t
+}
+
+// SMA is a helper function that returns the simple moving average indicator of the given interval and the window size.
+func (s *StandardIndicatorSet) SMA(iw types.IntervalWindow) *indicator.SMA {
+	inc := s.allocateSimpleIndicator(&indicator.SMA{IntervalWindow: iw}, iw)
+	return inc.(*indicator.SMA)
+}
+
+// EWMA is a helper function that returns the exponential weighed moving average indicator of the given interval and the window size.
+func (s *StandardIndicatorSet) EWMA(iw types.IntervalWindow) *indicator.EWMA {
+	inc := s.allocateSimpleIndicator(&indicator.EWMA{IntervalWindow: iw}, iw)
+	return inc.(*indicator.EWMA)
+}
+
+func (s *StandardIndicatorSet) PivotLow(iw types.IntervalWindow) *indicator.PivotLow {
+	inc := s.allocateSimpleIndicator(&indicator.PivotLow{IntervalWindow: iw}, iw)
+	return inc.(*indicator.PivotLow)
+}
+
+func (s *StandardIndicatorSet) ATR(iw types.IntervalWindow) *indicator.ATR {
+	inc := s.allocateSimpleIndicator(&indicator.ATR{IntervalWindow: iw}, iw)
+	return inc.(*indicator.ATR)
+}
+
+func (s *StandardIndicatorSet) ATRP(iw types.IntervalWindow) *indicator.ATRP {
+	inc := s.allocateSimpleIndicator(&indicator.ATRP{IntervalWindow: iw}, iw)
+	return inc.(*indicator.ATRP)
+}
+
+func (s *StandardIndicatorSet) EMV(iw types.IntervalWindow) *indicator.EMV {
+	inc := s.allocateSimpleIndicator(&indicator.EMV{IntervalWindow: iw}, iw)
+	return inc.(*indicator.EMV)
+}
+
+func (s *StandardIndicatorSet) CCI(iw types.IntervalWindow) *indicator.CCI {
+	inc := s.allocateSimpleIndicator(&indicator.CCI{IntervalWindow: iw}, iw)
+	return inc.(*indicator.CCI)
+}
+
+func (s *StandardIndicatorSet) HULL(iw types.IntervalWindow) *indicator.HULL {
+	inc := s.allocateSimpleIndicator(&indicator.HULL{IntervalWindow: iw}, iw)
+	return inc.(*indicator.HULL)
+}
+
+func (s *StandardIndicatorSet) STOCH(iw types.IntervalWindow) *indicator.STOCH {
+	inc, ok := s.stoch[iw]
+	if !ok {
+		inc = &indicator.STOCH{IntervalWindow: iw}
+		s.initAndBind(inc, iw)
+		s.stoch[iw] = inc
+	}
+
+	return inc
+}
+
 // BOLL returns the bollinger band indicator of the given interval, the window and bandwidth
-func (set *StandardIndicatorSet) BOLL(iw types.IntervalWindow, bandWidth float64) *indicator.BOLL {
+func (s *StandardIndicatorSet) BOLL(iw types.IntervalWindow, bandWidth float64) *indicator.BOLL {
 	iwb := types.IntervalWindowBandWidth{IntervalWindow: iw, BandWidth: bandWidth}
-	inc, ok := set.boll[iwb]
+	inc, ok := s.boll[iwb]
 	if !ok {
 		inc = &indicator.BOLL{IntervalWindow: iw, K: bandWidth}
-
+		s.initAndBind(inc, iw)
-		if klines, ok := set.store.KLinesOfInterval(iw.Interval); ok {
-			inc.LoadK(*klines)
-		}
 
 		if debugBOLL {
 			inc.OnUpdate(func(sma float64, upBand float64, downBand float64) {
-				logrus.Infof("%s BOLL %s: sma=%f up=%f down=%f", set.Symbol, iw.String(), sma, upBand, downBand)
+				logrus.Infof("%s BOLL %s: sma=%f up=%f down=%f", s.Symbol, iw.String(), sma, upBand, downBand)
 			})
 		}
-
+		s.boll[iwb] = inc
-		inc.BindK(set.stream, set.Symbol, iw.Interval)
-		set.boll[iwb] = inc
-	}
-
-	return inc
-}
-
-// SMA returns the simple moving average indicator of the given interval and the window size.
-func (set *StandardIndicatorSet) SMA(iw types.IntervalWindow) *indicator.SMA {
-	inc, ok := set.sma[iw]
-	if !ok {
-		inc = &indicator.SMA{IntervalWindow: iw}
-
-		if klines, ok := set.store.KLinesOfInterval(iw.Interval); ok {
-			inc.LoadK(*klines)
-		}
-
-		if debugSMA {
-			inc.OnUpdate(func(value float64) {
-				logrus.Infof("%s SMA %s: %f", set.Symbol, iw.String(), value)
-			})
-		}
-
-		inc.BindK(set.stream, set.Symbol, iw.Interval)
-		set.sma[iw] = inc
-	}
-
-	return inc
-}
-
-// EWMA returns the exponential weighed moving average indicator of the given interval and the window size.
-func (set *StandardIndicatorSet) EWMA(iw types.IntervalWindow) *indicator.EWMA {
-	inc, ok := set.ewma[iw]
-	if !ok {
-		inc = &indicator.EWMA{IntervalWindow: iw}
-
-		if klines, ok := set.store.KLinesOfInterval(iw.Interval); ok {
-			inc.LoadK(*klines)
-		}
-
-		if debugEWMA {
-			inc.OnUpdate(func(value float64) {
-				logrus.Infof("%s EWMA %s: value=%f", set.Symbol, iw.String(), value)
-			})
-		}
-
-		inc.BindK(set.stream, set.Symbol, iw.Interval)
-		set.ewma[iw] = inc
-	}
-
-	return inc
-}
-
-func (set *StandardIndicatorSet) STOCH(iw types.IntervalWindow) *indicator.STOCH {
-	inc, ok := set.stoch[iw]
-	if !ok {
-		inc = &indicator.STOCH{IntervalWindow: iw}
-
-		if klines, ok := set.store.KLinesOfInterval(iw.Interval); ok {
-			inc.LoadK(*klines)
-		}
-
-		inc.BindK(set.stream, set.Symbol, iw.Interval)
-		set.stoch[iw] = inc
 	}
 
 	return inc

pkg/bbgo/trader.go

@@ -292,7 +292,7 @@ func (trader *Trader) injectFields() error {
 					return fmt.Errorf("market of symbol %s not found", symbol)
 				}
 
-				indicatorSet, ok := session.StandardIndicatorSet(symbol)
+				indicatorSet := session.StandardIndicatorSet(symbol)
 				if !ok {
 					return fmt.Errorf("standardIndicatorSet of symbol %s not found", symbol)
 				}

pkg/indicator/atr.go

@@ -87,31 +87,3 @@ func (inc *ATR) PushK(k types.KLine) {
 	inc.EndTime = k.EndTime.Time()
 	inc.EmitUpdate(inc.Last())
 }
-
-func (inc *ATR) LoadK(allKlines []types.KLine) {
-	for _, k := range allKlines {
-		inc.PushK(k)
-	}
-}
-
-func (inc *ATR) BindK(target KLineClosedEmitter, symbol string, interval types.Interval) {
-	target.OnKLineClosed(types.KLineWith(symbol, interval, inc.PushK))
-}
-
-func (inc *ATR) CalculateAndUpdate(kLines []types.KLine) {
-	for _, k := range kLines {
-		inc.PushK(k)
-	}
-}
-
-func (inc *ATR) handleKLineWindowUpdate(interval types.Interval, window types.KLineWindow) {
-	if inc.Interval != interval {
-		return
-	}
-
-	inc.CalculateAndUpdate(window)
-}
-
-func (inc *ATR) Bind(updater KLineWindowUpdater) {
-	updater.OnKLineWindowUpdate(inc.handleKLineWindowUpdate)
-}

pkg/indicator/atr_test.go

@@ -61,7 +61,10 @@ func Test_calculateATR(t *testing.T) {
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
 			atr := &ATR{IntervalWindow: types.IntervalWindow{Window: tt.window}}
-			atr.CalculateAndUpdate(tt.kLines)
+			for _, k := range tt.kLines {
+				atr.PushK(k)
+			}
+
 			got := atr.Last()
 			diff := math.Trunc((got-tt.want)*100) / 100
 			if diff != 0 {

pkg/indicator/cci.go

@@ -78,7 +78,6 @@ func (inc *CCI) Length() int {
 
 var _ types.SeriesExtend = &CCI{}
 
-
 func (inc *CCI) PushK(k types.KLine) {
 	inc.Update(k.High.Add(k.Low).Add(k.Close).Div(three).Float64())
 }

pkg/indicator/emv.go

@@ -11,6 +11,7 @@ import (
 type EMV struct {
 	types.SeriesBase
 	types.IntervalWindow
+
 	prevH    float64
 	prevL    float64
 	Values   *SMA
@@ -25,6 +26,7 @@ func (inc *EMV) Update(high, low, vol float64) {
 	if inc.EMVScale == 0 {
 		inc.EMVScale = DefaultEMVScale
 	}
+
 	if inc.prevH == 0 || inc.Values == nil {
 		inc.SeriesBase.Series = inc
 		inc.prevH = high
@@ -32,6 +34,7 @@ func (inc *EMV) Update(high, low, vol float64) {
 		inc.Values = &SMA{IntervalWindow: inc.IntervalWindow}
 		return
 	}
+
 	distanceMoved := (high+low)/2. - (inc.prevH+inc.prevL)/2.
 	boxRatio := vol / inc.EMVScale / (high - low)
 	result := distanceMoved / boxRatio
@@ -66,29 +69,3 @@ var _ types.SeriesExtend = &EMV{}
 func (inc *EMV) PushK(k types.KLine) {
 	inc.Update(k.High.Float64(), k.Low.Float64(), k.Volume.Float64())
 }
-
-func (inc *EMV) CalculateAndUpdate(allKLines []types.KLine) {
-	if inc.Values == nil {
-		for _, k := range allKLines {
-			inc.PushK(k)
-			if inc.Length() > 0 {
-				inc.EmitUpdate(inc.Last())
-			}
-		}
-	} else {
-		k := allKLines[len(allKLines)-1]
-		inc.PushK(k)
-		inc.EmitUpdate(inc.Last())
-	}
-}
-
-func (inc *EMV) handleKLineWindowUpdate(interval types.Interval, window types.KLineWindow) {
-	if inc.Interval != interval {
-		return
-	}
-	inc.CalculateAndUpdate(window)
-}
-
-func (inc *EMV) Bind(updater KLineWindowUpdater) {
-	updater.OnKLineWindowUpdate(inc.handleKLineWindowUpdate)
-}

pkg/indicator/ewma.go

@@ -58,35 +58,6 @@ func (inc *EWMA) Length() int {
 	return len(inc.Values)
 }
 
-func (inc *EWMA) CalculateAndUpdate(allKLines []types.KLine) {
-	if len(inc.Values) == 0 {
-		for _, k := range allKLines {
-			inc.PushK(k)
-		}
-		inc.EmitUpdate(inc.Last())
-	} else {
-		k := allKLines[len(allKLines)-1]
-		inc.PushK(k)
-		inc.EmitUpdate(inc.Last())
-	}
-}
-
-func (inc *EWMA) handleKLineWindowUpdate(interval types.Interval, window types.KLineWindow) {
-	if inc.Interval != interval {
-		return
-	}
-
-	inc.CalculateAndUpdate(window)
-}
-
-func (inc *EWMA) Bind(updater KLineWindowUpdater) {
-	updater.OnKLineWindowUpdate(inc.handleKLineWindowUpdate)
-}
-
-func (inc *EWMA) BindK(target KLineClosedEmitter, symbol string, interval types.Interval) {
-	target.OnKLineClosed(types.KLineWith(symbol, interval, inc.PushK))
-}
-
 func (inc *EWMA) PushK(k types.KLine) {
 	if inc.EndTime != zeroTime && k.EndTime.Before(inc.EndTime) {
 		return
@@ -97,13 +68,6 @@ func (inc *EWMA) PushK(k types.KLine) {
 	inc.EmitUpdate(inc.Last())
 }
 
-func (inc *EWMA) LoadK(allKLines []types.KLine) {
-	for _, k := range allKLines {
-		inc.PushK(k)
-	}
-	inc.EmitUpdate(inc.Last())
-}
-
 func CalculateKLinesEMA(allKLines []types.KLine, priceF KLinePriceMapper, window int) float64 {
 	var multiplier = 2.0 / (float64(window) + 1)
 	return ewma(MapKLinePrice(allKLines, priceF), multiplier)

pkg/indicator/hull.go

@@ -19,6 +19,8 @@ type HULL struct {
 	updateCallbacks []func(value float64)
 }
 
+var _ types.SeriesExtend = &HULL{}
+
 func (inc *HULL) Update(value float64) {
 	if inc.result == nil {
 		inc.SeriesBase.Series = inc
@@ -52,33 +54,11 @@ func (inc *HULL) Length() int {
 	return inc.result.Length()
 }
 
-var _ types.SeriesExtend = &HULL{}
+func (inc *HULL) PushK(k types.KLine) {
-
+	if inc.ma1 != nil && inc.ma1.Length() > 0 && k.EndTime.Before(inc.ma1.EndTime) {
-// TODO: should we just ignore the possible overlapping?
-func (inc *HULL) CalculateAndUpdate(allKLines []types.KLine) {
-	doable := false
-	if inc.ma1 == nil || inc.ma1.Length() == 0 {
-		doable = true
-	}
-	for _, k := range allKLines {
-		if !doable && k.EndTime.After(inc.ma1.EndTime) {
-			doable = true
-		}
-		if doable {
-			inc.Update(k.Close.Float64())
-			inc.EmitUpdate(inc.Last())
-		}
-	}
-}
-
-func (inc *HULL) handleKLineWindowUpdate(interval types.Interval, window types.KLineWindow) {
-	if inc.Interval != interval {
 		return
 	}
 
-	inc.CalculateAndUpdate(window)
+	inc.Update(k.Close.Float64())
-}
+	inc.EmitUpdate(inc.Last())
-
-func (inc *HULL) Bind(updater KLineWindowUpdater) {
-	updater.OnKLineWindowUpdate(inc.handleKLineWindowUpdate)
 }

pkg/indicator/hull_test.go

@@ -4,9 +4,10 @@ import (
 	"encoding/json"
 	"testing"
 
+	"github.com/stretchr/testify/assert"
+
 	"github.com/c9s/bbgo/pkg/fixedpoint"
 	"github.com/c9s/bbgo/pkg/types"
-	"github.com/stretchr/testify/assert"
 )
 
 /*
@@ -26,6 +27,7 @@ func Test_HULL(t *testing.T) {
 	if err := json.Unmarshal(randomPrices, &input); err != nil {
 		panic(err)
 	}
+
 	tests := []struct {
 		name   string
 		kLines []types.KLine
@@ -44,8 +46,11 @@ func Test_HULL(t *testing.T) {
 
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			hull := HULL{IntervalWindow: types.IntervalWindow{Window: 16}}
+			hull := &HULL{IntervalWindow: types.IntervalWindow{Window: 16}}
-			hull.CalculateAndUpdate(tt.kLines)
+			for _, k := range tt.kLines {
+				hull.PushK(k)
+			}
+
 			last := hull.Last()
 			assert.InDelta(t, tt.want, last, Delta)
 			assert.InDelta(t, tt.next, hull.Index(1), Delta)

pkg/indicator/interface.go

@@ -22,10 +22,11 @@ type KLinePusher interface {
 	PushK(k types.KLine)
 }
 
-// KLineLoader provides an interface for API user to load history klines to the indicator.
+// Simple is the simple indicator that only returns one float64 value
-// The indicator implements its own way to calculate the values from the given history kline array.
+type Simple interface {
-type KLineLoader interface {
+	KLinePusher
-	LoadK(allKLines []types.KLine)
+	Last() float64
+	OnUpdate(f func(value float64))
 }
 
 type KLineCalculateUpdater interface {

pkg/indicator/low.go

@@ -0,0 +1,42 @@
+package indicator
+
+import (
+	"time"
+
+	"github.com/c9s/bbgo/pkg/types"
+)
+
+//go:generate callbackgen -type Low
+type Low struct {
+	types.IntervalWindow
+	types.SeriesBase
+
+	Values  types.Float64Slice
+	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())
+}
+
+func (inc *Low) LoadK(allKLines []types.KLine) {
+	for _, k := range allKLines {
+		inc.PushK(k)
+	}
+}

pkg/indicator/low_callbacks.go

@@ -0,0 +1,15 @@
+// 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/pivot_low.go

@@ -0,0 +1,74 @@
+package indicator
+
+import (
+	"fmt"
+	"time"
+
+	log "github.com/sirupsen/logrus"
+
+	"github.com/c9s/bbgo/pkg/types"
+)
+
+//go:generate callbackgen -type PivotLow
+type PivotLow struct {
+	types.IntervalWindow
+	types.SeriesBase
+
+	Lows    types.Float64Slice
+	Values  types.Float64Slice
+	EndTime time.Time
+
+	updateCallbacks []func(value float64)
+}
+
+func (inc *PivotLow) Update(value float64) {
+	if len(inc.Lows) == 0 {
+		inc.SeriesBase.Series = inc
+	}
+
+	inc.Lows.Push(value)
+
+	if len(inc.Lows) < inc.Window {
+		return
+	}
+
+	low, err := calculatePivotLow(inc.Lows, inc.Window)
+	if err != nil {
+		log.WithError(err).Errorf("can not calculate pivot low")
+		return
+	}
+
+	if low > 0.0 {
+		inc.Values.Push(low)
+	}
+}
+
+func (inc *PivotLow) 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())
+}
+
+
+func calculatePivotLow(lows types.Float64Slice, window int) (float64, error) {
+	length := len(lows)
+	if length == 0 || length < window {
+		return 0., fmt.Errorf("insufficient elements for calculating with window = %d", window)
+	}
+
+	var pv types.Float64Slice
+	for _, low := range lows {
+		pv.Push(low)
+	}
+
+	pl := 0.
+	if lows.Min() == lows.Index(int(window/2.)-1) {
+		pl = lows.Min()
+	}
+
+	return pl, nil
+}

pkg/indicator/pivotlow_callbacks.go

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

pkg/indicator/sma.go

@@ -76,28 +76,6 @@ func (inc *SMA) LoadK(allKLines []types.KLine) {
 	}
 }
 
-func (inc *SMA) CalculateAndUpdate(allKLines []types.KLine) {
-	if inc.rawValues == nil {
-		inc.LoadK(allKLines)
-	} else {
-		var last = allKLines[len(allKLines)-1]
-		inc.PushK(last)
-	}
-
-}
-
-func (inc *SMA) handleKLineWindowUpdate(interval types.Interval, window types.KLineWindow) {
-	if inc.Interval != interval {
-		return
-	}
-
-	inc.CalculateAndUpdate(window)
-}
-
-func (inc *SMA) Bind(updater KLineWindowUpdater) {
-	updater.OnKLineWindowUpdate(inc.handleKLineWindowUpdate)
-}
-
 func calculateSMA(kLines []types.KLine, window int, priceF KLinePriceMapper) (float64, error) {
 	length := len(kLines)
 	if length == 0 || length < window {

pkg/indicator/sma_test.go

@@ -4,9 +4,10 @@ import (
 	"encoding/json"
 	"testing"
 
+	"github.com/stretchr/testify/assert"
+
 	"github.com/c9s/bbgo/pkg/fixedpoint"
 	"github.com/c9s/bbgo/pkg/types"
-	"github.com/stretchr/testify/assert"
 )
 
 /*
@@ -52,7 +53,11 @@ func Test_SMA(t *testing.T) {
 			sma := SMA{
 				IntervalWindow: types.IntervalWindow{Window: 5},
 			}
-			sma.CalculateAndUpdate(tt.kLines)
+
+			for _, k := range tt.kLines {
+				sma.PushK(k)
+			}
+
 			assert.InDelta(t, tt.want, sma.Last(), Delta)
 			assert.InDelta(t, tt.next, sma.Index(1), Delta)
 			sma.Update(tt.update)

pkg/indicator/stoch.go

@@ -69,38 +69,6 @@ func (inc *STOCH) PushK(k types.KLine) {
 	inc.EmitUpdate(inc.LastK(), inc.LastD())
 }
 
-func (inc *STOCH) BindK(target KLineClosedEmitter, symbol string, interval types.Interval) {
-	target.OnKLineClosed(types.KLineWith(symbol, interval, inc.PushK))
-}
-
-func (inc *STOCH) LoadK(allKLines []types.KLine) {
-	for _, k := range allKLines {
-		inc.PushK(k)
-	}
-}
-
-func (inc *STOCH) CalculateAndUpdate(kLines []types.KLine) {
-	if len(kLines) < inc.Window || len(kLines) < DPeriod {
-		return
-	}
-
-	for _, k := range kLines {
-		inc.PushK(k)
-	}
-}
-
-func (inc *STOCH) handleKLineWindowUpdate(interval types.Interval, window types.KLineWindow) {
-	if inc.Interval != interval {
-		return
-	}
-
-	inc.CalculateAndUpdate(window)
-}
-
-func (inc *STOCH) Bind(updater KLineWindowUpdater) {
-	updater.OnKLineWindowUpdate(inc.handleKLineWindowUpdate)
-}
-
 func (inc *STOCH) GetD() types.Series {
 	return &inc.D
 }

pkg/indicator/stoch_test.go

@@ -56,7 +56,10 @@ func TestSTOCH_update(t *testing.T) {
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
 			kd := STOCH{IntervalWindow: types.IntervalWindow{Window: tt.window}}
-			kd.CalculateAndUpdate(tt.kLines)
+
+			for _, k := range tt.kLines {
+				kd.PushK(k)
+			}
 
 			got_k := kd.LastK()
 			diff_k := math.Trunc((got_k-tt.want_k)*100) / 100

pkg/strategy/emastop/strategy.go

@@ -25,7 +25,6 @@ func init() {
 }
 
 type Strategy struct {
-
 	SourceExchangeName string `json:"sourceExchange"`
 
 	TargetExchangeName string `json:"targetExchange"`
@@ -175,11 +174,7 @@ func (s *Strategy) handleOrderUpdate(order types.Order) {
 }
 
 func (s *Strategy) loadIndicator(sourceSession *bbgo.ExchangeSession) (types.Float64Indicator, error) {
-	var standardIndicatorSet, ok = sourceSession.StandardIndicatorSet(s.Symbol)
+	var standardIndicatorSet = sourceSession.StandardIndicatorSet(s.Symbol)
-	if !ok {
-		return nil, fmt.Errorf("standardIndicatorSet is nil, symbol %s", s.Symbol)
-	}
-
 	var iw = types.IntervalWindow{Interval: s.MovingAverageInterval, Window: s.MovingAverageWindow}
 
 	switch strings.ToUpper(s.MovingAverageType) {

pkg/strategy/funding/strategy.go

@@ -3,7 +3,6 @@ package funding
 import (
 	"context"
 	"errors"
-	"fmt"
 	"strings"
 
 	"github.com/sirupsen/logrus"
@@ -32,7 +31,7 @@ type Strategy struct {
 	Market              types.Market     `json:"-"`
 	Quantity            fixedpoint.Value `json:"quantity,omitempty"`
 	MaxExposurePosition fixedpoint.Value `json:"maxExposurePosition"`
-	//Interval            types.Interval   `json:"interval"`
+	// Interval            types.Interval   `json:"interval"`
 
 	FundingRate *struct {
 		High          fixedpoint.Value `json:"high"`
@@ -49,11 +48,11 @@ type Strategy struct {
 		// MovingAverageInterval is the interval of k-lines for the moving average indicator to calculate,
 		// it could be "1m", "5m", "1h" and so on.  note that, the moving averages are calculated from
 		// the k-line data we subscribed
-		//MovingAverageInterval types.Interval `json:"movingAverageInterval"`
+		// MovingAverageInterval types.Interval `json:"movingAverageInterval"`
 		//
-		//// MovingAverageWindow is the number of the window size of the moving average indicator.
+		// // MovingAverageWindow is the number of the window size of the moving average indicator.
-		//// The number of k-lines in the window. generally used window sizes are 7, 25 and 99 in the TradingView.
+		// // The number of k-lines in the window. generally used window sizes are 7, 25 and 99 in the TradingView.
-		//MovingAverageWindow int `json:"movingAverageWindow"`
+		// MovingAverageWindow int `json:"movingAverageWindow"`
 
 		MovingAverageIntervalWindow types.IntervalWindow `json:"movingAverageIntervalWindow"`
 
@@ -70,9 +69,9 @@ func (s *Strategy) ID() string {
 func (s *Strategy) Subscribe(session *bbgo.ExchangeSession) {
 	// session.Subscribe(types.BookChannel, s.Symbol, types.SubscribeOptions{})
 
-	//session.Subscribe(types.KLineChannel, s.Symbol, types.SubscribeOptions{
+	// session.Subscribe(types.KLineChannel, s.Symbol, types.SubscribeOptions{
 	//	Interval: string(s.Interval),
-	//})
+	// })
 
 	for _, detection := range s.SupportDetection {
 		session.Subscribe(types.KLineChannel, s.Symbol, types.SubscribeOptions{
@@ -93,23 +92,13 @@ func (s *Strategy) Validate() error {
 }
 
 func (s *Strategy) Run(ctx context.Context, orderExecutor bbgo.OrderExecutor, session *bbgo.ExchangeSession) error {
+	standardIndicatorSet := session.StandardIndicatorSet(s.Symbol)
 
-	standardIndicatorSet, ok := session.StandardIndicatorSet(s.Symbol)
-	if !ok {
-		return fmt.Errorf("standardIndicatorSet is nil, symbol %s", s.Symbol)
-	}
-	//binanceExchange, ok := session.Exchange.(*binance.Exchange)
-	//if !ok {
-	//	log.Error("exchange failed")
-	//}
 	if !session.Futures {
 		log.Error("futures not enabled in config for this strategy")
 		return nil
 	}
 
-	//if s.FundingRate != nil {
-	//	go s.listenToFundingRate(ctx, binanceExchange)
-	//}
 	premiumIndex, err := session.Exchange.(*binance.Exchange).QueryPremiumIndex(ctx, s.Symbol)
 	if err != nil {
 		log.Error("exchange does not support funding rate api")

pkg/strategy/pivotshort/breaklow.go

@@ -66,7 +66,7 @@ func (s *BreakLow) Bind(session *bbgo.ExchangeSession, orderExecutor *bbgo.Gener
 	position := orderExecutor.Position()
 	symbol := position.Symbol
 	store, _ := session.MarketDataStore(s.Symbol)
-	standardIndicator, _ := session.StandardIndicatorSet(s.Symbol)
+	standardIndicator := session.StandardIndicatorSet(s.Symbol)
 
 	s.lastLow = fixedpoint.Zero
 
@@ -120,17 +120,26 @@ func (s *BreakLow) Bind(session *bbgo.ExchangeSession, orderExecutor *bbgo.Gener
 	})
 
 	session.MarketDataStream.OnKLineClosed(types.KLineWith(symbol, s.Interval, func(kline types.KLine) {
+
 		lastLow := fixedpoint.NewFromFloat(s.pivot.LastLow())
 		if lastLow.IsZero() {
 			return
 		}
 
-		if lastLow.Compare(s.lastLow) != 0 {
+		if lastLow.Compare(s.lastLow) == 0 {
-			bbgo.Notify("%s new pivot low: %f", s.Symbol, s.pivot.LastLow())
+			return
 		}
 
 		s.lastLow = lastLow
 		s.pivotLowPrices = append(s.pivotLowPrices, s.lastLow)
+
+
+		// when position is opened, do not send pivot low notify
+		if position.IsOpened(kline.Close) {
+			return
+		}
+
+		bbgo.Notify("%s new pivot low: %f", s.Symbol, s.pivot.LastLow())
 	}))
 
 	session.MarketDataStream.OnKLineClosed(types.KLineWith(symbol, types.Interval1m, func(kline types.KLine) {

pkg/strategy/pricedrop/strategy.go

@@ -53,7 +53,7 @@ func (s *Strategy) Run(ctx context.Context, orderExecutor bbgo.OrderExecutor, se
 		return fmt.Errorf("market %s is not defined", s.Symbol)
 	}
 
-	standardIndicatorSet, ok := session.StandardIndicatorSet(s.Symbol)
+	standardIndicatorSet := session.StandardIndicatorSet(s.Symbol)
 	if !ok {
 		return fmt.Errorf("standardIndicatorSet is nil, symbol %s", s.Symbol)
 	}

pkg/strategy/skeleton/strategy.go

@@ -8,7 +8,6 @@ import (
 
 	"github.com/c9s/bbgo/pkg/bbgo"
 	"github.com/c9s/bbgo/pkg/fixedpoint"
-	"github.com/c9s/bbgo/pkg/indicator"
 	"github.com/c9s/bbgo/pkg/types"
 )
 
@@ -82,32 +81,11 @@ func (s *Strategy) Run(ctx context.Context, orderExecutor bbgo.OrderExecutor, se
 		s.State = &State{Counter: 1}
 	}
 
-	// Optional: You can get the market data store from session
+	indicators := session.StandardIndicatorSet(s.Symbol)
-	store, ok := session.MarketDataStore(s.Symbol)
+	atr := indicators.ATR(types.IntervalWindow{
-	if !ok {
-		return fmt.Errorf("market data store %s not found", s.Symbol)
-	}
-
-	// Initialize a custom indicator
-	atr := &indicator.ATR{
-		IntervalWindow: types.IntervalWindow{
 		Interval: types.Interval1m,
 		Window:   14,
-		},
+	})
-	}
-
-	// Bind the indicator to the market data store, so that when a new kline is received,
-	// the indicator will be updated.
-	atr.Bind(store)
-
-	// To get the past kline history, call KLinesOfInterval from the market data store
-	klines, ok := store.KLinesOfInterval(types.Interval1m)
-	if !ok {
-		return fmt.Errorf("market data store %s lkline not found", s.Symbol)
-	}
-
-	// Use the history data to initialize the indicator
-	atr.CalculateAndUpdate(*klines)
 
 	// To get the market information from the current session
 	// The market object provides the precision, MoQ (minimal of quantity) information

pkg/strategy/support/strategy.go

@@ -387,10 +387,7 @@ func (s *Strategy) Run(ctx context.Context, orderExecutor bbgo.OrderExecutor, se
 		log.Infof("adjusted minimal support volume to %s according to sensitivity %s", s.MinVolume.String(), s.Sensitivity.String())
 	}
 
-	standardIndicatorSet, ok := session.StandardIndicatorSet(s.Symbol)
+	standardIndicatorSet := session.StandardIndicatorSet(s.Symbol)
-	if !ok {
-		return fmt.Errorf("standardIndicatorSet is nil, symbol %s", s.Symbol)
-	}
 
 	if s.TriggerMovingAverage != zeroiw {
 		s.triggerEMA = standardIndicatorSet.EWMA(s.TriggerMovingAverage)

pkg/strategy/techsignal/strategy.go

@@ -3,7 +3,6 @@ package techsignal
 import (
 	"context"
 	"errors"
-	"fmt"
 	"strings"
 	"time"
 
@@ -145,10 +144,7 @@ func (s *Strategy) listenToFundingRate(ctx context.Context, exchange *binance.Ex
 }
 
 func (s *Strategy) Run(ctx context.Context, orderExecutor bbgo.OrderExecutor, session *bbgo.ExchangeSession) error {
-	standardIndicatorSet, ok := session.StandardIndicatorSet(s.Symbol)
+	standardIndicatorSet := session.StandardIndicatorSet(s.Symbol)
-	if !ok {
-		return fmt.Errorf("standardIndicatorSet is nil, symbol %s", s.Symbol)
-	}
 
 	if s.FundingRate != nil {
 		if binanceExchange, ok := session.Exchange.(*binance.Exchange); ok {

pkg/strategy/xmaker/strategy.go

@@ -681,7 +681,7 @@ func (s *Strategy) CrossRun(ctx context.Context, orderExecutionRouter bbgo.Order
 		return fmt.Errorf("maker session market %s is not defined", s.Symbol)
 	}
 
-	standardIndicatorSet, ok := s.sourceSession.StandardIndicatorSet(s.Symbol)
+	standardIndicatorSet := s.sourceSession.StandardIndicatorSet(s.Symbol)
 	if !ok {
 		return fmt.Errorf("%s standard indicator set not found", s.Symbol)
 	}