code/bybit-api
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #176 from tiagosiebler/topics

Browse Source 
Batch sub/unsubscribe to topics for spot v3
This commit is contained in:
Tiago
2022-09-28 10:28:57 +01:00
committed by GitHub
parent 2971dd7f2b a19200d311
commit 9b845eede3
5 changed files with 232 additions and 98 deletions
Download Patch File Download Diff File Expand all files Collapse all files

95
examples/ws-public.ts
View File

@@ -16,10 +16,10 @@ import { DefaultLogger, WS_KEY_MAP, WebsocketClient } from '../src';
// market: 'linear', // market: 'linear',
// market: 'inverse', // market: 'inverse',
// market: 'spot', // market: 'spot',
// market: 'spotv3', market: 'spotv3',
// market: 'usdcOption', // market: 'usdcOption',
// market: 'usdcPerp', // market: 'usdcPerp',
market: 'unifiedPerp', // market: 'unifiedPerp',
// market: 'unifiedOption', // market: 'unifiedOption',
}, },
logger logger
@@ -59,7 +59,73 @@ import { DefaultLogger, WS_KEY_MAP, WebsocketClient } from '../src';
// wsClient.subscribe('trade.BTCUSDT'); // wsClient.subscribe('trade.BTCUSDT');
// Spot V3 // Spot V3
// wsClient.subscribe('trade.BTCUSDT'); wsClient.subscribe('trade.BTCUSDT');
// Or an array of topics
// wsClient.subscribe([
// 'orderbook.40.BTCUSDT',
// 'orderbook.40.BTCUSDC',
// 'orderbook.40.USDCUSDT',
// 'orderbook.40.BTCDAI',
// 'orderbook.40.DAIUSDT',
// 'orderbook.40.ETHUSDT',
// 'orderbook.40.ETHUSDC',
// 'orderbook.40.ETHDAI',
// 'orderbook.40.XRPUSDT',
// 'orderbook.40.XRPUSDC',
// 'orderbook.40.EOSUSDT',
// 'orderbook.40.EOSUSDC',
// 'orderbook.40.DOTUSDT',
// 'orderbook.40.DOTUSDC',
// 'orderbook.40.XLMUSDT',
// 'orderbook.40.XLMUSDC',
// 'orderbook.40.LTCUSDT',
// 'orderbook.40.LTCUSDC',
// 'orderbook.40.DOGEUSDT',
// 'orderbook.40.DOGEUSDC',
// 'orderbook.40.BITUSDT',
// 'orderbook.40.BITUSDC',
// 'orderbook.40.BITDAI',
// 'orderbook.40.CHZUSDT',
// 'orderbook.40.CHZUSDC',
// 'orderbook.40.MANAUSDT',
// 'orderbook.40.MANAUSDC',
// 'orderbook.40.LINKUSDT',
// 'orderbook.40.LINKUSDC',
// 'orderbook.40.ICPUSDT',
// 'orderbook.40.ICPUSDC',
// 'orderbook.40.ADAUSDT',
// 'orderbook.40.ADAUSDC',
// 'orderbook.40.SOLUSDC',
// 'orderbook.40.SOLUSDT',
// 'orderbook.40.MATICUSDC',
// 'orderbook.40.MATICUSDT',
// 'orderbook.40.SANDUSDC',
// 'orderbook.40.SANDUSDT',
// 'orderbook.40.LUNCUSDC',
// 'orderbook.40.LUNCUSDT',
// 'orderbook.40.SLGUSDC',
// 'orderbook.40.SLGUSDT',
// 'orderbook.40.AVAXUSDC',
// 'orderbook.40.AVAXUSDT',
// 'orderbook.40.OPUSDC',
// 'orderbook.40.OPUSDT',
// 'orderbook.40.OKSEUSDC',
// 'orderbook.40.OKSEUSDT',
// 'orderbook.40.APEXUSDC',
// 'orderbook.40.APEXUSDT',
// 'orderbook.40.TRXUSDC',
// 'orderbook.40.TRXUSDT',
// 'orderbook.40.GMTUSDC',
// 'orderbook.40.GMTUSDT',
// 'orderbook.40.SHIBUSDC',
// 'orderbook.40.SHIBUSDT',
// 'orderbook.40.LDOUSDC',
// 'orderbook.40.LDOUSDT',
// 'orderbook.40.APEUSDC',
// 'orderbook.40.APEUSDT',
// 'orderbook.40.FILUSDC',
// 'orderbook.40.FILUSDT',
// ]);
// usdc options // usdc options
// wsClient.subscribe([ // wsClient.subscribe([
@@ -72,13 +138,30 @@ import { DefaultLogger, WS_KEY_MAP, WebsocketClient } from '../src';
// wsClient.subscribe('trade.BTCPERP'); // wsClient.subscribe('trade.BTCPERP');
// unified perps // unified perps
wsClient.subscribe('publicTrade.BTCUSDT'); // wsClient.subscribe('publicTrade.BTCUSDT');
// For spot v1 (the old, deprecated client), request public connection first then send required topics on 'open'
// Not necessary for spot v3
// wsClient.connectPublic();
// To unsubscribe from topics (after a 5 second delay, in this example):
// setTimeout(() => { // setTimeout(() => {
// console.log('unsubscribing'); // console.log('unsubscribing');
// wsClient.unsubscribe('trade.BTCUSDT'); // wsClient.unsubscribe('trade.BTCUSDT');
// }, 5 * 1000); // }, 5 * 1000);
// For spot, request public connection first then send required topics on 'open' // Topics are tracked per websocket type
// wsClient.connectPublic(); // Get a list of subscribed topics (e.g. for public v3 spot topics) (after a 5 second delay)
setTimeout(() => {
const publicSpotTopics = wsClient
.getWsStore()
.getTopics(WS_KEY_MAP.spotV3Public);
console.log('public spot topics: ', publicSpotTopics);
const privateSpotTopics = wsClient
.getWsStore()
.getTopics(WS_KEY_MAP.spotV3Private);
console.log('private spot topics: ', privateSpotTopics);
}, 5 * 1000);
})(); })();

2
package.json
View File

@@ -1,6 +1,6 @@
{ {
"name": "bybit-api", "name": "bybit-api",
"version": "3.0.1", "version": "3.0.2",
"description": "Complete & robust node.js SDK for Bybit's REST APIs and WebSockets, with TypeScript & integration tests.", "description": "Complete & robust node.js SDK for Bybit's REST APIs and WebSockets, with TypeScript & integration tests.",
"main": "lib/index.js", "main": "lib/index.js",
"types": "lib/index.d.ts", "types": "lib/index.d.ts",

22
src/util/websocket-util.ts
View File

@@ -256,6 +256,28 @@ export function getWsKeyForTopic(
} }
} }
export function getMaxTopicsPerSubscribeEvent(
market: APIMarket
): number | null {
switch (market) {
case 'inverse':
case 'linear':
case 'usdcOption':
case 'usdcPerp':
case 'unifiedOption':
case 'unifiedPerp':
case 'spot': {
return null;
}
case 'spotv3': {
return 10;
}
default: {
throw neverGuard(market, `getWsKeyForTopic(): Unhandled market`);
}
}
}
export function getUsdcWsKeyForTopic( export function getUsdcWsKeyForTopic(
topic: string, topic: string,
subGroup: 'option' | 'perp' subGroup: 'option' | 'perp'

198
src/websocket-client.ts
View File

@@ -30,6 +30,7 @@ import {
WS_BASE_URL_MAP, WS_BASE_URL_MAP,
getWsKeyForTopic, getWsKeyForTopic,
neverGuard, neverGuard,
getMaxTopicsPerSubscribeEvent,
} from './util'; } from './util';
import { USDCOptionClient } from './usdc-option-client'; import { USDCOptionClient } from './usdc-option-client';
import { USDCPerpetualClient } from './usdc-perpetual-client'; import { USDCPerpetualClient } from './usdc-perpetual-client';
@@ -108,11 +109,78 @@ export class WebsocketClient extends EventEmitter {
} }
/** /**
* Only used if we fetch exchange time before attempting auth. * Subscribe to topics & track/persist them. They will be automatically resubscribed to if the connection drops/reconnects.
* Disabled by default. * @param wsTopics topic or list of topics
* @param isPrivateTopic optional - the library will try to detect private topics, you can use this to mark a topic as private (if the topic isn't recognised yet)
*/
public subscribe(wsTopics: WsTopic[] | WsTopic, isPrivateTopic?: boolean) {
const topics = Array.isArray(wsTopics) ? wsTopics : [wsTopics];
topics.forEach((topic) =>
this.wsStore.addTopic(
getWsKeyForTopic(this.options.market, topic, isPrivateTopic),
topic
)
);
// attempt to send subscription topic per websocket
this.wsStore.getKeys().forEach((wsKey: WsKey) => {
// if connected, send subscription request
if (
this.wsStore.isConnectionState(wsKey, WsConnectionStateEnum.CONNECTED)
) {
return this.requestSubscribeTopics(wsKey, [
...this.wsStore.getTopics(wsKey),
]);
}
// start connection process if it hasn't yet begun. Topics are automatically subscribed to on-connect
if (
!this.wsStore.isConnectionState(
wsKey,
WsConnectionStateEnum.CONNECTING
) &&
!this.wsStore.isConnectionState(
wsKey,
WsConnectionStateEnum.RECONNECTING
)
) {
return this.connect(wsKey);
}
});
}
/**
* Unsubscribe from topics & remove them from memory. They won't be re-subscribed to if the connection reconnects.
* @param wsTopics topic or list of topics
* @param isPrivateTopic optional - the library will try to detect private topics, you can use this to mark a topic as private (if the topic isn't recognised yet)
*/
public unsubscribe(wsTopics: WsTopic[] | WsTopic, isPrivateTopic?: boolean) {
const topics = Array.isArray(wsTopics) ? wsTopics : [wsTopics];
topics.forEach((topic) =>
this.wsStore.deleteTopic(
getWsKeyForTopic(this.options.market, topic, isPrivateTopic),
topic
)
);
this.wsStore.getKeys().forEach((wsKey: WsKey) => {
// unsubscribe request only necessary if active connection exists
if (
this.wsStore.isConnectionState(wsKey, WsConnectionStateEnum.CONNECTED)
) {
this.requestUnsubscribeTopics(wsKey, [
...this.wsStore.getTopics(wsKey),
]);
}
});
}
/**
* @private Only used if we fetch exchange time before attempting auth. Disabled by default.
* I've removed this for ftx and it's working great, tempted to remove this here * I've removed this for ftx and it's working great, tempted to remove this here
*/ */
prepareRESTClient(): void { private prepareRESTClient(): void {
switch (this.options.market) { switch (this.options.market) {
case 'inverse': { case 'inverse': {
this.restClient = new InverseClient( this.restClient = new InverseClient(
@@ -174,6 +242,11 @@ export class WebsocketClient extends EventEmitter {
} }
} }
/** Get the WsStore that tracks websockets & topics */
public getWsStore(): WsStore {
return this.wsStore;
}
public isTestnet(): boolean { public isTestnet(): boolean {
return this.options.testnet === true; return this.options.testnet === true;
} }
@@ -503,13 +576,31 @@ export class WebsocketClient extends EventEmitter {
} }
/** /**
* Send WS message to subscribe to topics. * @private Use the `subscribe(topics)` method to subscribe to topics. Send WS message to subscribe to topics.
*/ */
private requestSubscribeTopics(wsKey: WsKey, topics: string[]) { private requestSubscribeTopics(wsKey: WsKey, topics: string[]) {
if (!topics.length) { if (!topics.length) {
return; return;
} }
const maxTopicsPerEvent = getMaxTopicsPerSubscribeEvent(
this.options.market
);
if (maxTopicsPerEvent && topics.length > maxTopicsPerEvent) {
this.logger.silly(
`Subscribing to topics in batches of ${maxTopicsPerEvent}`
);
for (var i = 0; i < topics.length; i += maxTopicsPerEvent) {
const batch = topics.slice(i, i + maxTopicsPerEvent);
this.logger.silly(`Subscribing to batch of ${batch.length}`);
this.requestSubscribeTopics(wsKey, batch);
}
this.logger.silly(
`Finished batch subscribing to ${topics.length} topics`
);
return;
}
const wsMessage = JSON.stringify({ const wsMessage = JSON.stringify({
req_id: topics.join(','), req_id: topics.join(','),
op: 'subscribe', op: 'subscribe',
@@ -520,12 +611,31 @@ export class WebsocketClient extends EventEmitter {
} }
/** /**
* Send WS message to unsubscribe from topics. * @private Use the `unsubscribe(topics)` method to unsubscribe from topics. Send WS message to unsubscribe from topics.
*/ */
private requestUnsubscribeTopics(wsKey: WsKey, topics: string[]) { private requestUnsubscribeTopics(wsKey: WsKey, topics: string[]) {
if (!topics.length) { if (!topics.length) {
return; return;
} }
const maxTopicsPerEvent = getMaxTopicsPerSubscribeEvent(
this.options.market
);
if (maxTopicsPerEvent && topics.length > maxTopicsPerEvent) {
this.logger.silly(
`Unsubscribing to topics in batches of ${maxTopicsPerEvent}`
);
for (var i = 0; i < topics.length; i += maxTopicsPerEvent) {
const batch = topics.slice(i, i + maxTopicsPerEvent);
this.logger.silly(`Unsubscribing to batch of ${batch.length}`);
this.requestUnsubscribeTopics(wsKey, batch);
}
this.logger.silly(
`Finished batch unsubscribing to ${topics.length} topics`
);
return;
}
const wsMessage = JSON.stringify({ const wsMessage = JSON.stringify({
op: 'unsubscribe', op: 'unsubscribe',
args: topics, args: topics,
@@ -622,11 +732,11 @@ export class WebsocketClient extends EventEmitter {
this.clearPongTimer(wsKey); this.clearPongTimer(wsKey);
const msg = JSON.parse((event && event.data) || event); const msg = JSON.parse((event && event.data) || event);
this.logger.silly('Received event', { // this.logger.silly('Received event', {
...loggerCategory, // ...loggerCategory,
wsKey, // wsKey,
msg: JSON.stringify(msg), // msg: JSON.stringify(msg),
}); // });
// TODO: cleanme // TODO: cleanme
if (msg['success'] || msg?.pong || isWsPong(msg)) { if (msg['success'] || msg?.pong || isWsPong(msg)) {
@@ -771,74 +881,6 @@ export class WebsocketClient extends EventEmitter {
); );
} }
/**
* Add topic/topics to WS subscription list
* @param wsTopics topic or list of topics
* @param isPrivateTopic optional - the library will try to detect private topics, you can use this to mark a topic as private (if the topic isn't recognised yet)
*/
public subscribe(wsTopics: WsTopic[] | WsTopic, isPrivateTopic?: boolean) {
const topics = Array.isArray(wsTopics) ? wsTopics : [wsTopics];
topics.forEach((topic) =>
this.wsStore.addTopic(
getWsKeyForTopic(this.options.market, topic, isPrivateTopic),
topic
)
);
// attempt to send subscription topic per websocket
this.wsStore.getKeys().forEach((wsKey: WsKey) => {
// if connected, send subscription request
if (
this.wsStore.isConnectionState(wsKey, WsConnectionStateEnum.CONNECTED)
) {
return this.requestSubscribeTopics(wsKey, [
...this.wsStore.getTopics(wsKey),
]);
}
// start connection process if it hasn't yet begun. Topics are automatically subscribed to on-connect
if (
!this.wsStore.isConnectionState(
wsKey,
WsConnectionStateEnum.CONNECTING
) &&
!this.wsStore.isConnectionState(
wsKey,
WsConnectionStateEnum.RECONNECTING
)
) {
return this.connect(wsKey);
}
});
}
/**
* Remove topic/topics from WS subscription list
* @param wsTopics topic or list of topics
* @param isPrivateTopic optional - the library will try to detect private topics, you can use this to mark a topic as private (if the topic isn't recognised yet)
*/
public unsubscribe(wsTopics: WsTopic[] | WsTopic, isPrivateTopic?: boolean) {
const topics = Array.isArray(wsTopics) ? wsTopics : [wsTopics];
topics.forEach((topic) =>
this.wsStore.deleteTopic(
getWsKeyForTopic(this.options.market, topic, isPrivateTopic),
topic
)
);
this.wsStore.getKeys().forEach((wsKey: WsKey) => {
// unsubscribe request only necessary if active connection exists
if (
this.wsStore.isConnectionState(wsKey, WsConnectionStateEnum.CONNECTED)
) {
this.requestUnsubscribeTopics(wsKey, [
...this.wsStore.getTopics(wsKey),
]);
}
});
}
/** @deprecated use "market: 'spotv3" client */ /** @deprecated use "market: 'spotv3" client */
public subscribePublicSpotTrades(symbol: string, binary?: boolean) { public subscribePublicSpotTrades(symbol: string, binary?: boolean) {
if (this.options.market !== 'spot') { if (this.options.market !== 'spot') {

13
test/linear/private.write.test.ts
View File

@@ -45,7 +45,6 @@ describe('Private Linear REST API POST Endpoints', () => {
}) })
).toMatchObject({ ).toMatchObject({
ret_code: API_ERROR_CODE.ORDER_NOT_FOUND_OR_TOO_LATE, ret_code: API_ERROR_CODE.ORDER_NOT_FOUND_OR_TOO_LATE,
ret_msg: 'order not exists or too late to cancel',
}); });
}); });
@@ -67,7 +66,6 @@ describe('Private Linear REST API POST Endpoints', () => {
}) })
).toMatchObject({ ).toMatchObject({
ret_code: API_ERROR_CODE.ORDER_NOT_FOUND_OR_TOO_LATE, ret_code: API_ERROR_CODE.ORDER_NOT_FOUND_OR_TOO_LATE,
ret_msg: 'order not exists or too late to replace',
}); });
}); });
@@ -88,7 +86,6 @@ describe('Private Linear REST API POST Endpoints', () => {
}) })
).toMatchObject({ ).toMatchObject({
ret_code: API_ERROR_CODE.INSUFFICIENT_BALANCE_FOR_ORDER_COST_LINEAR, ret_code: API_ERROR_CODE.INSUFFICIENT_BALANCE_FOR_ORDER_COST_LINEAR,
ret_msg: 'Insufficient wallet balance',
}); });
}); });
@@ -100,7 +97,6 @@ describe('Private Linear REST API POST Endpoints', () => {
}) })
).toMatchObject({ ).toMatchObject({
ret_code: API_ERROR_CODE.ORDER_NOT_FOUND_OR_TOO_LATE_LINEAR, ret_code: API_ERROR_CODE.ORDER_NOT_FOUND_OR_TOO_LATE_LINEAR,
ret_msg: 'order not exists or too late to cancel',
}); });
}); });
@@ -122,7 +118,6 @@ describe('Private Linear REST API POST Endpoints', () => {
}) })
).toMatchObject({ ).toMatchObject({
ret_code: API_ERROR_CODE.ORDER_NOT_FOUND_OR_TOO_LATE_LINEAR, ret_code: API_ERROR_CODE.ORDER_NOT_FOUND_OR_TOO_LATE_LINEAR,
ret_msg: 'order not exists or too late to replace',
}); });
}); });
@@ -135,7 +130,6 @@ describe('Private Linear REST API POST Endpoints', () => {
}) })
).toMatchObject({ ).toMatchObject({
ret_code: API_ERROR_CODE.AUTO_ADD_MARGIN_NOT_MODIFIED, ret_code: API_ERROR_CODE.AUTO_ADD_MARGIN_NOT_MODIFIED,
ret_msg: 'autoAddMargin not modified',
}); });
}); });
@@ -149,7 +143,6 @@ describe('Private Linear REST API POST Endpoints', () => {
}) })
).toMatchObject({ ).toMatchObject({
ret_code: API_ERROR_CODE.ISOLATED_NOT_MODIFIED_LINEAR, ret_code: API_ERROR_CODE.ISOLATED_NOT_MODIFIED_LINEAR,
ret_msg: 'Isolated not modified',
}); });
}); });
@@ -161,7 +154,6 @@ describe('Private Linear REST API POST Endpoints', () => {
}) })
).toMatchObject({ ).toMatchObject({
ret_code: API_ERROR_CODE.POSITION_MODE_NOT_MODIFIED, ret_code: API_ERROR_CODE.POSITION_MODE_NOT_MODIFIED,
ret_msg: 'position mode not modified',
}); });
}); });
@@ -173,7 +165,6 @@ describe('Private Linear REST API POST Endpoints', () => {
}) })
).toMatchObject({ ).toMatchObject({
ret_code: API_ERROR_CODE.SAME_SLTP_MODE_LINEAR, ret_code: API_ERROR_CODE.SAME_SLTP_MODE_LINEAR,
ret_msg: 'same tp sl mode2',
}); });
}); });
@@ -186,7 +177,6 @@ describe('Private Linear REST API POST Endpoints', () => {
}) })
).toMatchObject({ ).toMatchObject({
ret_code: API_ERROR_CODE.POSITION_SIZE_IS_ZERO, ret_code: API_ERROR_CODE.POSITION_SIZE_IS_ZERO,
ret_msg: 'position size is zero',
}); });
}); });
@@ -199,7 +189,6 @@ describe('Private Linear REST API POST Endpoints', () => {
}) })
).toMatchObject({ ).toMatchObject({
ret_code: API_ERROR_CODE.LEVERAGE_NOT_MODIFIED, ret_code: API_ERROR_CODE.LEVERAGE_NOT_MODIFIED,
ret_msg: 'leverage not modified',
}); });
}); });
@@ -212,7 +201,6 @@ describe('Private Linear REST API POST Endpoints', () => {
}) })
).toMatchObject({ ).toMatchObject({
ret_code: API_ERROR_CODE.CANNOT_SET_LINEAR_TRADING_STOP_FOR_ZERO_POS, ret_code: API_ERROR_CODE.CANNOT_SET_LINEAR_TRADING_STOP_FOR_ZERO_POS,
ret_msg: 'can not set tp/sl/ts for zero position',
}); });
}); });
@@ -225,7 +213,6 @@ describe('Private Linear REST API POST Endpoints', () => {
}) })
).toMatchObject({ ).toMatchObject({
ret_code: API_ERROR_CODE.RISK_ID_NOT_MODIFIED, ret_code: API_ERROR_CODE.RISK_ID_NOT_MODIFIED,
ret_msg: 'risk id not modified',
}); });
}); });
}); });