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 Alphas.EmaCrossAlphaModel import EmaCrossAlphaModel from Execution.ImmediateExecutionModel import ImmediateExecutionModel from Risk.NullRiskManagementModel import NullRiskManagementModel from Selection.QC500UniverseSelectionModel import QC500UniverseSelectionModel from SlopeBasedEquityMomentumAlphaModel import SlopeBasedEquityMomentumAlphaModel from Portfolio.EqualWeightingPortfolioConstructionModel import EqualWeightingPortfolioConstructionModel from RebalancingTimer import RebalancingTimer class BasicTemplateFrameworkAlgorithm(QCAlgorithmFramework): def Initialize(self): # Set requested data resolution self.UniverseSettings.Resolution = Resolution.Daily self.SetStartDate(2019, 2, 1) #Set Start Date self.SetEndDate(2019, 3, 6) #Set End Date self.SetCash(100000) #Set Strategy Cash self.rebalancingTimer = RebalancingTimer(self.Time,"monthly") # selection will run on mon/tues/thurs at 00:00/06:00/12:00/18:00 self.SetUniverseSelection(QC500UniverseSelectionModel()) self.SetAlpha(SlopeBasedEquityMomentumAlphaModel(indexAverageWindow = 0,rebalancingTimer=self.rebalancingTimer)) self.SetPortfolioConstruction(NullPortfolioConstructionModel()) self.SetPortfolioConstruction(EqualWeightingPortfolioConstructionModel()) self.SetExecution(ImmediateExecutionModel()) self.SetRiskManagement(NullRiskManagementModel()) def OnOrderEvent(self, orderEvent): if orderEvent.Status == OrderStatus.Filled: # self.Debug("Purchased Stock: {0}".format(orderEvent.Symbol)) pass
# QUANTCONNECT.COM - Democratizing Finance, Empowering Individuals. # Lean Algorithmic Trading Engine v2.0. Copyright 2014 QuantConnect Corporation. # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. import numpy as np from scipy import stats # using this for the reg slope from clr import AddReference AddReference("QuantConnect.Common") AddReference("QuantConnect.Algorithm") AddReference("QuantConnect.Algorithm.Framework") AddReference("QuantConnect.Indicators") from RebalancingTimer import * from QuantConnect import * from QuantConnect.Indicators import * from QuantConnect.Algorithm import * from QuantConnect.Algorithm.Framework import * from QuantConnect.Algorithm.Framework.Alphas import * class SlopeBasedEquityMomentumAlphaModel(AlphaModel): '''Defines a custom alpha model that uses the average of two momentum slopes''' def __init__(self, shortTermMomentumWindow = 60, longTermMomentumWindow = 90, minimumMomentum = 60, indexAverageWindow = 100, resolution = Resolution.Daily, rebalancingTimer = None): ''' Initializes a new instance of the MacdAlphaModel class Args: shortTermMomentumWindow: The short term momentum window longTermMomentumWindow: The long term momentum window minimumMomentum: The minimum momentum required for signal generation indexAverageWindow: The window over which the momentum of the reference index is average. If the value is >0 the index momentum is used as trend filter. resolution: The resolution of the momentum calculation''' self.shortTermMomentumWindow = shortTermMomentumWindow self.longTermMomentumWindow = longTermMomentumWindow self.minimumMomentum = minimumMomentum self.indexAverageWindow = indexAverageWindow self.resolution = resolution self.predictionInterval = Time.Multiply(Extensions.ToTimeSpan(self.resolution), self.longTermMomentumWindow) self.symbolData = {} self.rebalancingTimer = rebalancingTimer self.exchange = None def OnSecuritiesChanged(self, algorithm, changes): '''Event fired each time the we add/remove securities from the data feed. This initializes the MACD for each added security and cleans up the indicator for each removed security. Args: algorithm: The algorithm instance that experienced the change in securities changes: The security additions and removals from the algorithm''' for added in changes.AddedSecurities: self.symbolData[added.Symbol] = added self.exchange = added.Exchange for removed in changes.RemovedSecurities: data = self.symbolData.pop(removed.Symbol, None) def Slope(self,ts): ''' Args: Price time series. Returns: Annualized exponential regression slope ''' x = np.arange(len(ts)) log_ts = np.log(ts) slope, intercept, r_value, p_value, std_err = stats.linregress(x, log_ts) annualized_slope = (np.power(np.exp(slope), 250) - 1) * 100 return annualized_slope * (r_value ** 2) def Update(self, algorithm, data): ''' Determines an insight for each security based on two annualized slopes Args: algorithm: The algorithm instance data: The new data available Returns: The new insights generated''' insights = [] if not self.rebalancingTimer.isTimeToRebalance(algorithm,algorithm.Time, self.exchange): return insights # Check trend filter if enabled if self.indexAverageWindow>0: index_history = algorithm.History(Symbol.Create("SPY", SecurityType.Equity, Market.USA) , TimeSpan.FromDays(self.indexAverageWindow), self.resolution) index_sma = index_history['close'].mean() current_index = index_history['close'][-1] bull_market = current_index > index_sma if not bull_market: return insights for symbol,equity in self.symbolData.items(): if equity.Price == 0: continue shortTermBars = algorithm.History(symbol, TimeSpan.FromDays(self.shortTermMomentumWindow), self.resolution) longTermBars = algorithm.History(symbol, TimeSpan.FromDays(self.longTermMomentumWindow), self.resolution) shortTermSlope = self.Slope(shortTermBars['close']) longTermSlope = self.Slope(longTermBars['close']) meanMomentum = (shortTermSlope+longTermSlope)/2 if meanMomentum < self.minimumMomentum: continue insight = insights.append(Insight.Price(symbol, self.predictionInterval, InsightDirection.Up, meanMomentum, None)) return insights
class RebalancingTimer(PortfolioConstructionModel): def __init__(self,startTime, rebalanceFrequency): self.lastRebalance = startTime self.rebalanceFrequency = rebalanceFrequency self.timeToRebalance = True def isTimeToRebalance(self,algorithm,currentTime, securityExchange): if self.timeToRebalance and securityExchange.DateTimeIsOpen(currentTime): self.lastRebalance = currentTime self.timeToRebalance = False return True else: if "month" in self.rebalanceFrequency and self.lastRebalance.month != currentTime.month : self.timeToRebalance = True if "week" in self.rebalanceFrequency and self.lastRebalance.week != currentTime.week : self.timeToRebalance = True return False
# QUANTCONNECT.COM - Democratizing Finance, Empowering Individuals. # Lean Algorithmic Trading Engine v2.0. Copyright 2014 QuantConnect Corporation. # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. from clr import AddReference AddReference("QuantConnect.Common") AddReference("QuantConnect.Algorithm.Framework") from QuantConnect import Resolution, Extensions from QuantConnect.Algorithm.Framework.Alphas import * from QuantConnect.Algorithm.Framework.Portfolio import * from itertools import groupby from datetime import datetime, timedelta from pytz import utc UTCMIN = datetime.min.replace(tzinfo=utc) class VolatilityBasedPortfolioConstructionModel(PortfolioConstructionModel): '''Provides an implementation of IPortfolioConstructionModel that gives equal weighting to all securities. The target percent holdings of each security is 1/N where N is the number of securities. For insights of direction InsightDirection.Up, long targets are returned and for insights of direction InsightDirection.Down, short targets are returned.''' def __init__(self, resolution = Resolution.Daily): '''Initialize a new instance of EqualWeightingPortfolioConstructionModel Args: resolution: Rebalancing frequency''' self.insightCollection = InsightCollection() self.removedSymbols = [] self.nextExpiryTime = UTCMIN self.rebalancingTime = UTCMIN self.rebalancingPeriod = Extensions.ToTimeSpan(resolution) def CreateTargets(self, algorithm, insights): '''Create portfolio targets from the specified insights Args: algorithm: The algorithm instance insights: The insights to create portoflio targets from Returns: An enumerable of portoflio targets to be sent to the execution model''' targets = [] if (algorithm.UtcTime <= self.nextExpiryTime and algorithm.UtcTime <= self.rebalancingTime and len(insights) == 0 and self.removedSymbols is None): return targets self.insightCollection.AddRange(insights) # Create flatten target for each security that was removed from the universe if self.removedSymbols is not None: universeDeselectionTargets = [ PortfolioTarget(symbol, 0) for symbol in self.removedSymbols ] targets.extend(universeDeselectionTargets) self.removedSymbols = None # Get insight that haven't expired of each symbol that is still in the universe activeInsights = self.insightCollection.GetActiveInsights(algorithm.UtcTime) # Get the last generated active insight for each symbol lastActiveInsights = [] for symbol, g in groupby(activeInsights, lambda x: x.Symbol): lastActiveInsights.append(sorted(g, key = lambda x: x.GeneratedTimeUtc)[-1]) # give equal weighting to each security count = sum(x.Direction != InsightDirection.Flat for x in lastActiveInsights) percent = 0 if count == 0 else 1.0 / count errorSymbols = {} for insight in lastActiveInsights: target = PortfolioTarget.Percent(algorithm, insight.Symbol, insight.Direction * percent) if not target is None: targets.append(target) else: errorSymbols[insight.Symbol] = insight.Symbol # Get expired insights and create flatten targets for each symbol expiredInsights = self.insightCollection.RemoveExpiredInsights(algorithm.UtcTime) expiredTargets = [] for symbol, f in groupby(expiredInsights, lambda x: x.Symbol): if not self.insightCollection.HasActiveInsights(symbol, algorithm.UtcTime) and not symbol in errorSymbols: expiredTargets.append(PortfolioTarget(symbol, 0)) continue targets.extend(expiredTargets) self.nextExpiryTime = self.insightCollection.GetNextExpiryTime() if self.nextExpiryTime is None: self.nextExpiryTime = UTCMIN self.rebalancingTime = algorithm.UtcTime + self.rebalancingPeriod return targets def OnSecuritiesChanged(self, algorithm, changes): '''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''' # Get removed symbol and invalidate them in the insight collection self.removedSymbols = [x.Symbol for x in changes.RemovedSecurities] self.insightCollection.Clear(self.removedSymbols)