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"