Accessing ETFConstituentsData from the AlphaModel

Based on the information provided in the extracts, it's not clear why the ETFConstituentsData is returning an empty dictionary. However, it's important to note that the Update method in the AlphaModel must return a list of insights. If there are no insights, it should return an empty list.

Here is an example of how to structure the Update method:

class Alpha():
    def __init__(self):
        pass

    def Update(self, algorithm, data):
        return []

    def OnSecuritiesChanged(self, algorithm, changes):
        pass

In your case, you might want to ensure that ETFConstituentsData is properly set up and that the data is available at the time of the call. If the data is not available, it could result in an empty dictionary.

Unfortunately, without more information or context, it's difficult to provide a more specific solution. You might find it helpful to review the QuantConnect documentation or ask for help in the QuantConnect forum.

Here is a whole script. The symbols list is empty and I'm not sure why.

from QuantConnect.DataSource import *
from AlgorithmImports import *
from AlphaModel import *
import typing
from datetime import timedelta
from QuantConnect.Indicators import SimpleMovingAverage

class ChainedUniverseAlgorithm(QCAlgorithm):

    def Initialize(self):
        self.SetCash(100000)
        self.SetStartDate(2021, 1, 1)
        self.SetEndDate(2023, 12, 1)
        self.UniverseSettings.Asynchronous = True
        self.UniverseSettings.Resolution = Resolution.Daily
        self.Settings.MinimumOrderMarginPortfolioPercentage = 0.01
        etf = Symbol.Create("SPY", SecurityType.Equity, Market.USA)
        self.AddUniverseSelection(ETFConstituentsUniverseSelectionModel(etf, self.UniverseSettings,self.FilterETFConstituents))
        #
        self.SetBenchmark(etf)
        
        # Alpha Model
        self.AddAlpha(TechnicalAlphaModel())

        # Portfolio construction model
        self.SetPortfolioConstruction(InsightWeightingPortfolioConstructionModel())        
        
        # Risk model
        self.SetRiskManagement(NullRiskManagementModel())

        # Execution model
        self.SetExecution(ImmediateExecutionModel())
    
    def FilterETFConstituents(self, constituents):
        return [i.Symbol for i in constituents if i.Weight is not None]

class TechnicalAlphaModel(AlphaModel):
    def __init__(self):
        pass

    def Update(self, algorithm, data):
        insights = []
        c_data = []
        symbols = []
        constituents = data.Get(ETFConstituentData)
        #symbols = constituents.keys()
        #c_data = constituents.values()
        
        for kvp in constituents:
            symbol = kvp.Key
            constituentData = kvp.Value
            c_data.append(constituentData)
            symbols.append(symbol)
        algorithm.Debug(f'symbols {symbols}')
        
        for c in c_data:
            # for every constituent make a data object
            symbol = c.Symbol
            weight = c.Weight
            symboldata =  SymbolData(algorithm, c.Symbol)
            fast = symboldata.fast.Current.Value #self.Time, self.AdjustedPrice
            slow = symboldata.slow.Current.Value
            algorithm.Debug('Fast:' + str(fast) + 'Slow:' +str(slow))

            if fast > slow:   
                direction = InsightDirection.Up
            else:       
                direction = InsightDirection.Down
            
            algorithm.Debug(f'direction {direction}')
            insight = Insight(symbol, timedelta(days=1), 
                        InsightType.Price, direction, weight)
            insights.append(insight)
        return insights
    
    def OnSecuritiesChanged(self, algorithm, changes):
        pass

class SymbolData:
    def __init__(self, algorithm, symbol, fast=50, slow=200):
        self.slow = SimpleMovingAverage(symbol, slow)
        self.fast = SimpleMovingAverage(symbol, fast)
        self.ready = algorithm.WarmUpIndicator(symbol, self.slow, Resolution.Daily)

Based on the provided script, it seems like you are trying to access the ETF constituents data in your Update method of the TechnicalAlphaModel. However, the Update method is not the correct place to access this data.

The ETF constituents data should be accessed in the FilterETFConstituents method, which is called when the universe selection model is determining which securities to include in the universe.

Here is an example of how you might modify your FilterETFConstituents method to access the ETF constituents data:

def FilterETFConstituents(self, constituents):
    symbols = []
    for constituent in constituents:
        if constituent.Weight is not None:
            symbols.append(constituent.Symbol)
    return symbols

In this modified method, we iterate over the constituents and add the symbol of each constituent with a non-null weight to the symbols list. This list is then returned by the method, which means these symbols will be included in the universe.

Please note that the Update method in your AlphaModel should return a list of insights, not a list of symbols. If you want to generate insights based on the ETF constituents data, you should do this in the Update method of your AlphaModel, using the symbols that were included in the universe by the FilterETFConstituents method.

Unfortunately, without more information or context, it's difficult to provide a more specific solution. You might find it helpful to review the QuantConnect documentation or ask for help in the QuantConnect forum.

I'm trying to change your example:

Your example from github also doesn't work:

# QUANTCONNECT.COM - Democratizing Finance, Empowering Individuals.
# Lean Algorithmic Trading Engine v2.0. Copyright 2014 QuantConnect Corporation.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

import typing

from AlgorithmImports import *
from datetime import timedelta


### <summary>
### Example algorithm demonstrating the usage of the RSI indicator
### in combination with ETF constituents data to replicate the weighting
### of the ETF's assets in our own account.
### </summary>
class ETFConstituentUniverseRSIAlphaModelAlgorithm(QCAlgorithm):
    ### <summary>
    ### Initialize the data and resolution required, as well as the cash and start-end dates for your algorithm. All algorithms must initialized.
    ### </summary>
    def Initialize(self):
        self.SetStartDate(2020, 12, 1)
        self.SetEndDate(2021, 1, 31)
        self.SetCash(100000)

        self.SetAlpha(ConstituentWeightedRsiAlphaModel())
        self.SetPortfolioConstruction(InsightWeightingPortfolioConstructionModel())
        self.SetExecution(ImmediateExecutionModel())

        spy = self.AddEquity("SPY", Resolution.Hour).Symbol

        # We load hourly data for ETF constituents in this algorithm
        self.UniverseSettings.Resolution = Resolution.Hour
        self.Settings.MinimumOrderMarginPortfolioPercentage = 0.01

        self.AddUniverse(self.Universe.ETF(spy, self.UniverseSettings, self.FilterETFConstituents))

    ### <summary>
    ### Filters ETF constituents
    ### </summary>
    ### <param name="constituents">ETF constituents</param>
    ### <returns>ETF constituent Symbols that we want to include in the algorithm</returns>
    def FilterETFConstituents(self, constituents):
        return [i.Symbol for i in constituents if i.Weight is not None and i.Weight >= 0.001]


### <summary>
### Alpha model making use of the RSI indicator and ETF constituent weighting to determine
### which assets we should invest in and the direction of investment
### </summary>
class ConstituentWeightedRsiAlphaModel(AlphaModel):
    def __init__(self, maxTrades=None):
        self.rsiSymbolData = {}

    def Update(self, algorithm: QCAlgorithm, data: Slice):
        algoConstituents = []
        for barSymbol in data.Bars.Keys:
            if not algorithm.Securities[barSymbol].Cache.HasData(ETFConstituentData):
                continue

            constituentData = algorithm.Securities[barSymbol].Cache.GetData[ETFConstituentData]()
            algoConstituents.append(constituentData)

        if len(algoConstituents) == 0 or len(data.Bars) == 0:
            # Don't do anything if we have no data we can work with
            return []

        constituents = {i.Symbol:i for i in algoConstituents}

        for bar in data.Bars.Values:
            if bar.Symbol not in constituents:
                # Dealing with a manually added equity, which in this case is SPY
                continue

            if bar.Symbol not in self.rsiSymbolData:
                # First time we're initializing the RSI.
                # It won't be ready now, but it will be
                # after 7 data points
                constituent = constituents[bar.Symbol]
                self.rsiSymbolData[bar.Symbol] = SymbolData(bar.Symbol, algorithm, constituent, 7)

        allReady = all([sd.rsi.IsReady for sd in self.rsiSymbolData.values()])
        if not allReady:
            # We're still warming up the RSI indicators.
            return []

        insights = []

        for symbol, symbolData in self.rsiSymbolData.items():
            averageLoss = symbolData.rsi.AverageLoss.Current.Value
            averageGain = symbolData.rsi.AverageGain.Current.Value

            # If we've lost more than gained, then we think it's going to go down more
            direction = InsightDirection.Down if averageLoss > averageGain else InsightDirection.Up

            # Set the weight of the insight as the weight of the ETF's
            # holding. The InsightWeightingPortfolioConstructionModel
            # will rebalance our portfolio to have the same percentage
            # of holdings in our algorithm that the ETF has.
            insights.append(Insight.Price(
                symbol,
                timedelta(days=1),
                direction,
                float(averageLoss if direction == InsightDirection.Down else averageGain),
                weight=float(symbolData.constituent.Weight)
            ))
        
        return insights


class SymbolData:
    def __init__(self, symbol, algorithm, constituent, period):
        self.Symbol = symbol
        self.constituent = constituent
        self.rsi = algorithm.RSI(symbol, period, MovingAverageType.Exponential, Resolution.Hour)

Is it possible to have a human take a look?

Hi Amir

Sorry for the delay. You can try this:

class YourAlgorithmInstance(QCAlgorithm):
    def Initialize(self):
        ...
        self.AddAlpha(CustomAlphaModel(self, ...))
        self.AddUniverse(self.Universe.ETF(spy, self.UniverseSettings, self.FilterETFConstituents))
        self.etf_constituents_data = {}
    
    def FilterETFConstituents(self, constituents):
        self.etf_constituents_data = {c.Symbol: c for c in constituents}
        return [c.Symbol for c in constituents]

class CustomAlphaModel(AlphaModel):
    def __init__(self, algorithm, ...):
        self.algorithm = algorithm
        ...
    
    def Update(self, algorithm, data):
        # you can get the etf constituents data by
        for symbol, constituent_data in self.algorithm.etf_constituents_data.items():
            weight = constituent_data.Weight
            ...

Basically we're setting up a cache in the main algorithm instance, and set up a point to the main algorithm when constructing the custom alpha model. Then, we can have access to the cache through the pointer. Note that the cache will only be updated on every ETFConstituent filter call in the main algorithm.

Best
Louis