Introduction

Node.js tardis-dev library provides convenient access to tick-level historical and real-time cryptocurrency market data both in exchange native and normalized formats. Instead of callbacks it relies on async iteration (for await ...of) enabling composability features like seamless switching between real-time data streaming and historical data replay or computing derived data locally.

​

const tardis = require('tardis-dev')
const { replayNormalized, normalizeTrades, normalizeBookChanges } = tardis
​
const messages = replayNormalized(
  {
    exchange: 'bitmex',
    symbols: ['XBTUSD', 'ETHUSD'],
    from: '2019-05-01',
    to: '2019-05-02'
  },
  normalizeTrades,
  normalizeBookChanges
)
​
for await (const message of messages) {
  console.log(message)
}

const tardis = require('tardis-dev')
const { streamNormalized, normalizeTrades, normalizeBookChanges } = tardis
​
const messages = streamNormalized(
  {
    exchange: 'bitmex',
    symbols: ['XBTUSD', 'ETHUSD']
  },
  normalizeTrades,
  normalizeBookChanges
)
​
for await (const message of messages) {
  console.log(message)
}

​​​​

​

Features

• ​real-time streaming of tick-level market data with unified API for connecting directly to exchanges public WebSocket APIs without any intermediary/3rd party proxy

• ​historical tick-level market data replay backed by tardis.dev HTTP API — includes full order book depth snapshots plus incremental updates, tick-by-tick trades, historical open interest, funding, index, mark prices, liquidations and more

• support for both exchange native and normalized market data formats (consistent format for accessing market data across multiple exchanges — normalized trades, order book and ticker data)

• ​seamless switching between real-time streaming and historical market data replay thanks to async iterables providing unified way of consuming data messages

• transparent historical local data caching (cached data is stored on disk in compressed GZIP format and decompressed on demand when reading the data)

• 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, Huobi US, Bybit , OKCoin and HitBTC.

• automatic closed connections and stale connections reconnection logic for real-time streams

• ​combining multiple exchanges feeds into single one via combine helper function — synchronized historical market data replay and consolidated real-time data streaming from multiple exchanges

• ​computing derived data locally like order book imbalance, custom trade bars, book snapshots and more via compute helper function and computables, e.g., volume based bars, top 20 levels order book snapshots taken every 10 ms etc.

• ​full limit order book reconstruction both for real-time and historical data via OrderBook object

• fast and lightweight architecture — low memory footprint and no heavy in-memory buffering

• ​extensible mapping logic that allows adjusting normalized formats for specific needs

• built-in TypeScript support

​

​

​

Installation

Requires Node.js v12+ installed.

npm install tardis-dev --save

​

​

​

​

Debugging and logging

tardis-dev lib uses debug package for verbose logging and debugging purposes that can be enabled via DEBUG environment variable set to tardis-dev*.

​

​

​

Usage with TypeScript

Simply change from require

const { replay, stream } = require('tardis-dev')

to ES Modules import

import { replay, stream } from 'tardis-dev'

to enjoy first class TypeScript typings.

​

​

​

Replaying historical market data

​

replay(options)

Replays historical market data messages for given replay options in exchange native format. Historical market data is being fetched efficiently (in parallel) from the tardis.dev HTTP API and cached locally. Returns async iterable.

const { replay } = require('tardis-dev')
​
const messages = replay({
  exchange: 'bitmex',
  filters: [
    { channel: 'trade', symbols: ['XBTUSD'] },
    { channel: 'orderBookL2', symbols: ['XBTUSD'] }
  ],
  from: '2019-05-01',
  to: '2019-05-02'
})
​
for await (const message of messages) {
  console.log(message)
}

​​​​

​

​

replay options

​

type of messages provided by replay iterator (for await ...of)

type Message =
  | {
      localTimestamp: Date // local timestamp when message has been received
      message: any // message in exchange native data format
    }
    // when skipDecoding is set to true
  | {
      localTimestamp: Buffer
      message: Buffer
    }
​
// when withDisconnects is set to true whole message can be undefined (disconnect)
Message | undefined

​

sample message

{
  localTimestamp: 2019-05-01T00:00:07.013Z,
  message: {
    table: 'orderBookL2',
    action: 'update',
    data: [ { symbol: 'XBTUSD', id: 8799474100, side: 'Buy', size: 167340 } ]
  }
}

​

​

​

replayNormalized(options, ...normalizers)

Replays historical market data messages for given replay options and normalizes messages using normalizers provided as rest arguments. Historical market data is being fetched efficiently (in parallel) from the tardis.dev HTTP API and cached locally. Returns async iterable.

const tardis = require('tardis-dev')
const { replayNormalized, normalizeTrades, normalizeBookChanges } = tardis
​
const messages = replayNormalized(
  {
    exchange: 'bitmex',
    symbols: ['XBTUSD', 'ETHUSD'],
    from: '2019-05-01',
    to: '2019-05-02'
  },
  normalizeTrades,
  normalizeBookChanges
)
​
for await (const message of messages) {
  console.log(message)
}

​​​​

​

​

replay normalized options

​

Built-in normalizers

replayNormalized function accepts any number of normalizers as rest parameters that map from native exchange format to normalized data format. tardis-dev ships with built in ones that normalize trades, order book and derivative ticker data but also allows adding custom.

​

types of messages provided by replayNormalized iterator (for await ...of)

Message types and formats depend on specific normalizers provided to replayNormalized function and are documented in detail in data normalization section.

​

sample message

Sample message produced by normalizeBookChanges

{
  type: 'book_change',
  symbol: 'XBTUSD',
  exchange: 'bitmex',
  isSnapshot: false,
  bids: [{ price: 5263.5, amount: 1780043 }],
  asks: [],
  timestamp: 2019-05-01T00:00:04.430Z,
  localTimestamp: 2019-05-01T00:00:04.430Z
}

​

​

​

Streaming real-time market data

stream(options)

Streams real-time market data messages for given stream options in exchange native format. It connects directly to exchanges WebSocket APIs and transparently restarts closed, broken or stale connections (open connections without data being send for specified amount of time). Returns async iterable.

const { stream } = require('tardis-dev')
​
const messages = stream({
  exchange: 'bitmex',
  filters: [
    { channel: 'trade', symbols: ['XBTUSD'] },
    { channel: 'orderBookL2', symbols: ['XBTUSD'] }
  ]
})
​
for await (const message of messages) {
  console.log(message)
}

​​​​

​

​

stream options

​

type of messages provided by stream iterator (for await ...of)

type Message =
  | {
      localTimestamp: Date // local timestamp when message has been received
      message: any // message in exchange native data format
    }
    // when skipDecoding is set to true
  | {
      localTimestamp: Buffer
      message: Buffer
    }
​
// when withDisconnects is set to true whole message can be undefined
Message | undefined

​

sample message

{
  localTimestamp: 2019-10-22T09:24:51.025Z,
  message: {
    table: 'orderBookL2',
    action: 'update',
    data: [
      { symbol: 'XBTUSD', id: 8799172400, side: 'Sell', size: 369501 }
    ]
  }
}

​

​

​

streamNormalized(options, ...normalizers)

Streams real-time market data messages for given stream options and normalizes messages using provided normalizers provided as rest arguments. It connects directly to exchanges WebSocket APIs and transparently restarts closed, broken or stale connections (open connections without data being send for specified amount of time). Returns async iterable.

const tardis = require('tardis-dev')
const { streamNormalized, normalizeTrades, normalizeBookChanges } = tardis
​
const messages = streamNormalized(
  {
    exchange: 'bitmex',
    symbols: ['XBTUSD', 'ETHUSD']
  },
  normalizeTrades,
  normalizeBookChanges
)
​
for await (const message of messages) {
  console.log(message)
}

​​​​

​

​

stream normalized options

​

Built-in normalizers

streamNormalized function can accept any number of custom normalizers as rest parameters that map from native exchange format to normalized data format. tardis-dev ships with built in ones that normalize trades, order book and derivative ticker data but also allows adding custom ones.

​

types of messages provided by streamNormalized iterator (for await ...of)

Message types and formats depend on specific normalizers provided to streamNormalized function and are documented in detail in data normalization section.

​

sample message

Sample message produced by normalizeTrades

{
  type: 'trade',
  symbol: 'XBTUSD',
  exchange: 'bitmex',
  id: 'b1f4b309-80e2-1ffb-340b-2f7576f6ef0d',
  price: 7447.5,
  amount: 100,
  side: 'buy',
  timestamp: 2019-10-24T13:28:07.867Z,
  localTimestamp: 2019-10-24T13:28:07.938Z
}

​

​

​

Historical market data helpers

init(options)

When working with market data viareplay and replayNormalized functions by default only first day of each month of historical data is available for replay as well as locally cached historical data is stored in default location on disk (OS temp dir).

Init function allows providing apiKey received via email after ordering historical market data access via tardis.dev website as well as customcacheDir. ApiKey can also be provided directly via options of replay and replayNormalized functions - that overrides anything that was provided via init.

const { init } = require('tardis-dev')
​
init({
  apiKey: '<YOUR API KEY>',
  cacheDir: '<CUSTOM CACHE DIR>'
})

​

init options

​

​

​

getExchangeDetails(exchange)

Given exchange id provides exchange details (available symbols, availability dates, available channels, pricing info etc) provided by exchanges/:exchange API endpoint.

const { getExchangeDetails } = require('tardis-dev')
​
const bitmexExchangeDetails = await getExchangeDetails('bitmex')

​​​​

​

type of response returned by awaiting on getExchangeDetails

{
  id: string
  name: string
  enabled: boolean
  filterable: boolean
  availableSince: string
  availableSymbols: {
    id: string
    type: 'spot' | 'future' | 'perpetual' | 'option'
    availableSince: string
    availableTo?: string
  }[]
  availableChannels: string[]
  incidentReports: {
    from: string
    to: string
    status: string
    details: string
  }
}

​

​

​

getApiKeyAccessInfo(apiKey?)

Given apiKey provided as optional parameter or provided in init function provides information about what historical data is available for it - exchanges, date ranges, symbols.

const { getApiKeyAccessInfo } = require('tardis-dev')
​
const details = await getApiKeyAccessInfo()

​

type of response returned by awaiting on getApiKeyAccessInfo()

{
  exchange: string
  from: string
  to: string
  symbols: string[]
}[]

​

​

​

clearCache()

Clears local data cache dir.

const { clearCache } = require('tardis-dev')
​
await clearCache()

​

​

​

Data normalization

Data normalization allows consuming market data feeds from various exchanges in consistent format.

tardis-dev has following built-in normalizers that can be provided to replayNormalized or streamNormalized functions:

• ​normalizeTrades - provides normalized trade data

• normalizeBookChanges - provides normalized order book change data

• ​normalizeDerivativeTickers - provides normalized funding, index and mark price data

If you're interested in how exactly data is mapped from exchange native format to normalized one, please follow code in tardis-dev GitHub repository for each exchange and if you determined that mapping should be done differently please read "modifying built-in and adding custom normalizers" section.

const { streamNormalized, normalizeTrades, normalizeBookChanges,
  normalizeDerivativeTickers } = require('tardis-dev')
​
// or replayNormalized to replay normalized historical data
const messages = streamNormalized(
  {
    exchange: 'deribit',
    symbols: ['BTC-PERPETUAL']
  },
  normalizeTrades,
  normalizeBookChanges,
  normalizeDerivativeTickers
)
​
for await (const message of messages) {
  if (message.type === 'book_change') {
    // process normalized book change
  }
  if (message.type === 'trade') {
    // process normalized trade
  }
  if (message.type === 'derivative_ticker') {
    // process normalized derivative_ticker
  }
}

​​​​

​

normalizeTrades

When passed as an arg toreplayNormalized or streamNormalized function provides normalized trade data for all supported exchanges.

​

type of message provided by normalizeTrades

{
  type: 'trade'
  symbol: string
  exchange: string
  id: string | undefined
  price: number
  amount: number
  side: 'buy' | 'sell' | 'unknown' // liquidity taker side (aggressor)
  timestamp: Date
  localTimestamp: Date
}

​

sample normalized trade message

{
  type: 'trade',
  symbol: 'XBTUSD',
  exchange: 'bitmex',
  id: '282a0445-0e3a-abeb-f403-11003204ea1b',
  price: 7996,
  amount: 50,
  side: 'sell',
  timestamp: 2019-10-23T10:32:49.669Z,
  localTimestamp: 2019-10-23T10:32:49.740Z
}

​

​

​

normalizeBookChanges

When passed as an arg toreplayNormalized or streamNormalized function provides normalized book_change data for all supported exchanges.

It provides initial full order book snapshots (isSnapshot=true) plus incremental updates for each order book change as published by exchange.

​

type of message provided by normalizeBookChanges

{
  type: 'book_change'
  symbol: string
  exchange: string
  isSnapshot: boolean
  bids: { price: number; amount: number }[]
  asks: { price: number; amount: number }[]
  timestamp: Date
  localTimestamp: Date
}

​

sample normalized book_change message

{
  type: 'book_change',
  symbol: 'XBTUSD',
  exchange: 'bitmex',
  isSnapshot: false,
  bids: [],
  asks: [{ price: 7985, amount: 283318 }],
  timestamp: 2019-10-23T11:29:53.469Z,
  localTimestamp: 2019-10-23T11:29:53.469Z
}

​

​

​

normalizeDerivativeTickers

When passed as an arg toreplayNormalized or streamNormalized function provides normalized derivative_ticker data for supported exchanges that trade derivative instruments.

​

type of message provided by normalizeDerivativeTickers

{
  type: 'derivative_ticker'
  symbol: string
  exchange: string
  lastPrice: number | undefined
  openInterest: number | undefined
  fundingRate: number | undefined
  indexPrice: number | undefined
  markPrice: number | undefined
  timestamp: Date
  localTimestamp: Date
}

​

sample normalized derivative_ticker message

{
  type: 'derivative_ticker',
  symbol: 'BTC-PERPETUAL',
  exchange: 'deribit',
  lastPrice: 7987.5,
  openInterest: 84129491,
  fundingRate: -0.00001568,
  indexPrice: 7989.28,
  markPrice: 7987.56,
  timestamp: 2019-10-23T11:34:29.302Z,
  localTimestamp: 2019-10-23T11:34:29.416Z
}

​

​

​

disconnect message

When replayNormalized or streamNormalized functions options have withDisconnectMessages flag set to true and disconnect event occurred (eg.: WebSocket connection close) then disconnect message is being produced by the async iteratables returned by those functions.

​

type of disconnect message

{
  type: 'disconnect'
  exchange: string
  localTimestamp: Date
}

​

sample disconnect message

{
  type: 'disconnect',
  exchange: 'deribit',
  localTimestamp: 2019-10-23T11:34:29.416Z
}

​

​

​

​

Modifying built-in and adding custom normalizers

Intardis-dev data normalization is implemented via normalize factory functions provided to replayNormalized and streamNormalized functions. This design gives lots of flexibility by allowing replacing, extending and modifying built-in normalizers or adding new ones for new normalized data types without the need of forking the whole library.

Any normalize function provided to replayNormalized and streamNormalized functions needs to have following signature:

(exchange: string, localTimestamp: Date) => Mapper

Exchange is an exchange id for which mapper object needs to be returned for, localTimestamp is a date for which mapper is created (and is created after each disconnection). In most cases localTimestamp is not necessary for anything, but in certain cases like for example exchange API change it can be used to switch to different mapping logic like using new data channel that wasn't available until certain date.

Returned Mapper object has following signature:

{
  canHandle: (message: any) => boolean
​
  map(message: any, localTimestamp: Date): IterableIterator<Data> | undefined
​
  getFilters: (symbols?: string[]) => {channel: string, symbols?: string[]}[]
}

​

On every disconnection event that occurs normalize factory functions are called again to provide new Mapper objects with clean state if required (stateful mapping like BitMEX order book data that needs to persist state of mapping between price level id and price level and needs to 'reset' for each new connection). If mapper object is stateful it's required to always return new clean state object from normalize factory function or reset it's state in one way or another.

Normalized data returned by iterable iterator of Mapper.map method is expected to have a shape that has at least fields as described in normalized data type section below to play well with other tardis-dev functions like combine or compute.

normalized data type

{
  type: string
  symbol: string
  exchange: string
  timestamp: Date
  localTimestamp: Date
  name? : string | undefined
}

​

Adding custom normalizeLiquidations normalizer

Example implementation of custom normalizeLiquidations function that normalizes liquidations data for deribit exchange. Implementations for for other exchanges are left as an exercise for the reader.

​

type of messages provided by normalizeLiquidations

{
  type: 'liquidation'
  symbol: string
  exchange: string
  side: 'buy' | 'sell'
  amount: number
  price: number
  timestamp: Date
  localTimestamp: Date
}

implementation of deribitLiquidations mapper and normalizeLiquidations

// object that maps from deribit trades real-time channel to our custom
// liquidations messages if determines that trade was caused by liquidation
​
const deribitLiquidationsMapper = {
  
  // function that given message in deribit native format
  // https://docs.deribit.com/v2/#trades-instrument_name-interval
  // determines if such message can be mapped/normalized
  canHandle(message) {
    const params = message.params
    const channel = params !== undefined ? params.channel : undefined
    if (channel === undefined) {
      return false
    }
    return (
      channel.startsWith('trades') &&
      params.data.some(trade => trade.liquidation !== undefined)
    )
  },
  
  // given symbols returns filters that are provided 
  // as filters to replay & stream functions options
  // in our case we're interested in trades deribit channel
  // but it could be multiple channels in certain scenarions as well
  getFilters(symbols) {
    return [
      {
        channel: 'trades',
        symbols
      }
    ]
  },
​
  // map message that was determined that is can be handled via canHandle
  // to normalized liquidation message
  *map(message, localTimestamp) {
    for (const deribitLiquidationTrade of message.params.data) {
      if (deribitLiquidationTrade.liquidation === undefined) {
        continue
      }
​
      yield {
        type: 'liquidation',
        symbol: deribitLiquidationTrade.instrument_name,
        exchange: 'deribit',
        price: deribitLiquidationTrade.price,
        amount: deribitLiquidationTrade.amount,
        side: deribitLiquidationTrade.direction,
        timestamp: new Date(deribitLiquidationTrade.timestamp),
        localTimestamp: localTimestamp
      }
    }
  }
}
​
// provides factory function that given exchange returns mapper for it
// if such mapper if implemented
// in our case deribitLiquidationsMapper is stateless so we can return the same 
// instance every time
​
const normalizeLiquidations = (exchange, localTimestamp) => {
  if (exchange === 'deribit') {
    return deribitLiquidationsMapper
  }
  throw new Error(`normalizeLiquidations: ${exchange} not supported`)
}