Skip to content

Instantly share code, notes, and snippets.

Embed
What would you like to do?
GDAX WebSocket Rolling Stats
'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
}
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
You can’t perform that action at this time.