Overall Statistics |

Total Trades 105 Average Win 0.50% Average Loss -0.49% Compounding Annual Return -10.177% Drawdown 34.500% Expectancy -0.405 Net Profit -10.256% Sharpe Ratio -0.177 Probabilistic Sharpe Ratio 9.470% Loss Rate 71% Win Rate 29% Profit-Loss Ratio 1.02 Alpha -0.134 Beta 0.729 Annual Standard Deviation 0.275 Annual Variance 0.075 Information Ratio -0.9 Tracking Error 0.185 Treynor Ratio -0.067 Total Fees $318.51 |

class CalibratedTransdimensionalThrustAssembly(QCAlgorithm): def Initialize(self): # Set Start Date so that backtest has 5+ years of data self.SetStartDate(2019, 7, 9) # No need to set End Date as the final submission will be tested # up until the review date # Set $1m Strategy Cash to trade significant AUM self.SetCash(1000000) # Add a relevant benchmark, with the default being SPY self.AddEquity('SPY') self.SetBenchmark('SPY') # Use the Alpha Streams Brokerage Model, developed in conjunction with # funds to model their actual fees, costs, etc. # Please do not add any additional reality modelling, such as Slippage, Fees, Buying Power, etc. self.SetBrokerageModel(AlphaStreamsBrokerageModel()) self.AddAlpha(RsiAlphaModel()) self.SetUniverseSelection(CoarseFundamentalUniverseSelectionModel(self.CoarseSelectionFunction)) self.SetPortfolioConstruction(EqualWeightingPortfolioConstructionModel()) # sort the data by daily dollar volume and take the top '5' def CoarseSelectionFunction(self, coarse): # sort descending by daily dollar volume sortedByDollarVolume = sorted(coarse, key=lambda x: x.DollarVolume, reverse=True) # return the symbol objects of the top entries from our sorted collection return [ x.Symbol for x in sortedByDollarVolume[:5] ] def MyAlpha(self, data): pass class RsiAlphaModel(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 ={} resolutionString = Extensions.GetEnumString(resolution, Resolution) self.Name = '{}({},{})'.format(self.__class__.__name__, period, resolutionString) def Update(self, algorithm, data): '''Updates this alpha model with the latest data from the algorithm. This is called each time the algorithm receives data for subscribed securities Args: algorithm: The algorithm instance data: The new data available Returns: The new insights generated''' insights = [] for symbol, symbolData in self.symbolDataBySymbol.items(): 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): '''Cleans out old security data and initializes the RSI for any newly added securities. Event fired each time the we add/remove securities from the data feed Args: algorithm: The algorithm instance that experienced the change in securities changes: The security additions and removals from the algorithm''' # 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) 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): ''' Determines the new state. This is basically cross-over detection logic that includes considerations for bouncing using the configured bounce tolerance.''' 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: '''Contains data specific to a symbol required by this model''' 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