Home

Awesome

Logo

Cryptocurrency websocket extensions Build Status NuGet version

This is a library that provides extensions to cryptocurrency websocket exchange clients.

It helps to unify data models and usage of more clients together.

Releases and breaking changes

License:

Apache License 2.0

Features

Supported exchanges

LogoNameWebsocket client
bitfinexBitfinexbitfinex-client-websocket
bitmexBitMEXbitmex-client-websocket
binanceBinancebinance-client-websocket
coinbaseCoinbasecoinbase-client-websocket
bitstampBitstampbitstamp-client-websocket

Extensions

Order book

Usage:

var url = BitmexValues.ApiWebsocketUrl;
var communicator = new BitmexWebsocketCommunicator(url);
var client = new BitmexWebsocketClient(communicator);

var pair = "XBTUSD";

var source = new BitmexOrderBookSource(client);
var orderBook = new CryptoOrderBook(pair, source);

// orderBook.BidAskUpdatedStream.Subscribe(xxx)
orderBook.OrderBookUpdatedStream.Subscribe(quotes =>
{
    var currentBid = orderBook.BidPrice;
    var currentAsk = orderBook.AskPrice;

    var bids = orderBook.BidLevels;
    // xxx
});
        
await communicator.Start();

Trades

Orders (authenticated)

Position (authenticated)

Wallet (authenticated)


More usage examples:

Pull Requests are welcome!

Powerfull Rx.NET

Don't forget that you can do pretty nice things with reactive extensions and observables. For example, if you want to check latest bid/ask prices from all exchanges all together, you can do something like this:

Observable.CombineLatest(new[]
            {
                bitmexOrderBook.BidAskUpdatedStream,
                bitfinexOrderBook.BidAskUpdatedStream,
                binanceOrderBook.BidAskUpdatedStream,
            })
            .Subscribe(HandleQuoteChanged);

// Method HandleQuoteChanged(IList<CryptoQuotes> quotes)
// will be called on every exchange's price change

Multi-threading

Observables from Reactive Extensions are single threaded by default. It means that your code inside subscriptions is called synchronously and as soon as the message comes from websocket API. It brings a great advantage of not to worry about synchronization, but if your code takes a longer time to execute it will block the receiving method, buffer the messages and may end up losing messages. For that reason consider to handle messages on the other thread and unblock receiving thread as soon as possible. I've prepared a few examples for you:

Default behavior

Every subscription code is called on a main websocket thread. Every subscription is synchronized together. No parallel execution. It will block the receiving thread.

client
    .Streams
    .TradesStream
    .Subscribe(trade => { code1 });

client
    .Streams
    .BookStream
    .Subscribe(book => { code2 });

// 'code1' and 'code2' are called in a correct order, according to websocket flow
// ----- code1 ----- code1 ----- ----- code1
// ----- ----- code2 ----- code2 code2 -----

Parallel subscriptions

Every single subscription code is called on a separate thread. Every single subscription is synchronized, but different subscriptions are called in parallel.

client
    .Streams
    .TradesStream
    .ObserveOn(TaskPoolScheduler.Default)
    .Subscribe(trade => { code1 });

client
    .Streams
    .BookStream
    .ObserveOn(TaskPoolScheduler.Default)
    .Subscribe(book => { code2 });

// 'code1' and 'code2' are called in parallel, do not follow websocket flow
// ----- code1 ----- code1 ----- code1 -----
// ----- code2 code2 ----- code2 code2 code2

Parallel subscriptions with synchronization

In case you want to run your subscription code on the separate thread but still want to follow websocket flow through every subscription, use synchronization with gates:

private static readonly object GATE1 = new object();
client
    .Streams
    .TradesStream
    .ObserveOn(TaskPoolScheduler.Default)
    .Synchronize(GATE1)
    .Subscribe(trade => { code1 });

client
    .Streams
    .BookStream
    .ObserveOn(TaskPoolScheduler.Default)
    .Synchronize(GATE1)
    .Subscribe(book => { code2 });

// 'code1' and 'code2' are called concurrently and follow websocket flow
// ----- code1 ----- code1 ----- ----- code1
// ----- ----- code2 ----- code2 code2 ----

Async/Await integration

Using async/await in your subscribe methods is a bit tricky. Subscribe from Rx.NET doesn't await tasks, so it won't block stream execution and cause sometimes undesired concurrency. For example:

client
    .Streams
    .TradesStream
    .Subscribe(async trade => {
        // do smth 1
        await Task.Delay(5000); // waits 5 sec, could be HTTP call or something else
        // do smth 2
    });

That await Task.Delay won't block stream and subscribe method will be called multiple times concurrently. If you want to buffer messages and process them one-by-one, then use this:

client
    .Streams
    .TradesStream
    .Select(trade => Observable.FromAsync(async () => {
        // do smth 1
        await Task.Delay(5000); // waits 5 sec, could be HTTP call or something else
        // do smth 2
    }))
    .Concat() // executes sequentially
    .Subscribe();

If you want to process them concurrently (avoid synchronization), then use this

client
    .Streams
    .TradesStream
    .Select(trade => Observable.FromAsync(async () => {
        // do smth 1
        await Task.Delay(5000); // waits 5 sec, could be HTTP call or something else
        // do smth 2
    }))
    .Merge() // executes concurrently
    // .Merge(4) you can limit concurrency with a parameter
    // .Merge(1) is same as .Concat()
    // .Merge(0) is invalid (throws exception)
    .Subscribe();

More info on Github issue.

Don't worry about websocket connection, those sequential execution via .Concat() or .Merge(1) has no effect on receiving messages. It won't affect receiving thread, only buffers messages inside TradesStream.

But beware of producer-consumer problem when the consumer will be too slow. Here is a StackOverflow issue with an example how to ignore/discard buffered messages and always process only the last one.

Available for help

I do consulting, please don't hesitate to contact me if you have a custom solution you would like me to implement (web, m@mkotas.cz)