import { EventName , IBApiTickType , Stock } from '@stoqey/ib' ; import { isArray } from 'lodash' ; import isEmpty from 'lodash/isEmpty' ; import { IBKRConnection } from '../connection' ; import { ContractObject , ContractSummary , convertContractToContractDetailsParams , getContractDetails , } from '../contracts' ; import { IbkrEvents , IBKREVENTS , publishDataToTopic } from '../events' ; import { log , verbose } from '../log' ; import { getRadomReqId } from '../_utils/text.utils' ; import { PriceUpdatesEvent , TickPrice } from './price.interfaces' ; const ibEvents = IbkrEvents . Instance ; interface SymbolWithTicker { readonly tickerId : number ; readonly conId : number | undefined ; readonly symbol : string ; readonly tickTypes ? : readonly TickPrice [ ] ; } interface ReqPriceUpdates { readonly tickType ? : TickPrice | readonly TickPrice [ ] ; } interface ISubscribe { readonly contract : string | Partial < ContractObject > | ContractSummary ; readonly opt ? : ReqPriceUpdates ; } export class PriceUpdates { ib : any ; keyToTickerId : { [ key ] : number ; } = { } ; tickerIdToData : { [ tickerId ] : SymbolWithTicker ; } = { } ; private static _instance : PriceUpdates ; public static get Instance ( ) : PriceUpdates { return this . _instance || ( this . _instance = new this ( ) ) ; } private constructor ( ) { const that = this ; /**
         * When request to subscribe to market data
         * IBKREVENTS.SUBSCRIBE_PRICE_UPDATES
         */ ibEvents . on ( IBKREVENTS . SUBSCRIBE_PRICE_UPDATES , async ( data ) => { return that . subscribe ( data ) ; } ) ; /**
         * When request to unsubscribe to market data
         * IBKREVENTS.UNSUBSCRIBE_PRICE_UPDATES
         */ ibEvents . on ( IBKREVENTS . UNSUBSCRIBE_PRICE_UPDATES , async ( data ) => { return that . unsubscribe ( data ) ; } ) ; } /**
     * init
     */ public init ( ) { const ib = IBKRConnection . Instance . getIBKR ( ) ; this . ib = ib ; const that = this ; ib . on ( EventName . tickPrice , ( tickerId , tickType , price , _canAutoExecute ) => { const thisSymbol = that . tickerIdToData [ tickerId ] ; const tickTypeWords = IBApiTickType [ tickType ] ; log ( 'PriceUpdates.tickPrice' , `tickerId=${ tickerId }, tickType=${ tickType }/${ tickTypeWords }, price=${ price } ${ JSON . stringify ( { s : thisSymbol . symbol , ticker : thisSymbol . tickerId , } ) }` ) ; const tickTypesAllowed = thisSymbol . tickTypes ; const currentSymbol = thisSymbol && thisSymbol . symbol ; if ( isEmpty ( currentSymbol ) ) { log ( 'PriceUpdates.tickPrice' , `Symbol not found ${ JSON . stringify ( thisSymbol ) }` ) ; return ; } if ( tickTypesAllowed === undefined || tickTypesAllowed . includes ( tickTypeWords as any ) ) { log ( 'PriceUpdates.tickPrice' , `${ tickTypeWords }:PRICE ${ currentSymbol } => $${ price } tickerId = ${ tickerId }` ) ; const dataToPublish = { tickType : tickTypeWords as any , symbol : currentSymbol , price : price === - 1 ? null : price , date : new Date ( ) , } ; publishDataToTopic ( { topic : IBKREVENTS . ON_PRICE_UPDATES , data : dataToPublish , } ) ; } } ) ; } returns public async subscribe ( args ) { const that = this ; const tickTypes = args . opt ?. tickType ? isArray ( args . opt ?. tickType ) ? args . opt . tickType : [ args . opt . tickType ] : undefined ; const contract = await ( async ( ) => { const contractArg = typeof args . contract === 'string' ? new Stock ( args . contract ) : args . contract ; if ( isEmpty ( contractArg ) ) { return undefined ; } log ( 'contract before getting details' , contractArg ) ; if ( 'primaryExch' in contractArg ) { return contractArg ; } else { const contractDetailsParams = convertContractToContractDetailsParams ( contractArg ) ; const contractDetailsList = await getContractDetails ( contractDetailsParams ) ; if ( contractDetailsList . length !== 1 ) { log ( 'contract details are ambiguous, expected exactly one:' , contractDetailsList ) ; } return contractDetailsList [ 0 ] . summary ; } } ) ( ) ; if ( ! contract ) { return undefined ; } log ( 'contract to be queried:' , contract ) ; const includesForexOrOpt = [ 'CASH' , 'OPT' ] . includes ( contract . secType ) ; const symbol = includesForexOrOpt ? contract . localSymbol : contract . symbol ; if ( ! symbol ) { log ( 'PriceUpdates.subscribe' , `Symbol cannot be null` ) ; return undefined ; } if ( ! that . ib ) { that . init ( ) ; } const conId = contract . conId ; const key = conId . toString ( ) ; if ( that . keyToTickerId [ key ] ) { verbose ( 'PriceUpdates.subscribe' , `${ conId }/${ symbol } is already subscribed` ) ; return conId ; } const tickerIdToUse = getRadomReqId ( ) ; that . keyToTickerId [ key ] = tickerIdToUse ; that . tickerIdToData [ tickerIdToUse ] = { tickerId : tickerIdToUse , conId , symbol , tickTypes , } ; setImmediate ( ( ) => { that . ib . reqMktData ( tickerIdToUse , contract , '' , false , false ) ; log ( 'PriceUpdates.subscribe' , `${ conId }/${ symbol } is successfully subscribed` ) ; } ) ; return conId ; } public unsubscribeAllSymbols ( ) { const tickerIdToData = this . tickerIdToData ; this . tickerIdToData = { } ; this . keyToTickerId = { } ; const timeoutHandler = ( ) => { for ( const symbolTicker of Object . values ( tickerIdToData ) ) { log ( `cancelMktData ${ symbolTicker . symbol } tickerId=${ symbolTicker . tickerId }` ) ; this . ib . cancelMktData ( symbolTicker . tickerId ) ; } } ; setTimeout ( timeoutHandler , 1000 ) ; } /**
     * unsubscribe
     */ public unsubscribe ( contract ) { const key = contract . conId . toString ( ) ; const tickerId = this . keyToTickerId [ key ] ; if ( ! tickerId ) { return ; } const timeoutHandler = ( ) => { const symbolTicker = this . tickerIdToData [ tickerId ] ; if ( symbolTicker ) { log ( `cancelMktData ${ symbolTicker . symbol } tickerId=${ symbolTicker . tickerId }` ) ; this . ib . cancelMktData ( symbolTicker . tickerId ) ; } delete this . tickerIdToData [ tickerId ] ; delete this . keyToTickerId [ key ] ; } ; setTimeout ( timeoutHandler , 1000 ) ; } } export default PriceUpdates ;