michaelgrosner
commented
8 years ago Hey @josephfrazier, I do have some sample code I've started:
import ws = require('ws');
import Utils = require("./utils");
import moment = require("moment");
import Models = require("../common/models");
import _ = require("lodash");
import Interfaces = require("./interfaces");
import Messages = require("./messages");
import Persister = require("./persister");
import Config = require("./config");
import QuotingParameters = require("./quoting-parameters");
const Deque = require("collections/deque");
export type BitfinexMarketTrade = [number, string, string, number, number, number, number];
export interface IBitfinexSocket {
subscribe<T>(channel: string,
pair: string,
snapshot: (msgs: Models.Timestamped<T[]>) => void,
update: (msg: Models.Timestamped<T>) => void);
}
export function createBitfinexSocket(config: Config.IConfigProvider, timeProvider : Utils.ITimeProvider) : IBitfinexSocket {
const url = config.GetString("BitfinexWebsocketUrl");
if (url === null || url === "NULL") {
return {subscribe: () => {}};
}
else {
return new BitfinexWebsocket(config, timeProvider);
}
}
class BitfinexWebsocket implements IBitfinexSocket {
private _ws : ws = null;
private _log = Utils.log("bitfinex:websocket");
private _snapshotHandlers : {[key: string]: [string, string, (msgs: Models.Timestamped<any[]>) => void, (msg: Models.Timestamped<any>) => void]} = {};
private _channelIdsToChannels : {[channelId: string]: string} = {};
constructor(
private _config: Config.IConfigProvider,
private _timeProvider : Utils.ITimeProvider) {
this.createSocket();
}
private createSocket = () => {
this.discardSocket();
this._log.info("creating bitfinex websocket");
this._ws = new ws(this._config.GetString("BitfinexWebsocketUrl")); // "wss://api2.bitfinex.com:3000/ws"
this._ws.on("open", this.onOpen);
this._ws.on("message", this.onMessage);
this._ws.on("close", this.onClose);
this._ws.on("error", this.onError);
};
private discardSocket = () => {
if (this._ws === null) return;
this._log.info("discarding bitfinex websocket");
try {
this._ws.close();
}
finally {
this._ws = null;
}
};
private onOpen = () => {
this._log.info("connected to bitfinex.");
for (let v of _.values(this._snapshotHandlers)) {
this.sendSubscriptionRequest(v[0], v[1]);
}
}
private onClose = (code, msg) => {
this._log.warn("socket closed to bitfinex, attempting to reconnect in 15 seconds.", {code: code, msg: msg});
this.delayedReconnect();
};
private onError = (err) => {
this._log.error("error to bitfinex!", err);
this.delayedReconnect();
};
private delayedReconnect = () => {
this.discardSocket();
this._timeProvider.setTimeout(this.createSocket, moment.duration(15, 'seconds'));
};
private onMessage = (rawData) => {
try {
this.onMessageUnsafe(rawData);
}
catch (e) {
this._log.error("Unhandled exception handling Bitfinex data", rawData, e);
}
};
private onMessageUnsafe = (rawData) => {
const t = this._timeProvider.utcNow();
const parsed = JSON.parse(rawData);
if (parsed['event'] === "subscribed") {
const key = `${parsed['channel']}:${parsed['pair']}`;
this._channelIdsToChannels[parsed['chanId']] = key;
this._log.info("subscription response", parsed);
}
else {
const channelId = parsed[0];
const channelKey = this._channelIdsToChannels[channelId];
if (typeof channelKey === 'undefined') {
if (parsed['event'] === 'info')
this._log.info("info from bitfinex", parsed);
else
this._log.warn("received update for unknown channel", parsed);
return;
}
const handlers = this._snapshotHandlers[channelKey];
if (typeof handlers === 'undefined') {
this._log.warn("received update without handlers", parsed);
return;
}
if (_.isArray(parsed[1])) {
// snapshot
const handler = handlers[2];
if (handler === null) return;
handler(new Models.Timestamped(parsed[1], t));
}
else {
// incremental
if (parsed[1] === "hb") return;
const handler = handlers[3];
if (handler === null) return;
handler(new Models.Timestamped(parsed, t));
}
}
};
public subscribe = <T>(channel: string,
pair: string,
snapshot: (msgs: Models.Timestamped<T[]>) => void,
update: (msg: Models.Timestamped<T>) => void) => {
const key = `${channel}:${pair}`;
if (typeof this._snapshotHandlers[key] !== 'undefined') {
throw new Error(`Already subscribed for channel=${channel}, pair=${pair}`);
}
this._snapshotHandlers[key] = [channel, pair, snapshot, update];
if (this._ws.readyState === ws.OPEN) {
this.sendSubscriptionRequest(channel, pair);
}
}
private sendSubscriptionRequest = (channel: string, pair: string) => {
const data = {
"event": "subscribe",
"channel": channel,
"pair": pair
};
this._log.info("sending subscription request", data);
this._ws.send(JSON.stringify(data));
};
}This just writes to their websocket in a generic way, filling in the subscriptions for market data, positions, and order entry should be straightforward from their docs.
josephfrazier
gacelita
thekenshow
arrrgi
ntenko07
frankely
Bitfinex has added a WebSocket API (documentation), but tribeca currently doesn't support it. Would you be interested in reviewing/merging pull requests that add WebSocket support? I haven't yet familiarized myself with the code, but I thought I'd go ahead and ask before digging too far into it. Thanks!