WIP

fixup

indexer/packages/redis/src/caches/orderbook-mid-prices-cache.ts

@@ -0,0 +1,88 @@
+import { Callback, RedisClient } from 'redis';
+
+import {
+  addMarketPriceScript,
+  getMarketMedianScript,
+} from './scripts';
+
+// Cache of orderbook prices for each clob pair
+// Each price is cached for a 5 second window and in a ZSET
+export const ORDERBOOK_MID_PRICES_CACHE_KEY_PREFIX: string = 'v4/orderbook_mid_prices/';
+
+function getOrderbookMidPriceCacheKey(ticker: string): string {
+  return `${ORDERBOOK_MID_PRICES_CACHE_KEY_PREFIX}${ticker}`;
+}
+
+// Adds a price to the market prices cache
+export async function setPrice(
+  client: RedisClient,
+  ticker: string,
+  price: string,
+): Promise<void> {
+  // Number of keys for the lua script.
+  const numKeys: number = 1;
+
+  let evalAsync: (
+    marketCacheKey: string,
+  ) => Promise<void> = (marketCacheKey) => {
+
+    return new Promise<void>((resolve, reject) => {
+      const callback: Callback<void> = (
+        err: Error | null,
+      ) => {
+        if (err) {
+          return reject(err);
+        }
+        return resolve();
+      };
+
+      const nowSeconds = Math.floor(Date.now() / 1000); // Current time in seconds
+      client.evalsha(
+        addMarketPriceScript.hash,
+        numKeys,
+        marketCacheKey,
+        price,
+        nowSeconds,
+        callback,
+      );
+
+    });
+  };
+  evalAsync = evalAsync.bind(client);
+
+  return evalAsync(
+    getOrderbookMidPriceCacheKey(ticker),
+  );
+}
+
+export async function getMedianPrice(client: RedisClient, ticker: string): Promise<string | null> {
+  let evalAsync: (
+    marketCacheKey: string,
+  ) => Promise<string> = (
+    marketCacheKey,
+  ) => {
+    return new Promise((resolve, reject) => {
+      const callback: Callback<string> = (
+        err: Error | null,
+        results: string,
+      ) => {
+        if (err) {
+          return reject(err);
+        }
+        return resolve(results);
+      };
+
+      client.evalsha(
+        getMarketMedianScript.hash,
+        1,
+        marketCacheKey,
+        callback,
+      );
+    });
+  };
+  evalAsync = evalAsync.bind(client);
+
+  return evalAsync(
+    getOrderbookMidPriceCacheKey(ticker),
+  );
+}

indexer/packages/redis/src/caches/scripts.ts

@@ -63,6 +63,8 @@ export const removeOrderScript: LuaScript = newLuaScript('removeOrder', '../scri
 export const addCanceledOrderIdScript: LuaScript = newLuaScript('addCanceledOrderId', '../scripts/add_canceled_order_id.lua');
 export const addStatefulOrderUpdateScript: LuaScript = newLuaScript('addStatefulOrderUpdate', '../scripts/add_stateful_order_update.lua');
 export const removeStatefulOrderUpdateScript: LuaScript = newLuaScript('removeStatefulOrderUpdate', '../scripts/remove_stateful_order_update.lua');
+export const addMarketPriceScript: LuaScript = newLuaScript('addMarketPrice', '../scripts/add_market_price.lua');
+export const getMarketMedianScript: LuaScript = newLuaScript('getMarketMedianPrice', '../scripts/get_market_median_price.lua');
 
 export const allLuaScripts: LuaScript[] = [
   deleteZeroPriceLevelScript,
@@ -75,4 +77,6 @@ export const allLuaScripts: LuaScript[] = [
   addCanceledOrderIdScript,
   addStatefulOrderUpdateScript,
   removeStatefulOrderUpdateScript,
+  addMarketPriceScript,
+  getMarketMedianScript,
 ];

indexer/packages/redis/src/index.ts

@@ -12,6 +12,7 @@ export * as CanceledOrdersCache from './caches/canceled-orders-cache';
 export * as StatefulOrderUpdatesCache from './caches/stateful-order-updates-cache';
 export * as StateFilledQuantumsCache from './caches/state-filled-quantums-cache';
 export * as LeaderboardPnlProcessedCache from './caches/leaderboard-processed-cache';
+export * as OrderbookMidPricesCache from './caches/orderbook-mid-prices-cache';
 export { placeOrder } from './caches/place-order';
 export { removeOrder } from './caches/remove-order';
 export { updateOrder } from './caches/update-order';

indexer/packages/redis/src/scripts/add_market_price.lua

@@ -0,0 +1,17 @@
+-- Key for the ZSET storing price data
+local priceCacheKey = KEYS[1]
+-- Price to be added
+local price = tonumber(ARGV[1])
+-- Current timestamp
+local nowSeconds = tonumber(ARGV[2])
+-- Time window (5 seconds)
+local fiveSeconds = 5
+
+-- 1. Add the price to the sorted set (score is the current timestamp)
+redis.call("zadd", priceCacheKey, nowSeconds, price)
+
+-- 2. Remove any entries older than 5 seconds
+local cutoffTime = nowSeconds - fiveSeconds
+redis.call("zremrangebyscore", priceCacheKey, "-inf", cutoffTime)
+
+return true

indexer/packages/redis/src/scripts/get_market_median_price.lua

@@ -0,0 +1,23 @@
+-- Key for the sorted set storing price data
+local priceCacheKey = KEYS[1]
+
+-- Get all the prices from the sorted set (ascending order)
+local prices = redis.call('zrange', priceCacheKey, 0, -1)
+
+-- If no prices are found, return nil
+if #prices == 0 then
+  return nil
+end
+
+-- Calculate the middle index
+local middle = math.floor(#prices / 2)
+
+-- Calculate median
+if #prices % 2 == 0 then
+  -- If even, return the average of the two middle elements
+  local median = (tonumber(prices[middle]) + tonumber(prices[middle + 1])) / 2
+  return tostring(median)
+else
+  -- If odd, return the middle element
+  return prices[middle + 1]
+end

indexer/services/ender/src/lib/candles-generator.ts

@@ -20,7 +20,7 @@ import {
   TradeMessageContents,
   helpers,
 } from '@dydxprotocol-indexer/postgres';
-import { OrderbookLevelsCache } from '@dydxprotocol-indexer/redis';
+import { OrderbookMidPricesCache } from '@dydxprotocol-indexer/redis';
 import { CandleMessage } from '@dydxprotocol-indexer/v4-protos';
 import Big from 'big.js';
 import _ from 'lodash';
@@ -538,9 +538,9 @@ export async function getOrderbookMidPriceMap(): Promise<{ [ticker: string]: Ord
   const perpetualMarkets = Object.values(perpetualMarketRefresher.getPerpetualMarketsMap());
 
   const promises = perpetualMarkets.map(async (perpetualMarket: PerpetualMarketFromDatabase) => {
-    const price = await OrderbookLevelsCache.getOrderBookMidPrice(
-      perpetualMarket.ticker,
+    const price = await OrderbookMidPricesCache.getMedianPrice(
       redisClient,
+      perpetualMarket.ticker,
     );
     return { [perpetualMarket.ticker]: price === undefined ? undefined : price };
   });

indexer/services/roundtable/src/config.ts

@@ -59,6 +59,7 @@ export const configSchema = {
   LOOPS_ENABLED_LEADERBOARD_PNL_YEARLY: parseBoolean({ default: false }),
   LOOPS_ENABLED_UPDATE_WALLET_TOTAL_VOLUME: parseBoolean({ default: true }),
   LOOPS_ENABLED_DELETE_OLD_FIREBASE_NOTIFICATION_TOKENS: parseBoolean({ default: true }),
+  LOOPS_ENABLED_CACHE_ORDERBOOK_MID_PRICES: parseBoolean({ default: true }),
 
   // Loop Timing
   LOOPS_INTERVAL_MS_MARKET_UPDATER: parseInteger({
@@ -133,6 +134,9 @@ export const configSchema = {
   LOOPS_INTERVAL_MS_DELETE_FIREBASE_NOTIFICATION_TOKENS_MONTHLY: parseInteger({
     default: 30 * ONE_DAY_IN_MILLISECONDS,
   }),
+  LOOPS_INTERVAL_MS_CACHE_ORDERBOOK_MID_PRICES: parseInteger({
+    default: ONE_SECOND_IN_MILLISECONDS,
+  }),
 
   // Start delay
   START_DELAY_ENABLED: parseBoolean({ default: true }),

indexer/services/roundtable/src/index.ts

@@ -10,6 +10,7 @@ import {
   connect as connectToRedis,
 } from './helpers/redis';
 import aggregateTradingRewardsTasks from './tasks/aggregate-trading-rewards';
+import cacheOrderbookMidPrices from './tasks/cache-orderbook-mid-prices';
 import cancelStaleOrdersTask from './tasks/cancel-stale-orders';
 import createLeaderboardTask from './tasks/create-leaderboard';
 import createPnlTicksTask from './tasks/create-pnl-ticks';
@@ -265,6 +266,14 @@ async function start(): Promise<void> {
     );
   }
 
+  if (config.LOOPS_ENABLED_CACHE_ORDERBOOK_MID_PRICES) {
+    startLoop(
+      cacheOrderbookMidPrices,
+      'cache_orderbook_mid_prices',
+      config.LOOPS_INTERVAL_MS_DELETE_FIREBASE_NOTIFICATION_TOKENS_MONTHLY,
+    );
+  }
+
   logger.info({
     at: 'index',
     message: 'Successfully started',

indexer/services/roundtable/src/tasks/cache-orderbook-mid-prices.ts

@@ -0,0 +1,40 @@
+import {
+  logger,
+} from '@dydxprotocol-indexer/base';
+import {
+  PerpetualMarketFromDatabase,
+  PerpetualMarketTable,
+} from '@dydxprotocol-indexer/postgres';
+import {
+  OrderbookMidPricesCache,
+  OrderbookLevelsCache,
+} from '@dydxprotocol-indexer/redis';
+
+import { redisClient } from '../helpers/redis';
+
+/**
+ * Instrument data on the orderbook to be used for analytics.
+ */
+export default async function runTask(): Promise<void> {
+  const markets: PerpetualMarketFromDatabase[] = await PerpetualMarketTable.findAll({}, []);
+
+  for (const market of markets) {
+    try {
+      const price = await OrderbookLevelsCache.getOrderBookMidPrice(market.ticker, redisClient);
+      if (price) {
+        await OrderbookMidPricesCache.setPrice(redisClient, market.ticker, price);
+      } else {
+        logger.info({
+          at: 'cache-orderbook-mid-prices#runTask',
+          message: `undefined price for ${market.ticker}`,
+        });
+      }
+    } catch (error) {
+      logger.error({
+        at: 'cache-orderbook-mid-prices#runTask',
+        message: error.msg,
+        error,
+      });
+    }
+  }
+}