Overall Statistics |

Total Trades 19 Average Win 6.12% Average Loss -3.47% Compounding Annual Return 22.603% Drawdown 29.600% Expectancy 1.490 Net Profit 177.484% Sharpe Ratio 0.973 Probabilistic Sharpe Ratio 47.545% Loss Rate 10% Win Rate 90% Profit-Loss Ratio 1.77 Alpha 0.086 Beta 1.215 Annual Standard Deviation 0.239 Annual Variance 0.057 Information Ratio 0.633 Tracking Error 0.177 Treynor Ratio 0.191 Total Fees $24.74 |

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 BuyAndHoldAlphaCreationModel(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 Buy and Hold 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('(Alpha) 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) import numpy as np import pandas as pd from scipy.optimize import minimize class CustomOptimizationPortfolioConstructionModel(PortfolioConstructionModel): ''' Description: Allocate optimal weights to each security in order to optimize the portfolio objective function provided 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, objectiveFunction = 'std', rebalancingParam = False): ''' Description: Initialize a new instance of CustomOptimizationPortfolioConstructionModel Args: objectiveFunction: The function to optimize. If set to 'equal', it will just perform equal weighting 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 ''' if objectiveFunction != 'equal': # minWeight set to 0 to ensure long only weights self.optimizer = CustomPortfolioOptimizer(minWeight = 0, maxWeight = 1, objFunction = objectiveFunction) # initialize the optimizer self.optWeights = None self.objectiveFunction = objectiveFunction 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]) # check if we actually want to create new targets for the securities (check function ShouldCreateTargets for details) if self.ShouldCreateTargets(algorithm, self.optWeights, lastActiveInsights): # symbols with active insights lastActiveSymbols = [x.Symbol for x in lastActiveInsights] # get historical data for all symbols for the last 253 trading days (to get 252 returns) history = algorithm.History(lastActiveSymbols, 253, Resolution.Daily) # empty dictionary for calculations calculations = {} # iterate over all symbols and perform calculations for symbol in lastActiveSymbols: if (str(symbol) not in history.index or history.loc[str(symbol)].get('close') is None or history.loc[str(symbol)].get('close').isna().any()): algorithm.Log('(Portfolio) no historical data for: ' + str(symbol.Value)) continue else: # add symbol to calculations calculations[symbol] = SymbolData(symbol) try: # get series of log-returns calculations[symbol].CalculateLogReturnSeries(history) except Exception: algorithm.Log('(Portfolio) removing from calculations due to CalculateLogReturnSeries failing: ' + str(symbol.Value)) calculations.pop(symbol) continue # determine target percent for the given insights (check function DetermineTargetPercent for details) self.optWeights = self.DetermineTargetPercent(calculations, lastActiveInsights) if not self.optWeights.isnull().values.any(): algorithm.Log('(Portfolio) optimal weights: ' + str(self.optWeights)) errorSymbols = {} for symbol in lastActiveSymbols: if str(symbol) in self.optWeights: # avoid very small numbers and make them 0 if self.optWeights[str(symbol)] <= 1e-10: self.optWeights[str(symbol)] = 0 algorithm.Plot('Optimal Allocation', symbol.Value, float(self.optWeights[str(symbol)])) target = PortfolioTarget.Percent(algorithm, symbol, self.optWeights[str(symbol)]) if not target is None: targets.append(target) else: errorSymbols[symbol] = 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 ShouldCreateTargets(self, algorithm, optWeights, 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: optWeights: Series containing the current optimal weight for each security 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 and it has no existing optimal weight, then rebalance if (not algorithm.Portfolio[insight.Symbol].Invested and insight.Direction != InsightDirection.Flat and str(insight.Symbol) not in optWeights): 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 DetermineTargetPercent(self, calculations, lastActiveInsights): ''' Description: Determine the target percent for each symbol provided Args: calculations: Dictionary with calculations for symbols lastActiveInsights: Dictionary with calculations for symbols ''' if self.objectiveFunction == 'equal': # 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 result = {} for insight in lastActiveInsights: result[str(insight.Symbol)] = insight.Direction * percent weights = pd.Series(result) return weights else: # create a dictionary keyed by the symbols in calculations with a pandas.Series as value to create a dataframe of log-returns logReturnsDict = { str(symbol): symbolData.logReturnSeries for symbol, symbolData in calculations.items() } logReturnsDf = pd.DataFrame(logReturnsDict) # portfolio optimizer finds the optimal weights for the given data weights = self.optimizer.Optimize(historicalLogReturns = logReturnsDf) weights = pd.Series(weights, index = logReturnsDf.columns) return weights 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) class SymbolData: ''' Contain data specific to a symbol required by this model ''' def __init__(self, symbol): self.Symbol = symbol self.logReturnSeries = None def CalculateLogReturnSeries(self, history): ''' Calculate the log-returns series for each security ''' self.logReturnSeries = np.log(1 + history.loc[str(self.Symbol)]['close'].pct_change(periods = 1).dropna()) # 1-day log-returns ### class containing the CustomPortfolioOptimizer ----------------------------------------------------------------------------------------- class CustomPortfolioOptimizer: ''' Description: Implementation of a custom optimizer that calculates the weights for each asset to optimize a given objective function Details: Optimization can be: - Maximize Portfolio Return - Minimize Portfolio Standard Deviation - Maximize Portfolio Sharpe Ratio Constraints: - Weights must be between some given boundaries - Weights must sum to 1 ''' def __init__(self, minWeight = -1, maxWeight = 1, objFunction = 'std'): ''' Description: Initialize the CustomPortfolioOptimizer Args: minWeight(float): The lower bound on portfolio weights maxWeight(float): The upper bound on portfolio weights objFunction: The objective function to optimize (return, std, sharpe) ''' self.minWeight = minWeight self.maxWeight = maxWeight self.objFunction = objFunction def Optimize(self, historicalLogReturns, covariance = None): ''' Description: Perform portfolio optimization using a provided matrix of historical returns and covariance (optional) Args: historicalLogReturns: Matrix of historical log-returns where each column represents a security and each row log-returns for the given date/time (size: K x N) covariance: Multi-dimensional array of double with the portfolio covariance of returns (size: K x K) Returns: Array of double with the portfolio weights (size: K x 1) ''' # if no covariance is provided, calculate it using the historicalLogReturns if covariance is None: covariance = historicalLogReturns.cov() size = historicalLogReturns.columns.size # K x 1 x0 = np.array(size * [1. / size]) # apply equality constraints constraints = ({'type': 'eq', 'fun': lambda weights: self.GetBudgetConstraint(weights)}) opt = minimize(lambda weights: self.ObjectiveFunction(weights, historicalLogReturns, covariance), # Objective function x0, # Initial guess bounds = self.GetBoundaryConditions(size), # Bounds for variables constraints = constraints, # Constraints definition method = 'SLSQP') # Optimization method: Sequential Least Squares Programming return opt['x'] def ObjectiveFunction(self, weights, historicalLogReturns, covariance): ''' Description: Compute the objective function Args: weights: Portfolio weights historicalLogReturns: Matrix of historical log-returns covariance: Covariance matrix of historical log-returns ''' # calculate the annual return of portfolio annualizedPortfolioReturns = np.sum(historicalLogReturns.mean() * 252 * weights) # calculate the annual standard deviation of portfolio annualizedPortfolioStd = np.sqrt( np.dot(weights.T, np.dot(covariance * 252, weights)) ) if annualizedPortfolioStd == 0: raise ValueError(f'CustomPortfolioOptimizer.ObjectiveFunction: annualizedPortfolioStd cannot be zero. Weights: {weights}') # calculate annual sharpe ratio of portfolio annualizedPortfolioSharpeRatio = (annualizedPortfolioReturns / annualizedPortfolioStd) if self.objFunction == 'sharpe': return -annualizedPortfolioSharpeRatio # convert to negative to be minimized elif self.objFunction == 'return': return -annualizedPortfolioReturns # convert to negative to be minimized elif self.objFunction == 'std': return annualizedPortfolioStd else: raise ValueError(f'CustomPortfolioOptimizer.ObjectiveFunction: objFunction input has to be one of sharpe, return or std') def GetBoundaryConditions(self, size): ''' Create the boundary condition for the portfolio weights ''' return tuple((self.minWeight, self.maxWeight) for x in range(size)) def GetBudgetConstraint(self, weights): ''' Define a budget constraint: the sum of the weights equal to 1 ''' return np.sum(weights) - 1

### 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 BuyAndHoldAlphaCreation import BuyAndHoldAlphaCreationModel from CustomOptimizationPortfolioConstruction import CustomOptimizationPortfolioConstructionModel class BuyAndHoldFrameworkAlgorithm(QCAlgorithmFramework): ''' Trading Logic: This algorithm buys and holds the provided tickers from the start date until the end date Modules: Universe: Manual input of tickers Alpha: Constant creation of Up Insights every trading bar Portfolio: A choice between Equal Weighting, Maximize Portfolio Return, Minimize Portfolio Standard Deviation or Maximize Portfolio Sharpe Ratio - If some of the tickers did not exist at the start date, it will buy them when they first appeared in the market, in which case it will sell part of the existing securities in order to buy the new ones keeping an equally weighted portfolio - To rebalance the portfolio periodically to ensure optimal allocation, 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, 2, 1) # set end date self.SetCash(100000) # set strategy cash # add tickers to the list tickers = ['FB', 'AMZN', 'NFLX', 'GOOG'] # objective function for portfolio optimizer # options are: equal (Equal Weighting), return (Maximize Portfolio Return), std (Minimize Portfolio Standard Deviation), # and sharpe (Maximize Portfolio Sharpe Ratio) objectiveFunction = 'std' # rebalancing period (to enable rebalancing enter an integer for number of calendar days, e.g. 1, 7, 30, 365) rebalancingParam = 365 ### ----------------------------------------------------------------------------------- # 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 # initialize plot for optimal allocation allocationPlot = Chart('Optimal Allocation') symbols = [] # loop through the tickers list and create symbols for the universe for i in range(len(tickers)): symbols.append(Symbol.Create(tickers[i], SecurityType.Equity, Market.USA)) allocationPlot.AddSeries(Series(tickers[i], SeriesType.Line, '')) self.AddChart(allocationPlot) # select modules self.SetUniverseSelection(ManualUniverseSelectionModel(symbols)) self.SetAlpha(BuyAndHoldAlphaCreationModel()) self.SetPortfolioConstruction(CustomOptimizationPortfolioConstructionModel(objectiveFunction = objectiveFunction, 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)