#region imports
from AlgorithmImports import *
#endregion
# Your New Python File
class MomentumAlpha(AlphaModel):
def __init__(
self,
long_period=252, #12-month
short_period=21, #1-month
long_percent=0.2, #quintile
short_percent=0.2
):
self.long_period = long_period
self.short_period = short_period
self.long_percent = long_percent
self.short_percent = short_percent
self.mom_by_symbol = {}
self.last_month = -1
def update(self, algorithm, data):
# emit monthly insight for now
if self.last_month == algorithm.Time.month:
return []
self.last_month = algorithm.Time.month
mom_scores = []
for symbol in self.mom_by_symbol:
if self.mom_by_symbol[symbol].is_ready():
#algorithm.log(f"{algorithm.Time}: {symbol} is ready")
long_momentum = self.mom_by_symbol[symbol].get_momentum_percent(self.short_period, self.long_period-1)
short_momentum = self.mom_by_symbol[symbol].get_momentum_percent(0, self.short_period-1)
mom_scores.append([symbol, long_momentum, short_momentum])
# size_in_bytes = sys.getsizeof(self.mom_by_symbol[symbol])
# algorithm.log(f"mom rolling window size: {size_in_bytes}")
if not mom_scores:
return []
def assign_quintiles(data):
return pd.qcut(data, 5, labels=[1, 2, 3, 4, 5])
# algorithm.log(f"{algorithm.Time}: Number of assets available is {len(mom_scores)}")
mom_scores_df = pd.DataFrame(mom_scores, columns=['symbol', 'long_momentum', 'short_momentum'])
mom_scores_df['long_momentum_quintile'] = assign_quintiles(mom_scores_df['long_momentum'])
mom_scores_df['short_momentum_quintile'] = assign_quintiles(mom_scores_df['short_momentum'])
long_stocks = mom_scores_df[(mom_scores_df['long_momentum_quintile'] == 5) & (mom_scores_df['short_momentum_quintile'] == 1)]
short_stocks = mom_scores_df[(mom_scores_df['long_momentum_quintile'] == 1) & (mom_scores_df['short_momentum_quintile'] == 5)]
algorithm.log(f"Long Stocks: {long_stocks['symbol']}")
algorithm.log(f"Short Stocks: {short_stocks['symbol']}")
insights = []
for symbol in long_stocks['symbol'].unique():
insights.append(Insight.Price(symbol, timedelta(days=20), InsightDirection.Up, None, None))
for symbol in short_stocks['symbol'].unique():
insights.append(Insight.Price(symbol, timedelta(days=20), InsightDirection.Down, None, None))
return insights
def on_securities_changed(self, algorithm, changes):
for security in changes.RemovedSecurities:
# algorithm.log(f"{algorithm.time}: Removed {security.Symbol}")
if security.Symbol in self.mom_by_symbol:
del self.mom_by_symbol[security.Symbol]
for security in changes.AddedSecurities:
# algorithm.log(f"{algorithm.time}: Added {security.Symbol}")
if security not in self.mom_by_symbol:
history_by_symbol = algorithm.History(security.Symbol, self.long_period, Resolution.DAILY)
self.mom_by_symbol[security.Symbol] = LongMomentumShortReversalIndicator(security.Symbol, self.long_period, algorithm, history_by_symbol)
#register indicator for daily update
algorithm.RegisterIndicator(security.Symbol, self.mom_by_symbol[security.Symbol], Resolution.DAILY)
del history_by_symbol
class LongMomentumShortReversalIndicator(PythonIndicator):
def __init__(self, symbol, period, algorithm, history):
self.symbol = symbol
self.period = period
self.value = 0
self.rollingWindow = RollingWindow[float](self.period)
#warm up indicator with history
if not history.empty:
for _, tradebar in history.loc[symbol].iterrows():
self.update(tradebar)
# algorithm.log(f"{algorithm.Time}: {symbol}")
def update(self, bar):
self.rollingWindow.add(bar.close)
return self.rollingWindow.is_ready
def is_ready(self):
return self.rollingWindow.is_ready
# percentage momentum between time_0 and time_n calculated as:
# (price[time_0] - price[time_n]) / price[time_n]
def get_momentum_percent(self, time_0, time_n):
return 100 * (self.rollingWindow[time_0] - self.rollingWindow[time_n]) / self.rollingWindow[time_n]from AlgorithmImports import *
class TopCryptoStrategy(QCAlgorithm):
def Initialize(self):
self.SetStartDate(2024, 3, 1) # Set Start Date
self.SetEndDate(2024, 6, 1)
self.SetCash(100000) # Set Strategy Cash
# Define the symbols
self.crypto_symbols = ["BTCUSD"]
self.SetBenchmark("SPY")
# Attempt to add each cryptocurrency and stock
self.active_symbols = []
for symbol in self.crypto_symbols:
try:
self.AddCrypto(symbol, Resolution.MINUTE)
self.active_symbols.append(symbol)
except Exception as e:
self.Debug(f"Unable to add symbol: {symbol}. Exception: {e}")
# Define the technical indicators
self.supertrend1 = {}
self.supertrend2 = {}
self.rsi = {}
self.ema100 = {}
self.weekly_twap = {}
self.entry_prices = {}
for symbol in self.active_symbols:
self.supertrend1[symbol] = self.STR(symbol, 10, 10, MovingAverageType.Wilders)
self.supertrend2[symbol] = self.STR(symbol, 10, 3, MovingAverageType.Wilders)
self.rsi[symbol] = self.RSI(symbol, 10, MovingAverageType.Wilders, Resolution.MINUTE)
self.ema100[symbol] = self.EMA(symbol, 100, Resolution.MINUTE)
self.weekly_twap[symbol] = self.WeeklyTwap(symbol, 5)
self.entry_prices[symbol] = None
self.SetWarmUp(100, Resolution.MINUTE) # Warm up period for 100 days
def WeeklyTwap(self, symbol, num_weeks):
twap = self.SMA(symbol, num_weeks * 5, Resolution.MINUTE) # Assuming 5 trading days per week
return twap
def OnData(self, data):
if self.IsWarmingUp:
return
for symbol in self.active_symbols:
if not data.Bars.ContainsKey(symbol):
continue
bar = data.Bars[symbol]
# Get current values
current_price = bar.Close
supertrend1 = self.supertrend1[symbol].Current.Value
supertrend2 = self.supertrend2[symbol].Current.Value
rsi = self.rsi[symbol].Current.Value
ema100 = self.ema100[symbol].Current.Value
weekly_twap = self.weekly_twap[symbol].Current.Value
# Define factor based on asset type
factor = 1.2
# Entry condition
if self.entry_prices[symbol] is None:
if ( rsi < 99): # Use appropriate factor
self.Debug(f"{symbol}: Supertrend1={supertrend1}, Supertrend2={supertrend2}, RSI={rsi}, EMA100={ema100}, Weekly TWAP={weekly_twap}")
self.SetHoldings(symbol, -1.0)
self.entry_prices[symbol] = current_price
# Exit condition
elif rsi < 1:
self.Liquidate(symbol)
self.entry_prices[symbol] = None
