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

Merge pull request #138 from drift-labs/master

Browse Source 
cache priority fees
This commit is contained in:
Nour Alharithi
2024-04-04 16:44:07 -07:00
committed by GitHub
parent 377808df45 a993ee22e7
commit 4f44e625de
5 changed files with 276 additions and 249 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
package.json
View File

@@ -6,7 +6,7 @@
"license": "Apache-2.0",
"dependencies": {
"@coral-xyz/anchor": "^0.29.0",
"@drift-labs/sdk": "2.76.0-beta.6",
"@drift-labs/sdk": "2.76.0-beta.7",
"@opentelemetry/api": "^1.1.0",
"@opentelemetry/auto-instrumentations-node": "^0.31.1",
"@opentelemetry/exporter-prometheus": "^0.31.0",
@@ -56,6 +56,7 @@
"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",
"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",

250
src/index.ts
View File

@@ -9,9 +9,6 @@ import { Commitment, Connection, Keypair, PublicKey } from '@solana/web3.js';
import {
BulkAccountLoader,
DLOBNode,
DLOBOrder,
DLOBOrders,
DLOBOrdersCoder,
DLOBSubscriber,
DriftClient,
DriftClientSubscriptionConfig,
@@ -299,8 +296,6 @@ const main = async (): Promise<void> => {
const dlobProvider = getDLOBProviderFromOrderSubscriber(orderSubscriber);
const dlobCoder = DLOBOrdersCoder.create();
await driftClient.subscribe();
driftClient.eventEmitter.on('error', (e) => {
logger.info('clearing house error');
@@ -461,237 +456,24 @@ const main = async (): Promise<void> => {
app.get('/startup', handleStartup);
app.get('/', handleHealthCheck(2 * WS_FALLBACK_FETCH_INTERVAL, dlobProvider));
if (FEATURE_FLAGS.ENABLE_ORDERS_ENDPOINTS) {
app.get('/orders/json/raw', async (_req, res, next) => {
try {
// object with userAccount key and orders object serialized
const orders: Array<any> = [];
const oracles: Array<any> = [];
const slot = dlobProvider.getSlot();
app.get('/priorityFees', async (req, res, next) => {
try {
const { marketIndex, marketType } = req.query;
for (const market of driftClient.getPerpMarketAccounts()) {
const oracle = driftClient.getOracleDataForPerpMarket(
market.marketIndex
);
oracles.push({
marketIndex: market.marketIndex,
...oracle,
});
}
for (const {
userAccount,
publicKey,
} of dlobProvider.getUserAccounts()) {
for (const order of userAccount.orders) {
if (isVariant(order.status, 'init')) {
continue;
}
orders.push({
user: publicKey.toBase58(),
order: order,
});
}
}
// respond with orders
res.writeHead(200);
res.end(
JSON.stringify({
slot,
oracles,
orders,
})
);
} catch (e) {
next(e);
const fees = await redisClients[
parseInt(process.env.HELIUS_REDIS_HOST_INDEX) ?? 0
].client.get(`priorityFees_${marketType}_${marketIndex}`);
if (fees) {
res.status(200).json(JSON.parse(fees));
return;
} else {
res.writeHead(404);
res.end('Not found');
}
});
app.get('/orders/json', async (_req, res, next) => {
try {
// object with userAccount key and orders object serialized
const slot = dlobProvider.getSlot();
const orders: Array<any> = [];
const oracles: Array<any> = [];
for (const market of driftClient.getPerpMarketAccounts()) {
const oracle = driftClient.getOracleDataForPerpMarket(
market.marketIndex
);
const oracleHuman = {
marketIndex: market.marketIndex,
price: oracle.price.toString(),
slot: oracle.slot.toString(),
confidence: oracle.confidence.toString(),
hasSufficientNumberOfDataPoints:
oracle.hasSufficientNumberOfDataPoints,
maxPrice: oracle.maxPrice,
};
if (oracle.twap) {
oracleHuman['twap'] = oracle.twap.toString();
}
if (oracle.twapConfidence) {
oracleHuman['twapConfidence'] = oracle.twapConfidence.toString();
}
oracles.push(oracleHuman);
}
for (const {
userAccount,
publicKey,
} of dlobProvider.getUserAccounts()) {
for (const order of userAccount.orders) {
if (isVariant(order.status, 'init')) {
continue;
}
const orderHuman = {
status: getVariant(order.status),
orderType: getVariant(order.orderType),
marketType: getVariant(order.marketType),
slot: order.slot.toString(),
orderId: order.orderId,
userOrderId: order.userOrderId,
marketIndex: order.marketIndex,
price: order.price.toString(),
baseAssetAmount: order.baseAssetAmount.toString(),
baseAssetAmountFilled: order.baseAssetAmountFilled.toString(),
quoteAssetAmountFilled: order.quoteAssetAmountFilled.toString(),
direction: getVariant(order.direction),
reduceOnly: order.reduceOnly,
triggerPrice: order.triggerPrice.toString(),
triggerCondition: getVariant(order.triggerCondition),
existingPositionDirection: getVariant(
order.existingPositionDirection
),
postOnly: order.postOnly,
immediateOrCancel: order.immediateOrCancel,
oraclePriceOffset: order.oraclePriceOffset,
auctionDuration: order.auctionDuration,
auctionStartPrice: order.auctionStartPrice.toString(),
auctionEndPrice: order.auctionEndPrice.toString(),
maxTs: order.maxTs.toString(),
};
if (order.quoteAssetAmountFilled) {
orderHuman['quoteAssetAmount'] =
order.quoteAssetAmountFilled.toString();
orderHuman['quoteAssetAmountFilled'] =
order.quoteAssetAmountFilled.toString();
}
orders.push({
user: publicKey.toBase58(),
order: orderHuman,
});
}
}
// respond with orders
res.writeHead(200);
res.end(
JSON.stringify({
slot,
oracles,
orders,
})
);
} catch (err) {
next(err);
}
});
app.get('/orders/idl', async (_req, res, next) => {
try {
const dlobOrders: DLOBOrders = [];
for (const {
userAccount,
publicKey,
} of dlobProvider.getUserAccounts()) {
for (const order of userAccount.orders) {
if (isVariant(order.status, 'init')) {
continue;
}
dlobOrders.push({
user: publicKey,
order,
} as DLOBOrder);
}
}
res.writeHead(200);
res.end(dlobCoder.encode(dlobOrders));
} catch (err) {
next(err);
}
});
app.get('/orders/idlWithSlot', async (req, res, next) => {
try {
const { marketName, marketIndex, marketType } = req.query;
const { normedMarketType, normedMarketIndex, error } =
validateDlobQuery(
driftClient,
driftEnv,
marketType as string,
marketIndex as string,
marketName as string
);
const useFilter =
marketName !== undefined ||
marketIndex !== undefined ||
marketType !== undefined;
if (useFilter) {
if (
error ||
normedMarketType === undefined ||
normedMarketIndex === undefined
) {
res.status(400).send(error);
return;
}
}
const dlobOrders: DLOBOrders = [];
for (const {
userAccount,
publicKey,
} of dlobProvider.getUserAccounts()) {
for (const order of userAccount.orders) {
if (isVariant(order.status, 'init')) {
continue;
}
if (useFilter) {
if (
getVariant(order.marketType) !== getVariant(normedMarketType) ||
order.marketIndex !== normedMarketIndex
) {
continue;
}
}
dlobOrders.push({
user: publicKey,
order,
} as DLOBOrder);
}
}
res.end(
JSON.stringify({
slot: dlobProvider.getSlot(),
data: dlobCoder.encode(dlobOrders).toString('base64'),
})
);
} catch (err) {
next(err);
}
});
}
} catch (err) {
next(err);
}
});
app.get('/topMakers', async (req, res, next) => {
try {

247
src/publishers/priorityFeesPublisher.ts Normal file
View File

@@ -0,0 +1,247 @@
import { program } from 'commander';
import { Connection, Commitment, Keypair } from '@solana/web3.js';
import {
DriftClient,
initialize,
DriftEnv,
Wallet,
BulkAccountLoader,
getMarketsAndOraclesForSubscription,
} from '@drift-labs/sdk';
import { logger, setLogLevel } from '../utils/logger';
import { sleep } from '../utils/utils';
import express from 'express';
// import { handleHealthCheck } from '../core/metrics';
import { RedisClient } from '../utils/redisClient';
require('dotenv').config();
const stateCommitment: Commitment = 'confirmed';
const driftEnv = (process.env.ENV || 'devnet') as DriftEnv;
const commitHash = process.env.COMMIT;
const REDIS_HOST = process.env.REDIS_HOST || 'localhost';
const REDIS_PORT = process.env.REDIS_PORT || '6379';
const REDIS_PASSWORD = process.env.REDIS_PASSWORD;
// Set up express for health checks
const app = express();
//@ts-ignore
const sdkConfig = initialize({ env: process.env.ENV });
let driftClient: DriftClient;
const opts = program.opts();
setLogLevel(opts.debug ? 'debug' : 'info');
const token = process.env.TOKEN;
const endpoint = token
? process.env.ENDPOINT + `/${token}`
: process.env.ENDPOINT;
const wsEndpoint = process.env.WS_ENDPOINT;
const FEE_POLLING_FREQUENCY =
parseInt(process.env.FEE_POLLING_FREQUENCY) || 5000;
if (!endpoint.includes('helius')) {
throw new Error('We use helius for fee publisher fellas');
}
logger.info(`RPC endpoint: ${endpoint}`);
logger.info(`WS endpoint: ${wsEndpoint}`);
logger.info(`DriftEnv: ${driftEnv}`);
logger.info(`Commit: ${commitHash}`);
class PriorityFeeSubscriber {
endpoint: string;
perpMarketPubkeys: { marketIndex: number; pubkey: string }[];
spotMarketPubkeys: { marketIndex: number; pubkeys: string[] }[];
redisClient: RedisClient;
frequencyMs: number;
constructor(config: {
endpoint: string;
redisClient: RedisClient;
perpMarketPubkeys: { marketIndex: number; pubkey: string }[];
spotMarketPubkeys: { marketIndex: number; pubkeys: string[] }[];
frequencyMs?: number;
}) {
this.endpoint = config.endpoint;
this.perpMarketPubkeys = config.perpMarketPubkeys;
this.spotMarketPubkeys = config.spotMarketPubkeys;
this.redisClient = config.redisClient;
this.frequencyMs = config.frequencyMs ?? FEE_POLLING_FREQUENCY;
}
async subscribe() {
await this.fetchAndPushPriorityFees();
setInterval(async () => {
await this.fetchAndPushPriorityFees();
}, this.frequencyMs);
}
async fetchAndPushPriorityFees() {
const [resultPerp, resultSpot] = await Promise.all([
fetch(this.endpoint, {
method: 'POST',
headers: {
'Content-Type': 'application/json',
},
body: JSON.stringify(
this.perpMarketPubkeys.map((xx) => {
return {
jsonrpc: '2.0',
id: xx.marketIndex.toString(),
method: 'getPriorityFeeEstimate',
params: [
{
accountKeys: [xx.pubkey],
options: {
includeAllPriorityFeeLevels: true,
},
},
],
};
})
),
}),
fetch(this.endpoint, {
method: 'POST',
headers: {
'Content-Type': 'application/json',
},
body: JSON.stringify(
this.spotMarketPubkeys.map((xx) => {
return {
jsonrpc: '2.0',
id: (100 + xx.marketIndex).toString(),
method: 'getPriorityFeeEstimate',
params: [
{
accountKeys: xx.pubkeys,
options: {
includeAllPriorityFeeLevels: true,
},
},
],
};
})
),
}),
]);
const [dataPerp, dataSpot] = await Promise.all([
resultPerp.json(),
resultSpot.json(),
]);
dataPerp.forEach((result: any) => {
const marketIndex = parseInt(result['id']);
this.redisClient.client.publish(
`priorityFees_perp_${marketIndex}`,
JSON.stringify(result.result['priorityFeeLevels'])
);
this.redisClient.client.set(
`priorityFees_perp_${marketIndex}`,
JSON.stringify(result.result['priorityFeeLevels'])
);
});
dataSpot.forEach((result: any) => {
const marketIndex = parseInt(result['id']) - 100;
this.redisClient.client.publish(
`priorityFees_spot_${marketIndex}`,
JSON.stringify(result.result['priorityFeeLevels'])
);
this.redisClient.client.set(
`priorityFees_spot_${marketIndex}`,
JSON.stringify(result.result['priorityFeeLevels'])
);
});
}
}
const main = async () => {
const connection = new Connection(endpoint, {
wsEndpoint: wsEndpoint,
commitment: stateCommitment,
});
const { perpMarketIndexes, spotMarketIndexes, oracleInfos } =
getMarketsAndOraclesForSubscription(sdkConfig.ENV);
const driftClient = new DriftClient({
connection,
wallet: new Wallet(new Keypair()),
perpMarketIndexes,
spotMarketIndexes,
oracleInfos,
accountSubscription: {
type: 'polling',
accountLoader: new BulkAccountLoader(connection, stateCommitment, 0),
},
});
await driftClient.subscribe();
const perpMarketPubkeys = driftClient.getPerpMarketAccounts().map((acct) => {
return { marketIndex: acct.marketIndex, pubkey: acct.pubkey.toString() };
});
const usdcMarket = driftClient.getSpotMarketAccount(0).pubkey.toString();
const spotMarketPubkeys: { marketIndex: number; pubkeys: string[] }[] = [];
for (const market of sdkConfig.SPOT_MARKETS) {
const pubkeysForMarket = [usdcMarket];
const driftMarket = driftClient.getSpotMarketAccount(market.marketIndex);
pubkeysForMarket.push(driftMarket.pubkey.toString());
if (market.serumMarket) {
pubkeysForMarket.push(market.serumMarket.toString());
}
if (market.phoenixMarket) {
pubkeysForMarket.push(market.phoenixMarket.toString());
}
spotMarketPubkeys.push({
marketIndex: market.marketIndex,
pubkeys: pubkeysForMarket,
});
}
const redisClient = new RedisClient(REDIS_HOST, REDIS_PORT, REDIS_PASSWORD);
await redisClient.connect();
const priorityFeeSubscriber = new PriorityFeeSubscriber({
endpoint,
perpMarketPubkeys,
spotMarketPubkeys,
redisClient,
});
await priorityFeeSubscriber.subscribe();
const server = app.listen(8080);
// Default keepalive is 5s, since the AWS ALB timeout is 60 seconds, clients
// sometimes get 502s.
// https://shuheikagawa.com/blog/2019/04/25/keep-alive-timeout/
// https://stackoverflow.com/a/68922692
server.keepAliveTimeout = 61 * 1000;
server.headersTimeout = 65 * 1000;
console.log('Priority fee publisher Publishing Messages');
};
async function recursiveTryCatch(f: () => void) {
try {
await f();
} catch (e) {
console.error(e);
await sleep(15000);
await recursiveTryCatch(f);
}
}
recursiveTryCatch(() => main());
export { sdkConfig, endpoint, wsEndpoint, driftEnv, commitHash, driftClient };

17
src/wsConnectionManager.ts
View File

@@ -65,11 +65,13 @@ const getRedisChannelFromMessage = (message: any): string => {
throw new Error('Bad market specified');
}
switch (channel) {
switch (channel.toLowerCase()) {
case 'trades':
return `trades_${marketType}_${marketIndex}`;
case 'orderbook':
return `orderbook_${marketType}_${marketIndex}`;
case 'priorityfees':
return `priorityFees_${marketType}_${marketIndex}`;
case undefined:
default:
throw new Error('Bad channel specified');
@@ -145,17 +147,14 @@ async function main() {
JSON.stringify({
channel: requestChannel,
error:
'Error subscribing to channel with data: ' +
JSON.stringify(parsedMessage),
'Error subscribing to channel'
})
);
} else {
ws.close(
1003,
JSON.stringify({
error:
'Error subscribing to channel with data: ' +
JSON.stringify(parsedMessage),
error: 'Error subscribing to channel'
})
);
}
@@ -221,8 +220,7 @@ async function main() {
JSON.stringify({
channel: requestChannel,
error:
'Error unsubscribing from channel with data: ' +
JSON.stringify(parsedMessage),
'Error unsubscribing from channel'
})
);
} else {
@@ -230,8 +228,7 @@ async function main() {
1003,
JSON.stringify({
error:
'Error unsubscribing from channel with data: ' +
JSON.stringify(parsedMessage),
'Error unsubscribing from channel'
})
);
}

8
yarn.lock
View File

@@ -115,10 +115,10 @@
enabled "2.0.x"
kuler "^2.0.0"
"@drift-labs/sdk@2.76.0-beta.6":
version "2.76.0-beta.6"
resolved "https://registry.yarnpkg.com/@drift-labs/sdk/-/sdk-2.76.0-beta.6.tgz#c1f882f52cef4ef33f1ab9c5e346067cd3a325d0"
integrity sha512-0H5MD88lrCL2r4l6dBe8a9kaFrDeiHZfuxGq2Nf0O99V5UU57SJ236uM1OIhumQua4kptL6Mv+snKIZji6NgXQ==
"@drift-labs/sdk@2.76.0-beta.7":
version "2.76.0-beta.7"
resolved "https://registry.yarnpkg.com/@drift-labs/sdk/-/sdk-2.76.0-beta.7.tgz#aa910d2db9446ed825636da1c9984441eb5daa3c"
integrity sha512-nO8eVXQ3hWXJsKm++MRcyUZEXbwUjLLTVpEHvSCOE5Kaqc+z21y+Aqq+44PRMEljqNAE1ouf0Qe6hTsoaqhkZw==
dependencies:
"@coral-xyz/anchor" "0.28.1-beta.2"
"@ellipsis-labs/phoenix-sdk" "^1.4.2"