import { NormalizedData , Disconnect , Trade } from './types' ; import { CappedSet } from './handy' ; export async function * filter < T extends NormalizedData | Disconnect > ( messages , filter ) { for await ( const message of messages ) { if ( filter ( message ) ) { yield message ; } } } export function uniqueTradesOnly < T extends NormalizedData | Disconnect > ( { maxWindow , onDuplicateFound , skipStaleOlderThanSeconds } = { maxWindow : 500 } ) { const perSymbolQueues = { } as { [ key ] : CappedSet < string > ; } ; return ( message ) => { if ( message . type !== 'trade' ) { return true ; } else { const trade = message as unknown as Trade ; if ( trade . id === undefined || trade . symbol . startsWith ( '.' ) ) { return true ; } else { let alreadySeenTrades = perSymbolQueues [ trade . symbol ] ; if ( alreadySeenTrades === undefined ) { perSymbolQueues [ trade . symbol ] = new CappedSet < string > ( maxWindow ) ; alreadySeenTrades = perSymbolQueues [ trade . symbol ] ; } const isDuplicate = alreadySeenTrades . has ( trade . id ) ; const isStale = skipStaleOlderThanSeconds !== undefined && trade . localTimestamp . valueOf ( ) - trade . timestamp . valueOf ( ) > skipStaleOlderThanSeconds * 1000 ; if ( isDuplicate || isStale ) { if ( onDuplicateFound !== undefined ) { onDuplicateFound ( trade ) ; } alreadySeenTrades . remove ( trade . id ) ; alreadySeenTrades . add ( trade . id ) ; return false ; } else { alreadySeenTrades . add ( trade . id ) ; return true ; } } } } ; }