Overall Statistics |
Total Trades 0 Average Win 0% Average Loss 0% Compounding Annual Return 0% Drawdown 0% Expectancy 0 Net Profit 0% Sharpe Ratio 0 Loss Rate 0% Win Rate 0% Profit-Loss Ratio 0 Alpha 0 Beta 0 Annual Standard Deviation 0 Annual Variance 0 Information Ratio 0 Tracking Error 0 Treynor Ratio 0 Total Fees $0.00 |
from Risk.NullRiskManagementModel import NullRiskManagementModel from RsiAlphaModel import RsiAlphaModel class BasicTemplateFrameworkAlgorithm(QCAlgorithmFramework): def Initialize(self): # Set requested data resolution self.UniverseSettings.Resolution = Resolution.Daily self.SetStartDate(2014, 9, 9) #Set Start Date self.SetEndDate(2019, 3, 9) #Set End Date self.SetCash(100000) #Set Strategy Cash self.UniverseSettings.Resolution = Resolution.Daily symbols = [ Symbol.Create("SPY", SecurityType.Equity, Market.USA) ] # self.spy = self.AddEquity('SPY', Resolution.Daily) self.SetUniverseSelection( ManualUniverseSelectionModel(symbols) ) # self.rsi = self.RSI("SPY", 200, MovingAverageType.Simple, Resolution.Daily) # self.SetBenchmark("SPY") self.SetAlpha(RsiAlphaModel()) self.SetPortfolioConstruction(NullPortfolioConstructionModel()) self.SetExecution(NullExecutionModel()) self.SetRiskManagement(NullRiskManagementModel()) def OnOrderEvent(self, orderEvent): if orderEvent.Status == OrderStatus.Filled: self.Debug("Purchased Stock: {0}".format(orderEvent.Symbol)) pass
from clr import AddReference AddReference("QuantConnect.Algorithm") AddReference("QuantConnect.Algorithm.Framework") AddReference("QuantConnect.Indicators") AddReference("QuantConnect.Logging") AddReference("QuantConnect.Common") from QuantConnect import * from QuantConnect.Indicators import * from QuantConnect.Logging import Log from QuantConnect.Algorithm import * from QuantConnect.Algorithm.Framework import * from QuantConnect.Algorithm.Framework.Alphas import * from datetime import timedelta from enum import Enum class RsiAlphaModel(AlphaModel): '''Uses Wilder's RSI to create insights. Using default settings, a cross over below 30 or above 70 will trigger a new insight.''' def __init__(self, period = 14, resolution = Resolution.Daily): '''Initializes a new instance of the RsiAlphaModel class Args: period: The RSI indicator period''' 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