Hello All,
I was wondering if someone could help me with this:
I am trying to create daily consolidated prices (tradebars) rolling windows for a list of tickers, say : ["SPY", "TLT", "BND"]. Would this script do the trick? In the current script, I have two different classes, one for prices rolling windows, and another for the indicator rolling window. After which, I combine two dictionaries containing results of the indicator and prices rolling windows into one dictionary. I feel there should be an easier way of doing that resulting in a cleaner code.
How can we combine both the price daily rolling windows and the indicator (ADX in this case) daily rolling window in the same class?
Also, Is there a fast way to calculate, let's say, the 3 day high (i.e. the highest price the stock reached in the last 3 days?
The resolution in this code is hourly, also can be minutes.
The resolution for the price and indicator rolling windows should be daily.
Thanks a lot.
from collections import defaultdict
class MyAlgorithm(QCAlgorithm):
def Initialize(self):
self.SetStartDate(2019,1,1)
self.SetCash(100000)
self.SetBrokerageModel(BrokerageName.InteractiveBrokersBrokerage)
tickers = ["SPY", "TLT", "BND"]
# Creates empty dictionary to later hold daily Tradebar price rolling windows
self.symbolData_window = {}
# Creates empty dictionary to later hold ADX daily values
self.symbolData = {}
for ticker in tickers:
self.equity = self.AddEquity(ticker, Resolution.Hour)
self.equity.SetDataNormalizationMode(DataNormalizationMode.Raw)
self.equity.SetLeverage(1.0)
symbol_window = self.equity.Symbol
symbol = self.equity.Symbol
# Consolidating lower resolution data into daily data for daily rolling windows
consolidator_daily = TradeBarConsolidator(timedelta(days = 1))
consolidator_daily.DataConsolidated += self.OnDailyData
self.SubscriptionManager.AddConsolidator(symbol_window, consolidator_daily)
# Creating symbol data for respective symbol
self.symbolData_window[symbol_window] = SymbolData_window(self, symbol_window)
# Creates daily ADX rolling windows
self.symbolData[symbol] = SymbolData(self, symbol)
self.SetWarmUp(100)
# Adding daily bar data to 4, 3, 2 etc.. rolling windows
def OnDailyData(self, sender, bar):
self.symbolData_window[bar.Symbol].window4.Add(bar)
self.symbolData_window[bar.Symbol].window3.Add(bar)
self.symbolData_window[bar.Symbol].window2.Add(bar)
def OnData(self, data):
for symbol, value in self.symbolData.items():
if not data.ContainsKey(symbol.Value):
return
for symbol, value in self.symbolData_window.items():
if not data.ContainsKey(symbol.Value):
return
for symbol, value in self.symbolData_window.items():
value.window4.Add(data[symbol.Value])
value.window3.Add(data[symbol.Value])
value.window2.Add(data[symbol.Value])
if self.IsWarmingUp:
return
# ADX rolling window ready
if not all([symbol.IsReady for symbol in self.symbolData.values()]):
return
# Merging both dictionaries for ADX and daily rolling windows together on same key:
self.symbolData_combined = defaultdict(list)
for d in (self.symbolData, self.symbolData_window):
for key, value in d.items():
self.symbolData_combined[key].append(value)
for symbol, value in self.symbolData_combined.items():
if value[1].window4[0].High > value[1].window4[1].High > value[1].window4[2].High > value[1].window4[3].High and value[0].adxWindow[0] > value[0].adxWindow[1]:
self.SetHoldings(symbol, 1/3)
else:
if self.Portfolio[symbol].Invested == True:
self.Liquidate(symbol)
class SymbolData:
def __init__(self, algorithm, symbol):
self.algorithm = algorithm
self.symbol = symbol
# Defining indicator (ADX in this case)
self.adx = algorithm.ADX(symbol, 14, Resolution.Daily)
# Defining rolling window to hold indicator points
self.adxWindow = RollingWindow[IndicatorDataPoint](2)
# Setting event handler
self.adx.Updated += self.OnAdxUpdated
def OnAdxUpdated(self, sender, updated):
if self.adx.IsReady:
# Adding updated indicator data (ADX data) to rolling window
self.adxWindow.Add(updated)
@property
def IsReady(self):
return self.adx.IsReady and self.adxWindow.IsReady
class SymbolData_window:
def __init__(self, algorithm, symbol):
self.algorithm = algorithm
self.symbol = symbol
self.window4 = RollingWindow[TradeBar](4)
self.window3 = RollingWindow[TradeBar](3)
self.window2 = RollingWindow[TradeBar](2)
