Merge pull request #1299 from bailantaotao/edwin/add-server-time

pkg/exchange: add time to SliceOrderBook

pkg/cache/cache_test.go

@@ -100,4 +100,4 @@ func Test_loadMarketsFromMem(t *testing.T) {
 	}
 
 	globalMarketMemCache = newMarketMemCache() // reset the global cache
-}
+}

pkg/cmd/orderbook.go

@@ -71,9 +71,10 @@ var orderbookCmd = &cobra.Command{
 			}
 
 			if bid, ask, ok := orderBook.BestBidAndAsk(); ok {
-				log.Infof("ASK | %f x %f / %f x %f | BID",
+				log.Infof("ASK | %f x %f / %f x %f | BID | %s",
 					ask.Volume.Float64(), ask.Price.Float64(),
-					bid.Price.Float64(), bid.Volume.Float64())
+					bid.Price.Float64(), bid.Volume.Float64(),
+					book.Time.String())
 			}
 		})
 
@@ -84,9 +85,10 @@ var orderbookCmd = &cobra.Command{
 			orderBook.Update(book)
 
 			if bid, ask, ok := orderBook.BestBidAndAsk(); ok {
-				log.Infof("ASK | %f x %f / %f x %f | BID",
+				log.Infof("ASK | %f x %f / %f x %f | BID | %s",
 					ask.Volume.Float64(), ask.Price.Float64(),
-					bid.Price.Float64(), bid.Volume.Float64())
+					bid.Price.Float64(), bid.Volume.Float64(),
+					book.Time.String())
 			}
 		})
 

pkg/exchange/binance/exchange.go

@@ -1346,6 +1346,8 @@ func (e *Exchange) QueryDepth(ctx context.Context, symbol string) (snapshot type
 
 func convertDepth(snapshot types.SliceOrderBook, symbol string, finalUpdateID int64, response *binance.DepthResponse) (types.SliceOrderBook, int64, error) {
 	snapshot.Symbol = symbol
+	// empty time since the API does not provide time information.
+	snapshot.Time = time.Time{}
 	finalUpdateID = response.LastUpdateID
 	for _, entry := range response.Bids {
 		// entry.Price, Quantity: entry.Quantity

pkg/exchange/binance/parse.go

@@ -461,6 +461,7 @@ func (e *DepthEvent) String() (o string) {
 
 func (e *DepthEvent) OrderBook() (book types.SliceOrderBook, err error) {
 	book.Symbol = e.Symbol
+	book.Time = types.NewMillisecondTimestampFromInt(e.EventBase.Time).Time()
 
 	// already in descending order
 	book.Bids = e.Bids

pkg/exchange/binance/stream.go

@@ -88,6 +88,7 @@ func NewStream(ex *Exchange, client *binance.Client, futuresClient *futures.Clie
 		if ok {
 			err := f.AddUpdate(types.SliceOrderBook{
 				Symbol: e.Symbol,
+				Time:   types.NewMillisecondTimestampFromInt(e.EventBase.Time).Time(),
 				Bids:   e.Bids,
 				Asks:   e.Asks,
 			}, e.FirstUpdateID, e.FinalUpdateID)

pkg/exchange/bybit/stream.go

@@ -131,6 +131,7 @@ func (s *Stream) parseWebSocketEvent(in []byte) (interface{}, error) {
 			}
 
 			book.Type = e.WebSocketTopicEvent.Type
+			book.ServerTime = e.WebSocketTopicEvent.Ts.Time()
 			return &book, nil
 
 		case TopicTypeKLine:

pkg/exchange/bybit/stream_test.go

@@ -35,6 +35,7 @@ func getTestClientOrSkip(t *testing.T) *Stream {
 }
 
 func TestStream(t *testing.T) {
+	t.Skip()
 	s := getTestClientOrSkip(t)
 
 	t.Run("Auth test", func(t *testing.T) {
@@ -182,6 +183,7 @@ func TestStream_parseWebSocketEvent(t *testing.T) {
 			UpdateId:   fixedpoint.NewFromFloat(1854104),
 			SequenceId: fixedpoint.NewFromFloat(10559247733),
 			Type:       DataTypeDelta,
+			ServerTime: types.NewMillisecondTimestampFromInt(1691130685111).Time(),
 		}, *book)
 	})
 

pkg/exchange/bybit/types.go

@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"strconv"
 	"strings"
+	"time"
 
 	"github.com/c9s/bbgo/pkg/exchange/bybit/bybitapi"
 	"github.com/c9s/bbgo/pkg/fixedpoint"
@@ -117,14 +118,18 @@ type BookEvent struct {
 	SequenceId fixedpoint.Value `json:"seq"`
 
 	// internal use
-	// Type can be one of snapshot or delta. Copied from WebSocketTopicEvent.Type
+	// Copied from WebSocketTopicEvent.Type, WebSocketTopicEvent.Ts
+	// Type can be one of snapshot or delta.
 	Type DataType
+	// ServerTime using the websocket timestamp as server time. Since the event not provide server time information.
+	ServerTime time.Time
 }
 
 func (e *BookEvent) OrderBook() (snapshot types.SliceOrderBook) {
 	snapshot.Symbol = e.Symbol
 	snapshot.Bids = e.Bids
 	snapshot.Asks = e.Asks
+	snapshot.Time = e.ServerTime
 	return snapshot
 }
 

pkg/exchange/bybit/types_test.go

@@ -524,7 +524,7 @@ func TestTradeEvent_toGlobalTrade(t *testing.T) {
 			OrderId:         fmt.Sprintf("%d", expTrade.OrderID),
 			OrderLinkId:     "1691419101980",
 			Category:        "spot",
-			Symbol:          fmt.Sprintf("%s", expTrade.Symbol),
+			Symbol:          expTrade.Symbol,
 			ExecId:          fmt.Sprintf("%d", expTrade.ID),
 			ExecPrice:       expTrade.Price,
 			ExecQty:         expTrade.Quantity,

pkg/exchange/kucoin/exchange.go

@@ -438,6 +438,7 @@ func (e *Exchange) QueryDepth(ctx context.Context, symbol string) (types.SliceOr
 
 	return types.SliceOrderBook{
 		Symbol: toGlobalSymbol(symbol),
+		Time:   orderBook.Time.Time(),
 		Bids:   orderBook.Bids,
 		Asks:   orderBook.Asks,
 	}, sequence, nil

pkg/exchange/kucoin/stream.go

@@ -73,6 +73,7 @@ func (s *Stream) handleOrderBookL2Event(e *WebSocketOrderBookL2Event) {
 	if ok {
 		f.AddUpdate(types.SliceOrderBook{
 			Symbol: toGlobalSymbol(e.Symbol),
+			Time:   e.Time.Time(),
 			Bids:   e.Changes.Bids,
 			Asks:   e.Changes.Asks,
 		}, e.SequenceStart, e.SequenceEnd)

pkg/exchange/kucoin/websocket.go

@@ -80,6 +80,7 @@ type WebSocketOrderBookL2Event struct {
 		Asks types.PriceVolumeSlice `json:"asks"`
 		Bids types.PriceVolumeSlice `json:"bids"`
 	} `json:"changes"`
+	Time types.MillisecondTimestamp `json:"time"`
 }
 
 type WebSocketCandleEvent struct {

pkg/exchange/max/maxapi/public_parser.go

@@ -81,25 +81,25 @@ type KLineEvent struct {
 }
 
 /*
-{
-  "c": "kline",
-  "M": "btcusdt",
-  "e": "update",
-  "T": 1602999650179,
-  "k": {
-    "ST": 1602999900000,
-    "ET": 1602999900000,
-    "M": "btcusdt",
-    "R": "5m",
-    "O": "11417.21",
-    "H": "11417.21",
-    "L": "11417.21",
-    "C": "11417.21",
-    "v": "0",
-    "ti": 0,
-    "x": false
-  }
-}
+	{
+	  "c": "kline",
+	  "M": "btcusdt",
+	  "e": "update",
+	  "T": 1602999650179,
+	  "k": {
+	    "ST": 1602999900000,
+	    "ET": 1602999900000,
+	    "M": "btcusdt",
+	    "R": "5m",
+	    "O": "11417.21",
+	    "H": "11417.21",
+	    "L": "11417.21",
+	    "C": "11417.21",
+	    "v": "0",
+	    "ti": 0,
+	    "x": false
+	  }
+	}
 */
 type KLinePayload struct {
 	StartTime   int64  `json:"ST"`
@@ -175,6 +175,7 @@ func (e *BookEvent) Time() time.Time {
 
 func (e *BookEvent) OrderBook() (snapshot types.SliceOrderBook, err error) {
 	snapshot.Symbol = strings.ToUpper(e.Market)
+	snapshot.Time = e.Time()
 
 	for _, bid := range e.Bids {
 		pv, err := bid.PriceVolumePair()

pkg/exchange/max/stream.go

@@ -188,6 +188,7 @@ func (s *Stream) handleBookEvent(e max.BookEvent) {
 	}
 
 	newBook.Symbol = toGlobalSymbol(e.Market)
+	newBook.Time = e.Time()
 
 	switch e.Event {
 	case "snapshot":

pkg/exchange/okex/parse.go

@@ -85,6 +85,7 @@ func (data *BookEvent) BookTicker() types.BookTicker {
 func (data *BookEvent) Book() types.SliceOrderBook {
 	book := types.SliceOrderBook{
 		Symbol: data.Symbol,
+		Time:   types.NewMillisecondTimestampFromInt(data.MillisecondTimestamp).Time(),
 	}
 
 	for _, bid := range data.Bids {

pkg/migrations/mysql/migration_api_test.go

@@ -14,7 +14,7 @@ func TestGetMigrationsMap(t *testing.T) {
 
 func TestMergeMigrationsMap(t *testing.T) {
 	MergeMigrationsMap(map[int64]*rockhopper.Migration{
-		2: &rockhopper.Migration{},
-		3: &rockhopper.Migration{},
+		2: {},
+		3: {},
 	})
 }

pkg/migrations/sqlite3/migration_api_test.go

@@ -14,7 +14,7 @@ func TestGetMigrationsMap(t *testing.T) {
 
 func TestMergeMigrationsMap(t *testing.T) {
 	MergeMigrationsMap(map[int64]*rockhopper.Migration{
-		2: &rockhopper.Migration{},
-		3: &rockhopper.Migration{},
+		2: {},
+		3: {},
 	})
 }

pkg/strategy/tri/strategy.go

@@ -484,7 +484,7 @@ func (s *Strategy) executePath(ctx context.Context, session *bbgo.ExchangeSessio
 }
 
 func notifyUsdPnL(profit fixedpoint.Value) {
-	var title = fmt.Sprintf("Triangular Sum PnL ~= ")
+	var title = "Triangular Sum PnL ~= "
 	title += style.PnLEmojiSimple(profit) + " "
 	title += style.PnLSignString(profit) + " USD"
 	bbgo.Notify(title)

pkg/types/sliceorderbook.go

@@ -12,11 +12,14 @@ import (
 
 // SliceOrderBook is a general order book structure which could be used
 // for RESTful responses and websocket stream parsing
+//
 //go:generate callbackgen -type SliceOrderBook
 type SliceOrderBook struct {
 	Symbol string
 	Bids   PriceVolumeSlice
 	Asks   PriceVolumeSlice
+	// Time represents the server time. If empty, it indicates that the server does not provide this information.
+	Time time.Time
 
 	lastUpdateTime time.Time
 
@@ -162,6 +165,8 @@ func (b *SliceOrderBook) String() string {
 	sb.WriteString("BOOK ")
 	sb.WriteString(b.Symbol)
 	sb.WriteString("\n")
+	sb.WriteString(b.Time.Format(time.RFC1123))
+	sb.WriteString("\n")
 
 	if len(b.Asks) > 0 {
 		sb.WriteString("ASKS:\n")
@@ -187,6 +192,7 @@ func (b *SliceOrderBook) String() string {
 func (b *SliceOrderBook) CopyDepth(limit int) OrderBook {
 	var book SliceOrderBook
 	book.Symbol = b.Symbol
+	book.Time = b.Time
 	book.Bids = b.Bids.CopyDepth(limit)
 	book.Asks = b.Asks.CopyDepth(limit)
 	return &book
@@ -195,6 +201,7 @@ func (b *SliceOrderBook) CopyDepth(limit int) OrderBook {
 func (b *SliceOrderBook) Copy() OrderBook {
 	var book SliceOrderBook
 	book.Symbol = b.Symbol
+	book.Time = b.Time
 	book.Bids = b.Bids.Copy()
 	book.Asks = b.Asks.Copy()
 	return &book