GDAX WebSocket Rolling Stats

gdax-stream.js

'use strict'

const Gdax = require('gdax')
const h = require('highland')
const streamStatistics = require('stream-statistics')
const websocket = new Gdax.WebsocketClient(['BTC-USD'])

const tickerStream = exports.tickerStream = (s = websocket) =>
  h('message', s)
    .stopOnError(() =>
      tickerStream(new Gdax.WebsocketClient(['BTC-USD'])))

const matchedPrice = () => h.pipeline(
  h.where({ type: 'match' }),
  h.pluck('price')
)

const statStream = (label, duration, cb) => h.pipeline(
  h.through(matchedPrice()),
  h.batchWithTimeOrCount(duration, 100000),
  h.tap(orders => h(orders)
    .through(streamStatistics())
    .toArray(p => {
      cb(p[0], label)
    })
  )
)

exports.createStatsStream = function (label, duration, cb) {
  const stream = () => {
    tickerStream()
      .through(statStream(label, duration, cb))
      .stopOnError(() => {
        console.log(`${label} encountered an unexpected error, restarting`)
        stream()
      })
      .done(() => {
        console.log(`${label} stats stream ended unexpectedly, restarting`)
        stream()
      })
  }

  return stream
}