Initial overhaul of websocket connection and feeds (#189)

* Initial overhaul of websocket connection and feeds
* Added proxy support
* Piped to routines.go

* Added new websocket file in exchanges
Refactored orderbook handling into exchange_websocket.go
Added better error responses for binance_websocket.go
General clean for binance_websocket.go

* General fixes - bitfinex_websocket.go
Refactored orderbook cache code - bitfinex_websocket.go
Removed fatal error with unhandled type - routines.go

* Added general improvements to bitmex_websocket.go
Refactored orderbook handling to exchange_websocket.go
Added variable in Item struct in orderbook.go for looking up orders by ID

* Fix issue when routines are blocked due to Data Handler not started
Updated traffic handler
General fixes for bitstamp_websocket.go

* General fixes for coinbasepro_websocket.go

* General fixes for coinut_websocket.go
Fixed error return in exchange_websocket.go

* Removed comments in coinut_wrapper.go
Refactor orderbook logic from hitbtc_websocket.go to exchange_websocket.go

* General fixes

* Removed comments
General fixes

* Updated routines.go

* After rebase fix

* Fixed update config pairs in okcoin.go

* fixed config currency issue in okcoin.go for okcoin China

* exchange_websocket.go
*Removed unused const dec
*Removed state change routine
*Improved trafficMonitor routine
*Increased verbosity for error returns
*Removed uneeded mutex locks

exchange_websocket_test.go
*Added new tests for websocket and orderbook updating

routines.go
*Removed string cased

* Fixed race conditions on sync.waitgroup in exchanges_websocket.go

* Changes variable name in config.go

* Removes unnecessary comment

* Removes indefinite lock on error return

* Removes unnecessary comment

* Adds support for BTCC websocket
Drops support for BTCC REST

* Rewords comment in exchange_websocket.go
Moves types to poloniex_types.go

* Moves types to coinut_types.go

* Removes uneeded range for accessing array variables for coinbase_websocket.go
Removes comments in coinut_types.go

* Adds verbosity flag to GCT
Suppresses verbose output from routines.go

* Fixes setting proxy for REST and Websocket per exchange
Upgrades error handling
Drops unused *url.Url variable in exchange type

* Adds test for setting proxy

* Fixes bug that closes connection due to incorrect timeout time through a proxy connection

* Clarify verbose flag message

exchanges/binance/binance.go

@@ -22,7 +22,7 @@ type Binance struct {
 	exchange.Base
 	WebsocketConn *websocket.Conn
 
-	// valid string list that a required by the exchange
+	// Valid string list that is required by the exchange
 	validLimits    []int
 	validIntervals []TimeInterval
 }
@@ -61,7 +61,6 @@ func (b *Binance) SetDefaults() {
 	b.Name = "Binance"
 	b.Enabled = false
 	b.Verbose = false
-	b.Websocket = false
 	b.RESTPollingDelay = 10
 	b.RequestCurrencyPairFormat.Delimiter = ""
 	b.RequestCurrencyPairFormat.Uppercase = true
@@ -77,6 +76,7 @@ func (b *Binance) SetDefaults() {
 		common.NewHTTPClientWithTimeout(exchange.DefaultHTTPTimeout))
 	b.APIUrlDefault = apiURL
 	b.APIUrl = b.APIUrlDefault
+	b.WebsocketInit()
 }
 
 // Setup takes in the supplied exchange configuration details and sets params
@@ -91,7 +91,6 @@ func (b *Binance) Setup(exch config.ExchangeConfig) {
 		b.SetHTTPClientUserAgent(exch.HTTPUserAgent)
 		b.RESTPollingDelay = exch.RESTPollingDelay
 		b.Verbose = exch.Verbose
-		b.Websocket = exch.Websocket
 		b.BaseCurrencies = common.SplitStrings(exch.BaseCurrencies, ",")
 		b.AvailablePairs = common.SplitStrings(exch.AvailablePairs, ",")
 		b.EnabledPairs = common.SplitStrings(exch.EnabledPairs, ",")
@@ -111,6 +110,18 @@ func (b *Binance) Setup(exch config.ExchangeConfig) {
 		if err != nil {
 			log.Fatal(err)
 		}
+		err = b.SetClientProxyAddress(exch.ProxyAddress)
+		if err != nil {
+			log.Fatal(err)
+		}
+		err = b.WebsocketSetup(b.WSConnect,
+			exch.Name,
+			exch.Websocket,
+			binanceDefaultWebsocketURL,
+			exch.WebsocketURL)
+		if err != nil {
+			log.Fatal(err)
+		}
 	}
 }
 
@@ -199,6 +210,8 @@ func (b *Binance) GetOrderBook(obd OrderBookDataRequestParams) (OrderBook, error
 			}
 		}
 	}
+
+	orderbook.LastUpdateID = resp.LastUpdateID
 	return orderbook, nil
 }
 

exchanges/binance/binance_types.go

@@ -61,9 +61,10 @@ type OrderBookData struct {
 
 // OrderBook actual structured data that can be used for orderbook
 type OrderBook struct {
-	Code int
-	Msg  string
-	Bids []struct {
+	LastUpdateID int64
+	Code         int
+	Msg          string
+	Bids         []struct {
 		Price    float64
 		Quantity float64
 	}
@@ -73,6 +74,24 @@ type OrderBook struct {
 	}
 }
 
+// DepthUpdateParams is used as an embedded type for WebsocketDepthStream
+type DepthUpdateParams []struct {
+	PriceLevel float64
+	Quantity   float64
+	ingnore    []interface{}
+}
+
+// WebsocketDepthStream is the difference for the update depth stream
+type WebsocketDepthStream struct {
+	Event         string        `json:"e"`
+	Timestamp     int64         `json:"E"`
+	Pair          string        `json:"s"`
+	FirstUpdateID int64         `json:"U"`
+	LastUpdateID  int64         `json:"u"`
+	UpdateBids    []interface{} `json:"b"`
+	UpdateAsks    []interface{} `json:"a"`
+}
+
 // RecentTradeRequestParams represents Klines request data.
 type RecentTradeRequestParams struct {
 	Symbol string `json:"symbol"` // Required field. example LTCBTC, BTCUSDT

exchanges/binance/binance_websocket.go

@@ -1,98 +1,358 @@
 package binance
 
 import (
-	"log"
+	"errors"
+	"fmt"
 	"net/http"
+	"net/url"
+	"strconv"
 	"strings"
+	"sync"
 	"time"
 
 	"github.com/gorilla/websocket"
 	"github.com/thrasher-/gocryptotrader/common"
+	"github.com/thrasher-/gocryptotrader/currency/pair"
+	exchange "github.com/thrasher-/gocryptotrader/exchanges"
+	"github.com/thrasher-/gocryptotrader/exchanges/orderbook"
 )
 
 const (
 	binanceDefaultWebsocketURL = "wss://stream.binance.com:9443"
-	binancePingPeriod          = 20 * time.Second
 )
 
-// WebsocketClient starts and handles the websocket client connection
-func (b *Binance) WebsocketClient() {
-	for b.Enabled && b.Websocket {
-		var Dialer websocket.Dialer
-		var err error
-		// myenabledPairs := strings.ToLower(strings.Replace(strings.Join(b.EnabledPairs, "@ticker/"), "-", "", -1)) + "@trade"
+var lastUpdateID map[string]int64
+var m sync.Mutex
 
-		myenabledPairsTicker := strings.ToLower(strings.Replace(strings.Join(b.EnabledPairs, "@ticker/"), "-", "", -1)) + "@ticker"
-		myenabledPairsTrade := strings.ToLower(strings.Replace(strings.Join(b.EnabledPairs, "@trade/"), "-", "", -1)) + "@trade"
-		myenabledPairsKline := strings.ToLower(strings.Replace(strings.Join(b.EnabledPairs, "@kline_1m/"), "-", "", -1)) + "@kline_1m"
-		wsurl := b.WebsocketURL + "/stream?streams=" + myenabledPairsTicker + "/" + myenabledPairsTrade + "/" + myenabledPairsKline
+// SeedLocalCache seeds depth data
+func (b *Binance) SeedLocalCache(p pair.CurrencyPair) error {
+	var newOrderBook orderbook.Base
 
-		// b.WebsocketConn, _, err = Dialer.Dial(binanceDefaultWebsocketURL+myenabledPairs, http.Header{})
-		b.WebsocketConn, _, err = Dialer.Dial(wsurl, http.Header{})
+	formattedPair := exchange.FormatExchangeCurrency(b.Name, p)
 
+	orderbookNew, err := b.GetOrderBook(
+		OrderBookDataRequestParams{
+			Symbol: formattedPair.String(),
+			Limit:  1000,
+		})
+
+	if err != nil {
+		return err
+	}
+
+	m.Lock()
+	if lastUpdateID == nil {
+		lastUpdateID = make(map[string]int64)
+	}
+
+	lastUpdateID[formattedPair.String()] = orderbookNew.LastUpdateID
+	m.Unlock()
+
+	for _, bids := range orderbookNew.Bids {
+		newOrderBook.Bids = append(newOrderBook.Bids,
+			orderbook.Item{Amount: bids.Quantity, Price: bids.Price})
+	}
+	for _, Asks := range orderbookNew.Asks {
+		newOrderBook.Asks = append(newOrderBook.Asks,
+			orderbook.Item{Amount: Asks.Quantity, Price: Asks.Price})
+	}
+
+	newOrderBook.Pair = pair.NewCurrencyPairFromString(formattedPair.String())
+	newOrderBook.CurrencyPair = formattedPair.String()
+	newOrderBook.LastUpdated = time.Now()
+	newOrderBook.AssetType = "SPOT"
+
+	return b.Websocket.Orderbook.LoadSnapshot(newOrderBook, b.GetName())
+}
+
+// UpdateLocalCache updates and returns the most recent iteration of the orderbook
+func (b *Binance) UpdateLocalCache(ob WebsocketDepthStream) error {
+	m.Lock()
+	ID, ok := lastUpdateID[ob.Pair]
+	if !ok {
+		m.Unlock()
+		return errors.New("binance_websocket.go - Unable to find lastUpdateID")
+	}
+
+	if ob.LastUpdateID+1 <= ID || ID >= ob.LastUpdateID+1 {
+		// Drop update, out of order
+		m.Unlock()
+		return nil
+	}
+
+	lastUpdateID[ob.Pair] = ob.LastUpdateID
+	m.Unlock()
+
+	var updateBid, updateAsk []orderbook.Item
+
+	for _, bidsToUpdate := range ob.UpdateBids {
+		var priceToBeUpdated orderbook.Item
+		for i, bids := range bidsToUpdate.([]interface{}) {
+			switch i {
+			case 0:
+				priceToBeUpdated.Price, _ = strconv.ParseFloat(bids.(string), 64)
+			case 1:
+				priceToBeUpdated.Amount, _ = strconv.ParseFloat(bids.(string), 64)
+			}
+		}
+		updateBid = append(updateBid, priceToBeUpdated)
+	}
+
+	for _, asksToUpdate := range ob.UpdateAsks {
+		var priceToBeUpdated orderbook.Item
+		for i, asks := range asksToUpdate.([]interface{}) {
+			switch i {
+			case 0:
+				priceToBeUpdated.Price, _ = strconv.ParseFloat(asks.(string), 64)
+			case 1:
+				priceToBeUpdated.Amount, _ = strconv.ParseFloat(asks.(string), 64)
+			}
+		}
+		updateAsk = append(updateBid, priceToBeUpdated)
+	}
+
+	updatedTime := time.Unix(ob.Timestamp, 0)
+	currencyPair := pair.NewCurrencyPairFromString(ob.Pair)
+
+	return b.Websocket.Orderbook.Update(updateBid,
+		updateAsk,
+		currencyPair,
+		updatedTime,
+		b.GetName(),
+		"SPOT")
+}
+
+// WSConnect intiates a websocket connection
+func (b *Binance) WSConnect() error {
+	if !b.Websocket.IsEnabled() || !b.IsEnabled() {
+		return errors.New(exchange.WebsocketNotEnabled)
+	}
+
+	var Dialer websocket.Dialer
+	var err error
+
+	ticker := strings.ToLower(
+		strings.Replace(
+			strings.Join(b.EnabledPairs, "@ticker/"), "-", "", -1)) + "@ticker"
+	trade := strings.ToLower(
+		strings.Replace(
+			strings.Join(b.EnabledPairs, "@trade/"), "-", "", -1)) + "@trade"
+	kline := strings.ToLower(
+		strings.Replace(
+			strings.Join(b.EnabledPairs, "@kline_1m/"), "-", "", -1)) + "@kline_1m"
+	depth := strings.ToLower(
+		strings.Replace(
+			strings.Join(b.EnabledPairs, "@depth/"), "-", "", -1)) + "@depth"
+
+	wsurl := b.Websocket.GetWebsocketURL() +
+		"/stream?streams=" +
+		ticker +
+		"/" +
+		trade +
+		"/" +
+		kline +
+		"/" +
+		depth
+
+	if b.Websocket.GetProxyAddress() != "" {
+		url, err := url.Parse(b.Websocket.GetProxyAddress())
 		if err != nil {
-			log.Printf("%s Unable to connect to Websocket. Error: %s\n", b.Name, err)
-			continue
+			return fmt.Errorf("binance_websocket.go - Unable to connect to parse proxy address. Error: %s",
+				err)
 		}
 
-		if b.Verbose {
-			log.Printf("%s Connected to Websocket.\n", b.Name)
-		}
+		Dialer.Proxy = http.ProxyURL(url)
+	}
 
-		for b.Enabled && b.Websocket {
+	for _, ePair := range b.GetEnabledCurrencies() {
+		err := b.SeedLocalCache(ePair)
+		if err != nil {
+			return err
+		}
+	}
+
+	b.WebsocketConn, _, err = Dialer.Dial(wsurl, http.Header{})
+	if err != nil {
+		return fmt.Errorf("binance_websocket.go - Unable to connect to Websocket. Error: %s",
+			err)
+	}
+
+	go b.WsHandleData()
+
+	return nil
+}
+
+// WSReadData reads from the websocket connection
+func (b *Binance) WSReadData() {
+	b.Websocket.Wg.Add(1)
+
+	defer func() {
+		err := b.WebsocketConn.Close()
+		if err != nil {
+			b.Websocket.DataHandler <- fmt.Errorf("binance_websocket.go - Unable to to close Websocket connection. Error: %s",
+				err)
+		}
+		b.Websocket.Wg.Done()
+	}()
+
+	for {
+		select {
+		case <-b.Websocket.ShutdownC:
+			return
+
+		default:
 			msgType, resp, err := b.WebsocketConn.ReadMessage()
 			if err != nil {
-				log.Println(err)
-				break
+				b.Websocket.DataHandler <- fmt.Errorf("binance_websocket.go - Websocket Read Data. Error: %s",
+					err)
+				return
 			}
 
-			switch msgType {
+			b.Websocket.TrafficAlert <- struct{}{}
+			b.Websocket.Intercomm <- exchange.WebsocketResponse{Type: msgType, Raw: resp}
+		}
+	}
+}
+
+// WsHandleData handles websocket data from WsReadData
+func (b *Binance) WsHandleData() {
+	b.Websocket.Wg.Add(1)
+	defer b.Websocket.Wg.Done()
+
+	go b.WSReadData()
+
+	for {
+		select {
+		case <-b.Websocket.ShutdownC:
+			return
+
+		case read := <-b.Websocket.Intercomm:
+			switch read.Type {
 			case websocket.TextMessage:
 				multiStreamData := MultiStreamData{}
 
-				err := common.JSONDecode(resp, &multiStreamData)
-
+				err := common.JSONDecode(read.Raw, &multiStreamData)
 				if err != nil {
-					log.Println("Could not load multi stream data.", string(resp))
+					b.Websocket.DataHandler <- fmt.Errorf("binance_websocket.go - Could not load multi stream data: %s",
+						string(read.Raw))
 					continue
 				}
 
 				if strings.Contains(multiStreamData.Stream, "trade") {
 					trade := TradeStream{}
-					err := common.JSONDecode(multiStreamData.Data, &trade)
 
+					err := common.JSONDecode(multiStreamData.Data, &trade)
 					if err != nil {
-						log.Println("Could not convert to a TradeStream structure")
+						b.Websocket.DataHandler <- fmt.Errorf("binance_websocket.go - Could not unmarshal trade data: %s",
+							err)
 						continue
 					}
-					log.Println("Trade received", trade.Symbol, trade.TimeStamp, trade.TradeID, trade.Price, trade.Quantity)
+
+					price, err := strconv.ParseFloat(trade.Price, 64)
+					if err != nil {
+						b.Websocket.DataHandler <- fmt.Errorf("binance_websocket.go - price conversion error: %s",
+							err)
+						continue
+					}
+
+					amount, err := strconv.ParseFloat(trade.Quantity, 64)
+					if err != nil {
+						b.Websocket.DataHandler <- fmt.Errorf("binance_websocket.go - amount conversion error: %s",
+							err)
+						continue
+					}
+
+					b.Websocket.DataHandler <- exchange.TradeData{
+						CurrencyPair: pair.NewCurrencyPairFromString(trade.Symbol),
+						Timestamp:    time.Unix(0, trade.TimeStamp),
+						Price:        price,
+						Amount:       amount,
+						Exchange:     b.GetName(),
+						AssetType:    "SPOT",
+						Side:         trade.EventType,
+					}
+					continue
+
 				} else if strings.Contains(multiStreamData.Stream, "ticker") {
 					ticker := TickerStream{}
 
 					err := common.JSONDecode(multiStreamData.Data, &ticker)
 					if err != nil {
-						log.Println("Could not convert to a TickerStream structure")
+						b.Websocket.DataHandler <- fmt.Errorf("binance_websocket.go - Could not convert to a TickerStream structure %s",
+							err.Error())
 						continue
 					}
 
-					log.Println("Ticker received", ticker.Symbol, ticker.EventTime, ticker.TotalTradedVolume, ticker.LastTradeID)
+					var wsTicker exchange.TickerData
+
+					wsTicker.Timestamp = time.Unix(0, ticker.EventTime)
+					wsTicker.Pair = pair.NewCurrencyPairFromString(ticker.Symbol)
+					wsTicker.AssetType = "SPOT"
+					wsTicker.Exchange = b.GetName()
+					wsTicker.ClosePrice, _ = strconv.ParseFloat(ticker.CurrDayClose, 64)
+					wsTicker.Quantity, _ = strconv.ParseFloat(ticker.TotalTradedVolume, 64)
+					wsTicker.OpenPrice, _ = strconv.ParseFloat(ticker.OpenPrice, 64)
+					wsTicker.HighPrice, _ = strconv.ParseFloat(ticker.HighPrice, 64)
+					wsTicker.LowPrice, _ = strconv.ParseFloat(ticker.LowPrice, 64)
+
+					b.Websocket.DataHandler <- wsTicker
+					continue
+
 				} else if strings.Contains(multiStreamData.Stream, "kline") {
 					kline := KlineStream{}
 
 					err := common.JSONDecode(multiStreamData.Data, &kline)
 					if err != nil {
-						log.Println("Could not convert to a KlineStream structure")
+						b.Websocket.DataHandler <- fmt.Errorf("binance_websocket.go - Could not convert to a KlineStream structure %s",
+							err)
 						continue
 					}
 
-					log.Println("Kline received", kline.Symbol, kline.EventTime, kline.Kline.HighPrice, kline.Kline.LowPrice)
-				}
-				type MsgType struct {
-					MessageType string `json:"messageType"`
+					var wsKline exchange.KlineData
+
+					wsKline.Timestamp = time.Unix(0, kline.EventTime)
+					wsKline.Pair = pair.NewCurrencyPairFromString(kline.Symbol)
+					wsKline.AssetType = "SPOT"
+					wsKline.Exchange = b.GetName()
+					wsKline.StartTime = time.Unix(0, kline.Kline.StartTime)
+					wsKline.CloseTime = time.Unix(0, kline.Kline.CloseTime)
+					wsKline.Interval = kline.Kline.Interval
+					wsKline.OpenPrice, _ = strconv.ParseFloat(kline.Kline.OpenPrice, 64)
+					wsKline.ClosePrice, _ = strconv.ParseFloat(kline.Kline.ClosePrice, 64)
+					wsKline.HighPrice, _ = strconv.ParseFloat(kline.Kline.HighPrice, 64)
+					wsKline.LowPrice, _ = strconv.ParseFloat(kline.Kline.LowPrice, 64)
+					wsKline.Volume, _ = strconv.ParseFloat(kline.Kline.Volume, 64)
+
+					b.Websocket.DataHandler <- wsKline
+					continue
+
+				} else if common.StringContains(multiStreamData.Stream, "depth") {
+					depth := WebsocketDepthStream{}
+
+					err := common.JSONDecode(multiStreamData.Data, &depth)
+					if err != nil {
+						b.Websocket.DataHandler <- fmt.Errorf("binance_websocket.go - Could not convert to depthStream structure %s",
+							err)
+						continue
+					}
+
+					err = b.UpdateLocalCache(depth)
+					if err != nil {
+						b.Websocket.DataHandler <- fmt.Errorf("binance_websocket.go - UpdateLocalCache error: %s",
+							err)
+						continue
+					}
+
+					currencyPair := pair.NewCurrencyPairFromString(depth.Pair)
+
+					b.Websocket.DataHandler <- exchange.WebsocketOrderbookUpdate{
+						Pair:     currencyPair,
+						Asset:    "SPOT",
+						Exchange: b.GetName(),
+					}
+					continue
 				}
 			}
 		}
-		b.WebsocketConn.Close()
-		log.Printf("%s Websocket client disconnected.", b.Name)
 	}
 }

exchanges/binance/binance_wrapper.go

@@ -24,15 +24,11 @@ func (b *Binance) Start(wg *sync.WaitGroup) {
 // Run implements the OKEX wrapper
 func (b *Binance) Run() {
 	if b.Verbose {
-		log.Printf("%s Websocket: %s. (url: %s).\n", b.GetName(), common.IsEnabled(b.Websocket), b.WebsocketURL)
+		log.Printf("%s Websocket: %s. (url: %s).\n", b.GetName(), common.IsEnabled(b.Websocket.IsEnabled()), b.Websocket.GetWebsocketURL())
 		log.Printf("%s polling delay: %ds.\n", b.GetName(), b.RESTPollingDelay)
 		log.Printf("%s %d currencies enabled: %s.\n", b.GetName(), len(b.EnabledPairs), b.EnabledPairs)
 	}
 
-	if b.Websocket {
-		go b.WebsocketClient()
-	}
-
 	symbols, err := b.GetExchangeValidCurrencyPairs()
 	if err != nil {
 		log.Printf("%s Failed to get exchange info.\n", b.GetName())
@@ -193,3 +189,8 @@ func (b *Binance) WithdrawFiatExchangeFunds(currency pair.CurrencyItem, amount f
 func (b *Binance) WithdrawFiatExchangeFundsToInternationalBank(currency pair.CurrencyItem, amount float64) (string, error) {
 	return "", errors.New("not yet implemented")
 }
+
+// GetWebsocket returns a pointer to the exchange websocket
+func (b *Binance) GetWebsocket() (*exchange.Websocket, error) {
+	return b.Websocket, nil
+}