Overall Statistics |
Total Trades 21 Average Win 7.07% Average Loss 0% Compounding Annual Return 99.400% Drawdown 19.400% Expectancy 0 Net Profit 99.777% Sharpe Ratio 3.211 Probabilistic Sharpe Ratio 92.725% Loss Rate 0% Win Rate 100% Profit-Loss Ratio 0 Alpha 0.798 Beta 0.126 Annual Standard Deviation 0.256 Annual Variance 0.066 Information Ratio 1.693 Tracking Error 0.367 Treynor Ratio 6.552 Total Fees $77.62 |
class QInOut(QCAlgorithm): def Initialize(self): # ------------------------------------------- Parameters --------------------------------------- # Entry minute for the Trade algo after Market Open self.MinsAfter = int(self.GetParameter("minsAfter")) #Indicators' Days input self.SmaDays = int(self.GetParameter("smaDays")) 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.RcK = 2 self.PpoSlowDays = int(self.GetParameter("ppoSlowDays")) self.PpoFastDays = int(self.GetParameter("ppoFastDays")) # Indicators gaps and factors self.RsiLowGap = int(self.GetParameter("rsiLowGap")) self.RsiHighGap = int(self.GetParameter("rsiHighGap")) #Not Used ATM self.PpoGap = float(self.GetParameter("ppoGap")) self.SlopeGap = float(self.GetParameter("slopeGap")) self.MompGap = 100 # not in use ATM float(self.GetParameter("mompGap")) # MACD factors (Not Used ATM) self.MacdGap = float(self.GetParameter("macdGap")) self.MacdSignalGap = float(self.GetParameter("macdSignalGap")) self.MacdFactorOut = float(self.GetParameter("macdFactorOut")) self.MacdFactorIn = float(self.GetParameter("macdFactorIn")) self.MacdIn2Factor = float(self.GetParameter("macdIn2Factor")) # Trade time in minutes after market open self.MinDaysOut = int(self.GetParameter("daysOut")) # Select Out Equity list of assests self.InAssetSelector = int(self.GetParameter("inAssetSelector")) # Debug Algo Input parameters #self.Debug(f"After Market Open. rsiLow:{self.RsiLowGap} rsiHigh:{self.RsiHighGap} ppo:{self.PpoGap} rsiDays:{self.RsiDays} ppoFastDays:{self.PpoFastDays} ppoSlowDays:{self.PpoSlowDays}") self.StartYear = int(self.GetParameter("startYear")) self.EndYear = int(self.GetParameter("endYear")) # For Back testing self.SetStartDate(self.StartYear,1,1) self.SetEndDate(self.EndYear,12,31) self.StartCash = 100000 self.SetCash(self.StartCash) #self.Settings.FreePortfolioValuePercentage = self.CashOut --- No need ATM. Consider for Live self.SetWarmUp(timedelta(days=60)) # Minimum should be RSI period X 2 + 1 # --------------------------------------- Internal variables ----------------------------------- #For Plots on Equity Graph self.IndexEquityHistory = [] # now used for QQQ self.OutEquityHistory = [] # now used to TLT self.FirstDay = True self.IndexEquity_RSI_DAY_BEFORE = 0 self.IndexEquity_PPO_DAY_BEFORE = 0 self.IndexEquity_BIG_SMA_WINDOW_DAY_BEFORE = 0 self.IndexEquity_SMA_WINDOW_DAY_BEFORE = 0 self.IndexEquity_SLOPE_DAY_BEFORE = 0 self.IndexEquity_MOMP_DAY_BEFORE = 0 self.IndexEquity_SMA_DAY_BEFORE = 0 self.IndexEquity_MACD_DAY_BEFORE = 0 self.IndexEquity_MACD_Signal_DAY_BEFORE = 0 self.TickerTable = {} self.MyInOutIndicator = 1 # Default Enter the Market 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.Indicator_history = [] # Not Used ATM self.IndicatorRatio = 0 # Not Used ATM self.ReasonOut= "" # String to report reason to go out of the Market/InEquity # ---------------------------------------- Defining Tickers and collection Indicators --------------------------- self.SetBenchmark("SPY") # --- List to assets to buy when in the market (Buy equal weights) self.InAssets = {1:["QQQ"], 2:["MSFT","AAPL","FB","GOOG","AMZN"], 3:["ARKK"], 4:["CRSP","NTLA","EDIT"], 5:["SPY"] } self.IndexEquity = "QQQ" # Key index for In Out decisions self.OutEquity="TLT" # Out of market Asset. Consider making it a list of assets self.InAssetList = self.InAssets[self.InAssetSelector] # Asset List self.VXX = self.AddEquity("VXX", Resolution.Daily).Symbol # Used for Graph self.tickers = ["QQQ","TLT"] + self.InAssetList # ,"FDN","SPY","ARKK","SQQQ","GLD","PSQ","SH","EMTY", "XLU","XLP","TBT","IEF", "FDN","TLH"] for ticker in self.tickers: # --------- Add Equity self.AddEquity(ticker, Resolution.Minute) # --------- Add Equity indicators #smaLong = self.SMA(ticker, self.SmaLongDays, Resolution.Daily) #smaShort = self.SMA(ticker, self.SmaShortDays, Resolution.Daily) #momp = self.MOMP(ticker, 28, Resolution.Daily) #sma = self.SMA(ticker, self.SmaDays, Resolution.Daily) #ema = self.EMA(ticker, self.SmaDays, Resolution.Daily) rsi = self.RSI(ticker, self.RsiDays, MovingAverageType.DoubleExponential, Resolution.Daily) # MovingAverageType.Wilders, macd = self.MACD(ticker, self.MacdFastDays,self.MacdSlowDays,self.MacdSignalDays,MovingAverageType.Exponential, Resolution.Daily) rc = self.RC(ticker,self.RcDays, self.RcK, Resolution.Daily) ppo = self.PPO(ticker, self.PpoFastDays, self.PpoSlowDays, MovingAverageType.Exponential, Resolution.Daily) symbolData = SymbolData(ticker, rsi, rc, ppo, macd) self.TickerTable[ticker] = symbolData # --------------------------------------------------- Schedules -------------------------------------------- self.Schedule.On(self.DateRules.EveryDay("QQQ"),self.TimeRules.AfterMarketOpen("QQQ",self.MinsAfter), self.Trade) # ------------------------------------------------- On Data ----------------------------------------------------- def OnData(self, data): 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() if self.MyInOutIndicator == 1: self.TotalDaysIn +=1 NumberOfAssets = len(self.InAssetList) if self.Securities[self.OutEquity].Invested: self.Liquidate() for asset in self.InAssetList: if not self.Securities[asset].Invested: self.SetHoldings(asset, 1.00/NumberOfAssets, 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", asset +" In:" + str(self.Securities[asset].Price)) else: if not self.Securities[self.OutEquity].Invested: self.SetHoldings(self.OutEquity, 1.00, True,"Out: "+self.ReasonOut) self.Notify.Sms("+972542224488", self.IndexEquity + " Out:" + str(self.Securities[self.IndexEquity].Price)) self.AfterTradeUpdate() self.PlotIt() # -------------------------------- Decide IN or OUT ---------------------------------------- def GoldenCross(self): if abs (self.IndexEquity_MACD - self.IndexEquity_MACD_Signal) <0.1 and self.IndexEquity_MACD > self.IndexEquity_MACD_DAY_BEFORE: return True else: return False def DeathCross(self): if abs(self.IndexEquity_MACD - self.IndexEquity_MACD_Signal) < 0.1 and self.IndexEquity_MACD < self.IndexEquity_MACD_DAY_BEFORE: return True else: return False def DecideInOrOut(self): #if self.DeathCross(): # self.Debug(str(self.Time) + " DC OUT") #if self.GoldenCross(): # self.Debug(str(self.Time) + " GC IN") # Should we go OUT ? self.ReasonOut=" " #if self.DeathCross(): # self.ReasonOut = self.ReasonOut + f" MACD DC:{round(self.IndexEquity_MACD,2)}" if self.IndexEquity_PPO < self.PpoGap: self.ReasonOut = self.ReasonOut + f" PPO:{round(self.IndexEquity_PPO,2)}" if self.IndexEquity_RSI < self.RsiLowGap: self.ReasonOut = self.ReasonOut + f" RSI:{round(self.IndexEquity_RSI,2)}" if self.IndexEquity_SLOPE < self.SlopeGap: self.ReasonOut = self.ReasonOut + f" SLOPE:{round(self.IndexEquity_SLOPE,2)}" if self.ReasonOut != " ": self.MyInOutIndicator = 0 self.DaysOut = 0 # Zero the DaysOut counter # Should we get IN? elif True: # self.GoldenCross(): if self.DaysOut >= self.MinDaysOut: self.MyInOutIndicator = 1 self.DaysOut = 0 else: self.DaysOut +=1 # ---- Unless in Market out Cooldown else: if self.MyInOutIndicator == 0: self.DaysOut +=1 # ------------------------- Collect Indicators and trends before Trade --------------------------------- def CollectIndicatorsAndTrends(self): if not self.FirstDay: #self.IndexEquity_BIG_SMA_WINDOW_DAY_BEFORE = self.IndexEquity_BIG_SMA_WINDOW #self.IndexEquity_SMA_WINDOW_DAY_BEFORE = self.IndexEquity_SMALL_SMA_WINDOW #self.IndexEquity_MOMP_DAY_BEFORE = self.IndexEquity_MOMP #self.IndexEquity_SMA_DAY_BEFORE = self.IndexEquity_SMA self.IndexEquity_RSI_DAY_BEFORE = self.IndexEquity_RSI self.IndexEquity_PPO_DAY_BEFORE = self.IndexEquity_PPO self.IndexEquity_MACD_DAY_BEFORE = self.IndexEquity_MACD self.IndexEquity_MACD_Signal_DAY_BEFORE = self.IndexEquity_MACD_Signal self.IndexEquity_MACD_Histogram_DAY_BEFORE = self.IndexEquity_MACD_Histogram self.IndexEquity_SLOPE_DAY_BEFORE = self.IndexEquity_SLOPE #self.IndexEquity_BIG_SMA_WINDOW = self.TickerTable[self.IndexEquity].SmaLong.Current.Value #self.IndexEquity_SMALL_SMA_WINDOW = self.TickerTable[self.IndexEquity].SmaShort.Current.Value #self.IndexEquity_MOMP = self.TickerTable[self.IndexEquity].Momp.Current.Value #self.IndexEquity_SMA = self.TickerTable[self.IndexEquity].Sma.Current.Value #self.IndexEquity_EMA = self.TickerTable[self.IndexEquity].Ema.Current.Value self.IndexEquity_RSI = self.TickerTable[self.IndexEquity].Rsi.Current.Value self.IndexEquity_PPO = self.TickerTable[self.IndexEquity].Ppo.Current.Value self.IndexEquity_MACD = self.TickerTable[self.IndexEquity].Macd.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.FirstDay = False #self.Debug(f"{self.Time} PPO:{round(self.IndexEquity_PPO,2)} Slope:{round(self.IndexEquity_SLOPE,2)} RSI:{round(self.IndexEquity_RSI,2)}") # ------------------------------------ After Trade updates --------------------------------------------- def AfterTradeUpdate(self): #self.IndexEquity_BIG_SMA_WINDOW_DAY_BEFORE = self.IndexEquity_BIG_SMA_WINDOW #self.IndexEquity_SMA_WINDOW_DAY_BEFORE = self.IndexEquity_SMALL_SMA_WINDOW #self.IndexEquity_MOMP_DAY_BEFORE = self.IndexEquity_MOMP #self.IndexEquity_SMA_DAY_BEFORE = self.IndexEquity_SMA self.IndexEquity_RSI_DAY_BEFORE = self.IndexEquity_RSI self.IndexEquity_PPO_DAY_BEFORE = self.IndexEquity_PPO self.IndexEquity_MACD_DAY_BEFORE = self.IndexEquity_MACD self.IndexEquity_MACD_Signal_DAY_BEFORE = self.IndexEquity_MACD_Signal self.IndexEquity_SLOPE_DAY_BEFORE = self.IndexEquity_SLOPE self.Indicator_history.append(self.MyInOutIndicator) self.Indicator_history_changes = [x1 - x2 for x1, x2 in zip(self.Indicator_history[1:], self.Indicator_history)][-15:] try: self.IndicatorRatio = 1.0000 * (len([x for x in self.Indicator_history_changes if x != 0])/ len(self.Indicator_history_changes)) #if self.IndicatorRatio > 0: # self.Debug(str(self.Time) + str(self.IndicatorRatio)) except: pass # -------------------------------------------- Plot Function ------------------------------------------------ def PlotIt(self): self.Plot("In/Out Indicator","InOut",self.MyInOutIndicator) self.Plot("RSI","RSI",self.IndexEquity_RSI) self.Plot("PPO","PPO",self.IndexEquity_PPO) self.Plot("Indicators","SLOPE",self.IndexEquity_SLOPE) self.Plot("Indicators","MACD",self.IndexEquity_MACD) self.Plot("Indicators","MACD Signal",self.IndexEquity_MACD_Signal) self.Plot("Indicators","MACD Hist",self.IndexEquity_MACD_Histogram) try: self.Plot("VXX","TLT",self.Securities[self.VXX.Value].Price) except: pass # Benchmark Ploting 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) hist = self.History([self.OutEquity], 2, Resolution.Daily)['close'].unstack(level= 0).dropna() self.OutEquityHistory.append(hist[self.OutEquity].iloc[-1]) perf = self.OutEquityHistory[-1] / self.OutEquityHistory[0] * self.StartCash self.Plot("Strategy Equity", self.OutEquity, perf) # ----------------------------------------------- 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)}%") # ---------------------------------------------- SymbolData -------------------------------------------------- class SymbolData: def __init__(self, symbol, rsi, rc, ppo, macd): self.Symbol = symbol self.Rsi = rsi self.Macd = macd self.Rc = rc self.Ppo = ppo #self.Momp = momp #self.Sma = sma #self.Ema = ema #self.SmaLong = smaLong #self.SmaShort = smaShort