Merge pull request #1181 from c9s/feature/v2indicator

FEATURE: [indicator] add new ATR, RMA indicators with the new API design

pkg/datatype/floats/funcs_test.go

@@ -15,3 +15,9 @@ func TestHigher(t *testing.T) {
 	out := Higher([]float64{10.0, 11.0, 12.0, 13.0, 15.0}, 12.0)
 	assert.Equal(t, []float64{13.0, 15.0}, out)
 }
+
+func TestLSM(t *testing.T) {
+	slice := Slice{1., 2., 3., 4.}
+	slope := LSM(slice)
+	assert.Equal(t, 1.0, slope)
+}

pkg/datatype/floats/pivot.go

@@ -1,10 +1,10 @@
 package floats
 
 func (s Slice) Pivot(left, right int, f func(a, pivot float64) bool) (float64, bool) {
-	return CalculatePivot(s, left, right, f)
+	return FindPivot(s, left, right, f)
 }
 
-func CalculatePivot(values Slice, left, right int, f func(a, pivot float64) bool) (float64, bool) {
+func FindPivot(values Slice, left, right int, f func(a, pivot float64) bool) (float64, bool) {
 	length := len(values)
 
 	if right == 0 {

pkg/datatype/floats/slice.go

@@ -73,7 +73,10 @@ func (s Slice) Add(b Slice) (c Slice) {
 }
 
 func (s Slice) Sum() (sum float64) {
-	return floats.Sum(s)
+	for _, v := range s {
+		sum += v
+	}
+	return sum
 }
 
 func (s Slice) Mean() (mean float64) {
@@ -97,6 +100,18 @@ func (s Slice) Tail(size int) Slice {
 	return win
 }
 
+func (s Slice) Average() float64 {
+	if len(s) == 0 {
+		return 0.0
+	}
+
+	total := 0.0
+	for _, value := range s {
+		total += value
+	}
+	return total / float64(len(s))
+}
+
 func (s Slice) Diff() (values Slice) {
 	for i, v := range s {
 		if i == 0 {
@@ -171,19 +186,49 @@ func (s Slice) Addr() *Slice {
 func (s Slice) Last() float64 {
 	length := len(s)
 	if length > 0 {
-		return (s)[length-1]
+		return s[length-1]
 	}
 	return 0.0
 }
 
+// Index fetches the element from the end of the slice
+// WARNING: it does not start from 0!!!
 func (s Slice) Index(i int) float64 {
 	length := len(s)
-	if length-i <= 0 || i < 0 {
+	if i < 0 || length-1-i < 0 {
 		return 0.0
 	}
-	return (s)[length-i-1]
+	return s[length-1-i]
 }
 
 func (s Slice) Length() int {
 	return len(s)
 }
+
+func (s Slice) LSM() float64 {
+	return LSM(s)
+}
+
+// LSM is the least squares method for linear regression
+func LSM(values Slice) float64 {
+	var sumX, sumY, sumXSqr, sumXY = .0, .0, .0, .0
+
+	end := len(values) - 1
+	for i := end; i >= 0; i-- {
+		val := values[i]
+		per := float64(end - i + 1)
+		sumX += per
+		sumY += val
+		sumXSqr += per * per
+		sumXY += val * per
+	}
+
+	length := float64(len(values))
+	slope := (length*sumXY - sumX*sumY) / (length*sumXSqr - sumX*sumX)
+
+	average := sumY / length
+	tail := average - slope*sumX/length + slope
+	head := tail + slope*(length-1)
+	slope2 := (tail - head) / (length - 1)
+	return slope2
+}

pkg/indicator/ewmastream_callbacks.go

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

pkg/indicator/float64updater.go

@@ -1,6 +1,40 @@
 package indicator
 
+import (
+	"github.com/c9s/bbgo/pkg/datatype/floats"
+	"github.com/c9s/bbgo/pkg/types"
+)
+
 //go:generate callbackgen -type Float64Updater
 type Float64Updater struct {
 	updateCallbacks []func(v float64)
 }
+
+type Float64Series struct {
+	types.SeriesBase
+	Float64Updater
+	slice floats.Slice
+}
+
+func NewFloat64Series(v ...float64) Float64Series {
+	s := Float64Series{}
+	s.slice = v
+	s.SeriesBase.Series = s.slice
+	return s
+}
+
+func (f *Float64Series) Last() float64 {
+	return f.slice.Last()
+}
+
+func (f *Float64Series) Index(i int) float64 {
+	length := len(f.slice)
+	if length == 0 || length-i-1 < 0 {
+		return 0
+	}
+	return f.slice[length-i-1]
+}
+
+func (f *Float64Series) Length() int {
+	return len(f.slice)
+}

pkg/indicator/float64updater_callbacks.go

@@ -4,12 +4,12 @@ package indicator
 
 import ()
 
-func (F *Float64Updater) OnUpdate(cb func(v float64)) {
+func (f *Float64Updater) OnUpdate(cb func(v float64)) {
-	F.updateCallbacks = append(F.updateCallbacks, cb)
+	f.updateCallbacks = append(f.updateCallbacks, cb)
 }
 
-func (F *Float64Updater) EmitUpdate(v float64) {
+func (f *Float64Updater) EmitUpdate(v float64) {
-	for _, cb := range F.updateCallbacks {
+	for _, cb := range f.updateCallbacks {
 		cb(v)
 	}
 }

pkg/indicator/macd2.go

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

@@ -64,13 +64,13 @@ func (inc *PivotLow) PushK(k types.KLine) {
 }
 
 func calculatePivotHigh(highs floats.Slice, left, right int) (float64, bool) {
-	return floats.CalculatePivot(highs, left, right, func(a, pivot float64) bool {
+	return floats.FindPivot(highs, left, right, func(a, pivot float64) bool {
 		return a < pivot
 	})
 }
 
 func calculatePivotLow(lows floats.Slice, left, right int) (float64, bool) {
-	return floats.CalculatePivot(lows, left, right, func(a, pivot float64) bool {
+	return floats.FindPivot(lows, left, right, func(a, pivot float64) bool {
 		return a > pivot
 	})
 }

pkg/indicator/price.go

@@ -1,7 +1,6 @@
 package indicator
 
 import (
-	"github.com/c9s/bbgo/pkg/datatype/floats"
 	"github.com/c9s/bbgo/pkg/types"
 )
 
@@ -10,10 +9,8 @@ type KLineSubscription interface {
 }
 
 type PriceStream struct {
-	types.SeriesBase
+	Float64Series
-	Float64Updater
 
-	slice  floats.Slice
 	mapper KLineValueMapper
 }
 
@@ -26,12 +23,16 @@ func Price(source KLineSubscription, mapper KLineValueMapper) *PriceStream {
 
 	source.AddSubscriber(func(k types.KLine) {
 		v := s.mapper(k)
-		s.slice.Push(v)
+		s.PushAndEmit(v)
-		s.EmitUpdate(v)
 	})
 	return s
 }
 
+func (s *PriceStream) PushAndEmit(v float64) {
+	s.slice.Push(v)
+	s.EmitUpdate(v)
+}
+
 func ClosePrices(source KLineSubscription) *PriceStream {
 	return Price(source, KLineClosePriceMapper)
 }

pkg/indicator/subtract.go

@@ -0,0 +1,48 @@
+package indicator
+
+import (
+	"github.com/c9s/bbgo/pkg/datatype/floats"
+)
+
+// SubtractStream subscribes 2 upstream data, and then subtract these 2 values
+type SubtractStream struct {
+	Float64Series
+
+	a, b floats.Slice
+	i    int
+}
+
+// Subtract creates the SubtractStream object
+// subtract := Subtract(longEWMA, shortEWMA)
+func Subtract(a, b Float64Source) *SubtractStream {
+	s := &SubtractStream{
+		Float64Series: NewFloat64Series(),
+	}
+
+	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.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.EmitUpdate(f)
+		}
+	}
+}

pkg/indicator/types.go

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

pkg/indicator/util.go

@@ -1 +1,15 @@
 package indicator
+
+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.go

@@ -0,0 +1,19 @@
+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

@@ -0,0 +1,12 @@
+package indicator
+
+type ATRStream struct {
+	// embedded struct
+	*RMAStream
+}
+
+func ATR2(source KLineSubscription, window int) *ATRStream {
+	tr := TR2(source)
+	rma := RMA2(tr, window, true)
+	return &ATRStream{RMAStream: rma}
+}

pkg/indicator/v2_atr_test.go

@@ -0,0 +1,81 @@
+package indicator
+
+import (
+	"encoding/json"
+	"math"
+	"testing"
+
+	"github.com/c9s/bbgo/pkg/fixedpoint"
+	"github.com/c9s/bbgo/pkg/types"
+)
+
+/*
+python
+
+import pandas as pd
+import pandas_ta as ta
+
+	data = {
+	    "high": [40145.0, 40186.36, 40196.39, 40344.6, 40245.48, 40273.24, 40464.0, 40699.0, 40627.48, 40436.31, 40370.0, 40376.8, 40227.03, 40056.52, 39721.7, 39597.94, 39750.15, 39927.0, 40289.02, 40189.0],
+	    "low": [39870.71, 39834.98, 39866.31, 40108.31, 40016.09, 40094.66, 40105.0, 40196.48, 40154.99, 39800.0, 39959.21, 39922.98, 39940.02, 39632.0, 39261.39, 39254.63, 39473.91, 39555.51, 39819.0, 40006.84],
+	    "close": [40105.78, 39935.23, 40183.97, 40182.03, 40212.26, 40149.99, 40378.0, 40618.37, 40401.03, 39990.39, 40179.13, 40097.23, 40014.72, 39667.85, 39303.1, 39519.99,
+
+39693.79, 39827.96, 40074.94, 40059.84]
+}
+
+high = pd.Series(data['high'])
+low = pd.Series(data['low'])
+close = pd.Series(data['close'])
+result = ta.atr(high, low, close, length=14)
+print(result)
+*/
+func Test_ATR2(t *testing.T) {
+	var bytes = []byte(`{
+		"high": [40145.0, 40186.36, 40196.39, 40344.6, 40245.48, 40273.24, 40464.0, 40699.0, 40627.48, 40436.31, 40370.0, 40376.8, 40227.03, 40056.52, 39721.7, 39597.94, 39750.15, 39927.0, 40289.02, 40189.0], 
+		"low": [39870.71, 39834.98, 39866.31, 40108.31, 40016.09, 40094.66, 40105.0, 40196.48, 40154.99, 39800.0, 39959.21, 39922.98, 39940.02, 39632.0, 39261.39, 39254.63, 39473.91, 39555.51, 39819.0, 40006.84],
+		"close": [40105.78, 39935.23, 40183.97, 40182.03, 40212.26, 40149.99, 40378.0, 40618.37, 40401.03, 39990.39, 40179.13, 40097.23, 40014.72, 39667.85, 39303.1, 39519.99, 39693.79, 39827.96, 40074.94, 40059.84]
+	}`)
+
+	var buildKLines = func(bytes []byte) (kLines []types.KLine) {
+		var prices map[string][]fixedpoint.Value
+		_ = json.Unmarshal(bytes, &prices)
+		for i, h := range prices["high"] {
+			kLine := types.KLine{High: h, Low: prices["low"][i], Close: prices["close"][i]}
+			kLines = append(kLines, kLine)
+		}
+		return kLines
+	}
+
+	tests := []struct {
+		name   string
+		kLines []types.KLine
+		window int
+		want   float64
+	}{
+		{
+			name:   "test_binance_btcusdt_1h",
+			kLines: buildKLines(bytes),
+			window: 14,
+			want:   367.913903,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			stream := &types.StandardStream{}
+
+			kLines := KLines(stream)
+			atr := ATR2(kLines, tt.window)
+
+			for _, k := range tt.kLines {
+				stream.EmitKLineClosed(k)
+			}
+
+			got := atr.Last()
+			diff := math.Trunc((got-tt.want)*100) / 100
+			if diff != 0 {
+				t.Errorf("ATR2() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}

pkg/indicator/v2_ewma.go

@@ -0,0 +1,36 @@
+package indicator
+
+type EWMAStream struct {
+	Float64Series
+
+	window     int
+	multiplier float64
+}
+
+func EWMA2(source Float64Source, window int) *EWMAStream {
+	s := &EWMAStream{
+		Float64Series: NewFloat64Series(),
+		window:        window,
+		multiplier:    2.0 / float64(1+window),
+	}
+
+	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
+}

pkg/indicator/v2_rma.go

@@ -0,0 +1,69 @@
+package indicator
+
+type RMAStream struct {
+	// embedded structs
+	Float64Series
+
+	// config fields
+	Adjust bool
+
+	window        int
+	counter       int
+	sum, previous float64
+}
+
+func RMA2(source Float64Source, window int, adjust bool) *RMAStream {
+	s := &RMAStream{
+		Float64Series: NewFloat64Series(),
+		window:        window,
+		Adjust:        adjust,
+	}
+
+	if sub, ok := source.(Float64Subscription); ok {
+		sub.AddSubscriber(s.calculateAndPush)
+	} else {
+		source.OnUpdate(s.calculateAndPush)
+	}
+
+	return s
+}
+
+func (s *RMAStream) calculateAndPush(v float64) {
+	v2 := s.calculate(v)
+	s.slice.Push(v2)
+	s.EmitUpdate(v2)
+	s.truncate()
+}
+
+func (s *RMAStream) calculate(x float64) float64 {
+	lambda := 1 / float64(s.window)
+	tmp := 0.0
+	if s.counter == 0 {
+		s.sum = 1
+		tmp = x
+	} else {
+		if s.Adjust {
+			s.sum = s.sum*(1-lambda) + 1
+			tmp = s.previous + (x-s.previous)/s.sum
+		} else {
+			tmp = s.previous*(1-lambda) + x*lambda
+		}
+	}
+	s.counter++
+
+	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(tmp)
+	s.previous = tmp
+
+	return tmp
+}
+
+func (s *RMAStream) truncate() {
+	if len(s.slice) > MaxNumOfRMA {
+		s.slice = s.slice[MaxNumOfRMATruncateSize-1:]
+	}
+}

pkg/indicator/v2_rsi.go

@@ -0,0 +1,56 @@
+package indicator
+
+type RSIStream struct {
+	// embedded structs
+	Float64Series
+
+	// config fields
+	window int
+
+	// private states
+	source Float64Source
+}
+
+func RSI2(source Float64Source, window int) *RSIStream {
+	s := &RSIStream{
+		source:        source,
+		Float64Series: NewFloat64Series(),
+		window:        window,
+	}
+
+	if sub, ok := source.(Float64Subscription); ok {
+		sub.AddSubscriber(s.calculateAndPush)
+	} else {
+		source.OnUpdate(s.calculateAndPush)
+	}
+
+	return s
+}
+
+func (s *RSIStream) calculate(_ float64) float64 {
+	var gainSum, lossSum float64
+	var sourceLen = s.source.Length()
+	var limit = min(s.window, sourceLen)
+	for i := 0; i < limit; i++ {
+		value := s.source.Index(i)
+		prev := s.source.Index(i + 1)
+		change := value - prev
+		if change >= 0 {
+			gainSum += change
+		} else {
+			lossSum += -change
+		}
+	}
+
+	avgGain := gainSum / float64(limit)
+	avgLoss := lossSum / float64(limit)
+	rs := avgGain / avgLoss
+	rsi := 100.0 - (100.0 / (1.0 + rs))
+	return rsi
+}
+
+func (s *RSIStream) calculateAndPush(x float64) {
+	rsi := s.calculate(x)
+	s.slice.Push(rsi)
+	s.EmitUpdate(rsi)
+}

pkg/indicator/v2_rsi_test.go

@@ -0,0 +1,87 @@
+package indicator
+
+import (
+	"encoding/json"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+
+	"github.com/c9s/bbgo/pkg/datatype/floats"
+)
+
+func Test_RSI2(t *testing.T) {
+	// test case from https://school.stockcharts.com/doku.php?id=technical_indicators:relative_strength_index_rsi
+	var data = []byte(`[44.34, 44.09, 44.15, 43.61, 44.33, 44.83, 45.10, 45.42, 45.84, 46.08, 45.89, 46.03, 45.61, 46.28, 46.28, 46.00, 46.03, 46.41, 46.22, 45.64, 46.21, 46.25, 45.71, 46.45, 45.78, 45.35, 44.03, 44.18, 44.22, 44.57, 43.42, 42.66, 43.13]`)
+	var values []float64
+	err := json.Unmarshal(data, &values)
+	assert.NoError(t, err)
+
+	tests := []struct {
+		name   string
+		values []float64
+		window int
+		want   floats.Slice
+	}{
+		{
+			name:   "RSI",
+			values: values,
+			window: 14,
+			want: floats.Slice{
+				100.000000,
+				99.439336,
+				99.440090,
+				98.251826,
+				98.279242,
+				98.297781,
+				98.307626,
+				98.319149,
+				98.334036,
+				98.342426,
+				97.951933,
+				97.957908,
+				97.108036,
+				97.147514,
+				70.464135,
+				70.020964,
+				69.831224,
+				80.567686,
+				73.333333,
+				59.806295,
+				62.528217,
+				60.000000,
+				48.477752,
+				53.878407,
+				48.952381,
+				43.862816,
+				37.732919,
+				32.263514,
+				32.718121,
+				38.142620,
+				31.748252,
+				25.099602,
+				30.217670,
+			},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			// RSI2()
+			prices := &PriceStream{}
+			rsi := RSI2(prices, tt.window)
+
+			t.Logf("data length: %d", len(tt.values))
+			for _, price := range tt.values {
+				prices.PushAndEmit(price)
+			}
+
+			assert.Equal(t, floats.Slice(tt.values), prices.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])
+				}
+			}
+		})
+	}
+}

pkg/indicator/v2_test.go

@@ -9,7 +9,7 @@ import (
 	"github.com/c9s/bbgo/pkg/types"
 )
 
-func TestSubtract(t *testing.T) {
+func Test_v2_Subtract(t *testing.T) {
 	stream := &types.StandardStream{}
 	kLines := KLines(stream)
 	closePrices := ClosePrices(kLines)
@@ -25,6 +25,6 @@ func TestSubtract(t *testing.T) {
 	t.Logf("slowEMA: %+v", slowEMA.slice)
 
 	assert.Equal(t, len(subtract.a), len(subtract.b))
-	assert.Equal(t, len(subtract.a), len(subtract.c))
+	assert.Equal(t, len(subtract.a), len(subtract.slice))
-	assert.InDelta(t, subtract.c[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

@@ -0,0 +1,47 @@
+package indicator
+
+import (
+	"math"
+
+	"github.com/c9s/bbgo/pkg/types"
+)
+
+// This TRStream calculates the ATR first
+type TRStream struct {
+	// embedded struct
+	Float64Series
+
+	// private states
+	previousClose float64
+}
+
+func TR2(source KLineSubscription) *TRStream {
+	s := &TRStream{
+		Float64Series: NewFloat64Series(),
+	}
+
+	source.AddSubscriber(func(k types.KLine) {
+		s.calculateAndPush(k.High.Float64(), k.Low.Float64(), k.Close.Float64())
+	})
+	return s
+}
+
+func (s *TRStream) calculateAndPush(high, low, cls float64) {
+	if s.previousClose == .0 {
+		s.previousClose = cls
+		return
+	}
+
+	trueRange := high - low
+	hc := math.Abs(high - s.previousClose)
+	lc := math.Abs(low - s.previousClose)
+	if trueRange < hc {
+		trueRange = hc
+	}
+	if trueRange < lc {
+		trueRange = lc
+	}
+
+	s.previousClose = cls
+	s.EmitUpdate(trueRange)
+}

pkg/indicator/v2_tr_test.go

@@ -0,0 +1,82 @@
+package indicator
+
+import (
+	"encoding/json"
+	"math"
+	"testing"
+
+	"github.com/c9s/bbgo/pkg/fixedpoint"
+	"github.com/c9s/bbgo/pkg/types"
+)
+
+/*
+python
+
+import pandas as pd
+import pandas_ta as ta
+
+	data = {
+	    "high": [40145.0, 40186.36, 40196.39, 40344.6, 40245.48, 40273.24, 40464.0, 40699.0, 40627.48, 40436.31, 40370.0, 40376.8, 40227.03, 40056.52, 39721.7, 39597.94, 39750.15, 39927.0, 40289.02, 40189.0],
+	    "low": [39870.71, 39834.98, 39866.31, 40108.31, 40016.09, 40094.66, 40105.0, 40196.48, 40154.99, 39800.0, 39959.21, 39922.98, 39940.02, 39632.0, 39261.39, 39254.63, 39473.91, 39555.51, 39819.0, 40006.84],
+	    "close": [40105.78, 39935.23, 40183.97, 40182.03, 40212.26, 40149.99, 40378.0, 40618.37, 40401.03, 39990.39, 40179.13, 40097.23, 40014.72, 39667.85, 39303.1, 39519.99,
+
+39693.79, 39827.96, 40074.94, 40059.84]
+}
+
+high = pd.Series(data['high'])
+low = pd.Series(data['low'])
+close = pd.Series(data['close'])
+result = ta.atr(high, low, close, length=14)
+print(result)
+*/
+func Test_TR_and_RMA(t *testing.T) {
+	var bytes = []byte(`{
+		"high": [40145.0, 40186.36, 40196.39, 40344.6, 40245.48, 40273.24, 40464.0, 40699.0, 40627.48, 40436.31, 40370.0, 40376.8, 40227.03, 40056.52, 39721.7, 39597.94, 39750.15, 39927.0, 40289.02, 40189.0], 
+		"low": [39870.71, 39834.98, 39866.31, 40108.31, 40016.09, 40094.66, 40105.0, 40196.48, 40154.99, 39800.0, 39959.21, 39922.98, 39940.02, 39632.0, 39261.39, 39254.63, 39473.91, 39555.51, 39819.0, 40006.84],
+		"close": [40105.78, 39935.23, 40183.97, 40182.03, 40212.26, 40149.99, 40378.0, 40618.37, 40401.03, 39990.39, 40179.13, 40097.23, 40014.72, 39667.85, 39303.1, 39519.99, 39693.79, 39827.96, 40074.94, 40059.84]
+	}`)
+
+	var buildKLines = func(bytes []byte) (kLines []types.KLine) {
+		var prices map[string][]fixedpoint.Value
+		_ = json.Unmarshal(bytes, &prices)
+		for i, h := range prices["high"] {
+			kLine := types.KLine{High: h, Low: prices["low"][i], Close: prices["close"][i]}
+			kLines = append(kLines, kLine)
+		}
+		return kLines
+	}
+
+	tests := []struct {
+		name   string
+		kLines []types.KLine
+		window int
+		want   float64
+	}{
+		{
+			name:   "test_binance_btcusdt_1h",
+			kLines: buildKLines(bytes),
+			window: 14,
+			want:   367.913903,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			stream := &types.StandardStream{}
+
+			kLines := KLines(stream)
+			atr := TR2(kLines)
+			rma := RMA2(atr, tt.window, true)
+
+			for _, k := range tt.kLines {
+				stream.EmitKLineClosed(k)
+			}
+
+			got := rma.Last()
+			diff := math.Trunc((got-tt.want)*100) / 100
+			if diff != 0 {
+				t.Errorf("RMA(TR()) = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}

pkg/strategy/pivotshort/failedbreakhigh.go

@@ -341,7 +341,7 @@ func (s *FailedBreakHigh) detectMacdDivergence() {
 	var histogramPivots floats.Slice
 	for i := pivotWindow; i > 0 && i < len(histogramValues); i++ {
 		// find positive histogram and the top
-		pivot, ok := floats.CalculatePivot(histogramValues[0:i], pivotWindow, pivotWindow, func(a, pivot float64) bool {
+		pivot, ok := floats.FindPivot(histogramValues[0:i], pivotWindow, pivotWindow, func(a, pivot float64) bool {
 			return pivot > 0 && pivot > a
 		})
 		if ok {