Merge pull request #1445 from c9s/feature/xdepthmaker

IMPROVE: [xdepthmaker] add more improvements

.github/workflows/go.yml

@@ -52,7 +52,7 @@ jobs:
         # auto-start: "false"
 
     - name: Set up Go
-      uses: actions/setup-go@v3
+      uses: actions/setup-go@v4
       with:
         go-version: ${{ matrix.go-version }}
 

.github/workflows/golang-lint.yml

@@ -12,11 +12,11 @@ jobs:
     name: lint
     runs-on: ubuntu-latest
     steps:
-      - uses: actions/setup-go@v3
+      - uses: actions/setup-go@v4
         with:
-          go-version: 1.18
+          go-version: 1.21
       - uses: actions/checkout@v3
       - name: golangci-lint
         uses: golangci/golangci-lint-action@v3
         with:
-          version: v1.46.2
+          version: v1.54

pkg/bbgo/activeorderbook.go

@@ -392,10 +392,10 @@ func (b *ActiveOrderBook) add(order types.Order) {
 	if pendingOrder, ok := b.pendingOrderUpdates.Get(order.OrderID); ok {
 		// if the pending order update time is newer than the adding order
 		// we should use the pending order rather than the adding order.
-		// if pending order is older, than we should add the new one, and drop the pending order
+		// if the pending order is older, then we should add the new one, and drop the pending order
-		log.Infof("found pending order update")
+		log.Debugf("found pending order update: %+v", pendingOrder)
 		if isNewerOrderUpdate(pendingOrder, order) {
-			log.Infof("pending order update is newer")
+			log.Debugf("pending order update is newer: %+v", pendingOrder)
 			order = pendingOrder
 		}
 

pkg/strategy/xdepthmaker/strategy.go

@@ -13,6 +13,7 @@ import (
 
 	"github.com/c9s/bbgo/pkg/bbgo"
 	"github.com/c9s/bbgo/pkg/core"
+	"github.com/c9s/bbgo/pkg/exchange/retry"
 	"github.com/c9s/bbgo/pkg/fixedpoint"
 	"github.com/c9s/bbgo/pkg/types"
 	"github.com/c9s/bbgo/pkg/util"
@@ -24,7 +25,7 @@ var defaultMargin = fixedpoint.NewFromFloat(0.003)
 
 var Two = fixedpoint.NewFromInt(2)
 
-const priceUpdateTimeout = 30 * time.Second
+const priceUpdateTimeout = 5 * time.Minute
 
 const ID = "xdepthmaker"
 
@@ -237,7 +238,7 @@ type Strategy struct {
 
 	lastPrice fixedpoint.Value
 
-	stopC chan struct{}
+	stopC, authedC chan struct{}
 }
 
 func (s *Strategy) ID() string {
@@ -365,15 +366,40 @@ func (s *Strategy) CrossRun(
 		go s.runTradeRecover(ctx)
 	}
 
+	s.authedC = make(chan struct{}, 2)
+	bindAuthSignal(ctx, s.makerSession.UserDataStream, s.authedC)
+	bindAuthSignal(ctx, s.hedgeSession.UserDataStream, s.authedC)
+
 	go func() {
+		log.Infof("waiting for user data stream to get authenticated")
+		select {
+		case <-ctx.Done():
+			return
+		case <-s.authedC:
+		}
+
+		select {
+		case <-ctx.Done():
+			return
+		case <-s.authedC:
+		}
+
+		log.Infof("user data stream authenticated, start placing orders...")
+
 		posTicker := time.NewTicker(util.MillisecondsJitter(s.HedgeInterval.Duration(), 200))
 		defer posTicker.Stop()
 
 		fullReplenishTicker := time.NewTicker(util.MillisecondsJitter(s.FullReplenishInterval.Duration(), 200))
 		defer fullReplenishTicker.Stop()
 
+		// clean up the previous open orders
+		if err := s.cleanUpOpenOrders(ctx); err != nil {
+			log.WithError(err).Errorf("error cleaning up open orders")
+		}
+
 		s.updateQuote(ctx, 0)
 
+		lastOrderReplenishTime := time.Now()
 		for {
 			select {
 
@@ -387,6 +413,7 @@ func (s *Strategy) CrossRun(
 
 			case <-fullReplenishTicker.C:
 				s.updateQuote(ctx, 0)
+				lastOrderReplenishTime = time.Now()
 
 			case sig, ok := <-s.pricingBook.C:
 				// when any book change event happened
@@ -394,6 +421,10 @@ func (s *Strategy) CrossRun(
 					return
 				}
 
+				if time.Since(lastOrderReplenishTime) < 10*time.Second {
+					continue
+				}
+
 				switch sig.Type {
 				case types.BookSignalSnapshot:
 					s.updateQuote(ctx, 0)
@@ -402,6 +433,8 @@ func (s *Strategy) CrossRun(
 					s.updateQuote(ctx, 5)
 				}
 
+				lastOrderReplenishTime = time.Now()
+
 			case <-posTicker.C:
 				// For positive position and positive covered position:
 				// uncover position = +5 - +3 (covered position) = 2
@@ -598,31 +631,61 @@ func (s *Strategy) runTradeRecover(ctx context.Context) {
 	}
 }
 
-func (s *Strategy) generateMakerOrders(pricingBook *types.StreamOrderBook, maxLayer int) ([]types.SubmitOrder, error) {
+func (s *Strategy) generateMakerOrders(
-	bestBid, bestAsk, hasPrice := pricingBook.BestBidAndAsk()
+	pricingBook *types.StreamOrderBook, maxLayer int, availableBase fixedpoint.Value, availableQuote fixedpoint.Value,
+) ([]types.SubmitOrder, error) {
+	_, _, hasPrice := pricingBook.BestBidAndAsk()
 	if !hasPrice {
 		return nil, nil
 	}
 
-	bestBidPrice := bestBid.Price
-	bestAskPrice := bestAsk.Price
-
-	lastMidPrice := bestBidPrice.Add(bestAskPrice).Div(Two)
-	_ = lastMidPrice
-
 	var submitOrders []types.SubmitOrder
 	var accumulatedBidQuantity = fixedpoint.Zero
 	var accumulatedAskQuantity = fixedpoint.Zero
 	var accumulatedBidQuoteQuantity = fixedpoint.Zero
 
-	dupPricingBook := pricingBook.CopyDepth(0)
+	// copy the pricing book because during the generation the book data could change
+	dupPricingBook := pricingBook.Copy()
+
+	log.Infof("dupPricingBook: \n\tbids: %+v \n\tasks: %+v",
+		dupPricingBook.SideBook(types.SideTypeBuy),
+		dupPricingBook.SideBook(types.SideTypeSell))
+
+	log.Infof("pricingBook: \n\tbids: %+v \n\tasks: %+v",
+		pricingBook.SideBook(types.SideTypeBuy),
+		pricingBook.SideBook(types.SideTypeSell))
 
 	if maxLayer == 0 || maxLayer > s.NumLayers {
 		maxLayer = s.NumLayers
 	}
 
+	var availableBalances = map[types.SideType]fixedpoint.Value{
+		types.SideTypeBuy:  availableQuote,
+		types.SideTypeSell: availableBase,
+	}
+
 	for _, side := range []types.SideType{types.SideTypeBuy, types.SideTypeSell} {
+		sideBook := dupPricingBook.SideBook(side)
+		if sideBook.Len() == 0 {
+			log.Warnf("orderbook %s side is empty", side)
+			continue
+		}
+
+		availableSideBalance, ok := availableBalances[side]
+		if !ok {
+			log.Warnf("no available balance for side %s side", side)
+			continue
+		}
+
+	layerLoop:
 		for i := 1; i <= maxLayer; i++ {
+			// simple break, we need to check the market minNotional and minQuantity later
+			if !availableSideBalance.Eq(fixedpoint.PosInf) {
+				if availableSideBalance.IsZero() || availableSideBalance.Sign() < 0 {
+					break layerLoop
+				}
+			}
+
 			requiredDepthFloat, err := s.DepthScale.Scale(i)
 			if err != nil {
 				return nil, errors.Wrapf(err, "depthScale scale error")
@@ -631,7 +694,6 @@ func (s *Strategy) generateMakerOrders(pricingBook *types.StreamOrderBook, maxLa
 			// requiredDepth is the required depth in quote currency
 			requiredDepth := fixedpoint.NewFromFloat(requiredDepthFloat)
 
-			sideBook := dupPricingBook.SideBook(side)
 			index := sideBook.IndexByQuoteVolumeDepth(requiredDepth)
 
 			pvs := types.PriceVolumeSlice{}
@@ -641,7 +703,11 @@ func (s *Strategy) generateMakerOrders(pricingBook *types.StreamOrderBook, maxLa
 				pvs = sideBook[0 : index+1]
 			}
 
-			log.Infof("required depth: %f, pvs: %+v", requiredDepth.Float64(), pvs)
+			if len(pvs) == 0 {
+				continue
+			}
+
+			log.Infof("side: %s required depth: %f, pvs: %+v", side, requiredDepth.Float64(), pvs)
 
 			depthPrice, err := averageDepthPrice(pvs)
 			if err != nil {
@@ -678,12 +744,36 @@ func (s *Strategy) generateMakerOrders(pricingBook *types.StreamOrderBook, maxLa
 				accumulatedBidQuantity = accumulatedBidQuantity.Add(quantity)
 				quoteQuantity := fixedpoint.Mul(quantity, depthPrice)
 				quoteQuantity = quoteQuantity.Round(s.makerMarket.PricePrecision, fixedpoint.Up)
+
+				if !availableSideBalance.Eq(fixedpoint.PosInf) && availableSideBalance.Compare(quoteQuantity) <= 0 {
+					quoteQuantity = availableSideBalance
+					quantity = quoteQuantity.Div(depthPrice).Round(s.makerMarket.PricePrecision, fixedpoint.Down)
+				}
+
+				if quantity.Compare(s.makerMarket.MinQuantity) <= 0 || quoteQuantity.Compare(s.makerMarket.MinNotional) <= 0 {
+					break layerLoop
+				}
+
+				availableSideBalance = availableSideBalance.Sub(quoteQuantity)
+
 				accumulatedBidQuoteQuantity = accumulatedBidQuoteQuantity.Add(quoteQuantity)
 
 			case types.SideTypeSell:
 				quantity = quantity.Sub(accumulatedAskQuantity)
-				accumulatedAskQuantity = accumulatedAskQuantity.Add(quantity)
+				quoteQuantity := quantity.Mul(depthPrice)
 
+				// balance check
+				if !availableSideBalance.Eq(fixedpoint.PosInf) && availableSideBalance.Compare(quantity) <= 0 {
+					break layerLoop
+				}
+
+				if quantity.Compare(s.makerMarket.MinQuantity) <= 0 || quoteQuantity.Compare(s.makerMarket.MinNotional) <= 0 {
+					break layerLoop
+				}
+
+				availableSideBalance = availableSideBalance.Sub(quantity)
+
+				accumulatedAskQuantity = accumulatedAskQuantity.Add(quantity)
 			}
 
 			submitOrders = append(submitOrders, types.SubmitOrder{
@@ -747,20 +837,38 @@ func (s *Strategy) updateQuote(ctx context.Context, maxLayer int) {
 	bookLastUpdateTime := s.pricingBook.LastUpdateTime()
 
 	if _, err := s.bidPriceHeartBeat.Update(bestBid); err != nil {
-		log.WithError(err).Errorf("quote update error, %s price not updating, order book last update: %s ago",
+		log.WithError(err).Warnf("quote update error, %s price not updating, order book last update: %s ago",
 			s.Symbol,
 			time.Since(bookLastUpdateTime))
-		return
 	}
 
 	if _, err := s.askPriceHeartBeat.Update(bestAsk); err != nil {
-		log.WithError(err).Errorf("quote update error, %s price not updating, order book last update: %s ago",
+		log.WithError(err).Warnf("quote update error, %s price not updating, order book last update: %s ago",
 			s.Symbol,
 			time.Since(bookLastUpdateTime))
+	}
+
+	balances, err := s.MakerOrderExecutor.Session().Exchange.QueryAccountBalances(ctx)
+	if err != nil {
+		log.WithError(err).Errorf("balance query error")
 		return
 	}
 
-	submitOrders, err := s.generateMakerOrders(s.pricingBook, maxLayer)
+	log.Infof("balances: %+v", balances.NotZero())
+
+	quoteBalance, ok := balances[s.makerMarket.QuoteCurrency]
+	if !ok {
+		return
+	}
+
+	baseBalance, ok := balances[s.makerMarket.BaseCurrency]
+	if !ok {
+		return
+	}
+
+	log.Infof("quote balance: %s, base balance: %s", quoteBalance, baseBalance)
+
+	submitOrders, err := s.generateMakerOrders(s.pricingBook, maxLayer, baseBalance.Available, quoteBalance.Available)
 	if err != nil {
 		log.WithError(err).Errorf("generate order error")
 		return
@@ -780,6 +888,19 @@ func (s *Strategy) updateQuote(ctx context.Context, maxLayer int) {
 	s.orderStore.Add(createdOrders...)
 }
 
+func (s *Strategy) cleanUpOpenOrders(ctx context.Context) error {
+	openOrders, err := retry.QueryOpenOrdersUntilSuccessful(ctx, s.makerSession.Exchange, s.Symbol)
+	if err != nil {
+		return err
+	}
+
+	if err := s.makerSession.Exchange.CancelOrders(ctx, openOrders...); err != nil {
+		return err
+	}
+
+	return nil
+}
+
 func selectSessions2(
 	sessions map[string]*bbgo.ExchangeSession, n1, n2 string,
 ) (s1, s2 *bbgo.ExchangeSession, err error) {
@@ -820,3 +941,14 @@ func min(a, b int) int {
 
 	return b
 }
+
+func bindAuthSignal(ctx context.Context, stream types.Stream, c chan<- struct{}) {
+	stream.OnAuth(func() {
+		select {
+		case <-ctx.Done():
+			return
+		case c <- struct{}{}:
+		default:
+		}
+	})
+}

pkg/strategy/xdepthmaker/strategy_test.go

@@ -9,6 +9,7 @@ import (
 	"github.com/stretchr/testify/assert"
 
 	"github.com/c9s/bbgo/pkg/bbgo"
+	"github.com/c9s/bbgo/pkg/fixedpoint"
 	. "github.com/c9s/bbgo/pkg/testing/testhelper"
 	"github.com/c9s/bbgo/pkg/types"
 )
@@ -44,7 +45,7 @@ func TestStrategy_generateMakerOrders(t *testing.T) {
 	}
 
 	pricingBook := types.NewStreamBook("BTCUSDT")
-	pricingBook.OrderBook.Load(types.SliceOrderBook{
+	pricingBook.Load(types.SliceOrderBook{
 		Symbol: "BTCUSDT",
 		Bids: types.PriceVolumeSlice{
 			{Price: Number("25000.00"), Volume: Number("0.1")},
@@ -61,7 +62,7 @@ func TestStrategy_generateMakerOrders(t *testing.T) {
 		Time: time.Now(),
 	})
 
-	orders, err := s.generateMakerOrders(pricingBook, 0)
+	orders, err := s.generateMakerOrders(pricingBook, 0, fixedpoint.PosInf, fixedpoint.PosInf)
 	assert.NoError(t, err)
 	AssertOrdersPriceSideQuantity(t, []PriceSideQuantityAssert{
 		{Side: types.SideTypeBuy, Price: Number("25000"), Quantity: Number("0.04")},        // =~ $1000.00

pkg/types/orderbook.go

@@ -27,7 +27,8 @@ type MutexOrderBook struct {
 	sync.Mutex
 
 	Symbol string
-	OrderBook OrderBook
+
+	orderBook OrderBook
 }
 
 func NewMutexOrderBook(symbol string) *MutexOrderBook {
@@ -39,20 +40,27 @@ func NewMutexOrderBook(symbol string) *MutexOrderBook {
 
 	return &MutexOrderBook{
 		Symbol:    symbol,
-		OrderBook: book,
+		orderBook: book,
 	}
 }
 
 func (b *MutexOrderBook) IsValid() (ok bool, err error) {
 	b.Lock()
-	ok, err = b.OrderBook.IsValid()
+	ok, err = b.orderBook.IsValid()
 	b.Unlock()
 	return ok, err
 }
 
+func (b *MutexOrderBook) SideBook(sideType SideType) PriceVolumeSlice {
+	b.Lock()
+	sideBook := b.orderBook.SideBook(sideType)
+	b.Unlock()
+	return sideBook
+}
+
 func (b *MutexOrderBook) LastUpdateTime() time.Time {
 	b.Lock()
-	t := b.OrderBook.LastUpdateTime()
+	t := b.orderBook.LastUpdateTime()
 	b.Unlock()
 	return t
 }
@@ -60,8 +68,8 @@ func (b *MutexOrderBook) LastUpdateTime() time.Time {
 func (b *MutexOrderBook) BestBidAndAsk() (bid, ask PriceVolume, ok bool) {
 	var ok1, ok2 bool
 	b.Lock()
-	bid, ok1 = b.OrderBook.BestBid()
+	bid, ok1 = b.orderBook.BestBid()
-	ask, ok2 = b.OrderBook.BestAsk()
+	ask, ok2 = b.orderBook.BestAsk()
 	b.Unlock()
 	ok = ok1 && ok2
 	return bid, ask, ok
@@ -69,48 +77,49 @@ func (b *MutexOrderBook) BestBidAndAsk() (bid, ask PriceVolume, ok bool) {
 
 func (b *MutexOrderBook) BestBid() (pv PriceVolume, ok bool) {
 	b.Lock()
-	pv, ok = b.OrderBook.BestBid()
+	pv, ok = b.orderBook.BestBid()
 	b.Unlock()
 	return pv, ok
 }
 
 func (b *MutexOrderBook) BestAsk() (pv PriceVolume, ok bool) {
 	b.Lock()
-	pv, ok = b.OrderBook.BestAsk()
+	pv, ok = b.orderBook.BestAsk()
 	b.Unlock()
 	return pv, ok
 }
 
 func (b *MutexOrderBook) Load(book SliceOrderBook) {
 	b.Lock()
-	b.OrderBook.Load(book)
+	b.orderBook.Load(book)
 	b.Unlock()
 }
 
 func (b *MutexOrderBook) Reset() {
 	b.Lock()
-	b.OrderBook.Reset()
+	b.orderBook.Reset()
 	b.Unlock()
 }
 
 func (b *MutexOrderBook) CopyDepth(depth int) OrderBook {
 	b.Lock()
-	book := b.OrderBook.CopyDepth(depth)
+	defer b.Unlock()
-	b.Unlock()
+
-	return book
+	return b.orderBook.CopyDepth(depth)
 }
 
 func (b *MutexOrderBook) Copy() OrderBook {
 	b.Lock()
-	book := b.OrderBook.Copy()
+	defer b.Unlock()
-	b.Unlock()
+
-	return book
+	return b.orderBook.Copy()
 }
 
 func (b *MutexOrderBook) Update(update SliceOrderBook) {
 	b.Lock()
-	b.OrderBook.Update(update)
+	defer b.Unlock()
-	b.Unlock()
+
+	b.orderBook.Update(update)
 }
 
 type BookSignalType string

pkg/types/price_volume_slice.go

@@ -17,7 +17,7 @@ func (p PriceVolume) Equals(b PriceVolume) bool {
 }
 
 func (p PriceVolume) String() string {
-	return fmt.Sprintf("PriceVolume{ price: %s, volume: %s }", p.Price.String(), p.Volume.String())
+	return fmt.Sprintf("PriceVolume{ Price: %s, Volume: %s }", p.Price.String(), p.Volume.String())
 }
 
 type PriceVolumeSlice []PriceVolume

pkg/types/sliceorderbook_test.go

@@ -0,0 +1,27 @@
+package types
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestSliceOrderBook_CopyDepth(t *testing.T) {
+	b := &SliceOrderBook{
+		Bids: PriceVolumeSlice{
+			{Price: number(0.119), Volume: number(100.0)},
+			{Price: number(0.118), Volume: number(100.0)},
+			{Price: number(0.117), Volume: number(100.0)},
+			{Price: number(0.116), Volume: number(100.0)},
+		},
+		Asks: PriceVolumeSlice{
+			{Price: number(0.120), Volume: number(100.0)},
+			{Price: number(0.121), Volume: number(100.0)},
+			{Price: number(0.122), Volume: number(100.0)},
+		},
+	}
+
+	copied := b.CopyDepth(0)
+	assert.Equal(t, 3, len(copied.SideBook(SideTypeSell)))
+	assert.Equal(t, 4, len(copied.SideBook(SideTypeBuy)))
+}