| Overall Statistics |
|
Total Trades 60 Average Win 0.20% Average Loss -0.10% Compounding Annual Return 7.308% Drawdown 1.100% Expectancy 0.471 Net Profit 1.479% Sharpe Ratio 2.177 Probabilistic Sharpe Ratio 77.591% Loss Rate 50% Win Rate 50% Profit-Loss Ratio 1.94 Alpha 0.046 Beta 0.048 Annual Standard Deviation 0.023 Annual Variance 0.001 Information Ratio -0.305 Tracking Error 0.137 Treynor Ratio 1.059 Total Fees $60.00 Estimated Strategy Capacity $660000.00 Lowest Capacity Asset TAN V44Y4YYPI2XX Portfolio Turnover 19.59% |
#region imports
from AlgorithmImports import *
#endregion
class NadionResistanceShield(QCAlgorithm):
#class DataConsolidationAlgorithm(QCAlgorithm):
def Initialize(self):
self.SetStartDate(2023, 1, 1) # Set Start Date
#self.SetEndDate(2023, 1, 1)
self.SetCash(50000) # Set Strategy Cash
self.tickers = ["SPY", "NUE", "MSFT", "AMZN", "GOOG", "NVDA", "TAN"]
self.symbolDataBySymbol = {}
self.trade = True
self.vwaps = {}
self.stds = {}
# Before the open
for symbol in self.tickers:
self.AddEquity(symbol, Resolution.Minute)
self.Debug(f"Registering {symbol}")
'''For the below 3 EMA's, you can convert them to 4H bars using the colidator method'''
macd = self.MACD(symbol, 12, 26, 9, MovingAverageType.Exponential, Resolution.Daily, Field.Close)
ema10 = self.EMA(symbol, 10, Resolution.Hour, Field.Close)
sma200 = self.SMA(symbol, 200, Resolution.Daily, Field.Close)
sma20 = self.SMA(symbol, 20, Resolution.Daily, Field.Close)
sma50 = self.SMA(symbol, 50, Resolution.Daily, Field.Close)
sma7 = self.SMA(symbol, 7, Resolution.Hour, Field.Close)
#sma7 = self.SMA(symbol, 7, Resolution.Minute, Field.Close)
ema20 = self.EMA(symbol, 20, Resolution.Hour, Field.Close)
ema7 = self.EMA(symbol, 7, Resolution.Hour, Field.Close)
#ema7 = self.EMA(symbol, 7, Resolution.Minute, Field.Close)
ema9 = self.EMA(symbol, 9, Resolution.Hour, Field.Close)
rsi = self.RSI(symbol, 14, Resolution.Daily)
wilr_up = self.WILR(symbol, 10, Resolution.Daily)
wilr = self.WILR(symbol, 14, Resolution.Daily)
self.vwaps[symbol] = self.VWAP(symbol, 14, Resolution.Daily)
self.stds[symbol] = self.STD(symbol, 14, Resolution.Daily)
'''Consolidator method'''
smaConsolidate = ExponentialMovingAverage(20, MovingAverageType.Simple)
# create the 4 hour data consolidator
fourHourConsolidator = TradeBarConsolidator(timedelta(hours=4))
self.SubscriptionManager.AddConsolidator(symbol, fourHourConsolidator)
# register the 4 hour consolidated bar data to automatically update the indicator
self.RegisterIndicator(symbol, smaConsolidate, fourHourConsolidator)
symbolData = SymbolData(symbol, ema10, sma200, sma20, sma50, sma7, ema20, ema7, ema9, rsi, wilr, smaConsolidate)
self.symbolDataBySymbol[symbol] = symbolData
self.SetWarmUp(30, Resolution.Daily)
self.spy = self.AddEquity("SPY", Resolution.Daily)
self.Schedule.On(self.DateRules.EveryDay("SPY"),
self.TimeRules.AfterMarketOpen("SPY"),
self.tradeStart)
self.Schedule.On(self.DateRules.EveryDay("SPY"),
self.TimeRules.BeforeMarketClose("SPY"),
self.tradeEnd)
self.Schedule.On(self.DateRules.EveryDay(), self.TimeRules.At(10, 0), self.Trade)
#self.Schedule.On(self.DateRules.EveryDay("SPY"), self.AfterMarketOpen(10, 0), self.SellSignals)
#self.Schedule.On(self.DateRules.EveryDay("SPY"),self.TimeRules.Every(timedelta(minutes=10)),self.sellSignals)
self.Schedule.On(self.DateRules.EveryDay(), self.TimeRules.At(12, 0), self.Exit)
#self.AddRiskManagement(TrailingStopRiskManagementModel(0.03))
self.SetWarmUp(timedelta(days=50))
def tradeStart(self):
self.trade = True
def tradeEnd(self):
self.trade = False
def Trade(self):
if self.trade == False:
return
for symbol in self.tickers:
# Get the daily data for the symbol
history = self.History([symbol], 1, Resolution.Daily)
if not history.empty:
daily_data = history.loc[symbol]
# Get the open and close prices for the previous day
previous_open = daily_data["open"].iloc[0]
previous_close = daily_data["close"].iloc[0]
# Get the current price
current_price = self.Securities[symbol].Price
# Calculate the price gap and percentage change
price_gap = current_price - previous_close
percentage_change = price_gap / previous_close
# Check if the price has gapped up by more than 1%
if percentage_change > 0.02:
self.Log("{} gap up detected: {}%".format(symbol, percentage_change * 100))
# Enter a long position
self.SetHoldings(symbol, .25)
def buySignals(self):
if self.trade == False:
return
for symbol, symbolData in self.symbolDataBySymbol.items():
if not self.Portfolio[symbol].Invested and (self.Securities[symbol].Close > symbolData.ema7.Current.Value) and (symbolData.sma20.Current.Value > symbolData.sma50.Current.Value):
self.SetHoldings(symbol, .1, False, "Buy Signal")
# def sellSignals(self):
# if self.trade == False:
# return
# for symbol, symbolData in self.symbolDataBySymbol.items():
# if self.Portfolio[symbol].Invested and (self.Securities[symbol].Close < symbolData.ema7.Current.Value):
# self.Liquidate(symbol, "Sell Signal")
def Exit(self):
for symbol in self.tickers:
# Exit the position
self.Liquidate(symbol)
def OnData(self, data):
if self.IsWarmingUp:# or not all(std.IsReady for std in self.stds.values()):
return
for symbol in self.tickers:
price = self.Securities[symbol].Price
vwap = self.vwaps[symbol].Current.Value
std = self.stds[symbol].Current.Value
ub1 = vwap + std
ub2 = vwap + 2*std
lb1 = vwap - std
lb2 = vwap - 2*std
self.Plot(symbol, 'price', price)
self.Plot(symbol, 'vwap', vwap)
self.Plot(symbol, 'ub2', ub2)
self.Plot(symbol, 'ub1', ub1)
self.Plot(symbol, 'lb1', lb1)
self.Plot(symbol, 'lb2', lb2)
#Take Profit
if price > ub2:
self.Liquidate(symbol)
return
class SymbolData:
def __init__(self, symbol, ema10, sma200, sma20, sma50, sma7, ema20, ema7, ema9, rsi, wilr, smaConsolidate):
self.Symbol = symbol
self.ema10 = ema10
self.sma200 = sma200
self.sma20 = sma20
self.sma50 = sma50
self.sma7 = sma7
self.ema20 = ema20
self.ema7 = ema7
self.ema9 = ema9
self.rsi = rsi
self.wilr = wilr
#self.emaConsolidate = emaConsolidate
self.smaConsolidate = smaConsolidate
#region imports
from AlgorithmImports import *
#endregion
class NadionResistanceShield(QCAlgorithm):
#class DataConsolidationAlgorithm(QCAlgorithm):
def Initialize(self):
self.SetStartDate(2023, 1, 1) # Set Start Date
#self.SetEndDate(2023, 1, 1)
self.SetCash(50000) # Set Strategy Cash
self.tickers = ["SPY", "NUE", "MSFT", "AMZN", "GOOG", "NVDA", "TAN"]
self.symbolDataBySymbol = {}
self.trade = True
self.vwaps = {}
self.stds = {}
# Before the open
for symbol in self.tickers:
self.AddEquity(symbol, Resolution.Minute)
self.Debug(f"Registering {symbol}")
'''For the below 3 EMA's, you can convert them to 4H bars using the colidator method'''
macd = self.MACD(symbol, 12, 26, 9, MovingAverageType.Exponential, Resolution.Daily, Field.Close)
ema10 = self.EMA(symbol, 10, Resolution.Hour, Field.Close)
sma200 = self.SMA(symbol, 200, Resolution.Daily, Field.Close)
sma20 = self.SMA(symbol, 20, Resolution.Daily, Field.Close)
sma50 = self.SMA(symbol, 50, Resolution.Daily, Field.Close)
sma7 = self.SMA(symbol, 7, Resolution.Hour, Field.Close)
#sma7 = self.SMA(symbol, 7, Resolution.Minute, Field.Close)
ema20 = self.EMA(symbol, 20, Resolution.Hour, Field.Close)
ema7 = self.EMA(symbol, 7, Resolution.Hour, Field.Close)
#ema7 = self.EMA(symbol, 7, Resolution.Minute, Field.Close)
ema9 = self.EMA(symbol, 9, Resolution.Hour, Field.Close)
rsi = self.RSI(symbol, 14, Resolution.Daily)
wilr_up = self.WILR(symbol, 10, Resolution.Daily)
wilr = self.WILR(symbol, 14, Resolution.Daily)
self.vwaps[symbol] = self.VWAP(symbol, 14, Resolution.Daily)
self.stds[symbol] = self.STD(symbol, 14, Resolution.Daily)
'''Consolidator method'''
smaConsolidate = ExponentialMovingAverage(20, MovingAverageType.Simple)
# create the 4 hour data consolidator
fourHourConsolidator = TradeBarConsolidator(timedelta(hours=4))
self.SubscriptionManager.AddConsolidator(symbol, fourHourConsolidator)
# register the 4 hour consolidated bar data to automatically update the indicator
self.RegisterIndicator(symbol, smaConsolidate, fourHourConsolidator)
symbolData = SymbolData(symbol, ema10, sma200, sma20, sma50, sma7, ema20, ema7, ema9, rsi, wilr, smaConsolidate)
self.symbolDataBySymbol[symbol] = symbolData
self.SetWarmUp(30, Resolution.Daily)
self.spy = self.AddEquity("SPY", Resolution.Daily)
self.Schedule.On(self.DateRules.EveryDay("SPY"),
self.TimeRules.AfterMarketOpen("SPY"),
self.tradeStart)
self.Schedule.On(self.DateRules.EveryDay("SPY"),
self.TimeRules.BeforeMarketClose("SPY"),
self.tradeEnd)
self.Schedule.On(self.DateRules.EveryDay(), self.TimeRules.At(10, 0), self.Trade)
#self.Schedule.On(self.DateRules.EveryDay("SPY"), self.AfterMarketOpen(10, 0), self.SellSignals)
self.Schedule.On(self.DateRules.EveryDay("SPY"),self.TimeRules.Every(timedelta(minutes=10)),self.sellSignals)
self.Schedule.On(self.DateRules.EveryDay(), self.TimeRules.At(12, 0), self.Exit)
#self.AddRiskManagement(TrailingStopRiskManagementModel(0.03))
self.SetWarmUp(timedelta(days=50))
def tradeStart(self):
self.trade = True
def tradeEnd(self):
self.trade = False
def Trade(self):
if self.trade == False:
return
for symbol in self.tickers:
# Get the daily data for the symbol
history = self.History([symbol], 1, Resolution.Daily)
if not history.empty:
daily_data = history.loc[symbol]
# Get the open and close prices for the previous day
previous_open = daily_data["open"].iloc[0]
previous_close = daily_data["close"].iloc[0]
# Get the current price
current_price = self.Securities[symbol].Price
# Calculate the price gap and percentage change
price_gap = current_price - previous_close
percentage_change = price_gap / previous_close
# Check if the price has gapped up by more than 1%
if percentage_change > 0.02:
self.Log("{} gap up detected: {}%".format(symbol, percentage_change * 100))
# Enter a long position
self.SetHoldings(symbol, .25)
def buySignals(self):
if self.trade == False:
return
for symbol, symbolData in self.symbolDataBySymbol.items():
if not self.Portfolio[symbol].Invested and (self.Securities[symbol].Close > symbolData.ema7.Current.Value) and (symbolData.sma20.Current.Value > symbolData.sma50.Current.Value):
self.SetHoldings(symbol, .1, False, "Buy Signal")
# def sellSignals(self):
# if self.trade == False:
# return
# for symbol, symbolData in self.symbolDataBySymbol.items():
# if self.Portfolio[symbol].Invested and (self.Securities[symbol].Close < symbolData.ema7.Current.Value):
# self.Liquidate(symbol, "Sell Signal")
def sellSignals(self):
if self.trade == False:
return
invested_symbols = [x.Key for x in self.Portfolio if x.Value.Invested]
for symbol in invested_symbols:
if self.vwaps[symbol].Current.Value < self.Securities[symbol].Price and self.Securities[symbol].Price > (2 * self.stds[symbol].Current.Value):
self.Log("{} sell signal detected.".format(symbol))
self.Liquidate(symbol)
else:
self.Debug("{} has a VWAP of {} and current price is {}.".format(symbol, self.vwaps[symbol].Current.Value, self.Securities[symbol].Price))
def Exit(self):
for symbol in self.tickers:
# Exit the position
self.Liquidate(symbol)
def OnData(self, data):
if self.IsWarmingUp or not all(std.IsReady for std in self.stds.values()):
return
for symbol in self.tickers:
price = self.Securities[symbol].Price
vwap = self.vwaps[symbol].Current.Value
std = self.stds[symbol].Current.Value
ub1 = vwap + std
ub2 = vwap + 2*std
lb1 = vwap - std
lb2 = vwap - 2*std
self.Plot(symbol, 'price', price)
self.Plot(symbol, 'vwap', vwap)
self.Plot(symbol, 'ub2', ub2)
self.Plot(symbol, 'ub1', ub1)
self.Plot(symbol, 'lb1', lb1)
self.Plot(symbol, 'lb2', lb2)
class SymbolData:
def __init__(self, symbol, ema10, sma200, sma20, sma50, sma7, ema20, ema7, ema9, rsi, wilr, smaConsolidate):
self.Symbol = symbol
self.ema10 = ema10
self.sma200 = sma200
self.sma20 = sma20
self.sma50 = sma50
self.sma7 = sma7
self.ema20 = ema20
self.ema7 = ema7
self.ema9 = ema9
self.rsi = rsi
self.wilr = wilr
#self.emaConsolidate = emaConsolidate
self.smaConsolidate = smaConsolidate