ftx: trade update

pkg/cmd/orderbook.go

@@ -66,7 +66,7 @@ var orderbookCmd = &cobra.Command{
 	},
 }
 
-// go run ./cmd/bbgo orderupdate
+// go run ./cmd/bbgo orderupdate --session=ftx
 var orderUpdateCmd = &cobra.Command{
 	Use: "orderupdate",
 	RunE: func(cmd *cobra.Command, args []string) error {

pkg/cmd/trades.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 	"os"
+	"syscall"
 	"time"
 
 	"github.com/pkg/errors"
@@ -11,15 +12,10 @@ import (
 	"github.com/spf13/cobra"
 
 	"github.com/c9s/bbgo/pkg/bbgo"
+	"github.com/c9s/bbgo/pkg/cmd/cmdutil"
 	"github.com/c9s/bbgo/pkg/types"
 )
 
-func init() {
-	tradesCmd.Flags().String("session", "", "the exchange session name for querying balances")
-	tradesCmd.Flags().String("symbol", "", "the trading pair, like btcusdt")
-	RootCmd.AddCommand(tradesCmd)
-}
-
 // go run ./cmd/bbgo tradesCmd --session=ftx --symbol="BTC/USD"
 var tradesCmd = &cobra.Command{
 	Use:          "trades",
@@ -96,3 +92,77 @@ var tradesCmd = &cobra.Command{
 		return nil
 	},
 }
+
+// go run ./cmd/bbgo tradeupdate --session=ftx
+var tradeUpdateCmd = &cobra.Command{
+	Use: "tradeupdate",
+	RunE: func(cmd *cobra.Command, args []string) error {
+		ctx := context.Background()
+
+		configFile, err := cmd.Flags().GetString("config")
+		if err != nil {
+			return err
+		}
+
+		if len(configFile) == 0 {
+			return errors.New("--config option is required")
+		}
+
+		// if config file exists, use the config loaded from the config file.
+		// otherwise, use a empty config object
+		var userConfig *bbgo.Config
+		if _, err := os.Stat(configFile); err == nil {
+			// load successfully
+			userConfig, err = bbgo.Load(configFile, false)
+			if err != nil {
+				return err
+			}
+		} else if os.IsNotExist(err) {
+			// config file doesn't exist
+			userConfig = &bbgo.Config{}
+		} else {
+			// other error
+			return err
+		}
+
+		environ := bbgo.NewEnvironment()
+
+		if err := environ.ConfigureExchangeSessions(userConfig); err != nil {
+			return err
+		}
+
+		sessionName, err := cmd.Flags().GetString("session")
+		if err != nil {
+			return err
+		}
+
+		session, ok := environ.Session(sessionName)
+		if !ok {
+			return fmt.Errorf("session %s not found", sessionName)
+		}
+
+		s := session.Exchange.NewStream()
+		s.OnTradeUpdate(func(trade types.Trade) {
+			log.Infof("trade update: %+v", trade)
+		})
+
+		log.Infof("connecting...")
+		if err := s.Connect(ctx); err != nil {
+			return fmt.Errorf("failed to connect to %s", sessionName)
+		}
+		log.Infof("connected")
+
+		cmdutil.WaitForSignal(ctx, syscall.SIGINT, syscall.SIGTERM)
+		return nil
+	},
+}
+
+func init() {
+	tradesCmd.Flags().String("session", "", "the exchange session name for querying balances")
+	tradesCmd.Flags().String("symbol", "", "the trading pair, like btcusdt")
+
+	tradeUpdateCmd.Flags().String("session", "", "the exchange session name for querying balances")
+
+	RootCmd.AddCommand(tradesCmd)
+	RootCmd.AddCommand(tradeUpdateCmd)
+}

pkg/exchange/ftx/stream.go

@@ -50,6 +50,10 @@ func (s *Stream) subscribePrivateEvents() {
 		Operation: subscribe,
 		Channel:   privateOrdersChannel,
 	})
+	s.wsService.Subscribe(websocketRequest{
+		Operation: subscribe,
+		Channel:   privateTradesChannel,
+	})
 }
 
 func (s *Stream) SetPublicOnly() {

pkg/exchange/ftx/stream_message_handler.go

@@ -11,6 +11,7 @@ type messageHandler struct {
 }
 
 func (h *messageHandler) handleMessage(message []byte) {
+	logger.Infof("raw: %+v", string(message))
 	var r websocketResponse
 	if err := json.Unmarshal(message, &r); err != nil {
 		logger.WithError(err).Errorf("failed to unmarshal resp: %s", string(message))
@@ -22,6 +23,8 @@ func (h *messageHandler) handleMessage(message []byte) {
 		h.handleOrderBook(r)
 	case privateOrdersChannel:
 		h.handlePrivateOrders(r)
+	case privateTradesChannel:
+		h.handleTrades(r)
 	default:
 		if r.Type != errRespType {
 			logger.Errorf("unsupported message type: %+v", r.Type)
@@ -93,3 +96,23 @@ func (h *messageHandler) handlePrivateOrders(response websocketResponse) {
 	}
 	h.EmitOrderUpdate(globalOrder)
 }
+
+func (h *messageHandler) handleTrades(response websocketResponse) {
+	if response.Type == subscribedRespType {
+		h.handleSubscribedMessage(response)
+		return
+	}
+
+	r, err := response.toTradeUpdateResponse()
+	if err != nil {
+		logger.WithError(err).Errorf("failed to convert the trade update response")
+		return
+	}
+
+	t, err := toGlobalTrade(r.Data)
+	if err != nil {
+		logger.WithError(err).Errorf("failed to convert trade update to global trade ")
+		return
+	}
+	h.EmitTradeUpdate(t)
+}

pkg/exchange/ftx/stream_message_handler_test.go

@@ -0,0 +1,119 @@
+package ftx
+
+import (
+	"database/sql"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+
+	"github.com/c9s/bbgo/pkg/datatype"
+	"github.com/c9s/bbgo/pkg/types"
+)
+
+func Test_messageHandler_handleMessage(t *testing.T) {
+	t.Run("handle order update", func(t *testing.T) {
+		input := []byte(`
+{
+  "channel": "orders",
+  "type": "update",
+  "data": {
+    "id": 36379,
+    "clientId": null,
+    "market": "OXY-PERP",
+    "type": "limit",
+    "side": "sell",
+    "price": 2.7185,
+    "size": 1.0,
+    "status": "closed",
+    "filledSize": 1.0,
+    "remainingSize": 0.0,
+    "reduceOnly": false,
+    "liquidation": false,
+    "avgFillPrice": 2.7185,
+    "postOnly": false,
+    "ioc": false,
+    "createdAt": "2021-03-28T06:12:50.991447+00:00"
+  }
+}
+`)
+
+		h := &messageHandler{StandardStream: &types.StandardStream{}}
+		i := 0
+		h.OnOrderUpdate(func(order types.Order) {
+			i++
+			assert.Equal(t, types.Order{
+				SubmitOrder: types.SubmitOrder{
+					ClientOrderID: "",
+					Symbol:        "OXY-PERP",
+					Side:          types.SideTypeSell,
+					Type:          types.OrderTypeLimit,
+					Quantity:      1.0,
+					Price:         2.7185,
+					TimeInForce:   "GTC",
+				},
+				Exchange:         types.ExchangeFTX.String(),
+				OrderID:          36379,
+				Status:           types.OrderStatusFilled,
+				ExecutedQuantity: 1.0,
+				CreationTime:     datatype.Time(mustParseDatetime("2021-03-28T06:12:50.991447+00:00")),
+				UpdateTime:       datatype.Time(mustParseDatetime("2021-03-28T06:12:50.991447+00:00")),
+			}, order)
+		})
+		h.handleMessage(input)
+		assert.Equal(t, 1, i)
+	})
+
+	t.Run("handle trade update", func(t *testing.T) {
+		input := []byte(`
+{
+  "channel": "fills",
+  "type": "update",
+  "data": {
+    "id": 23427,
+    "market": "OXY-PERP",
+    "future": "OXY-PERP",
+    "baseCurrency": null,
+    "quoteCurrency": null,
+    "type": "order",
+    "side": "buy",
+    "price": 2.723,
+    "size": 1.0,
+    "orderId": 323789,
+    "time": "2021-03-28T06:12:34.702926+00:00",
+    "tradeId": 6276431,
+    "feeRate": 0.00056525,
+    "fee": 0.00153917575,
+    "feeCurrency": "USD",
+    "liquidity": "taker"
+  }
+}
+`)
+		h := &messageHandler{StandardStream: &types.StandardStream{}}
+		i := 0
+		h.OnTradeUpdate(func(trade types.Trade) {
+			i++
+			assert.Equal(t, types.Trade{
+				GID:           0,
+				ID:            6276431,
+				OrderID:       323789,
+				Exchange:      types.ExchangeFTX.String(),
+				Price:         2.723,
+				Quantity:      1.0,
+				QuoteQuantity: 2.723 * 1.0,
+				Symbol:        "OXY-PERP",
+				Side:          types.SideTypeBuy,
+				IsBuyer:       true,
+				IsMaker:       false,
+				Time:          datatype.Time(mustParseDatetime("2021-03-28T06:12:34.702926+00:00")),
+				Fee:           0.00153917575,
+				FeeCurrency:   "USD",
+				IsMargin:      false,
+				IsIsolated:    false,
+				StrategyID:    sql.NullString{},
+				PnL:           sql.NullFloat64{},
+			}, trade)
+		})
+		h.handleMessage(input)
+		assert.Equal(t, 1, i)
+	})
+}

pkg/exchange/ftx/websocket_messages.go

@@ -23,6 +23,7 @@ type channel string
 
 const orderBookChannel channel = "orderbook"
 const privateOrdersChannel channel = "orders"
+const privateTradesChannel channel = "fills"
 
 var errUnsupportedConversion = fmt.Errorf("unsupported conversion")
 
@@ -124,6 +125,24 @@ func (r websocketResponse) toOrderUpdateResponse() (orderUpdateResponse, error)
 	return o, nil
 }
 
+type tradeUpdateResponse struct {
+	mandatoryFields
+
+	Data fill `json:"data"`
+}
+
+func (r websocketResponse) toTradeUpdateResponse() (tradeUpdateResponse, error) {
+	if r.Channel != privateTradesChannel {
+		return tradeUpdateResponse{}, fmt.Errorf("type %s, channel %s: %w", r.Type, r.Channel, errUnsupportedConversion)
+	}
+	var t tradeUpdateResponse
+	if err := json.Unmarshal(r.Data, &t.Data); err != nil {
+		return tradeUpdateResponse{}, err
+	}
+	t.mandatoryFields = r.mandatoryFields
+	return t, nil
+}
+
 /*
 Private:
 	order: {"type": "subscribed", "channel": "orders"}