Merge pull request #1192 from c9s/feature/indicator-improvements

IMPROVE: improve order executor error checking, trailing stop and indicators

pkg/bbgo/exit_trailing_stop.go

@@ -59,15 +59,11 @@ func (s *TrailingStop2) Bind(session *ExchangeSession, orderExecutor *GeneralOrd
 	}))
 
 	if !IsBackTesting && enableMarketTradeStop {
-		session.MarketDataStream.OnMarketTrade(func(trade types.Trade) {
-			if trade.Symbol != position.Symbol {
-				return
-			}
-
+		session.MarketDataStream.OnMarketTrade(types.TradeWith(position.Symbol, func(trade types.Trade) {
 			if err := s.checkStopPrice(trade.Price, position); err != nil {
 				log.WithError(err).Errorf("error")
 			}
-		})
+		}))
 	}
 }
 
@@ -78,8 +74,10 @@ func (s *TrailingStop2) getRatio(price fixedpoint.Value, position *types.Positio
 		// for short position, it's:
 		//  (avg_cost - price) / avg_cost
 		return position.AverageCost.Sub(price).Div(position.AverageCost), nil
+
 	case types.SideTypeSell:
 		return price.Sub(position.AverageCost).Div(position.AverageCost), nil
+
 	default:
 		if position.IsLong() {
 			return price.Sub(position.AverageCost).Div(position.AverageCost), nil
@@ -174,12 +172,15 @@ func (s *TrailingStop2) triggerStop(price fixedpoint.Value) error {
 		s.activated = false
 		s.latestHigh = fixedpoint.Zero
 	}()
-	Notify("[TrailingStop] %s %s stop loss triggered. price: %f callback rate: %f", s.Symbol, s, price.Float64(), s.CallbackRate.Float64())
+
+	Notify("[TrailingStop] %s %s tailingStop is triggered. price: %f callbackRate: %s", s.Symbol, s.ActivationRatio.Percentage(), price.Float64(), s.CallbackRate.Percentage())
 	ctx := context.Background()
 	p := fixedpoint.One
 	if !s.ClosePosition.IsZero() {
 		p = s.ClosePosition
 	}
 
-	return s.orderExecutor.ClosePosition(ctx, p, "trailingStop")
+	tagName := fmt.Sprintf("trailingStop:activation=%s,callback=%s", s.ActivationRatio.Percentage(), s.CallbackRate.Percentage())
+
+	return s.orderExecutor.ClosePosition(ctx, p, tagName)
 }

pkg/bbgo/exit_trailing_stop_test.go

@@ -41,7 +41,7 @@ func TestTrailingStop_ShortPosition(t *testing.T) {
 		Type:             types.OrderTypeMarket,
 		Market:           market,
 		Quantity:         fixedpoint.NewFromFloat(1.0),
-		Tag:              "trailingStop",
+		Tag:              "trailingStop:activation=1%,callback=1%",
 		MarginSideEffect: types.SideEffectTypeAutoRepay,
 	})
 
@@ -119,7 +119,7 @@ func TestTrailingStop_LongPosition(t *testing.T) {
 		Type:             types.OrderTypeMarket,
 		Market:           market,
 		Quantity:         fixedpoint.NewFromFloat(1.0),
-		Tag:              "trailingStop",
+		Tag:              "trailingStop:activation=1%,callback=1%",
 		MarginSideEffect: types.SideEffectTypeAutoRepay,
 	})
 

pkg/bbgo/order_executor_general.go

@@ -268,7 +268,7 @@ func (e *GeneralOrderExecutor) reduceQuantityAndSubmitOrder(ctx context.Context,
 
 		submitOrder.Quantity = q
 		if e.position.Market.IsDustQuantity(submitOrder.Quantity, price) {
-			return nil, types.NewZeroAssetError(fmt.Errorf("dust quantity"))
+			return nil, types.NewZeroAssetError(fmt.Errorf("dust quantity, quantity = %f, price = %f", submitOrder.Quantity.Float64(), price.Float64()))
 		}
 
 		createdOrder, err2 := e.SubmitOrders(ctx, submitOrder)
@@ -334,10 +334,15 @@ func (e *GeneralOrderExecutor) NewOrderFromOpenPosition(ctx context.Context, opt
 				return nil, err
 			}
 
+			if price.IsZero() {
+				return nil, errors.New("unable to calculate quantity: zero price given")
+			}
+
 			quantity = quoteQuantity.Div(price)
 		}
+
 		if e.position.Market.IsDustQuantity(quantity, price) {
-			log.Warnf("dust quantity: %v", quantity)
+			log.Errorf("can not submit order: dust quantity, quantity = %f, price = %f", quantity.Float64(), price.Float64())
 			return nil, nil
 		}
 
@@ -389,9 +394,11 @@ func (e *GeneralOrderExecutor) OpenPosition(ctx context.Context, options OpenPos
 	if err != nil {
 		return nil, err
 	}
+
 	if submitOrder == nil {
 		return nil, nil
 	}
+
 	price := options.Price
 
 	side := "long"
@@ -399,7 +406,7 @@ func (e *GeneralOrderExecutor) OpenPosition(ctx context.Context, options OpenPos
 		side = "short"
 	}
 
-	Notify("Opening %s %s position with quantity %v at price %v", e.position.Symbol, side, submitOrder.Quantity, price)
+	Notify("Opening %s %s position with quantity %f at price %f", e.position.Symbol, side, submitOrder.Quantity.Float64(), price.Float64())
 
 	createdOrder, err := e.SubmitOrders(ctx, *submitOrder)
 	if err == nil {

pkg/indicator/float64series.go

@@ -0,0 +1,71 @@
+package indicator
+
+import (
+	"github.com/c9s/bbgo/pkg/datatype/floats"
+	"github.com/c9s/bbgo/pkg/types"
+)
+
+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(i int) float64 {
+	return f.slice.Last(i)
+}
+
+func (f *Float64Series) Index(i int) float64 {
+	return f.Last(i)
+}
+
+func (f *Float64Series) Length() int {
+	return len(f.slice)
+}
+
+func (f *Float64Series) Slice() floats.Slice {
+	return f.slice
+}
+
+func (f *Float64Series) PushAndEmit(x float64) {
+	f.slice.Push(x)
+	f.EmitUpdate(x)
+}
+
+func (f *Float64Series) Subscribe(source Float64Source, c func(x float64)) {
+	if sub, ok := source.(Float64Subscription); ok {
+		sub.AddSubscriber(c)
+	} else {
+		source.OnUpdate(c)
+	}
+}
+
+// Bind binds the source event to the target (Float64Calculator)
+// A Float64Calculator should be able to calculate the float64 result from a single float64 argument input
+func (f *Float64Series) Bind(source Float64Source, target Float64Calculator) {
+	var c func(x float64)
+
+	// optimize the truncation check
+	trc, canTruncate := target.(Float64Truncator)
+	if canTruncate {
+		c = func(x float64) {
+			y := target.Calculate(x)
+			target.PushAndEmit(y)
+			trc.Truncate()
+		}
+	} else {
+		c = func(x float64) {
+			y := target.Calculate(x)
+			target.PushAndEmit(y)
+		}
+	}
+
+	f.Subscribe(source, c)
+}

pkg/indicator/float64updater.go

@@ -1,72 +1,6 @@
 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(i int) float64 {
-	return f.slice.Last(i)
-}
-
-func (f *Float64Series) Index(i int) float64 {
-	return f.Last(i)
-}
-
-func (f *Float64Series) Length() int {
-	return len(f.slice)
-}
-
-func (f *Float64Series) PushAndEmit(x float64) {
-	f.slice.Push(x)
-	f.EmitUpdate(x)
-}
-
-func (f *Float64Series) Subscribe(source Float64Source, c func(x float64)) {
-	if sub, ok := source.(Float64Subscription); ok {
-		sub.AddSubscriber(c)
-	} else {
-		source.OnUpdate(c)
-	}
-}
-
-// Bind binds the source event to the target (Float64Calculator)
-// A Float64Calculator should be able to calculate the float64 result from a single float64 argument input
-func (f *Float64Series) Bind(source Float64Source, target Float64Calculator) {
-	var c func(x float64)
-
-	// optimize the truncation check
-	trc, canTruncate := target.(Float64Truncator)
-	if canTruncate {
-		c = func(x float64) {
-			y := target.Calculate(x)
-			target.PushAndEmit(y)
-			trc.Truncate()
-		}
-	} else {
-		c = func(x float64) {
-			y := target.Calculate(x)
-			target.PushAndEmit(y)
-		}
-	}
-
-	f.Subscribe(source, c)
-}

pkg/indicator/v2_atrp.go

@@ -0,0 +1,19 @@
+package indicator
+
+type ATRPStream struct {
+	Float64Series
+}
+
+func ATRP2(source KLineSubscription, window int) *ATRPStream {
+	s := &ATRPStream{}
+	tr := TR2(source)
+	atr := RMA2(tr, window, true)
+	atr.OnUpdate(func(x float64) {
+		// x is the last rma
+		k := source.Last(0)
+		cloze := k.Close.Float64()
+		atrp := x / cloze
+		s.PushAndEmit(atrp)
+	})
+	return s
+}

pkg/indicator/v2_cross.go

@@ -0,0 +1,59 @@
+package indicator
+
+import (
+	"github.com/c9s/bbgo/pkg/datatype/floats"
+)
+
+type CrossType float64
+
+const (
+	CrossOver  CrossType = 1.0
+	CrossUnder CrossType = -1.0
+)
+
+// CrossStream subscribes 2 upstreams, and calculate the cross signal
+type CrossStream struct {
+	Float64Series
+
+	a, b floats.Slice
+}
+
+// Cross creates the CrossStream object:
+//
+// cross := Cross(fastEWMA, slowEWMA)
+func Cross(a, b Float64Source) *CrossStream {
+	s := &CrossStream{
+		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 *CrossStream) calculate() {
+	if s.a.Length() != s.b.Length() {
+		return
+	}
+
+	current := s.a.Last(0) - s.b.Last(0)
+	previous := s.a.Last(1) - s.b.Last(1)
+
+	if previous == 0.0 {
+		return
+	}
+
+	// cross over or cross under
+	if current*previous < 0 {
+		if current > 0 {
+			s.PushAndEmit(float64(CrossOver))
+		} else {
+			s.PushAndEmit(float64(CrossUnder))
+		}
+	}
+}

pkg/indicator/v2_macd.go

@@ -5,8 +5,8 @@ type MACDStream struct {
 
 	shortWindow, longWindow, signalWindow int
 
-	fastEWMA, slowEWMA, signal *EWMAStream
-	histogram                  *SubtractStream
+	FastEWMA, SlowEWMA, Signal *EWMAStream
+	Histogram                  *SubtractStream
 }
 
 func MACD2(source Float64Source, shortWindow, longWindow, signalWindow int) *MACDStream {
@@ -21,9 +21,9 @@ func MACD2(source Float64Source, shortWindow, longWindow, signalWindow int) *MAC
 		shortWindow:    shortWindow,
 		longWindow:     longWindow,
 		signalWindow:   signalWindow,
-		fastEWMA:       fastEWMA,
-		slowEWMA:       slowEWMA,
-		signal:         signal,
-		histogram:      histogram,
+		FastEWMA:       fastEWMA,
+		SlowEWMA:       slowEWMA,
+		Signal:         signal,
+		Histogram:      histogram,
 	}
 }

pkg/indicator/v2_multiply.go

@@ -1,19 +1,42 @@
 package indicator
 
+import "github.com/c9s/bbgo/pkg/datatype/floats"
+
 type MultiplyStream struct {
 	Float64Series
-	multiplier float64
+	a, b floats.Slice
 }
 
-func Multiply(source Float64Source, multiplier float64) *MultiplyStream {
+func Multiply(a, b Float64Source) *MultiplyStream {
 	s := &MultiplyStream{
 		Float64Series: NewFloat64Series(),
-		multiplier:    multiplier,
 	}
-	s.Bind(source, s)
+
+	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 *MultiplyStream) Calculate(v float64) float64 {
-	return v * s.multiplier
+func (s *MultiplyStream) 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.Mul(tailB)
+		for _, f := range tailC {
+			s.slice.Push(f)
+			s.EmitUpdate(f)
+		}
+	}
 }

pkg/indicator/v2_price.go

@@ -6,6 +6,8 @@ import (
 
 type KLineSubscription interface {
 	AddSubscriber(f func(k types.KLine))
+	Length() int
+	Last(i int) *types.KLine
 }
 
 type PriceStream struct {

pkg/types/cross.go

@@ -0,0 +1,56 @@
+package types
+
+// The result structure that maps to the crossing result of `CrossOver` and `CrossUnder`
+// Accessible through BoolSeries interface
+type CrossResult struct {
+	a      Series
+	b      Series
+	isOver bool
+}
+
+func (c *CrossResult) Last() bool {
+	if c.Length() == 0 {
+		return false
+	}
+	if c.isOver {
+		return c.a.Last(0)-c.b.Last(0) > 0 && c.a.Last(1)-c.b.Last(1) < 0
+	} else {
+		return c.a.Last(0)-c.b.Last(0) < 0 && c.a.Last(1)-c.b.Last(1) > 0
+	}
+}
+
+func (c *CrossResult) Index(i int) bool {
+	if i >= c.Length() {
+		return false
+	}
+	if c.isOver {
+		return c.a.Last(i)-c.b.Last(i) > 0 && c.a.Last(i+1)-c.b.Last(i+1) < 0
+	} else {
+		return c.a.Last(i)-c.b.Last(i) < 0 && c.a.Last(i+1)-c.b.Last(i+1) > 0
+	}
+}
+
+func (c *CrossResult) Length() int {
+	la := c.a.Length()
+	lb := c.b.Length()
+	if la > lb {
+		return lb
+	}
+	return la
+}
+
+// a series cross above b series.
+// If in current KLine, a is higher than b, and in previous KLine, a is lower than b, then return true.
+// Otherwise return false.
+// If accessing index <= length, will always return false
+func CrossOver(a Series, b Series) BoolSeries {
+	return &CrossResult{a, b, true}
+}
+
+// a series cross under b series.
+// If in current KLine, a is lower than b, and in previous KLine, a is higher than b, then return true.
+// Otherwise return false.
+// If accessing index <= length, will always return false
+func CrossUnder(a Series, b Series) BoolSeries {
+	return &CrossResult{a, b, false}
+}

pkg/types/indicator.go

@@ -100,61 +100,6 @@ func NextCross(a Series, b Series, lookback int) (int, float64, bool) {
 	return int(math.Ceil(-indexf)), alpha1 + beta1*indexf, true
 }
 
-// The result structure that maps to the crossing result of `CrossOver` and `CrossUnder`
-// Accessible through BoolSeries interface
-type CrossResult struct {
-	a      Series
-	b      Series
-	isOver bool
-}
-
-func (c *CrossResult) Last() bool {
-	if c.Length() == 0 {
-		return false
-	}
-	if c.isOver {
-		return c.a.Last(0)-c.b.Last(0) > 0 && c.a.Last(1)-c.b.Last(1) < 0
-	} else {
-		return c.a.Last(0)-c.b.Last(0) < 0 && c.a.Last(1)-c.b.Last(1) > 0
-	}
-}
-
-func (c *CrossResult) Index(i int) bool {
-	if i >= c.Length() {
-		return false
-	}
-	if c.isOver {
-		return c.a.Last(i)-c.b.Last(i) > 0 && c.a.Last(i+1)-c.b.Last(i+1) < 0
-	} else {
-		return c.a.Last(i)-c.b.Last(i) < 0 && c.a.Last(i+1)-c.b.Last(i+1) > 0
-	}
-}
-
-func (c *CrossResult) Length() int {
-	la := c.a.Length()
-	lb := c.b.Length()
-	if la > lb {
-		return lb
-	}
-	return la
-}
-
-// a series cross above b series.
-// If in current KLine, a is higher than b, and in previous KLine, a is lower than b, then return true.
-// Otherwise return false.
-// If accessing index <= length, will always return false
-func CrossOver(a Series, b Series) BoolSeries {
-	return &CrossResult{a, b, true}
-}
-
-// a series cross under b series.
-// If in current KLine, a is lower than b, and in previous KLine, a is higher than b, then return true.
-// Otherwise return false.
-// If accessing index <= length, will always return false
-func CrossUnder(a Series, b Series) BoolSeries {
-	return &CrossResult{a, b, false}
-}
-
 func Highest(a Series, lookback int) float64 {
 	if lookback > a.Length() {
 		lookback = a.Length()

pkg/types/kline.go

@@ -627,6 +627,16 @@ func (k *KLineSeries) Length() int {
 
 var _ Series = &KLineSeries{}
 
+func TradeWith(symbol string, f func(trade Trade)) func(trade Trade) {
+	return func(trade Trade) {
+		if symbol != "" && trade.Symbol != symbol {
+			return
+		}
+
+		f(trade)
+	}
+}
+
 func KLineWith(symbol string, interval Interval, callback KLineCallback) KLineCallback {
 	return func(k KLine) {
 		if k.Symbol != symbol || (k.Interval != "" && k.Interval != interval) {

pkg/types/seriesbase_imp.go

@@ -85,11 +85,11 @@ func (s *SeriesBase) Dot(b interface{}, limit ...int) float64 {
 	return Dot(s, b, limit...)
 }
 
-func (s *SeriesBase) Array(limit ...int) (result []float64) {
+func (s *SeriesBase) Array(limit ...int) []float64 {
 	return Array(s, limit...)
 }
 
-func (s *SeriesBase) Reverse(limit ...int) (result floats.Slice) {
+func (s *SeriesBase) Reverse(limit ...int) floats.Slice {
 	return Reverse(s, limit...)
 }