Overall Statistics |
Total Trades 91 Average Win 2.19% Average Loss -1.62% Compounding Annual Return 20.861% Drawdown 18.900% Expectancy 0.444 Net Profit 28.559% Sharpe Ratio 0.734 Probabilistic Sharpe Ratio 33.648% Loss Rate 39% Win Rate 61% Profit-Loss Ratio 1.35 Alpha -0.012 Beta 1.023 Annual Standard Deviation 0.281 Annual Variance 0.079 Information Ratio -0.043 Tracking Error 0.169 Treynor Ratio 0.202 Total Fees $411.26 |
import operator from math import ceil,floor class Stockpicking(QCAlgorithm): def Initialize(self): # training period self.SetStartDate(2009,1,2) # Set Start Date self.SetEndDate(2010,5,2) # Set End Date # testing period #self.SetStartDate(2018,1,1) self.SetCash(50000) # Set Strategy Cash #self.Settings.FreePortfolioValuePercentage=0.2 # reserved cash buffer self.SetBrokerageModel(BrokerageName.InteractiveBrokersBrokerage) # for stock trading ################### self.CoarseSize = 300 # number of stocks in selected coarse universe self.FineSize = 10 # number of stocks in selected fine universe self.num_portfolios = 6 benchmark="SPY" ######################### # control flags self.flags={ # use dictionary to only transfer reference and keep unique value 'rebalance': True, # monthly rebalance of coarse and fine selection function 'fineSelected': False, # monthly rebalance of OnData function 'num_rebalanced': 0 # record the number of rebalancing times } self.last_rebalance=self.Time # auto initialization self.AddEquity(benchmark) # schedule symbol self.SetBenchmark(benchmark) self.AddUniverse(self.SelectCoarse, self.SelectFine) # universe selection happens only at midnight and cannot be changed # the resolution that added assets use self.UniverseSettings.Resolution = Resolution.Minute # there is no resolution above daily, use schedule instead self.Schedule.On(self.DateRules.MonthStart(benchmark), self.TimeRules.AfterMarketOpen(benchmark), Action(self.Rebalancing)) self.AddAlpha(SpAlpha(self.flags,benchmark)) self.SetPortfolioConstruction(SpPortfolio()) self.SetExecution(SpExecution()) self.rmvd_symbols=[] self.readded_symbols=[] self.rermvd_symbols=[] def SelectCoarse(self, coarse): if self.flags['rebalance']: CoarseWithFundamental = [x for x in coarse if x.HasFundamentalData] sortedByDollarVolume = sorted(CoarseWithFundamental, key=lambda x: x.DollarVolume, reverse=True) top = sortedByDollarVolume[:self.CoarseSize] return [c.Symbol for c in top] else: return Universe.Unchanged def SelectFine(self, fine): # will skip if Universe.Unchanged received if self.flags['rebalance']: self.flags['rebalance'] = False self.flags['fineSelected'] = True # filter the fine by deleting equities wit zero factor value filtered = [x for x in fine if x.EarningReports.TotalDividendPerShare.ThreeMonths and x.ValuationRatios.PriceChange1M and x.ValuationRatios.BookValuePerShare and x.ValuationRatios.FCFYield] sortedByfactor1 = sorted(filtered, key=lambda x: x.EarningReports.TotalDividendPerShare.ThreeMonths, reverse=True) # descending sortedByfactor2 = sorted(filtered, key=lambda x: x.ValuationRatios.PriceChange1M, reverse=False) sortedByfactor3 = sorted(filtered, key=lambda x: x.ValuationRatios.BookValuePerShare, reverse=True) sortedByfactor4 = sorted(filtered, key=lambda x: x.ValuationRatios.FCFYield, reverse=True) num_stocks = floor(len(filtered)/self.num_portfolios) stock_dict = {} for i,ele in enumerate(sortedByfactor1): rank1 = i rank2 = sortedByfactor2.index(ele) rank3 = sortedByfactor3.index(ele) rank4 = sortedByfactor4.index(ele) score = [ceil(rank1/num_stocks), ceil(rank2/num_stocks), ceil(rank3/num_stocks), ceil(rank4/num_stocks)] score = sum(score) stock_dict[ele] = score #self.Log("score" + str(score)) self.sorted_stock = sorted(stock_dict.items(), key=lambda d:d[1],reverse=True) sorted_symbol = [self.sorted_stock[i][0] for i in range(len(self.sorted_stock))] topFine = sorted_symbol[:self.FineSize] self.flags['num_rebalanced'] += 1 selection=[i.Symbol for i in topFine] tickers=[i.Value for i in selection] failed_to_add=[x.Value for x in selection if x in self.ActiveSecurities.Keys and\ not self.ActiveSecurities[x].Invested] if len(failed_to_add)>0: self.Debug(f"{self.Time.date()} | active but uninvested symbols to add: {[str(x) for x in failed_to_add][:5]}. ") return selection else: return Universe.Unchanged def OnData(self, data): pass # this event fires whenever we have changes to our universe def OnSecuritiesChanged(self, changes): for security in changes.RemovedSecurities: if security.Invested: self.Liquidate(security.Symbol) for security in changes.AddedSecurities: if security.Symbol in self.rmvd_symbols: self.readded_symbols.append(security.Symbol) for security in changes.RemovedSecurities: self.rmvd_symbols.append(security.Symbol) if security.Symbol in self.readded_symbols: self.rermvd_symbols.append(security.Symbol) self.Debug(f"{self.Time.date()} | readded symbols: {[str(x.Value) for x in self.readded_symbols][:5]}. ") self.Debug(f"{self.Time.date()} | reremoved symbols: {[str(x.Value) for x in self.rermvd_symbols][:5]}. ") def Rebalancing(self): if self.Time-self.last_rebalance > timedelta(days=20): self.last_rebalance=self.Time self.flags['rebalance'] = True #################################################### # Algorithm framework model that produces insights # #################################################### class SpAlpha(AlphaModel): def __init__(self,flags,benchmark): self.Name='StockpickingAlpha' self.flags=flags # global control flags input as reference self.changes = None # securities changes self.benchmark=benchmark def Update(self, algorithm, slice): # Updates this alpha model with the latest data from the algorithm. # This is called each time the algorithm receives data for subscribed securities # Generate insights on the securities in the universe. insights = [] if self.changes != None and self.flags['fineSelected'] and self.flags['num_rebalanced'] > 0 : self.flags['fineSelected'] = False # purchase added securities for security in self.changes.AddedSecurities: if security.Symbol.Value!=self.benchmark: insights.append( Insight.Price(security.Symbol, timedelta(days = 30), InsightDirection.Up, None, None, self.Name ) ) self.changes = None return insights def OnSecuritiesChanged(self, algorithm, changes): # Handle security changes in from your universe model. self.changes=changes ################################ # Portfolio construction model # ################################ class SpPortfolio(PortfolioConstructionModel): def __init__(self): self.changes = None # securities changes # Create list of PortfolioTarget objects from Insights def CreateTargets(self, algorithm, insights): targets=[] if self.changes != None: for Insight in insights: if Insight.Direction == InsightDirection.Up: targets.append(PortfolioTarget.Percent(algorithm, Insight.Symbol, 0.8/float(len(self.changes.AddedSecurities)))) self.changes = None return [t for t in targets if t is not None] # OPTIONAL: Security change details def OnSecuritiesChanged(self, algorithm, changes): # Security additions and removals are pushed here. # This can be used for setting up algorithm state. # changes.AddedSecurities: # changes.RemovedSecurities: self.changes=changes ################### # Execution Model # ################### class SpExecution: def __init__(self): pass # Fill the supplied portfolio targets efficiently def Execute(self, algorithm, targets): for Target in targets: ############### # open position algorithm.MarketOrder(Target.Symbol, Target.Quantity, True) # False: wait till filled # Optional: Securities changes event for handling new securities. def OnSecuritiesChanged(self, algorithm, changes): pass