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

strategy: xfunding: add profit stats and collect funding fee info

config/xfunding.yaml

@@ -30,11 +30,28 @@ crossExchangeStrategies:
 - xfunding:
     spotSession: binance
     futuresSession: binance_futures
+
+    ## symbol is the symbol name of the spot market and the futures market
+    ## todo: provide option to separate the futures market symbol
     symbol: ETHUSDT
+
+    ## interval is the interval for checking futures premium and the funding rate
+    interval: 1m
+
+    ## leverage is the leverage of the reverse futures position size.
+    ## for example, you can buy 1 BTC and short 3 BTC in the futures account with 3x leverage.
     leverage: 1.0
+
+    ## incrementalQuoteQuantity is the quote quantity per maker order when creating the positions
+    ## when in BTC-USDT 20 means 20 USDT, each buy order will hold 20 USDT quote amount.
     incrementalQuoteQuantity: 20
+
+    ## quoteInvestment is how much you want to invest to create your position.
+    ## for example, when 10k USDT is given as the quote investment, and the average executed price of your position is around BTC 18k
+    ## you will be holding around 0.555555 BTC
     quoteInvestment: 50
 
+    ## shortFundingRate is the funding rate range you want to create your position
     shortFundingRate:
       ## when funding rate is higher than this high value, the strategy will start buying spot and opening a short position
       high: 0.001%

pkg/bbgo/order_executor_general.go

@@ -155,8 +155,10 @@ func (e *GeneralOrderExecutor) BindProfitStats(profitStats *types.ProfitStats) {
 
 		profitStats.AddProfit(*profit)
 
-		Notify(profit)
+		if !e.disableNotify {
-		Notify(profitStats)
+			Notify(profit)
+			Notify(profitStats)
+		}
 	})
 }
 

pkg/cmd/margin.go

@@ -88,9 +88,9 @@ var marginLoansCmd = &cobra.Command{
 			return errors.New("session is not set")
 		}
 
-		marginHistoryService, ok := selectedSession.Exchange.(types.MarginHistory)
+		marginHistoryService, ok := selectedSession.Exchange.(types.MarginHistoryService)
 		if !ok {
-			return fmt.Errorf("exchange %s does not support MarginHistory service", selectedSession.ExchangeName)
+			return fmt.Errorf("exchange %s does not support MarginHistoryService service", selectedSession.ExchangeName)
 		}
 
 		now := time.Now()
@@ -127,9 +127,9 @@ var marginRepaysCmd = &cobra.Command{
 			return errors.New("session is not set")
 		}
 
-		marginHistoryService, ok := selectedSession.Exchange.(types.MarginHistory)
+		marginHistoryService, ok := selectedSession.Exchange.(types.MarginHistoryService)
 		if !ok {
-			return fmt.Errorf("exchange %s does not support MarginHistory service", selectedSession.ExchangeName)
+			return fmt.Errorf("exchange %s does not support MarginHistoryService service", selectedSession.ExchangeName)
 		}
 
 		now := time.Now()
@@ -166,9 +166,9 @@ var marginInterestsCmd = &cobra.Command{
 			return errors.New("session is not set")
 		}
 
-		marginHistoryService, ok := selectedSession.Exchange.(types.MarginHistory)
+		marginHistoryService, ok := selectedSession.Exchange.(types.MarginHistoryService)
 		if !ok {
-			return fmt.Errorf("exchange %s does not support MarginHistory service", selectedSession.ExchangeName)
+			return fmt.Errorf("exchange %s does not support MarginHistoryService service", selectedSession.ExchangeName)
 		}
 
 		now := time.Now()

pkg/exchange/batch/funding_fee.go

@@ -0,0 +1,41 @@
+package batch
+
+import (
+	"context"
+	"time"
+
+	"golang.org/x/time/rate"
+
+	"github.com/c9s/bbgo/pkg/exchange/binance/binanceapi"
+	"github.com/c9s/bbgo/pkg/types"
+)
+
+type BinanceFuturesIncomeHistoryService interface {
+	QueryFuturesIncomeHistory(ctx context.Context, symbol string, incomeType binanceapi.FuturesIncomeType, startTime, endTime *time.Time) ([]binanceapi.FuturesIncome, error)
+}
+
+type BinanceFuturesIncomeBatchQuery struct {
+	BinanceFuturesIncomeHistoryService
+}
+
+func (e *BinanceFuturesIncomeBatchQuery) Query(ctx context.Context, symbol string, incomeType binanceapi.FuturesIncomeType, startTime, endTime time.Time) (c chan binanceapi.FuturesIncome, errC chan error) {
+	query := &AsyncTimeRangedBatchQuery{
+		Type:        types.MarginInterest{},
+		Limiter:     rate.NewLimiter(rate.Every(3*time.Second), 1),
+		JumpIfEmpty: time.Hour * 24 * 30,
+		Q: func(startTime, endTime time.Time) (interface{}, error) {
+			return e.QueryFuturesIncomeHistory(ctx, symbol, incomeType, &startTime, &endTime)
+		},
+		T: func(obj interface{}) time.Time {
+			return time.Time(obj.(binanceapi.FuturesIncome).Time)
+		},
+		ID: func(obj interface{}) string {
+			interest := obj.(binanceapi.FuturesIncome)
+			return interest.Time.String()
+		},
+	}
+
+	c = make(chan binanceapi.FuturesIncome, 100)
+	errC = query.Query(ctx, c, startTime, endTime)
+	return c, errC
+}

pkg/exchange/batch/margin_interest.go

@@ -10,7 +10,7 @@ import (
 )
 
 type MarginInterestBatchQuery struct {
-	types.MarginHistory
+	types.MarginHistoryService
 }
 
 func (e *MarginInterestBatchQuery) Query(ctx context.Context, asset string, startTime, endTime time.Time) (c chan types.MarginInterest, errC chan error) {

pkg/exchange/batch/margin_liquidation.go

@@ -11,7 +11,7 @@ import (
 )
 
 type MarginLiquidationBatchQuery struct {
-	types.MarginHistory
+	types.MarginHistoryService
 }
 
 func (e *MarginLiquidationBatchQuery) Query(ctx context.Context, startTime, endTime time.Time) (c chan types.MarginLiquidation, errC chan error) {

pkg/exchange/batch/margin_loan.go

@@ -11,7 +11,7 @@ import (
 )
 
 type MarginLoanBatchQuery struct {
-	types.MarginHistory
+	types.MarginHistoryService
 }
 
 func (e *MarginLoanBatchQuery) Query(ctx context.Context, asset string, startTime, endTime time.Time) (c chan types.MarginLoan, errC chan error) {

pkg/exchange/batch/margin_repay.go

@@ -11,7 +11,7 @@ import (
 )
 
 type MarginRepayBatchQuery struct {
-	types.MarginHistory
+	types.MarginHistoryService
 }
 
 func (e *MarginRepayBatchQuery) Query(ctx context.Context, asset string, startTime, endTime time.Time) (c chan types.MarginRepay, errC chan error) {

pkg/exchange/binance/binanceapi/futures_get_income_history_request.go

@@ -35,7 +35,7 @@ type FuturesIncome struct {
 	Asset      string                     `json:"asset"`
 	Info       string                     `json:"info"`
 	Time       types.MillisecondTimestamp `json:"time"`
-	TranId     string                     `json:"tranId"`
+	TranId     int64                      `json:"tranId"`
 	TradeId    string                     `json:"tradeId"`
 }
 

pkg/exchange/binance/exchange.go

@@ -88,6 +88,8 @@ type Exchange struct {
 
 	// client2 is a newer version of the binance api client implemented by ourselves.
 	client2 *binanceapi.RestClient
+
+	futuresClient2 *binanceapi.FuturesRestClient
 }
 
 var timeSetterOnce sync.Once
@@ -111,17 +113,20 @@ func New(key, secret string) *Exchange {
 	}
 
 	client2 := binanceapi.NewClient(client.BaseURL)
+	futuresClient2 := binanceapi.NewFuturesRestClient(futuresClient.BaseURL)
 
 	ex := &Exchange{
-		key:           key,
+		key:            key,
-		secret:        secret,
+		secret:         secret,
-		client:        client,
+		client:         client,
-		futuresClient: futuresClient,
+		futuresClient:  futuresClient,
-		client2:       client2,
+		client2:        client2,
+		futuresClient2: futuresClient2,
 	}
 
 	if len(key) > 0 && len(secret) > 0 {
 		client2.Auth(key, secret)
+		futuresClient2.Auth(key, secret)
 
 		ctx := context.Background()
 		go timeSetterOnce.Do(func() {
@@ -1289,21 +1294,6 @@ func (e *Exchange) DefaultFeeRates() types.ExchangeFee {
 	}
 }
 
-func (e *Exchange) queryFuturesDepth(ctx context.Context, symbol string) (snapshot types.SliceOrderBook, finalUpdateID int64, err error) {
-	res, err := e.futuresClient.NewDepthService().Symbol(symbol).Do(ctx)
-	if err != nil {
-		return snapshot, finalUpdateID, err
-	}
-
-	response := &binance.DepthResponse{
-		LastUpdateID: res.LastUpdateID,
-		Bids:         res.Bids,
-		Asks:         res.Asks,
-	}
-
-	return convertDepth(snapshot, symbol, finalUpdateID, response)
-}
-
 // QueryDepth query the order book depth of a symbol
 func (e *Exchange) QueryDepth(ctx context.Context, symbol string) (snapshot types.SliceOrderBook, finalUpdateID int64, err error) {
 	if e.IsFutures {

pkg/exchange/binance/futures.go

@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"time"
 
+	"github.com/adshao/go-binance/v2"
 	"github.com/adshao/go-binance/v2/futures"
 	"github.com/google/uuid"
 	"go.uber.org/multierr"
@@ -349,3 +350,37 @@ func newFuturesClientOrderID(originalID string) (clientOrderID string) {
 
 	return clientOrderID
 }
+
+func (e *Exchange) queryFuturesDepth(ctx context.Context, symbol string) (snapshot types.SliceOrderBook, finalUpdateID int64, err error) {
+	res, err := e.futuresClient.NewDepthService().Symbol(symbol).Do(ctx)
+	if err != nil {
+		return snapshot, finalUpdateID, err
+	}
+
+	response := &binance.DepthResponse{
+		LastUpdateID: res.LastUpdateID,
+		Bids:         res.Bids,
+		Asks:         res.Asks,
+	}
+
+	return convertDepth(snapshot, symbol, finalUpdateID, response)
+}
+
+// QueryFuturesIncomeHistory queries the income history on the binance futures account
+// This is more binance futures specific API, the convert function is not designed yet.
+// TODO: consider other futures platforms and design the common data structure for this
+func (e *Exchange) QueryFuturesIncomeHistory(ctx context.Context, symbol string, incomeType binanceapi.FuturesIncomeType, startTime, endTime *time.Time) ([]binanceapi.FuturesIncome, error) {
+	req := e.futuresClient2.NewFuturesGetIncomeHistoryRequest()
+	req.Symbol(symbol)
+	req.IncomeType(incomeType)
+	if startTime != nil {
+		req.StartTime(*startTime)
+	}
+
+	if endTime != nil {
+		req.EndTime(*endTime)
+	}
+
+	resp, err := req.Do(ctx)
+	return resp, err
+}

pkg/service/margin.go

@@ -17,7 +17,7 @@ type MarginService struct {
 }
 
 func (s *MarginService) Sync(ctx context.Context, ex types.Exchange, asset string, startTime time.Time) error {
-	api, ok := ex.(types.MarginHistory)
+	api, ok := ex.(types.MarginHistoryService)
 	if !ok {
 		return nil
 	}
@@ -38,7 +38,7 @@ func (s *MarginService) Sync(ctx context.Context, ex types.Exchange, asset strin
 			Type:   types.MarginLoan{},
 			BatchQuery: func(ctx context.Context, startTime, endTime time.Time) (interface{}, chan error) {
 				query := &batch.MarginLoanBatchQuery{
-					MarginHistory: api,
+					MarginHistoryService: api,
 				}
 				return query.Query(ctx, asset, startTime, endTime)
 			},
@@ -55,7 +55,7 @@ func (s *MarginService) Sync(ctx context.Context, ex types.Exchange, asset strin
 			Type:   types.MarginRepay{},
 			BatchQuery: func(ctx context.Context, startTime, endTime time.Time) (interface{}, chan error) {
 				query := &batch.MarginRepayBatchQuery{
-					MarginHistory: api,
+					MarginHistoryService: api,
 				}
 				return query.Query(ctx, asset, startTime, endTime)
 			},
@@ -72,7 +72,7 @@ func (s *MarginService) Sync(ctx context.Context, ex types.Exchange, asset strin
 			Type:   types.MarginInterest{},
 			BatchQuery: func(ctx context.Context, startTime, endTime time.Time) (interface{}, chan error) {
 				query := &batch.MarginInterestBatchQuery{
-					MarginHistory: api,
+					MarginHistoryService: api,
 				}
 				return query.Query(ctx, asset, startTime, endTime)
 			},
@@ -90,7 +90,7 @@ func (s *MarginService) Sync(ctx context.Context, ex types.Exchange, asset strin
 			Type:   types.MarginLiquidation{},
 			BatchQuery: func(ctx context.Context, startTime, endTime time.Time) (interface{}, chan error) {
 				query := &batch.MarginLiquidationBatchQuery{
-					MarginHistory: api,
+					MarginHistoryService: api,
 				}
 				return query.Query(ctx, startTime, endTime)
 			},

pkg/service/sync.go

@@ -49,8 +49,8 @@ func (s *SyncService) SyncSessionSymbols(ctx context.Context, exchange types.Exc
 }
 
 func (s *SyncService) SyncMarginHistory(ctx context.Context, exchange types.Exchange, startTime time.Time, assets ...string) error {
-	if _, implemented := exchange.(types.MarginHistory); !implemented {
+	if _, implemented := exchange.(types.MarginHistoryService); !implemented {
-		log.Debugf("exchange %T does not support types.MarginHistory", exchange)
+		log.Debugf("exchange %T does not support types.MarginHistoryService", exchange)
 		return nil
 	}
 

pkg/strategy/xfunding/profitstats.go

@@ -0,0 +1,44 @@
+package xfunding
+
+import (
+	"fmt"
+	"time"
+
+	"github.com/c9s/bbgo/pkg/fixedpoint"
+	"github.com/c9s/bbgo/pkg/types"
+)
+
+type FundingFee struct {
+	Asset  string           `json:"asset"`
+	Amount fixedpoint.Value `json:"amount"`
+	Txn    int64            `json:"txn"`
+	Time   time.Time        `json:"time"`
+}
+
+type ProfitStats struct {
+	*types.ProfitStats
+
+	FundingFeeCurrency string           `json:"fundingFeeCurrency"`
+	TotalFundingFee    fixedpoint.Value `json:"totalFundingFee"`
+	FundingFeeRecords  []FundingFee     `json:"fundingFeeRecords"`
+	LastFundingFeeTxn  int64            `json:"lastFundingFeeTxn"`
+	LastFundingFeeTime time.Time        `json:"lastFundingFeeTime"`
+}
+
+func (s *ProfitStats) AddFundingFee(fee FundingFee) error {
+	if s.FundingFeeCurrency == "" {
+		s.FundingFeeCurrency = fee.Asset
+	} else if s.FundingFeeCurrency != fee.Asset {
+		return fmt.Errorf("unexpected error, funding fee currency is not matched, given: %s, wanted: %s", fee.Asset, s.FundingFeeCurrency)
+	}
+
+	if s.LastFundingFeeTxn == fee.Txn {
+		return errDuplicatedFundingFeeTxnId
+	}
+
+	s.FundingFeeRecords = append(s.FundingFeeRecords, fee)
+	s.TotalFundingFee = s.TotalFundingFee.Add(fee.Amount)
+	s.LastFundingFeeTxn = fee.Txn
+	s.LastFundingFeeTime = fee.Time
+	return nil
+}

pkg/strategy/xfunding/strategy.go

@@ -4,13 +4,14 @@ import (
 	"context"
 	"errors"
 	"fmt"
-	"strings"
 	"sync"
 	"time"
 
 	"github.com/sirupsen/logrus"
 
+	"github.com/c9s/bbgo/pkg/exchange/batch"
 	"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/util/backoff"
 
@@ -18,6 +19,14 @@ import (
 	"github.com/c9s/bbgo/pkg/types"
 )
 
+// WIP:
+// - track fee token price for cost
+// - buy enough BNB before creating positions
+// - transfer the rest BNB into the futures account
+// - add slack notification support
+// - use neutral position to calculate the position cost
+// - customize profit stats for this funding fee strategy
+
 const ID = "xfunding"
 
 // Position State Transitions:
@@ -34,8 +43,29 @@ const (
 	PositionClosing
 )
 
+type MovingAverageConfig 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"`
+}
+
 var log = logrus.WithField("strategy", ID)
 
+var errNotBinanceExchange = errors.New("not binance exchange, currently only support binance exchange")
+
+var errDuplicatedFundingFeeTxnId = errors.New("duplicated funding fee txn 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)
@@ -54,6 +84,15 @@ type State struct {
 	UsedQuoteInvestment fixedpoint.Value `json:"usedQuoteInvestment"`
 }
 
+func newState() *State {
+	return &State{
+		PositionState:       PositionClosed,
+		PendingBaseTransfer: fixedpoint.Zero,
+		TotalBaseTransfer:   fixedpoint.Zero,
+		UsedQuoteInvestment: fixedpoint.Zero,
+	}
+}
+
 func (s *State) Reset() {
 	s.PositionState = PositionClosed
 	s.PendingBaseTransfer = fixedpoint.Zero
@@ -68,7 +107,9 @@ type Strategy struct {
 	Environment *bbgo.Environment
 
 	// These fields will be filled from the config file (it translates YAML to JSON)
-	Symbol string       `json:"symbol"`
+	Symbol   string         `json:"symbol"`
+	Interval types.Interval `json:"interval"`
+
 	Market types.Market `json:"-"`
 
 	// Leverage is the leverage of the futures position
@@ -89,35 +130,25 @@ type Strategy struct {
 		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"`
+	ProfitStats *ProfitStats `persistence:"profit_stats"`
-	SpotPosition    *types.Position    `persistence:"spot_position"`
+
-	FuturesPosition *types.Position    `persistence:"futures_position"`
+	// SpotPosition is used for the spot position (usually long position)
+	// so that we know how much spot we have bought and the average cost of the spot.
+	SpotPosition *types.Position `persistence:"spot_position"`
+
+	// FuturesPosition is used for the futures position
+	// this position is the reverse side of the spot position, when spot position is long, then the futures position will be short.
+	// but the base quantity should be the same as the spot position
+	FuturesPosition *types.Position `persistence:"futures_position"`
+
+	// NeutralPosition is used for sharing spot/futures position
+	// when creating the spot position and futures position, there will be a spread between the spot position and the futures position.
+	// this neutral position can calculate the spread cost between these two positions
+	NeutralPosition *types.Position `persistence:"neutral_position"`
 
 	State *State `persistence:"state"`
 
@@ -128,6 +159,8 @@ type Strategy struct {
 	spotOrderExecutor, futuresOrderExecutor *bbgo.GeneralOrderExecutor
 	spotMarket, futuresMarket               types.Market
 
+	binanceFutures, binanceSpot *binance.Exchange
+
 	// positionType is the futures position type
 	// currently we only support short position for the positive funding rate
 	positionType types.PositionType
@@ -142,13 +175,8 @@ func (s *Strategy) CrossSubscribe(sessions map[string]*bbgo.ExchangeSession) {
 	spotSession := sessions[s.SpotSession]
 	futuresSession := sessions[s.FuturesSession]
 
-	spotSession.Subscribe(types.KLineChannel, s.Symbol, types.SubscribeOptions{
+	spotSession.Subscribe(types.KLineChannel, s.Symbol, types.SubscribeOptions{Interval: s.Interval})
-		Interval: types.Interval1m,
+	futuresSession.Subscribe(types.KLineChannel, s.Symbol, types.SubscribeOptions{Interval: s.Interval})
-	})
-
-	futuresSession.Subscribe(types.KLineChannel, s.Symbol, types.SubscribeOptions{
-		Interval: types.Interval1m,
-	})
 }
 
 func (s *Strategy) Subscribe(session *bbgo.ExchangeSession) {}
@@ -162,6 +190,10 @@ func (s *Strategy) Defaults() error {
 		s.MinHoldingPeriod = types.Duration(3 * 24 * time.Hour)
 	}
 
+	if s.Interval == "" {
+		s.Interval = types.Interval1m
+	}
+
 	s.positionType = types.PositionShort
 
 	return nil
@@ -192,31 +224,29 @@ func (s *Strategy) InstanceID() string {
 }
 
 func (s *Strategy) Run(ctx context.Context, orderExecutor bbgo.OrderExecutor, session *bbgo.ExchangeSession) error {
-	standardIndicatorSet := session.StandardIndicatorSet(s.Symbol)
+	// standardIndicatorSet := session.StandardIndicatorSet(s.Symbol)
-
+	/*
-	var ma types.Float64Indicator
+		var ma types.Float64Indicator
-	for _, detection := range s.SupportDetection {
+		for _, detection := range s.SupportDetection {
-
+			switch strings.ToLower(detection.MovingAverageType) {
-		switch strings.ToLower(detection.MovingAverageType) {
+			case "sma":
-		case "sma":
+				ma = standardIndicatorSet.SMA(types.IntervalWindow{
-			ma = standardIndicatorSet.SMA(types.IntervalWindow{
+					Interval: detection.MovingAverageIntervalWindow.Interval,
-				Interval: detection.MovingAverageIntervalWindow.Interval,
+					Window:   detection.MovingAverageIntervalWindow.Window,
-				Window:   detection.MovingAverageIntervalWindow.Window,
+				})
-			})
+			case "ema", "ewma":
-		case "ema", "ewma":
+				ma = standardIndicatorSet.EWMA(types.IntervalWindow{
-			ma = standardIndicatorSet.EWMA(types.IntervalWindow{
+					Interval: detection.MovingAverageIntervalWindow.Interval,
-				Interval: detection.MovingAverageIntervalWindow.Interval,
+					Window:   detection.MovingAverageIntervalWindow.Window,
-				Window:   detection.MovingAverageIntervalWindow.Window,
+				})
-			})
+			default:
-		default:
+				ma = standardIndicatorSet.EWMA(types.IntervalWindow{
-			ma = standardIndicatorSet.EWMA(types.IntervalWindow{
+					Interval: detection.MovingAverageIntervalWindow.Interval,
-				Interval: detection.MovingAverageIntervalWindow.Interval,
+					Window:   detection.MovingAverageIntervalWindow.Window,
-				Window:   detection.MovingAverageIntervalWindow.Window,
+				})
-			})
+			}
 		}
-	}
+	*/
-	_ = ma
-
 	return nil
 }
 
@@ -229,6 +259,17 @@ func (s *Strategy) CrossRun(ctx context.Context, orderExecutionRouter bbgo.Order
 	s.spotMarket, _ = s.spotSession.Market(s.Symbol)
 	s.futuresMarket, _ = s.futuresSession.Market(s.Symbol)
 
+	var ok bool
+	s.binanceFutures, ok = s.futuresSession.Exchange.(*binance.Exchange)
+	if !ok {
+		return errNotBinanceExchange
+	}
+
+	s.binanceSpot, ok = s.spotSession.Exchange.(*binance.Exchange)
+	if !ok {
+		return errNotBinanceExchange
+	}
+
 	// adjust QuoteInvestment
 	if b, ok := s.spotSession.Account.Balance(s.spotMarket.QuoteCurrency); ok {
 		originalQuoteInvestment := s.QuoteInvestment
@@ -246,32 +287,42 @@ func (s *Strategy) CrossRun(ctx context.Context, orderExecutionRouter bbgo.Order
 	}
 
 	if s.ProfitStats == nil || s.Reset {
-		s.ProfitStats = types.NewProfitStats(s.Market)
+		s.ProfitStats = &ProfitStats{
-	}
+			ProfitStats: types.NewProfitStats(s.Market),
-
+			// when receiving funding fee, the funding fee asset is the quote currency of that market.
-	if s.FuturesPosition == nil || s.Reset {
+			FundingFeeCurrency: s.futuresMarket.QuoteCurrency,
-		s.FuturesPosition = types.NewPositionFromMarket(s.futuresMarket)
+			TotalFundingFee:    fixedpoint.Zero,
+			FundingFeeRecords:  nil,
+		}
 	}
 
 	if s.SpotPosition == nil || s.Reset {
 		s.SpotPosition = types.NewPositionFromMarket(s.spotMarket)
 	}
 
+	if s.FuturesPosition == nil || s.Reset {
+		s.FuturesPosition = types.NewPositionFromMarket(s.futuresMarket)
+	}
+
+	if s.NeutralPosition == nil || s.Reset {
+		s.NeutralPosition = types.NewPositionFromMarket(s.futuresMarket)
+	}
+
 	if s.State == nil || s.Reset {
-		s.State = &State{
+		s.State = newState()
-			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())
+	log.Infof("loaded neutral position: %s", s.NeutralPosition.String())
 
-	binanceFutures := s.futuresSession.Exchange.(*binance.Exchange)
+	// sync funding fee txns
-	binanceSpot := s.spotSession.Exchange.(*binance.Exchange)
+	if !s.ProfitStats.LastFundingFeeTime.IsZero() {
-	_ = binanceSpot
+		s.syncFundingFeeRecords(ctx, s.ProfitStats.LastFundingFeeTime)
+	}
+
+	// TEST CODE:
+	// s.syncFundingFeeRecords(ctx, time.Now().Add(-3*24*time.Hour))
 
 	s.spotOrderExecutor = s.allocateOrderExecutor(ctx, s.spotSession, instanceID, s.SpotPosition)
 	s.spotOrderExecutor.TradeCollector().OnTrade(func(trade types.Trade, profit fixedpoint.Value, netProfit fixedpoint.Value) {
@@ -299,7 +350,7 @@ func (s *Strategy) CrossRun(ctx context.Context, orderExecutionRouter bbgo.Order
 			// 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)
+				return s.transferIn(ctx, s.binanceSpot, s.spotMarket.BaseCurrency, trade)
 			}); err != nil {
 				log.WithError(err).Errorf("spot-to-futures transfer in retry failed")
 				return
@@ -323,7 +374,7 @@ func (s *Strategy) CrossRun(ctx context.Context, orderExecutionRouter bbgo.Order
 		switch s.getPositionState() {
 		case PositionClosing:
 			if err := backoff.RetryGeneral(ctx, func() error {
-				return s.transferOut(ctx, binanceSpot, s.spotMarket.BaseCurrency, trade)
+				return s.transferOut(ctx, s.binanceSpot, s.spotMarket.BaseCurrency, trade)
 			}); err != nil {
 				log.WithError(err).Errorf("spot-to-futures transfer in retry failed")
 				return
@@ -332,21 +383,50 @@ func (s *Strategy) CrossRun(ctx context.Context, orderExecutionRouter bbgo.Order
 		}
 	})
 
-	s.futuresSession.MarketDataStream.OnKLineClosed(types.KLineWith(s.Symbol, types.Interval1m, func(kline types.KLine) {
+	s.futuresSession.MarketDataStream.OnKLineClosed(types.KLineWith(s.Symbol, s.Interval, func(kline types.KLine) {
-		s.queryAndDetectPremiumIndex(ctx, binanceFutures)
+		s.queryAndDetectPremiumIndex(ctx, s.binanceFutures)
 	}))
 
 	if binanceStream, ok := s.futuresSession.UserDataStream.(*binance.Stream); ok {
 		binanceStream.OnAccountUpdateEvent(func(e *binance.AccountUpdateEvent) {
-			log.Infof("onAccountUpdateEvent: %+v", e)
 			switch e.AccountUpdate.EventReasonType {
-
 			case binance.AccountUpdateEventReasonDeposit:
-
 			case binance.AccountUpdateEventReasonWithdraw:
-
 			case binance.AccountUpdateEventReasonFundingFee:
+				//  EventBase:{
+				// 		Event:ACCOUNT_UPDATE
+				// 		Time:1679760000932
+				// 	}
+				// 	Transaction:1679760000927
+				// 	AccountUpdate:{
+				// 			EventReasonType:FUNDING_FEE
+				// 			Balances:[{
+				// 					Asset:USDT
+				// 					WalletBalance:56.64251742
+				// 					CrossWalletBalance:56.64251742
+				// 					BalanceChange:-0.00037648
+				// 			}]
+				// 		}
+				// 	}
+				for _, b := range e.AccountUpdate.Balances {
+					if b.Asset != s.ProfitStats.FundingFeeCurrency {
+						continue
+					}
 
+					txnTime := time.UnixMilli(e.Time)
+					err := s.ProfitStats.AddFundingFee(FundingFee{
+						Asset:  b.Asset,
+						Amount: b.BalanceChange,
+						Txn:    e.Transaction,
+						Time:   txnTime,
+					})
+					if err != nil {
+						log.WithError(err).Error("unable to add funding fee to profitStats")
+					}
+				}
+
+				log.Infof("total collected funding fee: %f %s", s.ProfitStats.TotalFundingFee.Float64(), s.ProfitStats.FundingFeeCurrency)
+				bbgo.Sync(ctx, s)
 			}
 		})
 	}
@@ -384,6 +464,54 @@ func (s *Strategy) CrossRun(ctx context.Context, orderExecutionRouter bbgo.Order
 	return nil
 }
 
+func (s *Strategy) syncFundingFeeRecords(ctx context.Context, since time.Time) {
+	now := time.Now()
+
+	log.Infof("syncing funding fee records from the income history query: %s <=> %s", since, now)
+
+	defer log.Infof("sync funding fee records done")
+
+	q := batch.BinanceFuturesIncomeBatchQuery{
+		BinanceFuturesIncomeHistoryService: s.binanceFutures,
+	}
+
+	dataC, errC := q.Query(ctx, s.Symbol, binanceapi.FuturesIncomeFundingFee, since, now)
+	for {
+		select {
+		case <-ctx.Done():
+			return
+
+		case income, ok := <-dataC:
+			if !ok {
+				return
+			}
+
+			log.Infof("income: %+v", income)
+			switch income.IncomeType {
+			case binanceapi.FuturesIncomeFundingFee:
+				err := s.ProfitStats.AddFundingFee(FundingFee{
+					Asset:  income.Asset,
+					Amount: income.Income,
+					Txn:    income.TranId,
+					Time:   income.Time.Time(),
+				})
+				if err != nil {
+					log.WithError(err).Errorf("can not add funding fee record to ProfitStats")
+				}
+			}
+
+		case err, ok := <-errC:
+			if !ok {
+				return
+			}
+
+			log.WithError(err).Errorf("unable to query futures income history")
+			return
+
+		}
+	}
+}
+
 func (s *Strategy) queryAndDetectPremiumIndex(ctx context.Context, binanceFutures *binance.Exchange) {
 	premiumIndex, err := binanceFutures.QueryPremiumIndex(ctx, s.Symbol)
 	if err != nil {
@@ -810,11 +938,16 @@ func (s *Strategy) allocateOrderExecutor(ctx context.Context, session *bbgo.Exch
 	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)
 	})
+	orderExecutor.TradeCollector().OnTrade(func(trade types.Trade, _ fixedpoint.Value, _ fixedpoint.Value) {
+		s.ProfitStats.AddTrade(trade)
+
+		if profit, netProfit, madeProfit := s.NeutralPosition.AddTrade(trade); madeProfit {
+			p := s.NeutralPosition.NewProfit(trade, profit, netProfit)
+			s.ProfitStats.AddProfit(p)
+		}
+	})
 	return orderExecutor
 }

pkg/types/margin.go

@@ -105,8 +105,8 @@ type MarginLiquidation struct {
 	UpdatedTime      Time             `json:"updatedTime" db:"time"`
 }
 
-// MarginHistory provides the service of querying loan history and repay history
+// MarginHistoryService provides the service of querying loan history and repay history
-type MarginHistory interface {
+type MarginHistoryService interface {
 	QueryLoanHistory(ctx context.Context, asset string, startTime, endTime *time.Time) ([]MarginLoan, error)
 	QueryRepayHistory(ctx context.Context, asset string, startTime, endTime *time.Time) ([]MarginRepay, error)
 	QueryLiquidationHistory(ctx context.Context, startTime, endTime *time.Time) ([]MarginLiquidation, error)