From 1699bcdb1a3ea093b6d39989972489e6fd09a730 Mon Sep 17 00:00:00 2001 From: marcin Date: Mon, 3 Aug 2026 14:02:13 +0200 Subject: [PATCH 1/2] feat(aster-futures-integration): added aster futures --- src/consts.ts | 3 + src/mappers/asterfutures.ts | 376 ++++++++++++++++++++++++++++++++++++ src/mappers/index.ts | 2 + src/realtimefeeds/aster.ts | 15 +- src/realtimefeeds/index.ts | 3 +- test/mappers.test.ts | 295 ++++++++++++++++++++++++++++ 6 files changed, 690 insertions(+), 4 deletions(-) create mode 100644 src/mappers/asterfutures.ts diff --git a/src/consts.ts b/src/consts.ts index 6954f7a..9b75cf8 100644 --- a/src/consts.ts +++ b/src/consts.ts @@ -1,5 +1,6 @@ export const EXCHANGES = [ 'aster', + 'aster-futures', 'bitmex', 'deribit', 'binance-futures', @@ -67,6 +68,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_FUTURES_CHANNELS = ['aggTrade', 'ticker', 'depth', 'depthSnapshot', 'bookTicker', 'markPrice', 'forceOrder'] as const const BINANCE_DEX_CHANNELS = ['trades', 'marketDiff', 'depthSnapshot', 'ticker'] as const const BITFINEX_CHANNELS = ['trades', 'book', 'raw_book', 'ticker'] as const @@ -561,6 +563,7 @@ const POLYMARKET_CHANNELS = [ export const EXCHANGE_CHANNELS_INFO = { aster: ASTER_CHANNELS, + 'aster-futures': ASTER_FUTURES_CHANNELS, bitmex: BITMEX_CHANNELS, coinbase: COINBASE_CHANNELS, 'coinbase-international': COINBASE_INTERNATIONAL_CHANNELS, diff --git a/src/mappers/asterfutures.ts b/src/mappers/asterfutures.ts new file mode 100644 index 0000000..dc94950 --- /dev/null +++ b/src/mappers/asterfutures.ts @@ -0,0 +1,376 @@ +import { debug } from '../debug.ts' +import { asNumberOrUndefined, CircularBuffer, lowerCaseSymbols } from '../handy.ts' +import { BookChange, BookTicker, DerivativeTicker, Liquidation, Trade } from '../types.ts' +import { Mapper, PendingTickerInfoHelper } from './mapper.ts' +import { exchangeMappers, isRealTime } from './registry.ts' + +export const asterFuturesMappers = exchangeMappers({ + 'aster-futures': { + trades: () => new AsterFuturesTradesMapper(), + bookChanges: (localTimestamp) => + new AsterFuturesBookChangeMapper({ + ignoreBookSnapshotOverlapError: shouldIgnoreBookSnapshotOverlap(localTimestamp) + }), + derivativeTickers: () => new AsterFuturesDerivativeTickerMapper(), + liquidations: () => new AsterFuturesLiquidationsMapper(), + bookTickers: () => new AsterFuturesBookTickerMapper() + } +}) + +function shouldIgnoreBookSnapshotOverlap(date?: Date) { + if (process.env.IGNORE_BOOK_SNAPSHOT_OVERLAP_ERROR) { + return true + } + + return isRealTime(date) === false +} + +class AsterFuturesTradesMapper implements Mapper<'aster-futures', Trade> { + canHandle(message: AsterFuturesMessage) { + return message.stream?.endsWith('@aggTrade') === true + } + + getFilters(symbols?: string[]) { + return [{ channel: 'aggTrade', symbols: lowerCaseSymbols(symbols) } as const] + } + + *map({ data }: AsterFuturesMessage, localTimestamp: Date) { + const trade: Trade = { + type: 'trade', + symbol: data.s, + exchange: 'aster-futures', + id: String(data.a), + price: Number(data.p), + amount: Number(data.q), + side: data.m ? 'sell' : 'buy', + timestamp: new Date(data.T), + localTimestamp + } + + yield trade + } +} + +class AsterFuturesBookChangeMapper implements Mapper<'aster-futures', BookChange> { + private readonly symbolToDepthInfoMapping: { [key: string]: LocalDepthInfo } = {} + private readonly ignoreBookSnapshotOverlapError: boolean + + constructor({ ignoreBookSnapshotOverlapError }: { ignoreBookSnapshotOverlapError: boolean }) { + this.ignoreBookSnapshotOverlapError = ignoreBookSnapshotOverlapError + } + + canHandle(message: AsterFuturesMessage) { + 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 }: AsterFuturesMessage, 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-futures', + 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 AsterFuturesDepthData, localTimestamp) + if (bookChange !== undefined) { + yield bookChange + } + } else { + symbolDepthInfo.bufferedUpdates.append(data as AsterFuturesDepthData) + } + } + + private mapBookDepthUpdate(depthUpdateData: AsterFuturesDepthData, localTimestamp: Date): BookChange | undefined { + const depthContext = this.symbolToDepthInfoMapping[depthUpdateData.s]! + + if (depthUpdateData.u < depthContext.lastUpdateId!) { + return + } + + if (!depthContext.validatedFirstUpdate) { + if (depthUpdateData.U <= depthContext.lastUpdateId! && depthUpdateData.u >= depthContext.lastUpdateId!) { + depthContext.validatedFirstUpdate = true + } else if (this.ignoreBookSnapshotOverlapError) { + depthContext.validatedFirstUpdate = true + debug( + `Book depth snaphot has no overlap with first update, update ${JSON.stringify( + depthUpdateData + )}, lastUpdateId: ${depthContext.lastUpdateId!}, exchange aster-futures` + ) + } else { + throw new Error( + `Book depth snaphot has no overlap with first update, update ${JSON.stringify( + depthUpdateData + )}, lastUpdateId: ${depthContext.lastUpdateId!}, exchange aster-futures` + ) + } + } 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-futures` + ) + } + + depthContext.lastUpdateId = depthUpdateData.u + + return { + type: 'book_change', + symbol: depthUpdateData.s, + exchange: 'aster-futures', + isSnapshot: false, + bids: depthUpdateData.b.map(this.mapBookLevel), + asks: depthUpdateData.a.map(this.mapBookLevel), + timestamp: new Date(depthUpdateData.E), + localTimestamp + } + } + + private mapBookLevel(level: AsterFuturesBookLevel) { + return { + price: Number(level[0]), + amount: Number(level[1]) + } + } +} + +class AsterFuturesDerivativeTickerMapper implements Mapper<'aster-futures', DerivativeTicker> { + private readonly pendingTickerInfoHelper = new PendingTickerInfoHelper() + + canHandle(message: AsterFuturesMessage) { + return message.stream?.includes('@markPrice') === true || message.stream?.endsWith('@ticker') === true + } + + getFilters(symbols?: string[]) { + return [ + { channel: 'markPrice', symbols: lowerCaseSymbols(symbols) } as const, + { channel: 'ticker', symbols: lowerCaseSymbols(symbols) } as const + ] + } + + *map( + message: AsterFuturesMessage, + localTimestamp: Date + ): IterableIterator { + const pendingTickerInfo = this.pendingTickerInfoHelper.getPendingTickerInfo(message.data.s, 'aster-futures') + + if (message.data.e === 'markPriceUpdate') { + pendingTickerInfo.updateMarkPrice(Number(message.data.p)) + if (message.data.i !== undefined) { + pendingTickerInfo.updateIndexPrice(Number(message.data.i)) + } + if (message.data.r !== '' && message.data.T !== 0) { + pendingTickerInfo.updateFundingRate(Number(message.data.r)) + pendingTickerInfo.updateFundingTimestamp(new Date(message.data.T)) + } + pendingTickerInfo.updateTimestamp(new Date(message.data.E)) + } + + if (message.data.e === '24hrTicker') { + pendingTickerInfo.updateLastPrice(Number(message.data.c)) + pendingTickerInfo.updateTimestamp(new Date(message.data.E)) + } + + if (pendingTickerInfo.hasChanged()) { + yield pendingTickerInfo.getSnapshot(localTimestamp) + } + } +} + +class AsterFuturesLiquidationsMapper implements Mapper<'aster-futures', Liquidation> { + canHandle(message: AsterFuturesMessage) { + return message.stream?.endsWith('@forceOrder') === true + } + + getFilters(symbols?: string[]) { + return [{ channel: 'forceOrder', symbols: lowerCaseSymbols(symbols) } as const] + } + + *map({ data }: AsterFuturesMessage, localTimestamp: Date) { + const order = data.o + if (order.X !== 'FILLED') { + return + } + + const liquidation: Liquidation = { + type: 'liquidation', + symbol: order.s, + exchange: 'aster-futures', + id: undefined, + price: Number(order.p), + amount: Number(order.z), + side: order.S === 'SELL' ? 'sell' : 'buy', + timestamp: new Date(order.T), + localTimestamp + } + + yield liquidation + } +} + +class AsterFuturesBookTickerMapper implements Mapper<'aster-futures', BookTicker> { + canHandle(message: AsterFuturesMessage) { + return message.stream?.endsWith('@bookTicker') === true + } + + getFilters(symbols?: string[]) { + return [{ channel: 'bookTicker', symbols: lowerCaseSymbols(symbols) } as const] + } + + *map({ data }: AsterFuturesMessage, localTimestamp: Date) { + const ticker: BookTicker = { + type: 'book_ticker', + symbol: data.s, + exchange: 'aster-futures', + 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 AsterFuturesMessage = { + stream: string + data: T +} + +type AsterFuturesAggTradeData = { + e: 'aggTrade' + E: number + s: string + a: number + p: string + q: string + f: number + l: number + T: number + m: boolean +} + +type AsterFuturesDepthData = { + lastUpdateId: undefined + E: number + T: number + s: string + U: number + u: number + pu: number + b: AsterFuturesBookLevel[] + a: AsterFuturesBookLevel[] +} + +type AsterFuturesDepthSnapshotData = { + lastUpdateId: number + bids: AsterFuturesBookLevel[] + asks: AsterFuturesBookLevel[] + T?: number +} + +type AsterFuturesBookLevel = [string, string] + +type LocalDepthInfo = { + bufferedUpdates: CircularBuffer + snapshotProcessed?: boolean + lastUpdateId?: number + validatedFirstUpdate?: boolean +} + +type AsterFuturesTickerData = { + e: '24hrTicker' + E: number + s: string + c: string +} + +type AsterFuturesMarkPriceData = { + e: 'markPriceUpdate' + E: number + s: string + p: string + i?: string + r: string + T: number +} + +type AsterFuturesForceOrderData = { + e: 'forceOrder' + E: number + o: { + s: string + S: 'BUY' | 'SELL' + p: string + X: string + z: string + T: number + } +} + +type AsterFuturesBookTickerData = { + 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 1f7e4e6..d89343a 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 { asterFuturesMappers } from './asterfutures.ts' import { asterMappers } from './aster.ts' import { binanceMappers } from './binance.ts' import { binanceDexMappers } from './binancedex.ts' @@ -56,6 +57,7 @@ type Normalizer = ( { } ]) }) + + test('map aster futures messages', () => { + const asterFuturesMapper = createMapper('aster-futures', new Date()) + const localTimestamp = new Date('2026-08-03T10:00:00.000Z') + + expect( + asterFuturesMapper.map( + { + stream: 'btcusdt@aggTrade', + data: { + e: 'aggTrade', + E: 1568693103463, + s: 'BTCUSDT', + a: 181349, + p: '10223.74', + q: '0.236', + f: 181349, + l: 181349, + T: 1568693103463, + m: false + } + }, + localTimestamp + ) + ).toEqual([ + { + type: 'trade', + symbol: 'BTCUSDT', + exchange: 'aster-futures', + id: '181349', + price: 10223.74, + amount: 0.236, + side: 'buy', + timestamp: new Date('2019-09-17T04:05:03.463Z'), + localTimestamp + } + ]) + + expect( + asterFuturesMapper.map( + { + stream: 'btcusdt@depth@100ms', + data: { + e: 'depthUpdate', + E: 1573948821952, + T: 1573948821948, + s: 'BTCUSDT', + U: 687687944, + u: 687687946, + pu: 687687943, + b: [['8493.78', '0.162']], + a: [['4096.00000000', '2.42541900']] + } + }, + localTimestamp + ) + ).toEqual([]) + + expect( + asterFuturesMapper.map( + { + stream: 'btcusdt@depthSnapshot', + generated: true, + data: { + lastUpdateId: 687687945, + bids: [['8488.36', '1.501']], + asks: [['4096.00000000', '1.42541900']] + } + }, + localTimestamp + ) + ).toEqual([ + { + type: 'book_change', + symbol: 'BTCUSDT', + exchange: 'aster-futures', + isSnapshot: true, + bids: [ + { price: 8488.36, amount: 1.501 }, + { price: 8493.78, amount: 0.162 } + ], + asks: [{ price: 4096, amount: 2.425419 }], + timestamp: localTimestamp, + localTimestamp + } + ]) + + const asterFuturesOverlapEdgeMapper = createMapper('aster-futures', new Date()) + expect( + asterFuturesOverlapEdgeMapper.map( + { + stream: 'ethusdt@depthSnapshot', + generated: true, + data: { + lastUpdateId: 200, + bids: [['10', '1']], + asks: [['11', '2']] + } + }, + localTimestamp + ) + ).toHaveLength(1) + expect( + asterFuturesOverlapEdgeMapper.map( + { + stream: 'ethusdt@depth@100ms', + data: { + e: 'depthUpdate', + E: 1573948821952, + T: 1573948821948, + s: 'ETHUSDT', + U: 199, + u: 200, + pu: 198, + b: [['10', '3']], + a: [] + } + }, + localTimestamp + ) + ).toEqual([ + { + type: 'book_change', + symbol: 'ETHUSDT', + exchange: 'aster-futures', + isSnapshot: false, + bids: [{ price: 10, amount: 3 }], + asks: [], + timestamp: new Date('2019-11-17T00:00:21.952Z'), + localTimestamp + } + ]) + + expect( + asterFuturesMapper.map( + { + stream: 'btcusdt@depth@100ms', + data: { + e: 'depthUpdate', + E: 1573948822952, + T: 1573948822948, + s: 'BTCUSDT', + U: 687687947, + u: 687687948, + pu: 687687946, + b: [['8493.78', '0']], + a: [] + } + }, + localTimestamp + ) + ).toEqual([ + { + type: 'book_change', + symbol: 'BTCUSDT', + exchange: 'aster-futures', + isSnapshot: false, + bids: [{ price: 8493.78, amount: 0 }], + asks: [], + timestamp: new Date('2019-11-17T00:00:22.952Z'), + localTimestamp + } + ]) + + expect( + asterFuturesMapper.map( + { + stream: 'btcusdt@markPrice@1s', + data: { + e: 'markPriceUpdate', + E: 1597536008000, + s: 'BTCUSDT', + p: '11857.56000000', + i: '11851.86949091', + r: '0.00015640', + T: 1597564800000 + } + }, + localTimestamp + ) + ).toEqual([ + { + type: 'derivative_ticker', + symbol: 'BTCUSDT', + exchange: 'aster-futures', + lastPrice: undefined, + openInterest: undefined, + fundingRate: 0.0001564, + fundingTimestamp: new Date('2020-08-16T08:00:00.000Z'), + predictedFundingRate: undefined, + indexPrice: 11851.86949091, + markPrice: 11857.56, + timestamp: new Date('2020-08-16T00:00:08.000Z'), + localTimestamp + } + ]) + + expect( + asterFuturesMapper.map( + { + stream: 'btcusdt@ticker', + data: { e: '24hrTicker', E: 1568693103467, s: 'BTCUSDT', c: '10223.74' } + }, + localTimestamp + ) + ).toEqual([ + { + type: 'derivative_ticker', + symbol: 'BTCUSDT', + exchange: 'aster-futures', + lastPrice: 10223.74, + openInterest: undefined, + fundingRate: 0.0001564, + fundingTimestamp: new Date('2020-08-16T08:00:00.000Z'), + predictedFundingRate: undefined, + indexPrice: 11851.86949091, + markPrice: 11857.56, + timestamp: new Date('2020-08-16T00:00:08.000Z'), + localTimestamp + } + ]) + + expect( + asterFuturesMapper.map( + { + stream: 'btcusdt@forceOrder', + data: { + e: 'forceOrder', + E: 1584059031426, + o: { + s: 'BTCUSDT', + S: 'BUY', + o: 'LIMIT', + f: 'IOC', + q: '0.014', + p: '4793.91', + ap: '4706.04', + X: 'FILLED', + l: '0.015', + z: '0.014', + T: 1584059031421 + } + } + }, + localTimestamp + ) + ).toEqual([ + { + type: 'liquidation', + symbol: 'BTCUSDT', + exchange: 'aster-futures', + id: undefined, + price: 4793.91, + amount: 0.014, + side: 'buy', + timestamp: new Date('2020-03-13T00:23:51.421Z'), + localTimestamp + } + ]) + + expect( + asterFuturesMapper.map( + { + stream: 'btcusdt@bookTicker', + data: { + e: 'bookTicker', + u: 185130926750, + s: 'BTCUSDT', + b: '33134.42', + B: '0.170', + a: '33139.39', + A: '0.380', + E: 1612137603571 + } + }, + localTimestamp + ) + ).toEqual([ + { + type: 'book_ticker', + symbol: 'BTCUSDT', + exchange: 'aster-futures', + askAmount: 0.38, + askPrice: 33139.39, + bidPrice: 33134.42, + bidAmount: 0.17, + timestamp: new Date('2021-02-01T00:00:03.571Z'), + localTimestamp + } + ]) + }) + test('map bitfinex derivatives book ticker messages with trailing null placeholder', () => { const bitfinexDerivativesMapper = createMapper('bitfinex-derivatives') const mappedMessages = bitfinexDerivativesMapper.map( From 087b37cb6c3ec74b03dd39eb665c0d7e8217186a Mon Sep 17 00:00:00 2001 From: marcin Date: Thu, 20 Aug 2026 12:02:08 +0200 Subject: [PATCH 2/2] feat(aster-futures-integration): added missing subscriptions --- src/consts.ts | 13 ++- src/mappers/asterfutures.ts | 64 ++++++++----- src/realtimefeeds/aster.ts | 87 ++++++++++++++++-- test/__snapshots__/mappers.test.snapshot | 111 +++++++++++++++-------- test/aster-realtimefeed.test.ts | 40 ++++++-- test/mappers.test.ts | 45 ++++++++- 6 files changed, 280 insertions(+), 80 deletions(-) diff --git a/src/consts.ts b/src/consts.ts index 781027c..1ff1814 100644 --- a/src/consts.ts +++ b/src/consts.ts @@ -68,7 +68,18 @@ export const EXCHANGES = [ 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 ASTER_FUTURES_CHANNELS = ['aggTrade', 'ticker', 'depth', 'depthSnapshot', 'bookTicker', 'markPrice', 'forceOrder'] as const +const ASTER_FUTURES_CHANNELS = [ + 'trade', + 'aggTrade', + 'ticker', + 'depth', + 'depthSnapshot', + 'bookTicker', + 'markPrice', + 'forceOrder', + 'assetIndex', + 'openInterest' +] as const const BINANCE_DEX_CHANNELS = ['trades', 'marketDiff', 'depthSnapshot', 'ticker'] as const const BITFINEX_CHANNELS = ['trades', 'book', 'raw_book', 'ticker'] as const diff --git a/src/mappers/asterfutures.ts b/src/mappers/asterfutures.ts index dc94950..e1cf3ff 100644 --- a/src/mappers/asterfutures.ts +++ b/src/mappers/asterfutures.ts @@ -1,5 +1,6 @@ import { debug } from '../debug.ts' import { asNumberOrUndefined, CircularBuffer, lowerCaseSymbols } from '../handy.ts' +import type { AsterFuturesOpenInterestData } from '../realtimefeeds/aster.ts' import { BookChange, BookTicker, DerivativeTicker, Liquidation, Trade } from '../types.ts' import { Mapper, PendingTickerInfoHelper } from './mapper.ts' import { exchangeMappers, isRealTime } from './registry.ts' @@ -27,19 +28,19 @@ function shouldIgnoreBookSnapshotOverlap(date?: Date) { class AsterFuturesTradesMapper implements Mapper<'aster-futures', Trade> { canHandle(message: AsterFuturesMessage) { - return message.stream?.endsWith('@aggTrade') === true + return message.stream?.endsWith('@trade') === true || message.stream?.endsWith('@aggTrade') === true } getFilters(symbols?: string[]) { - return [{ channel: 'aggTrade', symbols: lowerCaseSymbols(symbols) } as const] + return [{ channel: 'trade', symbols: lowerCaseSymbols(symbols) } as const] } - *map({ data }: AsterFuturesMessage, localTimestamp: Date) { + *map({ data }: AsterFuturesMessage, localTimestamp: Date) { const trade: Trade = { type: 'trade', symbol: data.s, exchange: 'aster-futures', - id: String(data.a), + id: String(data.e === 'trade' ? data.t : data.a), price: Number(data.p), amount: Number(data.q), side: data.m ? 'sell' : 'buy', @@ -193,37 +194,47 @@ class AsterFuturesDerivativeTickerMapper implements Mapper<'aster-futures', Deri private readonly pendingTickerInfoHelper = new PendingTickerInfoHelper() canHandle(message: AsterFuturesMessage) { - return message.stream?.includes('@markPrice') === true || message.stream?.endsWith('@ticker') === true + return ( + message.stream?.includes('@markPrice') === true || + message.stream?.endsWith('@ticker') === true || + message.stream?.endsWith('@openInterest') === true + ) } getFilters(symbols?: string[]) { return [ { channel: 'markPrice', symbols: lowerCaseSymbols(symbols) } as const, - { channel: 'ticker', symbols: lowerCaseSymbols(symbols) } as const + { channel: 'ticker', symbols: lowerCaseSymbols(symbols) } as const, + { channel: 'openInterest', symbols: lowerCaseSymbols(symbols) } as const ] } *map( - message: AsterFuturesMessage, + { data }: AsterFuturesMessage, localTimestamp: Date ): IterableIterator { - const pendingTickerInfo = this.pendingTickerInfoHelper.getPendingTickerInfo(message.data.s, 'aster-futures') + const pendingTickerInfo = this.pendingTickerInfoHelper.getPendingTickerInfo('s' in data ? data.s : data.symbol, 'aster-futures') - if (message.data.e === 'markPriceUpdate') { - pendingTickerInfo.updateMarkPrice(Number(message.data.p)) - if (message.data.i !== undefined) { - pendingTickerInfo.updateIndexPrice(Number(message.data.i)) - } - if (message.data.r !== '' && message.data.T !== 0) { - pendingTickerInfo.updateFundingRate(Number(message.data.r)) - pendingTickerInfo.updateFundingTimestamp(new Date(message.data.T)) + if ('e' in data) { + if (data.e === 'markPriceUpdate') { + pendingTickerInfo.updateMarkPrice(Number(data.p)) + if (data.i !== undefined) { + pendingTickerInfo.updateIndexPrice(Number(data.i)) + } + if (data.r !== '' && data.T !== 0) { + pendingTickerInfo.updateFundingRate(Number(data.r)) + pendingTickerInfo.updateFundingTimestamp(new Date(data.T)) + } + pendingTickerInfo.updateTimestamp(new Date(data.E)) } - pendingTickerInfo.updateTimestamp(new Date(message.data.E)) - } - if (message.data.e === '24hrTicker') { - pendingTickerInfo.updateLastPrice(Number(message.data.c)) - pendingTickerInfo.updateTimestamp(new Date(message.data.E)) + if (data.e === '24hrTicker') { + pendingTickerInfo.updateLastPrice(Number(data.c)) + pendingTickerInfo.updateTimestamp(new Date(data.E)) + } + } else if ('openInterest' in data) { + pendingTickerInfo.updateOpenInterest(Number(data.openInterest)) + pendingTickerInfo.updateTimestamp(new Date(data.time)) } if (pendingTickerInfo.hasChanged()) { @@ -307,6 +318,17 @@ type AsterFuturesAggTradeData = { m: boolean } +type AsterFuturesTradeData = { + e: 'trade' + E: number + s: string + t: number + p: string + q: string + T: number + m: boolean +} + type AsterFuturesDepthData = { lastUpdateId: undefined E: number diff --git a/src/realtimefeeds/aster.ts b/src/realtimefeeds/aster.ts index 0297cab..1b3bf52 100644 --- a/src/realtimefeeds/aster.ts +++ b/src/realtimefeeds/aster.ts @@ -1,11 +1,12 @@ +import { Writable } from 'stream' import { batch, CircularBuffer, getJSON, wait } from '../handy.ts' import { Filter } from '../types.ts' -import { RealTimeFeedBase } from './realtimefeed.ts' +import { MultiConnectionRealTimeFeedBase, PoolingClientBase, RealTimeFeedBase } from './realtimefeed.ts' export class AsterRealTimeFeed extends RealTimeFeedBase { protected static readonly depthChannel = 'depth' protected static readonly depthSnapshotChannel = 'depthSnapshot' - protected static readonly depthStream = 'depth@100ms' + protected readonly depthStream: string = 'depth@100ms' private readonly pendingDepthSnapshotSymbols = new Set() private readonly bufferedDepthUpdates = new Map>() protected readonly wssURL: string = 'wss://sstream.asterdex.com/stream' @@ -19,7 +20,7 @@ export class AsterRealTimeFeed extends RealTimeFeedBase { 'bookTicker' ]) protected readonly channelMappings: { [key: string]: string | undefined } = { - [AsterRealTimeFeed.depthChannel]: AsterRealTimeFeed.depthStream + [AsterRealTimeFeed.depthChannel]: this.depthStream } protected mapToSubscribeMessages(filters: Filter[]): any[] { @@ -67,7 +68,7 @@ export class AsterRealTimeFeed extends RealTimeFeedBase { } protected override onMessage(message: any) { - if (message.stream?.endsWith(`@${AsterRealTimeFeed.depthStream}`) !== true || message.data?.s === undefined) { + if (message.stream?.endsWith(`@${this.depthStream}`) !== true || message.data?.s === undefined) { return } @@ -272,20 +273,48 @@ type AsterDepthUpdateData = { previousFinalUpdateId: number } -export class AsterFuturesRealTimeFeed extends AsterRealTimeFeed { +export class AsterFuturesRealTimeFeed extends MultiConnectionRealTimeFeedBase { + protected *_getRealTimeFeeds(exchange: string, filters: Filter[], timeoutIntervalMS?: number, onError?: (error: Error) => void) { + const webSocketFilters = filters.filter((filter) => filter.channel !== 'openInterest') + if (webSocketFilters.length > 0) { + yield new AsterFuturesWebSocketRealTimeFeed(exchange, webSocketFilters, timeoutIntervalMS, onError) + } + + const openInterestFilters = filters.filter((filter) => filter.channel === 'openInterest') + if (openInterestFilters.length > 0) { + for (const filter of openInterestFilters) { + if (!filter.symbols || filter.symbols.length === 0) { + throw new Error('AsterFuturesRealTimeFeed requires explicitly specified symbols when subscribing to live feed') + } + } + + yield new AsterFuturesOpenInterestClient( + exchange, + 'https://fapi.asterdex.com/fapi/v3', + openInterestFilters.flatMap((filter) => filter.symbols!), + onError + ) + } + } +} + +export class AsterFuturesWebSocketRealTimeFeed extends AsterRealTimeFeed { protected readonly wssURL: string = 'wss://fstream.asterdex.com/stream' - protected readonly httpURL: string = 'https://fapi.asterdex.com/fapi/v1' + protected readonly httpURL: string = 'https://fapi.asterdex.com/fapi/v3' + protected readonly depthStream = 'depth@0ms' protected readonly channels = new Set([ + 'trade', 'aggTrade', 'ticker', AsterRealTimeFeed.depthChannel, AsterRealTimeFeed.depthSnapshotChannel, 'markPrice', 'forceOrder', - 'bookTicker' + 'bookTicker', + 'assetIndex' ]) protected readonly channelMappings: { [key: string]: string | undefined } = { - [AsterRealTimeFeed.depthChannel]: AsterRealTimeFeed.depthStream, + [AsterRealTimeFeed.depthChannel]: this.depthStream, markPrice: 'markPrice@1s' } @@ -295,9 +324,49 @@ export class AsterFuturesRealTimeFeed extends AsterRealTimeFeed { continue } - return update.firstUpdateId !== undefined && update.firstUpdateId <= lastUpdateId && update.lastUpdateId >= lastUpdateId + return ( + (update.firstUpdateId !== undefined && update.firstUpdateId <= lastUpdateId && update.lastUpdateId >= lastUpdateId) || + update.previousFinalUpdateId === lastUpdateId + ) } return undefined } } + +class AsterFuturesOpenInterestClient extends PoolingClientBase { + constructor( + exchange: string, + private readonly httpURL: string, + private readonly instruments: string[], + onError?: (error: Error) => void + ) { + super(exchange, 6, onError) + } + + protected async poolDataToStream(outputStream: Writable) { + for (const instrument of this.instruments) { + if (outputStream.destroyed) { + return + } + + const response = await getJSON(`${this.httpURL}/openInterest?symbol=${instrument.toLowerCase()}`, { + timeout: 2500 + }) + + if (outputStream.writable) { + outputStream.write({ + stream: `${instrument.toLowerCase()}@openInterest`, + generated: true, + data: response.data + }) + } + } + } +} + +export type AsterFuturesOpenInterestData = { + symbol: string + openInterest: string + time: number +} diff --git a/test/__snapshots__/mappers.test.snapshot b/test/__snapshots__/mappers.test.snapshot index 96302a6..14bba7d 100644 --- a/test/__snapshots__/mappers.test.snapshot +++ b/test/__snapshots__/mappers.test.snapshot @@ -3801,6 +3801,54 @@ exports[`mappers > map ascendex messages 8`] = ` `; exports[`mappers > map aster futures messages 1`] = ` +[ + { + "type": "trade", + "symbol": "BTCUSDT", + "exchange": "aster-futures", + "id": "181350", + "price": 10224.5, + "amount": 0.125, + "side": "sell", + "timestamp": "2019-09-17T04:05:03.462Z", + "localTimestamp": "2026-08-03T10:00:00.000Z" + } +] +`; + +exports[`mappers > map aster futures messages 10`] = ` +[ + { + "type": "liquidation", + "symbol": "BTCUSDT", + "exchange": "aster-futures", + "id": undefined, + "price": 4793.91, + "amount": 0.014, + "side": "buy", + "timestamp": "2020-03-13T00:23:51.421Z", + "localTimestamp": "2026-08-03T10:00:00.000Z" + } +] +`; + +exports[`mappers > map aster futures messages 11`] = ` +[ + { + "type": "book_ticker", + "symbol": "BTCUSDT", + "exchange": "aster-futures", + "askAmount": 0.38, + "askPrice": 33139.39, + "bidPrice": 33134.42, + "bidAmount": 0.17, + "timestamp": "2021-02-01T00:00:03.571Z", + "localTimestamp": "2026-08-03T10:00:00.000Z" + } +] +`; + +exports[`mappers > map aster futures messages 2`] = ` [ { "type": "trade", @@ -3816,11 +3864,30 @@ exports[`mappers > map aster futures messages 1`] = ` ] `; -exports[`mappers > map aster futures messages 2`] = ` +exports[`mappers > map aster futures messages 3`] = ` [] `; -exports[`mappers > map aster futures messages 3`] = ` +exports[`mappers > map aster futures messages 4`] = ` +[ + { + "type": "derivative_ticker", + "symbol": "BTCUSDT", + "exchange": "aster-futures", + "lastPrice": undefined, + "openInterest": 6196.323, + "fundingRate": undefined, + "fundingTimestamp": undefined, + "predictedFundingRate": undefined, + "indexPrice": undefined, + "markPrice": undefined, + "timestamp": "2026-08-18T13:31:54.451Z", + "localTimestamp": "2026-08-03T10:00:00.000Z" + } +] +`; + +exports[`mappers > map aster futures messages 5`] = ` [ { "type": "book_change", @@ -3849,7 +3916,7 @@ exports[`mappers > map aster futures messages 3`] = ` ] `; -exports[`mappers > map aster futures messages 4`] = ` +exports[`mappers > map aster futures messages 6`] = ` [ { "type": "book_change", @@ -3869,7 +3936,7 @@ exports[`mappers > map aster futures messages 4`] = ` ] `; -exports[`mappers > map aster futures messages 5`] = ` +exports[`mappers > map aster futures messages 7`] = ` [ { "type": "book_change", @@ -3889,7 +3956,7 @@ exports[`mappers > map aster futures messages 5`] = ` ] `; -exports[`mappers > map aster futures messages 6`] = ` +exports[`mappers > map aster futures messages 8`] = ` [ { "type": "derivative_ticker", @@ -3908,7 +3975,7 @@ exports[`mappers > map aster futures messages 6`] = ` ] `; -exports[`mappers > map aster futures messages 7`] = ` +exports[`mappers > map aster futures messages 9`] = ` [ { "type": "derivative_ticker", @@ -3927,38 +3994,6 @@ exports[`mappers > map aster futures messages 7`] = ` ] `; -exports[`mappers > map aster futures messages 8`] = ` -[ - { - "type": "liquidation", - "symbol": "BTCUSDT", - "exchange": "aster-futures", - "id": undefined, - "price": 4793.91, - "amount": 0.014, - "side": "buy", - "timestamp": "2020-03-13T00:23:51.421Z", - "localTimestamp": "2026-08-03T10:00:00.000Z" - } -] -`; - -exports[`mappers > map aster futures messages 9`] = ` -[ - { - "type": "book_ticker", - "symbol": "BTCUSDT", - "exchange": "aster-futures", - "askAmount": 0.38, - "askPrice": 33139.39, - "bidPrice": 33134.42, - "bidAmount": 0.17, - "timestamp": "2021-02-01T00:00:03.571Z", - "localTimestamp": "2026-08-03T10:00:00.000Z" - } -] -`; - exports[`mappers > map aster messages 1`] = ` [ { diff --git a/test/aster-realtimefeed.test.ts b/test/aster-realtimefeed.test.ts index b398469..93f79d6 100644 --- a/test/aster-realtimefeed.test.ts +++ b/test/aster-realtimefeed.test.ts @@ -2,7 +2,7 @@ import type { AddressInfo } from 'net' import { createServer } from 'http' import { test } from 'node:test' import { assert, errorMessageIncludes } from './assertions.ts' -import { AsterFuturesRealTimeFeed, AsterRealTimeFeed } from '../dist/realtimefeeds/aster.js' +import { AsterFuturesWebSocketRealTimeFeed, AsterRealTimeFeed } from '../dist/realtimefeeds/aster.js' import { getRealTimeFeedFactory } from '../dist/realtimefeeds/index.js' import type { Filter } from '../dist/types.js' @@ -33,14 +33,14 @@ class TestAsterRealTimeFeed extends AsterRealTimeFeed { } } -class TestAsterFuturesRealTimeFeed extends AsterFuturesRealTimeFeed { +class TestAsterFuturesRealTimeFeed extends AsterFuturesWebSocketRealTimeFeed { protected readonly httpURL: string constructor( exchange: 'aster-futures', filters: Filter[], timeoutIntervalMS: number | undefined, - httpURL = 'https://fapi.asterdex.com/fapi/v1' + httpURL = 'https://fapi.asterdex.com/fapi/v3' ) { super(exchange, filters, timeoutIntervalMS) this.httpURL = httpURL @@ -165,6 +165,10 @@ test('map aster futures realtime subscriptions', () => { channel: 'depthSnapshot', symbols: ['btcusdt'] }, + { + channel: 'trade', + symbols: ['btcusdt'] + }, { channel: 'markPrice', symbols: ['btcusdt'] @@ -172,23 +176,37 @@ test('map aster futures realtime subscriptions', () => { { channel: 'forceOrder', symbols: ['btcusdt'] + }, + { + channel: 'assetIndex', + symbols: ['btcusd'] } ]), [ { method: 'SUBSCRIBE', - params: ['btcusdt@depth@100ms'], + params: ['btcusdt@depth@0ms'], id: 1 }, { method: 'SUBSCRIBE', - params: ['btcusdt@markPrice@1s'], + params: ['btcusdt@trade'], id: 2 }, { method: 'SUBSCRIBE', - params: ['btcusdt@forceOrder'], + params: ['btcusdt@markPrice@1s'], id: 3 + }, + { + method: 'SUBSCRIBE', + params: ['btcusdt@forceOrder'], + id: 4 + }, + { + method: 'SUBSCRIBE', + params: ['btcusd@assetIndex'], + id: 5 } ] ) @@ -292,7 +310,9 @@ test('retry aster futures manual depth snapshots until first update overlaps', a ] feed.map(filters) - feed.observe(createDepthUpdate({ symbol: 'BTCUSDT', firstUpdateId: 105, lastUpdateId: 105, previousFinalUpdateId: 106 })) + feed.observe( + createDepthUpdate({ symbol: 'BTCUSDT', firstUpdateId: 105, lastUpdateId: 105, previousFinalUpdateId: 106, stream: 'depth@0ms' }) + ) const snapshots = await feed.provideSnapshots(filters) @@ -317,15 +337,17 @@ function createDepthUpdate({ symbol, firstUpdateId, lastUpdateId, - previousFinalUpdateId + previousFinalUpdateId, + stream = 'depth@100ms' }: { symbol: string firstUpdateId?: number lastUpdateId: number previousFinalUpdateId: number + stream?: string }) { return { - stream: `${symbol.toLowerCase()}@depth@100ms`, + stream: `${symbol.toLowerCase()}@${stream}`, data: { e: 'depthUpdate', E: 1785230774524, diff --git a/test/mappers.test.ts b/test/mappers.test.ts index c9aaceb..0f7293b 100644 --- a/test/mappers.test.ts +++ b/test/mappers.test.ts @@ -2977,6 +2977,31 @@ describe('mappers', () => { const asterFuturesMapper = createMapper('aster-futures', new Date()) const localTimestamp = new Date('2026-08-03T10:00:00.000Z') + assert.deepEqual(normalizeTrades('aster-futures', localTimestamp).getFilters(['BTCUSDT']), [{ channel: 'trade', symbols: ['btcusdt'] }]) + assert.deepEqual(normalizeDerivativeTickers('aster-futures', localTimestamp).getFilters(['BTCUSDT']), [ + { channel: 'markPrice', symbols: ['btcusdt'] }, + { channel: 'ticker', symbols: ['btcusdt'] }, + { channel: 'openInterest', symbols: ['btcusdt'] } + ]) + snapshot( + asterFuturesMapper.map( + { + stream: 'btcusdt@trade', + data: { + e: 'trade', + E: 1568693103463, + s: 'BTCUSDT', + t: 181350, + p: '10224.50', + q: '0.125', + T: 1568693103462, + m: true + } + }, + localTimestamp + ) + ) + snapshot( asterFuturesMapper.map( { @@ -3018,6 +3043,22 @@ describe('mappers', () => { ) ) + const asterFuturesOpenInterestMapper = createMapper('aster-futures', new Date()) + snapshot( + asterFuturesOpenInterestMapper.map( + { + stream: 'btcusdt@openInterest', + generated: true, + data: { + symbol: 'BTCUSDT', + openInterest: '6196.323', + time: 1787059914451 + } + }, + localTimestamp + ) + ) + snapshot( asterFuturesMapper.map( { @@ -3052,7 +3093,7 @@ describe('mappers', () => { snapshot( asterFuturesOverlapEdgeMapper.map( { - stream: 'ethusdt@depth@100ms', + stream: 'ethusdt@depth@0ms', data: { e: 'depthUpdate', E: 1573948821952, @@ -3072,7 +3113,7 @@ describe('mappers', () => { snapshot( asterFuturesMapper.map( { - stream: 'btcusdt@depth@100ms', + stream: 'btcusdt@depth@0ms', data: { e: 'depthUpdate', E: 1573948822952,