import { Trade } from 'ccxt' ; import { timebucket } from '@magic8bot/timebucket' ; import { EventBusEmitter , EventBusListener } from '@magic8bot/event-bus' ; import { PeriodItem , eventBus , EVENT , wsServer , TradeCollection } from '@lib' ; import { StoreOpts } from '@m8bTypes' ; import { logger } from '../util/logger' ; const singleton = Symbol ( ) ; interface PeriodConf { periods : string [ ] ; lookbackSize : number ; } enum PERIOD_STATE { PREROLL , STOPPED , RUNNING } export class PeriodStore { public static get instance ( ) : PeriodStore { if ( ! this [ singleton ] ) this [ singleton ] = new PeriodStore ( ) ; return this [ singleton ] ; } public periods : Map < string , Map < string , PeriodItem [ ] > > = new Map ( ) ; private periodConfigs : Map < string , PeriodConf > = new Map ( ) ; private updateEmitters : Map < string , Map < string , EventBusEmitter < PeriodItem [ ] > > > = new Map ( ) ; private periodEmitters : Map < string , Map < string , EventBusEmitter < PeriodItem [ ] > > > = new Map ( ) ; private tradeEventTimeouts : Map < string , Map < string , NodeJS . Timer > > = new Map ( ) ; private periodTimer : Map < string , Map < string , NodeJS . Timer > > = new Map ( ) ; private periodStates : Map < string , PERIOD_STATE > = new Map ( ) ; private constructor ( ) { } public addSymbol ( { exchange , symbol , strategy } , conf ) { const idStr = this . makeIdStr ( { exchange , symbol , strategy } ) ; if ( this . periods . has ( idStr ) ) return ; logger . debug ( `Adding ${ idStr } to period store.` ) ; this . periodConfigs . set ( idStr , conf ) ; this . tradeEventTimeouts . set ( idStr , new Map ( ) ) ; this . periodTimer . set ( idStr , new Map ( ) ) ; this . periodStates . set ( idStr , PERIOD_STATE . PREROLL ) ; const tradeListener = eventBus . get ( EVENT . XCH_TRADE ) ( exchange ) ( symbol ) . listen ; tradeListener ( ( trade ) => this . addTrade ( idStr , trade ) ) ; eventBus . get ( EVENT . XCH_TRADE_PREROLL ) ( exchange ) ( symbol ) . listen ( ( trades ) => trades . forEach ( ( trade ) => this . addTrade ( idStr , trade , true ) ) ) ; this . updateEmitters . set ( idStr , new Map ( ) ) ; this . periodEmitters . set ( idStr , new Map ( ) ) ; this . periods . set ( idStr , new Map ( ) ) ; const updateEmitters = this . updateEmitters . get ( idStr ) ; const periodEmitters = this . periodEmitters . get ( idStr ) ; const periods = this . periods . get ( idStr ) ; conf . periods . forEach ( ( period ) => { updateEmitters . set ( period , ( periods ) => eventBus . get ( EVENT . PERIOD_UPDATE ) ( exchange ) ( symbol ) ( strategy ) . emit ( { period , periods } ) ) ; periodEmitters . set ( period , ( ) => eventBus . get ( EVENT . PERIOD_NEW ) ( exchange ) ( symbol ) ( strategy ) . emit ( period ) ) ; periods . set ( period , [ ] ) ; } ) ; } public start ( storeOpts ) { const idStr = this . makeIdStr ( storeOpts ) ; this . periodStates . set ( idStr , PERIOD_STATE . RUNNING ) ; const conf = this . periodConfigs . get ( idStr ) ; conf . periods . forEach ( ( period ) => { const timer = setTimeout ( ( ) => this . publishPeriod ( idStr , period ) , this . getPeriodTimeout ( idStr , period ) ) ; this . periodTimer . get ( idStr ) . set ( period , timer ) ; } ) ; } public stop ( storeOpts ) { const idStr = this . makeIdStr ( storeOpts ) ; this . periodStates . set ( idStr , PERIOD_STATE . STOPPED ) ; const conf = this . periodConfigs . get ( idStr ) ; conf . periods . forEach ( ( period ) => { clearTimeout ( this . periodTimer . get ( idStr ) . get ( period ) ) ; } ) ; } public addTrade ( idStr , trade , isPreroll = false ) { const periodState = this . periodStates . get ( idStr ) ; if ( periodState === PERIOD_STATE . STOPPED ) return ; const conf = this . periodConfigs . get ( idStr ) ; conf . periods . forEach ( ( period ) => { const periods = this . periods . get ( idStr ) . get ( period ) ; const { amount , price , timestamp } = trade ; const bucket = timebucket ( timestamp ) . resize ( period ) . toMilliseconds ( ) ; if ( ! periods . length ) this . newPeriod ( idStr , period , bucket ) ; const tradeBasedPeriodChange = isPreroll && periods . length && periods [ 0 ] . time < bucket ; if ( tradeBasedPeriodChange ) { this . checkPeriodWithoutTrades ( idStr , period ) ; this . emitTradeImmediate ( idStr , period ) ; this . periodEmitters . get ( idStr ) . get ( period ) ( ) ; this . newPeriod ( idStr , period , bucket ) ; } if ( ! periods [ 0 ] . open ) periods [ 0 ] . open = price ; periods [ 0 ] . trades ++ ; periods [ 0 ] . low = ! periods [ 0 ] . low ? price : Math . min ( price , periods [ 0 ] . low ) ; periods [ 0 ] . high = Math . max ( price , periods [ 0 ] . high ) ; periods [ 0 ] . close = price ; periods [ 0 ] . volume += amount ; if ( ! isPreroll ) this . emitTrades ( idStr , period ) ; } ) ; } public newPeriod ( idStr , period , bucket ) { const { lookbackSize } = this . periodConfigs . get ( idStr ) ; const periods = this . periods . get ( idStr ) . get ( period ) ; periods . unshift ( { close : null , high : null , low : null , open : null , time : bucket , volume : null , trades : 0 , bucket , } ) ; if ( periods . length > lookbackSize ) periods . pop ( ) ; } public clearPeriods ( idStr , period ) { this . periods . get ( idStr ) . set ( period , [ ] ) ; } param idStr private emitTrades ( idStr , period ) { if ( this . periodStates . get ( idStr ) === PERIOD_STATE . STOPPED ) return ; if ( this . tradeEventTimeouts . get ( idStr ) . has ( period ) ) return ; const fn = ( ) => this . emitTradeImmediate ( idStr , period ) ; this . tradeEventTimeouts . get ( idStr ) . set ( period , setTimeout ( fn , 100 ) ) ; } param idStr private emitTradeImmediate ( idStr , period , isPreroll = false ) { if ( this . periodStates . get ( idStr ) === PERIOD_STATE . STOPPED ) return ; const periods = this . periods . get ( idStr ) . get ( period ) ; this . updateEmitters . get ( idStr ) . get ( period ) ( [ ... periods ] ) ; if ( ! isPreroll ) { } if ( this . tradeEventTimeouts . get ( idStr ) . has ( period ) ) { clearTimeout ( this . tradeEventTimeouts . get ( idStr ) . get ( period ) ) ; this . tradeEventTimeouts . get ( idStr ) . delete ( period ) ; } } private makeIdStr ( { exchange , symbol , strategy } ) { return `${ exchange }.${ symbol }.${ strategy }` ; } private getPeriodTimeout ( idStr , period ) { const bucketEnd = timebucket ( period ) . getEndOfBucketMS ( ) ; const now = new Date ( ) . getTime ( ) ; logger . silly ( `Next ${ idStr } period will be published at ${ new Date ( bucketEnd ) . toTimeString ( ) }` ) ; return bucketEnd > now ? bucketEnd - now : 0 ; } private checkPeriodWithoutTrades ( idStr , period ) { if ( this . periodStates . get ( idStr ) === PERIOD_STATE . STOPPED ) return ; const periods = this . periods . get ( idStr ) . get ( period ) ; if ( periods . length < 2 ) return ; const [ thisPeriod , lastPeriod ] = periods ; const { close } = lastPeriod ; if ( ! thisPeriod . open && close ) { thisPeriod . open = close ; thisPeriod . high = close ; thisPeriod . low = close ; thisPeriod . close = close ; thisPeriod . volume = 0 ; logger . silly ( `Emitting PERIOD_UPDATE for empty period without any trade.` ) ; this . emitTrades ( idStr , period ) ; } } private publishPeriod ( idStr , period ) { clearTimeout ( this . periodTimer . get ( idStr ) . get ( period ) ) ; if ( this . periodStates . get ( idStr ) === PERIOD_STATE . STOPPED ) return ; this . checkPeriodWithoutTrades ( idStr , period ) ; this . periodEmitters . get ( idStr ) . get ( period ) ( ) ; wsServer . broadcast ( 'period-new' , { ... this . parseIdStr ( idStr ) , period : this . periods . get ( idStr ) . get ( period ) [ 0 ] } ) ; this . newPeriod ( idStr , period , timebucket ( period ) . toMilliseconds ( ) ) ; const fn = ( ) => this . publishPeriod ( idStr , period ) ; this . periodTimer . get ( idStr ) . set ( period , setTimeout ( fn , this . getPeriodTimeout ( idStr , period ) ) ) ; } private parseIdStr ( idStr ) { const [ exchange , symbol , strategy ] = idStr . split ( '.' ) ; return { exchange , symbol , strategy } ; } }