import { strict as assert } from 'assert' ; import { crawlIndex , IndexKlineMsg , IndexTickerMsg } from 'crypto-crawler/dist/crawler/okex' ; import fetchMarkets from 'crypto-markets' ; import path from 'path' ; import yargs from 'yargs' ; import { MsgWriter , Publisher , RotatedFileWriter } from '../utils' ; import { Heartbeat } from '../utils/heartbeat' ; import { createLogger } from '../utils/logger' ; import { REDIS_TOPIC_SPOT_INDEX_PRICE } from './common' ; const commandModule = { command : 'crawler_spot_index_price' , describe : 'Crawl Spot index price' , builder : ( yargs ) => yargs . options ( { } ) , handler : async ( ) => { const swapMarkets = ( await fetchMarkets ( 'OKEx' , 'Swap' ) ) . filter ( ( m ) => m . active ) ; const okexIndexPairs = swapMarkets . map ( ( m ) => m . pair ) ; const PAIRS = ( process . env . PAIRS || ' ' ) . split ( ' ' ) . filter ( ( x ) => x ) ; const pairs = PAIRS . length > 0 ? PAIRS . filter ( ( x ) => okexIndexPairs . includes ( x ) ) : okexIndexPairs ; assert . ok ( pairs . length > 0 ) ; const logger = createLogger ( `crawler-spot-index-price` ) ; const heartbeat = new Heartbeat ( logger , 60 ) ; assert . ok ( process . env . DATA_DIR , 'Please define a DATA_DIR environment variable in .envrc' ) ; const REDIS_URL = process . env . REDIS_URL || 'redis://localhost:6379' ; const publisher = new Publisher < IndexTickerMsg > ( REDIS_URL ) ; const klinePublisher = new Publisher < IndexKlineMsg > ( REDIS_URL ) ; const fileWriters = { } ; pairs . forEach ( ( pair ) => { fileWriters [ pair ] = new RotatedFileWriter ( path . join ( process . env . DATA_DIR ! , 'Index' , 'OKEx' , 'Price' ) , `OKEx.Spot.Index.Price.${ pair }.` ) ; } ) ; const klineFileWriters = new Map < string , MsgWriter > ( ) ; const mymap = new Map < string , IndexKlineMsg > ( ) ; crawlIndex ( pairs , [ 'Ticker' , 'Kline' ] , async ( msg ) : Promise < void > => { heartbeat . updateHeartbeat ( ) ; if ( ( msg as IndexKlineMsg ) . interval ) { const klineMsg = msg as IndexKlineMsg ; const key = `OKEx-${ klineMsg . pair }-${ klineMsg . interval }` ; if ( mymap . has ( key ) ) { const prev = mymap . get ( key ) ! ; if ( msg . timestamp > prev . timestamp ) { klinePublisher . publish ( `brick-mover:spot_index_price_kline_${ klineMsg . interval }` , prev ) ; if ( ! klineFileWriters . has ( key ) ) { klineFileWriters . set ( key , new RotatedFileWriter ( path . join ( process . env . DATA_DIR ! , 'Index' , 'OKEx' , 'Kline' ) , `OKEx.Spot.Index.Kline.${ klineMsg . interval }.${ klineMsg . pair }.` ) ) ; } klineFileWriters . get ( key ) ! . write ( prev ) ; } } mymap . set ( key , klineMsg ) ; } else { const tickerMsg = msg as IndexTickerMsg ; publisher . publish ( REDIS_TOPIC_SPOT_INDEX_PRICE , tickerMsg ) ; fileWriters [ tickerMsg . pair ] . write ( tickerMsg ) ; } } ) ; } , } ; export default commandModule ;