Merge pull request #1127 from c9s/c9s/strategy/funding

feature: strategy: xfunding

config/funding.yaml

@@ -1,31 +0,0 @@
----
-notifications:
-  slack:
-    defaultChannel: "dev-bbgo"
-    errorChannel: "bbgo-error"
-
-  switches:
-    trade: true
-    orderUpdate: true
-    submitOrder: true
-
-sessions:
-  binance:
-    exchange: binance
-    envVarPrefix: binance
-    futures: true
-
-exchangeStrategies:
-- on: binance
-  funding:
-    symbol: ETHUSDT
-    quantity: 0.0001
-    fundingRate:
-      high: 0.01%
-    supportDetection:
-      - interval: 1m
-        movingAverageType: EMA
-        movingAverageIntervalWindow:
-          interval: 15m
-          window: 60
-        minVolume: 8_000

config/grid2-max.yaml

@@ -11,7 +11,8 @@ notifications:
 sessions:
   max:
     exchange: max
-    envVarPrefix: max
+    envVarPrefix: MAX
+
 
 # example command:
 #    godotenv -f .env.local -- go run ./cmd/bbgo backtest --config config/grid2-max.yaml --base-asset-baseline

config/xfunding.yaml

@@ -0,0 +1,45 @@
+---
+notifications:
+  slack:
+    defaultChannel: "dev-bbgo"
+    errorChannel: "bbgo-error"
+
+  switches:
+    trade: true
+    orderUpdate: true
+    submitOrder: true
+
+persistence:
+  redis:
+    host: 127.0.0.1
+    port: 6379
+    db: 1
+
+sessions:
+  binance:
+    exchange: binance
+    envVarPrefix: BINANCE
+
+  binance_futures:
+    exchange: binance
+    envVarPrefix: BINANCE
+    futures: true
+
+crossExchangeStrategies:
+
+- xfunding:
+    spotSession: binance
+    futuresSession: binance_futures
+    symbol: ETHUSDT
+    leverage: 1.0
+    incrementalQuoteQuantity: 20
+    quoteInvestment: 50
+
+    shortFundingRate:
+      ## when funding rate is higher than this high value, the strategy will start buying spot and opening a short position
+      high: 0.001%
+      ## when funding rate is lower than this low value, the strategy will start closing futures position and sell the spot
+      low: -0.01%
+
+    ## reset will reset the spot/futures positions, the transfer stats and the position state.
+    # reset: true

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/bbgo/order_execution.go

@@ -54,7 +54,7 @@ func (e *ExchangeOrderExecutionRouter) SubmitOrdersTo(ctx context.Context, sessi
 		return nil, err
 	}
 
-	createdOrders, _, err := BatchPlaceOrder(ctx, es.Exchange, formattedOrders...)
+	createdOrders, _, err := BatchPlaceOrder(ctx, es.Exchange, nil, formattedOrders...)
 	return createdOrders, err
 }
 
@@ -94,7 +94,7 @@ func (e *ExchangeOrderExecutor) SubmitOrders(ctx context.Context, orders ...type
 		log.Infof("submitting order: %s", order.String())
 	}
 
-	createdOrders, _, err := BatchPlaceOrder(ctx, e.Session.Exchange, formattedOrders...)
+	createdOrders, _, err := BatchPlaceOrder(ctx, e.Session.Exchange, nil, formattedOrders...)
 	return createdOrders, err
 }
 
@@ -297,10 +297,13 @@ func (c *BasicRiskController) ProcessOrders(session *ExchangeSession, orders ...
 	return outOrders, nil
 }
 
+type OrderCallback func(order types.Order)
+
 // BatchPlaceOrder
-func BatchPlaceOrder(ctx context.Context, exchange types.Exchange, submitOrders ...types.SubmitOrder) (types.OrderSlice, []int, error) {
+func BatchPlaceOrder(ctx context.Context, exchange types.Exchange, orderCallback OrderCallback, submitOrders ...types.SubmitOrder) (types.OrderSlice, []int, error) {
 	var createdOrders types.OrderSlice
 	var err error
+
 	var errIndexes []int
 	for i, submitOrder := range submitOrders {
 		createdOrder, err2 := exchange.SubmitOrder(ctx, submitOrder)
@@ -309,6 +312,11 @@ func BatchPlaceOrder(ctx context.Context, exchange types.Exchange, submitOrders
 			errIndexes = append(errIndexes, i)
 		} else if createdOrder != nil {
 			createdOrder.Tag = submitOrder.Tag
+
+			if orderCallback != nil {
+				orderCallback(*createdOrder)
+			}
+
 			createdOrders = append(createdOrders, *createdOrder)
 		}
 	}
@@ -316,8 +324,6 @@ func BatchPlaceOrder(ctx context.Context, exchange types.Exchange, submitOrders
 	return createdOrders, errIndexes, err
 }
 
-type OrderCallback func(order types.Order)
-
 // BatchRetryPlaceOrder places the orders and retries the failed orders
 func BatchRetryPlaceOrder(ctx context.Context, exchange types.Exchange, errIdx []int, orderCallback OrderCallback, logger log.FieldLogger, submitOrders ...types.SubmitOrder) (types.OrderSlice, []int, error) {
 	if logger == nil {
@@ -329,26 +335,12 @@ func BatchRetryPlaceOrder(ctx context.Context, exchange types.Exchange, errIdx [
 
 	// if the errIdx is nil, then we should iterate all the submit orders
 	// allocate a variable for new error index
-	var errIdxNext []int
 	if len(errIdx) == 0 {
-		for i, submitOrder := range submitOrders {
+		var err2 error
-			createdOrder, err2 := exchange.SubmitOrder(ctx, submitOrder)
+		createdOrders, errIdx, err2 = BatchPlaceOrder(ctx, exchange, orderCallback, submitOrders...)
-			if err2 != nil {
+		if err2 != nil {
-				werr = multierr.Append(werr, err2)
+			werr = multierr.Append(werr, err2)
-				errIdxNext = append(errIdxNext, i)
-			} else if createdOrder != nil {
-				// if the order is successfully created, than we should copy the order tag
-				createdOrder.Tag = submitOrder.Tag
-
-				if orderCallback != nil {
-					orderCallback(*createdOrder)
-				}
-
-				createdOrders = append(createdOrders, *createdOrder)
-			}
 		}
-
-		errIdx = errIdxNext
 	}
 
 	timeoutCtx, cancelTimeout := context.WithTimeout(ctx, DefaultSubmitOrderRetryTimeout)
@@ -359,6 +351,7 @@ func BatchRetryPlaceOrder(ctx context.Context, exchange types.Exchange, errIdx [
 
 	// set backoff max retries to 101 because https://ja.wikipedia.org/wiki/101%E5%9B%9E%E7%9B%AE%E3%81%AE%E3%83%97%E3%83%AD%E3%83%9D%E3%83%BC%E3%82%BA
 	backoffMaxRetries := uint64(101)
+	var errIdxNext []int
 batchRetryOrder:
 	for retryRound := 0; len(errIdx) > 0 && retryRound < 10; retryRound++ {
 		// sleep for 200 millisecond between each retry

pkg/bbgo/order_executor_general.go

@@ -40,6 +40,7 @@ type GeneralOrderExecutor struct {
 
 	marginBaseMaxBorrowable, marginQuoteMaxBorrowable fixedpoint.Value
 
+	maxRetries    uint
 	disableNotify bool
 	closing       int64
 }
@@ -73,6 +74,10 @@ func (e *GeneralOrderExecutor) DisableNotify() {
 	e.disableNotify = true
 }
 
+func (e *GeneralOrderExecutor) SetMaxRetries(maxRetries uint) {
+	e.maxRetries = maxRetries
+}
+
 func (e *GeneralOrderExecutor) startMarginAssetUpdater(ctx context.Context) {
 	marginService, ok := e.session.Exchange.(types.MarginBorrowRepayService)
 	if !ok {
@@ -194,10 +199,12 @@ func (e *GeneralOrderExecutor) FastSubmitOrders(ctx context.Context, submitOrder
 	if err != nil {
 		return nil, err
 	}
-	createdOrders, errIdx, err := BatchPlaceOrder(ctx, e.session.Exchange, formattedOrders...)
+
+	createdOrders, errIdx, err := BatchPlaceOrder(ctx, e.session.Exchange, nil, formattedOrders...)
 	if len(errIdx) > 0 {
 		return nil, err
 	}
+
 	if IsBackTesting {
 		e.orderStore.Add(createdOrders...)
 		e.activeMakerOrders.Add(createdOrders...)
@@ -229,6 +236,11 @@ func (e *GeneralOrderExecutor) SubmitOrders(ctx context.Context, submitOrders ..
 		e.tradeCollector.Process()
 	}
 
+	if e.maxRetries == 0 {
+		createdOrders, _, err := BatchPlaceOrder(ctx, e.session.Exchange, orderCreateCallback, formattedOrders...)
+		return createdOrders, err
+	}
+
 	createdOrders, _, err := BatchRetryPlaceOrder(ctx, e.session.Exchange, nil, orderCreateCallback, e.logger, formattedOrders...)
 	return createdOrders, err
 }

pkg/bbgo/order_processor.go

@@ -33,7 +33,7 @@ func AdjustQuantityByMinAmount(quantity, currentPrice, minAmount fixedpoint.Valu
 	amount := currentPrice.Mul(quantity)
 	if amount.Compare(minAmount) < 0 {
 		ratio := minAmount.Div(amount)
-		quantity = quantity.Mul(ratio)
+		return quantity.Mul(ratio)
 	}
 
 	return quantity

pkg/cmd/strategy/builtin.go

@@ -16,7 +16,6 @@ import (
 	_ "github.com/c9s/bbgo/pkg/strategy/fixedmaker"
 	_ "github.com/c9s/bbgo/pkg/strategy/flashcrash"
 	_ "github.com/c9s/bbgo/pkg/strategy/fmaker"
-	_ "github.com/c9s/bbgo/pkg/strategy/funding"
 	_ "github.com/c9s/bbgo/pkg/strategy/grid"
 	_ "github.com/c9s/bbgo/pkg/strategy/grid2"
 	_ "github.com/c9s/bbgo/pkg/strategy/harmonic"
@@ -38,6 +37,7 @@ import (
 	_ "github.com/c9s/bbgo/pkg/strategy/trendtrader"
 	_ "github.com/c9s/bbgo/pkg/strategy/wall"
 	_ "github.com/c9s/bbgo/pkg/strategy/xbalance"
+	_ "github.com/c9s/bbgo/pkg/strategy/xfunding"
 	_ "github.com/c9s/bbgo/pkg/strategy/xgap"
 	_ "github.com/c9s/bbgo/pkg/strategy/xmaker"
 	_ "github.com/c9s/bbgo/pkg/strategy/xnav"

pkg/exchange/binance/binanceapi/client.go

@@ -233,6 +233,14 @@ func castPayload(payload interface{}) ([]byte, error) {
 		case []byte:
 			return v, nil
 
+		case map[string]interface{}:
+			var params = url.Values{}
+			for a, b := range v {
+				params.Add(a, fmt.Sprintf("%v", b))
+			}
+
+			return []byte(params.Encode()), nil
+
 		default:
 			body, err := json.Marshal(v)
 			return body, err

pkg/exchange/binance/binanceapi/futures_transfer_request.go

@@ -0,0 +1,33 @@
+package binanceapi
+
+import "github.com/c9s/requestgen"
+
+type FuturesTransferType int
+
+const (
+	FuturesTransferSpotToUsdtFutures FuturesTransferType = 1
+	FuturesTransferUsdtFuturesToSpot FuturesTransferType = 2
+
+	FuturesTransferSpotToCoinFutures FuturesTransferType = 3
+	FuturesTransferCoinFuturesToSpot FuturesTransferType = 4
+)
+
+type FuturesTransferResponse struct {
+	TranId int64 `json:"tranId"`
+}
+
+//go:generate requestgen -method POST -url "/sapi/v1/futures/transfer" -type FuturesTransferRequest -responseType .FuturesTransferResponse
+type FuturesTransferRequest struct {
+	client requestgen.AuthenticatedAPIClient
+
+	asset string `param:"asset"`
+
+	// amount is a decimal in string format
+	amount string `param:"amount"`
+
+	transferType FuturesTransferType `param:"type"`
+}
+
+func (c *RestClient) NewFuturesTransferRequest() *FuturesTransferRequest {
+	return &FuturesTransferRequest{client: c}
+}

pkg/exchange/binance/binanceapi/futures_transfer_request_requestgen.go

@@ -0,0 +1,178 @@
+// Code generated by "requestgen -method POST -url /sapi/v1/futures/transfer -type FuturesTransferRequest -responseType .FuturesTransferResponse"; DO NOT EDIT.
+
+package binanceapi
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"net/url"
+	"reflect"
+	"regexp"
+)
+
+func (f *FuturesTransferRequest) Asset(asset string) *FuturesTransferRequest {
+	f.asset = asset
+	return f
+}
+
+func (f *FuturesTransferRequest) Amount(amount string) *FuturesTransferRequest {
+	f.amount = amount
+	return f
+}
+
+func (f *FuturesTransferRequest) TransferType(transferType FuturesTransferType) *FuturesTransferRequest {
+	f.transferType = transferType
+	return f
+}
+
+// GetQueryParameters builds and checks the query parameters and returns url.Values
+func (f *FuturesTransferRequest) GetQueryParameters() (url.Values, error) {
+	var params = map[string]interface{}{}
+
+	query := url.Values{}
+	for _k, _v := range params {
+		query.Add(_k, fmt.Sprintf("%v", _v))
+	}
+
+	return query, nil
+}
+
+// GetParameters builds and checks the parameters and return the result in a map object
+func (f *FuturesTransferRequest) GetParameters() (map[string]interface{}, error) {
+	var params = map[string]interface{}{}
+	// check asset field -> json key asset
+	asset := f.asset
+
+	// assign parameter of asset
+	params["asset"] = asset
+	// check amount field -> json key amount
+	amount := f.amount
+
+	// assign parameter of amount
+	params["amount"] = amount
+	// check transferType field -> json key type
+	transferType := f.transferType
+
+	// TEMPLATE check-valid-values
+	switch transferType {
+	case FuturesTransferSpotToUsdtFutures, FuturesTransferUsdtFuturesToSpot, FuturesTransferSpotToCoinFutures, FuturesTransferCoinFuturesToSpot:
+		params["type"] = transferType
+
+	default:
+		return nil, fmt.Errorf("type value %v is invalid", transferType)
+
+	}
+	// END TEMPLATE check-valid-values
+
+	// assign parameter of transferType
+	params["type"] = transferType
+
+	return params, nil
+}
+
+// GetParametersQuery converts the parameters from GetParameters into the url.Values format
+func (f *FuturesTransferRequest) GetParametersQuery() (url.Values, error) {
+	query := url.Values{}
+
+	params, err := f.GetParameters()
+	if err != nil {
+		return query, err
+	}
+
+	for _k, _v := range params {
+		if f.isVarSlice(_v) {
+			f.iterateSlice(_v, func(it interface{}) {
+				query.Add(_k+"[]", fmt.Sprintf("%v", it))
+			})
+		} else {
+			query.Add(_k, fmt.Sprintf("%v", _v))
+		}
+	}
+
+	return query, nil
+}
+
+// GetParametersJSON converts the parameters from GetParameters into the JSON format
+func (f *FuturesTransferRequest) GetParametersJSON() ([]byte, error) {
+	params, err := f.GetParameters()
+	if err != nil {
+		return nil, err
+	}
+
+	return json.Marshal(params)
+}
+
+// GetSlugParameters builds and checks the slug parameters and return the result in a map object
+func (f *FuturesTransferRequest) GetSlugParameters() (map[string]interface{}, error) {
+	var params = map[string]interface{}{}
+
+	return params, nil
+}
+
+func (f *FuturesTransferRequest) applySlugsToUrl(url string, slugs map[string]string) string {
+	for _k, _v := range slugs {
+		needleRE := regexp.MustCompile(":" + _k + "\\b")
+		url = needleRE.ReplaceAllString(url, _v)
+	}
+
+	return url
+}
+
+func (f *FuturesTransferRequest) iterateSlice(slice interface{}, _f func(it interface{})) {
+	sliceValue := reflect.ValueOf(slice)
+	for _i := 0; _i < sliceValue.Len(); _i++ {
+		it := sliceValue.Index(_i).Interface()
+		_f(it)
+	}
+}
+
+func (f *FuturesTransferRequest) isVarSlice(_v interface{}) bool {
+	rt := reflect.TypeOf(_v)
+	switch rt.Kind() {
+	case reflect.Slice:
+		return true
+	}
+	return false
+}
+
+func (f *FuturesTransferRequest) GetSlugsMap() (map[string]string, error) {
+	slugs := map[string]string{}
+	params, err := f.GetSlugParameters()
+	if err != nil {
+		return slugs, nil
+	}
+
+	for _k, _v := range params {
+		slugs[_k] = fmt.Sprintf("%v", _v)
+	}
+
+	return slugs, nil
+}
+
+func (f *FuturesTransferRequest) Do(ctx context.Context) (*FuturesTransferResponse, error) {
+
+	params, err := f.GetParameters()
+	if err != nil {
+		return nil, err
+	}
+	query := url.Values{}
+
+	apiURL := "/sapi/v1/futures/transfer"
+
+	req, err := f.client.NewAuthenticatedRequest(ctx, "POST", apiURL, query, params)
+	if err != nil {
+		return nil, err
+	}
+
+	response, err := f.client.SendRequest(req)
+	if err != nil {
+		return nil, err
+	}
+
+	var apiResponse FuturesTransferResponse
+	if err := response.DecodeJSON(&apiResponse); err != nil {
+		return nil, err
+	}
+	return &apiResponse, nil
+}

pkg/exchange/binance/exchange.go

@@ -367,17 +367,45 @@ func (e *Exchange) QueryMarginBorrowHistory(ctx context.Context, asset string) e
 	return nil
 }
 
+func (e *Exchange) TransferFuturesAccountAsset(ctx context.Context, asset string, amount fixedpoint.Value, io types.TransferDirection) error {
+	req := e.client2.NewFuturesTransferRequest()
+	req.Asset(asset)
+	req.Amount(amount.String())
+
+	if io == types.TransferIn {
+		req.TransferType(binanceapi.FuturesTransferSpotToUsdtFutures)
+	} else if io == types.TransferOut {
+		req.TransferType(binanceapi.FuturesTransferUsdtFuturesToSpot)
+	} else {
+		return fmt.Errorf("unexpected transfer direction: %d given", io)
+	}
+
+	resp, err := req.Do(ctx)
+
+	switch io {
+	case types.TransferIn:
+		log.Infof("internal transfer (spot) => (futures) %s %s, transaction = %+v, err = %+v", amount.String(), asset, resp, err)
+	case types.TransferOut:
+		log.Infof("internal transfer (futures) => (spot) %s %s, transaction = %+v, err = %+v", amount.String(), asset, resp, err)
+	}
+
+	return err
+}
+
 // transferCrossMarginAccountAsset transfer asset to the cross margin account or to the main account
-func (e *Exchange) transferCrossMarginAccountAsset(ctx context.Context, asset string, amount fixedpoint.Value, io int) error {
+func (e *Exchange) transferCrossMarginAccountAsset(ctx context.Context, asset string, amount fixedpoint.Value, io types.TransferDirection) error {
 	req := e.client.NewMarginTransferService()
 	req.Asset(asset)
 	req.Amount(amount.String())
 
-	if io > 0 { // in
+	if io == types.TransferIn {
 		req.Type(binance.MarginTransferTypeToMargin)
-	} else if io < 0 { // out
+	} else if io == types.TransferOut {
 		req.Type(binance.MarginTransferTypeToMain)
+	} else {
+		return fmt.Errorf("unexpected transfer direction: %d given", io)
 	}
+
 	resp, err := req.Do(ctx)
 	if err != nil {
 		return err

pkg/interact/interact.go

@@ -247,9 +247,9 @@ func (it *Interact) Start(ctx context.Context) error {
 	}
 
 	for _, custom := range it.customInteractions {
-		log.Infof("checking %T custom interaction...", custom)
+		log.Debugf("checking %T custom interaction...", custom)
 		if initializer, ok := custom.(Initializer); ok {
-			log.Infof("initializing %T custom interaction...", custom)
+			log.Debugf("initializing %T custom interaction...", custom)
 			if err := initializer.Initialize(); err != nil {
 				return err
 			}

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
 	}
@@ -539,7 +539,7 @@ func (s *Strategy) klineHandler(ctx context.Context, kline types.KLine, counter
 	s.atr.PushK(kline)
 	atr := s.atr.Last()
 
-	price := kline.Close //s.getLastPrice()
+	price := kline.Close // s.getLastPrice()
 	pricef := price.Float64()
 	lowf := math.Min(kline.Low.Float64(), pricef)
 	highf := math.Max(kline.High.Float64(), pricef)

pkg/strategy/funding/strategy.go

@@ -1,200 +0,0 @@
-package funding
-
-import (
-	"context"
-	"errors"
-	"strings"
-
-	"github.com/sirupsen/logrus"
-
-	"github.com/c9s/bbgo/pkg/exchange/binance"
-	"github.com/c9s/bbgo/pkg/fixedpoint"
-
-	"github.com/c9s/bbgo/pkg/bbgo"
-	"github.com/c9s/bbgo/pkg/types"
-)
-
-const ID = "funding"
-
-var log = logrus.WithField("strategy", ID)
-
-func init() {
-	// Register the pointer of the strategy struct,
-	// so that bbgo knows what struct to be used to unmarshal the configs (YAML or JSON)
-	// Note: built-in strategies need to imported manually in the bbgo cmd package.
-	bbgo.RegisterStrategy(ID, &Strategy{})
-}
-
-type Strategy struct {
-	// 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"`
-	MaxExposurePosition fixedpoint.Value `json:"maxExposurePosition"`
-	// Interval            types.Interval   `json:"interval"`
-
-	FundingRate *struct {
-		High          fixedpoint.Value `json:"high"`
-		Neutral       fixedpoint.Value `json:"neutral"`
-		DiffThreshold fixedpoint.Value `json:"diffThreshold"`
-	} `json:"fundingRate"`
-
-	SupportDetection []struct {
-		Interval types.Interval `json:"interval"`
-		// MovingAverageType is the moving average indicator type that we want to use,
-		// it could be SMA or EWMA
-		MovingAverageType string `json:"movingAverageType"`
-
-		// MovingAverageInterval is the interval of k-lines for the moving average indicator to calculate,
-		// it could be "1m", "5m", "1h" and so on.  note that, the moving averages are calculated from
-		// the k-line data we subscribed
-		// MovingAverageInterval types.Interval `json:"movingAverageInterval"`
-		//
-		// // MovingAverageWindow is the number of the window size of the moving average indicator.
-		// // The number of k-lines in the window. generally used window sizes are 7, 25 and 99 in the TradingView.
-		// MovingAverageWindow int `json:"movingAverageWindow"`
-
-		MovingAverageIntervalWindow types.IntervalWindow `json:"movingAverageIntervalWindow"`
-
-		MinVolume fixedpoint.Value `json:"minVolume"`
-
-		MinQuoteVolume fixedpoint.Value `json:"minQuoteVolume"`
-	} `json:"supportDetection"`
-}
-
-func (s *Strategy) ID() string {
-	return ID
-}
-
-func (s *Strategy) Subscribe(session *bbgo.ExchangeSession) {
-	// session.Subscribe(types.BookChannel, s.Symbol, types.SubscribeOptions{})
-
-	// session.Subscribe(types.KLineChannel, s.Symbol, types.SubscribeOptions{
-	//	Interval: string(s.Interval),
-	// })
-
-	for _, detection := range s.SupportDetection {
-		session.Subscribe(types.KLineChannel, s.Symbol, types.SubscribeOptions{
-			Interval: detection.Interval,
-		})
-		session.Subscribe(types.KLineChannel, s.Symbol, types.SubscribeOptions{
-			Interval: detection.MovingAverageIntervalWindow.Interval,
-		})
-	}
-}
-
-func (s *Strategy) Validate() error {
-	if len(s.Symbol) == 0 {
-		return errors.New("symbol is required")
-	}
-
-	return nil
-}
-
-func (s *Strategy) Run(ctx context.Context, orderExecutor bbgo.OrderExecutor, session *bbgo.ExchangeSession) error {
-	standardIndicatorSet := session.StandardIndicatorSet(s.Symbol)
-
-	if !session.Futures {
-		log.Error("futures not enabled in config for this strategy")
-		return nil
-	}
-
-	premiumIndex, err := session.Exchange.(*binance.Exchange).QueryPremiumIndex(ctx, s.Symbol)
-	if err != nil {
-		log.Error("exchange does not support funding rate api")
-	}
-
-	var ma types.Float64Indicator
-	for _, detection := range s.SupportDetection {
-
-		switch strings.ToLower(detection.MovingAverageType) {
-		case "sma":
-			ma = standardIndicatorSet.SMA(types.IntervalWindow{
-				Interval: detection.MovingAverageIntervalWindow.Interval,
-				Window:   detection.MovingAverageIntervalWindow.Window,
-			})
-		case "ema", "ewma":
-			ma = standardIndicatorSet.EWMA(types.IntervalWindow{
-				Interval: detection.MovingAverageIntervalWindow.Interval,
-				Window:   detection.MovingAverageIntervalWindow.Window,
-			})
-		default:
-			ma = standardIndicatorSet.EWMA(types.IntervalWindow{
-				Interval: detection.MovingAverageIntervalWindow.Interval,
-				Window:   detection.MovingAverageIntervalWindow.Window,
-			})
-		}
-
-	}
-
-	session.MarketDataStream.OnKLineClosed(func(kline types.KLine) {
-		// skip k-lines from other symbols
-		if kline.Symbol != s.Symbol {
-			return
-		}
-		for _, detection := range s.SupportDetection {
-			var lastMA = ma.Last()
-
-			closePrice := kline.GetClose()
-			closePriceF := closePrice.Float64()
-			// skip if the closed price is under the moving average
-			if closePriceF < lastMA {
-				log.Infof("skip %s closed price %v < last ma %f", s.Symbol, closePrice, lastMA)
-				return
-			}
-
-			fundingRate := premiumIndex.LastFundingRate
-
-			if fundingRate.Compare(s.FundingRate.High) >= 0 {
-				bbgo.Notify("%s funding rate %s is too high! threshold %s",
-					s.Symbol,
-					fundingRate.Percentage(),
-					s.FundingRate.High.Percentage(),
-				)
-			} else {
-				log.Infof("skip funding rate is too low")
-				return
-			}
-
-			prettyBaseVolume := s.Market.BaseCurrencyFormatter()
-			prettyQuoteVolume := s.Market.QuoteCurrencyFormatter()
-
-			if detection.MinVolume.Sign() > 0 && kline.Volume.Compare(detection.MinVolume) > 0 {
-				bbgo.Notify("Detected %s %s resistance base volume %s > min base volume %s, quote volume %s",
-					s.Symbol, detection.Interval.String(),
-					prettyBaseVolume.FormatMoney(kline.Volume.Trunc()),
-					prettyBaseVolume.FormatMoney(detection.MinVolume.Trunc()),
-					prettyQuoteVolume.FormatMoney(kline.QuoteVolume.Trunc()),
-				)
-				bbgo.Notify(kline)
-
-				baseBalance, ok := session.GetAccount().Balance(s.Market.BaseCurrency)
-				if !ok {
-					return
-				}
-
-				if baseBalance.Available.Sign() > 0 && baseBalance.Total().Compare(s.MaxExposurePosition) < 0 {
-					log.Infof("opening a short position")
-					_, err := orderExecutor.SubmitOrders(ctx, types.SubmitOrder{
-						Symbol:   kline.Symbol,
-						Side:     types.SideTypeSell,
-						Type:     types.OrderTypeMarket,
-						Quantity: s.Quantity,
-					})
-					if err != nil {
-						log.WithError(err).Error("submit order error")
-					}
-				}
-			} else if detection.MinQuoteVolume.Sign() > 0 && kline.QuoteVolume.Compare(detection.MinQuoteVolume) > 0 {
-				bbgo.Notify("Detected %s %s resistance quote volume %s > min quote volume %s, base volume %s",
-					s.Symbol, detection.Interval.String(),
-					prettyQuoteVolume.FormatMoney(kline.QuoteVolume.Trunc()),
-					prettyQuoteVolume.FormatMoney(detection.MinQuoteVolume.Trunc()),
-					prettyBaseVolume.FormatMoney(kline.Volume.Trunc()),
-				)
-				bbgo.Notify(kline)
-			}
-		}
-	})
-	return nil
-}

pkg/strategy/xfunding/positionstate_string.go

@@ -0,0 +1,26 @@
+// Code generated by "stringer -type=PositionState"; 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[PositionClosed-0]
+	_ = x[PositionOpening-1]
+	_ = x[PositionReady-2]
+	_ = x[PositionClosing-3]
+}
+
+const _PositionState_name = "PositionClosedPositionOpeningPositionReadyPositionClosing"
+
+var _PositionState_index = [...]uint8{0, 14, 29, 42, 57}
+
+func (i PositionState) String() string {
+	if i < 0 || i >= PositionState(len(_PositionState_index)-1) {
+		return "PositionState(" + strconv.FormatInt(int64(i), 10) + ")"
+	}
+	return _PositionState_name[_PositionState_index[i]:_PositionState_index[i+1]]
+}

pkg/strategy/xfunding/strategy.go

@@ -0,0 +1,784 @@
+package xfunding
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"strings"
+	"sync"
+	"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"
+)
+
+const ID = "xfunding"
+
+// Position State Transitions:
+// NoOp -> Opening
+// Opening -> Ready -> Closing
+// Closing -> Closed -> Opening
+//go:generate stringer -type=PositionState
+type PositionState int
+
+const (
+	PositionClosed PositionState = iota
+	PositionOpening
+	PositionReady
+	PositionClosing
+)
+
+var log = logrus.WithField("strategy", ID)
+
+func init() {
+	// Register the pointer of the strategy struct,
+	// so that bbgo knows what struct to be used to unmarshal the configs (YAML or JSON)
+	// Note: built-in strategies need to imported manually in the bbgo cmd package.
+	bbgo.RegisterStrategy(ID, &Strategy{})
+}
+
+type State struct {
+	PositionStartTime time.Time `json:"positionStartTime"`
+
+	// PositionState is default to NoOp
+	PositionState PositionState
+
+	PendingBaseTransfer fixedpoint.Value `json:"pendingBaseTransfer"`
+	TotalBaseTransfer   fixedpoint.Value `json:"totalBaseTransfer"`
+	UsedQuoteInvestment fixedpoint.Value `json:"usedQuoteInvestment"`
+}
+
+func (s *State) Reset() {
+	s.PositionState = PositionClosed
+	s.PendingBaseTransfer = fixedpoint.Zero
+	s.TotalBaseTransfer = fixedpoint.Zero
+	s.UsedQuoteInvestment = fixedpoint.Zero
+}
+
+// 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:"-"`
+
+	// 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
+	IncrementalQuoteQuantity fixedpoint.Value `json:"incrementalQuoteQuantity"`
+
+	QuoteInvestment fixedpoint.Value `json:"quoteInvestment"`
+
+	MinHoldingPeriod types.Duration `json:"minHoldingPeriod"`
+
+	// ShortFundingRate is the funding rate range for short positions
+	// TODO: right now we don't support negative funding rate (long position) since it's rarer
+	ShortFundingRate *struct {
+		High fixedpoint.Value `json:"high"`
+		Low  fixedpoint.Value `json:"low"`
+	} `json:"shortFundingRate"`
+
+	SupportDetection []struct {
+		Interval types.Interval `json:"interval"`
+		// MovingAverageType is the moving average indicator type that we want to use,
+		// it could be SMA or EWMA
+		MovingAverageType string `json:"movingAverageType"`
+
+		// MovingAverageInterval is the interval of k-lines for the moving average indicator to calculate,
+		// it could be "1m", "5m", "1h" and so on.  note that, the moving averages are calculated from
+		// the k-line data we subscribed
+		// MovingAverageInterval types.Interval `json:"movingAverageInterval"`
+		//
+		// // MovingAverageWindow is the number of the window size of the moving average indicator.
+		// // The number of k-lines in the window. generally used window sizes are 7, 25 and 99 in the TradingView.
+		// MovingAverageWindow int `json:"movingAverageWindow"`
+
+		MovingAverageIntervalWindow types.IntervalWindow `json:"movingAverageIntervalWindow"`
+
+		MinVolume fixedpoint.Value `json:"minVolume"`
+
+		MinQuoteVolume fixedpoint.Value `json:"minQuoteVolume"`
+	} `json:"supportDetection"`
+
+	SpotSession    string `json:"spotSession"`
+	FuturesSession string `json:"futuresSession"`
+	Reset          bool   `json:"reset"`
+
+	ProfitStats     *types.ProfitStats `persistence:"profit_stats"`
+	SpotPosition    *types.Position    `persistence:"spot_position"`
+	FuturesPosition *types.Position    `persistence:"futures_position"`
+
+	State *State `persistence:"state"`
+
+	// mu is used for locking state
+	mu sync.Mutex
+
+	spotSession, futuresSession             *bbgo.ExchangeSession
+	spotOrderExecutor, futuresOrderExecutor *bbgo.GeneralOrderExecutor
+	spotMarket, futuresMarket               types.Market
+
+	// positionType is the futures position type
+	// currently we only support short position for the positive funding rate
+	positionType types.PositionType
+}
+
+func (s *Strategy) ID() string {
+	return ID
+}
+
+func (s *Strategy) CrossSubscribe(sessions map[string]*bbgo.ExchangeSession) {
+	// TODO: add safety check
+	spotSession := sessions[s.SpotSession]
+	futuresSession := sessions[s.FuturesSession]
+
+	spotSession.Subscribe(types.KLineChannel, s.Symbol, types.SubscribeOptions{
+		Interval: types.Interval1m,
+	})
+
+	futuresSession.Subscribe(types.KLineChannel, s.Symbol, types.SubscribeOptions{
+		Interval: types.Interval1m,
+	})
+}
+
+func (s *Strategy) Subscribe(session *bbgo.ExchangeSession) {}
+
+func (s *Strategy) Defaults() error {
+	if s.Leverage.IsZero() {
+		s.Leverage = fixedpoint.One
+	}
+
+	if s.MinHoldingPeriod == 0 {
+		s.MinHoldingPeriod = types.Duration(3 * 24 * time.Hour)
+	}
+
+	s.positionType = types.PositionShort
+
+	return nil
+}
+
+func (s *Strategy) Validate() error {
+	if len(s.Symbol) == 0 {
+		return errors.New("symbol is required")
+	}
+
+	if len(s.SpotSession) == 0 {
+		return errors.New("spotSession name is required")
+	}
+
+	if len(s.FuturesSession) == 0 {
+		return errors.New("futuresSession name is required")
+	}
+
+	if s.QuoteInvestment.IsZero() {
+		return errors.New("quoteInvestment can not be zero")
+	}
+
+	return nil
+}
+
+func (s *Strategy) InstanceID() string {
+	return fmt.Sprintf("%s-%s", ID, s.Symbol)
+}
+
+func (s *Strategy) Run(ctx context.Context, orderExecutor bbgo.OrderExecutor, session *bbgo.ExchangeSession) error {
+	standardIndicatorSet := session.StandardIndicatorSet(s.Symbol)
+
+	var ma types.Float64Indicator
+	for _, detection := range s.SupportDetection {
+
+		switch strings.ToLower(detection.MovingAverageType) {
+		case "sma":
+			ma = standardIndicatorSet.SMA(types.IntervalWindow{
+				Interval: detection.MovingAverageIntervalWindow.Interval,
+				Window:   detection.MovingAverageIntervalWindow.Window,
+			})
+		case "ema", "ewma":
+			ma = standardIndicatorSet.EWMA(types.IntervalWindow{
+				Interval: detection.MovingAverageIntervalWindow.Interval,
+				Window:   detection.MovingAverageIntervalWindow.Window,
+			})
+		default:
+			ma = standardIndicatorSet.EWMA(types.IntervalWindow{
+				Interval: detection.MovingAverageIntervalWindow.Interval,
+				Window:   detection.MovingAverageIntervalWindow.Window,
+			})
+		}
+	}
+	_ = ma
+
+	return nil
+}
+
+func (s *Strategy) CrossRun(ctx context.Context, orderExecutionRouter bbgo.OrderExecutionRouter, sessions map[string]*bbgo.ExchangeSession) error {
+	instanceID := s.InstanceID()
+
+	s.spotSession = sessions[s.SpotSession]
+	s.futuresSession = sessions[s.FuturesSession]
+
+	s.spotMarket, _ = s.spotSession.Market(s.Symbol)
+	s.futuresMarket, _ = s.futuresSession.Market(s.Symbol)
+
+	// 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 = types.NewProfitStats(s.Market)
+	}
+
+	if s.FuturesPosition == nil || s.Reset {
+		s.FuturesPosition = types.NewPositionFromMarket(s.futuresMarket)
+	}
+
+	if s.SpotPosition == nil || s.Reset {
+		s.SpotPosition = types.NewPositionFromMarket(s.spotMarket)
+	}
+
+	if s.State == nil || s.Reset {
+		s.State = &State{
+			PositionState:       PositionClosed,
+			PendingBaseTransfer: fixedpoint.Zero,
+			TotalBaseTransfer:   fixedpoint.Zero,
+			UsedQuoteInvestment: fixedpoint.Zero,
+		}
+	}
+
+	log.Infof("loaded spot position: %s", s.SpotPosition.String())
+	log.Infof("loaded futures position: %s", s.FuturesPosition.String())
+
+	binanceFutures := s.futuresSession.Exchange.(*binance.Exchange)
+	binanceSpot := s.spotSession.Exchange.(*binance.Exchange)
+	_ = binanceSpot
+
+	s.spotOrderExecutor = s.allocateOrderExecutor(ctx, s.spotSession, instanceID, s.SpotPosition)
+	s.spotOrderExecutor.TradeCollector().OnTrade(func(trade types.Trade, profit fixedpoint.Value, netProfit fixedpoint.Value) {
+		// we act differently on the spot account
+		// when opening a position, we place orders on the spot account first, then the futures account,
+		// and we need to accumulate the used quote amount
+		//
+		// when closing a position, we place orders on the futures account first, then the spot account
+		// we need to close the position according to its base quantity instead of quote quantity
+		if s.positionType != types.PositionShort {
+			return
+		}
+
+		switch s.State.PositionState {
+		case PositionOpening:
+			if trade.Side != types.SideTypeBuy {
+				log.Errorf("unexpected trade side: %+v, expecting BUY trade", trade)
+				return
+			}
+
+			s.mu.Lock()
+			s.State.UsedQuoteInvestment = s.State.UsedQuoteInvestment.Add(trade.QuoteQuantity)
+			s.mu.Unlock()
+
+			// if we have trade, try to query the balance and transfer the balance to the futures wallet account
+			// TODO: handle missing trades here. If the process crashed during the transfer, how to recover?
+			if err := backoff.RetryGeneral(ctx, func() error {
+				return s.transferIn(ctx, binanceSpot, s.spotMarket.BaseCurrency, trade)
+			}); err != nil {
+				log.WithError(err).Errorf("spot-to-futures transfer in retry failed")
+				return
+			}
+
+		case PositionClosing:
+			if trade.Side != types.SideTypeSell {
+				log.Errorf("unexpected trade side: %+v, expecting SELL trade", trade)
+				return
+			}
+
+		}
+	})
+
+	s.futuresOrderExecutor = s.allocateOrderExecutor(ctx, s.futuresSession, instanceID, s.FuturesPosition)
+	s.futuresOrderExecutor.TradeCollector().OnTrade(func(trade types.Trade, profit fixedpoint.Value, netProfit fixedpoint.Value) {
+		if s.positionType != types.PositionShort {
+			return
+		}
+
+		switch s.getPositionState() {
+		case PositionClosing:
+			if err := backoff.RetryGeneral(ctx, func() error {
+				return s.transferOut(ctx, binanceSpot, s.spotMarket.BaseCurrency, trade)
+			}); err != nil {
+				log.WithError(err).Errorf("spot-to-futures transfer in retry failed")
+				return
+			}
+
+		}
+	})
+
+	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)
+				s.sync(ctx)
+			}
+		}
+	}()
+
+	// 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
+	}
+
+	log.Info(premiumIndex)
+
+	if changed := s.detectPremiumIndex(premiumIndex); changed {
+		log.Infof("position state changed to -> %s %s", s.positionType, s.State.PositionState.String())
+	}
+}
+
+func (s *Strategy) sync(ctx context.Context) {
+	switch s.getPositionState() {
+	case PositionOpening:
+		s.increaseSpotPosition(ctx)
+		s.syncFuturesPosition(ctx)
+	case PositionClosing:
+		s.reduceFuturesPosition(ctx)
+		s.syncSpotPosition(ctx)
+	}
+}
+
+func (s *Strategy) reduceFuturesPosition(ctx context.Context) {
+	if s.notPositionState(PositionClosing) {
+		return
+	}
+
+	futuresBase := s.FuturesPosition.GetBase() // should be negative base quantity here
+
+	if futuresBase.Sign() > 0 {
+		// unexpected error
+		log.Errorf("unexpected futures position (got positive, expecting negative)")
+		return
+	}
+
+	_ = 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
+	}
+
+	if futuresBase.Compare(fixedpoint.Zero) < 0 {
+		orderPrice := ticker.Buy
+		orderQuantity := futuresBase.Abs()
+		orderQuantity = fixedpoint.Max(orderQuantity, s.futuresMarket.MinQuantity)
+		orderQuantity = s.futuresMarket.AdjustQuantityByMinNotional(orderQuantity, orderPrice)
+		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.SideTypeBuy,
+			Type:       types.OrderTypeLimitMaker,
+			Quantity:   orderQuantity,
+			Price:      orderPrice,
+			Market:     s.futuresMarket,
+			ReduceOnly: true,
+		})
+
+		if err != nil {
+			log.WithError(err).Errorf("can not submit order")
+			return
+		}
+
+		log.Infof("created orders: %+v", createdOrders)
+	}
+}
+
+// syncFuturesPosition syncs the futures position with the given spot position
+// when the spot is transferred successfully, sync futures position
+// compare spot position and futures position, increase the position size until they are the same size
+func (s *Strategy) syncFuturesPosition(ctx context.Context) {
+	if s.positionType != types.PositionShort {
+		return
+	}
+
+	if s.notPositionState(PositionOpening) {
+		return
+	}
+
+	spotBase := s.SpotPosition.GetBase()       // should be positive base quantity here
+	futuresBase := s.FuturesPosition.GetBase() // should be negative base quantity here
+
+	if spotBase.IsZero() || spotBase.Sign() < 0 {
+		// 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
+	}
+
+	_ = 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
+	}
+
+	// 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())
+
+	// max futures base position (without negative sign)
+	maxFuturesBasePosition := fixedpoint.Min(
+		spotBase.Mul(s.Leverage),
+		s.State.TotalBaseTransfer.Mul(s.Leverage))
+
+	// if - futures position < max futures position, increase it
+	if futuresBase.Neg().Compare(maxFuturesBasePosition) < 0 {
+		orderPrice := ticker.Sell
+		diffQuantity := maxFuturesBasePosition.Sub(futuresBase.Neg())
+
+		if diffQuantity.Sign() < 0 {
+			log.Errorf("unexpected negative position diff: %s", diffQuantity.String())
+			return
+		}
+
+		log.Infof("position diff quantity: %s", diffQuantity.String())
+
+		orderQuantity := fixedpoint.Max(diffQuantity, s.futuresMarket.MinQuantity)
+		orderQuantity = s.futuresMarket.AdjustQuantityByMinNotional(orderQuantity, orderPrice)
+		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,
+		})
+
+		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) {
+	if s.positionType != types.PositionShort {
+		return
+	}
+
+	if s.notPositionState(PositionClosing) {
+		return
+	}
+
+	spotBase := s.SpotPosition.GetBase()       // should be positive base quantity here
+	futuresBase := s.FuturesPosition.GetBase() // should be negative base quantity here
+
+	if spotBase.IsZero() {
+		s.setPositionState(PositionClosed)
+		return
+	}
+
+	// skip short spot position
+	if spotBase.Sign() < 0 {
+		return
+	}
+
+	log.Infof("spot/futures positions: %s (spot) <=> %s (futures)", spotBase.String(), futuresBase.String())
+
+	if futuresBase.Sign() > 0 {
+		// unexpected error
+		log.Errorf("unexpected futures position (got positive, expecting negative)")
+		return
+	}
+
+	_ = s.futuresOrderExecutor.GracefulCancel(ctx)
+
+	ticker, err := s.spotSession.Exchange.QueryTicker(ctx, s.Symbol)
+	if err != nil {
+		log.WithError(err).Errorf("can not query ticker")
+		return
+	}
+
+	if s.SpotPosition.IsDust(ticker.Sell) {
+		dust := s.SpotPosition.GetBase().Abs()
+		cost := s.SpotPosition.AverageCost
+
+		log.Warnf("spot dust loss: %f %s (average cost = %f)", dust.Float64(), s.spotMarket.BaseCurrency, cost.Float64())
+
+		s.SpotPosition.Reset()
+
+		s.setPositionState(PositionClosed)
+		return
+	}
+
+	// spot pos size > futures pos size ==> reduce spot position
+	if spotBase.Compare(futuresBase.Neg()) > 0 {
+		diffQuantity := spotBase.Sub(futuresBase.Neg())
+
+		if diffQuantity.Sign() < 0 {
+			log.Errorf("unexpected negative position diff: %s", diffQuantity.String())
+			return
+		}
+
+		orderPrice := ticker.Sell
+		orderQuantity := diffQuantity
+		if b, ok := s.spotSession.Account.Balance(s.spotMarket.BaseCurrency); ok {
+			orderQuantity = fixedpoint.Min(b.Available, orderQuantity)
+		}
+
+		// avoid increase the order size
+		if s.spotMarket.IsDustQuantity(orderQuantity, orderPrice) {
+			log.Infof("skip futures order with dust quantity %s, market = %+v", orderQuantity.String(), s.spotMarket)
+			return
+		}
+
+		createdOrders, err := s.spotOrderExecutor.SubmitOrders(ctx, types.SubmitOrder{
+			Symbol:   s.Symbol,
+			Side:     types.SideTypeSell,
+			Type:     types.OrderTypeLimitMaker,
+			Quantity: orderQuantity,
+			Price:    orderPrice,
+			Market:   s.futuresMarket,
+		})
+
+		if err != nil {
+			log.WithError(err).Errorf("can not submit spot order")
+			return
+		}
+
+		log.Infof("created spot orders: %+v", createdOrders)
+	}
+}
+
+func (s *Strategy) increaseSpotPosition(ctx context.Context) {
+	if s.positionType != types.PositionShort {
+		log.Errorf("funding long position type is not supported")
+		return
+	}
+
+	if s.notPositionState(PositionOpening) {
+		return
+	}
+
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	if s.State.UsedQuoteInvestment.Compare(s.QuoteInvestment) >= 0 {
+		// stop increase the position
+		s.setPositionState(PositionReady)
+
+		// DEBUG CODE - triggering closing position automatically
+		// s.startClosingPosition()
+		return
+	}
+
+	_ = s.spotOrderExecutor.GracefulCancel(ctx)
+
+	ticker, err := s.spotSession.Exchange.QueryTicker(ctx, s.Symbol)
+	if err != nil {
+		log.WithError(err).Errorf("can not query ticker")
+		return
+	}
+
+	leftQuota := s.QuoteInvestment.Sub(s.State.UsedQuoteInvestment)
+
+	orderPrice := ticker.Buy
+	orderQuantity := fixedpoint.Min(s.IncrementalQuoteQuantity, leftQuota).Div(orderPrice)
+
+	log.Infof("initial spot order quantity %s", orderQuantity.String())
+
+	orderQuantity = fixedpoint.Max(orderQuantity, s.spotMarket.MinQuantity)
+	orderQuantity = s.spotMarket.AdjustQuantityByMinNotional(orderQuantity, orderPrice)
+
+	if s.spotMarket.IsDustQuantity(orderQuantity, orderPrice) {
+		return
+	}
+
+	submitOrder := types.SubmitOrder{
+		Symbol:   s.Symbol,
+		Side:     types.SideTypeBuy,
+		Type:     types.OrderTypeLimitMaker,
+		Quantity: orderQuantity,
+		Price:    orderPrice,
+		Market:   s.spotMarket,
+	}
+
+	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
+	}
+
+	log.Infof("created orders: %+v", createdOrders)
+}
+
+func (s *Strategy) detectPremiumIndex(premiumIndex *types.PremiumIndex) bool {
+	fundingRate := premiumIndex.LastFundingRate
+
+	log.Infof("last %s funding rate: %s", s.Symbol, fundingRate.Percentage())
+
+	if s.ShortFundingRate == nil {
+		return false
+	}
+
+	switch s.getPositionState() {
+
+	case PositionClosed:
+		if fundingRate.Compare(s.ShortFundingRate.High) >= 0 {
+			log.Infof("funding rate %s is higher than the High threshold %s, start opening position...",
+				fundingRate.Percentage(), s.ShortFundingRate.High.Percentage())
+
+			s.startOpeningPosition(types.PositionShort, premiumIndex.Time)
+			return true
+		}
+
+	case PositionReady:
+		if fundingRate.Compare(s.ShortFundingRate.Low) <= 0 {
+			log.Infof("funding rate %s is lower than the Low threshold %s, start closing position...",
+				fundingRate.Percentage(), s.ShortFundingRate.Low.Percentage())
+
+			holdingPeriod := premiumIndex.Time.Sub(s.State.PositionStartTime)
+			if holdingPeriod < time.Duration(s.MinHoldingPeriod) {
+				log.Warnf("position holding period %s is less than %s, skip closing", holdingPeriod, s.MinHoldingPeriod.Duration())
+				return false
+			}
+
+			s.startClosingPosition()
+			return true
+		}
+	}
+
+	return false
+}
+
+func (s *Strategy) startOpeningPosition(pt types.PositionType, t time.Time) {
+	// only open a new position when there is no position
+	if s.notPositionState(PositionClosed) {
+		return
+	}
+
+	log.Infof("startOpeningPosition")
+	s.setPositionState(PositionOpening)
+
+	s.positionType = pt
+
+	// reset the transfer stats
+	s.State.PositionStartTime = t
+	s.State.PendingBaseTransfer = fixedpoint.Zero
+	s.State.TotalBaseTransfer = fixedpoint.Zero
+}
+
+func (s *Strategy) startClosingPosition() {
+	// we can't close a position that is not ready
+	if s.notPositionState(PositionReady) {
+		return
+	}
+
+	log.Infof("startClosingPosition")
+	s.setPositionState(PositionClosing)
+
+	// reset the transfer stats
+	s.State.PendingBaseTransfer = fixedpoint.Zero
+}
+
+func (s *Strategy) setPositionState(state PositionState) {
+	s.mu.Lock()
+	origState := s.State.PositionState
+	s.State.PositionState = state
+	s.mu.Unlock()
+	log.Infof("position state transition: %s -> %s", origState.String(), state.String())
+}
+
+func (s *Strategy) isPositionState(state PositionState) bool {
+	s.mu.Lock()
+	ret := s.State.PositionState == state
+	s.mu.Unlock()
+	return ret
+}
+
+func (s *Strategy) getPositionState() PositionState {
+	return s.State.PositionState
+}
+
+func (s *Strategy) notPositionState(state PositionState) bool {
+	s.mu.Lock()
+	ret := s.State.PositionState != state
+	s.mu.Unlock()
+	return ret
+}
+
+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) {
+		s.ProfitStats.AddTrade(trade)
+	})
+	orderExecutor.TradeCollector().OnPositionUpdate(func(position *types.Position) {
+		bbgo.Sync(ctx, s)
+	})
+	return orderExecutor
+}

pkg/strategy/xfunding/transfer.go

@@ -0,0 +1,127 @@
+package xfunding
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/c9s/bbgo/pkg/fixedpoint"
+	"github.com/c9s/bbgo/pkg/types"
+)
+
+type FuturesTransfer interface {
+	TransferFuturesAccountAsset(ctx context.Context, asset string, amount fixedpoint.Value, io types.TransferDirection) error
+	QueryAccountBalances(ctx context.Context) (types.BalanceMap, error)
+}
+
+func (s *Strategy) transferOut(ctx context.Context, ex FuturesTransfer, currency string, trade types.Trade) error {
+	// base asset needs BUY trades
+	if trade.Side != types.SideTypeBuy {
+		return nil
+	}
+
+	// if transfer done
+	if s.State.TotalBaseTransfer.IsZero() {
+		return nil
+	}
+
+	// de-leverage and get the collateral base quantity for transfer
+	quantity := trade.Quantity
+	quantity = quantity.Div(s.Leverage)
+
+	balances, err := s.futuresSession.Exchange.QueryAccountBalances(ctx)
+	if err != nil {
+		log.Infof("adding to pending base transfer: %s %s + %s", quantity.String(), currency, s.State.PendingBaseTransfer.String())
+		s.State.PendingBaseTransfer = s.State.PendingBaseTransfer.Add(quantity)
+		return err
+	}
+
+	b, ok := balances[currency]
+	if !ok {
+		log.Infof("adding to pending base transfer: %s %s + %s", quantity.String(), currency, s.State.PendingBaseTransfer.String())
+		s.State.PendingBaseTransfer = s.State.PendingBaseTransfer.Add(quantity)
+		return fmt.Errorf("%s balance not found", currency)
+	}
+
+	// add the previous pending base transfer and the current trade quantity
+	amount := s.State.PendingBaseTransfer.Add(quantity)
+
+	// try to transfer more if we enough balance
+	amount = fixedpoint.Min(amount, b.Available)
+
+	// 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("adding to pending base transfer: %s %s + %s ", quantity.String(), currency, 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, currency)
+	if err := ex.TransferFuturesAccountAsset(ctx, currency, 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)
+	return nil
+}
+
+func (s *Strategy) transferIn(ctx context.Context, ex FuturesTransfer, currency string, trade types.Trade) error {
+
+	// base asset needs BUY trades
+	if trade.Side == types.SideTypeSell {
+		return nil
+	}
+
+	balances, err := s.spotSession.Exchange.QueryAccountBalances(ctx)
+	if err != nil {
+		return err
+	}
+
+	b, ok := balances[currency]
+	if !ok {
+		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, we can adjust the quantity here
+	quantity := trade.Quantity
+	if b.Available.Compare(quantity) < 0 {
+		log.Infof("adding to pending base transfer: %s %s", quantity, currency)
+		s.State.PendingBaseTransfer = s.State.PendingBaseTransfer.Add(quantity)
+		return nil
+	}
+
+	amount := s.State.PendingBaseTransfer.Add(quantity)
+
+	pos := s.SpotPosition.GetBase().Abs()
+	rest := pos.Sub(s.State.TotalBaseTransfer)
+
+	if rest.Sign() < 0 {
+		return nil
+	}
+
+	amount = fixedpoint.Min(rest, amount)
+
+	log.Infof("transfering in futures account asset %s %s", amount, currency)
+	if err := ex.TransferFuturesAccountAsset(ctx, currency, 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
+}

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/types/channel.go

@@ -2,8 +2,18 @@ package types
 
 type Channel string
 
-var BookChannel = Channel("book")
+const (
-var KLineChannel = Channel("kline")
+	BookChannel        = Channel("book")
-var BookTickerChannel = Channel("bookticker")
+	KLineChannel       = Channel("kline")
-var MarketTradeChannel = Channel("trade")
+	BookTickerChannel  = Channel("bookTicker")
-var AggTradeChannel = Channel("aggTrade")
+	MarketTradeChannel = Channel("trade")
+	AggTradeChannel    = Channel("aggTrade")
+
+	// channels for futures
+	MarkPriceChannel = Channel("markPrice")
+
+	LiquidationOrderChannel = Channel("liquidationOrder")
+
+	// ContractInfoChannel is the contract info provided by the exchange
+	ContractInfoChannel = Channel("contractInfo")
+)

pkg/types/market.go

@@ -2,6 +2,7 @@ package types
 
 import (
 	"math"
+	"strconv"
 
 	"github.com/leekchan/accounting"
 
@@ -59,7 +60,14 @@ func (m Market) IsDustQuantity(quantity, price fixedpoint.Value) bool {
 
 // TruncateQuantity uses the step size to truncate floating number, in order to avoid the rounding issue
 func (m Market) TruncateQuantity(quantity fixedpoint.Value) fixedpoint.Value {
-	return fixedpoint.MustNewFromString(m.FormatQuantity(quantity))
+	var ts = m.StepSize.Float64()
+	var prec = int(math.Round(math.Log10(ts) * -1.0))
+	var pow10 = math.Pow10(prec)
+
+	qf := math.Trunc(quantity.Float64() * pow10)
+	qf = qf / pow10
+	qs := strconv.FormatFloat(qf, 'f', prec, 64)
+	return fixedpoint.MustNewFromString(qs)
 }
 
 func (m Market) TruncatePrice(price fixedpoint.Value) fixedpoint.Value {
@@ -136,6 +144,23 @@ func (m Market) CanonicalizeVolume(val fixedpoint.Value) float64 {
 	return math.Trunc(p*val.Float64()) / p
 }
 
+// AdjustQuantityByMinNotional adjusts the quantity to make the amount greater than the given minAmount
+func (m Market) AdjustQuantityByMinNotional(quantity, currentPrice fixedpoint.Value) fixedpoint.Value {
+	// modify quantity for the min amount
+	quantity = m.TruncateQuantity(quantity)
+	amount := currentPrice.Mul(quantity)
+	if amount.Compare(m.MinNotional) < 0 {
+		ratio := m.MinNotional.Div(amount)
+		quantity = quantity.Mul(ratio)
+
+		ts := m.StepSize.Float64()
+		prec := int(math.Round(math.Log10(ts) * -1.0))
+		return quantity.Round(prec, fixedpoint.Up)
+	}
+
+	return quantity
+}
+
 type MarketMap map[string]Market
 
 func (m MarketMap) Add(market Market) {

pkg/types/market_test.go

@@ -191,3 +191,53 @@ func Test_formatQuantity(t *testing.T) {
 		})
 	}
 }
+
+func TestMarket_TruncateQuantity(t *testing.T) {
+	market := Market{
+		StepSize: fixedpoint.NewFromFloat(0.0001),
+	}
+
+	testCases := []struct {
+		input  string
+		expect string
+	}{
+		{"0.00573961", "0.0057"},
+		{"0.00579961", "0.0057"},
+		{"0.0057", "0.0057"},
+	}
+
+	for _, testCase := range testCases {
+		q := fixedpoint.MustNewFromString(testCase.input)
+		q2 := market.TruncateQuantity(q)
+		assert.Equalf(t, testCase.expect, q2.String(), "input: %s stepSize: %s", testCase.input, market.StepSize.String())
+	}
+
+}
+
+func TestMarket_AdjustQuantityByMinNotional(t *testing.T) {
+
+	market := Market{
+		Symbol:          "ETHUSDT",
+		StepSize:        fixedpoint.NewFromFloat(0.0001),
+		MinQuantity:     fixedpoint.NewFromFloat(0.0001),
+		MinNotional:     fixedpoint.NewFromFloat(10.0),
+		VolumePrecision: 8,
+		PricePrecision:  2,
+	}
+
+	// Quantity:0.00573961 Price:1750.99
+	testCases := []struct {
+		input  string
+		expect string
+	}{
+		{"0.00573961", "0.0058"},
+	}
+
+	price := fixedpoint.NewFromFloat(1750.99)
+	for _, testCase := range testCases {
+		q := fixedpoint.MustNewFromString(testCase.input)
+		q2 := market.AdjustQuantityByMinNotional(q, price)
+		assert.Equalf(t, testCase.expect, q2.String(), "input: %s stepSize: %s", testCase.input, market.StepSize.String())
+		assert.False(t, market.IsDustQuantity(q2, price))
+	}
+}

pkg/types/premiumindex.go

@@ -1,6 +1,7 @@
 package types
 
 import (
+	"fmt"
 	"time"
 
 	"github.com/c9s/bbgo/pkg/fixedpoint"
@@ -13,3 +14,7 @@ type PremiumIndex struct {
 	NextFundingTime time.Time        `json:"nextFundingTime"`
 	Time            time.Time        `json:"time"`
 }
+
+func (i *PremiumIndex) String() string {
+	return fmt.Sprintf("PremiumIndex | %s | %.4f | %s | %s | NEXT FUNDING TIME: %s", i.Symbol, i.MarkPrice.Float64(), i.LastFundingRate.Percentage(), i.Time, i.NextFundingTime)
+}

pkg/types/transfer.go

@@ -0,0 +1,8 @@
+package types
+
+type TransferDirection int
+
+const (
+	TransferIn  TransferDirection = 1
+	TransferOut TransferDirection = -1
+)

pkg/util/backoff/general.go

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