Files
gocryptotrader/exchanges/hitbtc/hitbtc_websocket.go
Ryan O'Hara-Reid cad7586e98 exchange/websocket, gateio: Rename/export package again, add websocket request functions for futures trading (#1603)
* gateio: Add multi asset websocket support WIP.

* meow

* Add tests and shenanigans

* integrate flushing and for enabling/disabling pairs from rpc shenanigans

* some changes

* linter: fixes strikes again.

* Change name ConnectionAssociation -> ConnectionCandidate for better clarity on purpose. Change connections map to point to candidate to track subscriptions for future dynamic connections holder and drop struct ConnectionDetails.

* Add subscription tests (state functional)

* glorious:nits + proxy handling

* Spelling

* linter: fixerino

* instead of nil, dont do nil.

* clean up nils

* cya nils

* don't need to set URL or check if its running

* stream match update

* update tests

* linter: fix

* glorious: nits + handle context cancellations

* stop ping handler routine leak

* * Fix bug where reader routine on error that is not a disconnection error but websocket frame error or anything really makes the reader routine return and then connection never cycles and the buffer gets filled.
* Handle reconnection via an errors.Is check which is simpler and in that scope allow for quick disconnect reconnect without waiting for connection cycle.
* Dial now uses code from DialContext but just calls context.Background()
* Don't allow reader to return on parse binary response error. Just output error and return a non nil response

* Allow rollback on connect on any error across all connections

* fix shadow jutsu

* glorious/gk: nitters - adds in ws mock server

* linter: fix

* fix deadlock on connection as the previous channel had no reader and would hang connection reader for eternity.

* glorious: whooops

* gk: nits

* Leak issue and edge case

* Websocket: Add SendMessageReturnResponses

* whooooooopsie

* gk: nitssssss

* Update exchanges/stream/stream_match.go

Co-authored-by: Gareth Kirwan <gbjkirwan@gmail.com>

* Update exchanges/stream/stream_match_test.go

Co-authored-by: Gareth Kirwan <gbjkirwan@gmail.com>

* linter: appease the linter gods

* gk: nits

* gk: drain brain

* started

* more changes before merge match pr

* gateio: still building out

* gateio: finish spot

* fix up tests in gateio

* Add tests for stream package

* rm unused field

* glorious: nits

* rn files, specifically set function names to asset and offload routing to websocket type.

* linter: fix

* Add futures websocket request support

* gateio: integrate with IBOTExchange (cherry pick my nose)

* linter: fix

* glorious: nits

* add counter and update gateio

* fix collision issue

* Update exchanges/stream/websocket.go

Co-authored-by: Scott <gloriousCode@users.noreply.github.com>

* glorious: nits

* add tests

* linter: fix

* After merge

* Add error connection info

* upgrade to upstream merge

* Fix edge case where it does not reconnect made by an already closed connection

* stream coverage

* glorious: nits

* glorious: nits removed asset error handling in stream package

* linter: fix

* rm block

* Add basic readme

* fix asset enabled flush cycle for multi connection

* spella: fix

* linter: fix

* Add glorious suggestions, fix some race thing

* reinstate name before any routine gets spawned

* stop on error in mock tests

* glorious: nits

* Set correct price

* glorious: nits found in CI build

* Add test for drain, bumped wait times as there seems to be something happening on macos CI builds, used context.WithTimeout because its instant.

* mutex across shutdown and connect for protection

* lint: fix

* test time withoffset, reinstate stop

* fix whoops

* const trafficCheckInterval; rm testmain

* y

* fix lint

* bump time check window

* stream: fix intermittant test failures while testing routines and remove code that is not needed.

* spells

* cant do what I did

* protect race due to routine.

* update testURL

* use mock websocket connection instead of test URL's

* linter: fix

* remove url because its throwing errors on CI builds

* connections drop all the time, don't need to worry about not being able to echo back ws data as it can be easily reviewed _test file side.

* remove another superfluous url thats not really set up for this

* spawn overwatch routine when there is no errors, inline checker instead of waiting for a time period, add sleep inline with echo handler as this is really quick and wanted to ensure that latency is handing correctly

* linter: fixerino uperino

* fix ID bug, why I do this, I don't know.

* glorious: panix

* linter: things

* whoops

* dont need to make consecutive Unix() calls

* websocket: fix potential panic on error and no responses and adding waitForResponses

* rm json parser and handle in json package instead

* in favour of json package unmarshalling

* linter: fix

* linter: fix again

* * change field name OutboundRequestSignature to WrapperDefinedConnectionSignature for agnostic inbound and outbound connections.
* change method name GetOutboundConnection to GetConnection for agnostic inbound and outbound connections.
* drop outbound field map for improved performance just using a range and field check (less complex as well)
* change field name connections to connectionToWrapper for better clarity

* spells and magic and wands

* glorious: nits

* comparable check for signature

* mv err var

* glorious: nits and stuff

* attempt to fix race

* glorious: nits

* gk: nits; engine log cleanup

* gk: nits; OCD

* gk: nits; move function change file names

* gk: nits; 🚀

* gk: nits; convert variadic function and message inspection to interface and include a specific function for that handling so as to not need nil on every call

* gk: nits; continued

* gk: engine nits; rm loaded exchange

* gk: nits; drop WebsocketLoginResponse

* stream: Add match method EnsureMatchWithData

* gk: nits; rn Inspect to IsFinal

* gk: nits; rn to MessageFilter

* linter: fix

* gateio: update rate limit definitions (cherry-pick)

* Add test and missing

* Shared REST rate limit definitions with Websocket service, set lookup item to nil for systems that do not require rate limiting; add glorious nit

* integrate rate limits for websocket trading spot

* conform to match upstream changes

* standardise names to upstream style

* fix wrapper standards test when sending a auth request through a websocket connection

* whoops

* Update exchanges/gateio/gateio_types.go

Co-authored-by: Scott <gloriousCode@users.noreply.github.com>

* glorious: nits

* linter: fix

* linter: overload

* whoops

* spelling fixes on recent merge

* glorious: nits

* linter: fix?

* glorious: nits

* gk: assert errors touched

* gk: unexport derive functions

* gk: nitssssssss

* fix test

* gk: nitters v1

* gk: http status

* gk/nits: Add getAssetFromFuturesPair

* gk: nits single response when submitting

* gk: new pair with delimiter in tests

* gk: param update slice to slice of pointers

* gk: add asset type in params, includes t.Context() for tests

* linter: fix

* linter: fix

* fix merge whoopsie

* glorious: nits

* gk: nit

* shift over to websocket package error

* internal/exchange/websocket -> exchange/websocket

* PEAK OCD!

* appease the OCD gods

* thrasher: nits

---------

Co-authored-by: shazbert <ryan.oharareid@thrasher.io>
Co-authored-by: Gareth Kirwan <gbjkirwan@gmail.com>
Co-authored-by: Scott <gloriousCode@users.noreply.github.com>
2025-04-11 16:47:33 +10:00

854 lines
24 KiB
Go

package hitbtc
import (
"context"
"errors"
"fmt"
"net/http"
"strconv"
"strings"
"text/template"
"time"
"github.com/Masterminds/sprig/v3"
gws "github.com/gorilla/websocket"
"github.com/thrasher-corp/gocryptotrader/common"
"github.com/thrasher-corp/gocryptotrader/common/crypto"
"github.com/thrasher-corp/gocryptotrader/currency"
"github.com/thrasher-corp/gocryptotrader/encoding/json"
"github.com/thrasher-corp/gocryptotrader/exchange/websocket"
"github.com/thrasher-corp/gocryptotrader/exchanges/asset"
"github.com/thrasher-corp/gocryptotrader/exchanges/kline"
"github.com/thrasher-corp/gocryptotrader/exchanges/order"
"github.com/thrasher-corp/gocryptotrader/exchanges/orderbook"
"github.com/thrasher-corp/gocryptotrader/exchanges/request"
"github.com/thrasher-corp/gocryptotrader/exchanges/subscription"
"github.com/thrasher-corp/gocryptotrader/exchanges/ticker"
"github.com/thrasher-corp/gocryptotrader/exchanges/trade"
"github.com/thrasher-corp/gocryptotrader/log"
)
const (
hitbtcWebsocketAddress = "wss://api.hitbtc.com/api/2/ws"
rpcVersion = "2.0"
errAuthFailed = 1002
)
var subscriptionNames = map[string]string{
subscription.TickerChannel: "Ticker",
subscription.OrderbookChannel: "Orderbook",
subscription.CandlesChannel: "Candles",
subscription.AllTradesChannel: "Trades",
subscription.MyAccountChannel: "Reports",
}
var defaultSubscriptions = subscription.List{
{Enabled: true, Asset: asset.Spot, Channel: subscription.TickerChannel},
{Enabled: true, Asset: asset.Spot, Channel: subscription.OrderbookChannel},
{Enabled: true, Asset: asset.Spot, Channel: subscription.CandlesChannel, Interval: kline.ThirtyMin, Levels: 100},
{Enabled: true, Asset: asset.Spot, Channel: subscription.AllTradesChannel, Levels: 100},
{Enabled: true, Asset: asset.Spot, Channel: subscription.MyAccountChannel, Authenticated: true},
}
// WsConnect starts a new connection with the websocket API
func (h *HitBTC) WsConnect() error {
if !h.Websocket.IsEnabled() || !h.IsEnabled() {
return websocket.ErrWebsocketNotEnabled
}
var dialer gws.Dialer
err := h.Websocket.Conn.Dial(&dialer, http.Header{})
if err != nil {
return err
}
h.Websocket.Wg.Add(1)
go h.wsReadData()
if h.Websocket.CanUseAuthenticatedEndpoints() {
err = h.wsLogin(context.TODO())
if err != nil {
log.Errorf(log.ExchangeSys, "%v - authentication failed: %v\n", h.Name, err)
}
}
return nil
}
// wsReadData receives and passes on websocket messages for processing
func (h *HitBTC) wsReadData() {
defer h.Websocket.Wg.Done()
for {
resp := h.Websocket.Conn.ReadMessage()
if resp.Raw == nil {
return
}
err := h.wsHandleData(resp.Raw)
if err != nil {
h.Websocket.DataHandler <- err
}
}
}
func (h *HitBTC) wsGetTableName(respRaw []byte) (string, error) {
var init capture
err := json.Unmarshal(respRaw, &init)
if err != nil {
return "", err
}
if init.Error.Code == errAuthFailed {
h.Websocket.SetCanUseAuthenticatedEndpoints(false)
}
if init.ID > 0 {
if h.Websocket.Match.IncomingWithData(init.ID, respRaw) {
return "", nil
}
}
if init.Error.Message != "" || init.Error.Code != 0 {
return "", fmt.Errorf("code: %d, Message: %s",
init.Error.Code,
init.Error.Message)
}
if _, ok := init.Result.(bool); ok {
return "", nil
}
if init.Method != "" {
return init.Method, nil
}
switch resultType := init.Result.(type) {
case map[string]any:
if reportType, ok := resultType["reportType"].(string); ok {
return reportType, nil
}
// check for ids - means it was a specific request
// and can't go through normal processing
if responseID, ok := resultType["id"].(string); ok {
if responseID != "" {
return "", nil
}
}
case []any:
if len(resultType) == 0 {
h.Websocket.DataHandler <- fmt.Sprintf("No data returned. ID: %v", init.ID)
return "", nil
}
data, ok := resultType[0].(map[string]any)
if !ok {
return "", errors.New("unable to type assert data")
}
if _, ok := data["clientOrderId"]; ok {
return "order", nil
} else if _, ok := data["available"]; ok {
return "trading", nil
}
}
h.Websocket.DataHandler <- websocket.UnhandledMessageWarning{Message: h.Name + websocket.UnhandledMessage + string(respRaw)}
return "", nil
}
func (h *HitBTC) wsHandleData(respRaw []byte) error {
name, err := h.wsGetTableName(respRaw)
if err != nil {
return err
}
switch name {
case "":
return nil
case "ticker":
var wsTicker WsTicker
err := json.Unmarshal(respRaw, &wsTicker)
if err != nil {
return err
}
pairs, err := h.GetEnabledPairs(asset.Spot)
if err != nil {
return err
}
format, err := h.GetPairFormat(asset.Spot, true)
if err != nil {
return err
}
p, err := currency.NewPairFromFormattedPairs(wsTicker.Params.Symbol,
pairs,
format)
if err != nil {
return err
}
h.Websocket.DataHandler <- &ticker.Price{
ExchangeName: h.Name,
Open: wsTicker.Params.Open,
Volume: wsTicker.Params.Volume,
QuoteVolume: wsTicker.Params.VolumeQuote,
High: wsTicker.Params.High,
Low: wsTicker.Params.Low,
Bid: wsTicker.Params.Bid,
Ask: wsTicker.Params.Ask,
Last: wsTicker.Params.Last,
LastUpdated: wsTicker.Params.Timestamp,
AssetType: asset.Spot,
Pair: p,
}
case "snapshotOrderbook":
var obSnapshot WsOrderbook
err := json.Unmarshal(respRaw, &obSnapshot)
if err != nil {
return err
}
err = h.WsProcessOrderbookSnapshot(&obSnapshot)
if err != nil {
return err
}
case "updateOrderbook":
var obUpdate WsOrderbook
err := json.Unmarshal(respRaw, &obUpdate)
if err != nil {
return err
}
err = h.WsProcessOrderbookUpdate(&obUpdate)
if err != nil {
return err
}
case "snapshotTrades", "updateTrades":
if !h.IsSaveTradeDataEnabled() {
return nil
}
var tradeSnapshot WsTrade
err := json.Unmarshal(respRaw, &tradeSnapshot)
if err != nil {
return err
}
var trades []trade.Data
p, err := currency.NewPairFromString(tradeSnapshot.Params.Symbol)
if err != nil {
return &order.ClassificationError{
Exchange: h.Name,
Err: err,
}
}
for i := range tradeSnapshot.Params.Data {
side, err := order.StringToOrderSide(tradeSnapshot.Params.Data[i].Side)
if err != nil {
return &order.ClassificationError{
Exchange: h.Name,
Err: err,
}
}
trades = append(trades, trade.Data{
Timestamp: tradeSnapshot.Params.Data[i].Timestamp,
Exchange: h.Name,
CurrencyPair: p,
AssetType: asset.Spot,
Price: tradeSnapshot.Params.Data[i].Price,
Amount: tradeSnapshot.Params.Data[i].Quantity,
Side: side,
TID: strconv.FormatInt(tradeSnapshot.Params.Data[i].ID, 10),
})
}
return trade.AddTradesToBuffer(trades...)
case "activeOrders":
var o wsActiveOrdersResponse
err := json.Unmarshal(respRaw, &o)
if err != nil {
return err
}
for i := range o.Params {
err = h.wsHandleOrderData(&o.Params[i])
if err != nil {
return err
}
}
case "trading":
var trades WsGetTradingBalanceResponse
err := json.Unmarshal(respRaw, &trades)
if err != nil {
return err
}
h.Websocket.DataHandler <- trades
case "report":
var o wsReportResponse
err := json.Unmarshal(respRaw, &o)
if err != nil {
return err
}
err = h.wsHandleOrderData(&o.OrderData)
if err != nil {
return err
}
case "order":
var o wsActiveOrderRequestResponse
err := json.Unmarshal(respRaw, &o)
if err != nil {
return err
}
for i := range o.OrderData {
err = h.wsHandleOrderData(&o.OrderData[i])
if err != nil {
return err
}
}
case "replaced", "canceled", "new":
var o wsOrderResponse
err := json.Unmarshal(respRaw, &o)
if err != nil {
return err
}
err = h.wsHandleOrderData(&o.OrderData)
if err != nil {
return err
}
default:
h.Websocket.DataHandler <- websocket.UnhandledMessageWarning{Message: h.Name + websocket.UnhandledMessage + string(respRaw)}
return nil
}
return nil
}
// WsProcessOrderbookSnapshot processes a full orderbook snapshot to a local cache
func (h *HitBTC) WsProcessOrderbookSnapshot(ob *WsOrderbook) error {
if len(ob.Params.Bid) == 0 || len(ob.Params.Ask) == 0 {
return errors.New("no orderbooks to process")
}
newOrderBook := orderbook.Base{
Bids: make(orderbook.Tranches, len(ob.Params.Bid)),
Asks: make(orderbook.Tranches, len(ob.Params.Ask)),
}
for i := range ob.Params.Bid {
newOrderBook.Bids[i] = orderbook.Tranche{
Amount: ob.Params.Bid[i].Size,
Price: ob.Params.Bid[i].Price,
}
}
for i := range ob.Params.Ask {
newOrderBook.Asks[i] = orderbook.Tranche{
Amount: ob.Params.Ask[i].Size,
Price: ob.Params.Ask[i].Price,
}
}
pairs, err := h.GetEnabledPairs(asset.Spot)
if err != nil {
return err
}
format, err := h.GetPairFormat(asset.Spot, true)
if err != nil {
return err
}
p, err := currency.NewPairFromFormattedPairs(ob.Params.Symbol,
pairs,
format)
if err != nil {
h.Websocket.DataHandler <- err
return err
}
newOrderBook.Asset = asset.Spot
newOrderBook.Pair = p
newOrderBook.Exchange = h.Name
newOrderBook.VerifyOrderbook = h.CanVerifyOrderbook
newOrderBook.LastUpdated = ob.Params.Timestamp
return h.Websocket.Orderbook.LoadSnapshot(&newOrderBook)
}
func (h *HitBTC) wsHandleOrderData(o *wsOrderData) error {
var trades []order.TradeHistory
if o.TradeID > 0 {
trades = append(trades, order.TradeHistory{
Price: o.TradePrice,
Amount: o.TradeQuantity,
Fee: o.TradeFee,
Exchange: h.Name,
TID: strconv.FormatFloat(o.TradeID, 'f', -1, 64),
Timestamp: o.UpdatedAt,
})
}
oType, err := order.StringToOrderType(o.Type)
if err != nil {
h.Websocket.DataHandler <- order.ClassificationError{
Exchange: h.Name,
OrderID: o.ID,
Err: err,
}
}
o.Status = strings.Replace(o.Status, "canceled", "cancelled", 1)
oStatus, err := order.StringToOrderStatus(o.Status)
if err != nil {
h.Websocket.DataHandler <- order.ClassificationError{
Exchange: h.Name,
OrderID: o.ID,
Err: err,
}
}
oSide, err := order.StringToOrderSide(o.Side)
if err != nil {
h.Websocket.DataHandler <- order.ClassificationError{
Exchange: h.Name,
OrderID: o.ID,
Err: err,
}
}
p, err := currency.NewPairFromString(o.Symbol)
if err != nil {
h.Websocket.DataHandler <- order.ClassificationError{
Exchange: h.Name,
OrderID: o.ID,
Err: err,
}
}
var a asset.Item
a, err = h.GetPairAssetType(p)
if err != nil {
return err
}
h.Websocket.DataHandler <- &order.Detail{
Price: o.Price,
Amount: o.Quantity,
ExecutedAmount: o.CumQuantity,
RemainingAmount: o.Quantity - o.CumQuantity,
Exchange: h.Name,
OrderID: o.ID,
Type: oType,
Side: oSide,
Status: oStatus,
AssetType: a,
Date: o.CreatedAt,
LastUpdated: o.UpdatedAt,
Pair: p,
Trades: trades,
}
return nil
}
// WsProcessOrderbookUpdate updates a local cache
func (h *HitBTC) WsProcessOrderbookUpdate(update *WsOrderbook) error {
if len(update.Params.Bid) == 0 && len(update.Params.Ask) == 0 {
// Periodically HitBTC sends empty updates which includes a sequence
// can return this as nil.
return nil
}
bids := make(orderbook.Tranches, len(update.Params.Bid))
for i := range update.Params.Bid {
bids[i] = orderbook.Tranche{
Price: update.Params.Bid[i].Price,
Amount: update.Params.Bid[i].Size,
}
}
asks := make(orderbook.Tranches, len(update.Params.Ask))
for i := range update.Params.Ask {
asks[i] = orderbook.Tranche{
Price: update.Params.Ask[i].Price,
Amount: update.Params.Ask[i].Size,
}
}
pairs, err := h.GetEnabledPairs(asset.Spot)
if err != nil {
return err
}
format, err := h.GetPairFormat(asset.Spot, true)
if err != nil {
return err
}
p, err := currency.NewPairFromFormattedPairs(update.Params.Symbol,
pairs,
format)
if err != nil {
return err
}
return h.Websocket.Orderbook.Update(&orderbook.Update{
Asks: asks,
Bids: bids,
Pair: p,
UpdateID: update.Params.Sequence,
Asset: asset.Spot,
UpdateTime: update.Params.Timestamp,
})
}
// generateSubscriptions returns a list of subscriptions from the configured subscriptions feature
func (h *HitBTC) generateSubscriptions() (subscription.List, error) {
return h.Features.Subscriptions.ExpandTemplates(h)
}
// GetSubscriptionTemplate returns a subscription channel template
func (h *HitBTC) GetSubscriptionTemplate(_ *subscription.Subscription) (*template.Template, error) {
return template.New("master.tmpl").Funcs(sprig.FuncMap()).Funcs(template.FuncMap{
"subToReq": subToReq,
"isSymbolChannel": isSymbolChannel,
}).Parse(subTplText)
}
const (
subscribeOp = "subscribe"
unsubscribeOp = "unsubscribe"
)
// Subscribe sends a websocket message to receive data from the channel
func (h *HitBTC) Subscribe(subs subscription.List) error {
return h.ParallelChanOp(subs, func(subs subscription.List) error { return h.manageSubs(subscribeOp, subs) }, 1)
}
// Unsubscribe sends a websocket message to stop receiving data from the channel
func (h *HitBTC) Unsubscribe(subs subscription.List) error {
return h.ParallelChanOp(subs, func(subs subscription.List) error { return h.manageSubs(unsubscribeOp, subs) }, 1)
}
func (h *HitBTC) manageSubs(op string, subs subscription.List) error {
var errs error
subs, errs = subs.ExpandTemplates(h)
for _, s := range subs {
r := WsRequest{
JSONRPCVersion: rpcVersion,
ID: h.Websocket.Conn.GenerateMessageID(false),
}
if err := json.Unmarshal([]byte(s.QualifiedChannel), &r); err != nil {
errs = common.AppendError(errs, err)
continue
}
r.Method = op + r.Method
err := h.Websocket.Conn.SendJSONMessage(context.TODO(), request.Unset, r) // v2 api does not return an ID with errors, so we don't use ReturnResponse
if err == nil {
if op == subscribeOp {
err = h.Websocket.AddSuccessfulSubscriptions(h.Websocket.Conn, s)
} else {
err = h.Websocket.RemoveSubscriptions(h.Websocket.Conn, s)
}
}
if err != nil {
errs = common.AppendError(errs, err)
}
}
return errs
}
// Unsubscribe sends a websocket message to stop receiving data from the channel
func (h *HitBTC) wsLogin(ctx context.Context) error {
if !h.IsWebsocketAuthenticationSupported() {
return fmt.Errorf("%v AuthenticatedWebsocketAPISupport not enabled", h.Name)
}
creds, err := h.GetCredentials(ctx)
if err != nil {
return err
}
h.Websocket.SetCanUseAuthenticatedEndpoints(true)
n := strconv.FormatInt(time.Now().Unix(), 10)
hmac, err := crypto.GetHMAC(crypto.HashSHA256,
[]byte(n),
[]byte(creds.Secret))
if err != nil {
return err
}
req := WsLoginRequest{
Method: "login",
Params: WsLoginData{
Algo: "HS256",
PKey: creds.Key,
Nonce: n,
Signature: crypto.HexEncodeToString(hmac),
},
ID: h.Websocket.Conn.GenerateMessageID(false),
}
err = h.Websocket.Conn.SendJSONMessage(context.TODO(), request.Unset, req)
if err != nil {
h.Websocket.SetCanUseAuthenticatedEndpoints(false)
return err
}
return nil
}
// wsPlaceOrder sends a websocket message to submit an order
func (h *HitBTC) wsPlaceOrder(pair currency.Pair, side string, price, quantity float64) (*WsSubmitOrderSuccessResponse, error) {
if !h.Websocket.CanUseAuthenticatedEndpoints() {
return nil, fmt.Errorf("%v not authenticated, cannot place order", h.Name)
}
id := h.Websocket.Conn.GenerateMessageID(false)
fPair, err := h.FormatExchangeCurrency(pair, asset.Spot)
if err != nil {
return nil, err
}
req := WsSubmitOrderRequest{
Method: "newOrder",
Params: WsSubmitOrderRequestData{
ClientOrderID: id,
Symbol: fPair.String(),
Side: strings.ToLower(side),
Price: price,
Quantity: quantity,
},
ID: id,
}
resp, err := h.Websocket.Conn.SendMessageReturnResponse(context.TODO(), request.Unset, id, req)
if err != nil {
return nil, fmt.Errorf("%v %v", h.Name, err)
}
var response WsSubmitOrderSuccessResponse
err = json.Unmarshal(resp, &response)
if err != nil {
return nil, fmt.Errorf("%v %v", h.Name, err)
}
if response.Error.Code > 0 || response.Error.Message != "" {
return &response, fmt.Errorf("%v Error:%v Message:%v", h.Name, response.Error.Code, response.Error.Message)
}
return &response, nil
}
// wsCancelOrder sends a websocket message to cancel an order
func (h *HitBTC) wsCancelOrder(clientOrderID string) (*WsCancelOrderResponse, error) {
if !h.Websocket.CanUseAuthenticatedEndpoints() {
return nil, fmt.Errorf("%v not authenticated, cannot place order", h.Name)
}
req := WsCancelOrderRequest{
Method: "cancelOrder",
Params: WsCancelOrderRequestData{
ClientOrderID: clientOrderID,
},
ID: h.Websocket.Conn.GenerateMessageID(false),
}
resp, err := h.Websocket.Conn.SendMessageReturnResponse(context.TODO(), request.Unset, req.ID, req)
if err != nil {
return nil, fmt.Errorf("%v %v", h.Name, err)
}
var response WsCancelOrderResponse
err = json.Unmarshal(resp, &response)
if err != nil {
return nil, fmt.Errorf("%v %v", h.Name, err)
}
if response.Error.Code > 0 || response.Error.Message != "" {
return &response, fmt.Errorf("%v Error:%v Message:%v", h.Name, response.Error.Code, response.Error.Message)
}
return &response, nil
}
// wsReplaceOrder sends a websocket message to replace an order
func (h *HitBTC) wsReplaceOrder(clientOrderID string, quantity, price float64) (*WsReplaceOrderResponse, error) {
if !h.Websocket.CanUseAuthenticatedEndpoints() {
return nil, fmt.Errorf("%v not authenticated, cannot place order", h.Name)
}
req := WsReplaceOrderRequest{
Method: "cancelReplaceOrder",
Params: WsReplaceOrderRequestData{
ClientOrderID: clientOrderID,
RequestClientID: strconv.FormatInt(time.Now().Unix(), 10),
Quantity: quantity,
Price: price,
},
ID: h.Websocket.Conn.GenerateMessageID(false),
}
resp, err := h.Websocket.Conn.SendMessageReturnResponse(context.TODO(), request.Unset, req.ID, req)
if err != nil {
return nil, fmt.Errorf("%v %v", h.Name, err)
}
var response WsReplaceOrderResponse
err = json.Unmarshal(resp, &response)
if err != nil {
return nil, fmt.Errorf("%v %v", h.Name, err)
}
if response.Error.Code > 0 || response.Error.Message != "" {
return &response, fmt.Errorf("%v Error:%v Message:%v", h.Name, response.Error.Code, response.Error.Message)
}
return &response, nil
}
// wsGetActiveOrders sends a websocket message to get all active orders
func (h *HitBTC) wsGetActiveOrders() (*wsActiveOrdersResponse, error) {
if !h.Websocket.CanUseAuthenticatedEndpoints() {
return nil, fmt.Errorf("%v not authenticated, cannot get active orders", h.Name)
}
req := WsReplaceOrderRequest{
Method: "getOrders",
Params: WsReplaceOrderRequestData{},
ID: h.Websocket.Conn.GenerateMessageID(false),
}
resp, err := h.Websocket.Conn.SendMessageReturnResponse(context.TODO(), request.Unset, req.ID, req)
if err != nil {
return nil, fmt.Errorf("%v %v", h.Name, err)
}
var response wsActiveOrdersResponse
err = json.Unmarshal(resp, &response)
if err != nil {
return nil, fmt.Errorf("%v %v", h.Name, err)
}
if response.Error.Code > 0 || response.Error.Message != "" {
return &response, fmt.Errorf("%v Error:%v Message:%v", h.Name, response.Error.Code, response.Error.Message)
}
return &response, nil
}
// wsGetTradingBalance sends a websocket message to get trading balance
func (h *HitBTC) wsGetTradingBalance() (*WsGetTradingBalanceResponse, error) {
if !h.Websocket.CanUseAuthenticatedEndpoints() {
return nil, fmt.Errorf("%v not authenticated, cannot place order", h.Name)
}
req := WsReplaceOrderRequest{
Method: "getTradingBalance",
Params: WsReplaceOrderRequestData{},
ID: h.Websocket.Conn.GenerateMessageID(false),
}
resp, err := h.Websocket.Conn.SendMessageReturnResponse(context.TODO(), request.Unset, req.ID, req)
if err != nil {
return nil, fmt.Errorf("%v %v", h.Name, err)
}
var response WsGetTradingBalanceResponse
err = json.Unmarshal(resp, &response)
if err != nil {
return nil, fmt.Errorf("%v %v", h.Name, err)
}
if response.Error.Code > 0 || response.Error.Message != "" {
return &response, fmt.Errorf("%v Error:%v Message:%v", h.Name, response.Error.Code, response.Error.Message)
}
return &response, nil
}
// wsGetCurrencies sends a websocket message to get trading balance
func (h *HitBTC) wsGetCurrencies(currencyItem currency.Code) (*WsGetCurrenciesResponse, error) {
req := WsGetCurrenciesRequest{
Method: "getCurrency",
Params: WsGetCurrenciesRequestParameters{
Currency: currencyItem,
},
ID: h.Websocket.Conn.GenerateMessageID(false),
}
resp, err := h.Websocket.Conn.SendMessageReturnResponse(context.TODO(), request.Unset, req.ID, req)
if err != nil {
return nil, fmt.Errorf("%v %v", h.Name, err)
}
var response WsGetCurrenciesResponse
err = json.Unmarshal(resp, &response)
if err != nil {
return nil, fmt.Errorf("%v %v", h.Name, err)
}
if response.Error.Code > 0 || response.Error.Message != "" {
return &response, fmt.Errorf("%v Error:%v Message:%v", h.Name, response.Error.Code, response.Error.Message)
}
return &response, nil
}
// wsGetSymbols sends a websocket message to get trading balance
func (h *HitBTC) wsGetSymbols(c currency.Pair) (*WsGetSymbolsResponse, error) {
fPair, err := h.FormatExchangeCurrency(c, asset.Spot)
if err != nil {
return nil, err
}
req := WsGetSymbolsRequest{
Method: "getSymbol",
Params: WsGetSymbolsRequestParameters{
Symbol: fPair.String(),
},
ID: h.Websocket.Conn.GenerateMessageID(false),
}
resp, err := h.Websocket.Conn.SendMessageReturnResponse(context.TODO(), request.Unset, req.ID, req)
if err != nil {
return nil, fmt.Errorf("%v %v", h.Name, err)
}
var response WsGetSymbolsResponse
err = json.Unmarshal(resp, &response)
if err != nil {
return nil, fmt.Errorf("%v %v", h.Name, err)
}
if response.Error.Code > 0 || response.Error.Message != "" {
return &response, fmt.Errorf("%v Error:%v Message:%v", h.Name, response.Error.Code, response.Error.Message)
}
return &response, nil
}
// wsGetSymbols sends a websocket message to get trading balance
func (h *HitBTC) wsGetTrades(c currency.Pair, limit int64, sort, by string) (*WsGetTradesResponse, error) {
fPair, err := h.FormatExchangeCurrency(c, asset.Spot)
if err != nil {
return nil, err
}
req := WsGetTradesRequest{
Method: "getTrades",
Params: WsGetTradesRequestParameters{
Symbol: fPair.String(),
Limit: limit,
Sort: sort,
By: by,
},
ID: h.Websocket.Conn.GenerateMessageID(false),
}
resp, err := h.Websocket.Conn.SendMessageReturnResponse(context.TODO(), request.Unset, req.ID, req)
if err != nil {
return nil, fmt.Errorf("%v %v", h.Name, err)
}
var response WsGetTradesResponse
err = json.Unmarshal(resp, &response)
if err != nil {
return nil, fmt.Errorf("%v %v", h.Name, err)
}
if response.Error.Code > 0 || response.Error.Message != "" {
return &response, fmt.Errorf("%v Error:%v Message:%v", h.Name, response.Error.Code, response.Error.Message)
}
return &response, nil
}
// subToReq returns the subscription as a map to populate WsRequest
func subToReq(s *subscription.Subscription, maybePair ...currency.Pair) *WsRequest {
name, ok := subscriptionNames[s.Channel]
if !ok {
panic(fmt.Errorf("%w: %s", subscription.ErrNotSupported, s.Channel))
}
r := &WsRequest{
Method: name,
}
if len(maybePair) != 0 {
r.Params = &WsParams{
Symbol: maybePair[0].String(),
Limit: s.Levels,
}
if s.Interval != 0 {
var err error
if r.Params.Period, err = formatExchangeKlineInterval(s.Interval); err != nil {
panic(err)
}
}
} else if s.Levels != 0 {
r.Params = &WsParams{
Limit: s.Levels,
}
}
return r
}
// isSymbolChannel returns if the channel expects receive a symbol
func isSymbolChannel(s *subscription.Subscription) bool {
return s.Channel != subscription.MyAccountChannel
}
const subTplText = `
{{- if isSymbolChannel $.S }}
{{ range $asset, $pairs := $.AssetPairs }}
{{- range $p := $pairs -}}
{{- subToReq $.S $p | mustToJson }}
{{ $.PairSeparator }}
{{- end }}
{{ $.AssetSeparator }}
{{- end }}
{{- else }}
{{- subToReq $.S | mustToJson }}
{{- end }}
`