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

Merge pull request #103 from drift-labs/master

Browse Source 
cooldown on rotation + RNDER
This commit is contained in:
Nour Alharithi
2024-02-26 12:52:58 -08:00
committed by GitHub
parent 3d17affa26 ae1a131a63
commit d8fd432f69
3 changed files with 121 additions and 141 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
package.json
View File

@@ -6,7 +6,7 @@
"license": "Apache-2.0",
"dependencies": {
"@coral-xyz/anchor": "^0.29.0",
"@drift-labs/sdk": "2.65.0-beta.9",
"@drift-labs/sdk": "2.65.0-beta.11",
"@opentelemetry/api": "^1.1.0",
"@opentelemetry/auto-instrumentations-node": "^0.31.1",
"@opentelemetry/exporter-prometheus": "^0.31.0",

220
src/index.ts
View File

@@ -8,7 +8,6 @@ import morgan from 'morgan';
import { Commitment, Connection, Keypair, PublicKey } from '@solana/web3.js';
import {
BN,
BulkAccountLoader,
DLOBNode,
DLOBOrder,
@@ -22,10 +21,10 @@ import {
Wallet,
getUserStatsAccountPublicKey,
getVariant,
groupL2,
initialize,
isVariant,
OrderSubscriber,
MarketType,
} from '@drift-labs/sdk';
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 SLOT_STALENESS_TOLERANCE =
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 rateLimitCallsPerSecond = 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`
);
// Handle redis client initialization and rotation maps
const redisClients: Array<RedisClient> = [];
const spotMarketRedisMap: Map<
number,
{ client: RedisClient; clientIndex: number }
{
client: RedisClient;
clientIndex: number;
lastRotationTime: number;
lock: boolean;
}
> = new Map();
const perpMarketRedisMap: Map<
number,
{ client: RedisClient; clientIndex: number }
{
client: RedisClient;
clientIndex: number;
lastRotationTime: number;
lock: boolean;
}
> = new Map();
if (useRedis) {
logger.info('Connecting to redis');
@@ -378,16 +389,73 @@ const main = async (): Promise<void> => {
spotMarketRedisMap.set(sdkConfig.SPOT_MARKETS[i].marketIndex, {
client: redisClients[0],
clientIndex: 0,
lastRotationTime: 0,
lock: false,
});
}
for (let i = 0; i < sdkConfig.PERP_MARKETS.length; i++) {
perpMarketRedisMap.set(sdkConfig.PERP_MARKETS[i].marketIndex, {
client: redisClients[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...`);
const initAllMarketSubscribersStart = Date.now();
MARKET_SUBSCRIBERS = await initializeAllMarketSubscribers(driftClient);
@@ -762,7 +830,6 @@ const main = async (): Promise<void> => {
includeVamm,
includePhoenix,
includeSerum,
grouping, // undefined or PRICE_PRECISION
includeOracle,
} = req.query;
@@ -780,20 +847,14 @@ const main = async (): Promise<void> => {
const isSpot = isVariant(normedMarketType, 'spot');
let adjustedDepth = depth ?? '10';
if (grouping !== undefined) {
// If grouping is also supplied, we want the entire book depth.
// we will apply depth after grouping
adjustedDepth = '-1';
}
const adjustedDepth = depth ?? '100';
let l2Formatted: any;
if (useRedis) {
if (
!isSpot &&
`${includeVamm}`?.toLowerCase() === 'true' &&
`${includeOracle}`?.toLowerCase() === 'true' &&
!grouping
`${includeOracle}`?.toLowerCase() === 'true'
) {
let redisL2: string;
const redisClient = perpMarketRedisMap.get(normedMarketIndex).client;
@@ -818,24 +879,16 @@ const main = async (): Promise<void> => {
l2Formatted = redisL2;
} else {
if (redisL2 && redisClients.length > 1) {
const nextClientIndex =
(perpMarketRedisMap.get(normedMarketIndex).clientIndex + 1) %
redisClients.length;
perpMarketRedisMap.set(normedMarketIndex, {
client: redisClients[nextClientIndex],
clientIndex: nextClientIndex,
});
console.log(
`Rotated redis client to index ${nextClientIndex} for perp market ${normedMarketIndex}`
);
if (canRotate(normedMarketType, normedMarketIndex)) {
rotateClient(normedMarketType, normedMarketIndex);
}
}
}
} else if (
isSpot &&
`${includeSerum}`?.toLowerCase() === 'true' &&
`${includePhoenix}`?.toLowerCase() === 'true' &&
`${includeOracle}`?.toLowerCase() === 'true' &&
!grouping
`${includeOracle}`?.toLowerCase() === 'true'
) {
let redisL2: string;
const redisClient = spotMarketRedisMap.get(normedMarketIndex).client;
@@ -860,16 +913,9 @@ const main = async (): Promise<void> => {
l2Formatted = redisL2;
} else {
if (redisL2 && redisClients.length > 1) {
const nextClientIndex =
(spotMarketRedisMap.get(normedMarketIndex).clientIndex + 1) %
redisClients.length;
spotMarketRedisMap.set(normedMarketIndex, {
client: redisClients[nextClientIndex],
clientIndex: nextClientIndex,
});
console.log(
`Rotated redis client to index ${nextClientIndex} for spot market ${normedMarketIndex}`
);
if (canRotate(normedMarketType, normedMarketIndex)) {
rotateClient(normedMarketType, normedMarketIndex);
}
}
}
}
@@ -900,27 +946,6 @@ const main = async (): Promise<void> => {
: [],
});
if (grouping) {
const finalDepth = depth ? parseInt(depth as string) : 10;
if (isNaN(parseInt(grouping as string))) {
res
.status(400)
.send('Bad Request: grouping must be a number if supplied');
return;
}
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') {
@@ -931,7 +956,7 @@ const main = async (): Promise<void> => {
normedMarketIndex
);
}
}
cacheHitCounter.add(1, {
miss: true,
});
@@ -953,7 +978,6 @@ const main = async (): Promise<void> => {
includePhoenix,
includeSerum,
includeOracle,
grouping, // undefined or PRICE_PRECISION
} = req.query;
const normedParams = normalizeBatchQueryParams({
@@ -965,7 +989,6 @@ const main = async (): Promise<void> => {
includePhoenix: includePhoenix as string | undefined,
includeSerum: includeSerum as string | undefined,
includeOracle: includeOracle as string | undefined,
grouping: grouping as string | undefined,
});
if (normedParams === undefined) {
@@ -977,6 +1000,8 @@ const main = async (): Promise<void> => {
return;
}
let hasError = false;
let errorMessage = '';
const l2s = await Promise.all(
normedParams.map(async (normedParam) => {
const { normedMarketType, normedMarketIndex, error } =
@@ -988,26 +1013,20 @@ const main = async (): Promise<void> => {
normedParam['marketName'] as string
);
if (error) {
res.status(400).send(`Bad Request: ${error}`);
hasError = true;
errorMessage = `Bad Request: ${error}`;
return;
}
const isSpot = isVariant(normedMarketType, 'spot');
let adjustedDepth = normedParam['depth'] ?? '10';
if (normedParam['grouping'] !== undefined) {
// If grouping is also supplied, we want the entire book depth.
// we will apply depth after grouping
adjustedDepth = '-1';
}
const adjustedDepth = normedParam['depth'] ?? '100';
let l2Formatted: any;
if (useRedis) {
if (
!isSpot &&
normedParam['includeVamm']?.toLowerCase() === 'true' &&
normedParam['includeOracle']?.toLowerCase() === 'true' &&
!normedParam['grouping']
normedParam['includeOracle']?.toLowerCase() === 'true'
) {
let redisL2: string;
const redisClient =
@@ -1034,25 +1053,16 @@ const main = async (): Promise<void> => {
l2Formatted = parsedRedisL2;
} else {
if (redisClients.length > 1) {
const nextClientIndex =
(perpMarketRedisMap.get(normedMarketIndex).clientIndex +
1) %
redisClients.length;
perpMarketRedisMap.set(normedMarketIndex, {
client: redisClients[nextClientIndex],
clientIndex: nextClientIndex,
});
console.log(
`Rotated redis client to index ${nextClientIndex} for perp market ${normedMarketIndex}`
);
if (canRotate(normedMarketType, normedMarketIndex)) {
rotateClient(normedMarketType, normedMarketIndex);
}
}
}
}
} else if (
isSpot &&
normedParam['includePhoenix']?.toLowerCase() === 'true' &&
normedParam['includeSerum']?.toLowerCase() === 'true' &&
!normedParam['grouping']
normedParam['includeSerum']?.toLowerCase() === 'true'
) {
let redisL2: string;
const redisClient =
@@ -1079,17 +1089,9 @@ const main = async (): Promise<void> => {
l2Formatted = parsedRedisL2;
} else {
if (redisClients.length > 1) {
const nextClientIndex =
(spotMarketRedisMap.get(normedMarketIndex).clientIndex +
1) %
redisClients.length;
spotMarketRedisMap.set(normedMarketIndex, {
client: redisClients[nextClientIndex],
clientIndex: nextClientIndex,
});
console.log(
`Rotated redis client to index ${nextClientIndex} for spot market ${normedMarketIndex}`
);
if (canRotate(normedMarketType, normedMarketIndex)) {
rotateClient(normedMarketType, normedMarketIndex);
}
}
}
}
@@ -1120,32 +1122,6 @@ const main = async (): Promise<void> => {
: [],
});
if (normedParam['grouping']) {
const finalDepth = normedParam['depth']
? parseInt(normedParam['depth'] as string)
: 10;
if (isNaN(parseInt(normedParam['grouping'] as string))) {
res
.status(400)
.send('Bad Request: grouping must be a number if supplied');
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') {
@@ -1156,7 +1132,6 @@ const main = async (): Promise<void> => {
normedMarketIndex
);
}
}
cacheHitCounter.add(1, {
miss: true,
});
@@ -1164,6 +1139,11 @@ const main = async (): Promise<void> => {
})
);
if (hasError) {
res.status(400).send(errorMessage);
return;
}
res.writeHead(200);
res.end(JSON.stringify({ l2s }));
} catch (err) {

8
yarn.lock
View File

@@ -115,10 +115,10 @@
enabled "2.0.x"
kuler "^2.0.0"
"@drift-labs/sdk@2.65.0-beta.9":
version "2.65.0-beta.9"
resolved "https://registry.yarnpkg.com/@drift-labs/sdk/-/sdk-2.65.0-beta.9.tgz#d278dd7ca50fe65191faa01e7700d6e1daa29928"
integrity sha512-KmWtpW82cyl/tODPNBT+AbOQi8MIdxtLkLwuDyN6pHeVz8Op+Jie5C7iZXTzqaTwrJah8C0fFdsLfK0RWHWMtw==
"@drift-labs/sdk@2.65.0-beta.11":
version "2.65.0-beta.11"
resolved "https://registry.yarnpkg.com/@drift-labs/sdk/-/sdk-2.65.0-beta.11.tgz#c991a4fb895f970bbf52362353ee50c5267b6124"
integrity sha512-R8bKH4/5PItamVMK04CLFt/hUI89oz9ciLfH1YAydoWuHMBCQi1VuU1YdboW+4/DKBDYYSBtSgt4AxrIO50uGw==
dependencies:
"@coral-xyz/anchor" "0.28.1-beta.2"
"@ellipsis-labs/phoenix-sdk" "^1.4.2"