Overall Statistics |
Total Trades 30 Average Win 0.42% Average Loss -0.15% Compounding Annual Return 10.670% Drawdown 4.600% Expectancy 0.675 Net Profit 1.586% Sharpe Ratio 0.816 Probabilistic Sharpe Ratio 46.608% Loss Rate 56% Win Rate 44% Profit-Loss Ratio 2.83 Alpha 0.111 Beta -0.442 Annual Standard Deviation 0.15 Annual Variance 0.022 Information Ratio 0.533 Tracking Error 0.278 Treynor Ratio -0.277 Total Fees $55.71 |
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(2010, 1, 1) # Set Start Date self.SetEndDate(2010, 2, 28) # Set End Date self.SetCash(100000) # Set Strategy Cash self.SetExecution(ImmediateExecutionModel()) self.SetPortfolioConstruction(EqualWeightingPortfolioConstructionModel()) self.SetRiskManagement(MaximumDrawdownPercentPerSecurity(0.05)) 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.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), period) self.symbolDataBySymbol = {} self.closeWindows = {} 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 previous_state = symbolData.State state = self.GetState(rsi, previous_state) if state != previous_state and rsi.IsReady: if state == State.TrippedLow: insights.append(Insight.Price(symbol, self.insightPeriod, InsightDirection.Up)) if state == State.TrippedHigh: insights.append(Insight.Price(symbol, self.insightPeriod, InsightDirection.Down)) 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, self.resolution) for symbol in addedSymbols: rsi = algorithm.RSI(symbol, self.period, MovingAverageType.Wilders, self.resolution) self.closeWindows[symbol] = RollingWindow[float](4) # need to figure out how to load historical data into rolling window if not history.empty: ticker = SymbolCache.GetTicker(symbol) if ticker not in history.index.levels[0]: Log.Trace(f'RsiAlphaModel.OnSecuritiesChanged: {ticker} not found in history data frame.') continue for tuple in history.loc[ticker].itertuples(): rsi.Update(tuple.Index, tuple.close) 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