Having alot of trouble trying to chain together a bunch of indicators to get a single value (couldn't figure it out so code attached as a snippet below).

Basically what I'm trying to do is a sequence of indicators chained together as follows ('=>' denotes the latest output of one indicator being passed as input to the next):

Standard deviation of closing prices measured over past 21 days => last standard deviation's value normalised as a percentile rank over the past 252 days worth of standard deviation measurements => percentile rank measurements smoothed with a 21 day simple moving average => last smoothed value's percentile rank over the past 250 days

Can anyone assist? I read Alex's custom indicator forum posts but couldn't piece together how a custom indicator would work (or not) with QC's indicator extenion methods. Also realise the percentile rank calaculation itself doesn't need to be an indicator per se - alternatively could just be a rolling window which we call stats.percentilescoreof on but I figured the difficulty was about the same...


import numpy as np
import datetime
from scipy import stats
from collections import deque

class RegimRanking(QCAlgorithm):

def Initialize(self):

self.reso = Resolution.Daily
self.spy = self.AddEquity("SPY",self.reso).Symbol

# Indicators
self.std = self.STD("SPY",21,self.reso) # Standard deviation of 21-day closes => pipe these data poitns to....
self.pct_rankSTD = IndicatorExtensions.Of(PercentRank("pct_rankSTD",252), self.std) # Percentrank STD over past 252 days => current value is piped to...
self.SMApct_rank = IndicatorExtensions.SMA(self.pct_rankSTD, 21) # 21 day SMA for smoothing => smoothed values piped to...
self.pct_rankSMA = IndicatorExtensions.Of(PercentRank("pct_rankSMA",250), self.SMApct_rank) # percent rank 250 of the SMA

self.RegisterIndicator("SPY", self.pct_rankSTD, self.reso)
self.RegisterIndicator("SPY", self.pct_rankSMA, self.reso)

def OnData(self):
## Manually warm up indicators
## Retrieve historical data for each symbol
history = self.History(self.spy, 252, Resolution.Daily)

if history.empty:

if str(self.spy) in history.index.get_level_values(0):
symbolVolumeHistory = history.loc[str(self.spy)]
if symbolVolumeHistory.empty:
self.Log("No history found")

# Update indicators manually
for time, row in history.loc["SPY"].iterrows():
self.std.Update(time, row["close"])

class PercentRank:
def __init__(self, name, period):
self.Name = name
#self.Time = datetime.min
self.Value = 0
self.IsReady = False
self.queue = deque(maxlen=period) # list-like

#def __repr__(self):
# return "{0} -> IsReady: {1}. Time: {2}. Value: {3}".format(self.Name, self.IsReady, self.Time, self.Value)

# Update method is mandatory
def Update(self, input):
self.Value = stats.percentileofscore(self.queue, input.Close)
self.Time = input.EndTime
self.IsReady = count == self.queue.maxlen
return self.IsReady