Add pvd strategy

config/pvdot.yaml

@@ -0,0 +1,21 @@
+---
+notifications:
+  slack:
+    defaultChannel: "bbgo"
+    errorChannel: "bbgo-error"
+
+exchangeStrategies:
+  - on: max
+    pvdot:
+      window: 720
+      interval: 1m
+      baseCurrency: TWD
+      quoteCurrencies:
+       - BTC
+       - ETH
+      ignoreLocked: true
+      threshold: 2%
+      # max amount to buy or sell per order
+      maxAmount: 10_000
+      verbose: true
+      dryRun: true

pkg/cmd/strategy/builtin.go

@@ -22,6 +22,7 @@ import (
 	_ "github.com/c9s/bbgo/pkg/strategy/pivotshort"
 	_ "github.com/c9s/bbgo/pkg/strategy/pricealert"
 	_ "github.com/c9s/bbgo/pkg/strategy/pricedrop"
+	_ "github.com/c9s/bbgo/pkg/strategy/pvd"
 	_ "github.com/c9s/bbgo/pkg/strategy/rebalance"
 	_ "github.com/c9s/bbgo/pkg/strategy/rsmaker"
 	_ "github.com/c9s/bbgo/pkg/strategy/schedule"

pkg/strategy/pvd/index.go

@@ -0,0 +1,108 @@
+package pvd
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"github.com/c9s/bbgo/pkg/bbgo"
+	"github.com/c9s/bbgo/pkg/fixedpoint"
+	"github.com/c9s/bbgo/pkg/types"
+)
+
+var zeroTime time.Time
+
+type PVDotSet struct {
+	types.IntervalWindow
+	session         *bbgo.ExchangeSession
+	BaseCurrency    string
+	QuoteCurrencies []string
+
+	Indicators map[string]*PVDot
+}
+
+func (i *PVDotSet) InitIndicators(ctx context.Context) error {
+	i.Indicators = make(map[string]*PVDot)
+	for _, quoteCurrency := range i.QuoteCurrencies {
+		symbol := quoteCurrency + i.BaseCurrency
+		i.Indicators[quoteCurrency] = &PVDot{Symbol: symbol, IntervalWindow: i.IntervalWindow}
+	}
+
+	fmt.Println(i.QuoteCurrencies)
+
+	for _, indicator := range i.Indicators {
+		endTime := time.Now()
+		options := types.KLineQueryOptions{Limit: i.Window, EndTime: &endTime}
+		klines, err := i.session.Exchange.QueryKLines(ctx, indicator.Symbol, i.Interval, options)
+		if err != nil {
+			return err
+		}
+		indicator.UpdateFromKLines(klines)
+	}
+	return nil
+}
+func (i *PVDotSet) getIndicator(symbol string) (*PVDot, error) {
+	for _, indicator := range i.Indicators {
+		if symbol == indicator.Symbol {
+			return indicator, nil
+		}
+	}
+	return nil, fmt.Errorf("indicator with symbol: %s not found", symbol)
+}
+
+func (i *PVDotSet) Update(kline types.KLine) error {
+	inc, err := i.getIndicator(kline.Symbol)
+	if err != nil {
+		return err
+	}
+	klines := []types.KLine{kline}
+	inc.UpdateFromKLines(klines)
+	return nil
+}
+
+func (i *PVDotSet) TargetWeights() map[string]fixedpoint.Value {
+	targetWeights := make(map[string]fixedpoint.Value)
+	for quoteCurrency, indicator := range i.Indicators {
+		targetWeights[quoteCurrency] = fixedpoint.NewFromFloat(indicator.Last())
+	}
+	return Normalize(targetWeights)
+}
+
+// price volume avg
+type PVDot struct {
+	Symbol string
+
+	types.IntervalWindow
+	Values  types.Float64Slice
+	Prices  types.Float64Slice
+	Volumes types.Float64Slice
+
+	EndTime         time.Time
+	UpdateCallbacks []func(value float64)
+}
+
+func (inc *PVDot) Last() float64 {
+	if len(inc.Values) == 0 {
+		return 0
+	}
+	return inc.Values[len(inc.Values)-1]
+}
+
+func (inc *PVDot) Update(price, volume float64) {
+	inc.Prices.Push(price)
+	inc.Volumes.Push(volume)
+
+	pva := inc.Prices.Tail(inc.Window).Dot(inc.Volumes.Tail(inc.Window)) / float64(inc.Window)
+	inc.Values.Push(pva)
+}
+
+func (inc *PVDot) UpdateFromKLines(klines []types.KLine) {
+
+	for _, k := range klines {
+		if inc.EndTime != zeroTime && k.EndTime.Before(inc.EndTime) {
+			continue
+		}
+		inc.Update(k.Close.Float64(), k.Volume.Float64())
+	}
+	inc.EndTime = klines[len(klines)-1].EndTime.Time()
+}

pkg/strategy/pvd/strategy.go

@@ -0,0 +1,237 @@
+package pvd
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/sirupsen/logrus"
+
+	"github.com/c9s/bbgo/pkg/bbgo"
+	"github.com/c9s/bbgo/pkg/fixedpoint"
+	"github.com/c9s/bbgo/pkg/types"
+)
+
+const ID = "pvdot"
+
+var log = logrus.WithField("strategy", ID)
+
+func init() {
+	bbgo.RegisterStrategy(ID, &Strategy{})
+}
+
+func Sum(m map[string]fixedpoint.Value) fixedpoint.Value {
+	sum := fixedpoint.NewFromFloat(0.0)
+	for _, v := range m {
+		sum = sum.Add(v)
+	}
+	return sum
+}
+
+func Normalize(m map[string]fixedpoint.Value) map[string]fixedpoint.Value {
+	sum := Sum(m)
+	if sum.Float64() == 1.0 {
+		return m
+	}
+
+	normalized := make(map[string]fixedpoint.Value)
+	for k, v := range m {
+		normalized[k] = v.Div(sum)
+	}
+	return normalized
+}
+
+func ElementwiseProduct(m1, m2 map[string]fixedpoint.Value) map[string]fixedpoint.Value {
+	m := make(map[string]fixedpoint.Value)
+	for k, v := range m1 {
+		m[k] = v.Mul(m2[k])
+	}
+	return m
+}
+
+type Strategy struct {
+	Notifiability *bbgo.Notifiability
+
+	Interval        types.Interval   `json:"interval"`
+	Window          int              `json:"window"`
+	BaseCurrency    string           `json:"baseCurrency"`
+	QuoteCurrencies []string         `json:"quoteCurrencies"`
+	Threshold       fixedpoint.Value `json:"threshold"`
+	IgnoreLocked    bool             `json:"ignoreLocked"`
+	Verbose         bool             `json:"verbose"`
+	DryRun          bool             `json:"dryRun"`
+
+	// max amount to buy or sell per order
+	MaxAmount fixedpoint.Value `json:"maxAmount"`
+
+	set PVDotSet
+}
+
+func (s *Strategy) ID() string {
+	return ID
+}
+
+func (s *Strategy) Validate() error {
+	if s.Threshold.Sign() < 0 {
+		return fmt.Errorf("threshold should not less than 0")
+	}
+
+	if s.MaxAmount.Sign() < 0 {
+		return fmt.Errorf("maxAmount shoud not less than 0")
+	}
+
+	return nil
+}
+
+func (s *Strategy) Subscribe(session *bbgo.ExchangeSession) {
+	for _, symbol := range s.getSymbols() {
+		session.Subscribe(types.KLineChannel, symbol, types.SubscribeOptions{Interval: s.Interval})
+	}
+}
+
+func (s *Strategy) Run(ctx context.Context, orderExecutor bbgo.OrderExecutor, session *bbgo.ExchangeSession) error {
+	iw := types.IntervalWindow{Interval: s.Interval, Window: s.Window}
+	s.set = PVDotSet{IntervalWindow: iw, session: session, BaseCurrency: s.BaseCurrency, QuoteCurrencies: s.QuoteCurrencies}
+	err := s.set.InitIndicators(ctx)
+	if err != nil {
+		return err
+	}
+
+	session.MarketDataStream.OnKLineClosed(func(kline types.KLine) {
+		s.set.Update(kline)
+		s.rebalance(ctx, orderExecutor, session)
+	})
+	return nil
+}
+
+func (s *Strategy) rebalance(ctx context.Context, orderExecutor bbgo.OrderExecutor, session *bbgo.ExchangeSession) {
+	targetWeights := s.set.TargetWeights()
+
+	prices, err := s.getPrices(ctx, session, targetWeights)
+	if err != nil {
+		return
+	}
+
+	balances := session.Account.Balances()
+	quantities := s.getQuantities(balances, targetWeights)
+	marketValues := ElementwiseProduct(prices, quantities)
+
+	orders := s.generateSubmitOrders(prices, marketValues, targetWeights)
+	for _, order := range orders {
+		log.Infof("generated submit order: %s", order.String())
+	}
+
+	if s.DryRun {
+		return
+	}
+
+	_, err = orderExecutor.SubmitOrders(ctx, orders...)
+	if err != nil {
+		log.WithError(err).Error("submit order error")
+		return
+	}
+}
+
+func (s *Strategy) getPrices(ctx context.Context, session *bbgo.ExchangeSession, targetWeights map[string]fixedpoint.Value) (map[string]fixedpoint.Value, error) {
+	prices := make(map[string]fixedpoint.Value)
+
+	for currency := range targetWeights {
+		if currency == s.BaseCurrency {
+			prices[currency] = fixedpoint.One
+			continue
+		}
+
+		symbol := currency + s.BaseCurrency
+		ticker, err := session.Exchange.QueryTicker(ctx, symbol)
+		if err != nil {
+			s.Notifiability.Notify("query ticker error: %s", err.Error())
+			log.WithError(err).Error("query ticker error")
+			return prices, err
+		}
+
+		prices[currency] = ticker.Last
+	}
+	return prices, nil
+}
+
+func (s *Strategy) getQuantities(balances types.BalanceMap, targetWeights map[string]fixedpoint.Value) map[string]fixedpoint.Value {
+	quantities := make(map[string]fixedpoint.Value)
+	for currency := range targetWeights {
+		if s.IgnoreLocked {
+			quantities[currency] = balances[currency].Total()
+		} else {
+			quantities[currency] = balances[currency].Available
+		}
+	}
+	return quantities
+}
+
+func (s *Strategy) generateSubmitOrders(prices, marketValues map[string]fixedpoint.Value, targetWeights map[string]fixedpoint.Value) []types.SubmitOrder {
+	var submitOrders []types.SubmitOrder
+
+	currentWeights := Normalize(marketValues)
+	totalValue := Sum(marketValues)
+
+	log.Infof("total value: %f", totalValue.Float64())
+
+	for currency, targetWeight := range targetWeights {
+		if currency == s.BaseCurrency {
+			continue
+		}
+		symbol := currency + s.BaseCurrency
+		currentWeight := currentWeights[currency]
+		currentPrice := prices[currency]
+		log.Infof("%s price: %v, current weight: %v, target weight: %v",
+			symbol,
+			currentPrice,
+			currentWeight,
+			targetWeight)
+
+		// calculate the difference between current weight and target weight
+		// if the difference is less than threshold, then we will not create the order
+		weightDifference := targetWeight.Sub(currentWeight)
+		if weightDifference.Abs().Compare(s.Threshold) < 0 {
+			log.Infof("%s weight distance |%v - %v| = |%v| less than the threshold: %v",
+				symbol,
+				currentWeight,
+				targetWeight,
+				weightDifference,
+				s.Threshold)
+			continue
+		}
+
+		quantity := weightDifference.Mul(totalValue).Div(currentPrice)
+
+		side := types.SideTypeBuy
+		if quantity.Sign() < 0 {
+			side = types.SideTypeSell
+			quantity = quantity.Abs()
+		}
+
+		if s.MaxAmount.Sign() > 0 {
+			quantity = bbgo.AdjustQuantityByMaxAmount(quantity, currentPrice, s.MaxAmount)
+			log.Infof("adjust the quantity %v (%s %s @ %v) by max amount %v",
+				quantity,
+				symbol,
+				side.String(),
+				currentPrice,
+				s.MaxAmount)
+		}
+
+		order := types.SubmitOrder{
+			Symbol:   symbol,
+			Side:     side,
+			Type:     types.OrderTypeMarket,
+			Quantity: quantity}
+
+		submitOrders = append(submitOrders, order)
+	}
+	return submitOrders
+}
+
+func (s *Strategy) getSymbols() []string {
+	var symbols []string
+	for _, c := range s.QuoteCurrencies {
+		symbols = append(symbols, c+s.BaseCurrency)
+	}
+	return symbols
+}

pkg/types/float_slice.go

@@ -0,0 +1,119 @@
+package types
+
+import "math"
+
+type Float64Slice []float64
+
+func (s *Float64Slice) Push(v float64) {
+	*s = append(*s, v)
+}
+
+func (s *Float64Slice) Pop(i int64) (v float64) {
+	v = (*s)[i]
+	*s = append((*s)[:i], (*s)[i+1:]...)
+	return v
+}
+
+func (s Float64Slice) Max() float64 {
+	m := -math.MaxFloat64
+	for _, v := range s {
+		m = math.Max(m, v)
+	}
+	return m
+}
+
+func (s Float64Slice) Min() float64 {
+	m := math.MaxFloat64
+	for _, v := range s {
+		m = math.Min(m, v)
+	}
+	return m
+}
+
+func (s Float64Slice) Sum() (sum float64) {
+	for _, v := range s {
+		sum += v
+	}
+	return sum
+}
+
+func (s Float64Slice) Mean() (mean float64) {
+	return s.Sum() / float64(len(s))
+}
+
+func (s Float64Slice) Tail(size int) Float64Slice {
+	length := len(s)
+	if length <= size {
+		win := make(Float64Slice, length)
+		copy(win, s)
+		return win
+	}
+
+	win := make(Float64Slice, size)
+	copy(win, s[length-size:])
+	return win
+}
+
+func (s Float64Slice) Diff() Float64Slice {
+	var values Float64Slice
+	for i, v := range s {
+		if i == 0 {
+			values.Push(0)
+			continue
+		}
+		values.Push(v - s[i-1])
+	}
+	return values
+}
+
+func (s Float64Slice) PositiveValuesOrZero() Float64Slice {
+	var values Float64Slice
+	for _, v := range s {
+		values.Push(math.Max(v, 0))
+	}
+	return values
+}
+
+func (s Float64Slice) NegativeValuesOrZero() Float64Slice {
+	var values Float64Slice
+	for _, v := range s {
+		values.Push(math.Min(v, 0))
+	}
+	return values
+}
+
+func (s Float64Slice) AbsoluteValues() Float64Slice {
+	var values Float64Slice
+	for _, v := range s {
+		values.Push(math.Abs(v))
+	}
+	return values
+}
+
+func (s Float64Slice) MulScalar(x float64) Float64Slice {
+	var values Float64Slice
+	for _, v := range s {
+		values.Push(v * x)
+	}
+	return values
+}
+
+func (s Float64Slice) DivScalar(x float64) Float64Slice {
+	var values Float64Slice
+	for _, v := range s {
+		values.Push(v / x)
+	}
+	return values
+}
+
+func (s Float64Slice) ElementwiseProduct(other Float64Slice) Float64Slice {
+	var values Float64Slice
+	for i, v := range s {
+		values.Push(v * other[i])
+	}
+	return values
+}
+
+func (s Float64Slice) Dot(other Float64Slice) float64 {
+	return s.ElementwiseProduct(other).Sum()
+}