Overall Statistics |
Total Trades 14635 Average Win 0.38% Average Loss -0.31% Compounding Annual Return 79.254% Drawdown 30.100% Expectancy 0.169 Net Profit 3453.443% Sharpe Ratio 2.422 Probabilistic Sharpe Ratio 99.301% Loss Rate 48% Win Rate 52% Profit-Loss Ratio 1.23 Alpha 0.69 Beta -0.073 Annual Standard Deviation 0.281 Annual Variance 0.079 Information Ratio 1.662 Tracking Error 0.333 Treynor Ratio -9.284 Total Fees $2486672.50 Estimated Strategy Capacity $190000.00 |
class MinusOneRisk(RiskManagementModel): def __init__(self, maximumDrawdownPercent = 0.1): self.maximumDrawdownPercent = -abs(maximumDrawdownPercent) self.liquidated = set() def ManageRisk(self, algorithm, targets): targets = [] algorithm.Log(len(self.liquidated)) for s in algorithm.Securities: security = s.Value if not security.Invested: continue pnl = security.Holdings.UnrealizedProfitPercent if pnl < self.maximumDrawdownPercent or security.Symbol in self.liquidated: # liquidate targets.append(PortfolioTarget(security.Symbol, 0)) if algorithm.Securities[security.Symbol].Invested: self.liquidated.add(security.Symbol) algorithm.Log(f"Liquidating {security.Symbol} to avoid drawdown") return targets
class MinusOneAlphaModel(AlphaModel): # Direction score close to 50 is good? https://www.quantconnect.com/forum/discussion/7082/alpha-score-vs-performance/p1 def __init__(self, algorithm): self.leverage = 1 self.algo = algorithm def Update(self, algorithm, data): # This is where insights are returned, which are then passed to the # Portfolio Construction, Risk, and Execution models. # The following Insight properties MUST be set before returning # - Symbol -- Secuirty Symbol # - Duration -- Time duration that the Insight is in effect # - Direction -- Direction of predicted price movement # - Weight -- Proportion of self.algo capital to be allocated to this Insight # onData gets called after onSecuritiesChanged here because we filtered the entire data first insights = [] for symbolData in self.algo.symbolDataDict.values(): symbol = symbolData.symbol symbolData.marketCap = self.algo.symbolToMarketCap.get(symbol,-1) # you need to update this in case some of the stocks persisted into next selection period, which won't be warmed up in onSecuritiesChanged security = self.algo.Securities[symbol] self.algo.symbolDataDict[symbol].update_closes(security.Close) self.algo.symbolDataDict[symbol].update_highs(security.High) self.algo.symbolDataDict[symbol].update_lows(security.Low) if not self.algo.selection_flag: return insights self.algo.selection_flag = False symbolDataList = list(self.algo.symbolDataDict.values()) # sort by market cap sorted_by_market_cap = [s.symbol for s in sorted(symbolDataList, key=lambda s:s.marketCap, reverse=True)] top = sorted_by_market_cap[:self.algo.top_by_market_cap_count//2] bottom = sorted_by_market_cap[-self.algo.top_by_market_cap_count//2:] # sort performance (from worst) for top market cap and performance (from best) for bottom market cap top_performances = {symbol : self.algo.symbolDataDict[symbol].period_return() for symbol in top if self.algo.symbolDataDict[symbol].is_ready()} bottom_performances = {symbol : self.algo.symbolDataDict[symbol].period_return() for symbol in bottom if self.algo.symbolDataDict[symbol].is_ready()} sorted_top_performances = [x[0] for x in sorted(top_performances.items(), key=lambda item: item[1])] sorted_bottom_performances = [x[0] for x in sorted(bottom_performances.items(), key=lambda item: item[1], reverse=True)] # select securities to short and long (50-50) self.algo.long = sorted_top_performances[:self.algo.stock_selection//2] self.algo.short = sorted_bottom_performances[:self.algo.stock_selection//2] invested = [x.Key for x in self.algo.Portfolio if x.Value.Invested] for symbol in invested: # if they are not to be selected again, then they are liquidated if symbol not in self.algo.long + self.algo.short: #self.algo.Liquidate(symbol) insights.append(Insight(symbol, timedelta(30), InsightType.Price, InsightDirection.Flat, None, None)) # reduce the insight valid period insight_period = timedelta(hours=6) insights += [Insight.Price(s, insight_period, InsightDirection.Up, None, None, None, self.leverage/len(self.algo.long)) for s in self.algo.long] insights += [Insight.Price(s, insight_period, InsightDirection.Down, None, None, None, self.leverage/len(self.algo.short)) for s in self.algo.short] self.algo.long.clear() self.algo.short.clear() return insights def OnSecuritiesChanged(self, algorithm, changes): #self.Debug("change") for removed in changes.RemovedSecurities: self.algo.symbolDataDict.pop(removed.Symbol, None) for security in changes.AddedSecurities: security.SetLeverage(self.leverage) symbol = security.Symbol daily_history = self.algo.History(symbol, self.algo.period+1, Resolution.Daily) if daily_history.empty: algorithm.Log(f"Not enough data for {symbol} yet") continue if symbol not in self.algo.symbolDataDict.keys(): symbolData = SymbolData(symbol, self.algo.period, self.algo) self.algo.symbolDataDict[symbol] = symbolData symbolData.warmup(daily_history) class SymbolData(): def __init__(self, symbol, period, algo): self.symbol = symbol self.marketCap = None self.algo = algo self.closes = RollingWindow[float](period+1) self.highs = RollingWindow[float](period+1) self.lows = RollingWindow[float](period+1) self.period = period def update_closes(self, close): self.closes.Add(close) def update_highs(self, high): self.highs.Add(high) def update_lows(self, low): self.lows.Add(low) def period_return(self): if self.closes[self.period] == 0: self.algo.Debug(self.symbol) return None return self.closes[0] / self.closes[self.period] - 1 def is_ready(self) -> bool: return self.closes.IsReady and self.highs.IsReady and self.lows.IsReady and self.closes[self.period] != 0 def warmup(self, daily_history): if daily_history.empty: return closes = daily_history.loc[self.symbol].close for time, c in closes.iteritems(): self.update_closes(c) highs = daily_history.loc[self.symbol].high for time, h in highs.iteritems(): self.update_highs(h) lows = daily_history.loc[self.symbol].low for time, l in lows.iteritems(): self.update_lows(l)
from alpha import MinusOneAlphaModel from risk import MinusOneRisk class MinusOneAlphaAlgorithm(QCAlgorithm): def Initialize(self): # The blocked section of code below is to remain UNCHANGED for the weekly competitions. # # Insight-weighting portfolio construction model: # - You can change the rebalancing date rules or portfolio bias # - For more info see https://github.com/QuantConnect/Lean/blob/master/Algorithm.Framework/Portfolio/InsightWeightingPortfolioConstructionModel.py # # Use the Alpha Streams Brokerage Model: # - Developed in conjunction with funds to model their actual fees, costs, etc. Please do not modify other models. ############################################################################################################################### self.SetStartDate(2015, 3, 1) # 5 years up to the submission date self.SetCash(1000000) # Set $1m Strategy Cash to trade significant AUM self.SetBenchmark('SPY') # SPY Benchmark self.SetBrokerageModel(AlphaStreamsBrokerageModel()) self.SetExecution(ImmediateExecutionModel()) self.SetPortfolioConstruction(InsightWeightingPortfolioConstructionModel()) ############################################################################################################################### # Do not change the code above self.symbol = self.AddEquity('SPY', Resolution.Daily).Symbol self.coarse_count = 500 self.stock_selection = 10 # 5 self.top_by_market_cap_count = 100 self.period = 2 self.SetWarmUp(timedelta(self.period)) self.long = [] self.short = [] # symbolData at each moment self.symbolDataDict = {} self.symbolToMarketCap = {} self.SetBenchmark("SPY") self.day = 1 self.selection_flag = False self.UniverseSettings.Resolution = Resolution.Daily self.AddUniverse(self.CoarseSelectionFunction, self.FineSelectionFunction) self.Schedule.On(self.DateRules.EveryDay("SPY"), self.TimeRules.BeforeMarketClose("SPY"), self.Selection) #self.AddRiskManagement(MaximumDrawdownPercentPortfolio(0.1,True)) self.AddRiskManagement(MaximumDrawdownPercentPerSecurity(maximumDrawdownPercent = 0.1)) # Add the alpha model and anything else you want below self.AddAlpha(MinusOneAlphaModel(self)) # Add a universe selection model self.AddUniverse(self.CoarseSelectionFunction, self.FineSelectionFunction) # def CoarseSelectionFunction(self, coarse): #self.Debug("coarse") if not self.selection_flag: # self.selection_flag is only true when day is 5 or it is a Friday. return Universe.Unchanged selected = sorted([x for x in coarse if x.HasFundamentalData and x.Market == 'usa' and x.Price > 1], key=lambda x: x.DollarVolume, reverse=True) selected = [x.Symbol for x in selected][:self.coarse_count] return selected def FineSelectionFunction(self, fine): # the long and short lists are updated daily #self.Debug("Fine") if not self.selection_flag: # self.selection_flag is only true when day is 5 or it is a Friday. return Universe.Unchanged fine = [x for x in fine if x.MarketCap != 0] sorted_by_market_cap = sorted(fine, key = lambda x:x.MarketCap, reverse = True) top_by_market_cap = { "top":[x.Symbol for x in sorted_by_market_cap[:self.top_by_market_cap_count//2]], "bottom":[x.Symbol for x in sorted_by_market_cap[-self.top_by_market_cap_count//2:]] } # get the top market cap and bottom market cap (50-50) filtered = top_by_market_cap["top"]+top_by_market_cap["bottom"] filteredFine = sorted_by_market_cap[:self.top_by_market_cap_count//2]+sorted_by_market_cap[-self.top_by_market_cap_count//2:] for f in filteredFine: self.symbolToMarketCap[f.Symbol] = f.MarketCap return filtered def Selection(self): if self.day == self.period: self.selection_flag = True self.day += 1 if self.day > self.period: self.day = 1