Merge pull request #1228 from c9s/c9s/move-v2-indicator

REFACTOR: move v2 indicators to the indicator/v2 package

pkg/bbgo/indicator_set.go

@@ -3,7 +3,7 @@ package bbgo
 import (
 	"github.com/sirupsen/logrus"
 
-	"github.com/c9s/bbgo/pkg/indicator"
+	"github.com/c9s/bbgo/pkg/indicator/v2"
 	"github.com/c9s/bbgo/pkg/types"
 )
 
@@ -16,8 +16,8 @@ type IndicatorSet struct {
 	store  *MarketDataStore
 
 	// caches
-	kLines      map[types.Interval]*indicator.KLineStream
+	kLines      map[types.Interval]*indicatorv2.KLineStream
-	closePrices map[types.Interval]*indicator.PriceStream
+	closePrices map[types.Interval]*indicatorv2.PriceStream
 }
 
 func NewIndicatorSet(symbol string, stream types.Stream, store *MarketDataStore) *IndicatorSet {
@@ -26,17 +26,17 @@ func NewIndicatorSet(symbol string, stream types.Stream, store *MarketDataStore)
 		store:  store,
 		stream: stream,
 
-		kLines:      make(map[types.Interval]*indicator.KLineStream),
+		kLines:      make(map[types.Interval]*indicatorv2.KLineStream),
-		closePrices: make(map[types.Interval]*indicator.PriceStream),
+		closePrices: make(map[types.Interval]*indicatorv2.PriceStream),
 	}
 }
 
-func (i *IndicatorSet) KLines(interval types.Interval) *indicator.KLineStream {
+func (i *IndicatorSet) KLines(interval types.Interval) *indicatorv2.KLineStream {
 	if kLines, ok := i.kLines[interval]; ok {
 		return kLines
 	}
 
-	kLines := indicator.KLines(i.stream, i.Symbol, interval)
+	kLines := indicatorv2.KLines(i.stream, i.Symbol, interval)
 	if kLinesWindow, ok := i.store.KLinesOfInterval(interval); ok {
 		kLines.BackFill(*kLinesWindow)
 	} else {
@@ -47,60 +47,60 @@ func (i *IndicatorSet) KLines(interval types.Interval) *indicator.KLineStream {
 	return kLines
 }
 
-func (i *IndicatorSet) OPEN(interval types.Interval) *indicator.PriceStream {
+func (i *IndicatorSet) OPEN(interval types.Interval) *indicatorv2.PriceStream {
-	return indicator.OpenPrices(i.KLines(interval))
+	return indicatorv2.OpenPrices(i.KLines(interval))
 }
 
-func (i *IndicatorSet) HIGH(interval types.Interval) *indicator.PriceStream {
+func (i *IndicatorSet) HIGH(interval types.Interval) *indicatorv2.PriceStream {
-	return indicator.HighPrices(i.KLines(interval))
+	return indicatorv2.HighPrices(i.KLines(interval))
 }
 
-func (i *IndicatorSet) LOW(interval types.Interval) *indicator.PriceStream {
+func (i *IndicatorSet) LOW(interval types.Interval) *indicatorv2.PriceStream {
-	return indicator.LowPrices(i.KLines(interval))
+	return indicatorv2.LowPrices(i.KLines(interval))
 }
 
-func (i *IndicatorSet) CLOSE(interval types.Interval) *indicator.PriceStream {
+func (i *IndicatorSet) CLOSE(interval types.Interval) *indicatorv2.PriceStream {
 	if closePrices, ok := i.closePrices[interval]; ok {
 		return closePrices
 	}
 
-	closePrices := indicator.ClosePrices(i.KLines(interval))
+	closePrices := indicatorv2.ClosePrices(i.KLines(interval))
 	i.closePrices[interval] = closePrices
 	return closePrices
 }
 
-func (i *IndicatorSet) VOLUME(interval types.Interval) *indicator.PriceStream {
+func (i *IndicatorSet) VOLUME(interval types.Interval) *indicatorv2.PriceStream {
-	return indicator.Volumes(i.KLines(interval))
+	return indicatorv2.Volumes(i.KLines(interval))
 }
 
-func (i *IndicatorSet) RSI(iw types.IntervalWindow) *indicator.RSIStream {
+func (i *IndicatorSet) RSI(iw types.IntervalWindow) *indicatorv2.RSIStream {
-	return indicator.RSI2(i.CLOSE(iw.Interval), iw.Window)
+	return indicatorv2.RSI2(i.CLOSE(iw.Interval), iw.Window)
 }
 
-func (i *IndicatorSet) EMA(iw types.IntervalWindow) *indicator.EWMAStream {
+func (i *IndicatorSet) EMA(iw types.IntervalWindow) *indicatorv2.EWMAStream {
 	return i.EWMA(iw)
 }
 
-func (i *IndicatorSet) EWMA(iw types.IntervalWindow) *indicator.EWMAStream {
+func (i *IndicatorSet) EWMA(iw types.IntervalWindow) *indicatorv2.EWMAStream {
-	return indicator.EWMA2(i.CLOSE(iw.Interval), iw.Window)
+	return indicatorv2.EWMA2(i.CLOSE(iw.Interval), iw.Window)
 }
 
-func (i *IndicatorSet) STOCH(iw types.IntervalWindow, dPeriod int) *indicator.StochStream {
+func (i *IndicatorSet) STOCH(iw types.IntervalWindow, dPeriod int) *indicatorv2.StochStream {
-	return indicator.Stoch2(i.KLines(iw.Interval), iw.Window, dPeriod)
+	return indicatorv2.Stoch(i.KLines(iw.Interval), iw.Window, dPeriod)
 }
 
-func (i *IndicatorSet) BOLL(iw types.IntervalWindow, k float64) *indicator.BOLLStream {
+func (i *IndicatorSet) BOLL(iw types.IntervalWindow, k float64) *indicatorv2.BOLLStream {
-	return indicator.BOLL2(i.CLOSE(iw.Interval), iw.Window, k)
+	return indicatorv2.BOLL(i.CLOSE(iw.Interval), iw.Window, k)
 }
 
-func (i *IndicatorSet) MACD(interval types.Interval, shortWindow, longWindow, signalWindow int) *indicator.MACDStream {
+func (i *IndicatorSet) MACD(interval types.Interval, shortWindow, longWindow, signalWindow int) *indicatorv2.MACDStream {
-	return indicator.MACD2(i.CLOSE(interval), shortWindow, longWindow, signalWindow)
+	return indicatorv2.MACD2(i.CLOSE(interval), shortWindow, longWindow, signalWindow)
 }
 
-func (i *IndicatorSet) ATR(interval types.Interval, window int) *indicator.ATRStream {
+func (i *IndicatorSet) ATR(interval types.Interval, window int) *indicatorv2.ATRStream {
-	return indicator.ATR2(i.KLines(interval), window)
+	return indicatorv2.ATR2(i.KLines(interval), window)
 }
 
-func (i *IndicatorSet) ATRP(interval types.Interval, window int) *indicator.ATRPStream {
+func (i *IndicatorSet) ATRP(interval types.Interval, window int) *indicatorv2.ATRPStream {
-	return indicator.ATRP2(i.KLines(interval), window)
+	return indicatorv2.ATRP2(i.KLines(interval), window)
 }

pkg/indicator/ewma.go

@@ -76,9 +76,9 @@ func (inc *EWMA) PushK(k types.KLine) {
 	inc.EmitUpdate(inc.Last(0))
 }
 
-func CalculateKLinesEMA(allKLines []types.KLine, priceF KLineValueMapper, window int) float64 {
+func CalculateKLinesEMA(allKLines []types.KLine, priceF types.KLineValueMapper, window int) float64 {
 	var multiplier = 2.0 / (float64(window) + 1)
-	return ewma(MapKLinePrice(allKLines, priceF), multiplier)
+	return ewma(types.MapKLinePrice(allKLines, priceF), multiplier)
 }
 
 // see https://www.investopedia.com/ask/answers/122314/what-exponential-moving-average-ema-formula-and-how-ema-calculated.asp

pkg/indicator/ewma_test.go

@@ -1027,7 +1027,7 @@ func buildKLines(prices []fixedpoint.Value) (klines []types.KLine) {
 func Test_calculateEWMA(t *testing.T) {
 	type args struct {
 		allKLines []types.KLine
-		priceF    KLineValueMapper
+		priceF    types.KLineValueMapper
 		window    int
 	}
 	var input []fixedpoint.Value
@@ -1043,7 +1043,7 @@ func Test_calculateEWMA(t *testing.T) {
 			name: "ETHUSDT EMA 7",
 			args: args{
 				allKLines: buildKLines(input),
-				priceF:    KLineClosePriceMapper,
+				priceF:    types.KLineClosePriceMapper,
 				window:    7,
 			},
 			want: 571.72, // with open price, binance desktop returns 571.45, trading view returns 570.8957, for close price, binance mobile returns 571.72
@@ -1052,7 +1052,7 @@ func Test_calculateEWMA(t *testing.T) {
 			name: "ETHUSDT EMA 25",
 			args: args{
 				allKLines: buildKLines(input),
-				priceF:    KLineClosePriceMapper,
+				priceF:    types.KLineClosePriceMapper,
 				window:    25,
 			},
 			want: 571.30,
@@ -1061,7 +1061,7 @@ func Test_calculateEWMA(t *testing.T) {
 			name: "ETHUSDT EMA 99",
 			args: args{
 				allKLines: buildKLines(input),
-				priceF:    KLineClosePriceMapper,
+				priceF:    types.KLineClosePriceMapper,
 				window:    99,
 			},
 			want: 577.62, // binance mobile uses 577.58

pkg/indicator/float64updater_callbacks.go

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

@@ -6070,7 +6070,7 @@ func Test_GHFilter(t *testing.T) {
 func Test_GHFilterEstimationAccurate(t *testing.T) {
 	type args struct {
 		allKLines []types.KLine
-		priceF    KLineValueMapper
+		priceF    types.KLineValueMapper
 		window    int
 	}
 	var klines []types.KLine

pkg/indicator/mapper.go

@@ -1,33 +0,0 @@
-package indicator
-
-import "github.com/c9s/bbgo/pkg/types"
-
-type KLineValueMapper func(k types.KLine) float64
-
-func KLineOpenPriceMapper(k types.KLine) float64 {
-	return k.Open.Float64()
-}
-
-func KLineClosePriceMapper(k types.KLine) float64 {
-	return k.Close.Float64()
-}
-
-func KLineTypicalPriceMapper(k types.KLine) float64 {
-	return (k.High.Float64() + k.Low.Float64() + k.Close.Float64()) / 3.
-}
-
-func KLinePriceVolumeMapper(k types.KLine) float64 {
-	return k.Close.Mul(k.Volume).Float64()
-}
-
-func KLineVolumeMapper(k types.KLine) float64 {
-	return k.Volume.Float64()
-}
-
-func MapKLinePrice(kLines []types.KLine, f KLineValueMapper) (prices []float64) {
-	for _, k := range kLines {
-		prices = append(prices, f(k))
-	}
-
-	return prices
-}

pkg/indicator/pivot.go

@@ -52,7 +52,7 @@ func (inc *Pivot) CalculateAndUpdate(klines []types.KLine) {
 
 	recentT := klines[end-(inc.Window-1) : end+1]
 
-	l, h, err := calculatePivot(recentT, inc.Window, KLineLowPriceMapper, KLineHighPriceMapper)
+	l, h, err := calculatePivot(recentT, inc.Window, types.KLineLowPriceMapper, types.KLineHighPriceMapper)
 	if err != nil {
 		log.WithError(err).Error("can not calculate pivots")
 		return
@@ -90,7 +90,7 @@ func (inc *Pivot) Bind(updater KLineWindowUpdater) {
 	updater.OnKLineWindowUpdate(inc.handleKLineWindowUpdate)
 }
 
-func calculatePivot(klines []types.KLine, window int, valLow KLineValueMapper, valHigh KLineValueMapper) (float64, float64, error) {
+func calculatePivot(klines []types.KLine, window int, valLow types.KLineValueMapper, valHigh types.KLineValueMapper) (float64, float64, error) {
 	length := len(klines)
 	if length == 0 || length < window {
 		return 0., 0., fmt.Errorf("insufficient elements for calculating with window = %d", window)
@@ -115,11 +115,3 @@ func calculatePivot(klines []types.KLine, window int, valLow KLineValueMapper, v
 
 	return pl, ph, nil
 }
-
-func KLineLowPriceMapper(k types.KLine) float64 {
-	return k.Low.Float64()
-}
-
-func KLineHighPriceMapper(k types.KLine) float64 {
-	return k.High.Float64()
-}

pkg/indicator/types.go

@@ -1,22 +0,0 @@
-package indicator
-
-import "github.com/c9s/bbgo/pkg/types"
-
-type Float64Calculator interface {
-	Calculate(x float64) float64
-	PushAndEmit(x float64)
-}
-
-type Float64Source interface {
-	types.Series
-	OnUpdate(f func(v float64))
-}
-
-type Float64Subscription interface {
-	types.Series
-	AddSubscriber(f func(v float64))
-}
-
-type Float64Truncator interface {
-	Truncate()
-}

pkg/indicator/util.go

@@ -7,7 +7,7 @@ func max(x, y int) int {
 	return y
 }
 
-func min(x, y int) int {
+func Min(x, y int) int {
 	if x < y {
 		return x
 	}

pkg/indicator/v2.go

@@ -1,19 +0,0 @@
-package indicator
-
-/*
-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)
-*/

pkg/indicator/v2/atr.go

@@ -1,4 +1,4 @@
-package indicator
+package indicatorv2
 
 type ATRStream struct {
 	// embedded struct

pkg/indicator/v2/atr_test.go

@@ -1,4 +1,4 @@
-package indicator
+package indicatorv2
 
 import (
 	"encoding/json"

pkg/indicator/v2/atrp.go

@@ -1,12 +1,14 @@
-package indicator
+package indicatorv2
+
+import "github.com/c9s/bbgo/pkg/types"
 
 type ATRPStream struct {
-	*Float64Series
+	*types.Float64Series
 }
 
 func ATRP2(source KLineSubscription, window int) *ATRPStream {
 	s := &ATRPStream{
-		Float64Series: NewFloat64Series(),
+		Float64Series: types.NewFloat64Series(),
 	}
 	tr := TR2(source)
 	atr := RMA2(tr, window, true)

pkg/indicator/v2/boll.go

@@ -1,10 +1,14 @@
-package indicator
+package indicatorv2
+
+import (
+	"github.com/c9s/bbgo/pkg/types"
+)
 
 type BOLLStream struct {
 	// the band series
-	*Float64Series
+	*types.Float64Series
 
-	UpBand, DownBand *Float64Series
+	UpBand, DownBand *types.Float64Series
 
 	window int
 	k      float64
@@ -20,15 +24,15 @@ type BOLLStream struct {
 //
 //	-> calculate SMA
 //	-> calculate stdDev -> calculate bandWidth -> get latest SMA -> upBand, downBand
-func BOLL2(source Float64Source, window int, k float64) *BOLLStream {
+func BOLL(source types.Float64Source, window int, k float64) *BOLLStream {
 	// bind these indicators before our main calculator
 	sma := SMA2(source, window)
-	stdDev := StdDev2(source, window)
+	stdDev := StdDev(source, window)
 
 	s := &BOLLStream{
-		Float64Series: NewFloat64Series(),
+		Float64Series: types.NewFloat64Series(),
-		UpBand:        NewFloat64Series(),
+		UpBand:        types.NewFloat64Series(),
-		DownBand:      NewFloat64Series(),
+		DownBand:      types.NewFloat64Series(),
 		window:        window,
 		k:             k,
 		SMA:           sma,

pkg/indicator/v2/cma.go

@@ -0,0 +1,28 @@
+package indicatorv2
+
+import (
+	"github.com/c9s/bbgo/pkg/indicator"
+	"github.com/c9s/bbgo/pkg/types"
+)
+
+type CMAStream struct {
+	*types.Float64Series
+}
+
+func CMA2(source types.Float64Source) *CMAStream {
+	s := &CMAStream{
+		Float64Series: types.NewFloat64Series(),
+	}
+	s.Bind(source, s)
+	return s
+}
+
+func (s *CMAStream) Calculate(x float64) float64 {
+	l := float64(s.Slice.Length())
+	cma := (s.Slice.Last(0)*l + x) / (l + 1.)
+	return cma
+}
+
+func (s *CMAStream) Truncate() {
+	s.Slice.Truncate(indicator.MaxNumOfEWMA)
+}

pkg/indicator/v2/cross.go

@@ -1,7 +1,8 @@
-package indicator
+package indicatorv2
 
 import (
 	"github.com/c9s/bbgo/pkg/datatype/floats"
+	"github.com/c9s/bbgo/pkg/types"
 )
 
 type CrossType float64
@@ -13,7 +14,7 @@ const (
 
 // CrossStream subscribes 2 upstreams, and calculate the cross signal
 type CrossStream struct {
-	*Float64Series
+	*types.Float64Series
 
 	a, b floats.Slice
 }
@@ -21,9 +22,9 @@ type CrossStream struct {
 // Cross creates the CrossStream object:
 //
 // cross := Cross(fastEWMA, slowEWMA)
-func Cross(a, b Float64Source) *CrossStream {
+func Cross(a, b types.Float64Source) *CrossStream {
 	s := &CrossStream{
-		Float64Series: NewFloat64Series(),
+		Float64Series: types.NewFloat64Series(),
 	}
 	a.OnUpdate(func(v float64) {
 		s.a.Push(v)

pkg/indicator/v2/ewma.go

@@ -1,15 +1,17 @@
-package indicator
+package indicatorv2
+
+import "github.com/c9s/bbgo/pkg/types"
 
 type EWMAStream struct {
-	*Float64Series
+	*types.Float64Series
 
 	window     int
 	multiplier float64
 }
 
-func EWMA2(source Float64Source, window int) *EWMAStream {
+func EWMA2(source types.Float64Source, window int) *EWMAStream {
 	s := &EWMAStream{
-		Float64Series: NewFloat64Series(),
+		Float64Series: types.NewFloat64Series(),
 		window:        window,
 		multiplier:    2.0 / float64(1+window),
 	}
@@ -18,7 +20,7 @@ func EWMA2(source Float64Source, window int) *EWMAStream {
 }
 
 func (s *EWMAStream) Calculate(v float64) float64 {
-	last := s.slice.Last(0)
+	last := s.Slice.Last(0)
 	if last == 0.0 {
 		return v
 	}

pkg/indicator/v2/klines.go

@@ -1,4 +1,4 @@
-package indicator
+package indicatorv2
 
 import "github.com/c9s/bbgo/pkg/types"
 
@@ -60,3 +60,9 @@ func KLines(source types.Stream, symbol string, interval types.Interval) *KLineS
 
 	return s
 }
+
+type KLineSubscription interface {
+	AddSubscriber(f func(k types.KLine))
+	Length() int
+	Last(i int) *types.KLine
+}

pkg/indicator/v2/klinestream_callbacks.go

@@ -1,6 +1,6 @@
 // Code generated by "callbackgen -type KLineStream"; DO NOT EDIT.
 
-package indicator
+package indicatorv2
 
 import (
 	"github.com/c9s/bbgo/pkg/types"

pkg/indicator/v2/macd.go

@@ -1,4 +1,8 @@
-package indicator
+package indicatorv2
+
+import (
+	"github.com/c9s/bbgo/pkg/types"
+)
 
 type MACDStream struct {
 	*SubtractStream
@@ -9,7 +13,7 @@ type MACDStream struct {
 	Histogram                  *SubtractStream
 }
 
-func MACD2(source Float64Source, shortWindow, longWindow, signalWindow int) *MACDStream {
+func MACD2(source types.Float64Source, shortWindow, longWindow, signalWindow int) *MACDStream {
 	// bind and calculate these first
 	fastEWMA := EWMA2(source, shortWindow)
 	slowEWMA := EWMA2(source, longWindow)

pkg/indicator/v2/macd_test.go

@@ -1,4 +1,4 @@
-package indicator
+package indicatorv2
 
 import (
 	"encoding/json"
@@ -21,6 +21,14 @@ fast = s.ewm(span=12, adjust=False).mean()
 print(fast - slow)
 */
 
+func buildKLines(prices []fixedpoint.Value) (klines []types.KLine) {
+	for _, p := range prices {
+		klines = append(klines, types.KLine{Close: p})
+	}
+
+	return klines
+}
+
 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

pkg/indicator/v2/multiply.go

@@ -1,15 +1,18 @@
-package indicator
+package indicatorv2
 
-import "github.com/c9s/bbgo/pkg/datatype/floats"
+import (
+	"github.com/c9s/bbgo/pkg/datatype/floats"
+	"github.com/c9s/bbgo/pkg/types"
+)
 
 type MultiplyStream struct {
-	*Float64Series
+	*types.Float64Series
 	a, b floats.Slice
 }
 
-func Multiply(a, b Float64Source) *MultiplyStream {
+func Multiply(a, b types.Float64Source) *MultiplyStream {
 	s := &MultiplyStream{
-		Float64Series: NewFloat64Series(),
+		Float64Series: types.NewFloat64Series(),
 	}
 
 	a.OnUpdate(func(v float64) {
@@ -29,13 +32,13 @@ func (s *MultiplyStream) calculate() {
 		return
 	}
 
-	if s.a.Length() > s.slice.Length() {
+	if s.a.Length() > s.Slice.Length() {
-		var numNewElems = 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.Mul(tailB)
 		for _, f := range tailC {
-			s.slice.Push(f)
+			s.Slice.Push(f)
 			s.EmitUpdate(f)
 		}
 	}

pkg/indicator/v2/pivothigh.go

@@ -0,0 +1,34 @@
+package indicatorv2
+
+import (
+	"github.com/c9s/bbgo/pkg/datatype/floats"
+	"github.com/c9s/bbgo/pkg/types"
+)
+
+type PivotHighStream struct {
+	*types.Float64Series
+	rawValues           floats.Slice
+	window, rightWindow int
+}
+
+func PivotHigh2(source types.Float64Source, window, rightWindow int) *PivotHighStream {
+	s := &PivotHighStream{
+		Float64Series: types.NewFloat64Series(),
+		window:        window,
+		rightWindow:   rightWindow,
+	}
+
+	s.Subscribe(source, func(x float64) {
+		s.rawValues.Push(x)
+		if low, ok := s.calculatePivotHigh(s.rawValues, s.window, s.rightWindow); ok {
+			s.PushAndEmit(low)
+		}
+	})
+	return s
+}
+
+func (s *PivotHighStream) calculatePivotHigh(highs floats.Slice, left, right int) (float64, bool) {
+	return floats.FindPivot(highs, left, right, func(a, pivot float64) bool {
+		return a < pivot
+	})
+}

pkg/indicator/v2/pivotlow.go

@@ -0,0 +1,34 @@
+package indicatorv2
+
+import (
+	"github.com/c9s/bbgo/pkg/datatype/floats"
+	"github.com/c9s/bbgo/pkg/types"
+)
+
+type PivotLowStream struct {
+	*types.Float64Series
+	rawValues           floats.Slice
+	window, rightWindow int
+}
+
+func PivotLow(source types.Float64Source, window, rightWindow int) *PivotLowStream {
+	s := &PivotLowStream{
+		Float64Series: types.NewFloat64Series(),
+		window:        window,
+		rightWindow:   rightWindow,
+	}
+
+	s.Subscribe(source, func(x float64) {
+		s.rawValues.Push(x)
+		if low, ok := s.calculatePivotLow(s.rawValues, s.window, s.rightWindow); ok {
+			s.PushAndEmit(low)
+		}
+	})
+	return s
+}
+
+func (s *PivotLowStream) calculatePivotLow(lows floats.Slice, left, right int) (float64, bool) {
+	return floats.FindPivot(lows, left, right, func(a, pivot float64) bool {
+		return a > pivot
+	})
+}

pkg/indicator/v2/price.go

@@ -1,24 +1,18 @@
-package indicator
+package indicatorv2
 
 import (
 	"github.com/c9s/bbgo/pkg/types"
 )
 
-type KLineSubscription interface {
-	AddSubscriber(f func(k types.KLine))
-	Length() int
-	Last(i int) *types.KLine
-}
-
 type PriceStream struct {
-	*Float64Series
+	*types.Float64Series
 
-	mapper KLineValueMapper
+	mapper types.KLineValueMapper
 }
 
-func Price(source KLineSubscription, mapper KLineValueMapper) *PriceStream {
+func Price(source KLineSubscription, mapper types.KLineValueMapper) *PriceStream {
 	s := &PriceStream{
-		Float64Series: NewFloat64Series(),
+		Float64Series: types.NewFloat64Series(),
 		mapper:        mapper,
 	}
 
@@ -37,37 +31,37 @@ func Price(source KLineSubscription, mapper KLineValueMapper) *PriceStream {
 func (s *PriceStream) AddSubscriber(f func(v float64)) {
 	s.OnUpdate(f)
 
-	if len(s.slice) == 0 {
+	if len(s.Slice) == 0 {
 		return
 	}
 
 	// push historical value to the subscriber
-	for _, v := range s.slice {
+	for _, v := range s.Slice {
 		f(v)
 	}
 }
 
 func (s *PriceStream) PushAndEmit(v float64) {
-	s.slice.Push(v)
+	s.Slice.Push(v)
 	s.EmitUpdate(v)
 }
 
 func ClosePrices(source KLineSubscription) *PriceStream {
-	return Price(source, KLineClosePriceMapper)
+	return Price(source, types.KLineClosePriceMapper)
 }
 
 func LowPrices(source KLineSubscription) *PriceStream {
-	return Price(source, KLineLowPriceMapper)
+	return Price(source, types.KLineLowPriceMapper)
 }
 
 func HighPrices(source KLineSubscription) *PriceStream {
-	return Price(source, KLineHighPriceMapper)
+	return Price(source, types.KLineHighPriceMapper)
 }
 
 func OpenPrices(source KLineSubscription) *PriceStream {
-	return Price(source, KLineOpenPriceMapper)
+	return Price(source, types.KLineOpenPriceMapper)
 }
 
 func Volumes(source KLineSubscription) *PriceStream {
-	return Price(source, KLineVolumeMapper)
+	return Price(source, types.KLineVolumeMapper)
 }

pkg/indicator/v2/rma.go

@@ -1,8 +1,15 @@
-package indicator
+package indicatorv2
+
+import (
+	"github.com/c9s/bbgo/pkg/types"
+)
+
+const MaxNumOfRMA = 1000
+const MaxNumOfRMATruncateSize = 500
 
 type RMAStream struct {
 	// embedded structs
-	*Float64Series
+	*types.Float64Series
 
 	// config fields
 	Adjust bool
@@ -12,9 +19,9 @@ type RMAStream struct {
 	sum, previous float64
 }
 
-func RMA2(source Float64Source, window int, adjust bool) *RMAStream {
+func RMA2(source types.Float64Source, window int, adjust bool) *RMAStream {
 	s := &RMAStream{
-		Float64Series: NewFloat64Series(),
+		Float64Series: types.NewFloat64Series(),
 		window:        window,
 		Adjust:        adjust,
 	}
@@ -41,17 +48,17 @@ func (s *RMAStream) Calculate(x float64) float64 {
 
 	if s.counter < s.window {
 		// we can use x, but we need to use 0. to make the same behavior as the result from python pandas_ta
-		s.slice.Push(0)
+		s.Slice.Push(0)
 	}
 
-	s.slice.Push(tmp)
+	s.Slice.Push(tmp)
 	s.previous = tmp
 
 	return tmp
 }
 
 func (s *RMAStream) Truncate() {
-	if len(s.slice) > MaxNumOfRMA {
+	if len(s.Slice) > MaxNumOfRMA {
-		s.slice = s.slice[MaxNumOfRMATruncateSize-1:]
+		s.Slice = s.Slice[MaxNumOfRMATruncateSize-1:]
 	}
 }

pkg/indicator/v2/rsi.go

@@ -1,20 +1,24 @@
-package indicator
+package indicatorv2
+
+import (
+	"github.com/c9s/bbgo/pkg/types"
+)
 
 type RSIStream struct {
 	// embedded structs
-	*Float64Series
+	*types.Float64Series
 
 	// config fields
 	window int
 
 	// private states
-	source Float64Source
+	source types.Float64Source
 }
 
-func RSI2(source Float64Source, window int) *RSIStream {
+func RSI2(source types.Float64Source, window int) *RSIStream {
 	s := &RSIStream{
 		source:        source,
-		Float64Series: NewFloat64Series(),
+		Float64Series: types.NewFloat64Series(),
 		window:        window,
 	}
 	s.Bind(source, s)
@@ -42,3 +46,17 @@ func (s *RSIStream) Calculate(_ float64) float64 {
 	rsi := 100.0 - (100.0 / (1.0 + rs))
 	return rsi
 }
+
+func max(x, y int) int {
+	if x > y {
+		return x
+	}
+	return y
+}
+
+func min(x, y int) int {
+	if x < y {
+		return x
+	}
+	return y
+}

pkg/indicator/v2/rsi_test.go

@@ -1,4 +1,4 @@
-package indicator
+package indicatorv2
 
 import (
 	"encoding/json"
@@ -75,11 +75,11 @@ func Test_RSI2(t *testing.T) {
 				prices.PushAndEmit(price)
 			}
 
-			assert.Equal(t, floats.Slice(tt.values), prices.slice)
+			assert.Equal(t, floats.Slice(tt.values), prices.Slice)
 
-			if assert.Equal(t, len(tt.want), len(rsi.slice)) {
+			if assert.Equal(t, len(tt.want), len(rsi.Slice)) {
 				for i, v := range tt.want {
-					assert.InDelta(t, v, rsi.slice[i], 0.000001, "Expected rsi.slice[%d] to be %v, but got %v", i, v, rsi.slice[i])
+					assert.InDelta(t, v, rsi.Slice[i], 0.000001, "Expected rsi.slice[%d] to be %v, but got %v", i, v, rsi.Slice[i])
 				}
 			}
 		})

pkg/indicator/v2/sma.go

@@ -1,16 +1,20 @@
-package indicator
+package indicatorv2
 
-import "github.com/c9s/bbgo/pkg/types"
+import (
+	"github.com/c9s/bbgo/pkg/types"
+)
+
+const MaxNumOfSMA = 5_000
 
 type SMAStream struct {
-	*Float64Series
+	*types.Float64Series
 	window    int
 	rawValues *types.Queue
 }
 
-func SMA2(source Float64Source, window int) *SMAStream {
+func SMA2(source types.Float64Source, window int) *SMAStream {
 	s := &SMAStream{
-		Float64Series: NewFloat64Series(),
+		Float64Series: types.NewFloat64Series(),
 		window:        window,
 		rawValues:     types.NewQueue(window),
 	}
@@ -25,5 +29,5 @@ func (s *SMAStream) Calculate(v float64) float64 {
 }
 
 func (s *SMAStream) Truncate() {
-	s.slice = s.slice.Truncate(MaxNumOfSMA)
+	s.Slice = s.Slice.Truncate(MaxNumOfSMA)
 }

pkg/indicator/v2/stddev.go

@@ -1,9 +1,9 @@
-package indicator
+package indicatorv2
 
 import "github.com/c9s/bbgo/pkg/types"
 
 type StdDevStream struct {
-	*Float64Series
+	*types.Float64Series
 
 	rawValues *types.Queue
 
@@ -11,9 +11,9 @@ type StdDevStream struct {
 	multiplier float64
 }
 
-func StdDev2(source Float64Source, window int) *StdDevStream {
+func StdDev(source types.Float64Source, window int) *StdDevStream {
 	s := &StdDevStream{
-		Float64Series: NewFloat64Series(),
+		Float64Series: types.NewFloat64Series(),
 		rawValues:     types.NewQueue(window),
 		window:        window,
 	}

pkg/indicator/v2/stoch.go

@@ -1,10 +1,12 @@
-package indicator
+package indicatorv2
 
 import (
 	"github.com/c9s/bbgo/pkg/datatype/floats"
 	"github.com/c9s/bbgo/pkg/types"
 )
 
+const DPeriod int = 3
+
 // Stochastic Oscillator
 // - https://www.investopedia.com/terms/s/stochasticoscillator.asp
 //
@@ -30,7 +32,7 @@ type StochStream struct {
 }
 
 // Stochastic Oscillator
-func Stoch2(source KLineSubscription, window, dPeriod int) *StochStream {
+func Stoch(source KLineSubscription, window, dPeriod int) *StochStream {
 	highPrices := HighPrices(source)
 	lowPrices := LowPrices(source)
 
@@ -42,8 +44,8 @@ func Stoch2(source KLineSubscription, window, dPeriod int) *StochStream {
 	}
 
 	source.AddSubscriber(func(kLine types.KLine) {
-		lowest := s.lowPrices.slice.Tail(s.window).Min()
+		lowest := s.lowPrices.Slice.Tail(s.window).Min()
-		highest := s.highPrices.slice.Tail(s.window).Max()
+		highest := s.highPrices.Slice.Tail(s.window).Max()
 
 		var k float64 = 50.0
 		var d float64 = 0.0
@@ -53,7 +55,7 @@ func Stoch2(source KLineSubscription, window, dPeriod int) *StochStream {
 		}
 
 		s.K.Push(k)
-	
+
 		d = s.K.Tail(s.dPeriod).Mean()
 		s.D.Push(d)
 		s.EmitUpdate(k, d)

pkg/indicator/v2/stoch_test.go

@@ -1,4 +1,4 @@
-package indicator
+package indicatorv2
 
 import (
 	"encoding/json"
@@ -58,7 +58,7 @@ func TestSTOCH2_update(t *testing.T) {
 		t.Run(tt.name, func(t *testing.T) {
 			stream := &types.StandardStream{}
 			kLines := KLines(stream, "", "")
-			kd := Stoch2(kLines, tt.window, DPeriod)
+			kd := Stoch(kLines, tt.window, DPeriod)
 
 			for _, k := range tt.kLines {
 				stream.EmitKLineClosed(k)

pkg/indicator/v2/stochstream_callbacks.go

@@ -1,6 +1,6 @@
 // Code generated by "callbackgen -type StochStream"; DO NOT EDIT.
 
-package indicator
+package indicatorv2
 
 import ()
 

pkg/indicator/v2/subtract.go

@@ -1,12 +1,13 @@
-package indicator
+package indicatorv2
 
 import (
 	"github.com/c9s/bbgo/pkg/datatype/floats"
+	"github.com/c9s/bbgo/pkg/types"
 )
 
 // SubtractStream subscribes 2 upstream data, and then subtract these 2 values
 type SubtractStream struct {
-	*Float64Series
+	*types.Float64Series
 
 	a, b floats.Slice
 	i    int
@@ -14,9 +15,9 @@ type SubtractStream struct {
 
 // Subtract creates the SubtractStream object
 // subtract := Subtract(longEWMA, shortEWMA)
-func Subtract(a, b Float64Source) *SubtractStream {
+func Subtract(a, b types.Float64Source) *SubtractStream {
 	s := &SubtractStream{
-		Float64Series: NewFloat64Series(),
+		Float64Series: types.NewFloat64Series(),
 	}
 
 	a.OnUpdate(func(v float64) {
@@ -35,13 +36,13 @@ func (s *SubtractStream) calculate() {
 		return
 	}
 
-	if s.a.Length() > s.slice.Length() {
+	if s.a.Length() > s.Slice.Length() {
-		var numNewElems = 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.slice.Push(f)
+			s.Slice.Push(f)
 			s.EmitUpdate(f)
 		}
 	}

pkg/indicator/v2/subtract_test.go

@@ -1,4 +1,4 @@
-package indicator
+package indicatorv2
 
 import (
 	"testing"
@@ -21,10 +21,10 @@ func Test_v2_Subtract(t *testing.T) {
 		stream.EmitKLineClosed(types.KLine{Close: fixedpoint.NewFromFloat(19_000.0 + i)})
 	}
 
-	t.Logf("fastEMA: %+v", fastEMA.slice)
+	t.Logf("fastEMA: %+v", fastEMA.Slice)
-	t.Logf("slowEMA: %+v", slowEMA.slice)
+	t.Logf("slowEMA: %+v", slowEMA.Slice)
 
 	assert.Equal(t, len(subtract.a), len(subtract.b))
-	assert.Equal(t, len(subtract.a), len(subtract.slice))
+	assert.Equal(t, len(subtract.a), len(subtract.Slice))
-	assert.InDelta(t, subtract.slice[0], subtract.a[0]-subtract.b[0], 0.0001)
+	assert.InDelta(t, subtract.Slice[0], subtract.a[0]-subtract.b[0], 0.0001)
 }

pkg/indicator/v2/tr.go

@@ -1,4 +1,4 @@
-package indicator
+package indicatorv2
 
 import (
 	"math"
@@ -9,7 +9,7 @@ import (
 // This TRStream calculates the ATR first
 type TRStream struct {
 	// embedded struct
-	*Float64Series
+	*types.Float64Series
 
 	// private states
 	previousClose float64
@@ -17,7 +17,7 @@ type TRStream struct {
 
 func TR2(source KLineSubscription) *TRStream {
 	s := &TRStream{
-		Float64Series: NewFloat64Series(),
+		Float64Series: types.NewFloat64Series(),
 	}
 
 	source.AddSubscriber(func(k types.KLine) {

pkg/indicator/v2/tr_test.go

@@ -1,4 +1,4 @@
-package indicator
+package indicatorv2
 
 import (
 	"encoding/json"

pkg/indicator/v2_cma.go

@@ -1,23 +0,0 @@
-package indicator
-
-type CMAStream struct {
-	*Float64Series
-}
-
-func CMA2(source Float64Source) *CMAStream {
-	s := &CMAStream{
-		Float64Series: NewFloat64Series(),
-	}
-	s.Bind(source, s)
-	return s
-}
-
-func (s *CMAStream) Calculate(x float64) float64 {
-	l := float64(s.slice.Length())
-	cma := (s.slice.Last(0)*l + x) / (l + 1.)
-	return cma
-}
-
-func (s *CMAStream) Truncate() {
-	s.slice.Truncate(MaxNumOfEWMA)
-}

pkg/indicator/v2_pivothigh.go

@@ -1,27 +0,0 @@
-package indicator
-
-import (
-	"github.com/c9s/bbgo/pkg/datatype/floats"
-)
-
-type PivotHighStream struct {
-	*Float64Series
-	rawValues           floats.Slice
-	window, rightWindow int
-}
-
-func PivotHigh2(source Float64Source, window, rightWindow int) *PivotHighStream {
-	s := &PivotHighStream{
-		Float64Series: NewFloat64Series(),
-		window:        window,
-		rightWindow:   rightWindow,
-	}
-
-	s.Subscribe(source, func(x float64) {
-		s.rawValues.Push(x)
-		if low, ok := calculatePivotHigh(s.rawValues, s.window, s.rightWindow); ok {
-			s.PushAndEmit(low)
-		}
-	})
-	return s
-}

pkg/indicator/v2_pivotlow.go

@@ -1,27 +0,0 @@
-package indicator
-
-import (
-	"github.com/c9s/bbgo/pkg/datatype/floats"
-)
-
-type PivotLowStream struct {
-	*Float64Series
-	rawValues           floats.Slice
-	window, rightWindow int
-}
-
-func PivotLow2(source Float64Source, window, rightWindow int) *PivotLowStream {
-	s := &PivotLowStream{
-		Float64Series: NewFloat64Series(),
-		window:        window,
-		rightWindow:   rightWindow,
-	}
-
-	s.Subscribe(source, func(x float64) {
-		s.rawValues.Push(x)
-		if low, ok := calculatePivotLow(s.rawValues, s.window, s.rightWindow); ok {
-			s.PushAndEmit(low)
-		}
-	})
-	return s
-}

pkg/indicator/volatility.go

@@ -58,7 +58,7 @@ func (inc *Volatility) CalculateAndUpdate(allKLines []types.KLine) {
 
 	var recentT = allKLines[end-(inc.Window-1) : end+1]
 
-	volatility, err := calculateVOLATILITY(recentT, inc.Window, KLineClosePriceMapper)
+	volatility, err := calculateVOLATILITY(recentT, inc.Window, types.KLineClosePriceMapper)
 	if err != nil {
 		log.WithError(err).Error("can not calculate volatility")
 		return
@@ -86,7 +86,7 @@ func (inc *Volatility) Bind(updater KLineWindowUpdater) {
 	updater.OnKLineWindowUpdate(inc.handleKLineWindowUpdate)
 }
 
-func calculateVOLATILITY(klines []types.KLine, window int, priceF KLineValueMapper) (float64, error) {
+func calculateVOLATILITY(klines []types.KLine, window int, priceF types.KLineValueMapper) (float64, error) {
 	length := len(klines)
 	if length == 0 || length < window {
 		return 0.0, fmt.Errorf("insufficient elements for calculating VOL with window = %d", window)

pkg/indicator/vwap.go

@@ -71,7 +71,7 @@ func (inc *VWAP) Length() int {
 var _ types.SeriesExtend = &VWAP{}
 
 func (inc *VWAP) PushK(k types.KLine) {
-	inc.Update(KLineTypicalPriceMapper(k), k.Volume.Float64())
+	inc.Update(types.KLineTypicalPriceMapper(k), k.Volume.Float64())
 }
 
 func (inc *VWAP) CalculateAndUpdate(allKLines []types.KLine) {
@@ -99,7 +99,7 @@ func (inc *VWAP) Bind(updater KLineWindowUpdater) {
 	updater.OnKLineWindowUpdate(inc.handleKLineWindowUpdate)
 }
 
-func calculateVWAP(klines []types.KLine, priceF KLineValueMapper, window int) float64 {
+func calculateVWAP(klines []types.KLine, priceF types.KLineValueMapper, window int) float64 {
 	vwap := VWAP{IntervalWindow: types.IntervalWindow{Window: window}}
 	for _, k := range klines {
 		vwap.Update(priceF(k), k.Volume.Float64())

pkg/indicator/vwap_test.go

@@ -63,7 +63,7 @@ func Test_calculateVWAP(t *testing.T) {
 
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			priceF := KLineTypicalPriceMapper
+			priceF := types.KLineTypicalPriceMapper
 			got := calculateVWAP(tt.kLines, priceF, tt.window)
 			diff := math.Trunc((got-tt.want)*100) / 100
 			if diff != 0 {

pkg/risk/riskcontrol/circuit_break.go

@@ -4,14 +4,14 @@ import (
 	log "github.com/sirupsen/logrus"
 
 	"github.com/c9s/bbgo/pkg/fixedpoint"
-	"github.com/c9s/bbgo/pkg/indicator"
+	"github.com/c9s/bbgo/pkg/indicator/v2"
 	"github.com/c9s/bbgo/pkg/types"
 )
 
 type CircuitBreakRiskControl struct {
 	// Since price could be fluctuated large,
 	// use an EWMA to smooth it in running time
-	price         *indicator.EWMAStream
+	price         *indicatorv2.EWMAStream
 	position      *types.Position
 	profitStats   *types.ProfitStats
 	lossThreshold fixedpoint.Value
@@ -19,7 +19,7 @@ type CircuitBreakRiskControl struct {
 
 func NewCircuitBreakRiskControl(
 	position *types.Position,
-	price *indicator.EWMAStream,
+	price *indicatorv2.EWMAStream,
 	lossThreshold fixedpoint.Value,
 	profitStats *types.ProfitStats) *CircuitBreakRiskControl {
 

pkg/risk/riskcontrol/circuit_break_test.go

@@ -6,12 +6,11 @@ import (
 	"github.com/stretchr/testify/assert"
 
 	"github.com/c9s/bbgo/pkg/fixedpoint"
-	"github.com/c9s/bbgo/pkg/indicator"
+	indicatorv2 "github.com/c9s/bbgo/pkg/indicator/v2"
 	"github.com/c9s/bbgo/pkg/types"
 )
 
 func Test_IsHalted(t *testing.T) {
-
 	var (
 		price          = 30000.00
 		realizedPnL    = fixedpoint.NewFromFloat(-100.0)
@@ -19,7 +18,7 @@ func Test_IsHalted(t *testing.T) {
 	)
 
 	window := types.IntervalWindow{Window: 30, Interval: types.Interval1m}
-	priceEWMA := indicator.EWMA2(nil, window.Window)
+	priceEWMA := indicatorv2.EWMA2(nil, window.Window)
 	priceEWMA.PushAndEmit(price)
 
 	cases := []struct {

pkg/strategy/bollmaker/dynamic_spread.go

@@ -7,6 +7,7 @@ import (
 
 	"github.com/c9s/bbgo/pkg/bbgo"
 	"github.com/c9s/bbgo/pkg/indicator"
+	indicatorv2 "github.com/c9s/bbgo/pkg/indicator/v2"
 	"github.com/c9s/bbgo/pkg/types"
 )
 
@@ -28,7 +29,7 @@ type DynamicSpreadSettings struct {
 }
 
 // Initialize dynamic spreads and preload SMAs
-func (ds *DynamicSpreadSettings) Initialize(symbol string, session *bbgo.ExchangeSession, neutralBoll, defaultBoll *indicator.BOLLStream) {
+func (ds *DynamicSpreadSettings) Initialize(symbol string, session *bbgo.ExchangeSession, neutralBoll, defaultBoll *indicatorv2.BOLLStream) {
 	switch {
 	case ds.AmpSpreadSettings != nil:
 		ds.AmpSpreadSettings.initialize(symbol, session)
@@ -164,10 +165,10 @@ type DynamicSpreadBollWidthRatioSettings struct {
 	// A positive number. The greater factor, the sharper weighting function. Default set to 1.0 .
 	Sensitivity float64 `json:"sensitivity"`
 
-	defaultBoll, neutralBoll *indicator.BOLLStream
+	defaultBoll, neutralBoll *indicatorv2.BOLLStream
 }
 
-func (ds *DynamicSpreadBollWidthRatioSettings) initialize(neutralBoll, defaultBoll *indicator.BOLLStream) {
+func (ds *DynamicSpreadBollWidthRatioSettings) initialize(neutralBoll, defaultBoll *indicatorv2.BOLLStream) {
 	ds.neutralBoll = neutralBoll
 	ds.defaultBoll = defaultBoll
 	if ds.Sensitivity <= 0. {

pkg/strategy/bollmaker/strategy.go

@@ -6,7 +6,7 @@ import (
 	"math"
 	"sync"
 
-	"github.com/c9s/bbgo/pkg/indicator"
+	indicatorv2 "github.com/c9s/bbgo/pkg/indicator/v2"
 	"github.com/c9s/bbgo/pkg/util"
 
 	"github.com/pkg/errors"
@@ -158,10 +158,10 @@ type Strategy struct {
 	groupID uint32
 
 	// defaultBoll is the BOLLINGER indicator we used for predicting the price.
-	defaultBoll *indicator.BOLLStream
+	defaultBoll *indicatorv2.BOLLStream
 
 	// neutralBoll is the neutral price section
-	neutralBoll *indicator.BOLLStream
+	neutralBoll *indicatorv2.BOLLStream
 
 	// StrategyController
 	bbgo.StrategyController

pkg/strategy/bollmaker/trend.go

@@ -1,6 +1,8 @@
 package bollmaker
 
-import "github.com/c9s/bbgo/pkg/indicator"
+import (
+	indicatorv2 "github.com/c9s/bbgo/pkg/indicator/v2"
+)
 
 type PriceTrend string
 
@@ -11,7 +13,7 @@ const (
 	UnknownTrend PriceTrend = "unknown"
 )
 
-func detectPriceTrend(inc *indicator.BOLLStream, price float64) PriceTrend {
+func detectPriceTrend(inc *indicatorv2.BOLLStream, price float64) PriceTrend {
 	if inBetween(price, inc.DownBand.Last(0), inc.UpBand.Last(0)) {
 		return NeutralTrend
 	}

pkg/strategy/factorzoo/factors/price_mean_reversion.go

@@ -82,13 +82,13 @@ func (inc *PMR) PushK(k types.KLine) {
 		return
 	}
 
-	inc.Update(indicator.KLineClosePriceMapper(k))
+	inc.Update(types.KLineClosePriceMapper(k))
 	inc.EndTime = k.EndTime.Time()
 	inc.EmitUpdate(inc.Last(0))
 }
 
 func CalculateKLinesPMR(allKLines []types.KLine, window int) float64 {
-	return pmr(indicator.MapKLinePrice(allKLines, indicator.KLineClosePriceMapper), window)
+	return pmr(types.MapKLinePrice(allKLines, types.KLineClosePriceMapper), window)
 }
 
 func pmr(prices []float64, window int) float64 {

pkg/strategy/factorzoo/factors/price_volume_divergence.go

@@ -89,13 +89,13 @@ func (inc *PVD) PushK(k types.KLine) {
 		return
 	}
 
-	inc.Update(indicator.KLineClosePriceMapper(k), indicator.KLineVolumeMapper(k))
+	inc.Update(types.KLineClosePriceMapper(k), types.KLineVolumeMapper(k))
 	inc.EndTime = k.EndTime.Time()
 	inc.EmitUpdate(inc.Last(0))
 }
 
 func CalculateKLinesPVD(allKLines []types.KLine, window int) float64 {
-	return pvd(indicator.MapKLinePrice(allKLines, indicator.KLineClosePriceMapper), indicator.MapKLinePrice(allKLines, indicator.KLineVolumeMapper), window)
+	return pvd(types.MapKLinePrice(allKLines, types.KLineClosePriceMapper), types.MapKLinePrice(allKLines, types.KLineVolumeMapper), window)
 }
 
 func pvd(prices []float64, volumes []float64, window int) float64 {

pkg/strategy/factorzoo/factors/return_rate.go

@@ -81,7 +81,7 @@ func (inc *RR) PushK(k types.KLine) {
 		return
 	}
 
-	inc.Update(indicator.KLineClosePriceMapper(k))
+	inc.Update(types.KLineClosePriceMapper(k))
 	inc.EndTime = k.EndTime.Time()
 	inc.EmitUpdate(inc.Last(0))
 }

pkg/strategy/fmaker/A18.go

@@ -42,7 +42,7 @@ func (inc *A18) CalculateAndUpdate(klines []types.KLine) {
 
 	var recentT = klines[end-(inc.Window-1) : end+1]
 
-	val, err := calculateA18(recentT, indicator.KLineClosePriceMapper)
+	val, err := calculateA18(recentT, types.KLineClosePriceMapper)
 	if err != nil {
 		log.WithError(err).Error("can not calculate")
 		return

pkg/strategy/fmaker/A2.go

@@ -42,7 +42,7 @@ func (inc *A2) CalculateAndUpdate(klines []types.KLine) {
 
 	var recentT = klines[end-(inc.Window-1) : end+1]
 
-	val, err := calculateA2(recentT, KLineLowPriceMapper, KLineHighPriceMapper, indicator.KLineClosePriceMapper)
+	val, err := calculateA2(recentT, KLineLowPriceMapper, KLineHighPriceMapper, types.KLineClosePriceMapper)
 	if err != nil {
 		log.WithError(err).Error("can not calculate")
 		return

pkg/strategy/fmaker/A3.go

@@ -43,7 +43,7 @@ func (inc *A3) CalculateAndUpdate(klines []types.KLine) {
 
 	var recentT = klines[end-(inc.Window-1) : end+1]
 
-	val, err := calculateA3(recentT, KLineLowPriceMapper, KLineHighPriceMapper, indicator.KLineClosePriceMapper)
+	val, err := calculateA3(recentT, KLineLowPriceMapper, KLineHighPriceMapper, types.KLineClosePriceMapper)
 	if err != nil {
 		log.WithError(err).Error("can not calculate pivots")
 		return

pkg/strategy/fmaker/A34.go

@@ -42,7 +42,7 @@ func (inc *A34) CalculateAndUpdate(klines []types.KLine) {
 
 	var recentT = klines[end-(inc.Window-1) : end+1]
 
-	val, err := calculateA34(recentT, indicator.KLineClosePriceMapper)
+	val, err := calculateA34(recentT, types.KLineClosePriceMapper)
 	if err != nil {
 		log.WithError(err).Error("can not calculate pivots")
 		return

pkg/strategy/fmaker/R.go

@@ -46,7 +46,7 @@ func (inc *R) CalculateAndUpdate(klines []types.KLine) {
 
 	var recentT = klines[end-(inc.Window-1) : end+1]
 
-	val, err := calculateR(recentT, indicator.KLineOpenPriceMapper, indicator.KLineClosePriceMapper)
+	val, err := calculateR(recentT, types.KLineOpenPriceMapper, types.KLineClosePriceMapper)
 	if err != nil {
 		log.WithError(err).Error("can not calculate pivots")
 		return

pkg/strategy/fmaker/S0.go

@@ -42,7 +42,7 @@ func (inc *S0) CalculateAndUpdate(klines []types.KLine) {
 
 	var recentT = klines[end-(inc.Window-1) : end+1]
 
-	val, err := calculateS0(recentT, indicator.KLineClosePriceMapper)
+	val, err := calculateS0(recentT, types.KLineClosePriceMapper)
 	if err != nil {
 		log.WithError(err).Error("can not calculate")
 		return

pkg/strategy/fmaker/S1.go

@@ -40,7 +40,7 @@ func (inc *S1) CalculateAndUpdate(klines []types.KLine) {
 
 	var recentT = klines[end-(inc.Window-1) : end+1]
 
-	correlation, err := calculateS1(recentT, inc.Window, KLineAmplitudeMapper, indicator.KLineVolumeMapper)
+	correlation, err := calculateS1(recentT, inc.Window, KLineAmplitudeMapper, types.KLineVolumeMapper)
 	if err != nil {
 		log.WithError(err).Error("can not calculate correlation")
 		return

pkg/strategy/fmaker/S2.go

@@ -40,7 +40,7 @@ func (inc *S2) CalculateAndUpdate(klines []types.KLine) {
 
 	var recentT = klines[end-(inc.Window-1) : end+1]
 
-	correlation, err := calculateS2(recentT, inc.Window, indicator.KLineOpenPriceMapper, indicator.KLineVolumeMapper)
+	correlation, err := calculateS2(recentT, inc.Window, types.KLineOpenPriceMapper, types.KLineVolumeMapper)
 	if err != nil {
 		log.WithError(err).Error("can not calculate correlation")
 		return

pkg/strategy/fmaker/S3.go

@@ -42,7 +42,7 @@ func (inc *S3) CalculateAndUpdate(klines []types.KLine) {
 
 	var recentT = klines[end-(inc.Window-1) : end+1]
 
-	val, err := calculateS3(recentT, indicator.KLineClosePriceMapper, indicator.KLineOpenPriceMapper)
+	val, err := calculateS3(recentT, types.KLineClosePriceMapper, types.KLineOpenPriceMapper)
 	if err != nil {
 		log.WithError(err).Error("can not calculate")
 		return

pkg/strategy/fmaker/S4.go

@@ -42,7 +42,7 @@ func (inc *S4) CalculateAndUpdate(klines []types.KLine) {
 
 	var recentT = klines[end-(inc.Window-1) : end+1]
 
-	val, err := calculateS4(recentT, indicator.KLineClosePriceMapper)
+	val, err := calculateS4(recentT, types.KLineClosePriceMapper)
 	if err != nil {
 		log.WithError(err).Error("can not calculate")
 		return

pkg/strategy/fmaker/S5.go

@@ -42,7 +42,7 @@ func (inc *S5) CalculateAndUpdate(klines []types.KLine) {
 
 	var recentT = klines[end-(inc.Window-1) : end+1]
 
-	val, err := calculateS5(recentT, indicator.KLineVolumeMapper)
+	val, err := calculateS5(recentT, types.KLineVolumeMapper)
 	if err != nil {
 		log.WithError(err).Error("can not calculate pivots")
 		return

pkg/strategy/fmaker/S6.go

@@ -42,7 +42,7 @@ func (inc *S6) CalculateAndUpdate(klines []types.KLine) {
 
 	var recentT = klines[end-(inc.Window-1) : end+1]
 
-	val, err := calculateS6(recentT, indicator.KLineHighPriceMapper, indicator.KLineLowPriceMapper, indicator.KLineClosePriceMapper, indicator.KLineVolumeMapper)
+	val, err := calculateS6(recentT, types.KLineHighPriceMapper, types.KLineLowPriceMapper, types.KLineClosePriceMapper, types.KLineVolumeMapper)
 	if err != nil {
 		log.WithError(err).Error("can not calculate")
 		return

pkg/strategy/fmaker/S7.go

@@ -42,7 +42,7 @@ func (inc *S7) CalculateAndUpdate(klines []types.KLine) {
 
 	var recentT = klines[end-(inc.Window-1) : end+1]
 
-	val, err := calculateS7(recentT, indicator.KLineOpenPriceMapper, indicator.KLineClosePriceMapper)
+	val, err := calculateS7(recentT, types.KLineOpenPriceMapper, types.KLineClosePriceMapper)
 	if err != nil {
 		log.WithError(err).Error("can not calculate")
 		return

pkg/strategy/harmonic/shark.go

@@ -72,7 +72,7 @@ func (inc *SHARK) PushK(k types.KLine) {
 		return
 	}
 
-	inc.Update(indicator.KLineHighPriceMapper(k), indicator.KLineLowPriceMapper(k), indicator.KLineClosePriceMapper(k))
+	inc.Update(types.KLineHighPriceMapper(k), types.KLineLowPriceMapper(k), types.KLineClosePriceMapper(k))
 	inc.EndTime = k.EndTime.Time()
 	inc.EmitUpdate(inc.Last(0))
 }

pkg/strategy/irr/neg_return_rate.go

@@ -74,7 +74,7 @@ func (inc *NRR) PushK(k types.KLine) {
 		return
 	}
 
-	inc.Update(indicator.KLineOpenPriceMapper(k), indicator.KLineClosePriceMapper(k))
+	inc.Update(types.KLineOpenPriceMapper(k), types.KLineClosePriceMapper(k))
 	inc.EndTime = k.EndTime.Time()
 	inc.EmitUpdate(inc.Last(0))
 }

pkg/strategy/rsicross/strategy.go

@@ -9,7 +9,7 @@ import (
 
 	"github.com/c9s/bbgo/pkg/bbgo"
 	"github.com/c9s/bbgo/pkg/fixedpoint"
-	"github.com/c9s/bbgo/pkg/indicator"
+	"github.com/c9s/bbgo/pkg/indicator/v2"
 	"github.com/c9s/bbgo/pkg/strategy/common"
 	"github.com/c9s/bbgo/pkg/types"
 )
@@ -52,10 +52,10 @@ func (s *Strategy) Run(ctx context.Context, orderExecutor bbgo.OrderExecutor, se
 
 	fastRsi := session.Indicators(s.Symbol).RSI(types.IntervalWindow{Interval: s.Interval, Window: s.FastWindow})
 	slowRsi := session.Indicators(s.Symbol).RSI(types.IntervalWindow{Interval: s.Interval, Window: s.SlowWindow})
-	rsiCross := indicator.Cross(fastRsi, slowRsi)
+	rsiCross := indicatorv2.Cross(fastRsi, slowRsi)
 	rsiCross.OnUpdate(func(v float64) {
-		switch indicator.CrossType(v) {
+		switch indicatorv2.CrossType(v) {
-		case indicator.CrossOver:
+		case indicatorv2.CrossOver:
 			opts := s.OpenPositionOptions
 			opts.Long = true
 
@@ -69,7 +69,7 @@ func (s *Strategy) Run(ctx context.Context, orderExecutor bbgo.OrderExecutor, se
 				logErr(err, "unable to open position")
 			}
 
-		case indicator.CrossUnder:
+		case indicatorv2.CrossUnder:
 			if err := s.OrderExecutor.ClosePosition(ctx, fixedpoint.One); err != nil {
 				logErr(err, "failed to close position")
 			}

pkg/strategy/scmaker/intensity.go

@@ -2,24 +2,24 @@ package scmaker
 
 import (
 	"github.com/c9s/bbgo/pkg/fixedpoint"
-	"github.com/c9s/bbgo/pkg/indicator"
+	indicatorv2 "github.com/c9s/bbgo/pkg/indicator/v2"
 	"github.com/c9s/bbgo/pkg/types"
 )
 
 type IntensityStream struct {
-	*indicator.Float64Series
+	*types.Float64Series
 
-	Buy, Sell *indicator.RMAStream
+	Buy, Sell *indicatorv2.RMAStream
 	window    int
 }
 
-func Intensity(source indicator.KLineSubscription, window int) *IntensityStream {
+func Intensity(source indicatorv2.KLineSubscription, window int) *IntensityStream {
 	s := &IntensityStream{
-		Float64Series: indicator.NewFloat64Series(),
+		Float64Series: types.NewFloat64Series(),
 		window:        window,
 
-		Buy:  indicator.RMA2(indicator.NewFloat64Series(), window, false),
+		Buy:  indicatorv2.RMA2(types.NewFloat64Series(), window, false),
-		Sell: indicator.RMA2(indicator.NewFloat64Series(), window, false),
+		Sell: indicatorv2.RMA2(types.NewFloat64Series(), window, false),
 	}
 
 	threshold := fixedpoint.NewFromFloat(100.0)

pkg/strategy/scmaker/strategy.go

@@ -10,7 +10,7 @@ import (
 
 	"github.com/c9s/bbgo/pkg/bbgo"
 	"github.com/c9s/bbgo/pkg/fixedpoint"
-	"github.com/c9s/bbgo/pkg/indicator"
+	. "github.com/c9s/bbgo/pkg/indicator/v2"
 	"github.com/c9s/bbgo/pkg/risk/riskcontrol"
 	"github.com/c9s/bbgo/pkg/strategy/common"
 	"github.com/c9s/bbgo/pkg/types"
@@ -73,8 +73,8 @@ type Strategy struct {
 	liquidityScale bbgo.Scale
 
 	// indicators
-	ewma      *indicator.EWMAStream
+	ewma      *EWMAStream
-	boll      *indicator.BOLLStream
+	boll      *BOLLStream
 	intensity *IntensityStream
 
 	positionRiskControl     *riskcontrol.PositionRiskControl
@@ -172,7 +172,7 @@ func (s *Strategy) Run(ctx context.Context, orderExecutor bbgo.OrderExecutor, se
 	return nil
 }
 
-func (s *Strategy) preloadKLines(inc *indicator.KLineStream, session *bbgo.ExchangeSession, symbol string, interval types.Interval) {
+func (s *Strategy) preloadKLines(inc *KLineStream, session *bbgo.ExchangeSession, symbol string, interval types.Interval) {
 	if store, ok := session.MarketDataStore(symbol); ok {
 		if kLinesData, ok := store.KLinesOfInterval(interval); ok {
 			for _, k := range *kLinesData {
@@ -183,23 +183,23 @@ func (s *Strategy) preloadKLines(inc *indicator.KLineStream, session *bbgo.Excha
 }
 
 func (s *Strategy) initializeMidPriceEMA(session *bbgo.ExchangeSession) {
-	kLines := indicator.KLines(session.MarketDataStream, s.Symbol, s.MidPriceEMA.Interval)
+	kLines := KLines(session.MarketDataStream, s.Symbol, s.MidPriceEMA.Interval)
-	s.ewma = indicator.EWMA2(indicator.ClosePrices(kLines), s.MidPriceEMA.Window)
+	s.ewma = EWMA2(ClosePrices(kLines), s.MidPriceEMA.Window)
 
 	s.preloadKLines(kLines, session, s.Symbol, s.MidPriceEMA.Interval)
 }
 
 func (s *Strategy) initializeIntensityIndicator(session *bbgo.ExchangeSession) {
-	kLines := indicator.KLines(session.MarketDataStream, s.Symbol, s.StrengthInterval)
+	kLines := KLines(session.MarketDataStream, s.Symbol, s.StrengthInterval)
 	s.intensity = Intensity(kLines, 10)
 
 	s.preloadKLines(kLines, session, s.Symbol, s.StrengthInterval)
 }
 
 func (s *Strategy) initializePriceRangeBollinger(session *bbgo.ExchangeSession) {
-	kLines := indicator.KLines(session.MarketDataStream, s.Symbol, s.PriceRangeBollinger.Interval)
+	kLines := KLines(session.MarketDataStream, s.Symbol, s.PriceRangeBollinger.Interval)
-	closePrices := indicator.ClosePrices(kLines)
+	closePrices := ClosePrices(kLines)
-	s.boll = indicator.BOLL2(closePrices, s.PriceRangeBollinger.Window, s.PriceRangeBollinger.K)
+	s.boll = BOLL(closePrices, s.PriceRangeBollinger.Window, s.PriceRangeBollinger.K)
 
 	s.preloadKLines(kLines, session, s.Symbol, s.PriceRangeBollinger.Interval)
 }

pkg/types/float64updater.go

@@ -1,4 +1,4 @@
-package indicator
+package types
 
 //go:generate callbackgen -type Float64Updater
 type Float64Updater struct {

pkg/types/float64updater_callbacks.go

@@ -0,0 +1,15 @@
+// Code generated by "callbackgen -type Float64Updater"; DO NOT EDIT.
+
+package types
+
+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/types/kline.go

@@ -645,3 +645,41 @@ func KLineWith(symbol string, interval Interval, callback KLineCallback) KLineCa
 		callback(k)
 	}
 }
+
+type KLineValueMapper func(k KLine) float64
+
+func KLineOpenPriceMapper(k KLine) float64 {
+	return k.Open.Float64()
+}
+
+func KLineClosePriceMapper(k KLine) float64 {
+	return k.Close.Float64()
+}
+
+func KLineTypicalPriceMapper(k KLine) float64 {
+	return (k.High.Float64() + k.Low.Float64() + k.Close.Float64()) / 3.
+}
+
+func KLinePriceVolumeMapper(k KLine) float64 {
+	return k.Close.Mul(k.Volume).Float64()
+}
+
+func KLineVolumeMapper(k KLine) float64 {
+	return k.Volume.Float64()
+}
+
+func MapKLinePrice(kLines []KLine, f KLineValueMapper) (prices []float64) {
+	for _, k := range kLines {
+		prices = append(prices, f(k))
+	}
+
+	return prices
+}
+
+func KLineLowPriceMapper(k KLine) float64 {
+	return k.Low.Float64()
+}
+
+func KLineHighPriceMapper(k KLine) float64 {
+	return k.High.Float64()
+}

pkg/types/series_float64.go

@@ -1,25 +1,24 @@
-package indicator
+package types
 
 import (
 	"github.com/c9s/bbgo/pkg/datatype/floats"
-	"github.com/c9s/bbgo/pkg/types"
 )
 
 type Float64Series struct {
-	types.SeriesBase
+	SeriesBase
 	Float64Updater
-	slice floats.Slice
+	Slice floats.Slice
 }
 
 func NewFloat64Series(v ...float64) *Float64Series {
 	s := &Float64Series{}
-	s.slice = v
+	s.Slice = v
-	s.SeriesBase.Series = s.slice
+	s.SeriesBase.Series = s.Slice
 	return s
 }
 
 func (f *Float64Series) Last(i int) float64 {
-	return f.slice.Last(i)
+	return f.Slice.Last(i)
 }
 
 func (f *Float64Series) Index(i int) float64 {
@@ -27,15 +26,11 @@ func (f *Float64Series) Index(i int) float64 {
 }
 
 func (f *Float64Series) Length() int {
-	return len(f.slice)
+	return len(f.Slice)
-}
-
-func (f *Float64Series) Slice() floats.Slice {
-	return f.slice
 }
 
 func (f *Float64Series) PushAndEmit(x float64) {
-	f.slice.Push(x)
+	f.Slice.Push(x)
 	f.EmitUpdate(x)
 }
 
@@ -71,3 +66,22 @@ func (f *Float64Series) Bind(source Float64Source, target Float64Calculator) {
 		f.Subscribe(source, c)
 	}
 }
+
+type Float64Calculator interface {
+	Calculate(x float64) float64
+	PushAndEmit(x float64)
+}
+
+type Float64Source interface {
+	Series
+	OnUpdate(f func(v float64))
+}
+
+type Float64Subscription interface {
+	Series
+	AddSubscriber(f func(v float64))
+}
+
+type Float64Truncator interface {
+	Truncate()
+}