/* ============================================================

* node-binance-api

* https://github.com/jaggedsoft/node-binance-api

* ============================================================

* Copyright 2017-, Jon Eyrick

* Released under the MIT License

* ============================================================

* @module jaggedsoft/node-binance-api

* @return {object} instance to class object */

let api = function Binance( options = {} ) {

if ( !new.target ) return new api( options ); // Legacy support for calling the constructor without 'new'

let Binance = this; // eslint-disable-line consistent-this

const WebSocket = require( 'ws' );

const request = require( 'request' );

const crypto = require( 'crypto' );

const file = require( 'fs' );

const url = require( 'url' );

const JSONbig = require( 'json-bigint' );

const HttpsProxyAgent = require( 'https-proxy-agent' );

const SocksProxyAgent = require( 'socks-proxy-agent' );

const stringHash = require( 'string-hash' );

const async = require( 'async' );

let base = 'https://api.binance.com/api/';

let wapi = 'https://api.binance.com/wapi/';

let sapi = 'https://api.binance.com/sapi/';

let fapi = 'https://fapi.binance.com/fapi/';

let dapi = 'https://dapi.binance.com/dapi/';

let fapiTest = 'https://testnet.binancefuture.com/fapi/';

let dapiTest = 'https://testnet.binancefuture.com/dapi/';

let fstream = 'wss://fstream.binance.com/stream?streams=';

let fstreamSingle = 'wss://fstream.binance.com/ws/';

let fstreamSingleTest = 'wss://stream.binancefuture.com/ws/';

let fstreamTest = 'wss://stream.binancefuture.com/stream?streams=';

let dstream = 'wss://dstream.binance.com/stream?streams=';

let dstreamSingle = 'wss://dstream.binance.com/ws/';

let dstreamSingleTest = 'wss://dstream.binancefuture.com/ws/';

let dstreamTest = 'wss://dstream.binancefuture.com/stream?streams=';

let stream = 'wss://stream.binance.com:9443/ws/';

let combineStream = 'wss://stream.binance.com:9443/stream?streams=';

const userAgent = 'Mozilla/4.0 (compatible; Node Binance API)';

const contentType = 'application/x-www-form-urlencoded';

Binance.subscriptions = {};

Binance.futuresSubscriptions = {};

Binance.futuresInfo = {};

Binance.futuresMeta = {};

Binance.futuresTicks = {};

Binance.futuresRealtime = {};

Binance.futuresKlineQueue = {};

Binance.deliverySubscriptions = {};

Binance.deliveryInfo = {};

Binance.deliveryMeta = {};

Binance.deliveryTicks = {};

Binance.deliveryRealtime = {};

Binance.deliveryKlineQueue = {};

Binance.depthCache = {};

Binance.depthCacheContext = {};

Binance.ohlcLatest = {};

Binance.klineQueue = {};

Binance.ohlc = {};

const default_options = {

recvWindow: 5000,

useServerTime: false,

reconnect: true,

keepAlive: true,

verbose: false,

test: false,

hedgeMode: false,

localAddress: false,

family: false,

log: function ( ...args ) {

console.log( Array.prototype.slice.call( args ) );

}

};

Binance.options = default_options;

Binance.info = {

usedWeight: 0,

futuresLatency: false,

lastRequest: false,

lastURL: false,

statusCode: 0,

orderCount1s: 0,

orderCount1m: 0,

orderCount1h: 0,

orderCount1d: 0,

timeOffset: 0

};

Binance.socketHeartbeatInterval = null;

if ( options ) setOptions( options );

function setOptions( opt = {}, callback = false ) {

if ( typeof opt === 'string' ) { // Pass json config filename

Binance.options = JSON.parse( file.readFileSync( opt ) );

} else Binance.options = opt;

if ( typeof Binance.options.recvWindow === 'undefined' ) Binance.options.recvWindow = default_options.recvWindow;

if ( typeof Binance.options.useServerTime === 'undefined' ) Binance.options.useServerTime = default_options.useServerTime;

if ( typeof Binance.options.reconnect === 'undefined' ) Binance.options.reconnect = default_options.reconnect;

if ( typeof Binance.options.test === 'undefined' ) Binance.options.test = default_options.test;

if ( typeof Binance.options.hedgeMode === 'undefined' ) Binance.options.hedgeMode = default_options.hedgeMode;

if ( typeof Binance.options.log === 'undefined' ) Binance.options.log = default_options.log;

if ( typeof Binance.options.verbose === 'undefined' ) Binance.options.verbose = default_options.verbose;

if ( typeof Binance.options.keepAlive === 'undefined' ) Binance.options.keepAlive = default_options.keepAlive;

if ( typeof Binance.options.localAddress === 'undefined' ) Binance.options.localAddress = default_options.localAddress;

if ( typeof Binance.options.family === 'undefined' ) Binance.options.family = default_options.family;

if ( typeof Binance.options.urls !== 'undefined' ) {

const { urls } = Binance.options;

if ( typeof urls.base === 'string' ) base = urls.base;

if ( typeof urls.wapi === 'string' ) wapi = urls.wapi;

if ( typeof urls.sapi === 'string' ) sapi = urls.sapi;

if ( typeof urls.fapi === 'string' ) fapi = urls.fapi;

if ( typeof urls.fapiTest === 'string' ) fapiTest = urls.fapiTest;

if ( typeof urls.stream === 'string' ) stream = urls.stream;

if ( typeof urls.combineStream === 'string' ) combineStream = urls.combineStream;

if ( typeof urls.fstream === 'string' ) fstream = urls.fstream;

if ( typeof urls.fstreamSingle === 'string' ) fstreamSingle = urls.fstreamSingle;

if ( typeof urls.fstreamTest === 'string' ) fstreamTest = urls.fstreamTest;

if ( typeof urls.fstreamSingleTest === 'string' ) fstreamSingleTest = urls.fstreamSingleTest;

if ( typeof urls.dstream === 'string' ) dstream = urls.dstream;

if ( typeof urls.dstreamSingle === 'string' ) dstreamSingle = urls.dstreamSingle;

if ( typeof urls.dstreamTest === 'string' ) dstreamTest = urls.dstreamTest;

if ( typeof urls.dstreamSingleTest === 'string' ) dstreamSingleTest = urls.dstreamSingleTest;

}

if ( Binance.options.useServerTime ) {

publicRequest( base + 'v3/time', {}, function ( error, response ) {

Binance.info.timeOffset = response.serverTime - new Date().getTime();

//Binance.options.log("server time set: ", response.serverTime, Binance.info.timeOffset);

if ( callback ) callback();

} );

} else if ( callback ) callback();

return this;

}

/**

* Replaces socks connection uri hostname with IP address

* @param {string} connString - socks connection string

* @return {string} modified string with ip address

*/

const proxyReplacewithIp = connString => {

return connString;

}

/**

* Returns an array in the form of [host, port]

* @param {string} connString - connection string

* @return {array} array of host and port

*/

const parseProxy = connString => {

let arr = connString.split( '/' );

let host = arr[2].split( ':' )[0];

let port = arr[2].split( ':' )[1];

return [ arr[0], host, port ];

}

/**

* Checks to see of the object is iterable

* @param {object} obj - The object check

* @return {boolean} true or false is iterable

*/

const isIterable = obj => {

if ( obj === null ) return false;

return typeof obj[Symbol.iterator] === 'function';

}

const addProxy = opt => {

if ( Binance.options.proxy ) {

const proxyauth = Binance.options.proxy.auth ? `${ Binance.options.proxy.auth.username }:${ Binance.options.proxy.auth.password }@` : '';

opt.proxy = `http://${ proxyauth }${ Binance.options.proxy.host }:${ Binance.options.proxy.port }`;

}

return opt;

}

const reqHandler = cb => ( error, response, body ) => {

Binance.info.lastRequest = new Date().getTime();

if ( response ) {

Binance.info.statusCode = response.statusCode || 0;

if ( response.request ) Binance.info.lastURL = response.request.uri.href;

if ( response.headers ) {

Binance.info.usedWeight = response.headers['x-mbx-used-weight-1m'] || 0;

Binance.info.orderCount1s = response.headers['x-mbx-order-count-1s'] || 0;

Binance.info.orderCount1m = response.headers['x-mbx-order-count-1m'] || 0;

Binance.info.orderCount1h = response.headers['x-mbx-order-count-1h'] || 0;

Binance.info.orderCount1d = response.headers['x-mbx-order-count-1d'] || 0;

}

}

if ( !cb ) return;

if ( error ) return cb( error, {} );

if ( response && response.statusCode !== 200 ) return cb( response, {} );

return cb( null, JSONbig.parse( body ) );

}

const proxyRequest = ( opt, cb ) => {

const req = request( addProxy( opt ), reqHandler( cb ) ).on('error', (err) => { cb( err, {} ) });

return req;

}

const reqObj = ( url, data = {}, method = 'GET', key ) => ( {

url: url,

qs: data,

method: method,

family: Binance.options.family,

localAddress: Binance.options.localAddress,

timeout: Binance.options.recvWindow,

forever: Binance.options.keepAlive,

headers: {

'User-Agent': userAgent,

'Content-type': contentType,

'X-MBX-APIKEY': key || ''

}

} )

const reqObjPOST = ( url, data = {}, method = 'POST', key ) => ( {

url: url,

form: data,

method: method,

family: Binance.options.family,

localAddress: Binance.options.localAddress,

timeout: Binance.options.recvWindow,

forever: Binance.options.keepAlive,

qsStringifyOptions: {

arrayFormat: 'repeat'

},

headers: {

'User-Agent': userAgent,

'Content-type': contentType,

'X-MBX-APIKEY': key || ''

}

} )

/**

* Create a http request to the public API

* @param {string} url - The http endpoint

* @param {object} data - The data to send

* @param {function} callback - The callback method to call

* @param {string} method - the http method

* @return {undefined}

*/

const publicRequest = ( url, data = {}, callback, method = 'GET' ) => {

let opt = reqObj( url, data, method );

proxyRequest( opt, callback );

};

// XXX: This one works with array (e.g. for dust.transfer)

// XXX: I _guess_ we could use replace this function with the `qs` module

const makeQueryString = q =>

Object.keys( q )

.reduce( ( a, k ) => {

if ( Array.isArray( q[k] ) ) {

q[k].forEach( v => {

a.push( k + "=" + encodeURIComponent( v ) )

} )

} else if ( q[k] !== undefined ) {

a.push( k + "=" + encodeURIComponent( q[k] ) );

}

return a;

}, [] )

.join( "&" );

/**

* Create a http request to the public API

* @param {string} url - The http endpoint

* @param {object} data - The data to send

* @param {function} callback - The callback method to call

* @param {string} method - the http method

* @return {undefined}

*/

const apiRequest = ( url, data = {}, callback, method = 'GET' ) => {

requireApiKey( 'apiRequest' );

let opt = reqObj(

url,

data,

method,

Binance.options.APIKEY

);

proxyRequest( opt, callback );

};

// Check if API key is empty or invalid

const requireApiKey = function( source = 'requireApiKey', fatalError = true ) {

if ( !Binance.options.APIKEY ) {

if ( fatalError ) throw Error( `${ source }: Invalid API Key!` );

return false;

}

return true;

}

// Check if API secret is present

const requireApiSecret = function( source = 'requireApiSecret', fatalError = true ) {

if ( !Binance.options.APIKEY ) {

if ( fatalError ) throw Error( `${ source }: Invalid API Key!` );

return false;

}

if ( !Binance.options.APISECRET ) {

if ( fatalError ) throw Error( `${ source }: Invalid API Secret!` );

return false;

}

return true;

}

/**

* Make market request

* @param {string} url - The http endpoint

* @param {object} data - The data to send

* @param {function} callback - The callback method to call

* @param {string} method - the http method

* @return {undefined}

*/

const marketRequest = ( url, data = {}, callback, method = 'GET' ) => {

requireApiKey( 'marketRequest' );

let query = makeQueryString( data );

let opt = reqObj(

url + ( query ? '?' + query : '' ),

data,

method,

Binance.options.APIKEY

);

proxyRequest( opt, callback );

};

/**

* Create a signed http request

* @param {string} url - The http endpoint

* @param {object} data - The data to send

* @param {function} callback - The callback method to call

* @param {string} method - the http method

* @param {boolean} noDataInSignature - Prevents data from being added to signature

* @return {undefined}

*/

const signedRequest = ( url, data = {}, callback, method = 'GET', noDataInSignature = false ) => {

requireApiSecret( 'signedRequest' );

data.timestamp = new Date().getTime() + Binance.info.timeOffset;

if ( typeof data.recvWindow === 'undefined' ) data.recvWindow = Binance.options.recvWindow;

let query = method === 'POST' && noDataInSignature ? '' : makeQueryString( data );

let signature = crypto.createHmac( 'sha256', Binance.options.APISECRET ).update( query ).digest( 'hex' ); // set the HMAC hash header

if ( method === 'POST' ) {

let opt = reqObjPOST(

url,

data,

method,

Binance.options.APIKEY

);

opt.form.signature = signature;

proxyRequest( opt, callback );

} else {

let opt = reqObj(

url + '?' + query + '&signature=' + signature,

data,

method,

Binance.options.APIKEY

);

proxyRequest( opt, callback );

}

};

/**

* Create a signed spot order

* @param {string} side - BUY or SELL

* @param {string} symbol - The symbol to buy or sell

* @param {string} quantity - The quantity to buy or sell

* @param {string} price - The price per unit to transact each unit at

* @param {object} flags - additional order settings

* @param {function} callback - the callback function

* @return {undefined}

*/

const order = ( side, symbol, quantity, price, flags = {}, callback = false ) => {

let endpoint = flags.type === 'OCO' ? 'v3/order/oco' : 'v3/order';

if ( Binance.options.test ) endpoint += '/test';

let opt = {

symbol: symbol,

side: side,

type: 'LIMIT',

quantity: quantity

};

if ( typeof flags.type !== 'undefined' ) opt.type = flags.type;

if ( opt.type.includes( 'LIMIT' ) ) {

opt.price = price;

if ( opt.type !== 'LIMIT_MAKER' ) {

opt.timeInForce = 'GTC';

}

}

if ( opt.type === 'OCO' ) {

opt.price = price;

opt.stopLimitPrice = flags.stopLimitPrice;

opt.stopLimitTimeInForce = 'GTC';

delete opt.type;

if ( typeof flags.listClientOrderId !== 'undefined' ) opt.listClientOrderId = flags.listClientOrderId;

if ( typeof flags.limitClientOrderId !== 'undefined' ) opt.limitClientOrderId = flags.limitClientOrderId;

if ( typeof flags.stopClientOrderId !== 'undefined' ) opt.stopClientOrderId = flags.stopClientOrderId;

}

if ( typeof flags.timeInForce !== 'undefined' ) opt.timeInForce = flags.timeInForce;

if ( typeof flags.newOrderRespType !== 'undefined' ) opt.newOrderRespType = flags.newOrderRespType;

if ( typeof flags.newClientOrderId !== 'undefined' ) opt.newClientOrderId = flags.newClientOrderId;

/*

* STOP_LOSS

* STOP_LOSS_LIMIT

* TAKE_PROFIT

* TAKE_PROFIT_LIMIT

* LIMIT_MAKER

*/

if ( typeof flags.icebergQty !== 'undefined' ) opt.icebergQty = flags.icebergQty;

if ( typeof flags.stopPrice !== 'undefined' ) {

opt.stopPrice = flags.stopPrice;

if ( opt.type === 'LIMIT' ) throw Error( 'stopPrice: Must set "type" to one of the following: STOP_LOSS, STOP_LOSS_LIMIT, TAKE_PROFIT, TAKE_PROFIT_LIMIT' );

}

signedRequest( base + endpoint, opt, ( error, response ) => {

if ( !response ) {

if ( callback ) callback( error, response );

else Binance.options.log( 'Order() error:', error );

return;

}

if ( typeof response.msg !== 'undefined' && response.msg === 'Filter failure: MIN_NOTIONAL' ) {

Binance.options.log( 'Order quantity too small. See exchangeInfo() for minimum amounts' );

}

if ( callback ) callback( error, response );

else Binance.options.log( side + '(' + symbol + ',' + quantity + ',' + price + ') ', response );

}, 'POST' );

};

/**

* Create a signed margin order

* @param {string} side - BUY or SELL

* @param {string} symbol - The symbol to buy or sell

* @param {string} quantity - The quantity to buy or sell

* @param {string} price - The price per unit to transact each unit at

* @param {object} flags - additional order settings

* @param {function} callback - the callback function

* @return {undefined}

*/

const marginOrder = ( side, symbol, quantity, price, flags = {}, callback = false ) => {

let endpoint = 'v1/margin/order';

if ( Binance.options.test ) endpoint += '/test';

let opt = {

symbol: symbol,

side: side,

type: 'LIMIT',

quantity: quantity

};

if ( typeof flags.type !== 'undefined' ) opt.type = flags.type;

if (typeof flags.isIsolated !== 'undefined') opt.isIsolated = flags.isIsolated;

if ( opt.type.includes( 'LIMIT' ) ) {

opt.price = price;

if ( opt.type !== 'LIMIT_MAKER' ) {

opt.timeInForce = 'GTC';

}

}

if ( typeof flags.timeInForce !== 'undefined' ) opt.timeInForce = flags.timeInForce;

if ( typeof flags.newOrderRespType !== 'undefined' ) opt.newOrderRespType = flags.newOrderRespType;

if ( typeof flags.newClientOrderId !== 'undefined' ) opt.newClientOrderId = flags.newClientOrderId;

if ( typeof flags.sideEffectType !== 'undefined' ) opt.sideEffectType = flags.sideEffectType;

/*

* STOP_LOSS

* STOP_LOSS_LIMIT

* TAKE_PROFIT

* TAKE_PROFIT_LIMIT

*/

if ( typeof flags.icebergQty !== 'undefined' ) opt.icebergQty = flags.icebergQty;

if ( typeof flags.stopPrice !== 'undefined' ) {

opt.stopPrice = flags.stopPrice;

if ( opt.type === 'LIMIT' ) throw Error( 'stopPrice: Must set "type" to one of the following: STOP_LOSS, STOP_LOSS_LIMIT, TAKE_PROFIT, TAKE_PROFIT_LIMIT' );

}

signedRequest( sapi + endpoint, opt, function ( error, response ) {

if ( !response ) {

if ( callback ) callback( error, response );

else Binance.options.log( 'Order() error:', error );

return;

}

if ( typeof response.msg !== 'undefined' && response.msg === 'Filter failure: MIN_NOTIONAL' ) {

Binance.options.log( 'Order quantity too small. See exchangeInfo() for minimum amounts' );

}

if ( callback ) callback( error, response );

else Binance.options.log( side + '(' + symbol + ',' + quantity + ',' + price + ') ', response );

}, 'POST' );

};

// Futures internal functions

const futuresOrder = async ( side, symbol, quantity, price = false, params = {} ) => {

params.symbol = symbol;

params.side = side;

if ( quantity ) params.quantity = quantity;

// if in the binance futures setting Hedged mode is active, positionSide parameter is mandatory

if( typeof params.positionSide === 'undefined' && Binance.options.hedgeMode ){

params.positionSide = side === 'BUY' ? 'LONG' : 'SHORT';

}

// LIMIT STOP MARKET STOP_MARKET TAKE_PROFIT TAKE_PROFIT_MARKET

// reduceOnly stopPrice

if ( price ) {

params.price = price;

if ( typeof params.type === 'undefined' ) params.type = 'LIMIT';

} else {

if ( typeof params.type === 'undefined' ) params.type = 'MARKET';

}

if ( !params.timeInForce && ( params.type.includes( 'LIMIT' ) || params.type === 'STOP' || params.type === 'TAKE_PROFIT' ) ) {

params.timeInForce = 'GTX'; // Post only by default. Use GTC for limit orders.

}

return promiseRequest( 'v1/order', params, { base:fapi, type:'TRADE', method:'POST' } );

};

const deliveryOrder = async ( side, symbol, quantity, price = false, params = {} ) => {

params.symbol = symbol;

params.side = side;

params.quantity = quantity;

// if in the binance futures setting Hedged mode is active, positionSide parameter is mandatory

if( Binance.options.hedgeMode ){

params.positionSide = side === 'BUY' ? 'LONG' : 'SHORT';

}

// LIMIT STOP MARKET STOP_MARKET TAKE_PROFIT TAKE_PROFIT_MARKET

// reduceOnly stopPrice

if ( price ) {

params.price = price;

if ( typeof params.type === 'undefined' ) params.type = 'LIMIT';

} else {

if ( typeof params.type === 'undefined' ) params.type = 'MARKET';

}

if ( !params.timeInForce && ( params.type.includes( 'LIMIT' ) || params.type === 'STOP' || params.type === 'TAKE_PROFIT' ) ) {

params.timeInForce = 'GTX'; // Post only by default. Use GTC for limit orders.

}

return promiseRequest( 'v1/order', params, { base:dapi, type:'TRADE', method:'POST' } );

};

const promiseRequest = async ( url, data = {}, flags = {} ) => {

return new Promise( ( resolve, reject ) => {

let query = '', headers = {

'User-Agent': userAgent,

'Content-type': 'application/x-www-form-urlencoded'

};

if ( typeof flags.method === 'undefined' ) flags.method = 'GET'; // GET POST PUT DELETE

if ( typeof flags.type === 'undefined' ) flags.type = false; // TRADE, SIGNED, MARKET_DATA, USER_DATA, USER_STREAM

else {

if ( typeof data.recvWindow === 'undefined' ) data.recvWindow = Binance.options.recvWindow;

requireApiKey( 'promiseRequest' );

headers['X-MBX-APIKEY'] = Binance.options.APIKEY;

}

let baseURL = typeof flags.base === 'undefined' ? base : flags.base;

if ( Binance.options.test && baseURL === fapi ) baseURL = fapiTest;

if ( Binance.options.test && baseURL === dapi ) baseURL = dapiTest;

let opt = {

headers,

url: baseURL + url,

method: flags.method,

timeout: Binance.options.recvWindow,

followAllRedirects: true

};

if ( flags.type === 'SIGNED' || flags.type === 'TRADE' || flags.type === 'USER_DATA' ) {

if ( !requireApiSecret( 'promiseRequest' ) ) return reject( 'promiseRequest: Invalid API Secret!' );

data.timestamp = new Date().getTime() + Binance.info.timeOffset;

query = makeQueryString( data );

data.signature = crypto.createHmac( 'sha256', Binance.options.APISECRET ).update( query ).digest( 'hex' ); // HMAC hash header

opt.url = `${ baseURL }${ url }?${ query }&signature=${ data.signature }`;

}

opt.qs = data;

/*if ( flags.method === 'POST' ) {

opt.form = data;

} else {

opt.qs = data;

}*/

try {

request( addProxy( opt ), ( error, response, body ) => {

if ( error ) return reject( error );

try {

Binance.info.lastRequest = new Date().getTime();

if ( response ) {

Binance.info.statusCode = response.statusCode || 0;

if ( response.request ) Binance.info.lastURL = response.request.uri.href;

if ( response.headers ) {

Binance.info.usedWeight = response.headers['x-mbx-used-weight-1m'] || 0;

Binance.info.futuresLatency = response.headers['x-response-time'] || 0;

}

}

if ( !error && response.statusCode == 200 ) return resolve( JSONbig.parse( body ) );

if ( typeof response.body !== 'undefined' ) {

return resolve( JSONbig.parse( response.body ) );

}

return reject( response );

} catch ( err ) {

return reject( `promiseRequest error #${ response.statusCode }` );

}

} ).on( 'error', reject );

} catch ( err ) {

return reject( err );

}

} );

};

/**

* No-operation function

* @return {undefined}

*/

const noop = () => { }; // Do nothing.

/**

* Reworked Tuitio's heartbeat code into a shared single interval tick

* @return {undefined}

*/

const socketHeartbeat = () => {

/* Sockets removed from `subscriptions` during a manual terminate()

will no longer be at risk of having functions called on them */

for ( let endpointId in Binance.subscriptions ) {

const ws = Binance.subscriptions[endpointId];

if ( ws.isAlive ) {

ws.isAlive = false;

if ( ws.readyState === WebSocket.OPEN ) ws.ping( noop );

} else {

if ( Binance.options.verbose ) Binance.options.log( 'Terminating inactive/broken WebSocket: ' + ws.endpoint );

if ( ws.readyState === WebSocket.OPEN ) ws.terminate();

}

}

};

/**

* Called when socket is opened, subscriptions are registered for later reference

* @param {function} opened_callback - a callback function

* @return {undefined}

*/

const handleSocketOpen = function ( opened_callback ) {

this.isAlive = true;

if ( Object.keys( Binance.subscriptions ).length === 0 ) {

Binance.socketHeartbeatInterval = setInterval( socketHeartbeat, 30000 );

}

Binance.subscriptions[this.endpoint] = this;

if ( typeof opened_callback === 'function' ) opened_callback( this.endpoint );

};

/**

* Called when socket is closed, subscriptions are de-registered for later reference

* @param {boolean} reconnect - true or false to reconnect the socket

* @param {string} code - code associated with the socket

* @param {string} reason - string with the response

* @return {undefined}

*/

const handleSocketClose = function ( reconnect, code, reason ) {

delete Binance.subscriptions[this.endpoint];

if ( Binance.subscriptions && Object.keys( Binance.subscriptions ).length === 0 ) {

clearInterval( Binance.socketHeartbeatInterval );

}

Binance.options.log( 'WebSocket closed: ' + this.endpoint +

( code ? ' (' + code + ')' : '' ) +

( reason ? ' ' + reason : '' ) );

if ( Binance.options.reconnect && this.reconnect && reconnect ) {

if ( this.endpoint && parseInt( this.endpoint.length, 10 ) === 60 ) Binance.options.log( 'Account data WebSocket reconnecting...' );

else Binance.options.log( 'WebSocket reconnecting: ' + this.endpoint + '...' );

try {

reconnect();

} catch ( error ) {

Binance.options.log( 'WebSocket reconnect error: ' + error.message );

}

}

};

/**

* Called when socket errors

* @param {object} error - error object message

* @return {undefined}

*/

const handleSocketError = function ( error ) {

/* Errors ultimately result in a `close` event.

see: https://github.com/websockets/ws/blob/828194044bf247af852b31c49e2800d557fedeff/lib/websocket.js#L126 */

Binance.options.log( 'WebSocket error: ' + this.endpoint +

( error.code ? ' (' + error.code + ')' : '' ) +

( error.message ? ' ' + error.message : '' ) );

};

/**

* Called on each socket heartbeat

* @return {undefined}

*/

const handleSocketHeartbeat = function () {

this.isAlive = true;

};

/**

* Used to subscribe to a single websocket endpoint

* @param {string} endpoint - endpoint to connect to

* @param {function} callback - the function to call when information is received

* @param {boolean} reconnect - whether to reconnect on disconnect

* @param {object} opened_callback - the function to call when opened

* @return {WebSocket} - websocket reference

*/

const subscribe = function ( endpoint, callback, reconnect = false, opened_callback = false ) {

let httpsproxy = process.env.https_proxy || false;

let socksproxy = process.env.socks_proxy || false;

let ws = false;

if ( socksproxy !== false ) {

socksproxy = proxyReplacewithIp( socksproxy );

if ( Binance.options.verbose ) Binance.options.log( 'using socks proxy server ' + socksproxy );

let agent = new SocksProxyAgent( {

protocol: parseProxy( socksproxy )[0],

host: parseProxy( socksproxy )[1],

port: parseProxy( socksproxy )[2]

} );

ws = new WebSocket( stream + endpoint, { agent: agent } );

} else if ( httpsproxy !== false ) {

let config = url.parse( httpsproxy );

let agent = new HttpsProxyAgent( config );

if ( Binance.options.verbose ) Binance.options.log( 'using proxy server ' + agent );

ws = new WebSocket( stream + endpoint, { agent: agent } );

} else {

ws = new WebSocket( stream + endpoint );

}

if ( Binance.options.verbose ) Binance.options.log( 'Subscribed to ' + endpoint );

ws.reconnect = Binance.options.reconnect;

ws.endpoint = endpoint;

ws.isAlive = false;

ws.on( 'open', handleSocketOpen.bind( ws, opened_callback ) );

ws.on( 'pong', handleSocketHeartbeat );

ws.on( 'error', handleSocketError );

ws.on( 'close', handleSocketClose.bind( ws, reconnect ) );

ws.on( 'message', data => {

try {

callback( JSON.parse( data ) );

} catch ( error ) {

Binance.options.log( 'Parse error: ' + error.message );

}

} );

return ws;

};

/**

* Used to subscribe to a combined websocket endpoint

* @param {string} streams - streams to connect to

* @param {function} callback - the function to call when information is received

* @param {boolean} reconnect - whether to reconnect on disconnect

* @param {object} opened_callback - the function to call when opened

* @return {WebSocket} - websocket reference

*/

const subscribeCombined = function ( streams, callback, reconnect = false, opened_callback = false ) {

let httpsproxy = process.env.https_proxy || false;

let socksproxy = process.env.socks_proxy || false;

const queryParams = streams.join( '/' );

let ws = false;

if ( socksproxy !== false ) {

socksproxy = proxyReplacewithIp( socksproxy );

if ( Binance.options.verbose ) Binance.options.log( 'using socks proxy server ' + socksproxy );

let agent = new SocksProxyAgent( {

protocol: parseProxy( socksproxy )[0],

host: parseProxy( socksproxy )[1],

port: parseProxy( socksproxy )[2]

} );

ws = new WebSocket( combineStream + queryParams, { agent: agent } );

} else if ( httpsproxy !== false ) {

if ( Binance.options.verbose ) Binance.options.log( 'using proxy server ' + httpsproxy );

let config = url.parse( httpsproxy );

let agent = new HttpsProxyAgent( config );

ws = new WebSocket( combineStream + queryParams, { agent: agent } );

} else {

ws = new WebSocket( combineStream + queryParams );

}

ws.reconnect = Binance.options.reconnect;

ws.endpoint = stringHash( queryParams );

ws.isAlive = false;

if ( Binance.options.verbose ) {

Binance.options.log( 'CombinedStream: Subscribed to [' + ws.endpoint + '] ' + queryParams );

}

ws.on( 'open', handleSocketOpen.bind( ws, opened_callback ) );

ws.on( 'pong', handleSocketHeartbeat );

ws.on( 'error', handleSocketError );

ws.on( 'close', handleSocketClose.bind( ws, reconnect ) );

ws.on( 'message', data => {

try {

callback( JSON.parse( data ).data );

} catch ( error ) {

Binance.options.log( 'CombinedStream: Parse error: ' + error.message );

}

} );

return ws;

};

/**

* Used to terminate a web socket

* @param {string} endpoint - endpoint identifier associated with the web socket

* @param {boolean} reconnect - auto reconnect after termination

* @return {undefined}

*/

const terminate = function ( endpoint, reconnect = false ) {

let ws = Binance.subscriptions[endpoint];

if ( !ws ) return;

ws.removeAllListeners( 'message' );

ws.reconnect = reconnect;

ws.terminate();

}

/**

* Futures heartbeat code with a shared single interval tick

* @return {undefined}

*/

const futuresSocketHeartbeat = () => {

/* Sockets removed from subscriptions during a manual terminate()

will no longer be at risk of having functions called on them */

for ( let endpointId in Binance.futuresSubscriptions ) {

const ws = Binance.futuresSubscriptions[endpointId];

if ( ws.isAlive ) {

ws.isAlive = false;

if ( ws.readyState === WebSocket.OPEN ) ws.ping( noop );

} else {

if ( Binance.options.verbose ) Binance.options.log( `Terminating zombie futures WebSocket: ${ ws.endpoint }` );

if ( ws.readyState === WebSocket.OPEN ) ws.terminate();

}

}

};

/**

* Called when a futures socket is opened, subscriptions are registered for later reference

* @param {function} openCallback - a callback function

* @return {undefined}

*/

const handleFuturesSocketOpen = function ( openCallback ) {

this.isAlive = true;

if ( Object.keys( Binance.futuresSubscriptions ).length === 0 ) {

Binance.socketHeartbeatInterval = setInterval( futuresSocketHeartbeat, 30000 );

}

Binance.futuresSubscriptions[this.endpoint] = this;

if ( typeof openCallback === 'function' ) openCallback( this.endpoint );

};

/**

* Called when futures websocket is closed, subscriptions are de-registered for later reference

* @param {boolean} reconnect - true or false to reconnect the socket

* @param {string} code - code associated with the socket

* @param {string} reason - string with the response

* @return {undefined}

*/

const handleFuturesSocketClose = function ( reconnect, code, reason ) {

delete Binance.futuresSubscriptions[this.endpoint];

if ( Binance.futuresSubscriptions && Object.keys( Binance.futuresSubscriptions ).length === 0 ) {

clearInterval( Binance.socketHeartbeatInterval );

}

Binance.options.log( 'Futures WebSocket closed: ' + this.endpoint +

( code ? ' (' + code + ')' : '' ) +

( reason ? ' ' + reason : '' ) );

if ( Binance.options.reconnect && this.reconnect && reconnect ) {

if ( this.endpoint && parseInt( this.endpoint.length, 10 ) === 60 ) Binance.options.log( 'Futures account data WebSocket reconnecting...' );

else Binance.options.log( 'Futures WebSocket reconnecting: ' + this.endpoint + '...' );

try {

reconnect();

} catch ( error ) {

Binance.options.log( 'Futures WebSocket reconnect error: ' + error.message );

}

}

};

/**

* Called when a futures websocket errors

* @param {object} error - error object message

* @return {undefined}

*/

const handleFuturesSocketError = function ( error ) {

Binance.options.log( 'Futures WebSocket error: ' + this.endpoint +

( error.code ? ' (' + error.code + ')' : '' ) +

( error.message ? ' ' + error.message : '' ) );

};

/**

* Called on each futures socket heartbeat

* @return {undefined}

*/

const handleFuturesSocketHeartbeat = function () {

this.isAlive = true;

};

/**

* Used to subscribe to a single futures websocket endpoint

* @param {string} endpoint - endpoint to connect to

* @param {function} callback - the function to call when information is received

* @param {object} params - Optional reconnect {boolean} (whether to reconnect on disconnect), openCallback {function}, id {string}

* @return {WebSocket} - websocket reference

*/

const futuresSubscribeSingle = function ( endpoint, callback, params = {} ) {

if ( typeof params === 'boolean' ) params = { reconnect: params };

if ( !params.reconnect ) params.reconnect = false;

if ( !params.openCallback ) params.openCallback = false;

if ( !params.id ) params.id = false;

let httpsproxy = process.env.https_proxy || false;

let socksproxy = process.env.socks_proxy || false;

let ws = false;

if ( socksproxy !== false ) {

socksproxy = proxyReplacewithIp( socksproxy );

if ( Binance.options.verbose ) Binance.options.log( `futuresSubscribeSingle: using socks proxy server: ${ socksproxy }` );

let agent = new SocksProxyAgent( {

protocol: parseProxy( socksproxy )[0],

host: parseProxy( socksproxy )[1],

port: parseProxy( socksproxy )[2]

} );

ws = new WebSocket( ( Binance.options.test ? fstreamSingleTest : fstreamSingle ) + endpoint, { agent } );

} else if ( httpsproxy !== false ) {

if ( Binance.options.verbose ) Binance.options.log( `futuresSubscribeSingle: using proxy server: ${ agent }` );

let config = url.parse( httpsproxy );

let agent = new HttpsProxyAgent( config );

ws = new WebSocket( ( Binance.options.test ? fstreamSingleTest : fstreamSingle ) + endpoint, { agent } );

} else {

ws = new WebSocket( ( Binance.options.test ? fstreamSingleTest : fstreamSingle ) + endpoint );

}

if ( Binance.options.verbose ) Binance.options.log( 'futuresSubscribeSingle: Subscribed to ' + endpoint );

ws.reconnect = Binance.options.reconnect;

ws.endpoint = endpoint;

ws.isAlive = false;

ws.on( 'open', handleFuturesSocketOpen.bind( ws, params.openCallback ) );

ws.on( 'pong', handleFuturesSocketHeartbeat );

ws.on( 'error', handleFuturesSocketError );

ws.on( 'close', handleFuturesSocketClose.bind( ws, params.reconnect ) );

ws.on( 'message', data => {

try {

callback( JSON.parse( data ) );

} catch ( error ) {

Binance.options.log( 'Parse error: ' + error.message );

}

} );

return ws;

};

/**

* Used to subscribe to a combined futures websocket endpoint

* @param {string} streams - streams to connect to

* @param {function} callback - the function to call when information is received

* @param {object} params - Optional reconnect {boolean} (whether to reconnect on disconnect), openCallback {function}, id {string}

* @return {WebSocket} - websocket reference

*/

const futuresSubscribe = function ( streams, callback, params = {} ) {

if ( typeof streams === 'string' ) return futuresSubscribeSingle( streams, callback, params );

if ( typeof params === 'boolean' ) params = { reconnect: params };

if ( !params.reconnect ) params.reconnect = false;

if ( !params.openCallback ) params.openCallback = false;

if ( !params.id ) params.id = false;

let httpsproxy = process.env.https_proxy || false;

let socksproxy = process.env.socks_proxy || false;

const queryParams = streams.join( '/' );

let ws = false;

if ( socksproxy !== false ) {

socksproxy = proxyReplacewithIp( socksproxy );

if ( Binance.options.verbose ) Binance.options.log( `futuresSubscribe: using socks proxy server ${ socksproxy }` );

let agent = new SocksProxyAgent( {

protocol: parseProxy( socksproxy )[0],

host: parseProxy( socksproxy )[1],

port: parseProxy( socksproxy )[2]

} );

ws = new WebSocket( ( Binance.options.test ? fstreamTest : fstream ) + queryParams, { agent } );

} else if ( httpsproxy !== false ) {

if ( Binance.options.verbose ) Binance.options.log( `futuresSubscribe: using proxy server ${ httpsproxy }` );

let config = url.parse( httpsproxy );

let agent = new HttpsProxyAgent( config );

ws = new WebSocket( ( Binance.options.test ? fstreamTest : fstream ) + queryParams, { agent } );

} else {

ws = new WebSocket( ( Binance.options.test ? fstreamTest : fstream ) + queryParams );

}

ws.reconnect = Binance.options.reconnect;

ws.endpoint = stringHash( queryParams );

ws.isAlive = false;

if ( Binance.options.verbose ) {

Binance.options.log( `futuresSubscribe: Subscribed to [${ ws.endpoint }] ${ queryParams }` );

}

ws.on( 'open', handleFuturesSocketOpen.bind( ws, params.openCallback ) );

ws.on( 'pong', handleFuturesSocketHeartbeat );

ws.on( 'error', handleFuturesSocketError );

ws.on( 'close', handleFuturesSocketClose.bind( ws, params.reconnect ) );

ws.on( 'message', data => {

try {

callback( JSON.parse( data ).data );

} catch ( error ) {

Binance.options.log( `futuresSubscribe: Parse error: ${ error.message }` );

}

} );

return ws;

};

/**

* Used to terminate a futures websocket

* @param {string} endpoint - endpoint identifier associated with the web socket

* @param {boolean} reconnect - auto reconnect after termination

* @return {undefined}

*/

const futuresTerminate = function ( endpoint, reconnect = false ) {

let ws = Binance.futuresSubscriptions[endpoint];

if ( !ws ) return;

ws.removeAllListeners( 'message' );

ws.reconnect = reconnect;

ws.terminate();

}

/**

* Combines all futures OHLC data with the latest update

* @param {string} symbol - the symbol

* @param {string} interval - time interval

* @return {array} - interval data for given symbol

*/

const futuresKlineConcat = ( symbol, interval ) => {

let output = Binance.futuresTicks[symbol][interval];

if ( typeof Binance.futuresRealtime[symbol][interval].time === 'undefined' ) return output;

const time = Binance.futuresRealtime[symbol][interval].time;

const last_updated = Object.keys( Binance.futuresTicks[symbol][interval] ).pop();

if ( time >= last_updated ) {

output[time] = Binance.futuresRealtime[symbol][interval];

//delete output[time].time;

output[last_updated].isFinal = true;

output[time].isFinal = false;

}

return output;

};