| Overall Statistics |
|
Total Orders 169 Average Win 1.55% Average Loss -1.52% Compounding Annual Return 41.498% Drawdown 27.700% Expectancy 0.596 Start Equity 1000000 End Equity 3091955.41 Net Profit 209.196% Sharpe Ratio 1.25 Sortino Ratio 1.496 Probabilistic Sharpe Ratio 66.528% Loss Rate 21% Win Rate 79% Profit-Loss Ratio 1.02 Alpha 0.288 Beta -0.177 Annual Standard Deviation 0.221 Annual Variance 0.049 Information Ratio 0.75 Tracking Error 0.276 Treynor Ratio -1.561 Total Fees $1287.86 Estimated Strategy Capacity $680000000.00 Lowest Capacity Asset AMZN R735QTJ8XC9X Portfolio Turnover 2.02% |
# region imports
from AlgorithmImports import *
from scipy.stats import linregress
import numpy as np
# endregion
class SimpleDynamicMomentumAlgorithm(QCAlgorithm):
def Initialize(self):
self.SetStartDate(2021, 1, 1) # Set start date
self.SetEndDate(datetime.now()) # Set end date
self.SetCash(1000000) # Set initial capital
# Define the universe of assets
self.symbols = [self.AddEquity(ticker, Resolution.Daily).Symbol for ticker in ["AAPL", "MSFT", "GOOGL", "AMZN", "META"]]
self.lookback = 90 # Lookback period for momentum calculation (e.g., 3 months)
self.rebalance_period = 30 # Rebalance period (e.g., monthly)
self.next_rebalance = self.Time + timedelta(days=self.rebalance_period)
self.stop_loss_percentage = 0.175 # 17.5% stop-loss
self.entry_prices = {} # Store the entry prices for positions
self.highest_prices = {} # Store the highest price reached by a stock for trailing stop loss
# Market index to gauge overall market conditions
self.market = self.AddEquity("SPY", Resolution.Daily).Symbol
# Moving averages for market condition
self.short_sma = self.SMA(self.market, 50, Resolution.Daily)
self.long_sma = self.SMA(self.market, 200, Resolution.Daily)
self.Schedule.On(self.DateRules.EveryDay(), self.TimeRules.At(10, 0), self.Rebalance)
def OnData(self, data):
self.UpdateTrailingStopLoss(data)
def UpdateTrailingStopLoss(self, data):
for symbol in list(self.entry_prices.keys()):
if symbol in data and data[symbol] is not None:
current_price = data[symbol].Price
# Update the highest price reached
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)
# Calculate trailing stop price
trailing_stop_price = self.highest_prices[symbol] * (1 - self.stop_loss_percentage)
# Check if the current price is below the stop price
if current_price < trailing_stop_price:
self.Liquidate(symbol)
self.Debug(f"Trailing stop-loss triggered for {symbol.Value} at {current_price}")
del self.entry_prices[symbol]
del self.highest_prices[symbol]
# Calculate momentum using annualized exponential regression slope
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 # Assuming 252 trading days in a year
return annualized_slope
def Rebalance(self):
if self.Time < self.next_rebalance:
return
if self.short_sma.Current.Value > self.long_sma.Current.Value:
long_weight = 0.99
short_weight = 0.01
else:
long_weight = 0.01
short_weight = 0.99
momentum = {}
for symbol in self.symbols:
history = self.History(symbol, self.lookback, Resolution.Daily)
if not history.empty:
momentum[symbol] = self.calculate_momentum(history)
sorted_symbols = sorted(momentum.items(), key=lambda x: x[1], reverse=True)
num_long = int(len(sorted_symbols) * long_weight)
num_short = int(len(sorted_symbols) * short_weight)
long_symbols = [symbol for symbol, mom in sorted_symbols[:num_long]]
short_symbols = [symbol for symbol, mom in sorted_symbols[-num_short:]]
long_weight_per_position = long_weight / num_long if num_long > 0 else 0
short_weight_per_position = short_weight / num_short if num_short > 0 else 0
for symbol in self.symbols:
if symbol in long_symbols:
self.SetHoldings(symbol, long_weight_per_position)
self.entry_prices[symbol] = self.Securities[symbol].Price * (1 - self.stop_loss_percentage)
elif symbol in short_symbols:
self.SetHoldings(symbol, -short_weight_per_position)
self.entry_prices[symbol] = self.Securities[symbol].Price * (1 + self.stop_loss_percentage)
else:
self.Liquidate(symbol)
if symbol in self.entry_prices:
del self.entry_prices[symbol]
self.next_rebalance = self.Time + timedelta(days=self.rebalance_period)
def OnEndOfAlgorithm(self):
self.Debug("Algorithm finished running.")