Feature: support Fair Market Value (#35)

* feat: handle new FairMarketValue WS msg
refactor: use BaseClass with repeat properties

* feat: additional logs

* feat: add subscription on Fair Market Value

* refactor: subscribe on FMV if business always

* feat: use business only fro trade

* refactor: log message in ProcessMessage

* test:feat: fair market value as Tick or TradeBar

* feat: index value updates

* fix: miss subscribe on OI updates

* feat: validate OI and fair market value

* feat: implement aggregator minutes

* feat: use consolidator in smart way
refactor: use enum eventType instead of Prefixes naming for ws channels

* remove: `SetUsingAggregates` from AggrManager

* refactor: use EventType in all subscription methods

* refactor: reset eventType in AggregatorManager

* feat: implement license type logic

* fix: ParseLicenseType more implicit use Individual license type

* fix: use ws endpoint url by licenseType explicitly

* fix: reset `_usingEventType` by default in finally block

* refactor: subscribe logic in PolygonSubscriptionManager

* feat: use _subscriptionsDataConfigs with specific configs data to reuse and another part of app

* feat: handle not unavailable connection exception

* feat: add condition to catch(PolygonAuthenticationException) in Subscribe

Author: Romazes

QuantConnect.Polygon.Tests/PolygonDataProviderBaseTests.cs

@@ -240,7 +240,7 @@ protected static Task StartEnumeratorProcessing(
             }, TaskContinuationOptions.OnlyOnFaulted);
         }
 
-        protected SubscriptionDataConfig GetSubscriptionDataConfig<T>(Symbol symbol, Resolution resolution)
+        public static SubscriptionDataConfig GetSubscriptionDataConfig<T>(Symbol symbol, Resolution resolution)
         {
             return new SubscriptionDataConfig(
                 typeof(T),

QuantConnect.Polygon.Tests/PolygonDataProviderFairMarketValueTests.cs

@@ -0,0 +1,197 @@
+/*
+ * QUANTCONNECT.COM - Democratizing Finance, Empowering Individuals.
+ * Lean Algorithmic Trading Engine v2.0. Copyright 2014 QuantConnect Corporation.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+*/
+
+using System;
+using System.Linq;
+using NUnit.Framework;
+using System.Threading;
+using QuantConnect.Data;
+using QuantConnect.Tests;
+using QuantConnect.Logging;
+using QuantConnect.Data.Market;
+using QuantConnect.Configuration;
+using System.Collections.Generic;
+using QuantConnect.Lean.Engine.DataFeeds;
+using QuantConnect.Lean.Engine.DataFeeds.Enumerators;
+
+namespace QuantConnect.Lean.DataSource.Polygon.Tests
+{
+    [TestFixture]
+    [Explicit("Tests are dependent on network and take long")]
+    public class PolygonDataProviderFairMarketValueTests
+    {
+        private List<SubscriptionDataConfig> GetConfigs(Resolution resolution)
+        {
+            var symbols = new Symbol[]
+            {
+                Symbols.AAPL,
+                Symbols.MSFT,
+                Symbol.CreateOption(Symbols.AAPL, Symbols.AAPL.ID.Market, SecurityType.Option.DefaultOptionStyle(), OptionRight.Call, 227.5m, new(2025, 08, 29)),
+                Symbol.CreateOption(Symbols.MSFT, Symbols.MSFT.ID.Market, SecurityType.Option.DefaultOptionStyle(), OptionRight.Call, 502.5m, new(2025, 08, 29)),
+                //Symbols.SPX,
+                Symbol.CreateOption(Symbols.SPX, "SPXW", Symbols.SPX.ID.Market, SecurityType.IndexOption.DefaultOptionStyle(), OptionRight.Call, 6485m, new(2025, 08, 29))
+            };
+
+            return [.. symbols.SelectMany(symbol => GetSubscriptionDataConfigs(symbol, resolution))];
+        }
+
+        [TestCase(Resolution.Tick, 10)]
+        [TestCase(Resolution.Second, 5)]
+        [TestCase(Resolution.Minute, 1)]
+        public void StreamDataOnDifferentSecuritiesTypes(Resolution resolution, int expectedReceiveAmountData)
+        {
+            var mockDateTimeAfterOpenExchange = DateTime.UtcNow.Date.AddHours(9).AddMinutes(30).AddSeconds(59).ConvertToUtc(TimeZones.NewYork);
+            TestablePolygonDataProvider.TimeProviderInstance = new ManualTimeProvider(mockDateTimeAfterOpenExchange);
+            using var polygon = new TestablePolygonDataProvider(Config.Get("polygon-api-key"), Config.Get("polygon-license-type"));
+
+            var configs = GetConfigs(resolution);
+
+            var receivedData = new Dictionary<Symbol, List<BaseData>>();
+
+            var resetEvent = new AutoResetEvent(false);
+            var receivedAllData = default(bool);
+
+            foreach (var config in configs)
+            {
+                receivedData[config.Symbol] = [];
+
+                polygon.Subscribe(config, (sender, args) =>
+                {
+                    if (args is NewDataAvailableEventArgs newData && newData.DataPoint is BaseData bd)
+                    {
+                        Log.Trace($"{bd}. Time span: {bd.Time} - {bd.EndTime}");
+                        lock (receivedData)
+                        {
+                            receivedData[bd.Symbol].Add(bd);
+
+                            if ((bd is Tick t && t.TickType == TickType.OpenInterest) || bd is OpenInterest)
+                            {
+                                // Prevent from repeating request
+                                TestablePolygonDataProvider.TimeProviderInstance.SetCurrentTimeUtc(DateTime.UtcNow);
+                            }
+
+                            if (!receivedAllData && receivedData.Values.All(x => x.Count >= expectedReceiveAmountData))
+                            {
+                                receivedAllData = true;
+                                resetEvent.Set();
+                            }
+                        }
+                    }
+                });
+            }
+
+            Assert.IsTrue(resetEvent.WaitOne(TimeSpan.FromSeconds(120)));
+
+            Log.Trace("Unsubscribing symbols");
+            foreach (var config in configs)
+            {
+                polygon.Unsubscribe(config);
+            }
+
+            resetEvent.WaitOne(TimeSpan.FromSeconds(10));
+
+            var dataSummary = receivedData.ToDictionary(pair => pair.Key, pair => pair.Value.Count);
+            Log.Trace($"Received {receivedData.Count} symbols with data points -> {string.Join(", ", dataSummary.Select(kvp => $"{kvp.Key}: {kvp.Value}"))}");
+
+            Assert.IsTrue(receivedData.Values.All(d => d.Count >= expectedReceiveAmountData));
+
+            foreach (var (symbol, baseData) in receivedData.Where(x => x.Key.SecurityType.IsOption()))
+            {
+                var openInterestDetected = default(bool);
+                foreach (var data in baseData)
+                {
+                    if (data is Tick t && t.TickType == TickType.OpenInterest)
+                    {
+                        openInterestDetected = true;
+                        break;
+                    }
+                }
+                Assert.IsTrue(openInterestDetected);
+            }
+
+            foreach (var (symbol, baseData) in receivedData)
+            {
+                foreach (var data in baseData)
+                {
+                    switch (data)
+                    {
+                        case OpenInterest oi:
+                            Assert.Greater(oi.Value, 0m);
+                            break;
+                        case Tick t:
+                            switch (t.TickType)
+                            {
+                                case TickType.OpenInterest:
+                                    Assert.Greater(t.Value, 0m);
+                                    break;
+                                case TickType.Trade:
+                                    Assert.Greater(t.LastPrice, 0m);
+                                    Assert.AreEqual(t.Quantity, 0m);
+                                    break;
+                                default:
+                                    throw new NotSupportedException();
+                            }
+                            break;
+                        case TradeBar tb:
+                            Assert.Greater(tb.Open, 0m);
+                            Assert.Greater(tb.High, 0m);
+                            Assert.Greater(tb.Low, 0m);
+                            Assert.Greater(tb.Close, 0m);
+                            if (resolution >= Resolution.Minute)
+                            {
+                                // For 1-minute or higher aggregates, volume is expected (may be zero).
+                                // Example: {"ev":"AM","sym":"O:MSFT250829C00502500","v":0,"av":358,...}
+                                Assert.GreaterOrEqual(tb.Volume, 0m);
+                            }
+                            else
+                            {
+                                // For tick/fmv data, volume is not reported and should remain zero
+                                Assert.AreEqual(tb.Volume, 0m);
+                            }
+                            break;
+                        default:
+                            throw new NotSupportedException();
+                    }
+                }
+            }
+        }
+
+        private static List<SubscriptionDataConfig> GetSubscriptionDataConfigs(Symbol symbol, Resolution resolution)
+        {
+            var subs = new List<SubscriptionDataConfig>();
+            if (resolution == Resolution.Tick)
+            {
+                subs.Add(new SubscriptionDataConfig(PolygonDataProviderBaseTests.GetSubscriptionDataConfig<Tick>(symbol, resolution), tickType: TickType.Trade));
+
+                if (symbol.SecurityType.IsOption())
+                {
+                    subs.Add(new SubscriptionDataConfig(PolygonDataProviderBaseTests.GetSubscriptionDataConfig<Tick>(symbol, resolution), tickType: TickType.OpenInterest));
+                }
+
+                return subs;
+            }
+
+            subs.Add(PolygonDataProviderBaseTests.GetSubscriptionDataConfig<TradeBar>(symbol, resolution));
+
+            if (symbol.SecurityType.IsOption())
+            {
+                subs.Add(PolygonDataProviderBaseTests.GetSubscriptionDataConfig<OpenInterest>(symbol, resolution));
+            }
+
+            return subs;
+        }
+    }
+}

QuantConnect.Polygon.Tests/PolygonOpenInterestProcessorManagerTests.cs

@@ -33,11 +33,9 @@ public class PolygonOpenInterestProcessorManagerTests : PolygonDataProviderBaseT
 
         private readonly PolygonSymbolMapper symbolMapper = new();
 
-        private readonly PolygonAggregationManager dataAggregator = new();
-
         private readonly ManualTimeProvider _timeProviderInstance = new();
 
-        private readonly EventBasedDataQueueHandlerSubscriptionManager _subscriptionManager = new()
+        private readonly PolygonSubscriptionManager _subscriptionManager = new([], 0, null)
         {
             SubscribeImpl = (symbols, _) => { return true; },
             UnsubscribeImpl = (symbols, _) => { return true; }
@@ -53,6 +51,7 @@ public void GetOpenInterestWithHugeAmountSymbols()
             var resetEvent = new AutoResetEvent(false);
             var cancellationTokenSource = new CancellationTokenSource();
             var _activeEnumerators = new ConcurrentDictionary<Symbol, IEnumerator<BaseData>>();
+            var dataAggregator = new PolygonAggregationManager(_subscriptionManager);
 
             using var polygon = new PolygonDataProvider(ApiKey);
 

QuantConnect.Polygon.Tests/TestablePolygonDataProvider.cs

@@ -14,15 +14,27 @@
  * limitations under the License.
 */
 
+using QuantConnect.Lean.Engine.DataFeeds;
+using System;
+
 namespace QuantConnect.Lean.DataSource.Polygon.Tests
 {
     public class TestablePolygonDataProvider : PolygonDataProvider
     {
+        public static ManualTimeProvider TimeProviderInstance = new(DateTime.UtcNow);
+
+        protected override ITimeProvider TimeProvider => TimeProviderInstance;
+
         public PolygonSubscriptionManager SubscriptionManager => _subscriptionManager;
 
         public TestablePolygonDataProvider(string apiKey, int maxSubscriptionsPerWebSocket)
             : base(apiKey, maxSubscriptionsPerWebSocket)
         {
         }
+
+        public TestablePolygonDataProvider(string apiKey, string licenseType, bool streamingEnabled = true)
+            : base(apiKey, streamingEnabled, licenseType)
+        {
+        }
     }
 }

QuantConnect.Polygon.Tests/config.json

@@ -7,7 +7,7 @@
   "job-organization-id": "",
 
   "polygon-api-key": "",
-  "polygon-subscription-plan": "Advanced",
+  "polygon-license-type": "Individual",
   "polygon-api-url": "https://api.polygon.io",
   "polygon-ws-url": "wss://socket.polygon.io",
   // For launchpad:

QuantConnect.Polygon/Enum/EventType.cs

@@ -0,0 +1,58 @@
+/*
+ * QUANTCONNECT.COM - Democratizing Finance, Empowering Individuals.
+ * Lean Algorithmic Trading Engine v2.0. Copyright 2014 QuantConnect Corporation.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+*/
+
+namespace QuantConnect.Lean.DataSource.Polygon;
+
+/// <summary>
+/// Defines the types of market data events that can be streamed via WebSocket,
+/// covering various levels of aggregation and security types.
+/// </summary>
+public enum EventType
+{
+    /// <summary>
+    /// No event type. Used when no subscription is active or the event is undefined.
+    /// </summary>
+    None = 0,
+
+    /// <summary>
+    /// Minute-level aggregated OHLC (Open, High, Low, Close) and volume data.
+    /// </summary>
+    AM = 1,
+
+    /// <summary>
+    /// Second-level aggregated OHLC (Open, High, Low, Close) and volume data.
+    /// </summary>
+    A = 2,
+
+    /// <summary>
+    /// Tick-by-tick trade data for supported securities.
+    /// </summary>
+    T = 3,
+
+    /// <summary>
+    /// National Best Bid and Offer (NBBO) quote data for supported securities.
+    /// </summary>
+    Q = 4,
+
+    /// <summary>
+    /// Real-time Fair Market Value (FMV) data for supported securities.
+    /// </summary>
+    FMV = 5,
+
+    /// <summary>
+    /// Real-time index value updates for specified indexes.
+    /// </summary>
+    V = 6
+}

QuantConnect.Polygon/Enum/LicenseType.cs

@@ -0,0 +1,32 @@
+/*
+ * QUANTCONNECT.COM - Democratizing Finance, Empowering Individuals.
+ * Lean Algorithmic Trading Engine v2.0. Copyright 2014 QuantConnect Corporation.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+*/
+
+namespace QuantConnect.Lean.DataSource.Polygon;
+
+/// <summary>
+/// Defines the available license types for a Polygon subscription.
+/// </summary>
+public enum LicenseType
+{
+    /// <summary>
+    /// Represents an individual subscription plan, intended for single users.
+    /// </summary>
+    Individual = 0,
+
+    /// <summary>
+    /// Represents a business subscription plan, intended for organizations or commercial use.
+    /// </summary>
+    Business = 1
+}

QuantConnect.Polygon/Exception/UnsupportedTickTypeForLicenseException.cs

@@ -0,0 +1,34 @@
+/*
+ * QUANTCONNECT.COM - Democratizing Finance, Empowering Individuals.
+ * Lean Algorithmic Trading Engine v2.0. Copyright 2014 QuantConnect Corporation.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+*/
+
+namespace QuantConnect.Lean.DataSource.Polygon;
+
+/// <summary>
+/// Exception thrown when a tick type is not supported for the current Polygon license.
+/// </summary>
+public class UnsupportedTickTypeForLicenseException : Exception
+{
+    /// <summary>
+    /// Initializes a new instance of the <see cref="UnsupportedTickTypeForLicenseException"/> class
+    /// with the specified tick type and license type.
+    /// </summary>
+    /// <param name="tickType">The tick type that was requested (e.g., Trade, Quote).</param>
+    /// <param name="licenseType">The license type of the current Polygon subscription (Individual or Business).</param>
+    public UnsupportedTickTypeForLicenseException(string tickType, LicenseType licenseType)
+        : base($"TickType '{tickType}' is not supported for license type '{licenseType}'.")
+    {
+    }
+}