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.SetStartDate(2020,1,11) #self.SetEndDate(2020,3,30) self.SetCash(10000) 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: return 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.queue.appendleft(input.Close) self.Time = input.EndTime self.IsReady = count == self.queue.maxlen return self.IsReady #https://docs.scipy.org/doc/scipy/reference/generated/scipy.stats.percentileofscore.html