Support dynamic CCXT headers for exchanges such as Coinbase (#314)

* 1 - add infrastructure for HeaderFn to networking lib and wiring into ccxt.go

* 2 - make header function method with STATIC type and ability to pass in custom mappings to framework with wiring for CCXT

* 3 - support backward compatible case of not having any pre-specified function int he exchange header value, added LIST_OF_HACKS.md

* 4 - add dynamic header functions to support coinbase pro

* 5 - base64 encode signature for coinbase pro, use COINBASEPRO prefix instead of COINBASE

* 6 - show all function names when there is an error in MakeHeaderFn

* 7 - update sample config files with sample config entries for coinbase, mark it as a tested exchange

* 8 - updated Exchanges section of README.md

Author: nikhilsaraf

LIST_OF_HACKS.md

@@ -0,0 +1,11 @@
+# LIST OF HACKS
+
+## Awating v2.0
+
+Incomplete list of hacks in the codebase that should be fixed before upgrading to v2.0 which will change the API to Kelp in some way
+
+- LOH-1 - support backward-compatible case of not having any pre-specified function
+
+## Workarounds
+
+Incomplete list of workaround hacks in the codebase that should be fixed at some point

README.md

@@ -208,11 +208,13 @@ For more details, check out the [examples section](#configuration-files-1) of th
 
 Exchange integrations provide data to trading strategies and allow you to [hedge][hedge] your positions on different exchanges. The following [exchange integrations](plugins) are available **out of the box** with Kelp:
 
-- sdex ([source](plugins/sdex.go)): The [Stellar Decentralized Exchange][sdex]
-- kraken ([source](plugins/krakenExchange.go)): [Kraken][kraken]
-- binance (_`"ccxt-binance"`_) ([source](plugins/ccxtExchange.go)): Binance via CCXT - only supports priceFeeds and mirroring (buysell, sell, and mirror strategy)
-- poloniex (_`"ccxt-poloniex"`_) ([source](plugins/ccxtExchange.go)): Poloniex via CCXT - only supports priceFeeds and mirroring (buysell, sell, and mirror strategy)
-- bittrex (_`"ccxt-bittrex"`_) ([source](plugins/ccxtExchange.go)): Bittrex via CCXT - only supports priceFeeds and mirroring (buysell, sell, and mirror strategy)
+- sdex (_`"sdex"`_) ([source](plugins/sdex.go)): The [Stellar Decentralized Exchange][sdex]
+- kraken (_`"kraken"`_) ([source](plugins/krakenExchange.go)): [Kraken][kraken] - recommended to use `ccxt-kraken` instead
+- kraken (via CCXT) (_`"ccxt-kraken"`_) ([source](plugins/ccxtExchange.go)): Kraken via CCXT - full two-way integration (tested)
+- binance (via CCXT) (_`"ccxt-binance"`_) ([source](plugins/ccxtExchange.go)): Binance via CCXT - full two-way integration (tested)
+- coinbasepro (via CCXT) (_`"ccxt-coinbasepro"`_) ([source](plugins/ccxtExchange.go)): Coinbase Pro via CCXT - full two-way integration (tested)
+- poloniex (via CCXT) (_`"ccxt-poloniex"`_) ([source](plugins/ccxtExchange.go)): Poloniex via CCXT - only tested on priceFeeds and one-way mirroring
+- bittrex (via CCXT) (_`"ccxt-bittrex"`_) ([source](plugins/ccxtExchange.go)): Bittrex via CCXT - only tested on priceFeeds and onw-way mirroring
 
 ## Plugins
 

examples/configs/trader/sample_trader.cfg

@@ -113,16 +113,26 @@ MAX_OP_FEE_STROOPS=5000
 # if your exchange requires additional parameters during initialization, list them here (only ccxt supported currently)
 # Note that some CCXT exchanges require additional parameters, e.g. coinbase pro requires a "password"
 #[[EXCHANGE_PARAMS]]
-#PARAM=""
-#VALUE=""
+#PARAM="password"
+#VALUE="<coinbasepro-api-passphrase-here>"
 #[[EXCHANGE_PARAMS]]
 #PARAM=""
 #VALUE=""
 
 # if your exchange requires additional parameters as http headers, list them here (only ccxt supported currently)
+# e.g., coinbase pro requires CB-ACCESS-KEY, CB-ACCESS-SIGN, CB-ACCESS-TIMESTAMP, and CB-ACCESS-PASSPHRASE
 #[[EXCHANGE_HEADERS]]
-#HEADER=""
-#VALUE=""
+#HEADER="CB-ACCESS-KEY"
+#VALUE="STATIC:<coinbasepro-api-key-here>"
+#[[EXCHANGE_HEADERS]]
+#HEADER="CB-ACCESS-SIGN"
+#VALUE="COINBASEPRO__CB-ACCESS-SIGN:<coinbasepro-api-secret-here>"
+#[[EXCHANGE_HEADERS]]
+#HEADER="CB-ACCESS-TIMESTAMP"
+#VALUE="TIMESTAMP:"             # leave the value as "TIMESTAMP:" for coinbasepro
+#[[EXCHANGE_HEADERS]]
+#HEADER="CB-ACCESS-PASSPHRASE"
+#VALUE="STATIC:<coinbasepro-passphrase-here>"
 #[[EXCHANGE_HEADERS]]
 #HEADER=""
 #VALUE=""

plugins/factory.go

@@ -180,7 +180,9 @@ func getExchanges() map[string]ExchangeContainer {
 func loadExchanges() {
 	// marked as tested if key exists in this map (regardless of bool value)
 	testedCcxtExchanges := map[string]bool{
-		"binance": true,
+		"kraken":      true,
+		"binance":     true,
+		"coinbasepro": true,
 	}
 
 	exchanges = &map[string]ExchangeContainer{

support/networking/headerFn.go

@@ -0,0 +1,68 @@
+package networking
+
+import (
+	"fmt"
+	"strings"
+
+	"github.com/stellar/kelp/support/utils"
+)
+
+// HeaderFn represents a function that transforms headers
+type HeaderFn func(string, string, string) string // (string httpMethod, string requestPath, string body)
+
+// makeStaticHeaderFn is a convenience method
+func makeStaticHeaderFn(value string) (HeaderFn, error) {
+	// need to convert to HeaderFn to work as a api.ExchangeHeader.Value
+	return HeaderFn(func(method string, requestPath string, body string) string {
+		return value
+	}), nil
+}
+
+// HeaderFnFactory is a factory method for the HeaderFn
+type HeaderFnFactory func(string) (HeaderFn, error)
+
+var defaultMappings = map[string]HeaderFnFactory{
+	"STATIC": HeaderFnFactory(makeStaticHeaderFn),
+}
+
+func headerFnNames(maps ...map[string]HeaderFnFactory) []string {
+	names := []string{}
+	for _, m := range maps {
+		if m != nil {
+			for k, _ := range m {
+				names = append(names, k)
+			}
+		}
+	}
+	return utils.Dedupe(names)
+}
+
+// MakeHeaderFn is a factory method that makes a HeaderFn
+func MakeHeaderFn(value string, primaryMappings map[string]HeaderFnFactory) (HeaderFn, error) {
+	numSeparators := strings.Count(value, ":")
+
+	if numSeparators == 0 {
+		// LOH-1 - support backward-compatible case of not having any pre-specified function
+		return makeStaticHeaderFn(value)
+	} else if numSeparators != 1 {
+		names := headerFnNames(primaryMappings, defaultMappings)
+		return nil, fmt.Errorf("invalid format of header value (%s), needs exactly one colon (:) to separate the header function from the input value to that function. list of available header functions: [%s]", value, strings.Join(names, ", "))
+	}
+
+	valueParts := strings.Split(value, ":")
+	fnType := valueParts[0]
+	fnInputValue := valueParts[1]
+
+	if primaryMappings != nil {
+		if makeHeaderFn, ok := primaryMappings[fnType]; ok {
+			return makeHeaderFn(fnInputValue)
+		}
+	}
+
+	if makeHeaderFn, ok := defaultMappings[fnType]; ok {
+		return makeHeaderFn(fnInputValue)
+	}
+
+	names := headerFnNames(primaryMappings, defaultMappings)
+	return nil, fmt.Errorf("invalid function prefix (%s) as part of header value (%s). list of available header functions: [%s]", fnType, value, strings.Join(names, ", "))
+}

support/networking/network.go

@@ -11,6 +11,32 @@ import (
 	"strings"
 )
 
+// JSONRequestDynamicHeaders submits an HTTP web request and parses the response into the responseData object as JSON
+func JSONRequestDynamicHeaders(
+	httpClient *http.Client,
+	method string,
+	reqURL string,
+	data string,
+	headers map[string]HeaderFn,
+	responseData interface{}, // the passed in responseData should be a pointer
+	errorKey string,
+) error {
+	headersMap := map[string]string{}
+	for header, fn := range headers {
+		headersMap[header] = fn(method, reqURL, data)
+	}
+
+	return JSONRequest(
+		httpClient,
+		method,
+		reqURL,
+		data,
+		headersMap,
+		responseData,
+		errorKey,
+	)
+}
+
 // JSONRequest submits an HTTP web request and parses the response into the responseData object as JSON
 func JSONRequest(
 	httpClient *http.Client,

support/sdk/ccxt.go

@@ -36,7 +36,7 @@ type Ccxt struct {
 	exchangeName string
 	instanceName string
 	markets      map[string]CcxtMarket
-	headersMap   map[string]string
+	headersMap   map[string]networking.HeaderFn
 }
 
 // CcxtMarket represents the result of a LoadMarkets call
@@ -162,9 +162,14 @@ func (c *Ccxt) initialize(apiKey api.ExchangeAPIKey, params []api.ExchangeParam,
 	}
 	c.markets = markets
 
-	headersMap := map[string]string{}
+	headersMap := map[string]networking.HeaderFn{}
+	ccxtHeaderMappings := makeHeaderMappingsFromNewTimestamp()
 	for _, header := range headers {
-		headersMap[header.Header] = header.Value
+		headerFn, e := networking.MakeHeaderFn(header.Value, ccxtHeaderMappings)
+		if e != nil {
+			return fmt.Errorf("unable to make header function with key (%s) and value (%s): %s", header.Header, header.Value, e)
+		}
+		headersMap[header.Header] = headerFn
 	}
 	c.headersMap = headersMap
 
@@ -216,7 +221,7 @@ func (c *Ccxt) newInstance(apiKey api.ExchangeAPIKey, params []api.ExchangeParam
 	}
 
 	var newInstance map[string]interface{}
-	e = networking.JSONRequest(c.httpClient, "POST", ccxtBaseURL+pathExchanges+"/"+c.exchangeName, string(jsonData), c.headersMap, &newInstance, "error")
+	e = networking.JSONRequestDynamicHeaders(c.httpClient, "POST", ccxtBaseURL+pathExchanges+"/"+c.exchangeName, string(jsonData), c.headersMap, &newInstance, "error")
 	if e != nil {
 		return fmt.Errorf("error in web request when creating new exchange instance for exchange '%s': %s", c.exchangeName, e)
 	}
@@ -233,7 +238,7 @@ func (c *Ccxt) symbolExists(tradingPair string) error {
 	url := ccxtBaseURL + pathExchanges + "/" + c.exchangeName + "/" + c.instanceName
 	// decode generic data (see "https://blog.golang.org/json-and-go#TOC_4.")
 	var exchangeOutput interface{}
-	e := networking.JSONRequest(c.httpClient, "GET", url, "", c.headersMap, &exchangeOutput, "error")
+	e := networking.JSONRequestDynamicHeaders(c.httpClient, "GET", url, "", c.headersMap, &exchangeOutput, "error")
 	if e != nil {
 		return fmt.Errorf("error fetching details of exchange instance (exchange=%s, instanceName=%s): %s", c.exchangeName, c.instanceName, e)
 	}
@@ -284,7 +289,7 @@ func (c *Ccxt) FetchTicker(tradingPair string) (map[string]interface{}, error) {
 	url := ccxtBaseURL + pathExchanges + "/" + c.exchangeName + "/" + c.instanceName + "/fetchTicker"
 	// decode generic data (see "https://blog.golang.org/json-and-go#TOC_4.")
 	var output interface{}
-	e = networking.JSONRequest(c.httpClient, "POST", url, string(data), c.headersMap, &output, "error")
+	e = networking.JSONRequestDynamicHeaders(c.httpClient, "POST", url, string(data), c.headersMap, &output, "error")
 	if e != nil {
 		return nil, fmt.Errorf("error fetching tickers for trading pair '%s': %s", tradingPair, e)
 	}
@@ -324,7 +329,7 @@ func (c *Ccxt) FetchOrderBook(tradingPair string, limit *int) (map[string][]Ccxt
 	url := ccxtBaseURL + pathExchanges + "/" + c.exchangeName + "/" + c.instanceName + "/fetchOrderBook"
 	// decode generic data (see "https://blog.golang.org/json-and-go#TOC_4.")
 	var output interface{}
-	e = networking.JSONRequest(c.httpClient, "POST", url, string(data), c.headersMap, &output, "error")
+	e = networking.JSONRequestDynamicHeaders(c.httpClient, "POST", url, string(data), c.headersMap, &output, "error")
 	if e != nil {
 		return nil, fmt.Errorf("error fetching orderbook for trading pair '%s': %s", tradingPair, e)
 	}
@@ -385,7 +390,7 @@ func (c *Ccxt) FetchTrades(tradingPair string) ([]CcxtTrade, error) {
 	url := ccxtBaseURL + pathExchanges + "/" + c.exchangeName + "/" + c.instanceName + "/fetchTrades"
 	// decode generic data (see "https://blog.golang.org/json-and-go#TOC_4.")
 	output := []CcxtTrade{}
-	e = networking.JSONRequest(c.httpClient, "POST", url, string(data), c.headersMap, &output, "error")
+	e = networking.JSONRequestDynamicHeaders(c.httpClient, "POST", url, string(data), c.headersMap, &output, "error")
 	if e != nil {
 		return nil, fmt.Errorf("error fetching trades for trading pair '%s': %s", tradingPair, e)
 	}
@@ -418,7 +423,7 @@ func (c *Ccxt) FetchMyTrades(tradingPair string, limit int, maybeCursorStart int
 	url := ccxtBaseURL + pathExchanges + "/" + c.exchangeName + "/" + c.instanceName + "/fetchMyTrades"
 	// decode generic data (see "https://blog.golang.org/json-and-go#TOC_4.")
 	output := []CcxtTrade{}
-	e = networking.JSONRequest(c.httpClient, "POST", url, string(data), c.headersMap, &output, "error")
+	e = networking.JSONRequestDynamicHeaders(c.httpClient, "POST", url, string(data), c.headersMap, &output, "error")
 	if e != nil {
 		return nil, fmt.Errorf("error fetching trades for trading pair '%s': %s", tradingPair, e)
 	}
@@ -437,7 +442,7 @@ func (c *Ccxt) FetchBalance() (map[string]CcxtBalance, error) {
 	url := ccxtBaseURL + pathExchanges + "/" + c.exchangeName + "/" + c.instanceName + "/fetchBalance"
 	// decode generic data (see "https://blog.golang.org/json-and-go#TOC_4.")
 	var output interface{}
-	e := networking.JSONRequest(c.httpClient, "POST", url, "", c.headersMap, &output, "error")
+	e := networking.JSONRequestDynamicHeaders(c.httpClient, "POST", url, "", c.headersMap, &output, "error")
 	if e != nil {
 		return nil, fmt.Errorf("error fetching balance: %s", e)
 	}
@@ -503,7 +508,7 @@ func (c *Ccxt) FetchOpenOrders(tradingPairs []string) (map[string][]CcxtOpenOrde
 	url := ccxtBaseURL + pathExchanges + "/" + c.exchangeName + "/" + c.instanceName + "/fetchOpenOrders"
 	// decode generic data (see "https://blog.golang.org/json-and-go#TOC_4.")
 	var output interface{}
-	e = networking.JSONRequest(c.httpClient, "POST", url, string(data), c.headersMap, &output, "error")
+	e = networking.JSONRequestDynamicHeaders(c.httpClient, "POST", url, string(data), c.headersMap, &output, "error")
 	if e != nil {
 		return nil, fmt.Errorf("error fetching open orders: %s", e)
 	}
@@ -559,7 +564,7 @@ func (c *Ccxt) CreateLimitOrder(tradingPair string, side string, amount float64,
 	url := ccxtBaseURL + pathExchanges + "/" + c.exchangeName + "/" + c.instanceName + "/createOrder"
 	// decode generic data (see "https://blog.golang.org/json-and-go#TOC_4.")
 	var output interface{}
-	e = networking.JSONRequest(c.httpClient, "POST", url, string(data), c.headersMap, &output, "error")
+	e = networking.JSONRequestDynamicHeaders(c.httpClient, "POST", url, string(data), c.headersMap, &output, "error")
 	if e != nil {
 		return nil, fmt.Errorf("error creating order: %s", e)
 	}
@@ -598,7 +603,7 @@ func (c *Ccxt) CancelOrder(orderID string, tradingPair string) (*CcxtOpenOrder,
 	url := ccxtBaseURL + pathExchanges + "/" + c.exchangeName + "/" + c.instanceName + "/cancelOrder"
 	// decode generic data (see "https://blog.golang.org/json-and-go#TOC_4.")
 	var output interface{}
-	e = networking.JSONRequest(c.httpClient, "POST", url, string(data), c.headersMap, &output, "error")
+	e = networking.JSONRequestDynamicHeaders(c.httpClient, "POST", url, string(data), c.headersMap, &output, "error")
 	if e != nil {
 		return nil, fmt.Errorf("error canceling order: %s", e)
 	}

support/sdk/headerFnMappings.go

@@ -0,0 +1,56 @@
+package sdk
+
+import (
+	"crypto/hmac"
+	"crypto/sha256"
+	"encoding/base64"
+	"fmt"
+	"strconv"
+	"strings"
+	"time"
+
+	"github.com/stellar/kelp/support/networking"
+)
+
+type ccxtMapper struct {
+	timestamp int64
+}
+
+// makeHeaderMappingsFromNewTimestamp creates a new ccxtMapper so the timestamp can be consistent across HeaderFns and returns the required map
+func makeHeaderMappingsFromNewTimestamp() map[string]networking.HeaderFnFactory {
+	c := &ccxtMapper{
+		timestamp: time.Now().Unix(),
+	}
+
+	return map[string]networking.HeaderFnFactory{
+		"COINBASEPRO__CB-ACCESS-SIGN": networking.HeaderFnFactory(c.coinbaseSignFn),
+		"TIMESTAMP":                   networking.HeaderFnFactory(c.timestampFn),
+	}
+}
+
+func (c *ccxtMapper) coinbaseSignFn(base64EncodedSigningKey string) (networking.HeaderFn, error) {
+	base64DecodedSigningKey, e := base64.StdEncoding.DecodeString(base64EncodedSigningKey)
+	if e != nil {
+		return nil, fmt.Errorf("could not decode signing key (%s): %s", base64EncodedSigningKey, e)
+	}
+
+	// return this inline method casted as a HeaderFn to work as a headerValue
+	return networking.HeaderFn(func(method string, requestPath string, body string) string {
+		uppercaseMethod := strings.ToUpper(method)
+		payload := fmt.Sprintf("%d%s%s%s", c.timestamp, uppercaseMethod, requestPath, body)
+
+		// sign
+		mac := hmac.New(sha256.New, base64DecodedSigningKey)
+		mac.Write([]byte(payload))
+		signature := mac.Sum(nil)
+		base64EncodedSignature := base64.StdEncoding.EncodeToString(signature)
+
+		return base64EncodedSignature
+	}), nil
+}
+
+func (c *ccxtMapper) timestampFn(_ string) (networking.HeaderFn, error) {
+	return networking.HeaderFn(func(method string, requestPath string, body string) string {
+		return strconv.FormatInt(c.timestamp, 10)
+	}), nil
+}

support/utils/functions.go

@@ -355,3 +355,17 @@ func StringSet(list []string) map[string]bool {
 	}
 	return m
 }
+
+// Dedupe removes duplicates from the list
+func Dedupe(list []string) []string {
+	seen := map[string]bool{}
+	out := []string{}
+
+	for _, elem := range list {
+		if _, ok := seen[elem]; !ok {
+			out = append(out, elem)
+			seen[elem] = true
+		}
+	}
+	return out
+}