# -*- 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 from ccxt.base.types import Any, Balances, Int, Order, OrderBook, Position, Str, Strings, Ticker, Trade from ccxt.async_support.base.ws.client import Client from typing import List from ccxt.base.errors import ExchangeError class bullish(ccxt.async_support.bullish): def describe(self) -> Any: return self.deep_extend(super(bullish, self).describe(), { 'has': { 'ws': True, 'watchTicker': True, 'watchTickers': False, 'watchOrderBook': True, 'watchOrders': True, 'watchTrades': True, 'watchPositions': True, 'watchMyTrades': True, 'watchBalance': True, 'watchOHLCV': False, }, 'urls': { 'api': { 'ws': { 'public': 'wss://api.exchange.bullish.com', 'private': 'wss://api.exchange.bullish.com/trading-api/v1/private-data', }, }, 'test': { 'ws': { 'public': 'wss://api.simnext.bullish-test.com', 'private': 'wss://api.simnext.bullish-test.com/trading-api/v1/private-data', }, }, }, 'options': { 'ws': { 'cookies': {}, }, }, 'streaming': { 'ping': self.ping, 'keepAlive': 99000, # disconnect after 100 seconds of inactivity }, }) def request_id(self): requestId = self.sum(self.safe_integer(self.options, 'requestId', 0), 1) self.options['requestId'] = requestId return requestId def ping(self, client: Client): # bullish does not support built-in ws protocol-level ping-pong # https://api.exchange.bullish.com/docs/api/rest/trading-api/v2/#overview--keep-websocket-open id = str(self.request_id()) return { 'jsonrpc': '2.0', 'type': 'command', 'method': 'keepalivePing', 'params': {}, 'id': id, } def handle_pong(self, client: Client, message): # # { # "id": "7", # "jsonrpc": "2.0", # "result": { # "responseCodeName": "OK", # "responseCode": "200", # "message": "Keep alive pong" # } # } # client.lastPong = self.milliseconds() return message # current line is for transpilation compatibility async def watch_public(self, url: str, messageHash: str, request={}, params={}) -> Any: id = str(self.request_id()) message = { 'jsonrpc': '2.0', 'type': 'command', 'method': 'subscribe', 'params': request, 'id': id, } fullUrl = self.urls['api']['ws']['public'] + url return await self.watch(fullUrl, messageHash, self.deep_extend(message, params), messageHash) async def watch_private(self, messageHash: str, subscribeHash: str, request={}, params={}) -> Any: url = self.urls['api']['ws']['private'] token = await self.handleToken() cookies = { 'JWT_COOKIE': token, } self.options['ws']['cookies'] = cookies id = str(self.request_id()) message = { 'jsonrpc': '2.0', 'type': 'command', 'method': 'subscribe', 'params': request, 'id': id, } result = await self.watch(url, messageHash, self.deep_extend(message, params), subscribeHash) return result async def watch_trades(self, symbol: str, since: Int = None, limit: Int = None, params={}) -> List[Trade]: """ get the list of most recent trades for a particular symbol https://api.exchange.bullish.com/docs/api/rest/trading-api/v2/#overview--unified-anonymous-trades-websocket-unauthenticated :param str symbol: 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 ` """ await self.load_markets() market = self.market(symbol) messageHash = 'trades::' + market['symbol'] url = '/trading-api/v1/market-data/trades' request: Any = { 'topic': 'anonymousTrades', 'symbol': market['id'], } trades = await self.watch_public(url, messageHash, request, params) if self.newUpdates: limit = trades.getLimit(symbol, limit) return self.filter_by_since_limit(trades, since, limit, 'timestamp', True) def handle_trades(self, client: Client, message): # # { # "type": "snapshot", # "dataType": "V1TAAnonymousTradeUpdate", # "data": { # "trades": [ # { # "tradeId": "100086000000609304", # "isTaker": True, # "price": "104889.2063", # "createdAtTimestamp": "1749124509118", # "quantity": "0.01000000", # "publishedAtTimestamp": "1749124531466", # "side": "BUY", # "createdAtDatetime": "2025-06-05T11:55:09.118Z", # "symbol": "BTCUSDC" # } # ], # "createdAtTimestamp": "1749124509118", # "publishedAtTimestamp": "1749124531466", # "symbol": "BTCUSDC" # } # } # data = self.safe_dict(message, 'data', {}) marketId = self.safe_string(data, 'symbol') symbol = self.safe_symbol(marketId) market = self.market(symbol) rawTrades = self.safe_list(data, 'trades', []) trades = self.parse_trades(rawTrades, market) if not (symbol in self.trades): limit = self.safe_integer(self.options, 'tradesLimit', 1000) tradesArrayCache = ArrayCache(limit) self.trades[symbol] = tradesArrayCache tradesArray = self.trades[symbol] for i in range(0, len(trades)): tradesArray.append(trades[i]) self.trades[symbol] = tradesArray messageHash = 'trades::' + market['symbol'] client.resolve(tradesArray, messageHash) 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://api.exchange.bullish.com/docs/api/rest/trading-api/v2/#overview--anonymous-market-data-price-tick-unauthenticated :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 ` """ await self.load_markets() market = self.market(symbol) symbol = market['symbol'] url = self.urls['api']['ws']['public'] + '/trading-api/v1/market-data/tick/' + market['id'] messageHash = 'ticker::' + symbol return await self.watch(url, messageHash, params, messageHash) # no need to send a subscribe message, the server sends a ticker update on connect def handle_ticker(self, client: Client, message): # # { # "type": "update", # "dataType": "V1TATickerResponse", # "data": { # "askVolume": "0.00100822", # "average": "104423.1806", # "baseVolume": "472.83799258", # "bestAsk": "104324.6000", # "bestBid": "104324.5000", # "bidVolume": "0.00020146", # "change": "-198.4864", # "close": "104323.9374", # "createdAtTimestamp": "1749132838951", # "publishedAtTimestamp": "1749132838955", # "high": "105966.6577", # "last": "104323.9374", # "lastTradeDatetime": "2025-06-05T14:13:56.111Z", # "lastTradeSize": "0.02396100", # "low": "104246.6662", # "open": "104522.4238", # "percentage": "-0.19", # "quoteVolume": "49662592.6712", # "symbol": "BTC-USDC-PERP", # "type": "ticker", # "vwap": "105030.6996", # "currentPrice": "104324.7747", # "ammData": [ # { # "feeTierId": "1", # "currentPrice": "104324.7747", # "baseReservesQuantity": "8.27911366", # "quoteReservesQuantity": "1067283.0234", # "bidSpreadFee": "0.00000000", # "askSpreadFee": "0.00000000" # } # ], # "createdAtDatetime": "2025-06-05T14:13:58.951Z", # "markPrice": "104289.6884", # "fundingRate": "-0.000192", # "openInterest": "92.24146651" # } # } # updateType = self.safe_string(message, 'type', '') data = self.safe_dict(message, 'data', {}) marketId = self.safe_string(data, 'symbol') market = self.safe_market(marketId) symbol = market['symbol'] parsed = None if (updateType == 'snapshot'): parsed = self.parse_ticker(data, market) elif updateType == 'update': ticker = self.safe_dict(self.tickers, symbol, {}) rawTicker = self.safe_dict(ticker, 'info', {}) merged = self.extend(rawTicker, data) parsed = self.parse_ticker(merged, market) self.tickers[symbol] = parsed messageHash = 'ticker::' + symbol client.resolve(self.tickers[symbol], messageHash) 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://api.exchange.bullish.com/docs/api/rest/trading-api/v2/#overview--multi-orderbook-websocket-unauthenticated :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 ` indexed by market symbols """ await self.load_markets() market = self.market(symbol) url = '/trading-api/v1/market-data/orderbook' messageHash = 'orderbook::' + market['symbol'] request: dict = { 'topic': 'l2Orderbook', # 'l2Orderbook' returns only snapshots while 'l1Orderbook' returns only updates 'symbol': market['id'], } orderbook = await self.watch_public(url, messageHash, request, params) return orderbook.limit() def handle_order_book(self, client: Client, message): # # { # "type": "snapshot", # "dataType": "V1TALevel2", # "data": { # "timestamp": "1749372632028", # "bids": [ # "105523.3000", # "0.00046045", # ], # "asks": [ # "105523.4000", # "0.00117112", # ], # "publishedAtTimestamp": "1749372632073", # "datetime": "2025-06-08T08:50:32.028Z", # "sequenceNumberRange": [1967862061, 1967862062], # "symbol": "BTCUSDC" # } # } # # current channel is 'l2Orderbook' which returns only snapshots data = self.safe_dict(message, 'data', {}) marketId = self.safe_string(data, 'symbol') symbol = self.safe_symbol(marketId) messageHash = 'orderbook::' + symbol timestamp = self.safe_integer(data, 'timestamp') if not (symbol in self.orderbooks): self.orderbooks[symbol] = self.order_book() orderbook = self.orderbooks[symbol] bids = self.separate_bids_or_asks(self.safe_list(data, 'bids', [])) asks = self.separate_bids_or_asks(self.safe_list(data, 'asks', [])) snapshot = { 'bids': bids, 'asks': asks, } parsed = self.parse_order_book(snapshot, symbol, timestamp) sequenceNumberRange = self.safe_list(data, 'sequenceNumberRange', []) if len(sequenceNumberRange) > 0: lastIndex = len(sequenceNumberRange) - 1 parsed['nonce'] = self.safe_integer(sequenceNumberRange, lastIndex) orderbook.reset(parsed) self.orderbooks[symbol] = orderbook client.resolve(orderbook, messageHash) def separate_bids_or_asks(self, entry): result = [] # 300 = '54885.0000000' # 301 = '0.06141566' # 302 ='53714.0000000' for i in range(0, len(entry)): if i % 2 != 0: continue price = self.safe_string(entry, i) amount = self.safe_string(entry, i + 1) result.append([price, amount]) return result 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://api.exchange.bullish.com/docs/api/rest/trading-api/v2/#overview--private-data-websocket-authenticated :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 str [params.tradingAccountId]: the trading account id to fetch entries for :returns dict[]: a list of `order structures ` """ await self.load_markets() subscribeHash = 'orders' messageHash = subscribeHash if symbol is not None: symbol = self.symbol(symbol) messageHash = messageHash + '::' + symbol request: dict = { 'topic': 'orders', } tradingAccountId = self.safe_string(params, 'tradingAccountId') if tradingAccountId is not None: request['tradingAccountId'] = tradingAccountId params = self.omit(params, 'tradingAccountId') orders = await self.watch_private(messageHash, subscribeHash, request, params) if self.newUpdates: limit = orders.getLimit(symbol, limit) return self.filter_by_symbol_since_limit(orders, symbol, since, limit, True) def handle_orders(self, client: Client, message): # snapshot # { # "type": "snapshot", # "tradingAccountId": "111309424211255", # "dataType": "V1TAOrder", # "data": [...] # could be an empty list or a list of orders # } # # update # { # "type": "update", # "tradingAccountId": "111309424211255", # "dataType": "V1TAOrder", # "data": { # "status": "OPEN", # "createdAtTimestamp": "1751893427971", # "quoteFee": "0.000000", # "stopPrice": null, # "quantityFilled": "0.00000000", # "handle": null, # "clientOrderId": null, # "quantity": "0.10000000", # "margin": False, # "side": "BUY", # "createdAtDatetime": "2025-07-07T13:03:47.971Z", # "isLiquidation": False, # "borrowedQuoteQuantity": null, # "borrowedBaseQuantity": null, # "timeInForce": "GTC", # "borrowedQuantity": null, # "baseFee": "0.000000", # "quoteAmount": "0.0000000", # "price": "0.0000000", # "statusReason": "Order accepted", # "type": "MKT", # "statusReasonCode": 6014, # "allowBorrow": False, # "orderId": "862317981870850049", # "publishedAtTimestamp": "1751893427975", # "symbol": "ETHUSDT", # "averageFillPrice": null # } # } # type = self.safe_string(message, 'type') rawOrders = [] if type == 'update': data = self.safe_dict(message, 'data', {}) rawOrders.append(data) # update is a single order else: rawOrders = self.safe_list(message, 'data', []) # snapshot is a list of orders if len(rawOrders) > 0: if self.orders is None: limit = self.safe_integer(self.options, 'ordersLimit', 1000) self.orders = ArrayCacheBySymbolById(limit) orders = self.orders symbols: dict = {} for i in range(0, len(rawOrders)): rawOrder = rawOrders[i] parsedOrder = self.parse_order(rawOrder) orders.append(parsedOrder) symbol = self.safe_string(parsedOrder, 'symbol') symbols[symbol] = True messageHash = 'orders' client.resolve(orders, messageHash) keys = list(symbols.keys()) for i in range(0, len(keys)): hashSymbol = keys[i] symbolMessageHash = messageHash + '::' + hashSymbol client.resolve(self.orders, symbolMessageHash) 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://api.exchange.bullish.com/docs/api/rest/trading-api/v2/#overview--private-data-websocket-authenticated :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 :param str [params.tradingAccountId]: the trading account id to fetch entries for :returns dict[]: a list of `trade structures ` """ await self.load_markets() subscribeHash = 'myTrades' messageHash = subscribeHash if symbol is not None: symbol = self.symbol(symbol) messageHash += '::' + symbol request: dict = { 'topic': 'trades', } tradingAccountId = self.safe_string(params, 'tradingAccountId') if tradingAccountId is not None: request['tradingAccountId'] = tradingAccountId params = self.omit(params, 'tradingAccountId') trades = await self.watch_private(messageHash, subscribeHash, request, params) if self.newUpdates: limit = trades.getLimit(symbol, limit) return self.filter_by_since_limit(trades, since, limit, 'timestamp', True) def handle_my_trades(self, client: Client, message): # # snapshot # { # "type": "snapshot", # "tradingAccountId": "111309424211255", # "dataType": "V1TATrade", # "data": [...] # could be an empty list or a list of trades # } # # update # { # "type": "update", # "tradingAccountId": "111309424211255", # "dataType": "V1TATrade", # "data": { # "clientOtcTradeId": null, # "tradeId": "100203000003940164", # "baseFee": "0.00000000", # "isTaker": True, # "quoteAmount": "253.6012195", # "price": "2536.0121950", # "createdAtTimestamp": "1751914859840", # "quoteFee": "0.0000000", # "tradeRebateAmount": null, # "tradeRebateAssetSymbol": null, # "handle": null, # "otcTradeId": null, # "otcMatchId": null, # "orderId": "862407873644725249", # "quantity": "0.10000000", # "publishedAtTimestamp": "1751914859843", # "side": "SELL", # "createdAtDatetime": "2025-07-07T19:00:59.840Z", # "symbol": "ETHUSDT" # } # } # type = self.safe_string(message, 'type') rawTrades = [] if type == 'update': data = self.safe_dict(message, 'data', {}) rawTrades.append(data) # update is a single trade else: rawTrades = self.safe_list(message, 'data', []) # snapshot is a list of trades if len(rawTrades) > 0: if self.myTrades is None: limit = self.safe_integer(self.options, 'tradesLimit', 1000) self.myTrades = ArrayCacheBySymbolById(limit) trades = self.myTrades symbols: dict = {} for i in range(0, len(rawTrades)): rawTrade = rawTrades[i] parsedTrade = self.parse_trade(rawTrade) trades.append(parsedTrade) symbol = self.safe_string(parsedTrade, 'symbol') symbols[symbol] = True messageHash = 'myTrades' client.resolve(trades, messageHash) keys = list(symbols.keys()) for i in range(0, len(keys)): hashSymbol = keys[i] symbolMessageHash = messageHash + '::' + hashSymbol client.resolve(self.myTrades, symbolMessageHash) async def watch_balance(self, params={}) -> Balances: """ watch balance and get the amount of funds available for trading or funds locked in orders https://api.exchange.bullish.com/docs/api/rest/trading-api/v2/#overview--private-data-websocket-authenticated :param dict [params]: extra parameters specific to the exchange API endpoint :param str [params.tradingAccountId]: the trading account id to fetch entries for :returns dict: a `balance structure ` """ await self.load_markets() request: dict = { 'topic': 'assetAccounts', } messageHash = 'balance' tradingAccountId = self.safe_string(params, 'tradingAccountId') if tradingAccountId is not None: params = self.omit(params, 'tradingAccountId') request['tradingAccountId'] = tradingAccountId messageHash += '::' + tradingAccountId return await self.watch_private(messageHash, messageHash, request, params) def handle_balance(self, client: Client, message): # # snapshot # { # "type": "snapshot", # "tradingAccountId": "111309424211255", # "dataType": "V1TAAssetAccount", # "data": [ # { # "updatedAtTimestamp": "1751989627509", # "borrowedQuantity": "0.0000", # "tradingAccountId": "111309424211255", # "loanedQuantity": "0.0000", # "lockedQuantity": "0.0000", # "assetId": "5", # "assetSymbol": "USDC", # "publishedAtTimestamp": "1751989627512", # "availableQuantity": "999672939.8767", # "updatedAtDatetime": "2025-07-08T15:47:07.509Z" # } # ] # } # # update # { # "type": "update", # "tradingAccountId": "111309424211255", # "dataType": "V1TAAssetAccount", # "data": { # "updatedAtTimestamp": "1751989627509", # "borrowedQuantity": "0.0000", # "tradingAccountId": "111309424211255", # "loanedQuantity": "0.0000", # "lockedQuantity": "0.0000", # "assetId": "5", # "assetSymbol": "USDC", # "publishedAtTimestamp": "1751989627512", # "availableQuantity": "999672939.8767", # "updatedAtDatetime": "2025-07-08T15:47:07.509Z" # } # } # tradingAccountId = self.safe_string(message, 'tradingAccountId') if not (tradingAccountId in self.balance): self.balance[tradingAccountId] = {} messageType = self.safe_string(message, 'type') if messageType == 'snapshot': data = self.safe_list(message, 'data', []) self.balance[tradingAccountId] = self.parse_balance(data) else: data = self.safe_dict(message, 'data', {}) assetId = self.safe_string(data, 'assetSymbol') account = self.account() account['total'] = self.safe_string(data, 'availableQuantity') account['used'] = self.safe_string(data, 'lockedQuantity') code = self.safe_currency_code(assetId) self.balance[tradingAccountId][code] = account self.balance[tradingAccountId]['info'] = message self.balance[tradingAccountId] = self.safe_balance(self.balance[tradingAccountId]) messageHash = 'balance' tradingAccountIdHash = '::' + tradingAccountId client.resolve(self.balance[tradingAccountId], messageHash) client.resolve(self.balance[tradingAccountId], messageHash + tradingAccountIdHash) async def watch_positions(self, symbols: Strings = None, since: Int = None, limit: Int = None, params={}) -> List[Position]: """ https://api.exchange.bullish.com/docs/api/rest/trading-api/v2/#overview--private-data-websocket-authenticated 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 ` """ await self.load_markets() subscribeHash = 'positions' messageHash = subscribeHash if not self.is_empty(symbols): symbols = self.market_symbols(symbols) messageHash += '::' + ','.join(symbols) request: dict = { 'topic': 'derivativesPositionsV2', } positions = await self.watch_private(messageHash, subscribeHash, request, params) if self.newUpdates: return positions return self.filter_by_symbols_since_limit(positions, symbols, since, limit, True) def handle_positions(self, client: Client, message): # exchange does not return messages for sandbox mode # current method is implemented blindly # todo: check if self works with not-sandbox mode messageType = self.safe_string(message, 'type') rawPositions = [] if messageType == 'update': data = self.safe_dict(message, 'data', {}) rawPositions.append(data) else: rawPositions = self.safe_list(message, 'data', []) if self.positions is None: self.positions = ArrayCacheBySymbolBySide() positions = self.positions newPositions = [] for i in range(0, len(rawPositions)): rawPosition = rawPositions[i] position = self.parse_position(rawPosition) positions.append(position) newPositions.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(',') symbolPositions = self.filter_by_array(newPositions, 'symbol', symbols, False) if not self.is_empty(symbolPositions): client.resolve(symbolPositions, messageHash) client.resolve(positions, 'positions') def handle_error_message(self, client: Client, message): # # { # "data": { # "errorCode": 401, # "errorCodeName": "UNAUTHORIZED", # "message": "Unable to authenticate; JWT is missing/invalid or unauthorised to access account" # }, # "dataType": "V1TAErrorResponse", # "type": "error" # } # data = self.safe_dict(message, 'data', {}) feedback = self.id + ' ' + self.json(data) try: errorCode = self.safe_string(data, 'errorCode') errorCodeName = self.safe_string(data, 'errorCodeName') self.throw_exactly_matched_exception(self.exceptions['exact'], errorCode, feedback) self.throw_broadly_matched_exception(self.exceptions['broad'], errorCodeName, feedback) raise ExchangeError(feedback) # unknown message except Exception as e: client.reject(e) def handle_message(self, client: Client, message): dataType = self.safe_string(message, 'dataType') result = self.safe_dict(message, 'result') if result is not None: response = self.safe_string(result, 'message') if response == 'Keep alive pong': self.handle_pong(client, message) elif dataType is not None: if dataType == 'V1TAAnonymousTradeUpdate': self.handle_trades(client, message) if dataType == 'V1TATickerResponse': self.handle_ticker(client, message) if dataType == 'V1TALevel2': self.handle_order_book(client, message) if dataType == 'V1TAOrder': self.handle_orders(client, message) if dataType == 'V1TATrade': self.handle_my_trades(client, message) if dataType == 'V1TAAssetAccount': self.handle_balance(client, message) if dataType == 'V1TAErrorResponse': self.handle_error_message(client, message)