xfunding: fix transfer and refactoring more methods

config/xfunding.yaml

@@ -12,20 +12,22 @@ notifications:
 sessions:
   binance:
     exchange: binance
-    envVarPrefix: binance
+    envVarPrefix: BINANCE
+
+  binance_futures:
+    exchange: binance
+    envVarPrefix: BINANCE
     futures: true
 
-exchangeStrategies:
-- on: binance
-  funding:
+crossExchangeStrategies:
+
+- xfunding:
+    spotSession: binance
+    futuresSession: binance_futures
     symbol: ETHUSDT
-    quantity: 0.0001
-    fundingRate:
-      high: 0.01%
-    supportDetection:
-      - interval: 1m
-        movingAverageType: EMA
-        movingAverageIntervalWindow:
-          interval: 15m
-          window: 60
-        minVolume: 8_000
+    leverage: 1.0
+    incrementalQuoteQuantity: 11
+    quoteInvestment: 110
+    shortFundingRate:
+      high: 0.000%
+      low: -0.01%

go.mod

@@ -13,6 +13,7 @@ require (
 	github.com/c9s/rockhopper v1.2.2-0.20220617053729-ffdc87df194b
 	github.com/cenkalti/backoff/v4 v4.2.0
 	github.com/cheggaaa/pb/v3 v3.0.8
+	github.com/cloudflare/cfssl v0.0.0-20190808011637-b1ec8c586c2a
 	github.com/codingconcepts/env v0.0.0-20200821220118-a8fbf8d84482
 	github.com/ethereum/go-ethereum v1.10.23
 	github.com/evanphx/json-patch/v5 v5.6.0

go.sum

@@ -103,6 +103,7 @@ github.com/chzyer/logex v1.1.10/go.mod h1:+Ywpsq7O8HXn0nuIou7OrIPyXbp3wmkHB+jjWR
 github.com/chzyer/readline v0.0.0-20180603132655-2972be24d48e/go.mod h1:nSuG5e5PlCu98SY8svDHJxuZscDgtXS6KTTbou5AhLI=
 github.com/chzyer/test v0.0.0-20180213035817-a1ea475d72b1/go.mod h1:Q3SI9o4m/ZMnBNeIyt5eFwwo7qiLfzFZmjNmxjkiQlU=
 github.com/client9/misspell v0.3.4/go.mod h1:qj6jICC3Q7zFZvVWo7KLAzC3yx5G7kyvSDkc90ppPyw=
+github.com/cloudflare/cfssl v0.0.0-20190808011637-b1ec8c586c2a h1:ym8P2+ZvUvVtpLzy8wFLLvdggUIU31mvldvxixQQI2o=
 github.com/cloudflare/cfssl v0.0.0-20190808011637-b1ec8c586c2a/go.mod h1:yMWuSON2oQp+43nFtAV/uvKQIFpSPerB57DCt9t8sSA=
 github.com/cncf/udpa/go v0.0.0-20191209042840-269d4d468f6f/go.mod h1:M8M6+tZqaGXZJjfX53e64911xZQV5JYwmTeXPW+k8Sc=
 github.com/cncf/udpa/go v0.0.0-20201120205902-5459f2c99403/go.mod h1:WmhPx2Nbnhtbo57+VJT5O0JRkEi1Wbu0z5j0R8u5Hbk=

pkg/strategy/drift/strategy.go

@@ -177,7 +177,7 @@ func (s *Strategy) SubmitOrder(ctx context.Context, submitOrder types.SubmitOrde
 	if err != nil {
 		return nil, err
 	}
-	createdOrders, errIdx, err := bbgo.BatchPlaceOrder(ctx, s.Session.Exchange, formattedOrder)
+	createdOrders, errIdx, err := bbgo.BatchPlaceOrder(ctx, s.Session.Exchange, nil, formattedOrder)
 	if len(errIdx) > 0 {
 		return nil, err
 	}

pkg/strategy/xfunding/positionaction_string.go

@@ -0,0 +1,25 @@
+// Code generated by "stringer -type=PositionAction"; DO NOT EDIT.
+
+package xfunding
+
+import "strconv"
+
+func _() {
+	// An "invalid array index" compiler error signifies that the constant values have changed.
+	// Re-run the stringer command to generate them again.
+	var x [1]struct{}
+	_ = x[PositionNoOp-0]
+	_ = x[PositionOpening-1]
+	_ = x[PositionClosing-2]
+}
+
+const _PositionAction_name = "PositionNoOpPositionOpeningPositionClosing"
+
+var _PositionAction_index = [...]uint8{0, 12, 27, 42}
+
+func (i PositionAction) String() string {
+	if i < 0 || i >= PositionAction(len(_PositionAction_index)-1) {
+		return "PositionAction(" + strconv.FormatInt(int64(i), 10) + ")"
+	}
+	return _PositionAction_name[_PositionAction_index[i]:_PositionAction_index[i+1]]
+}

pkg/strategy/xfunding/strategy.go

@@ -5,11 +5,13 @@ import (
 	"errors"
 	"fmt"
 	"strings"
+	"time"
 
 	"github.com/sirupsen/logrus"
 
 	"github.com/c9s/bbgo/pkg/exchange/binance"
 	"github.com/c9s/bbgo/pkg/fixedpoint"
+	"github.com/c9s/bbgo/pkg/util/backoff"
 
 	"github.com/c9s/bbgo/pkg/bbgo"
 	"github.com/c9s/bbgo/pkg/types"
@@ -17,6 +19,7 @@ import (
 
 const ID = "xfunding"
 
+//go:generate stringer -type=PositionAction
 type PositionAction int
 
 const (
@@ -34,13 +37,18 @@ func init() {
 	bbgo.RegisterStrategy(ID, &Strategy{})
 }
 
+// Strategy is the xfunding fee strategy
+// Right now it only supports short position in the USDT futures account.
+// When opening the short position, it uses spot account to buy inventory, then transfer the inventory to the futures account as collateral assets.
 type Strategy struct {
 	Environment *bbgo.Environment
 
 	// These fields will be filled from the config file (it translates YAML to JSON)
 	Symbol string       `json:"symbol"`
 	Market types.Market `json:"-"`
-	Quantity fixedpoint.Value `json:"quantity,omitempty"`
+
+	// Leverage is the leverage of the futures position
+	Leverage fixedpoint.Value `json:"leverage,omitempty"`
 
 	// IncrementalQuoteQuantity is used for opening position incrementally with a small fixed quote quantity
 	// for example, 100usdt per order
@@ -129,6 +137,11 @@ func (s *Strategy) Subscribe(session *bbgo.ExchangeSession) {
 	}
 }
 
+func (s *Strategy) Defaults() error {
+	s.Leverage = fixedpoint.One
+	return nil
+}
+
 func (s *Strategy) Validate() error {
 	if len(s.Symbol) == 0 {
 		return errors.New("symbol is required")
@@ -236,8 +249,8 @@ func (s *Strategy) CrossRun(ctx context.Context, orderExecutionRouter bbgo.Order
 		if s.positionType == types.PositionShort {
 			switch s.positionAction {
 			case PositionOpening:
-				if trade.Side != types.SideTypeSell {
-					log.Errorf("unexpected trade side: %+v, expecting SELL trade", trade)
+				if trade.Side != types.SideTypeBuy {
+					log.Errorf("unexpected trade side: %+v, expecting BUY trade", trade)
 					return
 				}
 
@@ -248,15 +261,20 @@ func (s *Strategy) CrossRun(ctx context.Context, orderExecutionRouter bbgo.Order
 				}
 
 				// 1) if we have trade, try to query the balance and transfer the balance to the futures wallet account
-				// balances, err := binanceSpot.QueryAccountBalances(ctx)
+				// TODO: handle missing trades here. If the process crashed during the transfer, how to recover?
+				if err := backoff.RetryGeneric(ctx, func() error {
+					return s.transferIn(ctx, binanceSpot, trade)
+				}); err != nil {
+					log.WithError(err).Errorf("transfer in retry failed")
+					return
+				}
 
 				// 2) transferred successfully, sync futures position
-
-				// 3) compare spot position and futures position, increase the position size until they are the same size
+				// compare spot position and futures position, increase the position size until they are the same size
 
 			case PositionClosing:
-				if trade.Side != types.SideTypeBuy {
-					log.Errorf("unexpected trade side: %+v, expecting BUY trade", trade)
+				if trade.Side != types.SideTypeSell {
+					log.Errorf("unexpected trade side: %+v, expecting SELL trade", trade)
 					return
 				}
 
@@ -267,42 +285,167 @@ func (s *Strategy) CrossRun(ctx context.Context, orderExecutionRouter bbgo.Order
 	s.futuresOrderExecutor = s.allocateOrderExecutor(ctx, s.futuresSession, instanceID, s.FuturesPosition)
 
 	s.futuresSession.MarketDataStream.OnKLineClosed(types.KLineWith(s.Symbol, types.Interval1m, func(kline types.KLine) {
+		// s.queryAndDetectPremiumIndex(ctx, binanceFutures)
+	}))
+
+	go func() {
+		ticker := time.NewTicker(10 * time.Second)
+		defer ticker.Stop()
+
+		for {
+			select {
+			case <-ctx.Done():
+				return
+
+			case <-ticker.C:
+				s.queryAndDetectPremiumIndex(ctx, binanceFutures)
+
+			}
+		}
+	}()
+
+	// TODO: use go routine and time.Ticker to trigger spot sync and futures sync
+	/*
+		s.spotSession.MarketDataStream.OnKLineClosed(types.KLineWith(s.Symbol, types.Interval1m, func(k types.KLine) {
+		}))
+	*/
+
+	return nil
+}
+
+func (s *Strategy) queryAndDetectPremiumIndex(ctx context.Context, binanceFutures *binance.Exchange) {
 	premiumIndex, err := binanceFutures.QueryPremiumIndex(ctx, s.Symbol)
 	if err != nil {
 		log.WithError(err).Error("premium index query error")
 		return
 	}
 
-		s.detectPremiumIndex(premiumIndex)
-	}))
+	log.Infof("premiumIndex: %+v", premiumIndex)
 
-	s.spotSession.MarketDataStream.OnKLineClosed(types.KLineWith(s.Symbol, types.Interval1m, func(k types.KLine) {
-		// TODO: use go routine and time.Ticker
+	if changed := s.detectPremiumIndex(premiumIndex); changed {
+		log.Infof("position action: %s %s", s.positionType, s.positionAction.String())
 		s.triggerPositionAction(ctx)
-	}))
-
-	return nil
+	}
 }
 
 // TODO: replace type binance.Exchange with an interface
 func (s *Strategy) transferIn(ctx context.Context, ex *binance.Exchange, trade types.Trade) error {
+	currency := s.spotMarket.BaseCurrency
+
+	// base asset needs BUY trades
+	if trade.Side == types.SideTypeSell {
+		return nil
+	}
+
 	balances, err := ex.QueryAccountBalances(ctx)
 	if err != nil {
 		return err
 	}
 
-	b, ok := balances[s.spotMarket.BaseCurrency]
+	b, ok := balances[currency]
 	if !ok {
-		return nil
+		return fmt.Errorf("%s balance not found", currency)
 	}
 
-	// TODO: according to the fee, we might not be able to get enough balance greater than the trade quantity
+	// TODO: according to the fee, we might not be able to get enough balance greater than the trade quantity, we can adjust the quantity here
 	if b.Available.Compare(trade.Quantity) >= 0 {
-
+		log.Infof("transfering futures account asset %s %s", trade.Quantity, currency)
+		if err := ex.TransferFuturesAccountAsset(ctx, currency, trade.Quantity, types.TransferIn); err != nil {
+			log.WithError(err).Errorf("spot-to-futures transfer error")
+			return err
 		}
+	}
+
 	return nil
 }
 
+func (s *Strategy) triggerPositionAction(ctx context.Context) {
+	switch s.positionAction {
+	case PositionOpening:
+		s.syncSpotPosition(ctx)
+		s.syncFuturesPosition(ctx)
+	case PositionClosing:
+		s.syncFuturesPosition(ctx)
+		s.syncSpotPosition(ctx)
+	}
+}
+
+func (s *Strategy) syncFuturesPosition(ctx context.Context) {
+	_ = s.futuresOrderExecutor.GracefulCancel(ctx)
+
+	ticker, err := s.futuresSession.Exchange.QueryTicker(ctx, s.Symbol)
+	if err != nil {
+		log.WithError(err).Errorf("can not query ticker")
+		return
+	}
+
+	switch s.positionAction {
+
+	case PositionClosing:
+
+	case PositionOpening:
+
+		if s.positionType != types.PositionShort {
+			return
+		}
+
+		spotBase := s.SpotPosition.GetBase()       // should be positive base quantity here
+		futuresBase := s.FuturesPosition.GetBase() // should be negative base quantity here
+
+		if spotBase.IsZero() {
+			// skip when spot base is zero
+			return
+		}
+
+		log.Infof("position comparision: %s (spot) <=> %s (futures)", spotBase.String(), futuresBase.String())
+
+		if futuresBase.Sign() > 0 {
+			// unexpected error
+			log.Errorf("unexpected futures position (got positive, expecting negative)")
+			return
+		}
+
+		// compare with the spot position and increase the position
+		quoteValue, err := bbgo.CalculateQuoteQuantity(ctx, s.futuresSession, s.futuresMarket.QuoteCurrency, s.Leverage)
+		if err != nil {
+			log.WithError(err).Errorf("can not calculate futures account quote value")
+			return
+		}
+		log.Infof("calculated futures account quote value = %s", quoteValue.String())
+
+		if spotBase.Sign() > 0 && futuresBase.Neg().Compare(spotBase) < 0 {
+			orderPrice := ticker.Sell
+			diffQuantity := spotBase.Sub(futuresBase.Neg().Mul(s.Leverage))
+
+			log.Infof("position diff quantity: %s", diffQuantity.String())
+
+			orderQuantity := fixedpoint.Max(diffQuantity, s.futuresMarket.MinQuantity)
+			orderQuantity = bbgo.AdjustQuantityByMinAmount(orderQuantity, orderPrice, s.futuresMarket.MinNotional)
+			if s.futuresMarket.IsDustQuantity(orderQuantity, orderPrice) {
+				log.Infof("skip futures order with dust quantity %s, market = %+v", orderQuantity.String(), s.futuresMarket)
+				return
+			}
+
+			createdOrders, err := s.futuresOrderExecutor.SubmitOrders(ctx, types.SubmitOrder{
+				Symbol:   s.Symbol,
+				Side:     types.SideTypeSell,
+				Type:     types.OrderTypeLimitMaker,
+				Quantity: orderQuantity,
+				Price:    orderPrice,
+				Market:   s.futuresMarket,
+				// TimeInForce: types.TimeInForceGTC,
+			})
+
+			if err != nil {
+				log.WithError(err).Errorf("can not submit order")
+				return
+			}
+
+			log.Infof("created orders: %+v", createdOrders)
+		}
+	}
+}
+
 func (s *Strategy) syncSpotPosition(ctx context.Context) {
 	ticker, err := s.spotSession.Exchange.QueryTicker(ctx, s.Symbol)
 	if err != nil {
@@ -318,26 +461,32 @@ func (s *Strategy) syncSpotPosition(ctx context.Context) {
 	switch s.positionAction {
 
 	case PositionClosing:
+		// TODO: compare with the futures position and reduce the position
 
 	case PositionOpening:
-		if s.usedQuoteInvestment.IsZero() || s.usedQuoteInvestment.Compare(s.QuoteInvestment) >= 0 {
-			// stop
+		if s.usedQuoteInvestment.Compare(s.QuoteInvestment) >= 0 {
 			return
 		}
 
 		leftQuote := s.QuoteInvestment.Sub(s.usedQuoteInvestment)
-		orderPrice := ticker.Sell
+		orderPrice := ticker.Buy
 		orderQuantity := fixedpoint.Min(s.IncrementalQuoteQuantity, leftQuote).Div(orderPrice)
 		orderQuantity = fixedpoint.Max(orderQuantity, s.spotMarket.MinQuantity)
-		createdOrders, err := s.spotOrderExecutor.SubmitOrders(ctx, types.SubmitOrder{
+
+		_ = s.spotOrderExecutor.GracefulCancel(ctx)
+
+		submitOrder := types.SubmitOrder{
 			Symbol:   s.Symbol,
-			Side:        types.SideTypeSell,
+			Side:     types.SideTypeBuy,
 			Type:     types.OrderTypeLimitMaker,
 			Quantity: orderQuantity,
 			Price:    orderPrice,
 			Market:   s.spotMarket,
-			TimeInForce: types.TimeInForceGTC,
-		})
+		}
+
+		log.Infof("placing spot order: %+v", submitOrder)
+
+		createdOrders, err := s.spotOrderExecutor.SubmitOrders(ctx, submitOrder)
 		if err != nil {
 			log.WithError(err).Errorf("can not submit order")
 			return
@@ -347,30 +496,28 @@ func (s *Strategy) syncSpotPosition(ctx context.Context) {
 	}
 }
 
-func (s *Strategy) triggerPositionAction(ctx context.Context) {
-	switch s.positionAction {
-	case PositionOpening:
-		s.syncSpotPosition(ctx)
-
-	case PositionClosing:
-
-	}
-}
-
-func (s *Strategy) detectPremiumIndex(premiumIndex *types.PremiumIndex) {
+func (s *Strategy) detectPremiumIndex(premiumIndex *types.PremiumIndex) (changed bool) {
 	fundingRate := premiumIndex.LastFundingRate
+
+	log.Infof("last %s funding rate: %s", s.Symbol, fundingRate.Percentage())
+
 	if s.ShortFundingRate != nil {
 		if fundingRate.Compare(s.ShortFundingRate.High) >= 0 {
 			s.positionAction = PositionOpening
 			s.positionType = types.PositionShort
+			changed = true
 		} else if fundingRate.Compare(s.ShortFundingRate.Low) <= 0 {
 			s.positionAction = PositionClosing
+			changed = true
 		}
 	}
+
+	return changed
 }
 
 func (s *Strategy) allocateOrderExecutor(ctx context.Context, session *bbgo.ExchangeSession, instanceID string, position *types.Position) *bbgo.GeneralOrderExecutor {
 	orderExecutor := bbgo.NewGeneralOrderExecutor(session, s.Symbol, ID, instanceID, position)
+	orderExecutor.SetMaxRetries(0)
 	orderExecutor.BindEnvironment(s.Environment)
 	orderExecutor.Bind()
 	orderExecutor.TradeCollector().OnTrade(func(trade types.Trade, _, _ fixedpoint.Value) {

pkg/strategy/xgap/strategy.go

@@ -331,7 +331,7 @@ func (s *Strategy) CrossRun(ctx context.Context, _ bbgo.OrderExecutionRouter, se
 						s.tradingMarket.MinNotional.Mul(NotionModifier).Div(price))
 				}
 
-				createdOrders, _, err := bbgo.BatchPlaceOrder(ctx, tradingSession.Exchange, types.SubmitOrder{
+				createdOrders, _, err := bbgo.BatchPlaceOrder(ctx, tradingSession.Exchange, nil, types.SubmitOrder{
 					Symbol:   s.Symbol,
 					Side:     types.SideTypeBuy,
 					Type:     types.OrderTypeLimit,

pkg/util/backoff/generic.go

@@ -0,0 +1,18 @@
+package backoff
+
+import (
+	"context"
+
+	"github.com/cenkalti/backoff/v4"
+)
+
+var MaxRetries uint64 = 101
+
+func RetryGeneric(ctx context.Context, op backoff.Operation) (err error) {
+	err = backoff.Retry(op, backoff.WithContext(
+		backoff.WithMaxRetries(
+			backoff.NewExponentialBackOff(),
+			MaxRetries),
+		ctx))
+	return err
+}