From 4d39b39f759e13a4f97edb9ad2e317030fd740be Mon Sep 17 00:00:00 2001 From: Adam Fraser Date: Tue, 6 Aug 2024 16:11:51 -0400 Subject: [PATCH] Add orderbook data to candles table Revert "Revert "Candles HLOC Data (#1887)"" This reverts commit 79345d69f6e8779ad41581ca46f5ed31d2175461. Revert "Revert "Adam/ct 1059 address issue with orderbookmidprice calculation (#2012)"" This reverts commit d10317a30b65662dda90f8517fa16ec43d0b088e. Explicity create nullable columns --- .../postgres/__tests__/helpers/constants.ts | 2 + .../__tests__/stores/candle-table.test.ts | 7 +- ...ndles_add_mid_book_price_open_and_close.ts | 19 ++ .../postgres/src/models/candle-model.ts | 6 + .../postgres/src/types/candle-types.ts | 4 + .../postgres/src/types/db-model-types.ts | 2 + .../caches/orderbook-levels-cache.test.ts | 133 +++++++++ .../src/caches/orderbook-levels-cache.ts | 24 ++ .../comlink/public/api-documentation.md | 8 + indexer/services/comlink/public/swagger.json | 8 + .../ender/__tests__/helpers/redis-helpers.ts | 18 ++ .../__tests__/lib/candles-generator.test.ts | 276 +++++++++++++++++- .../ender/src/lib/candles-generator.ts | 111 ++++++- .../src/lib/athena-ddl-tables/candles.ts | 4 + 14 files changed, 598 insertions(+), 24 deletions(-) create mode 100644 indexer/packages/postgres/src/db/migrations/migration_files/20240806152937_candles_add_mid_book_price_open_and_close.ts diff --git a/indexer/packages/postgres/__tests__/helpers/constants.ts b/indexer/packages/postgres/__tests__/helpers/constants.ts index 28e4524a4c3..fef71972968 100644 --- a/indexer/packages/postgres/__tests__/helpers/constants.ts +++ b/indexer/packages/postgres/__tests__/helpers/constants.ts @@ -725,6 +725,8 @@ export const defaultCandle: CandleCreateObject = { usdVolume: '2200000', trades: 300, startingOpenInterest: '200000', + orderbookMidPriceOpen: '11500', + orderbookMidPriceClose: '12500', }; export const defaultCandleId: string = CandleTable.uuid( diff --git a/indexer/packages/postgres/__tests__/stores/candle-table.test.ts b/indexer/packages/postgres/__tests__/stores/candle-table.test.ts index 9f6f31ce46c..aab085081e1 100644 --- a/indexer/packages/postgres/__tests__/stores/candle-table.test.ts +++ b/indexer/packages/postgres/__tests__/stores/candle-table.test.ts @@ -65,12 +65,11 @@ describe('CandleTable', () => { const updatedCandle: CandleUpdateObject = { id: defaultCandleId, open: '100', + orderbookMidPriceClose: '200', + orderbookMidPriceOpen: '300', }; - await CandleTable.update({ - id: defaultCandleId, - open: '100', - }); + await CandleTable.update(updatedCandle); const candle: CandleFromDatabase | undefined = await CandleTable.findById( defaultCandleId, diff --git a/indexer/packages/postgres/src/db/migrations/migration_files/20240806152937_candles_add_mid_book_price_open_and_close.ts b/indexer/packages/postgres/src/db/migrations/migration_files/20240806152937_candles_add_mid_book_price_open_and_close.ts new file mode 100644 index 00000000000..c97c972d3fc --- /dev/null +++ b/indexer/packages/postgres/src/db/migrations/migration_files/20240806152937_candles_add_mid_book_price_open_and_close.ts @@ -0,0 +1,19 @@ +import * as Knex from 'knex'; + +export async function up(knex: Knex): Promise { + return knex + .schema + .alterTable('candles', (table) => { + table.decimal('orderbookMidPriceOpen', null).nullable(); + table.decimal('orderbookMidPriceClose', null).nullable(); + }); +} + +export async function down(knex: Knex): Promise { + return knex + .schema + .alterTable('candles', (table) => { + table.dropColumn('orderbookMidPriceOpen'); + table.dropColumn('orderbookMidPriceClose'); + }); +} diff --git a/indexer/packages/postgres/src/models/candle-model.ts b/indexer/packages/postgres/src/models/candle-model.ts index ec256318be1..a17cd2f4e89 100644 --- a/indexer/packages/postgres/src/models/candle-model.ts +++ b/indexer/packages/postgres/src/models/candle-model.ts @@ -50,6 +50,8 @@ export default class CandleModel extends Model { usdVolume: { type: 'string', pattern: NonNegativeNumericPattern }, trades: { type: 'integer' }, startingOpenInterest: { type: 'string', pattern: NonNegativeNumericPattern }, + orderbookMidPriceOpen: { type: ['string', 'null'], pattern: NonNegativeNumericPattern }, + orderbookMidPriceClose: { type: ['string', 'null'], pattern: NonNegativeNumericPattern }, }, }; } @@ -77,4 +79,8 @@ export default class CandleModel extends Model { trades!: number; startingOpenInterest!: string; + + orderbookMidPriceOpen?: string; + + orderbookMidPriceClose?: string; } diff --git a/indexer/packages/postgres/src/types/candle-types.ts b/indexer/packages/postgres/src/types/candle-types.ts index 87f30dbf320..a54a2a12857 100644 --- a/indexer/packages/postgres/src/types/candle-types.ts +++ b/indexer/packages/postgres/src/types/candle-types.ts @@ -12,6 +12,8 @@ export interface CandleCreateObject { usdVolume: string; trades: number; startingOpenInterest: string; + orderbookMidPriceOpen: string | undefined; + orderbookMidPriceClose: string | undefined; } export interface CandleUpdateObject { @@ -24,6 +26,8 @@ export interface CandleUpdateObject { usdVolume?: string; trades?: number; startingOpenInterest?: string; + orderbookMidPriceOpen?: string; + orderbookMidPriceClose?: string; } export enum CandleResolution { diff --git a/indexer/packages/postgres/src/types/db-model-types.ts b/indexer/packages/postgres/src/types/db-model-types.ts index a6f4e1192e2..04365785146 100644 --- a/indexer/packages/postgres/src/types/db-model-types.ts +++ b/indexer/packages/postgres/src/types/db-model-types.ts @@ -198,6 +198,8 @@ export interface CandleFromDatabase extends IdBasedModelFromDatabase { usdVolume: string; trades: number; startingOpenInterest: string; + orderbookMidPriceOpen?: string | null; + orderbookMidPriceClose?: string | null; } export interface PnlTicksFromDatabase extends IdBasedModelFromDatabase { diff --git a/indexer/packages/redis/__tests__/caches/orderbook-levels-cache.test.ts b/indexer/packages/redis/__tests__/caches/orderbook-levels-cache.test.ts index af077cf1715..603e332865f 100644 --- a/indexer/packages/redis/__tests__/caches/orderbook-levels-cache.test.ts +++ b/indexer/packages/redis/__tests__/caches/orderbook-levels-cache.test.ts @@ -12,6 +12,7 @@ import { deleteZeroPriceLevel, getLastUpdatedKey, deleteStalePriceLevel, + getOrderBookMidPrice, } from '../../src/caches/orderbook-levels-cache'; import { OrderSide } from '@dydxprotocol-indexer/postgres'; import { OrderbookLevels, PriceLevel } from '../../src/types'; @@ -685,4 +686,136 @@ describe('orderbookLevelsCache', () => { expect(size).toEqual('10'); }); }); + + describe('getMidPrice', () => { + beforeEach(() => { + jest.restoreAllMocks(); + jest.restoreAllMocks(); + }); + afterEach(() => { + jest.restoreAllMocks(); + jest.restoreAllMocks(); + }); + + it('returns the correct mid price', async () => { + await Promise.all([ + updatePriceLevel({ + ticker, + side: OrderSide.BUY, + humanPrice: '45200', + sizeDeltaInQuantums: '2000', + client, + }), + updatePriceLevel({ + ticker, + side: OrderSide.BUY, + humanPrice: '45100', + sizeDeltaInQuantums: '2000', + client, + }), + updatePriceLevel({ + ticker, + side: OrderSide.BUY, + humanPrice: '45300', + sizeDeltaInQuantums: '2000', + client, + }), + updatePriceLevel({ + ticker, + side: OrderSide.SELL, + humanPrice: '45500', + sizeDeltaInQuantums: '2000', + client, + }), + updatePriceLevel({ + ticker, + side: OrderSide.SELL, + humanPrice: '45400', + sizeDeltaInQuantums: '2000', + client, + }), + updatePriceLevel({ + ticker, + side: OrderSide.SELL, + humanPrice: '45600', + sizeDeltaInQuantums: '2000', + client, + }), + ]); + + const midPrice = await getOrderBookMidPrice(ticker, client); + expect(midPrice).toEqual('45350'); + }); + }); + + it('returns the correct mid price for very small numbers', async () => { + await Promise.all([ + updatePriceLevel({ + ticker, + side: OrderSide.SELL, + humanPrice: '0.000000002346', + sizeDeltaInQuantums: '2000', + client, + }), + updatePriceLevel({ + ticker, + side: OrderSide.BUY, + humanPrice: '0.000000002344', + sizeDeltaInQuantums: '2000', + client, + }), + ]); + + const midPrice = await getOrderBookMidPrice(ticker, client); + expect(midPrice).toEqual('0.000000002345'); + }); + + it('returns the approprite amount of decimal precision', async () => { + await Promise.all([ + updatePriceLevel({ + ticker, + side: OrderSide.SELL, + humanPrice: '1.02', + sizeDeltaInQuantums: '2000', + client, + }), + updatePriceLevel({ + ticker, + side: OrderSide.BUY, + humanPrice: '1.01', + sizeDeltaInQuantums: '2000', + client, + }), + ]); + + const midPrice = await getOrderBookMidPrice(ticker, client); + expect(midPrice).toEqual('1.015'); + }); + + it('returns undefined if there are no bids or asks', async () => { + await updatePriceLevel({ + ticker, + side: OrderSide.SELL, + humanPrice: '45400', + sizeDeltaInQuantums: '2000', + client, + }); + + const midPrice = await getOrderBookMidPrice(ticker, client); + expect(midPrice).toBeUndefined(); + }); + + it('returns undefined if humanPrice is NaN', async () => { + await updatePriceLevel({ + ticker, + side: OrderSide.SELL, + humanPrice: 'nan', + sizeDeltaInQuantums: '2000', + client, + }); + + const midPrice = await getOrderBookMidPrice(ticker, client); + + expect(midPrice).toBeUndefined(); + }); }); diff --git a/indexer/packages/redis/src/caches/orderbook-levels-cache.ts b/indexer/packages/redis/src/caches/orderbook-levels-cache.ts index f094d80d7b4..cb75fda5845 100644 --- a/indexer/packages/redis/src/caches/orderbook-levels-cache.ts +++ b/indexer/packages/redis/src/caches/orderbook-levels-cache.ts @@ -529,3 +529,27 @@ function convertToPriceLevels( }; }); } + +export async function getOrderBookMidPrice( + ticker: string, + client: RedisClient, +): Promise { + const levels = await getOrderBookLevels(ticker, client, { + removeZeros: true, + sortSides: true, + uncrossBook: true, + limitPerSide: 1, + }); + + if (levels.bids.length === 0 || levels.asks.length === 0) { + return undefined; + } + + const bestAsk = Big(levels.asks[0].humanPrice); + const bestBid = Big(levels.bids[0].humanPrice); + + if (bestAsk === undefined || bestBid === undefined) { + return undefined; + } + return bestBid.plus(bestAsk).div(2).toFixed(); +} diff --git a/indexer/services/comlink/public/api-documentation.md b/indexer/services/comlink/public/api-documentation.md index 9134a47e869..83001d797f8 100644 --- a/indexer/services/comlink/public/api-documentation.md +++ b/indexer/services/comlink/public/api-documentation.md @@ -692,6 +692,8 @@ fetch(`${baseURL}/candles/perpetualMarkets/{ticker}?resolution=1MIN`, "usdVolume": "string", "trades": 0.1, "startingOpenInterest": "string", + "orderbookMidPriceOpen": "string", + "orderbookMidPriceClose": "string", "id": "string" } ] @@ -3655,6 +3657,8 @@ This operation does not require authentication "usdVolume": "string", "trades": 0.1, "startingOpenInterest": "string", + "orderbookMidPriceOpen": "string", + "orderbookMidPriceClose": "string", "id": "string" } @@ -3675,6 +3679,8 @@ This operation does not require authentication |usdVolume|string|true|none|none| |trades|number(double)|true|none|none| |startingOpenInterest|string|true|none|none| +|orderbookMidPriceOpen|string¦null|false|none|none| +|orderbookMidPriceClose|string¦null|false|none|none| |id|string|true|none|none| ## CandleResponse @@ -3699,6 +3705,8 @@ This operation does not require authentication "usdVolume": "string", "trades": 0.1, "startingOpenInterest": "string", + "orderbookMidPriceOpen": "string", + "orderbookMidPriceClose": "string", "id": "string" } ] diff --git a/indexer/services/comlink/public/swagger.json b/indexer/services/comlink/public/swagger.json index e55c2d4f751..2974e402d24 100644 --- a/indexer/services/comlink/public/swagger.json +++ b/indexer/services/comlink/public/swagger.json @@ -303,6 +303,14 @@ "startingOpenInterest": { "type": "string" }, + "orderbookMidPriceOpen": { + "type": "string", + "nullable": true + }, + "orderbookMidPriceClose": { + "type": "string", + "nullable": true + }, "id": { "type": "string" } diff --git a/indexer/services/ender/__tests__/helpers/redis-helpers.ts b/indexer/services/ender/__tests__/helpers/redis-helpers.ts index 3af8f0c1f0a..7f2ec30800a 100644 --- a/indexer/services/ender/__tests__/helpers/redis-helpers.ts +++ b/indexer/services/ender/__tests__/helpers/redis-helpers.ts @@ -1,5 +1,7 @@ +import { OrderSide } from '@dydxprotocol-indexer/postgres'; import { NextFundingCache, + OrderbookLevelsCache, StateFilledQuantumsCache, } from '@dydxprotocol-indexer/redis'; import Big from 'big.js'; @@ -29,3 +31,19 @@ export async function expectStateFilledQuantums( expect(stateFilledQuantums).toBeDefined(); expect(stateFilledQuantums).toEqual(quantums); } + +export async function updatePriceLevel( + ticker: string, + price: string, + side: OrderSide, +): Promise { + const quantums: string = '30'; + + await OrderbookLevelsCache.updatePriceLevel({ + ticker, + side, + humanPrice: price, + sizeDeltaInQuantums: quantums, + client: redisClient, + }); +} diff --git a/indexer/services/ender/__tests__/lib/candles-generator.test.ts b/indexer/services/ender/__tests__/lib/candles-generator.test.ts index 6723dc4d56d..d1c257b80b6 100644 --- a/indexer/services/ender/__tests__/lib/candles-generator.test.ts +++ b/indexer/services/ender/__tests__/lib/candles-generator.test.ts @@ -18,6 +18,7 @@ import { testMocks, Transaction, helpers, + OrderSide, } from '@dydxprotocol-indexer/postgres'; import { CandleMessage, CandleMessage_Resolution } from '@dydxprotocol-indexer/v4-protos'; import Big from 'big.js'; @@ -26,11 +27,14 @@ import { clearCandlesMap, getCandlesMap, startCandleCache, } from '../../src/caches/candle-cache'; import config from '../../src/config'; -import { CandlesGenerator } from '../../src/lib/candles-generator'; +import { CandlesGenerator, getOrderbookMidPriceMap } from '../../src/lib/candles-generator'; import { KafkaPublisher } from '../../src/lib/kafka-publisher'; import { ConsolidatedKafkaEvent } from '../../src/lib/types'; import { defaultTradeContent, defaultTradeKafkaEvent } from '../helpers/constants'; import { contentToSingleTradeMessage, createConsolidatedKafkaEventFromTrade } from '../helpers/kafka-publisher-helpers'; +import { updatePriceLevel } from '../helpers/redis-helpers'; +import { redisClient } from '../../src/helpers/redis/redis-controller'; +import { redis } from '@dydxprotocol-indexer/redis'; describe('candleHelper', () => { beforeAll(async () => { @@ -48,6 +52,7 @@ describe('candleHelper', () => { await dbHelpers.clearData(); clearCandlesMap(); jest.clearAllMocks(); + await redis.deleteAllAsync(redisClient); }); afterAll(async () => { @@ -71,6 +76,8 @@ describe('candleHelper', () => { ).toString(), trades: 2, startingOpenInterest: '0', + orderbookMidPriceClose: undefined, + orderbookMidPriceOpen: undefined, }; const startedAt: IsoString = helpers.calculateNormalizedCandleStartTime( testConstants.createdDateTime, @@ -106,6 +113,10 @@ describe('candleHelper', () => { defaultTradeKafkaEvent2, ]); + // Create Orderbook levels to set orderbookMidPrice open & close + await updatePriceLevel('BTC-USD', '100000', OrderSide.BUY); + await updatePriceLevel('BTC-USD', '110000', OrderSide.SELL); + await runUpdateCandles(publisher); // Verify postgres is updated @@ -122,6 +133,8 @@ describe('candleHelper', () => { id: CandleTable.uuid(currentStartedAt, defaultCandle.ticker, resolution), startedAt: currentStartedAt, resolution, + orderbookMidPriceClose: '105000', + orderbookMidPriceOpen: '105000', }; }, ); @@ -142,6 +155,9 @@ describe('candleHelper', () => { defaultTradeKafkaEvent2, ]); + await updatePriceLevel('BTC-USD', '80000', OrderSide.BUY); + await updatePriceLevel('BTC-USD', '81000', OrderSide.SELL); + // Create Perpetual Position to set open position const openInterest: string = '100'; await createOpenPosition(openInterest); @@ -163,6 +179,8 @@ describe('candleHelper', () => { startedAt: currentStartedAt, resolution, startingOpenInterest: openInterest, + orderbookMidPriceClose: '80500', + orderbookMidPriceOpen: '80500', }; }, ); @@ -180,6 +198,8 @@ describe('candleHelper', () => { const startingOpenInterest: string = '200'; const baseTokenVolume: string = '10'; const usdVolume: string = Big(existingPrice).times(baseTokenVolume).toString(); + const orderbookMidPriceClose = '7500'; + const orderbookMidPriceOpen = '8000'; await Promise.all( _.map(Object.values(CandleResolution), (resolution: CandleResolution) => { const currentStartedAt: IsoString = helpers.calculateNormalizedCandleStartTime( @@ -199,6 +219,8 @@ describe('candleHelper', () => { usdVolume, trades: existingTrades, startingOpenInterest, + orderbookMidPriceClose, + orderbookMidPriceOpen, }); }), ); @@ -234,6 +256,8 @@ describe('candleHelper', () => { usdVolume: Big(defaultCandle.usdVolume).plus(usdVolume).toString(), trades: existingTrades + 2, startingOpenInterest, + orderbookMidPriceClose, + orderbookMidPriceOpen, }; }, ); @@ -248,7 +272,7 @@ describe('candleHelper', () => { it.each([ [ - 'creates empty candles', // description + 'creates empty candle', // description { // initial candle startedAt: previousStartedAt, ticker: testConstants.defaultPerpetualMarket.ticker, @@ -261,6 +285,8 @@ describe('candleHelper', () => { usdVolume: '10000', trades: existingTrades, startingOpenInterest: existingStartingOpenInterest, + orderbookMidPriceClose: undefined, + orderbookMidPriceOpen: undefined, }, '100', // open interest false, // block contains trades @@ -277,7 +303,11 @@ describe('candleHelper', () => { usdVolume: '0', trades: 0, startingOpenInterest: '100', + orderbookMidPriceClose: '1000', + orderbookMidPriceOpen: '1000', }, + true, + 1000, ], [ 'creates new candle if existing candle is from a past normalized candle start time', // description @@ -293,6 +323,8 @@ describe('candleHelper', () => { usdVolume: '10000', trades: existingTrades, startingOpenInterest: existingStartingOpenInterest, + orderbookMidPriceClose: '3000', + orderbookMidPriceOpen: '3500', }, '100', // open interest true, // block contains trades @@ -302,7 +334,11 @@ describe('candleHelper', () => { startedAt, resolution: CandleResolution.ONE_MINUTE, startingOpenInterest: '100', + orderbookMidPriceClose: '1000', + orderbookMidPriceOpen: '1000', }, + true, // contains kafka messages + 1000, // orderbook mid price ], [ 'updates empty candle', // description @@ -318,6 +354,8 @@ describe('candleHelper', () => { usdVolume: '0', trades: 0, startingOpenInterest: existingStartingOpenInterest, + orderbookMidPriceClose: undefined, + orderbookMidPriceOpen: undefined, }, '100', // open interest true, // block contains trades @@ -327,14 +365,20 @@ describe('candleHelper', () => { startedAt, resolution: CandleResolution.ONE_MINUTE, startingOpenInterest: existingStartingOpenInterest, + orderbookMidPriceClose: null, + orderbookMidPriceOpen: null, }, + true, // contains kafka messages + 1000, // orderbook mid price ], [ 'does nothing when there are no trades and no existing candle', // description - undefined, + undefined, // initial candle '100', // open interest false, // block contains trades - undefined, + undefined, // expected candle + true, // contains kafka messages + 1000, // orderbook mid price ], [ 'does not update candle when there are no trades and an existing candle', // description @@ -350,6 +394,8 @@ describe('candleHelper', () => { usdVolume: '10000', trades: existingTrades, startingOpenInterest: existingStartingOpenInterest, + orderbookMidPriceClose: '5000', + orderbookMidPriceOpen: '6000', }, '100', // open interest false, // block contains trades @@ -366,8 +412,11 @@ describe('candleHelper', () => { usdVolume: '10000', trades: existingTrades, startingOpenInterest: existingStartingOpenInterest, + orderbookMidPriceClose: '5000', + orderbookMidPriceOpen: '6000', }, false, // contains kafka messages + 1000, ], ])('Successfully %s', async ( _description: string, @@ -376,7 +425,12 @@ describe('candleHelper', () => { blockContainsTrades: boolean, expectedCandle: CandleFromDatabase | undefined, containsKafkaMessages: boolean = true, + orderbookMidPrice: number, ) => { + const midPriceSpread = 10; + await updatePriceLevel('BTC-USD', String(orderbookMidPrice + midPriceSpread), OrderSide.SELL); + await updatePriceLevel('BTC-USD', String(orderbookMidPrice - midPriceSpread), OrderSide.BUY); + if (initialCandle !== undefined) { await CandleTable.create(initialCandle); } @@ -410,6 +464,220 @@ describe('candleHelper', () => { await validateCandlesCache(); expectTimingStats(); }); + + it('Updates previous candle orderBookMidPriceClose if startTime is past candle resolution', async () => { + // Create existing candles + const existingPrice: string = '7000'; + const startingOpenInterest: string = '200'; + const baseTokenVolume: string = '10'; + const usdVolume: string = Big(existingPrice).times(baseTokenVolume).toString(); + const orderbookMidPriceClose = '7500'; + const orderbookMidPriceOpen = '8000'; + await Promise.all( + _.map(Object.values(CandleResolution), (resolution: CandleResolution) => { + return CandleTable.create({ + startedAt: previousStartedAt, + ticker: testConstants.defaultPerpetualMarket.ticker, + resolution, + low: existingPrice, + high: existingPrice, + open: existingPrice, + close: existingPrice, + baseTokenVolume, + usdVolume, + trades: existingTrades, + startingOpenInterest, + orderbookMidPriceClose, + orderbookMidPriceOpen, + }); + }), + ); + await startCandleCache(); + + // Update Orderbook levels + await updatePriceLevel('BTC-USD', '10010', OrderSide.SELL); + await updatePriceLevel('BTC-USD', '10000', OrderSide.BUY); + + const publisher: KafkaPublisher = new KafkaPublisher(); + publisher.addEvents([ + defaultTradeKafkaEvent, + defaultTradeKafkaEvent2, + ]); + + // Create new candles, with trades + await runUpdateCandles(publisher).then(async () => { + + // Verify previous candles have orderbookMidPriceClose updated + const previousExpectedCandles: CandleFromDatabase[] = _.map( + Object.values(CandleResolution), + (resolution: CandleResolution) => { + return { + id: CandleTable.uuid(previousStartedAt, defaultCandle.ticker, resolution), + startedAt: previousStartedAt, + ticker: defaultCandle.ticker, + resolution, + low: existingPrice, + high: existingPrice, + open: existingPrice, + close: existingPrice, + baseTokenVolume, + usdVolume, + trades: existingTrades, + startingOpenInterest, + orderbookMidPriceClose: '10005', + orderbookMidPriceOpen, + }; + }, + ); + await verifyCandlesInPostgres(previousExpectedCandles); + }); + + // Verify new candles were created + const expectedCandles: CandleFromDatabase[] = _.map( + Object.values(CandleResolution), + (resolution: CandleResolution) => { + const currentStartedAt: IsoString = helpers.calculateNormalizedCandleStartTime( + testConstants.createdDateTime, + resolution, + ).toISO(); + + return { + id: CandleTable.uuid(currentStartedAt, defaultCandle.ticker, resolution), + startedAt: currentStartedAt, + ticker: defaultCandle.ticker, + resolution, + low: '10000', + high: defaultPrice2, + open: '10000', + close: defaultPrice2, + baseTokenVolume: '20', + usdVolume: '250000', + trades: 2, + startingOpenInterest: '0', + orderbookMidPriceClose: '10005', + orderbookMidPriceOpen: '10005', + }; + }, + ); + await verifyCandlesInPostgres(expectedCandles); + await validateCandlesCache(); + expectTimingStats(); + }); + + it('creates an empty candle and updates the previous candle orderBookMidPriceClose if startTime is past candle resolution', async () => { + // Create existing candles + const existingPrice: string = '7000'; + const startingOpenInterest: string = '200'; + const baseTokenVolume: string = '10'; + const usdVolume: string = Big(existingPrice).times(baseTokenVolume).toString(); + const orderbookMidPriceClose = '7500'; + const orderbookMidPriceOpen = '8000'; + + await Promise.all( + _.map(Object.values(CandleResolution), (resolution: CandleResolution) => { + return CandleTable.create({ + startedAt: previousStartedAt, + ticker: testConstants.defaultPerpetualMarket.ticker, + resolution, + low: existingPrice, + high: existingPrice, + open: existingPrice, + close: existingPrice, + baseTokenVolume, + usdVolume, + trades: existingTrades, + startingOpenInterest, + orderbookMidPriceClose, + orderbookMidPriceOpen, + }); + }), + ); + await startCandleCache(); + + // Update Orderbook levels + await updatePriceLevel('BTC-USD', '10010', OrderSide.SELL); + await updatePriceLevel('BTC-USD', '10000', OrderSide.BUY); + + const publisher: KafkaPublisher = new KafkaPublisher(); + publisher.addEvents([]); + + // Create new candles, without trades + await runUpdateCandles(publisher); + + // Verify previous candles have orderbookMidPriceClose updated + const previousExpectedCandles: CandleFromDatabase[] = _.map( + Object.values(CandleResolution), + (resolution: CandleResolution) => { + return { + id: CandleTable.uuid(previousStartedAt, defaultCandle.ticker, resolution), + startedAt: previousStartedAt, + ticker: defaultCandle.ticker, + resolution, + low: existingPrice, + high: existingPrice, + open: existingPrice, + close: existingPrice, + baseTokenVolume, + usdVolume, + trades: existingTrades, + startingOpenInterest, + orderbookMidPriceClose: '10005', + orderbookMidPriceOpen, + }; + }, + ); + await verifyCandlesInPostgres(previousExpectedCandles); + + // Verify new empty candle was created + const expectedCandles: CandleFromDatabase[] = _.map( + Object.values(CandleResolution), + (resolution: CandleResolution) => { + const currentStartedAt: IsoString = helpers.calculateNormalizedCandleStartTime( + testConstants.createdDateTime, + resolution, + ).toISO(); + + return { + id: CandleTable.uuid(currentStartedAt, defaultCandle.ticker, resolution), + startedAt: currentStartedAt, + ticker: defaultCandle.ticker, + resolution, + low: existingPrice, + high: existingPrice, + open: existingPrice, + close: existingPrice, + baseTokenVolume: '0', + usdVolume: '0', + trades: 0, + startingOpenInterest: '0', + orderbookMidPriceClose: '10005', + orderbookMidPriceOpen: '10005', + }; + }, + ); + await verifyCandlesInPostgres(expectedCandles); + + }); + + it('successfully creates an orderbook price map for each market', async () => { + await updatePriceLevel('BTC-USD', '100000', OrderSide.BUY); + await updatePriceLevel('BTC-USD', '110000', OrderSide.SELL); + + await updatePriceLevel('ISO-USD', '110000', OrderSide.BUY); + await updatePriceLevel('ISO-USD', '120000', OrderSide.SELL); + + await updatePriceLevel('ETH-USD', '100000', OrderSide.BUY); + await updatePriceLevel('ETH-USD', '200000', OrderSide.SELL); + + const map = await getOrderbookMidPriceMap(); + expect(map).toEqual({ + 'BTC-USD': '105000', + 'ETH-USD': '150000', + 'ISO-USD': '115000', + 'ISO2-USD': undefined, + 'SHIB-USD': undefined, + }); + }); }); async function createOpenPosition( diff --git a/indexer/services/ender/src/lib/candles-generator.ts b/indexer/services/ender/src/lib/candles-generator.ts index 209557fad57..00fd1ab5981 100644 --- a/indexer/services/ender/src/lib/candles-generator.ts +++ b/indexer/services/ender/src/lib/candles-generator.ts @@ -20,6 +20,7 @@ import { TradeMessageContents, helpers, } from '@dydxprotocol-indexer/postgres'; +import { OrderbookLevelsCache } from '@dydxprotocol-indexer/redis'; import { CandleMessage } from '@dydxprotocol-indexer/v4-protos'; import Big from 'big.js'; import _ from 'lodash'; @@ -27,6 +28,7 @@ import { DateTime } from 'luxon'; import { getCandle } from '../caches/candle-cache'; import config from '../config'; +import { redisClient } from '../helpers/redis/redis-controller'; import { KafkaPublisher } from './kafka-publisher'; import { ConsolidatedKafkaEvent, SingleTradeMessage } from './types'; @@ -40,6 +42,9 @@ type BlockCandleUpdate = { usdVolume: string; trades: number; }; + +type OrderbookMidPrice = string | undefined; + type OpenInterestMap = { [ticker: string]: string }; export class CandlesGenerator { @@ -100,7 +105,6 @@ export class CandlesGenerator { if (ticker === undefined) { throw Error(`Could not find ticker for clobPairId: ${tradeMessage.clobPairId}`); } - // There should only be a single trade in SingleTradeMessage const contents: TradeMessageContents = JSON.parse(tradeMessage.contents); const tradeContent: TradeContent = contents.trades[0]; @@ -167,6 +171,7 @@ export class CandlesGenerator { const promises: Promise[] = []; const openInterestMap: OpenInterestMap = await this.getOpenInterestMap(); + const orderbookMidPriceMap = await getOrderbookMidPriceMap(); _.forEach( Object.values(perpetualMarketRefresher.getPerpetualMarketsMap()), (perpetualMarket: PerpetualMarketFromDatabase) => { @@ -176,11 +181,12 @@ export class CandlesGenerator { _.forEach( Object.values(CandleResolution), (resolution: CandleResolution) => { - promises.push(this.createUpdateOrPassPostgresCandle( + promises.push(...this.createUpdateOrPassPostgresCandle( blockCandleUpdate, perpetualMarket.ticker, resolution, openInterestMap, + orderbookMidPriceMap[perpetualMarket.ticker], )); }, ); @@ -200,17 +206,26 @@ export class CandlesGenerator { * Cases: * - Candle doesn't exist & there is no block update - do nothing * - Candle doesn't exist & there is a block update - create candle - * - Candle exists & !sameStartTime & there is a block update - create candle - * - Candle exists & !sameStartTime & there is no block update - create empty candle + * - Candle exists & !sameStartTime & there is a block update - create candle, + * update previous candle orderbookMidPriceClose + * - Candle exists & !sameStartTime & there is no block update - create empty candle, + * update previous candle orderbookMidPriceClose * - Candle exists & sameStartTime & no block update - do nothing * - Candle exists & sameStartTime & block update - update candle + * + * The orderbookMidPriceClose/Open are updated for each candle at the start and end of + * each resolution period. + * Whenever we create a new candle we set the orderbookMidPriceClose/Open + * If there is a previous candle & we're creating a new one (this occurs at the + * beginning of a resolution period) set the previous candles orderbookMidPriceClose */ - private async createUpdateOrPassPostgresCandle( + private createUpdateOrPassPostgresCandle( blockCandleUpdate: BlockCandleUpdate | undefined, ticker: string, resolution: CandleResolution, openInterestMap: OpenInterestMap, - ): Promise { + orderbookMidPrice: OrderbookMidPrice, + ): Promise[] { const currentStartTime: DateTime = CandlesGenerator.calculateNormalizedCandleStartTime( this.blockTimestamp, resolution, @@ -224,48 +239,59 @@ export class CandlesGenerator { if (existingCandle === undefined) { // - Candle doesn't exist & there is no block update - do nothing if (blockCandleUpdate === undefined) { - return; + return []; } // - Candle doesn't exist & there is a block update - create candle - return this.createCandleInPostgres( + return [this.createCandleInPostgres( currentStartTime, blockCandleUpdate, ticker, resolution, openInterestMap, - ); + orderbookMidPrice, + )]; } const sameStartTime: boolean = existingCandle.startedAt === currentStartTime.toISO(); if (!sameStartTime) { // - Candle exists & !sameStartTime & there is a block update - create candle + // update previous candle orderbookMidPriceClose + + const previousCandleUpdate = this.updateCandleWithOrderbookMidPriceInPostgres( + existingCandle, + orderbookMidPrice, + ); + if (blockCandleUpdate !== undefined) { - return this.createCandleInPostgres( + return [previousCandleUpdate, this.createCandleInPostgres( currentStartTime, blockCandleUpdate, ticker, resolution, openInterestMap, - ); + orderbookMidPrice, + )]; } // - Candle exists & !sameStartTime & there is no block update - create empty candle - return this.createEmptyCandleInPostgres( + // update previous candle orderbookMidPriceClose/Open + return [previousCandleUpdate, this.createEmptyCandleInPostgres( currentStartTime, ticker, resolution, openInterestMap, existingCandle, - ); + orderbookMidPrice, + )]; } if (blockCandleUpdate === undefined) { // - Candle exists & sameStartTime & no block update - do nothing - return; + return []; } // - Candle exists & sameStartTime & block update - update candle - return this.updateCandleInPostgres( + return [this.updateCandleInPostgres( existingCandle, blockCandleUpdate, - ); + )]; } /** @@ -344,6 +370,7 @@ export class CandlesGenerator { ticker: string, resolution: CandleResolution, openInterestMap: OpenInterestMap, + orderbookMidPrice: OrderbookMidPrice, ): Promise { const candle: CandleCreateObject = { startedAt: startedAt.toISO(), @@ -357,6 +384,8 @@ export class CandlesGenerator { usdVolume: blockCandleUpdate.usdVolume, trades: blockCandleUpdate.trades, startingOpenInterest: openInterestMap[ticker] ?? '0', + orderbookMidPriceClose: orderbookMidPrice, + orderbookMidPriceOpen: orderbookMidPrice, }; return CandleTable.create(candle, this.writeOptions); @@ -373,6 +402,7 @@ export class CandlesGenerator { resolution: CandleResolution, openInterestMap: OpenInterestMap, existingCandle: CandleFromDatabase, + orderbookMidPrice: OrderbookMidPrice, ): Promise { const candle: CandleCreateObject = { startedAt: startedAt.toISO(), @@ -386,6 +416,8 @@ export class CandlesGenerator { usdVolume: '0', trades: 0, startingOpenInterest: openInterestMap[ticker] ?? '0', + orderbookMidPriceClose: orderbookMidPrice, + orderbookMidPriceOpen: orderbookMidPrice, }; return CandleTable.create(candle, this.writeOptions); @@ -411,6 +443,8 @@ export class CandlesGenerator { baseTokenVolume: blockCandleUpdate.baseTokenVolume, usdVolume: blockCandleUpdate.usdVolume, trades: blockCandleUpdate.trades, + orderbookMidPriceOpen: existingCandle.orderbookMidPriceOpen ?? undefined, + orderbookMidPriceClose: existingCandle.orderbookMidPriceClose ?? undefined, }, this.writeOptions, ) as Promise; @@ -432,6 +466,28 @@ export class CandlesGenerator { ).toFixed(), usdVolume: Big(existingCandle.usdVolume).plus(blockCandleUpdate.usdVolume).toFixed(), trades: existingCandle.trades + blockCandleUpdate.trades, + orderbookMidPriceClose: existingCandle.orderbookMidPriceClose ?? undefined, + orderbookMidPriceOpen: existingCandle.orderbookMidPriceOpen ?? undefined, + }; + + return CandleTable.update(candle, this.writeOptions) as Promise; + } + + private async updateCandleWithOrderbookMidPriceInPostgres( + existingCandle: CandleFromDatabase, + orderbookMidPrice: OrderbookMidPrice, + ): Promise { + + const candle: CandleUpdateObject = { + id: existingCandle.id, + low: existingCandle.low, + high: existingCandle.high, + close: existingCandle.close, + baseTokenVolume: existingCandle.baseTokenVolume, + usdVolume: existingCandle.usdVolume, + trades: existingCandle.trades, + orderbookMidPriceOpen: existingCandle.orderbookMidPriceOpen ?? undefined, + orderbookMidPriceClose: orderbookMidPrice, }; return CandleTable.update(candle, this.writeOptions) as Promise; @@ -473,3 +529,26 @@ export class CandlesGenerator { return DateTime.fromSeconds(normalizedTimeSeconds).toUTC(); } } + +/** + * Get the cached orderbook mid price for a given ticker +*/ +export async function getOrderbookMidPriceMap(): Promise<{ [ticker: string]: OrderbookMidPrice; }> { + const perpetualMarkets = Object.values(perpetualMarketRefresher.getPerpetualMarketsMap()); + + const promises = perpetualMarkets.map(async (perpetualMarket: PerpetualMarketFromDatabase) => { + const price = await OrderbookLevelsCache.getOrderBookMidPrice( + perpetualMarket.ticker, + redisClient, + ); + return { [perpetualMarket.ticker]: price === undefined ? undefined : price }; + }); + + const pricesArray = await Promise.all(promises); + const priceMap: { [ticker: string]: OrderbookMidPrice; } = {}; + pricesArray.forEach((price) => { + Object.assign(priceMap, price); + }); + + return priceMap; +} diff --git a/indexer/services/roundtable/src/lib/athena-ddl-tables/candles.ts b/indexer/services/roundtable/src/lib/athena-ddl-tables/candles.ts index c9bcc36cf09..54904a006b5 100644 --- a/indexer/services/roundtable/src/lib/athena-ddl-tables/candles.ts +++ b/indexer/services/roundtable/src/lib/athena-ddl-tables/candles.ts @@ -19,6 +19,8 @@ const RAW_TABLE_COLUMNS: string = ` \`usdVolume\` string, \`trades\` int, \`startingOpenInterest\` string + \`orderbookMidPriceOpen\` string + \`orderbookMidPriceClose\` string `; const TABLE_COLUMNS: string = ` "id", @@ -33,6 +35,8 @@ const TABLE_COLUMNS: string = ` ${castToDouble('usdVolume')}, "trades", ${castToDouble('startingOpenInterest')} + ${castToDouble('orderbookMidPriceOpen')} + ${castToDouble('orderbookMidPriceClose')} `; export function generateRawTable(tablePrefix: string, rdsExportIdentifier: string): string {