| Overall Statistics |
|
Total Trades 191 Average Win 0.18% Average Loss -0.30% Compounding Annual Return 6.128% Drawdown 4.000% Expectancy -0.200 Net Profit 0.834% Sharpe Ratio 0.398 Probabilistic Sharpe Ratio 40.247% Loss Rate 51% Win Rate 49% Profit-Loss Ratio 0.62 Alpha 0.08 Beta -0.091 Annual Standard Deviation 0.158 Annual Variance 0.025 Information Ratio -0.705 Tracking Error 0.186 Treynor Ratio -0.694 Total Fees $196.41 Estimated Strategy Capacity $28000000.00 Lowest Capacity Asset ZY XNTGMD4HZ0X1 |
import pandas as pd
class mean_variance(QCAlgorithm):
def Initialize(self):
#User input area:
self.ema_fast_span = 10 #Fast_ema_period
self.ema_slow_span = 50 #Slow ema period
self.amount = 0.05 #Maximum percent of portfolio to invest long/short
self.portfolio_long_limit = 10 #Maximum number of stocks to long
self.portfolio_short_limit = 10 #Maximum number of stocks to short
self.manual_list = ['AAPL', 'ABBV', 'ABT', 'ACN', 'ADBE', 'AIG', 'AMGN', 'AMT', 'AMZN', 'AVGO',
'AXP', 'BA', 'BAC', 'BIIB', 'BK', 'BKNG', 'BLK', 'BMY', 'BRK.B', 'C', 'CAT',
'CHTR', 'CL', 'CMCSA', 'COF', 'COP', 'COST', 'CRM', 'CSCO', 'CVS', 'CVX', 'DD',
'DHR', 'DIS', 'DOW', 'DUK', 'EMR', 'EXC', 'F', 'FB', 'FDX', 'GD', 'GE', 'GILD',
'GM', 'GOOG', 'GOOGL', 'GS', 'HD', 'HON', 'IBM', 'INTC', 'JNJ', 'JPM', 'KHC', 'KO',
'LIN', 'LLY', 'LMT', 'LOW', 'MA', 'MCD', 'MDLZ', 'MDT', 'MET', 'MMM', 'MO', 'MRK', 'MS',
'MSFT', 'NEE', 'NFLX', 'NKE', 'NVDA', 'ORCL', 'PEP', 'PFE', 'PG', 'PM', 'PYPL', 'QCOM', 'RTX',
'SBUX', 'SO', 'SPG', 'T', 'TGT', 'TMO', 'TMUS', 'TSLA', 'TXN', 'UNH', 'UNP', 'UPS', 'USB', 'V',
'VZ', 'WBA', 'WFC', 'WMT', 'XOM']
#QC setup
self.SetStartDate(2021,7,1)
self.SetCash(50000)
self.AddUniverse(self.Coarse, self.Fine)
self.UniverseSettings.Resolution = Resolution.Daily
#Variables
self.symbols = []
self.long_list = []
self.short_list = []
self.holdings = {}
self.ema_fast = {}
self.ema_slow = {}
self.prices = {}
self.long_limit = False
self.short_limit = False
def Coarse(self, coarse):
filtered = [x for x in coarse if x.HasFundamentalData
and x.DollarVolume > 1000000
or x.Symbol.Value in self.manual_list]
sortedStocks = sorted(filtered, key = lambda x: x.DollarVolume, reverse = True)
return [x.Symbol for x in sortedStocks][:50] #When debugging, I only used 50 stocks in universe to speed up backtesting
def Fine(self, fine):
return [x.Symbol for x in fine
if x.CompanyReference.PrimaryExchangeID == "NAS"
and x.CompanyReference.CountryId == "USA"]
def OnSecuritiesChanged(self, changes):
for stock in changes.RemovedSecurities:
symbol = stock.Symbol
self.Liquidate(symbol)
if symbol in self.symbols:
self.symbols.remove(symbol)
self.ema_fast.pop(symbol, None)
self.ema_slow.pop(symbol, None)
self.history = self.History([stock.Symbol for stock in changes.AddedSecurities], 100, Resolution.Daily)
for stock in changes.AddedSecurities:
symbol = stock.Symbol
if symbol in self.history.index:
if symbol not in self.symbols:
close = self.history.loc[symbol]["close"].to_list()
self.symbols.append(symbol)
self.prices[symbol] = close
close_prices = {}
close_prices["close"] = close
ema = pd.DataFrame(close_prices , columns = ["close"])
ema["EMA_fast"] = ema["close"].ewm(span=self.ema_fast_span,min_periods=0,adjust=False,ignore_na=False).mean()
ema_fast = ema["EMA_fast"].to_list()
self.ema_fast[symbol] = ema_fast
ema["EMA_slow"] = ema["close"].ewm(span=self.ema_slow_span,min_periods=0,adjust=False,ignore_na=False).mean()
ema_slow = ema["EMA_slow"].to_list()
self.ema_slow[symbol] = ema_slow
for i in self.symbols:
if self.ema_fast[i][-1] > self.ema_slow[i][-1]:
self.long_list.append(i)
elif self.ema_slow[i][-1] > self.ema_fast[i][-1]:
self.short_list.append(i)
else:
0
for i in self.long_list:
self.Debug(str(i) + " long symbol")
for i in self.short_list:
self.Debug(str(i) + " short symbol")
def OnData(self,data):
self.UpdateData(data)
self.CheckLiquidate()
for i in self.symbols:
if i not in self.holdings:
if self.ema_fast[i][-1] > self.ema_slow[i][-1] and self.long_limit == False:
self.SetHoldings(i, self.amount)
if self.ema_slow[i][-1] > self.ema_fast[i][-1] and self.short_limit == False:
self.SetHoldings(i, -self.amount)
def UpdateData(self, data):
self.data = data
self.tradebars = data.Bars
for symbol in self.symbols:
if not self.data.ContainsKey(symbol):
continue
if not self.data.Bars.ContainsKey(symbol):
continue
self.prices[symbol].append(self.tradebars[symbol].Close)
close = self.prices[symbol]
close_prices = {}
close_prices["close"] = close
ema = pd.DataFrame(close_prices , columns = ["close"])
ema["EMA_fast"] = ema["close"].ewm(span=self.ema_fast_span,min_periods=0,adjust=False,ignore_na=False).mean()
ema_fast = ema["EMA_fast"].to_list()
self.ema_fast[symbol] = ema_fast
ema["EMA_slow"] = ema["close"].ewm(span=self.ema_slow_span,min_periods=0,adjust=False,ignore_na=False).mean()
ema_slow = ema["EMA_slow"].to_list()
self.ema_slow[symbol] = ema_slow
def CheckLiquidate(self):
self.holdings = {}
items_to_pop = []
for kvp in self.Portfolio:
security_holding = kvp.Value
if security_holding.Invested:
symbol = security_holding.Symbol
quantity = security_holding.Quantity
self.holdings[symbol] = quantity
if len(self.holdings) > 0:
for i in self.holdings:
if self.holdings[i] > 0:
if i in self.ema_slow and self.ema_slow[i][-1] > self.ema_fast[i][-1]:
self.Liquidate(i)
items_to_pop.append(i)
else:
0
if self.holdings[i] < 0:
if i in self.ema_slow and self.ema_fast[i][-1] > self.ema_slow[i][-1]:
self.Liquidate(i)
items_to_pop.append(i)
else:
0
for i in items_to_pop:
self.holdings.pop(i)
short_limit = 0
long_limit = 0
for i in self.holdings:
if self.holdings[i] < 0:
short_limit+= 1
elif self.holdings[i] > 0:
long_limit += 1
else:
0
if short_limit >= self.portfolio_short_limit:
self.short_limit = True
else:
self.short_limit = False
if long_limit >= self.portfolio_long_limit:
self.long_limit = True
else:
self.long_limit = False# import numpy as np
# import pandas as pd
# from scipy.optimize import minimize
# import statsmodels.formula.api as sm
# class mean_variance(QCAlgorithm):
# def __init__(self):
# def Coarse(self, coarse):
# filtered = [x for x in coarse if x.HasFundamentalData
# and x.DollarVolume > 1000000
# or x.Symbol.Value in self.manual_list]
# sortedStocks = sorted(filtered, key = lambda x: x.DollarVolume, reverse = True)
# return [x.Symbol for x in sortedStocks][:50] #When debugging, I only used 50 stocks in universe to speed up backtesting
# def Fine(self, fine):
# return [x.Symbol for x in fine
# if x.CompanyReference.PrimaryExchangeID == "NAS"
# and x.CompanyReference.CountryId == "USA"]
# def OnSecuritiesChanged(self, changes):
# for stock in changes.RemovedSecurities:
# symbol = stock.Symbol
# self.Liquidate(symbol)
# if symbol in self.symbols:
# self.symbols.remove(symbol)
# self.ema_fast.pop(symbol, None)
# self.ema_slow.pop(symbol, None)
# self.history = self.History([stock.Symbol for stock in changes.AddedSecurities], 100, Resolution.Daily)
# for stock in changes.AddedSecurities:
# symbol = stock.Symbol
# if symbol in self.history.index:
# if symbol not in self.symbols:
# close = self.history.loc[symbol]["close"].to_list()
# self.symbols.append(symbol)
# self.prices[symbol] = close
# close_prices = {}
# close_prices["close"] = close
# ema = pd.DataFrame(close_prices , columns = ["close"])
# ema["EMA_fast"] = ema["close"].ewm(span=self.ema_fast_span,min_periods=0,adjust=False,ignore_na=False).mean()
# ema_fast = ema["EMA_fast"].to_list()
# self.ema_fast[symbol] = ema_fast
# ema["EMA_slow"] = ema["close"].ewm(span=self.ema_slow_span,min_periods=0,adjust=False,ignore_na=False).mean()
# ema_slow = ema["EMA_slow"].to_list()
# self.ema_slow[symbol] = ema_slow
# '''
# self.symbols = ["SPY","MMM", "AXP", "AAPL", "BA", "CAT", "CVX", "CSCO","KO",
# "DIS","DD","XOM","GE","GS","HD","IBM","INTC","JPM","MCD",
# "MRK","MSFT","NKE","PFE","PG","TRV","UTX","UNH","VZ","V","WMT"]
# '''
# self.slow_symbols = ema_slow
# self.slow_symbols = ema_fast
# self.num = 21*12
# self.reb_feq = 21
# self.count = 0
# def get_history(self,symbol):
# prices = []
# dates = []
# for i in self.history:
# bar = i[symbol]
# prices.append(np.log(float(bar.Close)))
# dates.append(bar.EndTime)
# symbol.df = pd.DataFrame({'log_price':prices},index = dates)
# symbol.df['log_return'] = symbol.df['log_price'].diff()
# symbol.df = symbol.df.dropna()
# def regression(self):
# for i in self.symbols:
# df = pd.DataFrame({'%s'%str(i):i.df['log_return'], 'SPY':self.spy.df['log_return']})
# i.model = sm.ols(formula = '%s ~ SPY'%str(i), data = df).fit()
# i.intercept = i.model.params[0]
# i.beta = i.model.params[1]
# i.one_month = sum(i.df['log_return'].tail(21))
# def Initialize(self):
# self.SetStartDate(2014,1,1)
# self.SetEndDate(2017,1,1)
# self.SetCash(50000)
# self.long_list = []
# self.short_list = []
# for i in range(len(self.symbols)):
# equity = self.AddEquity(self.symbols[i],Resolution.Daily).Symbol
# self.symbols[i] = equity
# self.history = self.History(self.num, Resolution.Daily)
# for i in self.symbols:
# self.get_history(i)
# i.leng = i.df.shape[0]
# i.mean = np.mean(i.df['log_return'])
# i.std = np.std(i.df['log_return'])
# i.price_list = []
# i.dates_list = []
# self.spy = self.symbols[0]
# self.regression()
# def OnData(self,data):
# # if not self.Securities[self.symbols[0]].Exchange.ExchangeOpen:
# # return
# if self.count == 0:
# # calculate alpha#
# for i in self.symbols:
# i.alpha = i.one_month - i.intercept - i.beta*self.spy.one_month
# #Getting long list and short list
# self.long_list = []
# self.short_list = []
# for i in self.symbols:
# if self.ema_fast[i][-1] > self.ema_slow[i][-1]:
# self.long_list.append(i)
# elif self.ema_slow[i][-1] > self.ema_fast[i][-1]:
# self.short_list.append(i)
# else:
# 0
# #############
# self.long_list = [x for x in self.symbols]
# #The following lines are CAPM part, and we don't use them for this strategy.
# self.long_list = [x for x in self.symbols if x.alpha < 0]
# self.long_list.sort(key = lambda x: x.alpha)
# self.long_list = self.long_list[:10]
# self.short_list = [x for x in self.symbols if x.alpha > 0]
# self.short_list.sort(key = lambda x: x.alpha, reverse = True)
# self.short_list = self.short_list[:10]
# #portfolio optimization#
# self.ticker_list = [str(x) for x in self.long_list]
# self.mean_list = [x.mean for x in self.long_list]
# self.cov_matrix = np.cov([x.df['log_return'] for x in self.long_list])
# self.port = optimizer(self.ticker_list,self.mean_list,self.cov_matrix)
# self.port.optimize()
# self.Log(str(self.port.opt_df))
# self.Log(str([str(x) for x in self.long_list]))
# # self.Log(str([str(x) for x in self.short_list]))
# for i in self.long_list:
# self.SetHoldings(i,self.port.opt_df.ix[str(i)])
# for i in self.short_list:
# self.SetHoldings(i,-1/len(self.short_list))
# self.count += 1
# return
# if self.count < self.reb_feq:
# for i in self.symbols:
# try:
# i.price_list.append(np.log(float(data[i].Close)))
# i.dates_list.append(data[i].EndTime)
# except:
# self.Log(str(i))
# self.count += 1
# return
# if self.count == self.reb_feq:
# for i in self.symbols:
# try:
# i.price_list.append(np.log(float(data[i].Close)))
# i.dates_list.append(data[i].EndTime)
# df = pd.DataFrame({'log_price':i.price_list},index = i.dates_list)
# df = df.diff().dropna()
# df = pd.concat([i.df,df]).tail(self.num)
# except:
# pass
# self.regression()
# self.Liquidate()
# self.count = 0
# return
# class optimizer(object):
# def __init__(self,ticker_list, mean_list,cov_matrix):
# self.tickers = ticker_list
# self.mean_list = mean_list
# self.cov_matrix = cov_matrix
# def optimize(self):
# leng = len(self.tickers)
# def target(x, sigma, mean):
# sr_inv = (np.sqrt(np.dot(np.dot(x.T,sigma),x)*252))/((x.dot(mean))*252)
# return sr_inv
# x = np.ones(leng)/leng
# mean = self.mean_list
# sigma = self.cov_matrix
# c = ({'type':'eq','fun':lambda x: sum(x) - 1},
# {'type':'ineq','fun':lambda x: 2 - sum([abs(i) for i in x])})
# bounds = [(-1,1) for i in range(leng)]
# res = minimize(target, x, args = (sigma,mean),method = 'SLSQP',constraints = c,bounds = bounds)
# self.opt_weight = res.x
# self.exp_return = np.dot(self.mean_list,res.x)*252
# self.std = np.sqrt(np.dot(np.dot(res.x.T,sigma),res.x)*252)
# self.opt_df = pd.DataFrame({'weight':res.x},index = self.tickers)
# self.opt_df.index = self.opt_df.index.map(str)
# def update(self,ticker_list, mean_list,cov_matrix):
# self.tickers = ticker_list
# self.mean_list = mean_list
# self.cov_matrix = cov_matrix
# self.optimize()