okex: parse and convert account information

examples/okex-book/main.go

@@ -62,14 +62,13 @@ var rootCmd = &cobra.Command{
 		log.Infof("funding rate: %+v", fundingRate)
 
 		log.Infof("ACCOUNT BALANCES:")
-		balanceSummaries, err := client.AccountBalances()
+		account, err := client.AccountBalances()
 		if err != nil {
 			return err
 		}
 
-		for _, balanceSummary := range balanceSummaries {
-			log.Infof("%+v", balanceSummary)
-		}
+		log.Infof("%+v", account)
+
 
 		log.Infof("ASSET BALANCES:")
 		assetBalances, err := client.AssetBalances()

pkg/exchange/okex/convert.go

@@ -36,23 +36,21 @@ func toGlobalTicker(marketTicker okexapi.MarketTicker) *types.Ticker {
 	}
 }
 
-func toGlobalBalance(balanceSummaries []okexapi.BalanceSummary) types.BalanceMap {
+func toGlobalBalance(account *okexapi.Account) types.BalanceMap {
 	var balanceMap = types.BalanceMap{}
-	for _, balanceSummary := range balanceSummaries {
-		for _, balanceDetail := range balanceSummary.Details {
-			balanceMap[balanceDetail.Currency] = types.Balance{
-				Currency:  balanceDetail.Currency,
-				Available: balanceDetail.CashBalance,
-				Locked:    balanceDetail.Frozen,
-			}
+	for _, balanceDetail := range account.Details {
+		balanceMap[balanceDetail.Currency] = types.Balance{
+			Currency:  balanceDetail.Currency,
+			Available: balanceDetail.CashBalance,
+			Locked:    balanceDetail.Frozen,
 		}
 	}
 	return balanceMap
 }
 
 type WebsocketSubscription struct {
-	Channel      string `json:"channel"`
-	InstrumentID string `json:"instId,omitempty"`
+	Channel        string `json:"channel"`
+	InstrumentID   string `json:"instId,omitempty"`
 	InstrumentType string `json:"instType,omitempty"`
 }
 

pkg/exchange/okex/exchange.go

@@ -126,7 +126,7 @@ func (e *Exchange) PlatformFeeCurrency() string {
 }
 
 func (e *Exchange) QueryAccount(ctx context.Context) (*types.Account, error) {
-	balanceSummaries, err := e.client.AccountBalances()
+	accountBalance, err := e.client.AccountBalances()
 	if err != nil {
 		return nil, err
 	}
@@ -135,7 +135,7 @@ func (e *Exchange) QueryAccount(ctx context.Context) (*types.Account, error) {
 		AccountType: "SPOT",
 	}
 
-	var balanceMap = toGlobalBalance(balanceSummaries)
+	var balanceMap = toGlobalBalance(accountBalance)
 	account.UpdateBalances(balanceMap)
 	return &account, nil
 }

pkg/exchange/okex/okexapi/client.go

@@ -205,15 +205,13 @@ type BalanceDetail struct {
 	UnrealizedProfitAndLoss fixedpoint.Value           `json:"upl"`
 }
 
-type BalanceSummary struct {
+type Account struct {
 	TotalEquityInUSD fixedpoint.Value `json:"totalEq"`
 	UpdateTime       string           `json:"uTime"`
 	Details          []BalanceDetail  `json:"details"`
 }
 
-type BalanceSummaryList []BalanceSummary
-
-func (c *RestClient) AccountBalances() (BalanceSummaryList, error) {
+func (c *RestClient) AccountBalances() (*Account, error) {
 	req, err := c.newAuthenticatedRequest("GET", "/api/v5/account/balance", nil, nil)
 	if err != nil {
 		return nil, err
@@ -225,15 +223,20 @@ func (c *RestClient) AccountBalances() (BalanceSummaryList, error) {
 	}
 
 	var balanceResponse struct {
-		Code    string           `json:"code"`
-		Message string           `json:"msg"`
-		Data    []BalanceSummary `json:"data"`
+		Code    string    `json:"code"`
+		Message string    `json:"msg"`
+		Data    []Account `json:"data"`
 	}
+
 	if err := response.DecodeJSON(&balanceResponse); err != nil {
 		return nil, err
 	}
 
-	return balanceResponse.Data, nil
+	if len(balanceResponse.Data) == 0 {
+		return nil, errors.New("empty account data")
+	}
+
+	return &balanceResponse.Data[0], nil
 }
 
 type AssetBalance struct {

pkg/exchange/okex/parse.go

@@ -1,12 +1,14 @@
 package okex
 
 import (
+	"encoding/json"
 	"errors"
 	"fmt"
 	"strconv"
 	"strings"
 	"time"
 
+	"github.com/c9s/bbgo/pkg/exchange/okex/okexapi"
 	"github.com/c9s/bbgo/pkg/fixedpoint"
 	"github.com/c9s/bbgo/pkg/types"
 	"github.com/valyala/fastjson"
@@ -113,7 +115,8 @@ func parseBookEntry(v *fastjson.Value) (*BookEntry, error) {
 	}, nil
 }
 
-func parseBookData(instrumentId string, v *fastjson.Value) (*BookData, error) {
+func parseBookData(v *fastjson.Value) (*BookData, error) {
+	instrumentId := string(v.GetStringBytes("arg", "instId"))
 	data := v.GetArray("data")
 	if len(data) == 0 {
 		return nil, errors.New("empty data payload")
@@ -198,7 +201,8 @@ func (c *Candle) KLine() types.KLine {
 	}
 }
 
-func parseCandle(channel, instrumentID string, v *fastjson.Value) (*Candle, error) {
+func parseCandle(channel string, v *fastjson.Value) (*Candle, error) {
+	instrumentID := string(v.GetStringBytes("arg", "instId"))
 	data, err := v.Get("data").Array()
 	if err != nil {
 		return nil, err
@@ -271,17 +275,36 @@ func parseCandle(channel, instrumentID string, v *fastjson.Value) (*Candle, erro
 	}, nil
 }
 
+func parseAccount(v *fastjson.Value) (*okexapi.Account, error) {
+	data := v.Get("data").MarshalTo(nil)
+
+	var accounts []okexapi.Account
+	err := json.Unmarshal(data, &accounts)
+	if err != nil {
+		return nil, err
+	}
+
+	if len(accounts) == 0 {
+		return nil, errors.New("empty account data")
+	}
+
+	return &accounts[0], nil
+}
+
 func parseData(v *fastjson.Value) (interface{}, error) {
-	instrumentId := string(v.GetStringBytes("arg", "instId"))
+
 	channel := string(v.GetStringBytes("arg", "channel"))
 
 	switch channel {
 	case "books":
-		return parseBookData(instrumentId, v)
+		return parseBookData(v)
+
+	case "account":
+		return parseAccount(v)
 
 	default:
 		if strings.HasPrefix(channel, "candle") {
-			return parseCandle(channel, instrumentId, v)
+			return parseCandle(channel, v)
 		}
 
 	}

pkg/exchange/okex/stream.go

@@ -40,6 +40,7 @@ type Stream struct {
 	candleDataCallbacks []func(candle Candle)
 	bookDataCallbacks   []func(book BookData)
 	eventCallbacks      []func(event WebSocketEvent)
+	accountCallbacks    []func(account okexapi.Account)
 
 	lastCandle map[CandleKey]Candle
 }
@@ -84,6 +85,11 @@ func NewStream(client *okexapi.RestClient) *Stream {
 		}
 	})
 
+	stream.OnAccount(func(account okexapi.Account) {
+		balances := toGlobalBalance(&account)
+		stream.EmitBalanceSnapshot(balances)
+	})
+
 	stream.OnEvent(func(event WebSocketEvent) {
 		log.Infof("event: %+v", event)
 		switch event.Event {
@@ -313,6 +319,10 @@ func (s *Stream) read(ctx context.Context) {
 
 				case *Candle:
 					s.EmitCandleData(*et)
+
+				case *okexapi.Account:
+					s.EmitAccount(*et)
+
 				}
 			}
 		}

pkg/exchange/okex/stream_callbacks.go

@@ -2,7 +2,9 @@
 
 package okex
 
-import ()
+import (
+	"github.com/c9s/bbgo/pkg/exchange/okex/okexapi"
+)
 
 func (s *Stream) OnCandleData(cb func(candle Candle)) {
 	s.candleDataCallbacks = append(s.candleDataCallbacks, cb)
@@ -34,10 +36,22 @@ func (s *Stream) EmitEvent(event WebSocketEvent) {
 	}
 }
 
+func (s *Stream) OnAccount(cb func(account okexapi.Account)) {
+	s.accountCallbacks = append(s.accountCallbacks, cb)
+}
+
+func (s *Stream) EmitAccount(account okexapi.Account) {
+	for _, cb := range s.accountCallbacks {
+		cb(account)
+	}
+}
+
 type StreamEventHub interface {
 	OnCandleData(cb func(candle Candle))
 
 	OnBookData(cb func(book BookData))
 
 	OnEvent(cb func(event WebSocketEvent))
+
+	OnAccount(cb func(account okexapi.Account))
 }