tardis-dev - npm Package Compare versions

Comparing version 7.1.1 to 7.2.0

dist/apikeyaccessinfo.d.ts

 import { Exchange } from './types';
 export declare function getApiKeyAccessInfo(apiKey?: string): Promise<{
-    exchange: "bitmex" | "binance" | "binance-futures" | "deribit" | "bitstamp" | "coinbase" | "cryptofacilities" | "kraken" | "bitfinex" | "bitfinex-derivatives" | "okex" | "bitflyer" | "ftx" | "gemini" | "binance-us" | "binance-jersey" | "binance-dex";
+    exchange: "bitmex" | "deribit" | "binance" | "binance-futures" | "ftx" | "okex" | "huobi" | "huobi-dm" | "bitflyer" | "bitstamp" | "coinbase" | "cryptofacilities" | "gemini" | "kraken" | "bitfinex" | "bitfinex-derivatives" | "binance-dex" | "binance-jersey" | "binance-us" | "huobi-us";
     from: string;
@@ -5,0 +5,0 @@ to: string;

dist/apikeyaccessinfo.js

@@ -12,3 +12,3 @@ "use strict";
     const apiKeyAccessInfo = await got_1.default
-        .get(`${options.endpoint}/v1/api-key-info`, {
+        .get(`${options.endpoint}/api-key-info`, {
         headers: {
@@ -15,0 +15,0 @@ Authorization: `Bearer ${apiKeyToCheck}`

dist/combine.js

@@ -41,2 +41,3 @@ "use strict";
         // based on https://github.com/fraxken/combine-async-iterators
+        console.warn('Important! using combine for real-time streams may cause memory leaks due to Node.js issue with Promise.race handling - https://github.com/nodejs/node/issues/29385');
         while (true) {
@@ -43,0 +44,0 @@ // this does not handle iterators that are finite,

dist/consts.d.ts

@@ -1,2 +0,2 @@
-export declare const EXCHANGES: readonly ["bitmex", "binance", "binance-futures", "deribit", "bitstamp", "coinbase", "cryptofacilities", "kraken", "bitfinex", "bitfinex-derivatives", "okex", "bitflyer", "ftx", "gemini", "binance-us", "binance-jersey", "binance-dex"];
+export declare const EXCHANGES: readonly ["bitmex", "deribit", "binance", "binance-futures", "ftx", "okex", "huobi", "huobi-dm", "bitflyer", "bitstamp", "coinbase", "cryptofacilities", "gemini", "kraken", "bitfinex", "bitfinex-derivatives", "binance-dex", "binance-jersey", "binance-us", "huobi-us"];
 export declare const EXCHANGE_CHANNELS_INFO: {
@@ -20,3 +20,6 @@ bitmex: readonly ["trade", "orderBookL2", "liquidation", "connected", "announcement", "chat", "publicNotifications", "instrument", "settlement", "funding", "insurance", "orderBookL2_25", "quote", "quoteBin1m", "quoteBin5m", "quoteBin1h", "quoteBin1d", "tradeBin1m", "tradeBin5m", "tradeBin1h", "tradeBin1d"];
     'bitfinex-derivatives': readonly ["trades", "book", "status"];
+    huobi: readonly ["depth", "detail", "trade", "bbo"];
+    'huobi-dm': readonly ["depth", "detail", "trade"];
+    'huobi-us': readonly ["depth", "detail", "trade"];
 };
 //# sourceMappingURL=consts.d.ts.map

dist/consts.js

@@ -5,18 +5,21 @@ "use strict";
     'bitmex',
+    'deribit',
     'binance',
     'binance-futures',
-    'deribit',
+    'ftx',
+    'okex',
+    'huobi',
+    'huobi-dm',
+    'bitflyer',
     'bitstamp',
     'coinbase',
     'cryptofacilities',
+    'gemini',
     'kraken',
     'bitfinex',
     'bitfinex-derivatives',
-    'okex',
-    'bitflyer',
-    'ftx',
-    'gemini',
+    'binance-dex',
+    'binance-jersey',
     'binance-us',
-    'binance-jersey',
-    'binance-dex'
+    'huobi-us'
 ];
@@ -99,2 +102,5 @@ const BINANCE_CHANNELS = ['trade', 'ticker', 'depth', 'miniTicker', 'depthSnapshot', 'bookTicker'];
 const BITFINEX_DERIV_CHANNELS = ['trades', 'book', 'status'];
+const HUOBI_CHANNELS = ['depth', 'detail', 'trade', 'bbo'];
+const HUOBI_US_CHANNELS = ['depth', 'detail', 'trade'];
+const HUOBI_DM_CHANNELS = ['depth', 'detail', 'trade'];
 exports.EXCHANGE_CHANNELS_INFO = {
@@ -117,4 +123,7 @@ bitmex: BITMEX_CHANNELS,
     'binance-futures': BINANCE_FUTURES_CHANNELS,
-    'bitfinex-derivatives': BITFINEX_DERIV_CHANNELS
+    'bitfinex-derivatives': BITFINEX_DERIV_CHANNELS,
+    huobi: HUOBI_CHANNELS,
+    'huobi-dm': HUOBI_DM_CHANNELS,
+    'huobi-us': HUOBI_US_CHANNELS
 };
 //# sourceMappingURL=consts.js.map

dist/exchangedetails.js

@@ -10,3 +10,3 @@ "use strict";
     const options = options_1.getOptions();
-    const exchangeDetails = await got_1.default.get(`${options.endpoint}/v1/exchanges/${exchange}`).json();
+    const exchangeDetails = await got_1.default.get(`${options.endpoint}/exchanges/${exchange}`).json();
     return exchangeDetails;
@@ -13,0 +13,0 @@ }

dist/handy.d.ts

@@ -20,4 +20,4 @@ import { Mapper } from './mappers';
     message: any;
-} | undefined>, createMappers: () => Mapper<any, any>[], symbols: string[] | undefined, withDisconnectMessages: boolean | undefined): AsyncGenerator<any, void, unknown>;
+} | undefined>, mappers: Mapper<any, any>[], createMappers: (localTimestamp: Date) => Mapper<any, any>[], withDisconnectMessages: boolean | undefined, filter?: (symbol: string) => boolean): AsyncGenerator<any, void, unknown>;
 export declare function getFilters<T extends Exchange>(mappers: Mapper<T, any>[], symbols?: string[]): FilterForExchange[T][];
 //# sourceMappingURL=handy.d.ts.map

dist/handy.js

@@ -73,5 +73,5 @@ "use strict";
 exports.take = take;
-async function* normalizeMessages(exchange, messages, createMappers, symbols, withDisconnectMessages) {
+async function* normalizeMessages(exchange, messages, mappers, createMappers, withDisconnectMessages, filter) {
     let previousLocalTimestamp;
-    let mappersForExchange = createMappers();
+    let mappersForExchange = mappers;
     if (mappersForExchange.length === 0) {
@@ -84,3 +84,3 @@ throw new Error(`Can't normalize data without any normalizers provided`);
             // lets create new mappers with clean state for 'new connection'
-            mappersForExchange = createMappers();
+            mappersForExchange = undefined;
             // if flag withDisconnectMessages is set, yield disconnect message
@@ -97,2 +97,5 @@ if (withDisconnectMessages === true && previousLocalTimestamp !== undefined) {
         }
+        if (mappersForExchange === undefined) {
+            mappersForExchange = createMappers(messageWithTimestamp.localTimestamp);
+        }
         previousLocalTimestamp = messageWithTimestamp.localTimestamp;
@@ -106,5 +109,8 @@ for (const mapper of mappersForExchange) {
                 for (const message of mappedMessages) {
-                    if (symbolsInclude(symbols, message.symbol)) {
+                    if (filter === undefined) {
                         yield message;
                     }
+                    else if (filter(message.symbol)) {
+                        yield message;
+                    }
                 }
@@ -140,5 +146,2 @@ }
 exports.getFilters = getFilters;
-function symbolsInclude(symbols, symbol) {
-    return symbols === undefined || symbols.length === 0 || symbols.includes(symbol);
-}
 //# sourceMappingURL=handy.js.map

dist/mappers/binance.js

@@ -13,5 +13,3 @@ "use strict";
     getFilters(symbols) {
-        if (symbols !== undefined) {
-            symbols = symbols.map(s => s.toLocaleLowerCase());
-        }
+        symbols = lowerCaseSymbols(symbols);
         return [
@@ -50,5 +48,3 @@ {
     getFilters(symbols) {
-        if (symbols !== undefined) {
-            symbols = symbols.map(s => s.toLocaleLowerCase());
-        }
+        symbols = lowerCaseSymbols(symbols);
         return [
@@ -159,5 +155,3 @@ {
     getFilters(symbols) {
-        if (symbols !== undefined) {
-            symbols = symbols.map(s => s.toLocaleLowerCase());
-        }
+        symbols = lowerCaseSymbols(symbols);
         return [
@@ -229,5 +223,3 @@ {
     getFilters(symbols) {
-        if (symbols !== undefined) {
-            symbols = symbols.map(s => s.toLocaleLowerCase());
-        }
+        symbols = lowerCaseSymbols(symbols);
         return [
@@ -259,2 +251,8 @@ {
 exports.BinanceFuturesDerivativeTickerMapper = BinanceFuturesDerivativeTickerMapper;
+function lowerCaseSymbols(symbols) {
+    if (symbols !== undefined) {
+        return symbols.map(s => s.toLocaleLowerCase());
+    }
+    return;
+}
 //# sourceMappingURL=binance.js.map

dist/mappers/bitstamp.js

@@ -12,5 +12,3 @@ "use strict";
     getFilters(symbols) {
-        if (symbols !== undefined) {
-            symbols = symbols.map(s => s.toLocaleLowerCase());
-        }
+        symbols = lowerCaseSymbols(symbols);
         return [
@@ -50,5 +48,3 @@ {
     getFilters(symbols) {
-        if (symbols !== undefined) {
-            symbols = symbols.map(s => s.toLocaleLowerCase());
-        }
+        symbols = lowerCaseSymbols(symbols);
         return [
@@ -131,2 +127,8 @@ {
 exports.BitstampBookChangeMapper = BitstampBookChangeMapper;
+function lowerCaseSymbols(symbols) {
+    if (symbols !== undefined) {
+        return symbols.map(s => s.toLocaleLowerCase());
+    }
+    return;
+}
 //# sourceMappingURL=bitstamp.js.map

dist/mappers/index.d.ts

 import { BookChange, DerivativeTicker, Trade } from '../types';
 import { Mapper } from './mapper';
 export * from './mapper';
-export declare const normalizeTrades: <T extends "bitmex" | "binance" | "binance-futures" | "deribit" | "bitstamp" | "coinbase" | "cryptofacilities" | "kraken" | "bitfinex" | "bitfinex-derivatives" | "okex" | "bitflyer" | "ftx" | "gemini" | "binance-us" | "binance-jersey" | "binance-dex">(exchange: T) => Mapper<T, Trade>;
-export declare const normalizeBookChanges: <T extends "bitmex" | "binance" | "binance-futures" | "deribit" | "bitstamp" | "coinbase" | "cryptofacilities" | "kraken" | "bitfinex" | "bitfinex-derivatives" | "okex" | "bitflyer" | "ftx" | "gemini" | "binance-us" | "binance-jersey" | "binance-dex">(exchange: T) => Mapper<T, BookChange>;
-export declare const normalizeDerivativeTickers: <T extends "bitmex" | "binance-futures" | "deribit" | "cryptofacilities" | "bitfinex-derivatives" | "okex">(exchange: T) => Mapper<T, DerivativeTicker>;
+export declare const normalizeTrades: <T extends "bitmex" | "deribit" | "binance" | "binance-futures" | "ftx" | "okex" | "huobi" | "huobi-dm" | "bitflyer" | "bitstamp" | "coinbase" | "cryptofacilities" | "gemini" | "kraken" | "bitfinex" | "bitfinex-derivatives" | "binance-dex" | "binance-jersey" | "binance-us" | "huobi-us">(exchange: T, _localTimestamp: Date) => Mapper<T, Trade>;
+export declare const normalizeBookChanges: <T extends "bitmex" | "deribit" | "binance" | "binance-futures" | "ftx" | "okex" | "huobi" | "huobi-dm" | "bitflyer" | "bitstamp" | "coinbase" | "cryptofacilities" | "gemini" | "kraken" | "bitfinex" | "bitfinex-derivatives" | "binance-dex" | "binance-jersey" | "binance-us" | "huobi-us">(exchange: T, _localTimestamp: Date) => Mapper<T, BookChange>;
+export declare const normalizeDerivativeTickers: <T extends "bitmex" | "deribit" | "binance-futures" | "okex" | "cryptofacilities" | "bitfinex-derivatives">(exchange: T, _localTimestamp: Date) => Mapper<T, DerivativeTicker>;
 //# sourceMappingURL=index.d.ts.map

dist/mappers/index.js

@@ -19,2 +19,3 @@ "use strict";
 const okex_1 = require("./okex");
+const huobi_1 = require("./huobi");
 __export(require("./mapper"));
@@ -38,3 +39,6 @@ const tradesMappers = {
     kraken: () => kraken_1.krakenTradesMapper,
-    okex: () => okex_1.okexTradesMapper
+    okex: () => okex_1.okexTradesMapper,
+    huobi: () => new huobi_1.HuobiTradesMapper('huobi'),
+    'huobi-dm': () => new huobi_1.HuobiTradesMapper('huobi-dm'),
+    'huobi-us': () => new huobi_1.HuobiTradesMapper('huobi-us')
 };
@@ -58,3 +62,6 @@ const bookChangeMappers = {
     kraken: () => kraken_1.krakenBookChangeMapper,
-    okex: () => okex_1.okexBookChangeMapper
+    okex: () => okex_1.okexBookChangeMapper,
+    huobi: () => new huobi_1.HuobiBookChangeMapper('huobi'),
+    'huobi-dm': () => new huobi_1.HuobiBookChangeMapper('huobi-dm'),
+    'huobi-us': () => new huobi_1.HuobiBookChangeMapper('huobi-us')
 };
@@ -69,3 +76,3 @@ const derivativeTickersMappers = {
 };
-exports.normalizeTrades = (exchange) => {
+exports.normalizeTrades = (exchange, _localTimestamp) => {
     const createTradesMapper = tradesMappers[exchange];
@@ -77,3 +84,3 @@ if (createTradesMapper === undefined) {
 };
-exports.normalizeBookChanges = (exchange) => {
+exports.normalizeBookChanges = (exchange, _localTimestamp) => {
     const createBookChangesMapper = bookChangeMappers[exchange];
@@ -85,3 +92,3 @@ if (createBookChangesMapper === undefined) {
 };
-exports.normalizeDerivativeTickers = (exchange) => {
+exports.normalizeDerivativeTickers = (exchange, _localTimestamp) => {
     const createDerivativeTickerMapper = derivativeTickersMappers[exchange];
@@ -88,0 +95,0 @@ if (createDerivativeTickerMapper === undefined) {

dist/mappers/mapper.d.ts

@@ -7,3 +7,3 @@ import { DerivativeTicker, Exchange, FilterForExchange, NormalizedData } from '../types';
 };
-export declare type MapperFactory<T extends Exchange, U extends NormalizedData> = (exchange: T) => Mapper<T, U>;
+export declare type MapperFactory<T extends Exchange, U extends NormalizedData> = (exchange: T, localTimestamp: Date) => Mapper<T, U>;
 export declare class PendingTickerInfoHelper {
@@ -10,0 +10,0 @@ private readonly _pendingTickers;

dist/options.js

@@ -10,3 +10,3 @@ "use strict";
 const defaultOptions = {
-    endpoint: 'https://tardis.dev/api',
+    endpoint: 'https://tardis.dev/api/v1',
     cacheDir: path_1.default.join(os_1.default.tmpdir(), '.tardis-cache'),
@@ -13,0 +13,0 @@ apiKey: ''

dist/realtimefeeds/binance.d.ts

 import { Filter } from '../types';
 import { RealTimeFeedBase } from './realtimefeed';
-export declare class BinanceRealTimeFeed extends RealTimeFeedBase {
-    protected wssURL: string;
-    protected httpURL: string;
+declare abstract class BinanceRealTimeFeedBase extends RealTimeFeedBase {
+    protected abstract wssURL: string;
+    protected abstract httpURL: string;
     protected bookUpdateSpeed: string;
     protected mapToSubscribeMessages(filters: Filter<string>[]): string | any[];
     protected messageIsError(message: any): boolean;
-    protected provideManualSnapshots: (filters: Filter<string>[], snapshotsBuffer: any[], shouldCancel: () => boolean) => Promise<void>;
+    protected provideManualSnapshots(filters: Filter<string>[], snapshotsBuffer: any[], shouldCancel: () => boolean): Promise<void>;
 }
-export declare class BinanceJerseyRealTimeFeed extends BinanceRealTimeFeed {
+export declare class BinanceRealTimeFeed extends BinanceRealTimeFeedBase {
     protected wssURL: string;
     protected httpURL: string;
 }
-export declare class BinanceUSRealTimeFeed extends BinanceRealTimeFeed {
+export declare class BinanceJerseyRealTimeFeed extends BinanceRealTimeFeedBase {
     protected wssURL: string;
     protected httpURL: string;
 }
-export declare class BinanceFuturesRealTimeFeed extends BinanceRealTimeFeed {
+export declare class BinanceUSRealTimeFeed extends BinanceRealTimeFeedBase {
     protected wssURL: string;
     protected httpURL: string;
+}
+export declare class BinanceFuturesRealTimeFeed extends BinanceRealTimeFeedBase {
+    protected wssURL: string;
+    protected httpURL: string;
     protected bookUpdateSpeed: string;
 }
+export {};
 //# sourceMappingURL=binance.d.ts.map

dist/realtimefeeds/binance.js

@@ -8,27 +8,6 @@ "use strict";
 const realtimefeed_1 = require("./realtimefeed");
-class BinanceRealTimeFeed extends realtimefeed_1.RealTimeFeedBase {
+class BinanceRealTimeFeedBase extends realtimefeed_1.RealTimeFeedBase {
     constructor() {
         super(...arguments);
-        this.wssURL = 'wss://stream.binance.com:9443';
-        this.httpURL = 'https://api.binance.com/api/v1';
         this.bookUpdateSpeed = '@100ms';
-        this.provideManualSnapshots = async (filters, snapshotsBuffer, shouldCancel) => {
-            const depthSnapshotFilter = filters.find(f => f.channel === 'depthSnapshot');
-            if (!depthSnapshotFilter) {
-                return;
-            }
-            for (let symbol of depthSnapshotFilter.symbols) {
-                if (shouldCancel()) {
-                    return;
-                }
-                this.debug('requesting manual snapshot for: %s', symbol);
-                const depthSnapshotResponse = await got_1.default.get(`${this.httpURL}/depth?symbol=${symbol.toUpperCase()}&limit=1000`).json();
-                const snapshot = {
-                    stream: `${symbol}@depthSnapshot`,
-                    generated: true,
-                    data: depthSnapshotResponse
-                };
-                snapshotsBuffer.push(snapshot);
-            }
-        };
     }
@@ -57,5 +36,32 @@ mapToSubscribeMessages(filters) {
     }
+    async provideManualSnapshots(filters, snapshotsBuffer, shouldCancel) {
+        const depthSnapshotFilter = filters.find(f => f.channel === 'depthSnapshot');
+        if (!depthSnapshotFilter) {
+            return;
+        }
+        for (let symbol of depthSnapshotFilter.symbols) {
+            if (shouldCancel()) {
+                return;
+            }
+            this.debug('requesting manual snapshot for: %s', symbol);
+            const depthSnapshotResponse = await got_1.default.get(`${this.httpURL}/depth?symbol=${symbol.toUpperCase()}&limit=1000`).json();
+            const snapshot = {
+                stream: `${symbol}@depthSnapshot`,
+                generated: true,
+                data: depthSnapshotResponse
+            };
+            this.debug('requested manual snapshot for: %s successfully', symbol);
+            snapshotsBuffer.push(snapshot);
+        }
+    }
 }
+class BinanceRealTimeFeed extends BinanceRealTimeFeedBase {
+    constructor() {
+        super(...arguments);
+        this.wssURL = 'wss://stream.binance.com:9443';
+        this.httpURL = 'https://api.binance.com/api/v1';
+    }
+}
 exports.BinanceRealTimeFeed = BinanceRealTimeFeed;
-class BinanceJerseyRealTimeFeed extends BinanceRealTimeFeed {
+class BinanceJerseyRealTimeFeed extends BinanceRealTimeFeedBase {
     constructor() {
@@ -68,3 +74,3 @@ super(...arguments);
 exports.BinanceJerseyRealTimeFeed = BinanceJerseyRealTimeFeed;
-class BinanceUSRealTimeFeed extends BinanceRealTimeFeed {
+class BinanceUSRealTimeFeed extends BinanceRealTimeFeedBase {
     constructor() {
@@ -77,3 +83,3 @@ super(...arguments);
 exports.BinanceUSRealTimeFeed = BinanceUSRealTimeFeed;
-class BinanceFuturesRealTimeFeed extends BinanceRealTimeFeed {
+class BinanceFuturesRealTimeFeed extends BinanceRealTimeFeedBase {
     constructor() {
@@ -80,0 +86,0 @@ super(...arguments);

dist/realtimefeeds/binancedex.d.ts

@@ -12,4 +12,4 @@ import { Filter } from '../types';
     protected messageIsError(message: any): boolean;
-    protected provideManualSnapshots: (filters: Filter<string>[], snapshotsBuffer: any[], shouldCancel: () => boolean) => Promise<void>;
+    protected provideManualSnapshots(filters: Filter<string>[], snapshotsBuffer: any[], shouldCancel: () => boolean): Promise<void>;
 }
 //# sourceMappingURL=binancedex.d.ts.map

dist/realtimefeeds/binancedex.js

@@ -13,24 +13,2 @@ "use strict";
         this.httpURL = 'https://dex.binance.org/api/v1';
-        this.provideManualSnapshots = async (filters, snapshotsBuffer, shouldCancel) => {
-            const depthSnapshotFilter = filters.find(f => f.channel === 'depthSnapshot');
-            if (!depthSnapshotFilter) {
-                return;
-            }
-            for (let symbol of depthSnapshotFilter.symbols) {
-                if (shouldCancel()) {
-                    return;
-                }
-                this.debug('requesting manual snapshot for: %s', symbol);
-                const depthSnapshotResponse = await got_1.default.get(`${this.httpURL}/depth?symbol=${symbol}&limit=1000`).json();
-                const snapshot = {
-                    stream: `depthSnapshot`,
-                    generated: true,
-                    data: {
-                        symbol,
-                        ...depthSnapshotResponse
-                    }
-                };
-                snapshotsBuffer.push(snapshot);
-            }
-        };
     }
@@ -57,4 +35,27 @@ mapToSubscribeMessages(filters) {
     }
+    async provideManualSnapshots(filters, snapshotsBuffer, shouldCancel) {
+        const depthSnapshotFilter = filters.find(f => f.channel === 'depthSnapshot');
+        if (!depthSnapshotFilter) {
+            return;
+        }
+        for (let symbol of depthSnapshotFilter.symbols) {
+            if (shouldCancel()) {
+                return;
+            }
+            this.debug('requesting manual snapshot for: %s', symbol);
+            const depthSnapshotResponse = await got_1.default.get(`${this.httpURL}/depth?symbol=${symbol}&limit=1000`).json();
+            const snapshot = {
+                stream: `depthSnapshot`,
+                generated: true,
+                data: {
+                    symbol,
+                    ...depthSnapshotResponse
+                }
+            };
+            this.debug('requested manual snapshot for: %s successfully', symbol);
+            snapshotsBuffer.push(snapshot);
+        }
+    }
 }
 exports.BinanceDexRealTimeFeed = BinanceDexRealTimeFeed;
 //# sourceMappingURL=binancedex.js.map

dist/realtimefeeds/bitstamp.d.ts

@@ -8,4 +8,4 @@ import { Filter } from '../types';
     protected messageIsError(message: any): boolean;
-    protected provideManualSnapshots: (filters: Filter<string>[], snapshotsBuffer: any[], shouldCancel: () => boolean) => Promise<void>;
+    protected provideManualSnapshots(filters: Filter<string>[], snapshotsBuffer: any[], shouldCancel: () => boolean): Promise<void>;
 }
 //# sourceMappingURL=bitstamp.d.ts.map

dist/realtimefeeds/bitstamp.js

@@ -13,23 +13,2 @@ "use strict";
         this.httpURL = 'https://www.bitstamp.net/api/v2';
-        this.provideManualSnapshots = async (filters, snapshotsBuffer, shouldCancel) => {
-            // does not work currently on node v12 due to https://github.com/nodejs/node/issues/27711
-            const orderBookFilter = filters.find(f => f.channel === 'diff_order_book');
-            if (!orderBookFilter) {
-                return;
-            }
-            for (let symbol of orderBookFilter.symbols) {
-                if (shouldCancel()) {
-                    return;
-                }
-                this.debug('requesting manual snapshot for: %s', symbol);
-                const depthSnapshotResponse = await got_1.default.get(`${this.httpURL}/order_book/${symbol}?group=1`).json();
-                const snapshot = {
-                    data: depthSnapshotResponse,
-                    event: 'snapshot',
-                    channel: `diff_order_book_${symbol}`,
-                    generated: true
-                };
-                snapshotsBuffer.push(snapshot);
-            }
-        };
     }
@@ -56,4 +35,29 @@ mapToSubscribeMessages(filters) {
     }
+    async provideManualSnapshots(filters, snapshotsBuffer, shouldCancel) {
+        const orderBookFilter = filters.find(f => f.channel === 'diff_order_book');
+        if (!orderBookFilter) {
+            return;
+        }
+        // does not work currently on node v12 due to https://github.com/nodejs/node/issues/27711
+        console.warn(`Due to Node 12 updated http parser and not spec compliant headers being returned by Bitstamp,
+       book snapshots do not work currently for Bitstamp real-time stream.
+       As a workaround try running node with -http-parser=legacy flag`);
+        for (let symbol of orderBookFilter.symbols) {
+            if (shouldCancel()) {
+                return;
+            }
+            this.debug('requesting manual snapshot for: %s', symbol);
+            const depthSnapshotResponse = await got_1.default.get(`${this.httpURL}/order_book/${symbol}?group=1`).json();
+            const snapshot = {
+                data: depthSnapshotResponse,
+                event: 'snapshot',
+                channel: `diff_order_book_${symbol}`,
+                generated: true
+            };
+            this.debug('requested manual snapshot for: %s successfully', symbol);
+            snapshotsBuffer.push(snapshot);
+        }
+    }
 }
 exports.BitstampRealTimeFeed = BitstampRealTimeFeed;
 //# sourceMappingURL=bitstamp.js.map

dist/realtimefeeds/cryptofacilities.js

@@ -7,3 +7,3 @@ "use strict";
         super(...arguments);
-        this.wssURL = 'wss://api.cryptofacilities.com/ws/v1';
+        this.wssURL = 'wss://www.cryptofacilities.com/ws/v1';
     }
@@ -10,0 +10,0 @@ mapToSubscribeMessages(filters) {

dist/realtimefeeds/deribit.d.ts

 import { Filter, FilterForExchange } from '../types';
 import { RealTimeFeedBase } from './realtimefeed';
+import WebSocket from 'ws';
 export declare class DeribitRealTimeDataFeed extends RealTimeFeedBase {
@@ -8,3 +9,6 @@ protected wssURL: string;
     protected messageIsError(message: any): boolean;
+    protected onConnected(ws: WebSocket): void;
+    protected messageIsHeartbeat(msg: any): boolean;
+    protected onMessage(msg: any, ws: WebSocket): void;
 }
 //# sourceMappingURL=deribit.d.ts.map

dist/realtimefeeds/deribit.js

@@ -36,4 +36,30 @@ "use strict";
     }
+    onConnected(ws) {
+        // set heartbeat so deribit won't close connection prematurely
+        // https://docs.deribit.com/v2/#public-set_heartbeat
+        ws.send(JSON.stringify({
+            jsonrpc: '2.0',
+            method: 'public/set_heartbeat',
+            id: 0,
+            params: {
+                interval: 10
+            }
+        }));
+    }
+    messageIsHeartbeat(msg) {
+        return msg.method === 'heartbeat';
+    }
+    onMessage(msg, ws) {
+        // respond with public/test message to keep connection alive
+        if (msg.params !== undefined && msg.params.type === 'test_request') {
+            ws.send(JSON.stringify({
+                jsonrpc: '2.0',
+                method: 'public/test',
+                id: 0,
+                params: {}
+            }));
+        }
+    }
 }
 exports.DeribitRealTimeDataFeed = DeribitRealTimeDataFeed;
 //# sourceMappingURL=deribit.js.map

dist/realtimefeeds/index.js

@@ -19,2 +19,3 @@ "use strict";
 const okex_1 = require("./okex");
+const huobi_1 = require("./huobi");
 __export(require("./realtimefeed"));
@@ -38,3 +39,6 @@ const realTimeFeedsMap = {
     kraken: () => new kraken_1.KrakenRealTimeFeed('kraken'),
-    okex: () => new okex_1.OkexRealTimeFeed('okex')
+    okex: () => new okex_1.OkexRealTimeFeed('okex'),
+    'huobi-dm': () => new huobi_1.HuobiDMRealTimeFeed('huobi-dm'),
+    'huobi-us': () => new huobi_1.HuobiUSRealTimeFeed('huobi-us'),
+    huobi: () => new huobi_1.HuobiRealTimeFeed('huobi')
 };
@@ -41,0 +45,0 @@ function getRealTimeFeedFactory(exchange) {

dist/realtimefeeds/okex.d.ts

@@ -5,3 +5,3 @@ import { Filter } from '../types';
     protected wssURL: string;
-    protected messagesNeedDecompression: boolean;
+    protected decompress: (message: any) => any;
     protected mapToSubscribeMessages(filters: Filter<string>[]): string | any[];
@@ -8,0 +8,0 @@ protected messageIsError(message: any): boolean;

dist/realtimefeeds/okex.js

 "use strict";
 Object.defineProperty(exports, "__esModule", { value: true });
+const zlib_1 = require("zlib");
 const realtimefeed_1 = require("./realtimefeed");
+const pongBuffer = Buffer.from('pong');
 class OkexRealTimeFeed extends realtimefeed_1.RealTimeFeedBase {
@@ -8,3 +10,9 @@ constructor() {
         this.wssURL = 'wss://real.okex.com:8443/ws/v3';
-        this.messagesNeedDecompression = true;
+        this.decompress = (message) => {
+            message = zlib_1.inflateRawSync(message);
+            if (message.equals(pongBuffer)) {
+                return;
+            }
+            return message;
+        };
     }
@@ -11,0 +19,0 @@ mapToSubscribeMessages(filters) {

dist/realtimefeeds/realtimefeed.d.ts

@@ -16,9 +16,12 @@ import dbg from 'debug';
     stream(filters: Filter<string>[]): AsyncGenerator<any, void, unknown>;
+    private _onConnectionOpen;
     protected abstract readonly wssURL: string;
     protected abstract mapToSubscribeMessages(filters: Filter<string>[]): string | any[];
     protected abstract messageIsError(message: any): boolean;
-    protected provideManualSnapshots?: (filters: Filter<string>[], snapshotsBuffer: any[], shouldCancel: () => boolean) => void;
-    protected onMessage?: (msg: any, ws: WebSocket) => void;
-    protected messagesNeedDecompression: boolean;
+    protected messageIsHeartbeat(_msg: any): boolean;
+    protected provideManualSnapshots(_filters: Filter<string>[], _snapshotsBuffer: any[], _shouldCancel: () => boolean): Promise<void>;
+    protected onMessage(_msg: any, _ws: WebSocket): void;
+    protected onConnected(_ws: WebSocket): void;
+    protected decompress?: (msg: any) => any;
 }
 //# sourceMappingURL=realtimefeed.d.ts.map

dist/realtimefeeds/realtimefeed.js

@@ -7,12 +7,7 @@ "use strict";
 const debug_1 = __importDefault(require("debug"));
-const util_1 = require("util");
 const ws_1 = __importDefault(require("ws"));
-const zlib_1 = __importDefault(require("zlib"));
 const handy_1 = require("../handy");
-const inflateRaw = util_1.promisify(zlib_1.default.inflateRaw);
-const pongBuffer = Buffer.from('pong');
 class RealTimeFeedBase {
     constructor(exchange) {
         this.exchange = exchange;
-        this.messagesNeedDecompression = false;
         this.debug = debug_1.default(`tardis-dev:realtime:${exchange}`);
@@ -26,8 +21,7 @@ }
         this.debug('starting streaming: %o filters, subscribe messages: %o', filters, subscribeMessages);
-        const subscribeViaURL = typeof subscribeMessages === 'string';
         let retries = 0;
         while (true) {
-            let timerid;
+            let staleConnectionCheckTID;
             try {
-                const address = subscribeViaURL ? `${this.wssURL}${subscribeMessages}` : this.wssURL;
+                const address = typeof subscribeMessages === 'string' ? `${this.wssURL}${subscribeMessages}` : this.wssURL;
                 this.debug('estabilishing connection to %s', address);
@@ -37,24 +31,17 @@ const ws = new ws_1.default(address, { perMessageDeflate: false });
                 let receivedMessagesCount = 0;
-                ws.once('open', async () => {
-                    this.debug('estabilished connection to %s', address);
-                    if (!subscribeViaURL) {
-                        for (const message of subscribeMessages) {
-                            this.debug('subscribing to %o', message);
-                            ws.send(JSON.stringify(message));
-                        }
-                    }
-                    if (this.provideManualSnapshots !== undefined) {
-                        await handy_1.wait(handy_1.ONE_SEC_IN_MS);
-                        this.provideManualSnapshots(filters, snapshotsToReturn, () => ws.readyState === ws_1.default.CLOSED);
-                    }
+                ws.onopen = this._onConnectionOpen({
+                    address,
+                    subscribeMessages,
+                    snapshotsToReturn,
+                    filters
                 });
                 if (this.timeoutIntervalMS !== undefined) {
                     // set up timer that checks against open, but stale connections that do not return any data
-                    timerid = setInterval(() => {
+                    staleConnectionCheckTID = setInterval(() => {
                         if (receivedMessagesCount === 0) {
                             this.debug('did not received any messages within %d ms timeout, restarting...', this.timeoutIntervalMS);
                             ws.terminate();
-                            if (timerid !== undefined) {
-                                clearInterval(timerid);
-                                timerid = undefined;
+                            if (staleConnectionCheckTID !== undefined) {
+                                clearInterval(staleConnectionCheckTID);
+                                staleConnectionCheckTID = undefined;
                             }
@@ -71,6 +58,5 @@ }
                 for await (let message of realtimeMessagesStream) {
-                    receivedMessagesCount++;
-                    if (this.messagesNeedDecompression) {
-                        message = (await inflateRaw(message));
-                        if (message.equals(pongBuffer)) {
+                    if (this.decompress !== undefined) {
+                        message = this.decompress(message);
+                        if (message === undefined) {
                             continue;
@@ -83,5 +69,8 @@ }
                     }
-                    if (this.onMessage !== undefined) {
-                        this.onMessage(messageDeserialized, ws);
+                    // exclude heaartbeat messages from  received messages counter
+                    // connection could still be stale even if only heartbeats are provided without any data
+                    if (this.messageIsHeartbeat(messageDeserialized) === false) {
+                        receivedMessagesCount++;
                     }
+                    this.onMessage(messageDeserialized, ws);
                     yield messageDeserialized;
@@ -96,3 +85,3 @@ if (retries > 0) {
                         }
-                        snapshotsToReturn = [];
+                        snapshotsToReturn.length = 0;
                     }
@@ -114,5 +103,5 @@ }
             finally {
-                if (timerid !== undefined) {
-                    clearInterval(timerid);
-                    timerid = undefined;
+                if (staleConnectionCheckTID !== undefined) {
+                    clearInterval(staleConnectionCheckTID);
+                    staleConnectionCheckTID = undefined;
                 }
@@ -122,4 +111,29 @@ }
     }
+    _onConnectionOpen({ address, filters, snapshotsToReturn, subscribeMessages }) {
+        return async ({ target }) => {
+            this.debug('estabilished connection to %s', address);
+            if (Array.isArray(subscribeMessages)) {
+                for (const message of subscribeMessages) {
+                    this.debug('subscribing to %o', message);
+                    target.send(JSON.stringify(message));
+                }
+            }
+            this.onConnected(target);
+            try {
+                await handy_1.wait(2 * handy_1.ONE_SEC_IN_MS);
+                await this.provideManualSnapshots(filters, snapshotsToReturn, () => target.readyState === ws_1.default.CLOSED);
+            }
+            catch (e) {
+                this.debug('providing manual snapshots error: %o, closing connection...', e);
+            }
+        };
+    }
+    messageIsHeartbeat(_msg) {
+        return false;
+    }
+    async provideManualSnapshots(_filters, _snapshotsBuffer, _shouldCancel) { }
+    onMessage(_msg, _ws) { }
+    onConnected(_ws) { }
 }
 exports.RealTimeFeedBase = RealTimeFeedBase;
 //# sourceMappingURL=realtimefeed.js.map

dist/replay.js

@@ -127,5 +127,7 @@ "use strict";
     }
-    const createMappers = () => normalizers.map(m => m(exchange));
+    const fromDate = handy_1.parseAsUTCDate(from);
+    const createMappers = (localTimestamp) => normalizers.map(m => m(exchange, localTimestamp));
     const nonFilterableExchanges = ['bitfinex', 'bitfinex-derivatives'];
-    const filters = nonFilterableExchanges.includes(exchange) ? [] : handy_1.getFilters(createMappers(), symbols);
+    const mappers = createMappers(fromDate);
+    const filters = nonFilterableExchanges.includes(exchange) ? [] : handy_1.getFilters(mappers, symbols);
     const messages = replay({
@@ -139,3 +141,8 @@ exchange,
     });
-    return handy_1.normalizeMessages(exchange, messages, createMappers, symbols, withDisconnectMessages);
+    // filter normalized messages by symbol as some exchanges do not provide server side filtering so we could end up with messages
+    // for symbols we've not requested for
+    const filter = (symbol) => {
+        return symbols === undefined || symbols.length === 0 || symbols.includes(symbol);
+    };
+    return handy_1.normalizeMessages(exchange, messages, mappers, createMappers, withDisconnectMessages, filter);
 }
@@ -142,0 +149,0 @@ exports.replayNormalized = replayNormalized;

dist/stream.js

@@ -31,4 +31,5 @@ "use strict";
     }
-    const createMappers = () => normalizers.map(m => m(exchange));
-    const filters = handy_1.getFilters(createMappers(), symbols);
+    const createMappers = (localTimestamp) => normalizers.map(m => m(exchange, localTimestamp));
+    const mappers = createMappers(new Date());
+    const filters = handy_1.getFilters(mappers, symbols);
     const messages = stream({
@@ -40,3 +41,3 @@ exchange,
     });
-    return handy_1.normalizeMessages(exchange, messages, createMappers, symbols, withDisconnectMessages);
+    return handy_1.normalizeMessages(exchange, messages, mappers, createMappers, withDisconnectMessages);
 }
@@ -43,0 +44,0 @@ exports.streamNormalized = streamNormalized;

dist/worker.js

@@ -99,3 +99,3 @@ "use strict";
 async function reliablyFetchAndCacheSlice({ exchange, fromDate, endpoint, apiKey }, offset, filters, sliceCachePath) {
-    let url = `${endpoint}/v1/data-feeds/${exchange}?from=${fromDate.toISOString()}&offset=${offset}`;
+    let url = `${endpoint}/data-feeds/${exchange}?from=${fromDate.toISOString()}&offset=${offset}`;
     if (filters.length > 0) {
@@ -102,0 +102,0 @@ url += `&filters=${encodeURIComponent(JSON.stringify(filters))}`;

package.json

 {
   "name": "tardis-dev",
-  "version": "7.1.1",
+  "version": "7.2.0",
   "engines": {
@@ -5,0 +5,0 @@ "node": ">=12"

README.md

@@ -56,3 +56,3 @@ # tardis-dev
 
-- support for top cryptocurrency exchanges: BitMEX, Binance, Binance Futures, Deribit, Bitfinex, bitFlyer, Bitstamp, Coinbase Pro, Crypto Facilities, Gemini, FTX, Kraken and OKEx.
+- support for top cryptocurrency exchanges: BitMEX, Deribit, Binance, Binance Futures, FTX, OKEx, Huobi Global, Huobi DM, bitFlyer, Bitstamp, Coinbase Pro, Crypto Facilities, Gemini, Kraken, Bitfinex and Huobi US.
 
@@ -85,3 +85,3 @@ <br/>
 
-- built-in TypeScript support
+- [built-in TypeScript support](https://docs.tardis.dev/api/node-js#usage-with-typescript)
 
@@ -88,0 +88,0 @@ <br/>

src/apikeyaccessinfo.ts

@@ -10,3 +10,3 @@ import got from 'got'
   const apiKeyAccessInfo = await got
-    .get(`${options.endpoint}/v1/api-key-info`, {
+    .get(`${options.endpoint}/api-key-info`, {
       headers: {
@@ -13,0 +13,0 @@ Authorization: `Bearer ${apiKeyToCheck}`

src/combine.ts

@@ -56,2 +56,5 @@ import { ONE_SEC_IN_MS } from './handy'
     // based on https://github.com/fraxken/combine-async-iterators
+    console.warn(
+      'Important! using combine for real-time streams may cause memory leaks due to Node.js issue with Promise.race handling - https://github.com/nodejs/node/issues/29385'
+    )
 
@@ -58,0 +61,0 @@ while (true) {

src/consts.ts

 export const EXCHANGES = [
   'bitmex',
+  'deribit',
   'binance',
   'binance-futures',
-  'deribit',
+  'ftx',
+  'okex',
+  'huobi',
+  'huobi-dm',
+  'bitflyer',
   'bitstamp',
   'coinbase',
   'cryptofacilities',
+  'gemini',
   'kraken',
   'bitfinex',
   'bitfinex-derivatives',
-  'okex',
-  'bitflyer',
-  'ftx',
-  'gemini',
+  'binance-dex',
+  'binance-jersey',
   'binance-us',
-  'binance-jersey',
-  'binance-dex'
+  'huobi-us'
 ] as const
@@ -112,2 +115,8 @@
 
+const HUOBI_CHANNELS = ['depth', 'detail', 'trade', 'bbo'] as const
+
+const HUOBI_US_CHANNELS = ['depth', 'detail', 'trade'] as const
+
+const HUOBI_DM_CHANNELS = ['depth', 'detail', 'trade'] as const
+
 export const EXCHANGE_CHANNELS_INFO = {
@@ -130,3 +139,6 @@ bitmex: BITMEX_CHANNELS,
   'binance-futures': BINANCE_FUTURES_CHANNELS,
-  'bitfinex-derivatives': BITFINEX_DERIV_CHANNELS
+  'bitfinex-derivatives': BITFINEX_DERIV_CHANNELS,
+  huobi: HUOBI_CHANNELS,
+  'huobi-dm': HUOBI_DM_CHANNELS,
+  'huobi-us': HUOBI_US_CHANNELS
 }

src/exchangedetails.ts

@@ -7,3 +7,3 @@ import got from 'got'
   const options = getOptions()
-  const exchangeDetails = await got.get(`${options.endpoint}/v1/exchanges/${exchange}`).json()
+  const exchangeDetails = await got.get(`${options.endpoint}/exchanges/${exchange}`).json()
 
@@ -10,0 +10,0 @@ return exchangeDetails as ExchangeDetails<T>

src/handy.ts

@@ -79,9 +79,9 @@ import { createHash } from 'crypto'
   messages: AsyncIterableIterator<{ localTimestamp: Date; message: any } | undefined>,
-  createMappers: () => Mapper<any, any>[],
-  symbols: string[] | undefined,
-  withDisconnectMessages: boolean | undefined
+  mappers: Mapper<any, any>[],
+  createMappers: (localTimestamp: Date) => Mapper<any, any>[],
+  withDisconnectMessages: boolean | undefined,
+  filter?: (symbol: string) => boolean
 ) {
   let previousLocalTimestamp: Date | undefined
-  let mappersForExchange = createMappers()
-
+  let mappersForExchange: Mapper<any, any>[] | undefined = mappers
   if (mappersForExchange.length === 0) {
@@ -95,3 +95,3 @@ throw new Error(`Can't normalize data without any normalizers provided`)
       // lets create new mappers with clean state for 'new connection'
-      mappersForExchange = createMappers()
+      mappersForExchange = undefined
 
@@ -111,2 +111,6 @@ // if flag withDisconnectMessages is set, yield disconnect message
 
+    if (mappersForExchange === undefined) {
+      mappersForExchange = createMappers(messageWithTimestamp.localTimestamp)
+    }
+
     previousLocalTimestamp = messageWithTimestamp.localTimestamp
@@ -122,4 +126,6 @@
         for (const message of mappedMessages) {
-          if (symbolsInclude(symbols, message.symbol)) {
+          if (filter === undefined) {
             yield message
+          } else if (filter(message.symbol)) {
+            yield message
           }
@@ -159,5 +165,1 @@ }
 }
-
-function symbolsInclude(symbols: string[] | undefined, symbol: string) {
-  return symbols === undefined || symbols.length === 0 || symbols.includes(symbol)
-}

src/mappers/binance.ts

@@ -14,5 +14,3 @@ import { BookChange, DerivativeTicker, Exchange, FilterForExchange, Trade } from '../types'
   getFilters(symbols?: string[]) {
-    if (symbols !== undefined) {
-      symbols = symbols.map(s => s.toLocaleLowerCase())
-    }
+    symbols = lowerCaseSymbols(symbols)
 
@@ -56,5 +54,3 @@ return [
   getFilters(symbols?: string[]) {
-    if (symbols !== undefined) {
-      symbols = symbols.map(s => s.toLocaleLowerCase())
-    }
+    symbols = lowerCaseSymbols(symbols)
 
@@ -182,5 +178,3 @@ return [
   getFilters(symbols?: string[]) {
-    if (symbols !== undefined) {
-      symbols = symbols.map(s => s.toLocaleLowerCase())
-    }
+    symbols = lowerCaseSymbols(symbols)
 
@@ -263,5 +257,3 @@ return [
   getFilters(symbols?: string[]): FilterForExchange['binance-futures'][] {
-    if (symbols !== undefined) {
-      symbols = symbols.map(s => s.toLocaleLowerCase())
-    }
+    symbols = lowerCaseSymbols(symbols)
 
@@ -300,2 +292,9 @@ return [
 
+function lowerCaseSymbols(symbols?: string[]) {
+  if (symbols !== undefined) {
+    return symbols.map(s => s.toLocaleLowerCase())
+  }
+  return
+}
+
 type BinanceResponse<T> = {
@@ -302,0 +301,0 @@ stream: string

src/mappers/bitstamp.ts

@@ -16,5 +16,3 @@ import { BookChange, Trade } from '../types'
   getFilters(symbols?: string[]) {
-    if (symbols !== undefined) {
-      symbols = symbols.map(s => s.toLocaleLowerCase())
-    }
+    symbols = lowerCaseSymbols(symbols)
 
@@ -58,5 +56,3 @@ return [
   getFilters(symbols?: string[]) {
-    if (symbols !== undefined) {
-      symbols = symbols.map(s => s.toLocaleLowerCase())
-    }
+    symbols = lowerCaseSymbols(symbols)
 
@@ -153,2 +149,9 @@ return [
 
+function lowerCaseSymbols(symbols?: string[]) {
+  if (symbols !== undefined) {
+    return symbols.map(s => s.toLocaleLowerCase())
+  }
+  return
+}
+
 type BitstampTrade = {
@@ -155,0 +158,0 @@ event: 'trade'

src/mappers/index.ts

@@ -22,2 +22,3 @@ import { BookChange, DerivativeTicker, Trade } from '../types'
 import { okexBookChangeMapper, OkexDerivativeTickerMapper, okexTradesMapper } from './okex'
+import { HuobiTradesMapper, HuobiBookChangeMapper } from './huobi'
 
@@ -43,3 +44,6 @@ export * from './mapper'
   kraken: () => krakenTradesMapper,
-  okex: () => okexTradesMapper
+  okex: () => okexTradesMapper,
+  huobi: () => new HuobiTradesMapper('huobi'),
+  'huobi-dm': () => new HuobiTradesMapper('huobi-dm'),
+  'huobi-us': () => new HuobiTradesMapper('huobi-us')
 }
@@ -64,3 +68,6 @@
   kraken: () => krakenBookChangeMapper,
-  okex: () => okexBookChangeMapper
+  okex: () => okexBookChangeMapper,
+  huobi: () => new HuobiBookChangeMapper('huobi'),
+  'huobi-dm': () => new HuobiBookChangeMapper('huobi-dm'),
+  'huobi-us': () => new HuobiBookChangeMapper('huobi-us')
 }
@@ -77,3 +84,3 @@
 
-export const normalizeTrades = <T extends keyof typeof tradesMappers>(exchange: T): Mapper<T, Trade> => {
+export const normalizeTrades = <T extends keyof typeof tradesMappers>(exchange: T, _localTimestamp: Date): Mapper<T, Trade> => {
   const createTradesMapper = tradesMappers[exchange]
@@ -88,3 +95,6 @@
 
-export const normalizeBookChanges = <T extends keyof typeof bookChangeMappers>(exchange: T): Mapper<T, BookChange> => {
+export const normalizeBookChanges = <T extends keyof typeof bookChangeMappers>(
+  exchange: T,
+  _localTimestamp: Date
+): Mapper<T, BookChange> => {
   const createBookChangesMapper = bookChangeMappers[exchange]
@@ -99,3 +109,6 @@
 
-export const normalizeDerivativeTickers = <T extends keyof typeof derivativeTickersMappers>(exchange: T): Mapper<T, DerivativeTicker> => {
+export const normalizeDerivativeTickers = <T extends keyof typeof derivativeTickersMappers>(
+  exchange: T,
+  _localTimestamp: Date
+): Mapper<T, DerivativeTicker> => {
   const createDerivativeTickerMapper = derivativeTickersMappers[exchange]
@@ -102,0 +115,0 @@

src/mappers/mapper.ts

@@ -11,3 +11,3 @@ import { DerivativeTicker, Exchange, FilterForExchange, NormalizedData } from '../types'
 
-export type MapperFactory<T extends Exchange, U extends NormalizedData> = (exchange: T) => Mapper<T, U>
+export type MapperFactory<T extends Exchange, U extends NormalizedData> = (exchange: T, localTimestamp: Date) => Mapper<T, U>
 
@@ -14,0 +14,0 @@ type Writeable<T> = { -readonly [P in keyof T]: T[P] }

src/options.ts

@@ -6,3 +6,3 @@ import os from 'os'
 const defaultOptions: Options = {
-  endpoint: 'https://tardis.dev/api',
+  endpoint: 'https://tardis.dev/api/v1',
   cacheDir: path.join(os.tmpdir(), '.tardis-cache'),
@@ -9,0 +9,0 @@ apiKey: ''

src/realtimefeeds/binance.ts

@@ -5,5 +5,5 @@ import got from 'got'
 
-export class BinanceRealTimeFeed extends RealTimeFeedBase {
-  protected wssURL = 'wss://stream.binance.com:9443'
-  protected httpURL = 'https://api.binance.com/api/v1'
+abstract class BinanceRealTimeFeedBase extends RealTimeFeedBase {
+  protected abstract wssURL: string
+  protected abstract httpURL: string
   protected bookUpdateSpeed = '@100ms'
@@ -38,3 +38,3 @@
 
-  protected provideManualSnapshots = async (filters: Filter<string>[], snapshotsBuffer: any[], shouldCancel: () => boolean) => {
+  protected async provideManualSnapshots(filters: Filter<string>[], snapshotsBuffer: any[], shouldCancel: () => boolean) {
     const depthSnapshotFilter = filters.find(f => f.channel === 'depthSnapshot')
@@ -59,2 +59,3 @@ if (!depthSnapshotFilter) {
       }
+      this.debug('requested manual snapshot for: %s successfully', symbol)
 
@@ -66,3 +67,8 @@ snapshotsBuffer.push(snapshot)
 
-export class BinanceJerseyRealTimeFeed extends BinanceRealTimeFeed {
+export class BinanceRealTimeFeed extends BinanceRealTimeFeedBase {
+  protected wssURL = 'wss://stream.binance.com:9443'
+  protected httpURL = 'https://api.binance.com/api/v1'
+}
+
+export class BinanceJerseyRealTimeFeed extends BinanceRealTimeFeedBase {
   protected wssURL = 'wss://stream.binance.je:9443'
@@ -72,3 +78,3 @@ protected httpURL = 'https://api.binance.je/api/v1'
 
-export class BinanceUSRealTimeFeed extends BinanceRealTimeFeed {
+export class BinanceUSRealTimeFeed extends BinanceRealTimeFeedBase {
   protected wssURL = 'wss://stream.binance.us:9443'
@@ -78,3 +84,3 @@ protected httpURL = 'https://api.binance.us/api/v1'
 
-export class BinanceFuturesRealTimeFeed extends BinanceRealTimeFeed {
+export class BinanceFuturesRealTimeFeed extends BinanceRealTimeFeedBase {
   protected wssURL = 'wss://fstream.binance.com'
@@ -81,0 +87,0 @@ protected httpURL = 'https://fapi.binance.com/fapi/v1'

src/realtimefeeds/binancedex.ts

@@ -33,3 +33,3 @@ import got from 'got'
 
-  protected provideManualSnapshots = async (filters: Filter<string>[], snapshotsBuffer: any[], shouldCancel: () => boolean) => {
+  protected async provideManualSnapshots(filters: Filter<string>[], snapshotsBuffer: any[], shouldCancel: () => boolean) {
     const depthSnapshotFilter = filters.find(f => f.channel === 'depthSnapshot')
@@ -58,2 +58,4 @@ if (!depthSnapshotFilter) {
 
+      this.debug('requested manual snapshot for: %s successfully', symbol)
+
       snapshotsBuffer.push(snapshot)
@@ -60,0 +62,0 @@ }

src/realtimefeeds/bitstamp.ts

@@ -32,5 +32,3 @@ import got from 'got'
 
-  protected provideManualSnapshots = async (filters: Filter<string>[], snapshotsBuffer: any[], shouldCancel: () => boolean) => {
-    // does not work currently on node v12 due to https://github.com/nodejs/node/issues/27711
-
+  protected async provideManualSnapshots(filters: Filter<string>[], snapshotsBuffer: any[], shouldCancel: () => boolean) {
     const orderBookFilter = filters.find(f => f.channel === 'diff_order_book')
@@ -41,2 +39,7 @@ if (!orderBookFilter) {
 
+    // does not work currently on node v12 due to https://github.com/nodejs/node/issues/27711
+    console.warn(`Due to Node 12 updated http parser and not spec compliant headers being returned by Bitstamp,
+       book snapshots do not work currently for Bitstamp real-time stream.
+       As a workaround try running node with -http-parser=legacy flag`)
+
     for (let symbol of orderBookFilter.symbols!) {
@@ -57,2 +60,3 @@ if (shouldCancel()) {
       }
+      this.debug('requested manual snapshot for: %s successfully', symbol)
 
@@ -59,0 +63,0 @@ snapshotsBuffer.push(snapshot)

src/realtimefeeds/cryptofacilities.ts

@@ -5,3 +5,3 @@ import { Filter } from '../types'
 export class CryptofacilitiesRealTimeFeed extends RealTimeFeedBase {
-  protected wssURL = 'wss://api.cryptofacilities.com/ws/v1'
+  protected wssURL = 'wss://www.cryptofacilities.com/ws/v1'
 
@@ -8,0 +8,0 @@ protected mapToSubscribeMessages(filters: Filter<string>[]): string | any[] {

src/realtimefeeds/deribit.ts

 import { Filter, FilterForExchange } from '../types'
 import { RealTimeFeedBase } from './realtimefeed'
+import WebSocket from 'ws'
 
@@ -38,2 +39,35 @@ export class DeribitRealTimeDataFeed extends RealTimeFeedBase {
   }
+
+  protected onConnected(ws: WebSocket) {
+    // set heartbeat so deribit won't close connection prematurely
+    // https://docs.deribit.com/v2/#public-set_heartbeat
+    ws.send(
+      JSON.stringify({
+        jsonrpc: '2.0',
+        method: 'public/set_heartbeat',
+        id: 0,
+        params: {
+          interval: 10
+        }
+      })
+    )
+  }
+
+  protected messageIsHeartbeat(msg: any) {
+    return msg.method === 'heartbeat'
+  }
+
+  protected onMessage(msg: any, ws: WebSocket) {
+    // respond with public/test message to keep connection alive
+    if (msg.params !== undefined && msg.params.type === 'test_request') {
+      ws.send(
+        JSON.stringify({
+          jsonrpc: '2.0',
+          method: 'public/test',
+          id: 0,
+          params: {}
+        })
+      )
+    }
+  }
 }

src/realtimefeeds/index.ts

@@ -16,2 +16,3 @@ import { Exchange } from '../types'
 import { RealTimeFeed } from './realtimefeed'
+import { HuobiRealTimeFeed, HuobiDMRealTimeFeed, HuobiUSRealTimeFeed } from './huobi'
 
@@ -39,3 +40,6 @@ export * from './realtimefeed'
   kraken: () => new KrakenRealTimeFeed('kraken'),
-  okex: () => new OkexRealTimeFeed('okex')
+  okex: () => new OkexRealTimeFeed('okex'),
+  'huobi-dm': () => new HuobiDMRealTimeFeed('huobi-dm'),
+  'huobi-us': () => new HuobiUSRealTimeFeed('huobi-us'),
+  huobi: () => new HuobiRealTimeFeed('huobi')
 }
@@ -42,0 +46,0 @@

src/realtimefeeds/okex.ts

@@ -0,8 +1,19 @@
+import { inflateRawSync } from 'zlib'
 import { Filter } from '../types'
 import { RealTimeFeedBase } from './realtimefeed'
 
+const pongBuffer = Buffer.from('pong')
+
 export class OkexRealTimeFeed extends RealTimeFeedBase {
   protected wssURL = 'wss://real.okex.com:8443/ws/v3'
-  protected messagesNeedDecompression = true
 
+  protected decompress = (message: any) => {
+    message = inflateRawSync(message) as Buffer
+    if (message.equals(pongBuffer)) {
+      return
+    }
+
+    return message
+  }
+
   protected mapToSubscribeMessages(filters: Filter<string>[]): string | any[] {
@@ -9,0 +20,0 @@ const args = filters

src/realtimefeeds/realtimefeed.ts

 import dbg from 'debug'
-import { promisify } from 'util'
 import WebSocket from 'ws'
-import zlib from 'zlib'
 import { ONE_SEC_IN_MS, wait } from '../handy'
 import { Exchange, Filter } from '../types'
-const inflateRaw = promisify(zlib.inflateRaw)
 
-const pongBuffer = Buffer.from('pong')
-
 export type RealTimeFeed = {
@@ -33,9 +28,8 @@ stream(filters: Filter<string>[]): AsyncIterableIterator<object | undefined>
 
-    const subscribeViaURL = typeof subscribeMessages === 'string'
     let retries = 0
 
     while (true) {
-      let timerid: NodeJS.Timeout | undefined
+      let staleConnectionCheckTID: NodeJS.Timeout | undefined
       try {
-        const address = subscribeViaURL ? `${this.wssURL}${subscribeMessages}` : this.wssURL
+        const address = typeof subscribeMessages === 'string' ? `${this.wssURL}${subscribeMessages}` : this.wssURL
         this.debug('estabilishing connection to %s', address)
@@ -47,15 +41,8 @@
         let receivedMessagesCount = 0
-        ws.once('open', async () => {
-          this.debug('estabilished connection to %s', address)
-          if (!subscribeViaURL) {
-            for (const message of subscribeMessages) {
-              this.debug('subscribing to %o', message)
-              ws.send(JSON.stringify(message))
-            }
-          }
 
-          if (this.provideManualSnapshots !== undefined) {
-            await wait(ONE_SEC_IN_MS)
-            this.provideManualSnapshots(filters, snapshotsToReturn, () => ws.readyState === WebSocket.CLOSED)
-          }
+        ws.onopen = this._onConnectionOpen({
+          address,
+          subscribeMessages,
+          snapshotsToReturn,
+          filters
         })
@@ -65,9 +52,9 @@
           // set up timer that checks against open, but stale connections that do not return any data
-          timerid = setInterval(() => {
+          staleConnectionCheckTID = setInterval(() => {
             if (receivedMessagesCount === 0) {
               this.debug('did not received any messages within %d ms timeout, restarting...', this.timeoutIntervalMS)
               ws.terminate()
-              if (timerid !== undefined) {
-                clearInterval(timerid)
-                timerid = undefined
+              if (staleConnectionCheckTID !== undefined) {
+                clearInterval(staleConnectionCheckTID)
+                staleConnectionCheckTID = undefined
               }
@@ -86,7 +73,5 @@ }
         for await (let message of realtimeMessagesStream) {
-          receivedMessagesCount++
-
-          if (this.messagesNeedDecompression) {
-            message = (await inflateRaw(message)) as Buffer
-            if (message.equals(pongBuffer)) {
+          if (this.decompress !== undefined) {
+            message = this.decompress(message)
+            if (message === undefined) {
               continue
@@ -102,6 +87,10 @@ }
 
-          if (this.onMessage !== undefined) {
-            this.onMessage(messageDeserialized, ws)
+          // exclude heaartbeat messages from  received messages counter
+          // connection could still be stale even if only heartbeats are provided without any data
+          if (this.messageIsHeartbeat(messageDeserialized) === false) {
+            receivedMessagesCount++
           }
 
+          this.onMessage(messageDeserialized, ws)
+
           yield messageDeserialized
@@ -118,5 +107,7 @@
             }
-            snapshotsToReturn = []
+
+            snapshotsToReturn.length = 0
           }
         }
+
         this.debug('connection closed, restarting...')
@@ -136,5 +127,5 @@ // websocket connection has been closed notify about it by yielding undefined
       } finally {
-        if (timerid !== undefined) {
-          clearInterval(timerid)
-          timerid = undefined
+        if (staleConnectionCheckTID !== undefined) {
+          clearInterval(staleConnectionCheckTID)
+          staleConnectionCheckTID = undefined
         }
@@ -145,2 +136,34 @@ }
 
+  private _onConnectionOpen({
+    address,
+    filters,
+    snapshotsToReturn,
+    subscribeMessages
+  }: {
+    address: string
+    subscribeMessages: string | any[]
+    filters: Filter<string>[]
+    snapshotsToReturn: any[]
+  }) {
+    return async ({ target }: WebSocket.OpenEvent) => {
+      this.debug('estabilished connection to %s', address)
+
+      if (Array.isArray(subscribeMessages)) {
+        for (const message of subscribeMessages) {
+          this.debug('subscribing to %o', message)
+          target.send(JSON.stringify(message))
+        }
+      }
+
+      this.onConnected(target)
+
+      try {
+        await wait(2 * ONE_SEC_IN_MS)
+        await this.provideManualSnapshots(filters, snapshotsToReturn, () => target.readyState === WebSocket.CLOSED)
+      } catch (e) {
+        this.debug('providing manual snapshots error: %o, closing connection...', e)
+      }
+    }
+  }
+
   protected abstract readonly wssURL: string
@@ -150,5 +173,13 @@ protected abstract mapToSubscribeMessages(filters: Filter<string>[]): string | any[]
 
-  protected provideManualSnapshots?: (filters: Filter<string>[], snapshotsBuffer: any[], shouldCancel: () => boolean) => void
-  protected onMessage?: (msg: any, ws: WebSocket) => void
-  protected messagesNeedDecompression = false
+  protected messageIsHeartbeat(_msg: any) {
+    return false
+  }
+
+  protected async provideManualSnapshots(_filters: Filter<string>[], _snapshotsBuffer: any[], _shouldCancel: () => boolean) {}
+
+  protected onMessage(_msg: any, _ws: WebSocket) {}
+
+  protected onConnected(_ws: WebSocket) {}
+
+  protected decompress?: (msg: any) => any
 }

src/replay.ts

@@ -168,5 +168,7 @@ import { createReadStream } from 'fs-extra'
 
-  const createMappers = () => normalizers.map(m => m(exchange))
+  const fromDate = parseAsUTCDate(from)
+  const createMappers = (localTimestamp: Date) => normalizers.map(m => m(exchange, localTimestamp))
   const nonFilterableExchanges = ['bitfinex', 'bitfinex-derivatives']
-  const filters = nonFilterableExchanges.includes(exchange) ? [] : getFilters(createMappers(), symbols)
+  const mappers = createMappers(fromDate)
+  const filters = nonFilterableExchanges.includes(exchange) ? [] : getFilters(mappers, symbols)
 
@@ -182,3 +184,9 @@ const messages = replay({
 
-  return normalizeMessages(exchange, messages, createMappers, symbols, withDisconnectMessages)
+  // filter normalized messages by symbol as some exchanges do not provide server side filtering so we could end up with messages
+  // for symbols we've not requested for
+  const filter = (symbol: string) => {
+    return symbols === undefined || symbols.length === 0 || symbols.includes(symbol)
+  }
+
+  return normalizeMessages(exchange, messages, mappers, createMappers, withDisconnectMessages, filter)
 }
@@ -185,0 +193,0 @@

src/stream.ts

@@ -50,4 +50,5 @@ import { getFilters, normalizeMessages } from './handy'
 
-  const createMappers = () => normalizers.map(m => m(exchange))
-  const filters = getFilters(createMappers(), symbols)
+  const createMappers = (localTimestamp: Date) => normalizers.map(m => m(exchange, localTimestamp))
+  const mappers = createMappers(new Date())
+  const filters = getFilters(mappers, symbols)
 
@@ -61,3 +62,3 @@ const messages = stream({
 
-  return normalizeMessages(exchange, messages, createMappers, symbols, withDisconnectMessages)
+  return normalizeMessages(exchange, messages, mappers, createMappers, withDisconnectMessages)
 }
@@ -64,0 +65,0 @@

src/worker.ts

@@ -119,3 +119,3 @@ import crypto from 'crypto'
 ) {
-  let url = `${endpoint}/v1/data-feeds/${exchange}?from=${fromDate.toISOString()}&offset=${offset}`
+  let url = `${endpoint}/data-feeds/${exchange}?from=${fromDate.toISOString()}&offset=${offset}`
 
@@ -122,0 +122,0 @@ if (filters.length > 0) {

No alert changes

Improved metrics

Total package byte prevSize

540272

increased by6.32%

Number of package files

259

increased by4.02%

Lines of code

9418

increased by7.16%

No dependency changes