| Overall Statistics |
|
Total Orders 1043 Average Win 0.21% Average Loss -0.25% Compounding Annual Return 5.094% Drawdown 12.200% Expectancy 0.086 Start Equity 100000 End Equity 110923.96 Net Profit 10.924% Sharpe Ratio -0.107 Sortino Ratio -0.126 Probabilistic Sharpe Ratio 13.410% Loss Rate 41% Win Rate 59% Profit-Loss Ratio 0.83 Alpha -0.1 Beta 0.697 Annual Standard Deviation 0.112 Annual Variance 0.013 Information Ratio -1.526 Tracking Error 0.09 Treynor Ratio -0.017 Total Fees $1038.66 Estimated Strategy Capacity $0 Lowest Capacity Asset KMI UU1M6GZF9OPX Portfolio Turnover 6.37% |
# region imports
from AlgorithmImports import *
from scipy.stats import linregress
import numpy as np
# endregion
### SECTOR ROTATION DYNAMIC MOMENTUM ###
class SectorRotationDynamicMomentum(QCAlgorithm):
def Initialize(self):
self.SetStartDate(2023, 1, 1)
self.SetEndDate(2025, 4, 15)
self.SetCash(100_000)
self.leverage = 1.0
self.free_portfolio_value = 0.02
self.SetBrokerageModel(BrokerageName.QuantConnectBrokerage, AccountType.MARGIN)
self.symbols = []
# manually selected on FinVIiz
tickers = [
"AAPL", "ACN", "ADBE", "ADI", "AKAM", "AMAT", "AMD", "ANET", "APH", "AVGO",
"CDNS", "CRM", "CRWD", "CSCO", "CTSH", "DELL", "FI", "FIS", "FSLR", "FTNT",
"MSFT", "NVDA", "GOOGL",
"A", "ABBV", "ABT", "AMGN", "BAX", "BDX", "BMY", "BSX", "CAH", "CNC",
"COO", "CVS", "DHR", "DXCM", "EW", "GEHC", "GILD", "HOLX", "ISRG", "JNJ",
"LLY", "UNH",
"AFL", "AIG", "APO", "AXP", "BAC", "BK", "BX", "C", "CFG", "CME",
"COF", "DFS", "FITB", "GS", "ICE", "JPM", "KKR", "MA", "MET",
"APA", "BKR", "COP", "CTRA", "CVX", "DVN", "EOG", "EQT", "EXE", "FANG",
"HAL", "HES", "KMI", "MPC", "OKE", "OXY", "PSX", "SLB", "TPL", "TRGP",
"ALLE", "AME", "AOS", "AXON", "BA", "BLDR", "CARR", "CAT",
"CHRW", "CMI", "CPRT", "CSX", "CTAS", "DAL", "DE", "DOV",
"EFX", "EMR", "ETN", "EXPD",
"AEE", "AEP", "AES", "ATO", "AWK", "CEG", "CMS", "CNP", "D", "DTE",
"DUK", "ED", "EIX", "ES", "ETR", "EVRG", "EXC", "FE", "LNT", "NEE",
"AMT", "ARE", "AVB", "BXP", "CBRE", "CCI", "CPT", "CSGP", "DLR", "DOC",
"EQIX", "EQR", "ESS", "EXR", "FRT", "HST", "INVH", "IRM", "KIM", "MAA"
]
for ticker in tickers:
symbol = self.AddEquity(ticker, Resolution.Daily, leverage=self.leverage).Symbol
self.symbols.append(symbol)
self.sector_etfs = {
"Technology": self.AddEquity("XLK", Resolution.Daily).Symbol,
"Healthcare": self.AddEquity("XLV", Resolution.Daily).Symbol,
"Financials": self.AddEquity("XLF", Resolution.Daily).Symbol,
"Energy": self.AddEquity("XLE", Resolution.Daily).Symbol,
"Industrials": self.AddEquity("XLI", Resolution.Daily).Symbol,
"Utilities": self.AddEquity("XLU", Resolution.Daily).Symbol,
"Real Estate": self.AddEquity("XLRE", Resolution.Daily).Symbol,
}
self.stock_sector_map = {
# Technology
"AAPL": "Technology", "ACN": "Technology", "ADBE": "Technology", "ADI": "Technology",
"AKAM": "Technology", "AMAT": "Technology", "AMD": "Technology", "ANET": "Technology",
"APH": "Technology", "AVGO": "Technology", "CDNS": "Technology", "CRM": "Technology",
"CRWD": "Technology", "CSCO": "Technology", "CTSH": "Technology", "DELL": "Technology",
"FI": "Technology", "FIS": "Technology", "FSLR": "Technology", "FTNT": "Technology",
"MSFT": "Technology", "NVDA": "Technology", "GOOGL": "Technology",
# Healthcare
"A": "Healthcare", "ABBV": "Healthcare", "ABT": "Healthcare", "AMGN": "Healthcare",
"BAX": "Healthcare", "BDX": "Healthcare", "BMY": "Healthcare", "BSX": "Healthcare",
"CAH": "Healthcare", "CNC": "Healthcare", "COO": "Healthcare", "CVS": "Healthcare",
"DHR": "Healthcare", "DXCM": "Healthcare", "EW": "Healthcare", "GEHC": "Healthcare",
"GILD": "Healthcare", "HOLX": "Healthcare", "ISRG": "Healthcare", "JNJ": "Healthcare",
"LLY": "Healthcare", "UNH": "Healthcare",
# Financials
"AFL": "Financials", "AIG": "Financials", "APO": "Financials", "AXP": "Financials",
"BAC": "Financials", "BK": "Financials", "BX": "Financials", "C": "Financials",
"CFG": "Financials", "CME": "Financials", "COF": "Financials", "DFS": "Financials",
"FITB": "Financials", "GS": "Financials", "ICE": "Financials", "JPM": "Financials",
"KKR": "Financials", "MA": "Financials", "MET": "Financials",
# Energy
"APA": "Energy", "BKR": "Energy", "COP": "Energy", "CTRA": "Energy", "CVX": "Energy",
"DVN": "Energy", "EOG": "Energy", "EQT": "Energy", "EXE": "Energy", "FANG": "Energy",
"HAL": "Energy", "HES": "Energy", "KMI": "Energy", "MPC": "Energy", "OKE": "Energy",
"OXY": "Energy", "PSX": "Energy", "SLB": "Energy", "TPL": "Energy", "TRGP": "Energy",
# Industrials
"ALLE": "Industrials", "AME": "Industrials", "AOS": "Industrials", "AXON": "Industrials",
"BA": "Industrials", "BLDR": "Industrials", "CARR": "Industrials", "CAT": "Industrials",
"CHRW": "Industrials", "CMI": "Industrials", "CPRT": "Industrials", "CSX": "Industrials",
"CTAS": "Industrials", "DAL": "Industrials", "DE": "Industrials", "DOV": "Industrials",
"EFX": "Industrials", "EMR": "Industrials", "ETN": "Industrials", "EXPD": "Industrials",
# Utilities
"AEE": "Utilities", "AEP": "Utilities", "AES": "Utilities", "ATO": "Utilities", "AWK": "Utilities",
"CEG": "Utilities", "CMS": "Utilities", "CNP": "Utilities", "D": "Utilities", "DTE": "Utilities",
"DUK": "Utilities", "ED": "Utilities", "EIX": "Utilities", "ES": "Utilities", "ETR": "Utilities",
"EVRG": "Utilities", "EXC": "Utilities", "FE": "Utilities", "LNT": "Utilities", "NEE": "Utilities",
# Real Estate
"AMT": "Real Estate", "ARE": "Real Estate", "AVB": "Real Estate", "BXP": "Real Estate", "CBRE": "Real Estate",
"CCI": "Real Estate", "CPT": "Real Estate", "CSGP": "Real Estate", "DLR": "Real Estate", "DOC": "Real Estate",
"EQIX": "Real Estate", "EQR": "Real Estate", "ESS": "Real Estate", "EXR": "Real Estate", "FRT": "Real Estate",
"HST": "Real Estate", "INVH": "Real Estate", "IRM": "Real Estate", "KIM": "Real Estate", "MAA": "Real Estate",
}
self.market = self.AddEquity("SPY", Resolution.Daily).Symbol
self.short_sma = self.SMA(self.market, 50, Resolution.Daily)
self.long_sma = self.SMA(self.market, 200, Resolution.Daily)
self.lookback = 120
self.rebalance_period = 1
self.next_rebalance = self.Time + timedelta(days=self.rebalance_period)
self.entry_prices = {}
self.highest_prices = {}
self.entry_times = {}
self.SetWarmUp(200)
self.Schedule.On(self.DateRules.EveryDay(), self.TimeRules.At(10, 0), self.Rebalance)
def OnData(self, data):
self.UpdateTrailingStopLoss(data)
def UpdateTrailingStopLoss(self, data):
to_remove = []
spy_returns = self.History(self.market, 60, Resolution.Daily)['close'].pct_change()
spy_volatility = spy_returns.ewm(span=20).std().iloc[-1] * np.sqrt(252)
if spy_volatility < 0.2:
stop_multiplier = 1.2
elif spy_volatility > 0.3:
stop_multiplier = 2.0
else:
stop_multiplier = 1.5
for symbol in list(self.entry_prices.keys()):
if symbol not in data or data[symbol] is None or not self.Securities[symbol].IsTradable:
continue
current_price = data[symbol].Price
entry_price = self.entry_prices[symbol]
entry_time = self.entry_times.get(symbol, self.Time)
holding_period = (self.Time - entry_time).days
if symbol not in self.highest_prices:
self.highest_prices[symbol] = current_price
else:
self.highest_prices[symbol] = max(self.highest_prices[symbol], current_price)
atr = self.ATR(symbol, 20, Resolution.Daily)
if not atr.IsReady:
continue
trailing_stop_price = self.highest_prices[symbol] - stop_multiplier * atr.Current.Value
if current_price >= entry_price * 1.10:
self.Liquidate(symbol)
self.Debug(f"Profit target hit for {symbol.Value} at {current_price}")
to_remove.append(symbol)
continue
if current_price < trailing_stop_price:
self.Liquidate(symbol)
self.Debug(f"Trailing stop-loss triggered for {symbol.Value} at {current_price} with multiplier {stop_multiplier}")
to_remove.append(symbol)
continue
if holding_period >= 20:
self.Liquidate(symbol)
self.Debug(f"Time stop triggered for {symbol.Value} after {holding_period} days at {current_price}")
to_remove.append(symbol)
continue
for symbol in to_remove:
if symbol in self.entry_prices:
del self.entry_prices[symbol]
if symbol in self.highest_prices:
del self.highest_prices[symbol]
if symbol in self.entry_times:
del self.entry_times[symbol]
def calculate_momentum(self, history):
log_prices = np.log(history['close'])
days = np.arange(len(log_prices))
slope, _, _, _, _ = linregress(days, log_prices)
annualized_slope = slope * 252
total_return = history['close'].iloc[-1] / history['close'].iloc[0] - 1
volatility = np.std(history['close'].pct_change()) * np.sqrt(252)
sharpe_like = total_return / volatility if volatility != 0 else 0
combined_score = 0.5 * annualized_slope + 0.5 * sharpe_like
return combined_score
def calculate_inverse_volatility_weights(self, selected_symbols):
volatilities = {}
for symbol in selected_symbols:
history = self.History(symbol, self.lookback, Resolution.Daily)
if not history.empty:
returns = history['close'].pct_change()
volatility = returns.ewm(span=20).std().iloc[-1] * np.sqrt(252)
if volatility > 0:
volatilities[symbol] = 1 / volatility
total = sum(volatilities.values())
weights = {symbol: weight / total for symbol, weight in volatilities.items()}
return weights
def Rebalance(self):
if self.Time < self.next_rebalance or self.IsWarmingUp:
return
# Sector Rotation Step
sector_momentum = {}
for sector, etf_symbol in self.sector_etfs.items():
history = self.History(etf_symbol, 63, Resolution.Daily)
if not history.empty:
return_3m = history['close'].iloc[-1] / history['close'].iloc[0] - 1
sector_momentum[sector] = return_3m
top_sectors = sorted(sector_momentum.items(), key=lambda x: x[1], reverse=True)[:3]
selected_sectors = [sector for sector, momentum in top_sectors]
eligible_symbols = [
symbol for symbol in self.symbols
if self.stock_sector_map.get(symbol.Value, "") in selected_sectors
]
# Normal Rebalance
spy_returns = self.History(self.market, 60, Resolution.Daily)['close'].pct_change()
spy_volatility = spy_returns.ewm(span=20).std().iloc[-1] * np.sqrt(252)
if spy_volatility < 0.2:
self.rebalance_period = 10
else:
self.rebalance_period = 3
if self.short_sma.Current.Value > self.long_sma.Current.Value:
base_long_exposure = 0.8
else:
base_long_exposure = 0.2
volatility_scaling_factor = max(0.2, min(1.0, 0.3 / spy_volatility))
final_long_exposure = base_long_exposure * volatility_scaling_factor * (1 - self.free_portfolio_value)
momentum = {}
for symbol in eligible_symbols:
history = self.History(symbol, self.lookback, Resolution.Daily)
if not history.empty:
score = self.calculate_momentum(history)
if score > 0.05:
momentum[symbol] = score
sorted_symbols = sorted(momentum.items(), key=lambda x: x[1], reverse=True)
top_long_symbols = [symbol for symbol, score in sorted_symbols[:10]]
long_weights = self.calculate_inverse_volatility_weights(top_long_symbols)
for symbol in self.symbols:
if symbol in top_long_symbols:
self.SetHoldings(symbol, long_weights.get(symbol, 0) * final_long_exposure * self.leverage)
self.entry_prices[symbol] = self.Securities[symbol].Price
self.entry_times[symbol] = self.Time
else:
self.Liquidate(symbol)
if symbol in self.entry_prices:
del self.entry_prices[symbol]
if symbol in self.entry_times:
del self.entry_times[symbol]
self.next_rebalance = self.Time + timedelta(days=self.rebalance_period)
def OnEndOfAlgorithm(self):
self.Debug("Algorithm finished running.")