Overall Statistics |
Total Trades 124 Average Win 0.61% Average Loss -0.62% Compounding Annual Return -5.385% Drawdown 19.700% Expectancy -0.291 Net Profit -8.697% Sharpe Ratio -0.572 Probabilistic Sharpe Ratio 1.706% Loss Rate 64% Win Rate 36% Profit-Loss Ratio 0.99 Alpha 0 Beta 0 Annual Standard Deviation 0.085 Annual Variance 0.007 Information Ratio -0.572 Tracking Error 0.085 Treynor Ratio 0 Total Fees $129.71 |
import numpy as np import time class FFFiveFactorAdapted(QCAlgorithm): ''' Stocks Selecting Strategy based on Fama French 5 Factors Model Reference: https://tevgeniou.github.io/EquityRiskFactors/bibliography/FiveFactor.pdf ''' def Initialize(self): self.SetStartDate(2018, 1, 1) # Set Start Date self.SetEndDate(2019, 8, 24) # Set End Date self.SetCash(100000) # Set Strategy Cash self.Log("here41") self.UniverseSettings.Resolution = Resolution.Daily self.AddUniverse(self.CoarseSelectionFunction, self.FineSelectionFunction) self.Log("here51") # self.num_coarse = 10000 # Number of symbols selected at Coarse Selection self.fraction = 20 # Number of stocks to long self.longSymbols = [] # Contains the stocks we'd like to long self.nextLiquidate = self.Time # Initialize last trade time self.rebalance_days = 180 # Set the weights of each factor self.beta_m = 4 # 3 better than 2? 3 better. 4 better than 3? 4 better. 5>4? about even. 6>5? 5. self.beta_s = 1 self.beta_h = 0 # putting to 0? 0 better than 1 self.beta_r = 1 # 2 better than 1.5? 1.5. 1>1.5? slightly self.beta_g = 3 # putting to zero not good, from 1 to 2: 2 better, from 2 to 3? 3 better. 3vs 4? 3 better self.beta_c = 2 # better at 2 than 3, 2 better than 1 def CoarseSelectionFunction(self, coarse): '''Drop securities which have no fundamental data or have too low prices. Select those with highest by dollar volume''' if self.Time < self.nextLiquidate: return Universe.Unchanged self.Log("here31") selected = sorted([x for x in coarse if x.HasFundamentalData], # and x.Price > 1 key=lambda x: x.DollarVolume, reverse=True) return [x.Symbol for x in selected] def FineSelectionFunction(self, fine): '''Select securities with highest score on Fama French 5 factors''' # Select stocks with these 5 factors: # MKT -- Book value per share: Value # SMB -- TotalEquity: Size # HML -- Operation profit margin: Quality # RMW -- ROE: Profitability # RMW2 -- gross margin # CMA -- TotalAssetsGrowth: Investment Pattern if self.Time >= self.nextLiquidate: self.Log("here21") filtered = [x for x in fine if x.ValuationRatios.PBRatio and x.FinancialStatements.BalanceSheet.TotalEquity and x.OperationRatios.GrossMargin and x.ValuationRatios.PERatio > 0 and x.ValuationRatios.EVToEBITDA > 0 and x.CompanyReference.CountryId != "CHN"] # and x.CompanyReference.BusinessCountryID != "CHN" self.Log("nr filtered: " + str(len(filtered))) # for x in filtered: # if x.ValuationRatios.PERatio < 4: # self.Log(str(x.Symbol)[:4] + str(x.ValuationRatios.PERatio)) # self.Log(str(x.CompanyReference.BusinessCountryID)) # self.Log(str(x.CompanyReference.CountryId)) # Sort by factors sortedByMkt1 = sorted(filtered, key=lambda x: x.ValuationRatios.PBRatio) sortedByMkt2 = sorted(filtered, key=lambda x: x.ValuationRatios.PERatio) sortedByMkt4 = sorted(filtered, key=lambda x: x.ValuationRatios.EVToEBITDA, reverse=False) sortedByMkt3 = sorted(filtered, key=lambda x: x.ValuationRatios.FCFRatio) sortedBySmb1 = sorted(filtered, key=lambda x: x.FinancialStatements.BalanceSheet.TotalEquity.Value) sortedByHml1 = sorted(filtered, key=lambda x: x.ValuationRatios.PayoutRatio, reverse=True) sortedByRmw1 = sorted(filtered, key=lambda x: x.OperationRatios.GrossMargin.Value, reverse=True) sortedByRmw2 = sorted(filtered, key=lambda x: x.OperationRatios.ROE.Value, reverse=True) sortedByRmw3 = sorted(filtered, key=lambda x: x.OperationRatios.ROA.Value, reverse=True) sortedByCma1 = sorted(filtered, key=lambda x: abs(x.OperationRatios.TotalAssetsGrowth.Value), reverse=False) sortedByCma2 = sorted(filtered, key=lambda x: x.OperationRatios.DebttoAssets.Value, reverse=False) stockBySymbol = {} # Get the rank based on 5 factors for every stock for index, stock in enumerate(sortedBySmb1): mktRank = self.beta_m * index smbRank1 = self.beta_s * sortedByMkt1.index(stock) smbRank2 = self.beta_s * sortedByMkt2.index(stock) smbRank3 = self.beta_s * sortedByMkt3.index(stock) smbRank4 = self.beta_s * sortedByMkt4.index(stock) hmlRank = self.beta_h * sortedByHml1.index(stock) rmwRank1 = self.beta_r * sortedByRmw1.index(stock) rmwRank2 = self.beta_r * sortedByRmw2.index(stock) rmwRank3 = self.beta_r * sortedByRmw3.index(stock) cmaRank1 = self.beta_c * sortedByCma1.index(stock) cmaRank2 = self.beta_g * sortedByCma2.index(stock) avgRank = np.mean([mktRank, smbRank1, smbRank2, smbRank3, smbRank4, hmlRank, rmwRank1, rmwRank2, rmwRank3, cmaRank1, cmaRank2]) stockBySymbol[stock.Symbol] = avgRank self.Log("here11") sorted_dict = sorted(stockBySymbol.items(), key = lambda x: x[1], reverse = False) sectors = [MorningstarSectorCode.BasicMaterials, MorningstarSectorCode.ConsumerCyclical, MorningstarSectorCode.FinancialServices, MorningstarSectorCode.RealEstate, MorningstarSectorCode.ConsumerDefensive, MorningstarSectorCode.Healthcare, MorningstarSectorCode.Utilities, MorningstarSectorCode.CommunicationServices, MorningstarSectorCode.Energy, MorningstarSectorCode.Industrials, MorningstarSectorCode.Technology] self.longSymbols = [] self.Log("here12") for i, sector in enumerate(sectors): sector_symbols = [] # filtered or fine??? sector_stocks = [x.Symbol for x in fine if x.AssetClassification.MorningstarSectorCode == sector] for x in sorted_dict: if x[0] in sector_stocks: sector_symbols.append(x[0]) self.longSymbols += sector_symbols[:int(len(sector_stocks)/self.fraction)] self.Log("here13") return self.longSymbols def OnData(self, data): '''Rebalance Every self.rebalance_days''' # Liquidate stocks in the end of every month self.Log("here1") # if len(self.longSymbols)!= 0: # for symbol in self.longSymbols: # self.AddEquity(symbol) if self.Time >= self.nextLiquidate: for holding in self.Portfolio.Values: # If the holding is in the long/short list for the next month, don't liquidate if holding.Symbol in self.longSymbols: continue # If the holding is not in the list, liquidate if holding.Invested: self.Liquidate(holding.Symbol) count = len(self.longSymbols) self.Log("here2") # It means the long & short lists for the month have been cleared if count == 0: return self.Log("here3") # Open long position at the start of every month for symbol in self.longSymbols: # stock_price = data[symbol].Price self.MarketOrder(symbol, 100) # self.SetHoldings(symbol, 1/count) self.Log("here4") # Set the Liquidate Date self.nextLiquidate = self.Time + timedelta(self.rebalance_days) self.Log("amount:" + str(len(self.longSymbols))) self.Log("invested: " + str([symbol.Value for symbol in self.longSymbols])) # After opening positions, clear the long & short symbol lists until next universe selection self.longSymbols.clear() # symbols = [x[0] for x in sorted_dict] # quality_symbols = [x for x in symbols[:self.num_quality]] # # Get the quarterly returns for each symbol # history = self.History(quality_symbols, self.momentum_days, Resolution.Daily) # history = history.close.unstack(level = 0) #.drop_duplicates() # rankByQuarterReturn = self.GetQuarterlyReturn(history) # for symbol in quality_symbols: # if symbol not in rankByQuarterReturn.keys(): # rankByQuarterReturn[symbol] = self.num_coarse/2 # stockBySymbol = {} # for index, stock in enumerate(quality_symbols): # stockBySymbol[stock] = np.mean([rankByQuarterReturn[stock]]) # sorted_dict = sorted(stockBySymbol.items(), key = lambda x: x[1], reverse = False) # def GetQuarterlyReturn(self, history): # ''' # Get the rank of securities based on their quarterly return from historical close prices # Return: dictionary # ''' # # Get quarterly returns for all symbols # # (The first row divided by the last row) # returns = history.iloc[0] / history.iloc[-1] # # Transform them to dictionary structure # returns = returns.to_dict() # # Get the rank of the returns (key: symbol; value: rank) # # (The symbol with the 1st quarterly return ranks the 1st, etc.) # ranked = sorted(returns, key = returns.get, reverse = True) # return {symbol: rank for rank, symbol in enumerate(ranked, 1)}
import numpy as np import time class FFFiveFactorAdapted(QCAlgorithm): ''' Stocks Selecting Strategy based on Fama French 5 Factors Model Reference: https://tevgeniou.github.io/EquityRiskFactors/bibliography/FiveFactor.pdf ''' def Initialize(self): self.SetStartDate(2010, 1, 1) # Set Start Date self.SetEndDate(2020, 8, 24) # Set End Date self.SetCash(100000) # Set Strategy Cash self.UniverseSettings.Resolution = Resolution.Daily self.AddUniverse(self.CoarseSelectionFunction, self.FineSelectionFunction) # self.num_coarse = 10000 # Number of symbols selected at Coarse Selection self.fraction = 20 # Number of stocks to long self.longSymbols = [] # Contains the stocks we'd like to long self.nextLiquidate = self.Time # Initialize last trade time self.rebalance_days = 340 # Set the weights of each factor self.beta_m = 4 def CoarseSelectionFunction(self, coarse): '''Drop securities which have no fundamental data or have too low prices. Select those with highest by dollar volume''' if self.Time < self.nextLiquidate: return Universe.Unchanged self.Log("here31") selected = sorted([x for x in coarse if x.HasFundamentalData], # and x.Price > 1 key=lambda x: x.DollarVolume, reverse=True) return [x.Symbol for x in selected] def FineSelectionFunction(self, fine): '''Select securities with highest score on Fama French 5 factors''' # Select stocks with these 5 factors: # MKT -- Book value per share: Value # SMB -- TotalEquity: Size # HML -- Operation profit margin: Quality # RMW -- ROE: Profitability # RMW2 -- gross margin # CMA -- TotalAssetsGrowth: Investment Pattern if self.Time >= self.nextLiquidate: filtered = [x for x in fine if x.ValuationRatios.PBRatio and x.FinancialStatements.BalanceSheet.TotalEquity and x.OperationRatios.GrossMargin and x.ValuationRatios.PERatio and x.ValuationRatios.EVToEBITDA] # and x.CompanyReference.BusinessCountryID != "CHN" self.Log("nr filtered: " + str(len(filtered))) # Sort by factors sortedBySmb1 = sorted(filtered, key=lambda x: x.FinancialStatements.BalanceSheet.TotalEquity.Value) stockBySymbol = {} # Get the rank based on 5 factors for every stock for index, stock in enumerate(sortedBySmb1): mktRank = self.beta_m * index avgRank = np.mean([mktRank]) stockBySymbol[stock.Symbol] = avgRank sorted_dict = sorted(stockBySymbol.items(), key = lambda x: x[1], reverse = False) sectors = [MorningstarSectorCode.BasicMaterials, MorningstarSectorCode.ConsumerCyclical, MorningstarSectorCode.FinancialServices, MorningstarSectorCode.RealEstate, MorningstarSectorCode.ConsumerDefensive, MorningstarSectorCode.Healthcare, MorningstarSectorCode.Utilities, MorningstarSectorCode.CommunicationServices, MorningstarSectorCode.Energy, MorningstarSectorCode.Industrials, MorningstarSectorCode.Technology] self.longSymbols = [] for i, sector in enumerate(sectors): sector_symbols = [] # filtered or fine??? sector_stocks = [x.Symbol for x in fine if x.AssetClassification.MorningstarSectorCode == sector] for x in sorted_dict: if x[0] in sector_stocks: sector_symbols.append(x[0]) self.longSymbols += sector_symbols[:int(len(sector_stocks)/self.fraction)] return self.longSymbols def OnData(self, data): '''Rebalance Every self.rebalance_days''' # Liquidate stocks in the end of every month if self.Time >= self.nextLiquidate: for holding in self.Portfolio.Values: # If the holding is in the long/short list for the next month, don't liquidate if holding.Symbol in self.longSymbols: continue # If the holding is not in the list, liquidate if holding.Invested: self.Liquidate(holding.Symbol) count = len(self.longSymbols) # It means the long & short lists for the month have been cleared if count == 0: return # Open long position at the start of every month for symbol in self.longSymbols: self.SetHoldings(symbol, 1/count) # Set the Liquidate Date self.nextLiquidate = self.Time + timedelta(self.rebalance_days) # After opening positions, clear the long & short symbol lists until next universe selection self.longSymbols.clear()
import numpy as np import time class FFFiveFactorAdapted(QCAlgorithm): ''' Stocks Selecting Strategy based on Fama French 5 Factors Model Reference: https://tevgeniou.github.io/EquityRiskFactors/bibliography/FiveFactor.pdf ''' def Initialize(self): self.SetStartDate(2000, 1, 1) # Set Start Date self.SetEndDate(2020, 8, 21) # Set End Date self.SetCash(100000) # Set Strategy Cash self.UniverseSettings.Resolution = Resolution.Daily self.AddUniverse(self.CoarseSelectionFunction, self.FineSelectionFunction) # self.num_coarse = 10000 # Number of symbols selected at Coarse Selection self.fraction = 20 # Number of stocks to long self.longSymbols = [] # Contains the stocks we'd like to long self.nextLiquidate = self.Time # Initialize last trade time self.rebalance_days = 180 # Set the weights of each factor self.beta_m = 2 self.beta_s = 1 self.beta_h = 0 # putting to 0? 0 better than 1 self.beta_r = 1.5 self.beta_g = 3 # putting to zero not good, from 1 to 2: 2 better, from 2 to 3? 3 better self.beta_c = 2 # better at 2 than 3, 2 better than 1 def CoarseSelectionFunction(self, coarse): '''Drop securities which have no fundamental data or have too low prices. Select those with highest by dollar volume''' if self.Time < self.nextLiquidate: return Universe.Unchanged selected = sorted([x for x in coarse if x.HasFundamentalData ], # and x.Price > 1 key=lambda x: x.DollarVolume, reverse=True) return [x.Symbol for x in selected] def FineSelectionFunction(self, fine): '''Select securities with highest score on Fama French 5 factors''' # Select stocks with these 5 factors: # MKT -- Book value per share: Value # SMB -- TotalEquity: Size # HML -- Operation profit margin: Quality # RMW -- ROE: Profitability # RMW2 -- gross margin # CMA -- TotalAssetsGrowth: Investment Pattern if self.Time >= self.nextLiquidate: filtered = [x for x in fine if x.ValuationRatios.PBRatio and x.FinancialStatements.BalanceSheet.TotalEquity and x.OperationRatios.GrossMargin and x.ValuationRatios.PERatio > 0 and x.ValuationRatios.EVToEBITDA > 0] # and x.OperationRatios.TotalAssetsGrowth.Value > 0 # assetgrowth higher than 0??? self.Log("nr filtered: " + str(len(filtered))) # Sort by factors sortedByMkt1 = sorted(filtered, key=lambda x: x.ValuationRatios.PBRatio) sortedByMkt2 = sorted(filtered, key=lambda x: x.ValuationRatios.PERatio) sortedByMkt4 = sorted(filtered, key=lambda x: x.ValuationRatios.EVToEBITDA, reverse=False) sortedByMkt3 = sorted(filtered, key=lambda x: x.ValuationRatios.FCFRatio) sortedBySmb1 = sorted(filtered, key=lambda x: x.FinancialStatements.BalanceSheet.TotalEquity.Value) sortedByHml1 = sorted(filtered, key=lambda x: x.ValuationRatios.PayoutRatio, reverse=True) sortedByRmw1 = sorted(filtered, key=lambda x: x.OperationRatios.GrossMargin.Value, reverse=True) sortedByRmw2 = sorted(filtered, key=lambda x: x.OperationRatios.ROE.Value, reverse=True) sortedByRmw3 = sorted(filtered, key=lambda x: x.OperationRatios.ROA.Value, reverse=True) sortedByCma1 = sorted(filtered, key=lambda x: abs(x.OperationRatios.TotalAssetsGrowth.Value), reverse=False) sortedByCma2 = sorted(filtered, key=lambda x: x.OperationRatios.DebttoAssets.Value, reverse=False) stockBySymbol = {} # Get the rank based on 5 factors for every stock for index, stock in enumerate(sortedBySmb1): mktRank = self.beta_m * index smbRank1 = self.beta_s * sortedByMkt1.index(stock) smbRank2 = self.beta_s * sortedByMkt2.index(stock) smbRank3 = self.beta_s * sortedByMkt3.index(stock) smbRank4 = self.beta_s * sortedByMkt4.index(stock) hmlRank = self.beta_h * sortedByHml1.index(stock) rmwRank1 = self.beta_r * sortedByRmw1.index(stock) rmwRank2 = self.beta_r * sortedByRmw2.index(stock) rmwRank3 = self.beta_r * sortedByRmw3.index(stock) cmaRank1 = self.beta_c * sortedByCma1.index(stock) cmaRank2 = self.beta_g * sortedByCma2.index(stock) avgRank = np.mean([mktRank, smbRank1, smbRank2, smbRank3, smbRank4, hmlRank, rmwRank1, rmwRank2, rmwRank3, cmaRank1, cmaRank2]) stockBySymbol[stock.Symbol] = avgRank sorted_dict = sorted(stockBySymbol.items(), key = lambda x: x[1], reverse = False) sectors = [MorningstarSectorCode.BasicMaterials, MorningstarSectorCode.ConsumerCyclical, MorningstarSectorCode.FinancialServices, MorningstarSectorCode.RealEstate, MorningstarSectorCode.ConsumerDefensive, MorningstarSectorCode.Healthcare, MorningstarSectorCode.Utilities, MorningstarSectorCode.CommunicationServices, MorningstarSectorCode.Energy, MorningstarSectorCode.Industrials, MorningstarSectorCode.Technology] self.longSymbols = [] for i, sector in enumerate(sectors): sector_symbols = [] # filtered or fine??? sector_stocks = [x.Symbol for x in fine if x.AssetClassification.MorningstarSectorCode == sector] for x in sorted_dict: if x[0] in sector_stocks: sector_symbols.append(x[0]) self.longSymbols += sector_symbols[:int(len(sector_stocks)/self.fraction)] # Materials_symbols = [] # for x in sorted_dict: # if x[0] in Materials: # Materials_symbols.append(x[0]) # ConsumerCyclical_symbols = [] # for x in sorted_dict: # if x[0] in ConsumerCyclical: # ConsumerCyclical_symbols.append(x[0]) # Financial_symbols = [] # for x in sorted_dict: # if x[0] in Financial: # Financial_symbols.append(x[0]) # RealEstate_symbols = [] # for x in sorted_dict: # if x[0] in RealEstate: # RealEstate_symbols.append(x[0]) # ConsumerDefensive_symbols = [] # for x in sorted_dict: # if x[0] in ConsumerDefensive: # ConsumerDefensive_symbols.append(x[0]) # Healthcare_symbols = [] # for x in sorted_dict: # if x[0] in Healthcare: # Healthcare_symbols.append(x[0]) # Utilities_symbols = [] # for x in sorted_dict: # if x[0] in Utilities: # Utilities_symbols.append(x[0]) # Communication_symbols = [] # for x in sorted_dict: # if x[0] in Communication: # Communication_symbols.append(x[0]) # Energy_symbols = [] # for x in sorted_dict: # if x[0] in Energy: # Energy_symbols.append(x[0]) # Industrials_symbols = [] # for x in sorted_dict: # if x[0] in Industrials: # Industrials_symbols.append(x[0]) # Technology_symbols = [] # for x in sorted_dict: # if x[0] in Technology: # Technology_symbols.append(x[0]) # # Pick the stocks with the highest scores to long # self.longSymbols= Materials_symbols[:int(len(Materials_symbols)/self.fraction)] + ConsumerCyclical_symbols[:int(len(ConsumerCyclical_symbols)/self.fraction)] + Financial_symbols[:int(len(Financial_symbols)/self.fraction)] + RealEstate_symbols[:int(len(RealEstate_symbols)/self.fraction)] + ConsumerDefensive_symbols[:int(len(ConsumerDefensive_symbols)/self.fraction)] + Healthcare_symbols[:int(len(Healthcare_symbols)/self.fraction)] + Utilities_symbols[:int(len(Utilities_symbols)/self.fraction)] + Communication_symbols[:int(len(Communication_symbols)/self.fraction)] + Energy_symbols[:int(len(Energy_symbols)/self.fraction)] + Industrials_symbols[:int(len(Industrials_symbols)/self.fraction)] + Technology_symbols[:int(len(Technology_symbols)/self.fraction)] ## one sector: # one_sector = [] # for x in sorted_dict: # if x[0] in Technology: # one_sector.append(x[0]) # self.longSymbols= one_sector[:self.num_long] return self.longSymbols def OnData(self, data): '''Rebalance Every self.rebalance_days''' # Liquidate stocks in the end of every month if self.Time >= self.nextLiquidate: for holding in self.Portfolio.Values: # If the holding is in the long/short list for the next month, don't liquidate if holding.Symbol in self.longSymbols: continue # If the holding is not in the list, liquidate if holding.Invested: self.Liquidate(holding.Symbol) count = len(self.longSymbols) # It means the long & short lists for the month have been cleared if count == 0: return # Open long position at the start of every month for symbol in self.longSymbols: self.SetHoldings(symbol, 1/count) # Set the Liquidate Date self.nextLiquidate = self.Time + timedelta(self.rebalance_days) self.Log("amount:" + str(len(self.longSymbols))) self.Log("invested: " + str([symbol.Value for symbol in self.longSymbols])) # After opening positions, clear the long & short symbol lists until next universe selection self.longSymbols.clear() # symbols = [x[0] for x in sorted_dict] # quality_symbols = [x for x in symbols[:self.num_quality]] # # Get the quarterly returns for each symbol # history = self.History(quality_symbols, self.momentum_days, Resolution.Daily) # history = history.close.unstack(level = 0) #.drop_duplicates() # rankByQuarterReturn = self.GetQuarterlyReturn(history) # for symbol in quality_symbols: # if symbol not in rankByQuarterReturn.keys(): # rankByQuarterReturn[symbol] = self.num_coarse/2 # stockBySymbol = {} # for index, stock in enumerate(quality_symbols): # stockBySymbol[stock] = np.mean([rankByQuarterReturn[stock]]) # sorted_dict = sorted(stockBySymbol.items(), key = lambda x: x[1], reverse = False) # def GetQuarterlyReturn(self, history): # ''' # Get the rank of securities based on their quarterly return from historical close prices # Return: dictionary # ''' # # Get quarterly returns for all symbols # # (The first row divided by the last row) # returns = history.iloc[0] / history.iloc[-1] # # Transform them to dictionary structure # returns = returns.to_dict() # # Get the rank of the returns (key: symbol; value: rank) # # (The symbol with the 1st quarterly return ranks the 1st, etc.) # ranked = sorted(returns, key = returns.get, reverse = True) # return {symbol: rank for rank, symbol in enumerate(ranked, 1)}