Back

Custom Indicators On Consolidated Data

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

Update Backtest







Hi there, one thing, I can see is that you are registering the indicator with:

self.RegisterIndicator("SPY", self.custom, Resolution.Minute)

From what I have seen in the examples you need to register it with the handler like

self.RegisterIndicator("SPY", self.custom, OnDataConsolidated)

If that works you also don't need to update it manually as far as I am aware. You can just call the latest value with:

self.Debug("Custom: {}".format(self.custom.Current.Value))

Apologies if my advice does not work for you.  I have not been on the platform long :)

0

Hi,

First of all, thank you very much for your help, it's really appreciated.

So both of your suggestions worked! A few comments for the record:

Here I had to use self.RegisterIndicator("SPY", self.custom, consolidator), instead of self.RegisterIndicator("SPY", self.custom, OnDataConsolidated). And then I also removed this Update line  self.custom.Update(bar.EndTime, bar.Close) as you suggested and just called self.custom.Value in the boolean for the signals.

Thanks again!

Emilio

0

Update Backtest





0

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.


Loading...

This discussion is closed