Revert "Fix: OpenInterestProcessorManger (request data each day 8 AM)… (#31)

* Revert "Fix: OpenInterestProcessorManger (request data each day 8 AM) (#29)"

This reverts commit 51367b9.

* Minor unit test building fix

Author: jhonabreul

QuantConnect.Polygon.Tests/PolygonOpenInterestProcessorManagerTests.cs

@@ -14,7 +14,6 @@
  */
 
 using System;
-using System.Linq;
 using NUnit.Framework;
 using System.Threading;
 using QuantConnect.Data;
@@ -37,15 +36,23 @@ public class PolygonOpenInterestProcessorManagerTests : PolygonDataProviderBaseT
 
         private readonly ManualTimeProvider _timeProviderInstance = new();
 
-        private readonly Lock _locker = new();
+        private readonly EventBasedDataQueueHandlerSubscriptionManager _subscriptionManager = new()
+        {
+            SubscribeImpl = (symbols, _) => { return true; },
+            UnsubscribeImpl = (symbols, _) => { return true; }
+        };
+
+        private object _locker = new();
 
-        [Test]
-        public void GetOpenInterestOfOptionSymbolsByPolygonOpenInterestProcessorManager()
+        [TestCase("2024-09-16T09:30:59", true, Description = "Market: After Opening")]
+        [TestCase("2024-09-16T15:28:59", true, Description = "Market: Before Closing")]
+        [TestCase("2024-09-16T16:28:59", false, Description = "Market: Closed")]
+        public void GetOpenInterestInDifferentTimeExchangeTime(string mockDateTime, bool isShouldReturnData)
         {
-            var resetEvent = new ManualResetEvent(false);
+            var waitOneDelay = isShouldReturnData ? TimeSpan.FromSeconds(30) : TimeSpan.FromSeconds(5);
+            var resetEvent = new AutoResetEvent(false);
             var cancellationTokenSource = new CancellationTokenSource();
             var optionContractsConfigs = GetConfigs();
-            var amountOfConfigs = optionContractsConfigs.Count;
 
             var symbolOpenInterest = new ConcurrentDictionary<Symbol, decimal>();
             Action<BaseData> callback = (baseData) =>
@@ -59,42 +66,41 @@ public void GetOpenInterestOfOptionSymbolsByPolygonOpenInterestProcessorManager(
                 {
                     symbolOpenInterest[baseData.Symbol] = baseData.Value;
 
-                    if (symbolOpenInterest.Count == amountOfConfigs)
+                    if (symbolOpenInterest.Count > 5)
                     {
                         resetEvent.Set();
                     }
                 }
             };
 
-            _timeProviderInstance.SetCurrentTimeUtc(DateTime.UtcNow.ConvertFromUtc(TimeZones.NewYork).Date.AddHours(8).AddSeconds(-5).ConvertToUtc(TimeZones.NewYork));
-            var processor = new PolygonOpenInterestProcessorManager(_timeProviderInstance, _restApiClient, symbolMapper, dataAggregator, GetTickTime);
-
-            processor.AddSymbols([.. optionContractsConfigs.Select(x => x.Symbol)]);
-
             foreach (var config in optionContractsConfigs)
             {
+                _subscriptionManager.Subscribe(config);
                 ProcessFeed(
                     Subscribe(dataAggregator, config, (sender, args) => { }),
                     cancellationTokenSource.Token,
                     callback: callback
                     );
             }
 
-            // Internal delay to respect the 1-minute request interval for OpenInterest data.
-            resetEvent.WaitOne(TimeSpan.FromSeconds(70), cancellationTokenSource.Token);
-
-            Assert.Greater(symbolOpenInterest.Count, 0);
-
-            // Advance the current UTC time by 1 day to simulate the next day.
-            // This ensures that any new OpenInterest requests are validated against the updated time.
-            _timeProviderInstance.SetCurrentTimeUtc(DateTime.UtcNow.AddDays(1));
-
-            resetEvent.Reset();
-            symbolOpenInterest.Clear();
+            var mockDateTimeAfterOpenExchange = DateTime.Parse(mockDateTime).ConvertTo(TimeZones.NewYork, TimeZones.Utc);
+            _timeProviderInstance.SetCurrentTimeUtc(mockDateTimeAfterOpenExchange);
+            var processor = new PolygonOpenInterestProcessorManager(_timeProviderInstance, _restApiClient, symbolMapper, _subscriptionManager, dataAggregator, GetTickTime);
+            processor.ScheduleNextRun();
+            resetEvent.WaitOne(waitOneDelay, cancellationTokenSource.Token);
 
-            resetEvent.WaitOne(TimeSpan.FromSeconds(30), cancellationTokenSource.Token);
-
-            Assert.Greater(symbolOpenInterest.Count, 0);
+            if (isShouldReturnData)
+            {
+                Assert.Greater(symbolOpenInterest.Count, 0);
+                foreach (var (symbol, openInterest) in symbolOpenInterest)
+                {
+                    Assert.Greater(openInterest, 0);
+                }
+            }
+            else
+            {
+                Assert.Zero(symbolOpenInterest.Count);
+            }
 
             cancellationTokenSource.Cancel();
             cancellationTokenSource.Dispose();
@@ -106,7 +112,7 @@ protected override List<SubscriptionDataConfig> GetConfigs(Resolution resolution
         {
             var configs = new List<SubscriptionDataConfig>();
 
-            var expiryContractDate = new DateTime(2025, 05, 16);
+            var expiryContractDate = new DateTime(2024, 09, 20);
             var strikesAAPL = new decimal[] { 100m, 105m, 110m, 115m, 120m, 125m, 130m, 135m, 140m, 145m };
 
             foreach (var strike in strikesAAPL)

QuantConnect.Polygon/PolygonDataProvider.cs

@@ -58,8 +58,6 @@ public partial class PolygonDataProvider : IDataQueueHandler
 
         protected PolygonSubscriptionManager _subscriptionManager;
 
-        private PolygonOpenInterestProcessorManager _polygonOpenInterestProcessorManager;
-
         private List<ExchangeMapping> _exchangeMappings;
         private readonly PolygonSymbolMapper _symbolMapper = new();
         private readonly MarketHoursDatabase _marketHoursDatabase = MarketHoursDatabase.FromDataFolder();
@@ -152,17 +150,16 @@ private void Initialize(string apiKey, int maxSubscriptionsPerWebSocket, bool st
             // Initialize the exchange mappings
             _exchangeMappings = FetchExchangeMappings();
 
-            _polygonOpenInterestProcessorManager = new PolygonOpenInterestProcessorManager(TimeProvider, RestApiClient, _symbolMapper, _dataAggregator, GetTickTime);
-
             // Initialize the subscription manager if this instance is going to be used as a data queue handler
             if (streamingEnabled)
             {
                 _subscriptionManager = new PolygonSubscriptionManager(
                     _supportedSecurityTypes,
                     maxSubscriptionsPerWebSocket,
-                    (securityType) => new PolygonWebSocketClientWrapper(_apiKey, _symbolMapper, securityType, OnMessage),
-                    _polygonOpenInterestProcessorManager);
+                    (securityType) => new PolygonWebSocketClientWrapper(_apiKey, _symbolMapper, securityType, OnMessage));
             }
+            var openInterestManager = new PolygonOpenInterestProcessorManager(TimeProvider, RestApiClient, _symbolMapper, _subscriptionManager, _dataAggregator, GetTickTime);
+            openInterestManager.ScheduleNextRun();
         }
 
         #region IDataQueueHandler implementation
@@ -328,7 +325,10 @@ private void ProcessTrade(TradeMessage trade)
             var time = GetTickTime(symbol, trade.Timestamp);
             // TODO: Map trade.Conditions to Lean sale conditions
             var tick = new Tick(time, symbol, string.Empty, GetExchangeCode(trade.ExchangeID), trade.Size, trade.Price);
-            ProcessTickWithOpenInterestTick(tick);
+            lock (_dataAggregator)
+            {
+                _dataAggregator.Update(tick);
+            }
         }
 
         /// <summary>
@@ -342,30 +342,9 @@ private void ProcessQuote(QuoteMessage quote)
             // Note: Polygon's quotes have bid/ask exchange IDs, but Lean only has one exchange per tick. We'll use the bid exchange.
             var tick = new Tick(time, symbol, string.Empty, GetExchangeCode(quote.BidExchangeID),
                 quote.BidSize, quote.BidPrice, quote.AskSize, quote.AskPrice);
-            ProcessTickWithOpenInterestTick(tick);
-        }
-
-        /// <summary>
-        /// Processes a tick and its corresponding open interest tick, if available.
-        /// </summary>
-        /// <param name="mainTick">The primary tick (trade or quote) to process.</param>
-        /// <remarks>
-        /// This method retrieves the open interest tick corresponding to the symbol and time of the provided tick.
-        /// If an open interest tick is found, both the open interest tick and the main tick are passed to the data aggregator
-        /// in a thread-safe manner.
-        /// We do this to sync the open interest tick (which is one for the whole day) with the quote/trade tick.
-        /// </remarks>
-        private void ProcessTickWithOpenInterestTick(Tick mainTick)
-        {
-            var openInterest = _polygonOpenInterestProcessorManager.GetOpenInterestTick(mainTick.Symbol, mainTick.Time);
             lock (_dataAggregator)
             {
-                if (openInterest != null)
-                {
-                    _dataAggregator.Update(openInterest);
-                }
-
-                _dataAggregator.Update(mainTick);
+                _dataAggregator.Update(tick);
             }
         }