Merge pull request #1568 from c9s/c9s/xfunding-fix

FIX: [xfunding] add many fixes and improvements

go.mod

@@ -7,7 +7,7 @@ go 1.20
 require (
 	github.com/DATA-DOG/go-sqlmock v1.5.0
 	github.com/Masterminds/squirrel v1.5.3
-	github.com/adshao/go-binance/v2 v2.4.2
+	github.com/adshao/go-binance/v2 v2.4.5
 	github.com/c-bata/goptuna v0.8.1
 	github.com/c9s/requestgen v1.3.6
 	github.com/c9s/rockhopper/v2 v2.0.3-0.20240124055428-2473c6221858
@@ -25,7 +25,7 @@ require (
 	github.com/gofrs/flock v0.8.1
 	github.com/golang/mock v1.6.0
 	github.com/google/uuid v1.4.0
-	github.com/gorilla/websocket v1.5.0
+	github.com/gorilla/websocket v1.5.1
 	github.com/heroku/rollrus v0.2.0
 	github.com/jedib0t/go-pretty/v6 v6.5.3
 	github.com/jmoiron/sqlx v1.3.4
@@ -142,13 +142,13 @@ require (
 	go.opentelemetry.io/otel/metric v0.19.0 // indirect
 	go.opentelemetry.io/otel/trace v0.19.0 // indirect
 	golang.org/x/arch v0.3.0 // indirect
-	golang.org/x/crypto v0.18.0 // indirect
+	golang.org/x/crypto v0.21.0 // indirect
 	golang.org/x/exp v0.0.0-20240112132812-db7319d0e0e3 // indirect
 	golang.org/x/image v0.5.0 // indirect
 	golang.org/x/mod v0.14.0 // indirect
-	golang.org/x/net v0.20.0 // indirect
+	golang.org/x/net v0.22.0 // indirect
-	golang.org/x/sys v0.16.0 // indirect
+	golang.org/x/sys v0.18.0 // indirect
-	golang.org/x/term v0.16.0 // indirect
+	golang.org/x/term v0.18.0 // indirect
 	golang.org/x/text v0.14.0 // indirect
 	golang.org/x/tools v0.17.0 // indirect
 	google.golang.org/appengine v1.6.7 // indirect

go.sum

@@ -52,6 +52,8 @@ github.com/VividCortex/ewma v1.1.1 h1:MnEK4VOv6n0RSY4vtRe3h11qjxL3+t0B8yOL8iMXdc
 github.com/VividCortex/ewma v1.1.1/go.mod h1:2Tkkvm3sRDVXaiyucHiACn4cqf7DpdyLvmxzcbUokwA=
 github.com/adshao/go-binance/v2 v2.4.2 h1:NBNMUyXrci45v3sr0RkZosiBYSw1/yuqCrJNkyEM8U0=
 github.com/adshao/go-binance/v2 v2.4.2/go.mod h1:41Up2dG4NfMXpCldrDPETEtiOq+pHoGsFZ73xGgaumo=
+github.com/adshao/go-binance/v2 v2.4.5 h1:V3KpolmS9a7TLVECSrl2gYm+GGBSxhVk9ILaxvOTOVw=
+github.com/adshao/go-binance/v2 v2.4.5/go.mod h1:41Up2dG4NfMXpCldrDPETEtiOq+pHoGsFZ73xGgaumo=
 github.com/ajstarks/svgo v0.0.0-20180226025133-644b8db467af/go.mod h1:K08gAheRH3/J6wwsYMMT4xOr94bZjxIelGM0+d/wbFw=
 github.com/alecthomas/template v0.0.0-20160405071501-a0175ee3bccc/go.mod h1:LOuyumcjzFXgccqObfd/Ljyb9UuFJ6TxHnclSeseNhc=
 github.com/alecthomas/template v0.0.0-20190718012654-fb15b899a751/go.mod h1:LOuyumcjzFXgccqObfd/Ljyb9UuFJ6TxHnclSeseNhc=
@@ -292,6 +294,8 @@ github.com/gorilla/mux v1.8.0/go.mod h1:DVbg23sWSpFRCP0SfiEN6jmj59UnW/n46BH5rLB7
 github.com/gorilla/websocket v1.4.2/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE=
 github.com/gorilla/websocket v1.5.0 h1:PPwGk2jz7EePpoHN/+ClbZu8SPxiqlu12wZP/3sWmnc=
 github.com/gorilla/websocket v1.5.0/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE=
+github.com/gorilla/websocket v1.5.1 h1:gmztn0JnHVt9JZquRuzLw3g4wouNVzKL15iLr/zn/QY=
+github.com/gorilla/websocket v1.5.1/go.mod h1:x3kM2JMyaluk02fnUJpQuwD2dCS5NDG2ZHL0uE0tcaY=
 github.com/grpc-ecosystem/go-grpc-middleware v1.0.0/go.mod h1:FiyG127CGDf3tlThmgyCl78X/SZQqEOJBCDaAfeWzPs=
 github.com/grpc-ecosystem/go-grpc-prometheus v1.2.0/go.mod h1:8NvIoxWQoOIhqOTXgfV/d3M/q6VIi02HzZEHgUlZvzk=
 github.com/grpc-ecosystem/grpc-gateway v1.9.0/go.mod h1:vNeuVxBJEsws4ogUvrchl83t/GYV9WGTSLVdBhOQFDY=
@@ -715,6 +719,8 @@ golang.org/x/crypto v0.0.0-20210921155107-089bfa567519/go.mod h1:GvvjBRRGRdwPK5y
 golang.org/x/crypto v0.0.0-20220622213112-05595931fe9d/go.mod h1:IxCIyHEi3zRg3s0A5j5BB6A9Jmi73HwBIUl50j+osU4=
 golang.org/x/crypto v0.18.0 h1:PGVlW0xEltQnzFZ55hkuX5+KLyrMYhHld1YHO4AKcdc=
 golang.org/x/crypto v0.18.0/go.mod h1:R0j02AL6hcrfOiy9T4ZYp/rcWeMxM3L6QYxlOuEG1mg=
+golang.org/x/crypto v0.21.0 h1:X31++rzVUdKhX5sWmSOFZxx8UW/ldWx55cbf08iNAMA=
+golang.org/x/crypto v0.21.0/go.mod h1:0BP7YvVV9gBbVKyeTG0Gyn+gZm94bibOW5BjDEYAOMs=
 golang.org/x/exp v0.0.0-20180321215751-8460e604b9de/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA=
 golang.org/x/exp v0.0.0-20180807140117-3d87b88a115f/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA=
 golang.org/x/exp v0.0.0-20190121172915-509febef88a4/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA=
@@ -806,6 +812,8 @@ golang.org/x/net v0.0.0-20211112202133-69e39bad7dc2/go.mod h1:9nx3DQGgdP8bBQD5qx
 golang.org/x/net v0.0.0-20220722155237-a158d28d115b/go.mod h1:XRhObCWvk6IyKnWLug+ECip1KBveYUHfp+8e9klMJ9c=
 golang.org/x/net v0.20.0 h1:aCL9BSgETF1k+blQaYUBx9hJ9LOGP3gAVemcZlf1Kpo=
 golang.org/x/net v0.20.0/go.mod h1:z8BVo6PvndSri0LbOE3hAn0apkU+1YvI6E70E9jsnvY=
+golang.org/x/net v0.22.0 h1:9sGLhx7iRIHEiX0oAJ3MRZMUCElJgy7Br1nO+AMN3Tc=
+golang.org/x/net v0.22.0/go.mod h1:JKghWKKOSdJwpW2GEx0Ja7fmaKnMsbu+MWVZTokSYmg=
 golang.org/x/oauth2 v0.0.0-20180821212333-d2e6202438be/go.mod h1:N/0e6XlmueqKjAGxoOufVs8QHGRruUQn6yWY3a++T0U=
 golang.org/x/oauth2 v0.0.0-20190226205417-e64efc72b421/go.mod h1:gOpvHmFTYa4IltrdGE7lF6nIHvwfUNPOp7c8zoXwtLw=
 golang.org/x/oauth2 v0.0.0-20190604053449-0f29369cfe45/go.mod h1:gOpvHmFTYa4IltrdGE7lF6nIHvwfUNPOp7c8zoXwtLw=
@@ -892,10 +900,14 @@ golang.org/x/sys v0.0.0-20220811171246-fbc7d0a398ab/go.mod h1:oPkhp1MJrh7nUepCBc
 golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.16.0 h1:xWw16ngr6ZMtmxDyKyIgsE93KNKz5HKmMa3b8ALHidU=
 golang.org/x/sys v0.16.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
+golang.org/x/sys v0.18.0 h1:DBdB3niSjOA/O0blCZBqDefyWNYveAYMNF1Wum0DYQ4=
+golang.org/x/sys v0.18.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
 golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
 golang.org/x/term v0.0.0-20210927222741-03fcf44c2211/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8=
 golang.org/x/term v0.16.0 h1:m+B6fahuftsE9qjo0VWp2FW0mB3MTJvR0BaMQrq0pmE=
 golang.org/x/term v0.16.0/go.mod h1:yn7UURbUtPyrVJPGPq404EukNFxcm/foM+bV/bfcDsY=
+golang.org/x/term v0.18.0 h1:FcHjZXDMxI8mM3nwhX9HlKop4C0YQvCVCdwYl2wOtE8=
+golang.org/x/term v0.18.0/go.mod h1:ILwASektA3OnRv7amZ1xhE/KTR+u50pbXfZ03+6Nx58=
 golang.org/x/text v0.0.0-20170915032832-14c0d48ead0c/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
 golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
 golang.org/x/text v0.3.1-0.20180807135948-17ff2d5776d2/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=

pkg/strategy/common/profit_fixer.go

@@ -19,14 +19,11 @@ type ProfitFixerConfig struct {
 
 // ProfitFixer implements a trade-history-based profit fixer
 type ProfitFixer struct {
-	market types.Market
-
 	sessions map[string]types.ExchangeTradeHistoryService
 }
 
-func NewProfitFixer(market types.Market) *ProfitFixer {
+func NewProfitFixer() *ProfitFixer {
 	return &ProfitFixer{
-		market:   market,
 		sessions: make(map[string]types.ExchangeTradeHistoryService),
 	}
 }
@@ -48,7 +45,7 @@ func (f *ProfitFixer) batchQueryTrades(
 	})
 }
 
-func (f *ProfitFixer) aggregateAllTrades(ctx context.Context, market types.Market, since, until time.Time) ([]types.Trade, error) {
+func (f *ProfitFixer) aggregateAllTrades(ctx context.Context, symbol string, since, until time.Time) ([]types.Trade, error) {
 	var mu sync.Mutex
 	var allTrades = make([]types.Trade, 0, 1000)
 
@@ -58,8 +55,8 @@ func (f *ProfitFixer) aggregateAllTrades(ctx context.Context, market types.Marke
 		sessionName := n
 		service := s
 		g.Go(func() error {
-			log.Infof("batch querying %s trade history from %s since %s until %s", market.Symbol, sessionName, since.String(), until.String())
+			log.Infof("batch querying %s trade history from %s since %s until %s", symbol, sessionName, since.String(), until.String())
-			trades, err := f.batchQueryTrades(subCtx, service, f.market.Symbol, since, until)
+			trades, err := f.batchQueryTrades(subCtx, service, symbol, since, until)
 			if err != nil {
 				log.WithError(err).Errorf("unable to batch query trades for fixer")
 				return err
@@ -80,9 +77,11 @@ func (f *ProfitFixer) aggregateAllTrades(ctx context.Context, market types.Marke
 	return allTrades, nil
 }
 
-func (f *ProfitFixer) Fix(ctx context.Context, since, until time.Time, stats *types.ProfitStats, position *types.Position) error {
+func (f *ProfitFixer) Fix(
+	ctx context.Context, symbol string, since, until time.Time, stats *types.ProfitStats, position *types.Position,
+) error {
 	log.Infof("starting profitFixer with time range %s <=> %s", since, until)
-	allTrades, err := f.aggregateAllTrades(ctx, f.market, since, until)
+	allTrades, err := f.aggregateAllTrades(ctx, symbol, since, until)
 	if err != nil {
 		return err
 	}

pkg/strategy/xdepthmaker/strategy.go

@@ -333,7 +333,7 @@ func (s *Strategy) CrossRun(
 		s.CrossExchangeMarketMakingStrategy.Position = types.NewPositionFromMarket(makerMarket)
 		s.CrossExchangeMarketMakingStrategy.ProfitStats = types.NewProfitStats(makerMarket)
 
-		fixer := common.NewProfitFixer(makerMarket)
+		fixer := common.NewProfitFixer()
 		if ss, ok := makerSession.Exchange.(types.ExchangeTradeHistoryService); ok {
 			log.Infof("adding makerSession %s to profitFixer", makerSession.Name)
 			fixer.AddExchange(makerSession.Name, ss)
@@ -344,7 +344,11 @@ func (s *Strategy) CrossRun(
 			fixer.AddExchange(hedgeSession.Name, ss)
 		}
 
-		if err2 := fixer.Fix(ctx, s.ProfitFixerConfig.TradesSince.Time(), time.Now(), s.CrossExchangeMarketMakingStrategy.ProfitStats, s.CrossExchangeMarketMakingStrategy.Position); err2 != nil {
+		if err2 := fixer.Fix(ctx, makerMarket.Symbol,
+			s.ProfitFixerConfig.TradesSince.Time(),
+			time.Now(),
+			s.CrossExchangeMarketMakingStrategy.ProfitStats,
+			s.CrossExchangeMarketMakingStrategy.Position); err2 != nil {
 			return err2
 		}
 

pkg/strategy/xfunding/strategy.go

@@ -13,6 +13,7 @@ import (
 	"github.com/c9s/bbgo/pkg/exchange/binance"
 	"github.com/c9s/bbgo/pkg/exchange/binance/binanceapi"
 	"github.com/c9s/bbgo/pkg/fixedpoint"
+	"github.com/c9s/bbgo/pkg/strategy/common"
 	"github.com/c9s/bbgo/pkg/util/backoff"
 
 	"github.com/c9s/bbgo/pkg/bbgo"
@@ -137,6 +138,8 @@ type Strategy struct {
 	// Reset your position info
 	Reset bool `json:"reset"`
 
+	ProfitFixerConfig *common.ProfitFixerConfig `json:"profitFixer"`
+
 	// CloseFuturesPosition can be enabled to close the futures position and then transfer the collateral asset back to the spot account.
 	CloseFuturesPosition bool `json:"closeFuturesPosition"`
 
@@ -258,7 +261,9 @@ func (s *Strategy) Run(ctx context.Context, orderExecutor bbgo.OrderExecutor, se
 	return nil
 }
 
-func (s *Strategy) CrossRun(ctx context.Context, orderExecutionRouter bbgo.OrderExecutionRouter, sessions map[string]*bbgo.ExchangeSession) error {
+func (s *Strategy) CrossRun(
+	ctx context.Context, orderExecutionRouter bbgo.OrderExecutionRouter, sessions map[string]*bbgo.ExchangeSession,
+) error {
 	instanceID := s.InstanceID()
 
 	s.spotSession = sessions[s.SpotSession]
@@ -286,22 +291,6 @@ func (s *Strategy) CrossRun(ctx context.Context, orderExecutionRouter bbgo.Order
 		return err
 	}
 
-	// adjust QuoteInvestment
-	if b, ok := s.spotSession.Account.Balance(s.spotMarket.QuoteCurrency); ok {
-		originalQuoteInvestment := s.QuoteInvestment
-
-		// adjust available quote with the fee rate
-		available := b.Available.Mul(fixedpoint.NewFromFloat(1.0 - (0.01 * 0.075)))
-		s.QuoteInvestment = fixedpoint.Min(available, s.QuoteInvestment)
-
-		if originalQuoteInvestment.Compare(s.QuoteInvestment) != 0 {
-			log.Infof("adjusted quoteInvestment from %s to %s according to the balance",
-				originalQuoteInvestment.String(),
-				s.QuoteInvestment.String(),
-			)
-		}
-	}
-
 	if s.ProfitStats == nil || s.Reset {
 		s.ProfitStats = &ProfitStats{
 			ProfitStats: types.NewProfitStats(s.Market),
@@ -332,7 +321,46 @@ func (s *Strategy) CrossRun(ctx context.Context, orderExecutionRouter bbgo.Order
 		s.State = newState()
 	}
 
-	if err := s.checkAndRestorePositionRisks(ctx); err != nil {
+	if s.ProfitFixerConfig != nil {
+		log.Infof("profitFixer is enabled, start fixing with config: %+v", s.ProfitFixerConfig)
+
+		s.SpotPosition = types.NewPositionFromMarket(s.spotMarket)
+		s.FuturesPosition = types.NewPositionFromMarket(s.futuresMarket)
+		s.ProfitStats.ProfitStats = types.NewProfitStats(s.Market)
+
+		since := s.ProfitFixerConfig.TradesSince.Time()
+		now := time.Now()
+
+		spotFixer := common.NewProfitFixer()
+		if ss, ok := s.spotSession.Exchange.(types.ExchangeTradeHistoryService); ok {
+			spotFixer.AddExchange(s.spotSession.Name, ss)
+		}
+
+		if err2 := spotFixer.Fix(ctx, s.Symbol,
+			since, now,
+			s.ProfitStats.ProfitStats,
+			s.SpotPosition); err2 != nil {
+			return err2
+		}
+
+		futuresFixer := common.NewProfitFixer()
+		if ss, ok := s.futuresSession.Exchange.(types.ExchangeTradeHistoryService); ok {
+			futuresFixer.AddExchange(s.futuresSession.Name, ss)
+		}
+
+		if err2 := futuresFixer.Fix(ctx, s.Symbol,
+			since, now,
+			s.ProfitStats.ProfitStats,
+			s.FuturesPosition); err2 != nil {
+			return err2
+		}
+
+		bbgo.Notify("Fixed spot position", s.SpotPosition)
+		bbgo.Notify("Fixed futures position", s.FuturesPosition)
+		bbgo.Notify("Fixed profit stats", s.ProfitStats.ProfitStats)
+	}
+
+	if err := s.syncPositionRisks(ctx); err != nil {
 		return err
 	}
 
@@ -348,7 +376,7 @@ func (s *Strategy) CrossRun(ctx context.Context, orderExecutionRouter bbgo.Order
 	bbgo.Notify("Spot Position", s.SpotPosition)
 	bbgo.Notify("Futures Position", s.FuturesPosition)
 	bbgo.Notify("Neutral Position", s.NeutralPosition)
-	bbgo.Notify("State", s.State.PositionState)
+	bbgo.Notify("State: %s", s.State.PositionState.String())
 
 	// sync funding fee txns
 	s.syncFundingFeeRecords(ctx, s.ProfitStats.LastFundingFeeTime)
@@ -357,6 +385,31 @@ func (s *Strategy) CrossRun(ctx context.Context, orderExecutionRouter bbgo.Order
 	// s.syncFundingFeeRecords(ctx, time.Now().Add(-3*24*time.Hour))
 
 	switch s.State.PositionState {
+	case PositionClosed:
+		// adjust QuoteInvestment according to the available quote balance
+		// ONLY when the position is not opening
+		if b, ok := s.spotSession.Account.Balance(s.spotMarket.QuoteCurrency); ok {
+			originalQuoteInvestment := s.QuoteInvestment
+
+			// adjust available quote with the fee rate
+			spotFeeRate := 0.075
+			availableQuoteWithoutFee := b.Available.Mul(fixedpoint.NewFromFloat(1.0 - (spotFeeRate * 0.01)))
+
+			s.QuoteInvestment = fixedpoint.Min(availableQuoteWithoutFee, s.QuoteInvestment)
+
+			if originalQuoteInvestment.Compare(s.QuoteInvestment) != 0 {
+				log.Infof("adjusted quoteInvestment from %f to %f according to the balance",
+					originalQuoteInvestment.Float64(),
+					s.QuoteInvestment.Float64(),
+				)
+			}
+		}
+	default:
+	}
+
+	switch s.State.PositionState {
+	case PositionReady:
+
 	case PositionOpening:
 		// transfer all base assets from the spot account into the spot account
 		if err := s.transferIn(ctx, s.binanceSpot, s.spotMarket.BaseCurrency, fixedpoint.Zero); err != nil {
@@ -368,6 +421,7 @@ func (s *Strategy) CrossRun(ctx context.Context, orderExecutionRouter bbgo.Order
 		if err := s.transferOut(ctx, s.binanceSpot, s.spotMarket.BaseCurrency, fixedpoint.Zero); err != nil {
 			log.WithError(err).Errorf("futures asset transfer out error")
 		}
+
 	}
 
 	s.spotOrderExecutor = s.allocateOrderExecutor(ctx, s.spotSession, instanceID, s.SpotPosition)
@@ -735,12 +789,14 @@ func (s *Strategy) syncFuturesPosition(ctx context.Context) {
 
 	if futuresBase.Sign() > 0 {
 		// unexpected error
-		log.Errorf("unexpected futures position (got positive, expecting negative)")
+		log.Errorf("unexpected futures position, got positive number (long), expecting negative number (short)")
 		return
 	}
 
+	// cancel the previous futures order
 	_ = s.futuresOrderExecutor.GracefulCancel(ctx)
 
+	// get the latest ticker price
 	ticker, err := s.futuresSession.Exchange.QueryTicker(ctx, s.Symbol)
 	if err != nil {
 		log.WithError(err).Errorf("can not query ticker")
@@ -753,6 +809,7 @@ func (s *Strategy) syncFuturesPosition(ctx context.Context) {
 		log.WithError(err).Errorf("can not calculate futures account quote value")
 		return
 	}
+
 	log.Infof("calculated futures account quote value = %s", quoteValue.String())
 	if quoteValue.IsZero() {
 		return
@@ -796,12 +853,10 @@ func (s *Strategy) syncFuturesPosition(ctx context.Context) {
 	orderQuantity = fixedpoint.Max(diffQuantity, s.minQuantity)
 	orderQuantity = s.futuresMarket.AdjustQuantityByMinNotional(orderQuantity, orderPrice)
 
-	/*
+	if s.futuresMarket.IsDustQuantity(orderQuantity, orderPrice) {
-		if s.futuresMarket.IsDustQuantity(orderQuantity, orderPrice) {
+		log.Warnf("unexpected dust quantity, skip futures order with dust quantity %s, market = %+v", orderQuantity.String(), s.futuresMarket)
-			log.Warnf("unexpected dust quantity, skip futures order with dust quantity %s, market = %+v", orderQuantity.String(), s.futuresMarket)
+		return
-			return
+	}
-		}
-	*/
 
 	submitOrder := types.SubmitOrder{
 		Symbol:   s.Symbol,
@@ -814,7 +869,7 @@ func (s *Strategy) syncFuturesPosition(ctx context.Context) {
 	createdOrders, err := s.futuresOrderExecutor.SubmitOrders(ctx, submitOrder)
 
 	if err != nil {
-		log.WithError(err).Errorf("can not submit spot order: %+v", submitOrder)
+		log.WithError(err).Errorf("can not submit futures order: %+v", submitOrder)
 		return
 	}
 
@@ -1083,7 +1138,9 @@ func (s *Strategy) notPositionState(state PositionState) bool {
 	return ret
 }
 
-func (s *Strategy) allocateOrderExecutor(ctx context.Context, session *bbgo.ExchangeSession, instanceID string, position *types.Position) *bbgo.GeneralOrderExecutor {
+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)
@@ -1141,7 +1198,7 @@ func (s *Strategy) checkAndFixMarginMode(ctx context.Context) error {
 	return nil
 }
 
-func (s *Strategy) checkAndRestorePositionRisks(ctx context.Context) error {
+func (s *Strategy) syncPositionRisks(ctx context.Context) error {
 	futuresClient := s.binanceFutures.GetFuturesClient()
 	req := futuresClient.NewFuturesGetPositionRisksRequest()
 	req.Symbol(s.Symbol)

pkg/strategy/xfunding/transfer.go

@@ -43,104 +43,111 @@ func (s *Strategy) resetTransfer(ctx context.Context, ex FuturesTransfer, asset
 
 func (s *Strategy) transferOut(ctx context.Context, ex FuturesTransfer, asset string, quantity fixedpoint.Value) error {
 	// if transfer done
+	// TotalBaseTransfer here is the rest quantity we need to transfer
+	// (total spot -> futures transfer amount) is recorded in this variable.
+	//
+	// TotalBaseTransfer == 0 means we have nothing to transfer.
 	if s.State.TotalBaseTransfer.IsZero() {
 		return nil
 	}
 
-	balances, err := s.futuresSession.Exchange.QueryAccountBalances(ctx)
+	quantity = quantity.Add(s.State.PendingBaseTransfer)
+
+	// A simple protection here -- we can only transfer the rest quota (total base transfer) back to spot
+	quantity = fixedpoint.Min(s.State.TotalBaseTransfer, quantity)
+
+	available, pending, err := s.queryAvailableTransfer(ctx, s.futuresSession.Exchange, asset, quantity)
 	if err != nil {
-		log.Infof("balance query error, adding to pending base transfer: %s %s + %s", quantity.String(), asset, s.State.PendingBaseTransfer.String())
+		s.State.PendingBaseTransfer = quantity
-		s.State.PendingBaseTransfer = s.State.PendingBaseTransfer.Add(quantity)
 		return err
 	}
 
-	b, ok := balances[asset]
+	s.State.PendingBaseTransfer = pending
-	if !ok {
-		log.Infof("balance not found, adding to pending base transfer: %s %s + %s", quantity.String(), asset, s.State.PendingBaseTransfer.String())
-		s.State.PendingBaseTransfer = s.State.PendingBaseTransfer.Add(quantity)
-		return fmt.Errorf("%s balance not found", asset)
-	}
 
-	log.Infof("found futures balance: %+v", b)
+	log.Infof("transfering out futures account asset %f %s", available.Float64(), asset)
-
+	if err := ex.TransferFuturesAccountAsset(ctx, asset, available, types.TransferOut); err != nil {
-	// add the previous pending base transfer and the current trade quantity
+		s.State.PendingBaseTransfer = s.State.PendingBaseTransfer.Add(available)
-	amount := b.MaxWithdrawAmount
-	if !quantity.IsZero() {
-		amount = s.State.PendingBaseTransfer.Add(quantity)
-	}
-
-	// try to transfer more if we enough balance
-	amount = fixedpoint.Min(amount, b.MaxWithdrawAmount)
-
-	// we can only transfer the rest quota (total base transfer)
-	amount = fixedpoint.Min(s.State.TotalBaseTransfer, amount)
-
-	// 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 amount.IsZero() {
-		log.Infof("zero amount, adding to pending base transfer: %s %s + %s ", quantity.String(), asset, s.State.PendingBaseTransfer.String())
-		s.State.PendingBaseTransfer = s.State.PendingBaseTransfer.Add(quantity)
-		return nil
-	}
-
-	// de-leverage and get the collateral base quantity
-	collateralBase := s.FuturesPosition.GetBase().Abs().Div(s.Leverage)
-	_ = collateralBase
-
-	// if s.State.TotalBaseTransfer.Compare(collateralBase)
-
-	log.Infof("transfering out futures account asset %s %s", amount, asset)
-	if err := ex.TransferFuturesAccountAsset(ctx, asset, amount, types.TransferOut); err != nil {
 		return err
 	}
 
-	// reset pending transfer
-	s.State.PendingBaseTransfer = fixedpoint.Zero
-
 	// reduce the transfer in the total base transfer
-	s.State.TotalBaseTransfer = s.State.TotalBaseTransfer.Sub(amount)
+	s.State.TotalBaseTransfer = s.State.TotalBaseTransfer.Sub(available)
 	return nil
 }
 
+// transferIn transfers the asset from the spot account to the futures account
 func (s *Strategy) transferIn(ctx context.Context, ex FuturesTransfer, asset string, quantity fixedpoint.Value) error {
-	balances, err := s.spotSession.Exchange.QueryAccountBalances(ctx)
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	// add the pending transfer and reset the pending transfer
+	quantity = s.State.PendingBaseTransfer.Add(quantity)
+
+	available, pending, err := s.queryAvailableTransfer(ctx, s.spotSession.Exchange, asset, quantity)
 	if err != nil {
+		s.State.PendingBaseTransfer = quantity
 		return err
 	}
 
+	s.State.PendingBaseTransfer = pending
+
+	if available.IsZero() {
+		return fmt.Errorf("unable to transfer zero %s from spot wallet to futures wallet", asset)
+	}
+
+	log.Infof("transfering %f %s from the spot wallet into futures wallet...", available.Float64(), asset)
+	if err := ex.TransferFuturesAccountAsset(ctx, asset, available, types.TransferIn); err != nil {
+		s.State.PendingBaseTransfer = s.State.PendingBaseTransfer.Add(available)
+		return err
+	}
+
+	// record the transfer in the total base transfer
+	s.State.TotalBaseTransfer = s.State.TotalBaseTransfer.Add(available)
+	return nil
+}
+
+func (s *Strategy) queryAvailableTransfer(
+	ctx context.Context, ex types.Exchange, asset string, quantity fixedpoint.Value,
+) (available, pending fixedpoint.Value, err error) {
+	available = fixedpoint.Zero
+	pending = fixedpoint.Zero
+
+	// query spot balances to validate the quantity
+	balances, err := ex.QueryAccountBalances(ctx)
+	if err != nil {
+		return available, pending, err
+	}
+
 	b, ok := balances[asset]
 	if !ok {
-		return fmt.Errorf("%s balance not found", asset)
+		return available, pending, fmt.Errorf("%s balance not found", asset)
 	}
 
-	// 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
+	log.Infof("loaded %s balance: %+v", asset, b)
-	if !quantity.IsZero() && b.Available.Compare(quantity) < 0 {
+
-		log.Infof("adding to pending base transfer: %s %s", quantity, asset)
+	// if quantity = 0, we will transfer all available balance into the futures wallet
-		s.State.PendingBaseTransfer = s.State.PendingBaseTransfer.Add(quantity)
+	if quantity.IsZero() {
-		return nil
+		quantity = b.Available
 	}
 
-	amount := b.Available
+	limit := b.Available
-	if !quantity.IsZero() {
+	if b.MaxWithdrawAmount.Sign() > 0 {
-		amount = s.State.PendingBaseTransfer.Add(quantity)
+		limit = fixedpoint.Min(b.MaxWithdrawAmount, limit)
 	}
 
-	pos := s.SpotPosition.GetBase().Abs()
+	if limit.Compare(quantity) < 0 {
-	rest := pos.Sub(s.State.TotalBaseTransfer)
+		log.Infof("%s available balance is not enough for transfer (%f < %f)",
-	if rest.Sign() < 0 {
+			asset,
-		return nil
+			b.Available.Float64(),
+			quantity.Float64())
+
+		available = fixedpoint.Min(limit, quantity)
+		pending = quantity.Sub(available)
+		log.Infof("adjusted transfer quantity from %f to %f", quantity.Float64(), available.Float64())
+		return available, pending, nil
 	}
 
-	amount = fixedpoint.Min(rest, amount)
+	available = quantity
-
+	pending = fixedpoint.Zero
-	log.Infof("transfering in futures account asset %s %s", amount, asset)
+	return available, pending, nil
-	if err := ex.TransferFuturesAccountAsset(ctx, asset, amount, types.TransferIn); err != nil {
-		return err
-	}
-
-	// reset pending transfer
-	s.State.PendingBaseTransfer = fixedpoint.Zero
-
-	// record the transfer in the total base transfer
-	s.State.TotalBaseTransfer = s.State.TotalBaseTransfer.Add(amount)
-	return nil
 }