Overall Statistics |
Total Trades 69 Average Win 4.15% Average Loss -4.81% Compounding Annual Return -6.482% Drawdown 46.500% Expectancy -0.131 Net Profit -28.250% Sharpe Ratio -0.137 Probabilistic Sharpe Ratio 0.559% Loss Rate 53% Win Rate 47% Profit-Loss Ratio 0.86 Alpha -0.161 Beta 1.103 Annual Standard Deviation 0.253 Annual Variance 0.064 Information Ratio -0.727 Tracking Error 0.206 Treynor Ratio -0.031 Total Fees $406.97 |
from clr import AddReference AddReference("System") AddReference("QuantConnect.Common") AddReference("QuantConnect.Indicators") AddReference("QuantConnect.Algorithm.Framework") from QuantConnect.Data.UniverseSelection import * from Selection.FundamentalUniverseSelectionModel import FundamentalUniverseSelectionModel import numpy as np class SmallCapsLowPERatioUniverseSelectionModel(FundamentalUniverseSelectionModel): ''' Description: This Universe model selects Small Cap stocks with low P/E Ratio (in the 1st percentile) Details: The important thing to understand here is the internal flow of the Universe module: 1) SelectCoarse filters stocks with price above $5 2) SelectFine further filters those stocks by fundamental data. In this case, we use Market Cap and P/E Ratio ''' def __init__(self, filterFineData = True, universeSettings = None, securityInitializer = None): super().__init__(filterFineData, universeSettings, securityInitializer) self.periodCheck = -1 # initialize a variable to check when the period changes def SelectCoarse(self, algorithm, coarse): ''' Coarse selection based on price and volume ''' # this ensures the universe selection only runs once a year if algorithm.Time.year == self.periodCheck: return Universe.Unchanged self.periodCheck = algorithm.Time.year # securities must have fundamental data (to avoid ETFs) # securities must have last price above $5 filterCoarse = [x for x in coarse if x.HasFundamentalData and x.Price > 5] algorithm.Log('stocks with fundamental data and price above 5: ' + str(len(filterCoarse))) coarseSelection = [x.Symbol for x in filterCoarse] # return coarseSelection symbols ready for fundamental data filtering below return coarseSelection def SelectFine(self, algorithm, fine): ''' Fine selection based on fundamental data ''' filterFine = [] # select small caps only (market cap between $300 million and $2 billion) for x in fine: marketCap = (x.EarningReports.BasicAverageShares.ThreeMonths * x.EarningReports.BasicEPS.TwelveMonths * x.ValuationRatios.PERatio) if marketCap > 3e8 and marketCap < 2e9: filterFine.append(x) algorithm.Log('total number of small caps: ' + str(len(filterFine))) # now calculate the PE Ratio 1st percentile peRatios = [x.ValuationRatios.PERatio for x in filterFine] lowestPERatioPercentile = np.percentile(peRatios, 1) # filter stocks in the 1st PE Ratio percentile lowestPERatio = list(filter(lambda x: x.ValuationRatios.PERatio <= lowestPERatioPercentile, filterFine)) algorithm.Log('small caps in the 1st PE Ratio percentile: ' + str(len(lowestPERatio))) for x in lowestPERatio: algorithm.Log('stock: ' + str(x.Symbol.Value) + ', current PE Ratio: ' + str(x.ValuationRatios.PERatio)) fineSelection = [x.Symbol for x in lowestPERatio] # return fineSelection ready for Alpha module return fineSelection
### PRODUCT INFORMATION -------------------------------------------------------------------------------- # Copyright InnoQuantivity.com, granted to the public domain. # Use entirely at your own risk. # This algorithm contains open source code from other sources and no claim is being made to such code. # Do not remove this copyright notice. ### ---------------------------------------------------------------------------------------------------- from SmallCapsLowPERatioUniverseSelection import SmallCapsLowPERatioUniverseSelectionModel from LongOnlyConstantAlphaCreation import LongOnlyConstantAlphaCreationModel from CustomEqualWeightingPortfolioConstruction import CustomEqualWeightingPortfolioConstructionModel class LongOnlySmallCapsLowPERatioFrameworkAlgorithm(QCAlgorithmFramework): ''' Trading Logic: This algorithm buys at the start of every year Small Caps with low P/E Ratio Universe: Dynamically selects stocks at the start of each year based on: - Price above $5 - Small Caps (Market Cap between $300 million and $2 billion) - Then select stocks in the 1st percentile of Price To Earnings Ratio (PE Ratio) Alpha: Constant creation of Up Insights every trading bar Portfolio: Equal Weighting (allocate equal amounts of portfolio % to each security) - To rebalance the portfolio periodically to ensure equal weighting, change the rebalancingParam below Execution: Immediate Execution with Market Orders Risk: Null ''' def Initialize(self): ### user-defined inputs -------------------------------------------------------------- self.SetStartDate(2015, 1, 1) # set start date #self.SetEndDate(2019, 1, 4) # set end date self.SetCash(100000) # set strategy cash # True/False to enable/disable filtering by fundamental data filterFineData = True # rebalancing period (to enable rebalancing enter an integer for number of days, e.g. 1, 7, 30, 365) rebalancingParam = False ### ----------------------------------------------------------------------------------- # set the brokerage model for slippage and fees self.SetSecurityInitializer(self.CustomSecurityInitializer) self.SetBrokerageModel(AlphaStreamsBrokerageModel()) # set requested data resolution and disable fill forward data self.UniverseSettings.Resolution = Resolution.Daily self.UniverseSettings.FillForward = False # select modules self.SetUniverseSelection(SmallCapsLowPERatioUniverseSelectionModel(filterFineData = filterFineData)) self.SetAlpha(LongOnlyConstantAlphaCreationModel()) self.SetPortfolioConstruction(CustomEqualWeightingPortfolioConstructionModel(rebalancingParam = rebalancingParam)) self.SetExecution(ImmediateExecutionModel()) self.SetRiskManagement(NullRiskManagementModel()) def CustomSecurityInitializer(self, security): ''' Description: Initialize the security with adjusted prices Args: security: Security which characteristics we want to change ''' security.SetDataNormalizationMode(DataNormalizationMode.Adjusted)
from clr import AddReference AddReference("System") AddReference("QuantConnect.Common") AddReference("QuantConnect.Algorithm") AddReference("QuantConnect.Algorithm.Framework") from System import * from QuantConnect import * from QuantConnect.Algorithm import * from QuantConnect.Algorithm.Framework import * from QuantConnect.Algorithm.Framework.Alphas import AlphaModel, Insight, InsightType, InsightDirection class LongOnlyConstantAlphaCreationModel(AlphaModel): ''' Description: This Alpha model creates InsightDirection.Up (to go Long) for a duration of 1 day, every day for all active securities in our Universe Details: The important thing to understand here is the concept of Insight: - A prediction about the future of the security, indicating an expected Up, Down or Flat move - This prediction has an expiration time/date, meaning we think the insight holds for some amount of time - In the case of a constant long-only strategy, we are just updating every day the Up prediction for another extra day - In other words, every day we are making the conscious decision of staying invested in the security one more day ''' def __init__(self, resolution = Resolution.Daily): self.insightExpiry = Time.Multiply(Extensions.ToTimeSpan(resolution), 0.25) # insight duration self.insightDirection = InsightDirection.Up # insight direction self.securities = [] # list to store securities to consider def Update(self, algorithm, data): insights = [] # list to store the new insights to be created # loop through securities and generate insights for security in self.securities: # check if there's new data for the security or we're already invested # if there's no new data but we're invested, we keep updating the insight since we don't really need to place orders if data.ContainsKey(security.Symbol) or algorithm.Portfolio[security.Symbol].Invested: # append the insights list with the prediction for each symbol insights.append(Insight.Price(security.Symbol, self.insightExpiry, self.insightDirection)) else: algorithm.Log('excluding this security due to missing data: ' + str(security.Symbol.Value)) return insights def OnSecuritiesChanged(self, algorithm, changes): ''' Description: 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 ''' # add new securities for added in changes.AddedSecurities: self.securities.append(added) # remove securities for removed in changes.RemovedSecurities: if removed in self.securities: self.securities.remove(removed)
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 CustomEqualWeightingPortfolioConstructionModel(PortfolioConstructionModel): ''' Description: Provide a custom implementation of IPortfolioConstructionModel that gives equal weighting to all active securities Details: - The target percent holdings of each security is 1/N where N is the number of securities with active Up/Down insights - For InsightDirection.Up, long targets are returned - For InsightDirection.Down, short targets are returned - For InsightDirection.Flat, closing position targets are returned ''' def __init__(self, rebalancingParam = False): ''' Description: Initialize a new instance of CustomEqualWeightingPortfolioConstructionModel Args: rebalancingParam: Integer indicating the number of days for rebalancing (default set to False, no rebalance) - Independent of this parameter, the portfolio will be rebalanced when a security is added/removed/changed direction ''' self.insightCollection = InsightCollection() self.removedSymbols = [] self.nextExpiryTime = UTCMIN self.rebalancingTime = UTCMIN # if the rebalancing parameter is not False but a positive integer # convert rebalancingParam to timedelta and create rebalancingFunc if rebalancingParam > 0: self.rebalancing = True rebalancingParam = timedelta(days = rebalancingParam) self.rebalancingFunc = lambda dt: dt + rebalancingParam else: self.rebalancing = rebalancingParam def CreateTargets(self, algorithm, insights): ''' Description: Create portfolio targets from the specified insights Args: algorithm: The algorithm instance insights: The insights to create portfolio targets from Returns: An enumerable of portfolio targets to be sent to the execution model ''' targets = [] # check if we have new insights coming from the alpha model or if some existing insights have expired # or if we have removed symbols from the universe if (len(insights) == 0 and algorithm.UtcTime <= self.nextExpiryTime and self.removedSymbols is None): return targets # here we get the new insights and add them to our insight collection for insight in insights: self.insightCollection.Add(insight) # 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]) # determine target percent for the given insights (check function DetermineTargetPercent for details) percents = self.DetermineTargetPercent(lastActiveInsights) errorSymbols = {} # check if we actually want to create new targets for the securities (check function ShouldCreateTargets for details) if self.ShouldCreateTargets(algorithm, lastActiveInsights): for insight in lastActiveInsights: target = PortfolioTarget.Percent(algorithm, insight.Symbol, percents[insight]) if not target is None: targets.append(target) else: errorSymbols[insight.Symbol] = insight.Symbol # update rebalancing time if self.rebalancing: self.rebalancingTime = self.rebalancingFunc(algorithm.UtcTime) # 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) # here we update the next expiry date in the insight collection self.nextExpiryTime = self.insightCollection.GetNextExpiryTime() if self.nextExpiryTime is None: self.nextExpiryTime = UTCMIN return targets def DetermineTargetPercent(self, lastActiveInsights): ''' Description: Determine the target percent from each insight Args: lastActiveInsights: The active insights to generate a target from ''' result = {} # 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 for insight in lastActiveInsights: result[insight] = insight.Direction * percent return result def ShouldCreateTargets(self, algorithm, lastActiveInsights): ''' Description: Determine whether we should rebalance the portfolio to keep equal weighting when: - It is time to rebalance regardless - We want to include some new security in the portfolio - We want to modify the direction of some existing security Args: lastActiveInsights: The last active insights to check ''' # it is time to rebalance if self.rebalancing and algorithm.UtcTime >= self.rebalancingTime: return True for insight in lastActiveInsights: # if there is an insight for a new security that's not invested, then rebalance if not algorithm.Portfolio[insight.Symbol].Invested and insight.Direction != InsightDirection.Flat: return True # if there is an insight to close a long position, then rebalance elif algorithm.Portfolio[insight.Symbol].IsLong and insight.Direction != InsightDirection.Up: return True # if there is an insight to close a short position, then rebalance elif algorithm.Portfolio[insight.Symbol].IsShort and insight.Direction != InsightDirection.Down: return True else: continue return False def OnSecuritiesChanged(self, algorithm, changes): ''' Description: 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)