Hello All,
I am tyring to create an algorithm that involves the use of two rolling windows: one for close data and the other for RSI. I am not sure how to pull from my historical data to warm up the RSI rolling window. My code is shown below. Line 67 in the "OnSecuritiesChanged" function is where I am trying to warm up the RSI rolling window. I'm at a loss on this, and I have found the QC documentation to be somewhat lacking, so I will greatly appreciate any help that the community can offer.
from Execution.ImmediateExecutionModel import ImmediateExecutionModel
from Portfolio.EqualWeightingPortfolioConstructionModel import EqualWeightingPortfolioConstructionModel
from Risk.MaximumDrawdownPercentPerSecurity import MaximumDrawdownPercentPerSecurity
from Selection.QC500UniverseSelectionModel import QC500UniverseSelectionModel
class SimpleRSITestQC500Universe(QCAlgorithm):
def Initialize(self):
self.SetStartDate(2010, 1, 1) # Set Start Date
self.SetEndDate(2010, 2, 28) # Set End Date
self.SetCash(100000) # Set Strategy Cash
self.SetExecution(ImmediateExecutionModel())
self.SetPortfolioConstruction(EqualWeightingPortfolioConstructionModel())
self.SetRiskManagement(MaximumDrawdownPercentPerSecurity(0.05))
symbols = [ Symbol.Create("SPY", SecurityType.Equity, Market.USA), Symbol.Create("GE", SecurityType.Equity, Market.USA), Symbol.Create("BA", SecurityType.Equity, Market.USA) ]
self.SetUniverseSelection(ManualUniverseSelectionModel(symbols))
self.UniverseSettings.Resolution = Resolution.Daily
self.AddAlpha(RsiAlphaModelTest())
class RsiAlphaModelTest(AlphaModel):
def __init__(self, period = 14, resolution = Resolution.Daily):
self.period = period
self.resolution = resolution
self.insightPeriod = Time.Multiply(Extensions.ToTimeSpan(resolution), period)
self.symbolDataBySymbol = {}
self.closeWindows = {}
self.rsiWindows = {}
resolutionString = Extensions.GetEnumString(resolution, Resolution)
self.Name = '{}({},{})'.format(self.__class__.__name__, period, resolutionString)
def Update(self, algorithm, data):
insights = []
for symbol, symbolData in self.symbolDataBySymbol.items():
if data.ContainsKey(symbol) and data[symbol] is not None:
self.closeWindows[symbol].Add(data[symbol].Close)
if self.closeWindows[symbol].Count>2:
algorithm.Debug(self.closeWindows[symbol][2])
rsi = symbolData.RSI
self.rsiWindows[symbol].Add(rsi)
previous_state = symbolData.State
state = self.GetState(rsi, previous_state)
if state != previous_state and rsi.IsReady:
if state == State.TrippedLow:
insights.append(Insight.Price(symbol, self.insightPeriod, InsightDirection.Up))
if state == State.TrippedHigh:
insights.append(Insight.Price(symbol, self.insightPeriod, InsightDirection.Down))
symbolData.State = state
return insights
def OnSecuritiesChanged(self, algorithm, changes):
# clean up data for removed securities
symbols = [ x.Symbol for x in changes.RemovedSecurities ]
if len(symbols) > 0:
for subscription in algorithm.SubscriptionManager.Subscriptions:
if subscription.Symbol in symbols:
self.symbolDataBySymbol.pop(subscription.Symbol, None)
subscription.Consolidators.Clear()
# initialize data for added securities
addedSymbols = [ x.Symbol for x in changes.AddedSecurities if x.Symbol not in self.symbolDataBySymbol]
if len(addedSymbols) == 0: return
history = algorithm.History(addedSymbols, self.period, self.resolution)
for symbol in addedSymbols:
rsi = algorithm.RSI(symbol, self.period, MovingAverageType.Wilders, self.resolution)
#rsi.Updated += self.RsiUpdated(symbol=symbol, sender=sender, updated=updated)
self.rsiWindows[symbol] = RollingWindow[IndicatorDataPoint](20)
self.closeWindows[symbol] = RollingWindow[float](self.period)
symbolTradeBarsHistory = history.loc[symbol]
symbolClose = symbolTradeBarsHistory["close"]
symbolTime = symbolTradeBarsHistory["time"]
for historyIndex in range(self.period):
self.closeWindows[symbol].Add(symbolClose[historyIndex])
self.rsiWindows[symbol].Add(symbolTime[historyIndex],symbolClose[historyIndex])
if not history.empty:
ticker = SymbolCache.GetTicker(symbol)
if ticker not in history.index.levels[0]:
Log.Trace(f'RsiAlphaModel.OnSecuritiesChanged: {ticker} not found in history data frame.')
continue
for tuple in history.loc[ticker].itertuples():
rsi.Update(tuple.Index, tuple.close)
self.symbolDataBySymbol[symbol] = SymbolData(symbol, rsi)
symbolTradeBarsHistory = None
symbolClose = None
for k in self.closeWindows.keys():
algorithm.Debug(str(k) + ' ' + str(self.closeWindows[k][0]) + ' ' + str(self.closeWindows[k][1]) + ' ' + str(self.closeWindows[k][2]) + ' ' + str(self.closeWindows[k][3]) + ' ' + str(self.closeWindows[k][4]) + ' ' + str(self.closeWindows[k][5]))
def GetState(self, rsi, previous):
if rsi.Current.Value > 70:
return State.TrippedHigh
if rsi.Current.Value < 30:
return State.TrippedLow
if previous == State.TrippedLow:
if rsi.Current.Value > 35:
return State.Middle
if previous == State.TrippedHigh:
if rsi.Current.Value < 65:
return State.Middle
return previous
#def RsiUpdated(self, symbol, sender, updated):
# self.rsiWindows[symbol].Add(updated)
class SymbolData:
def __init__(self, symbol, rsi):
self.Symbol = symbol
self.RSI = rsi
self.State = State.Middle
class State(Enum):
'''Defines the state. This is used to prevent signal spamming and aid in bounce detection.'''
TrippedLow = 0
Middle = 1
TrippedHigh = 2
Shile Wen
Hi Garrett,
Since indicator values are actually floats themselves, we would first need to change the type of the RSI RollingWindow to a float. Because we need to warm up our indicators alongside our RollingWindows, we need to increase the amount of historical data we use so that once we warmup our indicators, we can then also warm up our RollingWindows. I’ve shown the necessary adjustments in the backtest.
Best,
Shile Wen
Garrett Grow
Thank you! That helped immensely! Maybe I could bother you or the community for one more question. I want to liquidate my position for a symbol after all of its insights expire. I think I need to use "self.IsActive" somehow, but I don't know how to use it. Apologies for the newb questions.
Shile Wen
Hi Garrett,
If our algorithm uses a Portfolio Construction Model that inherits from PorfolioConstructionModel class (e.g.: EqualWeightingPorfolioConstructionModel or InsightWeightingPorfolioConstructionModel), then the algorithm generally will automatically liquidate positions after the insight expires.
Best,
Shile Wen
Garrett Grow
The material on this website is provided for informational purposes only and does not constitute an offer to sell, a solicitation to buy, or a recommendation or endorsement for any security or strategy, nor does it constitute an offer to provide investment advisory services by QuantConnect. In addition, the material offers no opinion with respect to the suitability of any security or specific investment. QuantConnect makes no guarantees as to the accuracy or completeness of the views expressed in the website. The views are subject to change, and may have become unreliable for various reasons, including changes in market conditions or economic circumstances. All investments involve risk, including loss of principal. You should consult with an investment professional before making any investment decisions.
To unlock posting to the community forums please complete at least 30% of Boot Camp.
You can continue your Boot Camp training progress from the terminal. We hope to see you in the community soon!