diff --git a/CryptoExchange.Net/CryptoExchange.Net.xml b/CryptoExchange.Net/CryptoExchange.Net.xml index ca429f4..45fcd78 100644 --- a/CryptoExchange.Net/CryptoExchange.Net.xml +++ b/CryptoExchange.Net/CryptoExchange.Net.xml @@ -871,6 +871,16 @@ The price of the entry + + + Interface for order book entries + + + + + Sequence of the update + + Interface for websocket interaction @@ -1649,16 +1659,6 @@ Buffer entry for order book - - - The first sequence number of the entries - - - - - The last sequence number of the entries - - List of asks @@ -1669,6 +1669,41 @@ List of bids + + + First update id + + + + + List of asks + + + + + List of bids + + + + + First update id + + + + + Last update id + + + + + List of asks + + + + + List of bids + + Base for order book implementations @@ -1699,6 +1734,11 @@ The log + + + If order book is set + + The status of the order book. Order book is up to date when the status is `Synced` @@ -1722,7 +1762,7 @@ Event when order book was updated, containing the changed bids and asks. Be careful! It can generate a lot of events at high-liquidity markets - + @@ -1815,12 +1855,27 @@ List of asks List of bids + + + Update the order book using a single id for an update + + + + + - Update the order book with entries + Update the order book using a first/last update id + + + + + + + + + Update the order book using sequenced entries - First sequence number - Last sequence number List of bids List of asks @@ -1829,13 +1884,21 @@ Check and empty the process buffer; see what entries to update the book with - + Update order book with an entry + Sequence number of the update Type of entry The entry + + + Wait until the order book has been set + + Max wait time + + Dispose the order book @@ -2815,148 +2878,5 @@ - - - Specifies that is allowed as an input even if the - corresponding type disallows it. - - - - - Initializes a new instance of the class. - - - - - Specifies that is disallowed as an input even if the - corresponding type allows it. - - - - - Initializes a new instance of the class. - - - - - Specifies that a method that will never return under any circumstance. - - - - - Initializes a new instance of the class. - - - - - Specifies that the method will not return if the associated - parameter is passed the specified value. - - - - - Gets the condition parameter value. - Code after the method is considered unreachable by diagnostics if the argument - to the associated parameter matches this value. - - - - - Initializes a new instance of the - class with the specified parameter value. - - - The condition parameter value. - Code after the method is considered unreachable by diagnostics if the argument - to the associated parameter matches this value. - - - - - Specifies that an output may be even if the - corresponding type disallows it. - - - - - Initializes a new instance of the class. - - - - - Specifies that when a method returns , - the parameter may be even if the corresponding type disallows it. - - - - - Gets the return value condition. - If the method returns this value, the associated parameter may be . - - - - - Initializes the attribute with the specified return value condition. - - - The return value condition. - If the method returns this value, the associated parameter may be . - - - - - Specifies that an output is not even if the - corresponding type allows it. - - - - - Initializes a new instance of the class. - - - - - Specifies that the output will be non- if the - named parameter is non-. - - - - - Gets the associated parameter name. - The output will be non- if the argument to the - parameter specified is non-. - - - - - Initializes the attribute with the associated parameter name. - - - The associated parameter name. - The output will be non- if the argument to the - parameter specified is non-. - - - - - Specifies that when a method returns , - the parameter will not be even if the corresponding type allows it. - - - - - Gets the return value condition. - If the method returns this value, the associated parameter will not be . - - - - - Initializes the attribute with the specified return value condition. - - - The return value condition. - If the method returns this value, the associated parameter will not be . - - diff --git a/CryptoExchange.Net/Interfaces/ISymbolOrderBookEntry.cs b/CryptoExchange.Net/Interfaces/ISymbolOrderBookEntry.cs index 419f41a..3aa9f97 100644 --- a/CryptoExchange.Net/Interfaces/ISymbolOrderBookEntry.cs +++ b/CryptoExchange.Net/Interfaces/ISymbolOrderBookEntry.cs @@ -14,4 +14,15 @@ /// decimal Price { get; set; } } + + /// + /// Interface for order book entries + /// + public interface ISymbolOrderSequencedBookEntry: ISymbolOrderBookEntry + { + /// + /// Sequence of the update + /// + long Sequence { get; set; } + } } diff --git a/CryptoExchange.Net/OrderBook/ProcessBufferEntry.cs b/CryptoExchange.Net/OrderBook/ProcessBufferEntry.cs index 397f53a..b5ebfe9 100644 --- a/CryptoExchange.Net/OrderBook/ProcessBufferEntry.cs +++ b/CryptoExchange.Net/OrderBook/ProcessBufferEntry.cs @@ -9,13 +9,41 @@ namespace CryptoExchange.Net.OrderBook public class ProcessBufferEntry { /// - /// The first sequence number of the entries + /// List of asks /// - public long FirstSequence { get; set; } + public IEnumerable Asks { get; set; } = new List(); /// - /// The last sequence number of the entries + /// List of bids /// - public long LastSequence { get; set; } + public IEnumerable Bids { get; set; } = new List(); + } + + public class ProcessBufferSingleSequenceEntry + { + /// + /// First update id + /// + public long UpdateId { get; set; } + /// + /// List of asks + /// + public IEnumerable Asks { get; set; } = new List(); + /// + /// List of bids + /// + public IEnumerable Bids { get; set; } = new List(); + } + + public class ProcessBufferRangeSequenceEntry + { + /// + /// First update id + /// + public long FirstUpdateId { get; set; } + /// + /// Last update id + /// + public long LastUpdateId { get; set; } /// /// List of asks /// diff --git a/CryptoExchange.Net/OrderBook/SymbolOrderBook.cs b/CryptoExchange.Net/OrderBook/SymbolOrderBook.cs index 4597e7d..cc890c8 100644 --- a/CryptoExchange.Net/OrderBook/SymbolOrderBook.cs +++ b/CryptoExchange.Net/OrderBook/SymbolOrderBook.cs @@ -19,7 +19,7 @@ namespace CryptoExchange.Net.OrderBook /// /// The process buffer, used while syncing /// - protected readonly List processBuffer; + protected readonly List processBuffer; private readonly object bookLock = new object(); /// /// The ask list @@ -43,7 +43,10 @@ namespace CryptoExchange.Net.OrderBook /// protected Log log; - private bool bookSet; + /// + /// If order book is set + /// + protected bool bookSet; /// /// The status of the order book. Order book is up to date when the status is `Synced` @@ -78,7 +81,7 @@ namespace CryptoExchange.Net.OrderBook public event Action? OnStatusChange; /// /// Event when order book was updated, containing the changed bids and asks. Be careful! It can generate a lot of events at high-liquidity markets - /// + /// public event Action, IEnumerable>? OnOrderBookUpdate; /// /// Timestamp of the last update @@ -156,7 +159,7 @@ namespace CryptoExchange.Net.OrderBook throw new ArgumentNullException(nameof(options)); Id = options.OrderBookName; - processBuffer = new List(); + processBuffer = new List(); sequencesAreConsecutive = options.SequenceNumbersAreConsecutive; Symbol = symbol; Status = OrderBookStatus.Disconnected; @@ -259,11 +262,11 @@ namespace CryptoExchange.Net.OrderBook /// The last update sequence number /// List of asks /// List of bids - protected void SetInitialOrderBook(long orderBookSequenceNumber, IEnumerable askList, IEnumerable bidList) + protected void SetInitialOrderBook(long orderBookSequenceNumber, IEnumerable bidList, IEnumerable askList) { lock (bookLock) { - if (Status == OrderBookStatus.Connecting) + if (Status == OrderBookStatus.Connecting || Status == OrderBookStatus.Disconnected) return; asks.Clear(); @@ -278,99 +281,213 @@ namespace CryptoExchange.Net.OrderBook AskCount = asks.Count; BidCount = asks.Count; - CheckProcessBuffer(); bookSet = true; LastOrderBookUpdate = DateTime.UtcNow; + log.Write(LogVerbosity.Debug, $"{Id} order book {Symbol} data set: {BidCount} bids, {AskCount} asks. #{orderBookSequenceNumber}"); + CheckProcessBuffer(); OnOrderBookUpdate?.Invoke(bidList, askList); - log.Write(LogVerbosity.Debug, $"{Id} order book {Symbol} data set: {BidCount} bids, {AskCount} asks"); } } /// - /// Update the order book with entries + /// Update the order book using a single id for an update /// - /// First sequence number - /// Last sequence number - /// List of bids - /// List of asks - protected void UpdateOrderBook(long firstSequenceNumber, long lastSequenceNumber, IEnumerable bids, IEnumerable asks) + /// + /// + /// + protected void UpdateOrderBook(long rangeUpdateId, IEnumerable bids, IEnumerable asks) { lock (bookLock) { - if (lastSequenceNumber < LastSequenceNumber) + if (Status == OrderBookStatus.Connecting || Status == OrderBookStatus.Disconnected) return; if (!bookSet) { - var entry = new ProcessBufferEntry + processBuffer.Add(new ProcessBufferSingleSequenceEntry() { - FirstSequence = firstSequenceNumber, - LastSequence = lastSequenceNumber, + UpdateId = rangeUpdateId, Asks = asks, Bids = bids - }; - processBuffer.Add(entry); - log.Write(LogVerbosity.Debug, $"{Id} order book {Symbol} update before synced; buffering"); - } - else if (sequencesAreConsecutive && firstSequenceNumber > LastSequenceNumber + 1) - { - // Out of sync - log.Write(LogVerbosity.Warning, $"{Id} order book {Symbol} out of sync, reconnecting"); - subscription!.Reconnect().Wait(); + }); + log.Write(LogVerbosity.Debug, $"{Id} order book {Symbol} update buffered #{rangeUpdateId}"); } else { - foreach (var entry in asks) - ProcessUpdate(OrderBookEntryType.Ask, entry); - foreach (var entry in bids) - ProcessUpdate(OrderBookEntryType.Bid, entry); - LastSequenceNumber = lastSequenceNumber; CheckProcessBuffer(); - LastOrderBookUpdate = DateTime.UtcNow; + ProcessSingleSequenceUpdates(rangeUpdateId, bids, asks); OnOrderBookUpdate?.Invoke(bids, asks); - log.Write(LogVerbosity.Debug, $"{Id} order book {Symbol} update: {asks.Count()} asks, {bids.Count()} bids processed"); } } } + /// + /// Update the order book using a first/last update id + /// + /// + /// + /// + /// + protected void UpdateOrderBook(long firstUpdateId, long lastUpdateId, IEnumerable bids, IEnumerable asks) + { + lock (bookLock) + { + if (Status == OrderBookStatus.Connecting || Status == OrderBookStatus.Disconnected) + return; + + if (!bookSet) + { + processBuffer.Add(new ProcessBufferRangeSequenceEntry() + { + Asks = asks, + Bids = bids, + FirstUpdateId = firstUpdateId, + LastUpdateId = lastUpdateId + }); + log.Write(LogVerbosity.Debug, $"{Id} order book {Symbol} update buffered #{firstUpdateId}-{lastUpdateId}"); + } + else + { + CheckProcessBuffer(); + ProcessRangeUpdates(firstUpdateId, lastUpdateId, bids, asks); + OnOrderBookUpdate?.Invoke(bids, asks); + } + } + } + + /// + /// Update the order book using sequenced entries + /// + /// List of bids + /// List of asks + protected void UpdateOrderBook(IEnumerable bids, IEnumerable asks) + { + lock (bookLock) + { + if (!bookSet) + { + processBuffer.Add(new ProcessBufferEntry + { + Asks = asks, + Bids = bids + }); + log.Write(LogVerbosity.Debug, $"{Id} order book {Symbol} update buffered #{Math.Min(bids.Min(b => b.Sequence), asks.Min(a => a.Sequence))}-{Math.Max(bids.Max(b => b.Sequence), asks.Max(a => a.Sequence))}"); + } + else + { + CheckProcessBuffer(); + ProcessUpdates(bids, asks); + OnOrderBookUpdate?.Invoke(bids, asks); + } + } + } + + private void ProcessUpdates(IEnumerable bids, IEnumerable asks) + { + var entries = new Dictionary(); + foreach (var entry in asks.OrderBy(a => a.Sequence)) + entries.Add(entry, OrderBookEntryType.Ask); + foreach (var entry in bids.OrderBy(a => a.Sequence)) + entries.Add(entry, OrderBookEntryType.Bid); + + foreach (var entry in entries.OrderBy(e => e.Key.Sequence)) + { + if(ProcessUpdate(entry.Key.Sequence, entry.Value, entry.Key)) + LastSequenceNumber = entry.Key.Sequence; + log.Write(LogVerbosity.Debug, $"{Id} order book {Symbol} update #{LastSequenceNumber}"); + } + } + + private void ProcessRangeUpdates(long firstUpdateId, long lastUpdateId, IEnumerable bids, IEnumerable asks) + { + if (lastUpdateId < LastSequenceNumber) + { + log.Write(LogVerbosity.Debug, $"{Id} order book {Symbol} update skipped #{firstUpdateId}-{lastUpdateId}"); + return; + } + + foreach (var entry in bids) + ProcessUpdate(LastSequenceNumber + 1, OrderBookEntryType.Bid, entry); + + foreach (var entry in asks) + ProcessUpdate(LastSequenceNumber + 1, OrderBookEntryType.Ask, entry); + + LastSequenceNumber = lastUpdateId; + log.Write(LogVerbosity.Debug, $"{Id} order book {Symbol} update processed #{firstUpdateId}-{lastUpdateId}"); + } + + private void ProcessSingleSequenceUpdates(long updateId, IEnumerable bids, IEnumerable asks) + { + foreach (var entry in bids) + { + if (!ProcessUpdate(updateId, OrderBookEntryType.Bid, entry)) + return; + } + + foreach (var entry in asks) + { + if (!ProcessUpdate(updateId, OrderBookEntryType.Ask, entry)) + return; + } + + LastSequenceNumber = updateId; + log.Write(LogVerbosity.Debug, $"{Id} order book {Symbol} update processed #{LastSequenceNumber}"); + } + /// /// Check and empty the process buffer; see what entries to update the book with /// protected void CheckProcessBuffer() { - foreach (var bufferEntry in processBuffer.OrderBy(b => b.FirstSequence).ToList()) + var pbList = processBuffer.ToList(); + if(pbList.Count > 0) + log.Write(LogVerbosity.Debug, "Processing buffered updates"); + + foreach (var bufferEntry in pbList) { - if (bufferEntry.LastSequence < LastSequenceNumber) - { - processBuffer.Remove(bufferEntry); - continue; - } - - if (bufferEntry.FirstSequence > LastSequenceNumber + 1) - break; - - foreach (var entry in bufferEntry.Asks) - ProcessUpdate(OrderBookEntryType.Ask, entry); - foreach (var entry in bufferEntry.Bids) - ProcessUpdate(OrderBookEntryType.Bid, entry); + if (bufferEntry is ProcessBufferEntry pbe) + ProcessUpdates(pbe.Bids, pbe.Asks); + else if(bufferEntry is ProcessBufferRangeSequenceEntry pbrse) + ProcessRangeUpdates(pbrse.FirstUpdateId, pbrse.LastUpdateId, pbrse.Bids, pbrse.Asks); + else if (bufferEntry is ProcessBufferSingleSequenceEntry pbsse) + ProcessSingleSequenceUpdates(pbsse.UpdateId, pbsse.Bids, pbsse.Asks); processBuffer.Remove(bufferEntry); - LastSequenceNumber = bufferEntry.LastSequence; } } /// /// Update order book with an entry /// + /// Sequence number of the update /// Type of entry /// The entry - protected virtual void ProcessUpdate(OrderBookEntryType type, ISymbolOrderBookEntry entry) + protected virtual bool ProcessUpdate(long sequence, OrderBookEntryType type, ISymbolOrderBookEntry entry) { + if (Status != OrderBookStatus.Syncing && Status != OrderBookStatus.Synced) + return false; + + if (sequence <= LastSequenceNumber) + { + log.Write(LogVerbosity.Debug, $"{Id} order book {Symbol} update skipped #{sequence}"); + return false; + } + + if (sequencesAreConsecutive && sequence > LastSequenceNumber + 1) + { + // Out of sync + log.Write(LogVerbosity.Warning, $"{Id} order book {Symbol} out of sync (expected { LastSequenceNumber + 1}, was {sequence}), reconnecting"); + Status = OrderBookStatus.Connecting; + subscription?.Reconnect(); + return false; + } + + LastOrderBookUpdate = DateTime.UtcNow; var listToChange = type == OrderBookEntryType.Ask ? asks : bids; if (entry.Quantity == 0) { if (!listToChange.ContainsKey(entry.Price)) - return; + return true; listToChange.Remove(entry.Price); if (type == OrderBookEntryType.Ask) AskCount--; @@ -389,6 +506,27 @@ namespace CryptoExchange.Net.OrderBook listToChange[entry.Price].Quantity = entry.Quantity; } } + + return true; + } + + /// + /// Wait until the order book has been set + /// + /// Max wait time + /// + protected async Task> WaitForSetOrderBook(int timeout) + { + var startWait = DateTime.UtcNow; + while (!bookSet && Status == OrderBookStatus.Syncing) + { + if ((DateTime.UtcNow - startWait).TotalMilliseconds > timeout) + return new CallResult(false, new ServerError("Timeout while waiting for data")); + + await Task.Delay(10).ConfigureAwait(false); + } + + return new CallResult(true, null); } ///