import * as moment from 'moment' ; import { InfluxDB } from 'influx' ; import { logger } from '../../logger' ; import { OHLCV } from '../Exchange/Exchange' ; import { MEASUREMENT_OHLC , MEASUREMENT_OHLC_FILLED } from './constants' ; import { tagsToString } from './helpers' ; import { sleep } from '../helpers' ; export interface InfluxConfig { host : string ; port : number ; stockDatabase : string ; eventDatabase : string ; } export class Influx { private influx : InfluxDB ; constructor ( public conf ) { this . influx = new InfluxDB ( { host : this . conf . host , port : this . conf . port , } ) ; } public async init ( ) { try { await this . createDatabaseIfNotExist ( this . conf . stockDatabase ) ; await this . createDatabaseIfNotExist ( this . conf . eventDatabase ) ; logger . info ( `[INFLUXDB] Connection successful: ${ this . conf . host }:${ this . conf . port }` ) ; } catch ( error ) { logger . error ( new Error ( `[INFLUXDB] Connection error: ${ this . conf . host }:${ this . conf . port }` ) ) ; throw error ; } } param { { symbol : string } } tags param { OHLCV [ ] } data memberof public async writeOHLC ( tags , data , measurement = MEASUREMENT_OHLC ) { const points = data . map ( ( { time , open , high , low , close , volume } ) => { return { measurement , tags , fields : { open , high , low , close , volume } , timestamp : moment ( time ) . toDate ( ) . getTime ( ) , } ; } ) ; await this . influx . writePoints ( points , { database : this . conf . stockDatabase , precision : 'ms' , } ) ; } param { { symbol : string } } tags param { string } aggregatedTime memberof public async getOHLC ( tags , aggregatedTime = '1m' ) { try { const ret = await this . influx . query ( `SELECT first(open) as open, max(high) as high, min(low) as low, last(close) as close
         FROM ${ MEASUREMENT_OHLC }
         WHERE ${ tagsToString ( tags ) }
         GROUP BY time(${ aggregatedTime })` , { database : this . conf . stockDatabase , } ) ; return ret ; } catch ( error ) { logger . error ( error ) ; throw new Error ( '[INFLUX] Problem while fetching OHLC data' ) ; } } param { string } measurement param { { [ name : string ] : string } } tags param { string } aggregatedTime returns { string [ ] } memberof public async getSeriesGap ( measurement , tags ? , aggregatedTime = '1m' ) { const query = `SELECT * FROM (
        SELECT max(close) as close 
        FROM ${ measurement }
        ${ tags ? `WHERE ${ tagsToString ( tags ) }` : '' }
        GROUP BY time(${ aggregatedTime }) fill(-1)
      ) WHERE close = -1` ; try { const ret = await this . influx . query ( query , { database : this . conf . stockDatabase , } ) ; return ret . map ( ( el ) => el . time . _nanoISO ) ; } catch ( error ) { logger . error ( error ) ; throw new Error ( `[INFLUX] Problem during query fetchGap\n${ query }` ) ; } } param { string } measurement param { { [ name : string ] : string } } tags returns { Promise < number > } memberof public async count ( measurement , tags ? ) { const query = `
      SELECT count(close)
      FROM ${ measurement }
      ${ tags ? `WHERE ${ tagsToString ( tags ) }` : '' }
    ` ; try { const ret = ( < any > await this . influx . query ( query , { database : this . conf . stockDatabase , } ) ) [ 0 ] ; return ret ? ret . count : 0 ; } catch ( error ) { logger . error ( error ) ; throw new Error ( `[INFLUX] Problem with count query\n${ query }` ) ; } } public async refreshOHLCFILLED ( tags , force = false , date ? ) { try { const query = ( time ? ) => `
        SELECT first(open) as open, max(high) as high, min(low) as low, last(close) as close, sum(volume) as volume
        INTO ${ MEASUREMENT_OHLC_FILLED }
        FROM ${ MEASUREMENT_OHLC }
        WHERE ${ tagsToString ( tags ) }
        ${ time ? `AND time >= '${ time }'` : '' }
        GROUP BY time(1m), * fill(linear)
      ` ; if ( force ) { await this . influx . query ( query ( date ) , { database : this . conf . stockDatabase } ) ; } else { const ret = await this . getSeriesGap ( MEASUREMENT_OHLC_FILLED ) ; if ( ret . length > 0 ) { const start = moment ( ret [ 0 ] ) . subtract ( 100 , 'm' ) . utc ( ) . format ( ) ; await this . influx . query ( query ( start ) , { database : this . conf . stockDatabase } ) ; } } } catch ( error ) { logger . error ( error ) ; throw new Error ( `[INFLUX] Problem while refreshing ${ MEASUREMENT_OHLC_FILLED }` ) ; } } public async dropSerie ( measurement , tags ) { await this . influx . query ( `DROP SERIES FROM "${ measurement }" WHERE ${ tagsToString ( tags ) }` , { database : this . conf . stockDatabase , } ) ; } private param { string } name memberof private async createDatabaseIfNotExist ( name ) { try { const databases = await this . influx . getDatabaseNames ( ) ; if ( ! databases . includes ( name ) ) await this . influx . createDatabase ( name ) ; } catch ( error ) { throw error ; } } }