import { debug } from '../debug' ; import { CircularBuffer , lowerCaseSymbols } from '../handy' ; import { BookChange , BookTicker , DerivativeTicker , Exchange , FilterForExchange , Liquidation , Trade } from '../types' ; import { Mapper , PendingTickerInfoHelper } from './mapper' ; export class BinanceTradesMapper implements Mapper < 'binance' | 'binance-jersey' | 'binance-us' | 'binance-futures' | 'binance-delivery' , Trade > { constructor ( private readonly _exchange ) { } canHandle ( message ) { if ( message . stream === undefined ) { return false ; } return message . stream . endsWith ( '@trade' ) ; } getFilters ( symbols ? ) { symbols = lowerCaseSymbols ( symbols ) ; return [ { channel : 'trade' , symbols } as const ] ; } * map ( binanceTradeResponse , localTimestamp ) { const binanceTrade = binanceTradeResponse . data ; const isOffBookTrade = binanceTrade . X === 'INSURANCE_FUND' || binanceTrade . X === 'ADL' ; if ( isOffBookTrade ) { return ; } const trade = { type : 'trade' , symbol : binanceTrade . s , exchange : this . _exchange , id : String ( binanceTrade . t ) , price : Number ( binanceTrade . p ) , amount : Number ( binanceTrade . q ) , side : binanceTrade . m ? 'sell' : 'buy' , timestamp : new Date ( binanceTrade . T ) , localTimestamp : localTimestamp } ; yield trade ; } } export class BinanceBookChangeMapper implements Mapper < 'binance' | 'binance-jersey' | 'binance-us' | 'binance-futures' | 'binance-delivery' , BookChange > { protected readonly symbolToDepthInfoMapping : { [ key ] : LocalDepthInfo ; } = { } ; constructor ( protected readonly exchange , protected readonly ignoreBookSnapshotOverlapError ) { } canHandle ( message ) { if ( message . stream === undefined ) { return false ; } return message . stream . includes ( '@depth' ) ; } getFilters ( symbols ? ) { symbols = lowerCaseSymbols ( symbols ) ; return [ { channel : 'depth' , symbols } as const , { channel : 'depthSnapshot' , symbols } as const ] ; } * map ( message , localTimestamp ) { const symbol = message . stream . split ( '@' ) [ 0 ] . toUpperCase ( ) ; if ( this . symbolToDepthInfoMapping [ symbol ] === undefined ) { this . symbolToDepthInfoMapping [ symbol ] = { bufferedUpdates : new CircularBuffer < BinanceDepthData > ( 2000 ) } ; } const symbolDepthInfo = this . symbolToDepthInfoMapping [ symbol ] ; const snapshotAlreadyProcessed = symbolDepthInfo . snapshotProcessed ; if ( message . data . lastUpdateId !== undefined ) { if ( snapshotAlreadyProcessed ) { return ; } const binanceDepthSnapshotData = message . data ; symbolDepthInfo . lastUpdateId = binanceDepthSnapshotData . lastUpdateId ; symbolDepthInfo . snapshotProcessed = true ; for ( const update of symbolDepthInfo . bufferedUpdates . items ( ) ) { const bookChange = this . mapBookDepthUpdate ( update , localTimestamp ) ; if ( bookChange !== undefined ) { for ( const bid of update . b ) { const matchingBid = binanceDepthSnapshotData . bids . find ( ( b ) => b [ 0 ] === bid [ 0 ] ) ; if ( matchingBid !== undefined ) { matchingBid [ 1 ] = bid [ 1 ] ; } else { binanceDepthSnapshotData . bids . push ( bid ) ; } } for ( const ask of update . a ) { const matchingAsk = binanceDepthSnapshotData . asks . find ( ( a ) => a [ 0 ] === ask [ 0 ] ) ; if ( matchingAsk !== undefined ) { matchingAsk [ 1 ] = ask [ 1 ] ; } else { binanceDepthSnapshotData . asks . push ( ask ) ; } } } } symbolDepthInfo . bufferedUpdates . clear ( ) ; const bookChange = { type : 'book_change' , symbol , exchange : this . exchange , isSnapshot : true , bids : binanceDepthSnapshotData . bids . map ( this . mapBookLevel ) , asks : binanceDepthSnapshotData . asks . map ( this . mapBookLevel ) , timestamp : binanceDepthSnapshotData . T !== undefined ? new Date ( binanceDepthSnapshotData . T ) : localTimestamp , localTimestamp } ; yield bookChange ; } else if ( snapshotAlreadyProcessed ) { const bookChange = this . mapBookDepthUpdate ( message . data as BinanceDepthData , localTimestamp ) ; if ( bookChange !== undefined ) { yield bookChange ; } } else { const binanceDepthUpdateData = message . data as BinanceDepthData ; symbolDepthInfo . bufferedUpdates . append ( binanceDepthUpdateData ) ; } } protected mapBookDepthUpdate ( binanceDepthUpdateData , localTimestamp ) { const depthContext = this . symbolToDepthInfoMapping [ binanceDepthUpdateData . s ] ! ; const lastUpdateId = depthContext . lastUpdateId ! ; if ( binanceDepthUpdateData . u <= lastUpdateId ) { return ; } if ( ! depthContext . validatedFirstUpdate ) { const bookSnapshotIsEmpty = lastUpdateId == - 1 ; if ( ( binanceDepthUpdateData . U <= lastUpdateId + 1 && binanceDepthUpdateData . u >= lastUpdateId + 1 ) || bookSnapshotIsEmpty ) { depthContext . validatedFirstUpdate = true ; } else { const message = `Book depth snaphot has no overlap with first update, update ${ JSON . stringify ( binanceDepthUpdateData ) }, lastUpdateId: ${ lastUpdateId }, exchange ${ this . exchange }` ; if ( this . ignoreBookSnapshotOverlapError ) { depthContext . validatedFirstUpdate = true ; debug ( message ) ; } else { throw new Error ( message ) ; } } } return { type : 'book_change' , symbol : binanceDepthUpdateData . s , exchange : this . exchange , isSnapshot : false , bids : binanceDepthUpdateData . b . map ( this . mapBookLevel ) , asks : binanceDepthUpdateData . a . map ( this . mapBookLevel ) , timestamp : new Date ( binanceDepthUpdateData . E ) , localTimestamp : localTimestamp } ; } protected mapBookLevel ( level ) { const price = Number ( level [ 0 ] ) ; const amount = Number ( level [ 1 ] ) ; return { price , amount } ; } } export class BinanceFuturesBookChangeMapper extends BinanceBookChangeMapper implements Mapper < 'binance-futures' | 'binance-delivery' , BookChange > { constructor ( protected readonly exchange , protected readonly ignoreBookSnapshotOverlapError ) { super ( exchange , ignoreBookSnapshotOverlapError ) ; } protected mapBookDepthUpdate ( binanceDepthUpdateData , localTimestamp ) { const depthContext = this . symbolToDepthInfoMapping [ binanceDepthUpdateData . s ] ! ; const lastUpdateId = depthContext . lastUpdateId ! ; if ( binanceDepthUpdateData . u < lastUpdateId ) { return ; } if ( ! depthContext . validatedFirstUpdate ) { if ( binanceDepthUpdateData . U <= lastUpdateId && binanceDepthUpdateData . u >= lastUpdateId ) { depthContext . validatedFirstUpdate = true ; } else { const message = `Book depth snaphot has no overlap with first update, update ${ JSON . stringify ( binanceDepthUpdateData ) }, lastUpdateId: ${ lastUpdateId }, exchange ${ this . exchange }` ; if ( this . ignoreBookSnapshotOverlapError ) { depthContext . validatedFirstUpdate = true ; debug ( message ) ; } else { throw new Error ( message ) ; } } } return { type : 'book_change' , symbol : binanceDepthUpdateData . s , exchange : this . exchange , isSnapshot : false , bids : binanceDepthUpdateData . b . map ( this . mapBookLevel ) , asks : binanceDepthUpdateData . a . map ( this . mapBookLevel ) , timestamp : new Date ( binanceDepthUpdateData . T ) , localTimestamp : localTimestamp } ; } } export class BinanceFuturesDerivativeTickerMapper implements Mapper < 'binance-futures' | 'binance-delivery' , DerivativeTicker > { private readonly pendingTickerInfoHelper = new PendingTickerInfoHelper ( ) ; private readonly _indexPrices = new Map < string , number > ( ) ; constructor ( protected readonly exchange ) { } canHandle ( message ) { if ( message . stream === undefined ) { return false ; } return ( message . stream . includes ( '@markPrice' ) || message . stream . endsWith ( '@ticker' ) || message . stream . endsWith ( '@openInterest' ) || message . stream . includes ( '@indexPrice' ) ) ; } getFilters ( symbols ? ) { symbols = lowerCaseSymbols ( symbols ) ; const filters = [ { channel : 'markPrice' , symbols } as const , { channel : 'ticker' , symbols } as const , { channel : 'openInterest' , symbols } as const ] ; if ( this . exchange === 'binance-delivery' ) { filters . push ( { channel : 'indexPrice' as any , symbols : symbols !== undefined ? symbols . map ( ( s ) => s . split ( '_' ) [ 0 ] ) : undefined } ) ; } return filters ; } * map ( message , localTimestamp ) { if ( message . data . e === 'indexPriceUpdate' ) { this . _indexPrices . set ( message . data . i , Number ( message . data . p ) ) ; } else { const symbol = 's' in message . data ? message . data . s : message . data . symbol ; const pendingTickerInfo = this . pendingTickerInfoHelper . getPendingTickerInfo ( symbol , this . exchange ) ; const lastIndexPrice = this . _indexPrices . get ( symbol . split ( '_' ) [ 0 ] ) ; if ( lastIndexPrice !== undefined ) { pendingTickerInfo . updateIndexPrice ( lastIndexPrice ) ; } if ( message . data . e === 'markPriceUpdate' ) { if ( 'r' in message . data && message . data . r !== '' && message . data . T !== 0 ) { pendingTickerInfo . updateFundingRate ( Number ( message . data . r ) ) ; pendingTickerInfo . updateFundingTimestamp ( new Date ( message . data . T ! ) ) ; } if ( message . data . i !== undefined ) { pendingTickerInfo . updateIndexPrice ( Number ( message . data . i ) ) ; } pendingTickerInfo . updateMarkPrice ( Number ( message . data . p ) ) ; pendingTickerInfo . updateTimestamp ( new Date ( message . data . E ) ) ; } if ( message . data . e === '24hrTicker' ) { pendingTickerInfo . updateLastPrice ( Number ( message . data . c ) ) ; pendingTickerInfo . updateTimestamp ( new Date ( message . data . E ) ) ; } if ( 'openInterest' in message . data ) { pendingTickerInfo . updateOpenInterest ( Number ( message . data . openInterest ) ) ; } if ( pendingTickerInfo . hasChanged ( ) ) { yield pendingTickerInfo . getSnapshot ( localTimestamp ) ; } } } } export class BinanceLiquidationsMapper implements Mapper < 'binance-futures' | 'binance-delivery' , Liquidation > { constructor ( private readonly _exchange ) { } canHandle ( message ) { if ( message . stream === undefined ) { return false ; } return message . stream . endsWith ( '@forceOrder' ) ; } getFilters ( symbols ? ) { symbols = lowerCaseSymbols ( symbols ) ; return [ { channel : 'forceOrder' , symbols } as const ] ; } * map ( binanceTradeResponse , localTimestamp ) { const binanceLiquidation = binanceTradeResponse . data . o ; if ( binanceLiquidation . X !== 'FILLED' ) { return ; } const liquidation = { type : 'liquidation' , symbol : binanceLiquidation . s , exchange : this . _exchange , id : undefined , price : Number ( binanceLiquidation . ap ) , amount : Number ( binanceLiquidation . z ) , side : binanceLiquidation . S === 'SELL' ? 'sell' : 'buy' , timestamp : new Date ( binanceLiquidation . T ) , localTimestamp : localTimestamp } ; yield liquidation ; } } export class BinanceBookTickerMapper implements Mapper < 'binance-futures' | 'binance-delivery' | 'binance' , BookTicker > { constructor ( private readonly _exchange ) { } canHandle ( message ) { if ( message . stream === undefined ) { return false ; } return message . stream . endsWith ( '@bookTicker' ) ; } getFilters ( symbols ? ) { symbols = lowerCaseSymbols ( symbols ) ; return [ { channel : 'bookTicker' , symbols } as const ] ; } * map ( binanceBookTickerResponse , localTimestamp ) { const binanceBookTicker = binanceBookTickerResponse . data ; const ticker = { type : 'book_ticker' , symbol : binanceBookTicker . s , exchange : this . _exchange , askAmount : binanceBookTicker . A !== undefined ? Number ( binanceBookTicker . A ) : undefined , askPrice : binanceBookTicker . a !== undefined ? Number ( binanceBookTicker . a ) : undefined , bidPrice : binanceBookTicker . b !== undefined ? Number ( binanceBookTicker . b ) : undefined , bidAmount : binanceBookTicker . B !== undefined ? Number ( binanceBookTicker . B ) : undefined , timestamp : binanceBookTicker . T !== undefined ? new Date ( binanceBookTicker . T ) : localTimestamp , localTimestamp : localTimestamp } ; yield ticker ; } } type BinanceResponse < T > = { stream : string ; data : T ; } ; type BinanceTradeData = { s : string ; t : number ; p : string ; q : string ; T : number ; m : true ; X ? : 'INSURANCE_FUND' | 'MARKET' | 'ADL' ; } ; type BinanceBookLevel = [ string , string ] ; type BinanceDepthData = { lastUpdateId : undefined ; E : number ; s : string ; U : number ; u : number ; b : BinanceBookLevel [ ] ; a : BinanceBookLevel [ ] ; } ; type BinanceFuturesDepthData = BinanceDepthData & { pu : number ; T : number ; } ; type BinanceDepthSnapshotData = { lastUpdateId : number ; bids : BinanceBookLevel [ ] ; asks : BinanceBookLevel [ ] ; T ? : number ; } ; type LocalDepthInfo = { bufferedUpdates : CircularBuffer < BinanceDepthData > ; snapshotProcessed ? : boolean ; lastUpdateId ? : number ; validatedFirstUpdate ? : boolean ; } ; type BinanceFuturesMarkPriceData = { e : 'markPriceUpdate' ; s : string ; E : number ; p : string ; r ? : string ; T ? : number ; i ? : string ; } ; type BinanceFuturesTickerData = { e : '24hrTicker' ; E : number ; s : string ; c : string ; } ; type BinanceFuturesOpenInterestData = { e : undefined ; symbol : string ; openInterest : string ; } ; type BinanceFuturesIndexPriceData = { e : 'indexPriceUpdate' ; E : 1591261236000 ; i : string ; p : string ; } ; type BinanceFuturesForceOrderData = { o : { s : string ; S : string ; q : string ; p : string ; ap : string ; X : 'FILLED' ; T : 1568014460893 ; z : string ; } ; } ; type BinanceBookTickerData = { u : number ; s : string ; b : string ; B : string ; a : string ; A : string ; T ? : number ; } ;