Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

okx: decouple outbound trading functionality from multiplexer #1600

Open
wants to merge 20 commits into
base: master
Choose a base branch
from
Open

okx: decouple outbound trading functionality from multiplexer #1600

Show file tree
Hide file tree
Changes from 4 commits
Commits
Show all changes
20 commits
Select commit Hold shift + click to select a range
91dd7de
decouple outbound trading function from multiplexer
Aug 2, 2024
76be905
linter: fix
Aug 2, 2024
e60dae5
samuael: adds patch thanks @samuael
Aug 2, 2024
d653bd8
rm: prints
Aug 2, 2024
66e75db
Merge branch 'master' into okx_websocket_req
Aug 15, 2024
398fad0
glorious: nits
Aug 15, 2024
8a53e5d
linter: fix
Aug 15, 2024
7d2a93d
Merge branch 'master' into okx_websocket_req
Sep 4, 2024
c7e002f
rm semaphore
Sep 4, 2024
2c1d6ee
poulate context after upstream merge
Sep 4, 2024
76d2f4a
use counter for outbound messages after upstream merge
Sep 4, 2024
6251a5e
Shift todo to relevant function for future update
Sep 4, 2024
75eeed4
Merge branch 'master' into okx_websocket_req
Sep 15, 2024
3f5d83a
Merge branch 'master' into okx_websocket_req
Sep 19, 2024
63cb18f
Merge branch 'master' into okx_websocket_req
Oct 2, 2024
67f8818
Merge branch 'master' into okx_websocket_req
Oct 10, 2024
ab89312
glorious: nits
Oct 10, 2024
f76cf7d
glorious: nits
Oct 13, 2024
3923a78
Merge branch 'master' into okx_websocket_req
Nov 7, 2024
f9a7beb
Merge branch 'master' into okx_websocket_req
Nov 11, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
113 changes: 33 additions & 80 deletions exchanges/okx/okx.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -29,14 +29,13 @@ import (
// Okx is the overarching type across this package
type Okx struct {
exchange.Base
WsResponseMultiplexer wsRequestDataChannelsMultiplexer

// WsRequestSemaphore channel is used to block write operation on the websocket connection to reduce contention; a kind of bounded parallelism.
// it is made to hold up to 20 integers so that up to 20 write operations can be called over the websocket connection at a time.
// and when the operation is completed the thread releases (consumes) one value from the channel so that the other waiting operation can enter.
// ok.WsRequestSemaphore <- 1
// defer func() { <-ok.WsRequestSemaphore }()
WsRequestSemaphore chan int
WsRequestSemaphore chan int // TODO: Remove when rate limiting PR is merged.
}

const (
Expand Down Expand Up @@ -322,7 +321,6 @@ var (
errNoCandlestickDataFound = errors.New("no candlesticks data found")
errInvalidWebsocketEvent = errors.New("invalid websocket event")
errMissingValidChannelInformation = errors.New("missing channel information")
errNilArgument = errors.New("nil argument is not acceptable")
errNoOrderParameterPassed = errors.New("no order parameter was passed")
errMaxRfqOrdersToCancel = errors.New("no more than 100 Rfq cancel order parameter is allowed")
errMalformedData = errors.New("malformed data")
Expand All @@ -338,7 +336,6 @@ var (
errInvalidIPAddress = errors.New("invalid ip address")
errInvalidAPIKeyPermission = errors.New("invalid API Key permission")
errInvalidResponseParam = errors.New("invalid response parameter, response must be non-nil pointer")
errEmptyPlaceOrderResponse = errors.New("empty place order response")
errTooManyArgument = errors.New("too many cancel request params")
errIncompleteCurrencyPair = errors.New("incomplete currency pair")
errInvalidDuration = errors.New("invalid grid contract duration, only '7D', '30D', and '180D' are allowed")
Expand All @@ -347,6 +344,7 @@ var (
errOnlyThreeMonthsSupported = errors.New("only three months of trade data retrieval supported")
errOnlyOneResponseExpected = errors.New("one response item expected")
errNoInstrumentFound = errors.New("no instrument found")
errWebsocketDataNotMatchedWithID = errors.New("websocket data not matched with id")
)

/************************************ MarketData Endpoints *************************************************/
Expand Down Expand Up @@ -393,11 +391,8 @@ func (ok *Okx) OrderTypeString(orderType order.Type) (string, error) {

// PlaceOrder place an order only if you have sufficient funds.
func (ok *Okx) PlaceOrder(ctx context.Context, arg *PlaceOrderRequestParam, a asset.Item) (*OrderData, error) {
if arg == nil {
return nil, errNilArgument
}
arg.AssetType = a
err := ok.validatePlaceOrderParams(arg)
err := arg.Validate()
if err != nil {
return nil, err
}
Expand All @@ -407,77 +402,34 @@ func (ok *Okx) PlaceOrder(ctx context.Context, arg *PlaceOrderRequestParam, a as
if len(resp) != 1 {
return nil, err
}
return nil, fmt.Errorf("error code:%s message: %v", resp[0].SCode, resp[0].SMessage)
return nil, fmt.Errorf("error code: %d message: %v", resp[0].SCode, resp[0].SMessage)
}
if len(resp) == 1 {
return &resp[0], nil
}
return nil, errNoValidResponseFromServer
}

func (ok *Okx) validatePlaceOrderParams(arg *PlaceOrderRequestParam) error {
if arg.InstrumentID == "" {
return errMissingInstrumentID
}
arg.Side = strings.ToLower(arg.Side)
if arg.Side != order.Buy.Lower() && arg.Side != order.Sell.Lower() {
return fmt.Errorf("%w %s", errInvalidOrderSide, arg.Side)
}
if arg.TradeMode != "" &&
arg.TradeMode != TradeModeCross &&
arg.TradeMode != TradeModeIsolated &&
arg.TradeMode != TradeModeCash {
return fmt.Errorf("%w %s", errInvalidTradeModeValue, arg.TradeMode)
}
if arg.PositionSide != "" {
if (arg.PositionSide == positionSideLong || arg.PositionSide == positionSideShort) &&
(arg.AssetType != asset.Futures && arg.AssetType != asset.PerpetualSwap) {
return errors.New("invalid position mode, 'long' or 'short' for Futures/SWAP, otherwise 'net'(default) are allowed")
}
}
arg.OrderType = strings.ToLower(arg.OrderType)
if arg.OrderType == order.OptimalLimitIOC.Lower() &&
(arg.AssetType != asset.Futures && arg.AssetType != asset.PerpetualSwap) {
return errors.New("\"optimal_limit_ioc\": market order with immediate-or-cancel order (applicable only to Futures and Perpetual swap)")
}
if arg.OrderType != OkxOrderMarket &&
arg.OrderType != OkxOrderLimit &&
arg.OrderType != OkxOrderPostOnly &&
arg.OrderType != OkxOrderFOK &&
arg.OrderType != OkxOrderIOC &&
arg.OrderType != OkxOrderOptimalLimitIOC {
return fmt.Errorf("%w %v", errInvalidOrderType, arg.OrderType)
}
if arg.Amount <= 0 {
return errInvalidAmount
}
if arg.QuantityType != "" && arg.QuantityType != "base_ccy" && arg.QuantityType != "quote_ccy" {
return errors.New("only base_ccy and quote_ccy quantity types are supported")
}
return nil
}

// PlaceMultipleOrders to place orders in batches. Maximum 20 orders can be placed at a time. Request parameters should be passed in the form of an array.
func (ok *Okx) PlaceMultipleOrders(ctx context.Context, args []PlaceOrderRequestParam) ([]OrderData, error) {
if len(args) == 0 {
return nil, errNoOrderParameterPassed
}
var err error
for x := range args {
err = ok.validatePlaceOrderParams(&args[x])
err := args[x].Validate()
if err != nil {
return nil, err
}
}
var resp []OrderData
err = ok.SendHTTPRequest(ctx, exchange.RestSpot, placeMultipleOrdersEPL, http.MethodPost, placeMultipleOrderURL, &args, &resp, true)
err := ok.SendHTTPRequest(ctx, exchange.RestSpot, placeMultipleOrdersEPL, http.MethodPost, placeMultipleOrderURL, &args, &resp, true)
if err != nil {
if len(resp) == 0 {
return nil, err
}
var errs error
for x := range resp {
errs = common.AppendError(errs, fmt.Errorf("error code:%s message: %v", resp[x].SCode, resp[x].SMessage))
errs = common.AppendError(errs, fmt.Errorf("error code: %d message: %v", resp[x].SCode, resp[x].SMessage))
}
return nil, errs
}
Expand All @@ -498,7 +450,7 @@ func (ok *Okx) CancelSingleOrder(ctx context.Context, arg CancelOrderRequestPara
if len(resp) != 1 {
return nil, err
}
return nil, fmt.Errorf("error code:%s message: %v", resp[0].SCode, resp[0].SMessage)
return nil, fmt.Errorf("error code: %d message: %v", resp[0].SCode, resp[0].SMessage)
}
if len(resp) == 1 {
return &resp[0], nil
Expand Down Expand Up @@ -527,8 +479,8 @@ func (ok *Okx) CancelMultipleOrders(ctx context.Context, args []CancelOrderReque
}
var errs error
for x := range resp {
if resp[x].SCode != "0" {
errs = common.AppendError(errs, fmt.Errorf("error code:%s message: %v", resp[x].SCode, resp[x].SMessage))
if resp[x].SCode != 0 {
errs = common.AppendError(errs, fmt.Errorf("error code: %d message: %v", resp[x].SCode, resp[x].SMessage))
}
}
return nil, errs
Expand Down Expand Up @@ -599,7 +551,7 @@ func (ok *Okx) ClosePositions(ctx context.Context, arg *ClosePositionsRequestPar
// GetOrderDetail retrieves order details given instrument id and order identification
func (ok *Okx) GetOrderDetail(ctx context.Context, arg *OrderDetailRequestParam) (*OrderDetail, error) {
if arg == nil {
return nil, errNilArgument
return nil, fmt.Errorf("%T: %w", arg, common.ErrNilPointer)
}
params := url.Values{}
if arg.InstrumentID == "" {
Expand Down Expand Up @@ -628,7 +580,7 @@ func (ok *Okx) GetOrderDetail(ctx context.Context, arg *OrderDetailRequestParam)
// GetOrderList retrieves all incomplete orders under the current account.
func (ok *Okx) GetOrderList(ctx context.Context, arg *OrderListRequestParams) ([]OrderDetail, error) {
if arg == nil {
return nil, errNilArgument
return nil, fmt.Errorf("%T: %w", arg, common.ErrNilPointer)
}
params := url.Values{}
if arg.InstrumentType != "" {
Expand Down Expand Up @@ -792,7 +744,7 @@ func (ok *Okx) PlaceAlgoOrder(ctx context.Context, arg *AlgoOrderParams) (*AlgoO
// PlaceStopOrder to place stop order
func (ok *Okx) PlaceStopOrder(ctx context.Context, arg *AlgoOrderParams) (*AlgoOrder, error) {
if arg == nil {
return nil, errNilArgument
return nil, fmt.Errorf("%T: %w", arg, common.ErrNilPointer)
}
if arg.OrderType != "conditional" {
return nil, errInvalidOrderType
Expand All @@ -809,7 +761,7 @@ func (ok *Okx) PlaceStopOrder(ctx context.Context, arg *AlgoOrderParams) (*AlgoO
// PlaceTrailingStopOrder to place trailing stop order
func (ok *Okx) PlaceTrailingStopOrder(ctx context.Context, arg *AlgoOrderParams) (*AlgoOrder, error) {
if arg == nil {
return nil, errNilArgument
return nil, fmt.Errorf("%T: %w", arg, common.ErrNilPointer)
}
if arg.OrderType != "move_order_stop" {
return nil, errInvalidOrderType
Expand All @@ -823,7 +775,7 @@ func (ok *Okx) PlaceTrailingStopOrder(ctx context.Context, arg *AlgoOrderParams)
// PlaceIcebergOrder to place iceburg algo order
func (ok *Okx) PlaceIcebergOrder(ctx context.Context, arg *AlgoOrderParams) (*AlgoOrder, error) {
if arg == nil {
return nil, errNilArgument
return nil, fmt.Errorf("%T: %w", arg, common.ErrNilPointer)
}
if arg.OrderType != "iceberg" {
return nil, errInvalidOrderType
Expand Down Expand Up @@ -857,7 +809,7 @@ func (ok *Okx) PlaceTWAPOrder(ctx context.Context, arg *AlgoOrderParams) (*AlgoO
// TriggerAlgoOrder fetches algo trigger orders for SWAP market types.
func (ok *Okx) TriggerAlgoOrder(ctx context.Context, arg *AlgoOrderParams) (*AlgoOrder, error) {
if arg == nil {
return nil, errNilArgument
return nil, fmt.Errorf("%T: %w", arg, common.ErrNilPointer)
}
if arg.OrderType != "trigger" {
return nil, errInvalidOrderType
Expand All @@ -878,8 +830,8 @@ func (ok *Okx) TriggerAlgoOrder(ctx context.Context, arg *AlgoOrderParams) (*Alg
// A maximum of 10 orders can be canceled at a time.
// Request parameters should be passed in the form of an array.
func (ok *Okx) CancelAdvanceAlgoOrder(ctx context.Context, args []AlgoOrderCancelParams) ([]AlgoOrder, error) {
if args == nil {
return nil, errNilArgument
if len(args) == 0 {
return nil, fmt.Errorf("%T: %w", args, common.ErrNilPointer)
}
return ok.cancelAlgoOrder(ctx, args, cancelAdvancedAlgoOrder, cancelAdvanceAlgoOrderEPL)
}
Expand All @@ -888,8 +840,8 @@ func (ok *Okx) CancelAdvanceAlgoOrder(ctx context.Context, args []AlgoOrderCance
// A maximum of 10 orders can be canceled at a time.
// Request parameters should be passed in the form of an array.
func (ok *Okx) CancelAlgoOrder(ctx context.Context, args []AlgoOrderCancelParams) ([]AlgoOrder, error) {
if args == nil {
return nil, errNilArgument
if len(args) == 0 {
return nil, fmt.Errorf("%T: %w", args, common.ErrNilPointer)
}
return ok.cancelAlgoOrder(ctx, args, cancelAlgoOrder, cancelAlgoOrderEPL)
}
Expand Down Expand Up @@ -1271,7 +1223,7 @@ func (ok *Okx) CancelAllQuotes(ctx context.Context) (time.Time, error) {
// GetRfqs retrieves details of Rfqs that the user is a counterparty to (either as the creator or the receiver of the Rfq).
func (ok *Okx) GetRfqs(ctx context.Context, arg *RfqRequestParams) ([]RfqResponse, error) {
if arg == nil {
return nil, errNilArgument
return nil, fmt.Errorf("%T: %w", arg, common.ErrNilPointer)
}
params := url.Values{}
if arg.RfqID != "" {
Expand Down Expand Up @@ -1299,7 +1251,7 @@ func (ok *Okx) GetRfqs(ctx context.Context, arg *RfqRequestParams) ([]RfqRespons
// GetQuotes retrieves all Quotes that the user is a counterparty to (either as the creator or the receiver).
func (ok *Okx) GetQuotes(ctx context.Context, arg *QuoteRequestParams) ([]QuoteResponse, error) {
if arg == nil {
return nil, errNilArgument
return nil, fmt.Errorf("%T: %w", arg, common.ErrNilPointer)
}
params := url.Values{}
if arg.RfqID != "" {
Expand Down Expand Up @@ -1333,7 +1285,7 @@ func (ok *Okx) GetQuotes(ctx context.Context, arg *QuoteRequestParams) ([]QuoteR
// GetRfqTrades retrieves the executed trades that the user is a counterparty to (either as the creator or the receiver).
func (ok *Okx) GetRfqTrades(ctx context.Context, arg *RfqTradesRequestParams) ([]RfqTradeResponse, error) {
if arg == nil {
return nil, errNilArgument
return nil, fmt.Errorf("%T: %w", arg, common.ErrNilPointer)
}
params := url.Values{}
if arg.RfqID != "" {
Expand Down Expand Up @@ -1541,7 +1493,7 @@ func (ok *Okx) GetCurrencyDepositHistory(ctx context.Context, currency, depositI
// Withdrawal to perform a withdrawal action. Sub-account does not support withdrawal.
func (ok *Okx) Withdrawal(ctx context.Context, input *WithdrawalInput) (*WithdrawalResponse, error) {
if input == nil {
return nil, errNilArgument
return nil, fmt.Errorf("%T: %w", input, common.ErrNilPointer)
}
var resp []WithdrawalResponse
switch {
Expand Down Expand Up @@ -1659,7 +1611,7 @@ func (ok *Okx) GetSavingBalance(ctx context.Context, currency string) ([]SavingB
// SavingsPurchaseOrRedemption creates a purchase or redemption instance
func (ok *Okx) SavingsPurchaseOrRedemption(ctx context.Context, arg *SavingsPurchaseRedemptionInput) (*SavingsPurchaseRedemptionResponse, error) {
if arg == nil {
return nil, errNilArgument
return nil, fmt.Errorf("%T: %w", arg, common.ErrNilPointer)
}
arg.ActionType = strings.ToLower(arg.ActionType)
switch {
Expand Down Expand Up @@ -1781,7 +1733,7 @@ func (ok *Okx) GetConvertCurrencyPair(ctx context.Context, fromCurrency, toCurre
// EstimateQuote retrieves quote estimation detail result given the base and quote currency.
func (ok *Okx) EstimateQuote(ctx context.Context, arg *EstimateQuoteRequestInput) (*EstimateQuoteResponse, error) {
if arg == nil {
return nil, errNilArgument
return nil, fmt.Errorf("%T: %w", arg, common.ErrNilPointer)
}
if arg.BaseCurrency == "" {
return nil, errors.New("missing base currency")
Expand Down Expand Up @@ -1813,7 +1765,7 @@ func (ok *Okx) EstimateQuote(ctx context.Context, arg *EstimateQuoteRequestInput
// ConvertTrade converts a base currency to quote currency.
func (ok *Okx) ConvertTrade(ctx context.Context, arg *ConvertTradeInput) (*ConvertTradeResponse, error) {
if arg == nil {
return nil, errNilArgument
return nil, fmt.Errorf("%T: %w", arg, common.ErrNilPointer)
}
if arg.BaseCurrency == "" {
return nil, errors.New("missing base currency")
Expand Down Expand Up @@ -1953,7 +1905,7 @@ func (ok *Okx) GetBillsDetail3Months(ctx context.Context, arg *BillsDetailQueryP
// GetBillsDetail retrieves the bills of the account.
func (ok *Okx) GetBillsDetail(ctx context.Context, arg *BillsDetailQueryParameter, route string) ([]BillsDetailResponse, error) {
if arg == nil {
return nil, errNilArgument
return nil, fmt.Errorf("%T: %w", arg, common.ErrNilPointer)
}
params := url.Values{}
if arg.InstrumentType != "" {
Expand Down Expand Up @@ -2413,13 +2365,14 @@ func (ok *Okx) ViewSubAccountList(ctx context.Context, enable bool, subaccountNa

// ResetSubAccountAPIKey applies to master accounts only and master accounts APIKey must be linked to IP addresses.
func (ok *Okx) ResetSubAccountAPIKey(ctx context.Context, arg *SubAccountAPIKeyParam) (*SubAccountAPIKeyResponse, error) {
params := url.Values{}
if arg == nil {
return nil, errNilArgument
return nil, fmt.Errorf("%T: %w", arg, common.ErrNilPointer)
}
if arg.SubAccountName == "" {
return nil, errInvalidSubAccountName
}
// TODO: Not used
params := url.Values{}
params.Set("subAcct", arg.SubAccountName)
if arg.APIKey == "" {
return nil, errInvalidAPIKey
Expand Down Expand Up @@ -2553,7 +2506,7 @@ func (ok *Okx) GetCustodyTradingSubaccountList(ctx context.Context, subaccountNa
// PlaceGridAlgoOrder place spot grid algo order.
func (ok *Okx) PlaceGridAlgoOrder(ctx context.Context, arg *GridAlgoOrder) (*GridAlgoOrderIDResponse, error) {
if arg == nil {
return nil, errNilArgument
return nil, fmt.Errorf("%T: %w", arg, common.ErrNilPointer)
}
if arg.InstrumentID == "" {
return nil, errMissingInstrumentID
Expand Down Expand Up @@ -3646,7 +3599,7 @@ func (ok *Okx) GetPublicUnderlyings(ctx context.Context, instrumentType string)
// GetInsuranceFundInformation returns insurance fund balance information.
func (ok *Okx) GetInsuranceFundInformation(ctx context.Context, arg *InsuranceFundInformationRequestParams) (*InsuranceFundInformation, error) {
if arg == nil {
return nil, errNilArgument
return nil, fmt.Errorf("%T: %w", arg, common.ErrNilPointer)
}
params := url.Values{}
if arg.InstrumentType == "" {
Expand Down
Loading
Loading