diff --git a/package.json b/package.json index 477827e..c2f06c5 100644 --- a/package.json +++ b/package.json @@ -1,75 +1,76 @@ { - "name": "@drift-labs/dlob-server", - "version": "0.1.0", - "author": "wphan", - "main": "lib/index.js", - "license": "Apache-2.0", - "dependencies": { - "@coral-xyz/anchor": "^0.29.0", - "@drift/common": "file:./drift-common/common-ts", - "@drift-labs/sdk": "file:./drift-common/protocol/sdk", - "@opentelemetry/api": "^1.1.0", - "@opentelemetry/auto-instrumentations-node": "^0.31.1", - "@opentelemetry/exporter-prometheus": "^0.31.0", - "@opentelemetry/sdk-node": "^0.31.0", - "@project-serum/anchor": "^0.19.1-beta.1", - "@project-serum/serum": "^0.13.65", - "@solana/web3.js": "^1.73.3", - "@triton-one/yellowstone-grpc": "^0.3.0", - "@types/redis": "^4.0.11", - "@types/ws": "^8.5.8", - "async-mutex": "^0.4.0", - "axios": "^1.6.2", - "commander": "^9.4.0", - "compression": "^1.7.4", - "cors": "^2.8.5", - "dotenv": "^10.0.0", - "express": "^4.18.2", - "ioredis": "^5.3.2", - "morgan": "^1.10.0", - "prom-client": "^15.0.0", - "redis": "^4.6.10", - "response-time": "^2.3.2", - "rxjs": "^7.8.1", - "socket.io-redis": "^6.1.1", - "typescript": "4.5.4", - "undici": "^6.16.1", - "winston": "^3.8.1", - "ws": "^8.14.2" - }, - "devDependencies": { - "@types/k6": "^0.45.0", - "@typescript-eslint/eslint-plugin": "^4.28.0", - "@typescript-eslint/parser": "^4.28.0", - "eslint": "^7.29.0", - "eslint-config-prettier": "^8.3.0", - "eslint-plugin-prettier": "^3.4.0", - "husky": "^7.0.4", - "k6": "^0.0.0", - "prettier": "^2.4.1", - "ts-node": "^10.9.1" - }, - "scripts": { - "prepare": "husky install", - "build": "yarn clean && tsc", - "clean": "rm -rf lib", - "start": "node lib/index.js", - "dev": "ts-node src/index.ts", - "server-lite": "ts-node src/serverLite.ts", - "dlob-publish": "ts-node src/publishers/dlobPublisher.ts", - "trades-publish": "ts-node src/publishers/tradesPublisher.ts", - "fees-publish": "ts-node src/publishers/priorityFeesPublisher.ts", - "pnl-publish": "ts-node src/scripts/publishUnsettledPnlUsers.ts", - "ws-manager": "ts-node src/wsConnectionManager.ts", - "ws-manager:inspect": "yarn build && node --inspect ./lib/wsConnectionManager.js", - "dev:inspect": "yarn build && node --inspect ./lib/index.js", - "dev:debug": "yarn build && node --inspect-brk --inspect=2230 ./lib/index.js", - "example": "ts-node example/client.ts", - "exampleWithSlot": "ts-node example/clientWithSlot.ts", - "prettify": "prettier --check './src/**/*.ts'", - "prettify:fix": "prettier --write './src/**/*.ts'", - "lint": "eslint . --ext ts --quiet", - "lint:fix": "eslint . --ext ts --fix", - "playground": "ts-node src/playground.ts" - } -} \ No newline at end of file + "name": "@drift-labs/dlob-server", + "version": "0.1.0", + "author": "wphan", + "main": "lib/index.js", + "license": "Apache-2.0", + "dependencies": { + "@coral-xyz/anchor": "^0.29.0", + "@drift/common": "file:./drift-common/common-ts", + "@drift-labs/sdk": "file:./drift-common/protocol/sdk", + "@opentelemetry/api": "^1.1.0", + "@opentelemetry/auto-instrumentations-node": "^0.31.1", + "@opentelemetry/exporter-prometheus": "^0.31.0", + "@opentelemetry/sdk-node": "^0.31.0", + "@project-serum/anchor": "^0.19.1-beta.1", + "@project-serum/serum": "^0.13.65", + "@solana/web3.js": "^1.73.3", + "@triton-one/yellowstone-grpc": "^0.3.0", + "@types/redis": "^4.0.11", + "@types/ws": "^8.5.8", + "async-mutex": "^0.4.0", + "axios": "^1.6.2", + "bottleneck": "^2.19.5", + "commander": "^9.4.0", + "compression": "^1.7.4", + "cors": "^2.8.5", + "dotenv": "^10.0.0", + "express": "^4.18.2", + "ioredis": "^5.3.2", + "morgan": "^1.10.0", + "prom-client": "^15.0.0", + "redis": "^4.6.10", + "response-time": "^2.3.2", + "rxjs": "^7.8.1", + "socket.io-redis": "^6.1.1", + "typescript": "4.5.4", + "undici": "^6.16.1", + "winston": "^3.8.1", + "ws": "^8.14.2" + }, + "devDependencies": { + "@types/k6": "^0.45.0", + "@typescript-eslint/eslint-plugin": "^4.28.0", + "@typescript-eslint/parser": "^4.28.0", + "eslint": "^7.29.0", + "eslint-config-prettier": "^8.3.0", + "eslint-plugin-prettier": "^3.4.0", + "husky": "^7.0.4", + "k6": "^0.0.0", + "prettier": "^2.4.1", + "ts-node": "^10.9.1" + }, + "scripts": { + "prepare": "husky install", + "build": "yarn clean && tsc", + "clean": "rm -rf lib", + "start": "node lib/index.js", + "dev": "ts-node src/index.ts", + "server-lite": "ts-node src/serverLite.ts", + "dlob-publish": "ts-node src/publishers/dlobPublisher.ts", + "trades-publish": "ts-node src/publishers/tradesPublisher.ts", + "fees-publish": "ts-node src/publishers/priorityFeesPublisher.ts", + "pnl-publish": "ts-node src/scripts/publishUnsettledPnlUsers.ts", + "ws-manager": "ts-node src/wsConnectionManager.ts", + "ws-manager:inspect": "yarn build && node --inspect ./lib/wsConnectionManager.js", + "dev:inspect": "yarn build && node --inspect ./lib/index.js", + "dev:debug": "yarn build && node --inspect-brk --inspect=2230 ./lib/index.js", + "example": "ts-node example/client.ts", + "exampleWithSlot": "ts-node example/clientWithSlot.ts", + "prettify": "prettier --check './src/**/*.ts'", + "prettify:fix": "prettier --write './src/**/*.ts'", + "lint": "eslint . --ext ts --quiet", + "lint:fix": "eslint . --ext ts --fix", + "playground": "ts-node src/playground.ts" + } +} diff --git a/src/index.ts b/src/index.ts index 023fb3f..54c3935 100644 --- a/src/index.ts +++ b/src/index.ts @@ -657,7 +657,10 @@ const main = async (): Promise => { return; } - const redisClient = perpMarketRedisMap.get(marketIndex).client; + const redisClient = redisClients.find( + (client) => + client.forceGetClient().options.keyPrefix === RedisClientPrefix.DLOB + ); const redisResponseGainers = await redisClient.getRaw( `perp_market_${marketIndex}_gainers` diff --git a/src/scripts/publishUnsettledPnlUsers.ts b/src/scripts/publishUnsettledPnlUsers.ts index edbf923..d1cac5a 100644 --- a/src/scripts/publishUnsettledPnlUsers.ts +++ b/src/scripts/publishUnsettledPnlUsers.ts @@ -1,4 +1,4 @@ -import { RedisClient, RedisClientPrefix } from '@drift/common'; +import { RedisClient, RedisClientPrefix, sleep } from '@drift/common'; import { BigNum, DriftClient, @@ -11,11 +11,11 @@ import { Wallet, ZERO, calculateClaimablePnl, - calculatePositionPNL, decodeUser, } from '@drift-labs/sdk'; import { Connection, Keypair } from '@solana/web3.js'; import { logger } from '../utils/logger'; +import Bottleneck from 'bottleneck'; const dotenv = require('dotenv'); dotenv.config(); @@ -29,10 +29,19 @@ type UserPnlMap = { }; }; +type AllPnlUsers = { + [perpMarketIndex: number]: { + users: { userPubKey: string; pnl: number }[]; + }; +}; + const startTime = Date.now(); const endpoint = process.env.ENDPOINT as string; const wsEndpoint = process.env.WS_ENDPOINT as string; const driftEnv = process.env.ENV as string; +const chunkSize = Number(process.env.CHUNK_SIZE) || 100; +const sleepTimeMs = Number(process.env.SLEEP_TIME_MS) || 500; +const delayMs = Number(process.env.DEFAULT_DELAY_MS) || 100; const connection = new Connection(endpoint, { commitment: 'confirmed', @@ -45,6 +54,8 @@ const driftClient = new DriftClient({ }); const userMapRedisClient = new RedisClient({ + host: process.env.ELASTICACHE_USERMAP_HOST ?? process.env.ELASTICACHE_HOST, + port: process.env.ELASTICACHE_USERMAP_PORT ?? process.env.ELASTICACHE_PORT, prefix: RedisClientPrefix.USER_MAP, }); @@ -56,14 +67,41 @@ const main = async () => { // get the users from usermap redis client await driftClient.subscribe(); + const limiter = new Bottleneck({ + maxConcurrent: 10, + minTime: delayMs, + }); + const userStrings = await userMapRedisClient.lRange('user_pubkeys', 0, -1); - const redisUsers = await Promise.all( - userStrings.map((userStr) => getUserFromRedis(userStr)) - ); + const totalCount = userStrings.length; + let finishedCount = 0; + let allNonZeroPnls = {}; - // construct an object with the top 20/bottom 20 unsettled in each perp market - const userPnlMap = createMarketSpecificPnlLeaderboards(redisUsers); + logger.info(`Starting loop for total of ${totalCount} users`); + + while (finishedCount < totalCount) { + const userChunk = userStrings.slice( + finishedCount, + finishedCount + chunkSize + ); + + limiter.schedule(async () => { + const redisUsers = await Promise.all( + userChunk.map((userStr) => getUserFromRedis(userStr)) + ); + // add all the nonzero pnl users from each market in chunks + allNonZeroPnls = buildUserMarketLists(redisUsers, allNonZeroPnls); + }); + + finishedCount += chunkSize; + + logger.info(`Wrote ${finishedCount} users, sleeping for ${sleepTimeMs}ms`); + + await sleep(sleepTimeMs); + } + + const userPnlMap = createMarketPnlLeaderboards(allNonZeroPnls); const success = await writeToDlobRedis(userPnlMap); @@ -73,6 +111,12 @@ const main = async () => { } in ${Date.now() - startTime} ms` ); + await sleep(5000); + + logger.info('==== TEST PRINT LOSERS FROM BTC-PERP ===='); + const losers = await dlobRedisClient.getRaw(`perp_market_1_losers`); + logger.info(losers); + process.exit(); }; @@ -84,30 +128,66 @@ const writeToDlobRedis = async (userPnlMap: UserPnlMap): Promise => { const losersRedisKey = `perp_market_${perpMarketIndex}_losers`; // write the new lists - await dlobRedisClient.setRaw( - gainersRedisKey, - JSON.stringify(userPnlMap[perpMarketIndex].gain) - ); - await dlobRedisClient.setRaw( - losersRedisKey, - JSON.stringify(userPnlMap[perpMarketIndex].loss) - ); - - return; + await Promise.all([ + dlobRedisClient.setRaw( + gainersRedisKey, + JSON.stringify(userPnlMap[perpMarketIndex].gain ?? []) + ), + dlobRedisClient.setRaw( + losersRedisKey, + JSON.stringify(userPnlMap[perpMarketIndex].loss ?? []) + ), + ]).then(([_winnersResult, _losersResult]) => { + logger.info( + `Wrote ${userPnlMap[perpMarketIndex].gain.length} users to redis key: ${gainersRedisKey}` + ); + logger.info( + `Wrote ${userPnlMap[perpMarketIndex].loss.length} users to redis key: ${losersRedisKey}` + ); + }); }) ); return true; } catch (e) { - console.log('Error writing to dlob redis client: ', e); + logger.info(`Error writing to dlob redis client: ${e}`); return false; } }; -const createMarketSpecificPnlLeaderboards = ( - redisUsers: { user: User; bufferString: string }[] -): UserPnlMap => { - const pnlMap = {}; +const createMarketPnlLeaderboards = (allPnlUsers: AllPnlUsers): UserPnlMap => { + let pnlMap = {}; + + Object.keys(allPnlUsers).forEach((perpMarketIndex) => { + try { + const sortedPnls = allPnlUsers[perpMarketIndex].users.sort( + (a, b) => b.pnl - a.pnl + ); + // store the top 20 winners and losers in each perp market + pnlMap[perpMarketIndex] = { + gain: sortedPnls.slice(0, 20), + loss: sortedPnls.slice(-20).reverse(), + }; + } catch (e) { + logger.info( + `Error in ${PerpMarkets[driftEnv][perpMarketIndex].symbol}: ${e}` + ); + + pnlMap[perpMarketIndex] = { + gain: [], + loss: [], + }; + } + }); + + return pnlMap; +}; + +const buildUserMarketLists = ( + redisUsers: { user: User; bufferString: string }[], + allPnlUsers: AllPnlUsers +): AllPnlUsers => { + let newPnlUsers = allPnlUsers; const usdcSpotMarket = driftClient.getSpotMarketAccount( QUOTE_SPOT_MARKET_INDEX @@ -123,7 +203,7 @@ const createMarketSpecificPnlLeaderboards = ( perpMarket.marketIndex ); - const allNonZeroPnls = redisUsers.flatMap((redisUser) => { + const allNonZeroPnlsInMarket = redisUsers.flatMap((redisUser) => { try { const perpPosition = redisUser.user.getPerpPosition( perpMarket.marketIndex @@ -144,22 +224,13 @@ const createMarketSpecificPnlLeaderboards = ( false )[0]; - let marketPnl = calculatePositionPNL( + const marketPnl = calculateClaimablePnl( perpMarketAccount, + usdcSpotMarket, perpPositionWithLpSettle, - true, oraclePriceData ); - if (marketPnl.gt(ZERO)) { - marketPnl = calculateClaimablePnl( - perpMarketAccount, - usdcSpotMarket, - perpPositionWithLpSettle, - oraclePriceData - ); - } - return marketPnl.eq(ZERO) ? [] : { @@ -175,19 +246,21 @@ const createMarketSpecificPnlLeaderboards = ( } }); - const sortedPnls = allNonZeroPnls.sort((a, b) => b.pnl - a.pnl); - - // store the top 20 winners and losers in each perp market - pnlMap[perpMarket.marketIndex] = { - gain: sortedPnls.slice(0, 20), - loss: sortedPnls.slice(-20, -1), - }; + if (!allPnlUsers[perpMarket.marketIndex]) { + newPnlUsers[perpMarket.marketIndex] = { users: allNonZeroPnlsInMarket }; + } else { + newPnlUsers[perpMarket.marketIndex] = { + users: allPnlUsers[perpMarket.marketIndex].users.concat( + allNonZeroPnlsInMarket + ), + }; + } } catch (e) { logger.error(`Could not fetch PnLs for ${perpMarket.symbol}: `, e); } } - return pnlMap; + return newPnlUsers; }; const getUserFromRedis = async (userAccountStr: string) => {