diff --git a/src/consts.ts b/src/consts.ts index b5c7039..b506452 100644 --- a/src/consts.ts +++ b/src/consts.ts @@ -1,4 +1,5 @@ export const EXCHANGES = [ + 'aster', 'bitmex', 'deribit', 'binance-futures', @@ -64,6 +65,8 @@ export const EXCHANGES = [ ] as const const BINANCE_CHANNELS = ['trade', 'aggTrade', 'ticker', 'depth', 'depthSnapshot', 'bookTicker', 'recentTrades', 'borrowInterest'] as const + +const ASTER_CHANNELS = ['trade', 'aggTrade', 'ticker', 'depth', 'depthSnapshot', 'bookTicker'] as const const BINANCE_DEX_CHANNELS = ['trades', 'marketDiff', 'depthSnapshot', 'ticker'] as const const BITFINEX_CHANNELS = ['trades', 'book', 'raw_book', 'ticker'] as const @@ -558,6 +561,7 @@ const POLYMARKET_CHANNELS = [ ] as const export const EXCHANGE_CHANNELS_INFO = { + aster: ASTER_CHANNELS, bitmex: BITMEX_CHANNELS, coinbase: COINBASE_CHANNELS, 'coinbase-international': COINBASE_INTERNATIONAL_CHANNELS, diff --git a/src/mappers/aster.ts b/src/mappers/aster.ts new file mode 100644 index 0000000..f324a51 --- /dev/null +++ b/src/mappers/aster.ts @@ -0,0 +1,263 @@ +import { asNumberOrUndefined, CircularBuffer, lowerCaseSymbols } from '../handy.ts' +import { BookChange, BookTicker, Trade } from '../types.ts' +import { Mapper } from './mapper.ts' +import { exchangeMappers, isRealTime } from './registry.ts' + +export const asterMappers = exchangeMappers({ + aster: { + trades: () => new AsterTradesMapper(), + bookChanges: (localTimestamp) => + new AsterBookChangeMapper({ + ignoreBookSnapshotOverlapError: shouldIgnoreBookSnapshotOverlap(localTimestamp) + }), + bookTickers: () => new AsterBookTickerMapper() + } +}) + +function shouldIgnoreBookSnapshotOverlap(date?: Date) { + if (process.env.IGNORE_BOOK_SNAPSHOT_OVERLAP_ERROR) { + return true + } + + return isRealTime(date) === false +} + +class AsterTradesMapper implements Mapper<'aster', Trade> { + canHandle(message: AsterMessage) { + return message.stream?.endsWith('@trade') === true + } + + getFilters(symbols?: string[]) { + return [{ channel: 'trade', symbols: lowerCaseSymbols(symbols) } as const] + } + + *map({ data }: AsterMessage, localTimestamp: Date) { + const trade: Trade = { + type: 'trade', + symbol: data.s, + exchange: 'aster', + id: String(data.t), + price: Number(data.p), + amount: Number(data.q), + side: data.m ? 'sell' : 'buy', + timestamp: new Date(data.T), + localTimestamp + } + + yield trade + } +} + +class AsterBookChangeMapper implements Mapper<'aster', BookChange> { + private readonly symbolToDepthInfoMapping: { [key: string]: LocalDepthInfo } = {} + private readonly ignoreBookSnapshotOverlapError: boolean + + constructor({ ignoreBookSnapshotOverlapError }: { ignoreBookSnapshotOverlapError: boolean }) { + this.ignoreBookSnapshotOverlapError = ignoreBookSnapshotOverlapError + } + + canHandle(message: AsterMessage) { + return message.stream?.includes('@depth') === true + } + + getFilters(symbols?: string[]) { + return [ + { channel: 'depth', symbols: lowerCaseSymbols(symbols) } as const, + { channel: 'depthSnapshot', symbols: lowerCaseSymbols(symbols) } as const + ] + } + + *map({ stream, data }: AsterMessage, localTimestamp: Date) { + const symbol = stream.split('@')[0].toUpperCase() + + if (this.symbolToDepthInfoMapping[symbol] === undefined) { + this.symbolToDepthInfoMapping[symbol] = { + bufferedUpdates: new CircularBuffer(2000) + } + } + + const symbolDepthInfo = this.symbolToDepthInfoMapping[symbol] + + if (data.lastUpdateId !== undefined) { + if (symbolDepthInfo.snapshotProcessed) { + return + } + + symbolDepthInfo.lastUpdateId = data.lastUpdateId + symbolDepthInfo.snapshotProcessed = true + + for (const update of symbolDepthInfo.bufferedUpdates.items()) { + const bookChange = this.mapBookDepthUpdate(update, localTimestamp) + if (bookChange !== undefined) { + for (const bid of update.b) { + const matchingBid = data.bids.find((b) => b[0] === bid[0]) + if (matchingBid !== undefined) { + matchingBid[1] = bid[1] + } else { + data.bids.push(bid) + } + } + + for (const ask of update.a) { + const matchingAsk = data.asks.find((a) => a[0] === ask[0]) + if (matchingAsk !== undefined) { + matchingAsk[1] = ask[1] + } else { + data.asks.push(ask) + } + } + } + } + + symbolDepthInfo.bufferedUpdates.clear() + + const bookChange: BookChange = { + type: 'book_change', + symbol, + exchange: 'aster', + isSnapshot: true, + bids: data.bids.map(this.mapBookLevel), + asks: data.asks.map(this.mapBookLevel), + timestamp: data.T !== undefined ? new Date(data.T) : localTimestamp, + localTimestamp + } + + yield bookChange + } else if (symbolDepthInfo.snapshotProcessed) { + const bookChange = this.mapBookDepthUpdate(data as AsterDepthData, localTimestamp) + if (bookChange !== undefined) { + yield bookChange + } + } else { + symbolDepthInfo.bufferedUpdates.append(data as AsterDepthData) + } + } + + private mapBookDepthUpdate(depthUpdateData: AsterDepthData, localTimestamp: Date): BookChange | undefined { + const depthContext = this.symbolToDepthInfoMapping[depthUpdateData.s]! + + if (depthUpdateData.u <= depthContext.lastUpdateId!) { + return + } + + if (!depthContext.validatedFirstUpdate) { + if ( + (depthUpdateData.pu <= depthContext.lastUpdateId! && depthUpdateData.u >= depthContext.lastUpdateId!) || + depthContext.lastUpdateId! == -1 + ) { + depthContext.validatedFirstUpdate = true + } else if (this.ignoreBookSnapshotOverlapError) { + depthContext.validatedFirstUpdate = true + } else { + throw new Error( + `Book depth snaphot has no overlap with first update, update ${JSON.stringify( + depthUpdateData + )}, lastUpdateId: ${depthContext.lastUpdateId!}, exchange aster` + ) + } + } else if (depthUpdateData.pu !== depthContext.lastUpdateId!) { + throw new Error( + `Book depth update has a sequence gap, update ${JSON.stringify( + depthUpdateData + )}, lastUpdateId: ${depthContext.lastUpdateId!}, exchange aster` + ) + } + + depthContext.lastUpdateId = depthUpdateData.u + + return { + type: 'book_change', + symbol: depthUpdateData.s, + exchange: 'aster', + isSnapshot: false, + bids: depthUpdateData.b.map(this.mapBookLevel), + asks: depthUpdateData.a.map(this.mapBookLevel), + timestamp: new Date(depthUpdateData.E), + localTimestamp + } + } + private mapBookLevel(level: AsterBookLevel) { + return { + price: Number(level[0]), + amount: Number(level[1]) + } + } +} + +class AsterBookTickerMapper implements Mapper<'aster', BookTicker> { + canHandle(message: AsterMessage) { + return message.stream?.endsWith('@bookTicker') === true + } + + getFilters(symbols?: string[]) { + return [{ channel: 'bookTicker', symbols: lowerCaseSymbols(symbols) } as const] + } + + *map({ data }: AsterMessage, localTimestamp: Date) { + const ticker: BookTicker = { + type: 'book_ticker', + symbol: data.s, + exchange: 'aster', + askAmount: asNumberOrUndefined(data.A), + askPrice: asNumberOrUndefined(data.a), + bidPrice: asNumberOrUndefined(data.b), + bidAmount: asNumberOrUndefined(data.B), + timestamp: data.E !== undefined ? new Date(data.E) : localTimestamp, + localTimestamp + } + + yield ticker + } +} + +type AsterMessage = { + stream: string + data: T +} + +type AsterTradeData = { + s: string + t: number + p: string + q: string + T: number + m: boolean +} + +type AsterDepthData = { + lastUpdateId: undefined + E: number + T: number + s: string + U: number + u: number + pu: number + b: AsterBookLevel[] + a: AsterBookLevel[] +} + +type AsterDepthSnapshotData = { + lastUpdateId: number + bids: AsterBookLevel[] + asks: AsterBookLevel[] + T?: number +} + +type AsterBookLevel = [string, string] + +type LocalDepthInfo = { + bufferedUpdates: CircularBuffer + snapshotProcessed?: boolean + lastUpdateId?: number + validatedFirstUpdate?: boolean +} + +type AsterBookTickerData = { + u: number + s: string + b: string + B: string + a: string + A: string + E?: number +} diff --git a/src/mappers/index.ts b/src/mappers/index.ts index 1f1c715..1f7e4e6 100644 --- a/src/mappers/index.ts +++ b/src/mappers/index.ts @@ -1,5 +1,6 @@ import { BookChange, BookTicker, DerivativeTicker, Exchange, Liquidation, NormalizedData, OptionSummary, Trade } from '../types.ts' import { ascendexMappers } from './ascendex.ts' +import { asterMappers } from './aster.ts' import { binanceMappers } from './binance.ts' import { binanceDexMappers } from './binancedex.ts' import { binanceEuropeanOptionsMappers } from './binanceeuropeanoptions.ts' @@ -55,6 +56,7 @@ type Normalizer = (() + private readonly bufferedDepthUpdates = new Map>() + protected readonly wssURL: string = 'wss://sstream.asterdex.com/stream' + protected readonly httpURL: string = 'https://sapi.asterdex.com/api/v3' + private readonly channels = new Set([ + 'trade', + 'aggTrade', + 'ticker', + AsterRealTimeFeed.depthChannel, + AsterRealTimeFeed.depthSnapshotChannel, + 'bookTicker' + ]) + private readonly channelMappings: { [key: string]: string | undefined } = { + [AsterRealTimeFeed.depthChannel]: AsterRealTimeFeed.depthStream + } + + protected mapToSubscribeMessages(filters: Filter[]): any[] { + const filtersWithSymbols = filters.map>>((filter) => { + if (!this.channels.has(filter.channel)) { + throw new Error(`AsterRealTimeFeed unsupported channel ${filter.channel}`) + } + + if (!filter.symbols || filter.symbols.length === 0) { + throw new Error('AsterRealTimeFeed requires explicitly specified symbols when subscribing to live feed') + } + + return filter as Required> + }) + + const depthSnapshotFilters = filtersWithSymbols.filter((filter) => filter.channel === AsterRealTimeFeed.depthSnapshotChannel) + this.validateDepthSnapshotFilters(filtersWithSymbols, depthSnapshotFilters) + this.resetDepthSnapshotTracking(depthSnapshotFilters) + + return filtersWithSymbols + .filter((f) => f.channel !== AsterRealTimeFeed.depthSnapshotChannel) + .map((filter, index) => { + return { + method: 'SUBSCRIBE', + params: filter.symbols.map((symbol) => `${symbol.toLowerCase()}@${this.channelMappings[filter.channel] ?? filter.channel}`), + id: index + 1 + } + }) + } + + protected messageIsError(message: any): boolean { + if (message.result === null) { + return false + } + + if (message.stream !== undefined) { + return false + } + + if (message.code !== undefined || message.error !== undefined) { + return true + } + + return false + } + + protected override onMessage(message: any) { + if (message.stream?.endsWith(`@${AsterRealTimeFeed.depthStream}`) !== true || message.data?.s === undefined) { + return + } + + const symbol = message.data.s.toUpperCase() + if (this.pendingDepthSnapshotSymbols.has(symbol) === false) { + return + } + + const lastUpdateId = Number(message.data.u) + const previousFinalUpdateId = Number(message.data.pu) + if (Number.isFinite(lastUpdateId) === false || Number.isFinite(previousFinalUpdateId) === false) { + return + } + + const bufferedUpdates = this.bufferedDepthUpdates.get(symbol) ?? new CircularBuffer(2000) + bufferedUpdates.append({ + lastUpdateId, + previousFinalUpdateId + }) + this.bufferedDepthUpdates.set(symbol, bufferedUpdates) + } + + protected async provideManualSnapshots(filters: Filter[], shouldCancel: () => boolean) { + const depthSnapshotFilter = filters.find((f) => f.channel === AsterRealTimeFeed.depthSnapshotChannel) + if (!depthSnapshotFilter) { + return + } + + for (const symbolsBatch of batch(depthSnapshotFilter.symbols!, 4)) { + if (shouldCancel()) { + return + } + + this.debug('requesting manual snapshots for: %s', symbolsBatch) + + await Promise.all( + symbolsBatch.map(async (symbol) => { + if (shouldCancel()) { + return + } + + await this.provideManualSnapshot(symbol, shouldCancel) + }) + ) + + await wait(100) + this.debug('requested manual snapshots successfully for: %s', symbolsBatch) + } + + this.debug('requested all manual snapshots successfully') + } + + private resetDepthSnapshotTracking(filters: Required>[]) { + this.pendingDepthSnapshotSymbols.clear() + this.bufferedDepthUpdates.clear() + + for (const filter of filters) { + for (const symbol of filter.symbols) { + const upperCaseSymbol = symbol.toUpperCase() + this.pendingDepthSnapshotSymbols.add(upperCaseSymbol) + this.bufferedDepthUpdates.set(upperCaseSymbol, new CircularBuffer(2000)) + } + } + } + + private validateDepthSnapshotFilters(filters: Required>[], depthSnapshotFilters: Required>[]) { + if (depthSnapshotFilters.length === 0) { + return + } + + const depthSymbols = new Set( + filters + .filter((filter) => filter.channel === AsterRealTimeFeed.depthChannel) + .flatMap((filter) => filter.symbols.map((symbol) => symbol.toUpperCase())) + ) + + for (const filter of depthSnapshotFilters) { + for (const symbol of filter.symbols) { + if (depthSymbols.has(symbol.toUpperCase()) === false) { + throw new Error( + `AsterRealTimeFeed requires ${AsterRealTimeFeed.depthChannel} for every ${AsterRealTimeFeed.depthSnapshotChannel} symbol` + ) + } + } + } + } + + private async provideManualSnapshot(symbol: string, shouldCancel: () => boolean) { + const maxSnapshotRounds = 4 + const maxSnapshotAttemptsPerRound = 3 + const normalizedSymbol = symbol.toUpperCase() + + for (let round = 0; round < maxSnapshotRounds; round++) { + for (let attempt = 1; attempt <= maxSnapshotAttemptsPerRound; attempt++) { + if (shouldCancel()) { + return + } + + const { data } = await getJSON(`${this.httpURL}/depth?symbol=${symbol}&limit=1000`) + if (this.snapshotResponseIsValid(data) === false) { + if (attempt < maxSnapshotAttemptsPerRound) { + await wait(attempt * 1000) + } + continue + } + + const hasOverlap = await this.waitForSnapshotOverlap(normalizedSymbol, data.lastUpdateId) + + if (shouldCancel()) { + return + } + + if (hasOverlap === false) { + this.trimBufferedUpdates(normalizedSymbol) + if (attempt < maxSnapshotAttemptsPerRound) { + await wait(attempt * 1000) + } + continue + } + + if (hasOverlap === true || attempt === maxSnapshotAttemptsPerRound) { + this.manualSnapshotsBuffer.push(this.createManualSnapshot(symbol, data)) + this.pendingDepthSnapshotSymbols.delete(normalizedSymbol) + this.bufferedDepthUpdates.delete(normalizedSymbol) + return + } + } + } + + throw new Error(`AsterRealTimeFeed could not align depth snapshot for ${normalizedSymbol}`) + } + + private async waitForSnapshotOverlap(symbol: string, lastUpdateId: number) { + let hasOverlap = this.validateSnapshotOverlap(this.bufferedDepthUpdates.get(symbol), lastUpdateId) + for (let attempt = 0; attempt < 60; attempt++) { + if (hasOverlap !== undefined) { + return hasOverlap + } + + await wait(100) + hasOverlap = this.validateSnapshotOverlap(this.bufferedDepthUpdates.get(symbol), lastUpdateId) + } + + return hasOverlap + } + + protected validateSnapshotOverlap(bufferedUpdates: CircularBuffer | undefined, lastUpdateId: number) { + for (const update of bufferedUpdates?.items() ?? []) { + if (update.lastUpdateId < lastUpdateId) { + continue + } + + return update.previousFinalUpdateId <= lastUpdateId && update.lastUpdateId >= lastUpdateId + } + + return undefined + } + + private trimBufferedUpdates(symbol: string) { + const bufferedUpdates = this.bufferedDepthUpdates.get(symbol) + if (bufferedUpdates === undefined || bufferedUpdates.count <= 100) { + return + } + + const trimmed = new CircularBuffer(2000) + for (const update of [...bufferedUpdates.items()].slice(-100)) { + trimmed.append(update) + } + this.bufferedDepthUpdates.set(symbol, trimmed) + } + + private snapshotResponseIsValid(data: AsterDepthSnapshotData) { + return Number.isFinite(data.lastUpdateId) && Array.isArray(data.asks) && Array.isArray(data.bids) + } + + private createManualSnapshot(symbol: string, data: AsterDepthSnapshotData): AsterDepthSnapshotMessage { + return { + stream: `${symbol.toLowerCase()}@${AsterRealTimeFeed.depthSnapshotChannel}`, + generated: true, + data + } + } +} + +type AsterDepthSnapshotData = { + lastUpdateId: number + bids: string[][] + asks: string[][] +} + +type AsterDepthSnapshotMessage = { + stream: string + generated: true + data: AsterDepthSnapshotData +} + +type AsterDepthUpdateData = { + lastUpdateId: number + previousFinalUpdateId: number +} diff --git a/src/realtimefeeds/index.ts b/src/realtimefeeds/index.ts index 9560552..eabbe72 100644 --- a/src/realtimefeeds/index.ts +++ b/src/realtimefeeds/index.ts @@ -1,4 +1,5 @@ import { Exchange, Filter } from '../types.ts' +import { AsterRealTimeFeed } from './aster.ts' import { BinanceFuturesRealTimeFeed, BinanceJerseyRealTimeFeed, @@ -62,6 +63,7 @@ export * from './realtimefeed.ts' const realTimeFeedsMap: { [key in Exchange]?: RealTimeFeed } = { + aster: AsterRealTimeFeed, bitmex: BitmexRealTimeFeed, binance: BinanceRealTimeFeed, 'binance-jersey': BinanceJerseyRealTimeFeed, diff --git a/test/__snapshots__/mappers.test.snapshot b/test/__snapshots__/mappers.test.snapshot index 1ed79d1..835d350 100644 --- a/test/__snapshots__/mappers.test.snapshot +++ b/test/__snapshots__/mappers.test.snapshot @@ -3800,6 +3800,108 @@ exports[`mappers > map ascendex messages 8`] = ` ] `; +exports[`mappers > map aster messages 1`] = ` +[ + { + "type": "trade", + "symbol": "BTCUSDT", + "exchange": "aster", + "id": "12345", + "price": 5.1, + "amount": 0.2, + "side": "sell", + "timestamp": "2020-06-04T09:00:35.999Z", + "localTimestamp": "2026-07-29T00:00:01.000Z" + } +] +`; + +exports[`mappers > map aster messages 2`] = ` +[ + { + "type": "book_change", + "symbol": "BTCUSDT", + "exchange": "aster", + "isSnapshot": true, + "bids": [ + { + "price": 5, + "amount": 1.2 + } + ], + "asks": [ + { + "price": 5.2, + "amount": 2.4 + } + ], + "timestamp": "2020-06-04T09:00:36.000Z", + "localTimestamp": "2026-07-29T00:00:01.000Z" + } +] +`; + +exports[`mappers > map aster messages 3`] = ` +[ + { + "type": "book_change", + "symbol": "BTCUSDT", + "exchange": "aster", + "isSnapshot": false, + "bids": [ + { + "price": 5.1, + "amount": 3 + } + ], + "asks": [ + { + "price": 5.3, + "amount": 4 + } + ], + "timestamp": "2020-06-04T09:00:36.100Z", + "localTimestamp": "2026-07-29T00:00:01.000Z" + } +] +`; + +exports[`mappers > map aster messages 4`] = ` +[ + { + "type": "book_change", + "symbol": "BTCUSDT", + "exchange": "aster", + "isSnapshot": false, + "bids": [ + { + "price": 5.4, + "amount": 6 + } + ], + "asks": [], + "timestamp": "2020-06-04T09:00:36.200Z", + "localTimestamp": "2026-07-29T00:00:01.000Z" + } +] +`; + +exports[`mappers > map aster messages 5`] = ` +[ + { + "type": "book_ticker", + "symbol": "BTCUSDT", + "exchange": "aster", + "askAmount": 2.4, + "askPrice": 5.2, + "bidPrice": 5.1, + "bidAmount": 1.2, + "timestamp": "2026-07-29T00:00:01.000Z", + "localTimestamp": "2026-07-29T00:00:01.000Z" + } +] +`; + exports[`mappers > map binance delivery messages 1`] = ` [] `; diff --git a/test/aster-realtimefeed.test.ts b/test/aster-realtimefeed.test.ts new file mode 100644 index 0000000..5e0cac3 --- /dev/null +++ b/test/aster-realtimefeed.test.ts @@ -0,0 +1,257 @@ +import type { AddressInfo } from 'net' +import { createServer } from 'http' +import { test } from 'node:test' +import { assert, errorMessageIncludes } from './assertions.ts' +import { AsterRealTimeFeed } from '../dist/realtimefeeds/aster.js' +import { getRealTimeFeedFactory } from '../dist/realtimefeeds/index.js' +import type { Filter } from '../dist/index.js' + +class TestAsterRealTimeFeed extends AsterRealTimeFeed { + protected readonly httpURL: string + + constructor( + exchange: 'aster', + filters: Filter[], + timeoutIntervalMS: number | undefined, + httpURL = 'https://sapi.asterdex.com/api/v3' + ) { + super(exchange, filters, timeoutIntervalMS) + this.httpURL = httpURL + } + + map(filters: Filter[]) { + return this.mapToSubscribeMessages(filters) + } + + observe(message: any) { + this.onMessage(message) + } + + async provideSnapshots(filters: Filter[], shouldCancel = () => false) { + await this.provideManualSnapshots(filters, shouldCancel) + return this.manualSnapshotsBuffer + } +} + +test('register aster realtime feeds', () => { + assert.ok(getRealTimeFeedFactory('aster')) +}) + +test('map aster realtime subscriptions', () => { + const feed = new TestAsterRealTimeFeed('aster', [], undefined) + + assert.deepEqual( + feed.map([ + { + channel: 'depth', + symbols: ['btcusdt'] + }, + { + channel: 'depthSnapshot', + symbols: ['btcusdt'] + }, + { + channel: 'trade', + symbols: ['btcusdt', 'ethusdt'] + }, + { + channel: 'aggTrade', + symbols: ['btcusdt'] + } + ]), + [ + { + method: 'SUBSCRIBE', + params: ['btcusdt@depth@100ms'], + id: 1 + }, + { + method: 'SUBSCRIBE', + params: ['btcusdt@trade', 'ethusdt@trade'], + id: 2 + }, + { + method: 'SUBSCRIBE', + params: ['btcusdt@aggTrade'], + id: 3 + } + ] + ) +}) + +test('aster snapshot filters require matching depth filters', () => { + const feed = new TestAsterRealTimeFeed('aster', [], undefined) + + assert.throws( + () => + feed.map([ + { + channel: 'depthSnapshot', + symbols: ['BTCUSDT'] + } + ]), + errorMessageIncludes('AsterRealTimeFeed requires depth for every depthSnapshot symbol') + ) + + assert.throws( + () => + feed.map([ + { + channel: 'depth', + symbols: ['ETHUSDT'] + }, + { + channel: 'depthSnapshot', + symbols: ['BTCUSDT'] + } + ]), + errorMessageIncludes('AsterRealTimeFeed requires depth for every depthSnapshot symbol') + ) +}) + +test('aster realtime rejects unsupported channels', () => { + const feed = new TestAsterRealTimeFeed('aster', [], undefined) + + assert.throws( + () => + feed.map([ + { + channel: 'unsupported', + symbols: ['BTCUSDT'] + } + ]), + errorMessageIncludes('AsterRealTimeFeed unsupported channel unsupported') + ) +}) + +test('provide aster manual depth snapshots after buffered update overlaps', async () => { + const server = await startSnapshotServer([{ lastUpdateId: 100, asks: [['100.1', '1.2']], bids: [['99.9', '0.5']] }]) + const feed = new TestAsterRealTimeFeed('aster', [], undefined, server.url) + + try { + const filters = [ + { + channel: 'depth', + symbols: ['BTCUSDT'] + }, + { + channel: 'depthSnapshot', + symbols: ['BTCUSDT'] + } + ] + + feed.map(filters) + feed.observe(createDepthUpdate({ symbol: 'BTCUSDT', lastUpdateId: 101, previousFinalUpdateId: 100 })) + + const snapshots = await feed.provideSnapshots(filters) + + assert.deepEqual(snapshots, [ + { + stream: 'btcusdt@depthSnapshot', + generated: true, + data: { + lastUpdateId: 100, + asks: [['100.1', '1.2']], + bids: [['99.9', '0.5']] + } + } + ]) + } finally { + await server.close() + } +}) + +test('retry aster manual depth snapshots until buffered update overlaps', async () => { + const server = await startSnapshotServer([ + { lastUpdateId: 102, asks: [['100.1', '1.2']], bids: [['99.9', '0.5']] }, + { lastUpdateId: 104, asks: [['100.2', '1.2']], bids: [['99.8', '0.5']] } + ]) + const feed = new TestAsterRealTimeFeed('aster', [], undefined, server.url) + + try { + const filters = [ + { + channel: 'depth', + symbols: ['BTCUSDT'] + }, + { + channel: 'depthSnapshot', + symbols: ['BTCUSDT'] + } + ] + + feed.map(filters) + feed.observe(createDepthUpdate({ symbol: 'BTCUSDT', lastUpdateId: 105, previousFinalUpdateId: 104 })) + + const snapshots = await feed.provideSnapshots(filters) + + assert.equal(server.requestsCount, 2) + assert.deepEqual(snapshots, [ + { + stream: 'btcusdt@depthSnapshot', + generated: true, + data: { + lastUpdateId: 104, + asks: [['100.2', '1.2']], + bids: [['99.8', '0.5']] + } + } + ]) + } finally { + await server.close() + } +}) + +function createDepthUpdate({ + symbol, + lastUpdateId, + previousFinalUpdateId +}: { + symbol: string + lastUpdateId: number + previousFinalUpdateId: number +}) { + return { + stream: `${symbol.toLowerCase()}@depth@100ms`, + data: { + e: 'depthUpdate', + E: 1785230774524, + T: 1785230774522, + s: symbol, + U: previousFinalUpdateId + 1, + u: lastUpdateId, + pu: previousFinalUpdateId, + b: [], + a: [] + } + } +} + +async function startSnapshotServer(responses: AsterTestDepthSnapshotResponse[]) { + let requestsCount = 0 + const server = createServer((request, response) => { + assert.equal(request.url, '/depth?symbol=BTCUSDT&limit=1000') + const body = responses[Math.min(requestsCount, responses.length - 1)] + requestsCount++ + + response.writeHead(200, { 'Content-Type': 'application/json' }) + response.end(JSON.stringify(body)) + }) + + await new Promise((resolve) => server.listen(0, resolve)) + const { port } = server.address() as AddressInfo + + return { + url: `http://127.0.0.1:${port}`, + get requestsCount() { + return requestsCount + }, + close: () => new Promise((resolve, reject) => server.close((error) => (error ? reject(error) : resolve()))) + } +} + +type AsterTestDepthSnapshotResponse = { + lastUpdateId: number + bids: string[][] + asks: string[][] +} diff --git a/test/mappers.test.ts b/test/mappers.test.ts index aaadab9..cab28bf 100644 --- a/test/mappers.test.ts +++ b/test/mappers.test.ts @@ -44,6 +44,7 @@ const exchangesWithDerivativeInfo: Exchange[] = [ ] const exchangesWithBookTickerInfo: Exchange[] = [ + 'aster', 'ascendex', 'binance', 'binance-futures', @@ -2849,6 +2850,125 @@ describe('mappers', () => { } }) + test('map aster messages', () => { + const asterMapper = createMapper('aster', new Date()) + const localTimestamp = new Date('2026-07-29T00:00:01.000Z') + + snapshot( + asterMapper.map( + { + stream: 'btcusdt@trade', + data: { + e: 'trade', + E: 1591261236000, + s: 'BTCUSDT', + t: 12345, + p: '5.1', + q: '0.2', + T: 1591261235999, + m: true + } + }, + localTimestamp + ) + ) + + snapshot( + asterMapper.map( + { + stream: 'btcusdt@depthSnapshot', + generated: true, + data: { + lastUpdateId: 100, + bids: [['5.0', '1.2']], + asks: [['5.2', '2.4']], + T: 1591261236000 + } + }, + localTimestamp + ) + ) + + snapshot( + asterMapper.map( + { + stream: 'btcusdt@depth@100ms', + data: { + e: 'depthUpdate', + E: 1591261236100, + T: 1591261236099, + s: 'BTCUSDT', + U: 120, + u: 122, + pu: 100, + b: [['5.1', '3']], + a: [['5.3', '4']] + } + }, + localTimestamp + ) + ) + + snapshot( + asterMapper.map( + { + stream: 'btcusdt@depth@100ms', + data: { + e: 'depthUpdate', + E: 1591261236200, + T: 1591261236199, + s: 'BTCUSDT', + U: 130, + u: 131, + pu: 122, + b: [['5.4', '6']], + a: [] + } + }, + localTimestamp + ) + ) + + assert.throws( + () => + Array.from( + asterMapper.map( + { + stream: 'btcusdt@depth@100ms', + data: { + e: 'depthUpdate', + E: 1591261236300, + T: 1591261236299, + s: 'BTCUSDT', + U: 140, + u: 141, + pu: 129, + b: [], + a: [] + } + }, + localTimestamp + )! + ), + errorMessageIncludes('Book depth update has a sequence gap') + ) + snapshot( + asterMapper.map( + { + stream: 'btcusdt@bookTicker', + data: { + u: 400900217, + s: 'BTCUSDT', + b: '5.1', + B: '1.2', + a: '5.2', + A: '2.4' + } + }, + localTimestamp + ) + ) + }) test('map bitfinex derivatives book ticker messages with trailing null placeholder', () => { const bitfinexDerivativesMapper = createMapper('bitfinex-derivatives') const mappedMessages = bitfinexDerivativesMapper.map(