Merge pull request #19 from drift-labs/fills-publisher

Trades publisher

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",

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['marketName'] = marketName?.toUpperCase();
-		l2Formatted['marketType'] = l2Args.marketType;
+		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)
+		);
 	}
 }

src/publishers/dlobPublisher.ts

@@ -12,10 +12,10 @@ import {
 	BulkAccountLoader,
 } from '@drift-labs/sdk';
 
-import { logger, setLogLevel } from './utils/logger';
+import { logger, setLogLevel } from '../utils/logger';
-import { sleep } from './utils/utils';
+import { sleep } from '../utils/utils';
-import { DLOBSubscriberIO } from './dlob-subscriber/DLOBSubscriberIO';
+import { DLOBSubscriberIO } from '../dlob-subscriber/DLOBSubscriberIO';
-import { RedisClient } from './utils/redisClient';
+import { RedisClient } from '../utils/redisClient';
 
 require('dotenv').config();
 const driftEnv = (process.env.ENV || 'devnet') as DriftEnv;

src/publishers/tradesPublisher.ts

@@ -0,0 +1,208 @@
+import { program } from 'commander';
+
+import { Connection, Commitment, PublicKey, Keypair } from '@solana/web3.js';
+
+import {
+	DriftClient,
+	initialize,
+	DriftEnv,
+	SlotSubscriber,
+	UserMap,
+	Wallet,
+	BulkAccountLoader,
+	EventSubscriber,
+	OrderAction,
+	convertToNumber,
+	BASE_PRECISION,
+	QUOTE_PRECISION,
+	PRICE_PRECISION,
+	getVariant,
+} 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: getVariant(fill.marketType),
+					filler: fill.filler?.toBase58(),
+					takerFee: convertToNumber(fill.takerFee, QUOTE_PRECISION),
+					makerFee: convertToNumber(fill.makerFee, QUOTE_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: getVariant(fill.takerOrderDirection),
+					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: getVariant(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,
+					action: 'fill',
+					actionExplanation: getVariant(fill.actionExplanation),
+					referrerReward: convertToNumber(fill.referrerReward, QUOTE_PRECISION),
+				};
+			})
+		)
+		.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 };

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.toUpperCase() === marketName
+		).marketIndex;
+	} else if (marketType === 'perp') {
+		marketIndex = PerpMarkets[driftEnv].find(
+			(market) => market.symbol.toUpperCase() === 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,60 @@ async function main() {
 
 			switch (messageType) {
 				case 'subscribe': {
-					const channel = parsedMessage.channel;
+					let redisChannel: string;
-					if (!subscribedChannels.has(channel)) {
+					try {
-						console.log('Trying to subscribe to channel', channel);
+						redisChannel = getRedisChannelFromMessage(parsedMessage);
+					} catch (error) {
+						ws.send(JSON.stringify({ error: error.message }));
+						return;
+					}
+
+					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,
+										redisChannel,
-										error: `Error subscribing to channel: ${channel}`,
+										error: `Error subscribing to channel: ${redisChannel}`,
 									})
 								);
 								return;
 							});
 					}
 
-					if (!channelSubscribers.get(channel)) {
+					if (!channelSubscribers.get(redisChannel)) {
 						const subscribers = new Set<WebSocket>();
-						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(
+					if (redisChannel.includes('orderbook')) {
-						`last_update_${channel}`
+						const lastMessage = await lastMessageRetriever.client.get(
-					);
+							`last_update_${redisChannel}`
-					if (lastMessage !== null) {
+						);
-						ws.send(JSON.stringify({ channel, data: lastMessage }));
+						if (lastMessage !== null) {
+							ws.send(JSON.stringify({ redisChannel, data: lastMessage }));
+						}
 					}
 					break;
 				}
 				case 'unsubscribe': {
-					const channel = parsedMessage.channel;
+					let redisChannel: string;
-					const subscribers = channelSubscribers.get(channel);
+					try {
+						redisChannel = getRedisChannelFromMessage(parsedMessage);
+					} catch (error) {
+						ws.send(JSON.stringify({ error: error.message }));
+						return;
+					}
+					const subscribers = channelSubscribers.get(redisChannel);
 					if (subscribers) {
-						channelSubscribers.get(channel).delete(ws);
+						channelSubscribers.get(redisChannel).delete(ws);
 					}
 					break;
 				}

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"