diff --git a/src/athena/README.md b/src/athena/README.md new file mode 100644 index 0000000..ebe9a2f --- /dev/null +++ b/src/athena/README.md @@ -0,0 +1,348 @@ +# Athena Integration + +This module provides connectivity to AWS Athena for querying historical Drift protocol data. + +## Setup + +### Prerequisites + +1. AWS credentials configured with access to Athena +2. Environment variables: + - `AWS_REGION` (optional, defaults to `us-east-1`) + - `AWS_ACCESS_KEY_ID` (required) + - `AWS_SECRET_ACCESS_KEY` (required) + - `ATHENA_DATABASE` (optional, defaults to `staging-archive`) + - `ATHENA_OUTPUT_BUCKET` (optional, defaults to `staging-data-ingestion-bucket`) + +### Installation + +The required AWS SDK packages are already included in `package.json`: +```bash +yarn install +``` + +## Usage + +### Basic Query Example + +```typescript +import { Athena } from './athena/client'; + +const { query } = Athena(); + +const results = await query(` + SELECT * FROM eventtype_traderecord + WHERE markettype = 'perp' + LIMIT 10 +`); + +console.log(results); +``` + +### Using the Fill Quality Analytics Repository + +```typescript +import { FillQualityAnalyticsRepository } from './athena/repositories/fillQualityAnalytics'; + +const repository = FillQualityAnalyticsRepository(); + +// Get taker fill vs oracle basis points for ALL perp markets +// Time range: last 7 days +const nowMs = Date.now(); +const weekAgoMs = nowMs - 7 * 24 * 60 * 60 * 1000; + +const results = await repository.getTakerFillVsOracleBps( + weekAgoMs, // from (unix timestamp in milliseconds) + nowMs, // to (unix timestamp in milliseconds) + 9, // baseDecimals (default: 9) + 60 // smoothingMinutes (default: 60) +); + +// Results contain time series data with columns: +// - Time: timestamp +// - MarketIndex: the perp market index +// - TakerBuyBpsFromOracle_1e0: average for buy orders < $1000 +// - TakerBuyBpsFromOracle_1e3: average for buy orders $1k-$10k +// - TakerBuyBpsFromOracle_1e4: average for buy orders $10k-$100k +// - TakerBuyBpsFromOracle_1e5: average for buy orders $100k-$1M +// - TakerBuyBpsFromOracle_1e6: average for buy orders > $1M +// - (Similar columns for sell orders) +// - Zero: baseline (always 0) + +// Data is partitioned by MarketIndex, so you can filter/group by market +const solPerpData = results.filter(r => r.MarketIndex === '0'); +console.log(solPerpData); +``` + +### Batch Queries + +```typescript +import { Athena } from './athena/client'; + +const { batchQuery } = Athena(); + +const { results, errors } = await batchQuery({ + queries: [ + { query: 'SELECT COUNT(*) FROM eventtype_traderecord' }, + { query: 'SELECT COUNT(*) FROM eventtype_orderrecord' }, + ] +}); + +console.log('Results:', results); +console.log('Errors:', errors); +``` + +### Custom Database/Bucket + +```typescript +import { Athena } from './athena/client'; + +const { query } = Athena({ + overrideDatabaseName: 'my_custom_db', + overrideBucketName: 'my-custom-bucket' +}); +``` + +## Architecture + +### Client (`client.ts`) +- Handles Athena query execution +- Manages query lifecycle (start, poll, retrieve results) +- Implements rate limiting via Bottleneck +- Provides batch query support + +### Utilities (`utils.ts`) +- `getTimePartition`: Generate time partition filters for queries +- `getTimeRangeAndPartitions`: Generate time range and partition filters + +### Repositories +- **Fill Quality Analytics** (`repositories/fillQualityAnalytics.ts`): Analyzes taker fill quality vs oracle prices, bucketed by order size cohorts + +## Query Best Practices + +1. **Always use partition filters** to minimize data scanned: + ```sql + WHERE year = '2024' AND month = '01' AND day = '15' + ``` + +2. **Use the utility functions** for generating time-based filters: + ```typescript + import { getTimeRangeAndPartitions } from './athena/utils'; + + const query = ` + ${getTimeRangeAndPartitions(from, to)} + SELECT * FROM eventtype_traderecord + JOIN valid_partitions vp ON ... + `; + ``` + +3. **Monitor costs**: Each query logs bytes scanned. Athena charges per TB scanned. + +4. **Use batch queries** when running multiple independent queries to improve throughput. + +## Error Handling + +The client automatically retries failed queries up to 3 times with exponential backoff. Queries timeout after 5 minutes (300s) by default. + +```typescript +try { + const results = await query(queryString); +} catch (error) { + console.error('Query failed:', error.message); + // Handle error appropriately +} +``` + +## Adding New Repositories + +To add a new analytics repository: + +1. Create a new file in `repositories/` (e.g., `myAnalytics.ts`) +2. Define TypeScript interfaces for your results +3. Create a repository function that uses the Athena client +4. Export from `index.ts` + +Example: + +```typescript +// repositories/myAnalytics.ts +import { Athena } from '../client'; + +export interface MyResult { + field1: string; + field2: string; +} + +export const MyAnalyticsRepository = () => { + const { query } = Athena(); + + const getMyData = async (from: number, to: number): Promise => { + const queryString = ` + SELECT field1, field2 + FROM my_table + WHERE ts BETWEEN ${from} AND ${to} + `; + + const results = await query(queryString); + + return results.map((r) => ({ + field1: r.field1 || '', + field2: r.field2 || '', + })); + }; + + return { getMyData }; +}; +``` + +```typescript +// index.ts +export * from './repositories/myAnalytics'; +``` + +## Integration with DLOBPublisher + +The DLOBPublisher (`src/publishers/dlobPublisher.ts`) includes automatic fetching and caching of fill quality analytics data to Redis. + +### Configuration + +Enable fill quality analytics via environment variables: + +```bash +# Enable the feature (default: false) +ENABLE_FILL_QUALITY_ANALYTICS=true + +# How often to fetch and update analytics (default: 300000 = 5 minutes) +FILL_QUALITY_ANALYTICS_INTERVAL=300000 + +# Historical data lookback window (default: 86400000 = 24 hours) +FILL_QUALITY_ANALYTICS_LOOKBACK_MS=86400000 + +# Rolling average smoothing window in minutes (default: 60) +FILL_QUALITY_ANALYTICS_SMOOTHING_MINUTES=60 +``` + +### How it Works + +1. **On Startup**: Immediately fetches fill quality analytics for all perp markets +2. **Periodic Updates**: Re-fetches data every `FILL_QUALITY_ANALYTICS_INTERVAL` milliseconds +3. **Redis Storage**: Stores results in Redis with the following keys: + - `taker_fill_vs_oracle_bps:market:{marketIndex}` - Individual market data + - `taker_fill_vs_oracle_bps:summary` - Summary with metadata + +### Redis Data Format + +Per-market keys contain: +```json +{ + "time": "2024-01-15 10:30:00.000", + "marketIndex": "0", + "takerBuyBpsFromOracle": { + "all": "5.23", + "1e0": "8.45", + "1e3": "6.12", + "1e4": "4.89", + "1e5": "3.56", + "1e6": "2.34" + }, + "takerSellBpsFromOracle": { + "all": "-4.87", + "1e0": "-7.23", + "1e3": "-5.45", + "1e4": "-4.12", + "1e5": "-3.01", + "1e6": "-2.15" + }, + "updatedAt": "2024-01-15T10:35:00.000Z" +} +``` + +Summary key contains: +```json +{ + "markets": ["0", "1", "2"], + "lastUpdated": "2024-01-15T10:35:00.000Z", + "lookbackMs": 86400000, + "smoothingMinutes": 60 +} +``` + +### Accessing the Data + +Downstream services can read the cached analytics from Redis without querying Athena directly: + +```typescript +import { RedisClient } from '@drift/common/clients'; + +const redis = new RedisClient(); +await redis.connect(); + +// Get summary +const summary = await redis.get('taker_fill_vs_oracle_bps:summary'); +console.log(JSON.parse(summary)); + +// Get data for SOL-PERP (market index 0) +const solData = await redis.get('taker_fill_vs_oracle_bps:market:0'); +console.log(JSON.parse(solData)); +``` + +This architecture allows multiple services to access expensive analytics queries without hitting Athena repeatedly. + +## Testing + +### Test Redis Polling + +To verify that fill quality analytics data is being written to and updated in Redis: + +```bash +# Start the DLOBPublisher with fill quality analytics enabled +ENABLE_FILL_QUALITY_ANALYTICS=true yarn dlob-publish + +# In another terminal, run the Redis polling test +yarn test:fill-quality-redis +``` + +The polling test script will: +- Connect to Redis +- Poll every 10 seconds for markets 0, 1, 2 (SOL, BTC, ETH) +- Display the summary and per-market data +- Track when data is updated (shows 🆕 NEW UPDATE when timestamps change) +- Validate data structure integrity +- Show data age and warn if stale (>10 minutes old) + +**Example output:** +``` +========== Polling iteration 1 ========== + +📋 Summary: + Markets available: 0, 1, 2 + Last updated: 2024-01-15T10:35:00.000Z + Lookback: 86400000ms (24h) + Smoothing: 60 minutes + +📊 Market Data: + 🆕 NEW UPDATE - SOL-PERP (Market 0) + Data timestamp: 2024-01-15 10:30:00.000 + Updated at: 2024-01-15T10:35:00.000Z + Age: 15s ago + Taker Buy (all): 5.23 bps + Taker Sell (all): -4.87 bps + Large orders (>$1M): Buy=2.34bps, Sell=-2.15bps + + 🆕 NEW UPDATE - BTC-PERP (Market 1) + Data timestamp: 2024-01-15 10:30:00.000 + Updated at: 2024-01-15T10:35:00.000Z + Age: 15s ago + Taker Buy (all): 4.12 bps + Taker Sell (all): -3.89 bps + Large orders (>$1M): Buy=1.89bps, Sell=-1.76bps + +📈 Update Tracking: + ✅ All 3 markets have data + +🔍 Validation: + ✅ All data structures are valid +``` + +The script will continue polling until you press `Ctrl+C`. + diff --git a/src/athena/client.ts b/src/athena/client.ts index c0494f7..2cd1de2 100644 --- a/src/athena/client.ts +++ b/src/athena/client.ts @@ -42,16 +42,24 @@ const parseRow = (row: Row, columnInfo: ColumnInfo[]): QueryResult => { const athena = new AthenaClient({ region: process.env.AWS_REGION, - retryStrategy: new ConfiguredRetryStrategy(3, (attempt: number) => 100 + attempt * 1000), + retryStrategy: new ConfiguredRetryStrategy( + 3, + (attempt: number) => 100 + attempt * 1000 + ), }); export const Athena = ({ overrideDatabaseName, overrideBucketName, }: { overrideDatabaseName?: string; overrideBucketName?: string } = {}) => { - const database = overrideDatabaseName ?? process.env.ATHENA_DATABASE ?? DEFAULT_ATHENA_DATABASE; + const database = + overrideDatabaseName ?? + process.env.ATHENA_DATABASE ?? + DEFAULT_ATHENA_DATABASE; const outputLocation = `s3://${ - overrideBucketName ?? process.env.ATHENA_OUTPUT_BUCKET ?? DEFAULT_ATHENA_OUTPUT_BUCKET + overrideBucketName ?? + process.env.ATHENA_OUTPUT_BUCKET ?? + DEFAULT_ATHENA_OUTPUT_BUCKET }/athena`; const startQuery = async (query: string, params?: Record) => { @@ -104,7 +112,9 @@ export const Athena = ({ while (Date.now() - startTime < timeout) { const execution = await getQueryExecution(queryExecutionId); logger.info( - `Total bytes scanned: ${(execution?.Statistics?.DataScannedInBytes ?? 0) / 1024}kb` + `Total bytes scanned: ${ + (execution?.Statistics?.DataScannedInBytes ?? 0) / 1024 + }kb` ); const state = execution?.Status?.State; @@ -112,7 +122,9 @@ export const Athena = ({ case QueryExecutionState.SUCCEEDED: return; case QueryExecutionState.FAILED: - throw new Error(`Query failed: ${execution?.Status?.StateChangeReason}`); + throw new Error( + `Query failed: ${execution?.Status?.StateChangeReason}` + ); case QueryExecutionState.CANCELLED: throw new Error('Query was cancelled'); default: @@ -135,15 +147,22 @@ export const Athena = ({ ); }; - const getAllQueryResults = async (queryExecutionId: string): Promise => { + const getAllQueryResults = async ( + queryExecutionId: string + ): Promise => { let nextToken: string | undefined; const allResults: QueryResult[] = []; do { const results = await getQueryResults(queryExecutionId, nextToken); - if (results.ResultSet?.Rows && results.ResultSet.ResultSetMetadata?.ColumnInfo) { + if ( + results.ResultSet?.Rows && + results.ResultSet.ResultSetMetadata?.ColumnInfo + ) { const columnInfo = results.ResultSet.ResultSetMetadata.ColumnInfo; - const rows = nextToken ? results.ResultSet.Rows : results.ResultSet.Rows.slice(1); + const rows = nextToken + ? results.ResultSet.Rows + : results.ResultSet.Rows.slice(1); const parsedRows = rows.map((row) => parseRow(row, columnInfo)); allResults.push(...parsedRows); } @@ -183,10 +202,14 @@ export const Athena = ({ await Promise.all( queries.map(async ({ query: queryString, params }, index) => { try { - const result = await limiter.schedule(() => query(queryString, params)); + const result = await limiter.schedule(() => + query(queryString, params) + ); results[index] = result; } catch (error) { - logger.error(`Error in batch query ${index}: ${queryString}, Error: ${error}`); + logger.error( + `Error in batch query ${index}: ${queryString}, Error: ${error}` + ); errors[index] = error as Error; } }) @@ -212,4 +235,3 @@ export const Athena = ({ waitForQueryCompletion, }; }; - diff --git a/src/athena/index.ts b/src/athena/index.ts index e9fb402..ae965fc 100644 --- a/src/athena/index.ts +++ b/src/athena/index.ts @@ -1,4 +1,3 @@ export * from './client'; export * from './utils'; export * from './repositories/fillQualityAnalytics'; - diff --git a/src/athena/repositories/fillQualityAnalytics.ts b/src/athena/repositories/fillQualityAnalytics.ts index 03c75ab..291d241 100644 --- a/src/athena/repositories/fillQualityAnalytics.ts +++ b/src/athena/repositories/fillQualityAnalytics.ts @@ -1,7 +1,6 @@ import { Athena } from '../client'; export interface TakerFillVsOracleBpsResult { - Time: string; MarketIndex: string; TakerBuyBpsFromOracle_ALL: string | null; TakerSellBpsFromOracle_ALL: string | null; @@ -18,6 +17,27 @@ export interface TakerFillVsOracleBpsResult { Zero: string; } +export type TakerFillVsOracleBpsRedisResult = { + marketIndex: string; + takerBuyBpsFromOracle: { + all: string | null; + '1e0': string | null; + '1e3': string | null; + '1e4': string | null; + '1e5': string | null; + '1e6': string | null; + }; + takerSellBpsFromOracle: { + all: string | null; + '1e0': string | null; + '1e3': string | null; + '1e4': string | null; + '1e5': string | null; + '1e6': string | null; + }; + updatedAtTs: number; +}; + export const FillQualityAnalyticsRepository = () => { const { query } = Athena(); @@ -255,7 +275,6 @@ ORDER BY MarketIndex; const results = await query(queryString); return results.map((result) => ({ - Time: result.Time || '', MarketIndex: result.MarketIndex || '', TakerBuyBpsFromOracle_ALL: result.TakerBuyBpsFromOracle_ALL, TakerSellBpsFromOracle_ALL: result.TakerSellBpsFromOracle_ALL, @@ -277,4 +296,3 @@ ORDER BY MarketIndex; getTakerFillVsOracleBps, }; }; - diff --git a/src/athena/utils.ts b/src/athena/utils.ts index 484ce10..4fd9ec9 100644 --- a/src/athena/utils.ts +++ b/src/athena/utils.ts @@ -37,4 +37,3 @@ export const getTimeRangeAndPartitions = ( ) AS ${valid_partitions_label}(year, month, day) )`; }; - diff --git a/src/index.ts b/src/index.ts index 9651ee5..f3ae323 100644 --- a/src/index.ts +++ b/src/index.ts @@ -50,8 +50,9 @@ import FEATURE_FLAGS from './utils/featureFlags'; import { getDLOBProviderFromOrderSubscriber } from './dlobProvider'; import { setGlobalDispatcher, Agent } from 'undici'; import { HermesClient } from '@pythnetwork/hermes-client'; -import { COMMON_UI_UTILS } from '@drift/common'; +import { COMMON_UI_UTILS, ENUM_UTILS } from '@drift/common'; import { AuctionParamArgs } from './utils/types'; +import { TakerFillVsOracleBpsRedisResult } from './athena/repositories/fillQualityAnalytics'; setGlobalDispatcher( new Agent({ @@ -947,6 +948,7 @@ const main = async (): Promise => { forceUpToSlippage, maxLeverageSelected, maxLeverageOrderSize, + version, } = req.query; // Validate required parameters @@ -959,6 +961,31 @@ const main = async (): Promise => { return; } + const apiVersion = version ? parseInt(version as string) : 1; + + let redisFillQualityInfo: TakerFillVsOracleBpsRedisResult | undefined; + if (apiVersion === 2) { + const redisKey = `taker_fill_vs_oracle_bps:market:${marketIndex}`; + try { + const redisValue = await fetchFromRedis( + redisKey, + (responses) => responses[0] as any + ); + if (redisValue) { + const parsed = JSON.parse(redisValue); + redisFillQualityInfo = + typeof parsed === 'string' ? JSON.parse(parsed) : parsed; + } + // Fall through to existing logic below + } catch (err) { + logger.error( + `Version 2: Error fetching redis stats for market ${marketIndex}:`, + err + ); + // Fall through to existing logic below + } + } + // Parse and validate values const parsedMarketIndex = parseInt(marketIndex as string); if (isNaN(parsedMarketIndex)) { @@ -1029,7 +1056,9 @@ const main = async (): Promise => { inputParams, driftClient, fetchFromRedis, - selectMostRecentBySlot + selectMostRecentBySlot, + redisFillQualityInfo, + apiVersion ); if (!result.success) { @@ -1043,6 +1072,24 @@ const main = async (): Promise => { result.data.marketOrderParams ); + // Log final auction prices for debugging + logger.info( + JSON.stringify({ + event: 'auction_params_derived', + marketIndex: parsedMarketIndex, + direction: direction, + apiVersion, + finalAuctionParams: { + auctionStartPrice: auctionParams.auctionStartPrice?.toString(), + auctionEndPrice: auctionParams.auctionEndPrice?.toString(), + price: auctionParams.price?.toString(), + oraclePriceOffset: auctionParams.oraclePriceOffset?.toString(), + auctionDuration: auctionParams.auctionDuration?.toString(), + orderType: ENUM_UTILS.toStr(auctionParams.orderType), + }, + }) + ); + const response = { data: { params: formatAuctionParamsForResponse(auctionParams), diff --git a/src/publishers/dlobPublisher.ts b/src/publishers/dlobPublisher.ts index adf7b9c..cf91ae0 100644 --- a/src/publishers/dlobPublisher.ts +++ b/src/publishers/dlobPublisher.ts @@ -43,6 +43,10 @@ import { handleHealthCheck } from '../core/middleware'; import { setGlobalDispatcher, Agent } from 'undici'; import { Metrics } from '../core/metricsV2'; import { OrderSubscriberFiltered } from '../dlob-subscriber/OrderSubscriberFiltered'; +import { + FillQualityAnalyticsRepository, + TakerFillVsOracleBpsRedisResult, +} from '../athena/repositories/fillQualityAnalytics'; setGlobalDispatcher( new Agent({ @@ -93,6 +97,10 @@ const tobStuckGauge = metricsV2.addGauge( 'tob_stuck_duration', 'Duration TOB has been stuck for each market' ); +const takerFillBpsFromOracleGauge = metricsV2.addGauge( + 'taker_fill_bps_from_oracle', + 'Taker fill BPS from oracle by market, side, and cohort' +); metricsV2.finalizeObservables(); //@ts-ignore @@ -109,7 +117,7 @@ const endpoint = process.env.ENDPOINT; const grpcEndpoint = useGrpc ? process.env.GRPC_ENDPOINT ?? endpoint + `/${token}` : ''; -const grpcClient = process.env.GRPC_CLIENT ?? 'yellowstone' +const grpcClient = process.env.GRPC_CLIENT ?? 'yellowstone'; const wsEndpoint = process.env.WS_ENDPOINT; const useOrderSubscriber = @@ -122,6 +130,16 @@ const WS_FALLBACK_FETCH_INTERVAL = 60_000; const KILLSWITCH_SLOT_DIFF_THRESHOLD = parseInt(process.env.KILLSWITCH_SLOT_DIFF_THRESHOLD) || 200; +// Fill Quality Analytics configuration +const ENABLE_FILL_QUALITY_ANALYTICS = + process.env.ENABLE_FILL_QUALITY_ANALYTICS?.toLowerCase() === 'true'; +const FILL_QUALITY_ANALYTICS_INTERVAL = + parseInt(process.env.FILL_QUALITY_ANALYTICS_INTERVAL) || 300_000; // 5 minutes default +const FILL_QUALITY_ANALYTICS_LOOKBACK_MS = + parseInt(process.env.FILL_QUALITY_ANALYTICS_LOOKBACK_MS) || 86_400_000; // 24 hours default +const FILL_QUALITY_ANALYTICS_SMOOTHING_MINUTES = + parseInt(process.env.FILL_QUALITY_ANALYTICS_SMOOTHING_MINUTES) || 60; + // TOB monitoring configuration - defaults to true if not set const ENABLE_TOB_MONITORING = !process.env.ENABLE_TOB_MONITORING || @@ -169,6 +187,22 @@ if (ENABLE_TOB_MONITORING) { `TOB Monitoring Markets: ${TOB_MONITORING_ENABLED_PERP_MARKETS.join(', ')}` ); } +logger.info( + `Fill Quality Analytics: ${ + ENABLE_FILL_QUALITY_ANALYTICS ? 'enabled' : 'disabled' + }` +); +if (ENABLE_FILL_QUALITY_ANALYTICS) { + logger.info( + `Fill Quality Analytics Interval: ${FILL_QUALITY_ANALYTICS_INTERVAL}ms` + ); + logger.info( + `Fill Quality Analytics Lookback: ${FILL_QUALITY_ANALYTICS_LOOKBACK_MS}ms` + ); + logger.info( + `Fill Quality Analytics Smoothing: ${FILL_QUALITY_ANALYTICS_SMOOTHING_MINUTES} minutes` + ); +} let MARKET_SUBSCRIBERS: SubscriberLookup = {}; @@ -504,8 +538,7 @@ const main = async () => { 'grpc.keepalive_timeout_ms': 1_000, 'grpc.keepalive_permit_without_calls': 1, }, - client: grpcClient - + client: grpcClient, }, commitment: stateCommitment, }; @@ -817,6 +850,196 @@ const main = async () => { // Start TOB monitoring setInterval(checkTobForStuckOrders, TOB_CHECK_INTERVAL); + // Track last known values for fill quality metrics (per market/side/cohort) + const lastFillQualityValues = new Map(); + + // Helper function to update fill quality metric with null tracking + const updateFillQualityMetric = ( + value: string | number | null | undefined, + marketIndex: string | number, + side: string, + cohort: string + ) => { + const marketIndexStr = marketIndex.toString(); + const key = `${marketIndexStr}:${side}:${cohort}`; + const isNull = value === null || value === undefined; + + let valueToUse: number; + if (isNull) { + // Use last known value, or 0 if no previous value exists + valueToUse = lastFillQualityValues.get(key) || 0; + logger.debug( + `Null value for market ${marketIndexStr} ${side} ${cohort}, using last known value: ${valueToUse}` + ); + } else { + // Use current value and update the last known value + valueToUse = Number(value); + lastFillQualityValues.set(key, valueToUse); + } + + takerFillBpsFromOracleGauge.setLatestValue(valueToUse, { + marketIndex: marketIndexStr, + side, + cohort, + null: isNull ? 'true' : 'false', + }); + }; + + // Fill Quality Analytics function - fetch and store in Redis + const fetchAndStoreFillQualityAnalytics = async () => { + if (!ENABLE_FILL_QUALITY_ANALYTICS) { + return; + } + + try { + logger.info('Starting fill quality analytics fetch'); + const startTime = Date.now(); + + const fillQualityRepo = FillQualityAnalyticsRepository(); + const toMs = Date.now(); + const fromMs = toMs - FILL_QUALITY_ANALYTICS_LOOKBACK_MS; + + const results = await fillQualityRepo.getTakerFillVsOracleBps( + fromMs, + toMs, + 9, // baseDecimals + FILL_QUALITY_ANALYTICS_SMOOTHING_MINUTES + ); + + logger.info( + `Fetched fill quality analytics for ${results.length} markets in ${ + Date.now() - startTime + }ms` + ); + + // Store results in Redis - one key per market + const startRedisSet = Date.now(); + for (const result of results) { + const redisKey = `taker_fill_vs_oracle_bps:market:${result.MarketIndex}`; + const redisValue = JSON.stringify({ + marketIndex: result.MarketIndex, + takerBuyBpsFromOracle: { + all: result.TakerBuyBpsFromOracle_ALL, + '1e0': result.TakerBuyBpsFromOracle_1e0, + '1e3': result.TakerBuyBpsFromOracle_1e3, + '1e4': result.TakerBuyBpsFromOracle_1e4, + '1e5': result.TakerBuyBpsFromOracle_1e5, + '1e6': result.TakerBuyBpsFromOracle_1e6, + }, + takerSellBpsFromOracle: { + all: result.TakerSellBpsFromOracle_ALL, + '1e0': result.TakerSellBpsFromOracle_1e0, + '1e3': result.TakerSellBpsFromOracle_1e3, + '1e4': result.TakerSellBpsFromOracle_1e4, + '1e5': result.TakerSellBpsFromOracle_1e5, + '1e6': result.TakerSellBpsFromOracle_1e6, + }, + updatedAtTs: Date.now(), + } as TakerFillVsOracleBpsRedisResult); + + await redisClient.set(redisKey, redisValue); + + updateFillQualityMetric( + result.TakerBuyBpsFromOracle_ALL, + result.MarketIndex, + 'buy', + 'all' + ); + updateFillQualityMetric( + result.TakerBuyBpsFromOracle_1e0, + result.MarketIndex, + 'buy', + '1e0' + ); + updateFillQualityMetric( + result.TakerBuyBpsFromOracle_1e3, + result.MarketIndex, + 'buy', + '1e3' + ); + updateFillQualityMetric( + result.TakerBuyBpsFromOracle_1e4, + result.MarketIndex, + 'buy', + '1e4' + ); + updateFillQualityMetric( + result.TakerBuyBpsFromOracle_1e5, + result.MarketIndex, + 'buy', + '1e5' + ); + updateFillQualityMetric( + result.TakerBuyBpsFromOracle_1e6, + result.MarketIndex, + 'buy', + '1e6' + ); + updateFillQualityMetric( + result.TakerSellBpsFromOracle_ALL, + result.MarketIndex, + 'sell', + 'all' + ); + updateFillQualityMetric( + result.TakerSellBpsFromOracle_1e0, + result.MarketIndex, + 'sell', + '1e0' + ); + updateFillQualityMetric( + result.TakerSellBpsFromOracle_1e3, + result.MarketIndex, + 'sell', + '1e3' + ); + updateFillQualityMetric( + result.TakerSellBpsFromOracle_1e4, + result.MarketIndex, + 'sell', + '1e4' + ); + updateFillQualityMetric( + result.TakerSellBpsFromOracle_1e5, + result.MarketIndex, + 'sell', + '1e5' + ); + updateFillQualityMetric( + result.TakerSellBpsFromOracle_1e6, + result.MarketIndex, + 'sell', + '1e6' + ); + + logger.debug( + `Stored fill quality analytics for market ${result.MarketIndex}` + ); + } + logger.info( + `Successfully stored fill quality analytics for ${ + results.length + } markets in Redis in ${Date.now() - startRedisSet}ms` + ); + } catch (error) { + logger.error('Error fetching/storing fill quality analytics:', error); + } + }; + + // Run fill quality analytics fetch immediately on startup, then on interval + if (ENABLE_FILL_QUALITY_ANALYTICS) { + fetchAndStoreFillQualityAnalytics().catch((error) => { + logger.error('Initial fill quality analytics fetch failed:', error); + }); + setInterval( + fetchAndStoreFillQualityAnalytics, + FILL_QUALITY_ANALYTICS_INTERVAL + ); + logger.info( + `Fill quality analytics will run every ${FILL_QUALITY_ANALYTICS_INTERVAL}ms` + ); + } + const handleStartup = async (_req, res, _next) => { if (driftClient.isSubscribed && dlobProvider.size() > 0) { res.writeHead(200); diff --git a/src/utils/utils.ts b/src/utils/utils.ts index 9b3eea1..ec0f7ae 100644 --- a/src/utils/utils.ts +++ b/src/utils/utils.ts @@ -34,6 +34,7 @@ import { wsMarketArgs } from 'src/dlob-subscriber/DLOBSubscriberIO'; import { DEFAULT_AUCTION_PARAMS, MID_MAJOR_MARKETS } from './constants'; import { AuctionParamArgs } from './types'; import { COMMON_MATH, ENUM_UTILS } from '@drift/common'; +import { TakerFillVsOracleBpsRedisResult } from '../athena/repositories/fillQualityAnalytics'; export const GROUPING_OPTIONS = [1, 10, 100, 500, 1000]; export const GROUPING_DEPENDENCIES = { @@ -312,10 +313,17 @@ export function publishGroupings( asks: aggregatedAsks, }); - - if(['SOL-PERP', 'BTC-PERP', 'ETH-PERP'].includes(l2Formatted_grouped20.marketName) && aggregatedBids.length !== 20 || aggregatedAsks.length !== 20) { - logger.error(`Error aggregating dlob levels: group=${group}, bids=${fullAggregatedBids.length}, asks=${fullAggregatedAsks.length}`) - logger.error(`Response: ${JSON.stringify(l2Formatted_grouped20)}`) + if ( + (['SOL-PERP', 'BTC-PERP', 'ETH-PERP'].includes( + l2Formatted_grouped20.marketName + ) && + aggregatedBids.length !== 20) || + aggregatedAsks.length !== 20 + ) { + logger.error( + `Error aggregating dlob levels: group=${group}, bids=${fullAggregatedBids.length}, asks=${fullAggregatedAsks.length}` + ); + logger.error(`Response: ${JSON.stringify(l2Formatted_grouped20)}`); } redisClient.publish( @@ -911,6 +919,8 @@ export const getEstimatedPrices = async ( * @param driftClient - DriftClient instance (optional, for price calculation) * @param fetchFromRedis - Redis fetch function (optional, for price calculation) * @param selectMostRecentBySlot - Slot selection function (optional, for price calculation) + * @param fillQualityInfo - Fill quality analytics data (version 2 only) + * @param apiVersion - API version (1 or 2) * @returns Object formatted for deriveMarketOrderParams function or error response */ export const mapToMarketOrderParams = async ( @@ -920,7 +930,9 @@ export const mapToMarketOrderParams = async ( key: string, selectionCriteria: (responses: any) => any ) => Promise, - selectMostRecentBySlot?: (responses: any[]) => any + selectMostRecentBySlot?: (responses: any[]) => any, + fillQualityInfo?: TakerFillVsOracleBpsRedisResult, + apiVersion: number = 1 ): Promise<{ success: boolean; data?: { @@ -960,6 +972,15 @@ export const mapToMarketOrderParams = async ( let estimatedPrices; let processedSlippageTolerance = params.slippageTolerance; + // Track debug info for logging + let debugInfo = { + originalOraclePrice: null as string | null, + adjustedOraclePrice: null as string | null, + isCrossed: false, + fillQualityBps: null as number | null, + appliedV2Adjustment: false, + }; + if (driftClient && fetchFromRedis && selectMostRecentBySlot) { // Get L2 orderbook data using the utility function const redisL2 = await fetchL2FromRedis( @@ -980,6 +1001,73 @@ export const mapToMarketOrderParams = async ( redisL2 ); + // Store original oracle for debugging + debugInfo.originalOraclePrice = estimatedPrices.oraclePrice.toString(); + + // VERSION 2: Adjust oracle price based on fill quality when orderbook is crossed + if (apiVersion === 2 && fillQualityInfo && redisL2) { + try { + // Convert raw L2 to formatted L2 for cross detection + const l2Formatted = convertRawL2ToBN(redisL2); + + const isSpot = isVariant(marketType, 'spot'); + const oracleData = isSpot + ? driftClient.getOracleDataForSpotMarket(params.marketIndex) + : driftClient.getOracleDataForPerpMarket(params.marketIndex); + const oraclePrice = oracleData.price ?? ZERO; + + // Detect if orderbook is crossed + const spreadInfo = COMMON_MATH.calculateSpreadBidAskMark( + l2Formatted, + oraclePrice + ); + + const isCrossed = + spreadInfo.bestBidPrice && + spreadInfo.bestAskPrice && + spreadInfo.bestBidPrice.gte(spreadInfo.bestAskPrice); + + debugInfo.isCrossed = isCrossed; + + if (isCrossed) { + // Get fill quality metric based on direction + const fillQualityBpsStr = + direction === PositionDirection.LONG + ? fillQualityInfo.takerBuyBpsFromOracle?.all + : fillQualityInfo.takerSellBpsFromOracle?.all; + + if ( + fillQualityBpsStr && + fillQualityBpsStr !== 'null' && + fillQualityBpsStr !== null + ) { + const fillQualityBps = Math.round( + parseFloat(fillQualityBpsStr) * 100 + ); // mul 100 to maintain 2 sigfigs (1.23 bps = 0.000123) + debugInfo.fillQualityBps = fillQualityBps; + + if (!isNaN(fillQualityBps)) { + const adjustment = oraclePrice + .muln(fillQualityBps) + .divn(10000 * 100); // adjustmentFactor is in bps + const adjustedOraclePrice = oraclePrice.add(adjustment); + + // Update the oracle price in estimatedPrices + estimatedPrices.oraclePrice = adjustedOraclePrice; + debugInfo.adjustedOraclePrice = adjustedOraclePrice.toString(); + debugInfo.appliedV2Adjustment = true; + } + } + } + } catch (error) { + logger.warn( + 'Version 2: Failed to apply fill quality adjustment, using standard oracle:', + error + ); + // Fall through to use unadjusted oracle + } + } + // Handle dynamic slippage tolerance calculation if needed if (params.slippageTolerance === undefined) { // Convert raw L2 to formatted L2 for slippage calculation @@ -1028,6 +1116,62 @@ export const mapToMarketOrderParams = async ( baseAmount = amount.mul(BASE_PRECISION).div(estimatedPrices.entryPrice); } + // Comprehensive debug logging + logger.info( + JSON.stringify({ + event: 'auction_params_calculated', + requestParams: { + marketIndex: params.marketIndex, + marketType: params.marketType, + direction: direction === PositionDirection.LONG ? 'long' : 'short', + amount: params.amount, + assetType: params.assetType, + slippageTolerance: params.slippageTolerance, + apiVersion, + }, + priceDiscovery: { + originalOraclePrice: debugInfo.originalOraclePrice, + finalOraclePrice: estimatedPrices.oraclePrice.toString(), + bestPrice: estimatedPrices.bestPrice.toString(), + entryPrice: estimatedPrices.entryPrice.toString(), + worstPrice: estimatedPrices.worstPrice.toString(), + markPrice: estimatedPrices.markPrice.toString(), + priceImpactBps: estimatedPrices.priceImpact.toString(), + }, + v2CrossDetection: + apiVersion === 2 + ? { + isCrossed: debugInfo.isCrossed, + fillQualityBps: debugInfo.fillQualityBps, + adjustedOraclePrice: debugInfo.adjustedOraclePrice, + appliedAdjustment: debugInfo.appliedV2Adjustment, + fillQualityData: fillQualityInfo + ? { + takerBuyBpsAll: fillQualityInfo.takerBuyBpsFromOracle?.all, + takerSellBpsAll: + fillQualityInfo.takerSellBpsFromOracle?.all, + } + : null, + } + : undefined, + auctionConfig: { + duration: params.auctionDuration, + startPriceOffset: params.auctionStartPriceOffset, + startPriceOffsetFrom: params.auctionStartPriceOffsetFrom, + endPriceOffset: params.auctionEndPriceOffset, + endPriceOffsetFrom: params.auctionEndPriceOffsetFrom, + slippageTolerance: processedSlippageTolerance, + isOracleOrder: params.isOracleOrder, + reduceOnly: params.reduceOnly ?? false, + allowInfSlippage: params.allowInfSlippage ?? false, + }, + calculatedValues: { + baseAmount: baseAmount.toString(), + slippageToleranceFinal: processedSlippageTolerance, + }, + }) + ); + return { success: true, data: {