Merge branch 'master' into engine

exchanges/huobi/huobi.go

@@ -64,9 +64,10 @@ const (
 // HUOBI is the overarching type across this package
 type HUOBI struct {
 	exchange.Base
-	AccountID     string
-	WebsocketConn *websocket.Conn
-	wsRequestMtx  sync.Mutex
+	AccountID                  string
+	WebsocketConn              *websocket.Conn
+	AuthenticatedWebsocketConn *websocket.Conn
+	wsRequestMtx               sync.Mutex
 }
 
 // GetSpotKline returns kline data

exchanges/huobi/huobi_test.go

@@ -9,12 +9,15 @@ import (
 	"strconv"
 	"strings"
 	"testing"
+	"time"
 
+	"github.com/gorilla/websocket"
 	"github.com/thrasher-/gocryptotrader/common"
 	"github.com/thrasher-/gocryptotrader/common/crypto"
 	"github.com/thrasher-/gocryptotrader/config"
 	"github.com/thrasher-/gocryptotrader/currency"
 	exchange "github.com/thrasher-/gocryptotrader/exchanges"
+	"github.com/thrasher-/gocryptotrader/exchanges/sharedtestvalues"
 )
 
 // Please supply you own test keys here for due diligence testing.
@@ -26,6 +29,7 @@ const (
 )
 
 var h HUOBI
+var wsSetupRan bool
 
 func TestSetDefaults(t *testing.T) {
 	h.SetDefaults()
@@ -39,12 +43,54 @@ func TestSetup(t *testing.T) {
 		t.Error("Test Failed - Huobi Setup() init error")
 	}
 	hConfig.API.AuthenticatedSupport = true
+	hConfig.API.AuthenticatedWebsocketSupport = true
 	hConfig.API.Credentials.Key = apiKey
 	hConfig.API.Credentials.Secret = apiSecret
 
 	h.Setup(hConfig)
 }
 
+func setupWsTests(t *testing.T) {
+	if wsSetupRan {
+		return
+	}
+	TestSetDefaults(t)
+	TestSetup(t)
+	if !h.Websocket.IsEnabled() && !h.API.AuthenticatedWebsocketSupport || !areTestAPIKeysSet() {
+		t.Skip(exchange.WebsocketNotEnabled)
+	}
+	var err error
+	var dialer websocket.Dialer
+	comms = make(chan WsMessage, sharedtestvalues.WebsocketChannelOverrideCapacity)
+	h.Websocket.DataHandler = sharedtestvalues.GetWebsocketInterfaceChannelOverride()
+	h.Websocket.TrafficAlert = sharedtestvalues.GetWebsocketStructChannelOverride()
+	go h.WsHandleData()
+	err = h.wsAuthenticatedDial(&dialer)
+	if err != nil {
+		t.Error(err)
+	}
+	err = h.wsLogin()
+	if err != nil {
+		t.Error(err)
+	}
+	timer := time.NewTimer(sharedtestvalues.WebsocketResponseDefaultTimeout)
+	select {
+	case response := <-h.Websocket.DataHandler:
+		switch respType := response.(type) {
+		case WsAuthenticatedDataResponse:
+			if respType.ErrorCode > 0 {
+				t.Error(respType)
+			}
+		case error:
+			t.Error(respType)
+		}
+	case <-timer.C:
+		t.Error("Websocket did not receive a response")
+	}
+	timer.Stop()
+	wsSetupRan = true
+}
+
 func TestGetSpotKline(t *testing.T) {
 	t.Parallel()
 	_, err := h.GetSpotKline(KlinesRequestParams{
@@ -592,3 +638,50 @@ func TestGetDepositAddress(t *testing.T) {
 		t.Error("Test Failed - GetDepositAddress() error cannot be nil")
 	}
 }
+
+// TestWsGetAccountsList connects to WS, logs in, gets account list
+func TestWsGetAccountsList(t *testing.T) {
+	setupWsTests(t)
+	h.wsGetAccountsList(currency.NewPairFromString("ethbtc"))
+	timer := time.NewTimer(sharedtestvalues.WebsocketResponseDefaultTimeout)
+	select {
+	case response := <-h.Websocket.DataHandler:
+		switch respType := response.(type) {
+		case WsAuthenticatedAccountsListResponse:
+			if respType.ErrorCode > 0 {
+				t.Error(respType)
+			}
+		case error:
+			t.Error(respType)
+		}
+	case <-timer.C:
+		t.Error("Websocket did not receive a response")
+	}
+	timer.Stop()
+}
+
+// TestWsGetOrderList connects to WS, logs in, gets order list
+func TestWsGetOrderList(t *testing.T) {
+	setupWsTests(t)
+	h.wsGetOrdersList(1, currency.NewPairFromString("ethbtc"))
+	timer := time.NewTimer(sharedtestvalues.WebsocketResponseDefaultTimeout)
+	select {
+	case <-h.Websocket.DataHandler:
+	case <-timer.C:
+		t.Error("Websocket did not receive a response")
+	}
+	timer.Stop()
+}
+
+// TestWsGetOrderDetails connects to WS, logs in, gets order details
+func TestWsGetOrderDetails(t *testing.T) {
+	setupWsTests(t)
+	h.wsGetOrderDetails("123")
+	timer := time.NewTimer(sharedtestvalues.WebsocketResponseDefaultTimeout)
+	select {
+	case <-h.Websocket.DataHandler:
+	case <-timer.C:
+		t.Error("Websocket did not receive a response")
+	}
+	timer.Stop()
+}

exchanges/huobi/huobi_types.go

@@ -1,5 +1,7 @@
 package huobi
 
+import "github.com/thrasher-/gocryptotrader/currency"
+
 // Response stores the Huobi response information
 type Response struct {
 	Status       string `json:"status"`
@@ -271,13 +273,13 @@ type WsRequest struct {
 // WsResponse defines a response from the websocket connection when there
 // is an error
 type WsResponse struct {
-	TS           int64  `json:"ts"`
-	Status       string `json:"status"`
-	ErrorCode    string `json:"err-code"`
-	ErrorMessage string `json:"err-msg"`
-	Ping         int64  `json:"ping"`
-	Channel      string `json:"ch"`
-	Subscribed   string `json:"subbed"`
+	TS           int64       `json:"ts"`
+	Status       string      `json:"status"`
+	ErrorCode    interface{} `json:"err-code"`
+	ErrorMessage string      `json:"err-msg"`
+	Ping         int64       `json:"ping"`
+	Channel      string      `json:"ch"`
+	Subscribed   string      `json:"subbed"`
 }
 
 // WsHeartBeat defines a heartbeat request
@@ -323,9 +325,201 @@ type WsTrade struct {
 		Data      []struct {
 			Amount    float64 `json:"amount"`
 			Timestamp int64   `json:"ts"`
-			ID        float64 `json:"id,string"`
+			ID        float64 `json:"id"`
 			Price     float64 `json:"price"`
 			Direction string  `json:"direction"`
 		} `json:"data"`
 	}
 }
+
+// WsAuthenticationRequest data for login
+type WsAuthenticationRequest struct {
+	Op               string `json:"op"`
+	AccessKeyID      string `json:"AccessKeyId"`
+	SignatureMethod  string `json:"SignatureMethod"`
+	SignatureVersion string `json:"SignatureVersion"`
+	Timestamp        string `json:"Timestamp"`
+	Signature        string `json:"Signature"`
+}
+
+// WsMessage defines read data from the websocket connection
+type WsMessage struct {
+	Raw []byte
+	URL string
+}
+
+// WsAuthenticatedSubscriptionRequest request for subscription on authenticated connection
+type WsAuthenticatedSubscriptionRequest struct {
+	Op               string `json:"op"`
+	AccessKeyID      string `json:"AccessKeyId"`
+	SignatureMethod  string `json:"SignatureMethod"`
+	SignatureVersion string `json:"SignatureVersion"`
+	Timestamp        string `json:"Timestamp"`
+	Signature        string `json:"Signature"`
+	Topic            string `json:"topic"`
+}
+
+// WsAuthenticatedAccountsListRequest request for account list authenticated connection
+type WsAuthenticatedAccountsListRequest struct {
+	Op               string        `json:"op"`
+	AccessKeyID      string        `json:"AccessKeyId"`
+	SignatureMethod  string        `json:"SignatureMethod"`
+	SignatureVersion string        `json:"SignatureVersion"`
+	Timestamp        string        `json:"Timestamp"`
+	Signature        string        `json:"Signature"`
+	Topic            string        `json:"topic"`
+	Symbol           currency.Pair `json:"symbol"`
+}
+
+// WsAuthenticatedOrderDetailsRequest request for order details authenticated connection
+type WsAuthenticatedOrderDetailsRequest struct {
+	Op               string `json:"op"`
+	AccessKeyID      string `json:"AccessKeyId"`
+	SignatureMethod  string `json:"SignatureMethod"`
+	SignatureVersion string `json:"SignatureVersion"`
+	Timestamp        string `json:"Timestamp"`
+	Signature        string `json:"Signature"`
+	Topic            string `json:"topic"`
+	OrderID          string `json:"order-id"`
+}
+
+// WsAuthenticatedOrdersListRequest request for orderslist authenticated connection
+type WsAuthenticatedOrdersListRequest struct {
+	Op               string        `json:"op"`
+	AccessKeyID      string        `json:"AccessKeyId"`
+	SignatureMethod  string        `json:"SignatureMethod"`
+	SignatureVersion string        `json:"SignatureVersion"`
+	Timestamp        string        `json:"Timestamp"`
+	Signature        string        `json:"Signature"`
+	Topic            string        `json:"topic"`
+	States           string        `json:"states"`
+	AccountID        int64         `json:"account-id"`
+	Symbol           currency.Pair `json:"symbol"`
+}
+
+// WsAuthenticatedDataResponse response from authenticated connection
+type WsAuthenticatedDataResponse struct {
+	Op           string `json:"op,omitempty"`
+	Ts           int64  `json:"ts,omitempty"`
+	Topic        string `json:"topic,omitempty"`
+	ErrorCode    int64  `json:"err-code,omitempty"`
+	ErrorMessage string `json:"err-msg,omitempty"`
+	Ping         int64  `json:"ping,omitempty"`
+	CID          string `json:"cid,omitempty"`
+}
+
+// WsAuthenticatedAccountsResponse response from Accounts authenticated subscription
+type WsAuthenticatedAccountsResponse struct {
+	WsAuthenticatedDataResponse
+	Data WsAuthenticatedAccountsResponseData `json:"data"`
+}
+
+// WsAuthenticatedAccountsResponseData account data
+type WsAuthenticatedAccountsResponseData struct {
+	Event string                                    `json:"event"`
+	List  []WsAuthenticatedAccountsResponseDataList `json:"list"`
+}
+
+// WsAuthenticatedAccountsResponseDataList detailed account data
+type WsAuthenticatedAccountsResponseDataList struct {
+	AccountID int64   `json:"account-id"`
+	Currency  string  `json:"currency"`
+	Type      string  `json:"type"`
+	Balance   float64 `json:"balance,string"`
+}
+
+// WsAuthenticatedOrdersUpdateResponse response from OrdersUpdate authenticated subscription
+type WsAuthenticatedOrdersUpdateResponse struct {
+	WsAuthenticatedDataResponse
+	Data WsAuthenticatedOrdersUpdateResponseData `json:"data"`
+}
+
+// WsAuthenticatedOrdersUpdateResponseData order  updatedata
+type WsAuthenticatedOrdersUpdateResponseData struct {
+	UnfilledAmount   float64       `json:"unfilled-amount,string"`
+	FilledAmount     float64       `json:"filled-amount,string"`
+	Price            float64       `json:"price,string"`
+	OrderID          int64         `json:"order-id"`
+	Symbol           currency.Pair `json:"symbol"`
+	MatchID          int64         `json:"match-id"`
+	FilledCashAmount float64       `json:"filled-cash-amount,string"`
+	Role             string        `json:"role"`
+	OrderState       string        `json:"order-state"`
+}
+
+// WsAuthenticatedOrdersResponse response from Orders authenticated subscription
+type WsAuthenticatedOrdersResponse struct {
+	WsAuthenticatedDataResponse
+	Data []WsAuthenticatedOrdersResponseData `json:"data"`
+}
+
+// WsAuthenticatedOrdersResponseData order data
+type WsAuthenticatedOrdersResponseData struct {
+	SeqID            int64         `json:"seq-id"`
+	OrderID          int64         `json:"order-id"`
+	Symbol           currency.Pair `json:"symbol"`
+	AccountID        int64         `json:"account-id"`
+	OrderAmount      float64       `json:"order-amount,string"`
+	OrderPrice       float64       `json:"order-price,string"`
+	CreatedAt        int64         `json:"created-at"`
+	OrderType        string        `json:"order-type"`
+	OrderSource      string        `json:"order-source"`
+	OrderState       string        `json:"order-state"`
+	Role             string        `json:"role"`
+	Price            float64       `json:"price,string"`
+	FilledAmount     float64       `json:"filled-amount,string"`
+	UnfilledAmount   float64       `json:"unfilled-amount,string"`
+	FilledCashAmount float64       `json:"filled-cash-amount,string"`
+	FilledFees       float64       `json:"filled-fees,string"`
+}
+
+// WsAuthenticatedAccountsListResponse response from AccountsList authenticated endpoint
+type WsAuthenticatedAccountsListResponse struct {
+	WsAuthenticatedDataResponse
+	Data []WsAuthenticatedAccountsListResponseData `json:"data"`
+}
+
+// WsAuthenticatedAccountsListResponseData account data
+type WsAuthenticatedAccountsListResponseData struct {
+	ID    int64                                         `json:"id"`
+	Type  string                                        `json:"type"`
+	State string                                        `json:"state"`
+	List  []WsAuthenticatedAccountsListResponseDataList `json:"list"`
+}
+
+// WsAuthenticatedAccountsListResponseDataList detailed account data
+type WsAuthenticatedAccountsListResponseDataList struct {
+	Currency string  `json:"currency"`
+	Type     string  `json:"type"`
+	Balance  float64 `json:"balance,string"`
+}
+
+// WsAuthenticatedOrdersListResponse response from OrdersList authenticated endpoint
+type WsAuthenticatedOrdersListResponse struct {
+	WsAuthenticatedDataResponse
+	Data []WsAuthenticatedOrdersListResponseData `json:"data"`
+}
+
+// WsAuthenticatedOrdersListResponseData contains order details
+type WsAuthenticatedOrdersListResponseData struct {
+	ID               int64         `json:"id"`
+	Symbol           currency.Pair `json:"symbol"`
+	AccountID        int64         `json:"account-id"`
+	Amount           float64       `json:"amount,string"`
+	Price            float64       `json:"price,string"`
+	CreatedAt        int64         `json:"created-at"`
+	Type             string        `json:"type"`
+	FilledAmount     float64       `json:"filled-amount,string"`
+	FilledCashAmount float64       `json:"filled-cash-amount,string"`
+	FilledFees       float64       `json:"filled-fees,string"`
+	FinishedAt       int64         `json:"finished-at"`
+	Source           string        `json:"source"`
+	State            string        `json:"state"`
+	CanceledAt       int64         `json:"canceled-at"`
+}
+
+// WsAuthenticatedOrderDetailResponse response from OrderDetail authenticated endpoint
+type WsAuthenticatedOrderDetailResponse struct {
+	WsAuthenticatedDataResponse
+	Data WsAuthenticatedOrdersListResponseData `json:"data"`
+}

exchanges/huobi/huobi_websocket.go

@@ -13,6 +13,7 @@ import (
 
 	"github.com/gorilla/websocket"
 	"github.com/thrasher-/gocryptotrader/common"
+	"github.com/thrasher-/gocryptotrader/common/crypto"
 	"github.com/thrasher-/gocryptotrader/currency"
 	exchange "github.com/thrasher-/gocryptotrader/exchanges"
 	"github.com/thrasher-/gocryptotrader/exchanges/asset"
@@ -21,12 +22,33 @@ import (
 )
 
 const (
-	huobiSocketIOAddress = "wss://api.huobi.pro/hbus/ws"
-	wsMarketKline        = "market.%s.kline.1min"
-	wsMarketDepth        = "market.%s.depth.step0"
-	wsMarketTrade        = "market.%s.trade.detail"
+	baseWSURL = "wss://api.huobi.pro"
+
+	wsMarketURL   = baseWSURL + "/ws"
+	wsMarketKline = "market.%s.kline.1min"
+	wsMarketDepth = "market.%s.depth.step0"
+	wsMarketTrade = "market.%s.trade.detail"
+
+	wsAccountsOrdersEndPoint = "/ws/v1"
+	wsAccountsList           = "accounts.list"
+	wsOrdersList             = "orders.list"
+	wsOrdersDetail           = "orders.detail"
+	wsAccountsOrdersURL      = baseWSURL + wsAccountsOrdersEndPoint
+	wsAccountListEndpoint    = wsAccountsOrdersEndPoint + "/" + wsAccountsList
+	wsOrdersListEndpoint     = wsAccountsOrdersEndPoint + "/" + wsOrdersList
+	wsOrdersDetailEndpoint   = wsAccountsOrdersEndPoint + "/" + wsOrdersDetail
+
+	wsDateTimeFormatting = "2006-01-02T15:04:05"
+
+	signatureMethod  = "HmacSHA256"
+	signatureVersion = "2"
+	requestOp        = "req"
+	authOp           = "auth"
 )
 
+// Instantiates a communications channel between websocket connections
+var comms = make(chan WsMessage, 1)
+
 // WsConnect initiates a new websocket connection
 func (h *HUOBI) WsConnect() error {
 	if !h.Websocket.IsEnabled() || !h.IsEnabled() {
@@ -44,140 +66,264 @@ func (h *HUOBI) WsConnect() error {
 		dialer.Proxy = http.ProxyURL(proxy)
 	}
 
-	var err error
-	h.WebsocketConn, _, err = dialer.Dial(h.Websocket.GetWebsocketURL(), http.Header{})
+	err := h.wsDial(&dialer)
 	if err != nil {
 		return err
 	}
+	err = h.wsAuthenticatedDial(&dialer)
+	if err != nil {
+		log.Errorf("%v - authenticated dial failed: %v", h.Name, err)
+	}
+	err = h.wsLogin()
+	if err != nil {
+		log.Errorf("%v - authentication failed: %v", h.Name, err)
+	}
 
 	go h.WsHandleData()
-
 	h.GenerateDefaultSubscriptions()
+
 	return nil
 }
 
-// WsReadData reads data from the websocket connection
-func (h *HUOBI) WsReadData() (exchange.WebsocketResponse, error) {
-	_, resp, err := h.WebsocketConn.ReadMessage()
+func (h *HUOBI) wsDial(dialer *websocket.Dialer) error {
+	var err error
+	var conStatus *http.Response
+	h.WebsocketConn, conStatus, err = dialer.Dial(wsMarketURL, http.Header{})
 	if err != nil {
-		return exchange.WebsocketResponse{}, err
+		return fmt.Errorf("%v %v %v Error: %v", wsMarketURL, conStatus, conStatus.StatusCode, err)
 	}
+	go h.wsMultiConnectionFunnel(h.WebsocketConn, wsMarketURL)
+	return nil
+}
 
-	h.Websocket.TrafficAlert <- struct{}{}
-
-	b := bytes.NewReader(resp)
-	gReader, err := gzip.NewReader(b)
+func (h *HUOBI) wsAuthenticatedDial(dialer *websocket.Dialer) error {
+	if !h.GetAuthenticatedAPISupport(exchange.WebsocketAuthentication) {
+		return fmt.Errorf("%v AuthenticatedWebsocketAPISupport not enabled", h.Name)
+	}
+	var err error
+	var conStatus *http.Response
+	h.AuthenticatedWebsocketConn, conStatus, err = dialer.Dial(wsAccountsOrdersURL, http.Header{})
 	if err != nil {
-		return exchange.WebsocketResponse{}, err
+		return fmt.Errorf("%v %v %v Error: %v", wsAccountsOrdersURL, conStatus, conStatus.StatusCode, err)
 	}
+	go h.wsMultiConnectionFunnel(h.AuthenticatedWebsocketConn, wsAccountsOrdersURL)
+	return nil
+}
 
-	unzipped, err := ioutil.ReadAll(gReader)
-	if err != nil {
-		return exchange.WebsocketResponse{}, err
+// wsMultiConnectionFunnel manages data from multiple endpoints and passes it to a channel
+func (h *HUOBI) wsMultiConnectionFunnel(ws *websocket.Conn, url string) {
+	h.Websocket.Wg.Add(1)
+	defer h.Websocket.Wg.Done()
+	for {
+		select {
+		case <-h.Websocket.ShutdownC:
+			return
+		default:
+			_, resp, err := ws.ReadMessage()
+			if err != nil {
+				h.Websocket.DataHandler <- err
+				return
+			}
+			h.Websocket.TrafficAlert <- struct{}{}
+			b := bytes.NewReader(resp)
+			gReader, err := gzip.NewReader(b)
+			if err != nil {
+				h.Websocket.DataHandler <- err
+				return
+			}
+			unzipped, err := ioutil.ReadAll(gReader)
+			if err != nil {
+				h.Websocket.DataHandler <- err
+				return
+			}
+			err = gReader.Close()
+			if err != nil {
+				h.Websocket.DataHandler <- err
+				return
+			}
+			comms <- WsMessage{Raw: unzipped, URL: url}
+		}
 	}
-	gReader.Close()
-
-	return exchange.WebsocketResponse{Raw: unzipped}, nil
 }
 
 // WsHandleData handles data read from the websocket connection
 func (h *HUOBI) WsHandleData() {
 	h.Websocket.Wg.Add(1)
-
-	defer func() {
-		h.Websocket.Wg.Done()
-	}()
-
+	defer h.Websocket.Wg.Done()
 	for {
 		select {
 		case <-h.Websocket.ShutdownC:
 			return
-
-		default:
-			resp, err := h.WsReadData()
-			if err != nil {
-				h.Websocket.DataHandler <- err
-				return
+		case resp := <-comms:
+			if h.Verbose {
+				log.Debugf("%v: %v: %v", h.Name, resp.URL, string(resp.Raw))
 			}
-
-			var init WsResponse
-			err = common.JSONDecode(resp.Raw, &init)
-			if err != nil {
-				h.Websocket.DataHandler <- err
-				continue
+			switch resp.URL {
+			case wsMarketURL:
+				h.wsHandleMarketData(resp)
+			case wsAccountsOrdersURL:
+				h.wsHandleAuthenticatedData(resp)
 			}
+		}
+	}
+}
 
-			if init.Status == "error" {
-				h.Websocket.DataHandler <- fmt.Errorf("huobi.go Websocker error %s %s",
-					init.ErrorCode,
-					init.ErrorMessage)
-				continue
-			}
+func (h *HUOBI) wsHandleAuthenticatedData(resp WsMessage) {
+	var init WsAuthenticatedDataResponse
+	err := common.JSONDecode(resp.Raw, &init)
+	if err != nil {
+		h.Websocket.DataHandler <- err
+		return
+	}
+	if init.ErrorCode > 0 {
+		if init.ErrorMessage == "api-signature-not-valid" {
+			h.Websocket.SetCanUseAuthenticatedEndpoints(false)
+		}
+		h.Websocket.DataHandler <- fmt.Errorf("%v %v Websocket error %v %s",
+			h.Name,
+			resp.URL,
+			init.ErrorCode,
+			init.ErrorMessage)
+		return
+	}
+	if init.Ping != 0 {
+		err = h.WebsocketConn.WriteJSON(`{"pong":1337}`)
+		if err != nil {
+			log.Error(err)
+		}
+		return
+	}
 
-			if init.Subscribed != "" {
-				continue
-			}
+	if init.Op == "sub" {
+		if h.Verbose {
+			log.Debugf("%v: %v: Successfully subscribed to %v", h.Name, resp.URL, init.Topic)
+		}
+		return
+	}
 
-			if init.Ping != 0 {
-				err = h.WebsocketConn.WriteJSON(`{"pong":1337}`)
-				if err != nil {
-					log.Error(err)
-				}
-				continue
-			}
+	switch {
+	case strings.EqualFold(init.Op, authOp):
+		h.Websocket.SetCanUseAuthenticatedEndpoints(true)
+		var response WsAuthenticatedDataResponse
+		err := common.JSONDecode(resp.Raw, &response)
+		if err != nil {
+			h.Websocket.DataHandler <- err
+		}
+		h.Websocket.DataHandler <- response
+	case strings.EqualFold(init.Topic, "accounts"):
+		var response WsAuthenticatedAccountsResponse
+		err := common.JSONDecode(resp.Raw, &response)
+		if err != nil {
+			h.Websocket.DataHandler <- err
+		}
+		h.Websocket.DataHandler <- response
+	case strings.Contains(init.Topic, "orders") &&
+		strings.Contains(init.Topic, "update"):
+		var response WsAuthenticatedOrdersUpdateResponse
+		err := common.JSONDecode(resp.Raw, &response)
+		if err != nil {
+			h.Websocket.DataHandler <- err
+		}
+		h.Websocket.DataHandler <- response
+	case strings.Contains(init.Topic, "orders"):
+		var response WsAuthenticatedOrdersResponse
+		err := common.JSONDecode(resp.Raw, &response)
+		if err != nil {
+			h.Websocket.DataHandler <- err
+		}
+		h.Websocket.DataHandler <- response
+	case strings.EqualFold(init.Topic, wsAccountsList):
+		var response WsAuthenticatedAccountsListResponse
+		err := common.JSONDecode(resp.Raw, &response)
+		if err != nil {
+			h.Websocket.DataHandler <- err
+		}
+		h.Websocket.DataHandler <- response
+	case strings.EqualFold(init.Topic, wsOrdersList):
+		var response WsAuthenticatedOrdersListResponse
+		err := common.JSONDecode(resp.Raw, &response)
+		if err != nil {
+			h.Websocket.DataHandler <- err
+		}
+		h.Websocket.DataHandler <- response
+	case strings.EqualFold(init.Topic, wsOrdersDetail):
+		var response WsAuthenticatedOrderDetailResponse
+		err := common.JSONDecode(resp.Raw, &response)
+		if err != nil {
+			h.Websocket.DataHandler <- err
+		}
+		h.Websocket.DataHandler <- response
+	}
+}
 
-			switch {
-			case strings.Contains(init.Channel, "depth"):
-				var depth WsDepth
-				err := common.JSONDecode(resp.Raw, &depth)
-				if err != nil {
-					h.Websocket.DataHandler <- err
-					continue
-				}
+func (h *HUOBI) wsHandleMarketData(resp WsMessage) {
+	var init WsResponse
+	err := common.JSONDecode(resp.Raw, &init)
+	if err != nil {
+		h.Websocket.DataHandler <- err
+		return
+	}
+	if init.Status == "error" {
+		h.Websocket.DataHandler <- fmt.Errorf("%v %v Websocket error %s %s",
+			h.Name,
+			resp.URL,
+			init.ErrorCode,
+			init.ErrorMessage)
+		return
+	}
+	if init.Subscribed != "" {
+		return
+	}
+	if init.Ping != 0 {
+		err = h.WebsocketConn.WriteJSON(`{"pong":1337}`)
+		if err != nil {
+			log.Error(err)
+		}
+		return
+	}
 
-				data := strings.Split(depth.Channel, ".")
-
-				h.WsProcessOrderbook(&depth, data[1])
-
-			case strings.Contains(init.Channel, "kline"):
-				var kline WsKline
-				err := common.JSONDecode(resp.Raw, &kline)
-				if err != nil {
-					h.Websocket.DataHandler <- err
-					continue
-				}
-
-				data := strings.Split(kline.Channel, ".")
-
-				h.Websocket.DataHandler <- exchange.KlineData{
-					Timestamp:  time.Unix(0, kline.Timestamp),
-					Exchange:   h.GetName(),
-					AssetType:  asset.Spot,
-					Pair:       currency.NewPairFromString(data[1]),
-					OpenPrice:  kline.Tick.Open,
-					ClosePrice: kline.Tick.Close,
-					HighPrice:  kline.Tick.High,
-					LowPrice:   kline.Tick.Low,
-					Volume:     kline.Tick.Volume,
-				}
-
-			case strings.Contains(init.Channel, "trade"):
-				var trade WsTrade
-				err := common.JSONDecode(resp.Raw, &trade)
-				if err != nil {
-					h.Websocket.DataHandler <- err
-					continue
-				}
-
-				data := strings.Split(trade.Channel, ".")
-
-				h.Websocket.DataHandler <- exchange.TradeData{
-					Exchange:     h.GetName(),
-					AssetType:    asset.Spot,
-					CurrencyPair: currency.NewPairFromString(data[1]),
-					Timestamp:    time.Unix(0, trade.Tick.Timestamp),
-				}
-			}
+	switch {
+	case strings.Contains(init.Channel, "depth"):
+		var depth WsDepth
+		err := common.JSONDecode(resp.Raw, &depth)
+		if err != nil {
+			h.Websocket.DataHandler <- err
+			return
+		}
+		data := strings.Split(depth.Channel, ".")
+		h.WsProcessOrderbook(&depth, data[1])
+	case strings.Contains(init.Channel, "kline"):
+		var kline WsKline
+		err := common.JSONDecode(resp.Raw, &kline)
+		if err != nil {
+			h.Websocket.DataHandler <- err
+			return
+		}
+		data := strings.Split(kline.Channel, ".")
+		h.Websocket.DataHandler <- exchange.KlineData{
+			Timestamp:  time.Unix(0, kline.Timestamp),
+			Exchange:   h.GetName(),
+			AssetType:  "SPOT",
+			Pair:       currency.NewPairFromString(data[1]),
+			OpenPrice:  kline.Tick.Open,
+			ClosePrice: kline.Tick.Close,
+			HighPrice:  kline.Tick.High,
+			LowPrice:   kline.Tick.Low,
+			Volume:     kline.Tick.Volume,
+		}
+	case strings.Contains(init.Channel, "trade"):
+		var trade WsTrade
+		err := common.JSONDecode(resp.Raw, &trade)
+		if err != nil {
+			h.Websocket.DataHandler <- err
+			return
+		}
+		data := strings.Split(trade.Channel, ".")
+		h.Websocket.DataHandler <- exchange.TradeData{
+			Exchange:     h.GetName(),
+			AssetType:    "SPOT",
+			CurrencyPair: currency.NewPairFromString(data[1]),
+			Timestamp:    time.Unix(0, trade.Tick.Timestamp),
 		}
 	}
 }
@@ -222,8 +368,14 @@ func (h *HUOBI) WsProcessOrderbook(ob *WsDepth, symbol string) error {
 // GenerateDefaultSubscriptions Adds default subscriptions to websocket to be handled by ManageSubscriptions()
 func (h *HUOBI) GenerateDefaultSubscriptions() {
 	var channels = []string{wsMarketKline, wsMarketDepth, wsMarketTrade}
+	var subscriptions []exchange.WebsocketChannelSubscription
+	if h.Websocket.CanUseAuthenticatedEndpoints() {
+		channels = append(channels, "orders.%v", "orders.%v.update")
+		subscriptions = append(subscriptions, exchange.WebsocketChannelSubscription{
+			Channel: "accounts",
+		})
+	}
 	enabledCurrencies := h.GetEnabledPairs(asset.Spot)
-	subscriptions := []exchange.WebsocketChannelSubscription{}
 	for i := range channels {
 		for j := range enabledCurrencies {
 			enabledCurrencies[j].Delimiter = ""
@@ -239,11 +391,11 @@ func (h *HUOBI) GenerateDefaultSubscriptions() {
 
 // Subscribe sends a websocket message to receive data from the channel
 func (h *HUOBI) Subscribe(channelToSubscribe exchange.WebsocketChannelSubscription) error {
-	subscriptionRequest := WsRequest{Subscribe: channelToSubscribe.Channel}
-	if h.Verbose {
-		log.Debugf("Subscription: %v", subscriptionRequest)
+	if strings.Contains(channelToSubscribe.Channel, "orders.") ||
+		strings.Contains(channelToSubscribe.Channel, "accounts") {
+		return h.wsAuthenticatedSubscribe("sub", wsAccountsOrdersEndPoint+channelToSubscribe.Channel, channelToSubscribe.Channel)
 	}
-	subscription, err := common.JSONEncode(subscriptionRequest)
+	subscription, err := common.JSONEncode(WsRequest{Subscribe: channelToSubscribe.Channel})
 	if err != nil {
 		return err
 	}
@@ -252,6 +404,10 @@ func (h *HUOBI) Subscribe(channelToSubscribe exchange.WebsocketChannelSubscripti
 
 // Unsubscribe sends a websocket message to stop receiving data from the channel
 func (h *HUOBI) Unsubscribe(channelToSubscribe exchange.WebsocketChannelSubscription) error {
+	if strings.Contains(channelToSubscribe.Channel, "orders.") ||
+		strings.Contains(channelToSubscribe.Channel, "accounts") {
+		return h.wsAuthenticatedSubscribe("unsub", wsAccountsOrdersEndPoint+channelToSubscribe.Channel, channelToSubscribe.Channel)
+	}
 	subscription, err := common.JSONEncode(WsRequest{Unsubscribe: channelToSubscribe.Channel})
 	if err != nil {
 		return err
@@ -268,3 +424,125 @@ func (h *HUOBI) wsSend(data []byte) error {
 	}
 	return h.WebsocketConn.WriteMessage(websocket.TextMessage, data)
 }
+
+func (h *HUOBI) wsLogin() error {
+	if !h.GetAuthenticatedAPISupport(exchange.WebsocketAuthentication) {
+		return fmt.Errorf("%v AuthenticatedWebsocketAPISupport not enabled", h.Name)
+	}
+	h.Websocket.SetCanUseAuthenticatedEndpoints(true)
+	timestamp := time.Now().UTC().Format(wsDateTimeFormatting)
+	request := WsAuthenticationRequest{
+		Op:               authOp,
+		AccessKeyID:      h.API.Credentials.Key,
+		SignatureMethod:  signatureMethod,
+		SignatureVersion: signatureVersion,
+		Timestamp:        timestamp,
+	}
+	hmac := h.wsGenerateSignature(timestamp, wsAccountsOrdersEndPoint)
+	request.Signature = crypto.Base64Encode(hmac)
+	err := h.wsAuthenticatedSend(request)
+	if err != nil {
+		h.Websocket.SetCanUseAuthenticatedEndpoints(false)
+		return err
+	}
+	return nil
+}
+
+func (h *HUOBI) wsAuthenticatedSend(request interface{}) error {
+	h.wsRequestMtx.Lock()
+	defer h.wsRequestMtx.Unlock()
+	encodedRequest, err := common.JSONEncode(request)
+	if err != nil {
+		return err
+	}
+	if h.Verbose {
+		log.Debugf("%v sending Authenticated message to websocket %s", h.Name, string(encodedRequest))
+	}
+	return h.AuthenticatedWebsocketConn.WriteMessage(websocket.TextMessage, encodedRequest)
+}
+
+func (h *HUOBI) wsGenerateSignature(timestamp, endpoint string) []byte {
+	values := url.Values{}
+	values.Set("AccessKeyId", h.API.Credentials.Key)
+	values.Set("SignatureMethod", signatureMethod)
+	values.Set("SignatureVersion", signatureVersion)
+	values.Set("Timestamp", timestamp)
+	host := "api.huobi.pro"
+	payload := fmt.Sprintf("%s\n%s\n%s\n%s",
+		"GET", host, endpoint, values.Encode())
+	return crypto.GetHMAC(crypto.HashSHA256, []byte(payload), []byte(h.API.Credentials.Secret))
+}
+
+func (h *HUOBI) wsAuthenticatedSubscribe(operation, endpoint, topic string) error {
+	timestamp := time.Now().UTC().Format(wsDateTimeFormatting)
+	request := WsAuthenticatedSubscriptionRequest{
+		Op:               operation,
+		AccessKeyID:      h.API.Credentials.Key,
+		SignatureMethod:  signatureMethod,
+		SignatureVersion: signatureVersion,
+		Timestamp:        timestamp,
+		Topic:            topic,
+	}
+	hmac := h.wsGenerateSignature(timestamp, endpoint)
+	request.Signature = crypto.Base64Encode(hmac)
+	return h.wsAuthenticatedSend(request)
+}
+
+func (h *HUOBI) wsGetAccountsList(pair currency.Pair) error {
+	if !h.Websocket.CanUseAuthenticatedEndpoints() {
+		return fmt.Errorf("%v not authenticated cannot get accounts list", h.Name)
+	}
+	timestamp := time.Now().UTC().Format(wsDateTimeFormatting)
+	request := WsAuthenticatedAccountsListRequest{
+		Op:               requestOp,
+		AccessKeyID:      h.API.Credentials.Key,
+		SignatureMethod:  signatureMethod,
+		SignatureVersion: signatureVersion,
+		Timestamp:        timestamp,
+		Topic:            wsAccountsList,
+		Symbol:           pair,
+	}
+	hmac := h.wsGenerateSignature(timestamp, wsAccountListEndpoint)
+	request.Signature = crypto.Base64Encode(hmac)
+	return h.wsAuthenticatedSend(request)
+}
+
+func (h *HUOBI) wsGetOrdersList(accountID int64, pair currency.Pair) error {
+	if !h.Websocket.CanUseAuthenticatedEndpoints() {
+		return fmt.Errorf("%v not authenticated cannot get orders list", h.Name)
+	}
+	timestamp := time.Now().UTC().Format(wsDateTimeFormatting)
+	request := WsAuthenticatedOrdersListRequest{
+		Op:               requestOp,
+		AccessKeyID:      h.API.Credentials.Key,
+		SignatureMethod:  signatureMethod,
+		SignatureVersion: signatureVersion,
+		Timestamp:        timestamp,
+		Topic:            wsOrdersList,
+		AccountID:        accountID,
+		Symbol:           pair.Lower(),
+		States:           "submitted,partial-filled",
+	}
+	hmac := h.wsGenerateSignature(timestamp, wsOrdersListEndpoint)
+	request.Signature = crypto.Base64Encode(hmac)
+	return h.wsAuthenticatedSend(request)
+}
+
+func (h *HUOBI) wsGetOrderDetails(orderID string) error {
+	if !h.Websocket.CanUseAuthenticatedEndpoints() {
+		return fmt.Errorf("%v not authenticated cannot get order details", h.Name)
+	}
+	timestamp := time.Now().UTC().Format(wsDateTimeFormatting)
+	request := WsAuthenticatedOrderDetailsRequest{
+		Op:               requestOp,
+		AccessKeyID:      h.API.Credentials.Key,
+		SignatureMethod:  signatureMethod,
+		SignatureVersion: signatureVersion,
+		Timestamp:        timestamp,
+		Topic:            wsOrdersDetail,
+		OrderID:          orderID,
+	}
+	hmac := h.wsGenerateSignature(timestamp, wsOrdersDetailEndpoint)
+	request.Signature = crypto.Base64Encode(hmac)
+	return h.wsAuthenticatedSend(request)
+}

exchanges/huobi/huobi_wrapper.go

@@ -117,7 +117,7 @@ func (h *HUOBI) Setup(exch *config.ExchangeConfig) error {
 		exch.Name,
 		exch.Features.Enabled.Websocket,
 		exch.Verbose,
-		huobiSocketIOAddress,
+		wsMarketURL,
 		exch.API.Endpoints.WebsocketURL)
 }
 
@@ -133,7 +133,7 @@ func (h *HUOBI) Start(wg *sync.WaitGroup) {
 // Run implements the HUOBI wrapper
 func (h *HUOBI) Run() {
 	if h.Verbose {
-		log.Debugf("%s Websocket: %s (url: %s).\n", h.GetName(), common.IsEnabled(h.Websocket.IsEnabled()), huobiSocketIOAddress)
+		log.Debugf("%s Websocket: %s (url: %s).\n", h.GetName(), common.IsEnabled(h.Websocket.IsEnabled()), wsMarketURL)
 		h.PrintEnabledPairs()
 	}
 
@@ -635,3 +635,13 @@ func (h *HUOBI) UnsubscribeToWebsocketChannels(channels []exchange.WebsocketChan
 	h.Websocket.UnsubscribeToChannels(channels)
 	return nil
 }
+
+// GetSubscriptions returns a copied list of subscriptions
+func (h *HUOBI) GetSubscriptions() ([]exchange.WebsocketChannelSubscription, error) {
+	return h.Websocket.GetSubscriptions(), nil
+}
+
+// AuthenticateWebsocket sends an authentication message to the websocket
+func (h *HUOBI) AuthenticateWebsocket() error {
+	return h.wsLogin()
+}