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)

Then I realized it was because of the bar.EndTime argument. However, when I removed that argument, and therefore use self.custom.Update(bar.Close), the error changes to this one:

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