| Overall Statistics |
|
Total Orders 10249 Average Win 1.49% Average Loss -1.02% Compounding Annual Return 34.085% Drawdown 65.600% Expectancy 0.289 Start Equity 100000 End Equity 581919.70 Net Profit 481.920% Sharpe Ratio 0.734 Sortino Ratio 0.787 Probabilistic Sharpe Ratio 17.574% Loss Rate 47% Win Rate 53% Profit-Loss Ratio 1.45 Alpha 0.211 Beta 1.017 Annual Standard Deviation 0.43 Annual Variance 0.185 Information Ratio 0.536 Tracking Error 0.396 Treynor Ratio 0.31 Total Fees $4566.16 Estimated Strategy Capacity $490000000.00 Lowest Capacity Asset EDA TGRALZT9E5ID Portfolio Turnover 8.55% |
from AlgorithmImports import *
class MomentumStockPortfolioAlgorithm(QCAlgorithm):
def Initialize(self):
self.SetStartDate(2019, 1, 1)
self.SetEndDate(2025, 1, 1)
self.SetCash(100000)
self.Debug("Algorithm initialized with start date: 2023-01-01 and cash: 100000")
# Set warm-up period
self.SetWarmUp(252, Resolution.Daily)
# Add SPY for market regime filtering
self.spy = self.AddEquity("SPY", Resolution.Daily).Symbol
self.spy_100MA = self.SMA(self.spy, 100, Resolution.Daily)
# Dictionary to store SymbolData for each symbol
self.symbol_data_by_symbol: Dict[Symbol, SymbolData] = {}
# List to track the selected universe
self.selected_symbols: List[Symbol] = []
# List to track the top momentum stocks
self.momentum_symbols: List[Symbol] = []
# Universe Settings
self.UniverseSettings.Resolution = Resolution.Daily
self.UniverseSettings.ExtendedMarketHours = False
self.UniverseSettings.DataNormalizationMode = DataNormalizationMode.SplitAdjusted
self.Debug(f"Universe settings configured: Resolution = {self.UniverseSettings.Resolution}, "
f"ExtendedMarketHours = {self.UniverseSettings.ExtendedMarketHours}, "
f"DataNormalizationMode = {self.UniverseSettings.DataNormalizationMode}")
# Universe schedule (first trading day of each month)
self.UniverseSettings.Schedule.On(self.DateRules.MonthStart())
# Add a custom universe
self.AddUniverse(self.filter_universe, self.log_filtered_universe)
# Schedule daily momentum calculations
self.Schedule.On(self.DateRules.EveryDay(self.spy),
self.TimeRules.BeforeMarketOpen(self.spy, 15),
self.CalculateMomentumScores)
def filter_universe(self, fundamental):
""" Filters the universe based on custom rules """
return [
x.Symbol for x in sorted(
[
f for f in fundamental
if f.HasFundamentalData
and f.Price > 10
and f.DollarVolume > 10_000_000
and f.MarketCap > 500_000_000
],
key=lambda f: f.DollarVolume,
reverse=True
)[:100] # Limit to top 50 symbols by DollarVolume
]
def log_filtered_universe(self, selected):
""" Logs the filtered universe and updates the selected symbols list """
selected_list = list(selected) # Convert selected to a list
tickers = [s.Symbol.Value for s in selected_list] # Extract tickers from the Symbol property of Fundamental objects
self.Debug(f"Selected {len(selected_list)} stocks in FundamentalSelection: {', '.join(tickers)}")
self.selected_symbols = [s.Symbol for s in selected_list] # Update selectedSymbols with Symbol objects
return [s.Symbol for s in selected_list] # Return only Symbol objects
def CalculateMomentumScores(self):
if self.IsWarmingUp:
return # Skip trading until warm-up is complete
"""Calculate momentum scores for the selected universe using Smoothness Ratio and Maximum Drawdown."""
results = []
for symbol in self.selected_symbols:
if symbol not in self.symbol_data_by_symbol:
self.symbol_data_by_symbol[symbol] = SymbolData(self, symbol)
data = self.symbol_data_by_symbol[symbol]
if data.IsReady:
# Fetch historical data for 3 months
history = self.History(symbol, 63, Resolution.Daily)['close']
if len(history) < 63:
continue
# Calculate Momentum 3-1
price_t_3 = history.iloc[-63] # Price 3 months ago
price_t_1 = history.iloc[-21] # Price 1 month ago
momentum_3_1 = ((price_t_1 - price_t_3) / price_t_3) * 100 # Convert to percentage
if momentum_3_1 < 0:
continue # Exclude stocks with negative momentum
# Calculate Maximum Drawdown (MDD)
running_max = history.cummax()
drawdowns = (history - running_max) / running_max
max_drawdown = drawdowns.min()
if max_drawdown < -0.05: # Exclude stocks with drawdown > 20%
continue
# Calculate Smoothness Ratio
daily_returns = history.pct_change().dropna()
rolling_std = daily_returns.std()
rolling_std_pct = rolling_std * 100 # Convert to percentage
smoothness_ratio = momentum_3_1 / rolling_std_pct if rolling_std_pct > 0 else 0
# Append results
results.append({
"Symbol": symbol,
"Momentum3-1": momentum_3_1,
"SmoothnessRatio": smoothness_ratio,
"MaxDrawdown": max_drawdown
})
# Sort results by Smoothness Ratio (descending)
self.momentum_scores = sorted(results, key=lambda x: x["SmoothnessRatio"], reverse=True)
# Log top 10 symbols and their metrics
top_10 = self.momentum_scores[:10]
log_message = "Top 10 Momentum Stocks (3-1): "
for rank, result in enumerate(top_10):
log_message += f"Rank {rank + 1}: {result['Symbol'].Value} (Momentum: {result['Momentum3-1']:.2f}%, "
log_message += f"Smoothness: {result['SmoothnessRatio']:.2f}, Drawdown: {result['MaxDrawdown']:.2f}); "
self.Debug(log_message)
# Update momentum symbols
self.momentum_symbols = [result["Symbol"] for result in top_10]
def PositionSizing(self, price):
""" Calculate position sizing based on portfolio value and price """
cash_to_invest = self.Portfolio.TotalPortfolioValue * 0.1 # 10% allocation
return int(cash_to_invest / price)
def OnData(self, data):
if self.IsWarmingUp:
return # Skip trading until warm-up is complete
# Log momentum symbols ready for trading tomorrow
if self.momentum_symbols:
self.Debug(f"Momentum symbols calculated today: {[symbol.Value for symbol in self.momentum_symbols]}")
else:
self.Debug("No momentum symbols calculated today.")
# List to track symbols meeting the trading criteria
ready_to_trade = []
# ===== Manage Existing Portfolio Positions =====
for holding in self.Portfolio.Values:
symbol = holding.Symbol
if not holding.Invested:
continue
if symbol not in self.symbol_data_by_symbol or self.symbol_data_by_symbol[symbol] is None:
continue
symbolData = self.symbol_data_by_symbol[symbol]
if not symbolData.IsReady:
continue
price = symbolData.Price
sma50 = symbolData.SMA50.Current.Value
if price < sma50: # Exit logic: Price drops below SMA50
self.MarketOnOpenOrder(symbol, -holding.Quantity) # Exit using MOO
self.Log(f"Exiting position in {symbol} with MOO order - Price: {price:.2f}, 50 SMA: {sma50:.2f}")
else:
self.Log(f"Managing position in {symbol} - Holding valid: Price: {price:.2f} > SMA50: {sma50:.2f}")
# ===== Check New Positions Based on Coarse and Fine Selection =====
for symbol, symbolData in self.symbol_data_by_symbol.items():
if symbolData is None or not symbolData.IsReady:
continue
symbolData.Update(data)
price = symbolData.Price
sma50 = symbolData.SMA50.Current.Value
sma150 = symbolData.SMA150.Current.Value
sma200 = symbolData.SMA200.Current.Value
high_52 = symbolData.High52.Current.Value
low_52 = symbolData.Low52.Current.Value
threshold_30_above_low = low_52 * 1.3
threshold_25_within_high = high_52 * 0.75
if price > sma50 > sma150 > sma200:
if price > threshold_30_above_low and price >= threshold_25_within_high:
# Add to the list of symbols ready to trade
if not self.Portfolio[symbol].Invested:
ready_to_trade.append(symbol)
quantity = self.PositionSizing(price)
if quantity > 0:
self.MarketOnOpenOrder(symbol, quantity)
symbolData.SetLong()
self.Log(f"TTP: True - Entered long position in {symbol} - Price: {price:.2f} > SMA50: {sma50:.2f} > SMA150 {sma150:.2f} > SMA200 {sma200:.2f}. Price > 52W Low: {low_52:.2f} by 30% and Within 52W High: {high_52:.2f} by 25%")
# Log the symbols ready to trade tomorrow
if ready_to_trade:
self.Debug(f"Symbols meeting trading criteria today, ready for tomorrow: {[symbol.Value for symbol in ready_to_trade]}")
else:
self.Debug("No symbols meet the trading criteria today for new positions.")
def is_bullish_market(self):
"""Check if the market is bullish based on SPY SMA100."""
if not self.spy_100MA.IsReady:
return False
spy_price = self.Securities[self.spy].Price
return spy_price > self.spy_100MA.Current.Value
def OnSecuritiesChanged(self, changes):
# Handle added securities
for security in changes.AddedSecurities:
symbol = security.Symbol
if symbol not in self.symbol_data_by_symbol:
symbolData = SymbolData(self, symbol)
security.SetSlippageModel(ConstantSlippageModel(0.001)) # Example: Setting slippage model
self.symbol_data_by_symbol[symbol] = symbolData
# Handle removed securities
for security in changes.RemovedSecurities:
symbol = security.Symbol
if symbol in self.symbol_data_by_symbol:
try:
self.symbol_data_by_symbol[symbol].Dispose()
except Exception as e:
self.Debug(f"Error disposing SymbolData for {symbol.Value}: {e}")
finally:
del self.symbol_data_by_symbol[symbol]
def on_warmup_finished(self):
self.Debug("OnWarmUpFinished method triggered.")
self.Debug("Warm-up finished. Running momentum score calculation.")
self.CalculateMomentumScores()
class SymbolData:
def __init__(self, algorithm, symbol):
self.algorithm = algorithm
self.symbol = symbol
self.Price = 0
self.IsLong = False
# Initialize indicators
self.ATR = algorithm.ATR(symbol, 20, MovingAverageType.Simple, Resolution.Daily)
self.SMA50 = algorithm.SMA(symbol, 50, Resolution.Daily)
self.SMA150 = algorithm.SMA(symbol, 150, Resolution.Daily)
self.SMA200 = algorithm.SMA(symbol, 200, Resolution.Daily)
self.High52 = algorithm.MAX(symbol, 252, Resolution.Daily)
self.Low52 = algorithm.MIN(symbol, 252, Resolution.Daily)
self.Reset()
@property
def IsReady(self):
return (
self.ATR.IsReady and
self.SMA50.IsReady and
self.SMA150.IsReady and
self.SMA200.IsReady and
self.High52 is not None and self.High52.IsReady and
self.Low52 is not None and self.Low52.IsReady
)
def Update(self, data):
if self.symbol in data.Bars:
self.Price = data.Bars[self.symbol].Close
def SetLong(self):
self.IsLong = True
def Reset(self):
self.IsLong = False
def Dispose(self):
# Deregister indicators to stop automatic updates
self.algorithm.deregister_indicator(self.ATR)
self.algorithm.deregister_indicator(self.SMA50)
self.algorithm.deregister_indicator(self.SMA150)
self.algorithm.deregister_indicator(self.SMA200)
if self.High52:
self.algorithm.deregister_indicator(self.High52)
if self.Low52:
self.algorithm.deregister_indicator(self.Low52)
# Reset attributes
self.IsLong = False
self.Price = 0
from AlgorithmImports import *
class MomentumStockPortfolioAlgorithm(QCAlgorithm):
def Initialize(self):
self.SetStartDate(2019, 1, 1)
self.SetEndDate(2025, 1, 1)
self.SetCash(100000)
self.Debug("Algorithm initialized with start date: 2023-01-01 and cash: 100000")
# Set warm-up period
self.SetWarmUp(7, Resolution.Daily)
# Add SPY for market regime filtering
self.spy = self.AddEquity("SPY", Resolution.Daily).Symbol
self.spy_100MA = self.SMA(self.spy, 100, Resolution.Daily)
# Warm up indicators for market regime filtering
self.warm_up_indicator(self.spy, self.spy_100MA, Resolution.Daily)
# Dictionary to store SymbolData for each symbol
self.symbol_data_by_symbol: Dict[Symbol, SymbolData] = {}
# List to track the selected universe and momentum stocks
self.selected_symbols: List[Symbol] = []
self.momentum_symbols: List[Symbol] = []
self.exit_symbols: List[Symbol] = []
# Universe Settings
self.UniverseSettings.Resolution = Resolution.Daily
self.UniverseSettings.ExtendedMarketHours = False
self.UniverseSettings.DataNormalizationMode = DataNormalizationMode.SplitAdjusted
self.UniverseSettings.Asynchronous = True
self.Debug(f"Universe settings configured: Resolution = {self.UniverseSettings.Resolution}, "
f"ExtendedMarketHours = {self.UniverseSettings.ExtendedMarketHours}, "
f"DataNormalizationMode = {self.UniverseSettings.DataNormalizationMode}")
# Universe schedule (first trading day of each month)
self.UniverseSettings.Schedule.On(self.DateRules.WeekStart())
# Use QuantConnect's built-in DollarVolume Universe for liquidity filtering
self.AddUniverse(self.Universe.DollarVolume.Top(20)) # Limiting to top 20 for testing
# Schedule weekly momentum calculations
self.Schedule.On(self.DateRules.WeekStart(self.spy),
self.TimeRules.BeforeMarketOpen(self.spy, 15),
self.CalculateMomentumScores)
# Schedule placing MarketOnOpen orders
self.Schedule.On(
self.DateRules.EveryDay(self.spy),
self.TimeRules.BeforeMarketOpen(self.spy, 2),
self.PlaceMarketOnOpenOrders
)
def on_warmup_finished(self):
self.Debug("OnWarmUpFinished method triggered.")
self.Debug("Warm-up finished. Running momentum score calculation.")
self.CalculateMomentumScores()
def is_bullish_market(self):
"""Check if the market is bullish based on SPY SMA100."""
if not self.spy_100MA.IsReady:
return False
spy_price = self.Securities[self.spy].Price
return spy_price > self.spy_100MA.Current.Value
def OnSecuritiesChanged(self, changes):
"""Handle securities added and removed from the universe."""
for security in changes.AddedSecurities:
symbol = security.Symbol
if symbol not in self.symbol_data_by_symbol:
# Initialize SymbolData
symbol_data = SymbolData(self, symbol)
self.symbol_data_by_symbol[symbol] = symbol_data
for security in changes.RemovedSecurities:
symbol = security.Symbol
if symbol in self.symbol_data_by_symbol:
self.symbol_data_by_symbol[symbol].Dispose()
del self.symbol_data_by_symbol[symbol]
self.Debug(f"Removed security: {symbol.Value}")
def CalculateMomentumScores(self):
if self.IsWarmingUp:
return # Skip trading until warm-up is complete
"""Calculate momentum scores using ROC42 and ATR40."""
top_momentum = [] # Maintain a fixed-size list of the top 10 momentum symbols
for symbol, symbol_data in self.symbol_data_by_symbol.items():
if not symbol_data.IsReady:
continue
# Calculate Momentum Score: ROC42 / ATR40
roc = symbol_data.ROC42.Current.Value
atr = symbol_data.ATR40.Current.Value
if atr > 0: # Avoid division by zero
momentum_score = roc / atr
# Insert into the sorted list of top momentum symbols
top_momentum.append((symbol, momentum_score))
top_momentum.sort(key=lambda x: x[1], reverse=True) # Sort by momentum score descending
if len(top_momentum) > 10:
top_momentum.pop() # Keep only the top 10
# Log top 10 symbols
log_message = "Top 10 Momentum Stocks: "
for rank, (symbol, momentum_score) in enumerate(top_momentum):
log_message += f"Rank {rank + 1}: {symbol.Value} (Momentum Score: {momentum_score:.2f}); "
self.Debug(log_message)
# Update momentum symbols
self.momentum_symbols = [symbol for symbol, _ in top_momentum]
def OnData(self, data):
if self.IsWarmingUp:
return
# Market Regime Filter: Exit all positions if SPY is below its 100 SMA
if not self.is_bullish_market():
if any(self.Portfolio[symbol].Invested for symbol in self.Portfolio.Keys):
self.exit_symbols = [symbol for symbol in self.Portfolio.Keys if self.Portfolio[symbol].Invested]
self.Debug("Market regime bearish. Liquidating positions.")
self.momentum_symbols = [] # Reset momentum symbols
return
self.exit_symbols = [] # Reset symbols for exit
current_holdings = {symbol.Key for symbol in self.Portfolio if self.Portfolio[symbol.Key].Invested}
new_holdings = set(self.momentum_symbols)
for symbol, symbol_data in self.symbol_data_by_symbol.items():
if not symbol_data.IsReady:
continue
symbol_data.Update(data)
price = symbol_data.Price
high_52 = symbol_data.High52.Current.Value
low_52 = symbol_data.Low52.Current.Value
sma150 = symbol_data.SMA150.Current.Value
sma200 = symbol_data.SMA200.Current.Value
threshold_30_above_low = low_52 * 1.3
threshold_25_within_high = high_52 * 0.75
# Exit condition: Price below SMA50 or symbol no longer in top 10
if symbol in current_holdings and (price < symbol_data.SMA50.Current.Value or symbol not in new_holdings):
self.exit_symbols.append(symbol)
# Entry criteria: Ensure price meets thresholds
if symbol not in current_holdings and price > threshold_30_above_low and price <= threshold_25_within_high:
if price > sma150 > sma200:
self.momentum_symbols.append(symbol)
def PlaceMarketOnOpenOrders(self):
""" Place MarketOnOpen orders for exit symbols and new entries """
for symbol in self.exit_symbols:
holding = self.Portfolio[symbol]
if holding.Invested:
self.MarketOnOpenOrder(symbol, -holding.Quantity)
self.Log(f"Exiting {symbol.Value} - Price: {holding.AveragePrice:.2f}")
for symbol in self.momentum_symbols:
if symbol in self.symbol_data_by_symbol:
symbol_data = self.symbol_data_by_symbol[symbol]
price = symbol_data.Price
if price > symbol_data.SMA50.Current.Value: # Entry condition
quantity = self.PositionSizing(price)
if quantity > 0:
self.MarketOnOpenOrder(symbol, quantity)
self.Log(f"Entering {symbol.Value} - Quantity: {quantity}, Price: {price:.2f}")
def PositionSizing(self, price):
""" Calculate position sizing based on portfolio value and price """
if price <= 0: # Prevent division by zero
self.Debug("PositionSizing: Invalid price encountered. Skipping position sizing.")
return 0
cash_to_invest = self.Portfolio.TotalPortfolioValue * 0.1 # 10% allocation
return int(cash_to_invest / price)
class SymbolData:
def __init__(self, algorithm, symbol):
self.algorithm = algorithm
self.symbol = symbol
self.Price = 0
self.IsLong = False
# Add indicators
self.ATR40 = algorithm.ATR(symbol, 40, MovingAverageType.Simple, Resolution.Daily)
self.ROC42 = algorithm.ROC(symbol, 42, Resolution.Daily) # 42-day Rate of Change
self.SMA50 = algorithm.SMA(symbol, 50, Resolution.Daily)
self.SMA150 = algorithm.SMA(symbol, 150, Resolution.Daily)
self.SMA200 = algorithm.SMA(symbol, 200, Resolution.Daily)
self.High52 = algorithm.MAX(symbol, 252, Resolution.Daily)
self.Low52 = algorithm.MIN(symbol, 252, Resolution.Daily)
# Warm up indicators using QuantConnect's warm_up_indicator method
algorithm.warm_up_indicator(symbol, self.ATR40, Resolution.Daily)
algorithm.warm_up_indicator(symbol, self.ROC42, Resolution.Daily)
algorithm.warm_up_indicator(symbol, self.SMA50, Resolution.Daily)
algorithm.warm_up_indicator(symbol, self.SMA150, Resolution.Daily)
algorithm.warm_up_indicator(symbol, self.SMA200, Resolution.Daily)
algorithm.warm_up_indicator(symbol, self.High52, Resolution.Daily)
algorithm.warm_up_indicator(symbol, self.Low52, Resolution.Daily)
# Register indicators for auto-update
algorithm.RegisterIndicator(symbol, self.ATR40, Resolution.Daily)
algorithm.RegisterIndicator(symbol, self.ROC42, Resolution.Daily)
algorithm.RegisterIndicator(symbol, self.SMA50, Resolution.Daily)
algorithm.RegisterIndicator(symbol, self.SMA150, Resolution.Daily)
algorithm.RegisterIndicator(symbol, self.SMA200, Resolution.Daily)
algorithm.RegisterIndicator(symbol, self.High52, Resolution.Daily)
algorithm.RegisterIndicator(symbol, self.Low52, Resolution.Daily)
self.Reset()
@property
def IsReady(self):
return (
self.ATR40.IsReady and
self.ROC42.IsReady and
self.SMA50.IsReady and
self.SMA150.IsReady and
self.SMA200.IsReady and
self.High52.IsReady and
self.Low52.IsReady
)
def Update(self, data):
if self.symbol in data.Bars:
self.Price = data.Bars[self.symbol].Close
def SetLong(self):
self.IsLong = True
def Reset(self):
self.IsLong = False
def Dispose(self):
# Deregister indicators to stop automatic updates
self.algorithm.deregister_indicator(self.ATR40)
self.algorithm.deregister_indicator(self.ROC42)
self.algorithm.deregister_indicator(self.SMA50)
self.algorithm.deregister_indicator(self.SMA150)
self.algorithm.deregister_indicator(self.SMA200)
if self.High52:
self.algorithm.deregister_indicator(self.High52)
if self.Low52:
self.algorithm.deregister_indicator(self.Low52)
# Reset attributes
self.IsLong = False
self.Price = 0
from AlgorithmImports import *
class MomentumStockPortfolioAlgorithm(QCAlgorithm):
def Initialize(self):
self.SetStartDate(2019, 1, 1)
self.SetEndDate(2025, 1, 1)
self.SetCash(100000)
self.Debug("Algorithm initialized with start date: 2023-01-01 and cash: 100000")
# Set warm-up period
self.SetWarmUp(7, Resolution.Daily)
# Add SPY for market regime filtering
self.spy = self.AddEquity("SPY", Resolution.Daily).Symbol
self.spy_100MA = self.SMA(self.spy, 100, Resolution.Daily)
# Warm up indicators for market regime filtering
self.warm_up_indicator(self.spy, self.spy_100MA, Resolution.Daily)
# Dictionary to store SymbolData for each symbol
self.symbol_data_by_symbol: Dict[Symbol, SymbolData] = {}
# List to track the selected universe and momentum stocks
self.selected_symbols: List[Symbol] = []
self.momentum_symbols: List[Symbol] = []
self.exit_symbols: List[Symbol] = []
# Universe Settings
self.UniverseSettings.Resolution = Resolution.Daily
self.UniverseSettings.ExtendedMarketHours = False
self.UniverseSettings.DataNormalizationMode = DataNormalizationMode.SplitAdjusted
self.UniverseSettings.Asynchronous = True
self.Debug(f"Universe settings configured: Resolution = {self.UniverseSettings.Resolution}, "
f"ExtendedMarketHours = {self.UniverseSettings.ExtendedMarketHours}, "
f"DataNormalizationMode = {self.UniverseSettings.DataNormalizationMode}")
# Universe schedule (first trading day of each month)
self.UniverseSettings.Schedule.On(self.DateRules.WeekStart())
# Use QuantConnect's built-in DollarVolume Universe for liquidity filtering
self.AddUniverse(self.Universe.DollarVolume.Top(20)) # Limiting to top 20 for testing
# Schedule weekly momentum calculations
self.Schedule.On(self.DateRules.WeekStart(self.spy),
self.TimeRules.BeforeMarketOpen(self.spy, 15),
self.CalculateMomentumScores)
# Schedule placing MarketOnOpen orders
self.Schedule.On(
self.DateRules.EveryDay(self.spy),
self.TimeRules.BeforeMarketOpen(self.spy, 2),
self.PlaceMarketOnOpenOrders
)
def on_warmup_finished(self):
self.Debug("OnWarmUpFinished method triggered.")
self.Debug("Warm-up finished. Running momentum score calculation.")
self.CalculateMomentumScores()
def is_bullish_market(self):
"""Check if the market is bullish based on SPY SMA100."""
if not self.spy_100MA.IsReady:
return False
spy_price = self.Securities[self.spy].Price
return spy_price > self.spy_100MA.Current.Value
def OnSecuritiesChanged(self, changes):
"""Handle securities added and removed from the universe."""
for security in changes.AddedSecurities:
symbol = security.Symbol
if symbol not in self.symbol_data_by_symbol:
# Initialize SymbolData
symbol_data = SymbolData(self, symbol)
self.symbol_data_by_symbol[symbol] = symbol_data
for security in changes.RemovedSecurities:
symbol = security.Symbol
if symbol in self.symbol_data_by_symbol:
self.symbol_data_by_symbol[symbol].Dispose()
del self.symbol_data_by_symbol[symbol]
self.Debug(f"Removed security: {symbol.Value}")
def CalculateMomentumScores(self):
if self.IsWarmingUp:
return # Skip trading until warm-up is complete
"""Calculate momentum scores using ROC42 and ATR40."""
top_momentum = [] # Maintain a fixed-size list of the top 10 momentum symbols
for symbol, symbol_data in self.symbol_data_by_symbol.items():
if not symbol_data.IsReady:
continue
# Calculate Momentum Score: ROC42 / ATR40
roc = symbol_data.ROC42.Current.Value
atr = symbol_data.ATR40.Current.Value
if atr > 0: # Avoid division by zero
momentum_score = roc / atr
# Insert into the sorted list of top momentum symbols
top_momentum.append((symbol, momentum_score))
top_momentum.sort(key=lambda x: x[1], reverse=True) # Sort by momentum score descending
if len(top_momentum) > 10:
top_momentum.pop() # Keep only the top 10
# Log top 10 symbols
log_message = "Top 10 Momentum Stocks: "
for rank, (symbol, momentum_score) in enumerate(top_momentum):
log_message += f"Rank {rank + 1}: {symbol.Value} (Momentum Score: {momentum_score:.2f}); "
self.Debug(log_message)
# Update momentum symbols
self.momentum_symbols = [symbol for symbol, _ in top_momentum]
def OnData(self, data):
if self.IsWarmingUp:
return
# Market Regime Filter: Exit all positions if SPY is below its 100 SMA
if not self.is_bullish_market():
if any(self.Portfolio[symbol].Invested for symbol in self.Portfolio.Keys):
self.exit_symbols = [symbol for symbol in self.Portfolio.Keys if self.Portfolio[symbol].Invested]
self.Debug("Market regime bearish. Liquidating positions.")
self.momentum_symbols = [] # Reset momentum symbols
return
self.exit_symbols = [] # Reset symbols for exit
current_holdings = {symbol.Key for symbol in self.Portfolio if self.Portfolio[symbol.Key].Invested}
new_holdings = set(self.momentum_symbols)
for symbol, symbol_data in self.symbol_data_by_symbol.items():
if not symbol_data.IsReady:
continue
symbol_data.Update(data)
price = symbol_data.Price
high_52 = symbol_data.High52.Current.Value
low_52 = symbol_data.Low52.Current.Value
sma150 = symbol_data.SMA150.Current.Value
sma200 = symbol_data.SMA200.Current.Value
threshold_30_above_low = low_52 * 1.3
threshold_25_within_high = high_52 * 0.75
# Exit condition: Price below SMA50 or symbol no longer in top 10
if symbol in current_holdings and (price < symbol_data.SMA50.Current.Value or symbol not in new_holdings):
self.exit_symbols.append(symbol)
# Entry criteria: Ensure price meets thresholds
if symbol not in current_holdings and price > threshold_30_above_low and price <= threshold_25_within_high:
if price > sma150 > sma200:
self.momentum_symbols.append(symbol)
def PlaceMarketOnOpenOrders(self):
""" Place MarketOnOpen orders for exit symbols and new entries """
for symbol in self.exit_symbols:
holding = self.Portfolio[symbol]
if holding.Invested:
self.MarketOnOpenOrder(symbol, -holding.Quantity)
self.Log(f"Exiting {symbol.Value} - Price: {holding.AveragePrice:.2f}")
for symbol in self.momentum_symbols:
if symbol in self.symbol_data_by_symbol:
symbol_data = self.symbol_data_by_symbol[symbol]
price = symbol_data.Price
if price > symbol_data.SMA50.Current.Value: # Entry condition
quantity = self.PositionSizing(price)
if quantity > 0:
self.MarketOnOpenOrder(symbol, quantity)
self.Log(f"Entering {symbol.Value} - Quantity: {quantity}, Price: {price:.2f}")
def PositionSizing(self, price):
""" Calculate position sizing based on portfolio value and price """
if price <= 0: # Prevent division by zero
self.Debug("PositionSizing: Invalid price encountered. Skipping position sizing.")
return 0
cash_to_invest = self.Portfolio.TotalPortfolioValue * 0.1 # 10% allocation
return int(cash_to_invest / price)
class SymbolData:
def __init__(self, algorithm, symbol):
self.algorithm = algorithm
self.symbol = symbol
self.Price = 0
self.IsLong = False
# Add indicators
self.ATR40 = algorithm.ATR(symbol, 40, MovingAverageType.Simple, Resolution.Daily)
self.ROC42 = algorithm.ROC(symbol, 42, Resolution.Daily) # 42-day Rate of Change
self.SMA50 = algorithm.SMA(symbol, 50, Resolution.Daily)
self.SMA150 = algorithm.SMA(symbol, 150, Resolution.Daily)
self.SMA200 = algorithm.SMA(symbol, 200, Resolution.Daily)
self.High52 = algorithm.MAX(symbol, 252, Resolution.Daily)
self.Low52 = algorithm.MIN(symbol, 252, Resolution.Daily)
# Warm up indicators using QuantConnect's warm_up_indicator method
algorithm.warm_up_indicator(symbol, self.ATR40, Resolution.Daily)
algorithm.warm_up_indicator(symbol, self.ROC42, Resolution.Daily)
algorithm.warm_up_indicator(symbol, self.SMA50, Resolution.Daily)
algorithm.warm_up_indicator(symbol, self.SMA150, Resolution.Daily)
algorithm.warm_up_indicator(symbol, self.SMA200, Resolution.Daily)
algorithm.warm_up_indicator(symbol, self.High52, Resolution.Daily)
algorithm.warm_up_indicator(symbol, self.Low52, Resolution.Daily)
# Register indicators for auto-update
algorithm.RegisterIndicator(symbol, self.ATR40, Resolution.Daily)
algorithm.RegisterIndicator(symbol, self.ROC42, Resolution.Daily)
algorithm.RegisterIndicator(symbol, self.SMA50, Resolution.Daily)
algorithm.RegisterIndicator(symbol, self.SMA150, Resolution.Daily)
algorithm.RegisterIndicator(symbol, self.SMA200, Resolution.Daily)
algorithm.RegisterIndicator(symbol, self.High52, Resolution.Daily)
algorithm.RegisterIndicator(symbol, self.Low52, Resolution.Daily)
self.Reset()
@property
def IsReady(self):
return (
self.ATR40.IsReady and
self.ROC42.IsReady and
self.SMA50.IsReady and
self.SMA150.IsReady and
self.SMA200.IsReady and
self.High52.IsReady and
self.Low52.IsReady
)
def Update(self, data):
if self.symbol in data.Bars:
self.Price = data.Bars[self.symbol].Close
def SetLong(self):
self.IsLong = True
def Reset(self):
self.IsLong = False
def Dispose(self):
# Deregister indicators to stop automatic updates
self.algorithm.deregister_indicator(self.ATR40)
self.algorithm.deregister_indicator(self.ROC42)
self.algorithm.deregister_indicator(self.SMA50)
self.algorithm.deregister_indicator(self.SMA150)
self.algorithm.deregister_indicator(self.SMA200)
if self.High52:
self.algorithm.deregister_indicator(self.High52)
if self.Low52:
self.algorithm.deregister_indicator(self.Low52)
# Reset attributes
self.IsLong = False
self.Price = 0
from AlgorithmImports import *
class MomentumStockPortfolioAlgorithm(QCAlgorithm):
def Initialize(self):
self.SetStartDate(2019, 1, 1)
self.SetEndDate(2025, 1, 1)
self.SetCash(100000)
self.Debug("Algorithm initialized with start date: 2023-01-01 and cash: 100000")
# Set warm-up period
self.SetWarmUp(252, Resolution.Daily)
# Add SPY for market regime filtering
self.spy = self.AddEquity("SPY", Resolution.Daily).Symbol
self.spy_100MA = self.SMA(self.spy, 100, Resolution.Daily)
# Dictionary to store SymbolData for each symbol
self.symbol_data_by_symbol: Dict[Symbol, SymbolData] = {}
# List to track the selected universe and momentum stocks
self.selected_symbols: List[Symbol] = []
self.momentum_symbols: List[Symbol] = []
self.exit_symbols: List[Symbol] = []
# Universe Settings
self.UniverseSettings.Resolution = Resolution.Daily
self.UniverseSettings.ExtendedMarketHours = False
self.UniverseSettings.DataNormalizationMode = DataNormalizationMode.SplitAdjusted
self.UniverseSettings.Asynchronous = True
self.Debug(f"Universe settings configured: Resolution = {self.UniverseSettings.Resolution}, "
f"ExtendedMarketHours = {self.UniverseSettings.ExtendedMarketHours}, "
f"DataNormalizationMode = {self.UniverseSettings.DataNormalizationMode}")
# Use QuantConnect's built-in DollarVolume Universe for liquidity filtering
self.AddUniverse(self.Universe.DollarVolume.Top(500))
# Schedule weekly momentum calculations
self.Schedule.On(self.DateRules.WeekStart(self.spy),
self.TimeRules.BeforeMarketOpen(self.spy, 15),
self.CalculateMomentumScores)
# Schedule placing MarketOnOpen orders
self.Schedule.On(
self.DateRules.EveryDay(self.spy),
self.TimeRules.BeforeMarketOpen(self.spy, 2),
self.PlaceMarketOnOpenOrders
)
def OnSecuritiesChanged(self, changes):
"""Handle securities added and removed from the universe."""
for security in changes.AddedSecurities:
symbol = security.Symbol
if symbol not in self.symbol_data_by_symbol:
self.symbol_data_by_symbol[symbol] = SymbolData(self, symbol)
# self.Debug(f"Added security: {symbol.Value}")
for security in changes.RemovedSecurities:
symbol = security.Symbol
if symbol in self.symbol_data_by_symbol:
self.symbol_data_by_symbol[symbol].Dispose()
del self.symbol_data_by_symbol[symbol]
# self.Debug(f"Removed security: {symbol.Value}")
def CalculateMomentumScores(self):
if self.IsWarmingUp:
return # Skip trading until warm-up is complete
"""Calculate momentum scores for the selected universe using Smoothness Ratio and Maximum Drawdown."""
# Fetch historical data for all selected symbols in one batch
history = self.History(list(self.symbol_data_by_symbol.keys()), 63, Resolution.Daily)
if history.empty:
return # Exit if no data is returned
results = []
for symbol in self.symbol_data_by_symbol.keys():
if symbol not in history.index.levels[0]:
continue
symbol_history = history.loc[symbol]
if len(symbol_history) < 63:
continue
# Calculate Momentum 3-1
price_t_3 = symbol_history['close'].iloc[-63] # Price 3 months ago
price_t_1 = symbol_history['close'].iloc[-21] # Price 1 month ago
momentum_3_1 = ((price_t_1 - price_t_3) / price_t_3) * 100 # Convert to percentage
if momentum_3_1 < 0:
continue # Exclude stocks with negative momentum
# Calculate Maximum Drawdown (MDD)
running_max = symbol_history['close'].cummax()
drawdowns = (symbol_history['close'] - running_max) / running_max
max_drawdown = drawdowns.min()
if max_drawdown < -0.05: # Exclude stocks with drawdown > 5%
continue
# Calculate Smoothness Ratio
daily_returns = symbol_history['close'].pct_change().dropna()
rolling_std = daily_returns.std()
rolling_std_pct = rolling_std * 100 # Convert to percentage
smoothness_ratio = momentum_3_1 / rolling_std_pct if rolling_std_pct > 0 else 0
# Append only necessary data for ranking
results.append({
"Symbol": symbol,
"SmoothnessRatio": smoothness_ratio
})
# Sort results by Smoothness Ratio (descending) and pick top 10
self.momentum_scores = sorted(results, key=lambda x: x["SmoothnessRatio"], reverse=True)[:10]
# Log top 10 symbols
top_10 = self.momentum_scores
log_message = "Top 10 Momentum Stocks (3-1): "
for rank, result in enumerate(top_10):
log_message += f"Rank {rank + 1}: {result['Symbol'].Value} (Smoothness: {result['SmoothnessRatio']:.2f}); "
self.Debug(log_message)
# Update momentum symbols
self.momentum_symbols = [result["Symbol"] for result in top_10]
def OnData(self, data):
if self.IsWarmingUp:
return
self.exit_symbols = [] # Reset symbols for exit
current_holdings = {symbol.Key for symbol in self.Portfolio if self.Portfolio[symbol.Key].Invested}
new_holdings = set(self.momentum_symbols)
for symbol, symbol_data in self.symbol_data_by_symbol.items():
if not symbol_data.IsReady:
continue
symbol_data.Update(data)
price = symbol_data.Price
high_52 = symbol_data.High52.Current.Value
low_52 = symbol_data.Low52.Current.Value
sma150 = symbol_data.SMA150.Current.Value
sma200 = symbol_data.SMA200.Current.Value
threshold_30_above_low = low_52 * 1.3
threshold_25_within_high = high_52 * 0.75
# Exit condition: Price below SMA50 or symbol no longer in top 10
if symbol in current_holdings and (price < symbol_data.SMA50.Current.Value or symbol not in new_holdings):
self.exit_symbols.append(symbol)
# Entry criteria: Ensure price meets thresholds
if symbol not in current_holdings and price > threshold_30_above_low and price <= threshold_25_within_high:
if price > sma150 > sma200:
self.momentum_symbols.append(symbol)
def PlaceMarketOnOpenOrders(self):
""" Place MarketOnOpen orders for exit symbols and new entries """
for symbol in self.exit_symbols:
holding = self.Portfolio[symbol]
if holding.Invested:
self.MarketOnOpenOrder(symbol, -holding.Quantity)
self.Log(f"Exiting {symbol.Value} - Price: {holding.AveragePrice:.2f}")
for symbol in self.momentum_symbols:
if symbol in self.symbol_data_by_symbol:
symbol_data = self.symbol_data_by_symbol[symbol]
price = symbol_data.Price
if price > symbol_data.SMA50.Current.Value: # Entry condition
quantity = self.PositionSizing(price)
if quantity > 0:
self.MarketOnOpenOrder(symbol, quantity)
self.Log(f"Entering {symbol.Value} - Quantity: {quantity}, Price: {price:.2f}")
def PositionSizing(self, price):
""" Calculate position sizing based on portfolio value and price """
if price <= 0: # Prevent division by zero
self.Debug("PositionSizing: Invalid price encountered. Skipping position sizing.")
return 0
cash_to_invest = self.Portfolio.TotalPortfolioValue * 0.1 # 10% allocation
return int(cash_to_invest / price)
class SymbolData:
def __init__(self, algorithm, symbol):
self.algorithm = algorithm
self.symbol = symbol
self.Price = 0
self.IsLong = False
# Add indicators
self.ATR = algorithm.ATR(symbol, 20, MovingAverageType.Simple, Resolution.Daily)
self.SMA50 = algorithm.SMA(symbol, 50, Resolution.Daily)
self.SMA150 = algorithm.SMA(symbol, 150, Resolution.Daily)
self.SMA200 = algorithm.SMA(symbol, 200, Resolution.Daily)
self.High52 = algorithm.MAX(symbol, 252, Resolution.Daily)
self.Low52 = algorithm.MIN(symbol, 252, Resolution.Daily)
self.ROC63 = algorithm.ROC(symbol, 63, Resolution.Daily) # 63-day Rate of Change
self.Reset()
@property
def IsReady(self):
return (
self.ATR.IsReady and
self.SMA50.IsReady and
self.SMA150.IsReady and
self.SMA200.IsReady and
self.High52.IsReady and
self.Low52.IsReady and
self.ROC63.IsReady
)
def Update(self, data):
if self.symbol in data.Bars:
self.Price = data.Bars[self.symbol].Close
def SetLong(self):
self.IsLong = True
def Reset(self):
self.IsLong = False
def Dispose(self):
# Deregister indicators to stop automatic updates
self.algorithm.deregister_indicator(self.ATR)
self.algorithm.deregister_indicator(self.SMA50)
self.algorithm.deregister_indicator(self.SMA150)
self.algorithm.deregister_indicator(self.SMA200)
if self.High52:
self.algorithm.deregister_indicator(self.High52)
if self.Low52:
self.algorithm.deregister_indicator(self.Low52)
# Reset attributes
self.IsLong = False
self.Price = 0
from AlgorithmImports import *
class MomentumStockPortfolioAlgorithm(QCAlgorithm):
def Initialize(self):
self.SetStartDate(2019, 1, 1)
self.SetEndDate(2025, 1, 1)
self.SetCash(100000)
self.Debug("Algorithm initialized with start date: 2023-01-01 and cash: 100000")
# Set warm-up period
self.SetWarmUp(252, Resolution.Daily)
# Add SPY for market regime filtering
self.spy = self.AddEquity("SPY", Resolution.Daily).Symbol
self.spy_100MA = self.SMA(self.spy, 100, Resolution.Daily)
# Dictionary to store SymbolData for each symbol
self.symbol_data_by_symbol: Dict[Symbol, SymbolData] = {}
# List to track the selected universe and momentum stocks
self.selected_symbols: List[Symbol] = []
self.momentum_symbols: List[Symbol] = []
self.exit_symbols: List[Symbol] = []
# Universe Settings
self.UniverseSettings.Resolution = Resolution.Daily
self.UniverseSettings.ExtendedMarketHours = False
self.UniverseSettings.DataNormalizationMode = DataNormalizationMode.SplitAdjusted
self.UniverseSettings.Asynchronous = True
self.Debug(f"Universe settings configured: Resolution = {self.UniverseSettings.Resolution}, "
f"ExtendedMarketHours = {self.UniverseSettings.ExtendedMarketHours}, "
f"DataNormalizationMode = {self.UniverseSettings.DataNormalizationMode}")
# Use QuantConnect's built-in DollarVolume Universe for liquidity filtering
self.AddUniverse(self.Universe.DollarVolume.Top(100))
# Schedule weekly momentum calculations
self.Schedule.On(self.DateRules.WeekStart(self.spy),
self.TimeRules.BeforeMarketOpen(self.spy, 15),
self.CalculateMomentumScores)
# Schedule placing MarketOnOpen orders
self.Schedule.On(
self.DateRules.EveryDay(self.spy),
self.TimeRules.BeforeMarketOpen(self.spy, 2),
self.PlaceMarketOnOpenOrders
)
def OnSecuritiesChanged(self, changes):
"""Handle securities added and removed from the universe."""
for security in changes.AddedSecurities:
symbol = security.Symbol
if symbol not in self.symbol_data_by_symbol:
self.symbol_data_by_symbol[symbol] = SymbolData(self, symbol)
self.Debug(f"Added security: {symbol.Value}")
for security in changes.RemovedSecurities:
symbol = security.Symbol
if symbol in self.symbol_data_by_symbol:
self.symbol_data_by_symbol[symbol].Dispose()
del self.symbol_data_by_symbol[symbol]
self.Debug(f"Removed security: {symbol.Value}")
def CalculateMomentumScores(self):
if self.IsWarmingUp:
return # Skip trading until warm-up is complete
"""Calculate momentum scores using ROC42 and ATR40."""
top_momentum = [] # Maintain a fixed-size list of the top 10 momentum symbols
for symbol, symbol_data in self.symbol_data_by_symbol.items():
if not symbol_data.IsReady:
continue
# Calculate Momentum Score: ROC42 / ATR40
roc = symbol_data.ROC42.Current.Value
atr = symbol_data.ATR40.Current.Value
if atr > 0: # Avoid division by zero
momentum_score = roc / atr
# Insert into the sorted list of top momentum symbols
top_momentum.append((symbol, momentum_score))
top_momentum.sort(key=lambda x: x[1], reverse=True) # Sort by momentum score descending
if len(top_momentum) > 10:
top_momentum.pop() # Keep only the top 10
# Log top 10 symbols
log_message = "Top 10 Momentum Stocks: "
for rank, (symbol, momentum_score) in enumerate(top_momentum):
log_message += f"Rank {rank + 1}: {symbol.Value} (Momentum Score: {momentum_score:.2f}); "
self.Debug(log_message)
# Update momentum symbols
self.momentum_symbols = [symbol for symbol, _ in top_momentum]
def OnData(self, data):
if self.IsWarmingUp:
return
self.exit_symbols = [] # Reset symbols for exit
current_holdings = {symbol.Key for symbol in self.Portfolio if self.Portfolio[symbol.Key].Invested}
new_holdings = set(self.momentum_symbols)
for symbol, symbol_data in self.symbol_data_by_symbol.items():
if not symbol_data.IsReady:
continue
symbol_data.Update(data)
price = symbol_data.Price
high_52 = symbol_data.High52.Current.Value
low_52 = symbol_data.Low52.Current.Value
sma150 = symbol_data.SMA150.Current.Value
sma200 = symbol_data.SMA200.Current.Value
threshold_30_above_low = low_52 * 1.3
threshold_25_within_high = high_52 * 0.75
# Exit condition: Price below SMA50 or symbol no longer in top 10
if symbol in current_holdings and (price < symbol_data.SMA50.Current.Value or symbol not in new_holdings):
self.exit_symbols.append(symbol)
# Entry criteria: Ensure price meets thresholds
if symbol not in current_holdings and price > threshold_30_above_low and price <= threshold_25_within_high:
if price > sma150 > sma200:
self.momentum_symbols.append(symbol)
def PlaceMarketOnOpenOrders(self):
""" Place MarketOnOpen orders for exit symbols and new entries """
for symbol in self.exit_symbols:
holding = self.Portfolio[symbol]
if holding.Invested:
self.MarketOnOpenOrder(symbol, -holding.Quantity)
self.Log(f"Exiting {symbol.Value} - Price: {holding.AveragePrice:.2f}")
for symbol in self.momentum_symbols:
if symbol in self.symbol_data_by_symbol:
symbol_data = self.symbol_data_by_symbol[symbol]
price = symbol_data.Price
if price > symbol_data.SMA50.Current.Value: # Entry condition
quantity = self.PositionSizing(price)
if quantity > 0:
self.MarketOnOpenOrder(symbol, quantity)
self.Log(f"Entering {symbol.Value} - Quantity: {quantity}, Price: {price:.2f}")
def PositionSizing(self, price):
""" Calculate position sizing based on portfolio value and price """
if price <= 0: # Prevent division by zero
self.Debug("PositionSizing: Invalid price encountered. Skipping position sizing.")
return 0
cash_to_invest = self.Portfolio.TotalPortfolioValue * 0.1 # 10% allocation
return int(cash_to_invest / price)
class SymbolData:
def __init__(self, algorithm, symbol):
self.algorithm = algorithm
self.symbol = symbol
self.Price = 0
self.IsLong = False
# Add indicators
self.ATR40 = algorithm.ATR(symbol, 40, MovingAverageType.Simple, Resolution.Daily)
self.ROC42 = algorithm.ROC(symbol, 42, Resolution.Daily) # 42-day Rate of Change
self.SMA50 = algorithm.SMA(symbol, 50, Resolution.Daily)
self.SMA150 = algorithm.SMA(symbol, 150, Resolution.Daily)
self.SMA200 = algorithm.SMA(symbol, 200, Resolution.Daily)
self.High52 = algorithm.MAX(symbol, 252, Resolution.Daily)
self.Low52 = algorithm.MIN(symbol, 252, Resolution.Daily)
self.Reset()
@property
def IsReady(self):
return (
self.ATR40.IsReady and
self.ROC42.IsReady and
self.SMA50.IsReady and
self.SMA150.IsReady and
self.SMA200.IsReady and
self.High52.IsReady and
self.Low52.IsReady
)
def Update(self, data):
if self.symbol in data.Bars:
self.Price = data.Bars[self.symbol].Close
def SetLong(self):
self.IsLong = True
def Reset(self):
self.IsLong = False
def Dispose(self):
# Deregister indicators to stop automatic updates
self.algorithm.deregister_indicator(self.ATR40)
self.algorithm.deregister_indicator(self.ROC42)
self.algorithm.deregister_indicator(self.SMA50)
self.algorithm.deregister_indicator(self.SMA150)
self.algorithm.deregister_indicator(self.SMA200)
if self.High52:
self.algorithm.deregister_indicator(self.High52)
if self.Low52:
self.algorithm.deregister_indicator(self.Low52)
# Reset attributes
self.IsLong = False
self.Price = 0
from AlgorithmImports import *
class MomentumStockPortfolioAlgorithm(QCAlgorithm):
def Initialize(self):
self.SetStartDate(2019, 1, 1)
self.SetEndDate(2025, 1, 1)
self.SetCash(100000)
self.Debug("Algorithm initialized with start date: 2023-01-01 and cash: 100000")
# Set warm-up period
self.SetWarmUp(252, Resolution.Daily)
# Add SPY for market regime filtering
self.spy = self.AddEquity("SPY", Resolution.Daily).Symbol
self.spy_100MA = self.SMA(self.spy, 100, Resolution.Daily)
# Dictionary to store SymbolData for each symbol
self.symbol_data_by_symbol: Dict[Symbol, SymbolData] = {}
# List to track the selected universe and momentum stocks
self.selected_symbols: List[Symbol] = []
self.momentum_symbols: List[Symbol] = []
self.exit_symbols: List[Symbol] = []
# Universe Settings
self.UniverseSettings.Resolution = Resolution.Daily
self.UniverseSettings.ExtendedMarketHours = False
self.UniverseSettings.DataNormalizationMode = DataNormalizationMode.SplitAdjusted
self.Debug(f"Universe settings configured: Resolution = {self.UniverseSettings.Resolution}, "
f"ExtendedMarketHours = {self.UniverseSettings.ExtendedMarketHours}, "
f"DataNormalizationMode = {self.UniverseSettings.DataNormalizationMode}")
# Universe schedule (first trading day of each month)
self.UniverseSettings.Schedule.On(self.DateRules.MonthStart())
# Add a custom universe
self.AddUniverse(self.filter_universe, self.log_filtered_universe)
# Schedule placing MarketOnOpen orders
self.Schedule.On(
self.DateRules.EveryDay(self.spy),
self.TimeRules.BeforeMarketOpen(self.spy, 2),
self.PlaceMarketOnOpenOrders
)
def filter_universe(self, fundamental):
""" Filters the universe based on custom rules """
return [
x.Symbol for x in sorted(
[
f for f in fundamental
if f.HasFundamentalData
and f.Price > 10
and f.DollarVolume > 10_000_000
and f.MarketCap > 500_000_000
],
key=lambda f: f.DollarVolume,
reverse=True
)[:100]
]
def log_filtered_universe(self, selected):
""" Logs the filtered universe and updates the selected symbols list """
selected_list = list(selected) # Convert selected to a list
tickers = [s.Symbol.Value for s in selected_list] # Extract tickers from the Symbol property of Fundamental objects
self.Debug(f"Selected {len(selected_list)} stocks in FundamentalSelection: {', '.join(tickers)}")
self.selected_symbols = [s.Symbol for s in selected_list] # Update selectedSymbols with Symbol objects
return [s.Symbol for s in selected_list] # Return only Symbol objects
def OnData(self, data):
if self.IsWarmingUp:
return
self.momentum_symbols = [] # Reset symbols for entry
self.exit_symbols = [] # Reset symbols for exit
for symbol, symbol_data in self.symbol_data_by_symbol.items():
if not symbol_data.IsReady:
continue
symbol_data.Update(data)
price = symbol_data.Price
# Entry condition: Momentum-based selection
if symbol not in self.Portfolio or not self.Portfolio[symbol].Invested:
sma50 = symbol_data.SMA50.Current.Value
sma150 = symbol_data.SMA150.Current.Value
sma200 = symbol_data.SMA200.Current.Value
if price > sma50 > sma150 > sma200:
self.momentum_symbols.append(symbol)
# Exit condition: Price below SMA50
if self.Portfolio[symbol].Invested:
sma50 = symbol_data.SMA50.Current.Value
if price < sma50:
self.exit_symbols.append(symbol)
def PlaceMarketOnOpenOrders(self):
""" Place MarketOnOpen orders for entry and exit symbols """
for symbol in self.exit_symbols:
holding = self.Portfolio[symbol]
if holding.Invested:
self.MarketOnOpenOrder(symbol, -holding.Quantity)
self.Log(f"Exiting {symbol.Value} - Price: {holding.AveragePrice:.2f}")
for symbol in self.momentum_symbols:
if symbol in self.symbol_data_by_symbol:
symbol_data = self.symbol_data_by_symbol[symbol]
price = symbol_data.Price
quantity = self.PositionSizing(price)
if quantity > 0:
self.MarketOnOpenOrder(symbol, quantity)
self.Log(f"Entering {symbol.Value} - Quantity: {quantity}, Price: {price:.2f}")
def PositionSizing(self, price):
""" Calculate position sizing based on portfolio value and price """
if price <= 0: # Prevent division by zero
self.Debug("PositionSizing: Invalid price encountered. Skipping position sizing.")
return 0
cash_to_invest = self.Portfolio.TotalPortfolioValue * 0.1 # 10% allocation
return int(cash_to_invest / price)
def OnSecuritiesChanged(self, changes):
for security in changes.AddedSecurities:
symbol = security.Symbol
if symbol not in self.symbol_data_by_symbol:
symbol_data = SymbolData(self, symbol)
self.symbol_data_by_symbol[symbol] = symbol_data
for security in changes.RemovedSecurities:
symbol = security.Symbol
if symbol in self.symbol_data_by_symbol:
self.symbol_data_by_symbol[symbol].Dispose()
del self.symbol_data_by_symbol[symbol]
class SymbolData:
def __init__(self, algorithm, symbol):
self.algorithm = algorithm
self.symbol = symbol
self.Price = 0
self.IsLong = False
self.ATR = algorithm.ATR(symbol, 20, MovingAverageType.Simple, Resolution.Daily)
self.SMA50 = algorithm.SMA(symbol, 50, Resolution.Daily)
self.SMA150 = algorithm.SMA(symbol, 150, Resolution.Daily)
self.SMA200 = algorithm.SMA(symbol, 200, Resolution.Daily)
self.High52 = algorithm.MAX(symbol, 252, Resolution.Daily)
self.Low52 = algorithm.MIN(symbol, 252, Resolution.Daily)
self.Reset()
@property
def IsReady(self):
return (
self.ATR.IsReady and
self.SMA50.IsReady and
self.SMA150.IsReady and
self.SMA200.IsReady and
self.High52 is not None and self.High52.IsReady and
self.Low52 is not None and self.Low52.IsReady
)
def Update(self, data):
if self.symbol in data.Bars:
self.Price = data.Bars[self.symbol].Close
def SetLong(self):
self.IsLong = True
def Reset(self):
self.IsLong = False
def Dispose(self):
# Deregister indicators to stop automatic updates
self.algorithm.deregister_indicator(self.ATR)
self.algorithm.deregister_indicator(self.SMA50)
self.algorithm.deregister_indicator(self.SMA150)
self.algorithm.deregister_indicator(self.SMA200)
if self.High52:
self.algorithm.deregister_indicator(self.High52)
if self.Low52:
self.algorithm.deregister_indicator(self.Low52)
# Reset attributes
self.IsLong = False
self.Price = 0
from AlgorithmImports import *
class MomentumStockPortfolioAlgorithm(QCAlgorithm):
def Initialize(self):
self.SetStartDate(2019, 1, 1)
self.SetEndDate(2025, 1, 1)
self.SetCash(100000)
self.Debug("Algorithm initialized with start date: 2023-01-01 and cash: 100000")
# Set warm-up period
self.SetWarmUp(252, Resolution.Daily)
# Add SPY for market regime filtering
self.spy = self.AddEquity("SPY", Resolution.Daily).Symbol
self.spy_100MA = self.SMA(self.spy, 100, Resolution.Daily)
# Dictionary to store SymbolData for each symbol
self.symbol_data_by_symbol: Dict[Symbol, SymbolData] = {}
# List to track the selected universe and momentum stocks
self.selected_symbols: List[Symbol] = []
self.momentum_symbols: List[Symbol] = []
self.exit_symbols: List[Symbol] = []
# Universe Settings
self.UniverseSettings.Resolution = Resolution.Daily
self.UniverseSettings.ExtendedMarketHours = False
self.UniverseSettings.DataNormalizationMode = DataNormalizationMode.SplitAdjusted
self.Debug(f"Universe settings configured: Resolution = {self.UniverseSettings.Resolution}, "
f"ExtendedMarketHours = {self.UniverseSettings.ExtendedMarketHours}, "
f"DataNormalizationMode = {self.UniverseSettings.DataNormalizationMode}")
# Universe schedule (first trading day of each month)
self.UniverseSettings.Schedule.On(self.DateRules.MonthStart())
# Add a custom universe
self.AddUniverse(self.filter_universe, self.log_filtered_universe)
# Schedule placing MarketOnOpen orders
self.Schedule.On(
self.DateRules.EveryDay(self.spy),
self.TimeRules.BeforeMarketOpen(self.spy, 2),
self.PlaceMarketOnOpenOrders
)
def filter_universe(self, fundamental):
""" Filters the universe based on custom rules """
return [
x.Symbol for x in sorted(
[
f for f in fundamental
if f.HasFundamentalData
and f.Price > 10
and f.DollarVolume > 10_000_000
and f.MarketCap > 500_000_000
],
key=lambda f: f.DollarVolume,
reverse=True
)[:100]
]
def log_filtered_universe(self, selected):
""" Logs the filtered universe and updates the selected symbols list """
selected_list = list(selected) # Convert selected to a list
tickers = [s.Symbol.Value for s in selected_list] # Extract tickers from the Symbol property of Fundamental objects
self.Debug(f"Selected {len(selected_list)} stocks in FundamentalSelection: {', '.join(tickers)}")
self.selected_symbols = [s.Symbol for s in selected_list] # Update selectedSymbols with Symbol objects
return [s.Symbol for s in selected_list] # Return only Symbol objects
def OnData(self, data):
if self.IsWarmingUp:
return
self.momentum_symbols = [] # Reset symbols for entry
self.exit_symbols = [] # Reset symbols for exit
for symbol, symbol_data in self.symbol_data_by_symbol.items():
if not symbol_data.IsReady:
continue
symbol_data.Update(data)
price = symbol_data.Price
# Entry condition: Momentum-based selection
if symbol not in self.Portfolio or not self.Portfolio[symbol].Invested:
sma50 = symbol_data.SMA50.Current.Value
sma150 = symbol_data.SMA150.Current.Value
sma200 = symbol_data.SMA200.Current.Value
high_52 = symbol_data.High52.Current.Value
low_52 = symbol_data.Low52.Current.Value
threshold_30_above_low = low_52 * 1.3
threshold_25_within_high = high_52 * 0.75
if price > sma50 > sma150 > sma200:
if price > threshold_30_above_low and price >= threshold_25_within_high:
self.momentum_symbols.append(symbol)
# Exit condition: Price below SMA50
if self.Portfolio[symbol].Invested:
sma50 = symbol_data.SMA50.Current.Value
if price < sma50:
self.exit_symbols.append(symbol)
def PlaceMarketOnOpenOrders(self):
""" Place MarketOnOpen orders for entry and exit symbols """
for symbol in self.exit_symbols:
holding = self.Portfolio[symbol]
if holding.Invested:
self.MarketOnOpenOrder(symbol, -holding.Quantity)
self.Log(f"Exiting {symbol.Value} - Price: {holding.AveragePrice:.2f}")
for symbol in self.momentum_symbols:
if symbol in self.symbol_data_by_symbol:
symbol_data = self.symbol_data_by_symbol[symbol]
price = symbol_data.Price
quantity = self.PositionSizing(price)
if quantity > 0:
self.MarketOnOpenOrder(symbol, quantity)
self.Log(f"Entering {symbol.Value} - Quantity: {quantity}, Price: {price:.2f}")
def PositionSizing(self, price):
""" Calculate position sizing based on portfolio value and price """
if price <= 0: # Prevent division by zero
self.Debug("PositionSizing: Invalid price encountered. Skipping position sizing.")
return 0
cash_to_invest = self.Portfolio.TotalPortfolioValue * 0.1 # 10% allocation
return int(cash_to_invest / price)
def OnSecuritiesChanged(self, changes):
for security in changes.AddedSecurities:
symbol = security.Symbol
if symbol not in self.symbol_data_by_symbol:
symbol_data = SymbolData(self, symbol)
self.symbol_data_by_symbol[symbol] = symbol_data
for security in changes.RemovedSecurities:
symbol = security.Symbol
if symbol in self.symbol_data_by_symbol:
self.symbol_data_by_symbol[symbol].Dispose()
del self.symbol_data_by_symbol[symbol]
class SymbolData:
def __init__(self, algorithm, symbol):
self.algorithm = algorithm
self.symbol = symbol
self.Price = 0
self.IsLong = False
self.ATR = algorithm.ATR(symbol, 20, MovingAverageType.Simple, Resolution.Daily)
self.SMA50 = algorithm.SMA(symbol, 50, Resolution.Daily)
self.SMA150 = algorithm.SMA(symbol, 150, Resolution.Daily)
self.SMA200 = algorithm.SMA(symbol, 200, Resolution.Daily)
self.High52 = algorithm.MAX(symbol, 252, Resolution.Daily)
self.Low52 = algorithm.MIN(symbol, 252, Resolution.Daily)
self.Reset()
@property
def IsReady(self):
return (
self.ATR.IsReady and
self.SMA50.IsReady and
self.SMA150.IsReady and
self.SMA200.IsReady and
self.High52 is not None and self.High52.IsReady and
self.Low52 is not None and self.Low52.IsReady
)
def Update(self, data):
if self.symbol in data.Bars:
self.Price = data.Bars[self.symbol].Close
def SetLong(self):
self.IsLong = True
def Reset(self):
self.IsLong = False
def Dispose(self):
# Deregister indicators to stop automatic updates
self.algorithm.deregister_indicator(self.ATR)
self.algorithm.deregister_indicator(self.SMA50)
self.algorithm.deregister_indicator(self.SMA150)
self.algorithm.deregister_indicator(self.SMA200)
if self.High52:
self.algorithm.deregister_indicator(self.High52)
if self.Low52:
self.algorithm.deregister_indicator(self.Low52)
# Reset attributes
self.IsLong = False
self.Price = 0
from AlgorithmImports import *
class MomentumStockPortfolioAlgorithm(QCAlgorithm):
def Initialize(self):
self.SetStartDate(2019, 1, 1)
self.SetEndDate(2025, 1, 1)
self.SetCash(100000)
self.Debug("Algorithm initialized with start date: 2023-01-01 and cash: 100000")
# Set warm-up period
self.SetWarmUp(252, Resolution.Daily)
# Add SPY for market regime filtering
self.spy = self.AddEquity("SPY", Resolution.Daily).Symbol
self.spy_100MA = self.SMA(self.spy, 100, Resolution.Daily)
# Dictionary to store SymbolData for each symbol
self.symbol_data_by_symbol: Dict[Symbol, SymbolData] = {}
# List to track the selected universe and momentum stocks
self.selected_symbols: List[Symbol] = []
self.momentum_symbols: List[Symbol] = []
self.exit_symbols: List[Symbol] = []
# Universe Settings
self.UniverseSettings.Resolution = Resolution.Daily
self.UniverseSettings.ExtendedMarketHours = False
self.UniverseSettings.DataNormalizationMode = DataNormalizationMode.SplitAdjusted
self.Debug(f"Universe settings configured: Resolution = {self.UniverseSettings.Resolution}, "
f"ExtendedMarketHours = {self.UniverseSettings.ExtendedMarketHours}, "
f"DataNormalizationMode = {self.UniverseSettings.DataNormalizationMode}")
# Add a custom universe
self.AddUniverse(self.filter_universe, self.log_filtered_universe)
# Schedule daily momentum calculations
self.Schedule.On(self.DateRules.EveryDay(self.spy),
self.TimeRules.BeforeMarketOpen(self.spy, 15),
self.CalculateMomentumScores)
# Schedule placing MarketOnOpen orders
self.Schedule.On(
self.DateRules.EveryDay(self.spy),
self.TimeRules.BeforeMarketOpen(self.spy, 2),
self.PlaceMarketOnOpenOrders
)
def filter_universe(self, fundamental):
""" Filters the universe based on custom rules """
return [
x.Symbol for x in sorted(
[
f for f in fundamental
if f.HasFundamentalData
and f.Price > 10
and f.DollarVolume > 10_000_000
and f.MarketCap > 500_000_000
],
key=lambda f: f.DollarVolume,
reverse=True
)[:100]
]
def log_filtered_universe(self, selected):
""" Logs the filtered universe and updates the selected symbols list """
selected_list = list(selected) # Convert selected to a list
tickers = [s.Symbol.Value for s in selected_list] # Extract tickers from the Symbol property of Fundamental objects
self.Debug(f"Selected {len(selected_list)} stocks in FundamentalSelection: {', '.join(tickers)}")
self.selected_symbols = [s.Symbol for s in selected_list] # Update selectedSymbols with Symbol objects
return [s.Symbol for s in selected_list] # Return only Symbol objects
def CalculateMomentumScores(self):
if self.IsWarmingUp:
return # Skip trading until warm-up is complete
"""Calculate momentum scores for the selected universe using Smoothness Ratio and Maximum Drawdown."""
# Fetch historical data for all selected symbols in one batch
history = self.History(self.selected_symbols, 63, Resolution.Daily)
if history.empty:
return # Exit if no data is returned
results = []
for symbol in self.selected_symbols:
if symbol not in history.index.levels[0]:
continue
symbol_history = history.loc[symbol]
if len(symbol_history) < 63:
continue
# Calculate Momentum 3-1
price_t_3 = symbol_history['close'].iloc[-63] # Price 3 months ago
price_t_1 = symbol_history['close'].iloc[-21] # Price 1 month ago
momentum_3_1 = ((price_t_1 - price_t_3) / price_t_3) * 100 # Convert to percentage
if momentum_3_1 < 0:
continue # Exclude stocks with negative momentum
# Calculate Maximum Drawdown (MDD)
running_max = symbol_history['close'].cummax()
drawdowns = (symbol_history['close'] - running_max) / running_max
max_drawdown = drawdowns.min()
if max_drawdown < -0.05: # Exclude stocks with drawdown > 5%
continue
# Calculate Smoothness Ratio
daily_returns = symbol_history['close'].pct_change().dropna()
rolling_std = daily_returns.std()
rolling_std_pct = rolling_std * 100 # Convert to percentage
smoothness_ratio = momentum_3_1 / rolling_std_pct if rolling_std_pct > 0 else 0
# Append only necessary data for ranking
results.append({
"Symbol": symbol,
"SmoothnessRatio": smoothness_ratio
})
# Sort results by Smoothness Ratio (descending) and pick top 10
self.momentum_scores = sorted(results, key=lambda x: x["SmoothnessRatio"], reverse=True)[:10]
# Log top 10 symbols
top_10 = self.momentum_scores
log_message = "Top 10 Momentum Stocks (3-1): "
for rank, result in enumerate(top_10):
log_message += f"Rank {rank + 1}: {result['Symbol'].Value} (Smoothness: {result['SmoothnessRatio']:.2f}); "
self.Debug(log_message)
# Update momentum symbols
self.momentum_symbols = [result["Symbol"] for result in top_10]
def OnData(self, data):
if self.IsWarmingUp:
return
self.exit_symbols = [] # Reset symbols for exit
for symbol, symbol_data in self.symbol_data_by_symbol.items():
if not symbol_data.IsReady:
continue
symbol_data.Update(data)
price = symbol_data.Price
# Exit condition: Price below SMA50
if self.Portfolio[symbol].Invested:
sma50 = symbol_data.SMA50.Current.Value
if price < sma50:
self.exit_symbols.append(symbol)
def PlaceMarketOnOpenOrders(self):
""" Place MarketOnOpen orders for exit symbols """
for symbol in self.exit_symbols:
holding = self.Portfolio[symbol]
if holding.Invested:
self.MarketOnOpenOrder(symbol, -holding.Quantity)
self.Log(f"Exiting {symbol.Value} - Price: {holding.AveragePrice:.2f}")
for symbol in self.momentum_symbols:
if symbol in self.symbol_data_by_symbol:
symbol_data = self.symbol_data_by_symbol[symbol]
price = symbol_data.Price
quantity = self.PositionSizing(price)
if quantity > 0:
self.MarketOnOpenOrder(symbol, quantity)
self.Log(f"Entering {symbol.Value} - Quantity: {quantity}, Price: {price:.2f}")
def PositionSizing(self, price):
""" Calculate position sizing based on portfolio value and price """
if price <= 0: # Prevent division by zero
self.Debug("PositionSizing: Invalid price encountered. Skipping position sizing.")
return 0
cash_to_invest = self.Portfolio.TotalPortfolioValue * 0.1 # 10% allocation
return int(cash_to_invest / price)
def OnSecuritiesChanged(self, changes):
for security in changes.AddedSecurities:
symbol = security.Symbol
if symbol not in self.symbol_data_by_symbol:
symbol_data = SymbolData(self, symbol)
self.symbol_data_by_symbol[symbol] = symbol_data
for security in changes.RemovedSecurities:
symbol = security.Symbol
if symbol in self.symbol_data_by_symbol:
self.symbol_data_by_symbol[symbol].Dispose()
del self.symbol_data_by_symbol[symbol]
class SymbolData:
def __init__(self, algorithm, symbol):
self.algorithm = algorithm
self.symbol = symbol
self.Price = 0
self.IsLong = False
# Add indicators
self.ATR = algorithm.ATR(symbol, 20, MovingAverageType.Simple, Resolution.Daily)
self.SMA50 = algorithm.SMA(symbol, 50, Resolution.Daily)
self.SMA150 = algorithm.SMA(symbol, 150, Resolution.Daily)
self.SMA200 = algorithm.SMA(symbol, 200, Resolution.Daily)
self.High52 = algorithm.MAX(symbol, 252, Resolution.Daily)
self.Low52 = algorithm.MIN(symbol, 252, Resolution.Daily)
self.ROC63 = algorithm.ROC(symbol, 63, Resolution.Daily) # 63-day Rate of Change
self.Reset()
@property
def IsReady(self):
return (
self.ATR.IsReady and
self.SMA50.IsReady and
self.SMA150.IsReady and
self.SMA200.IsReady and
self.High52.IsReady and
self.Low52.IsReady and
self.ROC63.IsReady
)
def Update(self, data):
if self.symbol in data.Bars:
self.Price = data.Bars[self.symbol].Close
def SetLong(self):
self.IsLong = True
def Reset(self):
self.IsLong = False
def Dispose(self):
# Deregister indicators to stop automatic updates
self.algorithm.deregister_indicator(self.ATR)
self.algorithm.deregister_indicator(self.SMA50)
self.algorithm.deregister_indicator(self.SMA150)
self.algorithm.deregister_indicator(self.SMA200)
if self.High52:
self.algorithm.deregister_indicator(self.High52)
if self.Low52:
self.algorithm.deregister_indicator(self.Low52)
# Reset attributes
self.IsLong = False
self.Price = 0