| Overall Statistics |
|
Total Trades 222 Average Win 0.95% Average Loss -0.74% Compounding Annual Return 147.530% Drawdown 12.400% Expectancy 0.578 Net Profit 53.902% Sharpe Ratio 3.491 Probabilistic Sharpe Ratio 90.392% Loss Rate 31% Win Rate 69% Profit-Loss Ratio 1.28 Alpha 0.624 Beta 1.438 Annual Standard Deviation 0.27 Annual Variance 0.073 Information Ratio 3.391 Tracking Error 0.213 Treynor Ratio 0.655 Total Fees $1074.00 Estimated Strategy Capacity $630000.00 Lowest Capacity Asset UVXY V0H08FY38ZFP Portfolio Turnover 30.14% |
from AlgorithmImports import *
import math
import pandas as pd
from cmath import sqrt
from clr import AddReference
AddReference("System")
AddReference("QuantConnect.Algorithm")
AddReference("QuantConnect.Common")
from System import *
from QuantConnect import *
from QuantConnect.Algorithm import *
from QuantConnect.Data.Custom import *
from QuantConnect.Python import PythonData
class IntelligentSkyRodent(QCAlgorithm):
def Initialize(self):
self.cash = 100000
self.buffer_pct = 0.03
self.SetStartDate(2023, 1, 1)
self.SetEndDate(2023, 6, 23)
self.SetCash(self.cash)
self.equities = ['TMV', 'VCIT', 'XLY', 'HIBL', 'XLK', 'XLP', 'SVXY', 'QID', 'TBF', 'TSLA', 'LQD', 'VTIP', 'EDV', 'STIP', 'SPTL', 'IEI', 'USDU', 'SQQQ', 'VIXM', 'SPXU', 'QQQ', 'BSV', 'TQQQ', 'SPY', 'DBC', 'SHV', 'IAU', 'VEA', 'UTSL', 'UVXY', 'UPRO', 'EFA', 'EEM', 'TLT', 'SHY', 'GLD', 'SLV', 'USO', 'WEAT', 'CORN', 'SH', 'DRN', 'PDBC', 'COMT', 'KOLD', 'BOIL', 'ESPO', 'PEJ', 'UGL', 'URE', 'VXX', 'UUP', 'BND', 'DUST', 'JDST', 'JNUG', 'GUSH', 'DBA', 'DBB', 'COM', 'PALL', 'AGQ', 'BAL', 'WOOD', 'URA', 'SCO', 'UCO', 'DBO', 'TAGS', 'CANE', 'REMX', 'COPX', 'IEF', 'SPDN', 'CHAD', 'DRIP', 'SPUU', 'INDL', 'BRZU', 'ERX', 'ERY', 'CWEB', 'CHAU', 'KORU', 'MEXX', 'EDZ', 'EURL', 'YINN', 'YANG', 'TNA', 'TZA', 'SPXL', 'SPXS', 'MIDU', 'TYD', 'TYO', 'TMF', 'TECL', 'TECS', 'SOXL', 'SOXS', 'LABU', 'LABD', 'RETL', 'DPST', 'DRV', 'PILL', 'CURE', 'FAZ', 'FAS', 'EWA', 'EWGS', 'EWG', 'EWP', 'EWQ', 'EWU', 'EWJ', 'EWI', 'EWN', 'ECC', 'NURE', 'VNQI', 'VNQ', 'VDC', 'VIS', 'VGT', 'VAW', 'VPU', 'VOX', 'VFH', 'VHT', 'VDE', 'SMH', 'DIA', 'UDOW', 'PSQ', 'SOXX', 'VTI', 'COST', 'UNH', 'SPHB', 'BTAL', 'VIXY', 'WEBL', 'WEBS', 'UBT', 'PST', 'TLH', 'QLD', 'SQM', 'SSO', 'SD', 'DGRO', 'SCHD', 'SGOL', 'TIP', 'DUG', 'EWZ', 'TBX', 'VGI', 'XLU', 'XLV', 'EUO', 'YCS', 'MVV', 'USD', 'BIL', 'TMF', 'EPI', 'IYK', 'DIG', 'AGG', 'PUI', 'UDN', 'QQQE', 'VTV', 'VOOG', 'VOOV']
self.MKT = self.AddEquity("SPY",Resolution.Daily).Symbol
self.mkt = []
for equity in self.equities:
self.AddEquity(equity,Resolution.Minute)
self.Securities[equity].SetDataNormalizationMode(DataNormalizationMode.Adjusted)
self.AddEquity('BIL',Resolution.Minute)
self.Securities['BIL'].SetDataNormalizationMode(DataNormalizationMode.TotalReturn)
self.PT1 = 0.27
self.PT2 = 0.67
self.HT1 = {str(i).zfill(2): 0 for i in range(1,10)}
self.HTS1 = {str(i).zfill(2): [] for i in range(1,10)}
self.HT2 = {str(i).zfill(2): 0 for i in range(1,10)}
self.HTS2 = {str(i).zfill(2): [] for i in range(1,10)}
self.Schedule.On(self.DateRules.EveryDay("SPY"),
self.TimeRules.BeforeMarketClose("SPY", 2),
self.FunctionBeforeMarketClose)
def RSI(self,equity,period):
extension = min(period*5,250)
r_w = RollingWindow[float](extension)
history = self.History(equity,extension - 1,Resolution.Daily)
for historical_bar in history:
r_w.Add(historical_bar.Close)
while r_w.Count < extension:
current_price = self.Securities[equity].Price
r_w.Add(current_price)
if r_w.IsReady:
average_gain = 0
average_loss = 0
gain = 0
loss = 0
for i in range(extension - 1,extension - period -1,-1):
gain += max(r_w[i-1] - r_w[i],0)
loss += abs(min(r_w[i-1] - r_w[i],0))
average_gain = gain/period
average_loss = loss/period
for i in range(extension - period - 1,0,-1):
average_gain = (average_gain*(period-1) + max(r_w[i-1] - r_w[i],0))/period
average_loss = (average_loss*(period-1) + abs(min(r_w[i-1] - r_w[i],0)))/period
if average_loss == 0:
return 100
else:
rsi = 100 - (100/(1 + average_gain / average_loss))
return rsi
else:
return None
def CumReturn(self,equity,period):
history = self.History(equity,period,Resolution.Daily)
closing_prices = pd.Series([bar.Close for bar in history])
current_price = self.Securities[equity].Price
closing_prices = closing_prices.append(pd.Series([current_price]))
first_price = closing_prices.iloc[0]
if first_price == 0:
return None
else:
return_val = (current_price / first_price) - 1
return return_val
def STD(self,equity,period):
r_w = RollingWindow[float](period + 1)
r_w_return = RollingWindow[float](period)
history = self.History(equity,period,Resolution.Daily)
for historical_bar in history:
r_w.Add(historical_bar.Close)
while r_w.Count < period + 1:
current_price = self.Securities[equity].Price
r_w.Add(current_price)
for i in range (period,0,-1):
daily_return = (r_w[i-1]/r_w[i] - 1)
r_w_return.Add(daily_return)
dfstd = pd.DataFrame({'r_w_return':r_w_return})
if r_w.IsReady:
std = dfstd['r_w_return'].std()
if std == 0:
return 0
else:
return std
else:
return 0
def MaxDD(self,equity,period):
history = self.History(equity,period - 1,Resolution.Daily)
closing_prices = pd.Series([bar.Close for bar in history])
current_price = self.Securities[equity].Price
closing_prices = closing_prices.append(pd.Series([current_price]))
rolling_max = closing_prices.cummax()
drawdowns = (rolling_max - closing_prices) / rolling_max
max_dd = drawdowns.min()
return max_dd
def SMA(self,equity,period):
r_w = RollingWindow[float](period)
history = self.History(equity,period - 1,Resolution.Daily)
for historical_bar in history:
r_w.Add(historical_bar.Close)
while r_w.Count < period:
current_price = self.Securities[equity].Price
r_w.Add(current_price)
if r_w.IsReady:
sma = sum(r_w) / period
return sma
else:
return 0
def IV(self,equity,period):
r_w = RollingWindow[float](period + 1)
r_w_return = RollingWindow[float](period)
history = self.History(equity,period,Resolution.Daily)
for historical_bar in history:
r_w.Add(historical_bar.Close)
while r_w.Count < period + 1:
current_price = self.Securities[equity].Price
r_w.Add(current_price)
for i in range (period,0,-1):
if r_w[i] == 0:
return 0
else:
daily_return = (r_w[i-1]/r_w[i] - 1)
r_w_return.Add(daily_return)
dfinverse = pd.DataFrame({'r_w_return':r_w_return})
if r_w.IsReady:
std = dfinverse['r_w_return'].std()
if std == 0:
return 0
else:
inv_vol = 1 / std
return inv_vol
else:
return 0
def SMADayRet(self,equity,period):
r_w = RollingWindow[float](period + 1)
r_w_return = RollingWindow[float](period)
history = self.History(equity,period,Resolution.Daily)
for historical_bar in history:
r_w.Add(historical_bar.Close)
while r_w.Count < period + 1:
current_price = self.Securities[equity].Price
r_w.Add(current_price)
for i in range (period,0,-1):
if r_w[i] == 0:
return None
daily_return = (r_w[i-1]/r_w[i] - 1)
r_w_return.Add(daily_return)
if r_w.IsReady:
smareturn = sum(r_w_return) / period
return smareturn
else:
return 0
def EMA(self,equity,period):
extension = period + 50
r_w = RollingWindow[float](extension)
history = self.History(equity,extension - 1,Resolution.Daily)
for historical_bar in history:
r_w.Add(historical_bar.Close)
while r_w.Count < extension:
current_price = self.Securities[equity].Price
r_w.Add(current_price)
if r_w.IsReady:
total_price = 0
for i in range(extension - 1,extension - period - 2,-1):
total_price += r_w[i]
average_price = total_price/period
for i in range(extension - period - 2,-1,-1):
average_price = r_w[i]*2/(period+1) + average_price*(1-2/(period+1))
return average_price
else:
return None
def Sort(self,sort_type,equities,period,reverse,number,multiplier):
self.PT = getattr(self,f"PT{number}") * multiplier
returns = {}
for equity in equities:
returns[equity] = getattr(self,sort_type)(equity,period)
s_e = sorted([item for item in returns.items() if item[1] is not None],key = lambda x: x[1],reverse = reverse)
t3e = s_e[:1]
ht = getattr(self,f"HT{number}")
hts = getattr(self,f"HTS{number}")
for i in ht.keys():
if ht[i] == 0:
ht[i] = self.PT
hts[i].append(t3e[0][0])
break
setattr(self,f"HT{number}",ht)
setattr(self,f"HTS{number}",hts)
def AH(self, equities, PTnumber, multiplier): #AppendHolding
if not isinstance(equities, list):
equities = [equities]
HT = getattr(self, f"HT{PTnumber}")
HTS = getattr(self, f"HTS{PTnumber}")
PT = getattr(self, f"PT{PTnumber}") * multiplier
for equity in equities:
for i in HT.keys():
if HT[i] == 0:
HT[i] = PT
HTS[i].append(equity)
break
def OnData (self,data):
pass
def FunctionBeforeMarketClose(self):
mkt_price = self.History(self.MKT,2,Resolution.Daily)['close'].unstack(level= 0).iloc[-1]
self.mkt.append(mkt_price)
mkt_perf = self.cash * self.mkt[-1] / self.mkt[0]
self.Plot('Strategy Equity',self.MKT,mkt_perf)
self.JamesDoge()
self.HolyGrailSimplified()
self.v41Pops()
self.ExecuteTrade()
def JamesDoge(self):
if self.Securities['SPY'].Price > self.SMA('SPY', 200):
if self.RSI('TQQQ', 14) > 75:
self.AH('UVXY', 1, 1)
else:
if self.RSI('SPXL', 10) > 80:
self.AH('UVXY', 1, 1)
else:
self.BuyTheDipsNasdaq()
else:
if self.CumReturn('QQQ', 60) <= -0.2:
self.SidewaysMarket()
else:
if self.RSI('TQQQ', 9) < 32:
if self.CumReturn('TQQQ', 2) >= self.CumReturn('TQQQ', 5):
self.SubStrategy1()
self.SubStrategy2()
else:
if self.RSI('SPY', 10) < 30:
self.Sort("RSI", ["SPXL", "SHY"], 10, False, 1, 1)
else:
if self.RSI('UVXY', 10) > 74:
if self.RSI('UVXY', 10) > 84:
self.Sort("RSI", ["BSV", "SOXS"], 10, True, 1, 1)
else:
self.AH('UVXY', 1, 1)
else:
if self.Securities['TQQQ'].Price > self.SMA('TQQQ', 20):
if self.RSI('SQQQ', 10) < 31:
self.AH('SQQQ', 1, 1)
else:
self.AH('SOXL', 1, 1)
else:
self.Sort("RSI", ["BSV", "SOXS"], 10, True, 1, 1)
else:
if self.RSI('SPY', 10) < 30:
self.Sort("RSI", ["SPXL", "SHY"], 10, False, 1, 1)
else:
if self.RSI('UVXY', 10) > 74:
if self.RSI('UVXY', 10) > 84:
self.Sort("RSI", ["BSV", "SQQQ"], 10, True, 1, 1)
else:
self.AH('UVXY', 1, 1)
else:
if self.Securities['TQQQ'].Price > self.SMA('TQQQ', 20):
if self.RSI('SQQQ', 10) < 31:
self.AH('SOXS', 1, 1)
else:
self.AH('SOXL', 1, 1)
else:
self.Sort("RSI", ["BSV", "SQQQ"], 10, True, 1, 1)
def BuyTheDipsNasdaq(self):
if self.CumReturn('QQQ', 5) < -0.06:
if self.CumReturn('TQQQ', 1) > 0.05:
self.AH('SQQQ', 1, 1)
else:
if self.RSI('TQQQ', 10) > 31:
self.AH('SQQQ', 1, 1)
else:
self.AH('SOXL', 1, 1)
else:
if self.RSI('QQQ', 10) > 80:
self.AH('SQQQ', 1, 1)
else:
if self.RSI('QQQ', 10) < 31:
self.AH('SOXL', 1, 1)
else:
self.AH('TQQQ', 1, 1)
def SidewaysMarket(self):
if self.CumReturn('UUP', 2) >= 0.01:
if self.CumReturn('TLT', 1) <= 0:
self.AH('SOXL', 1, 1)
else:
self.AH('SOXS', 1, 1)
else:
if self.Securities['SPY'].Price >= self.SMA('SPY', 3):
self.AH('TQQQ', 1, 1)
else:
self.AH('BSV', 1, 1)
def SubStrategy1(self):
self.Sort("RSI", ["TECL", "SOXL", "SHY"], 10, False, 1, 0.5)
def SubStrategy2(self):
self.Sort("RSI", ["SOXL", "SHY"], 5, False, 1, 0.5)
def HolyGrailSimplified(self):
if self.Securities['SPY'].Price > self.SMA('SPY', 200):
if self.RSI('QQQ', 10) > 79:
self.Sort("RSI", ["UVXY", "SQQQ"], 13, True, 2, 0.74)
elif self.RSI('SPY', 10) > 79:
self.Sort("RSI", ["UVXY", "SPXS"], 13, True, 2, 0.74)
else:
self.Sort("RSI", ["TECL", "STIP"], 11, False, 2, 0.74)
elif self.RSI('TQQQ', 10) < 31:
self.Sort("RSI", ["TECL", "SHY"], 11, False, 2, 0.74)
elif self.RSI('UPRO', 10) < 31:
self.Sort("RSI", ["UPRO", "SHY"], 11, False, 2, 0.74)
elif self.CumReturn('TQQQ', 6) < -0.11:
self.BuythedipsV2()
else:
if self.Securities['QLD'].Price > self.SMA('QLD', 20):
self.Sort("RSI", ["TECL", "STIP"], 11, False, 2, 0.74)
else:
self.Substrategy5()
self.Substrategy6()
def BuythedipsV2(self):
if self.CumReturn('TQQQ', 1) > 0.055:
self.Sort("RSI", ["UVXY", "SQQQ"], 11, True, 2, 0.74)
elif self.CumReturn('SQQQ', 1) > 0.028:
self.Substrategy1()
self.Substrategy2()
else:
self.HoldStocksBondsSOXLSHVTECLSTIP()
def Substrategy1(self):
self.Sort("RSI", ["TECL", "UDN"], 11, False, 2, 0.37)
def Substrategy2(self):
self.Sort("RSI", ["TECL", "TMV"], 11, True, 2, 0.37)
def HoldStocksBondsSOXLSHVTECLSTIP(self):
self.Substrategy3()
self.Substrategy4()
def Substrategy3(self):
self.Sort("RSI", ["SOXL", "SHV"], 11, False, 2, 0.5032)
def Substrategy4(self):
self.Sort("RSI", ["TECL", "STIP"], 11, False, 2, 0.2368)
def Substrategy5(self):
if self.SMADayRet('TLT', 20) > self.SMADayRet('UDN', 20):
self.Sort("RSI", ["TLT", "SQQQ"], 11, True, 2, 0.37)
else:
self.Sort("RSI", ["UUP", "SQQQ"], 10, False, 2, 0.37)
def Substrategy6(self):
self.Sort("RSI", ["UGL", "SQQQ"], 12, False, 2, 0.37)
def v41Pops(self):
if self.RSI('QQQE', 10) > 79:
self.Sort("RSI", ["UVXY", "VIXY"], 13, False, 2, 0.24)
elif self.RSI('VTV', 10) > 79:
self.Sort("RSI", ["UVXY", "VIXY"], 13, False, 2, 0.24)
elif self.RSI('VOX', 10) > 79:
self.Sort("RSI", ["UVXY", "VIXY"], 13, False, 2, 0.24)
elif self.MaxDD('SPY', 9) < 0:
self.Sort("RSI", ["UVXY", "VIXY"], 13, False, 2, 0.24)
elif self.RSI('TECL', 10) > 79:
self.Sort("RSI", ["UVXY", "VIXY"], 13, False, 2, 0.24)
elif self.RSI('VOOG', 10) > 79:
self.Sort("RSI", ["UVXY", "VIXY"], 13, False, 2, 0.24)
elif self.RSI('VOOV', 10) > 79:
self.Sort("RSI", ["UVXY", "VIXY"], 13, False, 2, 0.24)
elif self.RSI('XLP', 10) > 75:
self.Sort("RSI", ["UVXY", "VIXY"], 13, False, 2, 0.24)
elif self.RSI('TQQQ', 10) > 79:
self.Sort("RSI", ["UVXY", "VIXY"], 13, False, 2, 0.24)
elif self.RSI('XLY', 10) > 80:
self.Sort("RSI", ["UVXY", "VIXY"], 13, False, 2, 0.24)
elif self.RSI('FAS', 10) > 80:
self.Sort("RSI", ["UVXY", "VIXY"], 13, False, 2, 0.24)
elif self.RSI('SPY', 10) > 80:
self.Sort("RSI", ["UVXY", "VIXY"], 13, False, 2, 0.24)
elif self.CumReturn('TQQQ', 6) < -0.12:
if self.CumReturn('QQQ', 1) > 0.018:
self.Sort("RSI", ["UVXY", "VIXY"], 13, False, 2, 0.24)
else:
self.V15B()
else:
self.V15B()
def V15B(self):
if self.RSI('SPY', 6) > 75:
self.Sort("RSI", ["UVXY", "VIXY"], 13, True, 2, 0.24)
else:
if self.RSI('BIL', 5) < self.RSI('VCIT', 5):
self.AH('SOXL', 2, 0.24)
else:
if self.RSI('SVXY', 11) > 60:
self.Sort("RSI", ["TMF", "BIL"], 10, True, 2, 0.24)
else:
if self.SMADayRet('VIXM', 10) > self.SMADayRet('SVXY', 5):
self.Sort("MaxDD", ["UVXY", "VIXY", "IEF"], 3, False, 2, 0.24)
else:
if self.SMADayRet('EDV', 11) > 0:
self.Sort("RSI", ["SVXY", "TMF"], 21, True, 2, 0.24)
else:
self.Sort("RSI", ["SVXY", "TMV"], 21, True, 2, 0.24)
def ExecuteTrade(self):
group1 = {
'HTS': [self.HTS1[i][0] if len(self.HTS1[i]) == 1 else self.HTS1[i] for i in self.HTS1],
'HT': [self.HT1[i] for i in self.HT1]
}
df1 = pd.DataFrame(group1)
group2 = {
'HTS': [self.HTS2[i][0] if len(self.HTS2[i]) == 1 else self.HTS2[i] for i in self.HTS2],
'HT': [self.HT2[i] for i in self.HT2]
}
df2 = pd.DataFrame(group2)
df = pd.concat([df1, df2])
df['HTS'] = df['HTS'].astype(str)
result = df.groupby(['HTS']).sum().reset_index()
for equity in self.equities:
if all(not pd.isnull(result.iloc[i,0]) and not equity == result.iloc[i,0] for i in range(len(result))):
if self.Portfolio[equity].HoldStock:
self.Liquidate(equity)
output = "*****"
for i in range(len(result)):
if result.iloc[i,0]:
percentage = round(result.iloc[i,1] * 100,2)
output += "{}: {}% - ".format(result.iloc[i,0],percentage)
output = output.rstrip(" - ")
self.Log(output)
for i in range(len(result)):
if not result.iloc[i,1] == 0 and not result.iloc[i,0] == 'BIL':
percentage_equity = self.Portfolio[result.iloc[i,0]].HoldingsValue / self.Portfolio.TotalPortfolioValue
if result.iloc[i,1] < percentage_equity and abs(result.iloc[i,1] / percentage_equity - 1) > self.buffer_pct:
self.SetHoldings(result.iloc[i,0],result.iloc[i,1])
else:
pass
for i in range(len(result)):
if not result.iloc[i,1] == 0 and not result.iloc[i,0] == 'BIL':
percentage_equity = self.Portfolio[result.iloc[i,0]].HoldingsValue / self.Portfolio.TotalPortfolioValue
if result.iloc[i,1] > percentage_equity and abs(percentage_equity / result.iloc[i,1] - 1) > self.buffer_pct:
self.SetHoldings(result.iloc[i,0],result.iloc[i,1])
else:
pass
self.HT1 = {str(i).zfill(2): 0 for i in range(1,10)}
self.HTS1 = {str(i).zfill(2): [] for i in range(1,10)}
self.HT2 = {str(i).zfill(2): 0 for i in range(1,10)}
self.HTS2 = {str(i).zfill(2): [] for i in range(1,10)}