Filter out old trade data and don't emit invalid quotes (#17)

* Filter out old trade data and don't emit invalid quotes

* Filter repeated trade ids

QuantConnect.CoinbaseBrokerage/CoinbaseBrokerage.Messaging.cs

@@ -53,6 +53,11 @@ public partial class CoinbaseBrokerage
         /// </example>
         private readonly ConcurrentDictionary<Symbol, List<DefaultOrderBook>> _orderBooks = new();
 
+        /// <summary>
+        /// Sometimes coinbase likes to duplicate the trades, let's ignore old trade ids
+        /// </summary>
+        private readonly ConcurrentDictionary<Symbol, Tuple<long, DateTime>> _tradeIds = new();
+
         /// <summary>
         /// Represents a rate limiter for controlling the frequency of WebSocket operations.
         /// </summary>
@@ -273,6 +278,11 @@ private void Level2Snapshot(CoinbaseLevel2Event snapshotData)
 
                 orderBook.BestBidAskUpdated += OnBestBidAskUpdated;
 
+                if (orderBook.BestBidPrice == 0 && orderBook.BestAskPrice == 0)
+                {
+                    // nothing to emit, can happen with illiquid assets
+                    return;
+                }
                 EmitQuoteTick(orderBook.Symbol, orderBook.BestBidPrice, orderBook.BestBidSize, orderBook.BestAskPrice, orderBook.BestAskSize);
             }
         }
@@ -325,10 +335,21 @@ private void Level2Update(CoinbaseLevel2Event updateData)
 
         private void EmitTradeTick(CoinbaseMarketTradesEvent tradeUpdates)
         {
-            foreach (var trade in tradeUpdates.Trades)
+            // coinbase sends older data, as an update, seems they send the last 100 trades, so let's filter it out
+            // also order by time since they return in descending time and we want ascending
+            var dataFrontier = DateTime.UtcNow - TimeSpan.FromMinutes(5);
+            foreach (var trade in tradeUpdates.Trades.Where(x => x.Time.UtcDateTime > dataFrontier).OrderBy(x => x.Time))
             {
                 var symbol = _symbolMapper.GetLeanSymbol(trade.ProductId, SecurityType.Crypto, MarketName);
 
+                if (_tradeIds.TryGetValue(symbol, out var lastTradeData)
+                    // ignore old trade ids as long as they have an old timestamp too, just in case it restarted
+                    && lastTradeData.Item1 > trade.TradeId && lastTradeData.Item2 > trade.Time.UtcDateTime)
+                {
+                    continue;
+                }
+                _tradeIds[symbol] = new (trade.TradeId, trade.Time.UtcDateTime);
+
                 var tick = new Tick
                 {
                     Value = trade.Price.Value,

QuantConnect.CoinbaseBrokerage/Models/WebSocket/CoinbaseMarketTradesEvent.cs

@@ -29,7 +29,7 @@ public class CoinbaseMarketTradesEvent : WebSocketEvent
 public class Trade
 {
     [JsonProperty("trade_id")]
-    public string TradeId { get; set; }
+    public long TradeId { get; set; }
 
     [JsonProperty("product_id")]
     public string ProductId { get; set; }