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 EmaCrossAlphaModel.py 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 EmaCrossAlphaModel.py 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!

Emilio

-------------------------------

from clr import AddReference
AddReference("System")
AddReference("QuantConnect.Common")
AddReference("QuantConnect.Algorithm.Framework")
AddReference("QuantConnect.Indicators")
AddReference("QuantConnect.Common")

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
        Args:
            algorithm: The algorithm instance
            data: The new data available
        Returns:
            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
        Args:
            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
                
            else:
                # a security that was already initialized was re-added, reset the indicators
                symbolData.MA.Reset()

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

    @property
    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):
        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