From 16197130d3c0faf13e40bfa8c5d4b0d35c9d521c Mon Sep 17 00:00:00 2001 From: Nour Alharithi Date: Thu, 16 Nov 2023 10:54:50 -0800 Subject: [PATCH] new trades publisher --- package.json | 4 +- src/dlob-subscriber/DLOBSubscriberIO.ts | 13 +- .../dlobPublisher.ts} | 8 +- src/publishers/tradesPublisher.ts | 210 ++++++++++++++++++ src/wsConnectionManager.ts | 66 ++++-- yarn.lock | 12 + 6 files changed, 289 insertions(+), 24 deletions(-) rename src/{wsPublish.ts => publishers/dlobPublisher.ts} (93%) create mode 100644 src/publishers/tradesPublisher.ts diff --git a/package.json b/package.json index 6759e75..f171504 100644 --- a/package.json +++ b/package.json @@ -27,6 +27,7 @@ "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", "winston": "^3.8.1", @@ -50,7 +51,8 @@ "clean": "rm -rf lib", "start": "node lib/index.js", "dev": "ts-node src/index.ts", - "ws-publish": "ts-node src/wsPublish.ts", + "dlob-publish": "ts-node src/publishers/dlobPublisher.ts", + "trades-publish": "ts-node src/publishers/tradesPublisher.ts", "ws-manager": "ts-node src/wsConnectionManager.ts", "dev:inspect": "yarn build && node --inspect ./lib/index.js", "dev:debug": "yarn build && node --inspect-brk --inspect=2230 ./lib/index.js", diff --git a/src/dlob-subscriber/DLOBSubscriberIO.ts b/src/dlob-subscriber/DLOBSubscriberIO.ts index c4fe2f6..a61d2d7 100644 --- a/src/dlob-subscriber/DLOBSubscriberIO.ts +++ b/src/dlob-subscriber/DLOBSubscriberIO.ts @@ -9,10 +9,11 @@ import { MainnetSpotMarkets, MarketType, groupL2, + isVariant, } from '@drift-labs/sdk'; import { getOracleForMarket, l2WithBNToStrings } from '../utils/utils'; import { RedisClient } from '../utils/redisClient'; -import { driftEnv } from '../wsPublish'; +import { driftEnv } from '../publishers/dlobPublisher'; type wsMarketL2Args = { marketIndex: number; @@ -82,6 +83,7 @@ export class DLOBSubscriberIO extends DLOBSubscriber { const grouping = l2Args.grouping; const { marketName, ...l2FuncArgs } = l2Args; const l2 = this.getL2(l2FuncArgs); + const marketType = isVariant(l2Args.marketType, 'perp') ? 'perp' : 'spot'; let l2Formatted: any; if (grouping) { const groupingBN = new BN(grouping); @@ -101,14 +103,17 @@ export class DLOBSubscriberIO extends DLOBSubscriber { this.lastSeenL2Formatted .get(l2Args.marketType) ?.set(l2Args.marketIndex, JSON.stringify(l2Formatted)); - l2Formatted['marketName'] = marketName; - l2Formatted['marketType'] = l2Args.marketType; + l2Formatted['marketName'] = marketName?.toUpperCase(); + l2Formatted['marketType'] = marketType?.toLowerCase(); l2Formatted['marketIndex'] = l2Args.marketIndex; l2Formatted['oracle'] = getOracleForMarket( this.driftClient, l2Args.marketType, l2Args.marketIndex ); - this.redisClient.client.publish(marketName, JSON.stringify(l2Formatted)); + this.redisClient.client.publish( + `orderbook_${marketType}_${l2Args.marketIndex}`, + JSON.stringify(l2Formatted) + ); } } diff --git a/src/wsPublish.ts b/src/publishers/dlobPublisher.ts similarity index 93% rename from src/wsPublish.ts rename to src/publishers/dlobPublisher.ts index 69d84e1..4f7399f 100644 --- a/src/wsPublish.ts +++ b/src/publishers/dlobPublisher.ts @@ -12,10 +12,10 @@ import { BulkAccountLoader, } from '@drift-labs/sdk'; -import { logger, setLogLevel } from './utils/logger'; -import { sleep } from './utils/utils'; -import { DLOBSubscriberIO } from './dlob-subscriber/DLOBSubscriberIO'; -import { RedisClient } from './utils/redisClient'; +import { logger, setLogLevel } from '../utils/logger'; +import { sleep } from '../utils/utils'; +import { DLOBSubscriberIO } from '../dlob-subscriber/DLOBSubscriberIO'; +import { RedisClient } from '../utils/redisClient'; require('dotenv').config(); const driftEnv = (process.env.ENV || 'devnet') as DriftEnv; diff --git a/src/publishers/tradesPublisher.ts b/src/publishers/tradesPublisher.ts new file mode 100644 index 0000000..5283394 --- /dev/null +++ b/src/publishers/tradesPublisher.ts @@ -0,0 +1,210 @@ +import { program } from 'commander'; + +import { Connection, Commitment, PublicKey, Keypair } from '@solana/web3.js'; + +import { + DriftClient, + initialize, + DriftEnv, + SlotSubscriber, + UserMap, + Wallet, + BulkAccountLoader, + EventSubscriber, + OrderAction, + MarketType, + convertToNumber, + BASE_PRECISION, + QUOTE_PRECISION, + PRICE_PRECISION, + isVariant, +} from '@drift-labs/sdk'; + +import { logger, setLogLevel } from '../utils/logger'; +import { sleep } from '../utils/utils'; +import { RedisClient } from '../utils/redisClient'; +import { fromEvent, filter, map } from 'rxjs'; + +require('dotenv').config(); +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; + +//@ts-ignore +const sdkConfig = initialize({ env: process.env.ENV }); + +const stateCommitment: Commitment = 'processed'; +const ORDERBOOK_UPDATE_INTERVAL = 1000; + +let driftClient: DriftClient; + +const opts = program.opts(); +setLogLevel(opts.debug ? 'debug' : 'info'); + +const endpoint = process.env.ENDPOINT; +const wsEndpoint = process.env.WS_ENDPOINT; +logger.info(`RPC endpoint: ${endpoint}`); +logger.info(`WS endpoint: ${wsEndpoint}`); +logger.info(`DriftEnv: ${driftEnv}`); +logger.info(`Commit: ${commitHash}`); + +const main = async () => { + const wallet = new Wallet(new Keypair()); + const clearingHousePublicKey = new PublicKey(sdkConfig.DRIFT_PROGRAM_ID); + + const connection = new Connection(endpoint, { + wsEndpoint: wsEndpoint, + commitment: stateCommitment, + }); + + const bulkAccountLoader = new BulkAccountLoader( + connection, + stateCommitment, + ORDERBOOK_UPDATE_INTERVAL + ); + + driftClient = new DriftClient({ + connection, + wallet, + programID: clearingHousePublicKey, + accountSubscription: { + type: 'polling', + accountLoader: bulkAccountLoader, + }, + env: driftEnv, + userStats: true, + }); + + const slotSubscriber = new SlotSubscriber(connection, {}); + + const lamportsBalance = await connection.getBalance(wallet.publicKey); + logger.info( + `DriftClient ProgramId: ${driftClient.program.programId.toBase58()}` + ); + logger.info(`Wallet pubkey: ${wallet.publicKey.toBase58()}`); + logger.info(` . SOL balance: ${lamportsBalance / 10 ** 9}`); + + await driftClient.subscribe(); + driftClient.eventEmitter.on('error', (e) => { + logger.info('clearing house error'); + logger.error(e); + }); + + await slotSubscriber.subscribe(); + + const userMap = new UserMap( + driftClient, + driftClient.userAccountSubscriptionConfig, + false + ); + await userMap.subscribe(); + + const redisClient = new RedisClient(REDIS_HOST, REDIS_PORT, REDIS_PASSWORD); + await redisClient.connect(); + + const eventSubscriber = new EventSubscriber(connection, driftClient.program, { + maxTx: 8192, + maxEventsPerType: 4096, + orderBy: 'client', + commitment: 'confirmed', + logProviderConfig: { + type: 'polling', + frequency: 1000, + }, + }); + + await eventSubscriber.subscribe(); + + const eventObservable = fromEvent(eventSubscriber.eventEmitter, 'newEvent'); + eventObservable + .pipe( + filter( + (event) => + event.eventType === 'OrderActionRecord' && + JSON.stringify(event.action) === JSON.stringify(OrderAction.FILL) + ), + map((fill) => { + return { + ts: fill.ts.toNumber(), + marketIndex: fill.marketIndex, + marketType: isVariant(fill.marketType as MarketType, 'perp') + ? 'perp' + : 'spot', + filler: fill.filler?.toBase58(), + takerFee: convertToNumber(fill.takerFee, BASE_PRECISION), + makerFee: convertToNumber(fill.makerFee, BASE_PRECISION), + quoteAssetAmountSurplus: convertToNumber( + fill.quoteAssetAmountSurplus, + QUOTE_PRECISION + ), + baseAssetAmountFilled: convertToNumber( + fill.baseAssetAmountFilled, + BASE_PRECISION + ), + quoteAssetAmountFilled: convertToNumber( + fill.quoteAssetAmountFilled, + QUOTE_PRECISION + ), + taker: fill.taker?.toBase58(), + takerOrderId: fill.takerOrderId, + takerOrderDirection: isVariant(fill.takerOrderDirection, 'short') + ? 'short' + : 'long', + takerOrderBaseAssetAmount: convertToNumber( + fill.takerOrderBaseAssetAmount, + BASE_PRECISION + ), + takerOrderCumulativeBaseAssetAmountFilled: convertToNumber( + fill.takerOrderCumulativeBaseAssetAmountFilled, + BASE_PRECISION + ), + takerOrderCumulativeQuoteAssetAmountFilled: convertToNumber( + fill.takerOrderCumulativeQuoteAssetAmountFilled, + QUOTE_PRECISION + ), + maker: fill.maker?.toBase58(), + makerOrderId: fill.makerOrderId, + makerOrderDirection: fill.makerOrderDirection, + makerOrderBaseAssetAmount: convertToNumber( + fill.makerOrderBaseAssetAmount, + BASE_PRECISION + ), + makerOrderCumulativeBaseAssetAmountFilled: convertToNumber( + fill.makerOrderCumulativeBaseAssetAmountFilled, + BASE_PRECISION + ), + makerOrderCumulativeQuoteAssetAmountFilled: convertToNumber( + fill.makerOrderCumulativeQuoteAssetAmountFilled, + QUOTE_PRECISION + ), + oraclePrice: convertToNumber(fill.oraclePrice, PRICE_PRECISION), + txSig: fill.txSig, + slot: fill.slot, + }; + }) + ) + .subscribe((fillEvent) => { + redisClient.client.publish( + `trades_${fillEvent.marketType}_${fillEvent.marketIndex}`, + JSON.stringify(fillEvent) + ); + }); + + console.log('Publishing trades'); +}; + +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 }; diff --git a/src/wsConnectionManager.ts b/src/wsConnectionManager.ts index 7be6d1e..42cf309 100644 --- a/src/wsConnectionManager.ts +++ b/src/wsConnectionManager.ts @@ -6,9 +6,11 @@ import { WebSocket, WebSocketServer } from 'ws'; import { sleep } from './utils/utils'; import { RedisClient } from './utils/redisClient'; import { register, Gauge } from 'prom-client'; +import { DriftEnv, PerpMarkets, SpotMarkets } from '@drift-labs/sdk'; // Set up env constants require('dotenv').config(); +const driftEnv = (process.env.ENV || 'devnet') as DriftEnv; const app = express(); app.use(cors({ origin: '*' })); @@ -28,6 +30,40 @@ const REDIS_PORT = process.env.REDIS_PORT || '6379'; const WS_PORT = process.env.WS_PORT || '3000'; const REDIS_PASSWORD = process.env.REDIS_PASSWORD; +const getRedisChannelFromMessage = (message: any): string => { + const channel = message.channel; + const marketName = message.market?.toUpperCase(); + const marketType = message.marketType?.toLowerCase(); + if (!['spot', 'perp'].includes(marketType)) { + throw new Error('Bad market type specified'); + } + + let marketIndex: number; + if (marketType === 'spot') { + marketIndex = SpotMarkets[driftEnv].find( + (market) => market.symbol === marketName + ).marketIndex; + } else if (marketType === 'perp') { + marketIndex = PerpMarkets[driftEnv].find( + (market) => market.symbol === marketName + ).marketIndex; + } + + if (marketIndex === undefined || marketIndex === null) { + throw new Error('Bad market specified'); + } + + switch (channel) { + case 'trades': + return `trades_${marketType}_${marketIndex}`; + case 'orderbook': + return `orderbook_${marketType}_${marketIndex}`; + case undefined: + default: + throw new Error('Bad channel specified'); + } +}; + async function main() { const redisClient = new RedisClient(REDIS_HOST, REDIS_PORT, REDIS_PASSWORD); const lastMessageRetriever = new RedisClient( @@ -85,45 +121,45 @@ async function main() { switch (messageType) { case 'subscribe': { - const channel = parsedMessage.channel; - if (!subscribedChannels.has(channel)) { - console.log('Trying to subscribe to channel', channel); + const redisChannel = getRedisChannelFromMessage(parsedMessage); + if (!subscribedChannels.has(redisChannel)) { + console.log('Trying to subscribe to channel', redisChannel); redisClient.client - .subscribe(channel) + .subscribe(redisChannel) .then(() => { - subscribedChannels.add(channel); + subscribedChannels.add(redisChannel); }) .catch(() => { ws.send( JSON.stringify({ - channel, - error: `Error subscribing to channel: ${channel}`, + redisChannel, + error: `Error subscribing to channel: ${redisChannel}`, }) ); return; }); } - if (!channelSubscribers.get(channel)) { + if (!channelSubscribers.get(redisChannel)) { const subscribers = new Set(); - channelSubscribers.set(channel, subscribers); + channelSubscribers.set(redisChannel, subscribers); } - channelSubscribers.get(channel).add(ws); + channelSubscribers.get(redisChannel).add(ws); // Fetch and send last message const lastMessage = await lastMessageRetriever.client.get( - `last_update_${channel}` + `last_update_${redisChannel}` ); if (lastMessage !== null) { - ws.send(JSON.stringify({ channel, data: lastMessage })); + ws.send(JSON.stringify({ redisChannel, data: lastMessage })); } break; } case 'unsubscribe': { - const channel = parsedMessage.channel; - const subscribers = channelSubscribers.get(channel); + const redisChannel = getRedisChannelFromMessage(parsedMessage); + const subscribers = channelSubscribers.get(redisChannel); if (subscribers) { - channelSubscribers.get(channel).delete(ws); + channelSubscribers.get(redisChannel).delete(ws); } break; } diff --git a/yarn.lock b/yarn.lock index 6da178a..4b4066a 100644 --- a/yarn.lock +++ b/yarn.lock @@ -3590,6 +3590,13 @@ run-parallel@^1.1.9: dependencies: queue-microtask "^1.2.2" +rxjs@^7.8.1: + version "7.8.1" + resolved "https://registry.yarnpkg.com/rxjs/-/rxjs-7.8.1.tgz#6f6f3d99ea8044291efd92e7c7fcf562c4057543" + integrity sha512-AA3TVj+0A2iuIoQkWEK/tqFjBq2j+6PO6Y0zJcvzLAFhEFIO3HL0vls9hWLncZbAAbK0mar7oZ4V079I/qPMxg== + dependencies: + tslib "^2.1.0" + safe-buffer@5.1.2: version "5.1.2" resolved "https://registry.yarnpkg.com/safe-buffer/-/safe-buffer-5.1.2.tgz#991ec69d296e0313747d59bdfd2b745c35f8828d" @@ -3982,6 +3989,11 @@ tslib@^2.0.3, tslib@^2.4.0: resolved "https://registry.yarnpkg.com/tslib/-/tslib-2.4.1.tgz#0d0bfbaac2880b91e22df0768e55be9753a5b17e" integrity sha512-tGyy4dAjRIEwI7BzsB0lynWgOpfqjUdq91XXAlIWD2OwKBH7oCl/GZG/HT4BOHrTlPMOASlMQ7veyTqpmRcrNA== +tslib@^2.1.0: + version "2.6.2" + resolved "https://registry.yarnpkg.com/tslib/-/tslib-2.6.2.tgz#703ac29425e7b37cd6fd456e92404d46d1f3e4ae" + integrity sha512-AEYxH93jGFPn/a2iVAwW87VuUIkR1FVUKB77NwMF7nBTDkDrrT/Hpt/IrCJ0QXhW27jTBDcf5ZY7w6RiqTMw2Q== + tsutils@^3.21.0: version "3.21.0" resolved "https://registry.yarnpkg.com/tsutils/-/tsutils-3.21.0.tgz#b48717d394cea6c1e096983eed58e9d61715b623"