Overall Statistics |
Total Trades 7 Average Win 2.39% Average Loss 0% Compounding Annual Return 35.191% Drawdown 5.000% Expectancy 0 Net Profit 4.735% Sharpe Ratio 1.862 Probabilistic Sharpe Ratio 60.247% Loss Rate 0% Win Rate 100% Profit-Loss Ratio 0 Alpha -0.038 Beta 1.135 Annual Standard Deviation 0.169 Annual Variance 0.028 Information Ratio 0.047 Tracking Error 0.088 Treynor Ratio 0.277 Total Fees $16.75 |
from Selection.FundamentalUniverseSelectionModel import FundamentalUniverseSelectionModel CoarseUniverseSize = 3000 FineUniverseSize = 10 UniverseMinMarketCap= 50e9 UniverseMaxMarketCap = 500e9 PrimaryExchangeID = ["NAS"] UniverseSectors = [MorningstarSectorCode.Technology, MorningstarSectorCode.BasicMaterials, MorningstarSectorCode.ConsumerCyclical, MorningstarSectorCode.FinancialServices, MorningstarSectorCode.ConsumerDefensive, MorningstarSectorCode.Healthcare, MorningstarSectorCode.Utilities, MorningstarSectorCode.CommunicationServices, MorningstarSectorCode.Energy, MorningstarSectorCode.Industrials, MorningstarSectorCode.RealEstate] # ------------------------------ Universe Selection Model --------------------------------- class UniverseSelectionModel(FundamentalUniverseSelectionModel): def __init__(self, filterFineData = True, universeSettings = None, securityInitializer = None): super().__init__(filterFineData, universeSettings, securityInitializer) self.numberOfSymbolsCoarse = CoarseUniverseSize self.numberOfSymbolsFine = FineUniverseSize self.sorted1BySymbol = {} self.lastMonth = -1 self.lastYear = -1 # --------------------------------- Coarse -------------------------------------- def SelectCoarse(self, algorithm, coarse): #if algorithm.Time.month == self.lastMonth: # return Universe.Unchanged #self.lastMonth = algorithm.Time.month if algorithm.Time.year == self.lastYear: return Universe.Unchanged self.lastYear = algorithm.Time.year sorted1 = sorted([x for x in coarse if x.HasFundamentalData and x.Volume > 0 and x.Price >0], key = lambda x: x.DollarVolume, reverse=True)[:self.numberOfSymbolsCoarse] self.sorted1BySymbol = {x.Symbol:x.DollarVolume for x in sorted1} count = len(self.sorted1BySymbol) if count == 0: return Universe.Unchanged # return the symbol objects our sorted collection return list(self.sorted1BySymbol.keys()) # ---------------------------------- Fine --------------------------------------- def SelectFine(self, algorithm, fine): sorted2 = sorted([x for x in fine if x.CompanyReference.PrimaryExchangeID in PrimaryExchangeID and (algorithm.Time - x.SecurityReference.IPODate).days > 30 and x.AssetClassification.MorningstarSectorCode in UniverseSectors #and x.ValuationRatios.PERatio < 30 and x.MarketCap >= UniverseMinMarketCap and x.MarketCap <= UniverseMaxMarketCap], #and x.ValuationRatios.ForwardPERatio > 0 #and x.CompanyReference.CountryId == "USA" key = lambda x: x.MarketCap, reverse = True) rsiValues = {} for symbol in sorted2: stock = symbol.Symbol rsi = RelativeStrengthIndex(14) history = algorithm.History(stock, 14, Resolution.Daily) for tuple in history.loc[stock].itertuples(): rsi.Update(tuple.Index, tuple.close) rsiValue = rsi.Current.Value rsiValues[stock] = rsiValue sorted3 = sorted([x for x in sorted2], key = lambda x: rsiValues[x.Symbol], reverse = True) # fundametnal # market cap # sales growth .OperationRatios.RevenueGrowth.ThreeMonths # PE compared to sector # profitable # technical # RSI # PPO # SLOPE count = len(sorted3) if count == 0: return Universe.Unchanged algorithm.Debug(f"Fine:{count}") #for i in range(10): # algorithm.Debug(f"{i} {sorted3[i].Symbol} {rsiValues[sorted3[i].Symbol]}") # converts our list of fine to a list of Symbols sorted3BySymbol = [f.Symbol for f in sorted3] return sorted3BySymbol[:self.numberOfSymbolsFine]
# (c) QIO18 import Universe class QInOut(QCAlgorithm): def Initialize(self): # ------------------------------------------- Parameters --------------------------------------- self.InOut = True self.AlgoName = "QIO18" # --- List of assets to buy when in the market (Buy equal weights) FXO QQQJ PSCT self.InAssets = {0:["Universe"], 1:["QQQ"], 2:["SPY"], 3:["ARKK"], 4:["MSFT","AAPL","FB","GOOG","AMZN"], 5:["CRSP","NTLA","EDIT"], 6:["TQQQ"], 7:["WDAY","AMAT","TSLA"], 8:["AAPL","ADBE","ADI","ADP","ADSK","AEP","ALGN","ALXN","AMAT","AMD","AMGN","AMZN","ANSS","ASML","ATVI","AVGO","BIDU","BIIB","BKNG","CDNS","CDW","CERN","CHKP","CHTR","CMCSA","COST","CPRT","CSCO","CSX","CTAS","CTSH"], 9:["LMND","CRWD","ZM","PINS","FVRR","NFLX","MTCH","MASI","DIS","NVDA","ABNB"] # Fool } # --- List of assets to buy when OUT the market (Buy equal weights) self.OutAssets = {0:["Cash"], 1:["TLT"], 2:["TMF"], 3:["VXX"], 4:["TLT", "TMF", "VXX"], 5:["SQQQ"] } # --- List of IndexEquity self.IndexEquities = {1:"QQQ", 2:"SPY", 3:"ARKK" } self.IndicatorsActive = { 1: "RSI", 2: "RSI_PPO", 3: "RSI_PPO_SLOPE", 4: "RSI_PPO_SLOPE_MACD", 5: "ATR", 6: "MACD" } # ---------------------------------- Algorithm External Parameters ----------------------------- # Entry minute for the Trade algo after Market Open self.MinsAfter = int(self.GetParameter("minsAfter")) # Select IN and OUT Equity list of assests self.InAssetSelector = int(self.GetParameter("inAssetSelector")) # 0:Monthly selected Universe, 1:QQQ ... self.OutAssetSelector = int(self.GetParameter("OutAssetSelector")) # 1:TLT, 2:TMF ... self.IndexSelector = int(self.GetParameter("indexSelector")) # Indicators' Days input self.MacdFastDays = int(self.GetParameter("macdFastDays")) self.MacdSlowDays = int(self.GetParameter("macdSlowDays")) self.MacdSignalDays = int(self.GetParameter("macdSignalDays")) self.RsiDays = int(self.GetParameter("rsiDays")) self.RcDays = int(self.GetParameter("rcDays")) self.PpoSlowDays = int(self.GetParameter("ppoSlowDays")) self.PpoFastDays = int(self.GetParameter("ppoFastDays")) # Indicators gaps and factors self.RsiFactorGap = float(self.GetParameter("rsiFactorGap")) self.PpoGap = float(self.GetParameter("ppoGap")) self.PpoFactorGap = float(self.GetParameter("ppoFactorGap")) self.SlopeGap = float(self.GetParameter("slopeGap")) self.AtrDays = 18 # int(self.GetParameter("atrDays")) self.AtrFastDays = 2 # int(self.GetParameter("atrFastDays")) self.AtrSlowDays = 22 # int(self.GetParameter("atrSlowDays")) self.AtrGap = 1.3 # float(self.GetParameter("atrGap")) # MACD factors self.MacdFactorOutGap = float(self.GetParameter("macdFactorOutGap")) self.MacdHistOutGap = float(self.GetParameter("macdHistOutGap")) self.MacdHistInGap = 0 # = float(self.GetParameter("macdHistInGap")) self.SmaHourFactor = float(self.GetParameter("smaHourFactor")) # Trade time in minutes after market open self.MinDaysOut = int(self.GetParameter("daysOut")) # ------------------------------------ For Back Testing --------------------------------- self.StartYear = int(self.GetParameter("startYear")) self.EndYear = int(self.GetParameter("endYear")) self.SetStartDate(self.StartYear,1,1) self.SetEndDate(self.EndYear,12,31) self.StartCash = 100000 self.SetCash(self.StartCash) #self.Settings.FreePortfolioValuePercentage = 0.05 # For Live IB # -------------------------------------- Internal variables ---------------------------------- #For Plots on Equity Graph self.IndexEquityHistory = [] # now used for QQQ self.OutEquityHistory = [] # now used to TLT # For statistics self.AverageRSI = 0 self.TotalRSI = 0 self.TickerTable = {} self.MyInOutIndicator = 1 # Default Enter the Market self.IAMOut = False self.DaysOut = 0 # Cooldown days after exiting the Market self.TotalDays =0 # Counter for total days self.TotalDaysIn = 0 # Counter for days in the MArket self.ReasonOut= "" # String to report reason to go out of the Market/InEquity # ------------------------------- Sellecting Assets ---------------------------- self.IndexEquityTicker = self.IndexEquities[self.IndexSelector] self.InAssetList = self.InAssets[self.InAssetSelector] # IN Asset List self.SetAlpha(ConstantAlphaModel(InsightType.Price, InsightDirection.Up, timedelta(30))) if self.InAssetSelector == 0: self.UniverseSettings.Resolution = Resolution.Minute self.AddUniverseSelection(Universe.UniverseSelectionModel()) self.InAssetList = [] # Will be built in OnSecurities Change else: # Manually selected Asset lists for asset in self.InAssetList: self.AddEquity(asset, Resolution. Minute) self.SPY = self.AddEquity("SPY", Resolution.Daily).Symbol self.Fear = self.AddEquity("VXX", Resolution.Daily).Symbol self.IndexEquity = self.AddEquity(self.IndexEquityTicker, Resolution.Minute).Symbol # Key index for In Out decisions self.doNotEnter = [self.SPY, self.Fear, self.IndexEquity] self.OutAssetList = self.OutAssets[self.OutAssetSelector] # OUT Asset List Consider "SQQQ","GLD","PSQ","SH","EMTY", "XLU","XLP","TBT","IEF","TLH" "TMF"] for asset in self.OutAssetList: symbol = self.AddEquity(asset, Resolution.Minute).Symbol self.doNotEnter.append(symbol) # ---------------------------- Add IndexEquity Indicators ------------------------- rsi = self.RSI(self.IndexEquity, self.RsiDays, MovingAverageType.DoubleExponential, Resolution.Daily) rsiSMA = IndicatorExtensions.SMA(rsi,self.RsiDays) macd = self.MACD(self.IndexEquity, self.MacdFastDays,self.MacdSlowDays,self.MacdSignalDays,MovingAverageType.Exponential, Resolution.Daily) macdSMA = IndicatorExtensions.SMA(macd,9) rc = self.RC(self.IndexEquity,self.RcDays, 2, Resolution.Daily) ppo = self.PPO(self.IndexEquity, self.PpoFastDays, self.PpoSlowDays, MovingAverageType.Exponential, Resolution.Daily) ppoSMA = IndicatorExtensions.SMA(ppo,self.PpoFastDays) atr = self.ATR(self.IndexEquity, self.AtrDays, MovingAverageType.Exponential, Resolution.Daily) atrSMA_fast = IndicatorExtensions.SMA(atr,self.AtrFastDays) atrSMA_slow = IndicatorExtensions.SMA(atr,self.AtrSlowDays) smaHour = self.SMA(self.IndexEquity,24,Resolution.Hour) symbolData = SymbolData(self.IndexEquity, rsi, rsiSMA, macd, macdSMA, rc, ppo, ppoSMA, atr, atrSMA_fast, atrSMA_slow,smaHour) self.TickerTable[self.IndexEquity] = symbolData # ------------------------------------ Final setup ---------------------------------- self.SetBenchmark(self.SPY) # header data for graph str1 = f"{self.AlgoName} {str(self.StartYear)}-{str(self.EndYear)} {self.InAssetSelector}-{self.IndexEquity}-{self.OutAssetSelector} {self.InOut}" str2 = f"{self.RsiFactorGap}/{self.PpoGap}/{self.SlopeGap}/{self.MinDaysOut}/{self.MacdHistOutGap}/{self.MacdFactorOutGap}" self.SetRuntimeStatistic(str1, str2) self.SetWarmUp(timedelta(days=120)) # Minimum should be RSI period X 2 + 1 # --------------------------------------------------- Schedules -------------------------------------------- self.Schedule.On(self.DateRules.EveryDay(self.IndexEquityTicker),self.TimeRules.AfterMarketOpen(self.IndexEquityTicker,self.MinsAfter), self.Trade) # ------------------------------------------------- On Data ----------------------------------------------------- def OnData(self, data): self.allData = data # For checking assets existance in Time if self.IsWarmingUp: return if not self.TickerTable[self.IndexEquity].Rsi.IsReady: self.Debug("RSI not ready") return if not self.TickerTable[self.IndexEquity].Ppo.IsReady: self.Debug("PPO not ready") return if not self.TickerTable[self.IndexEquity].Macd.IsReady: self.Debug("MACD not ready") return if not self.TickerTable[self.IndexEquity].Rc.IsReady: self.Debug("RC not ready") return # --------------------------------------------------------------------------------------------------------------- # ---------------------------------------------- Trade Function ------------------------------------------------- # --------------------------------------------------------------------------------------------------------------- def Trade(self): self.TotalDays +=1 self.CollectIndicatorsAndTrends() self.DecideInOrOut() NumberOfInAssets = len(self.InAssetList) NumberOfOutAssets = len(self.OutAssetList) if self.MyInOutIndicator == 1: self.TotalDaysIn +=1 if self.IAMOut: self.Liquidate() self.IAMOut = False for asset in self.InAssetList: if not self.allData.ContainsKey(asset): # Asset does not exist in Data atm (ticker not traded ATM) ---- Cash will be left out !!! continue if not self.Securities[asset].Invested: # orderQuantity = self.CalculateOrderQuantity(asset, 1.00/NumberOfInAssets) #margin = self.Portfolio.GetMarginRemaining(asset, OrderDirection.Buy) #self.Debug("Security Margin " + str(asset) + " Margin: " + str(margin)) self.SetHoldings(asset, 1.00/NumberOfInAssets, False,f"PPO:{round(self.IndexEquity_PPO,2)} SLOPE:{round(self.IndexEquity_SLOPE,2)} RSI(Double Expo):{round(self.IndexEquity_RSI,2)}") self.Notify.Sms("+972542224488", self.AlgoName+" "+str(asset) +" In:" + str(self.Securities[asset].Price)) elif self.InOut: if not self.IAMOut: self.Liquidate() self.IAMOut = True if self.OutAssetSelector != 0: for asset in self.OutAssetList: if not self.allData.ContainsKey(asset): # Asset does not exist in Data atm (ticker not traded ATM) ---- Cash will be left out! continue self.SetHoldings(asset, 1.00/NumberOfOutAssets, False,"Out: "+self.ReasonOut) self.Notify.Sms("+972542224488", self.AlgoName+" "+str(self.IndexEquity) + " Out:" + str(self.Securities[self.IndexEquity].Price)) else: self.Notify.Sms("+972542224488", self.AlgoName+" "+str(self.IndexEquity) + " Out for Cash:" + str(self.Securities[self.IndexEquity].Price)) self.PlotIt() # -------------------------------- Decide IN or OUT ---------------------------------------- def GoldenCross(self): if self.IndexEquity_MACD > self.IndexEquity_MACD_SMA: return True else: return False def DeathCross(self): if abs(self.IndexEquity_MACD_Histogram) < self.MacdHistOutGap \ and self.IndexEquity_MACD < self.IndexEquity_MACD_SMA * self.MacdFactorOutGap: return True else: return False def DecideInOrOut(self): # Should we go OUT ? self.ReasonOut=" " if self.DeathCross(): self.ReasonOut = self.ReasonOut + f" MACD:{round(self.IndexEquity_MACD,2)}" #self.Debug(f" {self.Time} MACD SMA:{self.IndexEquity_SMA_HOUR} Price: {self.Securities[self.IndexEquity].Price}") #if self.Securities[self.IndexEquity].Price < self.IndexEquity_SMA_HOUR * self.SmaHourFactor: # self.ReasonOut = self.ReasonOut + f" SMA HOUR:{round(self.IndexEquity_SMA_HOUR,2)}" if self.IndexEquity_PPO < self.IndexEquity_PPO_SMA * self.PpoFactorGap and self.IndexEquity_PPO < self.PpoGap : self.ReasonOut = self.ReasonOut + f" PPO:{round(self.IndexEquity_PPO,2)}" #self.Debug(f" {self.Time} PPO SMA:{self.IndexEquity_SMA_HOUR} Price: {self.Securities[self.IndexEquity].Price}") if self.IndexEquity_RSI < self.IndexEquity_RSI_SMA * self.RsiFactorGap: self.ReasonOut = self.ReasonOut + f" RSI:{round(self.IndexEquity_RSI,2)}" #self.Debug(f" {self.Time} RSI SMA:{self.IndexEquity_SMA_HOUR} Price: {self.Securities[self.IndexEquity].Price}") if self.IndexEquity_SLOPE < self.SlopeGap: self.ReasonOut = self.ReasonOut + f" SLOPE:{round(self.IndexEquity_SLOPE,2)}" #self.Debug(f" {self.Time} SLOPE SMA:{self.IndexEquity_SMA_HOUR} Price: {self.Securities[self.IndexEquity].Price}") #if self.IndexEquity_ATR > self.IndexEquity_ATR_SMA_SLOW and self.IndexEquity_ATR > self.IndexEquity_ATR_SMA_FAST: # self.ReasonOut = self.ReasonOut + f" ATR:{round(self.IndexEquity_ATR,2)}" if self.ReasonOut != " ": self.MyInOutIndicator = 0 #if not self.IAMOut: self.DaysOut = 0 # Zero the DaysOut counter # Should we get IN? elif True: self.ReasonOut = " " if self.DaysOut >= self.MinDaysOut: self.MyInOutIndicator = 1 self.DaysOut = 0 else: self.DaysOut +=1 # ---- Unless in Market out Cooldown # ------------------------- Collect Indicators and trends before Trade --------------------------------- def CollectIndicatorsAndTrends(self): self.IndexEquity_RSI = self.TickerTable[self.IndexEquity].Rsi.Current.Value self.IndexEquity_RSI_SMA = self.TickerTable[self.IndexEquity].RsiSMA.Current.Value self.IndexEquity_PPO = self.TickerTable[self.IndexEquity].Ppo.Current.Value self.IndexEquity_PPO_SMA = self.TickerTable[self.IndexEquity].PpoSMA.Current.Value self.IndexEquity_MACD = self.TickerTable[self.IndexEquity].Macd.Current.Value self.IndexEquity_MACD_SMA = self.TickerTable[self.IndexEquity].MacdSMA.Current.Value self.IndexEquity_MACD_Signal = self.TickerTable[self.IndexEquity].Macd.Signal.Current.Value self.IndexEquity_MACD_Histogram = self.TickerTable[self.IndexEquity].Macd.Histogram.Current.Value self.IndexEquity_SLOPE = self.TickerTable[self.IndexEquity].Rc.Slope.Current.Value self.IndexEquity_ATR = self.TickerTable[self.IndexEquity].Atr.Current.Value self.IndexEquity_ATR_SMA_FAST = self.TickerTable[self.IndexEquity].AtrSMA_fast.Current.Value self.IndexEquity_ATR_SMA_SLOW = self.TickerTable[self.IndexEquity].AtrSMA_slow.Current.Value self.IndexEquity_SMA_HOUR = self.TickerTable[self.IndexEquity].SmaHour.Current.Value # for statistics self.TotalRSI+=self.IndexEquity_RSI self.AverageRSI = self.TotalRSI / self.TotalDays #self.Log(f"{self.Time} RSI:{round(self.IndexEquity_RSI,2)} RSISMA:{round(self.IndexEquity_RSI_SMA,2)} PPO:{round(self.IndexEquity_PPO,2)} Slope:{round(self.IndexEquity_SLOPE,2)} MACD:{round(self.IndexEquity_MACD,2)} MACDSMA:{round(self.IndexEquity_MACD_SMA,2)} MACDHIST:{round(self.IndexEquity_MACD_Histogram,2)}") # -------------------------------------------- Plot Function ------------------------------------------------ def PlotIt(self): self.Plot("In/Out Indicator","InOut",self.MyInOutIndicator*100) self.Plot("In/Out Indicator","RSI",self.IndexEquity_RSI) self.Plot("In/Out Indicator","RSI_SMA",self.IndexEquity_RSI_SMA) #self.Plot("PPO","PPO",self.IndexEquity_PPO) #self.Plot("RC Slope","SLOPE",self.IndexEquity_SLOPE) self.Plot("MACD","MACD",self.IndexEquity_MACD) self.Plot("MACD","MACD Signal",self.IndexEquity_MACD_Signal) self.Plot("MACD","MACD Hist",self.IndexEquity_MACD_Histogram) self.Plot("In Assets","In Assets",len(self.InAssetList)) #self.Plot("ATR","ATR",self.IndexEquity_ATR) try: self.Plot("Fear",self.Fear,self.Securities[self.Fear].Price) except: pass # Benchmark Ploting try: hist = self.History([self.IndexEquity], 2, Resolution.Daily)['close'].unstack(level= 0).dropna() self.IndexEquityHistory.append(hist[self.IndexEquity].iloc[-1]) perf = self.IndexEquityHistory[-1] / self.IndexEquityHistory[0] * self.StartCash self.Plot("Strategy Equity", self.IndexEquity, perf) except: # No data for this symbol pass try: hist = self.History([self.OutAssetList[0]], 2, Resolution.Daily)['close'].unstack(level= 0).dropna() self.OutEquityHistory.append(hist[self.OutAssetList[0]].iloc[-1]) perf = self.OutEquityHistory[-1] / self.OutEquityHistory[0] * self.StartCash self.Plot("Strategy Equity", self.OutAssetList[0], perf) except: # No data for this symbol pass # --------------------------------------------- On Securities Changed -------------------------------------------- def OnSecuritiesChanged(self, changes): self.changes = changes #self.Debug(f"OnSecuritiesChanged({self.Time}):: {changes}") if self.InAssetSelector == 0: for security in changes.RemovedSecurities: symbol = security.Symbol self.Liquidate(symbol) # ------ Need a smarter way ! if symbol in self.doNotEnter: continue if symbol in self.InAssetList: self.InAssetList.remove(symbol) for security in changes.AddedSecurities: symbol = security.Symbol if symbol in self.doNotEnter: continue if not symbol in self.InAssetList: self.InAssetList.append(symbol) #self.Debug(len(self.InAssetList)) # ---------------------------------------------- End of Algo -------------------------------------------------- def OnEndOfAlgorithm(self): self.Liquidate() if self.TotalDays>0: self.Debug(f"TPV:{round(self.Portfolio.TotalPortfolioValue,2)} Total Days:{self.TotalDays} Total Days In:{self.TotalDaysIn} {round(self.TotalDaysIn/self.TotalDays*100,2)}%") self.Debug(f"Average RSI: {self.AverageRSI}") # ---------------------------------------------- SymbolData -------------------------------------------------- class SymbolData: def __init__(self, symbol, rsi, rsiSMA, macd, macdSMA, rc, ppo, ppoSMA, atr, atrSMA_fast, atrSMA_slow, smaHour): self.Symbol = symbol self.Rsi = rsi self.RsiSMA =rsiSMA self.Macd = macd self.MacdSMA = macdSMA self.Rc = rc self.Ppo = ppo self.PpoSMA = ppoSMA self.Atr = atr self.AtrSMA_fast = atrSMA_fast self.AtrSMA_slow = atrSMA_slow self.SmaHour = smaHour