| Overall Statistics |
|
Total Trades 450 Average Win 0.03% Average Loss -0.01% Compounding Annual Return -17.547% Drawdown 1.100% Expectancy -0.348 Net Profit -0.930% Sharpe Ratio -5.019 Probabilistic Sharpe Ratio 11.344% Loss Rate 82% Win Rate 18% Profit-Loss Ratio 2.58 Alpha -0.102 Beta 0.088 Annual Standard Deviation 0.026 Annual Variance 0.001 Information Ratio 1.064 Tracking Error 0.166 Treynor Ratio -1.458 Total Fees $450.00 Estimated Strategy Capacity $9000.00 Lowest Capacity Asset BP XW4SISGA7ZOM|BP R735QTJ8XC9X |
#region imports
from AlgorithmImports import *
#endregion
class RetrospectiveBlueSeahorse(QCAlgorithm):
def Initialize(self):
# # # # # User input area # # # # #
# Set start date
self.SetStartDate(2022, 1, 13)
# Strategy cash
self.SetCash(100000)
# User input list of stocks
list_of_stocks = ["AMZN","AAPL","TSLA","BABA","AMD"]
# # # # # End user input area # # # # #
# Rolling RSI dictionary
self.rolling_rsi_dictionary = {}
# MACD dictionary
self.macd_dictionary = {}
# RSI dictionary
self.rsi_dictionary = {}
# SMA 14-period dictionary
self.sma_forteen_dictionary = {}
# SMA 26-period dictionary
self.sma_twentysix_dictionary = {}
# SMA 30-period dictionary
self.sma_thirty_dictionary = {}
# Yesterday close dictionary
self.yesterday_close = {}
# Symbol list
symbol_list = []
# Loop through stocks
for ticker in list_of_stocks:
# Initialize stock
symbol = self.AddEquity(ticker, Resolution.Minute).Symbol
# Add to symbol list
symbol_list.append(symbol)
# Minute history call
history = self.History(symbol_list, 100, Resolution.Minute)
# Daily history call
daily_history = self.History(symbol_list, 1, Resolution.Daily)
# Loop through added symbols
for symbol in symbol_list:
# Initialize security in rolling RSI dictionary
self.rolling_rsi_dictionary[symbol] = []
# MACD dictionary
self.macd_dictionary[symbol] = self.MACD(symbol, 12, 26, 9, MovingAverageType.Exponential, Resolution.Minute)
# RSI dictionary
self.rsi_dictionary[symbol] = self.RSI(symbol, 14, MovingAverageType.Simple, Resolution.Minute)
# SMA 14-period dictionary
self.sma_forteen_dictionary[symbol] = self.SMA(symbol, 14, Resolution.Minute)
# SMA 26-period dictionary
self.sma_twentysix_dictionary[symbol] = self.SMA(symbol, 26, Resolution.Minute)
# SMA 30-period dictionary
self.sma_thirty_dictionary[symbol] = self.SMA(symbol, 30, Resolution.Minute)
# # Yesterday close dictionary
self.yesterday_close[symbol] = daily_history.loc[symbol]["close"][-1]
# Loc history data for asset
asset_history = history.loc[symbol]
# Loop through history of asset
for time, row in asset_history.iterrows():
# Update macd indicator of asset
self.macd_dictionary[symbol].Update(time, row.close)
# Update rsi indicator for asset
self.rsi_dictionary[symbol].Update(time, row.close)
# Update SMA 14-period for asset
self.sma_forteen_dictionary[symbol].Update(time, row.close)
# Update SMA 26-period for asset
self.sma_twentysix_dictionary[symbol].Update(time, row.close)
# Update SMA 26-period for asset
self.sma_thirty_dictionary[symbol].Update(time, row.close)
# Add SPY to schedule events
self.AddEquity("SPY", Resolution.Minute)
# Schedule saving of market closing price
self.Schedule.On(self.DateRules.EveryDay("SPY"),
self.TimeRules.BeforeMarketClose("SPY", 0),
self.save_market_closing)
# Save market closing price function
def save_market_closing(self):
# Loop through all assets
for symbol in self.yesterday_close:
# Update close price
self.yesterday_close[symbol] = self.Securities[symbol].Close
def OnData(self, data):
# # # # # Long position logic # # # # #
# Loop through all assets
for symbol in self.rolling_rsi_dictionary:
# Update rolling RSI
self.rolling_rsi_dictionary[symbol].append(self.rsi_dictionary[symbol].Current.Value)
# Check if length of rolling RSI dictionary is greater than 60
if len(self.rolling_rsi_dictionary[symbol]) > 60:
# Cut length
self.rolling_rsi_dictionary[symbol] = self.rolling_rsi_dictionary[symbol][-60:]
# Check if MACD is greater than MACD signal
if self.macd_dictionary[symbol].Current.Value > self.macd_dictionary[symbol].Signal.Current.Value:
# Check if RSI is greater than its value 1 hour ago
if self.rsi_dictionary[symbol].Current.Value > self.rolling_rsi_dictionary[symbol][-60]:
# Check if 14-period SMA is greater than 30-period SMA
if self.sma_forteen_dictionary[symbol].Current.Value > self.sma_thirty_dictionary[symbol].Current.Value:
# Check if current close greater than yesterday close
if self.Securities[symbol].Close > self.yesterday_close[symbol]:
# If not invested
if not self.Portfolio[symbol].Invested:
# Get cash available
cash = self.Portfolio.Cash
# Get 25% of cash available
cash_allocation = cash * 0.25
# Calculate quantity
quantity = int(cash_allocation / self.Securities[symbol].Close)
# Submit market order
self.MarketOrder(symbol, quantity)
# # # # # Liquidation logic # # # # #
# Get holdings
holdings = [x.Key for x in self.Portfolio if x.Value.Invested]
# Loop through holdings
for symbol in holdings:
# If current MACD < MACD signal
if self.macd_dictionary[symbol].Current.Value < self.macd_dictionary[symbol].Signal.Current.Value:
# If RSI is less than its value 1 hour ago
if self.rsi_dictionary[symbol].Current.Value < self.rolling_rsi_dictionary[symbol][-1]:
# If 26-period SMA less than most recent close price
if self.sma_twentysix_dictionary[symbol].Current.Value < self.Securities[symbol].Close:
# Close position with market order
self.MarketOrder(symbol, -self.Portfolio[symbol].Quantity)#region imports
from AlgorithmImports import *
#endregion
class RetrospectiveBlueSeahorse(QCAlgorithm):
def Initialize(self):
# # # # # User input area # # # # #
# Set start date
self.SetStartDate(2022, 1, 13)
# Strategy cash
self.SetCash(25000)
# User input list of stocks
# # # Tickers need to be in capital letters # # #
list_of_stocks = ["AMZN","AAPL","TSLA","BABA","AMD","BP","F","GME","AMC"]
# # # # # End user input area # # # # #
# Rolling RSI dictionary
self.rolling_rsi_dictionary = {}
# MACD dictionary
self.macd_dictionary = {}
# RSI 6-minute dictionary
self.rsi_six_minute_dictionary = {}
# RSI 14-minute dictionary
self.rsi_forteen_minute_dictionary = {}
# SMA 14-period dictionary
self.sma_forteen_dictionary = {}
# SMA 26-period dictionary
self.sma_twentysix_dictionary = {}
# SMA 30-period dictionary
self.sma_thirty_dictionary = {}
# Yesterday close dictionary
self.yesterday_close = {}
# Symbol list
symbol_list = []
# Loop through stocks
for ticker in list_of_stocks:
# Initialize stock
# # # QuantConnect supports tick, second, minute, hour and daily resolution # # #
symbol = self.AddEquity(ticker, Resolution.Minute, "USA", True, 0, False).Symbol
# Add to symbol list
symbol_list.append(symbol)
# Minute history call
# # # The first parameter has to be a list # # #
history = self.History(symbol_list, 100, Resolution.Minute)
# Daily history call
daily_history = self.History(symbol_list, 1, Resolution.Daily)
# Loop through added symbols
for symbol in symbol_list:
# Initialize security in rolling RSI dictionary
self.rolling_rsi_dictionary[symbol] = []
# MACD dictionary
self.macd_dictionary[symbol] = self.MACD(symbol, 12, 26, 9, MovingAverageType.Exponential, Resolution.Minute, Field.Open)
# 6-minute RSI dictionary
self.rsi_six_minute_dictionary[symbol] = self.RSI(symbol, 6, MovingAverageType.Simple, Resolution.Minute)
# 14-minute RSI dictionary
self.rsi_forteen_minute_dictionary[symbol] = self.RSI(symbol, 14, MovingAverageType.Simple, Resolution.Minute)
# SMA 14-period dictionary
self.sma_forteen_dictionary[symbol] = self.SMA(symbol, 14, Resolution.Minute)
# SMA 26-period dictionary
self.sma_twentysix_dictionary[symbol] = self.SMA(symbol, 26, Resolution.Minute)
# SMA 30-period dictionary
self.sma_thirty_dictionary[symbol] = self.SMA(symbol, 30, Resolution.Minute)
# # Yesterday close dictionary
self.yesterday_close[symbol] = daily_history.loc[symbol]["close"][-1]
# Loc history data for asset
asset_history = history.loc[symbol]
# Loop through history of asset
for time, row in asset_history.iterrows():
# Update macd indicator of asset
self.macd_dictionary[symbol].Update(time, row.open)
# Update 6-minute rsi indicator for asset
self.rsi_six_minute_dictionary[symbol][symbol].Update(time, row.close)
# Update 14-minute rsi indicator for asset
self.rsi_forteen_minute_dictionary[symbol][symbol].Update(time, row.close)
# Update SMA 14-period for asset
self.sma_forteen_dictionary[symbol].Update(time, row.close)
# Update SMA 26-period for asset
self.sma_twentysix_dictionary[symbol].Update(time, row.close)
# Update SMA 26-period for asset
self.sma_thirty_dictionary[symbol].Update(time, row.close)
# Add SPY to schedule events
self.AddEquity("SPY", Resolution.Minute)
# Schedule saving of market closing price
self.Schedule.On(self.DateRules.EveryDay("SPY"),
self.TimeRules.BeforeMarketClose("SPY", 0),
self.save_market_closing)
# Save market closing price function
def save_market_closing(self):
# Loop through all assets
for symbol in self.yesterday_close:
# Update close price
self.yesterday_close[symbol] = self.Securities[symbol].Close
def OnData(self, data):
# # # # # Long position logic # # # # #
# Loop through all assets
for symbol in self.macd_dictionary:
# Check if MACD is greater than MACD signal
if self.macd_dictionary[symbol].Current.Value > self.macd_dictionary[symbol].Signal.Current.Value:
# Check if 14-period SMA is greater than 30-period SMA
if self.sma_forteen_dictionary[symbol].Current.Value > self.sma_thirty_dictionary[symbol].Current.Value:
# Check if current close greater than yesterday close
if self.Securities[symbol].Close > self.yesterday_close[symbol]:
# If not invested
if not self.Portfolio[symbol].Invested:
# Get cash available
cash = self.Portfolio.Cash
# Get 25% of cash available
cash_allocation = cash * 0.25
# Calculate quantity
quantity = int(cash_allocation / self.Securities[symbol].Close)
# If quantity is greater than 1
if quantity > 1:
# Send SMS to Yomi's phone number
self.Notify.Sms("+1234567890", "Buy signal for " + symbol.Value + " triggered")
# Submit market order
self.MarketOrder(symbol, quantity)
# # # # # Liquidation logic # # # # #
# Get holdings
holdings = [x.Key for x in self.Portfolio if x.Value.Invested]
# Loop through holdings
for symbol in holdings:
# If current MACD < MACD signal
if self.macd_dictionary[symbol].Current.Value < self.macd_dictionary[symbol].Signal.Current.Value:
# If 26-period SMA less than most recent close price
if self.sma_twentysix_dictionary[symbol].Current.Value < self.Securities[symbol].Close:
# Send SMS to Yomi's phone number
self.Notify.Sms("+1234567890", "Sell signal for " + symbol.Value + " triggered")
# Close position with market order
self.MarketOrder(symbol, -self.Portfolio[symbol].Quantity)#region imports
from AlgorithmImports import *
#endregion
class RetrospectiveBlueSeahorse(QCAlgorithm):
def Initialize(self):
# # # # # User input area # # # # #
# Set start date
self.SetStartDate(2022, 1, 15)
self.SetEndDate(2022, 2, 1)
# Strategy cash
self.SetCash(100000)
# User input list of stocks
list_of_stocks = ["BP","AAPL"]
# # # # # End user input area # # # # #
# Rolling RSI dictionary
self.rolling_rsi_dictionary = {}
# MACD dictionary
self.macd_dictionary = {}
# RSI dictionary
self.rsi_dictionary = {}
# SMA 14-period dictionary
self.sma_forteen_dictionary = {}
# SMA 26-period dictionary
self.sma_twentysix_dictionary = {}
# SMA 30-period dictionary
self.sma_thirty_dictionary = {}
# Yesterday close dictionary
self.yesterday_close = {}
# Symbols dictionary
self.symbols_dictionary = {}
# Invested dict
self.invested_dict = {}
# Loop through stocks
for ticker in list_of_stocks:
# Initialize stock
symbol = self.AddEquity(ticker, Resolution.Minute).Symbol
# Initialize security in rolling RSI dictionary
self.rolling_rsi_dictionary[symbol] = []
# MACD dictionary
self.macd_dictionary[symbol] = self.MACD(symbol, 12, 26, 9, MovingAverageType.Exponential, Resolution.Minute)
# RSI dictionary
self.rsi_dictionary[symbol] = self.RSI(symbol, 14, MovingAverageType.Simple, Resolution.Minute)
# SMA 14-period dictionary
self.sma_forteen_dictionary[symbol] = self.SMA(symbol, 14, Resolution.Minute)
# SMA 26-period dictionary
self.sma_twentysix_dictionary[symbol] = self.SMA(symbol, 26, Resolution.Minute)
# SMA 30-period dictionary
self.sma_thirty_dictionary[symbol] = self.SMA(symbol, 30, Resolution.Minute)
# Yesterday close dictionary
self.yesterday_close[symbol] = 0
# Add option
option = self.AddOption(ticker, Resolution.Minute)
# set our strike/expiry filter for this option chain
option.SetFilter(-10, +10, timedelta(5), timedelta(90))
# Store option symbol
self.symbols_dictionary[symbol] = option.Symbol
# Add SPY to schedule events
self.AddEquity("SPY", Resolution.Minute)
# Schedule saving of market closing price
self.Schedule.On(self.DateRules.EveryDay("SPY"),
self.TimeRules.BeforeMarketClose("SPY", 0),
self.save_market_closing)
# Set warmup
self.SetWarmup(timedelta(days = 10))
# Save market closing price function
def save_market_closing(self):
# Loop through all assets
for symbol in self.yesterday_close:
# Update close price
self.yesterday_close[symbol] = self.Securities[symbol].Close
def OnData(self, data):
# # # # # Long position logic # # # # #
# Loop through all assets
for symbol in self.rolling_rsi_dictionary:
# Update rolling RSI
self.rolling_rsi_dictionary[symbol].append(self.rsi_dictionary[symbol].Current.Value)
# Check if length of rolling RSI dictionary is greater than 60
if len(self.rolling_rsi_dictionary[symbol]) > 600:
# Cut length
self.rolling_rsi_dictionary[symbol] = self.rolling_rsi_dictionary[symbol][-600:]
# If not warming up
if not self.IsWarmingUp:
# Loop through all assets
for symbol in self.rolling_rsi_dictionary:
# Check if MACD is greater than MACD signal
if self.macd_dictionary[symbol].Current.Value > self.macd_dictionary[symbol].Signal.Current.Value:
# Check if RSI is greater than its value 1 hour ago
if self.rsi_dictionary[symbol].Current.Value > self.rolling_rsi_dictionary[symbol][0]:
# Check if 14-period SMA is greater than 30-period SMA
if self.sma_forteen_dictionary[symbol].Current.Value > self.sma_thirty_dictionary[symbol].Current.Value:
# Check if current close greater than yesterday close
if self.Securities[symbol].Close > self.yesterday_close[symbol]:
# If not invested
if not self.Portfolio[symbol].Invested:
for i in data.OptionChains:
if i.Key == self.symbols_dictionary[symbol]:
chain = i.Value
call = [x for x in chain if x.Right == OptionRight.Call]
put = [x for x in chain if x.Right == OptionRight.Put]
for i in call:
for j in put:
if i.Expiry == j.Expiry and i.Strike == j.Strike:
if not self.Portfolio[symbol].Invested:
if i.BidPrice + i.Strike > i.UnderlyingLastPrice + j.AskPrice:
self.MarketOrder(i.UnderlyingSymbol, 10) # long the underlying stocks
self.MarketOrder(j.Symbol, 1) # long put
self.MarketOrder(i.Symbol, -1) # short call
self.invested_dict[symbol] = [j.Symbol, i.Symbol]
elif i.AskPrice + i.Strike < i.UnderlyingLastPrice + j.BidPrice:
#quantity = self.CalculateOrderQuantity(self.euqity_symbol,0.6)
self.MarketOrder(i.UnderlyingSymbol, -10) # short the underlying stocks
self.MarketOrder(j.Symbol, -1) # short put
self.MarketOrder(i.Symbol, 1) # long call
self.invested_dict[symbol] = [j.Symbol, i.Symbol]
# # # # # Liquidation logic # # # # #
# Loop through holdings
for symbol in self.macd_dictionary:
# If symbol in invested dict
if symbol in self.invested_dict:
# If current MACD < MACD signal
if self.macd_dictionary[symbol].Current.Value < self.macd_dictionary[symbol].Signal.Current.Value:
# If RSI is less than its value 1 hour ago
if self.rsi_dictionary[symbol].Current.Value < self.rolling_rsi_dictionary[symbol][-60]:
# If 26-period SMA less than most recent close price
if self.sma_twentysix_dictionary[symbol].Current.Value < self.Securities[symbol].Close:
# Liquidate
self.Liquidate(symbol)
self.Liquidate(self.invested_dict[symbol][0])
self.Liquidate(self.invested_dict[symbol][1])
# Remove from invested dict
self.invested_dict.pop(symbol)
#region imports
from AlgorithmImports import *
#endregion
from datetime import timedelta
from math import floor,ceil
class BasicTemplateOptionsAlgorithm(QCAlgorithm):
def Initialize(self):
self.SetStartDate(2022, 1, 1)
#self.SetEndDate(2022, 9, 1)
self.SetCash(1000000)
equity = self.AddEquity("BP", Resolution.Minute)
option = self.AddOption("BP", Resolution.Minute)
self.symbol = option.Symbol
# set our strike/expiry filter for this option chain
option.SetFilter(-10, +10, timedelta(5), timedelta(90))
# use the underlying equity as the benchmark
self.SetBenchmark(equity.Symbol)
def OnData(self,slice):
if self.Portfolio.Invested: return
for i in slice.OptionChains:
if i.Key != self.symbol: continue
chain = i.Value
call = [x for x in chain if x.Right == OptionRight.Call]
put = [x for x in chain if x.Right == OptionRight.Put]
for i in call:
for j in put:
if i.Expiry == j.Expiry and i.Strike == j.Strike:
if self.Portfolio.Invested: return
if i.BidPrice + i.Strike > i.UnderlyingLastPrice + j.AskPrice:
self.MarketOrder(i.UnderlyingSymbol, 100) # long the underlying stocks
self.MarketOrder(j.Symbol, 1) # long put
self.MarketOrder(i.Symbol, -1) # short call
elif i.AskPrice + i.Strike < i.UnderlyingLastPrice + j.BidPrice:
#quantity = self.CalculateOrderQuantity(self.euqity_symbol,0.6)
self.MarketOrder(i.UnderlyingSymbol, -100) # short the underlying stocks
self.MarketOrder(j.Symbol, -1) # short put
self.MarketOrder(i.Symbol, 1) # long call
def OnOrderEvent(self, orderEvent):
self.Log(str(orderEvent))