Skip to content

Commit 901bcc5

Browse files
Joseph-Matteo-ScorsoneJoseph-Matteo-Scorsone
committed
Authentication and message flow finalized. Consolidated data properly for data aggregator. Data is now recieved in OnData. All log.Trace's are not commented out, they will print to the screen.
1 parent 2882977 commit 901bcc5

3 files changed

Lines changed: 363 additions & 363 deletions

File tree

‎Demonstration.cs‎

Lines changed: 46 additions & 38 deletions
Original file line numberDiff line numberDiff line change
@@ -14,61 +14,69 @@
1414
*
1515
*/
1616

17-
using QuantConnect.Data;
18-
using QuantConnect.Util;
19-
using QuantConnect.Orders;
2017
using QuantConnect.Algorithm;
21-
using QuantConnect.DataSource;
2218
using QuantConnect.Data.Market;
19+
using QuantConnect.Interfaces;
20+
using QuantConnect;
21+
using QuantConnect.Data;
2322
using QuantConnect.Securities.Future;
23+
using QuantConnect.Util;
24+
using System;
2425

25-
namespace QuantConnect.DataLibrary.Tests
26+
namespace QuantConnect.Algorithm.CSharp
2627
{
27-
/// <summary>
28-
/// Example algorithm using the custom data type as a source of alpha
29-
/// </summary>
30-
public class CustomDataAlgorithm : QCAlgorithm
28+
public class DatabentoFuturesTestAlgorithm : QCAlgorithm
3129
{
32-
private Future _esFuture;
30+
private Future _es;
3331

34-
/// <summary>
35-
/// Initialise the data and resolution required, as well as the cash and start-end dates for your algorithm. All algorithms must initialized.
36-
/// </summary>
3732
public override void Initialize()
3833
{
39-
SetStartDate(2021, 10, 07); //Set Start Date
40-
SetEndDate(2021, 10, 11); //Set End Date
41-
_esFuture = AddFuture("ES");
34+
Log("Algorithm Initialize");
35+
36+
SetStartDate(2025, 10, 1);
37+
SetStartDate(2025, 10, 16);
38+
SetCash(100000);
4239

43-
_esFuture.SetFilter(0, 182);
40+
var exp = new DateTime(2025, 12, 19);
41+
var symbol = QuantConnect.Symbol.CreateFuture("ES", Market.CME, exp);
42+
_es = AddFutureContract(symbol, Resolution.Minute, true, 1, true);
43+
Log($"_es: {_es}");
4444
}
4545

46-
/// <summary>
47-
/// OnData event is the primary entry point for your algorithm. Each new data point will be pumped in here.
48-
/// </summary>
49-
/// <param name="slice">Slice object keyed by symbol containing the stock data</param>
5046
public override void OnData(Slice slice)
5147
{
52-
Log($"_esFuture Open: {_esFuture.Open}");
53-
Log($"_esFuture High: {_esFuture.High}");
54-
Log($"_esFuture Low: {_esFuture.Low}");
55-
Log($"_esFuture Close: {_esFuture.Close}");
56-
57-
if (!Portfolio.Invested)
48+
if (!slice.HasData)
5849
{
59-
SetHoldings(_esFuture.Symbol, 1);
50+
Log("Slice has no data");
51+
return;
6052
}
61-
}
62-
63-
/// <summary>
64-
/// Order fill event handler. On an order fill update the resulting information is passed to this method.
65-
/// </summary>
66-
/// <param name="orderEvent">Order event details containing details of the events</param>
67-
public override void OnOrderEvent(OrderEvent orderEvent)
68-
{
69-
if (orderEvent.Status.IsFill())
53+
54+
Log($"OnData: Slice has {slice.Count} data points");
55+
56+
// For Tick resolution, check Ticks collection
57+
if (slice.Ticks.ContainsKey(_es.Symbol))
58+
{
59+
var ticks = slice.Ticks[_es.Symbol];
60+
Log($"Received {ticks.Count} ticks for {_es.Symbol}");
61+
62+
foreach (var tick in ticks)
63+
{
64+
if (tick.TickType == TickType.Trade)
65+
{
66+
Log($"Trade Tick - Price: {tick.Price}, Quantity: {tick.Quantity}, Time: {tick.Time}");
67+
}
68+
else if (tick.TickType == TickType.Quote)
69+
{
70+
Log($"Quote Tick - Bid: {tick.BidPrice}x{tick.BidSize}, Ask: {tick.AskPrice}x{tick.AskSize}, Time: {tick.Time}");
71+
}
72+
}
73+
}
74+
75+
// These won't have data for Tick resolution
76+
if (slice.Bars.ContainsKey(_es.Symbol))
7077
{
71-
Debug($"Purchased Stock: {orderEvent.Symbol}");
78+
var bar = slice.Bars[_es.Symbol];
79+
Log($"Bar - O:{bar.Open} H:{bar.High} L:{bar.Low} C:{bar.Close} V:{bar.Volume}");
7280
}
7381
}
7482
}

‎QuantConnect.DataBento/DataBentoDataProvider.cs‎

Lines changed: 86 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -23,6 +23,7 @@
2323
using QuantConnect.Configuration;
2424
using QuantConnect.Logging;
2525
using QuantConnect.Packets;
26+
using QuantConnect.Securities;
2627
using System.Threading.Tasks;
2728
using QuantConnect.Data.Market;
2829

@@ -67,7 +68,12 @@ public class DataBentoProvider : IDataQueueHandler
6768
private bool _unsupportedDataTypeMessageLogged;
6869
private bool _potentialUnsupportedResolutionMessageLogged;
6970

71+
private bool _sessionStarted = false;
72+
private readonly object _sessionLock = new object();
7073

74+
private readonly MarketHoursDatabase _marketHoursDatabase = MarketHoursDatabase.FromDataFolder();
75+
private readonly System.Collections.Concurrent.ConcurrentDictionary<Symbol, DateTimeZone> _symbolExchangeTimeZones = new();
76+
7177
/// <summary>
7278
/// Returns true if we're currently connected to the Data Provider
7379
/// </summary>
@@ -124,12 +130,30 @@ private void Initialize()
124130
if (_client?.IsConnected == true)
125131
{
126132
Log.Trace($"DataBentoProvider.SubscribeImpl(): Client is connected, attempting async subscribe for {symbol}");
127-
Task.Run(() =>
133+
Task.Run(async () =>
128134
{
129135
var success = _client.Subscribe(config.Symbol, config.Resolution, config.TickType);
130136
if (success)
131137
{
132138
Log.Trace($"DataBentoProvider.SubscribeImpl(): Successfully subscribed to {config.Symbol}");
139+
140+
// Start session after first successful subscription
141+
lock (_sessionLock)
142+
{
143+
if (!_sessionStarted)
144+
{
145+
Log.Trace("DataBentoProvider.SubscribeImpl(): Starting DataBento session to receive data");
146+
_sessionStarted = _client.StartSession();
147+
if (_sessionStarted)
148+
{
149+
Log.Trace("DataBentoProvider.SubscribeImpl(): Session started successfully - data should begin flowing");
150+
}
151+
else
152+
{
153+
Log.Error("DataBentoProvider.SubscribeImpl(): Failed to start session");
154+
}
155+
}
156+
}
133157
}
134158
else
135159
{
@@ -366,19 +390,58 @@ private bool IsSupported(SecurityType securityType, Type dataType, TickType tick
366390

367391
return true;
368392
}
393+
394+
/// <summary>
395+
/// Converts the given UTC time into the symbol security exchange time zone
396+
/// </summary>
397+
private DateTime GetTickTime(Symbol symbol, DateTime utcTime)
398+
{
399+
var exchangeTimeZone = _symbolExchangeTimeZones.GetOrAdd(symbol, sym =>
400+
{
401+
if (_marketHoursDatabase.TryGetEntry(sym.ID.Market, sym, sym.SecurityType, out var entry))
402+
{
403+
return entry.ExchangeHours.TimeZone;
404+
}
405+
// Futures default to Chicago
406+
return TimeZones.Chicago;
407+
});
408+
409+
return utcTime.ConvertFromUtc(exchangeTimeZone);
410+
}
411+
369412
// <summary>
370413
/// Handles data received from the live client
371414
/// </summary>
372415
private void OnDataReceived(object? sender, BaseData data)
373416
{
374-
Log.Trace($"DataBentoProvider.OnDataReceived(): Received data: {data}");
375417
try
376418
{
377-
_dataAggregator.Update(data);
419+
if (data is Tick tick)
420+
{
421+
tick.Time = GetTickTime(tick.Symbol, tick.Time);
422+
_dataAggregator.Update(tick);
423+
424+
Log.Trace($"DataBentoProvider.OnDataReceived(): Updated tick - Symbol: {tick.Symbol}, " +
425+
$"TickType: {tick.TickType}, Price: {tick.Value}, Quantity: {tick.Quantity}");
426+
}
427+
else if (data is TradeBar tradeBar)
428+
{
429+
tradeBar.Time = GetTickTime(tradeBar.Symbol, tradeBar.Time);
430+
tradeBar.EndTime = GetTickTime(tradeBar.Symbol, tradeBar.EndTime);
431+
_dataAggregator.Update(tradeBar);
432+
433+
Log.Trace($"DataBentoProvider.OnDataReceived(): Updated TradeBar - Symbol: {tradeBar.Symbol}, " +
434+
$"O:{tradeBar.Open} H:{tradeBar.High} L:{tradeBar.Low} C:{tradeBar.Close} V:{tradeBar.Volume}");
435+
}
436+
else
437+
{
438+
data.Time = GetTickTime(data.Symbol, data.Time);
439+
_dataAggregator.Update(data);
440+
}
378441
}
379442
catch (Exception ex)
380443
{
381-
Log.Error($"DataBentoProvider.OnDataReceived(): Error updating data aggregator: {ex.Message}");
444+
Log.Error($"DataBentoProvider.OnDataReceived(): Error updating data aggregator: {ex.Message}\n{ex.StackTrace}");
382445
}
383446
}
384447

@@ -391,13 +454,32 @@ private void OnConnectionStatusChanged(object? sender, bool isConnected)
391454

392455
if (isConnected)
393456
{
457+
// Reset session flag on reconnection
458+
lock (_sessionLock)
459+
{
460+
_sessionStarted = false;
461+
}
462+
394463
// Resubscribe to all active subscriptions
395464
Task.Run(() =>
396465
{
397466
foreach (var config in _activeSubscriptionConfigs)
398467
{
399468
_client.Subscribe(config.Symbol, config.Resolution, config.TickType);
400469
}
470+
471+
// Start session after resubscribing
472+
if (_activeSubscriptionConfigs.Any())
473+
{
474+
lock (_sessionLock)
475+
{
476+
if (!_sessionStarted)
477+
{
478+
Log.Trace("DataBentoProvider.OnConnectionStatusChanged(): Starting session after reconnection");
479+
_sessionStarted = _client.StartSession();
480+
}
481+
}
482+
}
401483
});
402484
}
403485
}

0 commit comments

Comments
 (0)