Skip to content

Fix missing quotes when subscribing to more than 30 symbols #41

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
Martin-Molinero merged 1 commit into from
Sep 17, 2026
Merged

Fix missing quotes when subscribing to more than 30 symbols #41

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
62 changes: 61 additions & 1 deletion QuantConnect.CoinbaseBrokerage.Tests/CoinbaseBrokerageDataQueueHandlerTests.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -253,6 +253,65 @@ public void SubscribeOnMultipleSymbols(List<SubscriptionDataConfig> liquidSymbol
cancelationToken.Cancel();
}

[Test]
public void StreamsQuotesForMoreSymbolsThanASingleConnectionAllows()
{
// coinbase rejects level2 subscriptions above 30 per connection, trades are not limited
var tickers = new[]
{
"BTC", "ETH", "SOL", "XRP", "ADA", "AVAX", "DOGE", "DOT", "LINK", "LTC", "BCH", "UNI", "AAVE", "ATOM", "NEAR",
"APT", "ARB", "OP", "SUI", "HBAR", "XLM", "ALGO", "FIL", "ICP", "ETC", "CRV", "MKR", "SNX", "COMP", "GRT",
"SHIB", "PEPE", "BONK", "INJ", "TIA", "SEI", "STX", "IMX", "RNDR", "FET", "LDO", "ENS", "SAND", "MANA", "AXS",
"CHZ", "XTZ", "EOS", "ZEC", "BAT", "ZRX", "YFI", "SUSHI", "1INCH", "ANKR", "SKL", "STORJ", "LRC", "MASK", "APE"
};

var symbolsWithQuotes = new HashSet<Symbol>();
var configs = new List<SubscriptionDataConfig>();
var enumerators = new List<IEnumerator<BaseData>>();
foreach (var ticker in tickers)
{
var symbol = Symbol.Create(ticker + "USD", SecurityType.Crypto, Market.Coinbase);
var config = new SubscriptionDataConfig(GetSubscriptionDataConfig<Tick>(symbol, Resolution.Tick), tickType: TickType.Quote);
var enumerator = _brokerage.Subscribe(config, (s, e) => { });
if (enumerator == null)
{
// not supported any more
continue;
}
configs.Add(config);
enumerators.Add(enumerator);
}

Assert.Greater(configs.Count, 40);

// a single consumer for all, 'ProcessFeed' would spin a thread for each
var expected = (int)(configs.Count * 0.9);
var end = DateTime.UtcNow.AddSeconds(120);
while (symbolsWithQuotes.Count < expected && DateTime.UtcNow < end)
{
foreach (var enumerator in enumerators)
{
while (enumerator.MoveNext() && enumerator.Current != null)
{
if (enumerator.Current is Tick { TickType: TickType.Quote } tick)
{
symbolsWithQuotes.Add(tick.Symbol);
}
}
}
Thread.Sleep(100);
}

foreach (var config in configs)
{
_brokerage.Unsubscribe(config);
}

Log.Trace($"Got quotes for {symbolsWithQuotes.Count} of {configs.Count} symbols");
var missing = configs.Select(x => x.Symbol).Where(x => !symbolsWithQuotes.Contains(x)).Select(x => x.Value);
Assert.GreaterOrEqual(symbolsWithQuotes.Count, expected, $"No quotes for: {string.Join(",", missing)}");
}

private static IEnumerable<IEnumerable<Symbol>> BitcoinTradingPairs
{
get
Expand Down Expand Up @@ -313,7 +372,8 @@ public void SubscribeOnDifferentUSDTickers(List<Symbol> symbols)
});
}

Assert.IsTrue(resetEvent.WaitOne(TimeSpan.FromSeconds(30), cancelationTokenSource.Token));
Assert.IsTrue(resetEvent.WaitOne(TimeSpan.FromSeconds(30), cancelationTokenSource.Token),
$"Data received: {string.Join(", ", dataReceivedForType.Select(x => $"{x.Key.Item2.Value} {x.Key.Item1.Name}={x.Value}"))}");

foreach (var config in configs)
{
Expand Down
141 changes: 112 additions & 29 deletions QuantConnect.CoinbaseBrokerage/CoinbaseBrokerage.Messaging.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -70,9 +70,19 @@ public partial class CoinbaseBrokerage
private RateGate _webSocketRateLimit = new(7, TimeSpan.FromSeconds(1));

/// <summary>
/// Represents an integer variable used to keep track of sequence numbers associated with WS feed messages.
/// The maximum symbols per connection, above it coinbase rejects level2 with 'too many L2 streams requested in a single session'
/// </summary>
private int _sequenceNumbers = 0;
private const int MaximumSymbolsPerWebSocket = 30;

/// <summary>
/// The connection we follow for each data stream, by channel and brokerage product id
/// </summary>
private readonly ConcurrentDictionary<(string, string), StreamOwner> _streamOwner = new();

/// <summary>
/// Keeps track of the sequence numbers associated with WS feed messages, by connection
/// </summary>
private readonly ConcurrentDictionary<object, int> _sequenceNumbers = new();

/// <summary>
/// Use to sync subscription process on WebSocket User Updates
Expand Down Expand Up @@ -120,29 +130,35 @@ protected override void OnMessage(object _, WebSocketMessage webSocketMessage)
//this means an error has occurred
if (channel == null)
{
Log.Debug($"{nameof(CoinbaseBrokerage)}.{nameof(OnMessage)}.ERROR: {data.Message}");
Log.Error($"{nameof(CoinbaseBrokerage)}.{nameof(OnMessage)}.ERROR: {data.Message}");
return;
}

var newSequenceNumbers = obj["sequence_num"].Value<int>();

var connection = (object)webSocketMessage.WebSocket ?? this;
_sequenceNumbers.TryGetValue(connection, out var sequenceNumbers);

// https://docs.cloud.coinbase.com/advanced-trade-api/docs/ws-overview#sequence-numbers
if (newSequenceNumbers != 0 && newSequenceNumbers != _sequenceNumbers + 1)
if (newSequenceNumbers != 0 && newSequenceNumbers != sequenceNumbers + 1)
{
OnMessage(new BrokerageMessageEvent(BrokerageMessageType.Warning, "SequenceNumbers",
$"{nameof(CoinbaseBrokerage)}.{nameof(OnMessage)}: sequence number mismatch. If Sequence numbers are greater that a message has been dropped else ones are less can be ignored or represent a message that has arrived out of order."));
}

_sequenceNumbers = newSequenceNumbers;
_sequenceNumbers[connection] = newSequenceNumbers;

switch (channel)
{
case CoinbaseWebSocketChannels.MarketTrades:
var message = obj.ToObject<CoinbaseWebSocketMessage<CoinbaseMarketTradesEvent>>();
if (message.Events[0].Type == WebSocketEventType.Update)
HandleIfStreamOwner(channel, message.Events[0].Trades.FirstOrDefault()?.ProductId, message.Events[0].Type, connection, () =>
{
EmitTradeTick(message.Events[0]);
}
if (message.Events[0].Type == WebSocketEventType.Update)
{
EmitTradeTick(message.Events[0]);
}
});
break;
case CoinbaseWebSocketChannels.User:
var orderUpdate = obj.ToObject<CoinbaseWebSocketMessage<CoinbaseUserEvent>>();
Expand All @@ -156,17 +172,20 @@ protected override void OnMessage(object _, WebSocketMessage webSocketMessage)
break;
case CoinbaseWebSocketChannels.Level2Response:
var level2Data = obj.ToObject<CoinbaseWebSocketMessage<CoinbaseLevel2Event>>();
switch (level2Data.Events[0].Type)
HandleIfStreamOwner(channel, level2Data.Events[0].ProductId, level2Data.Events[0].Type, connection, () =>
{
case WebSocketEventType.Snapshot:
Level2Snapshot(level2Data.Events[0]);
break;
case WebSocketEventType.Update:
Level2Update(level2Data.Events[0]);
break;
default:
throw new ArgumentException();
};
switch (level2Data.Events[0].Type)
{
case WebSocketEventType.Snapshot:
Level2Snapshot(level2Data.Events[0]);
break;
case WebSocketEventType.Update:
Level2Update(level2Data.Events[0]);
break;
default:
throw new ArgumentException();
};
});
break;
}
}
Expand All @@ -176,6 +195,25 @@ protected override void OnMessage(object _, WebSocketMessage webSocketMessage)
}
}

/// <summary>
/// Handles the message if the connection is the one we follow for the given stream, so data streamed by more than one is handled once
/// </summary>
private void HandleIfStreamOwner(string channel, string productId, WebSocketEventType eventType, object connection, Action handler)
{
var streamOwner = _streamOwner.GetOrAdd((channel, productId), _ => new StreamOwner());
lock (streamOwner)
{
if (eventType == WebSocketEventType.Snapshot || streamOwner.Connection == null)
{
streamOwner.Connection = connection;
}
if (streamOwner.Connection == connection)
{
handler();
}
}
}

/// <summary>
/// Handle order update based on WS message
/// </summary>
Expand Down Expand Up @@ -489,21 +527,33 @@ protected override bool Subscribe(IEnumerable<Symbol> symbols)
{
OnMessage(new BrokerageMessageEvent(BrokerageMessageType.Error, "SubscriptionOnWSFeed", "Failed to subscribe on `user update` channels"));
}

SubscribeSymbolsOnDataChannels(GetSubscribed().ToList());
}, _cancellationTokenSourceReSubscription.Token);

return true;
}

/// <summary>
/// Ends current subscriptions
/// Ends the symbol subscription in the given connection
/// </summary>
public bool Unsubscribe(IEnumerable<Symbol> leanSymbols)
private bool Unsubscribe(IWebSocket webSocket, Symbol symbol)
{
SubscribeSymbolsOnDataChannels(leanSymbols.ToList(), WebSocketSubscriptionType.Unsubscribe);
var productId = _symbolMapper.GetBrokerageSymbol(symbol);
foreach (var streamOwner in _streamOwner.Where(x => x.Key.Item2 == productId || x.Key.Item2 + "C" == productId).Select(x => x.Value))
{
lock (streamOwner)
{
if (streamOwner.Connection == webSocket)
{
streamOwner.Connection = null;
}
}
}

return true;
if (!webSocket.IsOpen)
{
return true;
}
return SubscribeSymbolsOnDataChannels(new List<Symbol> { symbol }, WebSocketSubscriptionType.Unsubscribe, webSocket);
}

/// <summary>
Expand All @@ -516,7 +566,7 @@ public bool Unsubscribe(IEnumerable<Symbol> leanSymbols)
/// invoking the <see cref="ManageChannelSubscription"/> method with the appropriate parameters.
/// </remarks>
/// <seealso cref="ManageChannelSubscription"/>
private bool SubscribeSymbolsOnDataChannels(List<Symbol> symbols, WebSocketSubscriptionType subscriptionType = WebSocketSubscriptionType.Subscribe)
private bool SubscribeSymbolsOnDataChannels(List<Symbol> symbols, WebSocketSubscriptionType subscriptionType = WebSocketSubscriptionType.Subscribe, IWebSocket webSocket = null)
{
var products = symbols.Select(symbol => _symbolMapper.GetBrokerageSymbol(symbol)).ToList();

Expand All @@ -529,7 +579,7 @@ private bool SubscribeSymbolsOnDataChannels(List<Symbol> symbols, WebSocketSubsc
{
foreach (var chunkProduct in products.Chunk(20))
{
ManageChannelSubscription(subscriptionType, channel, chunkProduct.ToList());
ManageChannelSubscription(subscriptionType, channel, chunkProduct.ToList(), webSocket);
}
}

Expand All @@ -544,14 +594,16 @@ private bool SubscribeSymbolsOnDataChannels(List<Symbol> symbols, WebSocketSubsc
/// <param name="productIds">Optional list of product IDs associated with the subscription.</param>
/// <exception cref="ArgumentException"></exception>
/// <exception cref="InvalidOperationException"></exception>
private void ManageChannelSubscription(WebSocketSubscriptionType subscriptionType, string channel, List<string> productIds = null)
/// <param name="webSocket">The connection to use, defaults to the main one</param>
private void ManageChannelSubscription(WebSocketSubscriptionType subscriptionType, string channel, List<string> productIds = null, IWebSocket webSocket = null)
{
if (string.IsNullOrWhiteSpace(channel))
{
throw new ArgumentException($"{nameof(CoinbaseBrokerage)}.{nameof(ManageChannelSubscription)}: ChannelRequired:", nameof(channel));
}

if (!IsConnected)
webSocket ??= WebSocket;
if (!webSocket.IsOpen)
{
throw new InvalidOperationException($"{nameof(CoinbaseBrokerage)}.{nameof(ManageChannelSubscription)}: WebSocketMustBeConnected");
}
Expand All @@ -565,7 +617,27 @@ private void ManageChannelSubscription(WebSocketSubscriptionType subscriptionTyp

_webSocketRateLimit.WaitToProceed();

WebSocket.Send(json);
webSocket.Send(json);
}

/// <summary>
/// Creates a connection for data subscriptions
/// </summary>
private WebSocketClientWrapper CreateDataWebSocket()
{
var webSocket = new WebSocketClientWrapper();
webSocket.Open += (_, _) => Task.Factory.StartNew(() =>
{
try
{
ManageChannelSubscription(WebSocketSubscriptionType.Subscribe, CoinbaseWebSocketChannels.Heartbeats, webSocket: webSocket);
}
catch (Exception ex)
{
Log.Error(ex, $"{nameof(CoinbaseBrokerage)}.{nameof(CreateDataWebSocket)}: failed to subscribe to heartbeats");
}
});
return webSocket;
}

/// <summary>
Expand All @@ -585,6 +657,17 @@ private Symbol GetSimilarSymbolUSDC(string productIdUSD)
return _symbolMapper.GetLeanSymbol(productIdUSD.Split('-')[0] + "-USDC", SecurityType.Crypto, MarketName);
}

/// <summary>
/// The connection we follow for a data stream
/// </summary>
private class StreamOwner
{
/// <summary>
/// The connection, null if none yet or it unsubscribed
/// </summary>
public object Connection { get; set; }
}

/// <summary>
/// The fills already reported to lean for a brokerage order
/// </summary>
Expand Down
15 changes: 10 additions & 5 deletions QuantConnect.CoinbaseBrokerage/CoinbaseBrokerage.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -163,11 +163,16 @@ protected void Initialize(string webSocketUrl, string name, string privateKey, s
_coinbaseApi = new CoinbaseApi(_symbolMapper, algorithm?.Portfolio, name, privateKey, restApiUrl);
OrderProvider = orderProvider;

SubscriptionManager = new EventBasedDataQueueHandlerSubscriptionManager()
{
SubscribeImpl = (symbols, _) => SubscribeSymbolsOnDataChannels(symbols.ToList()),
UnsubscribeImpl = (symbols, _) => Unsubscribe(symbols)
};
SubscriptionManager = new BrokerageMultiWebSocketSubscriptionManager(
webSocketUrl,
MaximumSymbolsPerWebSocket,
0,
null,
CreateDataWebSocket,
(webSocket, symbol) => SubscribeSymbolsOnDataChannels(new List<Symbol> { symbol }, webSocket: webSocket),
Unsubscribe,
message => OnMessage(this, message),
TimeSpan.Zero);
}

#region IBrokerage
Expand Down
Loading