Maintaining a Rolling Window of Moving Average Values in QuantConnect
Summary
This code example demonstrates how to maintain a rolling history of hourly SPY closing prices and calculate several moving averages with a technical-analysis library. During warm-up, it collects enough observations to fill a window sized from the selected indicator periods, then computes double exponential, exponential, and weighted moving averages. On subsequent bars it appends the new close, recalculates the latest indicator values, and trims the stored data to the window length.
The example illustrates indicator initialization and rolling data management in a QuantConnect algorithm, rather than a trading strategy: it does not generate orders or evaluate signals. Its sample dates span only a few days, and the algorithm logs the final rolling window and latest values. The labels and period settings include minor inconsistencies, so users should verify which moving average and period they intend to calculate before adapting it. No backtest performance or general trading conclusions are provided.
Key ideas
- The example stores a fixed-length history of hourly closing prices for one equity.
Tags
Full text
# CalibratedResistanceAtmosphericScrubbers
# CalibratedResistanceAtmosphericScrubbers
## Source (Apache-2.0)
```python
# 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 *
import talib
class CalibratedResistanceAtmosphericScrubbers(QCAlgorithm):
def initialize(self):
self.set_start_date(2020, 1, 2)
self.set_end_date(2020, 1, 6)
self.set_cash(100000)
self.add_equity("SPY", Resolution.HOUR)
self.rolling_window = pd.DataFrame()
self.dema_period = 3
self.sma_period = 3
self.wma_period = 3
self.window_size = self.dema_period * 2
self.set_warm_up(self.window_size)
def on_data(self, data):
if "SPY" not in data.bars:
return
close = data["SPY"].close
if self.is_warming_up:
# Add latest close to rolling window
row = pd.DataFrame({"close": [close]}, index=[data.time])
self.rolling_window = pd.concat([self.rolling_window, row]).iloc[-self.window_size:]
# If we have enough closing data to start calculating indicators...
if self.rolling_window.shape[0] == self.window_size:
closes = self.rolling_window['close'].values
# Add indicator columns to DataFrame
self.rolling_window['DEMA'] = talib.DEMA(closes, self.dema_period)
self.rolling_window['EMA'] = talib.EMA(closes, self.sma_period)
self.rolling_window['WMA'] = talib.WMA(closes, self.wma_period)
return
closes = np.append(self.rolling_window['close'].values, close)[-self.window_size:]
# Update talib indicators time series with the latest close
row = pd.DataFrame({"close": close,
"DEMA" : talib.DEMA(closes, self.dema_period)[-1],
"EMA" : talib.EMA(closes, self.sma_period)[-1],
"WMA" : talib.WMA(closes, self.wma_period)[-1]},
index=[data.time])
self.rolling_window = pd.concat([self.rolling_window, row]).iloc[-self.window_size:]
def on_end_of_algorithm(self):
self.log(f"\nRolling Window:\n{self.rolling_window.to_string()}\n")
self.log(f"\nLatest Values:\n{self.rolling_window.iloc[-1].to_string()}\n")
```Shown in full with attribution under the source's licence. Licence: Apache-2.0
This summary was written by Stratmill's research agent from the original; it is not a copy of the source.