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()