From 5493351f40e25d85469b459c2a5572e812145836 Mon Sep 17 00:00:00 2001 From: Nour Alharithi Date: Fri, 8 Dec 2023 17:09:50 -0800 Subject: [PATCH] fetch from redis --- src/dlob-subscriber/DLOBSubscriberIO.ts | 30 ++- src/index.ts | 263 +++++++++++++++++------- src/publishers/dlobPublisher.ts | 5 +- 3 files changed, 214 insertions(+), 84 deletions(-) diff --git a/src/dlob-subscriber/DLOBSubscriberIO.ts b/src/dlob-subscriber/DLOBSubscriberIO.ts index a7cea28..0a20194 100644 --- a/src/dlob-subscriber/DLOBSubscriberIO.ts +++ b/src/dlob-subscriber/DLOBSubscriberIO.ts @@ -62,8 +62,8 @@ export class DLOBSubscriberIO extends DLOBSubscriber { marketType: MarketType.PERP, marketName: market.symbol, depth: -1, - includeVamm: true, numVammOrders: 100, + includeVamm: true, updateOnChange: true, fallbackL2Generators: [], }); @@ -124,13 +124,39 @@ export class DLOBSubscriberIO extends DLOBSubscriber { l2Args.marketType, l2Args.marketIndex ); + + const l2Formatted_depth100 = Object.assign({}, l2Formatted, { + bids: l2Formatted.bids.slice(0, 100), + asks: l2Formatted.asks.slice(0, 100), + }); + const l2Formatted_depth20 = Object.assign({}, l2Formatted, { + bids: l2Formatted.bids.slice(0, 20), + asks: l2Formatted.asks.slice(0, 20), + }); + const l2Formatted_depth5 = Object.assign({}, l2Formatted, { + bids: l2Formatted.bids.slice(0, 5), + asks: l2Formatted.asks.slice(0, 5), + }); + this.redisClient.client.publish( `orderbook_${marketType}_${l2Args.marketIndex}`, JSON.stringify(l2Formatted) ); this.redisClient.client.set( `last_update_orderbook_${marketType}_${l2Args.marketIndex}`, - JSON.stringify(l2Formatted) + JSON.stringify(l2Formatted_depth100) + ); + this.redisClient.client.set( + `last_update_orderbook_${marketType}_${l2Args.marketIndex}_depth_100`, + JSON.stringify(l2Formatted_depth100) + ); + this.redisClient.client.set( + `last_update_orderbook_${marketType}_${l2Args.marketIndex}_depth_20`, + JSON.stringify(l2Formatted_depth20) + ); + this.redisClient.client.set( + `last_update_orderbook_${marketType}_${l2Args.marketIndex}_depth_5`, + JSON.stringify(l2Formatted_depth5) ); } } diff --git a/src/index.ts b/src/index.ts index a584c0a..e364c7d 100644 --- a/src/index.ts +++ b/src/index.ts @@ -28,8 +28,6 @@ import { initialize, isVariant, OrderSubscriber, - UserAccount, - Order, } from '@drift-labs/sdk'; import { logger, setLogLevel } from './utils/logger'; @@ -59,8 +57,14 @@ import { getDLOBProviderFromOrderSubscriber, getDLOBProviderFromUserMap, } from './dlobProvider'; +import { RedisClient } from './utils/redisClient'; require('dotenv').config(); + +const REDIS_HOST = process.env.REDIS_HOST || 'localhost'; +const REDIS_PORT = process.env.REDIS_PORT || '6379'; +const REDIS_PASSWORD = process.env.REDIS_PASSWORD; + const driftEnv = (process.env.ENV || 'devnet') as DriftEnv; const commitHash = process.env.COMMIT; //@ts-ignore @@ -343,6 +347,10 @@ const main = async () => { `DLOBSubscriber initialized in ${Date.now() - initDlobSubscriberStart} ms` ); + logger.info('Connecting to redis'); + const redisClient = new RedisClient(REDIS_HOST, REDIS_PORT, REDIS_PASSWORD); + await redisClient.connect(); + logger.info(`Initializing all market subscribers...`); const initAllMarketSubscribersStart = Date.now(); MARKET_SUBSCRIBERS = await initializeAllMarketSubscribers(driftClient); @@ -742,6 +750,55 @@ const main = async () => { adjustedDepth = '-1'; } + let l2Formatted: any; + if ( + !isSpot && + `${includeVamm}`.toLowerCase() === 'true' && + `${includeOracle}`.toLowerCase().toLowerCase() === 'true' && + !grouping + ) { + if (parseInt(adjustedDepth as string) === 5) { + l2Formatted = await redisClient.client.get( + `last_update_orderbook_perp_${normedMarketIndex}_depth_5` + ); + } else if (parseInt(adjustedDepth as string) === 20) { + l2Formatted = await redisClient.client.get( + `last_update_orderbook_perp_${normedMarketIndex}_depth_20` + ); + } else if (parseInt(adjustedDepth as string) === 100) { + l2Formatted = await redisClient.client.get( + `last_update_orderbook_perp_${normedMarketIndex}_depth_100` + ); + } + } else if ( + isSpot && + `${includeSerum}`.toLowerCase() === 'true' && + `${includePhoenix}`.toLowerCase() === 'true' && + `${includeOracle}`.toLowerCase() === 'true' && + !grouping + ) { + if (parseInt(adjustedDepth as string) === 5) { + l2Formatted = await redisClient.client.get( + `last_update_orderbook_spot_${normedMarketIndex}_depth_5` + ); + } else if (parseInt(adjustedDepth as string) === 20) { + l2Formatted = await redisClient.client.get( + `last_update_orderbook_spot_${normedMarketIndex}_depth_20` + ); + } else if (parseInt(adjustedDepth as string) === 100) { + console.log('100'); + l2Formatted = await redisClient.client.get( + `last_update_orderbook_spot_${normedMarketIndex}_depth_100` + ); + } + } + + if (l2Formatted) { + res.writeHead(200); + res.end(JSON.stringify(l2Formatted)); + return; + } + const l2 = dlobSubscriber.getL2({ marketIndex: normedMarketIndex, marketType: normedMarketType, @@ -783,7 +840,7 @@ const main = async () => { res.end(JSON.stringify(l2Formatted)); } else { // make the BNs into strings - const l2Formatted = l2WithBNToStrings(l2); + l2Formatted = l2WithBNToStrings(l2); if (`${includeOracle}`.toLowerCase() === 'true') { addOracletoResponse( l2Formatted, @@ -792,10 +849,10 @@ const main = async () => { normedMarketIndex ); } - - res.writeHead(200); - res.end(JSON.stringify(l2Formatted)); } + + res.writeHead(200); + res.end(JSON.stringify(l2Formatted)); } catch (err) { next(err); } @@ -836,86 +893,132 @@ const main = async () => { return; } - const l2s = normedParams.map((normedParam) => { - const { normedMarketType, normedMarketIndex, error } = - validateDlobQuery( - driftClient, - driftEnv, - normedParam['marketType'] as string, - normedParam['marketIndex'] as string, - normedParam['marketName'] as string - ); - if (error) { - res.status(400).send(`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 l2 = dlobSubscriber.getL2({ - marketIndex: normedMarketIndex, - marketType: normedMarketType, - depth: parseInt(adjustedDepth as string), - includeVamm: isSpot - ? false - : `${normedParam['includeVamm']}`.toLowerCase() === 'true', - fallbackL2Generators: isSpot - ? [ - `${normedParam['includePhoenix']}`.toLowerCase() === 'true' && - MARKET_SUBSCRIBERS[normedMarketIndex].phoenix, - `${normedParam['includeSerum']}`.toLowerCase() === 'true' && - MARKET_SUBSCRIBERS[normedMarketIndex].serum, - ].filter((a) => !!a) - : [], - }); - - 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'); + const l2s = await Promise.all( + normedParams.map(async (normedParam) => { + const { normedMarketType, normedMarketIndex, error } = + validateDlobQuery( + driftClient, + driftEnv, + normedParam['marketType'] as string, + normedParam['marketIndex'] as string, + normedParam['marketName'] as string + ); + if (error) { + res.status(400).send(`Bad Request: ${error}`); return; } - const groupingBN = new BN( - parseInt(normedParam['grouping'] as string) - ); - const l2Formatted = l2WithBNToStrings( - groupL2(l2, groupingBN, finalDepth) - ); - if (`${normedParam['includeOracle']}`.toLowerCase() === 'true') { - addOracletoResponse( - l2Formatted, - driftClient, - normedMarketType, - normedMarketIndex + 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'; + } + + let l2Formatted: any; + if ( + marketType === 'perp' && + normedParam['includeVamm'].toLowerCase() === 'true' && + normedParam['includeOracle'].toLowerCase() === 'true' && + !normedParam['grouping'] + ) { + if (parseInt(adjustedDepth as string) === 5) { + l2Formatted = await redisClient.client.get( + `last_update_orderbook_perp_${normedMarketIndex}_depth_5` + ); + } else if (parseInt(adjustedDepth as string) === 20) { + l2Formatted = await redisClient.client.get( + `last_update_orderbook_perp_${normedMarketIndex}_depth_20` + ); + } else if (parseInt(adjustedDepth as string) === 100) { + l2Formatted = await redisClient.client.get( + `last_update_orderbook_perp_${normedMarketIndex}_depth_100` + ); + } + } else if ( + marketType === 'spot' && + normedParam['includePhoenix'].toLowerCase() === 'true' && + normedParam['includeSerum'].toLowerCase() === 'true' && + !normedParam['grouping'] + ) { + if (parseInt(adjustedDepth as string) === 5) { + l2Formatted = await redisClient.client.get( + `last_update_orderbook_spot_${normedMarketIndex}_depth_5` + ); + } else if (parseInt(adjustedDepth as string) === 20) { + l2Formatted = await redisClient.client.get( + `last_update_orderbook_spot_${normedMarketIndex}_depth_20` + ); + } else if (parseInt(adjustedDepth as string) === 100) { + l2Formatted = await redisClient.client.get( + `last_update_orderbook_spot_${normedMarketIndex}_depth_100` + ); + } + } + + if (l2Formatted) { + return l2Formatted; + } + + const l2 = dlobSubscriber.getL2({ + marketIndex: normedMarketIndex, + marketType: normedMarketType, + depth: parseInt(adjustedDepth as string), + includeVamm: isSpot + ? false + : `${normedParam['includeVamm']}`.toLowerCase() === 'true', + fallbackL2Generators: isSpot + ? [ + `${normedParam['includePhoenix']}`.toLowerCase() === 'true' && + MARKET_SUBSCRIBERS[normedMarketIndex].phoenix, + `${normedParam['includeSerum']}`.toLowerCase() === 'true' && + MARKET_SUBSCRIBERS[normedMarketIndex].serum, + ].filter((a) => !!a) + : [], + }); + + 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') { + addOracletoResponse( + l2Formatted, + driftClient, + normedMarketType, + normedMarketIndex + ); + } } return l2Formatted; - } else { - // make the BNs into strings - const l2Formatted = l2WithBNToStrings(l2); - if (`${normedParam['includeOracle']}`.toLowerCase() === 'true') { - addOracletoResponse( - l2Formatted, - driftClient, - normedMarketType, - normedMarketIndex - ); - } - return l2Formatted; - } - }); + }) + ); res.writeHead(200); res.end(JSON.stringify({ l2s })); diff --git a/src/publishers/dlobPublisher.ts b/src/publishers/dlobPublisher.ts index 7aa15d6..b6049e6 100644 --- a/src/publishers/dlobPublisher.ts +++ b/src/publishers/dlobPublisher.ts @@ -36,8 +36,6 @@ import { GeyserOrderSubscriber } from '../grpc/OrderSubscriberGRPC'; require('dotenv').config(); const stateCommitment: Commitment = 'processed'; -const ORDERBOOK_UPDATE_INTERVAL = 400; -const WS_FALLBACK_FETCH_INTERVAL = ORDERBOOK_UPDATE_INTERVAL * 10; const driftEnv = (process.env.ENV || 'devnet') as DriftEnv; const commitHash = process.env.COMMIT; const REDIS_HOST = process.env.REDIS_HOST || 'localhost'; @@ -62,6 +60,9 @@ const useOrderSubscriber = const useGrpc = process.env.USE_GRPC?.toLowerCase() === 'true'; const useWebsocket = process.env.USE_WEBSOCKET?.toLowerCase() === 'true'; +const ORDERBOOK_UPDATE_INTERVAL = useGrpc ? 500 : 1000; +const WS_FALLBACK_FETCH_INTERVAL = 10_000; + logger.info(`RPC endpoint: ${endpoint}`); logger.info(`WS endpoint: ${wsEndpoint}`); logger.info(`DriftEnv: ${driftEnv}`);