Custom Indicators within a Custom AlphaModel (QCAlgorithmFramework)

Hi there,

I'm really getting into the Framework, but I would love to see an example of a customized AlphaModel in which a custom indicator is applied. Just like the example but instead of using the built-in EMA indicator, using a custom indicator created by the user. Below a dummy example in which I just combined the abovementioned with a CustomIndicator that only calcuilates a Moving Average. The backtest runs but it does not generate any insights.

Thank you very much for your help.

Loving the QCAlgortihmFramework!



from clr import AddReference

from QuantConnect import *
from QuantConnect.Indicators import *
from QuantConnect.Algorithm.Framework.Alphas import *

from QuantConnect.Data.Consolidators import *

import numpy as np
import pandas as pd
import decimal as d
from collections import deque
from datetime import datetime, timedelta

class TestAlphaModel3(AlphaModel):
    '''Alpha model that uses a custom indicator to create insights'''

# Initialize variables
    def __init__(self,
                n1 = 15,
                resolution = Resolution.Daily):
        self.n1 = n1
        self.resolution = resolution
        self.predictionInterval = Time.Multiply(Extensions.ToTimeSpan(resolution), n1)
        self.symbolDataBySymbol = {}

        resolutionString = Extensions.GetEnumString(resolution, Resolution)
        self.Name = '{}({},{})'.format(self.__class__.__name__, n1, resolutionString)
    def Update(self, algorithm, data):
        '''Updates this alpha model with the latest data from the algorithm.
        This is called each time the algorithm receives data for subscribed securities
            algorithm: The algorithm instance
            data: The new data available
            The new insights generated'''
        insights = []
        for symbol, symbolData in self.symbolDataBySymbol.items():
            if symbolData.MA.IsReady:

                if symbolData.MAup:
                    if symbolData.MA.Value < 100:
                        insights.append(Insight.Price(symbolData.Symbol, self.predictionInterval, InsightDirection.Down))

                elif symbolData.MAdown:
                    if symbolData.MA.Value > 100:
                        insights.append(Insight.Price(symbolData.Symbol, self.predictionInterval, InsightDirection.Up))

            symbolData.MAup = symbolData.MA.Value > 100

        return insights

    def OnSecuritiesChanged(self, algorithm, changes):
        '''Event fired each time the we add/remove securities from the data feed
            algorithm: The algorithm instance that experienced the change in securities
            changes: The security additions and removals from the algorithm'''
        for added in changes.AddedSecurities:
            symbolData = self.symbolDataBySymbol.get(added.Symbol)
            if symbolData is None:
                # create MA
                symbolData = SymbolData(added)
                symbolData.MA = CustomIndicator(added.Symbol, self.n1)
                self.symbolDataBySymbol[added.Symbol] = symbolData
                # a security that was already initialized was re-added, reset the indicators

class SymbolData:
    '''Contains data specific to a symbol required by this model'''
    def __init__(self, security):
        self.Security = security
        self.Symbol = security.Symbol
        self.MA = None

        # True if MA is above 100, otherwise false.
        # This is used to prevent emitting the same signal repeatedly
        self.MAup = False

    def MAdown(self):
        return not self.MAup
class CustomIndicator:
    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):
        count = len(self.queue)
        self.Time = input.EndTime
        self.Value = sum(self.queue) / count
        self.IsReady = count == self.queue.maxlen

Update Backtest

Hi Emilio, please see the attached example. The custom indicator needs to be updated with self.RegisterIndicator(). 


This is the orginal algorihm with SMA() indicator which provides the same insights with the above custom indicator framework algorithm.


Thank so much as always Jing! :)


Hi Jing,

Sorry to bother you again. I was trying to plot the indicators as usual using the below:

self.Plot("SMA", "Fast", symbolData.Fast.Value)

I placed that function under the def Update(self, algorithm, data)::, but didn't work


Runtime Error: AttributeError : 'TestAlphaModel2' object has no attribute 'Plot'
at Update in 66
AttributeError : 'TestAlphaModel2' object has no attribute 'Plot' (Open Stacktrace)

In general, what are the best practices when it comes to plotting indicators using the Framework?

Thank you for your help,



You can use 

algorithm.Plot("SMA", "Fast", symbolData.Fast.Value)

where "algorithm" in Update(self, algorithm, data) is "self" in the classic algorithm.


Update Backtest


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.


This discussion is closed