Overall Statistics |
Total Trades 10 Average Win 0% Average Loss 0% Compounding Annual Return 0.500% Drawdown 0.600% Expectancy 0 Net Profit 0.215% Sharpe Ratio 0.48 Probabilistic Sharpe Ratio 35.275% Loss Rate 0% Win Rate 0% Profit-Loss Ratio 0 Alpha 0.004 Beta -0.001 Annual Standard Deviation 0.009 Annual Variance 0 Information Ratio -0.201 Tracking Error 0.423 Treynor Ratio -3.142 Total Fees $10.00 Estimated Strategy Capacity $22000000000.00 Lowest Capacity Asset BA R735QTJ8XC9X |
from Execution.ImmediateExecutionModel import ImmediateExecutionModel from Portfolio.EqualWeightingPortfolioConstructionModel import EqualWeightingPortfolioConstructionModel from Risk.MaximumDrawdownPercentPerSecurity import MaximumDrawdownPercentPerSecurity from Selection.QC500UniverseSelectionModel import QC500UniverseSelectionModel class SimpleRSITestQC500Universe(QCAlgorithm): def Initialize(self): self.SetStartDate(2020, 1, 1) # Set Start Date self.SetEndDate(2020, 6, 5) # Set End Date self.SetCash(100000) # Set Strategy Cash self.SetExecution(ImmediateExecutionModel()) self.SetPortfolioConstruction(NullPortfolioConstructionModel()) self.SetRiskManagement(NullRiskManagementModel()) symbols = [ Symbol.Create("SPY", SecurityType.Equity, Market.USA), Symbol.Create("GE", SecurityType.Equity, Market.USA), Symbol.Create("BA", SecurityType.Equity, Market.USA) ] self.SetUniverseSelection(ManualUniverseSelectionModel(symbols)) self.UniverseSettings.Resolution = Resolution.Daily #self.AddUniverse(self.Universe.Index.QC500) self.AddAlpha(RsiAlphaModelTest()) class RsiAlphaModelTest(AlphaModel): def __init__(self, period = 14, resolution = Resolution.Daily): self.period = period self.resolution = resolution self.insightPeriod = Time.Multiply(Extensions.ToTimeSpan(resolution), 5) self.symbolDataBySymbol = {} self.closeWindows = {} self.rsiWindows = {} #self.insightCountUps = {} resolutionString = Extensions.GetEnumString(resolution, Resolution) self.Name = '{}({},{})'.format(self.__class__.__name__, period, resolutionString) def Update(self, algorithm, data): insights = [] for symbol, symbolData in self.symbolDataBySymbol.items(): if data.ContainsKey(symbol) and data[symbol] is not None: self.closeWindows[symbol].Add(data[symbol].Close) #if self.closeWindows[symbol].Count>2: #algorithm.Debug(self.closeWindows[symbol][2]) rsi = symbolData.RSI self.rsiWindows[symbol].Add(rsi.Current.Value) #self.insightCountUps[symbol] += 1 #if self.rsiWindows[symbol].IsReady: # plot oldest RSI value #algorithm.Plot('RSI', symbol.Value, self.rsiWindows[symbol][0]) previous_state = symbolData.State state = self.GetState(rsi, previous_state) if rsi.IsReady: if previous_state == State.TrippedLow and self.closeWindows[symbol][0] - self.closeWindows[symbol][1] >= 5: insights.append(Insight.Price(symbol, self.insightPeriod, InsightDirection.Up)) #self.insightCountUps[symbol] = 0 algorithm.MarketOrder(symbol, 1) if previous_state == State.TrippedHigh and self.closeWindows[symbol][0] - self.closeWindows[symbol][1] <= -5: insights.append(Insight.Price(symbol, self.insightPeriod, InsightDirection.Down)) #self.insightCountUps[symbol] = 0 algorithm.MarketOrder(symbol, -1) #if self.insightCountUps[symbol] >= 5: # algorithm.Liquidate(symbol) symbolData.State = state return insights def OnSecuritiesChanged(self, algorithm, changes): # clean up data for removed securities symbols = [ x.Symbol for x in changes.RemovedSecurities ] if len(symbols) > 0: for subscription in algorithm.SubscriptionManager.Subscriptions: if subscription.Symbol in symbols: self.symbolDataBySymbol.pop(subscription.Symbol, None) subscription.Consolidators.Clear() # initialize data for added securities addedSymbols = [ x.Symbol for x in changes.AddedSecurities if x.Symbol not in self.symbolDataBySymbol] if len(addedSymbols) == 0: return history = algorithm.History(addedSymbols, self.period + 20, self.resolution) for symbol in addedSymbols: rsi = algorithm.RSI(symbol, self.period, MovingAverageType.Wilders, self.resolution) self.rsiWindows[symbol] = RollingWindow[float](20) self.closeWindows[symbol] = RollingWindow[float](self.period) #self.insightCountUps[symbol] = 0 for tuple in history.loc[symbol].itertuples(): self.closeWindows[symbol].Add(tuple.close) rsi.Update(tuple.Index, tuple.close) if rsi.IsReady: self.rsiWindows[symbol].Add(rsi.Current.Value) self.symbolDataBySymbol[symbol] = SymbolData(symbol, rsi) def GetState(self, rsi, previous): if rsi.Current.Value > 70: return State.TrippedHigh if rsi.Current.Value < 30: return State.TrippedLow if previous == State.TrippedLow: if rsi.Current.Value > 35: return State.Middle if previous == State.TrippedHigh: if rsi.Current.Value < 65: return State.Middle return previous class SymbolData: def __init__(self, symbol, rsi): self.Symbol = symbol self.RSI = rsi self.State = State.Middle class State(Enum): '''Defines the state. This is used to prevent signal spamming and aid in bounce detection.''' TrippedLow = 0 Middle = 1 TrippedHigh = 2