Add Python version of IndicatorVolatilityModelAlgorithm

Algorithm.CSharp/IndicatorVolatilityModelAlgorithm.cs

@@ -128,7 +128,7 @@ private IIndicator UpdateIndicator(Security security, TradeBar bar)
         /// <summary>
         /// This is used by the regression test system to indicate which languages this algorithm is written in.
         /// </summary>
-        public List<Language> Languages { get; } = new() { Language.CSharp };
+        public List<Language> Languages { get; } = new() { Language.CSharp, Language.Python };
 
         /// <summary>
         /// Data Points count of all timeslices of algorithm

Algorithm.Python/IndicatorVolatilityModelAlgorithm.py

@@ -0,0 +1,85 @@
+# 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.
+
+from AlgorithmImports import *
+
+### <summary>
+### Algorithm illustrating the usage of the IndicatorVolatilityModel and
+### how to handle splits and dividends to avoid price discontinuities
+### </summary>
+class IndicatorVolatilityModelAlgorithm(QCAlgorithm):
+
+    _indicator_periods = 7
+
+    _data_normalization_mode = DataNormalizationMode.RAW
+
+    def initialize(self):
+        self.set_start_date(2014, 1, 1)
+        self.set_end_date(2014, 12, 31)
+        self.set_cash(100000)
+
+        equity = self.add_equity("AAPL", Resolution.DAILY, data_normalization_mode=self._data_normalization_mode)
+        self._aapl = equity.symbol
+
+        std = StandardDeviation(self._indicator_periods)
+        mean = SimpleMovingAverage(self._indicator_periods)
+        self._indicator = IndicatorExtensions.over(std, mean)
+
+        def update_indicator(security, data, indicator):
+            if data.price > 0:
+                std.update(data.time, data.price)
+                mean.update(data.time, data.price)
+
+        self._volatility_model = IndicatorVolatilityModel(self._indicator, update_indicator)
+        equity.set_volatility_model(self._volatility_model)
+
+        self._splits_and_dividends_count = 0
+        self._volatility_checked = False
+
+    def on_data(self, slice):
+        if slice.splits.contains_key(self._aapl) or slice.dividends.contains_key(self._aapl):
+            self._splits_and_dividends_count += 1
+
+            # On a split or dividend event, we need to reset and warm the indicator up as Lean does to BaseVolatilityModel's
+            # to avoid big jumps in volatility due to price discontinuities
+            self._indicator.reset()
+            equity = self.securities[self._aapl]
+            VolatilityModelExtensions.warm_up(
+                self._volatility_model,
+                self,
+                equity,
+                equity.resolution,
+                self._indicator_periods,
+                self._data_normalization_mode
+            )
+
+    def on_end_of_day(self, symbol):
+        if symbol != self._aapl or not self._indicator.is_ready:
+            return
+
+        self._volatility_checked = True
+
+        # This is expected only in this case, 0.05 is not a magical number of any kind.
+        # Just making sure we don't get big jumps on volatility
+        volatility = self.securities[self._aapl].volatility_model.volatility
+        if volatility <= 0 or volatility > 0.05:
+            raise RegressionTestException(
+                "Expected volatility to stay less than 0.05 (not big jumps due to price discontinuities on splits and dividends), "
+                f"but got {volatility}")
+
+    def on_end_of_algorithm(self):
+        if self._splits_and_dividends_count == 0:
+            raise RegressionTestException("Expected to get at least one split or dividend event")
+
+        if not self._volatility_checked:
+            raise RegressionTestException("Expected to check volatility at least once")