Merge branch 'master' into engine

2026-06-07 07:26:48 +00:00 · 2019-08-09 15:46:24 +10:00
parent e8c9a9a2a4 2078ba907f
commit 8ecd8a0c16
353 changed files with 4966 additions and 4196 deletions
--- a/exchanges/kraken/README.md
+++ b/exchanges/kraken/README.md
@@ -1,13 +1,13 @@
 # GoCryptoTrader package Kraken

-<img src="https://github.com/thrasher-/gocryptotrader/blob/master/web/src/assets/page-logo.png?raw=true" width="350px" height="350px" hspace="70">
+<img src="https://github.com/thrasher-corp/gocryptotrader/blob/master/web/src/assets/page-logo.png?raw=true" width="350px" height="350px" hspace="70">


-[![Build Status](https://travis-ci.org/thrasher-/gocryptotrader.svg?branch=master)](https://travis-ci.org/thrasher-/gocryptotrader)
-[![Software License](https://img.shields.io/badge/License-MIT-orange.svg?style=flat-square)](https://github.com/thrasher-/gocryptotrader/blob/master/LICENSE)
-[![GoDoc](https://godoc.org/github.com/thrasher-/gocryptotrader?status.svg)](https://godoc.org/github.com/thrasher-/gocryptotrader/exchanges/kraken)
-[![Coverage Status](http://codecov.io/github/thrasher-/gocryptotrader/coverage.svg?branch=master)](http://codecov.io/github/thrasher-/gocryptotrader?branch=master)
-[![Go Report Card](https://goreportcard.com/badge/github.com/thrasher-/gocryptotrader)](https://goreportcard.com/report/github.com/thrasher-/gocryptotrader)
+[![Build Status](https://travis-ci.org/thrasher-corp/gocryptotrader.svg?branch=master)](https://travis-ci.org/thrasher-corp/gocryptotrader)
+[![Software License](https://img.shields.io/badge/License-MIT-orange.svg?style=flat-square)](https://github.com/thrasher-corp/gocryptotrader/blob/master/LICENSE)
+[![GoDoc](https://godoc.org/github.com/thrasher-corp/gocryptotrader?status.svg)](https://godoc.org/github.com/thrasher-corp/gocryptotrader/exchanges/kraken)
+[![Coverage Status](http://codecov.io/github/thrasher-corp/gocryptotrader/coverage.svg?branch=master)](http://codecov.io/github/thrasher-corp/gocryptotrader?branch=master)
+[![Go Report Card](https://goreportcard.com/badge/github.com/thrasher-corp/gocryptotrader)](https://goreportcard.com/report/github.com/thrasher-corp/gocryptotrader)


 This kraken package is part of the GoCryptoTrader codebase.
@@ -26,7 +26,7 @@ Join our slack to discuss all things related to GoCryptoTrader! [GoCryptoTrader

 ### How to enable

-+ [Enable via configuration](https://github.com/thrasher-/gocryptotrader/tree/master/config#enable-exchange-via-config-example)
+ [Enable via configuration](https://github.com/thrasher-corp/gocryptotrader/tree/master/config#enable-exchange-via-config-example)

 + Individual package example below:

@@ -120,12 +120,12 @@ When submitting a PR, please abide by our coding guidelines:

 + Code must adhere to the official Go [formatting](https://golang.org/doc/effective_go.html#formatting) guidelines (i.e. uses [gofmt](https://golang.org/cmd/gofmt/)).
 + Code must be documented adhering to the official Go [commentary](https://golang.org/doc/effective_go.html#commentary) guidelines.
-+ Code must adhere to our [coding style](https://github.com/thrasher-/gocryptotrader/blob/master/doc/coding_style.md).
+ Code must adhere to our [coding style](https://github.com/thrasher-corp/gocryptotrader/blob/master/doc/coding_style.md).
 + Pull requests need to be based on and opened against the `master` branch.

 ## Donations

-<img src="https://github.com/thrasher-/gocryptotrader/blob/master/web/src/assets/donate.png?raw=true" hspace="70">
+<img src="https://github.com/thrasher-corp/gocryptotrader/blob/master/web/src/assets/donate.png?raw=true" hspace="70">

 If this framework helped you in any way, or you would like to support the developers working on it, please donate Bitcoin to:

--- a/exchanges/kraken/kraken.go
+++ b/exchanges/kraken/kraken.go
@@ -9,11 +9,11 @@ import (
 	"strings"
 	"sync"

-	"github.com/gorilla/websocket"
-	"github.com/thrasher-/gocryptotrader/common/crypto"
-	"github.com/thrasher-/gocryptotrader/currency"
-	exchange "github.com/thrasher-/gocryptotrader/exchanges"
-	log "github.com/thrasher-/gocryptotrader/logger"
+	"github.com/thrasher-corp/gocryptotrader/common/crypto"
+	"github.com/thrasher-corp/gocryptotrader/currency"
+	exchange "github.com/thrasher-corp/gocryptotrader/exchanges"
+	"github.com/thrasher-corp/gocryptotrader/exchanges/wshandler"
+	log "github.com/thrasher-corp/gocryptotrader/logger"
 )

 const (
@@ -54,7 +54,7 @@ const (
 // Kraken is the overarching type across the alphapoint package
 type Kraken struct {
 	exchange.Base
-	WebsocketConn *websocket.Conn
+	WebsocketConn *wshandler.WebsocketConnection
 	wsRequestMtx  sync.Mutex
 }

--- a/exchanges/kraken/kraken_test.go
+++ b/exchanges/kraken/kraken_test.go
@@ -2,17 +2,21 @@ package kraken

 import (
 	"fmt"
+	"net/http"
 	"strings"
 	"testing"

-	"github.com/thrasher-/gocryptotrader/common"
-	"github.com/thrasher-/gocryptotrader/config"
-	"github.com/thrasher-/gocryptotrader/currency"
-	exchange "github.com/thrasher-/gocryptotrader/exchanges"
-	"github.com/thrasher-/gocryptotrader/exchanges/sharedtestvalues"
+	"github.com/gorilla/websocket"
+	"github.com/thrasher-corp/gocryptotrader/common"
+	"github.com/thrasher-corp/gocryptotrader/config"
+	"github.com/thrasher-corp/gocryptotrader/currency"
+	exchange "github.com/thrasher-corp/gocryptotrader/exchanges"
+	"github.com/thrasher-corp/gocryptotrader/exchanges/sharedtestvalues"
+	"github.com/thrasher-corp/gocryptotrader/exchanges/wshandler"
 )

 var k Kraken
+var wsSetupRan bool

 // Please add your own APIkeys to do correct due diligence testing.
 const (
@@ -753,3 +757,42 @@ func TestOrderBookOutOfOrder(t *testing.T) {
 		t.Error("Expected out of order orderbook error")
 	}
 }
+
+func setupWsTests(t *testing.T) {
+	if wsSetupRan {
+		return
+	}
+	TestSetDefaults(t)
+	TestSetup(t)
+	if !k.Websocket.IsEnabled() && !k.API.AuthenticatedWebsocketSupport || !areTestAPIKeysSet() {
+		t.Skip(wshandler.WebsocketNotEnabled)
+	}
+	k.Websocket.DataHandler = sharedtestvalues.GetWebsocketInterfaceChannelOverride()
+	k.Websocket.TrafficAlert = sharedtestvalues.GetWebsocketStructChannelOverride()
+	k.WebsocketConn = &wshandler.WebsocketConnection{
+		ExchangeName:         k.Name,
+		URL:                  krakenWSURL,
+		Verbose:              k.Verbose,
+		ResponseMaxLimit:     exchange.DefaultWebsocketResponseMaxLimit,
+		ResponseCheckTimeout: exchange.DefaultWebsocketResponseCheckTimeout,
+	}
+	var dialer websocket.Dialer
+	err := k.WebsocketConn.Dial(&dialer, http.Header{})
+	if err != nil {
+		t.Fatal(err)
+	}
+	go k.WsHandleData()
+	wsSetupRan = true
+}
+
+// TestWebsocketSubscribe tests returning a message with an id
+func TestWebsocketSubscribe(t *testing.T) {
+	setupWsTests(t)
+	err := k.Subscribe(wshandler.WebsocketChannelSubscription{
+		Channel:  defaultSubscribedChannels[0],
+		Currency: currency.NewPairWithDelimiter("XBT", "USD", "/"),
+	})
+	if err != nil {
+		t.Error(err)
+	}
+}
--- a/exchanges/kraken/kraken_types.go
+++ b/exchanges/kraken/kraken_types.go
@@ -1,6 +1,6 @@
 package kraken

-import "github.com/thrasher-/gocryptotrader/currency"
+import "github.com/thrasher-corp/gocryptotrader/currency"

 // TimeResponse type
 type TimeResponse struct {
@@ -395,9 +395,14 @@ type WebsocketSubscriptionEventRequest struct {
 	Subscription WebsocketSubscriptionData `json:"subscription,omitempty"`
 }

+// WebsocketBaseEventRequest Just has an "event" property
+type WebsocketBaseEventRequest struct {
+	Event string `json:"event"` // eg "unsubscribe"
+}
+
 // WebsocketUnsubscribeByChannelIDEventRequest  handles WS unsubscribe events
 type WebsocketUnsubscribeByChannelIDEventRequest struct {
-	Event     string   `json:"event"`           // unsubscribe
+	WebsocketBaseEventRequest
 	RequestID int64    `json:"reqid,omitempty"` // Optional, client originated ID reflected in response message.
 	Pairs     []string `json:"pair,omitempty"`  // Array of currency pairs (pair1,pair2,pair3).
 	ChannelID int64    `json:"channelID,omitempty"`
@@ -412,7 +417,7 @@ type WebsocketSubscriptionData struct {

 // WebsocketEventResponse holds all data response types
 type WebsocketEventResponse struct {
-	Event        string                            `json:"event"`
+	WebsocketBaseEventRequest
 	Status       string                            `json:"status"`
 	Pair         currency.Pair                     `json:"pair,omitempty"`
 	RequestID    int64                             `json:"reqid,omitempty"` // Optional, client originated ID reflected in response message.
--- a/exchanges/kraken/kraken_websocket.go
+++ b/exchanges/kraken/kraken_websocket.go
@@ -1,26 +1,22 @@
 package kraken

 import (
-	"bytes"
-	"compress/flate"
 	"errors"
 	"fmt"
-	"io/ioutil"
 	"math"
 	"net/http"
-	"net/url"
 	"sort"
 	"strconv"
 	"sync"
 	"time"

 	"github.com/gorilla/websocket"
-	"github.com/thrasher-/gocryptotrader/common"
-	"github.com/thrasher-/gocryptotrader/currency"
-	exchange "github.com/thrasher-/gocryptotrader/exchanges"
-	"github.com/thrasher-/gocryptotrader/exchanges/asset"
-	"github.com/thrasher-/gocryptotrader/exchanges/orderbook"
-	log "github.com/thrasher-/gocryptotrader/logger"
+	"github.com/thrasher-corp/gocryptotrader/common"
+	"github.com/thrasher-corp/gocryptotrader/currency"
+	"github.com/thrasher-corp/gocryptotrader/exchanges/asset"
+	"github.com/thrasher-corp/gocryptotrader/exchanges/orderbook"
+	"github.com/thrasher-corp/gocryptotrader/exchanges/wshandler"
+	log "github.com/thrasher-corp/gocryptotrader/logger"
 )

 // List of all websocket channels to subscribe to
@@ -45,9 +41,8 @@ const (
 	krakenWsSpread             = "spread"
 	krakenWsOrderbook          = "book"
 	// Only supported asset type
-	krakenWsAssetType    = "SPOT"
 	orderbookBufferLimit = 3
-	krakenWsRateLimit    = 50 * time.Millisecond
+	krakenWsRateLimit    = 50
 )

 // orderbookMutex Ensures if two entries arrive at once, only one can be processed at a time
@@ -67,53 +62,15 @@ var subscribeToDefaultChannels = true
 // Format [[ticker,but-t4u],[orderbook,nce-btt]]
 var defaultSubscribedChannels = []string{krakenWsTicker, krakenWsTrade, krakenWsOrderbook, krakenWsOHLC, krakenWsSpread}

-// writeToWebsocket sends a message to the websocket endpoint
-func (k *Kraken) writeToWebsocket(message []byte) error {
-	k.wsRequestMtx.Lock()
-	defer k.wsRequestMtx.Unlock()
-	if k.Verbose {
-		log.Debugf(log.ExchangeSys, "%v Sending message to WS: %v",
-			k.Name,
-			string(message))
-	}
-	// Really basic WS rate limit
-	time.Sleep(krakenWsRateLimit)
-	return k.WebsocketConn.WriteMessage(websocket.TextMessage, message)
-}
-
 // WsConnect initiates a websocket connection
 func (k *Kraken) WsConnect() error {
 	if !k.Websocket.IsEnabled() || !k.IsEnabled() {
-		return errors.New(exchange.WebsocketNotEnabled)
+		return errors.New(wshandler.WebsocketNotEnabled)
 	}
-
 	var dialer websocket.Dialer
-	if k.Websocket.GetProxyAddress() != "" {
-		proxy, err := url.Parse(k.Websocket.GetProxyAddress())
-		if err != nil {
-			return err
-		}
-
-		dialer.Proxy = http.ProxyURL(proxy)
-	}
-
-	var err error
-	if k.Verbose {
-		log.Debugf(log.ExchangeSys, "%v Attempting to connect to %v",
-			k.Name,
-			k.Websocket.GetWebsocketURL())
-	}
-	k.WebsocketConn, _, err = dialer.Dial(k.Websocket.GetWebsocketURL(),
-		http.Header{})
+	err := k.WebsocketConn.Dial(&dialer, http.Header{})
 	if err != nil {
-		return fmt.Errorf("%s Unable to connect to Websocket. Error: %s",
-			k.Name,
-			err)
-	}
-	if k.Verbose {
-		log.Debugf(log.ExchangeSys, "%v Successful connection to %v",
-			k.Name,
-			k.Websocket.GetWebsocketURL())
+		return err
 	}
 	go k.WsHandleData()
 	go k.wsPingHandler()
@@ -124,35 +81,6 @@ func (k *Kraken) WsConnect() error {
 	return nil
 }

-// WsReadData reads data from the websocket connection
-func (k *Kraken) WsReadData() (exchange.WebsocketResponse, error) {
-	mType, resp, err := k.WebsocketConn.ReadMessage()
-	if err != nil {
-		return exchange.WebsocketResponse{}, err
-	}
-	k.Websocket.TrafficAlert <- struct{}{}
-	var standardMessage []byte
-	switch mType {
-	case websocket.TextMessage:
-		standardMessage = resp
-
-	case websocket.BinaryMessage:
-		reader := flate.NewReader(bytes.NewReader(resp))
-		standardMessage, err = ioutil.ReadAll(reader)
-		reader.Close()
-		if err != nil {
-			return exchange.WebsocketResponse{}, err
-		}
-	}
-	if k.Verbose {
-		log.Debugf(log.ExchangeSys, "%v Websocket message received: %v",
-			k.Name,
-			string(standardMessage))
-	}
-
-	return exchange.WebsocketResponse{Raw: standardMessage}, nil
-}
-
 // wsPingHandler sends a message "ping" every 27 to maintain the connection to the websocket
 func (k *Kraken) wsPingHandler() {
 	k.Websocket.Wg.Add(1)
@@ -164,14 +92,13 @@ func (k *Kraken) wsPingHandler() {
 		select {
 		case <-k.Websocket.ShutdownC:
 			return
-
 		case <-ticker.C:
-			pingEvent := fmt.Sprintf("{\"event\":\"%v\"}", krakenWsPing)
+			pingEvent := WebsocketBaseEventRequest{Event: krakenWsPing}
 			if k.Verbose {
 				log.Debugf(log.ExchangeSys, "%v sending ping",
 					k.Name)
 			}
-			err := k.writeToWebsocket([]byte(pingEvent))
+			err := k.WebsocketConn.SendMessage(pingEvent)
 			if err != nil {
 				k.Websocket.DataHandler <- err
 			}
@@ -191,18 +118,19 @@ func (k *Kraken) WsHandleData() {
 		case <-k.Websocket.ShutdownC:
 			return
 		default:
-			resp, err := k.WsReadData()
+			resp, err := k.WebsocketConn.ReadMessage()
 			if err != nil {
 				k.Websocket.DataHandler <- fmt.Errorf("%v WsHandleData: %v",
 					k.Name,
 					err)
 				return
 			}
+			k.Websocket.TrafficAlert <- struct{}{}
 			// event response handling
 			var eventResponse WebsocketEventResponse
 			err = common.JSONDecode(resp.Raw, &eventResponse)
 			if err == nil && eventResponse.Event != "" {
-				k.WsHandleEventResponse(&eventResponse)
+				k.WsHandleEventResponse(&eventResponse, resp.Raw)
 				continue
 			}
 			// Data response handling
@@ -260,7 +188,7 @@ func (k *Kraken) WsHandleDataResponse(response WebsocketDataResponse) {
 }

 // WsHandleEventResponse classifies the WS response and sends to appropriate handler
-func (k *Kraken) WsHandleEventResponse(response *WebsocketEventResponse) {
+func (k *Kraken) WsHandleEventResponse(response *WebsocketEventResponse, rawResponse []byte) {
 	switch response.Event {
 	case krakenWsHeartbeat:
 		if k.Verbose {
@@ -286,19 +214,9 @@ func (k *Kraken) WsHandleEventResponse(response *WebsocketEventResponse) {
 				k.Name, krakenWSSupportedVersion, response.WebsocketStatusResponse.Version)
 		}
 	case krakenWsSubscriptionStatus:
-		if k.Verbose {
-			log.Debugf(log.ExchangeSys, "%v Websocket subscription status data received",
-				k.Name)
-		}
+		k.WebsocketConn.AddResponseWithID(response.RequestID, rawResponse)
 		if response.Status != "subscribed" {
-			if response.RequestID > 0 {
-				k.Websocket.DataHandler <- fmt.Errorf("%v requestID: '%v'. Error: %v",
-					k.Name,
-					response.RequestID,
-					response.WebsocketErrorResponse.ErrorMessage)
-			} else {
-				k.Websocket.DataHandler <- fmt.Errorf(response.WebsocketErrorResponse.ErrorMessage)
-			}
+			k.Websocket.DataHandler <- fmt.Errorf("%v %v %v", k.Name, response.RequestID, response.WebsocketErrorResponse.ErrorMessage)
 			return
 		}
 		addNewSubscriptionChannelData(response)
@@ -360,10 +278,10 @@ func (k *Kraken) wsProcessTickers(channelData *WebsocketChannelData, data interf
 	lowPrice, _ := strconv.ParseFloat(lowData[0].(string), 64)
 	quantity, _ := strconv.ParseFloat(volumeData[0].(string), 64)

-	k.Websocket.DataHandler <- exchange.TickerData{
+	k.Websocket.DataHandler <- wshandler.TickerData{
 		Timestamp:  time.Now(),
 		Exchange:   k.Name,
-		AssetType:  krakenWsAssetType,
+		AssetType:  asset.Spot,
 		Pair:       channelData.Pair,
 		ClosePrice: closePrice,
 		OpenPrice:  openPrice,
@@ -406,8 +324,8 @@ func (k *Kraken) wsProcessTrades(channelData *WebsocketChannelData, data interfa
 		price, _ := strconv.ParseFloat(trade[0].(string), 64)
 		amount, _ := strconv.ParseFloat(trade[1].(string), 64)

-		k.Websocket.DataHandler <- exchange.TradeData{
-			AssetType:    krakenWsAssetType,
+		k.Websocket.DataHandler <- wshandler.TradeData{
+			AssetType:    asset.Spot,
 			CurrencyPair: channelData.Pair,
 			EventTime:    time.Now().Unix(),
 			Exchange:     k.Name,
@@ -435,7 +353,7 @@ func (k *Kraken) wsProcessOrderBook(channelData *WebsocketChannelData, data inte
 			if len(orderbookBuffer[channelData.ChannelID]) >= orderbookBufferLimit {
 				err := k.wsProcessOrderBookUpdate(channelData)
 				if err != nil {
-					subscriptionToRemove := exchange.WebsocketChannelSubscription{
+					subscriptionToRemove := wshandler.WebsocketChannelSubscription{
 						Channel:  krakenWsOrderbook,
 						Currency: channelData.Pair,
 					}
@@ -450,7 +368,7 @@ func (k *Kraken) wsProcessOrderBook(channelData *WebsocketChannelData, data inte
 func (k *Kraken) wsProcessOrderBookPartial(channelData *WebsocketChannelData, obData map[string]interface{}) {
 	ob := orderbook.Base{
 		Pair:      channelData.Pair,
-		AssetType: krakenWsAssetType,
+		AssetType: asset.Spot,
 	}
 	// Kraken ob data is timestamped per price, GCT orderbook data is timestamped per entry
 	// Using the highest last update time, we can attempt to respect both within a reasonable degree
@@ -498,9 +416,9 @@ func (k *Kraken) wsProcessOrderBookPartial(channelData *WebsocketChannelData, ob
 		return
 	}

-	k.Websocket.DataHandler <- exchange.WebsocketOrderbookUpdate{
+	k.Websocket.DataHandler <- wshandler.WebsocketOrderbookUpdate{
 		Exchange: k.Name,
-		Asset:    krakenWsAssetType,
+		Asset:    asset.Spot,
 		Pair:     channelData.Pair,
 	}

@@ -512,7 +430,7 @@ func (k *Kraken) wsProcessOrderBookPartial(channelData *WebsocketChannelData, ob

 func (k *Kraken) wsProcessOrderBookBuffer(channelData *WebsocketChannelData, obData map[string]interface{}) {
 	ob := orderbook.Base{
-		AssetType:    krakenWsAssetType,
+		AssetType:    asset.Spot,
 		ExchangeName: k.Name,
 		Pair:         channelData.Pair,
 	}
@@ -624,9 +542,9 @@ func (k *Kraken) wsProcessOrderBookUpdate(channelData *WebsocketChannelData) err
 		return err
 	}

-	k.Websocket.DataHandler <- exchange.WebsocketOrderbookUpdate{
+	k.Websocket.DataHandler <- wshandler.WebsocketOrderbookUpdate{
 		Exchange: k.Name,
-		Asset:    krakenWsAssetType,
+		Asset:    asset.Spot,
 		Pair:     channelData.Pair,
 	}
 	// Reset the buffer
@@ -765,8 +683,8 @@ func (k *Kraken) wsProcessCandles(channelData *WebsocketChannelData, data interf
 	closePrice, _ := strconv.ParseFloat(candleData[5].(string), 64)
 	volume, _ := strconv.ParseFloat(candleData[7].(string), 64)

-	k.Websocket.DataHandler <- exchange.KlineData{
-		AssetType: krakenWsAssetType,
+	k.Websocket.DataHandler <- wshandler.KlineData{
+		AssetType: asset.Spot,
 		Pair:      channelData.Pair,
 		Timestamp: time.Now(),
 		Exchange:  k.Name,
@@ -785,11 +703,11 @@ func (k *Kraken) wsProcessCandles(channelData *WebsocketChannelData, data interf
 // GenerateDefaultSubscriptions Adds default subscriptions to websocket to be handled by ManageSubscriptions()
 func (k *Kraken) GenerateDefaultSubscriptions() {
 	enabledCurrencies := k.GetEnabledPairs(asset.Spot)
-	var subscriptions []exchange.WebsocketChannelSubscription
+	var subscriptions []wshandler.WebsocketChannelSubscription
 	for i := range defaultSubscribedChannels {
 		for j := range enabledCurrencies {
 			enabledCurrencies[j].Delimiter = "/"
-			subscriptions = append(subscriptions, exchange.WebsocketChannelSubscription{
+			subscriptions = append(subscriptions, wshandler.WebsocketChannelSubscription{
 				Channel:  defaultSubscribedChannels[i],
 				Currency: enabledCurrencies[j],
 			})
@@ -799,39 +717,29 @@ func (k *Kraken) GenerateDefaultSubscriptions() {
 }

 // Subscribe sends a websocket message to receive data from the channel
-func (k *Kraken) Subscribe(channelToSubscribe exchange.WebsocketChannelSubscription) error {
+func (k *Kraken) Subscribe(channelToSubscribe wshandler.WebsocketChannelSubscription) error {
 	resp := WebsocketSubscriptionEventRequest{
 		Event: krakenWsSubscribe,
 		Pairs: []string{channelToSubscribe.Currency.String()},
 		Subscription: WebsocketSubscriptionData{
 			Name: channelToSubscribe.Channel,
 		},
+		RequestID: k.WebsocketConn.GenerateMessageID(true),
 	}
-	json, err := common.JSONEncode(resp)
-	if err != nil {
-		if k.Verbose {
-			log.Errorf(log.ExchangeSys, "%v subscribe error: %v", k.Name, err)
-		}
-		return err
-	}
-	return k.writeToWebsocket(json)
+	_, err := k.WebsocketConn.SendMessageReturnResponse(resp.RequestID, resp)
+	return err
 }

 // Unsubscribe sends a websocket message to stop receiving data from the channel
-func (k *Kraken) Unsubscribe(channelToSubscribe exchange.WebsocketChannelSubscription) error {
+func (k *Kraken) Unsubscribe(channelToSubscribe wshandler.WebsocketChannelSubscription) error {
 	resp := WebsocketSubscriptionEventRequest{
 		Event: krakenWsUnsubscribe,
 		Pairs: []string{channelToSubscribe.Currency.String()},
 		Subscription: WebsocketSubscriptionData{
 			Name: channelToSubscribe.Channel,
 		},
+		RequestID: k.WebsocketConn.GenerateMessageID(true),
 	}
-	json, err := common.JSONEncode(resp)
-	if err != nil {
-		if k.Verbose {
-			log.Errorf(log.ExchangeSys, "%v unsubscribe error: %v", k.Name, err)
-		}
-		return err
-	}
-	return k.writeToWebsocket(json)
+	_, err := k.WebsocketConn.SendMessageReturnResponse(resp.RequestID, resp)
+	return err
 }
--- a/exchanges/kraken/kraken_wrapper.go
+++ b/exchanges/kraken/kraken_wrapper.go
@@ -7,15 +7,16 @@ import (
 	"sync"
 	"time"

-	"github.com/thrasher-/gocryptotrader/common"
-	"github.com/thrasher-/gocryptotrader/config"
-	"github.com/thrasher-/gocryptotrader/currency"
-	exchange "github.com/thrasher-/gocryptotrader/exchanges"
-	"github.com/thrasher-/gocryptotrader/exchanges/asset"
-	"github.com/thrasher-/gocryptotrader/exchanges/orderbook"
-	"github.com/thrasher-/gocryptotrader/exchanges/request"
-	"github.com/thrasher-/gocryptotrader/exchanges/ticker"
-	log "github.com/thrasher-/gocryptotrader/logger"
+	"github.com/thrasher-corp/gocryptotrader/common"
+	"github.com/thrasher-corp/gocryptotrader/config"
+	"github.com/thrasher-corp/gocryptotrader/currency"
+	exchange "github.com/thrasher-corp/gocryptotrader/exchanges"
+	"github.com/thrasher-corp/gocryptotrader/exchanges/asset"
+	"github.com/thrasher-corp/gocryptotrader/exchanges/orderbook"
+	"github.com/thrasher-corp/gocryptotrader/exchanges/request"
+	"github.com/thrasher-corp/gocryptotrader/exchanges/ticker"
+	"github.com/thrasher-corp/gocryptotrader/exchanges/wshandler"
+	log "github.com/thrasher-corp/gocryptotrader/logger"
 )

 // GetDefaultConfig returns a default exchange config
@@ -91,15 +92,17 @@ func (k *Kraken) SetDefaults() {

 	k.API.Endpoints.URLDefault = krakenAPIURL
 	k.API.Endpoints.URL = k.API.Endpoints.URLDefault
-
-	k.WebsocketInit()
+	k.Websocket = wshandler.New()
 	k.API.Endpoints.WebsocketURL = krakenWSURL
-	k.Websocket.Functionality = exchange.WebsocketTickerSupported |
-		exchange.WebsocketTradeDataSupported |
-		exchange.WebsocketKlineSupported |
-		exchange.WebsocketOrderbookSupported |
-		exchange.WebsocketSubscribeSupported |
-		exchange.WebsocketUnsubscribeSupported
+	k.Websocket.Functionality = wshandler.WebsocketTickerSupported |
+		wshandler.WebsocketTradeDataSupported |
+		wshandler.WebsocketKlineSupported |
+		wshandler.WebsocketOrderbookSupported |
+		wshandler.WebsocketSubscribeSupported |
+		wshandler.WebsocketUnsubscribeSupported |
+		wshandler.WebsocketMessageCorrelationSupported
+	k.WebsocketResponseMaxLimit = exchange.DefaultWebsocketResponseMaxLimit
+	k.WebsocketResponseCheckTimeout = exchange.DefaultWebsocketResponseCheckTimeout
 }

 // Setup sets current exchange configuration
@@ -114,14 +117,29 @@ func (k *Kraken) Setup(exch *config.ExchangeConfig) error {
 		return err
 	}

-	return k.WebsocketSetup(k.WsConnect,
+	err = k.Websocket.Setup(k.WsConnect,
 		k.Subscribe,
 		k.Unsubscribe,
 		exch.Name,
 		exch.Features.Enabled.Websocket,
 		exch.Verbose,
 		krakenWSURL,
-		exch.API.Endpoints.WebsocketURL)
+		exch.API.Endpoints.WebsocketURL,
+		exch.API.AuthenticatedWebsocketSupport)
+	if err != nil {
+		return err
+	}
+
+	k.WebsocketConn = &wshandler.WebsocketConnection{
+		ExchangeName:         k.Name,
+		URL:                  k.Websocket.GetWebsocketURL(),
+		ProxyURL:             k.Websocket.GetProxyAddress(),
+		Verbose:              k.Verbose,
+		RateLimit:            krakenWsRateLimit,
+		ResponseCheckTimeout: exch.WebsocketResponseCheckTimeout,
+		ResponseMaxLimit:     exch.WebsocketResponseMaxLimit,
+	}
+	return nil
 }

 // Start starts the Kraken go routine
@@ -424,7 +442,7 @@ func (k *Kraken) WithdrawFiatFundsToInternationalBank(withdrawRequest *exchange.
 }

 // GetWebsocket returns a pointer to the exchange websocket
-func (k *Kraken) GetWebsocket() (*exchange.Websocket, error) {
+func (k *Kraken) GetWebsocket() (*wshandler.Websocket, error) {
 	return k.Websocket, nil
 }

@@ -516,20 +534,20 @@ func (k *Kraken) GetOrderHistory(getOrdersRequest *exchange.GetOrdersRequest) ([

 // SubscribeToWebsocketChannels appends to ChannelsToSubscribe
 // which lets websocket.manageSubscriptions handle subscribing
-func (k *Kraken) SubscribeToWebsocketChannels(channels []exchange.WebsocketChannelSubscription) error {
+func (k *Kraken) SubscribeToWebsocketChannels(channels []wshandler.WebsocketChannelSubscription) error {
 	k.Websocket.SubscribeToChannels(channels)
 	return nil
 }

 // UnsubscribeToWebsocketChannels removes from ChannelsToSubscribe
 // which lets websocket.manageSubscriptions handle unsubscribing
-func (k *Kraken) UnsubscribeToWebsocketChannels(channels []exchange.WebsocketChannelSubscription) error {
-	k.Websocket.UnsubscribeToChannels(channels)
+func (k *Kraken) UnsubscribeToWebsocketChannels(channels []wshandler.WebsocketChannelSubscription) error {
+	k.Websocket.RemoveSubscribedChannels(channels)
 	return nil
 }

 // GetSubscriptions returns a copied list of subscriptions
-func (k *Kraken) GetSubscriptions() ([]exchange.WebsocketChannelSubscription, error) {
+func (k *Kraken) GetSubscriptions() ([]wshandler.WebsocketChannelSubscription, error) {
 	return k.Websocket.GetSubscriptions(), nil
 }