diff --git a/package-lock.json b/package-lock.json index 77a10ce..a0bf78c 100644 --- a/package-lock.json +++ b/package-lock.json @@ -14,6 +14,7 @@ "follow-redirects": "^1.15.9", "https-proxy-agent": "^8.0.0", "p-map": "^7.0.4", + "protobufjs": "^8.6.2", "socks-proxy-agent": "^9.0.0", "ws": "^8.18.3" }, @@ -4602,6 +4603,12 @@ "node": ">=8" } }, + "node_modules/long": { + "version": "5.3.2", + "resolved": "https://registry.npmjs.org/long/-/long-5.3.2.tgz", + "integrity": "sha512-mNAgZ1GmyNhD7AuqnTG3/VQ26o760+ZYBPKjPvugO8+nLbYfX6TVpJPseBvopbdY+qpZ/lKUnmEc1LeZYS3QAA==", + "license": "Apache-2.0" + }, "node_modules/make-dir": { "version": "4.0.0", "resolved": "https://registry.npmjs.org/make-dir/-/make-dir-4.0.0.tgz", @@ -5035,6 +5042,18 @@ "node": ">= 6" } }, + "node_modules/protobufjs": { + "version": "8.6.2", + "resolved": "https://registry.npmjs.org/protobufjs/-/protobufjs-8.6.2.tgz", + "integrity": "sha512-CCERJxzRvKMeEdJSLwdQf40TXWNPc8M4RkN7j/lxY6FQB+4do8rETWqj60AqxP9n0XIsxnSefZ8uhAaGKg2njw==", + "license": "BSD-3-Clause", + "dependencies": { + "long": "^5.3.2" + }, + "engines": { + "node": ">=12.0.0" + } + }, "node_modules/pure-rand": { "version": "6.1.0", "resolved": "https://registry.npmjs.org/pure-rand/-/pure-rand-6.1.0.tgz", diff --git a/package.json b/package.json index 416b9dc..39ecc2e 100644 --- a/package.json +++ b/package.json @@ -73,6 +73,7 @@ "follow-redirects": "^1.15.9", "https-proxy-agent": "^8.0.0", "p-map": "^7.0.4", + "protobufjs": "^8.6.2", "socks-proxy-agent": "^9.0.0", "ws": "^8.18.3" }, diff --git a/src/consts.ts b/src/consts.ts index 5d3d9a0..b512bda 100644 --- a/src/consts.ts +++ b/src/consts.ts @@ -58,6 +58,7 @@ export const EXCHANGES = [ 'hyperliquid', 'lighter', 'bullish', + 'mexc', 'polymarket' ] as const @@ -526,6 +527,13 @@ const LIGHTER_CHANNELS = ['order_book', 'trade', 'ticker', 'market_stats', 'spot const BULLISH_CHANNELS = ['V1TALevel2', 'V1TALevel1', 'V1TAAnonymousTradeUpdate', 'V1TATickerResponse', 'V1TAIndexPrice'] as const +const MEXC_CHANNELS = [ + 'spot@public.aggre.deals.v3.api.pb@10ms', + 'spot@public.aggre.depth.v3.api.pb@10ms', + 'spot@public.aggre.depth.snapshot.v3.api.pb@10ms', + 'spot@public.aggre.bookTicker.v3.api.pb@10ms' +] as const + const POLYMARKET_CHANNELS = [ 'book', 'price_change', @@ -597,5 +605,6 @@ export const EXCHANGE_CHANNELS_INFO = { hyperliquid: HYPERLIQUID_CHANNELS, lighter: LIGHTER_CHANNELS, bullish: BULLISH_CHANNELS, + mexc: MEXC_CHANNELS, polymarket: POLYMARKET_CHANNELS } diff --git a/src/mappers/index.ts b/src/mappers/index.ts index 1f636fc..7dca811 100644 --- a/src/mappers/index.ts +++ b/src/mappers/index.ts @@ -33,6 +33,7 @@ import { kucoinMappers } from './kucoin.ts' import { kucoinFuturesMappers } from './kucoinfutures.ts' import { lighterMappers } from './lighter.ts' import { Mapper } from './mapper.ts' +import { mexcMappers } from './mexc.ts' import { okexMappers } from './okex.ts' import { phemexMappers } from './phemex.ts' import { poloniexMappers } from './poloniex.ts' @@ -85,6 +86,7 @@ const registeredMappers = mergeExchangeMappers( kucoinMappers, kucoinFuturesMappers, lighterMappers, + mexcMappers, okexMappers, phemexMappers, poloniexMappers, diff --git a/src/mappers/mexc.ts b/src/mappers/mexc.ts new file mode 100644 index 0000000..87e8d7d --- /dev/null +++ b/src/mappers/mexc.ts @@ -0,0 +1,324 @@ +import { debug } from '../debug.ts' +import { asNumberOrUndefined, CircularBuffer, upperCaseSymbols } from '../handy.ts' +import { BookChange, BookPriceLevel, BookTicker, Trade } from '../types.ts' +import { Mapper } from './mapper.ts' +import { exchangeMappers, isRealTime } from './registry.ts' + +export const mexcMappers = exchangeMappers({ + mexc: { + trades: () => new MexcTradesMapper(), + bookChanges: (localTimestamp) => new MexcBookChangeMapper({ ignoreBookSnapshotOverlapError: isRealTime(localTimestamp) === false }), + bookTickers: () => new MexcBookTickerMapper() + } +}) + +export class MexcTradesMapper implements Mapper<'mexc', Trade> { + private readonly channel = 'spot@public.aggre.deals.v3.api.pb@10ms' + + canHandle(message: MexcTradeMessage | MexcControlMessage) { + return message.channel?.startsWith(`${this.channel}@`) === true && 'publicAggreDeals' in message + } + + getFilters(symbols?: string[]) { + return [{ channel: this.channel, symbols: upperCaseSymbols(symbols) } as const] + } + + *map(message: MexcTradeMessage, localTimestamp: Date): IterableIterator { + for (const trade of message.publicAggreDeals.deals) { + yield { + type: 'trade', + symbol: message.symbol, + exchange: 'mexc', + id: trade.tradeId, + price: Number(trade.price), + amount: Number(trade.quantity), + side: trade.tradeType === MexcTradeType.Buy ? 'buy' : 'sell', + timestamp: new Date(Number(trade.time)), + localTimestamp + } + } + } +} + +export class MexcBookChangeMapper implements Mapper<'mexc', BookChange> { + private readonly channel = 'spot@public.aggre.depth.v3.api.pb@10ms' + private readonly snapshotChannel = 'spot@public.aggre.depth.snapshot.v3.api.pb@10ms' + private readonly symbolDepthInfo: Record = {} + private readonly ignoreBookSnapshotOverlapError: boolean + + constructor({ ignoreBookSnapshotOverlapError }: { ignoreBookSnapshotOverlapError: boolean }) { + this.ignoreBookSnapshotOverlapError = ignoreBookSnapshotOverlapError + } + + canHandle(message: MexcDepthSnapshotMessage | MexcDepthUpdateMessage) { + return message.generated === true + ? message.channel?.startsWith(`${this.snapshotChannel}@`) === true && 'publicAggreDepthsSnapshot' in message + : message.channel?.startsWith(`${this.channel}@`) === true && 'publicAggreDepths' in message + } + + getFilters(symbols?: string[]) { + const normalizedSymbols = upperCaseSymbols(symbols) + + return [ + { channel: this.channel, symbols: normalizedSymbols } as const, + { channel: this.snapshotChannel, symbols: normalizedSymbols } as const + ] + } + + *map(message: MexcDepthSnapshotMessage | MexcDepthUpdateMessage, localTimestamp: Date): IterableIterator { + const depthInfo = this.getDepthInfoFor(message.symbol) + if (message.generated === true) { + if (depthInfo.snapshotEmitted) { + return + } + + let currentBookVersion = Number(message.publicAggreDepthsSnapshot.lastUpdateId) + const bids = message.publicAggreDepthsSnapshot.bids.map(this.mapDepthSnapshotLevel) + const asks = message.publicAggreDepthsSnapshot.asks.map(this.mapDepthSnapshotLevel) + + for (const update of depthInfo.updates.items()) { + const fromVersion = Number(update.publicAggreDepths.fromVersion) + const toVersion = Number(update.publicAggreDepths.toVersion) + if (currentBookVersion >= toVersion) { + continue + } + + if (!depthInfo.isContinuityValidated) { + if (fromVersion > currentBookVersion + 1 || toVersion < currentBookVersion + 1) { + const message = `MEXC depth snapshot has no overlap with first update, update ${JSON.stringify(update)}, currentBookVersion: ${currentBookVersion}` + if (this.ignoreBookSnapshotOverlapError) { + depthInfo.isContinuityValidated = true + debug(message) + } else { + throw new Error(message) + } + } else { + depthInfo.isContinuityValidated = true + } + } + + for (const bid of update.publicAggreDepths.bids ?? []) { + this.applyLevel(bids, this.mapBookLevel(bid)) + } + for (const ask of update.publicAggreDepths.asks ?? []) { + this.applyLevel(asks, this.mapBookLevel(ask)) + } + currentBookVersion = toVersion + } + + depthInfo.updates.clear() + depthInfo.currentBookVersion = currentBookVersion + depthInfo.snapshotEmitted = true + + yield { + type: 'book_change', + symbol: message.symbol, + exchange: 'mexc', + isSnapshot: true, + bids, + asks, + timestamp: new Date(Number(message.publicAggreDepthsSnapshot.timestamp)), + localTimestamp + } + + return + } + + if (!depthInfo.snapshotEmitted) { + depthInfo.updates.append(message) + return + } + + const fromVersion = Number(message.publicAggreDepths.fromVersion) + const toVersion = Number(message.publicAggreDepths.toVersion) + if (toVersion <= depthInfo.currentBookVersion!) { + return + } + + if (!depthInfo.isContinuityValidated) { + if (fromVersion > depthInfo.currentBookVersion! + 1 || toVersion < depthInfo.currentBookVersion! + 1) { + const errorMessage = `MEXC depth snapshot has no overlap with first update, update ${JSON.stringify(message)}, currentBookVersion: ${depthInfo.currentBookVersion}` + if (this.ignoreBookSnapshotOverlapError) { + depthInfo.isContinuityValidated = true + debug(errorMessage) + } else { + throw new Error(errorMessage) + } + } else { + depthInfo.isContinuityValidated = true + } + } + + depthInfo.currentBookVersion = toVersion + + yield { + type: 'book_change', + symbol: message.symbol, + exchange: 'mexc', + isSnapshot: false, + bids: (message.publicAggreDepths.bids ?? []).map(this.mapBookLevel), + asks: (message.publicAggreDepths.asks ?? []).map(this.mapBookLevel), + timestamp: new Date(Number(message.sendTime)), + localTimestamp + } + } + + private applyLevel(bookSide: BookPriceLevel[], levelUpdate: BookPriceLevel) { + const existingIndex = bookSide.findIndex((level) => level.price === levelUpdate.price) + if (levelUpdate.amount === 0) { + if (existingIndex !== -1) { + bookSide.splice(existingIndex, 1) + } + return + } + + if (existingIndex === -1) { + bookSide.push(levelUpdate) + } else { + bookSide[existingIndex] = levelUpdate + } + } + + private mapBookLevel(level: MexcPriceLevel) { + return { + price: Number(level.price), + amount: Number(level.quantity) + } + } + + private mapDepthSnapshotLevel([price, quantity]: MexcDepthSnapshotLevel) { + return { + price: Number(price), + amount: Number(quantity) + } + } + + private getDepthInfoFor(symbol: string) { + if (this.symbolDepthInfo[symbol] === undefined) { + this.symbolDepthInfo[symbol] = { updates: new CircularBuffer(2000) } + } + + return this.symbolDepthInfo[symbol] + } +} + +export class MexcBookTickerMapper implements Mapper<'mexc', BookTicker> { + private readonly channel = 'spot@public.aggre.bookTicker.v3.api.pb@10ms' + + canHandle(message: MexcBookTickerMessage | MexcControlMessage) { + return message.channel?.startsWith(`${this.channel}@`) === true && 'publicAggreBookTicker' in message + } + + getFilters(symbols?: string[]) { + return [{ channel: this.channel, symbols: upperCaseSymbols(symbols) } as const] + } + + *map(message: MexcBookTickerMessage, localTimestamp: Date): IterableIterator { + yield { + type: 'book_ticker', + symbol: message.symbol, + exchange: 'mexc', + askAmount: asNumberOrUndefined(message.publicAggreBookTicker.askQuantity), + askPrice: asNumberOrUndefined(message.publicAggreBookTicker.askPrice), + bidAmount: asNumberOrUndefined(message.publicAggreBookTicker.bidQuantity), + bidPrice: asNumberOrUndefined(message.publicAggreBookTicker.bidPrice), + timestamp: new Date(Number(message.sendTime)), + localTimestamp + } + } +} + +type MexcControlMessage = { + channel?: undefined + id?: number + code?: number + msg?: string +} + +type MexcMappedChannel = + | 'spot@public.aggre.deals.v3.api.pb@10ms' + | 'spot@public.aggre.depth.v3.api.pb@10ms' + | 'spot@public.aggre.depth.snapshot.v3.api.pb@10ms' + | 'spot@public.aggre.bookTicker.v3.api.pb@10ms' +type MexcChannelWithSymbol = `${TChannel}@${string}` + +type MexcProtobufMessage = { + channel: MexcChannelWithSymbol + symbol: string + symbolId?: string + createTime?: string | number + sendTime: string | number +} + +type MexcTradeMessage = MexcProtobufMessage<'spot@public.aggre.deals.v3.api.pb@10ms'> & { + publicAggreDeals: { + eventType: 'spot@public.aggre.deals.v3.api.pb@10ms' + deals: MexcTrade[] + } +} + +type MexcTrade = { + price: string + quantity: string + tradeType: MexcTradeType + time: string | number + tradeId?: string +} + +enum MexcTradeType { + Buy = 1, + Sell = 2 +} + +export type MexcDepthSnapshotMessage = { + channel: MexcChannelWithSymbol<'spot@public.aggre.depth.snapshot.v3.api.pb@10ms'> + symbol: string + generated: true + publicAggreDepthsSnapshot: MexcAggreDepthsSnapshot +} + +type MexcAggreDepthsSnapshot = { + asks: MexcDepthSnapshotLevel[] + bids: MexcDepthSnapshotLevel[] + lastUpdateId: number + timestamp: number +} + +type MexcDepthSnapshotLevel = [string, string] + +type MexcDepthUpdateMessage = MexcProtobufMessage<'spot@public.aggre.depth.v3.api.pb@10ms'> & { + generated?: undefined + publicAggreDepths: MexcAggreDepths +} + +type MexcAggreDepths = { + eventType: 'spot@public.aggre.depth.v3.api.pb@10ms' + asks?: MexcPriceLevel[] + bids?: MexcPriceLevel[] + fromVersion: string + toVersion: string + lastOrderCreateTime?: string | number +} + +type MexcPriceLevel = { + price: string + quantity: string +} + +type MexcBookTickerMessage = MexcProtobufMessage<'spot@public.aggre.bookTicker.v3.api.pb@10ms'> & { + publicAggreBookTicker: { + bidPrice?: string + bidQuantity?: string + askPrice?: string + askQuantity?: string + version?: string + lastOrderCreateTime?: string | number + } +} + +type MexcDepthInfo = { + isContinuityValidated?: boolean + currentBookVersion?: number + snapshotEmitted?: boolean + updates: CircularBuffer +} diff --git a/src/realtimefeeds/index.ts b/src/realtimefeeds/index.ts index 60c2a09..e38b79a 100644 --- a/src/realtimefeeds/index.ts +++ b/src/realtimefeeds/index.ts @@ -54,6 +54,7 @@ import { CoinbaseInternationalRealTimeFeed } from './coinbaseinternational.ts' import { HyperliquidRealTimeFeed } from './hyperliquid.ts' import { LighterRealTimeFeed } from './lighter.ts' import { BullishRealTimeFeed } from './bullish.ts' +import { MexcRealTimeFeed } from './mexc.ts' import { PolymarketRealTimeFeed } from './polymarket.ts' export * from './realtimefeed.ts' @@ -120,6 +121,7 @@ const realTimeFeedsMap: { hyperliquid: HyperliquidRealTimeFeed, lighter: LighterRealTimeFeed, bullish: BullishRealTimeFeed, + mexc: MexcRealTimeFeed, polymarket: PolymarketRealTimeFeed } diff --git a/src/realtimefeeds/mexc.ts b/src/realtimefeeds/mexc.ts new file mode 100644 index 0000000..0a30013 --- /dev/null +++ b/src/realtimefeeds/mexc.ts @@ -0,0 +1,327 @@ +import protobuf from 'protobufjs' +import { CircularBuffer, getJSON, wait } from '../handy.ts' +import type { MexcDepthSnapshotMessage } from '../mappers/mexc.ts' +import { Filter } from '../types.ts' +import { RealTimeFeedBase } from './realtimefeed.ts' + +export class MexcRealTimeFeed extends RealTimeFeedBase { + private static readonly depthChannel = 'spot@public.aggre.depth.v3.api.pb@10ms' + private static readonly depthSnapshotChannel = 'spot@public.aggre.depth.snapshot.v3.api.pb@10ms' + private static readonly jsonObjectStart = '{'.charCodeAt(0) + private static pushDataV3ApiWrapper: protobuf.Type | undefined + /** + * MEXC protobuf docs at @see https://www.mexc.com/api-docs/spot-v3/websocket-market-streams/protocol-buffers-integration + * and schemas from @see https://github.com/mexcdevelop/websocket-proto, mirrored in tardis-recorder/datafeeds/src/mexc/proto. + */ + private static readonly pushDataV3ApiWrapperSchema = ` + syntax = "proto3"; + + message PushDataV3ApiWrapper { + string channel = 1; + oneof body { + PublicAggreDepthsV3Api publicAggreDepths = 313; + PublicAggreDealsV3Api publicAggreDeals = 314; + PublicAggreBookTickerV3Api publicAggreBookTicker = 315; + } + + optional string symbol = 3; + optional string symbolId = 4; + optional int64 createTime = 5; + optional int64 sendTime = 6; + } + + message PublicAggreDealsV3Api { + repeated PublicAggreDealsV3ApiItem deals = 1; + string eventType = 2; + } + + message PublicAggreDealsV3ApiItem { + string price = 1; + string quantity = 2; + int32 tradeType = 3; + int64 time = 4; + string tradeId = 5; + } + + message PublicAggreDepthsV3Api { + repeated PublicAggreDepthV3ApiItem asks = 1; + repeated PublicAggreDepthV3ApiItem bids = 2; + string eventType = 3; + string fromVersion = 4; + string toVersion = 5; + int64 lastOrderCreateTime = 6; + } + + message PublicAggreDepthV3ApiItem { + string price = 1; + string quantity = 2; + } + + message PublicAggreBookTickerV3Api { + string bidPrice = 1; + string bidQuantity = 2; + string askPrice = 3; + string askQuantity = 4; + string version = 5; + int64 lastOrderCreateTime = 6; + } + ` + + private readonly pendingDepthSnapshotSymbols = new Set() + private readonly bufferedDepthUpdates = new Map>() + protected readonly wssURL = 'wss://wbs-api.mexc.com/ws' + protected readonly httpURL: string = 'https://api.mexc.com' + private readonly channels = new Set([ + 'spot@public.aggre.deals.v3.api.pb@10ms', + MexcRealTimeFeed.depthChannel, + MexcRealTimeFeed.depthSnapshotChannel, + 'spot@public.aggre.bookTicker.v3.api.pb@10ms' + ]) + + protected override mapToSubscribeMessages(filters: Filter[]): any[] { + const filtersWithSymbols = filters.map>>((filter) => { + if (!this.channels.has(filter.channel)) { + throw new Error(`MexcRealTimeFeed unsupported channel ${filter.channel}`) + } + + if (!filter.symbols || filter.symbols.length === 0) { + throw new Error('MexcRealTimeFeed requires explicitly specified symbols when subscribing to live feed') + } + + return filter as Required> + }) + + const depthSnapshotFilters = filtersWithSymbols.filter((filter) => filter.channel === MexcRealTimeFeed.depthSnapshotChannel) + this.validateDepthSnapshotFilters(filtersWithSymbols, depthSnapshotFilters) + this.resetDepthSnapshotTracking(depthSnapshotFilters) + + return [ + { + method: 'SUBSCRIPTION', + params: filtersWithSymbols.flatMap((filter) => + filter.channel === MexcRealTimeFeed.depthSnapshotChannel + ? [] + : filter.symbols.map((symbol) => `${filter.channel}@${symbol.toUpperCase()}`) + ) + } + ] + } + + protected override parseMessage(message: Buffer): any { + const buffer = Buffer.isBuffer(message) ? message : Buffer.from(message) + + if (buffer.length > 0 && buffer[0] === MexcRealTimeFeed.jsonObjectStart) { + return JSON.parse(buffer.toString()) + } + + const pushDataV3ApiWrapper = this.getPushDataV3ApiWrapper() + return pushDataV3ApiWrapper.toObject(pushDataV3ApiWrapper.decode(buffer), { + longs: String, + arrays: true + }) + } + + protected override messageIsError(message: any): boolean { + return message.code !== undefined && message.code !== 0 + } + + protected override messageIsHeartbeat(message: any) { + return message.msg === 'PONG' + } + + protected override onMessage(message: any) { + if ( + message.channel?.startsWith(MexcRealTimeFeed.depthChannel) !== true || + message.symbol === undefined || + this.pendingDepthSnapshotSymbols.has(message.symbol) === false + ) { + return + } + + const firstVersion = Number(message.publicAggreDepths?.fromVersion) + const lastVersion = Number(message.publicAggreDepths?.toVersion) + if (Number.isFinite(firstVersion) === false || Number.isFinite(lastVersion) === false) { + return + } + + const bufferedUpdates = this.bufferedDepthUpdates.get(message.symbol) ?? new CircularBuffer(2000) + bufferedUpdates.append({ firstVersion, lastVersion }) + this.bufferedDepthUpdates.set(message.symbol, bufferedUpdates) + } + + protected override async provideManualSnapshots(filters: Filter[], shouldCancel: () => boolean) { + const depthFilter = filters.find((filter) => filter.channel === MexcRealTimeFeed.depthSnapshotChannel) + if (depthFilter === undefined) { + return + } + + this.debug('requesting manual snapshots for: %s', depthFilter.symbols!) + + for (const symbol of depthFilter.symbols!) { + await this.provideManualSnapshot(symbol.toUpperCase(), shouldCancel) + } + + this.debug('requested manual snapshots successfully for: %s ', depthFilter.symbols!) + } + + protected sendCustomPing = () => { + this.send({ method: 'PING' }) + } + + private getPushDataV3ApiWrapper() { + MexcRealTimeFeed.pushDataV3ApiWrapper ??= protobuf + .parse(MexcRealTimeFeed.pushDataV3ApiWrapperSchema) + .root.lookupType('PushDataV3ApiWrapper') + + return MexcRealTimeFeed.pushDataV3ApiWrapper + } + + 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 === MexcRealTimeFeed.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( + `MexcRealTimeFeed requires ${MexcRealTimeFeed.depthChannel} for every ${MexcRealTimeFeed.depthSnapshotChannel} symbol` + ) + } + } + } + } + + private async provideManualSnapshot(symbol: string, shouldCancel: () => boolean) { + const maxSnapshotRounds = 4 + const maxSnapshotAttemptsPerRound = 3 + + for (let round = 0; round < maxSnapshotRounds; round++) { + for (let attempt = 1; attempt <= maxSnapshotAttemptsPerRound; attempt++) { + if (shouldCancel()) { + return + } + + const { data } = await getJSON(`${this.httpURL}/api/v3/depth?symbol=${symbol}&limit=1000`) + if (this.snapshotResponseIsValid(data) === false) { + if (attempt < maxSnapshotAttemptsPerRound) { + await wait(attempt * 1000) + } + continue + } + + const hasOverlap = await this.waitForSnapshotOverlap(symbol, data.lastUpdateId) + + if (shouldCancel()) { + return + } + + if (hasOverlap === false) { + this.trimBufferedUpdates(symbol) + if (attempt < maxSnapshotAttemptsPerRound) { + await wait(attempt * 1000) + } + continue + } + + if (hasOverlap === true || attempt === maxSnapshotAttemptsPerRound) { + this.manualSnapshotsBuffer.push(this.createManualSnapshot(symbol, data)) + this.pendingDepthSnapshotSymbols.delete(symbol) + this.bufferedDepthUpdates.delete(symbol) + return + } + } + } + + throw new Error(`MexcRealTimeFeed could not align depth snapshot for ${symbol}`) + } + + private async waitForSnapshotOverlap(symbol: string, lastUpdateId: number) { + let hasOverlap = this.validateSnapshotOverlap(symbol, lastUpdateId) + for (let attempt = 0; attempt < 60; attempt++) { + if (hasOverlap !== undefined) { + return hasOverlap + } + + await wait(100) + hasOverlap = this.validateSnapshotOverlap(symbol, lastUpdateId) + } + + return hasOverlap + } + + private validateSnapshotOverlap(symbol: string, lastUpdateId: number) { + const bufferedUpdates = this.bufferedDepthUpdates.get(symbol) + for (const update of bufferedUpdates?.items() ?? []) { + if (update.lastVersion <= lastUpdateId) { + continue + } + + return update.firstVersion <= lastUpdateId + 1 && update.lastVersion >= lastUpdateId + 1 + } + + 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: MexcDepthSnapshotResponse) { + return Number.isFinite(data.lastUpdateId) && Array.isArray(data.asks) && Array.isArray(data.bids) + } + + private createManualSnapshot(symbol: string, data: MexcDepthSnapshotResponse): MexcDepthSnapshotMessage { + return { + channel: `${MexcRealTimeFeed.depthSnapshotChannel}@${symbol}`, + symbol, + generated: true, + publicAggreDepthsSnapshot: { + lastUpdateId: data.lastUpdateId, + asks: data.asks, + bids: data.bids, + timestamp: data.timestamp + } + } + } +} + +type MexcDepthSnapshotResponse = { + lastUpdateId: number + bids: [string, string][] + asks: [string, string][] + timestamp: number +} + +type MexcDepthUpdateData = { + firstVersion: number + lastVersion: number +} diff --git a/src/realtimefeeds/realtimefeed.ts b/src/realtimefeeds/realtimefeed.ts index 5ee978d..baa213c 100644 --- a/src/realtimefeeds/realtimefeed.ts +++ b/src/realtimefeeds/realtimefeed.ts @@ -117,7 +117,7 @@ export abstract class RealTimeFeedBase implements RealTimeFeedIterable { message = message.toString().replace(/"id":([0-9]+),/g, '"id":"$1",') as any } - const messageDeserialized = JSON.parse(message as any) + const messageDeserialized = this.parseMessage(message) if (this.messageIsError(messageDeserialized)) { if (this.isIgnoredError(messageDeserialized)) { @@ -233,6 +233,10 @@ export abstract class RealTimeFeedBase implements RealTimeFeedIterable { protected abstract messageIsError(message: any): boolean + protected parseMessage(message: Buffer): any { + return JSON.parse(message as any) + } + protected sendCustomPing: (() => void) | undefined = undefined protected isIgnoredError(_message: any) { diff --git a/test/__snapshots__/mexc-realtimefeed.test.ts.snap b/test/__snapshots__/mexc-realtimefeed.test.ts.snap new file mode 100644 index 0000000..c785e6f --- /dev/null +++ b/test/__snapshots__/mexc-realtimefeed.test.ts.snap @@ -0,0 +1,15 @@ +// Jest Snapshot v1, https://goo.gl/fbAQLP + +exports[`map mexc realtime subscriptions 1`] = ` +[ + { + "method": "SUBSCRIPTION", + "params": [ + "spot@public.aggre.deals.v3.api.pb@10ms@BTCUSDT", + "spot@public.aggre.deals.v3.api.pb@10ms@ETHUSDT", + "spot@public.aggre.depth.v3.api.pb@10ms@BTCUSDT", + "spot@public.aggre.bookTicker.v3.api.pb@10ms@BTCUSDT", + ], + }, +] +`; diff --git a/test/mapper-switches.test.ts b/test/mapper-switches.test.ts index 3460a91..253d2a9 100644 --- a/test/mapper-switches.test.ts +++ b/test/mapper-switches.test.ts @@ -78,6 +78,7 @@ test('normalizers keep existing exchange support matrix', () => { 'kucoin-futures', 'lighter', 'mango', + 'mexc', 'okcoin', 'okex', 'okex-futures', @@ -145,6 +146,7 @@ test('normalizers keep existing exchange support matrix', () => { 'kucoin-futures', 'lighter', 'mango', + 'mexc', 'okcoin', 'okex', 'okex-futures', @@ -265,6 +267,7 @@ test('normalizers keep existing exchange support matrix', () => { 'kucoin-futures', 'lighter', 'mango', + 'mexc', 'okcoin', 'okex', 'okex-futures', diff --git a/test/mappers.test.ts b/test/mappers.test.ts index 3821678..2683a8c 100644 --- a/test/mappers.test.ts +++ b/test/mappers.test.ts @@ -83,6 +83,7 @@ const exchangesWithBookTickerInfo: Exchange[] = [ 'hyperliquid', 'lighter', 'bullish', + 'mexc', 'polymarket' ] @@ -11013,6 +11014,648 @@ test('map lighter ticker messages', () => { } }) +test('map mexc messages', () => { + const localTimestamp = new Date('2026-05-27T00:00:00.000Z') + const mapper = createMapper('mexc', localTimestamp) + + expect( + mapper.map( + { + id: 0, + code: 0, + msg: 'spot@public.aggre.deals.v3.api.pb@10ms@BTCUSDT' + }, + localTimestamp + ) + ).toEqual([]) + + expect( + mapper.map( + { + channel: 'spot@public.aggre.deals.v3.api.pb@10ms@XENUSDT', + symbol: 'XENUSDT', + sendTime: '1781337091703' + }, + localTimestamp + ) + ).toEqual([]) + + expect( + mapper.map( + { + channel: 'spot@public.aggre.depth.v3.api.pb@10ms@XENUSDT', + symbol: 'XENUSDT', + sendTime: '1781337091703' + }, + localTimestamp + ) + ).toEqual([]) + + expect( + mapper.map( + { + channel: 'spot@public.aggre.deals.v3.api.pb@10ms@BTCUSDT', + symbol: 'BTCUSDT', + sendTime: '1710000000000', + publicAggreDeals: { + deals: [ + { + price: '100.1', + quantity: '0.2', + tradeType: 1, + time: '1710000000001', + tradeId: '698165549569396736X0_698165549569396737X0' + }, + { + price: '100.2', + quantity: '0.3', + tradeType: 2, + time: '1710000000002', + tradeId: '698165549569396738X0_698165549569396739X0' + } + ], + eventType: 'spot@public.aggre.deals.v3.api.pb@10ms' + } + }, + localTimestamp + ) + ).toEqual([ + { + type: 'trade', + symbol: 'BTCUSDT', + exchange: 'mexc', + id: '698165549569396736X0_698165549569396737X0', + price: 100.1, + amount: 0.2, + side: 'buy', + timestamp: new Date('2024-03-09T16:00:00.001Z'), + localTimestamp + }, + { + type: 'trade', + symbol: 'BTCUSDT', + exchange: 'mexc', + id: '698165549569396738X0_698165549569396739X0', + price: 100.2, + amount: 0.3, + side: 'sell', + timestamp: new Date('2024-03-09T16:00:00.002Z'), + localTimestamp + } + ]) + + expect( + mapper.map( + { + channel: 'spot@public.aggre.depth.v3.api.pb@10ms@BTCUSDT', + symbol: 'BTCUSDT', + sendTime: '1710000000000', + publicAggreDepths: { + asks: [], + bids: [{ price: '99.9', quantity: '0.5' }], + eventType: 'spot@public.aggre.depth.v3.api.pb@10ms', + fromVersion: '101', + toVersion: '101' + } + }, + localTimestamp + ) + ).toEqual([]) + + expect( + mapper.map( + { + channel: 'spot@public.aggre.depth.v3.api.pb@10ms@BTCUSDT', + symbol: 'BTCUSDT', + sendTime: '1710000000001', + publicAggreDepths: { + asks: [{ price: '100.1', quantity: '0' }], + bids: [{ price: '99.7', quantity: '1.1' }], + eventType: 'spot@public.aggre.depth.v3.api.pb@10ms', + fromVersion: '102', + toVersion: '102' + } + }, + localTimestamp + ) + ).toEqual([]) + + expect( + mapper.map( + { + channel: 'spot@public.aggre.depth.snapshot.v3.api.pb@10ms@BTCUSDT', + symbol: 'BTCUSDT', + generated: true, + publicAggreDepthsSnapshot: { + asks: [['100.1', '1.2']], + bids: [['99.8', '2.3']], + lastUpdateId: 100, + timestamp: 1710000000002 + } + }, + localTimestamp + ) + ).toEqual([ + { + type: 'book_change', + symbol: 'BTCUSDT', + exchange: 'mexc', + isSnapshot: true, + bids: [ + { price: 99.8, amount: 2.3 }, + { price: 99.9, amount: 0.5 }, + { price: 99.7, amount: 1.1 } + ], + asks: [], + timestamp: new Date('2024-03-09T16:00:00.002Z'), + localTimestamp + } + ]) + + expect( + mapper.map( + { + channel: 'spot@public.aggre.bookTicker.v3.api.pb@10ms@XENUSDT', + symbol: 'XENUSDT', + sendTime: '1781337091703' + }, + localTimestamp + ) + ).toEqual([]) + + expect( + mapper.map( + { + channel: 'spot@public.aggre.depth.v3.api.pb@10ms@BTCUSDT', + symbol: 'BTCUSDT', + sendTime: '1710000000003', + publicAggreDepths: { + asks: [{ price: '100.1', quantity: '0' }], + bids: [{ price: '99.7', quantity: '1.1' }], + eventType: 'spot@public.aggre.depth.v3.api.pb@10ms', + fromVersion: '103', + toVersion: '103' + } + }, + localTimestamp + ) + ).toEqual([ + { + type: 'book_change', + symbol: 'BTCUSDT', + exchange: 'mexc', + isSnapshot: false, + bids: [{ price: 99.7, amount: 1.1 }], + asks: [{ price: 100.1, amount: 0 }], + timestamp: new Date('2024-03-09T16:00:00.003Z'), + localTimestamp + } + ]) + + expect( + mapper.map( + { + channel: 'spot@public.aggre.depth.v3.api.pb@10ms@BTCUSDT', + symbol: 'BTCUSDT', + sendTime: '1710000000005', + publicAggreDepths: { + asks: [{ price: '100.3', quantity: '0.4' }], + eventType: 'spot@public.aggre.depth.v3.api.pb@10ms', + fromVersion: '104', + toVersion: '104' + } + }, + localTimestamp + ) + ).toEqual([ + { + type: 'book_change', + symbol: 'BTCUSDT', + exchange: 'mexc', + isSnapshot: false, + bids: [], + asks: [{ price: 100.3, amount: 0.4 }], + timestamp: new Date('2024-03-09T16:00:00.005Z'), + localTimestamp + } + ]) + + expect( + mapper.map( + { + channel: 'spot@public.aggre.depth.v3.api.pb@10ms@BTCUSDT', + symbol: 'BTCUSDT', + sendTime: '1710000000006', + publicAggreDepths: { + bids: [{ price: '99.6', quantity: '0.8' }], + eventType: 'spot@public.aggre.depth.v3.api.pb@10ms', + fromVersion: '105', + toVersion: '105' + } + }, + localTimestamp + ) + ).toEqual([ + { + type: 'book_change', + symbol: 'BTCUSDT', + exchange: 'mexc', + isSnapshot: false, + bids: [{ price: 99.6, amount: 0.8 }], + asks: [], + timestamp: new Date('2024-03-09T16:00:00.006Z'), + localTimestamp + } + ]) + + expect( + mapper.map( + { + channel: 'spot@public.aggre.bookTicker.v3.api.pb@10ms@BTCUSDT', + symbol: 'BTCUSDT', + sendTime: '1710000000004', + publicAggreBookTicker: { + bidPrice: '99.9', + bidQuantity: '1.2', + askPrice: '100.1', + askQuantity: '2.3' + } + }, + localTimestamp + ) + ).toEqual([ + { + type: 'book_ticker', + symbol: 'BTCUSDT', + exchange: 'mexc', + askAmount: 2.3, + askPrice: 100.1, + bidPrice: 99.9, + bidAmount: 1.2, + timestamp: new Date('2024-03-09T16:00:00.004Z'), + localTimestamp + } + ]) + + expect( + mapper.map( + { + channel: 'spot@public.aggre.bookTicker.v3.api.pb@10ms@BTCUSDT', + symbol: 'BTCUSDT', + sendTime: '1710000000005', + publicAggreBookTicker: { + bidPrice: '100', + bidQuantity: '4', + askPrice: '', + askQuantity: '' + } + }, + localTimestamp + ) + ).toEqual([ + { + type: 'book_ticker', + symbol: 'BTCUSDT', + exchange: 'mexc', + askAmount: undefined, + askPrice: undefined, + bidPrice: 100, + bidAmount: 4, + timestamp: new Date('2024-03-09T16:00:00.005Z'), + localTimestamp + } + ]) +}) + +test('map mexc buffered depth updates with omitted side', () => { + const localTimestamp = new Date('2026-06-18T00:00:00.000Z') + const mapper = createMapper('mexc', localTimestamp) + + expect( + mapper.map( + { + channel: 'spot@public.aggre.depth.v3.api.pb@10ms@BTCUSDT', + symbol: 'BTCUSDT', + sendTime: '1781740800130', + publicAggreDepths: { + asks: [ + { price: '64526.18', quantity: '0.32477055' }, + { price: '64849.35', quantity: '0.01209236' } + ], + eventType: 'spot@public.aggre.depth.v3.api.pb@10ms', + fromVersion: '74483042021', + toVersion: '74483042022' + } + }, + localTimestamp + ) + ).toEqual([]) + + expect( + mapper.map( + { + channel: 'spot@public.aggre.depth.snapshot.v3.api.pb@10ms@BTCUSDT', + symbol: 'BTCUSDT', + generated: true, + publicAggreDepthsSnapshot: { + asks: [['64530.00', '1']], + bids: [['64520.00', '1']], + lastUpdateId: 74483042020, + timestamp: 1781740800000 + } + }, + localTimestamp + ) + ).toEqual([ + { + type: 'book_change', + symbol: 'BTCUSDT', + exchange: 'mexc', + isSnapshot: true, + bids: [{ price: 64520, amount: 1 }], + asks: [ + { price: 64530, amount: 1 }, + { price: 64526.18, amount: 0.32477055 }, + { price: 64849.35, amount: 0.01209236 } + ], + timestamp: new Date('2026-06-18T00:00:00.000Z'), + localTimestamp + } + ]) +}) + +test('map mexc historical buffered depth updates without enforcing full sequence', () => { + const localTimestamp = new Date('2026-06-18T00:00:00.000Z') + const mapper = createMapper('mexc', localTimestamp) + + expect( + mapper.map( + { + channel: 'spot@public.aggre.depth.v3.api.pb@10ms@BTCUSDT', + symbol: 'BTCUSDT', + sendTime: '1781740800001', + publicAggreDepths: { + bids: [{ price: '99', quantity: '1' }], + eventType: 'spot@public.aggre.depth.v3.api.pb@10ms', + fromVersion: '101', + toVersion: '101' + } + }, + localTimestamp + ) + ).toEqual([]) + + expect( + mapper.map( + { + channel: 'spot@public.aggre.depth.v3.api.pb@10ms@BTCUSDT', + symbol: 'BTCUSDT', + sendTime: '1781740800002', + publicAggreDepths: { + asks: [{ price: '101', quantity: '1' }], + eventType: 'spot@public.aggre.depth.v3.api.pb@10ms', + fromVersion: '103', + toVersion: '103' + } + }, + localTimestamp + ) + ).toEqual([]) + + expect( + mapper.map( + { + channel: 'spot@public.aggre.depth.snapshot.v3.api.pb@10ms@BTCUSDT', + symbol: 'BTCUSDT', + generated: true, + publicAggreDepthsSnapshot: { + asks: [], + bids: [], + lastUpdateId: 100, + timestamp: 1781740800003 + } + }, + localTimestamp + ) + ).toEqual([ + { + type: 'book_change', + symbol: 'BTCUSDT', + exchange: 'mexc', + isSnapshot: true, + bids: [{ price: 99, amount: 1 }], + asks: [{ price: 101, amount: 1 }], + timestamp: new Date('2026-06-18T00:00:00.003Z'), + localTimestamp + } + ]) +}) + +test('map mexc realtime depth update throws when first update has no snapshot overlap', () => { + const localTimestamp = new Date() + const mapper = createMapper('mexc', localTimestamp) + + mapper.map( + { + channel: 'spot@public.aggre.depth.snapshot.v3.api.pb@10ms@BTCUSDT', + symbol: 'BTCUSDT', + generated: true, + publicAggreDepthsSnapshot: { + asks: [], + bids: [], + lastUpdateId: 100, + timestamp: Date.now() + } + }, + localTimestamp + ) + + expect(() => + mapper.map( + { + channel: 'spot@public.aggre.depth.v3.api.pb@10ms@BTCUSDT', + symbol: 'BTCUSDT', + sendTime: Date.now().toString(), + publicAggreDepths: { + asks: [{ price: '101', quantity: '1' }], + eventType: 'spot@public.aggre.depth.v3.api.pb@10ms', + fromVersion: '102', + toVersion: '102' + } + }, + localTimestamp + ) + ).toThrow('MEXC depth snapshot has no overlap with first update') +}) + +test('map mexc live captured messages', () => { + const localTimestamp = new Date('2026-06-23T09:46:45.000Z') + const mapper = createMapper('mexc', localTimestamp) + + expect( + mapper.map( + { + channel: 'spot@public.aggre.depth.snapshot.v3.api.pb@10ms@BTCUSDT', + symbol: 'BTCUSDT', + generated: true, + publicAggreDepthsSnapshot: { + asks: [], + bids: [], + lastUpdateId: 74891313262, + timestamp: 1782208005000 + } + }, + localTimestamp + ) + ).toEqual([ + { + type: 'book_change', + symbol: 'BTCUSDT', + exchange: 'mexc', + isSnapshot: true, + bids: [], + asks: [], + timestamp: new Date('2026-06-23T09:46:45.000Z'), + localTimestamp + } + ]) + + expect( + mapper.map( + { + channel: 'spot@public.aggre.depth.v3.api.pb@10ms@BTCUSDT', + symbol: 'BTCUSDT', + sendTime: '1782208005090', + publicAggreDepths: { + asks: [], + bids: [{ price: '61759.73', quantity: '0.0001778' }], + eventType: 'spot@public.aggre.depth.v3.api.pb@10ms', + fromVersion: '74891313263', + toVersion: '74891313263' + } + }, + localTimestamp + ) + ).toEqual([ + { + type: 'book_change', + symbol: 'BTCUSDT', + exchange: 'mexc', + isSnapshot: false, + bids: [{ price: 61759.73, amount: 0.0001778 }], + asks: [], + timestamp: new Date('2026-06-23T09:46:45.090Z'), + localTimestamp + } + ]) + + expect( + mapper.map( + { + channel: 'spot@public.aggre.bookTicker.v3.api.pb@10ms@BTCUSDT', + symbol: 'BTCUSDT', + sendTime: '1782208005090', + publicAggreBookTicker: { + bidPrice: '62383.56', + bidQuantity: '0.07956651', + askPrice: '62383.57', + askQuantity: '0.09556' + } + }, + localTimestamp + ) + ).toEqual([ + { + type: 'book_ticker', + symbol: 'BTCUSDT', + exchange: 'mexc', + askAmount: 0.09556, + askPrice: 62383.57, + bidPrice: 62383.56, + bidAmount: 0.07956651, + timestamp: new Date('2026-06-23T09:46:45.090Z'), + localTimestamp + } + ]) + + expect( + mapper.map( + { + channel: 'spot@public.aggre.deals.v3.api.pb@10ms@BTCUSDT', + symbol: 'BTCUSDT', + sendTime: '1782208006171', + publicAggreDeals: { + deals: [ + { price: '62383.56', quantity: '0.00008235', tradeType: 2, time: '1782208006167' }, + { price: '62383.57', quantity: '0.00019595', tradeType: 1, time: '1782208006167' }, + { price: '62383.57', quantity: '0.00059353', tradeType: 1, time: '1782208006167' }, + { price: '62383.56', quantity: '0.00027168', tradeType: 2, time: '1782208006168' }, + { price: '62383.57', quantity: '0.00002555', tradeType: 2, time: '1782208006168' } + ], + eventType: 'spot@public.aggre.deals.v3.api.pb@10ms' + } + }, + localTimestamp + ) + ).toEqual([ + { + type: 'trade', + symbol: 'BTCUSDT', + exchange: 'mexc', + id: undefined, + price: 62383.56, + amount: 0.00008235, + side: 'sell', + timestamp: new Date('2026-06-23T09:46:46.167Z'), + localTimestamp + }, + { + type: 'trade', + symbol: 'BTCUSDT', + exchange: 'mexc', + id: undefined, + price: 62383.57, + amount: 0.00019595, + side: 'buy', + timestamp: new Date('2026-06-23T09:46:46.167Z'), + localTimestamp + }, + { + type: 'trade', + symbol: 'BTCUSDT', + exchange: 'mexc', + id: undefined, + price: 62383.57, + amount: 0.00059353, + side: 'buy', + timestamp: new Date('2026-06-23T09:46:46.167Z'), + localTimestamp + }, + { + type: 'trade', + symbol: 'BTCUSDT', + exchange: 'mexc', + id: undefined, + price: 62383.56, + amount: 0.00027168, + side: 'sell', + timestamp: new Date('2026-06-23T09:46:46.168Z'), + localTimestamp + }, + { + type: 'trade', + symbol: 'BTCUSDT', + exchange: 'mexc', + id: undefined, + price: 62383.57, + amount: 0.00002555, + side: 'sell', + timestamp: new Date('2026-06-23T09:46:46.168Z'), + localTimestamp + } + ]) +}) + test('map polymarket messages', () => { const localTimestamp = new Date('2026-05-11T06:30:00.000Z') diff --git a/test/mexc-realtimefeed.test.ts b/test/mexc-realtimefeed.test.ts new file mode 100644 index 0000000..af95a93 --- /dev/null +++ b/test/mexc-realtimefeed.test.ts @@ -0,0 +1,386 @@ +import type { AddressInfo } from 'net' +import { createServer } from 'http' +import { Filter } from '../src/types.ts' +import { MexcRealTimeFeed } from '../src/realtimefeeds/mexc.ts' +import { getRealTimeFeedFactory } from '../src/realtimefeeds/index.ts' + +class TestMexcRealTimeFeed extends MexcRealTimeFeed { + protected readonly httpURL: string + + constructor(exchange: 'mexc', filters: Filter[], timeoutIntervalMS: number | undefined, httpURL = 'https://api.mexc.com') { + super(exchange, filters, timeoutIntervalMS) + this.httpURL = httpURL + } + + map(filters: Filter[]) { + return this.mapToSubscribeMessages(filters) + } + + parse(message: Buffer) { + return this.parseMessage(message) + } + + isError(message: any) { + return this.messageIsError(message) + } + + isHeartbeat(message: any) { + return this.messageIsHeartbeat(message) + } + + observe(message: any) { + this.onMessage(message) + } + + async provideSnapshots(filters: Filter[], shouldCancel = () => false) { + await this.provideManualSnapshots(filters, shouldCancel) + return this.manualSnapshotsBuffer + } +} + +test('register mexc realtime feed', () => { + expect(getRealTimeFeedFactory('mexc')).toBeDefined() +}) + +test('map mexc realtime subscriptions', () => { + const feed = new TestMexcRealTimeFeed('mexc', [], undefined) + + const subscribeMessages = feed.map([ + { + channel: 'spot@public.aggre.deals.v3.api.pb@10ms', + symbols: ['btcusdt', 'ETHUSDT'] + }, + { + channel: 'spot@public.aggre.depth.v3.api.pb@10ms', + symbols: ['BTCUSDT'] + }, + { + channel: 'spot@public.aggre.depth.snapshot.v3.api.pb@10ms', + symbols: ['BTCUSDT'] + }, + { + channel: 'spot@public.aggre.bookTicker.v3.api.pb@10ms', + symbols: ['BTCUSDT'] + } + ]) + + expect(subscribeMessages).toMatchSnapshot() +}) + +test('mexc realtime subscriptions require symbols', () => { + const feed = new TestMexcRealTimeFeed('mexc', [], undefined) + + expect(() => + feed.map([ + { + channel: 'spot@public.aggre.deals.v3.api.pb@10ms' + } + ]) + ).toThrow('MexcRealTimeFeed requires explicitly specified symbols when subscribing to live feed') +}) + +test('mexc realtime rejects unsupported channels', () => { + const feed = new TestMexcRealTimeFeed('mexc', [], undefined) + + expect(() => + feed.map([ + { + channel: 'unsupported', + symbols: ['BTCUSDT'] + } + ]) + ).toThrow('MexcRealTimeFeed unsupported channel unsupported') +}) + +test('mexc snapshot filters require matching depth filters', () => { + const feed = new TestMexcRealTimeFeed('mexc', [], undefined) + + expect(() => + feed.map([ + { + channel: 'spot@public.aggre.depth.snapshot.v3.api.pb@10ms', + symbols: ['BTCUSDT'] + } + ]) + ).toThrow( + 'MexcRealTimeFeed requires spot@public.aggre.depth.v3.api.pb@10ms for every spot@public.aggre.depth.snapshot.v3.api.pb@10ms symbol' + ) + + expect(() => + feed.map([ + { + channel: 'spot@public.aggre.depth.v3.api.pb@10ms', + symbols: ['ETHUSDT'] + }, + { + channel: 'spot@public.aggre.depth.snapshot.v3.api.pb@10ms', + symbols: ['BTCUSDT'] + } + ]) + ).toThrow( + 'MexcRealTimeFeed requires spot@public.aggre.depth.v3.api.pb@10ms for every spot@public.aggre.depth.snapshot.v3.api.pb@10ms symbol' + ) +}) + +test('classify mexc realtime control messages', () => { + const feed = new TestMexcRealTimeFeed('mexc', [], undefined) + + expect(feed.isHeartbeat({ msg: 'PONG' })).toBe(true) + expect(feed.isError({ code: 0, msg: 'spot@public.aggre.deals.v3.api.pb@10ms@BTCUSDT' })).toBe(false) + expect(feed.isError({ code: 1, msg: 'invalid request' })).toBe(true) +}) + +test('decode mexc realtime protobuf trade message', () => { + const feed = new TestMexcRealTimeFeed('mexc', [], undefined) + const trade = message(stringField(1, '100.1'), stringField(2, '0.2'), varintField(3, 1), varintField(4, 1710000000001)) + const publicAggreDeals = message(bytesField(1, trade), stringField(2, 'spot@public.aggre.deals.v3.api.pb@10ms')) + const wrapper = message( + stringField(1, 'spot@public.aggre.deals.v3.api.pb@10ms@BTCUSDT'), + stringField(3, 'BTCUSDT'), + varintField(6, 1710000000000), + bytesField(314, publicAggreDeals) + ) + + expect(feed.parse(wrapper)).toEqual({ + channel: 'spot@public.aggre.deals.v3.api.pb@10ms@BTCUSDT', + symbol: 'BTCUSDT', + sendTime: '1710000000000', + publicAggreDeals: { + deals: [ + { + price: '100.1', + quantity: '0.2', + tradeType: 1, + time: '1710000000001' + } + ], + eventType: 'spot@public.aggre.deals.v3.api.pb@10ms' + } + }) +}) + +test('decode mexc realtime protobuf depth message', () => { + const feed = new TestMexcRealTimeFeed('mexc', [], undefined) + const ask = message(stringField(1, '100.1'), stringField(2, '1.2')) + const bid = message(stringField(1, '99.9'), stringField(2, '0.5')) + const publicAggreDepths = message( + bytesField(1, ask), + bytesField(2, bid), + stringField(3, 'spot@public.aggre.depth.v3.api.pb@10ms'), + stringField(4, '101'), + stringField(5, '102') + ) + const wrapper = message( + stringField(1, 'spot@public.aggre.depth.v3.api.pb@10ms@BTCUSDT'), + stringField(3, 'BTCUSDT'), + varintField(6, 1710000000000), + bytesField(313, publicAggreDepths) + ) + + expect(feed.parse(wrapper)).toEqual({ + channel: 'spot@public.aggre.depth.v3.api.pb@10ms@BTCUSDT', + symbol: 'BTCUSDT', + sendTime: '1710000000000', + publicAggreDepths: { + asks: [{ price: '100.1', quantity: '1.2' }], + bids: [{ price: '99.9', quantity: '0.5' }], + eventType: 'spot@public.aggre.depth.v3.api.pb@10ms', + fromVersion: '101', + toVersion: '102' + } + }) +}) + +test('decode mexc realtime protobuf book ticker message', () => { + const feed = new TestMexcRealTimeFeed('mexc', [], undefined) + const publicAggreBookTicker = message(stringField(1, '99.9'), stringField(2, '1.2'), stringField(3, '100.1'), stringField(4, '2.3')) + const wrapper = message( + stringField(1, 'spot@public.aggre.bookTicker.v3.api.pb@10ms@BTCUSDT'), + stringField(3, 'BTCUSDT'), + varintField(6, 1710000000000), + bytesField(315, publicAggreBookTicker) + ) + + expect(feed.parse(wrapper)).toEqual({ + channel: 'spot@public.aggre.bookTicker.v3.api.pb@10ms@BTCUSDT', + symbol: 'BTCUSDT', + sendTime: '1710000000000', + publicAggreBookTicker: { + bidPrice: '99.9', + bidQuantity: '1.2', + askPrice: '100.1', + askQuantity: '2.3' + } + }) +}) + +test('provide mexc manual depth snapshots', async () => { + const server = await startSnapshotServer() + const feed = new TestMexcRealTimeFeed('mexc', [], undefined, server.url) + const originalDateNow = Date.now + + Date.now = () => 1710000000000 + + try { + const filters = [ + { + channel: 'spot@public.aggre.depth.v3.api.pb@10ms', + symbols: ['btcusdt'] + }, + { + channel: 'spot@public.aggre.depth.snapshot.v3.api.pb@10ms', + symbols: ['btcusdt'] + } + ] + + feed.map(filters) + feed.observe({ + channel: 'spot@public.aggre.depth.v3.api.pb@10ms@BTCUSDT', + symbol: 'BTCUSDT', + publicAggreDepths: { + fromVersion: '101', + toVersion: '101' + } + }) + + const snapshots = await feed.provideSnapshots(filters) + + expect(snapshots).toEqual([ + { + channel: 'spot@public.aggre.depth.snapshot.v3.api.pb@10ms@BTCUSDT', + symbol: 'BTCUSDT', + generated: true, + publicAggreDepthsSnapshot: { + lastUpdateId: 100, + asks: [['100.1', '1.2']], + bids: [['99.9', '0.5']], + timestamp: 1782235749967 + } + } + ]) + } finally { + Date.now = originalDateNow + await server.close() + } +}) + +test('retry mexc manual depth snapshots until buffered update overlaps', async () => { + const server = await startSnapshotServer([ + { lastUpdateId: 101, asks: [['100.1', '1.2']], bids: [['99.9', '0.5']], timestamp: 1782235749964 }, + { lastUpdateId: 102, asks: [['100.2', '1.2']], bids: [['99.8', '0.5']], timestamp: 1782235749965 }, + { lastUpdateId: 103, asks: [['100.3', '1.2']], bids: [['99.7', '0.5']], timestamp: 1782235749966 }, + { lastUpdateId: 104, asks: [['100.4', '1.2']], bids: [['99.6', '0.5']], timestamp: 1782235749967 } + ]) + const feed = new TestMexcRealTimeFeed('mexc', [], undefined, server.url) + const originalDateNow = Date.now + + Date.now = () => 1710000000000 + + try { + const filters = [ + { + channel: 'spot@public.aggre.depth.v3.api.pb@10ms', + symbols: ['btcusdt'] + }, + { + channel: 'spot@public.aggre.depth.snapshot.v3.api.pb@10ms', + symbols: ['btcusdt'] + } + ] + + feed.map(filters) + feed.observe({ + channel: 'spot@public.aggre.depth.v3.api.pb@10ms@BTCUSDT', + symbol: 'BTCUSDT', + publicAggreDepths: { + fromVersion: '105', + toVersion: '105' + } + }) + + const snapshots = await feed.provideSnapshots(filters) + + expect(server.requestsCount).toBe(4) + expect(snapshots).toEqual([ + { + channel: 'spot@public.aggre.depth.snapshot.v3.api.pb@10ms@BTCUSDT', + symbol: 'BTCUSDT', + generated: true, + publicAggreDepthsSnapshot: { + lastUpdateId: 104, + asks: [['100.4', '1.2']], + bids: [['99.6', '0.5']], + timestamp: 1782235749967 + } + } + ]) + } finally { + Date.now = originalDateNow + await server.close() + } +}) + +function message(...fields: Buffer[]) { + return Buffer.concat(fields) +} + +function stringField(fieldNumber: number, value: string) { + return bytesField(fieldNumber, Buffer.from(value)) +} + +function bytesField(fieldNumber: number, value: Buffer) { + return Buffer.concat([tag(fieldNumber, 2), varint(value.length), value]) +} + +function varintField(fieldNumber: number, value: number) { + return Buffer.concat([tag(fieldNumber, 0), varint(value)]) +} + +function tag(fieldNumber: number, wireType: number) { + return varint(fieldNumber * 8 + wireType) +} + +function varint(value: number) { + const bytes: number[] = [] + let remaining = BigInt(value) + while (remaining >= 0x80n) { + bytes.push(Number((remaining & 0x7fn) | 0x80n)) + remaining >>= 7n + } + bytes.push(Number(remaining)) + return Buffer.from(bytes) +} + +async function startSnapshotServer( + responses: MexcTestDepthSnapshotResponse[] = [ + { lastUpdateId: 100, asks: [['100.1', '1.2']], bids: [['99.9', '0.5']], timestamp: 1782235749967 } + ] +) { + let requestsCount = 0 + const server = createServer((request, response) => { + expect(request.url).toBe('/api/v3/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 MexcTestDepthSnapshotResponse = { + lastUpdateId: number + bids: string[][] + asks: string[][] + timestamp: number +} diff --git a/test/mexc.live.test.ts b/test/mexc.live.test.ts new file mode 100644 index 0000000..2464ada --- /dev/null +++ b/test/mexc.live.test.ts @@ -0,0 +1,69 @@ +import { normalizeBookChanges, normalizeBookTickers, normalizeTrades, streamNormalized } from '../dist/index.js' +import { describeLive } from './live.js' + +describeLive('mexc live', () => { + test('streams normalized BTCUSDT data for all mappers', async () => { + const messages = streamNormalized( + { + exchange: 'mexc', + symbols: ['BTCUSDT'], + timeoutIntervalMS: 20_000, + withDisconnectMessages: true + }, + normalizeTrades, + normalizeBookChanges, + normalizeBookTickers + ) + + const seen = { + trade: false, + bookSnapshot: false, + bookDelta: false, + bookTicker: false + } + let sawDisconnect = false + + try { + for await (const message of messages) { + if (message.type === 'disconnect') { + sawDisconnect = true + continue + } + + if (message.symbol !== 'BTCUSDT') { + continue + } + + if (message.type === 'trade') { + seen.trade = message.amount > 0 && Number.isFinite(message.price) + } + + if (message.type === 'book_ticker') { + seen.bookTicker = Number.isFinite(message.askPrice ?? NaN) || Number.isFinite(message.bidPrice ?? NaN) + } + + if (message.type === 'book_change' && message.isSnapshot) { + seen.bookSnapshot = message.asks.length > 0 || message.bids.length > 0 + } + + if (message.type === 'book_change' && !message.isSnapshot) { + seen.bookDelta = true + } + + if (Object.values(seen).every(Boolean)) { + break + } + } + } finally { + await messages.return?.() + } + + expect(sawDisconnect).toBe(false) + expect(seen).toEqual({ + trade: true, + bookSnapshot: true, + bookDelta: true, + bookTicker: true + }) + }, 40_000) +})