From 5a4ae3c567e5d1805f3d56b091eb67226462cb0d Mon Sep 17 00:00:00 2001 From: marcin Date: Thu, 11 Jun 2026 17:55:57 +0200 Subject: [PATCH 01/17] feat(mexc-integration): added mappers --- src/consts.ts | 8 ++ src/mappers/index.ts | 4 + src/mappers/mexc.ts | 254 +++++++++++++++++++++++++++++++++++++++++++ test/mappers.test.ts | 166 ++++++++++++++++++++++++++++ 4 files changed, 432 insertions(+) create mode 100644 src/mappers/mexc.ts diff --git a/src/consts.ts b/src/consts.ts index 63362ba..0c3c011 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,12 @@ 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.bookTicker.v3.api.pb@100ms' +] as const + const POLYMARKET_CHANNELS = [ 'book', 'price_change', @@ -597,5 +604,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 185f808..972dcaf 100644 --- a/src/mappers/index.ts +++ b/src/mappers/index.ts @@ -140,6 +140,7 @@ import { LighterLiquidationMapper, LighterTradesMapper } from './lighter.ts' +import { MexcBookChangeMapper, MexcBookTickerMapper, MexcTradesMapper } from './mexc.ts' import { krakenBookChangeMapper, krakenBookTickerMapper, krakenTradesMapper } from './kraken.ts' import { KucoinBookChangeMapper, KucoinBookTickerMapper, KucoinTradesMapper } from './kucoin.ts' import { @@ -354,6 +355,7 @@ const tradesMappers = { hyperliquid: () => new HyperliquidTradesMapper(), lighter: () => new LighterTradesMapper(), bullish: () => new BullishTradesMapper(), + mexc: () => new MexcTradesMapper(), polymarket: () => new PolymarketTradesMapper() } @@ -456,6 +458,7 @@ const bookChangeMappers = { hyperliquid: () => new HyperliquidBookChangeMapper(), lighter: () => new LighterBookChangeMapper(), bullish: () => new BullishBookChangeMapper(), + mexc: () => new MexcBookChangeMapper(), polymarket: () => new PolymarketBookChangeMapper() } @@ -606,6 +609,7 @@ const bookTickersMappers = { hyperliquid: () => new HyperliquidBookTickerMapper(), lighter: () => new LighterBookTickerMapper(), bullish: () => new BullishBookTickerMapper(), + mexc: () => new MexcBookTickerMapper(), 'binance-european-options': () => new BinanceEuropeanOptionsBookTickerMapper(), polymarket: () => new PolymarketBookTickerMapper() } diff --git a/src/mappers/mexc.ts b/src/mappers/mexc.ts new file mode 100644 index 0000000..e470404 --- /dev/null +++ b/src/mappers/mexc.ts @@ -0,0 +1,254 @@ +import { CircularBuffer, upperCaseSymbols } from '../handy.ts' +import { BookChange, BookPriceLevel, BookTicker, Trade } from '../types.ts' +import { Mapper } from './mapper.ts' + +export class MexcTradesMapper implements Mapper<'mexc', Trade> { + private readonly channel = 'spot@public.aggre.deals.v3.api.pb@10ms' + + canHandle(message: MexcTradeMessage) { + return message.channel === this.channel || message.channel.startsWith(`${this.channel}@`) + } + + 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: undefined, + 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 symbolDepthInfo: Record = {} + + canHandle(message: MexcDepthSnapshotMessage | MexcDepthUpdateMessage) { + return message.channel === this.channel || message.channel.startsWith(`${this.channel}@`) + } + + getFilters(symbols?: string[]) { + return [{ channel: this.channel, symbols: upperCaseSymbols(symbols) } as const] + } + + *map(message: MexcDepthSnapshotMessage | MexcDepthUpdateMessage, localTimestamp: Date): IterableIterator { + const depthInfo = this.getDepthInfoFor(message.symbol) + if (message.generated === true) { + if (depthInfo.snapshotEmitted) { + return + } + + const snapshotVersion = Number(message.publicAggreDepths.toVersion) + const bids = message.publicAggreDepths.bids.map(this.mapBookLevel) + const asks = message.publicAggreDepths.asks.map(this.mapBookLevel) + + for (const update of depthInfo.updates.items()) { + const fromVersion = Number(update.publicAggreDepths.fromVersion) + const toVersion = Number(update.publicAggreDepths.toVersion) + if (snapshotVersion + 1 < fromVersion || snapshotVersion >= toVersion) { + continue + } + 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)) + } + depthInfo.currentBookVersion = toVersion + } + + depthInfo.updates.clear() + depthInfo.currentBookVersion = depthInfo.currentBookVersion ?? snapshotVersion + depthInfo.snapshotEmitted = true + + yield { + type: 'book_change', + symbol: message.symbol, + exchange: 'mexc', + isSnapshot: true, + bids, + asks, + timestamp: new Date(Number(message.sendTime)), + 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) { + throw new Error( + `MEXC depth snapshot has no overlap with first update, update ${JSON.stringify(message)}, currentBookVersion: ${ + depthInfo.currentBookVersion + }` + ) + } + + 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 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@100ms' + + canHandle(message: MexcBookTickerMessage) { + return message.channel === this.channel || message.channel.startsWith(`${this.channel}@`) + } + + 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: Number(message.publicAggreBookTicker.askQuantity), + askPrice: Number(message.publicAggreBookTicker.askPrice), + bidAmount: Number(message.publicAggreBookTicker.bidQuantity), + bidPrice: Number(message.publicAggreBookTicker.bidPrice), + timestamp: new Date(Number(message.sendTime)), + localTimestamp + } + } +} + +type MexcMappedChannel = + | 'spot@public.aggre.deals.v3.api.pb@10ms' + | 'spot@public.aggre.depth.v3.api.pb@10ms' + | 'spot@public.aggre.bookTicker.v3.api.pb@100ms' +type MexcChannelWithSymbol = TChannel | `${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 +} + +enum MexcTradeType { + Buy = 1, + Sell = 2 +} + +type MexcDepthSnapshotMessage = MexcProtobufMessage<'spot@public.aggre.depth.v3.api.pb@10ms'> & { + generated: true + publicAggreDepths: MexcAggreDepths +} + +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 +} + +type MexcPriceLevel = { + price: string + quantity: string +} + +type MexcBookTickerMessage = MexcProtobufMessage<'spot@public.aggre.bookTicker.v3.api.pb@100ms'> & { + publicAggreBookTicker: { + bidPrice: string + bidQuantity: string + askPrice: string + askQuantity: string + } +} + +type MexcDepthInfo = { + isContinuityValidated?: boolean + currentBookVersion?: number + snapshotEmitted?: boolean + updates: CircularBuffer +} diff --git a/test/mappers.test.ts b/test/mappers.test.ts index 3821678..2a44b87 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,171 @@ 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( + { + 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' + }, + { + price: '100.2', + quantity: '0.3', + tradeType: 2, + time: '1710000000002' + } + ], + eventType: 'spot@public.aggre.deals.v3.api.pb@10ms' + } + }, + localTimestamp + ) + ).toEqual([ + { + type: 'trade', + symbol: 'BTCUSDT', + exchange: 'mexc', + id: undefined, + 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: undefined, + 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', + symbol: 'BTCUSDT', + sendTime: '1710000000002', + generated: true, + publicAggreDepths: { + asks: [{ price: '100.1', quantity: '1.2' }], + bids: [{ price: '99.8', quantity: '2.3' }], + eventType: 'spot@public.aggre.depth.v3.api.pb@10ms', + fromVersion: '100', + toVersion: '100' + } + }, + localTimestamp + ) + ).toEqual([ + { + type: 'book_change', + symbol: 'BTCUSDT', + exchange: 'mexc', + isSnapshot: true, + bids: [ + { price: 99.8, amount: 2.3 }, + { price: 99.9, amount: 0.5 } + ], + asks: [{ price: 100.1, amount: 1.2 }], + 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: '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: '102', + toVersion: '102' + } + }, + 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.bookTicker.v3.api.pb@100ms@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 + } + ]) +}) + test('map polymarket messages', () => { const localTimestamp = new Date('2026-05-11T06:30:00.000Z') From 90bdcfa3c24a957ae42357f4a9dfd5c083c8cd9f Mon Sep 17 00:00:00 2001 From: marcin Date: Thu, 11 Jun 2026 18:30:43 +0200 Subject: [PATCH 02/17] feat(mexc-integration): added possibility to override message parsing for real time feeds --- src/realtimefeeds/index.ts | 2 ++ src/realtimefeeds/realtimefeed.ts | 6 +++++- 2 files changed, 7 insertions(+), 1 deletion(-) 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/realtimefeed.ts b/src/realtimefeeds/realtimefeed.ts index 5ee978d..4b11236 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) { From e5d351c9150d9267de938e94607c4e4ffafa3fcd Mon Sep 17 00:00:00 2001 From: marcin Date: Thu, 11 Jun 2026 18:34:51 +0200 Subject: [PATCH 03/17] feat(mexc-integration): added real time feed --- src/realtimefeeds/mexc.ts | 278 +++++++++++++++++++++++++++++++++ test/mexc-realtimefeed.test.ts | 204 ++++++++++++++++++++++++ 2 files changed, 482 insertions(+) create mode 100644 src/realtimefeeds/mexc.ts create mode 100644 test/mexc-realtimefeed.test.ts diff --git a/src/realtimefeeds/mexc.ts b/src/realtimefeeds/mexc.ts new file mode 100644 index 0000000..511a1c0 --- /dev/null +++ b/src/realtimefeeds/mexc.ts @@ -0,0 +1,278 @@ +import { Filter } from '../types.ts' +import { RealTimeFeedBase } from './realtimefeed.ts' + +type ProtobufWireType = 0 | 1 | 2 | 3 | 4 | 5 + +export class MexcRealTimeFeed extends RealTimeFeedBase { + protected readonly wssURL = 'wss://wbs-api.mexc.com/ws' + private readonly channels = new Set([ + 'spot@public.aggre.deals.v3.api.pb@10ms', + 'spot@public.aggre.depth.v3.api.pb@10ms', + 'spot@public.aggre.bookTicker.v3.api.pb@100ms' + ]) + + protected 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> + }) + + return [ + { + method: 'SUBSCRIPTION', + params: filtersWithSymbols.flatMap((filter) => filter.symbols.map((symbol) => `${filter.channel}@${symbol.toUpperCase()}`)) + } + ] + } + + protected parseMessage(message: Buffer): any { + if (message.length > 0 && message[0] === 123) { + return JSON.parse(message.toString()) + } + + return decodePushDataV3ApiWrapper(message) + } + + protected messageIsError(message: any): boolean { + return message.code !== undefined && message.code !== 0 + } + + protected messageIsHeartbeat(message: any) { + return message.msg === 'PONG' + } + + protected sendCustomPing = () => { + this.send({ method: 'PING' }) + } +} + +function decodePushDataV3ApiWrapper(buffer: Buffer) { + const reader = new ProtobufReader(buffer) + const message: Record = {} + + while (!reader.done) { + const { fieldNumber, wireType } = reader.readTag() + + if (fieldNumber === 1 && wireType === 2) { + message.channel = reader.readString() + } else if (fieldNumber === 3 && wireType === 2) { + message.symbol = reader.readString() + } else if (fieldNumber === 4 && wireType === 2) { + message.symbolId = reader.readString() + } else if (fieldNumber === 5 && wireType === 0) { + message.createTime = reader.readInt64String() + } else if (fieldNumber === 6 && wireType === 0) { + message.sendTime = reader.readInt64String() + } else if (fieldNumber === 313 && wireType === 2) { + message.publicAggreDepths = decodePublicAggreDepths(reader.readBytes()) + } else if (fieldNumber === 314 && wireType === 2) { + message.publicAggreDeals = decodePublicAggreDeals(reader.readBytes()) + } else if (fieldNumber === 315 && wireType === 2) { + message.publicAggreBookTicker = decodePublicAggreBookTicker(reader.readBytes()) + } else { + reader.skip(wireType) + } + } + + return message +} + +function decodePublicAggreDeals(buffer: Uint8Array) { + const reader = new ProtobufReader(buffer) + const message: { deals: any[]; eventType?: string } = { deals: [] } + + while (!reader.done) { + const { fieldNumber, wireType } = reader.readTag() + if (fieldNumber === 1 && wireType === 2) { + message.deals.push(decodePublicAggreDeal(reader.readBytes())) + } else if (fieldNumber === 2 && wireType === 2) { + message.eventType = reader.readString() + } else { + reader.skip(wireType) + } + } + + return message +} + +function decodePublicAggreDeal(buffer: Uint8Array) { + const reader = new ProtobufReader(buffer) + const message: Record = {} + + while (!reader.done) { + const { fieldNumber, wireType } = reader.readTag() + if (fieldNumber === 1 && wireType === 2) { + message.price = reader.readString() + } else if (fieldNumber === 2 && wireType === 2) { + message.quantity = reader.readString() + } else if (fieldNumber === 3 && wireType === 0) { + message.tradeType = reader.readVarintNumber() + } else if (fieldNumber === 4 && wireType === 0) { + message.time = reader.readInt64String() + } else { + reader.skip(wireType) + } + } + + return message +} + +function decodePublicAggreDepths(buffer: Uint8Array) { + const reader = new ProtobufReader(buffer) + const message: { asks: any[]; bids: any[]; eventType?: string; fromVersion?: string; toVersion?: string } = { asks: [], bids: [] } + + while (!reader.done) { + const { fieldNumber, wireType } = reader.readTag() + if (fieldNumber === 1 && wireType === 2) { + message.asks.push(decodePublicAggreDepthLevel(reader.readBytes())) + } else if (fieldNumber === 2 && wireType === 2) { + message.bids.push(decodePublicAggreDepthLevel(reader.readBytes())) + } else if (fieldNumber === 3 && wireType === 2) { + message.eventType = reader.readString() + } else if (fieldNumber === 4 && wireType === 2) { + message.fromVersion = reader.readString() + } else if (fieldNumber === 5 && wireType === 2) { + message.toVersion = reader.readString() + } else { + reader.skip(wireType) + } + } + + return message +} + +function decodePublicAggreDepthLevel(buffer: Uint8Array) { + const reader = new ProtobufReader(buffer) + const message: Record = {} + + while (!reader.done) { + const { fieldNumber, wireType } = reader.readTag() + if (fieldNumber === 1 && wireType === 2) { + message.price = reader.readString() + } else if (fieldNumber === 2 && wireType === 2) { + message.quantity = reader.readString() + } else { + reader.skip(wireType) + } + } + + return message +} + +function decodePublicAggreBookTicker(buffer: Uint8Array) { + const reader = new ProtobufReader(buffer) + const message: Record = {} + + while (!reader.done) { + const { fieldNumber, wireType } = reader.readTag() + if (fieldNumber === 1 && wireType === 2) { + message.bidPrice = reader.readString() + } else if (fieldNumber === 2 && wireType === 2) { + message.bidQuantity = reader.readString() + } else if (fieldNumber === 3 && wireType === 2) { + message.askPrice = reader.readString() + } else if (fieldNumber === 4 && wireType === 2) { + message.askQuantity = reader.readString() + } else { + reader.skip(wireType) + } + } + + return message +} + +class ProtobufReader { + private offset = 0 + + constructor(private readonly buffer: Uint8Array) {} + + get done() { + return this.offset >= this.buffer.length + } + + readTag() { + const tag = this.readVarintNumber() + return { + fieldNumber: tag >>> 3, + wireType: (tag & 7) as ProtobufWireType + } + } + + readVarintNumber() { + return Number(this.readVarintBigInt()) + } + + readInt64String() { + return this.readVarintBigInt().toString() + } + + readString() { + return Buffer.from(this.readBytes()).toString('utf8') + } + + readBytes() { + const length = this.readVarintNumber() + const end = this.offset + length + if (end > this.buffer.length) { + throw new Error('MEXC protobuf message ended unexpectedly') + } + + const bytes = this.buffer.subarray(this.offset, end) + this.offset = end + return bytes + } + + skip(wireType: ProtobufWireType) { + if (wireType === 0) { + this.readVarintBigInt() + return + } + + if (wireType === 1) { + this.skipFixed(8) + return + } + + if (wireType === 2) { + this.readBytes() + return + } + + if (wireType === 5) { + this.skipFixed(4) + return + } + + throw new Error(`Unsupported MEXC protobuf wire type ${wireType}`) + } + + private readVarintBigInt() { + let shift = 0n + let value = 0n + + while (this.offset < this.buffer.length) { + const byte = this.buffer[this.offset++] + value |= BigInt(byte & 0x7f) << shift + if ((byte & 0x80) === 0) { + return value + } + shift += 7n + } + + throw new Error('MEXC protobuf varint ended unexpectedly') + } + + private skipFixed(bytes: number) { + this.offset += bytes + if (this.offset > this.buffer.length) { + throw new Error('MEXC protobuf message ended unexpectedly') + } + } +} diff --git a/test/mexc-realtimefeed.test.ts b/test/mexc-realtimefeed.test.ts new file mode 100644 index 0000000..97ecab9 --- /dev/null +++ b/test/mexc-realtimefeed.test.ts @@ -0,0 +1,204 @@ +import { Filter } from '../src/types.ts' +import { MexcRealTimeFeed } from '../src/realtimefeeds/mexc.ts' +import { getRealTimeFeedFactory } from '../src/realtimefeeds/index.ts' + +class TestMexcRealTimeFeed extends MexcRealTimeFeed { + 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) + } +} + +test('register mexc realtime feed', () => { + expect(getRealTimeFeedFactory('mexc')).toBeDefined() +}) + +test('map mexc realtime subscriptions', () => { + const feed = new TestMexcRealTimeFeed('mexc', [], undefined) + + expect( + 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.bookTicker.v3.api.pb@100ms', + symbols: ['BTCUSDT'] + } + ]) + ).toEqual([ + { + 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@100ms@BTCUSDT' + ] + } + ]) +}) + +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('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@100ms@BTCUSDT'), + stringField(3, 'BTCUSDT'), + varintField(6, 1710000000000), + bytesField(315, publicAggreBookTicker) + ) + + expect(feed.parse(wrapper)).toEqual({ + channel: 'spot@public.aggre.bookTicker.v3.api.pb@100ms@BTCUSDT', + symbol: 'BTCUSDT', + sendTime: '1710000000000', + publicAggreBookTicker: { + bidPrice: '99.9', + bidQuantity: '1.2', + askPrice: '100.1', + askQuantity: '2.3' + } + }) +}) + +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) +} From 19da8d7a518e7113d37cf1cdc9c82c0c732b09f9 Mon Sep 17 00:00:00 2001 From: marcin Date: Thu, 11 Jun 2026 19:02:42 +0200 Subject: [PATCH 04/17] feat(mexc-integration): using protobufjs --- package-lock.json | 19 ++ package.json | 1 + src/realtimefeeds/mexc.ts | 290 ++++++-------------------- test/mexcfutures-realtimefeed.test.ts | 68 ++++++ 4 files changed, 154 insertions(+), 224 deletions(-) create mode 100644 test/mexcfutures-realtimefeed.test.ts 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/realtimefeeds/mexc.ts b/src/realtimefeeds/mexc.ts index 511a1c0..6472616 100644 --- a/src/realtimefeeds/mexc.ts +++ b/src/realtimefeeds/mexc.ts @@ -1,9 +1,63 @@ +import protobuf from 'protobufjs' import { Filter } from '../types.ts' import { RealTimeFeedBase } from './realtimefeed.ts' -type ProtobufWireType = 0 | 1 | 2 | 3 | 4 | 5 - export class MexcRealTimeFeed extends RealTimeFeedBase { + 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; + } + + message PublicAggreDepthsV3Api { + repeated PublicAggreDepthV3ApiItem asks = 1; + repeated PublicAggreDepthV3ApiItem bids = 2; + string eventType = 3; + string fromVersion = 4; + string toVersion = 5; + } + + message PublicAggreDepthV3ApiItem { + string price = 1; + string quantity = 2; + } + + message PublicAggreBookTickerV3Api { + string bidPrice = 1; + string bidQuantity = 2; + string askPrice = 3; + string askQuantity = 4; + } + ` + protected readonly wssURL = 'wss://wbs-api.mexc.com/ws' private readonly channels = new Set([ 'spot@public.aggre.deals.v3.api.pb@10ms', @@ -37,7 +91,11 @@ export class MexcRealTimeFeed extends RealTimeFeedBase { return JSON.parse(message.toString()) } - return decodePushDataV3ApiWrapper(message) + const pushDataV3ApiWrapper = MexcRealTimeFeed.getPushDataV3ApiWrapper() + return pushDataV3ApiWrapper.toObject(pushDataV3ApiWrapper.decode(message), { + longs: String, + arrays: true + }) } protected messageIsError(message: any): boolean { @@ -51,228 +109,12 @@ export class MexcRealTimeFeed extends RealTimeFeedBase { protected sendCustomPing = () => { this.send({ method: 'PING' }) } -} - -function decodePushDataV3ApiWrapper(buffer: Buffer) { - const reader = new ProtobufReader(buffer) - const message: Record = {} - - while (!reader.done) { - const { fieldNumber, wireType } = reader.readTag() - - if (fieldNumber === 1 && wireType === 2) { - message.channel = reader.readString() - } else if (fieldNumber === 3 && wireType === 2) { - message.symbol = reader.readString() - } else if (fieldNumber === 4 && wireType === 2) { - message.symbolId = reader.readString() - } else if (fieldNumber === 5 && wireType === 0) { - message.createTime = reader.readInt64String() - } else if (fieldNumber === 6 && wireType === 0) { - message.sendTime = reader.readInt64String() - } else if (fieldNumber === 313 && wireType === 2) { - message.publicAggreDepths = decodePublicAggreDepths(reader.readBytes()) - } else if (fieldNumber === 314 && wireType === 2) { - message.publicAggreDeals = decodePublicAggreDeals(reader.readBytes()) - } else if (fieldNumber === 315 && wireType === 2) { - message.publicAggreBookTicker = decodePublicAggreBookTicker(reader.readBytes()) - } else { - reader.skip(wireType) - } - } - return message -} - -function decodePublicAggreDeals(buffer: Uint8Array) { - const reader = new ProtobufReader(buffer) - const message: { deals: any[]; eventType?: string } = { deals: [] } - - while (!reader.done) { - const { fieldNumber, wireType } = reader.readTag() - if (fieldNumber === 1 && wireType === 2) { - message.deals.push(decodePublicAggreDeal(reader.readBytes())) - } else if (fieldNumber === 2 && wireType === 2) { - message.eventType = reader.readString() - } else { - reader.skip(wireType) - } - } - - return message -} - -function decodePublicAggreDeal(buffer: Uint8Array) { - const reader = new ProtobufReader(buffer) - const message: Record = {} - - while (!reader.done) { - const { fieldNumber, wireType } = reader.readTag() - if (fieldNumber === 1 && wireType === 2) { - message.price = reader.readString() - } else if (fieldNumber === 2 && wireType === 2) { - message.quantity = reader.readString() - } else if (fieldNumber === 3 && wireType === 0) { - message.tradeType = reader.readVarintNumber() - } else if (fieldNumber === 4 && wireType === 0) { - message.time = reader.readInt64String() - } else { - reader.skip(wireType) - } - } + private static getPushDataV3ApiWrapper() { + MexcRealTimeFeed.pushDataV3ApiWrapper ??= protobuf + .parse(MexcRealTimeFeed.pushDataV3ApiWrapperSchema) + .root.lookupType('PushDataV3ApiWrapper') - return message -} - -function decodePublicAggreDepths(buffer: Uint8Array) { - const reader = new ProtobufReader(buffer) - const message: { asks: any[]; bids: any[]; eventType?: string; fromVersion?: string; toVersion?: string } = { asks: [], bids: [] } - - while (!reader.done) { - const { fieldNumber, wireType } = reader.readTag() - if (fieldNumber === 1 && wireType === 2) { - message.asks.push(decodePublicAggreDepthLevel(reader.readBytes())) - } else if (fieldNumber === 2 && wireType === 2) { - message.bids.push(decodePublicAggreDepthLevel(reader.readBytes())) - } else if (fieldNumber === 3 && wireType === 2) { - message.eventType = reader.readString() - } else if (fieldNumber === 4 && wireType === 2) { - message.fromVersion = reader.readString() - } else if (fieldNumber === 5 && wireType === 2) { - message.toVersion = reader.readString() - } else { - reader.skip(wireType) - } - } - - return message -} - -function decodePublicAggreDepthLevel(buffer: Uint8Array) { - const reader = new ProtobufReader(buffer) - const message: Record = {} - - while (!reader.done) { - const { fieldNumber, wireType } = reader.readTag() - if (fieldNumber === 1 && wireType === 2) { - message.price = reader.readString() - } else if (fieldNumber === 2 && wireType === 2) { - message.quantity = reader.readString() - } else { - reader.skip(wireType) - } - } - - return message -} - -function decodePublicAggreBookTicker(buffer: Uint8Array) { - const reader = new ProtobufReader(buffer) - const message: Record = {} - - while (!reader.done) { - const { fieldNumber, wireType } = reader.readTag() - if (fieldNumber === 1 && wireType === 2) { - message.bidPrice = reader.readString() - } else if (fieldNumber === 2 && wireType === 2) { - message.bidQuantity = reader.readString() - } else if (fieldNumber === 3 && wireType === 2) { - message.askPrice = reader.readString() - } else if (fieldNumber === 4 && wireType === 2) { - message.askQuantity = reader.readString() - } else { - reader.skip(wireType) - } - } - - return message -} - -class ProtobufReader { - private offset = 0 - - constructor(private readonly buffer: Uint8Array) {} - - get done() { - return this.offset >= this.buffer.length - } - - readTag() { - const tag = this.readVarintNumber() - return { - fieldNumber: tag >>> 3, - wireType: (tag & 7) as ProtobufWireType - } - } - - readVarintNumber() { - return Number(this.readVarintBigInt()) - } - - readInt64String() { - return this.readVarintBigInt().toString() - } - - readString() { - return Buffer.from(this.readBytes()).toString('utf8') - } - - readBytes() { - const length = this.readVarintNumber() - const end = this.offset + length - if (end > this.buffer.length) { - throw new Error('MEXC protobuf message ended unexpectedly') - } - - const bytes = this.buffer.subarray(this.offset, end) - this.offset = end - return bytes - } - - skip(wireType: ProtobufWireType) { - if (wireType === 0) { - this.readVarintBigInt() - return - } - - if (wireType === 1) { - this.skipFixed(8) - return - } - - if (wireType === 2) { - this.readBytes() - return - } - - if (wireType === 5) { - this.skipFixed(4) - return - } - - throw new Error(`Unsupported MEXC protobuf wire type ${wireType}`) - } - - private readVarintBigInt() { - let shift = 0n - let value = 0n - - while (this.offset < this.buffer.length) { - const byte = this.buffer[this.offset++] - value |= BigInt(byte & 0x7f) << shift - if ((byte & 0x80) === 0) { - return value - } - shift += 7n - } - - throw new Error('MEXC protobuf varint ended unexpectedly') - } - - private skipFixed(bytes: number) { - this.offset += bytes - if (this.offset > this.buffer.length) { - throw new Error('MEXC protobuf message ended unexpectedly') - } + return MexcRealTimeFeed.pushDataV3ApiWrapper } } diff --git a/test/mexcfutures-realtimefeed.test.ts b/test/mexcfutures-realtimefeed.test.ts new file mode 100644 index 0000000..bc74601 --- /dev/null +++ b/test/mexcfutures-realtimefeed.test.ts @@ -0,0 +1,68 @@ +import { Filter } from '../src/types.ts' +import { MexcFuturesRealTimeFeed } from '../src/realtimefeeds/mexcfutures.ts' + +class TestMexcFuturesRealTimeFeed extends MexcFuturesRealTimeFeed { + map(filters: Filter[]) { + return this.mapToSubscribeMessages(filters) + } +} + +test('map mexc futures stored push channels to subscription methods', () => { + const feed = new TestMexcFuturesRealTimeFeed('mexc-futures', [], undefined) + + expect( + feed.map([ + { + channel: 'push.deal', + symbols: ['BTC_USDT', 'ETH_USDT'] + }, + { + channel: 'push.depth', + symbols: ['BTC_USDT'] + }, + { + channel: 'push.funding.rate', + symbols: ['BTC_USDT'] + }, + { + channel: 'push.contract' + } + ]) + ).toEqual([ + { + method: 'sub.deal', + param: { symbol: 'BTC_USDT' }, + gzip: false + }, + { + method: 'sub.deal', + param: { symbol: 'ETH_USDT' }, + gzip: false + }, + { + method: 'sub.depth', + param: { symbol: 'BTC_USDT' }, + gzip: false + }, + { + method: 'sub.funding.rate', + param: { symbol: 'BTC_USDT' }, + gzip: false + }, + { + method: 'sub.contract' + } + ]) +}) + +test('mexc futures realtime subscriptions require symbols', () => { + const feed = new TestMexcFuturesRealTimeFeed('mexc-futures', [], undefined) + + expect(() => + feed.map([ + { + channel: 'push.deal' + } + ]) + ).toThrow('MexcFuturesRealTimeFeed requires explicitly specified symbols when subscribing to live feed') +}) From cb89dd6f29cab7d3b0df738a58218e55d096c94c Mon Sep 17 00:00:00 2001 From: marcin Date: Fri, 12 Jun 2026 09:09:37 +0200 Subject: [PATCH 05/17] feat(mexc-integration): need to remove this file since it was committed by mistake --- test/mexcfutures-realtimefeed.test.ts | 68 --------------------------- 1 file changed, 68 deletions(-) delete mode 100644 test/mexcfutures-realtimefeed.test.ts diff --git a/test/mexcfutures-realtimefeed.test.ts b/test/mexcfutures-realtimefeed.test.ts deleted file mode 100644 index bc74601..0000000 --- a/test/mexcfutures-realtimefeed.test.ts +++ /dev/null @@ -1,68 +0,0 @@ -import { Filter } from '../src/types.ts' -import { MexcFuturesRealTimeFeed } from '../src/realtimefeeds/mexcfutures.ts' - -class TestMexcFuturesRealTimeFeed extends MexcFuturesRealTimeFeed { - map(filters: Filter[]) { - return this.mapToSubscribeMessages(filters) - } -} - -test('map mexc futures stored push channels to subscription methods', () => { - const feed = new TestMexcFuturesRealTimeFeed('mexc-futures', [], undefined) - - expect( - feed.map([ - { - channel: 'push.deal', - symbols: ['BTC_USDT', 'ETH_USDT'] - }, - { - channel: 'push.depth', - symbols: ['BTC_USDT'] - }, - { - channel: 'push.funding.rate', - symbols: ['BTC_USDT'] - }, - { - channel: 'push.contract' - } - ]) - ).toEqual([ - { - method: 'sub.deal', - param: { symbol: 'BTC_USDT' }, - gzip: false - }, - { - method: 'sub.deal', - param: { symbol: 'ETH_USDT' }, - gzip: false - }, - { - method: 'sub.depth', - param: { symbol: 'BTC_USDT' }, - gzip: false - }, - { - method: 'sub.funding.rate', - param: { symbol: 'BTC_USDT' }, - gzip: false - }, - { - method: 'sub.contract' - } - ]) -}) - -test('mexc futures realtime subscriptions require symbols', () => { - const feed = new TestMexcFuturesRealTimeFeed('mexc-futures', [], undefined) - - expect(() => - feed.map([ - { - channel: 'push.deal' - } - ]) - ).toThrow('MexcFuturesRealTimeFeed requires explicitly specified symbols when subscribing to live feed') -}) From 9f57c8732c27f782f1ef7635a1c0e2ccede513a4 Mon Sep 17 00:00:00 2001 From: marcin Date: Fri, 12 Jun 2026 09:53:19 +0200 Subject: [PATCH 06/17] feat(mexc-integration): couple of fixes and improvements --- src/mappers/mexc.ts | 19 +++++++++++++------ src/realtimefeeds/mexc.ts | 15 +++++++++------ src/realtimefeeds/realtimefeed.ts | 2 +- test/mappers.test.ts | 11 +++++++++++ 4 files changed, 34 insertions(+), 13 deletions(-) diff --git a/src/mappers/mexc.ts b/src/mappers/mexc.ts index e470404..ae099b8 100644 --- a/src/mappers/mexc.ts +++ b/src/mappers/mexc.ts @@ -5,8 +5,8 @@ import { Mapper } from './mapper.ts' export class MexcTradesMapper implements Mapper<'mexc', Trade> { private readonly channel = 'spot@public.aggre.deals.v3.api.pb@10ms' - canHandle(message: MexcTradeMessage) { - return message.channel === this.channel || message.channel.startsWith(`${this.channel}@`) + canHandle(message: MexcTradeMessage | MexcControlMessage) { + return message.channel?.startsWith(`${this.channel}@`) === true || message.channel === this.channel } getFilters(symbols?: string[]) { @@ -34,8 +34,8 @@ export class MexcBookChangeMapper implements Mapper<'mexc', BookChange> { private readonly channel = 'spot@public.aggre.depth.v3.api.pb@10ms' private readonly symbolDepthInfo: Record = {} - canHandle(message: MexcDepthSnapshotMessage | MexcDepthUpdateMessage) { - return message.channel === this.channel || message.channel.startsWith(`${this.channel}@`) + canHandle(message: MexcDepthSnapshotMessage | MexcDepthUpdateMessage | MexcControlMessage) { + return message.channel?.startsWith(`${this.channel}@`) === true || message.channel === this.channel } getFilters(symbols?: string[]) { @@ -158,8 +158,8 @@ export class MexcBookChangeMapper implements Mapper<'mexc', BookChange> { export class MexcBookTickerMapper implements Mapper<'mexc', BookTicker> { private readonly channel = 'spot@public.aggre.bookTicker.v3.api.pb@100ms' - canHandle(message: MexcBookTickerMessage) { - return message.channel === this.channel || message.channel.startsWith(`${this.channel}@`) + canHandle(message: MexcBookTickerMessage | MexcControlMessage) { + return message.channel?.startsWith(`${this.channel}@`) === true || message.channel === this.channel } getFilters(symbols?: string[]) { @@ -181,6 +181,13 @@ export class MexcBookTickerMapper implements Mapper<'mexc', BookTicker> { } } +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' diff --git a/src/realtimefeeds/mexc.ts b/src/realtimefeeds/mexc.ts index 6472616..92edae7 100644 --- a/src/realtimefeeds/mexc.ts +++ b/src/realtimefeeds/mexc.ts @@ -3,6 +3,7 @@ import { Filter } from '../types.ts' import { RealTimeFeedBase } from './realtimefeed.ts' export class MexcRealTimeFeed extends RealTimeFeedBase { + 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 @@ -86,13 +87,15 @@ export class MexcRealTimeFeed extends RealTimeFeedBase { ] } - protected parseMessage(message: Buffer): any { - if (message.length > 0 && message[0] === 123) { - return JSON.parse(message.toString()) + protected 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 = MexcRealTimeFeed.getPushDataV3ApiWrapper() - return pushDataV3ApiWrapper.toObject(pushDataV3ApiWrapper.decode(message), { + const pushDataV3ApiWrapper = this.getPushDataV3ApiWrapper() + return pushDataV3ApiWrapper.toObject(pushDataV3ApiWrapper.decode(buffer), { longs: String, arrays: true }) @@ -110,7 +113,7 @@ export class MexcRealTimeFeed extends RealTimeFeedBase { this.send({ method: 'PING' }) } - private static getPushDataV3ApiWrapper() { + private getPushDataV3ApiWrapper() { MexcRealTimeFeed.pushDataV3ApiWrapper ??= protobuf .parse(MexcRealTimeFeed.pushDataV3ApiWrapperSchema) .root.lookupType('PushDataV3ApiWrapper') diff --git a/src/realtimefeeds/realtimefeed.ts b/src/realtimefeeds/realtimefeed.ts index 4b11236..baa213c 100644 --- a/src/realtimefeeds/realtimefeed.ts +++ b/src/realtimefeeds/realtimefeed.ts @@ -233,7 +233,7 @@ export abstract class RealTimeFeedBase implements RealTimeFeedIterable { protected abstract messageIsError(message: any): boolean - protected parseMessage(message: Buffer): any { + protected parseMessage(message: Buffer): any { return JSON.parse(message as any) } diff --git a/test/mappers.test.ts b/test/mappers.test.ts index 2a44b87..dec811e 100644 --- a/test/mappers.test.ts +++ b/test/mappers.test.ts @@ -11018,6 +11018,17 @@ 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( { From dfeb9e2fb00ebfcd6779e694cab076e45d95a67b Mon Sep 17 00:00:00 2001 From: marcin Date: Wed, 17 Jun 2026 13:00:02 +0200 Subject: [PATCH 07/17] feat(mexc-integration): fixed apply snapshot updates --- src/mappers/mexc.ts | 17 +++++++++++++---- 1 file changed, 13 insertions(+), 4 deletions(-) diff --git a/src/mappers/mexc.ts b/src/mappers/mexc.ts index 0957c6c..be580bb 100644 --- a/src/mappers/mexc.ts +++ b/src/mappers/mexc.ts @@ -58,27 +58,36 @@ export class MexcBookChangeMapper implements Mapper<'mexc', BookChange> { return } - const snapshotVersion = Number(message.publicAggreDepths.toVersion) + let currentBookVersion = Number(message.publicAggreDepths.toVersion) const bids = message.publicAggreDepths.bids.map(this.mapBookLevel) const asks = message.publicAggreDepths.asks.map(this.mapBookLevel) for (const update of depthInfo.updates.items()) { const fromVersion = Number(update.publicAggreDepths.fromVersion) const toVersion = Number(update.publicAggreDepths.toVersion) - if (snapshotVersion + 1 < fromVersion || snapshotVersion >= toVersion) { + if (currentBookVersion >= toVersion) { continue } + + if (fromVersion > currentBookVersion + 1 || toVersion < currentBookVersion + 1) { + throw new Error( + `MEXC depth snapshot has no overlap with buffered update, update ${JSON.stringify(update)}, currentBookVersion: ${ + currentBookVersion + }` + ) + } + 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)) } - depthInfo.currentBookVersion = toVersion + currentBookVersion = toVersion } depthInfo.updates.clear() - depthInfo.currentBookVersion = depthInfo.currentBookVersion ?? snapshotVersion + depthInfo.currentBookVersion = currentBookVersion depthInfo.snapshotEmitted = true yield { From 5048426523390ab0899a08371cb53899c5174595 Mon Sep 17 00:00:00 2001 From: marcin Date: Wed, 17 Jun 2026 13:00:23 +0200 Subject: [PATCH 08/17] feat(mexc-integration): fixed apply snapshot updates --- src/mappers/mexc.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/mappers/mexc.ts b/src/mappers/mexc.ts index be580bb..f429f6e 100644 --- a/src/mappers/mexc.ts +++ b/src/mappers/mexc.ts @@ -69,7 +69,7 @@ export class MexcBookChangeMapper implements Mapper<'mexc', BookChange> { continue } - if (fromVersion > currentBookVersion + 1 || toVersion < currentBookVersion + 1) { + if (currentBookVersion + 1 < fromVersion || currentBookVersion + 1 > toVersion) { throw new Error( `MEXC depth snapshot has no overlap with buffered update, update ${JSON.stringify(update)}, currentBookVersion: ${ currentBookVersion From 02466b926afd96a2bc1ecaaad9fca9e4749aba5f Mon Sep 17 00:00:00 2001 From: marcin Date: Wed, 17 Jun 2026 13:00:52 +0200 Subject: [PATCH 09/17] feat(mexc-integration): improved can handle checks, --- src/mappers/mexc.ts | 8 +++++--- 1 file changed, 5 insertions(+), 3 deletions(-) diff --git a/src/mappers/mexc.ts b/src/mappers/mexc.ts index f429f6e..0805d02 100644 --- a/src/mappers/mexc.ts +++ b/src/mappers/mexc.ts @@ -15,7 +15,7 @@ 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 || message.channel === this.channel + return (message.channel?.startsWith(`${this.channel}@`) === true || message.channel === this.channel) && 'publicAggreDeals' in message } getFilters(symbols?: string[]) { @@ -44,7 +44,7 @@ export class MexcBookChangeMapper implements Mapper<'mexc', BookChange> { private readonly symbolDepthInfo: Record = {} canHandle(message: MexcDepthSnapshotMessage | MexcDepthUpdateMessage | MexcControlMessage) { - return message.channel?.startsWith(`${this.channel}@`) === true || message.channel === this.channel + return (message.channel?.startsWith(`${this.channel}@`) === true || message.channel === this.channel) && 'publicAggreDepths' in message } getFilters(symbols?: string[]) { @@ -177,7 +177,9 @@ export class MexcBookTickerMapper implements Mapper<'mexc', BookTicker> { private readonly channel = 'spot@public.aggre.bookTicker.v3.api.pb@100ms' canHandle(message: MexcBookTickerMessage | MexcControlMessage) { - return message.channel?.startsWith(`${this.channel}@`) === true || message.channel === this.channel + return ( + (message.channel?.startsWith(`${this.channel}@`) === true || message.channel === this.channel) && 'publicAggreBookTicker' in message + ) } getFilters(symbols?: string[]) { From 994bcb1f605232669f371561fabae932d691c5fb Mon Sep 17 00:00:00 2001 From: marcin Date: Wed, 17 Jun 2026 16:15:52 +0200 Subject: [PATCH 10/17] feat(mexc-integration): streaming dat needs to have proper snapshots --- src/mappers/mexc.ts | 2 +- src/realtimefeeds/mexc.ts | 188 ++++++++++++++++++++++++++++++++- test/mappers.test.ts | 60 ++++++++++- test/mexc-realtimefeed.test.ts | 151 ++++++++++++++++++++++++++ 4 files changed, 392 insertions(+), 9 deletions(-) diff --git a/src/mappers/mexc.ts b/src/mappers/mexc.ts index 0805d02..c468d94 100644 --- a/src/mappers/mexc.ts +++ b/src/mappers/mexc.ts @@ -241,7 +241,7 @@ enum MexcTradeType { Sell = 2 } -type MexcDepthSnapshotMessage = MexcProtobufMessage<'spot@public.aggre.depth.v3.api.pb@10ms'> & { +export type MexcDepthSnapshotMessage = MexcProtobufMessage<'spot@public.aggre.depth.v3.api.pb@10ms'> & { generated: true publicAggreDepths: MexcAggreDepths } diff --git a/src/realtimefeeds/mexc.ts b/src/realtimefeeds/mexc.ts index 92edae7..23c14ec 100644 --- a/src/realtimefeeds/mexc.ts +++ b/src/realtimefeeds/mexc.ts @@ -1,4 +1,6 @@ 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' @@ -59,14 +61,17 @@ export class MexcRealTimeFeed extends RealTimeFeedBase { } ` + 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', 'spot@public.aggre.depth.v3.api.pb@10ms', 'spot@public.aggre.bookTicker.v3.api.pb@100ms' ]) - protected mapToSubscribeMessages(filters: Filter[]): any[] { + 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}`) @@ -79,6 +84,8 @@ export class MexcRealTimeFeed extends RealTimeFeedBase { return filter as Required> }) + this.resetDepthSnapshotTracking(filtersWithSymbols) + return [ { method: 'SUBSCRIPTION', @@ -87,7 +94,7 @@ export class MexcRealTimeFeed extends RealTimeFeedBase { ] } - protected parseMessage(message: Buffer): any { + protected override parseMessage(message: Buffer): any { const buffer = Buffer.isBuffer(message) ? message : Buffer.from(message) if (buffer.length > 0 && buffer[0] === MexcRealTimeFeed.jsonObjectStart) { @@ -101,14 +108,49 @@ export class MexcRealTimeFeed extends RealTimeFeedBase { }) } - protected messageIsError(message: any): boolean { + protected override messageIsError(message: any): boolean { return message.code !== undefined && message.code !== 0 } - protected messageIsHeartbeat(message: any) { + protected override messageIsHeartbeat(message: any) { return message.msg === 'PONG' } + protected override onMessage(message: any) { + if ( + message.channel?.startsWith('spot@public.aggre.depth.v3.api.pb@10ms') !== 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 === 'spot@public.aggre.depth.v3.api.pb@10ms') + 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' }) } @@ -120,4 +162,142 @@ export class MexcRealTimeFeed extends RealTimeFeedBase { return MexcRealTimeFeed.pushDataV3ApiWrapper } + + private resetDepthSnapshotTracking(filters: Required>[]) { + this.pendingDepthSnapshotSymbols.clear() + this.bufferedDepthUpdates.clear() + + for (const filter of filters) { + if (filter.channel !== 'spot@public.aggre.depth.v3.api.pb@10ms') { + continue + } + + for (const symbol of filter.symbols) { + const upperCaseSymbol = symbol.toUpperCase() + this.pendingDepthSnapshotSymbols.add(upperCaseSymbol) + this.bufferedDepthUpdates.set(upperCaseSymbol, new CircularBuffer(2000)) + } + } + } + + 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 { + const version = data.lastUpdateId.toString() + + return { + channel: 'spot@public.aggre.depth.v3.api.pb@10ms', + symbol, + sendTime: Date.now().toString(), + generated: true, + publicAggreDepths: { + asks: data.asks.map(this.mapDepthSnapshotLevel), + bids: data.bids.map(this.mapDepthSnapshotLevel), + eventType: 'spot@public.aggre.depth.v3.api.pb@10ms', + fromVersion: version, + toVersion: version + } + } + } + + private mapDepthSnapshotLevel([price, quantity]: [string, string]) { + return { price, quantity } + } +} + +type MexcDepthSnapshotResponse = { + lastUpdateId: number + bids: [string, string][] + asks: [string, string][] +} + +type MexcDepthUpdateData = { + firstVersion: number + lastVersion: number } diff --git a/test/mappers.test.ts b/test/mappers.test.ts index dec811e..3b2c8cd 100644 --- a/test/mappers.test.ts +++ b/test/mappers.test.ts @@ -11029,6 +11029,28 @@ test('map mexc messages', () => { ) ).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( { @@ -11098,6 +11120,24 @@ test('map mexc messages', () => { ) ).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( { @@ -11123,14 +11163,26 @@ test('map mexc messages', () => { isSnapshot: true, bids: [ { price: 99.8, amount: 2.3 }, - { price: 99.9, amount: 0.5 } + { price: 99.9, amount: 0.5 }, + { price: 99.7, amount: 1.1 } ], - asks: [{ price: 100.1, amount: 1.2 }], + asks: [], timestamp: new Date('2024-03-09T16:00:00.002Z'), localTimestamp } ]) + expect( + mapper.map( + { + channel: 'spot@public.aggre.bookTicker.v3.api.pb@100ms@XENUSDT', + symbol: 'XENUSDT', + sendTime: '1781337091703' + }, + localTimestamp + ) + ).toEqual([]) + expect( mapper.map( { @@ -11141,8 +11193,8 @@ test('map mexc messages', () => { 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' + fromVersion: '103', + toVersion: '103' } }, localTimestamp diff --git a/test/mexc-realtimefeed.test.ts b/test/mexc-realtimefeed.test.ts index 97ecab9..00262c2 100644 --- a/test/mexc-realtimefeed.test.ts +++ b/test/mexc-realtimefeed.test.ts @@ -1,8 +1,17 @@ +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) } @@ -18,6 +27,15 @@ class TestMexcRealTimeFeed extends MexcRealTimeFeed { 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', () => { @@ -172,6 +190,108 @@ test('decode mexc realtime protobuf book ticker message', () => { }) }) +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'] + } + ] + + 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.v3.api.pb@10ms', + symbol: 'BTCUSDT', + sendTime: '1710000000000', + generated: true, + 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: '100', + toVersion: '100' + } + } + ]) + } 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']] }, + { lastUpdateId: 102, asks: [['100.2', '1.2']], bids: [['99.8', '0.5']] }, + { lastUpdateId: 103, asks: [['100.3', '1.2']], bids: [['99.7', '0.5']] }, + { lastUpdateId: 104, asks: [['100.4', '1.2']], bids: [['99.6', '0.5']] } + ]) + 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'] + } + ] + + 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.v3.api.pb@10ms', + symbol: 'BTCUSDT', + sendTime: '1710000000000', + generated: true, + publicAggreDepths: { + asks: [{ price: '100.4', quantity: '1.2' }], + bids: [{ price: '99.6', quantity: '0.5' }], + eventType: 'spot@public.aggre.depth.v3.api.pb@10ms', + fromVersion: '104', + toVersion: '104' + } + } + ]) + } finally { + Date.now = originalDateNow + await server.close() + } +}) + function message(...fields: Buffer[]) { return Buffer.concat(fields) } @@ -202,3 +322,34 @@ function varint(value: number) { 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']] }] +) { + 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[][] +} From cfe5be42cd5c23c3cac156b1ed23e7a4c1fbb1b3 Mon Sep 17 00:00:00 2001 From: marcin Date: Wed, 17 Jun 2026 17:15:02 +0200 Subject: [PATCH 11/17] feat(mexc-integration): aligned snapshot channel with shape with regular exchange messages --- src/consts.ts | 2 +- src/mappers/mexc.ts | 16 ++++++------- src/realtimefeeds/mexc.ts | 4 ++-- test/mappers.test.ts | 6 ++--- test/mexc.live.test.ts | 47 +++++++++++++++++++++++++++++++++++++++ 5 files changed, 60 insertions(+), 15 deletions(-) create mode 100644 test/mexc.live.test.ts diff --git a/src/consts.ts b/src/consts.ts index 5073f80..0d95ae7 100644 --- a/src/consts.ts +++ b/src/consts.ts @@ -530,7 +530,7 @@ const BULLISH_CHANNELS = ['V1TALevel2', 'V1TALevel1', 'V1TAAnonymousTradeUpdate' const MEXC_CHANNELS = [ 'spot@public.aggre.deals.v3.api.pb@10ms', 'spot@public.aggre.depth.v3.api.pb@10ms', - 'spot@public.aggre.bookTicker.v3.api.pb@100ms' + 'spot@public.aggre.bookTicker.v3.api.pb@10ms' ] as const const POLYMARKET_CHANNELS = [ diff --git a/src/mappers/mexc.ts b/src/mappers/mexc.ts index c468d94..394941d 100644 --- a/src/mappers/mexc.ts +++ b/src/mappers/mexc.ts @@ -15,7 +15,7 @@ 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 || message.channel === this.channel) && 'publicAggreDeals' in message + return message.channel?.startsWith(`${this.channel}@`) === true && 'publicAggreDeals' in message } getFilters(symbols?: string[]) { @@ -44,7 +44,7 @@ export class MexcBookChangeMapper implements Mapper<'mexc', BookChange> { private readonly symbolDepthInfo: Record = {} canHandle(message: MexcDepthSnapshotMessage | MexcDepthUpdateMessage | MexcControlMessage) { - return (message.channel?.startsWith(`${this.channel}@`) === true || message.channel === this.channel) && 'publicAggreDepths' in message + return message.channel?.startsWith(`${this.channel}@`) === true && 'publicAggreDepths' in message } getFilters(symbols?: string[]) { @@ -174,12 +174,10 @@ export class MexcBookChangeMapper implements Mapper<'mexc', BookChange> { } export class MexcBookTickerMapper implements Mapper<'mexc', BookTicker> { - private readonly channel = 'spot@public.aggre.bookTicker.v3.api.pb@100ms' + private readonly channel = 'spot@public.aggre.bookTicker.v3.api.pb@10ms' canHandle(message: MexcBookTickerMessage | MexcControlMessage) { - return ( - (message.channel?.startsWith(`${this.channel}@`) === true || message.channel === this.channel) && 'publicAggreBookTicker' in message - ) + return message.channel?.startsWith(`${this.channel}@`) === true && 'publicAggreBookTicker' in message } getFilters(symbols?: string[]) { @@ -211,8 +209,8 @@ type MexcControlMessage = { type MexcMappedChannel = | 'spot@public.aggre.deals.v3.api.pb@10ms' | 'spot@public.aggre.depth.v3.api.pb@10ms' - | 'spot@public.aggre.bookTicker.v3.api.pb@100ms' -type MexcChannelWithSymbol = TChannel | `${TChannel}@${string}` + | 'spot@public.aggre.bookTicker.v3.api.pb@10ms' +type MexcChannelWithSymbol = `${TChannel}@${string}` type MexcProtobufMessage = { channel: MexcChannelWithSymbol @@ -264,7 +262,7 @@ type MexcPriceLevel = { quantity: string } -type MexcBookTickerMessage = MexcProtobufMessage<'spot@public.aggre.bookTicker.v3.api.pb@100ms'> & { +type MexcBookTickerMessage = MexcProtobufMessage<'spot@public.aggre.bookTicker.v3.api.pb@10ms'> & { publicAggreBookTicker: { bidPrice: string bidQuantity: string diff --git a/src/realtimefeeds/mexc.ts b/src/realtimefeeds/mexc.ts index 23c14ec..ccf124c 100644 --- a/src/realtimefeeds/mexc.ts +++ b/src/realtimefeeds/mexc.ts @@ -68,7 +68,7 @@ export class MexcRealTimeFeed extends RealTimeFeedBase { private readonly channels = new Set([ 'spot@public.aggre.deals.v3.api.pb@10ms', 'spot@public.aggre.depth.v3.api.pb@10ms', - 'spot@public.aggre.bookTicker.v3.api.pb@100ms' + 'spot@public.aggre.bookTicker.v3.api.pb@10ms' ]) protected override mapToSubscribeMessages(filters: Filter[]): any[] { @@ -272,7 +272,7 @@ export class MexcRealTimeFeed extends RealTimeFeedBase { const version = data.lastUpdateId.toString() return { - channel: 'spot@public.aggre.depth.v3.api.pb@10ms', + channel: `spot@public.aggre.depth.v3.api.pb@10ms@${symbol}`, symbol, sendTime: Date.now().toString(), generated: true, diff --git a/test/mappers.test.ts b/test/mappers.test.ts index 3b2c8cd..d9c2d5e 100644 --- a/test/mappers.test.ts +++ b/test/mappers.test.ts @@ -11141,7 +11141,7 @@ test('map mexc messages', () => { expect( mapper.map( { - channel: 'spot@public.aggre.depth.v3.api.pb@10ms', + channel: 'spot@public.aggre.depth.v3.api.pb@10ms@BTCUSDT', symbol: 'BTCUSDT', sendTime: '1710000000002', generated: true, @@ -11175,7 +11175,7 @@ test('map mexc messages', () => { expect( mapper.map( { - channel: 'spot@public.aggre.bookTicker.v3.api.pb@100ms@XENUSDT', + channel: 'spot@public.aggre.bookTicker.v3.api.pb@10ms@XENUSDT', symbol: 'XENUSDT', sendTime: '1781337091703' }, @@ -11215,7 +11215,7 @@ test('map mexc messages', () => { expect( mapper.map( { - channel: 'spot@public.aggre.bookTicker.v3.api.pb@100ms@BTCUSDT', + channel: 'spot@public.aggre.bookTicker.v3.api.pb@10ms@BTCUSDT', symbol: 'BTCUSDT', sendTime: '1710000000004', publicAggreBookTicker: { diff --git a/test/mexc.live.test.ts b/test/mexc.live.test.ts new file mode 100644 index 0000000..56b0fb5 --- /dev/null +++ b/test/mexc.live.test.ts @@ -0,0 +1,47 @@ +import { normalizeBookChanges, streamNormalized } from '../dist/index.js' +import { describeLive } from './live.js' + +describeLive('mexc live', () => { + test('streams normalized BTCUSDT book changes', async () => { + const messages = streamNormalized( + { + exchange: 'mexc', + symbols: ['BTCUSDT'], + timeoutIntervalMS: 20_000, + withDisconnectMessages: true + }, + normalizeBookChanges + ) + + let sawBookSnapshot = 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 === 'book_change') { + if (message.isSnapshot) { + sawBookSnapshot = message.asks.length > 0 || message.bids.length > 0 + } + } + + if (sawBookSnapshot) { + break + } + } + } finally { + await messages.return?.() + } + + expect(sawDisconnect).toBe(false) + expect(sawBookSnapshot).toBe(true) + }, 40_000) +}) From 7f2f9856f7e97eca98a1532af13b2fc2045e10ba Mon Sep 17 00:00:00 2001 From: marcin Date: Thu, 18 Jun 2026 08:52:59 +0200 Subject: [PATCH 12/17] feat(mexc-integration): adjusted tests --- .../mexc-realtimefeed.test.ts.snap | 15 +++++++ test/mexc-realtimefeed.test.ts | 44 +++++++------------ 2 files changed, 32 insertions(+), 27 deletions(-) create mode 100644 test/__snapshots__/mexc-realtimefeed.test.ts.snap 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/mexc-realtimefeed.test.ts b/test/mexc-realtimefeed.test.ts index 00262c2..eb47982 100644 --- a/test/mexc-realtimefeed.test.ts +++ b/test/mexc-realtimefeed.test.ts @@ -45,32 +45,22 @@ test('register mexc realtime feed', () => { test('map mexc realtime subscriptions', () => { const feed = new TestMexcRealTimeFeed('mexc', [], undefined) - expect( - 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.bookTicker.v3.api.pb@100ms', - symbols: ['BTCUSDT'] - } - ]) - ).toEqual([ + 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'] + }, { - 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@100ms@BTCUSDT' - ] + channel: 'spot@public.aggre.bookTicker.v3.api.pb@10ms', + symbols: ['BTCUSDT'] } ]) + + expect(subscribeMessages).toMatchSnapshot() }) test('mexc realtime subscriptions require symbols', () => { @@ -171,14 +161,14 @@ 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@100ms@BTCUSDT'), + 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@100ms@BTCUSDT', + channel: 'spot@public.aggre.bookTicker.v3.api.pb@10ms@BTCUSDT', symbol: 'BTCUSDT', sendTime: '1710000000000', publicAggreBookTicker: { @@ -219,7 +209,7 @@ test('provide mexc manual depth snapshots', async () => { expect(snapshots).toEqual([ { - channel: 'spot@public.aggre.depth.v3.api.pb@10ms', + channel: 'spot@public.aggre.depth.v3.api.pb@10ms@BTCUSDT', symbol: 'BTCUSDT', sendTime: '1710000000000', generated: true, @@ -273,7 +263,7 @@ test('retry mexc manual depth snapshots until buffered update overlaps', async ( expect(server.requestsCount).toBe(4) expect(snapshots).toEqual([ { - channel: 'spot@public.aggre.depth.v3.api.pb@10ms', + channel: 'spot@public.aggre.depth.v3.api.pb@10ms@BTCUSDT', symbol: 'BTCUSDT', sendTime: '1710000000000', generated: true, From 12efdf3bc05ec54e29cdc8c8bcf5e5cc8be998c1 Mon Sep 17 00:00:00 2001 From: marcin Date: Tue, 23 Jun 2026 16:44:18 +0200 Subject: [PATCH 13/17] feat(mexc-integration): fixed mapping --- src/mappers/mexc.ts | 60 ++++--- test/mappers.test.ts | 393 +++++++++++++++++++++++++++++++++++++++++ test/mexc.live.test.ts | 42 +++-- 3 files changed, 462 insertions(+), 33 deletions(-) diff --git a/src/mappers/mexc.ts b/src/mappers/mexc.ts index 394941d..9176f9d 100644 --- a/src/mappers/mexc.ts +++ b/src/mappers/mexc.ts @@ -1,12 +1,13 @@ +import { debug } from '../debug.ts' import { CircularBuffer, upperCaseSymbols } from '../handy.ts' import { BookChange, BookPriceLevel, BookTicker, Trade } from '../types.ts' import { Mapper } from './mapper.ts' -import { exchangeMappers } from './registry.ts' +import { exchangeMappers, isRealTime } from './registry.ts' export const mexcMappers = exchangeMappers({ mexc: { trades: () => new MexcTradesMapper(), - bookChanges: () => new MexcBookChangeMapper(), + bookChanges: (localTimestamp) => new MexcBookChangeMapper({ ignoreBookSnapshotOverlapError: isRealTime(localTimestamp) === false }), bookTickers: () => new MexcBookTickerMapper() } }) @@ -42,6 +43,11 @@ export class MexcTradesMapper implements Mapper<'mexc', Trade> { export class MexcBookChangeMapper implements Mapper<'mexc', BookChange> { private readonly channel = 'spot@public.aggre.depth.v3.api.pb@10ms' private readonly symbolDepthInfo: Record = {} + private readonly ignoreBookSnapshotOverlapError: boolean + + constructor({ ignoreBookSnapshotOverlapError }: { ignoreBookSnapshotOverlapError: boolean }) { + this.ignoreBookSnapshotOverlapError = ignoreBookSnapshotOverlapError + } canHandle(message: MexcDepthSnapshotMessage | MexcDepthUpdateMessage | MexcControlMessage) { return message.channel?.startsWith(`${this.channel}@`) === true && 'publicAggreDepths' in message @@ -59,8 +65,8 @@ export class MexcBookChangeMapper implements Mapper<'mexc', BookChange> { } let currentBookVersion = Number(message.publicAggreDepths.toVersion) - const bids = message.publicAggreDepths.bids.map(this.mapBookLevel) - const asks = message.publicAggreDepths.asks.map(this.mapBookLevel) + const bids = (message.publicAggreDepths.bids ?? []).map(this.mapBookLevel) + const asks = (message.publicAggreDepths.asks ?? []).map(this.mapBookLevel) for (const update of depthInfo.updates.items()) { const fromVersion = Number(update.publicAggreDepths.fromVersion) @@ -69,18 +75,24 @@ export class MexcBookChangeMapper implements Mapper<'mexc', BookChange> { continue } - if (currentBookVersion + 1 < fromVersion || currentBookVersion + 1 > toVersion) { - throw new Error( - `MEXC depth snapshot has no overlap with buffered update, update ${JSON.stringify(update)}, currentBookVersion: ${ - currentBookVersion - }` - ) + 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) { + for (const bid of update.publicAggreDepths.bids ?? []) { this.applyLevel(bids, this.mapBookLevel(bid)) } - for (const ask of update.publicAggreDepths.asks) { + for (const ask of update.publicAggreDepths.asks ?? []) { this.applyLevel(asks, this.mapBookLevel(ask)) } currentBookVersion = toVersion @@ -117,14 +129,16 @@ export class MexcBookChangeMapper implements Mapper<'mexc', BookChange> { if (!depthInfo.isContinuityValidated) { if (fromVersion > depthInfo.currentBookVersion! + 1 || toVersion < depthInfo.currentBookVersion! + 1) { - throw new Error( - `MEXC depth snapshot has no overlap with first update, update ${JSON.stringify(message)}, currentBookVersion: ${ - depthInfo.currentBookVersion - }` - ) + 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.isContinuityValidated = true } depthInfo.currentBookVersion = toVersion @@ -134,8 +148,8 @@ export class MexcBookChangeMapper implements Mapper<'mexc', BookChange> { symbol: message.symbol, exchange: 'mexc', isSnapshot: false, - bids: message.publicAggreDepths.bids.map(this.mapBookLevel), - asks: message.publicAggreDepths.asks.map(this.mapBookLevel), + bids: (message.publicAggreDepths.bids ?? []).map(this.mapBookLevel), + asks: (message.publicAggreDepths.asks ?? []).map(this.mapBookLevel), timestamp: new Date(Number(message.sendTime)), localTimestamp } @@ -251,8 +265,8 @@ type MexcDepthUpdateMessage = MexcProtobufMessage<'spot@public.aggre.depth.v3.ap type MexcAggreDepths = { eventType: 'spot@public.aggre.depth.v3.api.pb@10ms' - asks: MexcPriceLevel[] - bids: MexcPriceLevel[] + asks?: MexcPriceLevel[] + bids?: MexcPriceLevel[] fromVersion: string toVersion: string } diff --git a/test/mappers.test.ts b/test/mappers.test.ts index d9c2d5e..6b62355 100644 --- a/test/mappers.test.ts +++ b/test/mappers.test.ts @@ -11212,6 +11212,62 @@ test('map mexc messages', () => { } ]) + 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( { @@ -11242,6 +11298,343 @@ test('map mexc messages', () => { ]) }) +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.v3.api.pb@10ms@BTCUSDT', + symbol: 'BTCUSDT', + sendTime: '1781740800000', + generated: true, + publicAggreDepths: { + asks: [{ price: '64530.00', quantity: '1' }], + bids: [{ price: '64520.00', quantity: '1' }], + eventType: 'spot@public.aggre.depth.v3.api.pb@10ms', + fromVersion: '74483042020', + toVersion: '74483042020' + } + }, + 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.v3.api.pb@10ms@BTCUSDT', + symbol: 'BTCUSDT', + sendTime: '1781740800003', + generated: true, + publicAggreDepths: { + asks: [], + bids: [], + eventType: 'spot@public.aggre.depth.v3.api.pb@10ms', + fromVersion: '100', + toVersion: '100' + } + }, + 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.v3.api.pb@10ms@BTCUSDT', + symbol: 'BTCUSDT', + sendTime: Date.now().toString(), + generated: true, + publicAggreDepths: { + asks: [], + bids: [], + eventType: 'spot@public.aggre.depth.v3.api.pb@10ms', + fromVersion: '100', + toVersion: '100' + } + }, + 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.v3.api.pb@10ms@BTCUSDT', + symbol: 'BTCUSDT', + sendTime: '1782208005000', + generated: true, + publicAggreDepths: { + asks: [], + bids: [], + eventType: 'spot@public.aggre.depth.v3.api.pb@10ms', + fromVersion: '74891313262', + toVersion: '74891313262' + } + }, + 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.live.test.ts b/test/mexc.live.test.ts index 56b0fb5..2464ada 100644 --- a/test/mexc.live.test.ts +++ b/test/mexc.live.test.ts @@ -1,8 +1,8 @@ -import { normalizeBookChanges, streamNormalized } from '../dist/index.js' +import { normalizeBookChanges, normalizeBookTickers, normalizeTrades, streamNormalized } from '../dist/index.js' import { describeLive } from './live.js' describeLive('mexc live', () => { - test('streams normalized BTCUSDT book changes', async () => { + test('streams normalized BTCUSDT data for all mappers', async () => { const messages = streamNormalized( { exchange: 'mexc', @@ -10,10 +10,17 @@ describeLive('mexc live', () => { timeoutIntervalMS: 20_000, withDisconnectMessages: true }, - normalizeBookChanges + normalizeTrades, + normalizeBookChanges, + normalizeBookTickers ) - let sawBookSnapshot = false + const seen = { + trade: false, + bookSnapshot: false, + bookDelta: false, + bookTicker: false + } let sawDisconnect = false try { @@ -27,13 +34,23 @@ describeLive('mexc live', () => { continue } - if (message.type === 'book_change') { - if (message.isSnapshot) { - sawBookSnapshot = message.asks.length > 0 || message.bids.length > 0 - } + 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 (sawBookSnapshot) { + if (Object.values(seen).every(Boolean)) { break } } @@ -42,6 +59,11 @@ describeLive('mexc live', () => { } expect(sawDisconnect).toBe(false) - expect(sawBookSnapshot).toBe(true) + expect(seen).toEqual({ + trade: true, + bookSnapshot: true, + bookDelta: true, + bookTicker: true + }) }, 40_000) }) From c66debd0fb27075a694de1bd4889139768b55996 Mon Sep 17 00:00:00 2001 From: marcin Date: Tue, 23 Jun 2026 20:10:21 +0200 Subject: [PATCH 14/17] feat(mexc-futures-integration): updated proto schema --- src/mappers/mexc.ts | 6 +++++- src/realtimefeeds/mexc.ts | 4 ++++ test/mappers.test.ts | 10 ++++++---- 3 files changed, 15 insertions(+), 5 deletions(-) diff --git a/src/mappers/mexc.ts b/src/mappers/mexc.ts index 9176f9d..5e7b786 100644 --- a/src/mappers/mexc.ts +++ b/src/mappers/mexc.ts @@ -29,7 +29,7 @@ export class MexcTradesMapper implements Mapper<'mexc', Trade> { type: 'trade', symbol: message.symbol, exchange: 'mexc', - id: undefined, + id: trade.tradeId, price: Number(trade.price), amount: Number(trade.quantity), side: trade.tradeType === MexcTradeType.Buy ? 'buy' : 'sell', @@ -246,6 +246,7 @@ type MexcTrade = { quantity: string tradeType: MexcTradeType time: string | number + tradeId?: string } enum MexcTradeType { @@ -269,6 +270,7 @@ type MexcAggreDepths = { bids?: MexcPriceLevel[] fromVersion: string toVersion: string + lastOrderCreateTime?: string | number } type MexcPriceLevel = { @@ -282,6 +284,8 @@ type MexcBookTickerMessage = MexcProtobufMessage<'spot@public.aggre.bookTicker.v bidQuantity: string askPrice: string askQuantity: string + version?: string + lastOrderCreateTime?: string | number } } diff --git a/src/realtimefeeds/mexc.ts b/src/realtimefeeds/mexc.ts index ccf124c..f7aafdc 100644 --- a/src/realtimefeeds/mexc.ts +++ b/src/realtimefeeds/mexc.ts @@ -38,6 +38,7 @@ export class MexcRealTimeFeed extends RealTimeFeedBase { string quantity = 2; int32 tradeType = 3; int64 time = 4; + string tradeId = 5; } message PublicAggreDepthsV3Api { @@ -46,6 +47,7 @@ export class MexcRealTimeFeed extends RealTimeFeedBase { string eventType = 3; string fromVersion = 4; string toVersion = 5; + int64 lastOrderCreateTime = 6; } message PublicAggreDepthV3ApiItem { @@ -58,6 +60,8 @@ export class MexcRealTimeFeed extends RealTimeFeedBase { string bidQuantity = 2; string askPrice = 3; string askQuantity = 4; + string version = 5; + int64 lastOrderCreateTime = 6; } ` diff --git a/test/mappers.test.ts b/test/mappers.test.ts index 6b62355..4713568 100644 --- a/test/mappers.test.ts +++ b/test/mappers.test.ts @@ -11063,13 +11063,15 @@ test('map mexc messages', () => { price: '100.1', quantity: '0.2', tradeType: 1, - time: '1710000000001' + time: '1710000000001', + tradeId: '698165549569396736X0_698165549569396737X0' }, { price: '100.2', quantity: '0.3', tradeType: 2, - time: '1710000000002' + time: '1710000000002', + tradeId: '698165549569396738X0_698165549569396739X0' } ], eventType: 'spot@public.aggre.deals.v3.api.pb@10ms' @@ -11082,7 +11084,7 @@ test('map mexc messages', () => { type: 'trade', symbol: 'BTCUSDT', exchange: 'mexc', - id: undefined, + id: '698165549569396736X0_698165549569396737X0', price: 100.1, amount: 0.2, side: 'buy', @@ -11093,7 +11095,7 @@ test('map mexc messages', () => { type: 'trade', symbol: 'BTCUSDT', exchange: 'mexc', - id: undefined, + id: '698165549569396738X0_698165549569396739X0', price: 100.2, amount: 0.3, side: 'sell', From e31c5806132900ecfd12df4dbab3512767c75d05 Mon Sep 17 00:00:00 2001 From: marcin Date: Tue, 23 Jun 2026 20:18:12 +0200 Subject: [PATCH 15/17] feat(mexc-futures-integration): updated mapping --- src/mappers/mexc.ts | 18 +++++++++--------- test/mappers.test.ts | 29 +++++++++++++++++++++++++++++ 2 files changed, 38 insertions(+), 9 deletions(-) diff --git a/src/mappers/mexc.ts b/src/mappers/mexc.ts index 5e7b786..a4c9146 100644 --- a/src/mappers/mexc.ts +++ b/src/mappers/mexc.ts @@ -1,5 +1,5 @@ import { debug } from '../debug.ts' -import { CircularBuffer, upperCaseSymbols } from '../handy.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' @@ -203,10 +203,10 @@ export class MexcBookTickerMapper implements Mapper<'mexc', BookTicker> { type: 'book_ticker', symbol: message.symbol, exchange: 'mexc', - askAmount: Number(message.publicAggreBookTicker.askQuantity), - askPrice: Number(message.publicAggreBookTicker.askPrice), - bidAmount: Number(message.publicAggreBookTicker.bidQuantity), - bidPrice: Number(message.publicAggreBookTicker.bidPrice), + 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 } @@ -280,10 +280,10 @@ type MexcPriceLevel = { type MexcBookTickerMessage = MexcProtobufMessage<'spot@public.aggre.bookTicker.v3.api.pb@10ms'> & { publicAggreBookTicker: { - bidPrice: string - bidQuantity: string - askPrice: string - askQuantity: string + bidPrice?: string + bidQuantity?: string + askPrice?: string + askQuantity?: string version?: string lastOrderCreateTime?: string | number } diff --git a/test/mappers.test.ts b/test/mappers.test.ts index 4713568..37ff2af 100644 --- a/test/mappers.test.ts +++ b/test/mappers.test.ts @@ -11298,6 +11298,35 @@ test('map mexc messages', () => { 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', () => { From 49be4da348017f50724ef3b785885e2255104d7e Mon Sep 17 00:00:00 2001 From: marcin Date: Wed, 24 Jun 2026 00:12:26 +0200 Subject: [PATCH 16/17] feat(mexc-integration): adjusted to new snapshot shape --- src/mappers/mexc.ts | 37 ++++++++++++++++++++------ src/realtimefeeds/mexc.ts | 19 +++++--------- test/mappers.test.ts | 48 ++++++++++++++-------------------- test/mexc-realtimefeed.test.ts | 37 +++++++++++++------------- 4 files changed, 72 insertions(+), 69 deletions(-) diff --git a/src/mappers/mexc.ts b/src/mappers/mexc.ts index a4c9146..c6ca8a7 100644 --- a/src/mappers/mexc.ts +++ b/src/mappers/mexc.ts @@ -49,8 +49,11 @@ export class MexcBookChangeMapper implements Mapper<'mexc', BookChange> { this.ignoreBookSnapshotOverlapError = ignoreBookSnapshotOverlapError } - canHandle(message: MexcDepthSnapshotMessage | MexcDepthUpdateMessage | MexcControlMessage) { - return message.channel?.startsWith(`${this.channel}@`) === true && 'publicAggreDepths' in message + canHandle(message: MexcDepthSnapshotMessage | MexcDepthUpdateMessage) { + return ( + message.channel?.startsWith(`${this.channel}@`) === true && + (message.generated === true ? 'publicAggreDepthsSnapshot' in message : 'publicAggreDepths' in message) + ) } getFilters(symbols?: string[]) { @@ -64,9 +67,9 @@ export class MexcBookChangeMapper implements Mapper<'mexc', BookChange> { return } - let currentBookVersion = Number(message.publicAggreDepths.toVersion) - const bids = (message.publicAggreDepths.bids ?? []).map(this.mapBookLevel) - const asks = (message.publicAggreDepths.asks ?? []).map(this.mapBookLevel) + 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) @@ -109,7 +112,7 @@ export class MexcBookChangeMapper implements Mapper<'mexc', BookChange> { isSnapshot: true, bids, asks, - timestamp: new Date(Number(message.sendTime)), + timestamp: new Date(Number(message.publicAggreDepthsSnapshot.timestamp)), localTimestamp } @@ -178,6 +181,13 @@ export class MexcBookChangeMapper implements Mapper<'mexc', BookChange> { } } + 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) } @@ -254,11 +264,22 @@ enum MexcTradeType { Sell = 2 } -export type MexcDepthSnapshotMessage = MexcProtobufMessage<'spot@public.aggre.depth.v3.api.pb@10ms'> & { +export type MexcDepthSnapshotMessage = { + channel: MexcChannelWithSymbol<'spot@public.aggre.depth.v3.api.pb@10ms'> + symbol: string generated: true - publicAggreDepths: MexcAggreDepths + 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 diff --git a/src/realtimefeeds/mexc.ts b/src/realtimefeeds/mexc.ts index f7aafdc..767c8f2 100644 --- a/src/realtimefeeds/mexc.ts +++ b/src/realtimefeeds/mexc.ts @@ -273,32 +273,25 @@ export class MexcRealTimeFeed extends RealTimeFeedBase { } private createManualSnapshot(symbol: string, data: MexcDepthSnapshotResponse): MexcDepthSnapshotMessage { - const version = data.lastUpdateId.toString() - return { channel: `spot@public.aggre.depth.v3.api.pb@10ms@${symbol}`, symbol, - sendTime: Date.now().toString(), generated: true, - publicAggreDepths: { - asks: data.asks.map(this.mapDepthSnapshotLevel), - bids: data.bids.map(this.mapDepthSnapshotLevel), - eventType: 'spot@public.aggre.depth.v3.api.pb@10ms', - fromVersion: version, - toVersion: version + publicAggreDepthsSnapshot: { + lastUpdateId: data.lastUpdateId, + asks: data.asks, + bids: data.bids, + timestamp: data.timestamp } } } - - private mapDepthSnapshotLevel([price, quantity]: [string, string]) { - return { price, quantity } - } } type MexcDepthSnapshotResponse = { lastUpdateId: number bids: [string, string][] asks: [string, string][] + timestamp: number } type MexcDepthUpdateData = { diff --git a/test/mappers.test.ts b/test/mappers.test.ts index 37ff2af..375f1a3 100644 --- a/test/mappers.test.ts +++ b/test/mappers.test.ts @@ -11145,14 +11145,12 @@ test('map mexc messages', () => { { channel: 'spot@public.aggre.depth.v3.api.pb@10ms@BTCUSDT', symbol: 'BTCUSDT', - sendTime: '1710000000002', generated: true, - publicAggreDepths: { - asks: [{ price: '100.1', quantity: '1.2' }], - bids: [{ price: '99.8', quantity: '2.3' }], - eventType: 'spot@public.aggre.depth.v3.api.pb@10ms', - fromVersion: '100', - toVersion: '100' + publicAggreDepthsSnapshot: { + asks: [['100.1', '1.2']], + bids: [['99.8', '2.3']], + lastUpdateId: 100, + timestamp: 1710000000002 } }, localTimestamp @@ -11358,14 +11356,12 @@ test('map mexc buffered depth updates with omitted side', () => { { channel: 'spot@public.aggre.depth.v3.api.pb@10ms@BTCUSDT', symbol: 'BTCUSDT', - sendTime: '1781740800000', generated: true, - publicAggreDepths: { - asks: [{ price: '64530.00', quantity: '1' }], - bids: [{ price: '64520.00', quantity: '1' }], - eventType: 'spot@public.aggre.depth.v3.api.pb@10ms', - fromVersion: '74483042020', - toVersion: '74483042020' + publicAggreDepthsSnapshot: { + asks: [['64530.00', '1']], + bids: [['64520.00', '1']], + lastUpdateId: 74483042020, + timestamp: 1781740800000 } }, localTimestamp @@ -11431,14 +11427,12 @@ test('map mexc historical buffered depth updates without enforcing full sequence { channel: 'spot@public.aggre.depth.v3.api.pb@10ms@BTCUSDT', symbol: 'BTCUSDT', - sendTime: '1781740800003', generated: true, - publicAggreDepths: { + publicAggreDepthsSnapshot: { asks: [], bids: [], - eventType: 'spot@public.aggre.depth.v3.api.pb@10ms', - fromVersion: '100', - toVersion: '100' + lastUpdateId: 100, + timestamp: 1781740800003 } }, localTimestamp @@ -11465,14 +11459,12 @@ test('map mexc realtime depth update throws when first update has no snapshot ov { channel: 'spot@public.aggre.depth.v3.api.pb@10ms@BTCUSDT', symbol: 'BTCUSDT', - sendTime: Date.now().toString(), generated: true, - publicAggreDepths: { + publicAggreDepthsSnapshot: { asks: [], bids: [], - eventType: 'spot@public.aggre.depth.v3.api.pb@10ms', - fromVersion: '100', - toVersion: '100' + lastUpdateId: 100, + timestamp: Date.now() } }, localTimestamp @@ -11505,14 +11497,12 @@ test('map mexc live captured messages', () => { { channel: 'spot@public.aggre.depth.v3.api.pb@10ms@BTCUSDT', symbol: 'BTCUSDT', - sendTime: '1782208005000', generated: true, - publicAggreDepths: { + publicAggreDepthsSnapshot: { asks: [], bids: [], - eventType: 'spot@public.aggre.depth.v3.api.pb@10ms', - fromVersion: '74891313262', - toVersion: '74891313262' + lastUpdateId: 74891313262, + timestamp: 1782208005000 } }, localTimestamp diff --git a/test/mexc-realtimefeed.test.ts b/test/mexc-realtimefeed.test.ts index eb47982..27fddf7 100644 --- a/test/mexc-realtimefeed.test.ts +++ b/test/mexc-realtimefeed.test.ts @@ -211,14 +211,12 @@ test('provide mexc manual depth snapshots', async () => { { channel: 'spot@public.aggre.depth.v3.api.pb@10ms@BTCUSDT', symbol: 'BTCUSDT', - sendTime: '1710000000000', generated: true, - 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: '100', - toVersion: '100' + publicAggreDepthsSnapshot: { + lastUpdateId: 100, + asks: [['100.1', '1.2']], + bids: [['99.9', '0.5']], + timestamp: 1782235749967 } } ]) @@ -230,10 +228,10 @@ test('provide mexc manual depth snapshots', async () => { 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']] }, - { lastUpdateId: 102, asks: [['100.2', '1.2']], bids: [['99.8', '0.5']] }, - { lastUpdateId: 103, asks: [['100.3', '1.2']], bids: [['99.7', '0.5']] }, - { lastUpdateId: 104, asks: [['100.4', '1.2']], bids: [['99.6', '0.5']] } + { 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 @@ -265,14 +263,12 @@ test('retry mexc manual depth snapshots until buffered update overlaps', async ( { channel: 'spot@public.aggre.depth.v3.api.pb@10ms@BTCUSDT', symbol: 'BTCUSDT', - sendTime: '1710000000000', generated: true, - publicAggreDepths: { - asks: [{ price: '100.4', quantity: '1.2' }], - bids: [{ price: '99.6', quantity: '0.5' }], - eventType: 'spot@public.aggre.depth.v3.api.pb@10ms', - fromVersion: '104', - toVersion: '104' + publicAggreDepthsSnapshot: { + lastUpdateId: 104, + asks: [['100.4', '1.2']], + bids: [['99.6', '0.5']], + timestamp: 1782235749967 } } ]) @@ -314,7 +310,9 @@ function varint(value: number) { } async function startSnapshotServer( - responses: MexcTestDepthSnapshotResponse[] = [{ lastUpdateId: 100, asks: [['100.1', '1.2']], bids: [['99.9', '0.5']] }] + responses: MexcTestDepthSnapshotResponse[] = [ + { lastUpdateId: 100, asks: [['100.1', '1.2']], bids: [['99.9', '0.5']], timestamp: 1782235749967 } + ] ) { let requestsCount = 0 const server = createServer((request, response) => { @@ -342,4 +340,5 @@ type MexcTestDepthSnapshotResponse = { lastUpdateId: number bids: string[][] asks: string[][] + timestamp: number } From 1f9075c484183b0f94093d832687f6cd091c72c4 Mon Sep 17 00:00:00 2001 From: marcin Date: Wed, 24 Jun 2026 17:22:19 +0200 Subject: [PATCH 17/17] feat(mexc-integration): adjusted to new snapshot shape --- src/consts.ts | 1 + src/mappers/mexc.ts | 18 ++++++++----- src/realtimefeeds/mexc.ts | 47 ++++++++++++++++++++++++++-------- test/mappers.test.ts | 10 ++++---- test/mexc-realtimefeed.test.ts | 46 +++++++++++++++++++++++++++++++-- 5 files changed, 99 insertions(+), 23 deletions(-) diff --git a/src/consts.ts b/src/consts.ts index 0d95ae7..b512bda 100644 --- a/src/consts.ts +++ b/src/consts.ts @@ -530,6 +530,7 @@ const BULLISH_CHANNELS = ['V1TALevel2', 'V1TALevel1', 'V1TAAnonymousTradeUpdate' 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 diff --git a/src/mappers/mexc.ts b/src/mappers/mexc.ts index c6ca8a7..87e8d7d 100644 --- a/src/mappers/mexc.ts +++ b/src/mappers/mexc.ts @@ -42,6 +42,7 @@ export class MexcTradesMapper implements Mapper<'mexc', Trade> { 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 @@ -50,14 +51,18 @@ export class MexcBookChangeMapper implements Mapper<'mexc', BookChange> { } canHandle(message: MexcDepthSnapshotMessage | MexcDepthUpdateMessage) { - return ( - message.channel?.startsWith(`${this.channel}@`) === true && - (message.generated === true ? 'publicAggreDepthsSnapshot' in message : 'publicAggreDepths' in message) - ) + 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[]) { - return [{ channel: this.channel, symbols: upperCaseSymbols(symbols) } as const] + 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 { @@ -233,6 +238,7 @@ type MexcControlMessage = { 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}` @@ -265,7 +271,7 @@ enum MexcTradeType { } export type MexcDepthSnapshotMessage = { - channel: MexcChannelWithSymbol<'spot@public.aggre.depth.v3.api.pb@10ms'> + channel: MexcChannelWithSymbol<'spot@public.aggre.depth.snapshot.v3.api.pb@10ms'> symbol: string generated: true publicAggreDepthsSnapshot: MexcAggreDepthsSnapshot diff --git a/src/realtimefeeds/mexc.ts b/src/realtimefeeds/mexc.ts index 767c8f2..0a30013 100644 --- a/src/realtimefeeds/mexc.ts +++ b/src/realtimefeeds/mexc.ts @@ -5,6 +5,8 @@ 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 /** @@ -71,7 +73,8 @@ export class MexcRealTimeFeed extends RealTimeFeedBase { protected readonly httpURL: string = 'https://api.mexc.com' private readonly channels = new Set([ 'spot@public.aggre.deals.v3.api.pb@10ms', - 'spot@public.aggre.depth.v3.api.pb@10ms', + MexcRealTimeFeed.depthChannel, + MexcRealTimeFeed.depthSnapshotChannel, 'spot@public.aggre.bookTicker.v3.api.pb@10ms' ]) @@ -88,12 +91,18 @@ export class MexcRealTimeFeed extends RealTimeFeedBase { return filter as Required> }) - this.resetDepthSnapshotTracking(filtersWithSymbols) + const depthSnapshotFilters = filtersWithSymbols.filter((filter) => filter.channel === MexcRealTimeFeed.depthSnapshotChannel) + this.validateDepthSnapshotFilters(filtersWithSymbols, depthSnapshotFilters) + this.resetDepthSnapshotTracking(depthSnapshotFilters) return [ { method: 'SUBSCRIPTION', - params: filtersWithSymbols.flatMap((filter) => filter.symbols.map((symbol) => `${filter.channel}@${symbol.toUpperCase()}`)) + params: filtersWithSymbols.flatMap((filter) => + filter.channel === MexcRealTimeFeed.depthSnapshotChannel + ? [] + : filter.symbols.map((symbol) => `${filter.channel}@${symbol.toUpperCase()}`) + ) } ] } @@ -122,7 +131,7 @@ export class MexcRealTimeFeed extends RealTimeFeedBase { protected override onMessage(message: any) { if ( - message.channel?.startsWith('spot@public.aggre.depth.v3.api.pb@10ms') !== true || + message.channel?.startsWith(MexcRealTimeFeed.depthChannel) !== true || message.symbol === undefined || this.pendingDepthSnapshotSymbols.has(message.symbol) === false ) { @@ -141,7 +150,7 @@ export class MexcRealTimeFeed extends RealTimeFeedBase { } protected override async provideManualSnapshots(filters: Filter[], shouldCancel: () => boolean) { - const depthFilter = filters.find((filter) => filter.channel === 'spot@public.aggre.depth.v3.api.pb@10ms') + const depthFilter = filters.find((filter) => filter.channel === MexcRealTimeFeed.depthSnapshotChannel) if (depthFilter === undefined) { return } @@ -172,10 +181,6 @@ export class MexcRealTimeFeed extends RealTimeFeedBase { this.bufferedDepthUpdates.clear() for (const filter of filters) { - if (filter.channel !== 'spot@public.aggre.depth.v3.api.pb@10ms') { - continue - } - for (const symbol of filter.symbols) { const upperCaseSymbol = symbol.toUpperCase() this.pendingDepthSnapshotSymbols.add(upperCaseSymbol) @@ -184,6 +189,28 @@ export class MexcRealTimeFeed extends RealTimeFeedBase { } } + 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 @@ -274,7 +301,7 @@ export class MexcRealTimeFeed extends RealTimeFeedBase { private createManualSnapshot(symbol: string, data: MexcDepthSnapshotResponse): MexcDepthSnapshotMessage { return { - channel: `spot@public.aggre.depth.v3.api.pb@10ms@${symbol}`, + channel: `${MexcRealTimeFeed.depthSnapshotChannel}@${symbol}`, symbol, generated: true, publicAggreDepthsSnapshot: { diff --git a/test/mappers.test.ts b/test/mappers.test.ts index 375f1a3..2683a8c 100644 --- a/test/mappers.test.ts +++ b/test/mappers.test.ts @@ -11143,7 +11143,7 @@ test('map mexc messages', () => { expect( mapper.map( { - channel: 'spot@public.aggre.depth.v3.api.pb@10ms@BTCUSDT', + channel: 'spot@public.aggre.depth.snapshot.v3.api.pb@10ms@BTCUSDT', symbol: 'BTCUSDT', generated: true, publicAggreDepthsSnapshot: { @@ -11354,7 +11354,7 @@ test('map mexc buffered depth updates with omitted side', () => { expect( mapper.map( { - channel: 'spot@public.aggre.depth.v3.api.pb@10ms@BTCUSDT', + channel: 'spot@public.aggre.depth.snapshot.v3.api.pb@10ms@BTCUSDT', symbol: 'BTCUSDT', generated: true, publicAggreDepthsSnapshot: { @@ -11425,7 +11425,7 @@ test('map mexc historical buffered depth updates without enforcing full sequence expect( mapper.map( { - channel: 'spot@public.aggre.depth.v3.api.pb@10ms@BTCUSDT', + channel: 'spot@public.aggre.depth.snapshot.v3.api.pb@10ms@BTCUSDT', symbol: 'BTCUSDT', generated: true, publicAggreDepthsSnapshot: { @@ -11457,7 +11457,7 @@ test('map mexc realtime depth update throws when first update has no snapshot ov mapper.map( { - channel: 'spot@public.aggre.depth.v3.api.pb@10ms@BTCUSDT', + channel: 'spot@public.aggre.depth.snapshot.v3.api.pb@10ms@BTCUSDT', symbol: 'BTCUSDT', generated: true, publicAggreDepthsSnapshot: { @@ -11495,7 +11495,7 @@ test('map mexc live captured messages', () => { expect( mapper.map( { - channel: 'spot@public.aggre.depth.v3.api.pb@10ms@BTCUSDT', + channel: 'spot@public.aggre.depth.snapshot.v3.api.pb@10ms@BTCUSDT', symbol: 'BTCUSDT', generated: true, publicAggreDepthsSnapshot: { diff --git a/test/mexc-realtimefeed.test.ts b/test/mexc-realtimefeed.test.ts index 27fddf7..af95a93 100644 --- a/test/mexc-realtimefeed.test.ts +++ b/test/mexc-realtimefeed.test.ts @@ -54,6 +54,10 @@ test('map mexc realtime subscriptions', () => { 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'] @@ -88,6 +92,36 @@ test('mexc realtime rejects unsupported channels', () => { ).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) @@ -192,6 +226,10 @@ test('provide mexc manual depth snapshots', async () => { { channel: 'spot@public.aggre.depth.v3.api.pb@10ms', symbols: ['btcusdt'] + }, + { + channel: 'spot@public.aggre.depth.snapshot.v3.api.pb@10ms', + symbols: ['btcusdt'] } ] @@ -209,7 +247,7 @@ test('provide mexc manual depth snapshots', async () => { expect(snapshots).toEqual([ { - channel: 'spot@public.aggre.depth.v3.api.pb@10ms@BTCUSDT', + channel: 'spot@public.aggre.depth.snapshot.v3.api.pb@10ms@BTCUSDT', symbol: 'BTCUSDT', generated: true, publicAggreDepthsSnapshot: { @@ -243,6 +281,10 @@ test('retry mexc manual depth snapshots until buffered update overlaps', async ( { channel: 'spot@public.aggre.depth.v3.api.pb@10ms', symbols: ['btcusdt'] + }, + { + channel: 'spot@public.aggre.depth.snapshot.v3.api.pb@10ms', + symbols: ['btcusdt'] } ] @@ -261,7 +303,7 @@ test('retry mexc manual depth snapshots until buffered update overlaps', async ( expect(server.requestsCount).toBe(4) expect(snapshots).toEqual([ { - channel: 'spot@public.aggre.depth.v3.api.pb@10ms@BTCUSDT', + channel: 'spot@public.aggre.depth.snapshot.v3.api.pb@10ms@BTCUSDT', symbol: 'BTCUSDT', generated: true, publicAggreDepthsSnapshot: {