| Overall Statistics |
|
Total Trades 184 Average Win 2.78% Average Loss -1.26% Compounding Annual Return 206.262% Drawdown 20.700% Expectancy 1.051 Net Profit 194.896% Sharpe Ratio 3.462 Probabilistic Sharpe Ratio 92.121% Loss Rate 36% Win Rate 64% Profit-Loss Ratio 2.21 Alpha 1.206 Beta -1.096 Annual Standard Deviation 0.387 Annual Variance 0.15 Information Ratio 2.758 Tracking Error 0.53 Treynor Ratio -1.223 Total Fees $290.63 Estimated Strategy Capacity $440000.00 Lowest Capacity Asset TMF UBTUG7D0B7TX |
from AlgorithmImports import *
from collections import deque
class CustomDrawdown():
def __init__(self, algorithm, symbol, drawdown_period):
self.algorithm = algorithm
self.drawdown_period = drawdown_period
self.High_Queue = deque(maxlen=self.drawdown_period)
self.Low_Queue = deque(maxlen=self.drawdown_period)
self.symbol = symbol
self.Current_Value = None
self.First_Intraday_Update_Of_Day = True
self.Last_Main_Bar = None
self.New_Main_Bar = True
self.WarmUP = True
self.Updated_Today = False
def Receive_Main_Bar(self, bar):
if not self.First_Intraday_Update_Of_Day:
del self.High_Queue[0]
del self.Low_Queue[0]
self.New_Main_Bar = True
self.High_Queue.appendleft(bar.High)
self.Low_Queue.appendleft(bar.Low)
if len(self.High_Queue) == self.drawdown_period and len(self.Low_Queue) == self.drawdown_period:
self.Calculate_Drawdown()
def Receive_Sub_Bar(self, high, low):
if len(self.High_Queue) == self.drawdown_period and len(self.Low_Queue) == self.drawdown_period:
if self.First_Intraday_Update_Of_Day:
self.Last_Main_High = self.High_Queue[0]
self.Last_Main_Low = self.Low_Queue[0]
self.High_Queue.appendleft(high)
self.Low_Queue.appendleft(low)
self.First_Intraday_Update_Of_Day = False
else:
del self.High_Queue[0]
del self.Low_Queue[0]
self.High_Queue.appendleft(high)
self.Low_Queue.appendleft(low)
self.Calculate_Drawdown()
def Calculate_Drawdown(self):
#Drawdown needs to check that the LOW occurs AFTER the high! LOL
if len(self.High_Queue) == self.drawdown_period and len(self.Low_Queue) == self.drawdown_period:
high = 0
idx = None
#if self.algorithm.Time.hour < 23 and self.algorithm.Time.hour > 14 and self.symbol == "SPY" and self.drawdown_period == 8:
#self.algorithm.Debug(f"H {self.High_Queue} {self.symbol}")
#self.algorithm.Debug(f"L {self.Low_Queue} {self.symbol}")
for i in range(0, len(self.High_Queue)):
if self.High_Queue[i] > high:
high = self.High_Queue[i]
idx = i
low = 100000000
for i in range(0, idx+1):
if self.Low_Queue[i] < low:
low = self.Low_Queue[i]
#if self.algorithm.Time.hour < 23 and self.algorithm.Time.hour > 14 and self.symbol == "SPY" and self.drawdown_period == 8:
#self.algorithm.Debug(f"HH {high} LL {low}")
hh = high
ll = low
self.Current_Value = abs((ll - hh) / hh)
if self.New_Main_Bar:
self.First_Intraday_Update_Of_Day = True
self.New_Main_Bar = False
from AlgorithmImports import *
from collections import deque
class Intraday_Updating_SMA():
def __init__(self, algorithm, sma_period):
self.algorithm = algorithm
self.sma_period = sma_period
self.Bar_Queue = deque(maxlen=self.sma_period)
self.Current_Value = None
self.First_Intraday_Update_Of_Day = True
self.Last_Main_Bar = None
self.Updated_Today = False
self.New_Main_Bar = True
self.WarmUP = True
def Receive_Main_Bar(self, bar):
if not self.First_Intraday_Update_Of_Day:
del self.Bar_Queue[0]
self.New_Main_Bar = True
self.Bar_Queue.appendleft(bar.Close)
if len(self.Bar_Queue) == self.sma_period:
self.Calculate_SMA()
def Receive_Sub_Bar(self, bar):
if len(self.Bar_Queue) == self.sma_period:
if self.First_Intraday_Update_Of_Day:
self.Last_Main_Bar = self.Bar_Queue[0]
self.Bar_Queue.appendleft(bar)
self.First_Intraday_Update_Of_Day = False
else:
del self.Bar_Queue[0]
self.Bar_Queue.appendleft(bar)
self.Calculate_SMA()
def Calculate_SMA(self):
if len(self.Bar_Queue) == self.sma_period:
self.Current_Value = sum(self.Bar_Queue) / len(self.Bar_Queue)
if self.New_Main_Bar:
self.First_Intraday_Update_Of_Day = True
self.New_Main_Bar = False
from AlgorithmImports import *
from collections import deque
class Intraday_Updating_Cumulative_Return():
def __init__(self, algorithm, return_period):
self.algorithm = algorithm
self.return_period = return_period +1
self.Bar_Queue = deque(maxlen=self.return_period)
self.Current_Value = None
self.First_Intraday_Update_Of_Day = True
self.Last_Main_Bar = None
self.New_Main_Bar = True
self.WarmUP = True
self.Updated_Today = False
def Receive_Main_Bar(self, bar):
if not self.First_Intraday_Update_Of_Day:
del self.Bar_Queue[0]
self.New_Main_Bar = True
self.Bar_Queue.appendleft(bar.Close)
if len(self.Bar_Queue) == self.return_period:
self.Calculate_Cumulative_Return()
def Receive_Sub_Bar(self, bar):
if len(self.Bar_Queue) == self.return_period:
if self.First_Intraday_Update_Of_Day:
self.Last_Main_Bar = self.Bar_Queue[0]
self.Bar_Queue.appendleft(bar)
self.First_Intraday_Update_Of_Day = False
else:
del self.Bar_Queue[0]
self.Bar_Queue.appendleft(bar)
self.Calculate_Cumulative_Return()
def Calculate_Cumulative_Return(self):
if len(self.Bar_Queue) == self.return_period:
self.Current_Value = (self.Bar_Queue[0] - self.Bar_Queue[self.return_period-1]) / self.Bar_Queue[self.return_period-1]
if self.New_Main_Bar:
self.First_Intraday_Update_Of_Day = True
self.New_Main_Bar = False
"""
**
* OG v1.2 of Risk On/Risk Off Hedgefundie (No K-1) | okhi2u more TMF mod https://app.composer.trade/symphony/3yTnxyaq0AkD0ohTq1OQ/details
*
*
* Description: This Strategy rebalances ETF's based on various Indicators
**
"""
from AlgorithmImports import *
from custom_sma import Intraday_Updating_SMA
from custom_cumulative_return import Intraday_Updating_Cumulative_Return
from custom_rsi import Intraday_Updating_RSI
from custom_drawdown import CustomDrawdown
import config
class ETF_Rebalancing(QCAlgorithm):
def Initialize(self):
self.SetStartDate(config.BACKTEST_START_YEAR, config.BACKTEST_START_MONTH, config.BACKTEST_START_DAY) # Set Backtest Start Date
self.SetEndDate(config.BACKTEST_END_YEAR, config.BACKTEST_END_MONTH, config.BACKTEST_END_DAY) # Set Backtest End Date
self.SetCash(config.BACKTEST_ACCOUNT_CASH) # Set Backtest Strategy Cash
self.Rebalance_Threshold = config.REBALANCE_THRESHOLD/100
self.UniverseSettings.Leverage = 4
self.TECL = None
self.SOXL = None
self.FAS = None
self.TQQQ = None
self.UPRO = None
self.TMF = None
self.USDO = None
self.SQQQ = None
self.TBF = None
self.BIL = None
self.PHDG = None
self.TMV = None
self.UDOW = None
self.MOAT = None
self.BRKB = None
self.USMV = None
self.SPY = None
self.QQQ = None
self.SPXL = None
self.DBC = None
self.DBA = None
self.ICSH = None
self.TYD = None
self.URTY = None
self.XLU = None
self.NUGT = None
self.TYO = None
self.VIXM = None
self.IEF = None
self.SOXS = None
self.BND = None
self.VTI = None
self.SHY = None
self.USDU = None
self.TLT = None
self.UVXY = None
# if (config.CASH_BUFFER * ) != 0:
# (config.CASH_BUFFER * ) = (config.CASH_BUFFER * )/100
# else:
# (config.CASH_BUFFER * ) = 0
self.Sub_4_List = []
self.Symbol_List = []
self.ETFs = config.ETFS
self.Trade_Decision_Time = config.TRADE_DECISION_TIME
self.Order_Delay = config.ORDER_DELAY # Enable this for Delayed orders
self.Update_Intraday = False
self.Trade_Decision = False
for etf in self.ETFs:
try:
self.Symbol_List.append(self.AddEquity(etf, Resolution.Minute, dataNormalizationMode=DataNormalizationMode.Adjusted).Symbol)
except:
self.Debug(f"Unable to add stock {etf} to the algorithm")
self.Symbol_Dictionary = {}
for symbol in self.Symbol_List:
self.Symbol_Dictionary[symbol] = SymbolData(self, symbol)
self.Schedule.On(self.DateRules.EveryDay(), self.TimeRules.BeforeMarketClose("SPY", self.Trade_Decision_Time), self.Set_Update_Intraday)
self.Schedule.On(self.DateRules.EveryDay(), self.TimeRules.BeforeMarketClose("SPY", self.Trade_Decision_Time - self.Order_Delay), self.Delayed_Orders)
def Set_Update_Intraday(self):
self.Update_Intraday = True
self.Trade_Decision = True
def Orders(self):
for symbol in self.Symbol_Dictionary.keys():
Symbol_Object = self.Symbol_Dictionary[symbol]
if Symbol_Object.Current_Position < (Symbol_Object.Counter - self.Rebalance_Threshold) or Symbol_Object.Current_Position > (Symbol_Object.Counter + self.Rebalance_Threshold):
if Symbol_Object.Counter <= 0:
self.Liquidate(symbol)
if not config.USE_ORDER_DELAY:
for symbol in self.Symbol_Dictionary.keys():
Symbol_Object = self.Symbol_Dictionary[symbol]
if Symbol_Object.Current_Position < (Symbol_Object.Counter - self.Rebalance_Threshold) or Symbol_Object.Current_Position > (Symbol_Object.Counter + self.Rebalance_Threshold):
if Symbol_Object.Counter >= 0:
self.SetHoldings(symbol, Symbol_Object.Counter, tag=f"{Symbol_Object.Function}") # Disable this for Delayed orders
elif not self.Portfolio[symbol].Invested and Symbol_Object.Current_Position == 0 and Symbol_Object.Counter > 0:
self.SetHoldings(symbol, Symbol_Object.Counter, tag=f"{Symbol_Object.Function}")
# Enable this for Delayed orders
def Delayed_Orders(self):
if config.USE_ORDER_DELAY:
for symbol in self.Symbol_Dictionary.keys():
Symbol_Object = self.Symbol_Dictionary[symbol]
if Symbol_Object.Current_Position < (Symbol_Object.Counter - self.Rebalance_Threshold) or Symbol_Object.Current_Position > (Symbol_Object.Counter + self.Rebalance_Threshold):
if Symbol_Object.Counter >= 0:
self.SetHoldings(symbol, Symbol_Object.Counter, tag=f"{Symbol_Object.Function}")
elif not self.Portfolio[symbol].Invested and Symbol_Object.Current_Position == 0 and Symbol_Object.Counter > 0:
self.SetHoldings(symbol, Symbol_Object.Counter, tag=f"{Symbol_Object.Function}")
def Main_Strategy(self):
self.OG_v1_2_Risk_On_Off_Hedgefund()
def OG_v1_2_Risk_On_Off_Hedgefund(self):
if self.VIXM.Rsi_40.Current_Value > 69:
self.Risk_Off_Market_Crash()
else:
if self.BND.Return_60.Current_Value > self.BIL.Return_60.Current_Value:
self.Risk_On_Normal_Market_Conditions()
else:
if self.TLT.Return_20.Current_Value < self.BIL.Return_20.Current_Value:
self.Risk_Off_Rising_Rates()
else:
self.Risk_On_Falling_Rates_HFEA()
def Risk_Off_Market_Crash(self):
self.SHY.Counter += 1 - (config.CASH_BUFFER * 1)
def Risk_On_Normal_Market_Conditions(self):
self.Sub_Strategy_1()
self.TMF.Counter += 0.45 - (config.CASH_BUFFER * 0.45)
def Sub_Strategy_1(self):
rsi_list = [(self.TECL, self.TECL.Rsi_10.Current_Value),
(self.SOXL, self.SOXL.Rsi_10.Current_Value),
(self.FAS, self.FAS.Rsi_10.Current_Value),
(self.TQQQ, self.TQQQ.Rsi_10.Current_Value),
(self.UPRO, self.UPRO.Rsi_10.Current_Value)
]
sorted_rsi_list = sorted(rsi_list, key=lambda x: x[1], reverse=False)
counter_ = 0
for symbol, symbol_return in sorted_rsi_list:
if counter_ >= 3:
break
symbol.Counter += 0.183 - (config.CASH_BUFFER * 0.183)
counter_ += 1
def Risk_Off_Rising_Rates(self):
self.USDU.Counter += 0.5 - (config.CASH_BUFFER * 0.5)
self.Sub_Strategy_2()
def Sub_Strategy_2(self):
if self.SQQQ.Rsi_20.Current_Value < self.TBF.Rsi_20.Current_Value:
self.SQQQ.Counter += 0.5 - (config.CASH_BUFFER * 0.5)
else:
self.TBF.Counter += 0.5 - (config.CASH_BUFFER * 0.5)
def Risk_On_Falling_Rates_HFEA(self):
if self.SPY.Drawdown_10.Current_Value < 0.05:
self.HFEA_Refined_Risk_On()
else:
self.HFEA_Refined_Risk_Off()
def HFEA_Refined_Risk_On(self):
self.UPRO.Counter += 0.55 - (config.CASH_BUFFER * 0.55)
self.TMF.Counter += 0.45 - (config.CASH_BUFFER * 0.45)
def HFEA_Refined_Risk_Off(self):
self.BIL.Counter += 1 - (config.CASH_BUFFER * 1)
def Clear_Weightings(self):
portfolio_value = self.Portfolio.TotalPortfolioValue
for symbol in self.Symbol_Dictionary.keys():
Symbol_Object = self.Symbol_Dictionary[symbol]
Symbol_Object.Counter = 0
symbol_holding = self.Portfolio[symbol].AbsoluteHoldingsValue
allocation_per = symbol_holding / portfolio_value
Symbol_Object.Current_Position = allocation_per
def OnData(self, data: Slice):
for symbol in self.Symbol_Dictionary.keys():
Symbol_Object = self.Symbol_Dictionary[symbol]
if Symbol_Object is None:continue
if self.SHY is None and symbol == "SHY":
self.SHY = Symbol_Object
if self.TECL is None and symbol == "TECL":
self.TECL = Symbol_Object
if self.SOXL is None and symbol == "SOXL":
self.SOXL = Symbol_Object
if self.TQQQ is None and symbol == "TQQQ":
self.TQQQ = Symbol_Object
if self.UPRO is None and symbol == "UPRO":
self.UPRO = Symbol_Object
if self.TMF is None and symbol == "TMF":
self.TMF = Symbol_Object
if self.USDU is None and symbol == "USDU":
self.USDU = Symbol_Object
if self.SQQQ is None and symbol == "SQQQ":
self.SQQQ = Symbol_Object
if self.TBF is None and symbol == "TBF":
self.TBF = Symbol_Object
if self.BIL is None and symbol == "BIL":
self.BIL = Symbol_Object
if self.PHDG is None and symbol == "PHDG":
self.PHDG = Symbol_Object
if self.TMV is None and symbol == "TMV":
self.TMV = Symbol_Object
if self.UDOW is None and symbol == "UDOW":
self.UDOW = Symbol_Object
if self.MOAT is None and symbol == "MOAT":
self.MOAT = Symbol_Object
if self.BRKB is None and symbol == "BRK.B":
self.BRKB = Symbol_Object
if self.USMV is None and symbol == "USMV":
self.USMV = Symbol_Object
if self.SPY is None and symbol == "SPY":
self.SPY = Symbol_Object
if self.QQQ is None and symbol == "QQQ":
self.QQQ = Symbol_Object
if self.SPXL is None and symbol == "SPXL":
self.SPXL = Symbol_Object
if self.DBC is None and symbol == "DBC":
self.DBC = Symbol_Object
if self.DBA is None and symbol == "DBA":
self.DBA = Symbol_Object
if self.ICSH is None and symbol == "ICSH":
self.ICSH = Symbol_Object
if self.TYD is None and symbol == "TYD":
self.TYD = Symbol_Object
if self.URTY is None and symbol == "URTY":
self.URTY = Symbol_Object
if self.XLU is None and symbol == "XLU":
self.XLU = Symbol_Object
if self.NUGT is None and symbol == "NUGT":
self.NUGT = Symbol_Object
if self.TYO is None and symbol == "TYO":
self.TYO = Symbol_Object
if self.VIXM is None and symbol == "VIXM":
self.VIXM = Symbol_Object
if self.IEF is None and symbol == "IEF":
self.IEF = Symbol_Object
if self.SOXS is None and symbol == "SOXS":
self.SOXS = Symbol_Object
if self.BND is None and symbol == "BND":
self.BND = Symbol_Object
if self.VTI is None and symbol == "VTI":
self.VTI = Symbol_Object
if self.FAS is None and symbol == "FAS":
self.FAS = Symbol_Object
if self.TLT is None and symbol == "TLT":
self.TLT = Symbol_Object
if self.UVXY is None and symbol == "UVXY":
self.UVXY = Symbol_Object
if self.Update_Intraday:
if self.SHY is not None:
self.SHY.Update_With_Intraday_Bar(self.Securities["SHY"].Close)
if self.TECL is not None:
self.TECL.Update_With_Intraday_Bar(self.Securities["TECL"].Close)
if self.SOXL is not None:
self.SOXL.Update_With_Intraday_Bar(self.Securities["SOXL"].Close)
if self.FAS is not None:
self.FAS.Update_With_Intraday_Bar(self.Securities["FAS"].Close)
if self.TQQQ is not None:
self.TQQQ.Update_With_Intraday_Bar(self.Securities["TQQQ"].Close)
if self.UPRO is not None:
self.UPRO.Update_With_Intraday_Bar(self.Securities["UPRO"].Close)
if self.TMF is not None:
self.TMF.Update_With_Intraday_Bar(self.Securities["TMF"].Close)
if self.USDU is not None:
self.USDU.Update_With_Intraday_Bar(self.Securities["USDU"].Close)
if self.SQQQ is not None:
self.SQQQ.Update_With_Intraday_Bar(self.Securities["SQQQ"].Close)
if self.TBF is not None:
self.TBF.Update_With_Intraday_Bar(self.Securities["TBF"].Close)
if self.BIL is not None:
self.BIL.Update_With_Intraday_Bar(self.Securities["BIL"].Close)
if self.TMV is not None:
self.TMV.Update_With_Intraday_Bar(self.Securities["TMV"].Close)
if self.UDOW is not None:
self.UDOW.Update_With_Intraday_Bar(self.Securities["UDOW"].Close)
if self.MOAT is not None:
self.MOAT.Update_With_Intraday_Bar(self.Securities["MOAT"].Close)
if self.BRKB is not None:
self.BRKB.Update_With_Intraday_Bar(self.Securities["BRK.B"].Close)
if self.USMV is not None:
self.USMV.Update_With_Intraday_Bar(self.Securities["USMV"].Close)
if self.SPY is not None:
self.SPY.Update_With_Intraday_Bar(self.Securities["SPY"].Close)
if self.QQQ is not None:
self.QQQ.Update_With_Intraday_Bar(self.Securities["QQQ"].Close)
if self.SPXL is not None:
self.SPXL.Update_With_Intraday_Bar(self.Securities["SPXL"].Close)
if self.DBC is not None:
self.DBC.Update_With_Intraday_Bar(self.Securities["DBC"].Close)
if self.DBA is not None:
self.DBA.Update_With_Intraday_Bar(self.Securities["DBA"].Close)
if self.ICSH is not None:
self.ICSH.Update_With_Intraday_Bar(self.Securities["ICSH"].Close)
if self.TYD is not None:
self.TYD.Update_With_Intraday_Bar(self.Securities["TYD"].Close)
if self.URTY is not None:
self.URTY.Update_With_Intraday_Bar(self.Securities["URTY"].Close)
if self.XLU is not None:
self.XLU.Update_With_Intraday_Bar(self.Securities["XLU"].Close)
if self.NUGT is not None:
self.NUGT.Update_With_Intraday_Bar(self.Securities["NUGT"].Close)
if self.TYO is not None:
self.TYO.Update_With_Intraday_Bar(self.Securities["TYO"].Close)
if self.VIXM is not None:
self.VIXM.Update_With_Intraday_Bar(self.Securities["VIXM"].Close)
if self.IEF is not None:
self.IEF.Update_With_Intraday_Bar(self.Securities["IEF"].Close)
if self.SOXS is not None:
self.SOXS.Update_With_Intraday_Bar(self.Securities["SOXS"].Close)
if self.BND is not None:
self.BND.Update_With_Intraday_Bar(self.Securities["BND"].Close)
if self.VTI is not None:
self.VTI.Update_With_Intraday_Bar(self.Securities["VTI"].Close)
if self.TLT is not None:
self.TLT.Update_With_Intraday_Bar(self.Securities["TLT"].Close)
if self.UVXY is not None:
self.UVXY.Update_With_Intraday_Bar(self.Securities["UVXY"].Close)
if self.SHY is None or self.TECL is None or self.SOXL is None or self.FAS is None or self.TQQQ is None or self.UPRO is None or self.TMF is None or self.USDU is None or self.SQQQ is None or self.TBF is None or self.BIL is None or self.PHDG is None or self.TMV is None or self.UDOW is None or self.MOAT is None or self.BRKB is None or self.USMV is None or self.SPY is None or self.QQQ is None or self.SPXL is None or self.DBC is None or self.DBA is None or self.ICSH is None or self.TYD is None or self.URTY is None or self.XLU is None or self.NUGT is None or self.TYO is None or self.VIXM is None or self.IEF is None or self.SOXS is None or self.BND is None or self.VTI is None or self.TLT is None or self.UVXY is None:
return
self.Update_Intraday = False
if self.Trade_Decision:
self.Clear_Weightings()
self.Main_Strategy()
self.Orders()
self.Trade_Decision = False
class SymbolData():
def __init__(self, algorithm, symbol):
self.algorithm = algorithm
self.symbol = symbol
self.Daily_High = None
self.Daily_Low = None
self.Counter = 0
self.Rsi_5_Period = config.RSI_5_PERIOD
self.Rsi_10_Period = config.RSI_10_PERIOD
self.Rsi_15_Period = config.RSI_15_PERIOD
self.Rsi_16_Period = config.RSI_16_PERIOD
self.Rsi_18_Period = config.RSI_18_PERIOD
self.Rsi_20_Period = config.RSI_20_PERIOD
self.Rsi_40_Period = config.RSI_40_PERIOD
self.Rsi_60_Period = config.RSI_60_PERIOD
self.Rsi_200_Period = config.RSI_200_PERIOD
self.Return_5_Period = config.RETURN_5_PERIOD
self.Return_10_Period = config.RETURN_10_PERIOD
self.Return_15_Period = config.RETURN_15_PERIOD
self.Return_20_Period = config.RETURN_20_PERIOD
self.Return_30_Period = config.RETURN_30_PERIOD
self.Return_60_Period = config.RETURN_60_PERIOD
self.Return_70_Period = config.RETURN_70_PERIOD
self.Return_200_Period = config.RETURN_200_PERIOD
self.SMA_23_Period = config.SMA_23_PERIOD
self.SMA_150_Period = config.SMA_150_PERIOD
self.SMA_200_Period = config.SMA_200_PERIOD
self.Custom_Drawdown_5_Period = config.CUSTOM_DRAWDOWN_5_PERIOD
self.Custom_Drawdown_8_Period = config.CUSTOM_DRAWDOWN_8_PERIOD
self.Custom_Drawdown_10_Period = config.CUSTOM_DRAWDOWN_10_PERIOD
self.Function = ""
self.Current_Position = 0
self.Rsi_5 = Intraday_Updating_RSI(algorithm, self.Rsi_5_Period)
self.Rsi_10 = Intraday_Updating_RSI(algorithm, self.Rsi_10_Period)
self.Rsi_15 = Intraday_Updating_RSI(algorithm, self.Rsi_15_Period)
self.Rsi_16 = Intraday_Updating_RSI(algorithm, self.Rsi_16_Period)
self.Rsi_18 = Intraday_Updating_RSI(algorithm, self.Rsi_18_Period)
self.Rsi_20 = Intraday_Updating_RSI(algorithm, self.Rsi_20_Period)
self.Rsi_40 = Intraday_Updating_RSI(algorithm, self.Rsi_40_Period)
self.Rsi_60 = Intraday_Updating_RSI(algorithm, self.Rsi_60_Period)
self.Rsi_200 = Intraday_Updating_RSI(algorithm, self.Rsi_200_Period)
self.Return_5 = Intraday_Updating_Cumulative_Return(algorithm, self.Return_5_Period)
self.Return_10 = Intraday_Updating_Cumulative_Return(algorithm, self.Return_10_Period)
self.Return_15 = Intraday_Updating_Cumulative_Return(algorithm, self.Return_15_Period)
self.Return_20 = Intraday_Updating_Cumulative_Return(algorithm, self.Return_20_Period)
self.Return_30 = Intraday_Updating_Cumulative_Return(algorithm, self.Return_30_Period)
self.Return_60 = Intraday_Updating_Cumulative_Return(algorithm, self.Return_60_Period)
self.Return_70 = Intraday_Updating_Cumulative_Return(algorithm, self.Return_70_Period)
self.Return_200 = Intraday_Updating_Cumulative_Return(algorithm, self.Return_200_Period)
self.SMA_23 = Intraday_Updating_SMA(algorithm, self.SMA_23_Period)
self.SMA_150 = Intraday_Updating_SMA(algorithm, self.SMA_150_Period)
self.SMA_200 = Intraday_Updating_SMA(algorithm, self.SMA_200_Period)
self.Drawdown_5 = CustomDrawdown(algorithm, symbol, self.Custom_Drawdown_5_Period)
self.Drawdown_8 = CustomDrawdown(algorithm, symbol, self.Custom_Drawdown_8_Period)
self.Drawdown_10 = CustomDrawdown(algorithm, symbol, self.Custom_Drawdown_10_Period)
self.WarmUp = True
self.Daily_Bar_Consolidator = TradeBarConsolidator(timedelta(days=1))
self.algorithm.SubscriptionManager.AddConsolidator(self.symbol, self.Daily_Bar_Consolidator)
self.Daily_Bar_Consolidator.DataConsolidated += self.OnDataConsolidated
history = self.algorithm.History[TradeBar](self.symbol, 400, Resolution.Daily)
for bar in history:
self.Daily_Bar_Consolidator.Update(bar)
self.WarmUp = False
self.Minute_Bar_Consolidator = TradeBarConsolidator(timedelta(minutes=1))
self.algorithm.SubscriptionManager.AddConsolidator(self.symbol, self.Minute_Bar_Consolidator)
self.Minute_Bar_Consolidator.DataConsolidated += self.OnMinuteConsolidated
self.algorithm.Schedule.On(self.algorithm.DateRules.EveryDay(), self.algorithm.TimeRules.AfterMarketOpen(self.symbol, -10), self.Reset_HL)
def Reset_HL(self):
self.Daily_High = None
self.Daily_Low = None
def OnMinuteConsolidated(self, sender, bar):
if self.Daily_High is None:
self.Daily_High = bar.High
if self.Daily_Low is None:
self.Daily_Low = bar.Low
if self.Daily_High is not None and bar.High > self.Daily_High:
self.Daily_High = bar.High
if self.Daily_Low is not None and bar.Low < self.Daily_Low:
self.Daily_Low = bar.Low
def OnDataConsolidated(self, sender, bar):
self.Rsi_5.Receive_Main_Bar(bar)
self.Rsi_10.Receive_Main_Bar(bar)
self.Rsi_15.Receive_Main_Bar(bar)
self.Rsi_16.Receive_Main_Bar(bar)
self.Rsi_18.Receive_Main_Bar(bar)
self.Rsi_20.Receive_Main_Bar(bar)
self.Rsi_40.Receive_Main_Bar(bar)
self.Rsi_60.Receive_Main_Bar(bar)
self.Rsi_200.Receive_Main_Bar(bar)
self.Return_5.Receive_Main_Bar(bar)
self.Return_10.Receive_Main_Bar(bar)
self.Return_15.Receive_Main_Bar(bar)
self.Return_20.Receive_Main_Bar(bar)
self.Return_30.Receive_Main_Bar(bar)
self.Return_60.Receive_Main_Bar(bar)
self.Return_70.Receive_Main_Bar(bar)
self.Return_200.Receive_Main_Bar(bar)
self.SMA_23.Receive_Main_Bar(bar)
self.SMA_150.Receive_Main_Bar(bar)
self.SMA_200.Receive_Main_Bar(bar)
self.Drawdown_5.Receive_Main_Bar(bar)
self.Drawdown_8.Receive_Main_Bar(bar)
self.Drawdown_10.Receive_Main_Bar(bar)
self.Rsi_5.Updated_Today = True
self.Rsi_10.Updated_Today = True
self.Rsi_15.Updated_Today = True
self.Rsi_16.Updated_Today = True
self.Rsi_18.Updated_Today = True
self.Rsi_20.Updated_Today = True
self.Rsi_40.Updated_Today = True
self.Rsi_60.Updated_Today = True
self.Rsi_200.Updated_Today = True
self.Return_5.Updated_Today = True
self.Return_10.Updated_Today = True
self.Return_15.Updated_Today = True
self.Return_20.Updated_Today = True
self.Return_30.Updated_Today = True
self.Return_60.Updated_Today = True
self.Return_70.Updated_Today = True
self.Return_200.Updated_Today = True
self.SMA_23.Updated_Today = True
self.SMA_150.Updated_Today = True
self.SMA_200.Updated_Today = True
self.Drawdown_5.Updated_Today = True
self.Drawdown_8.Updated_Today = True
self.Drawdown_10.Updated_Today = True
def Update_With_Intraday_Bar(self, price):
if not self.WarmUp:
if self.Rsi_5.Updated_Today:
self.Rsi_5.Receive_Sub_Bar(price)
if self.Rsi_10.Updated_Today:
self.Rsi_10.Receive_Sub_Bar(price)
if self.Rsi_15.Updated_Today:
self.Rsi_15.Receive_Sub_Bar(price)
if self.Rsi_16.Updated_Today:
self.Rsi_16.Receive_Sub_Bar(price)
if self.Rsi_18.Updated_Today:
self.Rsi_18.Receive_Sub_Bar(price)
if self.Rsi_20.Updated_Today:
self.Rsi_20.Receive_Sub_Bar(price)
if self.Rsi_40.Updated_Today:
self.Rsi_40.Receive_Sub_Bar(price)
if self.Rsi_60.Updated_Today:
self.Rsi_60.Receive_Sub_Bar(price)
if self.Rsi_200.Updated_Today:
self.Rsi_200.Receive_Sub_Bar(price)
if self.Return_5.Updated_Today:
self.Return_5.Receive_Sub_Bar(price)
if self.Return_10.Updated_Today:
self.Return_10.Receive_Sub_Bar(price)
if self.Return_15.Updated_Today:
self.Return_15.Receive_Sub_Bar(price)
if self.Return_20.Updated_Today:
self.Return_20.Receive_Sub_Bar(price)
if self.Return_30.Updated_Today:
self.Return_30.Receive_Sub_Bar(price)
if self.Return_60.Updated_Today:
self.Return_60.Receive_Sub_Bar(price)
if self.Return_70.Updated_Today:
self.Return_70.Receive_Sub_Bar(price)
if self.Return_200.Updated_Today:
self.Return_200.Receive_Sub_Bar(price)
if self.SMA_23.Updated_Today:
self.SMA_23.Receive_Sub_Bar(price)
if self.SMA_150.Updated_Today:
self.SMA_150.Receive_Sub_Bar(price)
if self.SMA_200.Updated_Today:
self.SMA_200.Receive_Sub_Bar(price)
if self.Drawdown_5.Updated_Today:
self.Drawdown_5.Receive_Sub_Bar(self.Daily_High, self.Daily_Low)
if self.Drawdown_8.Updated_Today:
self.Drawdown_8.Receive_Sub_Bar(self.Daily_High, self.Daily_Low)
if self.Drawdown_10.Updated_Today:
self.Drawdown_10.Receive_Sub_Bar(self.Daily_High, self.Daily_Low)
def Reset(self):
self.Rsi_5.Updated_Today = False
self.Rsi_10.Updated_Today = False
self.Rsi_15.Updated_Today = False
self.Rsi_16.Updated_Today = False
self.Rsi_18.Updated_Today = False
self.Rsi_20.Updated_Today = False
self.Rsi_40.Updated_Today = False
self.Rsi_60.Updated_Today = False
self.Rsi_200.Updated_Today = False
self.Return_5.Updated_Today = False
self.Return_10.Updated_Today = False
self.Return_15.Updated_Today = False
self.Return_20.Updated_Today = False
self.Return_30.Updated_Today = False
self.Return_60.Updated_Today = False
self.Return_70.Updated_Today = False
self.Return_200.Updated_Today = False
self.SMA_23.Updated_Today = False
self.SMA_150.Updated_Today = False
self.SMA_200.Updated_Today = False
self.Drawdown_5.Updated_Today = False
self.Drawdown_8.Updated_Today = False
self.Drawdown_10.Updated_Today = False#region imports
from AlgorithmImports import *
#endregion
"""
"""
BACKTEST_START_YEAR = 2022 # Set start Year of the Backtest
BACKTEST_START_MONTH = 1 # Set start Month of the Backtest
BACKTEST_START_DAY = 3 # Set start Day of the Backtest
BACKTEST_END_YEAR = 2022 # Set end Year of the Backtest
BACKTEST_END_MONTH = 12 # Set end Month of the Backtest
BACKTEST_END_DAY = 30 # Set end Day of the Backtest
BACKTEST_ACCOUNT_CASH = 10000 # Set Backtest Strategy Cash
# List of ETF's that the algorithm will trade,
# !!! Changing these will break the code without the adequate changes in the main file
# If you need help in changing these please contact me
ETFS = ["SHY", "TECL", "SOXL", "FAS", "TQQQ", "UPRO", "TMF", "USDU", "SQQQ", "TBF", "BIL", "PHDG", "TMV", "UDOW", "MOAT",
"BRK.B", "USMV", "SPY", "QQQ", "SPXL", "DBC", "DBA", "ICSH", "TYD", "URTY", "XLU", "NUGT", "TYO", "VIXM", "IEF",
"SOXS","BND", "VTI", "TLT", "UVXY"]
REBALANCE_THRESHOLD = 5
# Cash Buffer in Percent, 0.05 = 5%
CASH_BUFFER = 0
# In minutes before market close
TRADE_DECISION_TIME = 2
USE_ORDER_DELAY = False
ORDER_DELAY = 1
# In minutes after original order
# If scheduled call is set to BeforeMarketClose keep the minus in the main file
# If scheduled call is set to AfterMarketOpen change the minus to a plus
# E.g. self.Schedule.On(self.DateRules.EveryDay(), self.TimeRules.BeforeMarketClose("SPY", self.Trade_Decision_Time - self.Order_Delay), self.Delayed_Orders)
# E.g. self.Schedule.On(self.DateRules.EveryDay(), self.TimeRules.AfterMarketOpen("SPY", self.Trade_Decision_Time + self.Order_Delay), self.Delayed_Orders)
# ORDER_DELAY = 1
# Period for the Daily RSI
RSI_5_PERIOD = 5
RSI_10_PERIOD = 10
RSI_15_PERIOD = 15
RSI_16_PERIOD = 16
RSI_18_PERIOD = 18
RSI_20_PERIOD = 20
RSI_40_PERIOD = 40
RSI_60_PERIOD = 60
RSI_200_PERIOD = 200
# Period for the Daily Cumulative Return
RETURN_5_PERIOD = 5
RETURN_10_PERIOD = 10
RETURN_15_PERIOD = 15
RETURN_20_PERIOD = 20
RETURN_30_PERIOD = 30
RETURN_60_PERIOD = 60
RETURN_70_PERIOD = 70
RETURN_200_PERIOD = 200
# Period for the Daily SMA
SMA_23_PERIOD = 23
SMA_150_PERIOD = 150
SMA_200_PERIOD = 200
CUSTOM_DRAWDOWN_5_PERIOD = 5
CUSTOM_DRAWDOWN_8_PERIOD = 8
CUSTOM_DRAWDOWN_10_PERIOD = 10
from AlgorithmImports import *
from collections import deque
class Intraday_Updating_RSI():
def __init__(self, algorithm, rsi_period):
self.algorithm = algorithm
self.rsi_period = rsi_period
self.Bar_Queue = deque(maxlen=2)
self.Average_Gain_Queue = deque(maxlen=self.rsi_period)
self.Average_Gain = None
self.Average_Loss_Queue = deque(maxlen=self.rsi_period)
self.Average_Loss = None
self.Change = None
self.Relative_Strength = None
self.Current_Value = None
self.Alpha = None
self.Sum_Gain = 0
self.Sum_Loss = 0
self.First_Intraday_Update_Of_Day = True
self.Last_Main_Bar = None
self.Last_Sum_Gain = None
self.Last_Sum_Loss = None
self.New_Main_Bar = True
self.WarmUP = True
self.Previous_Value = None
self.Updated_Today = False
def Receive_Main_Bar(self, bar):
if not self.First_Intraday_Update_Of_Day:
del self.Bar_Queue[0]
self.New_Main_Bar = True
self.Bar_Queue.appendleft(bar.Close)
if len(self.Bar_Queue) == 2:
self.Calculate_Change()
def Receive_Sub_Bar(self, bar):
if len(self.Bar_Queue) == 2:
if self.First_Intraday_Update_Of_Day:
self.Last_Main_Bar = self.Bar_Queue[1]
self.Last_Sum_Gain = self.Sum_Gain
self.Last_Sum_Loss = self.Sum_Loss
self.Bar_Queue.appendleft(bar)
self.First_Intraday_Update_Of_Day = False
else:
del self.Bar_Queue[0]
self.Bar_Queue.appendleft(bar)
self.Calculate_Change()
def Calculate_Change(self):
self.Change = self.Bar_Queue[0] - self.Bar_Queue[1]
self.Calculate_Average_Gain_Loss()
def Calculate_Average_Gain_Loss(self):
if self.Change >= 0:
gain = self.Change
else:
gain = 0
if self.Change < 0:
loss = abs(self.Change)
else:
loss = 0
self.Average_Gain_Queue.appendleft(gain)
self.Average_Loss_Queue.appendleft(loss)
if len(self.Average_Gain_Queue) == self.rsi_period and len(self.Average_Loss_Queue) == self.rsi_period:
self.Alpha = 1/self.rsi_period
self.Average_Gain = sum(self.Average_Gain_Queue) / len(self.Average_Gain_Queue)
self.Average_Loss = sum(self.Average_Loss_Queue) / len(self.Average_Loss_Queue)
if self.First_Intraday_Update_Of_Day:
if self.Sum_Gain == 0:
if self.WarmUP:
self.Sum_Gain = self.Average_Gain
else:
self.Sum_Gain = self.Last_Sum_Gain
else:
self.Sum_Gain = self.Alpha * gain + (1-self.Alpha) * self.Sum_Gain
if self.Sum_Loss == 0:
if self.WarmUP:
self.Sum_Loss = self.Average_Loss
else:
self.Sum_Loss = self.Last_Sum_Loss
else:
self.Sum_Loss = self.Alpha * loss + (1-self.Alpha) * self.Sum_Loss
else:
if self.Sum_Gain == 0:
self.Sum_Gain = self.Average_Gain
else:
self.Sum_Gain = self.Alpha * gain + (1-self.Alpha) * self.Last_Sum_Gain
if self.Sum_Loss == 0:
self.Sum_Loss = self.Average_Loss
else:
self.Sum_Loss = self.Alpha * loss + (1-self.Alpha) * self.Last_Sum_Loss
if self.New_Main_Bar:
self.First_Intraday_Update_Of_Day = True
self.Calculate_Relative_Strength()
def Calculate_Relative_Strength(self):
if self.Sum_Gain != 0 and self.Sum_Loss != 0:
self.Relative_Strength = self.Sum_Gain / self.Sum_Loss
if self.Relative_Strength is not None:
self.Current_Value = 100 - (100 / (1 + self.Relative_Strength))
self.New_Main_Bar = False