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

  1. class MainAlgorithm(QCAlgorithm):
  2. def Initialize(self):
  3. self.SetStartDate(2023, 1, 1)
  4. self.SetCash(params.INITIAL_CASH)
  5. self.UniverseSettings.Resolution = Resolution.Daily
  6. self.SetBenchmark("SPY")
  7. self.universe_selection = UniverseSelection(self, self.logger)
  1. # UniverseSelection.py
  2. from AlgorithmImports import *
  3. import parameters as params
  4. class UniverseSelection:
  5. def __init__(self, algorithm, logger):
  6. self.algorithm = algorithm
  7. self.logger = logger
  8. self.logger.log("Starting UniverseSelection initialization...")
  9. try:
  10. self.history = {}
  11. self.moving_averages_200 = {}
  12. self.moving_averages_20 = {}
  13. self.previous_close = {}
  14. self.atr = {}
  15. self.volatility = {}
  16. self.avg_volatility = {}
  17. self.logger.log("Finished UniverseSelection initialization.")
  18. except Exception as e:
  19. self.logger.log(f"Error during UniverseSelection initialization: {str(e)}")
  20. raise e
  21. def CoarseSelectionFunction(self, coarse):
  22. selected: List[str] = []
  23. for stock in coarse:
  24. if stock.Price > params.PRICE_THRESHOLD and stock.DollarVolume > params.DOLLAR_VOLUME_THRESHOLD:
  25. selected.append(stock.Symbol)
  26. # Initialize the historical data if not already done
  27. if stock.Symbol not in self.history:
  28. self.InitializeIndicators(stock.Symbol)
  29. return selected
  30. def InitializeIndicators(self, symbol):
  31. # Fetch the historical data
  32. history = self.algorithm.History(symbol, params.MOVING_AVERAGE_200_DAYS, Resolution.Daily)
  33. if not history.empty:
  34. # Calculate and store moving averages
  35. self.history[symbol] = history['close']
  36. self.moving_averages_200[symbol] = \
  37. history['close'].rolling(window=params.MOVING_AVERAGE_200_DAYS).mean().iloc[-1]
  38. self.moving_averages_20[symbol] = \
  39. history['close'].rolling(window=params.MOVING_AVERAGE_20_DAYS).mean().iloc[-1]
  40. self.previous_close[symbol] = history['close'].iloc[-2]
  41. # Calculate ATR (Average True Range)
  42. high = history['high']
  43. low = history['low']
  44. close = history['close'].shift(1)
  45. tr = (high - low).combine(close - high, max).combine(close - low, max)
  46. self.atr[symbol] = tr.rolling(window=14).mean().iloc[-1]
  47. # Calculate daily returns for volatility
  48. daily_returns = history['close'].pct_change().dropna()
  49. # Calculate and store volatility
  50. self.volatility[symbol] = daily_returns.std() * (252 ** 0.5) # Annualized volatility
  51. self.avg_volatility[symbol] = daily_returns.rolling(window=params.VOLATILITY_LOOKBACK_DAYS).std().mean() * (
  52. 252 ** 0.5)
  53. def UpdateDailyIndicators(self):
  54. # This function will be called at the end of each trading day to update indicators
  55. for symbol in self.algorithm.Securities.Keys:
  56. history = self.algorithm.History(symbol, params.MOVING_AVERAGE_200_DAYS, Resolution.Daily)
  57. if not history.empty:
  58. self.history[symbol] = history['close']
  59. self.moving_averages_200[symbol] = \
  60. history['close'].rolling(window=params.MOVING_AVERAGE_200_DAYS).mean().iloc[-1]
  61. self.moving_averages_20[symbol] = \
  62. history['close'].rolling(window=params.MOVING_AVERAGE_20_DAYS).mean().iloc[-1]
  63. self.previous_close[symbol] = history['close'].iloc[-2]
  64. # Update ATR
  65. high = history['high']
  66. low = history['low']
  67. close = history['close'].shift(1)
  68. tr = (high - low).combine(close - high, max).combine(close - low, max)
  69. self.atr[symbol] = tr.rolling(window=14).mean().iloc[-1]
  70. # Update volatility
  71. daily_returns = history['close'].pct_change().dropna()
  72. self.volatility[symbol] = daily_returns.std() * (252 ** 0.5)
  73. self.avg_volatility[symbol] = daily_returns.rolling(
  74. window=params.VOLATILITY_LOOKBACK_DAYS).std().mean() * (252 ** 0.5)
  75. def GetSelectedStocks(self):
  76. abnormal_dollar_volume_stocks = []
  77. for symbol in self.algorithm.Securities.Keys:
  78. if symbol in self.history:
  79. avg_dollar_volume = self.history[symbol].mean()
  80. current_dollar_volume = self.algorithm.Securities[symbol].Price * self.algorithm.Securities[
  81. symbol].Volume
  82. price_increase = (self.algorithm.Securities[symbol].Price - self.previous_close[symbol]) / \
  83. self.previous_close[symbol]
  84. if (current_dollar_volume > 2 * avg_dollar_volume and
  85. self.algorithm.Securities[symbol].Price > self.moving_averages_200[symbol] and
  86. self.algorithm.Securities[symbol].Price > self.moving_averages_20[symbol] and
  87. price_increase > params.MIN_PRICE_INCREASE_PERCENT and
  88. self.volatility[symbol] <= params.MAX_VOLATILITY_MULTIPLIER * self.avg_volatility[symbol]):
  89. abnormal_dollar_volume_stocks.append(symbol)
  90. return abnormal_dollar_volume_stocks
+ Expand

Author

Dharmesh Khalasi

September 2024