Got rid of my data type to use QuantConnects TradeBar and QuoteBar and Tick. Subscribe and Unsubscribe are now sync and return bools for success. GetHistory is the default implementation, no override. Fixed the wrong gateway url.

Author: Joseph-Matteo-Scorsone

QuantConnect.DataBento/DataBentoDataDownloader.cs

@@ -21,6 +21,7 @@
 using System.Globalization;
 using System.Collections.Generic;
 using CsvHelper;
+using CsvHelper.Configuration.Attributes;
 using QuantConnect.Data;
 using QuantConnect.Data.Market;
 using QuantConnect.Util;
@@ -41,6 +42,7 @@ public class DataBentoDataDownloader : IDataProvider
 
         /// <inheritdoc cref="MarketHoursDatabase" />
         private readonly MarketHoursDatabase _marketHoursDatabase;
+        private const decimal PriceScaleFactor = 1e-9m;
 
         /// <summary>
         /// Initializes a new instance of the <see cref="DataBentoDataDownloader"/>
@@ -104,32 +106,68 @@ public IEnumerable<BaseData> Get(DataDownloaderGetParameters parameters)
             using var reader = new StreamReader(stream);
             using var csv = new CsvReader(reader, CultureInfo.InvariantCulture);
 
-            // base data conversion
-            foreach (var record in csv.GetRecords<DatabentoTrade>())
+            if (tickType == TickType.Trade)
             {
-                if (resolution == Resolution.Tick)
+                foreach (var record in csv.GetRecords<DatabentoBar>())
                 {
-                    yield return new Tick
+                    if (resolution == Resolution.Tick)
+                    {
+                        yield return new Tick
+                        {
+                            Time = record.Timestamp,
+                            Symbol = symbol,
+                            Value = record.Price,
+                            Quantity = record.Size
+                        };
+                    }
+                    else
                     {
-                        Time = record.Timestamp,
-                        Symbol = symbol,
-                        Value = record.Price,
-                        Quantity = record.Size
-                    };
+                        yield return new TradeBar
+                        {
+                            Symbol = symbol,
+                            Time = record.Timestamp,
+                            Open = record.Open,
+                            High = record.High,
+                            Low = record.Low,
+                            Close = record.Close,
+                            Volume = record.Volume
+                        };
+                    }
                 }
-                else
+            }
+            else if (tickType == TickType.Quote)
+            {
+                foreach (var record in csv.GetRecords<DatabentoQuote>())
                 {
-                    yield return new DataBentoDataType
+                    var bidPrice = record.BidPrice * PriceScaleFactor;
+                    var askPrice = record.AskPrice * PriceScaleFactor;
+
+                    if (resolution == Resolution.Tick)
+                    {
+                        yield return new Tick
+                        {
+                            Time = record.Timestamp,
+                            Symbol = symbol,
+                            AskPrice = askPrice,
+                            BidPrice = bidPrice,
+                            AskSize = record.AskSize,
+                            BidSize = record.BidSize,
+                            TickType = TickType.Quote
+                        };
+                    }
+                    else
                     {
-                        Symbol = symbol,
-                        Time = record.Timestamp,
-                        Open = record.Open,
-                        High = record.High,
-                        Low = record.Low,
-                        Close = record.Close,
-                        Volume = record.Volume,
-                        Value = record.Close
-                    };
+                        var bidBar = new Bar(bidPrice, bidPrice, bidPrice, bidPrice);
+                        var askBar = new Bar(askPrice, askPrice, askPrice, askPrice);
+                        yield return new QuoteBar(
+                            record.Timestamp,
+                            symbol,
+                            bidBar,
+                            record.BidSize,
+                            askBar,
+                            record.AskSize
+                        );
+                    }
                 }
             }
         }
@@ -147,31 +185,33 @@ public void Dispose()
         /// </summary>
         private string GetSchema(Resolution resolution, TickType tickType)
         {
-            if (resolution == Resolution.Tick && tickType == TickType.Trade)
-                return "trades";
-
-            if (resolution == Resolution.Tick && tickType == TickType.Quote)
-                return "mbp-1"; // top of book
-
-            if (resolution == Resolution.Second)
-                return "ohlcv-1s";
-
-            if (resolution == Resolution.Minute)
-                return "ohlcv-1m";
-
-            if (resolution == Resolution.Hour)
-                return "ohlcv-1h";
-
-            if (resolution == Resolution.Daily)
-                return "ohlcv-1d";
+            if (tickType == TickType.Trade)
+            {
+                if (resolution == Resolution.Tick)
+                    return "trades";
+                if (resolution == Resolution.Second)
+                    return "ohlcv-1s";
+                if (resolution == Resolution.Minute)
+                    return "ohlcv-1m";
+                if (resolution == Resolution.Hour)
+                    return "ohlcv-1h";
+                if (resolution == Resolution.Daily)
+                    return "ohlcv-1d";
+            }
+            else if (tickType == TickType.Quote)
+            {
+                // top of book
+                if (resolution == Resolution.Tick || resolution == Resolution.Second || resolution == Resolution.Minute || resolution == Resolution.Hour || resolution == Resolution.Daily)
+                    return "mbp-1";
+            }
 
             throw new NotSupportedException($"Unsupported resolution {resolution} / {tickType}");
         }
 
         /// <summary>
         /// Map Lean Symbol to Databento symbol string for continous
         /// </summary>
-        private string MapSymbol(Symbol symbol)
+        private static string MapSymbol(Symbol symbol)
         {
             if (symbol.SecurityType == SecurityType.Future)
             {
@@ -183,7 +223,7 @@ private string MapSymbol(Symbol symbol)
 
 
         /// Class for parsing trade data from Databento
-        private class DatabentoTrade
+        private class DatabentoBar
         {
             public DateTime Timestamp { get; set; }
             public decimal Price { get; set; }
@@ -194,6 +234,19 @@ private class DatabentoTrade
             public decimal Close { get; set; }
             public decimal Volume { get; set; }
         }
+
+        private class DatabentoQuote
+        {
+            [Name("ts_event")]
+            public DateTime Timestamp { get; set; }
+            [Name("bid_px_00")]
+            public long BidPrice { get; set; }
+            [Name("bid_sz_00")]
+            public int BidSize { get; set; }
+            [Name("ask_px_00")]
+            public long AskPrice { get; set; }
+            [Name("ask_sz_00")]
+            public int AskSize { get; set; }
+        }
     }
 }
-

QuantConnect.DataBento/DataBentoDataProvider.cs

@@ -117,36 +117,6 @@ private void Initialize()
             });
         }
 
-        /// <inheritdoc cref="HistoryProviderBase.GetHistory(IEnumerable{HistoryRequest}, DateTimeZone)"/>
-        public override IEnumerable<Slice> GetHistory(IEnumerable<HistoryRequest> requests, DateTimeZone sliceTimeZone)
-        {
-            // Create subscription objects from the configs
-            var subscriptions = new List<Subscription>();
-            foreach (var request in requests)
-            {
-                // Retrieve the history for the current request
-                var history = GetHistory(request);
-
-                if (history == null)
-                {
-                    // If history is null, it indicates that the request contains wrong parameters
-                    // Handle the case where the request parameters are incorrect
-                    continue;
-                }
-
-                var subscription = CreateSubscription(request, history);
-                subscriptions.Add(subscription);
-            }
-
-            // Validate that at least one subscription is valid; otherwise, return null
-            if (subscriptions.Count == 0)
-            {
-                return null;
-            }
-
-            return CreateSliceEnumerableFromSubscriptions(subscriptions, sliceTimeZone);
-        }
-
         /// <summary>
         /// Subscribe to the specified configuration
         /// </summary>
@@ -166,9 +136,9 @@ public IEnumerator<BaseData> Subscribe(SubscriptionDataConfig dataConfig, EventH
             // Subscribe to live data if client is connected
             if (_client?.IsConnected == true)
             {
-                Task.Run(async () =>
+                Task.Run(() =>
                 {
-                    var success = await _client.Subscribe(dataConfig.Symbol, dataConfig.Resolution);
+                    var success = _client.Subscribe(dataConfig.Symbol, dataConfig.Resolution);
                     if (!success)
                     {
                         Log.Error($"DataBentoProvider.Subscribe(): Failed to subscribe to live data for {dataConfig.Symbol}");
@@ -191,9 +161,9 @@ public void Unsubscribe(SubscriptionDataConfig dataConfig)
             // Unsubscribe from live data if client is connected
             if (_client?.IsConnected == true)
             {
-                Task.Run(async () =>
+                Task.Run(() =>
                 {
-                    await _client.Unsubscribe(dataConfig.Symbol);
+                    _client.Unsubscribe(dataConfig.Symbol);
                 });
             }
         }
@@ -369,11 +339,11 @@ private void OnConnectionStatusChanged(object sender, bool isConnected)
             if (isConnected)
             {
                 // Resubscribe to all active subscriptions
-                Task.Run(async () =>
+                Task.Run(() =>
                 {
                     foreach (var config in _subscriptionManager.Subscriptions)
                     {
-                        await _client.Subscribe(config.Symbol, config.Resolution, config.TickType);
+                        _client.Subscribe(config.Symbol, config.Resolution);
                     }
                 });
             }