UniversalSelection

I wrote universal stock list with higher DollarVolume criteria and calling from main method but it is throwing an error. please correct and suggest the fix

class MainAlgorithm(QCAlgorithm):
    def Initialize(self):
        self.SetStartDate(2023, 1, 1)
        self.SetCash(params.INITIAL_CASH)
        self.UniverseSettings.Resolution = Resolution.Daily
        self.SetBenchmark("SPY")
       
        self.universe_selection = UniverseSelection(self, self.logger)

# UniverseSelection.py
from AlgorithmImports import *
import parameters as params
class UniverseSelection:
    def __init__(self, algorithm, logger):
        self.algorithm = algorithm
        self.logger = logger
        self.logger.log("Starting UniverseSelection initialization...")
        try:
            self.history = {}
            self.moving_averages_200 = {}
            self.moving_averages_20 = {}
            self.previous_close = {}
            self.atr = {}
            self.volatility = {}
            self.avg_volatility = {}
            self.logger.log("Finished UniverseSelection initialization.")
        except Exception as e:
            self.logger.log(f"Error during UniverseSelection initialization: {str(e)}")
            raise e
    def CoarseSelectionFunction(self, coarse):
        selected: List[str] = []
        for stock in coarse:
            if stock.Price > params.PRICE_THRESHOLD and stock.DollarVolume > params.DOLLAR_VOLUME_THRESHOLD:
                selected.append(stock.Symbol)
                # Initialize the historical data if not already done
                if stock.Symbol not in self.history:
                    self.InitializeIndicators(stock.Symbol)
        return selected
    def InitializeIndicators(self, symbol):
        # Fetch the historical data
        history = self.algorithm.History(symbol, params.MOVING_AVERAGE_200_DAYS, Resolution.Daily)
        if not history.empty:
            # Calculate and store moving averages
            self.history[symbol] = history['close']
            self.moving_averages_200[symbol] = \
            history['close'].rolling(window=params.MOVING_AVERAGE_200_DAYS).mean().iloc[-1]
            self.moving_averages_20[symbol] = \
            history['close'].rolling(window=params.MOVING_AVERAGE_20_DAYS).mean().iloc[-1]
            self.previous_close[symbol] = history['close'].iloc[-2]
            # Calculate ATR (Average True Range)
            high = history['high']
            low = history['low']
            close = history['close'].shift(1)
            tr = (high - low).combine(close - high, max).combine(close - low, max)
            self.atr[symbol] = tr.rolling(window=14).mean().iloc[-1]
            # Calculate daily returns for volatility
            daily_returns = history['close'].pct_change().dropna()
            # Calculate and store volatility
            self.volatility[symbol] = daily_returns.std() * (252 ** 0.5)  # Annualized volatility
            self.avg_volatility[symbol] = daily_returns.rolling(window=params.VOLATILITY_LOOKBACK_DAYS).std().mean() * (
                        252 ** 0.5)
    def UpdateDailyIndicators(self):
        # This function will be called at the end of each trading day to update indicators
        for symbol in self.algorithm.Securities.Keys:
            history = self.algorithm.History(symbol, params.MOVING_AVERAGE_200_DAYS, Resolution.Daily)
            if not history.empty:
                self.history[symbol] = history['close']
                self.moving_averages_200[symbol] = \
                history['close'].rolling(window=params.MOVING_AVERAGE_200_DAYS).mean().iloc[-1]
                self.moving_averages_20[symbol] = \
                history['close'].rolling(window=params.MOVING_AVERAGE_20_DAYS).mean().iloc[-1]
                self.previous_close[symbol] = history['close'].iloc[-2]
                # Update ATR
                high = history['high']
                low = history['low']
                close = history['close'].shift(1)
                tr = (high - low).combine(close - high, max).combine(close - low, max)
                self.atr[symbol] = tr.rolling(window=14).mean().iloc[-1]
                # Update volatility
                daily_returns = history['close'].pct_change().dropna()
                self.volatility[symbol] = daily_returns.std() * (252 ** 0.5)
                self.avg_volatility[symbol] = daily_returns.rolling(
                    window=params.VOLATILITY_LOOKBACK_DAYS).std().mean() * (252 ** 0.5)
    def GetSelectedStocks(self):
        abnormal_dollar_volume_stocks = []
        for symbol in self.algorithm.Securities.Keys:
            if symbol in self.history:
                avg_dollar_volume = self.history[symbol].mean()
                current_dollar_volume = self.algorithm.Securities[symbol].Price * self.algorithm.Securities[
                    symbol].Volume
                price_increase = (self.algorithm.Securities[symbol].Price - self.previous_close[symbol]) / \
                                 self.previous_close[symbol]
                if (current_dollar_volume > 2 * avg_dollar_volume and
                        self.algorithm.Securities[symbol].Price > self.moving_averages_200[symbol] and
                        self.algorithm.Securities[symbol].Price > self.moving_averages_20[symbol] and
                        price_increase > params.MIN_PRICE_INCREASE_PERCENT and
                        self.volatility[symbol] <= params.MAX_VOLATILITY_MULTIPLIER * self.avg_volatility[symbol]):
                    abnormal_dollar_volume_stocks.append(symbol)
        return abnormal_dollar_volume_stocks
+ Expand
- Collapse

from AlgorithmImports import *
import parameters as params
from UniverseSelection import UniverseSelection
class MainAlgorithm(QCAlgorithm):
def initialize(self):
self.set_start_date(2023, 1, 1)
self.set_cash(params.INITIAL_CASH)
# Set universe settings using the appropriate method
self.universe_settings.Resolution = Resolution.DAILY
self.set_benchmark("SPY")
# Initialize logger
self.logger = self.Debug
# Use a different attribute name to store UniverseSelection
self.my_universe_selection = UniverseSelection(self, self.logger)

+ Expand

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.

class UniverseSelection(UniverseSelectionModel):
def __init__(self, algorithm: QCAlgorithm, logger: Logger):
super().__init__()
self.algorithm = algorithm
self.logger = logger
self.logger.log("Starting UniverseSelection initialization...")
try:
self.history = {}
self.moving_averages_200 = {}
self.moving_averages_20 = {}
self.previous_close = {}
self.atr = {}
self.volatility = {}
self.avg_volatility = {}
self.logger.log("Finished UniverseSelection initialization.")
except BaseException as e:
self.logger.log(f"Error during UniverseSelection initialization: {str(e)}")
raise e
def CreateUniverses(self) -> List[Universe]:
# This method must return one or more universes
return [self.algorithm.AddUniverse(self.CoarseSelectionFunction)]

+ Expand

Mia Alissi INVESTOR

September 2024

Here are the necessary corrections and suggestions to fix the errors in your code:

Correct Method Names: Ensure method names like set_start_date, set_cash, and set_benchmark are in lowercase to match the QuantConnect API.
Fix Resolution Enum: Update Resolution.DAILY to correctly reference the enum value.
Define Parameters and UniverseSelection: Ensure that the parameters and UniverseSelection modules are correctly imported and defined. You need to create a parameters.py file with the necessary constants like INITIAL_CASH, PRICE_THRESHOLD, etc., and ensure UniverseSelection is available in your project.
Logger Initialization: Use self.Debug correctly by ensuring it's a method call. You can initialize the logger in the Initialize method.
Property Assignment: Avoid assigning to read-only properties like resolution in universe_settings. Use the appropriate method or configuration to set these values.

Here's how you can adjust your MainAlgorithm class:

from AlgorithmImports import *
import parameters as params
from UniverseSelection import UniverseSelection
class MainAlgorithm(QCAlgorithm):
    def initialize(self):
        self.set_start_date(2023, 1, 1)
        self.set_cash(params.INITIAL_CASH)
        # Set universe settings using the appropriate method
        self.universe_settings.Resolution = Resolution.DAILY
        self.set_benchmark("SPY")
        # Initialize logger
        self.logger = self.Debug
        # Use a different attribute name to store UniverseSelection
        self.my_universe_selection = UniverseSelection(self, self.logger)
+ Expand
- Collapse

Make sure to create a parameters.py file with the necessary constants and ensure that all imports are correctly set up.

Sources: Asset Growth Effect, Coarse Fine Universe Selection Benchmark, Momentum And State Of Market Filters, Short Term Reversal, Leveraged ETFs With Systematic Risk Management

Upvote

Dharmesh Khalasi INVESTOR

Do I need to inherit fromUniverseSelectionModel like below and implement CreateUniverses method ?

class UniverseSelection(UniverseSelectionModel):
    def __init__(self, algorithm: QCAlgorithm, logger: Logger):
        super().__init__()
        self.algorithm = algorithm
        self.logger = logger
        self.logger.log("Starting UniverseSelection initialization...")
        try:
            self.history = {}
            self.moving_averages_200 = {}
            self.moving_averages_20 = {}
            self.previous_close = {}
            self.atr = {}
            self.volatility = {}
            self.avg_volatility = {}
            self.logger.log("Finished UniverseSelection initialization.")
        except BaseException as e:
            self.logger.log(f"Error during UniverseSelection initialization: {str(e)}")
            raise e
    def CreateUniverses(self) -> List[Universe]:
        # This method must return one or more universes
        return [self.algorithm.AddUniverse(self.CoarseSelectionFunction)]
+ Expand
- Collapse

1 person upvoted this

Louis Szeto INVESTOR

QuantConnect | October 2024

Hi Dharmesh

Yes you do! Check out the docs of a custom universe selection model here.

Best
Louis

Platform

Radically Open-Source Algorithmic Trading Engine

Join Our Discord Channel

Quarterly Open-Source Trading Competition

Draft Discussions

Bookmarked Discussions

SEARCH DISCUSSIONS

359,800 Quants.

VOTE FOR UPCOMING FEATURES

JOIN OUR Community MAILING LIST

IN THIS RESEARCH

PARTICIPANTS

Actions

Join QuantConnect for Free