| Overall Statistics |
|
Total Orders 42 Average Win 7.60% Average Loss 0% Compounding Annual Return 11.410% Drawdown 2.700% Expectancy 0 Start Equity 100000 End Equity 115471 Net Profit 15.471% Sharpe Ratio 1.069 Sortino Ratio 0.363 Probabilistic Sharpe Ratio 85.888% Loss Rate 0% Win Rate 100% Profit-Loss Ratio 0 Alpha 0 Beta 0 Annual Standard Deviation 0.05 Annual Variance 0.002 Information Ratio 1.59 Tracking Error 0.05 Treynor Ratio 0 Total Fees $29.00 Estimated Strategy Capacity $1000.00 Lowest Capacity Asset CAT Y05J8JWKMFL2|CAT R735QTJ8XC9X Portfolio Turnover 0.03% |
# region imports
from AlgorithmImports import *
# endregion
class Sellpowerhouseputs(QCAlgorithm):
def Initialize(self):
# Locally Lean installs free sample data, to download more data please visit https://www.quantconnect.com/docs/v2/lean-cli/datasets/downloading-data
self.SetStartDate(2022, 1, 1)
self.SetEndDate(2023, 5, 1)
self.SetCash(100000)
self.SMAS200 = {}
self.SMAS50 = {}
self.RSIS = {}
self.RSISrollingWindows = {}
self.SMAS200rollingWindows = {}
self.SMAS50rollingWindows = {}
self.sma_rolling_length = 10
self.options = {}
self.MACDS = {}
self.sold_puts = {}
self.sold_calls = {}
self.WeakOrStrongs = {}
self.DaysAboveSMA = {}
self.DaysBelowSMA = {}
self.LoggedToday = {}
self.activeStocks = set()
self.popular2023stocks = ["TSLA", "AAPL", "MSFT", "NVDA", "AMZN", "META", "AMD", "NFLX", "UNH", "GOOG", "BA", "PYPL", "JPM", "XON", "VZ", "BAC",
"DIS", "TMUS", "AVGO", "BRKB", "CRM", "SQ", "NKE", "TMO", "JNJ", "COST", "HD", "V", "CSCO", "PEP", "WMT", "CMCSA"
,"SHOP", "INTU", "DHR", "DVN", "MU", "TXN", "SBUX", "PFE", "P", "SCH", "AMGN", "ACN", "COUP", "PXD", "CHEX", "CAT", "SNOW",
"CVS", "USB", "GS", "WFC", "BMY", "PCLN", "LLY", "GE", "AMAT", "UBER", "LRCX", "MDT", "CNTE", "CHTR", "DH", "F", "CI", "ALB",
"HUM", "PANW", "MPC", "DE", "GILD", "IBM", "FPL", "DG", "RIVN", "CMG", "MELI", "SLB", "ATH", "VLO", "EL", "LOW", "NOW", "MOH",
"LULU", "NEM", "D", "BLK", "ETSY", "MHP", "AMT", "ADI", "ORLY", "AXP", "EOG", "GM", "ACL", "RBLX", "TEAM", "CBRNA", "UPS", "LUV",
"ISRG", "AUD", "CEY", "WDAY", "CRWD", "MRVL", "PLUG", "MS", "UTX", "UNP", "HZNP", "ABT", "FDX", "KFT", "CEG", "AZO", "GPC", "REGN",
"SEDG", "FCX", "NOC", "FANG", "PINS", "MDB", "APD", "CAH", "ETR", "DLTR", "UAUA", "ABNB", "IDPH", "YUM", "CME", "TROW", "ALD", "HPE",
"PSX", "AR", "LVS", "NXPI", "DOCU", "ABX", "CF", "PCG", "TFC", "O", "WYNN", "KLAC", "COIN", "PM", "VRTX", "FSLR", "SO", "ZTS", "MMC",
"AAL", "APA", "DHI", "ZS", "TJX", "DAL", "PSA", "CCL", "EQT", "HAL", "ATVI", "HCA", "ON", "TWRS", "BDX", "ULTA", "PNC", "MMM", "DDOG",
"MCHP", "GIS", "MAR", "CSX", "INTC", "ENPH", "OXY", "QCOM", "PX", "KO", "ORCL", "LMT", "PG", "ADBE", "BX", "MA", "C", "ABBV", "MRK", "T", "MCD", "MRNA"]
for stock in self.popular2023stocks[:60]:
x = self.AddEquity(stock, Resolution.Daily)
self.AddEquity(str(x.Symbol).split()[0], Resolution.Daily)
self.activeStocks.add(x.Symbol)
self.options[x.Symbol] = self.AddOption(str(x.Symbol).split(" ")[0])
if x.Symbol not in self.SMAS200:
self.SMAS200[x.Symbol] = self.SMA(x.Symbol, 200, Resolution.Daily)
self.SMAS50[x.Symbol] = self.SMA(x.Symbol, 50, Resolution.Daily)
self.RSIS[x.Symbol] = self.RSI(x.Symbol, 30, MovingAverageType.Simple, Resolution.Daily)
self.RSISrollingWindows[x.Symbol] = RollingWindow[float](10)
self.SMAS200rollingWindows[x.Symbol] = RollingWindow[float](self.sma_rolling_length)
self.SMAS50rollingWindows[x.Symbol] = RollingWindow[float](self.sma_rolling_length)
self.MACDS[x.Symbol] = self.macd(x.Symbol, 50, 200, 9, MovingAverageType.Exponential)
# initialize SMA and RSI
history = self.History[TradeBar](x.Symbol, 200, Resolution.Daily)
for bar in history:
self.SMAS200[x.Symbol].Update(bar.EndTime, bar.close)
self.SMAS50[x.Symbol].Update(bar.EndTime, bar.close)
self.RSIS[x.Symbol].Update(bar.EndTime, bar.close)
self.MACDS[x.Symbol].Update(bar.EndTime, bar.close)
self.RSISrollingWindows[x.Symbol].Add(self.RSIS[x.Symbol].Current.Value)
self.SMAS200rollingWindows[x.Symbol].Add(self.SMAS200[x.Symbol].Current.Value)
self.SMAS50rollingWindows[x.Symbol].Add(self.SMAS50[x.Symbol].Current.Value)
if x.Symbol not in self.WeakOrStrongs:
self.WeakOrStrongs[x.Symbol] = "Neutral"
self.DaysAboveSMA[x.Symbol] = 0
self.DaysBelowSMA[x.Symbol] = 0
#self.AddUniverse(self.CoarseFilter, self.FineFilter)
self.UniverseSettings.Resolution = Resolution.Daily
#self.rebalanceTime = self.Time
self.UniverseSettings.DataNormalizationMode = DataNormalizationMode.Raw
self.set_brokerage_model(BrokerageName.InteractiveBrokersBrokerage, AccountType.Margin)
def OnData(self, data: Slice):
self.Log("len active: " + str(len(self.activeStocks)))
for symbol in self.activeStocks:
if not data.ContainsKey(symbol) or data[symbol] is None:
self.Log("No data for symbol: " + str(symbol))
return
#self.Log("working on symbol in data: " + str(symbol) + "len active: " + str(len(actives)))
if not self.SMAS200[symbol].IsReady or not self.RSIS[symbol].IsReady:
self.Log("SMA or RSI not ready for symbol: " + str(symbol))
return
self.RSISrollingWindows[symbol].Add(self.RSIS[symbol].Current.Value)
self.SMAS200rollingWindows[symbol].Add(self.SMAS200[symbol].Current.Value)
self.SMAS50rollingWindows[symbol].Add(self.SMAS50[symbol].Current.Value)
rsi_trend = "neutral"
# check if RSI trend is upward or downward
if self.RSISrollingWindows[symbol].IsReady:
if self.RSISrollingWindows[symbol][0] > self.RSISrollingWindows[symbol][13]:
rsi_trend = "upward"
self.Log("RSI trend is upward for symbol: " + str(symbol))
elif self.RSISrollingWindows[symbol][0] < self.RSISrollingWindows[symbol][13]:
rsi_trend = "downward"
self.Log("RSI trend is downward for symbol: " + str(symbol))
else:
self.Log("RSI trend is neutral for symbol: " + str(symbol))
recent_sma_cross = None
# check if the 50 day SMA has crossed the 200 day SMA
min_50 = min(self.SMAS50rollingWindows[symbol])
max_50 = max(self.SMAS50rollingWindows[symbol])
min_200 = min(self.SMAS200rollingWindows[symbol])
max_200 = max(self.SMAS200rollingWindows[symbol])
if min_50 < min_200 and max_50 > max_200:
recent_sma_cross = True
self.Log("50 day SMA has crossed the 200 day SMA for symbol: " + str(symbol))
else:
recent_sma_cross = False
#only update on the end of day
if self.time.hour == 0 and self.time.minute == 0:
if self.SMAS200[symbol].Current.Value < data[symbol].Close and self.SMAS50[symbol].Current.Value < data[symbol].Close:
self.DaysAboveSMA[symbol] += 1
self.DaysBelowSMA[symbol] = 0
elif self.SMAS200[symbol].Current.Value > data[symbol].Close and self.SMAS50[symbol].Current.Value > data[symbol].Close:
self.DaysBelowSMA[symbol] += 1
self.DaysAboveSMA[symbol] = 0
else:
self.DaysBelowSMA[symbol] = 0
self.DaysAboveSMA[symbol] = 0
if self.WeakOrStrongs[symbol] != "Weak" and (self.RSIS[symbol].Current.Value >= 65 or self.RSIS[symbol].Current.Value <=35) and self.DaysBelowSMA[symbol] >= 3 and rsi_trend == "downward" and self.SMAS50[symbol].Current.Value < self.SMAS200[symbol].Current.Value and self.MACDS[symbol].Current.Value < -1.5 and recent_sma_cross == True:
self.Log("symbol: " + str(symbol.value) + " emits a weak signal")
self.WeakOrStrongs[symbol] = "Weak"
elif self.WeakOrStrongs[symbol] != "Strong" and self.RSIS[symbol].Current.Value >= 40 and self.RSIS[symbol].Current.Value <= 60 and self.DaysAboveSMA[symbol] >= 3 and rsi_trend == "upward" and self.SMAS50[symbol].Current.Value > self.SMAS200[symbol].Current.Value and self.MACDS[symbol].Current.Value > 1.5 and recent_sma_cross == True:
self.Log("symbol: " + str(symbol.value) + " emits a strong signal")
self.WeakOrStrongs[symbol] = "Strong"
else:
if self.time.hour == 9 and self.time.minute == 35:
self.Log("symbol: " + str(symbol.value) + "self.DaysBelowSMA: " + str(self.DaysBelowSMA[symbol]) + " self.DaysAboveSMA: " + str(self.DaysAboveSMA[symbol]) + " self.RSI: " + str(self.RSIS[symbol].Current.Value) + " self.WeakOrStrong: " + str(self.WeakOrStrongs[symbol]))
self.Log("symbol: " + str(symbol) + " self.DaysBelowSMA: " + str(self.DaysBelowSMA[symbol]) + " self.DaysAboveSMA: " + str(self.DaysAboveSMA[symbol]) + " self.RSI: " + str(self.RSIS[symbol].Current.Value) + " self.WeakOrStrong: " + str(self.WeakOrStrongs[symbol]))
self.Log(" price: " + str(data[symbol].Close) + " SMA200: " + str(self.SMAS200[symbol].Current.Value) + " SMA50: " + str(self.SMAS50[symbol].Current.Value))
if self.WeakOrStrongs[symbol] == "Weak":
# sell calls
option = self.options[symbol]
option.set_filter(min_strike=-7, max_strike=7, min_expiry=timedelta(days=30), max_expiry=timedelta(days=90))
if symbol in self.sold_calls and len(self.sold_calls[symbol]) > 0:
return
# determine the standard deviation of the underlying security
history = self.History(symbol, 30, Resolution.Daily)
if history.empty:
return
returns = history["close"].dropna()
std = returns.std()
#self.Log("Standard deviation: " + str(std))
low_end = std
high_end = 1.5*std
#strikes(low_end, high_end).
chain = data.option_chains.get(option.Symbol)
if chain:
#self.Log("Chain found for: " + str(self.symbol))
# sell puts
expiry = max([x.expiry for x in chain])
self.Log("max epiry for symbol " + str(symbol) + " is " + str(expiry))
self.Log("min expiry for symbol " + str(symbol) + " is " + str(min([x.expiry for x in chain])))
chain = sorted([x for x in chain if x.expiry == expiry], key = lambda x: x.strike)
self.Log("min and max strikes: " + str(chain[0].strike) + " " + str(chain[-1].strike))
call_contracts = [x for x in chain if x.right == OptionRight.CALL]
if len(call_contracts) == 0:
return
# sell the put with the strike closest to the high end of the range
contract = call_contracts[-1]
for i in range(20):
self.Sell(contract.Symbol, 1)
self.Log("********* Sold call: " + str(contract.Symbol) + "strike: " + str(contract.Strike) + "expiry: " + str(contract.Expiry) + "current price: " + str(data[symbol].Close))
if symbol not in self.sold_calls:
self.sold_calls[symbol] = set()
self.sold_calls[symbol].add(contract.Symbol)
elif self.WeakOrStrongs[symbol] == "Strong":
option = self.options[symbol]
option.set_filter(min_strike=-7, max_strike=7, min_expiry=timedelta(days=30), max_expiry=timedelta(days=90))
if symbol in self.sold_puts and len(self.sold_puts[symbol]) > 0:
return
#if self.RSIS[symbol].Current.Value >60:
# return
# determine the standard deviation of the underlying security
history = self.History(symbol, 30, Resolution.Daily)
if history.empty:
return
returns = history["close"].dropna()
std = returns.std()
#self.Log("Standard deviation: " + str(std))
low_end = -1.5 * std
high_end = -std
#.strikes(low_end, high_end)
chain = data.option_chains.get(option.Symbol)
if chain:
#self.Log("Chain found for: " + str(self.symbol))
# sell puts
expiry = max([x.expiry for x in chain])
chain = sorted([x for x in chain if x.expiry == expiry], key = lambda x: x.strike)
self.Log("max epiry for symbol " + str(symbol) + " is " + str(expiry))
self.Log("min expiry for symbol " + str(symbol) + " is " + str(min([x.expiry for x in chain])))
self.Log("min and max strikes: " + str(chain[0].strike) + " " + str(chain[-1].strike))
put_contracts = [x for x in chain if x.right == OptionRight.PUT]
if len(put_contracts) == 0:
return
# sell the put with the strike closest to the high end of the range
contract = put_contracts[0]
quantity = self.CalculateOrderQuantity(contract.Symbol, .25)
self.Log("symbol: " + str(symbol) + " quantity: " + str(quantity) + " contract: " + str(contract.Symbol) + " strike: " + str(contract.Strike) + " expiry: " + str(contract.Expiry) + " current price: " + str(data[symbol].Close))
for i in range(20):
self.Sell(contract.Symbol, 1)
self.Log("********* Sold put: " + str(contract.Symbol) + "strike: " + str(contract.Strike) + "expiry: " + str(contract.Expiry) + "current price: " + str(data[symbol].Close))
if symbol not in self.sold_puts:
self.sold_puts[symbol] = set()
self.sold_puts[symbol].add(contract.Symbol)
if symbol in self.sold_puts and len(self.sold_puts[symbol]) > 0:
# if profit is 80% of the premium, buy back the put
for contract in self.sold_puts[symbol]:
if self.Portfolio[contract].UnrealizedProfitPercent > 0.8:
#self.SetHoldings(contract, 0)
self.sold_puts[symbol].remove(contract)
self.Log("Bought back put: " + str(contract.Symbol))
if symbol in self.sold_calls and len(self.sold_calls[symbol]) > 0:
# if profit is 80% of the premium, buy back the call
for contract in self.sold_calls[symbol]:
if self.Portfolio[contract].UnrealizedProfitPercent > 0.8:
self.SetHoldings(contract, 0)
#self.sold_calls[symbol].remove(contract)
self.Log("Bought back call: " + str(contract.Symbol))