Overall Statistics |
Total Trades 1299 Average Win 1.62% Average Loss -1.06% Compounding Annual Return 15.813% Drawdown 28.800% Expectancy 0.293 Net Profit 548.733% Sharpe Ratio 0.735 Probabilistic Sharpe Ratio 9.518% Loss Rate 49% Win Rate 51% Profit-Loss Ratio 1.54 Alpha 0.055 Beta 0.722 Annual Standard Deviation 0.166 Annual Variance 0.028 Information Ratio 0.219 Tracking Error 0.135 Treynor Ratio 0.169 Total Fees $1382.10 Estimated Strategy Capacity $910000.00 Lowest Capacity Asset EXPE TB0IQCZTBPT1 |
#region imports from AlgorithmImports import * #endregion ########################################################################################## # ETF Index Momentum Rebalancer # ----------------------------- # # Hold 'N' of the fastest moving stocks from the given ETF. Equally weightied # Rebalance every Day/Week/Month and cut losers with drawdown higher than X%. # # External Parameters, and defaults # ----------------------------------- # maxHoldings = 5 # Max number of positions to hold # lookbackInDays = 160 # Look at performance over last x days # rebalancePeriodIndex = 0 # 0:Monthly | 1:Weekly | 2:Daily # exitLosersPeriodIndex = 1 # irrelvant if the same as rebalance period index # exitLoserMaxDD = 10 # exit if ddown >= x%. Seems to do more harm than good. # maxPctEquity = 80 # % of equity to trade # @shock_and_awful # ########################################################################################## class ETFUniverse(QCAlgorithm): ## Main entry point for the algo ## ----------------------------- def Initialize(self): self.InitBacktestParams() self.InitExternalParams() self.InitAssets() self.InitAlgoParams() self.ScheduleRoutines() ## Set backtest params: dates, cash, etc. Called from Initialize(). ## ---------------------------------------------------------------- def InitBacktestParams(self): self.SetStartDate(2010, 1, 1) # Start Date # self.SetEndDate(2021, 1, 1) # End Date. Omit to run till present day self.SetCash(10000) # Set Strategy Cash self.EnableAutomaticIndicatorWarmUp = True ## Initialize external parameters. Called from Initialize(). ## --------------------------------------------------------- def InitExternalParams(self): self.maxPctEquity = float(self.GetParameter("maxPctEquity"))/100 self.maxHoldings = int(self.GetParameter("maxHoldings")) self.rebalancePeriodIndex = int(self.GetParameter("rebalancePeriodIndex")) self.lookbackInDays = int(self.GetParameter("lookbackInDays")) self.useETFWeights = bool(self.GetParameter("useETFWeights") == 1) self.exitLosersPeriodIndex = int(self.GetParameter("exitLosersPeriodIndex")) self.exitLoserMaxDD = -float(self.GetParameter("exitLoserMaxDD"))/100 ## Init assets: Symbol, broker model, universes, etc. Called from Initialize(). ## ---------------------------------------------------------------------------- def InitAssets(self): # Try diffferent ETF tickers like 'QQQ','SPY','XLF','EEM', etc self.ticker = 'QQQ' self.etfSymbol = self.AddEquity(self.ticker, Resolution.Hour).Symbol # Specify that we are using an ETF universe, and specify the data resolution # (ETF Universe versus some other universe, eg: based on fundamental critiera) self.AddUniverse(self.Universe.ETF(self.etfSymbol, self.UniverseSettings, self.ETFConstituentsFilter)) self.UniverseSettings.Resolution = Resolution.Hour # self.SetSecurityInitializer(lambda x: x.SetMarketPrice(self.GetLastKnownPrice(x))) self.SetSecurityInitializer(self.CustomSecurityInitializer) # TODO: Explore not trading if the ETF is below a critical SMA (eg 200) # Uncomment this code, and some additional code further below, to do so # ----------------------------------------------------------------------- # self.etfSMA = self.SMA(self.etfSymbol, 100, Resolution.Daily) ## Custom Security initializer, for reality modeling so you can ## better mimic real world conditions (eg: slippage, fees, etc) ## You can use your own reality modelling. read more in the docs ## quantconnect.com/docs/v2/writing-algorithms/reality-modeling/slippage/key-concepts ## --------------------------------------------------------------------------------- def CustomSecurityInitializer(self, security): security.SetMarketPrice(self.GetLastKnownPrice(security)) security.SetFeeModel(InteractiveBrokersFeeModel()) # Model IB's trading fees. security.SetSlippageModel(VolumeShareSlippageModel()) # Model slippage based on volume impact security.SetFillModel(LatestPriceFillModel()) # Model fills based on latest price ## Set algo params: Symbol, broker model, ticker, etc. Called from Initialize(). ## ----------------------------------------------------------------------------- def InitAlgoParams(self): # Flags to track and trigger rebalancing state self.timeToRebalance = True self.universeRepopulated = False # State vars self.symbolWeightDict = {} self.screenedSymbolData = {} self.ScreenedSymbols = [] # Interval Periods. Try Monthly, Weekly, Daily intervalPeriods = [IntervalEnum.MONTHLY, IntervalEnum.WEEKLY, IntervalEnum.DAILY] self.rebalancePeriod = intervalPeriods[self.rebalancePeriodIndex] self.exitLosersPeriod = intervalPeriods[self.exitLosersPeriodIndex] ## Schedule routine that we need to run on intervials ## Eg: rebalance every month, Exit losers every week, etc ## ------------------------------------------------------ def ScheduleRoutines(self): # Schedule rebalancing flag if( self.rebalancePeriod == IntervalEnum.MONTHLY ): self.Schedule.On( self.DateRules.MonthStart(self.etfSymbol), self.TimeRules.AfterMarketOpen(self.etfSymbol, 31), self.SetRebalanceFlag ) elif( self.rebalancePeriod == IntervalEnum.WEEKLY ): self.Schedule.On( self.DateRules.WeekStart(self.etfSymbol), self.TimeRules.AfterMarketOpen(self.etfSymbol, 31), self.SetRebalanceFlag ) # Schedule routines to exit losers if( self.exitLosersPeriod == IntervalEnum.WEEKLY ): self.Schedule.On( self.DateRules.WeekStart(self.etfSymbol), self.TimeRules.AfterMarketOpen(self.etfSymbol, 31), self.ExitLosers ) elif( self.exitLosersPeriod == IntervalEnum.DAILY ): self.Schedule.On( self.DateRules.EveryDay(self.etfSymbol), self.TimeRules.AfterMarketOpen(self.etfSymbol, 31), self.ExitLosers) # Buy screened symbols self.Schedule.On( self.DateRules.EveryDay(self.etfSymbol), self.TimeRules.AfterMarketOpen(self.etfSymbol, 60), self.BuyScreenedSymbols) ## This event handler receives the constituents of the ETF, and their weights. ## In here, if it is time to rebalance the portfolio, we will rank the stocks ## by momentum, and 'select' the top N positive movers. We dont use the weights atm. ## --------------------------------------------------------------------------------- def ETFConstituentsFilter(self, constituents): if( self.timeToRebalance ): # Create a dictionary , dict[symbol] = weight self.symbolWeightDict = {c.Symbol: c.Weight for c in constituents} # reset flags self.universeRepopulated = True self.timeToRebalance = False # Loop through the symbols, create a symbol data object (contains indicator calcs) for symbol in self.symbolWeightDict: if symbol not in self.screenedSymbolData: self.screenedSymbolData[symbol] = SymbolData(self, symbol, self.symbolWeightDict[symbol], \ self.lookbackInDays) # fetch recent history, then seed the symbol data object, # so indicators values can be calculated symbolData = self.screenedSymbolData[symbol] history = self.History[TradeBar](symbol, self.lookbackInDays+1, Resolution.Daily) symbolData.SeedTradeBarHistory(history) # - - Here is where you can add custom logic for signals. -- # - - right now the only entry criteria is positive momentum -- # - - to change this, update the 'ScreeningCriteriaMet' in SymbolData # - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - self.screenedSymbolData = {key: symData for key, symData in self.screenedSymbolData.items() if symData.ScreeningCriteriaMet} # Sort the symbols based on indicator values in the symbol data object. # right now we are using momentum. Might consider others in the future. momSorted = sorted(self.screenedSymbolData.items(), key=lambda x: x[1].MomentumValue, reverse=True)[:self.maxHoldings] # Add Symbols to the 'selected' list self.ScreenedSymbols = [x[0] for x in momSorted] return self.ScreenedSymbols else: return [] ## Called when the rebalance time interval is up ## ----------------------------------------------- def SetRebalanceFlag(self): self.timeToRebalance = True ## Open positions for screened symbols. ## ------------------------------------------ def BuyScreenedSymbols(self): # TODO: Explore not trading if the ETF is below a critical SMA (eg 200) # ---------------------------------------------------------------------- # if( self.Securities[self.etfSymbol].Price < self.etfSMA.Current.Value ): # if(self.Portfolio.Invested): # self.Liquidate(tag="ETF trading below critical SMA") # liquidate everything # return if (self.universeRepopulated): self.Liquidate() # liquidate everything self.symbolWeightDict = {} # reset weights # TODO: Explore using the relative weight of the assets for position sizing # ie: Respect the % of the assets in the ETF. Sof if you are trading SPY constituents # and AAPL had high momentum, it would have releatively large position size. # ------------------------------------------------------------------------------------------ # weightsSum = sum(self.screenedSymbolData[symbol].etfWeight for symbol in self.ScreenedSymbols) for symbol in self.ScreenedSymbols: # TODO: Uncomment Explore using relative weight. You'd use # symbolWeight = self.screenedSymbolData[symbol].etfWeight / weightsSum # respect weighting symbolWeight = 1 / len(self.ScreenedSymbols) # Equally weighted # Adjust to ensure we trade less than max pct of equity adjustedWeight = symbolWeight * self.maxPctEquity self.SetHoldings(symbol, adjustedWeight, tag=f"Momentum Pct: {round(self.screenedSymbolData[symbol].MomentumValue,2)}%") self.universeRepopulated = False def ExitLosers(self): # Loop through holdings, and exit any that are losing below the threshold for x in self.Portfolio: if x.Value.Invested: if( x.Value.UnrealizedProfitPercent <= self.exitLoserMaxDD): orderMsg = f"Unacceptable drawdown ({round(x.Value.UnrealizedProfitPercent*100,2)})% < {self.exitLoserMaxDD*100}" self.Liquidate(x.Key, tag=orderMsg) # TODO: Periodically check if a security is no longer in the ETF # --------------------------------------------------------------- # def RemoveDelistedSymbols(self, changes): # for investedSymbol in [x.Key for x in self.Portfolio if x.Value.Invested]: # if( investedSymbol not in self.symbolWeightDict.keys() ): # self.Liquidate(symbol, 'No longer in universe') ################################################## # Symbol Data Class -- # Data we need to persist for each Symvol ################################################## class SymbolData(): def __init__(self, algo, symbol, etfWeight, lookbackInDays): self.algo = algo self.symbol = symbol self.etfWeight = etfWeight self.momPct = MomentumPercent(lookbackInDays) def SeedTradeBarHistory(self,history): for tradeBar in history: self.momPct.Update(tradeBar.Time, tradeBar.Close) @property def MomentumValue(self): if(self.momPct.IsReady): return self.momPct.Current.Value else: return float('-inf') # - - Here is where you can add custom logic for signals. -- # - - right now the only entry criteria is positive momentum -- # - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - @property def ScreeningCriteriaMet(self): return (self.MomentumValue > 0 ) ############################### # Interval Enum ############################### class IntervalEnum(Enum): MONTHLY = "MONTHLY" WEEKLY = "WEEKLY" DAILY = "DAILY"