migrate pyth pull client to hermes client

package.json

@@ -14,7 +14,7 @@
 		"@opentelemetry/sdk-node": "0.31.0",
 		"@project-serum/anchor": "0.19.1-beta.1",
 		"@project-serum/serum": "0.13.65",
-		"@pythnetwork/price-service-client": "1.9.0",
+		"@pythnetwork/hermes-client": "2.0.0",
 		"@pythnetwork/pyth-lazer-sdk": "0.3.2",
 		"@solana/spl-token": "0.3.7",
 		"@solana/web3.js": "1.92.3",
@@ -83,5 +83,8 @@
 	},
 	"engines": {
 		"node": ">=20.18.0"
+	},
+	"resolutions": {
+		"jito-ts": "4.1.1"
 	}
 }

src/bots/pythCranker.ts

@@ -1,14 +1,11 @@
-import { Bot } from '../types';
+import { Bot, PythPullPriceData } from '../types';
 import { logger } from '../logger';
 import {
 	GlobalConfig,
 	PythCrankerBotConfig,
 	PythUpdateConfigs,
 } from '../config';
-import {
-	PriceFeed,
-	PriceServiceConnection,
-} from '@pythnetwork/price-service-client';
+import { HermesClient, PriceUpdate } from '@pythnetwork/hermes-client';
 import { PriceUpdateAccount } from '@pythnetwork/pyth-solana-receiver/lib/PythSolanaReceiver';
 import {
 	BlockhashSubscriber,
@@ -66,15 +63,16 @@ export type FeedIdToCrankInfo = {
 };
 
 export class PythCrankerBot implements Bot {
-	private priceServiceConnection: PriceServiceConnection;
+	private priceServiceConnection: HermesClient;
+	private pythEventSource: EventSource | undefined;
 	private feedIdsToCrank: FeedIdToCrankInfo[] = [];
 	private pythOracleClient: OracleClient;
 	readonly decodeFunc: (name: string, data: Buffer) => PriceUpdateAccount;
 
 	public name: string;
 	public dryRun: boolean;
 	private intervalMs: number;
-	private feedIdToPriceFeedMap: Map<string, PriceFeed> = new Map();
+	private feedIdToPriceUpdateMap: Map<string, PythPullPriceData> = new Map();
 	public defaultIntervalMs = 30_000;
 
 	private blockhashSubscriber: BlockhashSubscriber;
@@ -96,7 +94,7 @@ export class PythCrankerBot implements Bot {
 		if (!globalConfig.hermesEndpoint) {
 			throw new Error('Missing hermesEndpoint in global config');
 		}
-		this.priceServiceConnection = new PriceServiceConnection(
+		this.priceServiceConnection = new HermesClient(
 			globalConfig.hermesEndpoint,
 			{
 				timeout: 10_000,
@@ -233,12 +231,28 @@ export class PythCrankerBot implements Bot {
 			});
 		}
 
-		await this.priceServiceConnection.subscribePriceFeedUpdates(
-			this.feedIdsToCrank.map((x) => x.feedId),
-			(priceFeed) => {
-				this.feedIdToPriceFeedMap.set(priceFeed.id, priceFeed);
+		this.pythEventSource =
+			await this.priceServiceConnection.getPriceUpdatesStream(
+				this.feedIdsToCrank.map((x) => x.feedId)
+			);
+
+		this.pythEventSource.onmessage = async (event) => {
+			const priceUpdate = JSON.parse(event.data) as PriceUpdate;
+			if (priceUpdate.parsed) {
+				for (const update of priceUpdate.parsed) {
+					this.feedIdToPriceUpdateMap.set(update.id, {
+						expo: update.price.expo,
+						publishTime: update.price.publish_time,
+						price: +update.price.price as number,
+						conf: +update.price.conf,
+					});
+				}
+			} else {
+				logger.warn(
+					`Received invalid price update: ${JSON.stringify(priceUpdate)}`
+				);
 			}
-		);
+		};
 
 		this.priorityFeeSubscriber?.updateAddresses([
 			this.driftClient.getReceiverProgram().programId,
@@ -250,9 +264,7 @@ export class PythCrankerBot implements Bot {
 		this.feedIdsToCrank = [];
 		this.blockhashSubscriber.unsubscribe();
 		await this.driftClient.unsubscribe();
-		await this.priceServiceConnection.unsubscribePriceFeedUpdates(
-			this.feedIdsToCrank.map((x) => x.feedId)
-		);
+		this.pythEventSource?.close();
 	}
 
 	async startIntervalLoop(intervalMs = this.intervalMs): Promise<void> {
@@ -266,17 +278,13 @@ export class PythCrankerBot implements Bot {
 	}
 
 	async getVaaForPriceFeedIds(feedIds: string[]): Promise<string> {
-		const latestVaa = await this.priceServiceConnection.getLatestVaas(feedIds);
-		return latestVaa[0];
-	}
-
-	async getLatestPriceFeedUpdatesForFeedIds(
-		feedIds: string[]
-	): Promise<PriceFeed[] | undefined> {
-		const latestPrices = await this.priceServiceConnection.getLatestPriceFeeds(
-			feedIds
+		const latestVaa = await this.priceServiceConnection.getLatestPriceUpdates(
+			feedIds,
+			{
+				encoding: 'base64',
+			}
 		);
-		return latestPrices;
+		return latestVaa.binary.data[0];
 	}
 
 	private async getBlockhashForTx(): Promise<string> {
@@ -327,16 +335,15 @@ export class PythCrankerBot implements Bot {
 				);
 				return;
 			}
-			const pythnetPriceFeed = this.feedIdToPriceFeedMap.get(
+			const pythnetPriceData = this.feedIdToPriceUpdateMap.get(
 				trimFeedId(feedIdCrankInfo.feedId)
 			);
 
-			if (!pythnetPriceFeed || !onChainPriceFeed) {
+			if (!pythnetPriceData || !onChainPriceFeed) {
 				logger.info(`Missing price feed data for ${feedIdCrankInfo.feedId}`);
 				return;
 			}
 
-			const pythnetPriceData = pythnetPriceFeed.getPriceUnchecked();
 			const onChainPriceData =
 				this.pythOracleClient.getOraclePriceDataFromBuffer(result.data);
 			const onChainSlot = onChainPriceData.slot.toNumber();

src/bots/trigger.ts

@@ -54,7 +54,7 @@ import {
 } from '@solana/web3.js';
 import { PriceUpdateAccount } from '@pythnetwork/pyth-solana-receiver/lib/PythSolanaReceiver';
 import { PythLazerSubscriber } from '../pythLazerSubscriber';
-import { PriceServiceConnection } from '@pythnetwork/price-service-client';
+import { PythPriceFeedSubscriber } from '../pythPriceFeedSubscriber';
 
 const TRIGGER_ORDER_COOLDOWN_MS = 10000; // time to wait between triggering an order
 
@@ -101,15 +101,14 @@ function getPythPullFeedIdsToCrank(
 	const marketIdToFeedId: Map<string, string> = new Map();
 
 	for (const market of [...spotMarkets, ...perpMarkets]) {
-		if (!getVariant(market.oracleSource).toLowerCase().includes('pull')) {
-			continue;
-		}
-		if (market.pythFeedId === undefined) {
-			logger.warn(
-				`No pyth feed id for market ${
-					market.symbol
-				} with oracleSource ${getVariant(market.oracleSource)}`
-			);
+		const oracleSourceStr = getVariant(market.oracleSource).toLowerCase();
+		if (
+			!(
+				oracleSourceStr.includes('pull') &&
+				oracleSourceStr.includes('pyth') &&
+				market.pythFeedId !== undefined
+			)
+		) {
 			continue;
 		}
 
@@ -193,7 +192,7 @@ export class TriggerBot implements Bot {
 
 	private pythLazerClient?: PythLazerSubscriber;
 	private pythPullFeedIdsToCrank: FeedIdToCrankInfo[] = [];
-	private pythPullClient?: PriceServiceConnection;
+	private pythPullClient?: PythPriceFeedSubscriber;
 
 	// map from marketId (i.e. perp-0 or spot-0) to pyth feed id
 	private marketIdToPythPullFeedId: Map<string, string> = new Map();
@@ -269,7 +268,7 @@ export class TriggerBot implements Bot {
 				this.globalConfig.driftEnv
 			);
 
-			this.pythPullClient = new PriceServiceConnection(
+			this.pythPullClient = new PythPriceFeedSubscriber(
 				this.globalConfig.hermesEndpoint,
 				{
 					timeout: 10_000,
@@ -414,12 +413,8 @@ export class TriggerBot implements Bot {
 			this.pythPullClient
 		) {
 			await this.pythLazerClient.subscribe();
-			await this.pythPullClient.subscribePriceFeedUpdates(
-				this.pythPullFeedIdsToCrank.map((x) => x.feedId),
-				(priceFeed) => {
-					const p = priceFeed.getPriceUnchecked().getPriceAsNumberUnchecked();
-					this.pythPullFeedIdToPrice.set('0x' + priceFeed.id, p);
-				}
+			await this.pythPullClient.subscribe(
+				this.pythPullFeedIdsToCrank.map((x) => x.feedId)
 			);
 		}
 	}
@@ -522,7 +517,7 @@ export class TriggerBot implements Bot {
 			if (!feedId) {
 				return [];
 			}
-			const vaa = await this.pythPullClient?.getLatestVaas([feedId]);
+			const vaa = await this.pythPullClient?.getLatestCachedVaa(feedId);
 			if (!vaa) {
 				return [];
 			}

src/experimental-bots/entrypoint.ts

@@ -224,11 +224,7 @@ const runBot = async () => {
 	if (config.global.hermesEndpoint) {
 		pythPriceSubscriber = new PythPriceFeedSubscriber(
 			config.global.hermesEndpoint,
-			{
-				priceFeedRequestConfig: {
-					binary: true,
-				},
-			}
+			{}
 		);
 	}
 

src/experimental-bots/filler-common/dlobBuilder.ts

@@ -283,10 +283,7 @@ class DLOBBuilder {
 			auctionDuration: signedMsgOrderParams.auctionDuration,
 			auctionStartPrice: signedMsgOrderParams.auctionStartPrice,
 			auctionEndPrice: signedMsgOrderParams.auctionEndPrice,
-			immediateOrCancel:
-				(signedMsgOrderParams.bitFlags &
-					OrderParamsBitFlag.ImmediateOrCancel) !==
-				0,
+			immediateOrCancel: false,
 			direction: signedMsgOrderParams.direction,
 			postOnly: false,
 			oraclePriceOffset: signedMsgOrderParams.oraclePriceOffset ?? 0,

src/index.ts

@@ -1,6 +1,7 @@
 /* eslint-disable @typescript-eslint/no-non-null-assertion */
+import 'rpc-websockets/dist/lib/client';
 import { program, Option } from 'commander';
-import * as http from 'http';
+import * as http from 'node:http';
 
 import {
 	Connection,
@@ -455,11 +456,7 @@ const runBot = async () => {
 	if (config.global.hermesEndpoint) {
 		pythPriceSubscriber = new PythPriceFeedSubscriber(
 			config.global.hermesEndpoint,
-			{
-				priceFeedRequestConfig: {
-					binary: true,
-				},
-			}
+			{}
 		);
 	}
 

src/pythPriceFeedSubscriber.ts

@@ -1,23 +1,33 @@
+import { trimFeedId } from '@drift-labs/sdk';
 import {
-	PriceFeed,
-	PriceServiceConnection,
-	PriceServiceConnectionConfig,
-} from '@pythnetwork/price-service-client';
+	HermesClient,
+	HermesClientConfig,
+	PriceUpdate,
+} from '@pythnetwork/hermes-client';
 
-export class PythPriceFeedSubscriber extends PriceServiceConnection {
+export class PythPriceFeedSubscriber extends HermesClient {
 	protected latestPythVaas: Map<string, string> = new Map(); // priceFeedId -> vaa
+	protected streams?: EventSource[];
 
-	constructor(endpoint: string, config: PriceServiceConnectionConfig) {
+	constructor(endpoint: string, config: HermesClientConfig) {
 		super(endpoint, config);
 	}
 
 	async subscribe(feedIds: string[]) {
-		await super.subscribePriceFeedUpdates(feedIds, (priceFeed: PriceFeed) => {
-			if (priceFeed.vaa) {
-				const priceFeedId = '0x' + priceFeed.id;
-				this.latestPythVaas.set(priceFeedId, priceFeed.vaa);
-			}
-		});
+		for (const feedId of feedIds) {
+			const trimmedId = trimFeedId(feedId);
+			const stream = await this.getPriceUpdatesStream([feedId], {
+				encoding: 'base64',
+			});
+			stream.onmessage = async (event) => {
+				const data = JSON.parse(event.data) as PriceUpdate;
+				if (!data.parsed) {
+					console.error('Failed to parse VAA for feedId: ', feedId);
+					return;
+				}
+				this.latestPythVaas.set(trimmedId, data.binary.data[0]);
+			};
+		}
 	}
 
 	getLatestCachedVaa(feedId: string): string | undefined {

src/types.ts

@@ -1,5 +1,5 @@
 import { BN, PositionDirection } from '@drift-labs/sdk';
-import { PriceServiceConnection } from '@pythnetwork/price-service-client';
+import { HermesClient } from '@pythnetwork/hermes-client';
 
 export const constants = {
 	devnet: {
@@ -116,7 +116,7 @@ export interface Bot {
 	readonly name: string;
 	readonly dryRun: boolean;
 	readonly defaultIntervalMs?: number;
-	readonly pythConnection?: PriceServiceConnection;
+	readonly pythConnection?: HermesClient;
 
 	/**
 	 * Initialize the bot
@@ -138,3 +138,10 @@ export interface Bot {
 	 */
 	healthCheck: () => Promise<boolean>;
 }
+
+export type PythPullPriceData = {
+	conf: number;
+	price: number;
+	expo: number;
+	publishTime: number;
+};