c# - GDAX/Coinbase API Level3 orderbook - 跳过消息

Question

我正在使用 GDAX API Websocket Stream尝试创建完整的 LEVEL3 订单簿的副本。

我有一个使用 WebSocketSharp 的非常简单的实现，我基本上就是在做类似的事情。

private readonly WebSocket _webSocket = new WebSocket("wss://ws-feed.gdax.com");

_webSocket.OnMessage += WebSocket_OnMessage;
_webSocket.Connect();
_webSocket.Send(JsonConvert.SerializeObject(new BeginSubscriptionMessage()));

private void WebSocket_OnMessage(object sender, MessageEventArgs e)
{
    var message = JsonConvert.DeserializeObject<BaseMessage>(e.Data);
    switch (message.Type)
    {   
        case "match": //A trade occurred between two orders. 
            MatchMessage matchMessage = JsonConvert.DeserializeObject<MatchMessage>(e.Data);
            _receivedMatchQueue.Enqueue(matchMessage);
            break;
        case "received": //A valid order has been received and is now active. This message is emitted for every single valid order as soon as the matching engine receives it whether it fills immediately or not.
            ReceivedMessage receivedMessage = JsonConvert.DeserializeObject<ReceivedMessage>(e.Data);
            _receivedMessageQueue.Enqueue(receivedMessage);
            break;
        case "open": //The order is now open on the order book. This message will only be sent for orders which are not fully filled immediately. remaining_size will indicate how much of the order is unfilled and going on the book.
            OpenMessage openMessage = JsonConvert.DeserializeObject<OpenMessage>(e.Data);
            _receivedOpenQueue.Enqueue(openMessage);
            break;
        case "done": //The order is no longer on the order book. Sent for all orders for which there was a received message. This message can result from an order being canceled or filled. 
            DoneMessage doneMessage = JsonConvert.DeserializeObject<DoneMessage>(e.Data);
            _receivedDoneQueue.Enqueue(doneMessage);
            break;
        case "change": //Existing order has been changed
            ChangeMessage changeMessage = JsonConvert.DeserializeObject<ChangeMessage>(e.Data);
            _receivedChangeQueue.Enqueue(changeMessage);
            break;
        case "activate": //Stop order placed
            //Console.WriteLine("Stop Order Placed");
            //ActivateMessage activateMessage = JsonConvert.DeserializeObject<ActivateMessage>(e.Data);

            break;
        case "subscriptions":
            break;
        case "ticker":
            TickerMessage tickerMessage = JsonConvert.DeserializeObject<TickerMessage>(e.Data);
            _receivedTickerQueue.Enqueue(tickerMessage);
            break;
        case "l2update":

            break;
    }
}

我遇到的问题是，当我查看通过 RECEIVED 和 OPEN 消息收到的序列号时，我可以看到它们不是连续的，这 ( based on the following information ) 表明消息被跳过。

基本上你最终会得到这样的结果

Open Message SequenceId: 5359746354
Open Message SequenceId: 5359746358
Open Message SequenceId: 5359746361
Open Message SequenceId: 5359746363
Open Message SequenceId: 5359746365
Open Message SequenceId: 5359746370
Open Message SequenceId: 5359746372

我已经尝试在 Azure 上对此进行测试，只是为了确保它不是我这边的带宽限制，而且结果大致相似。

鉴于此，如果消息被丢弃，如何使用“完整”websocket 流构建一个完整的“实时”订单簿？我可以安全地忽略它们吗？或者我只是以某种方式清除孤立的值？

任何做过类似事情的人的任何建议都将不胜感激。

Answer 1

很可能消息没有被丢弃，您只是对这些序列号代表的“序列”有错误的印象。

如 api 文档中所述

Most feed messages contain a sequence number. Sequence numbers are increasing integer values for each product with every new message being exactly 1 sequence number than the one before it.

因此每个 channel 都有每个产品的单独序列号(如 ETH-USD)，而不是每个消息类型(如“打开”或“接收”)。假设您订阅了产品 ETH-USD 和 ETH-EUR 的“完整” channel 。那么你应该期待这样的序列:

receive `ETH-EUR` X
open `ETH-EUR` X + 1
receive `ETH-USD` Y
done `ETH-EUR` X + 2
open `ETH-USD` Y + 1

对于全 channel ，消息类型有:received、open、done、match、change、activate(注意ticker消息属于不同的 channel ，所以有不同的顺序)。因此，为确保不跳过任何消息，您需要跟踪所有这些消息类型，并确保您收到的每个产品的最后一个序列号正好比新序列号小 1(如果您订阅了多个产品)。

验证码:

class Program {
    private static readonly WebSocket _webSocket = new WebSocket("wss://ws-feed.gdax.com");
    private static long _lastSequence = 0;
    private static readonly HashSet<string> _expectedTypes = new HashSet<string>(
        new[] { "received", "open", "done", "match", "change", "activate" });

    static void Main(string[] args) {
        var subMsg = "{\"type\": \"subscribe\",\"product_ids\": [\"ETH-USD\"],\"channels\": [\"full\"]}";
        _webSocket.OnMessage += WebSocket_OnMessage;
        _webSocket.Connect();
        _webSocket.Send(subMsg);
        Console.ReadKey();
    }        

    private static void WebSocket_OnMessage(object sender, MessageEventArgs e) {
        var message = JsonConvert.DeserializeObject<BaseMessage>(e.Data);
        if (_expectedTypes.Contains(message.Type)) {
            lock (typeof(Program)) {
                if (_lastSequence == 0)
                    _lastSequence = message.Sequence;
                else {
                    if (message.Sequence > _lastSequence + 1) {
                        Debugger.Break(); // never hits, so nothing is dropped
                    }
                    _lastSequence = message.Sequence;
                }
            }
        }
    }
}

public class BaseMessage {
    [JsonProperty("type")]
    public string Type { get; set; }

    [JsonProperty("product_id")]
    public string ProductId { get; set; }

    [JsonProperty("sequence")]
    public long Sequence { get; set; }
}