xm/ccxt_with_mt5
Template
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity
Files
master
ccxt_with_mt5/ccxt/pro/bybit.py
lz_db 0fab423a18 add
2025-11-16 12:31:03 +08:00

2444 lines
108 KiB
Python
Raw Permalink Blame History

# -*- coding: utf-8 -*-
# PLEASE DO NOT EDIT THIS FILE, IT IS GENERATED AND WILL BE OVERWRITTEN:
# https://github.com/ccxt/ccxt/blob/master/CONTRIBUTING.md#how-to-contribute-code
import ccxt.async_support
from ccxt.async_support.base.ws.cache import ArrayCache, ArrayCacheBySymbolById, ArrayCacheBySymbolBySide, ArrayCacheByTimestamp
import asyncio
import hashlib
from ccxt.base.types import Any, Balances, Bool, Int, Liquidation, Num, Order, OrderBook, OrderSide, OrderType, Position, Str, Strings, Ticker, Tickers, Trade
from ccxt.async_support.base.ws.client import Client
from typing import List
from ccxt.base.errors import ExchangeError
from ccxt.base.errors import AuthenticationError
from ccxt.base.errors import ArgumentsRequired
from ccxt.base.errors import BadRequest
from ccxt.base.errors import NotSupported
class bybit(ccxt.async_support.bybit):
def describe(self) -> Any:
return self.deep_extend(super(bybit, self).describe(), {
'has': {
'ws': True,
'createOrderWs': True,
'editOrderWs': True,
'fetchOpenOrdersWs': False,
'fetchOrderWs': False,
'cancelOrderWs': True,
'cancelOrdersWs': False,
'cancelAllOrdersWs': False,
'fetchTradesWs': False,
'fetchBalanceWs': False,
'watchBalance': True,
'watchBidsAsks': True,
'watchLiquidations': True,
'watchLiquidationsForSymbols': False,
'watchMyLiquidations': False,
'watchMyLiquidationsForSymbols': False,
'watchMyTrades': True,
'watchOHLCV': True,
'watchOHLCVForSymbols': True,
'watchOrderBook': True,
'watchOrderBookForSymbols': True,
'watchOrders': True,
'watchTicker': True,
'watchTickers': True,
'watchTrades': True,
'watchPositions': True,
'watchTradesForSymbols': True,
'unWatchTicker': True,
'unWatchTickers': True,
'unWatchOHLCV': True,
'unWatchOHLCVForSymbols': True,
'unWatchOrderBook': True,
'unWatchOrderBookForSymbols': True,
'unWatchTrades': True,
'unWatchTradesForSymbols': True,
'unWatchMyTrades': True,
'unWatchOrders': True,
'unWatchPositions': True,
},
'urls': {
'api': {
'ws': {
'public': {
'spot': 'wss://stream.{hostname}/v5/public/spot',
'inverse': 'wss://stream.{hostname}/v5/public/inverse',
'option': 'wss://stream.{hostname}/v5/public/option',
'linear': 'wss://stream.{hostname}/v5/public/linear',
},
'private': {
'spot': {
'unified': 'wss://stream.{hostname}/v5/private',
'nonUnified': 'wss://stream.{hostname}/spot/private/v3',
},
'contract': 'wss://stream.{hostname}/v5/private',
'usdc': 'wss://stream.{hostname}/trade/option/usdc/private/v1',
'trade': 'wss://stream.bybit.com/v5/trade',
},
},
},
'test': {
'ws': {
'public': {
'spot': 'wss://stream-testnet.{hostname}/v5/public/spot',
'inverse': 'wss://stream-testnet.{hostname}/v5/public/inverse',
'linear': 'wss://stream-testnet.{hostname}/v5/public/linear',
'option': 'wss://stream-testnet.{hostname}/v5/public/option',
},
'private': {
'spot': {
'unified': 'wss://stream-testnet.{hostname}/v5/private',
'nonUnified': 'wss://stream-testnet.{hostname}/spot/private/v3',
},
'contract': 'wss://stream-testnet.{hostname}/v5/private',
'usdc': 'wss://stream-testnet.{hostname}/trade/option/usdc/private/v1',
'trade': 'wss://stream-testnet.bybit.com/v5/trade',
},
},
},
'demotrading': {
'ws': {
'public': {
'spot': 'wss://stream.{hostname}/v5/public/spot',
'inverse': 'wss://stream.{hostname}/v5/public/inverse',
'option': 'wss://stream.{hostname}/v5/public/option',
'linear': 'wss://stream.{hostname}/v5/public/linear',
},
'private': {
'spot': {
'unified': 'wss://stream-demo.{hostname}/v5/private',
'nonUnified': 'wss://stream-demo.{hostname}/spot/private/v3',
},
'contract': 'wss://stream-demo.{hostname}/v5/private',
'usdc': 'wss://stream-demo.{hostname}/trade/option/usdc/private/v1',
'trade': 'wss://stream-demo.bybit.com/v5/trade',
},
},
},
},
'options': {
'watchTicker': {
'name': 'tickers', # 'tickers' for 24hr statistical ticker or 'tickers_lt' for leverage token ticker
},
'watchPositions': {
'fetchPositionsSnapshot': True, # or False
'awaitPositionsSnapshot': True, # whether to wait for the positions snapshot before providing updates
},
'watchMyTrades': {
# filter execType: https://bybit-exchange.github.io/docs/api-explorer/v5/position/execution
'filterExecTypes': [
'Trade', 'AdlTrade', 'BustTrade', 'Settle',
],
},
'spot': {
'timeframes': {
'1m': '1m',
'3m': '3m',
'5m': '5m',
'15m': '15m',
'30m': '30m',
'1h': '1h',
'2h': '2h',
'4h': '4h',
'6h': '6h',
'12h': '12h',
'1d': '1d',
'1w': '1w',
'1M': '1M',
},
},
'contract': {
'timeframes': {
'1m': '1',
'3m': '3',
'5m': '5',
'15m': '15',
'30m': '30',
'1h': '60',
'2h': '120',
'4h': '240',
'6h': '360',
'12h': '720',
'1d': 'D',
'1w': 'W',
'1M': 'M',
},
},
},
'streaming': {
'ping': self.ping,
'keepAlive': 18000,
},
})
def request_id(self):
requestId = self.sum(self.safe_integer(self.options, 'requestId', 0), 1)
self.options['requestId'] = requestId
return requestId
async def get_url_by_market_type(self, symbol: Str = None, isPrivate=False, method: Str = None, params={}):
accessibility = 'private' if isPrivate else 'public'
isUsdcSettled = None
isSpot = None
type = None
market = None
url = self.urls['api']['ws']
if symbol is not None:
market = self.market(symbol)
isUsdcSettled = market['settle'] == 'USDC'
type = market['type']
else:
type, params = self.handle_market_type_and_params(method, None, params)
defaultSettle = self.safe_string(self.options, 'defaultSettle')
defaultSettle = self.safe_string_2(params, 'settle', 'defaultSettle', defaultSettle)
isUsdcSettled = (defaultSettle == 'USDC')
isSpot = (type == 'spot')
if isPrivate:
unified = await self.isUnifiedEnabled()
isUnifiedMargin = self.safe_bool(unified, 0, False)
isUnifiedAccount = self.safe_bool(unified, 1, False)
if isUsdcSettled and not isUnifiedMargin and not isUnifiedAccount:
url = url[accessibility]['usdc']
else:
url = url[accessibility]['contract']
else:
if isSpot:
url = url[accessibility]['spot']
elif (type == 'swap') or (type == 'future'):
subType = None
subType, params = self.handle_sub_type_and_params(method, market, params, 'linear')
url = url[accessibility][subType]
else:
# option
url = url[accessibility]['option']
url = self.implode_hostname(url)
return url
def clean_params(self, params):
params = self.omit(params, ['type', 'subType', 'settle', 'defaultSettle', 'unifiedMargin'])
return params
async def create_order_ws(self, symbol: str, type: OrderType, side: OrderSide, amount: float, price: Num = None, params={}):
"""
create a trade order
https://bybit-exchange.github.io/docs/v5/order/create-order
https://bybit-exchange.github.io/docs/v5/websocket/trade/guideline#createamendcancel-order
:param str symbol: unified symbol of the market to create an order in
:param str type: 'market' or 'limit'
:param str side: 'buy' or 'sell'
:param float amount: how much of currency you want to trade in units of base currency
:param float [price]: the price at which the order is to be fulfilled, in units of the quote currency, ignored in market orders
:param dict [params]: extra parameters specific to the exchange API endpoint
:param str [params.timeInForce]: "GTC", "IOC", "FOK"
:param bool [params.postOnly]: True or False whether the order is post-only
:param bool [params.reduceOnly]: True or False whether the order is reduce-only
:param str [params.positionIdx]: *contracts only* 0 for one-way mode, 1 buy side of hedged mode, 2 sell side of hedged mode
:param boolean [params.isLeverage]: *unified spot only* False then spot trading True then margin trading
:param str [params.tpslMode]: *contract only* 'full' or 'partial'
:param str [params.mmp]: *option only* market maker protection
:param str [params.triggerDirection]: *contract only* the direction for trigger orders, 'above' or 'below'
:param float [params.triggerPrice]: The price at which a trigger order is triggered at
:param float [params.stopLossPrice]: The price at which a stop loss order is triggered at
:param float [params.takeProfitPrice]: The price at which a take profit order is triggered at
:param dict [params.takeProfit]: *takeProfit object in params* containing the triggerPrice at which the attached take profit order will be triggered
:param float [params.takeProfit.triggerPrice]: take profit trigger price
:param dict [params.stopLoss]: *stopLoss object in params* containing the triggerPrice at which the attached stop loss order will be triggered
:param float [params.stopLoss.triggerPrice]: stop loss trigger price
:param str [params.trailingAmount]: the quote amount to trail away from the current market price
:param str [params.trailingTriggerPrice]: the price to trigger a trailing order, default uses the price argument
:returns dict: an `order structure <https://docs.ccxt.com/#/?id=order-structure>`
"""
await self.load_markets()
orderRequest = self.create_order_request(symbol, type, side, amount, price, params, True)
url = self.urls['api']['ws']['private']['trade']
await self.authenticate(url)
requestId = str(self.request_id())
request: dict = {
'op': 'order.create',
'reqId': requestId,
'args': [
orderRequest,
],
'header': {
'X-BAPI-TIMESTAMP': str(self.milliseconds()),
'X-BAPI-RECV-WINDOW': str(self.options['recvWindow']),
},
}
return await self.watch(url, requestId, request, requestId, True)
async def edit_order_ws(self, id: str, symbol: str, type: OrderType, side: OrderSide, amount: Num = None, price: Num = None, params={}):
"""
edit a trade order
https://bybit-exchange.github.io/docs/v5/order/amend-order
https://bybit-exchange.github.io/docs/v5/websocket/trade/guideline#createamendcancel-order
:param str id: cancel order id
:param str symbol: unified symbol of the market to create an order in
:param str type: 'market' or 'limit'
:param str side: 'buy' or 'sell'
:param float amount: how much of currency you want to trade in units of base currency
:param float price: the price at which the order is to be fulfilled, in units of the quote currency, ignored in market orders
:param dict [params]: extra parameters specific to the exchange API endpoint
:param float [params.triggerPrice]: The price that a trigger order is triggered at
:param float [params.stopLossPrice]: The price that a stop loss order is triggered at
:param float [params.takeProfitPrice]: The price that a take profit order is triggered at
:param dict [params.takeProfit]: *takeProfit object in params* containing the triggerPrice that the attached take profit order will be triggered
:param float [params.takeProfit.triggerPrice]: take profit trigger price
:param dict [params.stopLoss]: *stopLoss object in params* containing the triggerPrice that the attached stop loss order will be triggered
:param float [params.stopLoss.triggerPrice]: stop loss trigger price
:param str [params.triggerBy]: 'IndexPrice', 'MarkPrice' or 'LastPrice', default is 'LastPrice', required if no initial value for triggerPrice
:param str [params.slTriggerBy]: 'IndexPrice', 'MarkPrice' or 'LastPrice', default is 'LastPrice', required if no initial value for stopLoss
:param str [params.tpTriggerby]: 'IndexPrice', 'MarkPrice' or 'LastPrice', default is 'LastPrice', required if no initial value for takeProfit
:returns dict: an `order structure <https://docs.ccxt.com/#/?id=order-structure>`
"""
await self.load_markets()
orderRequest = self.edit_order_request(id, symbol, type, side, amount, price, params)
url = self.urls['api']['ws']['private']['trade']
await self.authenticate(url)
requestId = str(self.request_id())
request: dict = {
'op': 'order.amend',
'reqId': requestId,
'args': [
orderRequest,
],
'header': {
'X-BAPI-TIMESTAMP': str(self.milliseconds()),
'X-BAPI-RECV-WINDOW': str(self.options['recvWindow']),
},
}
return await self.watch(url, requestId, request, requestId, True)
async def cancel_order_ws(self, id: str, symbol: Str = None, params={}):
"""
cancels an open order
https://bybit-exchange.github.io/docs/v5/order/cancel-order
https://bybit-exchange.github.io/docs/v5/websocket/trade/guideline#createamendcancel-order
:param str id: order id
:param str symbol: unified symbol of the market the order was made in
:param dict [params]: extra parameters specific to the exchange API endpoint
:param boolean [params.trigger]: *spot only* whether the order is a trigger order
:param str [params.orderFilter]: *spot only* 'Order' or 'StopOrder' or 'tpslOrder'
:returns dict: An `order structure <https://docs.ccxt.com/#/?id=order-structure>`
"""
await self.load_markets()
if symbol is None:
raise ArgumentsRequired(self.id + ' cancelOrderWs() requires a symbol argument')
orderRequest = self.cancel_order_request(id, symbol, params)
url = self.urls['api']['ws']['private']['trade']
await self.authenticate(url)
requestId = str(self.request_id())
if 'orderFilter' in orderRequest:
del orderRequest['orderFilter']
request: dict = {
'op': 'order.cancel',
'reqId': requestId,
'args': [
orderRequest,
],
'header': {
'X-BAPI-TIMESTAMP': str(self.milliseconds()),
'X-BAPI-RECV-WINDOW': str(self.options['recvWindow']),
},
}
return await self.watch(url, requestId, request, requestId, True)
async def watch_ticker(self, symbol: str, params={}) -> Ticker:
"""
watches a price ticker, a statistical calculation with the information calculated over the past 24 hours for a specific market
https://bybit-exchange.github.io/docs/v5/websocket/public/ticker
https://bybit-exchange.github.io/docs/v5/websocket/public/etp-ticker
:param str symbol: unified symbol of the market to fetch the ticker for
:param dict [params]: extra parameters specific to the exchange API endpoint
:returns dict: a `ticker structure <https://docs.ccxt.com/#/?id=ticker-structure>`
"""
await self.load_markets()
market = self.market(symbol)
symbol = market['symbol']
messageHash = 'ticker:' + symbol
url = await self.get_url_by_market_type(symbol, False, 'watchTicker', params)
params = self.clean_params(params)
options = self.safe_value(self.options, 'watchTicker', {})
topic = self.safe_string(options, 'name', 'tickers')
if not market['spot'] and topic != 'tickers':
raise BadRequest(self.id + ' watchTicker() only supports name tickers for contract markets')
topic += '.' + market['id']
topics = [topic]
return await self.watch_topics(url, [messageHash], topics, params)
async def watch_tickers(self, symbols: Strings = None, params={}) -> Tickers:
"""
watches a price ticker, a statistical calculation with the information calculated over the past 24 hours for all markets of a specific list
https://bybit-exchange.github.io/docs/v5/websocket/public/ticker
https://bybit-exchange.github.io/docs/v5/websocket/public/etp-ticker
:param str[] symbols: unified symbol of the market to fetch the ticker for
:param dict [params]: extra parameters specific to the exchange API endpoint
:returns dict: a `ticker structure <https://docs.ccxt.com/#/?id=ticker-structure>`
"""
await self.load_markets()
symbols = self.market_symbols(symbols, None, False)
messageHashes = []
url = await self.get_url_by_market_type(symbols[0], False, 'watchTickers', params)
params = self.clean_params(params)
options = self.safe_value(self.options, 'watchTickers', {})
topic = self.safe_string(options, 'name', 'tickers')
marketIds = self.market_ids(symbols)
topics = []
for i in range(0, len(marketIds)):
marketId = marketIds[i]
topics.append(topic + '.' + marketId)
messageHashes.append('ticker:' + symbols[i])
ticker = await self.watch_topics(url, messageHashes, topics, params)
if self.newUpdates:
result: dict = {}
result[ticker['symbol']] = ticker
return result
return self.filter_by_array(self.tickers, 'symbol', symbols)
async def un_watch_tickers(self, symbols: Strings = None, params={}) -> Any:
"""
unWatches a price ticker
https://bybit-exchange.github.io/docs/v5/websocket/public/ticker
https://bybit-exchange.github.io/docs/v5/websocket/public/etp-ticker
:param str[] symbols: unified symbol of the market to fetch the ticker for
:param dict [params]: extra parameters specific to the exchange API endpoint
:returns dict: a `ticker structure <https://docs.ccxt.com/#/?id=ticker-structure>`
"""
await self.load_markets()
symbols = self.market_symbols(symbols, None, False)
options = self.safe_value(self.options, 'watchTickers', {})
topic = self.safe_string(options, 'name', 'tickers')
messageHashes = []
subMessageHashes = []
marketIds = self.market_ids(symbols)
topics = []
for i in range(0, len(marketIds)):
marketId = marketIds[i]
symbol = symbols[i]
topics.append(topic + '.' + marketId)
subMessageHashes.append('ticker:' + symbol)
messageHashes.append('unsubscribe:ticker:' + symbol)
url = await self.get_url_by_market_type(symbols[0], False, 'watchTickers', params)
return await self.un_watch_topics(url, 'ticker', symbols, messageHashes, subMessageHashes, topics, params)
async def un_watch_ticker(self, symbol: str, params={}) -> Any:
"""
unWatches a price ticker
https://bybit-exchange.github.io/docs/v5/websocket/public/ticker
https://bybit-exchange.github.io/docs/v5/websocket/public/etp-ticker
:param str[] symbol: unified symbol of the market to fetch the ticker for
:param dict [params]: extra parameters specific to the exchange API endpoint
:returns dict: a `ticker structure <https://docs.ccxt.com/#/?id=ticker-structure>`
"""
await self.load_markets()
return await self.un_watch_tickers([symbol], params)
def handle_ticker(self, client: Client, message):
#
# linear
# {
# "topic": "tickers.BTCUSDT",
# "type": "snapshot",
# "data": {
# "symbol": "BTCUSDT",
# "tickDirection": "PlusTick",
# "price24hPcnt": "0.017103",
# "lastPrice": "17216.00",
# "prevPrice24h": "16926.50",
# "highPrice24h": "17281.50",
# "lowPrice24h": "16915.00",
# "prevPrice1h": "17238.00",
# "markPrice": "17217.33",
# "indexPrice": "17227.36",
# "openInterest": "68744.761",
# "openInterestValue": "1183601235.91",
# "turnover24h": "1570383121.943499",
# "volume24h": "91705.276",
# "nextFundingTime": "1673280000000",
# "fundingRate": "-0.000212",
# "bid1Price": "17215.50",
# "bid1Size": "84.489",
# "ask1Price": "17216.00",
# "ask1Size": "83.020"
# },
# "cs": 24987956059,
# "ts": 1673272861686
# }
#
# option
# {
# "id": "tickers.BTC-6JAN23-17500-C-2480334983-1672917511074",
# "topic": "tickers.BTC-6JAN23-17500-C",
# "ts": 1672917511074,
# "data": {
# "symbol": "BTC-6JAN23-17500-C",
# "bidPrice": "0",
# "bidSize": "0",
# "bidIv": "0",
# "askPrice": "10",
# "askSize": "5.1",
# "askIv": "0.514",
# "lastPrice": "10",
# "highPrice24h": "25",
# "lowPrice24h": "5",
# "markPrice": "7.86976724",
# "indexPrice": "16823.73",
# "markPriceIv": "0.4896",
# "underlyingPrice": "16815.1",
# "openInterest": "49.85",
# "turnover24h": "446802.8473",
# "volume24h": "26.55",
# "totalVolume": "86",
# "totalTurnover": "1437431",
# "delta": "0.047831",
# "gamma": "0.00021453",
# "vega": "0.81351067",
# "theta": "-19.9115368",
# "predictedDeliveryPrice": "0",
# "change24h": "-0.33333334"
# },
# "type": "snapshot"
# }
#
# spot
# {
# "topic": "tickers.BTCUSDT",
# "ts": 1673853746003,
# "type": "snapshot",
# "cs": 2588407389,
# "data": {
# "symbol": "BTCUSDT",
# "lastPrice": "21109.77",
# "highPrice24h": "21426.99",
# "lowPrice24h": "20575",
# "prevPrice24h": "20704.93",
# "volume24h": "6780.866843",
# "turnover24h": "141946527.22907118",
# "price24hPcnt": "0.0196",
# "usdIndexPrice": "21120.2400136"
# }
# }
#
# lt ticker
# {
# "topic": "tickers_lt.EOS3LUSDT",
# "ts": 1672325446847,
# "type": "snapshot",
# "data": {
# "symbol": "EOS3LUSDT",
# "lastPrice": "0.41477848043290448",
# "highPrice24h": "0.435285472510871305",
# "lowPrice24h": "0.394601507960931382",
# "prevPrice24h": "0.431502290172376349",
# "price24hPcnt": "-0.0388"
# }
# }
# swap delta
# {
# "topic":"tickers.AAVEUSDT",
# "type":"delta",
# "data":{
# "symbol":"AAVEUSDT",
# "bid1Price":"112.89",
# "bid1Size":"2.12",
# "ask1Price":"112.90",
# "ask1Size":"5.02"
# },
# "cs":78039939929,
# "ts":1709210212704
# }
#
topic = self.safe_string(message, 'topic', '')
updateType = self.safe_string(message, 'type', '')
data = self.safe_dict(message, 'data', {})
isSpot = self.safe_string(data, 'usdIndexPrice') is not None
type = 'spot' if isSpot else 'contract'
symbol = None
parsed = None
if (updateType == 'snapshot'):
parsed = self.parse_ticker(data)
symbol = parsed['symbol']
elif updateType == 'delta':
topicParts = topic.split('.')
topicLength = len(topicParts)
marketId = self.safe_string(topicParts, topicLength - 1)
market = self.safe_market(marketId, None, None, type)
symbol = market['symbol']
# update the info in place
ticker = self.safe_dict(self.tickers, symbol, {})
rawTicker = self.safe_dict(ticker, 'info', {})
merged = self.extend(rawTicker, data)
parsed = self.parse_ticker(merged)
timestamp = self.safe_integer(message, 'ts')
parsed['timestamp'] = timestamp
parsed['datetime'] = self.iso8601(timestamp)
self.tickers[symbol] = parsed
messageHash = 'ticker:' + symbol
client.resolve(self.tickers[symbol], messageHash)
async def watch_bids_asks(self, symbols: Strings = None, params={}) -> Tickers:
"""
watches best bid & ask for symbols
https://bybit-exchange.github.io/docs/v5/websocket/public/orderbook
:param str[] symbols: unified symbol of the market to fetch the ticker for
:param dict [params]: extra parameters specific to the exchange API endpoint
:returns dict: a `ticker structure <https://docs.ccxt.com/#/?id=ticker-structure>`
"""
await self.load_markets()
symbols = self.market_symbols(symbols, None, False)
messageHashes = []
url = await self.get_url_by_market_type(symbols[0], False, 'watchBidsAsks', params)
params = self.clean_params(params)
marketIds = self.market_ids(symbols)
topics = []
for i in range(0, len(marketIds)):
marketId = marketIds[i]
topic = 'orderbook.1.' + marketId
topics.append(topic)
messageHashes.append('bidask:' + symbols[i])
ticker = await self.watch_topics(url, messageHashes, topics, params)
if self.newUpdates:
return ticker
return self.filter_by_array(self.bidsasks, 'symbol', symbols)
def parse_ws_bid_ask(self, orderbook, market=None):
timestamp = self.safe_integer(orderbook, 'timestamp')
bids = self.sort_by(self.aggregate(orderbook['bids']), 0)
asks = self.sort_by(self.aggregate(orderbook['asks']), 0)
bestBid = self.safe_list(bids, 0, [])
bestAsk = self.safe_list(asks, 0, [])
return self.safe_ticker({
'symbol': market['symbol'],
'timestamp': timestamp,
'datetime': self.iso8601(timestamp),
'ask': self.safe_number(bestAsk, 0),
'askVolume': self.safe_number(bestAsk, 1),
'bid': self.safe_number(bestBid, 0),
'bidVolume': self.safe_number(bestBid, 1),
'info': orderbook,
}, market)
async def watch_ohlcv(self, symbol: str, timeframe: str = '1m', since: Int = None, limit: Int = None, params={}) -> List[list]:
"""
watches historical candlestick data containing the open, high, low, and close price, and the volume of a market
https://bybit-exchange.github.io/docs/v5/websocket/public/kline
https://bybit-exchange.github.io/docs/v5/websocket/public/etp-kline
:param str symbol: unified symbol of the market to fetch OHLCV data for
:param str timeframe: the length of time each candle represents
:param int [since]: timestamp in ms of the earliest candle to fetch
:param int [limit]: the maximum amount of candles to fetch
:param dict [params]: extra parameters specific to the exchange API endpoint
:returns int[][]: A list of candles ordered, open, high, low, close, volume
"""
params['callerMethodName'] = 'watchOHLCV'
result = await self.watch_ohlcv_for_symbols([[symbol, timeframe]], since, limit, params)
return result[symbol][timeframe]
async def watch_ohlcv_for_symbols(self, symbolsAndTimeframes: List[List[str]], since: Int = None, limit: Int = None, params={}):
"""
watches historical candlestick data containing the open, high, low, and close price, and the volume of a market
https://bybit-exchange.github.io/docs/v5/websocket/public/kline
https://bybit-exchange.github.io/docs/v5/websocket/public/etp-kline
:param str[][] symbolsAndTimeframes: array of arrays containing unified symbols and timeframes to fetch OHLCV data for, example [['BTC/USDT', '1m'], ['LTC/USDT', '5m']]
:param int [since]: timestamp in ms of the earliest candle to fetch
:param int [limit]: the maximum amount of candles to fetch
:param dict [params]: extra parameters specific to the exchange API endpoint
:returns dict: A list of candles ordered, open, high, low, close, volume
"""
await self.load_markets()
symbols = self.get_list_from_object_values(symbolsAndTimeframes, 0)
marketSymbols = self.market_symbols(symbols, None, False, True, True)
firstSymbol = marketSymbols[0]
url = await self.get_url_by_market_type(firstSymbol, False, 'watchOHLCVForSymbols', params)
rawHashes = []
messageHashes = []
for i in range(0, len(symbolsAndTimeframes)):
data = symbolsAndTimeframes[i]
symbolString = self.safe_string(data, 0)
market = self.market(symbolString)
symbolString = market['symbol']
unfiedTimeframe = self.safe_string(data, 1)
timeframeId = self.safe_string(self.timeframes, unfiedTimeframe, unfiedTimeframe)
rawHashes.append('kline.' + timeframeId + '.' + market['id'])
messageHashes.append('ohlcv::' + symbolString + '::' + unfiedTimeframe)
symbol, timeframe, stored = await self.watch_topics(url, messageHashes, rawHashes, params)
if self.newUpdates:
limit = stored.getLimit(symbol, limit)
filtered = self.filter_by_since_limit(stored, since, limit, 0, True)
return self.create_ohlcv_object(symbol, timeframe, filtered)
async def un_watch_ohlcv_for_symbols(self, symbolsAndTimeframes: List[List[str]], params={}) -> Any:
"""
unWatches historical candlestick data containing the open, high, low, and close price, and the volume of a market
https://bybit-exchange.github.io/docs/v5/websocket/public/kline
https://bybit-exchange.github.io/docs/v5/websocket/public/etp-kline
:param str[][] symbolsAndTimeframes: array of arrays containing unified symbols and timeframes to fetch OHLCV data for, example [['BTC/USDT', '1m'], ['LTC/USDT', '5m']]
:param dict [params]: extra parameters specific to the exchange API endpoint
:returns dict: A list of candles ordered, open, high, low, close, volume
"""
await self.load_markets()
symbols = self.get_list_from_object_values(symbolsAndTimeframes, 0)
marketSymbols = self.market_symbols(symbols, None, False, True, True)
firstSymbol = marketSymbols[0]
url = await self.get_url_by_market_type(firstSymbol, False, 'watchOHLCVForSymbols', params)
rawHashes = []
subMessageHashes = []
messageHashes = []
for i in range(0, len(symbolsAndTimeframes)):
data = symbolsAndTimeframes[i]
symbolString = self.safe_string(data, 0)
market = self.market(symbolString)
symbolString = market['symbol']
unfiedTimeframe = self.safe_string(data, 1)
timeframeId = self.safe_string(self.timeframes, unfiedTimeframe, unfiedTimeframe)
rawHashes.append('kline.' + timeframeId + '.' + market['id'])
subMessageHashes.append('ohlcv::' + symbolString + '::' + unfiedTimeframe)
messageHashes.append('unsubscribe::ohlcv::' + symbolString + '::' + unfiedTimeframe)
subExtension = {
'symbolsAndTimeframes': symbolsAndTimeframes,
}
return await self.un_watch_topics(url, 'ohlcv', symbols, messageHashes, subMessageHashes, rawHashes, params, subExtension)
async def un_watch_ohlcv(self, symbol: str, timeframe: str = '1m', params={}) -> Any:
"""
unWatches historical candlestick data containing the open, high, low, and close price, and the volume of a market
https://bybit-exchange.github.io/docs/v5/websocket/public/kline
https://bybit-exchange.github.io/docs/v5/websocket/public/etp-kline
:param str symbol: unified symbol of the market to fetch OHLCV data for
:param str timeframe: the length of time each candle represents
:param dict [params]: extra parameters specific to the exchange API endpoint
:returns int[][]: A list of candles ordered, open, high, low, close, volume
"""
params['callerMethodName'] = 'watchOHLCV'
return await self.un_watch_ohlcv_for_symbols([[symbol, timeframe]], params)
def handle_ohlcv(self, client: Client, message):
#
# {
# "topic": "kline.5.BTCUSDT",
# "data": [
# {
# "start": 1672324800000,
# "end": 1672325099999,
# "interval": "5",
# "open": "16649.5",
# "close": "16677",
# "high": "16677",
# "low": "16608",
# "volume": "2.081",
# "turnover": "34666.4005",
# "confirm": False,
# "timestamp": 1672324988882
# }
# ],
# "ts": 1672324988882,
# "type": "snapshot"
# }
#
data = self.safe_value(message, 'data', {})
topic = self.safe_string(message, 'topic')
topicParts = topic.split('.')
topicLength = len(topicParts)
timeframeId = self.safe_string(topicParts, 1)
timeframe = self.find_timeframe(timeframeId)
marketId = self.safe_string(topicParts, topicLength - 1)
isSpot = client.url.find('spot') > -1
marketType = 'spot' if isSpot else 'contract'
market = self.safe_market(marketId, None, None, marketType)
symbol = market['symbol']
ohlcvsByTimeframe = self.safe_value(self.ohlcvs, symbol)
if ohlcvsByTimeframe is None:
self.ohlcvs[symbol] = {}
if self.safe_value(ohlcvsByTimeframe, timeframe) is None:
limit = self.safe_integer(self.options, 'OHLCVLimit', 1000)
self.ohlcvs[symbol][timeframe] = ArrayCacheByTimestamp(limit)
stored = self.ohlcvs[symbol][timeframe]
for i in range(0, len(data)):
parsed = self.parse_ws_ohlcv(data[i], market)
stored.append(parsed)
messageHash = 'ohlcv::' + symbol + '::' + timeframe
resolveData = [symbol, timeframe, stored]
client.resolve(resolveData, messageHash)
def parse_ws_ohlcv(self, ohlcv, market=None) -> list:
#
# {
# "start": 1670363160000,
# "end": 1670363219999,
# "interval": "1",
# "open": "16987.5",
# "close": "16987.5",
# "high": "16988",
# "low": "16987.5",
# "volume": "23.511",
# "turnover": "399396.344",
# "confirm": False,
# "timestamp": 1670363219614
# }
#
volumeIndex = 'turnover' if (market['inverse']) else 'volume'
return [
self.safe_integer(ohlcv, 'start'),
self.safe_number(ohlcv, 'open'),
self.safe_number(ohlcv, 'high'),
self.safe_number(ohlcv, 'low'),
self.safe_number(ohlcv, 'close'),
self.safe_number(ohlcv, volumeIndex),
]
async def watch_order_book(self, symbol: str, limit: Int = None, params={}) -> OrderBook:
"""
watches information on open orders with bid(buy) and ask(sell) prices, volumes and other data
https://bybit-exchange.github.io/docs/v5/websocket/public/orderbook
:param str symbol: unified symbol of the market to fetch the order book for
:param int [limit]: the maximum amount of order book entries to return.
:param dict [params]: extra parameters specific to the exchange API endpoint
:returns dict: A dictionary of `order book structures <https://docs.ccxt.com/#/?id=order-book-structure>` indexed by market symbols
"""
return await self.watch_order_book_for_symbols([symbol], limit, params)
async def watch_order_book_for_symbols(self, symbols: List[str], limit: Int = None, params={}) -> OrderBook:
"""
watches information on open orders with bid(buy) and ask(sell) prices, volumes and other data
https://bybit-exchange.github.io/docs/v5/websocket/public/orderbook
:param str[] symbols: unified array of symbols
:param int [limit]: the maximum amount of order book entries to return.
:param dict [params]: extra parameters specific to the exchange API endpoint
:returns dict: A dictionary of `order book structures <https://docs.ccxt.com/#/?id=order-book-structure>` indexed by market symbols
"""
await self.load_markets()
symbolsLength = len(symbols)
if symbolsLength == 0:
raise ArgumentsRequired(self.id + ' watchOrderBookForSymbols() requires a non-empty array of symbols')
symbols = self.market_symbols(symbols)
url = await self.get_url_by_market_type(symbols[0], False, 'watchOrderBook', params)
params = self.clean_params(params)
market = self.market(symbols[0])
if limit is None:
limit = 50 if (market['spot']) else 500
if market['option']:
limit = 100
else:
limits = {
'spot': [1, 50, 200, 1000],
'option': [25, 100],
'default': [1, 50, 200, 500, 1000],
}
selectedLimits = self.safe_list_2(limits, market['type'], 'default')
if not self.in_array(limit, selectedLimits):
raise BadRequest(self.id + ' watchOrderBookForSymbols(): for ' + market['type'] + ' markets limit can be one of: ' + self.json(selectedLimits))
topics = []
messageHashes = []
for i in range(0, len(symbols)):
symbol = symbols[i]
marketId = self.market_id(symbol)
topic = 'orderbook.' + str(limit) + '.' + marketId
topics.append(topic)
messageHash = 'orderbook:' + symbol
messageHashes.append(messageHash)
orderbook = await self.watch_topics(url, messageHashes, topics, params)
return orderbook.limit()
async def un_watch_order_book_for_symbols(self, symbols: List[str], params={}) -> Any:
"""
unsubscribe from the orderbook channel
https://bybit-exchange.github.io/docs/v5/websocket/public/orderbook
:param str[] symbols: unified symbol of the market to unwatch the trades for
:param dict [params]: extra parameters specific to the exchange API endpoint
:param int [params.limit]: orderbook limit, default is None
:returns dict: A dictionary of `order book structures <https://docs.ccxt.com/#/?id=order-book-structure>` indexed by market symbols
"""
await self.load_markets()
symbols = self.market_symbols(symbols, None, False)
channel = 'orderbook.'
limit = self.safe_integer(params, 'limit')
if limit is not None:
params = self.omit(params, 'limit')
else:
firstMarket = self.market(symbols[0])
limit = 50 if firstMarket['spot'] else 500
channel += str(limit)
subMessageHashes = []
messageHashes = []
topics = []
for i in range(0, len(symbols)):
symbol = symbols[i]
market = self.market(symbol)
marketId = market['id']
topic = channel + '.' + marketId
messageHashes.append('unsubscribe:orderbook:' + symbol)
subMessageHashes.append('orderbook:' + symbol)
topics.append(topic)
url = await self.get_url_by_market_type(symbols[0], False, 'watchOrderBook', params)
return await self.un_watch_topics(url, 'orderbook', symbols, messageHashes, subMessageHashes, topics, params)
async def un_watch_order_book(self, symbol: str, params={}) -> Any:
"""
unsubscribe from the orderbook channel
https://bybit-exchange.github.io/docs/v5/websocket/public/orderbook
:param str symbol: symbol of the market to unwatch the trades for
:param dict [params]: extra parameters specific to the exchange API endpoint
:param int [params.limit]: orderbook limit, default is None
:returns dict: A dictionary of `order book structures <https://docs.ccxt.com/#/?id=order-book-structure>` indexed by market symbols
"""
await self.load_markets()
return await self.un_watch_order_book_for_symbols([symbol], params)
def handle_order_book(self, client: Client, message):
#
# {
# "topic": "orderbook.50.BTCUSDT",
# "type": "snapshot",
# "ts": 1672304484978,
# "data": {
# "s": "BTCUSDT",
# "b": [
# ...,
# [
# "16493.50",
# "0.006"
# ],
# [
# "16493.00",
# "0.100"
# ]
# ],
# "a": [
# [
# "16611.00",
# "0.029"
# ],
# [
# "16612.00",
# "0.213"
# ],
# ],
# "u": 18521288,
# "seq": 7961638724
# }
# }
#
topic = self.safe_string(message, 'topic')
limit = topic.split('.')[1]
isSpot = client.url.find('spot') >= 0
type = self.safe_string(message, 'type')
isSnapshot = (type == 'snapshot')
data = self.safe_dict(message, 'data', {})
marketId = self.safe_string(data, 's')
marketType = 'spot' if isSpot else 'contract'
market = self.safe_market(marketId, None, None, marketType)
symbol = market['symbol']
timestamp = self.safe_integer(message, 'ts')
if not (symbol in self.orderbooks):
self.orderbooks[symbol] = self.order_book()
orderbook = self.orderbooks[symbol]
orderbook['symbol'] = symbol
if isSnapshot:
snapshot = self.parse_order_book(data, symbol, timestamp, 'b', 'a')
orderbook.reset(snapshot)
else:
asks = self.safe_list(data, 'a', [])
bids = self.safe_list(data, 'b', [])
self.handle_deltas(orderbook['asks'], asks)
self.handle_deltas(orderbook['bids'], bids)
orderbook['timestamp'] = timestamp
orderbook['datetime'] = self.iso8601(timestamp)
messageHash = 'orderbook' + ':' + symbol
self.orderbooks[symbol] = orderbook
client.resolve(orderbook, messageHash)
if limit == '1':
bidask = self.parse_ws_bid_ask(self.orderbooks[symbol], market)
newBidsAsks: dict = {}
newBidsAsks[symbol] = bidask
self.bidsasks[symbol] = bidask
client.resolve(newBidsAsks, 'bidask:' + symbol)
def handle_delta(self, bookside, delta):
bidAsk = self.parse_bid_ask(delta, 0, 1)
bookside.storeArray(bidAsk)
def handle_deltas(self, bookside, deltas):
for i in range(0, len(deltas)):
self.handle_delta(bookside, deltas[i])
async def watch_trades(self, symbol: str, since: Int = None, limit: Int = None, params={}) -> List[Trade]:
"""
watches information on multiple trades made in a market
https://bybit-exchange.github.io/docs/v5/websocket/public/trade
:param str symbol: unified market symbol of the market trades were made in
:param int [since]: the earliest time in ms to fetch trades for
:param int [limit]: the maximum number of trade structures to retrieve
:param dict [params]: extra parameters specific to the exchange API endpoint
:returns dict[]: a list of `trade structures <https://docs.ccxt.com/#/?id=trade-structure>`
"""
return await self.watch_trades_for_symbols([symbol], since, limit, params)
async def watch_trades_for_symbols(self, symbols: List[str], since: Int = None, limit: Int = None, params={}) -> List[Trade]:
"""
get the list of most recent trades for a list of symbols
https://bybit-exchange.github.io/docs/v5/websocket/public/trade
:param str[] symbols: unified symbol of the market to fetch trades for
:param int [since]: timestamp in ms of the earliest trade to fetch
:param int [limit]: the maximum amount of trades to fetch
:param dict [params]: extra parameters specific to the exchange API endpoint
:returns dict[]: a list of `trade structures <https://docs.ccxt.com/#/?id=public-trades>`
"""
await self.load_markets()
symbols = self.market_symbols(symbols)
symbolsLength = len(symbols)
if symbolsLength == 0:
raise ArgumentsRequired(self.id + ' watchTradesForSymbols() requires a non-empty array of symbols')
params = self.clean_params(params)
url = await self.get_url_by_market_type(symbols[0], False, 'watchTrades', params)
topics = []
messageHashes = []
for i in range(0, len(symbols)):
symbol = symbols[i]
market = self.market(symbol)
topic = 'publicTrade.' + market['id']
topics.append(topic)
messageHash = 'trade:' + symbol
messageHashes.append(messageHash)
trades = await self.watch_topics(url, messageHashes, topics, params)
if self.newUpdates:
first = self.safe_value(trades, 0)
tradeSymbol = self.safe_string(first, 'symbol')
limit = trades.getLimit(tradeSymbol, limit)
return self.filter_by_since_limit(trades, since, limit, 'timestamp', True)
async def un_watch_trades_for_symbols(self, symbols: List[str], params={}) -> Any:
"""
unsubscribe from the trades channel
https://bybit-exchange.github.io/docs/v5/websocket/public/trade
:param str[] symbols: unified symbol of the market to unwatch the trades for
:param dict [params]: extra parameters specific to the exchange API endpoint
:returns any: status of the unwatch request
"""
await self.load_markets()
symbols = self.market_symbols(symbols, None, False, True)
url = await self.get_url_by_market_type(symbols[0], False, 'unWatchTradesForSymbols', params)
messageHashes = []
topics = []
subMessageHashes = []
for i in range(0, len(symbols)):
symbol = symbols[i]
market = self.market(symbol)
topic = 'publicTrade.' + market['id']
topics.append(topic)
messageHash = 'unsubscribe:trade:' + symbol
messageHashes.append(messageHash)
subMessageHashes.append('trade:' + symbol)
return await self.un_watch_topics(url, 'trades', symbols, messageHashes, subMessageHashes, topics, params)
async def un_watch_trades(self, symbol: str, params={}) -> Any:
"""
unsubscribe from the trades channel
https://bybit-exchange.github.io/docs/v5/websocket/public/trade
:param str symbol: unified symbol of the market to unwatch the trades for
:param dict [params]: extra parameters specific to the exchange API endpoint
:returns any: status of the unwatch request
"""
await self.load_markets()
return await self.un_watch_trades_for_symbols([symbol], params)
def handle_trades(self, client: Client, message):
#
# {
# "topic": "publicTrade.BTCUSDT",
# "type": "snapshot",
# "ts": 1672304486868,
# "data": [
# {
# "T": 1672304486865,
# "s": "BTCUSDT",
# "S": "Buy",
# "v": "0.001",
# "p": "16578.50",
# "L": "PlusTick",
# "i": "20f43950-d8dd-5b31-9112-a178eb6023af",
# "BT": False
# }
# ]
# }
#
data = self.safe_value(message, 'data', {})
topic = self.safe_string(message, 'topic')
trades = data
parts = topic.split('.')
isSpot = client.url.find('spot') >= 0
marketType = 'spot' if (isSpot) else 'contract'
marketId = self.safe_string(parts, 1)
market = self.safe_market(marketId, None, None, marketType)
symbol = market['symbol']
stored = self.safe_value(self.trades, symbol)
if stored is None:
limit = self.safe_integer(self.options, 'tradesLimit', 1000)
stored = ArrayCache(limit)
self.trades[symbol] = stored
for j in range(0, len(trades)):
parsed = self.parse_ws_trade(trades[j], market)
stored.append(parsed)
messageHash = 'trade' + ':' + symbol
client.resolve(stored, messageHash)
def parse_ws_trade(self, trade, market=None):
#
# public
# {
# "T": 1672304486865,
# "s": "BTCUSDT",
# "S": "Buy",
# "v": "0.001",
# "p": "16578.50",
# "L": "PlusTick",
# "i": "20f43950-d8dd-5b31-9112-a178eb6023af",
# "BT": False
# }
#
# spot private
# {
# "e": "ticketInfo",
# "E": "1662348310386",
# "s": "BTCUSDT",
# "q": "0.001007",
# "t": "1662348310373",
# "p": "19842.02",
# "T": "2100000000002220938",
# "o": "1238261807653647872",
# "c": "spotx008",
# "O": "1238225004531834368",
# "a": "533287",
# "A": "642908",
# "m": False,
# "S": "BUY"
# }
#
id = self.safe_string_n(trade, ['i', 'T', 'v'])
isContract = ('BT' in trade)
marketType = 'contract' if isContract else 'spot'
if market is not None:
marketType = market['type']
marketId = self.safe_string(trade, 's')
market = self.safe_market(marketId, market, None, marketType)
symbol = market['symbol']
timestamp = self.safe_integer_2(trade, 't', 'T')
side = self.safe_string_lower(trade, 'S')
takerOrMaker = None
m = self.safe_value(trade, 'm')
if side is None:
side = 'buy' if m else 'sell'
else:
# spot private
takerOrMaker = m
price = self.safe_string(trade, 'p')
amount = self.safe_string_2(trade, 'q', 'v')
orderId = self.safe_string(trade, 'o')
return self.safe_trade({
'id': id,
'info': trade,
'timestamp': timestamp,
'datetime': self.iso8601(timestamp),
'symbol': symbol,
'order': orderId,
'type': None,
'side': side,
'takerOrMaker': takerOrMaker,
'price': price,
'amount': amount,
'cost': None,
'fee': None,
}, market)
def get_private_type(self, url):
if url.find('spot') >= 0:
return 'spot'
elif url.find('v5/private') >= 0:
return 'unified'
else:
return 'usdc'
async def watch_my_trades(self, symbol: Str = None, since: Int = None, limit: Int = None, params={}) -> List[Trade]:
"""
watches information on multiple trades made by the user
https://bybit-exchange.github.io/docs/v5/websocket/private/execution
https://bybit-exchange.github.io/docs/v5/websocket/private/fast-execution
:param str symbol: unified market symbol of the market orders were made in
:param int [since]: the earliest time in ms to fetch orders for
:param int [limit]: the maximum number of order structures to retrieve
:param dict [params]: extra parameters specific to the exchange API endpoint
:param boolean [params.unifiedMargin]: use unified margin account
:param boolean [params.executionFast]: use fast execution
:returns dict[]: a list of `order structures <https://docs.ccxt.com/#/?id=order-structure>`
"""
method = 'watchMyTrades'
messageHash = 'myTrades'
await self.load_markets()
if symbol is not None:
symbol = self.symbol(symbol)
messageHash += ':' + symbol
url = await self.get_url_by_market_type(symbol, True, method, params)
await self.authenticate(url)
topicByMarket: dict = {
'spot': 'ticketInfo',
'unified': 'execution',
'usdc': 'user.openapi.perp.trade',
}
topic = self.safe_value(topicByMarket, self.get_private_type(url))
executionFast = False
executionFast, params = self.handle_option_and_params(params, 'watchMyTrades', 'executionFast', False)
if executionFast:
topic = 'execution.fast'
trades = await self.watch_topics(url, [messageHash], [topic], params)
if self.newUpdates:
limit = trades.getLimit(symbol, limit)
return self.filter_by_symbol_since_limit(trades, symbol, since, limit, True)
async def un_watch_my_trades(self, symbol: Str = None, params={}) -> Any:
"""
unWatches information on multiple trades made by the user
https://bybit-exchange.github.io/docs/v5/websocket/private/execution
https://bybit-exchange.github.io/docs/v5/websocket/private/fast-execution
:param str symbol: unified market symbol of the market orders were made in
:param dict [params]: extra parameters specific to the exchange API endpoint
:param boolean [params.unifiedMargin]: use unified margin account
:param boolean [params.executionFast]: use fast execution
:returns dict[]: a list of `order structures <https://docs.ccxt.com/#/?id=order-structure>`
"""
method = 'watchMyTrades'
messageHash = 'unsubscribe:myTrades'
subHash = 'myTrades'
await self.load_markets()
if symbol is not None:
raise NotSupported(self.id + ' unWatchMyTrades() does not support a symbol parameter, you must unwatch all my trades')
url = await self.get_url_by_market_type(symbol, True, method, params)
await self.authenticate(url)
topicByMarket: dict = {
'spot': 'ticketInfo',
'unified': 'execution',
'usdc': 'user.openapi.perp.trade',
}
topic = self.safe_value(topicByMarket, self.get_private_type(url))
executionFast = False
executionFast, params = self.handle_option_and_params(params, 'watchMyTrades', 'executionFast', False)
if executionFast:
topic = 'execution.fast'
return await self.un_watch_topics(url, 'myTrades', [], [messageHash], [subHash], [topic], params)
def handle_my_trades(self, client: Client, message):
#
# spot
# {
# "type": "snapshot",
# "topic": "ticketInfo",
# "ts": "1662348310388",
# "data": [
# {
# "e": "ticketInfo",
# "E": "1662348310386",
# "s": "BTCUSDT",
# "q": "0.001007",
# "t": "1662348310373",
# "p": "19842.02",
# "T": "2100000000002220938",
# "o": "1238261807653647872",
# "c": "spotx008",
# "O": "1238225004531834368",
# "a": "533287",
# "A": "642908",
# "m": False,
# "S": "BUY"
# }
# ]
# }
# unified
# {
# "id": "592324803b2785-26fa-4214-9963-bdd4727f07be",
# "topic": "execution",
# "creationTime": 1672364174455,
# "data": [
# {
# "category": "linear",
# "symbol": "XRPUSDT",
# "execFee": "0.005061",
# "execId": "7e2ae69c-4edf-5800-a352-893d52b446aa",
# "execPrice": "0.3374",
# "execQty": "25",
# "execType": "Trade",
# "execValue": "8.435",
# "isMaker": False,
# "feeRate": "0.0006",
# "tradeIv": "",
# "markIv": "",
# "blockTradeId": "",
# "markPrice": "0.3391",
# "indexPrice": "",
# "underlyingPrice": "",
# "leavesQty": "0",
# "orderId": "f6e324ff-99c2-4e89-9739-3086e47f9381",
# "orderLinkId": "",
# "orderPrice": "0.3207",
# "orderQty": "25",
# "orderType": "Market",
# "stopOrderType": "UNKNOWN",
# "side": "Sell",
# "execTime": "1672364174443",
# "isLeverage": "0"
# }
# ]
# }
#
# execution.fast
#
# {
# "topic": "execution.fast",
# "creationTime": 1757405601981,
# "data": [
# {
# "category": "linear",
# "symbol": "BTCUSDT",
# "execId": "ffcac6ac-7571-536d-a28a-847dd7d08a0f",
# "execPrice": "112529.6",
# "execQty": "0.001",
# "orderId": "6e25ab73-7a55-4ae7-adc2-8ea95f167c85",
# "isMaker": False,
# "orderLinkId": "test-00001",
# "side": "Buy",
# "execTime": "1757405601977",
# "seq": 9515624038
# }
# ]
# }
#
topic = self.safe_string(message, 'topic')
spot = topic == 'ticketInfo'
executionFast = topic == 'execution.fast'
data = self.safe_value(message, 'data', [])
if not isinstance(data, list):
data = self.safe_value(data, 'result', [])
if self.myTrades is None:
limit = self.safe_integer(self.options, 'tradesLimit', 1000)
self.myTrades = ArrayCacheBySymbolById(limit)
trades = self.myTrades
symbols: dict = {}
filterExecTypes = self.handle_option('watchMyTrades', 'filterExecTypes', [])
for i in range(0, len(data)):
rawTrade = data[i]
parsed = None
if spot and not executionFast:
parsed = self.parse_ws_trade(rawTrade)
else:
# filter unified trades
execType = self.safe_string(rawTrade, 'execType', '')
if executionFast:
execType = 'Trade'
if not self.in_array(execType, filterExecTypes):
continue
parsed = self.parse_trade(rawTrade)
symbol = parsed['symbol']
symbols[symbol] = True
trades.append(parsed)
keys = list(symbols.keys())
for i in range(0, len(keys)):
currentMessageHash = 'myTrades:' + keys[i]
client.resolve(trades, currentMessageHash)
# non-symbol specific
messageHash = 'myTrades'
client.resolve(trades, messageHash)
async def watch_positions(self, symbols: Strings = None, since: Int = None, limit: Int = None, params={}) -> List[Position]:
"""
https://bybit-exchange.github.io/docs/v5/websocket/private/position
watch all open positions
:param str[] [symbols]: list of unified market symbols
:param int [since]: the earliest time in ms to fetch positions for
:param int [limit]: the maximum number of positions to retrieve
:param dict params: extra parameters specific to the exchange API endpoint
:returns dict[]: a list of `position structure <https://docs.ccxt.com/en/latest/manual.html#position-structure>`
"""
await self.load_markets()
method = 'watchPositions'
messageHash = ''
if not self.is_empty(symbols):
symbols = self.market_symbols(symbols)
messageHash = '::' + ','.join(symbols)
firstSymbol = self.safe_string(symbols, 0)
url = await self.get_url_by_market_type(firstSymbol, True, method, params)
messageHash = 'positions' + messageHash
client = self.client(url)
await self.authenticate(url)
self.set_positions_cache(client, symbols)
cache = self.positions
fetchPositionsSnapshot = self.handle_option('watchPositions', 'fetchPositionsSnapshot', True)
awaitPositionsSnapshot = self.handle_option('watchPositions', 'awaitPositionsSnapshot', True)
if fetchPositionsSnapshot and awaitPositionsSnapshot and cache is None:
snapshot = await client.future('fetchPositionsSnapshot')
return self.filter_by_symbols_since_limit(snapshot, symbols, since, limit, True)
topics = ['position']
newPositions = await self.watch_topics(url, [messageHash], topics, params)
if self.newUpdates:
return newPositions
return self.filter_by_symbols_since_limit(cache, symbols, since, limit, True)
def set_positions_cache(self, client: Client, symbols: Strings = None):
if self.positions is not None:
return
fetchPositionsSnapshot = self.handle_option('watchPositions', 'fetchPositionsSnapshot', True)
if fetchPositionsSnapshot:
messageHash = 'fetchPositionsSnapshot'
if not (messageHash in client.futures):
client.future(messageHash)
self.spawn(self.load_positions_snapshot, client, messageHash)
else:
self.positions = ArrayCacheBySymbolBySide()
async def load_positions_snapshot(self, client, messageHash):
# one ws channel gives positions for all types, for snapshot must load all positions
fetchFunctions = [
self.fetch_positions(None, {'type': 'swap', 'subType': 'linear'}),
self.fetch_positions(None, {'type': 'swap', 'subType': 'inverse'}),
]
promises = await asyncio.gather(*fetchFunctions)
self.positions = ArrayCacheBySymbolBySide()
cache = self.positions
for i in range(0, len(promises)):
positions = promises[i]
for ii in range(0, len(positions)):
position = positions[ii]
cache.append(position)
# don't remove the future from the .futures cache
future = client.futures[messageHash]
future.resolve(cache)
client.resolve(cache, 'position')
def handle_positions(self, client, message):
#
# {
# topic: 'position',
# id: '504b2671629b08e3c4f6960382a59363:3bc4028023786545:0:01',
# creationTime: 1694566055295,
# data: [{
# bustPrice: '15.00',
# category: 'inverse',
# createdTime: '1670083436351',
# cumRealisedPnl: '0.00011988',
# entryPrice: '19358.58553268',
# leverage: '10',
# liqPrice: '15.00',
# markPrice: '25924.00',
# positionBalance: '0.0000156',
# positionIdx: 0,
# positionMM: '0.001',
# positionIM: '0.0000015497',
# positionStatus: 'Normal',
# positionValue: '0.00015497',
# riskId: 1,
# riskLimitValue: '150',
# side: 'Buy',
# size: '3',
# stopLoss: '0.00',
# symbol: 'BTCUSD',
# takeProfit: '0.00',
# tpslMode: 'Full',
# tradeMode: 0,
# autoAddMargin: 1,
# trailingStop: '0.00',
# unrealisedPnl: '0.00003925',
# updatedTime: '1694566055293',
# adlRankIndicator: 3
# }]
# }
#
# each account is connected to a different endpoint
# and has exactly one subscriptionhash which is the account type
if self.positions is None:
self.positions = ArrayCacheBySymbolBySide()
cache = self.positions
newPositions = []
rawPositions = self.safe_value(message, 'data', [])
for i in range(0, len(rawPositions)):
rawPosition = rawPositions[i]
position = self.parse_position(rawPosition)
side = self.safe_string(position, 'side')
# hacky solution to handle closing positions
# without crashing, we should handle self properly later
newPositions.append(position)
if side is None or side == '':
# closing update, adding both sides to "reset" both sides
# since we don't know which side is being closed
position['side'] = 'long'
cache.append(position)
position['side'] = 'short'
cache.append(position)
position['side'] = None
else:
# regular update
cache.append(position)
messageHashes = self.find_message_hashes(client, 'positions::')
for i in range(0, len(messageHashes)):
messageHash = messageHashes[i]
parts = messageHash.split('::')
symbolsString = parts[1]
symbols = symbolsString.split(',')
positions = self.filter_by_array(newPositions, 'symbol', symbols, False)
if not self.is_empty(positions):
client.resolve(positions, messageHash)
client.resolve(newPositions, 'positions')
async def un_watch_positions(self, symbols: Strings = None, params={}) -> Any:
"""
unWatches all open positions
https://bybit-exchange.github.io/docs/v5/websocket/private/position
:param str[] [symbols]: list of unified market symbols
:param dict [params]: extra parameters specific to the exchange API endpoint
:returns dict: status of the unwatch request
"""
await self.load_markets()
method = 'watchPositions'
messageHash = 'unsubscribe:positions'
subHash = 'positions'
if not self.is_empty(symbols):
raise NotSupported(self.id + ' unWatchPositions() does not support a symbol parameter, you must unwatch all orders')
url = await self.get_url_by_market_type(None, True, method, params)
await self.authenticate(url)
topics = ['position']
return await self.un_watch_topics(url, 'positions', symbols, [messageHash], [subHash], topics, params)
async def watch_liquidations(self, symbol: str, since: Int = None, limit: Int = None, params={}) -> List[Liquidation]:
"""
watch the public liquidations of a trading pair
https://bybit-exchange.github.io/docs/v5/websocket/public/liquidation
:param str symbol: unified CCXT market symbol
:param int [since]: the earliest time in ms to fetch liquidations for
:param int [limit]: the maximum number of liquidation structures to retrieve
:param dict [params]: exchange specific parameters for the bitmex api endpoint
:param str [params.method]: exchange specific method, supported: liquidation, allLiquidation
:returns dict: an array of `liquidation structures <https://github.com/ccxt/ccxt/wiki/Manual#liquidation-structure>`
"""
await self.load_markets()
market = self.market(symbol)
symbol = market['symbol']
url = await self.get_url_by_market_type(symbol, False, 'watchLiquidations', params)
params = self.clean_params(params)
method = None
method, params = self.handle_option_and_params(params, 'watchLiquidations', 'method', 'liquidation')
messageHash = 'liquidations::' + symbol
topic = method + '.' + market['id']
newLiquidation = await self.watch_topics(url, [messageHash], [topic], params)
if self.newUpdates:
return newLiquidation
return self.filter_by_symbols_since_limit(self.liquidations, [symbol], since, limit, True)
def handle_liquidation(self, client: Client, message):
#
# {
# "data": {
# "price": "0.03803",
# "side": "Buy",
# "size": "1637",
# "symbol": "GALAUSDT",
# "updatedTime": 1673251091822
# },
# "topic": "liquidation.GALAUSDT",
# "ts": 1673251091822,
# "type": "snapshot"
# }
#
# {
# "topic": "allLiquidation.ROSEUSDT",
# "type": "snapshot",
# "ts": 1739502303204,
# "data": [
# {
# "T": 1739502302929,
# "s": "ROSEUSDT",
# "S": "Sell",
# "v": "20000",
# "p": "0.04499"
# }
# ]
# }
#
if isinstance(message['data'], list):
rawLiquidations = self.safe_list(message, 'data', [])
for i in range(0, len(rawLiquidations)):
rawLiquidation = rawLiquidations[i]
marketId = self.safe_string(rawLiquidation, 's')
market = self.safe_market(marketId, None, '', 'contract')
symbol = market['symbol']
liquidation = self.parse_ws_liquidation(rawLiquidation, market)
liquidations = self.safe_value(self.liquidations, symbol)
if liquidations is None:
limit = self.safe_integer(self.options, 'liquidationsLimit', 1000)
liquidations = ArrayCache(limit)
liquidations.append(liquidation)
self.liquidations[symbol] = liquidations
client.resolve([liquidation], 'liquidations')
client.resolve([liquidation], 'liquidations::' + symbol)
else:
rawLiquidation = self.safe_dict(message, 'data', {})
marketId = self.safe_string(rawLiquidation, 'symbol')
market = self.safe_market(marketId, None, '', 'contract')
symbol = market['symbol']
liquidation = self.parse_ws_liquidation(rawLiquidation, market)
liquidations = self.safe_value(self.liquidations, symbol)
if liquidations is None:
limit = self.safe_integer(self.options, 'liquidationsLimit', 1000)
liquidations = ArrayCache(limit)
liquidations.append(liquidation)
self.liquidations[symbol] = liquidations
client.resolve([liquidation], 'liquidations')
client.resolve([liquidation], 'liquidations::' + symbol)
def parse_ws_liquidation(self, liquidation, market=None):
#
# {
# "price": "0.03803",
# "side": "Buy",
# "size": "1637",
# "symbol": "GALAUSDT",
# "updatedTime": 1673251091822
# }
#
# {
# "T": 1739502302929,
# "s": "ROSEUSDT",
# "S": "Sell",
# "v": "20000",
# "p": "0.04499"
# }
#
marketId = self.safe_string_2(liquidation, 'symbol', 's')
market = self.safe_market(marketId, market, '', 'contract')
timestamp = self.safe_integer_2(liquidation, 'updatedTime', 'T')
return self.safe_liquidation({
'info': liquidation,
'symbol': market['symbol'],
'contracts': self.safe_number_2(liquidation, 'size', 'v'),
'contractSize': self.safe_number(market, 'contractSize'),
'price': self.safe_number_2(liquidation, 'price', 'p'),
'side': self.safe_string_lower(liquidation, 'side', 'S'),
'baseValue': None,
'quoteValue': None,
'timestamp': timestamp,
'datetime': self.iso8601(timestamp),
})
async def watch_orders(self, symbol: Str = None, since: Int = None, limit: Int = None, params={}) -> List[Order]:
"""
watches information on multiple orders made by the user
https://bybit-exchange.github.io/docs/v5/websocket/private/order
:param str symbol: unified market symbol of the market orders were made in
:param int [since]: the earliest time in ms to fetch orders for
:param int [limit]: the maximum number of order structures to retrieve
:param dict [params]: extra parameters specific to the exchange API endpoint
:returns dict[]: a list of `order structures <https://docs.ccxt.com/#/?id=order-structure>`
"""
await self.load_markets()
method = 'watchOrders'
messageHash = 'orders'
if symbol is not None:
symbol = self.symbol(symbol)
messageHash += ':' + symbol
url = await self.get_url_by_market_type(symbol, True, method, params)
await self.authenticate(url)
topicsByMarket: dict = {
'spot': ['order', 'stopOrder'],
'unified': ['order'],
'usdc': ['user.openapi.perp.order'],
}
topics = self.safe_value(topicsByMarket, self.get_private_type(url))
orders = await self.watch_topics(url, [messageHash], topics, params)
if self.newUpdates:
limit = orders.getLimit(symbol, limit)
return self.filter_by_symbol_since_limit(orders, symbol, since, limit, True)
async def un_watch_orders(self, symbol: Str = None, params={}) -> Any:
"""
unWatches information on multiple orders made by the user
https://bybit-exchange.github.io/docs/v5/websocket/private/order
:param str symbol: unified market symbol of the market orders were made in
:param dict [params]: extra parameters specific to the exchange API endpoint
:param boolean [params.unifiedMargin]: use unified margin account
:returns dict[]: a list of `order structures <https://docs.ccxt.com/#/?id=order-structure>`
"""
await self.load_markets()
method = 'watchOrders'
messageHash = 'unsubscribe:orders'
subHash = 'orders'
if symbol is not None:
raise NotSupported(self.id + ' unWatchOrders() does not support a symbol parameter, you must unwatch all orders')
url = await self.get_url_by_market_type(symbol, True, method, params)
await self.authenticate(url)
topicsByMarket: dict = {
'spot': ['order', 'stopOrder'],
'unified': ['order'],
'usdc': ['user.openapi.perp.order'],
}
topics = self.safe_value(topicsByMarket, self.get_private_type(url))
return await self.un_watch_topics(url, 'orders', [], [messageHash], [subHash], topics, params)
def handle_order_ws(self, client: Client, message):
#
# {
# "reqId":"1",
# "retCode":0,
# "retMsg":"OK",
# "op":"order.create",
# "data":{
# "orderId":"1673523595617593600",
# "orderLinkId":"1673523595617593601"
# },
# "header":{
# "X-Bapi-Limit":"20",
# "X-Bapi-Limit-Status":"19",
# "X-Bapi-Limit-Reset-Timestamp":"1714235558880",
# "Traceid":"584a06d373f2fdcb3a4dfdd81d27df11",
# "Timenow":"1714235558881"
# },
# "connId":"cojidqec0hv9fgvhtbt0-40e"
# }
#
messageHash = self.safe_string(message, 'reqId')
data = self.safe_dict(message, 'data')
order = self.parse_order(data)
client.resolve(order, messageHash)
def handle_order(self, client: Client, message):
#
# spot
# {
# "type": "snapshot",
# "topic": "order",
# "ts": "1662348310441",
# "data": [
# {
# "e": "order",
# "E": "1662348310441",
# "s": "BTCUSDT",
# "c": "spotx008",
# "S": "BUY",
# "o": "MARKET_OF_QUOTE",
# "f": "GTC",
# "q": "20",
# "p": "0",
# "X": "CANCELED",
# "i": "1238261807653647872",
# "M": "1238225004531834368",
# "l": "0.001007",
# "z": "0.001007",
# "L": "19842.02",
# "n": "0",
# "N": "BTC",
# "u": True,
# "w": True,
# "m": False,
# "O": "1662348310368",
# "Z": "19.98091414",
# "A": "0",
# "C": False,
# "v": "0",
# "d": "NO_LIQ",
# "t": "2100000000002220938"
# }
# ]
# }
# unified
# {
# "id": "5923240c6880ab-c59f-420b-9adb-3639adc9dd90",
# "topic": "order",
# "creationTime": 1672364262474,
# "data": [
# {
# "symbol": "ETH-30DEC22-1400-C",
# "orderId": "5cf98598-39a7-459e-97bf-76ca765ee020",
# "side": "Sell",
# "orderType": "Market",
# "cancelType": "UNKNOWN",
# "price": "72.5",
# "qty": "1",
# "orderIv": "",
# "timeInForce": "IOC",
# "orderStatus": "Filled",
# "orderLinkId": "",
# "lastPriceOnCreated": "",
# "reduceOnly": False,
# "leavesQty": "",
# "leavesValue": "",
# "cumExecQty": "1",
# "cumExecValue": "75",
# "avgPrice": "75",
# "blockTradeId": "",
# "positionIdx": 0,
# "cumExecFee": "0.358635",
# "createdTime": "1672364262444",
# "updatedTime": "1672364262457",
# "rejectReason": "EC_NoError",
# "stopOrderType": "",
# "triggerPrice": "",
# "takeProfit": "",
# "stopLoss": "",
# "tpTriggerBy": "",
# "slTriggerBy": "",
# "triggerDirection": 0,
# "triggerBy": "",
# "closeOnTrigger": False,
# "category": "option"
# }
# ]
# }
#
if self.orders is None:
limit = self.safe_integer(self.options, 'ordersLimit', 1000)
self.orders = ArrayCacheBySymbolById(limit)
orders = self.orders
rawOrders = self.safe_value(message, 'data', [])
first = self.safe_value(rawOrders, 0, {})
category = self.safe_string(first, 'category')
isSpot = category == 'spot'
if not isSpot:
rawOrders = self.safe_value(rawOrders, 'result', rawOrders)
symbols: dict = {}
for i in range(0, len(rawOrders)):
parsed = self.parse_order(rawOrders[i])
# if isSpot:
# parsed = self.parseWsSpotOrder(rawOrders[i])
# else:
# parsed = self.parse_order(rawOrders[i])
# }
symbol = parsed['symbol']
symbols[symbol] = True
orders.append(parsed)
symbolsArray = list(symbols.keys())
for i in range(0, len(symbolsArray)):
currentMessageHash = 'orders:' + symbolsArray[i]
client.resolve(orders, currentMessageHash)
messageHash = 'orders'
client.resolve(orders, messageHash)
async def watch_balance(self, params={}) -> Balances:
"""
watch balance and get the amount of funds available for trading or funds locked in orders
https://bybit-exchange.github.io/docs/v5/websocket/private/wallet
:param dict [params]: extra parameters specific to the exchange API endpoint
:returns dict: a `balance structure <https://docs.ccxt.com/#/?id=balance-structure>`
"""
await self.load_markets()
method = 'watchBalance'
messageHash = 'balances'
type = None
type, params = self.handle_market_type_and_params('watchBalance', None, params)
subType = None
subType, params = self.handle_sub_type_and_params('watchBalance', None, params)
unified = await self.isUnifiedEnabled()
isUnifiedMargin = self.safe_bool(unified, 0, False)
isUnifiedAccount = self.safe_bool(unified, 1, False)
url = await self.get_url_by_market_type(None, True, method, params)
await self.authenticate(url)
topicByMarket: dict = {
'spot': 'outboundAccountInfo',
'unified': 'wallet',
}
if isUnifiedAccount:
# unified account
if subType == 'inverse':
messageHash += ':contract'
else:
messageHash += ':unified'
if not isUnifiedMargin and not isUnifiedAccount:
# normal account using v5
if type == 'spot':
messageHash += ':spot'
else:
messageHash += ':contract'
if isUnifiedMargin:
# unified margin account using v5
if type == 'spot':
messageHash += ':spot'
else:
if subType == 'linear':
messageHash += ':unified'
else:
messageHash += ':contract'
topics = [self.safe_value(topicByMarket, self.get_private_type(url))]
return await self.watch_topics(url, [messageHash], topics, params)
def handle_balance(self, client: Client, message):
#
# spot
# {
# "type": "snapshot",
# "topic": "outboundAccountInfo",
# "ts": "1662107217641",
# "data": [
# {
# "e": "outboundAccountInfo",
# "E": "1662107217640",
# "T": True,
# "W": True,
# "D": True,
# "B": [
# {
# "a": "USDT",
# "f": "176.81254174",
# "l": "201.575"
# }
# ]
# }
# ]
# }
# unified
# {
# "id": "5923242c464be9-25ca-483d-a743-c60101fc656f",
# "topic": "wallet",
# "creationTime": 1672364262482,
# "data": [
# {
# "accountIMRate": "0.016",
# "accountMMRate": "0.003",
# "totalEquity": "12837.78330098",
# "totalWalletBalance": "12840.4045924",
# "totalMarginBalance": "12837.78330188",
# "totalAvailableBalance": "12632.05767702",
# "totalPerpUPL": "-2.62129051",
# "totalInitialMargin": "205.72562486",
# "totalMaintenanceMargin": "39.42876721",
# "coin": [
# {
# "coin": "USDC",
# "equity": "200.62572554",
# "usdValue": "200.62572554",
# "walletBalance": "201.34882644",
# "availableToWithdraw": "0",
# "availableToBorrow": "1500000",
# "borrowAmount": "0",
# "accruedInterest": "0",
# "totalOrderIM": "0",
# "totalPositionIM": "202.99874213",
# "totalPositionMM": "39.14289747",
# "unrealisedPnl": "74.2768991",
# "cumRealisedPnl": "-209.1544627",
# "bonus": "0"
# },
# {
# "coin": "BTC",
# "equity": "0.06488393",
# "usdValue": "1023.08402268",
# "walletBalance": "0.06488393",
# "availableToWithdraw": "0.06488393",
# "availableToBorrow": "2.5",
# "borrowAmount": "0",
# "accruedInterest": "0",
# "totalOrderIM": "0",
# "totalPositionIM": "0",
# "totalPositionMM": "0",
# "unrealisedPnl": "0",
# "cumRealisedPnl": "0",
# "bonus": "0"
# },
# {
# "coin": "ETH",
# "equity": "0",
# "usdValue": "0",
# "walletBalance": "0",
# "availableToWithdraw": "0",
# "availableToBorrow": "26",
# "borrowAmount": "0",
# "accruedInterest": "0",
# "totalOrderIM": "0",
# "totalPositionIM": "0",
# "totalPositionMM": "0",
# "unrealisedPnl": "0",
# "cumRealisedPnl": "0",
# "bonus": "0"
# },
# {
# "coin": "USDT",
# "equity": "11726.64664904",
# "usdValue": "11613.58597018",
# "walletBalance": "11728.54414904",
# "availableToWithdraw": "11723.92075829",
# "availableToBorrow": "2500000",
# "borrowAmount": "0",
# "accruedInterest": "0",
# "totalOrderIM": "0",
# "totalPositionIM": "2.72589075",
# "totalPositionMM": "0.28576575",
# "unrealisedPnl": "-1.8975",
# "cumRealisedPnl": "0.64782276",
# "bonus": "0"
# },
# {
# "coin": "EOS3L",
# "equity": "215.0570412",
# "usdValue": "0",
# "walletBalance": "215.0570412",
# "availableToWithdraw": "215.0570412",
# "availableToBorrow": "0",
# "borrowAmount": "0",
# "accruedInterest": "",
# "totalOrderIM": "0",
# "totalPositionIM": "0",
# "totalPositionMM": "0",
# "unrealisedPnl": "0",
# "cumRealisedPnl": "0",
# "bonus": "0"
# },
# {
# "coin": "BIT",
# "equity": "1.82",
# "usdValue": "0.48758257",
# "walletBalance": "1.82",
# "availableToWithdraw": "1.82",
# "availableToBorrow": "0",
# "borrowAmount": "0",
# "accruedInterest": "",
# "totalOrderIM": "0",
# "totalPositionIM": "0",
# "totalPositionMM": "0",
# "unrealisedPnl": "0",
# "cumRealisedPnl": "0",
# "bonus": "0"
# }
# ],
# "accountType": "UNIFIED"
# }
# ]
# }
#
if self.balance is None:
self.balance = {}
messageHash = 'balance'
topic = self.safe_value(message, 'topic')
info = None
rawBalances = []
account = None
if topic == 'outboundAccountInfo':
account = 'spot'
data = self.safe_value(message, 'data', [])
for i in range(0, len(data)):
B = self.safe_value(data[i], 'B', [])
rawBalances = self.array_concat(rawBalances, B)
info = rawBalances
if topic == 'wallet':
data = self.safe_value(message, 'data', {})
for i in range(0, len(data)):
result = self.safe_value(data, 0, {})
account = self.safe_string_lower(result, 'accountType')
rawBalances = self.array_concat(rawBalances, self.safe_value(result, 'coin', []))
info = data
for i in range(0, len(rawBalances)):
self.parse_ws_balance(rawBalances[i], account)
if account is not None:
if self.safe_value(self.balance, account) is None:
self.balance[account] = {}
self.balance[account]['info'] = info
timestamp = self.safe_integer(message, 'ts')
self.balance[account]['timestamp'] = timestamp
self.balance[account]['datetime'] = self.iso8601(timestamp)
self.balance[account] = self.safe_balance(self.balance[account])
messageHash = 'balances:' + account
client.resolve(self.balance[account], messageHash)
else:
self.balance['info'] = info
timestamp = self.safe_integer(message, 'ts')
self.balance['timestamp'] = timestamp
self.balance['datetime'] = self.iso8601(timestamp)
self.balance = self.safe_balance(self.balance)
messageHash = 'balances'
client.resolve(self.balance, messageHash)
def parse_ws_balance(self, balance, accountType=None):
#
# spot
# {
# "a": "USDT",
# "f": "176.81254174",
# "l": "201.575"
# }
# unified
# {
# "coin": "BTC",
# "equity": "0.06488393",
# "usdValue": "1023.08402268",
# "walletBalance": "0.06488393",
# "availableToWithdraw": "0.06488393",
# "availableToBorrow": "2.5",
# "borrowAmount": "0",
# "accruedInterest": "0",
# "totalOrderIM": "0",
# "totalPositionIM": "0",
# "totalPositionMM": "0",
# "unrealisedPnl": "0",
# "cumRealisedPnl": "0",
# "bonus": "0"
# }
#
account = self.account()
currencyId = self.safe_string_2(balance, 'a', 'coin')
code = self.safe_currency_code(currencyId)
account['free'] = self.safe_string_n(balance, ['availableToWithdraw', 'f', 'free', 'availableToWithdraw'])
account['used'] = self.safe_string_2(balance, 'l', 'locked')
account['total'] = self.safe_string(balance, 'walletBalance')
if accountType is not None:
if self.safe_value(self.balance, accountType) is None:
self.balance[accountType] = {}
self.balance[accountType][code] = account
else:
self.balance[code] = account
async def watch_topics(self, url, messageHashes, topics, params={}):
request: dict = {
'op': 'subscribe',
'req_id': self.request_id(),
'args': topics,
}
message = self.extend(request, params)
return await self.watch_multiple(url, messageHashes, message, messageHashes)
async def un_watch_topics(self, url: str, topic: str, symbols: Strings, messageHashes: List[str], subMessageHashes: List[str], topics, params={}, subExtension={}):
reqId = self.request_id()
request: dict = {
'op': 'unsubscribe',
'req_id': reqId,
'args': topics,
}
subscription = {
'id': reqId,
'topic': topic,
'messageHashes': messageHashes,
'subMessageHashes': subMessageHashes,
'symbols': symbols,
}
message = self.extend(request, params)
return await self.watch_multiple(url, messageHashes, message, messageHashes, self.extend(subscription, subExtension))
async def authenticate(self, url, params={}):
self.check_required_credentials()
messageHash = 'authenticated'
client = self.client(url)
future = client.reusableFuture(messageHash)
authenticated = self.safe_value(client.subscriptions, messageHash)
if authenticated is None:
expiresInt = self.milliseconds() + 10000
expires = self.number_to_string(expiresInt)
path = 'GET/realtime'
auth = path + expires
signature = self.hmac(self.encode(auth), self.encode(self.secret), hashlib.sha256, 'hex')
request: dict = {
'op': 'auth',
'args': [
self.apiKey, expires, signature,
],
}
message = self.extend(request, params)
self.watch(url, messageHash, message, messageHash)
return await future
def handle_error_message(self, client: Client, message) -> Bool:
#
# {
# "success": False,
# "ret_msg": "error:invalid op",
# "conn_id": "5e079fdd-9c7f-404d-9dbf-969d650838b5",
# "request": {op: '', args: null}
# }
#
# auth error
#
# {
# "success": False,
# "ret_msg": "error:USVC1111",
# "conn_id": "e73770fb-a0dc-45bd-8028-140e20958090",
# "request": {
# "op": "auth",
# "args": [
# "9rFT6uR4uz9Imkw4Wx",
# "1653405853543",
# "542e71bd85597b4db0290f0ce2d13ed1fd4bb5df3188716c1e9cc69a879f7889"
# ]
# }
#
# {code: '-10009', desc: "Invalid period!"}
#
# {
# "reqId":"1",
# "retCode":170131,
# "retMsg":"Insufficient balance.",
# "op":"order.create",
# "data":{
#
# },
# "header":{
# "X-Bapi-Limit":"20",
# "X-Bapi-Limit-Status":"19",
# "X-Bapi-Limit-Reset-Timestamp":"1714236608944",
# "Traceid":"3d7168a137bf32a947b7e5e6a575ac7f",
# "Timenow":"1714236608946"
# },
# "connId":"cojifin88smerbj9t560-406"
# }
#
code = self.safe_string_n(message, ['code', 'ret_code', 'retCode'])
try:
if code is not None and code != '0':
feedback = self.id + ' ' + self.json(message)
self.throw_exactly_matched_exception(self.exceptions['exact'], code, feedback)
msg = self.safe_string_2(message, 'retMsg', 'ret_msg')
self.throw_broadly_matched_exception(self.exceptions['broad'], msg, feedback)
raise ExchangeError(feedback)
success = self.safe_value(message, 'success')
if success is not None and not success:
ret_msg = self.safe_string(message, 'ret_msg')
request = self.safe_value(message, 'request', {})
op = self.safe_string(request, 'op')
if op == 'auth':
raise AuthenticationError('Authentication failed: ' + ret_msg)
else:
raise ExchangeError(self.id + ' ' + ret_msg)
return False
except Exception as error:
if isinstance(error, AuthenticationError):
messageHash = 'authenticated'
client.reject(error, messageHash)
if messageHash in client.subscriptions:
del client.subscriptions[messageHash]
else:
messageHash = self.safe_string(message, 'reqId')
client.reject(error, messageHash)
return True
def handle_message(self, client: Client, message):
topic = self.safe_string_2(message, 'topic', 'op', '')
if self.handle_error_message(client, message):
return
# contract pong
ret_msg = self.safe_string(message, 'ret_msg')
if (ret_msg == 'pong') or (topic == 'pong'):
self.handle_pong(client, message)
return
# spot pong
pong = self.safe_integer(message, 'pong')
if pong is not None:
self.handle_pong(client, message)
return
# pong
event = self.safe_string(message, 'event')
if event == 'sub' or (topic == 'subscribe'):
self.handle_subscription_status(client, message)
return
methods: dict = {
'orderbook': self.handle_order_book,
'kline': self.handle_ohlcv,
'order': self.handle_order,
'stopOrder': self.handle_order,
'ticker': self.handle_ticker,
'trade': self.handle_trades,
'publicTrade': self.handle_trades,
'depth': self.handle_order_book,
'wallet': self.handle_balance,
'outboundAccountInfo': self.handle_balance,
'execution': self.handle_my_trades,
'execution.fast': self.handle_my_trades,
'ticketInfo': self.handle_my_trades,
'user.openapi.perp.trade': self.handle_my_trades,
'position': self.handle_positions,
'liquidation': self.handle_liquidation,
'allLiquidation': self.handle_liquidation,
'pong': self.handle_pong,
'order.create': self.handle_order_ws,
'order.amend': self.handle_order_ws,
'order.cancel': self.handle_order_ws,
'auth': self.handle_authenticate,
'unsubscribe': self.handle_un_subscribe,
}
exacMethod = self.safe_value(methods, topic)
if exacMethod is not None:
exacMethod(client, message)
return
keys = list(methods.keys())
for i in range(0, len(keys)):
key = keys[i]
if topic.find(keys[i]) >= 0:
method = methods[key]
method(client, message)
return
# unified auth acknowledgement
type = self.safe_string(message, 'type')
if type == 'AUTH_RESP':
self.handle_authenticate(client, message)
def ping(self, client: Client):
return {
'req_id': self.request_id(),
'op': 'ping',
}
def handle_pong(self, client: Client, message):
#
# {
# "success": True,
# "ret_msg": "pong",
# "conn_id": "db3158a0-8960-44b9-a9de-ac350ee13158",
# "request": {op: "ping", args: null}
# }
#
# {pong: 1653296711335}
#
#
# {
# "req_id": "2",
# "op": "pong",
# "args": ["1757405570352"],
# "conn_id": "d266o6hqo29sqmnq4vk0-1yus1"
# }
#
client.lastPong = self.safe_integer(message, 'pong')
return message
def handle_authenticate(self, client: Client, message):
#
# {
# "success": True,
# "ret_msg": '',
# "op": "auth",
# "conn_id": "ce3dpomvha7dha97tvp0-2xh"
# }
#
# {
# "retCode":0,
# "retMsg":"OK",
# "op":"auth",
# "connId":"cojifin88smerbj9t560-404"
# }
#
# {
# "success": True,
# "ret_msg": "",
# "op": "auth",
# "conn_id": "d266o6hqo29sqmnq4vk0-1yus1"
# }
#
success = self.safe_value(message, 'success')
code = self.safe_integer(message, 'retCode')
messageHash = 'authenticated'
if success or code == 0:
future = self.safe_value(client.futures, messageHash)
future.resolve(True)
else:
error = AuthenticationError(self.id + ' ' + self.json(message))
client.reject(error, messageHash)
if messageHash in client.subscriptions:
del client.subscriptions[messageHash]
return message
def handle_subscription_status(self, client: Client, message):
#
# {
# "topic": "kline",
# "event": "sub",
# "params": {
# "symbol": "LTCUSDT",
# "binary": "false",
# "klineType": "1m",
# "symbolName": "LTCUSDT"
# },
# "code": "0",
# "msg": "Success"
# }
#
return message
def handle_un_subscribe(self, client: Client, message):
#
# {"success":true,"ret_msg":"","conn_id":"7188110e-6908-41e9-b863-6365127e92ad","req_id":"3","op":"unsubscribe"}
#
# client.subscription will be something like:
# {
# "publicTrade.LTCUSDT":true,
# "publicTrade.ADAUSDT":true,
# "unsubscribe:trade:LTC/USDT:USDT": {
# "id":4,
# "subHash": "trade:LTC/USDT"
# },
# }
reqId = self.safe_string(message, 'req_id')
keys = list(client.subscriptions.keys())
for i in range(0, len(keys)):
messageHash = keys[i]
if not (messageHash in client.subscriptions):
continue
# the previous iteration can have deleted the messageHash from the subscriptions
if messageHash.startswith('unsubscribe'):
subscription = client.subscriptions[messageHash]
subId = self.safe_string(subscription, 'id')
if reqId != subId:
continue
messageHashes = self.safe_list(subscription, 'messageHashes', [])
subMessageHashes = self.safe_list(subscription, 'subMessageHashes', [])
for j in range(0, len(messageHashes)):
unsubHash = messageHashes[j]
subHash = subMessageHashes[j]
usePrefix = (subHash == 'orders') or (subHash == 'myTrades') or (subHash == 'positions')
self.clean_unsubscription(client, subHash, unsubHash, usePrefix)
self.clean_cache(subscription)
return message
Reference in New Issue View Git Blame Copy Permalink