d0zingcat/gocryptotrader
1
0
Fork 0
mirror of https://github.com/d0zingcat/gocryptotrader.git synced 2026-06-06 07:26:47 +00:00
Code Issues Packages Projects Releases Wiki Activity

Feature: Websocket order handling (#446)

Browse Source 
* Initial changes, removing exchange name as an arg and puts it in the pointer struct. Adds case to ws routines

* Adds CancelAllOrders func, adds GetByExchangeAndID. Adds modify handler in routines.go

* initial poor attempts to have bitmex work with new datahandler handlers. fixes ordersides

* bitmex Completes new order

* Better bitmex handling, but not complete. Begins a gargantuan task of unifying order data structs. Sometimes an order update will contain lot's of information, so its best to be able to update all fields of our orders, rather than just an arbitrary subset. As a result, everything will be broken for the foreseeable future :glitch_crab:

* Removes old order handler which did nothing. Updates order properties for everything everywhere - now consistent. Changes order status. Adds asset type and wallet address to all order types

* Adds order updater to update only relevant fields since the object is generic, we don't know what fields are passed from what exchanges. Adds "lastupdated" field to order.Detail. Expands order cancellation for engine orders.

* Ensures that new orders are added to the ordermanager's order store. Saaa many comments. Internalises orderStore get func. Adds internalOrderID to orderdetail and adds websocket support for it

* Fixes a cancelAllOrders oopsie doopsie

* Adds potential func to update orderdetails from an orderdetail struct. Unsure if will keep.

* Begins btcmarkets implementation. Expands order "stringToOrder" funcs to allow for some more flexible string coversions. Removes order.Submit via websocket as it would cause unlimited order place issues :D

* Finishes btc markets without testing

* Adds untested ws auth func to btse

* Finises btse, fixes btcmarkets bug

* Adds coinbasepro support

* Fixes a few more fields in coinbase pro and readds the extra subs

* Begins work on coinbene. Plus theyve added a new ws connection yeee

* Wasted a bunch of time adding support to an additional websocket that isn't needed ;_; Fixed a bug in coinbasepro. Fully kitted out coinbene support. Updates order types with all fields

* Removes extra websocket connection ;_;

* Finishes gemini. Fixes order side unknown

* Adds okgroup support. Moves byte reading to another function to allow for unit testing. Updates routines to use pointers. Updates date update handling for order details

* Finishes order data for okgroup websocket, but starts the STRANGE process of converting all other websocket endpoints to be a little less silly

* Cleans up okroup websocket implementation. Fixes bug in Gemini

* Adds poloniex support. Updates ws order handling

* new bitmex support. Adds some tests now that its all in its own func. Fixes poloniex bug

* Begins work on authenticated binance websocket

* Attempts to track user data via binance websocket

* Maybe finishes Binance websocket support

* Begins adding test coverage to orders.go. Updates names of script properties to match updated

* Begins an experiment with code coverage. Fixes more rebase issues

* Completes orders coverage. Botches a few other things though. Fixes more scripting stuff

* All tests in engine package pass

* Adds some loevely routine tests

* Moves ordermanager to test Bot ordermanager
Adds lovely routine tests to ensure things that get sent to be handled the data handler are handled by the data handler by handling them

* Replaces "wsHandleData" with "wsReadData" as that's what its going to do now.

* Splits all wsHandleData into wsReadData and wsHandleData to allow for easy testing via sending []byte json examples to test proper functionality. Breaks so many tests

* Fixes majority of test issues. But data races which are tough on the engine package

* "Fixes" test by removing shutdown test. It interferes with too many things. Requires some thought

* Tests all the binance websocket points

* Adds better bitfinex websocket support.

* Adds testing for bitfinex, bitstamp and btcmarkets. Fixes websocket bugs encountered

* Adds BTSE ws tests. Fixes bugs in ws

* Adds coinbase pro tests. Fixes any issues

* Coinbene tests

* Starts to handle coinut. Runs into a problem conceptually regarding websocket roundtrip and orders. Both events need to happen without impacting eachother/racing

* Addresses a data race issue regarding websocket and bot order management submission - order submission locks at an earlier point to prevent routines.go from creating an order before order submission creates it. Updates rpcserver to use order management bot to submit orders.

* Finishes the hectic coinut testing

* Adds tests for gateio

* Fixes rebase issues. Updates tests to work without being overloaded

* Begins testing of gemini. fixes up minor issues

* ginishes gemini tests and fixes

* Adds hitbtc tests. Fixes all the many issues with hitbtc websocket

* Adds remaining tests. Increases default test channel limit again

* Begins work towards huobi tests

* Finishes huobi tests

* Fixed all mythical rebase adventures

* Begins kraken transformation

* Finishes kraken. Fixes coinbene leverage now that its changed

* Begins okgroup testing

* Adds okgroup ws tests

* Does some poloniex

* Fixes basic curreny issue by extracting to func

* Begins redesign of poloniex websocket datahandling. Completes authenticated handling, now onto unauth

* Finishes poloniex revision

* Finishes ZB additions

* Fixes data races

* Fixes rebase issues. Fixes bad kraken logic

* Fixes after reviewing code

* lint everywhere

* Fixes lingering lints

* lint

* Adds test coverage to order detail and modify updating

* Fixes linting

* Fixes huge int, fixes date tests

* Adds GetByExchange, adds test for it. Protects fakepass echange. Renames DisplayQty to DisplayQuantity. Removes verbose. Adds some websocket properties.  Updates bitmex asset type in test

* Addresses timestamps, type abbreviations, verbosity. Expands binance kline switch cases. Updates some websocket capabilities.

* Adds coverage to the stringToOrderType/Status functions introduced in PR

* Minor fixes addressing some time, error text and use of StringDataCompareInsensitive

* Introduces shiny new system which checks if there is an awaiting ID, if found, processes via wrapper method, else, goes through wsHandleData method. Removes weird locking system from wrapper/websocket data race. Updates bitfinex to properly handle websocket order requests and notifications

* Moves fakePassingExchange to test_helper. Fixes some order side implementations for trades. Botches a new error type

* Adds new error type to track and handle order classification errors separately

* Fully fleshes out ClassificationError for all instances of status conversion. Even in order trades and some wrapper functions

* Introduces common.SimpleTimeFormat for "2006-01-02 15:04:05". Fixes binance and bitfinex issues with auth endpoint use, map casting. Expands more order.ClassificationError usage. Fixes some more generic websocket response errors

* Future proofs order updating by utilising asset types. Expands testing to accomodate. Adds shiny new time type. Expands wrapper websocket functionality definitions

* minty linty

* Broken end of day code addressing basic nits on comments, returns and currency conversion

* Adds testing to btcmarkets websocket. Also updates websocket orderbook to use update instead

* Fixes fun rebase fun fun so fun

* Addresses minor nits regarding changed interface and comments

* Creates new function `GetRequestFormattedPairAndAssetType` to retrieve a currency pair and asset type based on a string. It will iterate over enabled pairs and compare them to formatted pairs and then return that pair if found.

* Fixes test

* Adds a single line to the end of the file, because that would be really bad if it wasn't there

* Updates fakepassexchange to not use params, updates test params, uses fatal in some tests where its important, updates order manager to have a rwmutex, removes some returns, improves ws key test for binance, updates properties to reflect their actual values, adds some more websocket properties

* Addresses binance switch linting

* Updates leverage property to int64

* Fixes what was broken
This commit is contained in:
Scott
2020-03-03 13:32:14 +11:00
committed by GitHub
parent 9d49184bc6
commit b686cf2e0e
142 changed files with 13867 additions and 6043 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
exchanges/gemini/gemini_live_test.go
View File

@@ -34,6 +34,8 @@ func TestMain(m *testing.M) {
log.Fatal("Gemini setup error", err)
}
g.API.Endpoints.URL = geminiSandboxAPIURL
g.Websocket.DataHandler = sharedtestvalues.GetWebsocketInterfaceChannelOverride()
g.Websocket.TrafficAlert = sharedtestvalues.GetWebsocketStructChannelOverride()
log.Printf(sharedtestvalues.LiveTesting, g.Name, g.API.Endpoints.URL)
os.Exit(m.Run())
}

3
exchanges/gemini/gemini_mock_test.go
View File

@@ -45,7 +45,8 @@ func TestMain(m *testing.M) {
g.HTTPClient = newClient
g.API.Endpoints.URL = serverDetails
g.Websocket.DataHandler = sharedtestvalues.GetWebsocketInterfaceChannelOverride()
g.Websocket.TrafficAlert = sharedtestvalues.GetWebsocketStructChannelOverride()
log.Printf(sharedtestvalues.MockTesting, g.Name, g.API.Endpoints.URL)
os.Exit(m.Run())
}

518
exchanges/gemini/gemini_test.go
View File

@@ -349,8 +349,8 @@ func TestFormatWithdrawPermissions(t *testing.T) {
func TestGetActiveOrders(t *testing.T) {
t.Parallel()
var getOrdersRequest = order.GetOrdersRequest{
OrderType: order.AnyType,
Currencies: []currency.Pair{
Type: order.AnyType,
Pairs: []currency.Pair{
currency.NewPair(currency.LTC, currency.BTC),
},
}
@@ -369,8 +369,8 @@ func TestGetActiveOrders(t *testing.T) {
func TestGetOrderHistory(t *testing.T) {
t.Parallel()
var getOrdersRequest = order.GetOrdersRequest{
OrderType: order.AnyType,
Currencies: []currency.Pair{currency.NewPair(currency.LTC, currency.BTC)},
Type: order.AnyType,
Pairs: []currency.Pair{currency.NewPair(currency.LTC, currency.BTC)},
}
_, err := g.GetOrderHistory(&getOrdersRequest)
@@ -402,11 +402,11 @@ func TestSubmitOrder(t *testing.T) {
Base: currency.LTC,
Quote: currency.BTC,
},
OrderSide: order.Buy,
OrderType: order.Limit,
Price: 10,
Amount: 1,
ClientID: "1234234",
Side: order.Buy,
Type: order.Limit,
Price: 10,
Amount: 1,
ClientID: "1234234",
}
response, err := g.SubmitOrder(orderSubmission)
@@ -426,7 +426,7 @@ func TestCancelExchangeOrder(t *testing.T) {
t.Skip("API keys set, canManipulateRealOrders false, skipping test")
}
var orderCancellation = &order.Cancel{
OrderID: "266029865",
ID: "266029865",
}
err := g.CancelOrder(orderCancellation)
@@ -448,10 +448,10 @@ func TestCancelAllExchangeOrders(t *testing.T) {
currencyPair := currency.NewPair(currency.LTC, currency.BTC)
var orderCancellation = &order.Cancel{
OrderID: "1",
ID: "1",
WalletAddress: core.BitcoinDonationAddress,
AccountID: "1",
CurrencyPair: currencyPair,
Pair: currencyPair,
}
resp, err := g.CancelAllOrders(orderCancellation)
@@ -553,7 +553,7 @@ func TestWsAuth(t *testing.T) {
t.Skip(wshandler.WebsocketNotEnabled)
}
var dialer websocket.Dialer
go g.WsHandleData()
go g.wsReadData()
err := g.WsSecureSubscribe(&dialer, geminiWsOrderEvents)
if err != nil {
t.Error(err)
@@ -569,3 +569,495 @@ func TestWsAuth(t *testing.T) {
}
timer.Stop()
}
func TestWsMissingRole(t *testing.T) {
pressXToJSON := []byte(`{
"result":"error",
"reason":"MissingRole",
"message":"To access this endpoint, you need to log in to the website and go to the settings page to assign one of these roles [FundManager] to API key wujB3szN54gtJ4QDhqRJ which currently has roles [Trader]"
}`)
err := g.wsHandleData(pressXToJSON, currency.NewPairFromString("BTCUSD"))
if err == nil {
t.Error("Expected error")
}
}
func TestWsOrderEventSubscriptionResponse(t *testing.T) {
pressXToJSON := []byte(`[ {
"type" : "accepted",
"order_id" : "372456298",
"event_id" : "372456299",
"client_order_id": "20170208_example",
"api_session" : "AeRLptFXoYEqLaNiRwv8",
"symbol" : "btcusd",
"side" : "buy",
"order_type" : "exchange limit",
"timestamp" : "1478203017",
"timestampms" : 1478203017455,
"is_live" : true,
"is_cancelled" : false,
"is_hidden" : false,
"avg_execution_price" : "0",
"original_amount" : "14.0296",
"price" : "1059.54"
} ]`)
err := g.wsHandleData(pressXToJSON, currency.NewPairFromString("BTCUSD"))
if err != nil {
t.Error(err)
}
pressXToJSON = []byte(`[{
"type": "accepted",
"order_id": "109535951",
"event_id": "109535952",
"api_session": "UI",
"symbol": "btcusd",
"side": "buy",
"order_type": "exchange limit",
"timestamp": "1547742904",
"timestampms": 1547742904989,
"is_live": true,
"is_cancelled": false,
"is_hidden": false,
"original_amount": "1",
"price": "3592.00",
"socket_sequence": 13
}]`)
err = g.wsHandleData(pressXToJSON, currency.NewPairFromString("BTCUSD"))
if err != nil {
t.Error(err)
}
pressXToJSON = []byte(`[{
"type": "accepted",
"order_id": "109964529",
"event_id": "109964530",
"api_session": "UI",
"symbol": "btcusd",
"side": "buy",
"order_type": "market buy",
"timestamp": "1547756076",
"timestampms": 1547756076644,
"is_live": false,
"is_cancelled": false,
"is_hidden": false,
"total_spend": "200.00",
"socket_sequence": 29
}]`)
err = g.wsHandleData(pressXToJSON, currency.NewPairFromString("BTCUSD"))
if err != nil {
t.Error(err)
}
pressXToJSON = []byte(`[{
"type": "accepted",
"order_id": "109964616",
"event_id": "109964617",
"api_session": "UI",
"symbol": "btcusd",
"side": "sell",
"order_type": "market sell",
"timestamp": "1547756893",
"timestampms": 1547756893937,
"is_live": true,
"is_cancelled": false,
"is_hidden": false,
"original_amount": "25",
"socket_sequence": 26
}]`)
err = g.wsHandleData(pressXToJSON, currency.NewPairFromString("BTCUSD"))
if err != nil {
t.Error(err)
}
pressXToJSON = []byte(`[ {
"type" : "accepted",
"order_id" : "6321",
"event_id" : "6322",
"api_session" : "UI",
"symbol" : "btcusd",
"side" : "sell",
"order_type" : "block_trade",
"timestamp" : "1478204198",
"timestampms" : 1478204198989,
"is_live" : true,
"is_cancelled" : false,
"is_hidden" : true,
"avg_execution_price" : "0",
"original_amount" : "500",
"socket_sequence" : 32307
} ]`)
err = g.wsHandleData(pressXToJSON, currency.NewPairFromString("BTCUSD"))
if err != nil {
t.Error(err)
}
}
func TestWsSubAck(t *testing.T) {
pressXToJSON := []byte(`{
"type": "subscription_ack",
"accountId": 5365,
"subscriptionId": "ws-order-events-5365-b8bk32clqeb13g9tk8p0",
"symbolFilter": [
"btcusd"
],
"apiSessionFilter": [
"UI"
],
"eventTypeFilter": [
"fill",
"closed"
]
}`)
err := g.wsHandleData(pressXToJSON, currency.NewPairFromString("BTCUSD"))
if err != nil {
t.Error(err)
}
}
func TestWsHeartbeat(t *testing.T) {
pressXToJSON := []byte(`{
"type": "heartbeat",
"timestampms": 1547742998508,
"sequence": 31,
"trace_id": "b8biknoqppr32kc7gfgg",
"socket_sequence": 37
}`)
err := g.wsHandleData(pressXToJSON, currency.NewPairFromString("BTCUSD"))
if err != nil {
t.Error(err)
}
}
func TestWsUnsubscribe(t *testing.T) {
pressXToJSON := []byte(`{
"type": "unsubscribe",
"subscriptions": [{
"name": "l2",
"symbols": [
"BTCUSD",
"ETHBTC"
]},
{"name": "candles_1m",
"symbols": [
"BTCUSD",
"ETHBTC"
]}
]
}`)
err := g.wsHandleData(pressXToJSON, currency.NewPairFromString("BTCUSD"))
if err != nil {
t.Error(err)
}
}
func TestWsTradeData(t *testing.T) {
pressXToJSON := []byte(`{
"type": "update",
"eventId": 5375547515,
"timestamp": 1547760288,
"timestampms": 1547760288001,
"socket_sequence": 15,
"events": [
{
"type": "trade",
"tid": 5375547515,
"price": "3632.54",
"amount": "0.1362819142",
"makerSide": "ask"
}
]
}`)
err := g.wsHandleData(pressXToJSON, currency.NewPairFromString("BTCUSD"))
if err != nil {
t.Error(err)
}
}
func TestWsAuctionData(t *testing.T) {
pressXToJSON := []byte(`{
"eventId": 371469414,
"socket_sequence":4009,
"timestamp":1486501200,
"timestampms":1486501200000,
"events": [
{
"amount": "1406",
"makerSide": "auction",
"price": "1048.75",
"tid": 371469414,
"type": "trade"
},
{
"auction_price": "1048.75",
"auction_quantity": "1406",
"eid": 371469414,
"highest_bid_price": "1050.98",
"lowest_ask_price": "1050.99",
"result": "success",
"time_ms": 1486501200000,
"type": "auction_result"
}
],
"type": "update"
}`)
err := g.wsHandleData(pressXToJSON, currency.NewPairFromString("BTCUSD"))
if err != nil {
t.Error(err)
}
}
func TestWsBlockTrade(t *testing.T) {
pressXToJSON := []byte(`{
"type":"update",
"eventId":1111597035,
"socket_sequence":8,
"timestamp":1501175027,
"timestampms":1501175027304,
"events":[
{
"type":"block_trade",
"tid":1111597035,
"price":"10100.00",
"amount":"1000"
}
]
}`)
err := g.wsHandleData(pressXToJSON, currency.NewPairFromString("BTCUSD"))
if err != nil {
t.Error(err)
}
}
func TestWsCandles(t *testing.T) {
pressXToJSON := []byte(`{
"type": "candles_15m_updates",
"symbol": "BTCUSD",
"changes": [
[
1561054500000,
9350.18,
9358.35,
9350.18,
9355.51,
2.07
],
[
1561053600000,
9357.33,
9357.33,
9350.18,
9350.18,
1.5900161
]
]
}`)
err := g.wsHandleData(pressXToJSON, currency.NewPairFromString("BTCUSD"))
if err != nil {
t.Error(err)
}
}
func TestWsAuctions(t *testing.T) {
pressXToJSON := []byte(`{
"eventId": 372481811,
"socket_sequence":23,
"timestamp": 1486591200,
"timestampms": 1486591200000,
"events": [
{
"auction_open_ms": 1486591200000,
"auction_time_ms": 1486674000000,
"first_indicative_ms": 1486673400000,
"last_cancel_time_ms": 1486673985000,
"type": "auction_open"
}
],
"type": "update"
}`)
err := g.wsHandleData(pressXToJSON, currency.NewPairFromString("BTCUSD"))
if err != nil {
t.Error(err)
}
pressXToJSON = []byte(`{
"type": "update",
"eventId": 2248762586,
"timestamp": 1510865640,
"timestampms": 1510865640122,
"socket_sequence": 177,
"events": [
{
"type": "auction_indicative",
"eid": 2248762586,
"result": "success",
"time_ms": 1510865640000,
"highest_bid_price": "7730.69",
"lowest_ask_price": "7730.7",
"collar_price": "7730.695",
"indicative_price": "7750",
"indicative_quantity": "45.43325086"
}
]
}`)
err = g.wsHandleData(pressXToJSON, currency.NewPairFromString("BTCUSD"))
if err != nil {
t.Error(err)
}
pressXToJSON = []byte(`{
"type": "update",
"eventId": 2248795680,
"timestamp": 1510866000,
"timestampms": 1510866000095,
"socket_sequence": 2920,
"events": [
{
"type": "trade",
"tid": 2248795680,
"price": "7763.23",
"amount": "55.95",
"makerSide": "auction"
},
{
"type": "auction_result",
"eid": 2248795680,
"result": "success",
"time_ms": 1510866000000,
"highest_bid_price": "7769",
"lowest_ask_price": "7769.01",
"collar_price": "7769.005",
"auction_price": "7763.23",
"auction_quantity": "55.95"
}
]
}`)
err = g.wsHandleData(pressXToJSON, currency.NewPairFromString("BTCUSD"))
if err != nil {
t.Error(err)
}
}
func TestWsMarketData(t *testing.T) {
pressXToJSON := []byte(`{
"type": "update",
"eventId": 5375461993,
"socket_sequence": 0,
"events": [
{
"type": "change",
"reason": "initial",
"price": "3641.61",
"delta": "0.83372051",
"remaining": "0.83372051",
"side": "bid"
},
{
"type": "change",
"reason": "initial",
"price": "3641.62",
"delta": "4.072",
"remaining": "4.072",
"side": "ask"
}
]
} `)
err := g.wsHandleData(pressXToJSON, currency.NewPairFromString("BTCUSD"))
if err != nil {
t.Error(err)
}
pressXToJSON = []byte(`{
"type": "update",
"eventId": 5375461993,
"socket_sequence": 0,
"events": [
{
"type": "change",
"reason": "initial",
"price": "3641.61",
"delta": "0.83372051",
"remaining": "0.83372051",
"side": "bid"
},
{
"type": "change",
"reason": "initial",
"price": "3641.62",
"delta": "4.072",
"remaining": "4.072",
"side": "ask"
}
]
} `)
err = g.wsHandleData(pressXToJSON, currency.NewPairFromString("BTCUSD"))
if err != nil {
t.Error(err)
}
pressXToJSON = []byte(`{
"type": "update",
"eventId": 5375503736,
"timestamp": 1547759964,
"timestampms": 1547759964051,
"socket_sequence": 2,
"events": [
{
"type": "change",
"side": "bid",
"price": "3628.01",
"remaining": "0",
"delta": "-2",
"reason": "cancel"
}
]
} `)
err = g.wsHandleData(pressXToJSON, currency.NewPairFromString("BTCUSD"))
if err != nil {
t.Error(err)
}
}
func TestResponseToStatus(t *testing.T) {
type TestCases struct {
Case string
Result order.Status
}
testCases := []TestCases{
{Case: "accepted", Result: order.New},
{Case: "booked", Result: order.Active},
{Case: "fill", Result: order.Filled},
{Case: "cancelled", Result: order.Cancelled},
{Case: "cancel_rejected", Result: order.Rejected},
{Case: "closed", Result: order.Filled},
{Case: "LOL", Result: order.UnknownStatus},
}
for i := range testCases {
result, _ := stringToOrderStatus(testCases[i].Case)
if result != testCases[i].Result {
t.Errorf("Exepcted: %v, received: %v", testCases[i].Result, result)
}
}
}
func TestResponseToOrderType(t *testing.T) {
type TestCases struct {
Case string
Result order.Type
}
testCases := []TestCases{
{Case: "exchange limit", Result: order.Limit},
{Case: "auction-only limit", Result: order.Limit},
{Case: "indication-of-interest limit", Result: order.Limit},
{Case: "market buy", Result: order.Market},
{Case: "market sell", Result: order.Market},
{Case: "block_trade", Result: order.Market},
{Case: "LOL", Result: order.UnknownType},
}
for i := range testCases {
result, _ := stringToOrderType(testCases[i].Case)
if result != testCases[i].Result {
t.Errorf("Exepcted: %v, received: %v", testCases[i].Result, result)
}
}
}

162
exchanges/gemini/gemini_types.go
View File

@@ -275,87 +275,29 @@ type WsHeartbeatResponse struct {
SocketSequence int64 `json:"socket_sequence"`
}
// WsActiveOrdersResponse contains active orders
type WsActiveOrdersResponse struct {
Type string `json:"type"`
OrderID string `json:"order_id"`
APISession string `json:"api_session"`
Symbol currency.Pair `json:"symbol"`
Side string `json:"side"`
OrderType string `json:"order_type"`
Timestamp string `json:"timestamp"`
Timestampms int64 `json:"timestampms"`
IsLive bool `json:"is_live"`
IsCancelled bool `json:"is_cancelled"`
IsHidden bool `json:"is_hidden"`
AvgExecutionPrice float64 `json:"avg_execution_price,string"`
ExecutedAmount float64 `json:"executed_amount,string"`
RemainingAmount float64 `json:"remaining_amount,string"`
OriginalAmount float64 `json:"original_amount,string"`
Price float64 `json:"price,string"`
SocketSequence int64 `json:"socket_sequence"`
}
// WsOrderRejectedResponse ws response
type WsOrderRejectedResponse struct {
Type string `json:"type"`
OrderID string `json:"order_id"`
EventID string `json:"event_id"`
Reason string `json:"reason"`
APISession string `json:"api_session"`
Symbol currency.Pair `json:"symbol"`
Side string `json:"side"`
OrderType string `json:"order_type"`
Timestamp string `json:"timestamp"`
Timestampms int64 `json:"timestampms"`
IsLive bool `json:"is_live"`
OriginalAmount float64 `json:"original_amount,string"`
Price float64 `json:"price,string"`
SocketSequence int64 `json:"socket_sequence"`
}
// WsOrderBookedResponse ws response
type WsOrderBookedResponse struct {
Type string `json:"type"`
OrderID string `json:"order_id"`
EventID string `json:"event_id"`
APISession string `json:"api_session"`
Symbol currency.Pair `json:"symbol"`
Side string `json:"side"`
OrderType string `json:"order_type"`
Timestamp string `json:"timestamp"`
Timestampms int64 `json:"timestampms"`
IsLive bool `json:"is_live"`
IsCancelled bool `json:"is_cancelled"`
IsHidden bool `json:"is_hidden"`
AvgExecutionPrice float64 `json:"avg_execution_price,string"`
ExecutedAmount float64 `json:"executed_amount,string"`
RemainingAmount float64 `json:"remaining_amount,string"`
OriginalAmount float64 `json:"original_amount,string"`
Price float64 `json:"price,string"`
SocketSequence int64 `json:"socket_sequence"`
}
// WsOrderFilledResponse ws response
type WsOrderFilledResponse struct {
Type string `json:"type"`
OrderID string `json:"order_id"`
APISession string `json:"api_session"`
Symbol currency.Pair `json:"symbol"`
Side string `json:"side"`
OrderType string `json:"order_type"`
Timestamp string `json:"timestamp"`
Timestampms int64 `json:"timestampms"`
// WsOrderResponse contains active orders
type WsOrderResponse struct {
IsLive bool `json:"is_live"`
IsCancelled bool `json:"is_cancelled"`
IsHidden bool `json:"is_hidden"`
SocketSequence int64 `json:"socket_sequence"`
Timestampms int64 `json:"timestampms"`
AvgExecutionPrice float64 `json:"avg_execution_price,string"`
ExecutedAmount float64 `json:"executed_amount,string"`
RemainingAmount float64 `json:"remaining_amount,string"`
OriginalAmount float64 `json:"original_amount,string"`
Price float64 `json:"price,string"`
EventID string `json:"event_id"`
CancelCommandID string `json:"cancel_command_id"`
Reason string `json:"reason"`
Type string `json:"type"`
OrderID string `json:"order_id"`
APISession string `json:"api_session"`
Symbol string `json:"symbol"`
Side string `json:"side"`
OrderType string `json:"order_type"`
Timestamp string `json:"timestamp"`
Fill WsOrderFilledData `json:"fill"`
SocketSequence int64 `json:"socket_sequence"`
}
// WsOrderFilledData ws response data
@@ -368,72 +310,16 @@ type WsOrderFilledData struct {
FeeCurrency string `json:"fee_currency"`
}
// WsOrderCancelledResponse ws response
type WsOrderCancelledResponse struct {
Type string `json:"type"`
OrderID string `json:"order_id"`
EventID string `json:"event_id"`
CancelCommandID string `json:"cancel_command_id,omitempty"`
Reason string `json:"reason"`
APISession string `json:"api_session"`
Symbol currency.Pair `json:"symbol"`
Side string `json:"side"`
OrderType string `json:"order_type"`
Timestamp string `json:"timestamp"`
Timestampms int64 `json:"timestampms"`
IsLive bool `json:"is_live"`
IsCancelled bool `json:"is_cancelled"`
IsHidden bool `json:"is_hidden"`
AvgExecutionPrice float64 `json:"avg_execution_price,string"`
ExecutedAmount float64 `json:"executed_amount,string"`
RemainingAmount float64 `json:"remaining_amount,string"`
OriginalAmount float64 `json:"original_amount,string"`
Price float64 `json:"price,string"`
SocketSequence int64 `json:"socket_sequence"`
type wsUnsubscribeResponse struct {
Type string `json:"type"`
Subscriptions []struct {
Name string `json:"name"`
Symbols []string `json:"symbols"`
} `json:"subscriptions"`
}
// WsOrderCancellationRejectedResponse ws response
type WsOrderCancellationRejectedResponse struct {
Type string `json:"type"`
OrderID string `json:"order_id"`
EventID string `json:"event_id"`
CancelCommandID string `json:"cancel_command_id"`
Reason string `json:"reason"`
APISession string `json:"api_session"`
Symbol currency.Pair `json:"symbol"`
Side string `json:"side"`
OrderType string `json:"order_type"`
Timestamp string `json:"timestamp"`
Timestampms int64 `json:"timestampms"`
IsLive bool `json:"is_live"`
IsCancelled bool `json:"is_cancelled"`
IsHidden bool `json:"is_hidden"`
AvgExecutionPrice float64 `json:"avg_execution_price,string"`
ExecutedAmount float64 `json:"executed_amount,string"`
RemainingAmount float64 `json:"remaining_amount,string"`
OriginalAmount float64 `json:"original_amount,string"`
Price float64 `json:"price,string"`
SocketSequence int64 `json:"socket_sequence"`
}
// WsOrderClosedResponse ws response
type WsOrderClosedResponse struct {
Type string `json:"type"`
OrderID string `json:"order_id"`
EventID string `json:"event_id"`
APISession string `json:"api_session"`
Symbol currency.Pair `json:"symbol"`
Side string `json:"side"`
OrderType string `json:"order_type"`
Timestamp string `json:"timestamp"`
Timestampms int64 `json:"timestampms"`
IsLive bool `json:"is_live"`
IsCancelled bool `json:"is_cancelled"`
IsHidden bool `json:"is_hidden"`
AvgExecutionPrice float64 `json:"avg_execution_price,string"`
ExecutedAmount float64 `json:"executed_amount,string"`
RemainingAmount float64 `json:"remaining_amount,string"`
OriginalAmount float64 `json:"original_amount,string"`
Price float64 `json:"price,string"`
SocketSequence int64 `json:"socket_sequence"`
type wsCandleResponse struct {
Type string `json:"type"`
Symbol string `json:"symbol"`
Changes [][]float64 `json:"changes"`
}

308
exchanges/gemini/gemini_websocket.go
View File

@@ -51,7 +51,7 @@ func (g *Gemini) WsConnect() error {
dialer.Proxy = http.ProxyURL(proxy)
}
go g.WsHandleData()
go g.wsReadData()
err := g.WsSecureSubscribe(&dialer, geminiWsOrderEvents)
if err != nil {
log.Errorf(log.ExchangeSys, "%v - authentication failed: %v\n", g.Name, err)
@@ -82,7 +82,7 @@ func (g *Gemini) WsSubscribe(dialer *websocket.Dialer) error {
return fmt.Errorf("%v Websocket connection %v error. Error %v",
g.Name, endpoint, err)
}
go g.WsReadData(connection, enabledCurrencies[i])
go g.wsFunnelConnectionData(connection, enabledCurrencies[i])
if len(enabledCurrencies)-1 == i {
return nil
}
@@ -126,13 +126,12 @@ func (g *Gemini) WsSecureSubscribe(dialer *websocket.Dialer, url string) error {
if err != nil {
return fmt.Errorf("%v Websocket connection %v error. Error %v", g.Name, endpoint, err)
}
go g.WsReadData(g.AuthenticatedWebsocketConn, currency.Pair{})
go g.wsFunnelConnectionData(g.AuthenticatedWebsocketConn, currency.Pair{})
return nil
}
// WsReadData reads from the websocket connection and returns the websocket
// response
func (g *Gemini) WsReadData(ws *wshandler.WebsocketConnection, c currency.Pair) {
// wsFunnelConnectionData receives data from multiple connections and passes it to wsReadData
func (g *Gemini) wsFunnelConnectionData(ws *wshandler.WebsocketConnection, c currency.Pair) {
g.Websocket.Wg.Add(1)
defer g.Websocket.Wg.Done()
for {
@@ -151,9 +150,8 @@ func (g *Gemini) WsReadData(ws *wshandler.WebsocketConnection, c currency.Pair)
}
}
// WsHandleData handles all the websocket data coming from the websocket
// connection
func (g *Gemini) WsHandleData() {
// wsReadData receives and passes on websocket messages for processing
func (g *Gemini) wsReadData() {
g.Websocket.Wg.Add(1)
defer g.Websocket.Wg.Done()
for {
@@ -165,98 +163,199 @@ func (g *Gemini) WsHandleData() {
if string(resp.Raw) == "[]" {
continue
}
var result map[string]interface{}
err := json.Unmarshal(resp.Raw, &result)
err := g.wsHandleData(resp.Raw, resp.Currency)
if err != nil {
g.Websocket.DataHandler <- fmt.Errorf("%v Error: %v, Raw: %v", g.Name, err, string(resp.Raw))
continue
}
switch result["type"] {
case "subscription_ack":
var result WsSubscriptionAcknowledgementResponse
err := json.Unmarshal(resp.Raw, &result)
if err != nil {
g.Websocket.DataHandler <- err
continue
}
g.Websocket.DataHandler <- result
case "initial":
var result WsSubscriptionAcknowledgementResponse
err := json.Unmarshal(resp.Raw, &result)
if err != nil {
g.Websocket.DataHandler <- err
continue
}
g.Websocket.DataHandler <- result
case "accepted":
var result WsActiveOrdersResponse
err := json.Unmarshal(resp.Raw, &result)
if err != nil {
g.Websocket.DataHandler <- err
continue
}
g.Websocket.DataHandler <- result
case "booked":
var result WsOrderBookedResponse
err := json.Unmarshal(resp.Raw, &result)
if err != nil {
g.Websocket.DataHandler <- err
continue
}
g.Websocket.DataHandler <- result
case "fill":
var result WsOrderFilledResponse
err := json.Unmarshal(resp.Raw, &result)
if err != nil {
g.Websocket.DataHandler <- err
continue
}
g.Websocket.DataHandler <- result
case "cancelled":
var result WsOrderCancelledResponse
err := json.Unmarshal(resp.Raw, &result)
if err != nil {
g.Websocket.DataHandler <- err
continue
}
g.Websocket.DataHandler <- result
case "closed":
var result WsOrderClosedResponse
err := json.Unmarshal(resp.Raw, &result)
if err != nil {
g.Websocket.DataHandler <- err
continue
}
g.Websocket.DataHandler <- result
case "heartbeat":
var result WsHeartbeatResponse
err := json.Unmarshal(resp.Raw, &result)
if err != nil {
g.Websocket.DataHandler <- err
continue
}
g.Websocket.DataHandler <- result
case "update":
if resp.Currency.IsEmpty() {
g.Websocket.DataHandler <- fmt.Errorf("%v - unhandled data %s",
g.Name, resp.Raw)
continue
}
var marketUpdate WsMarketUpdateResponse
err := json.Unmarshal(resp.Raw, &marketUpdate)
if err != nil {
g.Websocket.DataHandler <- err
continue
}
g.wsProcessUpdate(marketUpdate, resp.Currency)
default:
g.Websocket.DataHandler <- fmt.Errorf("%v - unhandled data %s",
g.Name, resp.Raw)
g.Websocket.DataHandler <- err
}
}
}
}
func (g *Gemini) wsHandleData(respRaw []byte, curr currency.Pair) error {
// only order details are sent in arrays
if strings.HasPrefix(string(respRaw), "[") {
var result []WsOrderResponse
err := json.Unmarshal(respRaw, &result)
if err != nil {
return err
}
for i := range result {
oSide, err := order.StringToOrderSide(result[i].Side)
if err != nil {
g.Websocket.DataHandler <- order.ClassificationError{
Exchange: g.Name,
OrderID: result[i].OrderID,
Err: err,
}
}
var oType order.Type
oType, err = stringToOrderType(result[i].OrderType)
if err != nil {
g.Websocket.DataHandler <- order.ClassificationError{
Exchange: g.Name,
OrderID: result[i].OrderID,
Err: err,
}
}
var oStatus order.Status
oStatus, err = stringToOrderStatus(result[i].Type)
if err != nil {
g.Websocket.DataHandler <- order.ClassificationError{
Exchange: g.Name,
OrderID: result[i].OrderID,
Err: err,
}
}
p := currency.NewPairFromString(result[i].Symbol)
var a asset.Item
a, err = g.GetPairAssetType(p)
if err != nil {
return err
}
g.Websocket.DataHandler <- &order.Detail{
HiddenOrder: result[i].IsHidden,
Price: result[i].Price,
Amount: result[i].OriginalAmount,
ExecutedAmount: result[i].ExecutedAmount,
RemainingAmount: result[i].RemainingAmount,
Exchange: g.Name,
ID: result[i].OrderID,
Type: oType,
Side: oSide,
Status: oStatus,
AssetType: a,
Date: time.Unix(0, result[i].Timestampms*int64(time.Millisecond)),
Pair: p,
}
}
return nil
}
var result map[string]interface{}
err := json.Unmarshal(respRaw, &result)
if err != nil {
return fmt.Errorf("%v Error: %v, Raw: %v", g.Name, err, string(respRaw))
}
if _, ok := result["type"]; ok {
switch result["type"] {
case "subscription_ack":
var result WsSubscriptionAcknowledgementResponse
err := json.Unmarshal(respRaw, &result)
if err != nil {
return err
}
g.Websocket.DataHandler <- result
case "unsubscribe":
var result wsUnsubscribeResponse
err := json.Unmarshal(respRaw, &result)
if err != nil {
return err
}
g.Websocket.DataHandler <- result
case "initial":
var result WsSubscriptionAcknowledgementResponse
err := json.Unmarshal(respRaw, &result)
if err != nil {
return err
}
g.Websocket.DataHandler <- result
case "heartbeat":
var result WsHeartbeatResponse
err := json.Unmarshal(respRaw, &result)
if err != nil {
return err
}
g.Websocket.DataHandler <- result
case "update":
if curr.IsEmpty() {
return fmt.Errorf("%v - `update` response error. Currency is empty %s",
g.Name, respRaw)
}
var marketUpdate WsMarketUpdateResponse
err := json.Unmarshal(respRaw, &marketUpdate)
if err != nil {
return err
}
g.wsProcessUpdate(marketUpdate, curr)
case "candles_1m_updates",
"candles_5m_updates",
"candles_15m_updates",
"candles_30m_updates",
"candles_1h_updates",
"candles_6h_updates",
"candles_1d_updates":
var candle wsCandleResponse
err := json.Unmarshal(respRaw, &result)
if err != nil {
return err
}
for i := range candle.Changes {
g.Websocket.DataHandler <- wshandler.KlineData{
Timestamp: time.Unix(int64(candle.Changes[i][0])*1000, 0),
Pair: curr,
AssetType: asset.Spot,
Exchange: g.Name,
Interval: result["type"].(string),
OpenPrice: candle.Changes[i][1],
ClosePrice: candle.Changes[i][4],
HighPrice: candle.Changes[i][2],
LowPrice: candle.Changes[i][3],
Volume: candle.Changes[i][5],
}
}
default:
g.Websocket.DataHandler <- wshandler.UnhandledMessageWarning{Message: g.Name + wshandler.UnhandledMessage + string(respRaw)}
return nil
}
} else if _, ok := result["result"]; ok {
switch result["result"].(string) {
case "error":
if _, ok := result["reason"]; ok {
if _, ok := result["message"]; ok {
return errors.New(result["reason"].(string) + " - " + result["message"].(string))
}
}
return fmt.Errorf("%v Unhandled websocket error %s", g.Name, respRaw)
default:
g.Websocket.DataHandler <- wshandler.UnhandledMessageWarning{Message: g.Name + wshandler.UnhandledMessage + string(respRaw)}
return nil
}
}
return nil
}
func stringToOrderStatus(status string) (order.Status, error) {
switch status {
case "accepted":
return order.New, nil
case "booked":
return order.Active, nil
case "fill":
return order.Filled, nil
case "cancelled":
return order.Cancelled, nil
case "cancel_rejected":
return order.Rejected, nil
case "closed":
return order.Filled, nil
default:
return order.UnknownStatus, errors.New(status + " not recognised as order status")
}
}
func stringToOrderType(oType string) (order.Type, error) {
switch oType {
case "exchange limit", "auction-only limit", "indication-of-interest limit":
return order.Limit, nil
case "market buy", "market sell", "block_trade":
// block trades are conducted off order-book, so their type is market, but would be considered a hidden trade
return order.Market, nil
default:
return order.UnknownType, errors.New(oType + " not recognised as order type")
}
}
// wsProcessUpdate handles order book data
func (g *Gemini) wsProcessUpdate(result WsMarketUpdateResponse, pair currency.Pair) {
if result.Timestamp == 0 && result.TimestampMS == 0 {
@@ -295,7 +394,15 @@ func (g *Gemini) wsProcessUpdate(result WsMarketUpdateResponse, pair currency.Pa
} else {
var asks, bids []orderbook.Item
for i := range result.Events {
if result.Events[i].Type == "trade" {
switch result.Events[i].Type {
case "trade":
tSide, err := order.StringToOrderSide(result.Events[i].MakerSide)
if err != nil {
g.Websocket.DataHandler <- order.ClassificationError{
Exchange: g.Name,
Err: err,
}
}
g.Websocket.DataHandler <- wshandler.TradeData{
Timestamp: time.Unix(0, result.Timestamp),
CurrencyPair: pair,
@@ -303,9 +410,9 @@ func (g *Gemini) wsProcessUpdate(result WsMarketUpdateResponse, pair currency.Pa
Exchange: g.Name,
Price: result.Events[i].Price,
Amount: result.Events[i].Amount,
Side: result.Events[i].MakerSide,
Side: tSide,
}
} else {
case "change":
item := orderbook.Item{
Amount: result.Events[i].Remaining,
Price: result.Events[i].Price,
@@ -315,8 +422,13 @@ func (g *Gemini) wsProcessUpdate(result WsMarketUpdateResponse, pair currency.Pa
} else {
bids = append(bids, item)
}
default:
g.Websocket.DataHandler <- fmt.Errorf("%s - Unhandled websocket update: %+v", g.Name, result)
}
}
if len(asks) == 0 && len(bids) == 0 {
return
}
err := g.Websocket.Orderbook.Update(&wsorderbook.WebsocketOrderbookUpdate{
Asks: asks,
Bids: bids,

51
exchanges/gemini/gemini_wrapper.go
View File

@@ -94,6 +94,9 @@ func (g *Gemini) SetDefaults() {
TradeFetching: true,
AuthenticatedEndpoints: true,
MessageSequenceNumbers: true,
Subscribe: true,
Unsubscribe: true,
KlineFetching: true,
},
WithdrawPermissions: exchange.AutoWithdrawCryptoWithAPIPermission |
exchange.AutoWithdrawCryptoWithSetup |
@@ -338,7 +341,7 @@ func (g *Gemini) SubmitOrder(s *order.Submit) (order.SubmitResponse, error) {
return submitOrderResponse, err
}
if s.OrderType != order.Limit {
if s.Type != order.Limit {
return submitOrderResponse,
errors.New("only limit orders are enabled through this exchange")
}
@@ -347,7 +350,7 @@ func (g *Gemini) SubmitOrder(s *order.Submit) (order.SubmitResponse, error) {
g.FormatExchangeCurrency(s.Pair, asset.Spot).String(),
s.Amount,
s.Price,
s.OrderSide.String(),
s.Side.String(),
"exchange limit")
if err != nil {
return submitOrderResponse, err
@@ -369,7 +372,7 @@ func (g *Gemini) ModifyOrder(action *order.Modify) (string, error) {
// CancelOrder cancels an order by its corresponding ID number
func (g *Gemini) CancelOrder(order *order.Cancel) error {
orderIDInt, err := strconv.ParseInt(order.OrderID, 10, 64)
orderIDInt, err := strconv.ParseInt(order.ID, 10, 64)
if err != nil {
return err
}
@@ -479,31 +482,31 @@ func (g *Gemini) GetActiveOrders(req *order.GetOrdersRequest) ([]order.Detail, e
ID: strconv.FormatInt(resp[i].OrderID, 10),
ExecutedAmount: resp[i].ExecutedAmount,
Exchange: g.Name,
OrderType: orderType,
OrderSide: side,
Type: orderType,
Side: side,
Price: resp[i].Price,
CurrencyPair: symbol,
OrderDate: orderDate,
Pair: symbol,
Date: orderDate,
})
}
order.FilterOrdersByTickRange(&orders, req.StartTicks, req.EndTicks)
order.FilterOrdersBySide(&orders, req.OrderSide)
order.FilterOrdersByType(&orders, req.OrderType)
order.FilterOrdersByCurrencies(&orders, req.Currencies)
order.FilterOrdersBySide(&orders, req.Side)
order.FilterOrdersByType(&orders, req.Type)
order.FilterOrdersByCurrencies(&orders, req.Pairs)
return orders, nil
}
// GetOrderHistory retrieves account order information
// Can Limit response to specific order status
func (g *Gemini) GetOrderHistory(req *order.GetOrdersRequest) ([]order.Detail, error) {
if len(req.Currencies) == 0 {
if len(req.Pairs) == 0 {
return nil, errors.New("currency must be supplied")
}
var trades []TradeHistory
for j := range req.Currencies {
resp, err := g.GetTradeHistory(g.FormatExchangeCurrency(req.Currencies[j],
for j := range req.Pairs {
resp, err := g.GetTradeHistory(g.FormatExchangeCurrency(req.Pairs[j],
asset.Spot).String(),
req.StartTicks.Unix())
if err != nil {
@@ -511,8 +514,8 @@ func (g *Gemini) GetOrderHistory(req *order.GetOrdersRequest) ([]order.Detail, e
}
for i := range resp {
resp[i].BaseCurrency = req.Currencies[j].Base.String()
resp[i].QuoteCurrency = req.Currencies[j].Quote.String()
resp[i].BaseCurrency = req.Pairs[j].Base.String()
resp[i].QuoteCurrency = req.Pairs[j].Quote.String()
trades = append(trades, resp[i])
}
}
@@ -523,21 +526,21 @@ func (g *Gemini) GetOrderHistory(req *order.GetOrdersRequest) ([]order.Detail, e
orderDate := time.Unix(trades[i].Timestamp, 0)
orders = append(orders, order.Detail{
Amount: trades[i].Amount,
ID: strconv.FormatInt(trades[i].OrderID, 10),
Exchange: g.Name,
OrderDate: orderDate,
OrderSide: side,
Fee: trades[i].FeeAmount,
Price: trades[i].Price,
CurrencyPair: currency.NewPairWithDelimiter(trades[i].BaseCurrency,
Amount: trades[i].Amount,
ID: strconv.FormatInt(trades[i].OrderID, 10),
Exchange: g.Name,
Date: orderDate,
Side: side,
Fee: trades[i].FeeAmount,
Price: trades[i].Price,
Pair: currency.NewPairWithDelimiter(trades[i].BaseCurrency,
trades[i].QuoteCurrency,
g.GetPairFormat(asset.Spot, false).Delimiter),
})
}
order.FilterOrdersByTickRange(&orders, req.StartTicks, req.EndTicks)
order.FilterOrdersBySide(&orders, req.OrderSide)
order.FilterOrdersBySide(&orders, req.Side)
return orders, nil
}