Merge pull request #1528 from c9s/c9s/rewrite-binance-depth-requestgen

REFACTOR: [binance] rewrite binance depth requestgen

pkg/exchange/binance/binanceapi/client_test.go

@@ -244,3 +244,19 @@ func TestClient_NewPlaceMarginOrderRequest(t *testing.T) {
 	assert.NotEmpty(t, res)
 	t.Logf("result: %+v", res)
 }
+
+func TestClient_GetDepth(t *testing.T) {
+	client := getTestClientOrSkip(t)
+	ctx := context.Background()
+
+	err := client.SetTimeOffsetFromServer(ctx)
+	assert.NoError(t, err)
+
+	req := client.NewGetDepthRequest().Symbol("BTCUSDT").Limit(1000)
+	resp, err := req.Do(ctx)
+	if assert.NoError(t, err) {
+		assert.NotNil(t, resp)
+		assert.NotEmpty(t, resp)
+		t.Logf("response: %+v", resp)
+	}
+}

pkg/exchange/binance/binanceapi/get_depth_request.go

@@ -0,0 +1,25 @@
+package binanceapi
+
+import (
+	"github.com/c9s/requestgen"
+
+	"github.com/c9s/bbgo/pkg/fixedpoint"
+)
+
+type Depth struct {
+	LastUpdateId int64                `json:"lastUpdateId"`
+	Bids         [][]fixedpoint.Value `json:"bids"`
+	Asks         [][]fixedpoint.Value `json:"asks"`
+}
+
+//go:generate requestgen -method GET -url "/api/v3/depth" -type GetDepthRequest -responseType .Depth
+type GetDepthRequest struct {
+	client requestgen.APIClient
+
+	symbol string `param:"symbol"`
+	limit  int    `param:"limit" defaultValue:"1000"`
+}
+
+func (c *RestClient) NewGetDepthRequest() *GetDepthRequest {
+	return &GetDepthRequest{client: c}
+}

pkg/exchange/binance/binanceapi/get_depth_request_requestgen.go

@@ -0,0 +1,190 @@
+// Code generated by "requestgen -method GET -url /api/v3/depth -type GetDepthRequest -responseType .Depth"; DO NOT EDIT.
+
+package binanceapi
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"net/url"
+	"reflect"
+	"regexp"
+)
+
+func (g *GetDepthRequest) Symbol(symbol string) *GetDepthRequest {
+	g.symbol = symbol
+	return g
+}
+
+func (g *GetDepthRequest) Limit(limit int) *GetDepthRequest {
+	g.limit = limit
+	return g
+}
+
+// GetQueryParameters builds and checks the query parameters and returns url.Values
+func (g *GetDepthRequest) 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 (g *GetDepthRequest) GetParameters() (map[string]interface{}, error) {
+	var params = map[string]interface{}{}
+	// check symbol field -> json key symbol
+	symbol := g.symbol
+
+	// assign parameter of symbol
+	params["symbol"] = symbol
+	// check limit field -> json key limit
+	limit := g.limit
+
+	// assign parameter of limit
+	params["limit"] = limit
+
+	return params, nil
+}
+
+// GetParametersQuery converts the parameters from GetParameters into the url.Values format
+func (g *GetDepthRequest) GetParametersQuery() (url.Values, error) {
+	query := url.Values{}
+
+	params, err := g.GetParameters()
+	if err != nil {
+		return query, err
+	}
+
+	for _k, _v := range params {
+		if g.isVarSlice(_v) {
+			g.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 (g *GetDepthRequest) GetParametersJSON() ([]byte, error) {
+	params, err := g.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 (g *GetDepthRequest) GetSlugParameters() (map[string]interface{}, error) {
+	var params = map[string]interface{}{}
+
+	return params, nil
+}
+
+func (g *GetDepthRequest) 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 (g *GetDepthRequest) 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 (g *GetDepthRequest) isVarSlice(_v interface{}) bool {
+	rt := reflect.TypeOf(_v)
+	switch rt.Kind() {
+	case reflect.Slice:
+		return true
+	}
+	return false
+}
+
+func (g *GetDepthRequest) GetSlugsMap() (map[string]string, error) {
+	slugs := map[string]string{}
+	params, err := g.GetSlugParameters()
+	if err != nil {
+		return slugs, nil
+	}
+
+	for _k, _v := range params {
+		slugs[_k] = fmt.Sprintf("%v", _v)
+	}
+
+	return slugs, nil
+}
+
+// GetPath returns the request path of the API
+func (g *GetDepthRequest) GetPath() string {
+	return "/api/v3/depth"
+}
+
+// Do generates the request object and send the request object to the API endpoint
+func (g *GetDepthRequest) Do(ctx context.Context) (*Depth, error) {
+
+	// empty params for GET operation
+	var params interface{}
+	query, err := g.GetParametersQuery()
+	if err != nil {
+		return nil, err
+	}
+
+	var apiURL string
+
+	apiURL = g.GetPath()
+
+	req, err := g.client.NewRequest(ctx, "GET", apiURL, query, params)
+	if err != nil {
+		return nil, err
+	}
+
+	response, err := g.client.SendRequest(req)
+	if err != nil {
+		return nil, err
+	}
+
+	var apiResponse Depth
+
+	type responseUnmarshaler interface {
+		Unmarshal(data []byte) error
+	}
+
+	if unmarshaler, ok := interface{}(&apiResponse).(responseUnmarshaler); ok {
+		if err := unmarshaler.Unmarshal(response.Body); err != nil {
+			return nil, err
+		}
+	} else {
+		// The line below checks the content type, however, some API server might not send the correct content type header,
+		// Hence, this is commented for backward compatibility
+		// response.IsJSON()
+		if err := response.DecodeJSON(&apiResponse); err != nil {
+			return nil, err
+		}
+	}
+
+	type responseValidator interface {
+		Validate() error
+	}
+
+	if validator, ok := interface{}(&apiResponse).(responseValidator); ok {
+		if err := validator.Validate(); err != nil {
+			return nil, err
+		}
+	}
+	return &apiResponse, nil
+}

pkg/exchange/binance/exchange.go

@@ -1346,15 +1346,32 @@ func (e *Exchange) QueryDepth(ctx context.Context, symbol string) (snapshot type
 		return e.queryFuturesDepth(ctx, symbol)
 	}
 
-	response, err := e.client.NewDepthService().Symbol(symbol).Limit(DefaultDepthLimit).Do(ctx)
+	response, err := e.client2.NewGetDepthRequest().Symbol(symbol).Limit(DefaultDepthLimit).Do(ctx)
 	if err != nil {
 		return snapshot, finalUpdateID, err
 	}
 
-	return convertDepth(snapshot, symbol, finalUpdateID, response)
+	return convertDepth(symbol, response)
 }
 
-func convertDepth(
+func convertDepth(symbol string, response *binanceapi.Depth) (snapshot types.SliceOrderBook, finalUpdateID int64, err error) {
+	snapshot.Symbol = symbol
+	snapshot.Time = time.Now()
+	snapshot.LastUpdateId = response.LastUpdateId
+
+	finalUpdateID = response.LastUpdateId
+	for _, entry := range response.Bids {
+		snapshot.Bids = append(snapshot.Bids, types.PriceVolume{Price: entry[0], Volume: entry[1]})
+	}
+
+	for _, entry := range response.Asks {
+		snapshot.Asks = append(snapshot.Asks, types.PriceVolume{Price: entry[0], Volume: entry[1]})
+	}
+
+	return snapshot, finalUpdateID, err
+}
+
+func convertDepthLegacy(
 	snapshot types.SliceOrderBook, symbol string, finalUpdateID int64, response *binance.DepthResponse,
 ) (types.SliceOrderBook, int64, error) {
 	snapshot.Symbol = symbol

pkg/exchange/binance/futures.go

@@ -15,7 +15,9 @@ import (
 	"github.com/c9s/bbgo/pkg/types"
 )
 
-func (e *Exchange) queryFuturesClosedOrders(ctx context.Context, symbol string, since, until time.Time, lastOrderID uint64) (orders []types.Order, err error) {
+func (e *Exchange) queryFuturesClosedOrders(
+	ctx context.Context, symbol string, since, until time.Time, lastOrderID uint64,
+) (orders []types.Order, err error) {
 	req := e.futuresClient.NewListOrdersService().Symbol(symbol)
 
 	if lastOrderID > 0 {
@@ -34,7 +36,9 @@ func (e *Exchange) queryFuturesClosedOrders(ctx context.Context, symbol string,
 	return toGlobalFuturesOrders(binanceOrders, false)
 }
 
-func (e *Exchange) TransferFuturesAccountAsset(ctx context.Context, asset string, amount fixedpoint.Value, io types.TransferDirection) error {
+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())
@@ -63,7 +67,7 @@ func (e *Exchange) TransferFuturesAccountAsset(ctx context.Context, asset string
 // Balance.Available = Wallet Balance(in Binance UI) - Used Margin
 // Balance.Locked = Used Margin
 func (e *Exchange) QueryFuturesAccount(ctx context.Context) (*types.Account, error) {
-	//account, err := e.futuresClient.NewGetAccountService().Do(ctx)
+	// account, err := e.futuresClient.NewGetAccountService().Do(ctx)
 	reqAccount := e.futuresClient2.NewFuturesGetAccountRequest()
 	account, err := reqAccount.Do(ctx)
 	if err != nil {
@@ -218,7 +222,9 @@ func (e *Exchange) submitFuturesOrder(ctx context.Context, order types.SubmitOrd
 	return createdOrder, err
 }
 
-func (e *Exchange) QueryFuturesKLines(ctx context.Context, symbol string, interval types.Interval, options types.KLineQueryOptions) ([]types.KLine, error) {
+func (e *Exchange) QueryFuturesKLines(
+	ctx context.Context, symbol string, interval types.Interval, options types.KLineQueryOptions,
+) ([]types.KLine, error) {
 
 	var limit = 1000
 	if options.Limit > 0 {
@@ -272,7 +278,9 @@ func (e *Exchange) QueryFuturesKLines(ctx context.Context, symbol string, interv
 	return kLines, nil
 }
 
-func (e *Exchange) queryFuturesTrades(ctx context.Context, symbol string, options *types.TradeQueryOptions) (trades []types.Trade, err error) {
+func (e *Exchange) queryFuturesTrades(
+	ctx context.Context, symbol string, options *types.TradeQueryOptions,
+) (trades []types.Trade, err error) {
 
 	var remoteTrades []*futures.AccountTrade
 	req := e.futuresClient.NewListAccountTradeService().
@@ -376,7 +384,7 @@ func (e *Exchange) queryFuturesDepth(ctx context.Context, symbol string) (snapsh
 		Asks:         res.Asks,
 	}
 
-	return convertDepth(snapshot, symbol, finalUpdateID, response)
+	return convertDepthLegacy(snapshot, symbol, finalUpdateID, response)
 }
 
 func (e *Exchange) GetFuturesClient() *binanceapi.FuturesRestClient {
@@ -386,7 +394,9 @@ func (e *Exchange) GetFuturesClient() *binanceapi.FuturesRestClient {
 // 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) {
+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)

pkg/exchange/binance/parse.go

@@ -12,10 +12,32 @@ import (
 	"github.com/adshao/go-binance/v2"
 	"github.com/valyala/fastjson"
 
+	"github.com/c9s/bbgo/pkg/exchange/binance/binanceapi"
 	"github.com/c9s/bbgo/pkg/fixedpoint"
 	"github.com/c9s/bbgo/pkg/types"
 )
 
+type EventType = string
+
+const (
+	EventTypeKLine                   EventType = "kline"
+	EventTypeOutboundAccountPosition EventType = "outboundAccountPosition"
+	EventTypeOutboundAccountInfo     EventType = "outboundAccountInfo"
+	EventTypeBalanceUpdate           EventType = "balanceUpdate"
+	EventTypeExecutionReport         EventType = "executionReport"
+	EventTypeDepthUpdate             EventType = "depthUpdate"
+	EventTypeListenKeyExpired        EventType = "listenKeyExpired"
+	EventTypeTrade                   EventType = "trade"
+	EventTypeAggTrade                EventType = "aggTrade"
+	EventTypeForceOrder              EventType = "forceOrder"
+
+	// Our side defines the following event types since binance doesn't
+	// define the event name from the server messages.
+	//
+	EventTypeBookTicker   EventType = "bookTicker"
+	EventTypePartialDepth EventType = "partialDepth"
+)
+
 type EventBase struct {
 	Event string                     `json:"e"` // event name
 	Time  types.MillisecondTimestamp `json:"E"` // event time
@@ -305,72 +327,84 @@ func parseWebSocketEvent(message []byte) (interface{}, error) {
 		return nil, err
 	}
 
-	// res, err := json.MarshalIndent(message, "", "  ")
-	// if err != nil {
-	//	log.Fatal(err)
-	// }
-	// str := strings.ReplaceAll(string(res), "\\", "")
-	// fmt.Println(str)
 	eventType := string(val.GetStringBytes("e"))
-	if eventType == "" && IsBookTicker(val) {
-		eventType = "bookTicker"
+	if eventType == "" {
+		if isBookTicker(val) {
+			eventType = EventTypeBookTicker
+		} else if isPartialDepth(val) {
+			eventType = EventTypePartialDepth
+		}
 	}
 
 	switch eventType {
-	case "kline":
-		var event KLineEvent
-		err := json.Unmarshal([]byte(message), &event)
+
+	case EventTypeOutboundAccountPosition:
+		var event OutboundAccountPositionEvent
+		err = json.Unmarshal(message, &event)
 		return &event, err
-	case "bookTicker":
+
+	case EventTypeOutboundAccountInfo:
+		var event OutboundAccountInfoEvent
+		err = json.Unmarshal(message, &event)
+		return &event, err
+
+	case EventTypeBalanceUpdate:
+		var event BalanceUpdateEvent
+		err = json.Unmarshal(message, &event)
+		return &event, err
+
+	case EventTypeExecutionReport:
+		var event ExecutionReportEvent
+		err = json.Unmarshal(message, &event)
+		return &event, err
+
+	case EventTypeDepthUpdate:
+		return parseDepthEvent(val)
+
+	case EventTypeTrade:
+		var event MarketTradeEvent
+		err = json.Unmarshal(message, &event)
+		return &event, err
+
+	case EventTypeBookTicker:
 		var event BookTickerEvent
-		err := json.Unmarshal([]byte(message), &event)
+		err := json.Unmarshal(message, &event)
 		event.Event = eventType
 		return &event, err
 
-	case "outboundAccountPosition":
-		var event OutboundAccountPositionEvent
-		err = json.Unmarshal([]byte(message), &event)
+	case EventTypePartialDepth:
+		var depth binanceapi.Depth
+		err := json.Unmarshal(message, &depth)
+		return &PartialDepthEvent{
+			EventBase: EventBase{
+				Event: EventTypePartialDepth,
+				Time:  types.MillisecondTimestamp(time.Now()),
+			},
+			Depth: depth,
+		}, err
+
+	case EventTypeKLine:
+		var event KLineEvent
+		err := json.Unmarshal(message, &event)
 		return &event, err
 
-	case "outboundAccountInfo":
-		var event OutboundAccountInfoEvent
-		err = json.Unmarshal([]byte(message), &event)
-		return &event, err
-
-	case "balanceUpdate":
-		var event BalanceUpdateEvent
-		err = json.Unmarshal([]byte(message), &event)
-		return &event, err
-
-	case "executionReport":
-		var event ExecutionReportEvent
-		err = json.Unmarshal([]byte(message), &event)
-		return &event, err
-
-	case "depthUpdate":
-		return parseDepthEvent(val)
-
-	case "listenKeyExpired":
+	case EventTypeListenKeyExpired:
 		var event ListenKeyExpired
-		err = json.Unmarshal([]byte(message), &event)
+		err = json.Unmarshal(message, &event)
 		return &event, err
 
-	case "trade":
-		var event MarketTradeEvent
-		err = json.Unmarshal([]byte(message), &event)
-		return &event, err
-
-	case "aggTrade":
+	case EventTypeAggTrade:
 		var event AggTradeEvent
-		err = json.Unmarshal([]byte(message), &event)
+		err = json.Unmarshal(message, &event)
 		return &event, err
-	case "forceOrder":
+
+	case EventTypeForceOrder:
 		var event ForceOrderEvent
-		err = json.Unmarshal([]byte(message), &event)
+		err = json.Unmarshal(message, &event)
 		return &event, err
 	}
 
-	// futures stream
+	// events for futures
 	switch eventType {
 
 	// futures market data stream
@@ -419,12 +453,17 @@ func parseWebSocketEvent(message []byte) (interface{}, error) {
 	return nil, fmt.Errorf("unsupported binance websocket message: %s", message)
 }
 
-// IsBookTicker document ref :https://binance-docs.github.io/apidocs/spot/en/#individual-symbol-book-ticker-streams
-// use key recognition because there's no identify in the content.
-func IsBookTicker(val *fastjson.Value) bool {
-	return !val.Exists("e") && val.Exists("u") &&
-		val.Exists("s") && val.Exists("b") &&
-		val.Exists("B") && val.Exists("a") && val.Exists("A")
+// isBookTicker document ref :https://binance-docs.github.io/apidocs/spot/en/#individual-symbol-book-ticker-streams
+// use key recognition because there's no identification in the content.
+func isBookTicker(val *fastjson.Value) bool {
+	return val.Exists("u") && val.Exists("s") &&
+		val.Exists("b") && val.Exists("B") &&
+		val.Exists("a") && val.Exists("A")
+}
+
+func isPartialDepth(val *fastjson.Value) bool {
+	return val.Exists("lastUpdateId") &&
+		val.Exists("bids") && val.Exists("bids")
 }
 
 type DepthEntry struct {
@@ -1090,19 +1129,47 @@ type AccountConfigUpdateEvent struct {
 	} `json:"ai"`
 }
 
+/*
+	{
+	  "lastUpdateId": 160,  // Last update ID
+	  "bids": [             // Bids to be updated
+	    [
+	      "0.0024",         // Price level to be updated
+	      "10"              // Quantity
+	    ]
+	  ],
+	  "asks": [             // Asks to be updated
+	    [
+	      "0.0026",         // Price level to be updated
+	      "100"             // Quantity
+	    ]
+	  ]
+	}
+*/
+type PartialDepthEvent struct {
+	EventBase
+
+	binanceapi.Depth
+}
+
+/*
+	{
+	  "u":400900217,     // order book updateId
+	  "s":"BNBUSDT",     // symbol
+	  "b":"25.35190000", // best bid price
+	  "B":"31.21000000", // best bid qty
+	  "a":"25.36520000", // best ask price
+	  "A":"40.66000000"  // best ask qty
+	}
+*/
 type BookTickerEvent struct {
 	EventBase
+	UpdateID int64            `json:"u"`
 	Symbol   string           `json:"s"`
 	Buy      fixedpoint.Value `json:"b"`
 	BuySize  fixedpoint.Value `json:"B"`
 	Sell     fixedpoint.Value `json:"a"`
 	SellSize fixedpoint.Value `json:"A"`
-	// "u":400900217,     // order book updateId
-	// "s":"BNBUSDT",     // symbol
-	// "b":"25.35190000", // best bid price
-	// "B":"31.21000000", // best bid qty
-	// "a":"25.36520000", // best ask price
-	// "A":"40.66000000"  // best ask qty
 }
 
 func (k *BookTickerEvent) BookTicker() types.BookTicker {

pkg/types/sliceorderbook.go

@@ -21,6 +21,11 @@ type SliceOrderBook struct {
 	// Time represents the server time. If empty, it indicates that the server does not provide this information.
 	Time time.Time
 
+	// LastUpdateId is the message id from the server
+	// this field is optional, not every exchange provides this information
+	// this is for binance right now.
+	LastUpdateId int64
+
 	lastUpdateTime time.Time
 
 	loadCallbacks   []func(book *SliceOrderBook)
@@ -136,7 +141,7 @@ func (b *SliceOrderBook) updateBids(pvs PriceVolumeSlice) {
 func (b *SliceOrderBook) update(book SliceOrderBook) {
 	b.updateBids(book.Bids)
 	b.updateAsks(book.Asks)
-	b.lastUpdateTime = time.Now()
+	b.lastUpdateTime = defaultTime(book.Time, time.Now)
 }
 
 func (b *SliceOrderBook) Reset() {