import * as moment from 'moment' ; import { sleep } from '../../helpers' ; import { IWatcherConfig , Watcher } from '../Watcher' ; import { Exchange , OHLCV } from '../../Exchange/Exchange' ; import { logger } from '../../../logger' ; import { MEASUREMENT_OHLC , MEASUREMENT_OHLC_FILLED } from '../../Influx/constants' ; export interface IMarketWatcherConfig extends IWatcherConfig { exchange : string ; base : string ; quote : string ; extra : { refreshInterval : number ; maxHistory : string ; } ; } export class extends Watcher export class MarketWatcher extends Watcher { public symbol : string ; public conf : IMarketWatcherConfig ; private exchange : Exchange ; private shouldStop : boolean = false ; private interval : { id : NodeJS . Timeout | undefined ; } = { id : undefined } ; constructor ( conf ) { super ( conf ) ; this . conf = Object . assign ( { type : 'MarketWatcher' , extra : { refreshInterval : 30000 , maxHistory : '2018-01-01T00:00:00Z' , } , } , conf ) ; this . checkConfig ( this . conf ) ; this . symbol = `${ this . conf . base }/${ this . conf . quote }` ; this . exchange = new Exchange ( { name : this . conf . exchange } ) ; } public async init ( ) { if ( ! ( await this . exchange . getExchangeInfo ( this . symbol ) ) ) { throw new Error ( `[WATCHER] symbol ${ this . symbol } doesn't exist on ${ this . conf . exchange }` ) ; } } returns { Promise < void > } memberof public async runWatcher ( ) { logger . info ( `[WATCHER] Watcher started on ${ this . conf . exchange }: ${ this . symbol }` ) ; const tags = { base : this . conf . base , quote : this . conf . quote , exchange : this . conf . exchange } ; this . shouldStop = false ; await this . checkTimeserie ( ) . catch ( error => { logger . error ( error ) ; } ) ; while ( ! this . shouldStop ) { try { const data = await this . exchange . getCandles ( this . symbol , { limit : 2 , } ) ; this . getInflux ( ) . writeOHLC ( { base : this . conf . base , quote : this . conf . quote , exchange : this . conf . exchange } , data ) . catch ( error => { throw error ; } ) ; const start = moment ( ) . subtract ( 100 , 'm' ) . utc ( ) . format ( ) ; await this . getInflux ( ) . refreshOHLCFILLED ( tags , true , start ) ; } catch ( error ) { logger . error ( error ) ; throw new Error ( `Error while running market watcher loop ${ this . conf . exchange } (${ this . symbol })` ) ; } await sleep ( this . conf . extra . refreshInterval , this . interval ) ; } } memberof public async stopWatcher ( flushData ) { logger . info ( `[WATCHER] Watcher stopped on ${ this . conf . exchange }: ${ this . symbol }` ) ; this . shouldStop = true ; if ( this . interval . id ) { clearTimeout ( this . interval . id ) ; this . interval . id = undefined ; } if ( flushData ) { const { base , quote , exchange } = this . conf ; const tags = { base , quote , exchange } ; [ MEASUREMENT_OHLC_FILLED , MEASUREMENT_OHLC ] . forEach ( measurement => this . getInflux ( ) . dropSerie ( measurement , tags ) . catch ( ( error ) => logger . error ( error ) ) ) ; } } private param { IMarketWatcherConfig } config memberof private checkConfig ( conf ) { if ( conf . extra . refreshInterval < 1000 ) { throw new Error ( '[WATCHER] Cannot create market watcher with refreshInterval < 1000 (1 seconde)' ) ; } } private returns { Promise < void > } memberof private async checkTimeserie ( ) { if ( ( await this . getInflux ( ) . count ( MEASUREMENT_OHLC , { base : this . conf . base , quote : this . conf . quote , exchange : this . conf . exchange , } ) ) === 0 ) { await this . fillHistory ( this . conf . extra . maxHistory ) ; } else await this . fillGap ( ) ; } private returns { Promise < void > } memberof private async fillHistory ( from ) { try { const batchSize = 500 ; const start = moment ( from ) ; const end = moment ( ) . subtract ( batchSize , 'm' ) ; let lastCandleFetch = null ; let data = [ ] ; const tags = { base : this . conf . base , quote : this . conf . quote , exchange : this . conf . exchange } ; while ( ! this . shouldStop && end . diff ( start , 'm' ) > batchSize ) { data = await this . exchange . getCandles ( this . symbol , { limit : batchSize , since : end . toDate ( ) . getTime ( ) , } ) ; if ( lastCandleFetch && data [ 0 ] . time === lastCandleFetch . time ) { await this . getInflux ( ) . refreshOHLCFILLED ( tags , true ) ; return ; } await this . getInflux ( ) . writeOHLC ( tags , data ) ; lastCandleFetch = data [ 0 ] ; end . subtract ( batchSize , 'm' ) ; } data = await this . exchange . getCandles ( this . symbol , { limit : batchSize , since : start . toDate ( ) . getTime ( ) , } ) ; await this . getInflux ( ) . writeOHLC ( tags , data ) ; await this . getInflux ( ) . refreshOHLCFILLED ( tags , true ) ; } catch ( error ) { logger . error ( error ) ; throw new Error ( `[Watcher] Error when filling history ${ this . conf . exchange } (${ this . symbol })` ) ; } } private memberof private async fillGap ( ) { try { const tags = { base : this . conf . base , quote : this . conf . quote , exchange : this . conf . exchange } ; const missingPoints = await this . getInflux ( ) . getSeriesGap ( MEASUREMENT_OHLC , tags , '1m' ) ; if ( missingPoints . length > 0 ) { const batchSize = 500 ; let i = 0 ; while ( ! this . shouldStop && i < missingPoints . length ) { let j = i + 1 ; const start = moment ( missingPoints [ i ] ) ; let next = moment ( missingPoints [ j ] ) ; while ( j < missingPoints . length && Math . abs ( start . diff ( next , 'm' ) ) < batchSize ) { j ++ ; next = moment ( missingPoints [ j ] ) ; } const data = await this . exchange . getCandles ( this . symbol , { limit : batchSize , since : start . toDate ( ) . getTime ( ) , } ) ; if ( data . length > 0 ) { await this . getInflux ( ) . writeOHLC ( tags , data ) ; await this . getInflux ( ) . writeOHLC ( tags , data , MEASUREMENT_OHLC_FILLED ) ; } i = j ; } logger . info ( `[Watcher] ${ missingPoints . length } missing points filled into InfluxDB for serie ${ this . conf . exchange } (${ this . symbol })` ) ; await this . getInflux ( ) . refreshOHLCFILLED ( tags ) ; } } catch ( error ) { logger . error ( error ) ; throw new Error ( `[Watcher] Error when filling gap ${ this . conf . exchange } (${ this . symbol })` ) ; } } }