trade/trade-drift-dlob
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #99 from drift-labs/throttle-rotation

Browse Source 
throttle rotation + more opts
This commit is contained in:
Nour Alharithi
2024-02-26 11:19:45 -08:00
committed by GitHub
parent cf3ca3aaae 510d94a4f8
commit 2d09378ef5
1 changed files with 107 additions and 135 deletions
Download Patch File Download Diff File Expand all files Collapse all files

242
src/index.ts
View File

@@ -8,7 +8,6 @@ import morgan from 'morgan';
import { Commitment, Connection, Keypair, PublicKey } from '@solana/web3.js'; import { Commitment, Connection, Keypair, PublicKey } from '@solana/web3.js';
import { import {
BN,
BulkAccountLoader, BulkAccountLoader,
DLOBNode, DLOBNode,
DLOBOrder, DLOBOrder,
@@ -22,10 +21,10 @@ import {
Wallet, Wallet,
getUserStatsAccountPublicKey, getUserStatsAccountPublicKey,
getVariant, getVariant,
groupL2,
initialize, initialize,
isVariant, isVariant,
OrderSubscriber, OrderSubscriber,
MarketType,
} from '@drift-labs/sdk'; } from '@drift-labs/sdk';
import { logger, setLogLevel } from './utils/logger'; import { logger, setLogLevel } from './utils/logger';
@@ -99,6 +98,7 @@ export const ORDERBOOK_UPDATE_INTERVAL = 1000;
const WS_FALLBACK_FETCH_INTERVAL = ORDERBOOK_UPDATE_INTERVAL * 10; const WS_FALLBACK_FETCH_INTERVAL = ORDERBOOK_UPDATE_INTERVAL * 10;
const SLOT_STALENESS_TOLERANCE = const SLOT_STALENESS_TOLERANCE =
parseInt(process.env.SLOT_STALENESS_TOLERANCE) || 20; parseInt(process.env.SLOT_STALENESS_TOLERANCE) || 20;
const ROTATION_COOLDOWN = parseInt(process.env.ROTATION_COOLDOWN) || 5000;
const useWebsocket = process.env.USE_WEBSOCKET?.toLowerCase() === 'true'; const useWebsocket = process.env.USE_WEBSOCKET?.toLowerCase() === 'true';
const rateLimitCallsPerSecond = process.env.RATE_LIMIT_CALLS_PER_SECOND const rateLimitCallsPerSecond = process.env.RATE_LIMIT_CALLS_PER_SECOND
? parseInt(process.env.RATE_LIMIT_CALLS_PER_SECOND) ? parseInt(process.env.RATE_LIMIT_CALLS_PER_SECOND)
@@ -353,14 +353,25 @@ const main = async (): Promise<void> => {
`DLOBSubscriber initialized in ${Date.now() - initDlobSubscriberStart} ms` `DLOBSubscriber initialized in ${Date.now() - initDlobSubscriberStart} ms`
); );
// Handle redis client initialization and rotation maps
const redisClients: Array<RedisClient> = []; const redisClients: Array<RedisClient> = [];
const spotMarketRedisMap: Map< const spotMarketRedisMap: Map<
number, number,
{ client: RedisClient; clientIndex: number } {
client: RedisClient;
clientIndex: number;
lastRotationTime: number;
lock: boolean;
}
> = new Map(); > = new Map();
const perpMarketRedisMap: Map< const perpMarketRedisMap: Map<
number, number,
{ client: RedisClient; clientIndex: number } {
client: RedisClient;
clientIndex: number;
lastRotationTime: number;
lock: boolean;
}
> = new Map(); > = new Map();
if (useRedis) { if (useRedis) {
logger.info('Connecting to redis'); logger.info('Connecting to redis');
@@ -378,16 +389,73 @@ const main = async (): Promise<void> => {
spotMarketRedisMap.set(sdkConfig.SPOT_MARKETS[i].marketIndex, { spotMarketRedisMap.set(sdkConfig.SPOT_MARKETS[i].marketIndex, {
client: redisClients[0], client: redisClients[0],
clientIndex: 0, clientIndex: 0,
lastRotationTime: 0,
lock: false,
}); });
} }
for (let i = 0; i < sdkConfig.PERP_MARKETS.length; i++) { for (let i = 0; i < sdkConfig.PERP_MARKETS.length; i++) {
perpMarketRedisMap.set(sdkConfig.PERP_MARKETS[i].marketIndex, { perpMarketRedisMap.set(sdkConfig.PERP_MARKETS[i].marketIndex, {
client: redisClients[0], client: redisClients[0],
clientIndex: 0, clientIndex: 0,
lastRotationTime: 0,
lock: false,
}); });
} }
} }
function canRotate(marketType: MarketType, marketIndex: number) {
if (isVariant(marketType, 'spot')) {
const state = spotMarketRedisMap.get(marketIndex);
if (state) {
const now = Date.now();
if (now - state.lastRotationTime > ROTATION_COOLDOWN && !state.lock) {
state.lastRotationTime = now;
return true;
}
}
} else {
const state = perpMarketRedisMap.get(marketIndex);
if (state) {
const now = Date.now();
if (now - state.lastRotationTime > ROTATION_COOLDOWN && !state.lock) {
state.lastRotationTime = now;
return true;
}
}
}
return false;
}
function rotateClient(marketType: MarketType, marketIndex: number) {
if (isVariant(marketType, 'spot')) {
const state = spotMarketRedisMap.get(marketIndex);
if (state) {
state.lock = true;
const nextClientIndex = (state.clientIndex + 1) % redisClients.length;
state.client = redisClients[nextClientIndex];
state.clientIndex = nextClientIndex;
logger.info(
`Rotated redis client to index ${nextClientIndex} for spot market ${marketIndex}`
);
state.lastRotationTime = Date.now();
state.lock = false;
}
} else {
const state = perpMarketRedisMap.get(marketIndex);
if (state) {
state.lock = true;
const nextClientIndex = (state.clientIndex + 1) % redisClients.length;
state.client = redisClients[nextClientIndex];
state.clientIndex = nextClientIndex;
logger.info(
`Rotated redis client to index ${nextClientIndex} for perp market ${marketIndex}`
);
state.lastRotationTime = Date.now();
state.lock = false;
}
}
}
logger.info(`Initializing all market subscribers...`); logger.info(`Initializing all market subscribers...`);
const initAllMarketSubscribersStart = Date.now(); const initAllMarketSubscribersStart = Date.now();
MARKET_SUBSCRIBERS = await initializeAllMarketSubscribers(driftClient); MARKET_SUBSCRIBERS = await initializeAllMarketSubscribers(driftClient);
@@ -762,7 +830,6 @@ const main = async (): Promise<void> => {
includeVamm, includeVamm,
includePhoenix, includePhoenix,
includeSerum, includeSerum,
grouping, // undefined or PRICE_PRECISION
includeOracle, includeOracle,
} = req.query; } = req.query;
@@ -780,20 +847,14 @@ const main = async (): Promise<void> => {
const isSpot = isVariant(normedMarketType, 'spot'); const isSpot = isVariant(normedMarketType, 'spot');
let adjustedDepth = depth ?? '10'; const adjustedDepth = depth ?? '100';
if (grouping !== undefined) {
// If grouping is also supplied, we want the entire book depth.
// we will apply depth after grouping
adjustedDepth = '-1';
}
let l2Formatted: any; let l2Formatted: any;
if (useRedis) { if (useRedis) {
if ( if (
!isSpot && !isSpot &&
`${includeVamm}`?.toLowerCase() === 'true' && `${includeVamm}`?.toLowerCase() === 'true' &&
`${includeOracle}`?.toLowerCase() === 'true' && `${includeOracle}`?.toLowerCase() === 'true'
!grouping
) { ) {
let redisL2: string; let redisL2: string;
const redisClient = perpMarketRedisMap.get(normedMarketIndex).client; const redisClient = perpMarketRedisMap.get(normedMarketIndex).client;
@@ -818,24 +879,16 @@ const main = async (): Promise<void> => {
l2Formatted = redisL2; l2Formatted = redisL2;
} else { } else {
if (redisL2 && redisClients.length > 1) { if (redisL2 && redisClients.length > 1) {
const nextClientIndex = if (canRotate(normedMarketType, normedMarketIndex)) {
(perpMarketRedisMap.get(normedMarketIndex).clientIndex + 1) % rotateClient(normedMarketType, normedMarketIndex);
redisClients.length; }
perpMarketRedisMap.set(normedMarketIndex, {
client: redisClients[nextClientIndex],
clientIndex: nextClientIndex,
});
console.log(
`Rotated redis client to index ${nextClientIndex} for perp market ${normedMarketIndex}`
);
} }
} }
} else if ( } else if (
isSpot && isSpot &&
`${includeSerum}`?.toLowerCase() === 'true' && `${includeSerum}`?.toLowerCase() === 'true' &&
`${includePhoenix}`?.toLowerCase() === 'true' && `${includePhoenix}`?.toLowerCase() === 'true' &&
`${includeOracle}`?.toLowerCase() === 'true' && `${includeOracle}`?.toLowerCase() === 'true'
!grouping
) { ) {
let redisL2: string; let redisL2: string;
const redisClient = spotMarketRedisMap.get(normedMarketIndex).client; const redisClient = spotMarketRedisMap.get(normedMarketIndex).client;
@@ -860,16 +913,9 @@ const main = async (): Promise<void> => {
l2Formatted = redisL2; l2Formatted = redisL2;
} else { } else {
if (redisL2 && redisClients.length > 1) { if (redisL2 && redisClients.length > 1) {
const nextClientIndex = if (canRotate(normedMarketType, normedMarketIndex)) {
(spotMarketRedisMap.get(normedMarketIndex).clientIndex + 1) % rotateClient(normedMarketType, normedMarketIndex);
redisClients.length; }
spotMarketRedisMap.set(normedMarketIndex, {
client: redisClients[nextClientIndex],
clientIndex: nextClientIndex,
});
console.log(
`Rotated redis client to index ${nextClientIndex} for spot market ${normedMarketIndex}`
);
} }
} }
} }
@@ -900,38 +946,17 @@ const main = async (): Promise<void> => {
: [], : [],
}); });
if (grouping) { // make the BNs into strings
const finalDepth = depth ? parseInt(depth as string) : 10; l2Formatted = l2WithBNToStrings(l2);
if (isNaN(parseInt(grouping as string))) { if (`${includeOracle}`.toLowerCase() === 'true') {
res addOracletoResponse(
.status(400) l2Formatted,
.send('Bad Request: grouping must be a number if supplied'); driftClient,
return; normedMarketType,
} normedMarketIndex
const groupingBN = new BN(parseInt(grouping as string));
const l2Formatted = l2WithBNToStrings(
groupL2(l2, groupingBN, finalDepth)
); );
if (`${includeOracle}`.toLowerCase() === 'true') {
addOracletoResponse(
l2Formatted,
driftClient,
normedMarketType,
normedMarketIndex
);
}
} else {
// make the BNs into strings
l2Formatted = l2WithBNToStrings(l2);
if (`${includeOracle}`.toLowerCase() === 'true') {
addOracletoResponse(
l2Formatted,
driftClient,
normedMarketType,
normedMarketIndex
);
}
} }
cacheHitCounter.add(1, { cacheHitCounter.add(1, {
miss: true, miss: true,
}); });
@@ -953,7 +978,6 @@ const main = async (): Promise<void> => {
includePhoenix, includePhoenix,
includeSerum, includeSerum,
includeOracle, includeOracle,
grouping, // undefined or PRICE_PRECISION
} = req.query; } = req.query;
const normedParams = normalizeBatchQueryParams({ const normedParams = normalizeBatchQueryParams({
@@ -965,7 +989,6 @@ const main = async (): Promise<void> => {
includePhoenix: includePhoenix as string | undefined, includePhoenix: includePhoenix as string | undefined,
includeSerum: includeSerum as string | undefined, includeSerum: includeSerum as string | undefined,
includeOracle: includeOracle as string | undefined, includeOracle: includeOracle as string | undefined,
grouping: grouping as string | undefined,
}); });
if (normedParams === undefined) { if (normedParams === undefined) {
@@ -994,20 +1017,13 @@ const main = async (): Promise<void> => {
const isSpot = isVariant(normedMarketType, 'spot'); const isSpot = isVariant(normedMarketType, 'spot');
let adjustedDepth = normedParam['depth'] ?? '10'; const adjustedDepth = normedParam['depth'] ?? '100';
if (normedParam['grouping'] !== undefined) {
// If grouping is also supplied, we want the entire book depth.
// we will apply depth after grouping
adjustedDepth = '-1';
}
let l2Formatted: any; let l2Formatted: any;
if (useRedis) { if (useRedis) {
if ( if (
!isSpot && !isSpot &&
normedParam['includeVamm']?.toLowerCase() === 'true' && normedParam['includeVamm']?.toLowerCase() === 'true' &&
normedParam['includeOracle']?.toLowerCase() === 'true' && normedParam['includeOracle']?.toLowerCase() === 'true'
!normedParam['grouping']
) { ) {
let redisL2: string; let redisL2: string;
const redisClient = const redisClient =
@@ -1034,25 +1050,16 @@ const main = async (): Promise<void> => {
l2Formatted = parsedRedisL2; l2Formatted = parsedRedisL2;
} else { } else {
if (redisClients.length > 1) { if (redisClients.length > 1) {
const nextClientIndex = if (canRotate(normedMarketType, normedMarketIndex)) {
(perpMarketRedisMap.get(normedMarketIndex).clientIndex + rotateClient(normedMarketType, normedMarketIndex);
1) % }
redisClients.length;
perpMarketRedisMap.set(normedMarketIndex, {
client: redisClients[nextClientIndex],
clientIndex: nextClientIndex,
});
console.log(
`Rotated redis client to index ${nextClientIndex} for perp market ${normedMarketIndex}`
);
} }
} }
} }
} else if ( } else if (
isSpot && isSpot &&
normedParam['includePhoenix']?.toLowerCase() === 'true' && normedParam['includePhoenix']?.toLowerCase() === 'true' &&
normedParam['includeSerum']?.toLowerCase() === 'true' && normedParam['includeSerum']?.toLowerCase() === 'true'
!normedParam['grouping']
) { ) {
let redisL2: string; let redisL2: string;
const redisClient = const redisClient =
@@ -1079,17 +1086,9 @@ const main = async (): Promise<void> => {
l2Formatted = parsedRedisL2; l2Formatted = parsedRedisL2;
} else { } else {
if (redisClients.length > 1) { if (redisClients.length > 1) {
const nextClientIndex = if (canRotate(normedMarketType, normedMarketIndex)) {
(spotMarketRedisMap.get(normedMarketIndex).clientIndex + rotateClient(normedMarketType, normedMarketIndex);
1) % }
redisClients.length;
spotMarketRedisMap.set(normedMarketIndex, {
client: redisClients[nextClientIndex],
clientIndex: nextClientIndex,
});
console.log(
`Rotated redis client to index ${nextClientIndex} for spot market ${normedMarketIndex}`
);
} }
} }
} }
@@ -1120,42 +1119,15 @@ const main = async (): Promise<void> => {
: [], : [],
}); });
if (normedParam['grouping']) { // make the BNs into strings
const finalDepth = normedParam['depth'] l2Formatted = l2WithBNToStrings(l2);
? parseInt(normedParam['depth'] as string) if (`${normedParam['includeOracle']}`.toLowerCase() === 'true') {
: 10; addOracletoResponse(
if (isNaN(parseInt(normedParam['grouping'] as string))) { l2Formatted,
res driftClient,
.status(400) normedMarketType,
.send('Bad Request: grouping must be a number if supplied'); normedMarketIndex
return;
}
const groupingBN = new BN(
parseInt(normedParam['grouping'] as string)
); );
l2Formatted = l2WithBNToStrings(
groupL2(l2, groupingBN, finalDepth)
);
if (`${normedParam['includeOracle']}`.toLowerCase() === 'true') {
addOracletoResponse(
l2Formatted,
driftClient,
normedMarketType,
normedMarketIndex
);
}
} else {
// make the BNs into strings
l2Formatted = l2WithBNToStrings(l2);
if (`${normedParam['includeOracle']}`.toLowerCase() === 'true') {
addOracletoResponse(
l2Formatted,
driftClient,
normedMarketType,
normedMarketIndex
);
}
} }
cacheHitCounter.add(1, { cacheHitCounter.add(1, {
miss: true, miss: true,