Hi there,
So I've been trying to combine Custom Indicators with the ability to consolidate bars to lower frequency. Please find below my first attempt. It gives me the following error under the "OnDataConsolidated" section:
Runtime Error: TypeError : Update() takes 2 positional arguments but 3 were given
at OnDataConsolidated in main.py:line 31
TypeError : Update() takes 2 positional arguments but 3 were given (Open Stacktrace)
Runtime Error: AttributeError : 'decimal.Decimal' object has no attribute 'Close'
at OnDataConsolidated in main.py:line 31
at Update in main.py:line 74
AttributeError : 'decimal.Decimal' object has no attribute 'Close' (Open Stacktrace)
Any help please? I think it could be a good template to have as it combines two very common features such as custom indicators and bars consolidation.
Thank you so much as always!
Emilio
--------------------------------
from collections import deque
from datetime import datetime, timedelta
from numpy import sum
### Demonstrates how to create a custom indicator and register it for automatic updated
class TEST(QCAlgorithm):
def Initialize(self):
self.SetStartDate(2013,10,7)
self.SetEndDate(2013,10,11)
self.AddEquity("SPY", Resolution.Minute)
# define our trade bar consolidator (2 hours). we can access the minute bar from the DataConsolidated events
consolidator = TradeBarConsolidator(timedelta(minutes = 120)) # 2 hours from minutes
# attach our event handler. The event handler is a function that will be called each time we produce a new consolidated piece of data.
consolidator.DataConsolidated += self.OnDataConsolidated
# this call adds our consolidator to the manager to receive updates from the engine
self.SubscriptionManager.AddConsolidator("SPY", consolidator)
# Create python custom indicator
self.custom = CustomSimpleMovingAverage('custom', 60)
self.RegisterIndicator("SPY", self.custom, Resolution.Minute)
# set warmup period
#self.SetWarmUp(30)
def OnDataConsolidated(self, sender, bar):
self.custom.Update(bar.EndTime, bar.Close)
# wait for our slow ema to fully initialize
if not self.custom.IsReady:
return
holdings = self.Portfolio["SPY"].Quantity
if holdings == 0 and bar.Close > self.custom.Value:
self.MarketOrder("SPY", 1)
# we only want to liquidate if we're currently long if the fast is less than the slow we'll liquidate our long
if holdings > 0 and bar.Close < self.custom.Value:
self.Liquidate("SPY")
def OnData(self, data):
'''OnData event is the primary entry point for your algorithm. Each new data point will be pumped in here.'''
pass
# Python implementation of SimpleMovingAverage
class CustomSimpleMovingAverage:
def __init__(self, name, period):
self.Name = name
self.Time = datetime.min
self.Value = 0
self.IsReady = False
self.queue = deque(maxlen=period)
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.queue.appendleft(input.Close)
count = len(self.queue)
self.Time = input.EndTime
self.Value = sum(self.queue) / count
self.IsReady = count == self.queue.maxlen