fetch from redis

This commit is contained in:
Nour Alharithi
2023-12-08 17:09:50 -08:00
parent 9abc31a32d
commit 5493351f40
3 changed files with 214 additions and 84 deletions

View File

@@ -62,8 +62,8 @@ export class DLOBSubscriberIO extends DLOBSubscriber {
marketType: MarketType.PERP, marketType: MarketType.PERP,
marketName: market.symbol, marketName: market.symbol,
depth: -1, depth: -1,
includeVamm: true,
numVammOrders: 100, numVammOrders: 100,
includeVamm: true,
updateOnChange: true, updateOnChange: true,
fallbackL2Generators: [], fallbackL2Generators: [],
}); });
@@ -124,13 +124,39 @@ export class DLOBSubscriberIO extends DLOBSubscriber {
l2Args.marketType, l2Args.marketType,
l2Args.marketIndex 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( this.redisClient.client.publish(
`orderbook_${marketType}_${l2Args.marketIndex}`, `orderbook_${marketType}_${l2Args.marketIndex}`,
JSON.stringify(l2Formatted) JSON.stringify(l2Formatted)
); );
this.redisClient.client.set( this.redisClient.client.set(
`last_update_orderbook_${marketType}_${l2Args.marketIndex}`, `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)
); );
} }
} }

View File

@@ -28,8 +28,6 @@ import {
initialize, initialize,
isVariant, isVariant,
OrderSubscriber, OrderSubscriber,
UserAccount,
Order,
} from '@drift-labs/sdk'; } from '@drift-labs/sdk';
import { logger, setLogLevel } from './utils/logger'; import { logger, setLogLevel } from './utils/logger';
@@ -59,8 +57,14 @@ import {
getDLOBProviderFromOrderSubscriber, getDLOBProviderFromOrderSubscriber,
getDLOBProviderFromUserMap, getDLOBProviderFromUserMap,
} from './dlobProvider'; } from './dlobProvider';
import { RedisClient } from './utils/redisClient';
require('dotenv').config(); 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 driftEnv = (process.env.ENV || 'devnet') as DriftEnv;
const commitHash = process.env.COMMIT; const commitHash = process.env.COMMIT;
//@ts-ignore //@ts-ignore
@@ -343,6 +347,10 @@ const main = async () => {
`DLOBSubscriber initialized in ${Date.now() - initDlobSubscriberStart} ms` `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...`); logger.info(`Initializing all market subscribers...`);
const initAllMarketSubscribersStart = Date.now(); const initAllMarketSubscribersStart = Date.now();
MARKET_SUBSCRIBERS = await initializeAllMarketSubscribers(driftClient); MARKET_SUBSCRIBERS = await initializeAllMarketSubscribers(driftClient);
@@ -742,6 +750,55 @@ const main = async () => {
adjustedDepth = '-1'; 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({ const l2 = dlobSubscriber.getL2({
marketIndex: normedMarketIndex, marketIndex: normedMarketIndex,
marketType: normedMarketType, marketType: normedMarketType,
@@ -783,7 +840,7 @@ const main = async () => {
res.end(JSON.stringify(l2Formatted)); res.end(JSON.stringify(l2Formatted));
} else { } else {
// make the BNs into strings // make the BNs into strings
const l2Formatted = l2WithBNToStrings(l2); l2Formatted = l2WithBNToStrings(l2);
if (`${includeOracle}`.toLowerCase() === 'true') { if (`${includeOracle}`.toLowerCase() === 'true') {
addOracletoResponse( addOracletoResponse(
l2Formatted, l2Formatted,
@@ -792,10 +849,10 @@ const main = async () => {
normedMarketIndex normedMarketIndex
); );
} }
res.writeHead(200);
res.end(JSON.stringify(l2Formatted));
} }
res.writeHead(200);
res.end(JSON.stringify(l2Formatted));
} catch (err) { } catch (err) {
next(err); next(err);
} }
@@ -836,86 +893,132 @@ const main = async () => {
return; return;
} }
const l2s = normedParams.map((normedParam) => { const l2s = await Promise.all(
const { normedMarketType, normedMarketIndex, error } = normedParams.map(async (normedParam) => {
validateDlobQuery( const { normedMarketType, normedMarketIndex, error } =
driftClient, validateDlobQuery(
driftEnv, driftClient,
normedParam['marketType'] as string, driftEnv,
normedParam['marketIndex'] as string, normedParam['marketType'] as string,
normedParam['marketName'] as string normedParam['marketIndex'] as string,
); normedParam['marketName'] as string
if (error) { );
res.status(400).send(`Bad Request: ${error}`); if (error) {
return; res.status(400).send(`Bad Request: ${error}`);
}
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');
return; return;
} }
const groupingBN = new BN(
parseInt(normedParam['grouping'] as string)
);
const l2Formatted = l2WithBNToStrings( const isSpot = isVariant(normedMarketType, 'spot');
groupL2(l2, groupingBN, finalDepth)
); let adjustedDepth = normedParam['depth'] ?? '10';
if (`${normedParam['includeOracle']}`.toLowerCase() === 'true') { if (normedParam['grouping'] !== undefined) {
addOracletoResponse( // If grouping is also supplied, we want the entire book depth.
l2Formatted, // we will apply depth after grouping
driftClient, adjustedDepth = '-1';
normedMarketType, }
normedMarketIndex
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; 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.writeHead(200);
res.end(JSON.stringify({ l2s })); res.end(JSON.stringify({ l2s }));

View File

@@ -36,8 +36,6 @@ import { GeyserOrderSubscriber } from '../grpc/OrderSubscriberGRPC';
require('dotenv').config(); require('dotenv').config();
const stateCommitment: Commitment = 'processed'; 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 driftEnv = (process.env.ENV || 'devnet') as DriftEnv;
const commitHash = process.env.COMMIT; const commitHash = process.env.COMMIT;
const REDIS_HOST = process.env.REDIS_HOST || 'localhost'; const REDIS_HOST = process.env.REDIS_HOST || 'localhost';
@@ -62,6 +60,9 @@ const useOrderSubscriber =
const useGrpc = process.env.USE_GRPC?.toLowerCase() === 'true'; const useGrpc = process.env.USE_GRPC?.toLowerCase() === 'true';
const useWebsocket = process.env.USE_WEBSOCKET?.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(`RPC endpoint: ${endpoint}`);
logger.info(`WS endpoint: ${wsEndpoint}`); logger.info(`WS endpoint: ${wsEndpoint}`);
logger.info(`DriftEnv: ${driftEnv}`); logger.info(`DriftEnv: ${driftEnv}`);