From a73cf6a478bf74a7bc28b378122dabb2bf4e5f5c Mon Sep 17 00:00:00 2001 From: marcinvalas Date: Thu, 30 Jul 2026 16:03:14 +0200 Subject: [PATCH 1/7] feat(aster-integration): added reply mappers --- src/consts.ts | 4 + src/mappers/aster.ts | 232 +++++++++++++++++++++++++++++++++++++++++++ src/mappers/index.ts | 4 + test/mappers.test.ts | 123 +++++++++++++++++++++++ 4 files changed, 363 insertions(+) create mode 100644 src/mappers/aster.ts diff --git a/src/consts.ts b/src/consts.ts index e9a38071..612b734c 100644 --- a/src/consts.ts +++ b/src/consts.ts @@ -1,4 +1,5 @@ export const EXCHANGES = [ + 'aster', 'bitmex', 'deribit', 'binance-futures', @@ -59,6 +60,8 @@ export const EXCHANGES = [ ] as const const BINANCE_CHANNELS = ['trade', 'aggTrade', 'ticker', 'depth', 'depthSnapshot', 'bookTicker', 'recentTrades', 'borrowInterest'] as const + +const ASTER_CHANNELS = ['trade', '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 @@ -518,6 +521,7 @@ const COINBASE_INTERNATIONAL_CHANNELS = ['INSTRUMENTS', 'MATCH', 'FUNDING', 'RIS const HYPERLIQUID_CHANNELS = ['l2Book', 'trades', 'activeAssetCtx', 'activeSpotAssetCtx', 'bbo'] 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 00000000..95dd7450 --- /dev/null +++ b/src/mappers/aster.ts @@ -0,0 +1,232 @@ +import { CircularBuffer, lowerCaseSymbols } from '../handy.ts' +import { BookChange, BookTicker, Trade } from '../types.ts' +import { Mapper } from './mapper.ts' + +export 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 + } +} + +export class AsterBookChangeMapper implements Mapper<'aster', BookChange> { + private readonly symbolToDepthInfoMapping: { [key: string]: LocalDepthInfo } = {} + + constructor(private readonly ignoreBookSnapshotOverlapError: boolean) {} + + 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.U <= depthContext.lastUpdateId! + 1 && depthUpdateData.u >= depthContext.lastUpdateId! + 1) || + 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` + ) + } + } + + 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]) + } + } +} + +export 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: data.A !== undefined ? Number(data.A) : undefined, + askPrice: data.a !== undefined ? Number(data.a) : undefined, + bidPrice: data.b !== undefined ? Number(data.b) : undefined, + bidAmount: data.B !== undefined ? Number(data.B) : undefined, + 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 056b5705..9b19b88c 100644 --- a/src/mappers/index.ts +++ b/src/mappers/index.ts @@ -1,6 +1,7 @@ import { ONE_SEC_IN_MS } from '../handy.ts' import { BookChange, DerivativeTicker, Liquidation, OptionSummary, BookTicker, Trade } from '../types.ts' import { AscendexBookChangeMapper, AscendexDerivativeTickerMapper, AscendexBookTickerMapper, AscendexTradesMapper } from './ascendex.ts' +import { AsterBookChangeMapper, AsterBookTickerMapper, AsterTradesMapper } from './aster.ts' import { BinanceBookChangeMapper, BinanceFuturesBookChangeMapper, @@ -236,6 +237,7 @@ const shouldUseBinanceEuropeanOptionsV2Mappers = (localTimestamp: Date) => { } const tradesMappers = { + aster: () => new AsterTradesMapper(), bitmex: () => bitmexTradesMapper, binance: () => new BinanceTradesMapper('binance'), 'binance-us': () => new BinanceTradesMapper('binance-us'), @@ -320,6 +322,7 @@ const tradesMappers = { } const bookChangeMappers = { + aster: (localTimestamp: Date) => new AsterBookChangeMapper(shouldIgnoreBookSnapshotOverlap(localTimestamp)), bitmex: () => new BitmexBookChangeMapper(), binance: (localTimestamp: Date) => new BinanceBookChangeMapper('binance', shouldIgnoreBookSnapshotOverlap(localTimestamp)), 'binance-us': (localTimestamp: Date) => new BinanceBookChangeMapper('binance-us', shouldIgnoreBookSnapshotOverlap(localTimestamp)), @@ -489,6 +492,7 @@ const liquidationsMappers = { } const bookTickersMappers = { + aster: () => new AsterBookTickerMapper(), binance: () => new BinanceBookTickerMapper('binance'), 'binance-futures': () => new BinanceBookTickerMapper('binance-futures'), 'binance-delivery': () => new BinanceBookTickerMapper('binance-delivery'), diff --git a/test/mappers.test.ts b/test/mappers.test.ts index 7c202476..615e87e6 100644 --- a/test/mappers.test.ts +++ b/test/mappers.test.ts @@ -39,6 +39,7 @@ const exchangesWithDerivativeInfo: Exchange[] = [ ] const exchangesWithBookTickerInfo: Exchange[] = [ + 'aster', 'ascendex', 'binance', 'binance-futures', @@ -2560,6 +2561,128 @@ describe('mappers', () => { } }) + test('map aster messages', () => { + const asterMapper = createMapper('aster', new Date('2026-07-29T00:00:00.000Z')) + const localTimestamp = new Date('2026-07-29T00:00:01.000Z') + + expect( + 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 + ) + ).toEqual([ + { + type: 'trade', + symbol: 'BTCUSDT', + exchange: 'aster', + id: '12345', + price: 5.1, + amount: 0.2, + side: 'sell', + timestamp: new Date('2020-06-04T09:00:35.999Z'), + localTimestamp + } + ]) + + expect( + asterMapper.map( + { + stream: 'btcusdt@depthSnapshot', + generated: true, + data: { + lastUpdateId: 100, + bids: [['5.0', '1.2']], + asks: [['5.2', '2.4']], + T: 1591261236000 + } + }, + localTimestamp + ) + ).toEqual([ + { + type: 'book_change', + symbol: 'BTCUSDT', + exchange: 'aster', + isSnapshot: true, + bids: [{ price: 5, amount: 1.2 }], + asks: [{ price: 5.2, amount: 2.4 }], + timestamp: new Date('2020-06-04T09:00:36.000Z'), + localTimestamp + } + ]) + + expect( + asterMapper.map( + { + stream: 'btcusdt@depth@100ms', + data: { + e: 'depthUpdate', + E: 1591261236100, + T: 1591261236099, + s: 'BTCUSDT', + U: 101, + u: 102, + pu: 100, + b: [['5.1', '3']], + a: [['5.3', '4']] + } + }, + localTimestamp + ) + ).toEqual([ + { + type: 'book_change', + symbol: 'BTCUSDT', + exchange: 'aster', + isSnapshot: false, + bids: [{ price: 5.1, amount: 3 }], + asks: [{ price: 5.3, amount: 4 }], + timestamp: new Date('2020-06-04T09:00:36.100Z'), + localTimestamp + } + ]) + + expect( + asterMapper.map( + { + stream: 'btcusdt@bookTicker', + data: { + u: 400900217, + s: 'BTCUSDT', + b: '5.1', + B: '1.2', + a: '5.2', + A: '2.4' + } + }, + localTimestamp + ) + ).toEqual([ + { + type: 'book_ticker', + symbol: 'BTCUSDT', + exchange: 'aster', + askAmount: 2.4, + askPrice: 5.2, + bidPrice: 5.1, + bidAmount: 1.2, + timestamp: localTimestamp, + localTimestamp + } + ]) + }) test('map binance messages', () => { const messages = [ { From 5cd408194d08873ceb6f3f1bc7f1f5a57068b180 Mon Sep 17 00:00:00 2001 From: marcinvalas Date: Fri, 31 Jul 2026 11:23:52 +0200 Subject: [PATCH 2/7] feat(aster-integration): added streaming --- src/mappers/aster.ts | 10 ++++- src/realtimefeeds/aster.ts | 81 ++++++++++++++++++++++++++++++++++++++ src/realtimefeeds/index.ts | 2 + test/mappers.test.ts | 58 +++++++++++++++++++++++++-- 4 files changed, 147 insertions(+), 4 deletions(-) create mode 100644 src/realtimefeeds/aster.ts diff --git a/src/mappers/aster.ts b/src/mappers/aster.ts index f11fb41f..d2fba714 100644 --- a/src/mappers/aster.ts +++ b/src/mappers/aster.ts @@ -142,7 +142,7 @@ class AsterBookChangeMapper implements Mapper<'aster', BookChange> { if (!depthContext.validatedFirstUpdate) { if ( - (depthUpdateData.U <= depthContext.lastUpdateId! + 1 && depthUpdateData.u >= depthContext.lastUpdateId! + 1) || + (depthUpdateData.pu <= depthContext.lastUpdateId! && depthUpdateData.u >= depthContext.lastUpdateId!) || depthContext.lastUpdateId! == -1 ) { depthContext.validatedFirstUpdate = true @@ -155,8 +155,16 @@ class AsterBookChangeMapper implements Mapper<'aster', BookChange> { )}, 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, diff --git a/src/realtimefeeds/aster.ts b/src/realtimefeeds/aster.ts new file mode 100644 index 00000000..04d73eec --- /dev/null +++ b/src/realtimefeeds/aster.ts @@ -0,0 +1,81 @@ +import { batch, getJSON, wait } from '../handy.ts' +import { Filter } from '../types.ts' +import { RealTimeFeedBase } from './realtimefeed.ts' + +export class AsterRealTimeFeed extends RealTimeFeedBase { + protected readonly wssURL = 'wss://sstream.asterdex.com/stream' + protected readonly httpURL = 'https://sapi.asterdex.com/api/v3' + private readonly channelMappings: { [key: string]: string | undefined } = { + depth: 'depth@100ms' + } + + protected mapToSubscribeMessages(filters: Filter[]): any[] { + return filters + .filter((f) => f.channel !== 'depthSnapshot') + .map((filter, index) => { + if (!filter.symbols || filter.symbols.length === 0) { + throw new Error('AsterRealTimeFeed requires explicitly specified symbols when subscribing to live feed') + } + + const channel = this.channelMappings[filter.channel] ?? filter.channel + + return { + method: 'SUBSCRIBE', + params: filter.symbols.map((symbol) => `${symbol.toLowerCase()}@${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 async provideManualSnapshots(filters: Filter[], shouldCancel: () => boolean) { + const depthSnapshotFilter = filters.find((f) => f.channel === 'depthSnapshot') + 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 + } + + const depthSnapshotResponse = await getJSON(`${this.httpURL}/depth?symbol=${symbol}&limit=1000`) + + this.manualSnapshotsBuffer.push({ + stream: `${symbol.toLowerCase()}@depthSnapshot`, + generated: true, + data: depthSnapshotResponse.data + }) + }) + ) + + await wait(100) + this.debug('requested manual snapshots successfully for: %s', symbolsBatch) + } + + this.debug('requested all manual snapshots successfully') + } +} diff --git a/src/realtimefeeds/index.ts b/src/realtimefeeds/index.ts index d69942eb..1f48dcad 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/mappers.test.ts b/test/mappers.test.ts index 85f80cde..c9bf0fbd 100644 --- a/test/mappers.test.ts +++ b/test/mappers.test.ts @@ -2630,7 +2630,7 @@ describe('mappers', () => { }) test('map aster messages', () => { - const asterMapper = createMapper('aster', new Date('2026-07-29T00:00:00.000Z')) + const asterMapper = createMapper('aster', new Date()) const localTimestamp = new Date('2026-07-29T00:00:01.000Z') expect( @@ -2700,8 +2700,8 @@ describe('mappers', () => { E: 1591261236100, T: 1591261236099, s: 'BTCUSDT', - U: 101, - u: 102, + U: 120, + u: 122, pu: 100, b: [['5.1', '3']], a: [['5.3', '4']] @@ -2722,6 +2722,58 @@ describe('mappers', () => { } ]) + expect( + 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 + ) + ).toEqual([ + { + type: 'book_change', + symbol: 'BTCUSDT', + exchange: 'aster', + isSnapshot: false, + bids: [{ price: 5.4, amount: 6 }], + asks: [], + timestamp: new Date('2020-06-04T09:00:36.200Z'), + localTimestamp + } + ]) + + expect(() => + 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 + )! + ) + ).toThrow('Book depth update has a sequence gap') expect( asterMapper.map( { From b9bed4d569b86a52cda2f8a9c8d3b0f414e37581 Mon Sep 17 00:00:00 2001 From: marcin Date: Mon, 3 Aug 2026 18:01:39 +0200 Subject: [PATCH 3/7] feat(aster-integration): cleanup --- src/mappers/aster.ts | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/src/mappers/aster.ts b/src/mappers/aster.ts index d2fba714..f324a514 100644 --- a/src/mappers/aster.ts +++ b/src/mappers/aster.ts @@ -1,4 +1,4 @@ -import { CircularBuffer, lowerCaseSymbols } from '../handy.ts' +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' @@ -198,10 +198,10 @@ class AsterBookTickerMapper implements Mapper<'aster', BookTicker> { type: 'book_ticker', symbol: data.s, exchange: 'aster', - askAmount: data.A !== undefined ? Number(data.A) : undefined, - askPrice: data.a !== undefined ? Number(data.a) : undefined, - bidPrice: data.b !== undefined ? Number(data.b) : undefined, - bidAmount: data.B !== undefined ? Number(data.B) : undefined, + 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 } From f54e14b2e7287b027c83771ba098e6535303788d Mon Sep 17 00:00:00 2001 From: marcin Date: Wed, 5 Aug 2026 20:31:58 +0200 Subject: [PATCH 4/7] feat(aster-integration): improved snapshot validation --- src/realtimefeeds/aster.ts | 228 +++++++++++++++++++++++++++--- test/aster-realtimefeed.test.ts | 239 ++++++++++++++++++++++++++++++++ 2 files changed, 447 insertions(+), 20 deletions(-) create mode 100644 test/aster-realtimefeed.test.ts diff --git a/src/realtimefeeds/aster.ts b/src/realtimefeeds/aster.ts index 04d73eec..eaed920a 100644 --- a/src/realtimefeeds/aster.ts +++ b/src/realtimefeeds/aster.ts @@ -1,27 +1,49 @@ -import { batch, getJSON, wait } from '../handy.ts' +import { batch, CircularBuffer, getJSON, wait } from '../handy.ts' import { Filter } from '../types.ts' import { RealTimeFeedBase } from './realtimefeed.ts' export class AsterRealTimeFeed extends RealTimeFeedBase { - protected readonly wssURL = 'wss://sstream.asterdex.com/stream' - protected readonly httpURL = 'https://sapi.asterdex.com/api/v3' + private static readonly depthChannel = 'depth' + private static readonly depthSnapshotChannel = 'depthSnapshot' + private static readonly depthStream = 'depth@100ms' + private readonly pendingDepthSnapshotSymbols = new Set() + 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', + 'ticker', + AsterRealTimeFeed.depthChannel, + AsterRealTimeFeed.depthSnapshotChannel, + 'bookTicker' + ]) private readonly channelMappings: { [key: string]: string | undefined } = { - depth: 'depth@100ms' + [AsterRealTimeFeed.depthChannel]: AsterRealTimeFeed.depthStream } protected mapToSubscribeMessages(filters: Filter[]): any[] { - return filters - .filter((f) => f.channel !== 'depthSnapshot') - .map((filter, index) => { - if (!filter.symbols || filter.symbols.length === 0) { - throw new Error('AsterRealTimeFeed requires explicitly specified symbols when subscribing to live feed') - } + 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 channel = this.channelMappings[filter.channel] ?? filter.channel + 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()}@${channel}`), + params: filter.symbols.map((symbol) => `${symbol.toLowerCase()}@${this.channelMappings[filter.channel] ?? filter.channel}`), id: index + 1 } }) @@ -43,8 +65,32 @@ export class AsterRealTimeFeed extends RealTimeFeedBase { 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 === 'depthSnapshot') + const depthSnapshotFilter = filters.find((f) => f.channel === AsterRealTimeFeed.depthSnapshotChannel) if (!depthSnapshotFilter) { return } @@ -62,13 +108,7 @@ export class AsterRealTimeFeed extends RealTimeFeedBase { return } - const depthSnapshotResponse = await getJSON(`${this.httpURL}/depth?symbol=${symbol}&limit=1000`) - - this.manualSnapshotsBuffer.push({ - stream: `${symbol.toLowerCase()}@depthSnapshot`, - generated: true, - data: depthSnapshotResponse.data - }) + await this.provideManualSnapshot(symbol, shouldCancel) }) ) @@ -78,4 +118,152 @@ export class AsterRealTimeFeed extends RealTimeFeedBase { 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/test/aster-realtimefeed.test.ts b/test/aster-realtimefeed.test.ts new file mode 100644 index 00000000..c93ad8c1 --- /dev/null +++ b/test/aster-realtimefeed.test.ts @@ -0,0 +1,239 @@ +import type { AddressInfo } from 'net' +import { createServer } from 'http' +import { AsterRealTimeFeed } from '../src/realtimefeeds/aster.ts' +import { getRealTimeFeedFactory } from '../src/realtimefeeds/index.ts' +import { Filter } from '../src/types.ts' + +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', () => { + expect(getRealTimeFeedFactory('aster')).toBeDefined() +}) + +test('map aster realtime subscriptions', () => { + const feed = new TestAsterRealTimeFeed('aster', [], undefined) + + expect( + feed.map([ + { + channel: 'depth', + symbols: ['btcusdt'] + }, + { + channel: 'depthSnapshot', + symbols: ['btcusdt'] + }, + { + channel: 'trade', + symbols: ['btcusdt', 'ethusdt'] + } + ]) + ).toEqual([ + { + method: 'SUBSCRIBE', + params: ['btcusdt@depth@100ms'], + id: 1 + }, + { + method: 'SUBSCRIBE', + params: ['btcusdt@trade', 'ethusdt@trade'], + id: 2 + } + ]) +}) + +test('aster snapshot filters require matching depth filters', () => { + const feed = new TestAsterRealTimeFeed('aster', [], undefined) + + expect(() => + feed.map([ + { + channel: 'depthSnapshot', + symbols: ['BTCUSDT'] + } + ]) + ).toThrow('AsterRealTimeFeed requires depth for every depthSnapshot symbol') + + expect(() => + feed.map([ + { + channel: 'depth', + symbols: ['ETHUSDT'] + }, + { + channel: 'depthSnapshot', + symbols: ['BTCUSDT'] + } + ]) + ).toThrow('AsterRealTimeFeed requires depth for every depthSnapshot symbol') +}) + +test('aster realtime rejects unsupported channels', () => { + const feed = new TestAsterRealTimeFeed('aster', [], undefined) + + expect(() => + feed.map([ + { + channel: 'unsupported', + symbols: ['BTCUSDT'] + } + ]) + ).toThrow('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) + + expect(snapshots).toEqual([ + { + 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) + + expect(server.requestsCount).toBe(2) + expect(snapshots).toEqual([ + { + 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) => { + expect(request.url).toBe('/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[][] +} From 7b2c43258e788cedc31e1191b897d72e606ec01b Mon Sep 17 00:00:00 2001 From: marcin Date: Wed, 19 Aug 2026 12:31:21 +0200 Subject: [PATCH 5/7] feat(aster-integration): fixed aster exposed channel --- src/consts.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/consts.ts b/src/consts.ts index 6090dad7..b506452d 100644 --- a/src/consts.ts +++ b/src/consts.ts @@ -66,7 +66,7 @@ export const EXCHANGES = [ const BINANCE_CHANNELS = ['trade', 'aggTrade', 'ticker', 'depth', 'depthSnapshot', 'bookTicker', 'recentTrades', 'borrowInterest'] as const -const ASTER_CHANNELS = ['trade', 'ticker', 'depth', 'depthSnapshot', 'bookTicker'] 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 From 9a578404090edfa7600ba733f2aa830f0081182a Mon Sep 17 00:00:00 2001 From: marcin Date: Wed, 19 Aug 2026 13:18:19 +0200 Subject: [PATCH 6/7] feat(aster-integration): fixed aster tests after back merge --- test/__snapshots__/mappers.test.snapshot | 102 +++++++++++++++++++ test/aster-realtimefeed.test.ts | 111 ++++++++++---------- test/mappers.test.ts | 123 +++++++---------------- 3 files changed, 196 insertions(+), 140 deletions(-) diff --git a/test/__snapshots__/mappers.test.snapshot b/test/__snapshots__/mappers.test.snapshot index 1ed79d17..835d3505 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 index c93ad8c1..76bfb40d 100644 --- a/test/aster-realtimefeed.test.ts +++ b/test/aster-realtimefeed.test.ts @@ -1,8 +1,10 @@ import type { AddressInfo } from 'net' import { createServer } from 'http' -import { AsterRealTimeFeed } from '../src/realtimefeeds/aster.ts' -import { getRealTimeFeedFactory } from '../src/realtimefeeds/index.ts' -import { Filter } from '../src/types.ts' +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 @@ -32,13 +34,13 @@ class TestAsterRealTimeFeed extends AsterRealTimeFeed { } test('register aster realtime feeds', () => { - expect(getRealTimeFeedFactory('aster')).toBeDefined() + assert.ok(getRealTimeFeedFactory('aster')) }) test('map aster realtime subscriptions', () => { const feed = new TestAsterRealTimeFeed('aster', [], undefined) - expect( + assert.deepEqual( feed.map([ { channel: 'depth', @@ -52,58 +54,65 @@ test('map aster realtime subscriptions', () => { channel: 'trade', symbols: ['btcusdt', 'ethusdt'] } - ]) - ).toEqual([ - { - method: 'SUBSCRIBE', - params: ['btcusdt@depth@100ms'], - id: 1 - }, - { - method: 'SUBSCRIBE', - params: ['btcusdt@trade', 'ethusdt@trade'], - id: 2 - } - ]) + ]), + [ + { + method: 'SUBSCRIBE', + params: ['btcusdt@depth@100ms'], + id: 1 + }, + { + method: 'SUBSCRIBE', + params: ['btcusdt@trade', 'ethusdt@trade'], + id: 2 + } + ] + ) }) test('aster snapshot filters require matching depth filters', () => { const feed = new TestAsterRealTimeFeed('aster', [], undefined) - expect(() => - feed.map([ - { - channel: 'depthSnapshot', - symbols: ['BTCUSDT'] - } - ]) - ).toThrow('AsterRealTimeFeed requires depth for every depthSnapshot symbol') - - expect(() => - feed.map([ - { - channel: 'depth', - symbols: ['ETHUSDT'] - }, - { - channel: 'depthSnapshot', - symbols: ['BTCUSDT'] - } - ]) - ).toThrow('AsterRealTimeFeed requires depth for every depthSnapshot symbol') + 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) - expect(() => - feed.map([ - { - channel: 'unsupported', - symbols: ['BTCUSDT'] - } - ]) - ).toThrow('AsterRealTimeFeed unsupported channel unsupported') + assert.throws( + () => + feed.map([ + { + channel: 'unsupported', + symbols: ['BTCUSDT'] + } + ]), + errorMessageIncludes('AsterRealTimeFeed unsupported channel unsupported') + ) }) test('provide aster manual depth snapshots after buffered update overlaps', async () => { @@ -127,7 +136,7 @@ test('provide aster manual depth snapshots after buffered update overlaps', asyn const snapshots = await feed.provideSnapshots(filters) - expect(snapshots).toEqual([ + assert.deepEqual(snapshots, [ { stream: 'btcusdt@depthSnapshot', generated: true, @@ -167,8 +176,8 @@ test('retry aster manual depth snapshots until buffered update overlaps', async const snapshots = await feed.provideSnapshots(filters) - expect(server.requestsCount).toBe(2) - expect(snapshots).toEqual([ + assert.equal(server.requestsCount, 2) + assert.deepEqual(snapshots, [ { stream: 'btcusdt@depthSnapshot', generated: true, @@ -212,7 +221,7 @@ function createDepthUpdate({ async function startSnapshotServer(responses: AsterTestDepthSnapshotResponse[]) { let requestsCount = 0 const server = createServer((request, response) => { - expect(request.url).toBe('/depth?symbol=BTCUSDT&limit=1000') + assert.equal(request.url, '/depth?symbol=BTCUSDT&limit=1000') const body = responses[Math.min(requestsCount, responses.length - 1)] requestsCount++ diff --git a/test/mappers.test.ts b/test/mappers.test.ts index 1df9a8eb..cab28bf2 100644 --- a/test/mappers.test.ts +++ b/test/mappers.test.ts @@ -2854,7 +2854,7 @@ describe('mappers', () => { const asterMapper = createMapper('aster', new Date()) const localTimestamp = new Date('2026-07-29T00:00:01.000Z') - expect( + snapshot( asterMapper.map( { stream: 'btcusdt@trade', @@ -2871,21 +2871,9 @@ describe('mappers', () => { }, localTimestamp ) - ).toEqual([ - { - type: 'trade', - symbol: 'BTCUSDT', - exchange: 'aster', - id: '12345', - price: 5.1, - amount: 0.2, - side: 'sell', - timestamp: new Date('2020-06-04T09:00:35.999Z'), - localTimestamp - } - ]) + ) - expect( + snapshot( asterMapper.map( { stream: 'btcusdt@depthSnapshot', @@ -2899,20 +2887,9 @@ describe('mappers', () => { }, localTimestamp ) - ).toEqual([ - { - type: 'book_change', - symbol: 'BTCUSDT', - exchange: 'aster', - isSnapshot: true, - bids: [{ price: 5, amount: 1.2 }], - asks: [{ price: 5.2, amount: 2.4 }], - timestamp: new Date('2020-06-04T09:00:36.000Z'), - localTimestamp - } - ]) + ) - expect( + snapshot( asterMapper.map( { stream: 'btcusdt@depth@100ms', @@ -2930,20 +2907,9 @@ describe('mappers', () => { }, localTimestamp ) - ).toEqual([ - { - type: 'book_change', - symbol: 'BTCUSDT', - exchange: 'aster', - isSnapshot: false, - bids: [{ price: 5.1, amount: 3 }], - asks: [{ price: 5.3, amount: 4 }], - timestamp: new Date('2020-06-04T09:00:36.100Z'), - localTimestamp - } - ]) + ) - expect( + snapshot( asterMapper.map( { stream: 'btcusdt@depth@100ms', @@ -2961,41 +2927,32 @@ describe('mappers', () => { }, localTimestamp ) - ).toEqual([ - { - type: 'book_change', - symbol: 'BTCUSDT', - exchange: 'aster', - isSnapshot: false, - bids: [{ price: 5.4, amount: 6 }], - asks: [], - timestamp: new Date('2020-06-04T09:00:36.200Z'), - localTimestamp - } - ]) - - expect(() => - 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 - )! - ) - ).toThrow('Book depth update has a sequence gap') - expect( + ) + + 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', @@ -3010,19 +2967,7 @@ describe('mappers', () => { }, localTimestamp ) - ).toEqual([ - { - type: 'book_ticker', - symbol: 'BTCUSDT', - exchange: 'aster', - askAmount: 2.4, - askPrice: 5.2, - bidPrice: 5.1, - bidAmount: 1.2, - timestamp: localTimestamp, - localTimestamp - } - ]) + ) }) test('map bitfinex derivatives book ticker messages with trailing null placeholder', () => { const bitfinexDerivativesMapper = createMapper('bitfinex-derivatives') From 2aa3aa0223a0e57132106b293431e8beb1dd17bf Mon Sep 17 00:00:00 2001 From: marcin Date: Wed, 19 Aug 2026 13:30:58 +0200 Subject: [PATCH 7/7] feat(aster-integration): added missing agg trade streaming channel --- src/realtimefeeds/aster.ts | 1 + test/aster-realtimefeed.test.ts | 9 +++++++++ 2 files changed, 10 insertions(+) diff --git a/src/realtimefeeds/aster.ts b/src/realtimefeeds/aster.ts index eaed920a..e4b1dbe1 100644 --- a/src/realtimefeeds/aster.ts +++ b/src/realtimefeeds/aster.ts @@ -12,6 +12,7 @@ export class AsterRealTimeFeed extends RealTimeFeedBase { protected readonly httpURL: string = 'https://sapi.asterdex.com/api/v3' private readonly channels = new Set([ 'trade', + 'aggTrade', 'ticker', AsterRealTimeFeed.depthChannel, AsterRealTimeFeed.depthSnapshotChannel, diff --git a/test/aster-realtimefeed.test.ts b/test/aster-realtimefeed.test.ts index 76bfb40d..5e0cac3f 100644 --- a/test/aster-realtimefeed.test.ts +++ b/test/aster-realtimefeed.test.ts @@ -53,6 +53,10 @@ test('map aster realtime subscriptions', () => { { channel: 'trade', symbols: ['btcusdt', 'ethusdt'] + }, + { + channel: 'aggTrade', + symbols: ['btcusdt'] } ]), [ @@ -65,6 +69,11 @@ test('map aster realtime subscriptions', () => { method: 'SUBSCRIBE', params: ['btcusdt@trade', 'ethusdt@trade'], id: 2 + }, + { + method: 'SUBSCRIBE', + params: ['btcusdt@aggTrade'], + id: 3 } ] )