| Overall Statistics |
|
Total Trades 727 Average Win 0.78% Average Loss -0.53% Compounding Annual Return 7.056% Drawdown 8.700% Expectancy 0.506 Net Profit 183.381% Sharpe Ratio 0.8 Probabilistic Sharpe Ratio 16.948% Loss Rate 39% Win Rate 61% Profit-Loss Ratio 1.46 Alpha 0.06 Beta 0.006 Annual Standard Deviation 0.076 Annual Variance 0.006 Information Ratio -0.086 Tracking Error 0.191 Treynor Ratio 10.808 Total Fees $2951.04 |
### 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 LongOnlyMomentumAlphaCreation import LongOnlyMomentumAlphaCreationModel
from OptimizationPortfolioConstruction import OptimizationPortfolioConstructionModel
from ImmediateExecutionWithLogs import ImmediateExecutionWithLogsModel
class KDAAssetAllocationFrameworkAlgorithm(QCAlgorithmFramework):
'''
Trading Logic:
- Implementation of https://quantstrattrader.wordpress.com/2019/01/24/right-now-its-kda-asset-allocation/
- This algorithm is a long-only dual momentum asset class strategy as described in the link above
Modules:
Universe: Manual input of tickers
Alpha:
- Calculates momentum score for each security at the end of every period (see Alpha module for details)
- Constant creation of Up Insights every trading bar during the period for the top securities
Portfolio: Minimum Variance (optimal weights to minimize portfolio variance)
- See Portfolio module for details
Execution: Immediate Execution with Market Orders
Risk: Null
'''
def Initialize(self):
### user-defined inputs --------------------------------------------------------------
# set timeframe for backtest and starting cash
self.SetStartDate(2005, 1, 1) # set start date
#self.SetEndDate(2016, 1, 1) # set end date
self.SetCash(100000) # set strategy cash
# add tickers for risky assets for momentum asset allocation
riskyTickers = ['SPY', # US equities
'VGK', # European equities
'EWJ', # Japanese equities
'EEM', # Emerging market equities
'VNQ', # US REITs
'RWX', # International REITs
'TLT', # US 30-year Treasuries
'DBC', # Commodities
'GLD', # Gold
]
# this ticker will also be part of risky assets for momentum asset allocation,
# but it's a special asset that will get the crash protection allocation when needed
crashProtectionTicker = ['IEF'] # US 10-year Treasuries
# add tickers for canary assets
canaryTickers = ["VWO", # Vanguard FTSE Emerging Markets ETF
"BND" # Vanguard Total Bond Market ETF
]
# number of top momentum securities to keep
topMomentum = 5
# select the logic for rebalancing period
# options are:
# - Date rules (for the first trading day of period): Expiry.EndOfDay, Expiry.EndOfWeek, Expiry.EndOfMonth, Expiry.EndOfQuarter, Expiry.EndOfYear
rebalancingPeriod = Expiry.EndOfMonth
# objective function for portfolio optimizer
# options are: return (maximize portfolio return), std (minimize portfolio Standard Deviation) and sharpe (maximize portfolio sharpe ratio)
objectiveFunction = 'std'
### -----------------------------------------------------------------------------------
# set the brokerage model for slippage and fees
self.SetBrokerageModel(AlphaStreamsBrokerageModel())
# set requested data resolution and disable fill forward data
self.UniverseSettings.Resolution = Resolution.Daily
# combine all lists of tickers
allTickers = riskyTickers + crashProtectionTicker + canaryTickers
# let's plot the series of optimal weights
optWeightsPlot = Chart('Chart Optimal Weights %')
symbols = []
# loop through the tickers list and create symbols for the universe
for i in range(len(allTickers)):
symbols.append(Symbol.Create(allTickers[i], SecurityType.Equity, Market.USA))
optWeightsPlot.AddSeries(Series(allTickers[i], SeriesType.Line, '%'))
self.AddChart(optWeightsPlot)
# select modules
self.SetUniverseSelection(ManualUniverseSelectionModel(symbols))
self.SetAlpha(LongOnlyMomentumAlphaCreationModel(riskyTickers = riskyTickers,
crashProtectionTicker = crashProtectionTicker,
canaryTickers = canaryTickers,
topMomentum = topMomentum,
rebalancingPeriod = rebalancingPeriod))
self.SetPortfolioConstruction(OptimizationPortfolioConstructionModel(crashProtectionTicker = crashProtectionTicker,
canaryTickers = canaryTickers,
topMomentum = topMomentum,
objectiveFunction = objectiveFunction,
rebalancingPeriod = rebalancingPeriod))
self.SetExecution(ImmediateExecutionWithLogsModel())
self.SetRiskManagement(NullRiskManagementModel())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)
UTCMAX = datetime.max.replace(tzinfo=utc)
from optimizer import CustomPortfolioOptimizer
import numpy as np
import pandas as pd
class OptimizationPortfolioConstructionModel(PortfolioConstructionModel):
'''
Description:
Allocate optimal weights to each security in order to optimize the portfolio objective function provided
Details:
- Two Canary Assets determine how much to invest in Risky Assets:
If both assets have positive absolute momentum => 100%
If only one has positive absolute momentum => 50%
If none have positive absolute momentum => 0%
* The remaining % from the above calculation will go to the Crash Protection Ticker, only if it has positive absolute momentum
- To calculate the weights for risky assets, we perform portfolio optimization with the following particularity:
We construct the correlation matrix using a 1-3-6-12 momentum weighting:
( last month correlation * 12 + last 3-month correlation * 4 + last 6-month correlation * 2 + last 12-month correlation ) / 19
'''
def __init__(self, crashProtectionTicker, canaryTickers, topMomentum, objectiveFunction = 'std', rebalancingPeriod = Expiry.EndOfMonth):
self.crashProtectionTicker = crashProtectionTicker # this ticker will also be part of momentum asset allocation,
# but it's a special asset that will get the crash protection allocation when needed
self.canaryTickers = canaryTickers # canary tickers to avoid in momentum calculations, but we need to subscribe to them
self.topMomentum = topMomentum # number of top momentum securities to keep
self.rebalancingPeriod = rebalancingPeriod # the rebalancing function
self.optimizer = CustomPortfolioOptimizer(minWeight = 0, maxWeight = 1, objFunction = objectiveFunction) # initialize the optimizer
self.insightCollection = InsightCollection()
self.nextExpiryTime = UTCMAX
self.rebalancingTime = None
def CreateTargets(self, algorithm, insights):
'''
Description:
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
'''
if self.rebalancingTime is None:
# get next rebalancing time
self.rebalancingTime = self.rebalancingPeriod(algorithm.Time)
targets = []
# check if we have new insights coming from the alpha model or if some existing insights have expired
if len(insights) == 0 and algorithm.UtcTime <= self.nextExpiryTime:
return targets
# here we get the new insights and add them to our insight collection
for insight in insights:
self.insightCollection.Add(insight)
# 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])
# symbols with active insights
lastActiveSymbols = [x.Symbol for x in lastActiveInsights]
### calculate targets -------------------------------------------------------------------------------------
if self.ShouldCreateTargets(algorithm, lastActiveSymbols):
algorithm.Log('(Portfolio) time to calculate the targets')
algorithm.Log('(Portfolio) number of active insights: ' + str(len(lastActiveSymbols)))
### get symbols ---------------------------------------------------------------------------------------
# top momentum symbols
topMomentumSymbols = [x.Symbol for x in lastActiveInsights]
# crash protection symbol
crashProtectionSymbol = [x.Symbol for x in algorithm.ActiveSecurities.Values if x.Symbol.Value in self.crashProtectionTicker]
# if active symbols are more than topMomentum, we need to remove the crashProtectionSymbol from topMomentumSymbols
if len(lastActiveSymbols) > self.topMomentum:
if crashProtectionSymbol[0] in topMomentumSymbols:
topMomentumSymbols.remove(crashProtectionSymbol[0])
else:
algorithm.Log('(Portfolio) lastActiveSymbols is bigger than topMomentum, but crashProtectionSymbol is not in topMomentumSymbols!')
# canary symbols
canarySymbols = [x.Symbol for x in algorithm.ActiveSecurities.Values if x.Symbol.Value in self.canaryTickers]
# combine the two lists to get all symbols for calculations
allSymbols = topMomentumSymbols + canarySymbols
### ----------------------------------------------------------------------------------------------------
# get historical data for all symbols
history = algorithm.History(allSymbols, 253, Resolution.Daily)
# empty dictionary for calculations
calculations = {}
# iterate over all symbols and perform calculations
for symbol in allSymbols:
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))
if symbol in lastActiveSymbols:
lastActiveSymbols.remove(symbol)
continue
else:
# add symbol to calculations
calculations[symbol] = SymbolData(symbol)
try:
# get momentum score
calculations[symbol].CalculateMomentumScore(history)
except Exception:
algorithm.Log('(Portfolio) removing from Portfolio calculations due to CalculateMomentumScore failing')
calculations.pop(symbol)
if symbol in lastActiveSymbols:
lastActiveSymbols.remove(symbol)
continue
# calculate the percentage of aggressive allocation for risky assets
pctAggressive = self.CalculatePctAggressive(calculations, canarySymbols)
algorithm.Log('(Portfolio) pctAggressive: ' + str(pctAggressive))
# calculate optimal weights
optWeights = self.DetermineTargetPercent(calculations, topMomentumSymbols, crashProtectionSymbol)
if not optWeights.isnull().values.any():
algorithm.Log('(Portfolio) optimal weights: ' + str(optWeights))
# apply percentage aggressive to the weights to get final weights
finalWeights = optWeights * pctAggressive
algorithm.Log('(Portfolio) final weights: ' + str(finalWeights))
# iterate over active symbols and create targets
for symbol in lastActiveSymbols:
# we allocate the rest to the crash protection asset
if symbol in crashProtectionSymbol and pctAggressive < 1:
finalWeights[str(symbol)] = finalWeights[str(symbol)] + (1 - pctAggressive)
algorithm.Log('(Portfolio) adding ' + str(1 - pctAggressive) + ' extra weight for ' + str(symbol.Value)
+ '; final weight: ' + str(finalWeights[str(symbol)]))
weight = finalWeights[str(symbol)]
target = PortfolioTarget.Percent(algorithm, symbol, weight)
algorithm.Plot('Chart Optimal Weights %', symbol.Value, float(finalWeights[str(symbol)]))
if not target is None:
targets.append(target)
### end of calculations --------------------------------------------------------------------------------
# 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):
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, lastActiveSymbols):
'''
Description:
Determine whether we should create new portfolio targets when:
- It's time to rebalance and there are active insights
Args:
lastActiveSymbols: Symbols for the last active securities
'''
if algorithm.Time >= self.rebalancingTime and len(lastActiveSymbols) > 0:
# update rebalancing time
self.rebalancingTime = self.rebalancingPeriod(algorithm.Time)
return True
else:
return False
def CalculatePctAggressive(self, calculations, canarySymbols):
'''
Description:
Calculate the percentage dedicated to risky assets
Args:
calculations: Dictionary with calculations for symbols
canarySymbols: Symbols for the canary assets
Returns:
Float with the percentage dedicated to risky assets
'''
# get a list with the canary assets that have positive absolute momentum
canaryPosMomList = list(filter(lambda x: x.momentumScore > 0 and x.Symbol in canarySymbols, calculations.values()))
# get the average positive (basically the options are 0, 0.5 or 1)
# this will be the allocation for risky assets
pctAggressive = len(canaryPosMomList) / len(canarySymbols)
return pctAggressive
def DetermineTargetPercent(self, calculations, topMomentumSymbols, crashProtectionSymbol):
'''
Description:
Determine the target percent for each symbol provided
Args:
calculations: Dictionary with calculations for symbols
topMomentumSymbols: Symbols for the top momentum assets
crashProtectionSymbol: Symbol for the crash protection asset
Returns:
Pandas series with the optimal weights for each symbol
'''
# 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): np.log(1 + symbolData.returnSeries) for symbol, symbolData in calculations.items() if symbol in topMomentumSymbols }
logReturnsDf = pd.DataFrame(logReturnsDict)
# create correlation matrix with 1-3-6-12 momentum weighting
corrMatrix = ( logReturnsDf.tail(21).corr() * 12 + logReturnsDf.tail(63).corr() * 4 + logReturnsDf.tail(126).corr() * 2 + logReturnsDf.tail(252).corr() ) / 19
# create standard deviation matrix using the 1-month standard deviation of returns
stdMatrix = pd.DataFrame(logReturnsDf.tail(21).std()) # column vector (one row per symbol and one single column with the standard deviation)
# get its transpose
stdMatrixTranspose = stdMatrix.T # row vector (one single row with the standard deviation and one column per symbol)
# compute the dot product between stdMatrix and its transpose to get the volatility matrix
volMatrix = stdMatrix.dot(stdMatrixTranspose) # square NxN matrix with the variances of each symbol on the diagonal and the product of stds on the off diagonal
# calculate the covariance matrix by doing element-wise multiplication of correlation matrix and volatility matrix
covMatrix = corrMatrix.multiply(volMatrix)
# portfolio optimizer finds the optimal weights for the given data
weights = self.optimizer.Optimize(historicalLogReturns = logReturnsDf, covariance = covMatrix)
weights = pd.Series(weights, index = logReturnsDf.columns)
# avoid very small numbers and make them 0
for symbol, weight in weights.items():
if weight <= 1e-10:
weights[str(symbol)] = 0
# add crashProtectionSymbol to the finalWeights series with 0 if not already there
if str(crashProtectionSymbol[0]) not in weights:
weights[str(crashProtectionSymbol[0])] = 0
return weights
class SymbolData:
''' Contain data specific to a symbol required by this model '''
def __init__(self, symbol):
self.Symbol = symbol
self.returnSeries = None
self.momentumScore = None
def CalculateMomentumScore(self, history):
''' Calculate the weighted average momentum score for each security '''
self.returnSeries = history.loc[str(self.Symbol)]['close'].pct_change(periods = 1).dropna() # 1-day returns for last year
cumRet1 = (self.returnSeries.tail(21).add(1).prod()) - 1 # 1-month momentum
cumRet3 = (self.returnSeries.tail(63).add(1).prod()) - 1 # 3-month momentum
cumRet6 = (self.returnSeries.tail(126).add(1).prod()) - 1 # 6-month momentum
cumRet12 = (self.returnSeries.tail(252).add(1).prod()) - 1 # 12-month momentum
self.momentumScore = (cumRet1 * 12 + cumRet3 * 4 + cumRet6 * 2 + cumRet12) # weighted average momentumimport numpy as np
from scipy.optimize import minimize
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) - 1from clr import AddReference
AddReference("System")
AddReference("QuantConnect.Common")
AddReference("QuantConnect.Algorithm")
AddReference("QuantConnect.Algorithm.Framework")
from System import *
from QuantConnect import *
from QuantConnect.Orders import *
from QuantConnect.Algorithm import *
from QuantConnect.Algorithm.Framework import *
from QuantConnect.Algorithm.Framework.Execution import *
from QuantConnect.Algorithm.Framework.Portfolio import *
import numpy as np
class ImmediateExecutionWithLogsModel(ExecutionModel):
'''
Description:
Custom implementation of IExecutionModel that immediately submits market orders to achieve the desired portfolio targets
Details:
This custom implementation includes logs with information about number of shares traded, prices, profit and profit percent
for both long and short positions.
'''
def __init__(self):
''' Initializes a new instance of the ImmediateExecutionModel class '''
self.targetsCollection = PortfolioTargetCollection()
def Execute(self, algorithm, targets):
'''
Description:
Immediately submits orders for the specified portfolio targets
Args:
algorithm: The algorithm instance
targets: The portfolio targets to be ordered
'''
self.targetsCollection.AddRange(targets)
if self.targetsCollection.Count > 0:
for target in self.targetsCollection.OrderByMarginImpact(algorithm):
# calculate remaining quantity to be ordered (this could be positive or negative)
unorderedQuantity = OrderSizing.GetUnorderedQuantity(algorithm, target)
# calculate the lot size for the security
lotSize = algorithm.ActiveSecurities[target.Symbol].SymbolProperties.LotSize
# this is the size of the order in terms of absolute number of shares
orderSize = abs(unorderedQuantity)
remainder = orderSize % lotSize
missingForLotSize = lotSize - remainder
# if the amount we are missing for +1 lot size is 1M part of a lot size
# we suppose its due to floating point error and round up
# Note: this is required to avoid a diff with C# equivalent
if missingForLotSize < (lotSize / 1000000):
remainder -= lotSize
# round down to even lot size
orderSize -= remainder
quantity = np.sign(unorderedQuantity) * orderSize
# check if quantity is greater than 1 share (in absolute value to account for shorts)
if quantity != 0:
# get the current holdings quantity, average price and cost
beforeHoldingsQuantity = algorithm.ActiveSecurities[target.Symbol].Holdings.Quantity
beforeHoldingsAvgPrice = algorithm.ActiveSecurities[target.Symbol].Holdings.AveragePrice
beforeHoldingsCost = algorithm.ActiveSecurities[target.Symbol].Holdings.HoldingsCost
# place market order
algorithm.MarketOrder(target.Symbol, quantity)
# get the new holdings quantity, average price and cost
newHoldingsQuantity = beforeHoldingsQuantity + quantity
newHoldingsAvgPrice = algorithm.ActiveSecurities[target.Symbol].Holdings.AveragePrice
newHoldingsCost = algorithm.ActiveSecurities[target.Symbol].Holdings.HoldingsCost
# this is just for market on open orders because the avg price and cost won't update until order gets filled
# so to avoid getting previous values we just make them zero
if newHoldingsAvgPrice == beforeHoldingsAvgPrice and newHoldingsCost == beforeHoldingsCost:
newHoldingsAvgPrice = 0
newHoldingsCost = 0
# calculate the profit percent and dollar profit when closing positions
lastPrice = algorithm.ActiveSecurities[target.Symbol].Price
if beforeHoldingsAvgPrice != 0 and lastPrice != 0:
# profit/loss percent for the trade
tradeProfitPercent = (((lastPrice / beforeHoldingsAvgPrice) - 1) * np.sign(beforeHoldingsQuantity)) * 100
# dollar profit/loss for the trade (when partially or entirely closing a position)
tradeDollarProfit = (lastPrice - beforeHoldingsAvgPrice) * (abs(quantity) * np.sign(beforeHoldingsQuantity))
# dollar profit/loss for the trade (when reversing a position from long/short to short/long)
tradeDollarProfitReverse = (lastPrice - beforeHoldingsAvgPrice) * beforeHoldingsQuantity
else:
tradeProfitPercent = 0
tradeDollarProfit = 0
tradeDollarProfitReverse = 0
### if we are not invested already the options are: ----------------------------------------------------------
# new holdings > 0 => going long
# new holdings < 0 => going short
if beforeHoldingsQuantity == 0:
if newHoldingsQuantity > 0:
algorithm.Log(str(target.Symbol.Value) + ': going long!'
+ ' current total holdings: ' + str(round(quantity, 0))
+ '; current average price: ' + str(round(newHoldingsAvgPrice, 4))
+ '; current total holdings cost: ' + str(round(newHoldingsCost, 2)))
else:
algorithm.Log(str(target.Symbol.Value) + ': going short!'
+ ' current total holdings: ' + str(round(quantity, 0))
+ '; average price: ' + str(round(newHoldingsAvgPrice, 4))
+ '; current total holdings cost: ' + str(round(newHoldingsCost, 2)))
### -----------------------------------------------------------------------------------------------------------
### if we are already long the security the options are: ------------------------------------------------------
# new quantity > 0 => adding to long position
# new quantity < 0 and new holdings < before holdings => partially selling long position
# new quantity < 0 and new holdings = 0 => closing entire long position
# new quantity < 0 and new holdings < 0 => closing entire long position and going short
elif beforeHoldingsQuantity > 0:
if quantity > 0:
algorithm.Log(str(target.Symbol.Value) + ': adding to current long position!'
+ ' additional shares: ' + str(round(quantity, 0))
+ '; current total holdings: ' + str(round(newHoldingsQuantity, 0))
+ '; current average price: ' + str(round(newHoldingsAvgPrice, 4))
+ '; current total holdings cost: ' + str(round(newHoldingsCost, 2)))
elif newHoldingsQuantity > 0 and newHoldingsQuantity < beforeHoldingsQuantity:
algorithm.Log(str(target.Symbol.Value) + ': selling part of current long position!'
+ ' selling shares: ' + str(round(-quantity, 0))
+ '; current total holdings: ' + str(round(newHoldingsQuantity, 0))
+ '; buying average price was: ' + str(round(beforeHoldingsAvgPrice, 4))
+ '; approx. selling average price is: ' + str(round(lastPrice, 4))
+ '; profit percent: ' + str(round(tradeProfitPercent, 4))
+ '; dollar profit: ' + str(round(tradeDollarProfit, 2)))
elif newHoldingsQuantity == 0:
algorithm.Log(str(target.Symbol.Value) + ': closing down entire current long position!'
+ ' selling shares: ' + str(round(-quantity, 0))
+ '; current total holdings: ' + str(round(newHoldingsQuantity, 0))
+ '; buying average price was: ' + str(round(beforeHoldingsAvgPrice, 4))
+ '; approx. selling average price is: ' + str(round(lastPrice, 4))
+ '; profit percent: ' + str(round(tradeProfitPercent, 4))
+ '; dollar profit: ' + str(round(tradeDollarProfit, 2)))
elif newHoldingsQuantity < 0:
algorithm.Log(str(target.Symbol.Value) + ': closing down entire current long position and going short!'
+ ' selling shares to close long: ' + str(round(beforeHoldingsQuantity, 0))
+ '; profit percent on long position: ' + str(round(tradeProfitPercent, 4))
+ '; dollar profit on long position: ' + str(round(tradeDollarProfitReverse, 2))
+ '; selling shares to go short: ' + str(round(-newHoldingsQuantity, 0))
+ '; current total holdings: ' + str(round(newHoldingsQuantity, 0))
+ '; current average price: ' + str(round(newHoldingsAvgPrice, 4))
+ '; current total holdings cost: ' + str(round(newHoldingsCost, 2)))
### --------------------------------------------------------------------------------------------------------------
### if we are already short the security the options are: --------------------------------------------------------
# new quantity < 0 => adding to short position
# new quantity > 0 and new holdings > before holdings => partially buying back short position
# new quantity > 0 and new holdings = 0 => closing entire short position
# new quantity > 0 and new holdings > 0 => closing entire short position and going long
elif beforeHoldingsQuantity < 0:
if quantity < 0:
algorithm.Log(str(target.Symbol.Value) + ': adding to current short position!'
+ ' additional shares: ' + str(round(quantity, 0))
+ '; current total holdings: ' + str(round(newHoldingsQuantity, 0))
+ '; current average price: ' + str(round(newHoldingsAvgPrice, 4))
+ '; current total holdings cost: ' + str(round(newHoldingsCost, 2)))
elif newHoldingsQuantity < 0 and newHoldingsQuantity > beforeHoldingsQuantity:
algorithm.Log(str(target.Symbol.Value) + ': buying back part of current short position!'
+ ' buying back shares: ' + str(round(quantity, 0))
+ '; current total holdings: ' + str(round(newHoldingsQuantity, 0))
+ '; shorting average price was: ' + str(round(beforeHoldingsAvgPrice, 4))
+ '; approx. buying back average price is: ' + str(round(lastPrice, 4))
+ '; profit percent: ' + str(round(tradeProfitPercent, 4))
+ '; dollar profit: ' + str(round(tradeDollarProfit, 2)))
elif newHoldingsQuantity == 0:
algorithm.Log(str(target.Symbol.Value) + ': closing down entire current short position!'
+ ' buying back shares: ' + str(round(quantity, 0))
+ '; current total holdings: ' + str(round(newHoldingsQuantity, 0))
+ '; shorting average price was: ' + str(round(beforeHoldingsAvgPrice, 4))
+ '; approx. buying back average price is: ' + str(round(lastPrice, 4))
+ '; profit percent: ' + str(round(tradeProfitPercent, 4))
+ '; dollar profit: ' + str(round(tradeDollarProfit, 2)))
elif newHoldingsQuantity > 0:
algorithm.Log(str(target.Symbol.Value) + ': closing down entire current short position and going long!'
+ ' buying back shares to close short: ' + str(round(-beforeHoldingsQuantity, 0))
+ '; profit percent on short position: ' + str(round(tradeProfitPercent, 4))
+ '; dollar profit on short position: ' + str(round(tradeDollarProfitReverse, 2))
+ '; buying shares to go long: ' + str(round(newHoldingsQuantity, 0))
+ '; current total holdings: ' + str(round(newHoldingsQuantity, 0))
+ '; current average price: ' + str(round(newHoldingsAvgPrice, 4))
+ '; current total holdings cost: ' + str(round(newHoldingsCost, 2)))
### ---------------------------------------------------------------------------------------------------------------
self.targetsCollection.ClearFulfilled(algorithm)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 LongOnlyMomentumAlphaCreationModel(AlphaModel):
'''
Description:
- Every N days, this Alpha model calculates the momentum score of each risky asset in the Universe
The momentum score is a weighted average of cumulative returns: (1-month * 12) + (3-month * 4) + (6-month * 2) + (12-month * 1)
- This Alpha model then creates InsightDirection.Up (to go Long) for a duration of a trading bar, every day for the selected top momentum securities
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, riskyTickers, crashProtectionTicker, canaryTickers, topMomentum = 5, rebalancingPeriod = Expiry.EndOfMonth):
self.riskyTickers = riskyTickers # risky tickers to use for momentum asset allocation
self.crashProtectionTicker = crashProtectionTicker # this ticker will also be part of momentum asset allocation,
# but it's a special asset that will get the crash protection allocation when needed
self.canaryTickers = canaryTickers # canary tickers to avoid in momentum calculations, but we need to subscribe to them
self.topMomentum = topMomentum # number of top momentum securities to keep
self.rebalancingPeriod = rebalancingPeriod # the rebalancing function
self.insightExpiry = Time.Multiply(Extensions.ToTimeSpan(Resolution.Daily), 0.25) # insight duration
self.insightDirection = InsightDirection.Up # insight direction
self.securities = [] # list to store securities to consider
self.topMomentumSecurities = {} # empty dictionary to store top momentum securities
self.rebalancingTime = None
def Update(self, algorithm, data):
if self.rebalancingTime is None:
# get next rebalancing time
self.rebalancingTime = self.rebalancingPeriod(algorithm.Time)
### calculate momentum scores (every N number of trading days) --------------------------------------------------------
if algorithm.Time >= self.rebalancingTime:
algorithm.Log('(Alpha) time to calculate the momentum securities')
### get symbols ---------------------------------------------------------------------------------------------------
# risky symbols
riskySymbols = [x.Symbol for x in self.securities if x.Symbol.Value in self.riskyTickers]
algorithm.Log('(Alpha) number of risky assets: ' + str(len(riskySymbols)))
# crash protection symbol
crashProtectionSymbol = [x.Symbol for x in self.securities if x.Symbol.Value in self.crashProtectionTicker]
algorithm.Log('(Alpha) number of crash protection assets: ' + str(len(crashProtectionSymbol)))
# combine the two lists to get relevant symbols for momentum calculations
relevantSymbols = riskySymbols + crashProtectionSymbol
algorithm.Log('(Alpha) number of relevant assets for trading: ' + str(len(relevantSymbols)))
# canary symbols
canarySymbols = [x.Symbol for x in self.securities if x.Symbol.Value in self.canaryTickers]
algorithm.Log('(Alpha) number of canary assets: ' + str(len(canarySymbols)))
# combine all lists to get all symbols for calculations
allSymbols = relevantSymbols + canarySymbols
algorithm.Log('(Alpha) total number of assets considered for calculations: ' + str(len(allSymbols)))
### make momentum calculations ---------------------------------------------------------------------------------------
# create empty dictionary to store calculations
calculations = {}
if len(allSymbols) > 0:
# get historical prices for symbols
history = algorithm.History(allSymbols, 253, Resolution.Daily)
for symbol in allSymbols:
# if symbol has no historical data continue the loop
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('(Alpha) no historical data for: ' + str(symbol.Value))
continue
else:
# add symbol to calculations
calculations[symbol] = SymbolData(symbol)
try:
# get momentum score
calculations[symbol].CalculateMomentumScore(history)
except Exception:
algorithm.Log('(Alpha) removing from Alpha calculations due to CalculateMomentumScore failing')
calculations.pop(symbol)
continue
calculatedSymbols = [x for x in calculations]
algorithm.Log('(Alpha) checking the number of calculated symbols: ' + str(len(calculatedSymbols)))
### get the top momentum securities among risky assets (including crash protection asset) ---------------------------
# perform Absolute Momentum: get the securities with positive momentum
positiveMomentumSecurities = list(filter(lambda x: x.momentumScore > 0 and x.Symbol in relevantSymbols, calculations.values()))
# perform Relative Momentum: sort descending by momentum score and select the top n
self.topMomentumSecurities = sorted(positiveMomentumSecurities, key = lambda x: x.momentumScore, reverse = True)[:self.topMomentum]
### get percentage dedicated to risky assets ------------------------------------------------------------------------
pctAggressive = self.CalculatePctAggressive(calculations, canarySymbols)
algorithm.Log('(Alpha) pctAggressive: ' + str(pctAggressive))
### if percentage aggressive is less than 1 ------------------------------------------------------------
### we need to add the crashProtectionSecurity if it has positive absolute momentum
crashProtectionSecurity = [x for x in self.securities if x.Symbol.Value in self.crashProtectionTicker]
positiveMomentumSymbols = [x.Symbol for x in positiveMomentumSecurities]
topMomentumSymbols = [x.Symbol for x in self.topMomentumSecurities]
# if percentage aggressive is 0,
# we only generate insights for the crash protection asset if it has positive absolute momentum;
# if not, we don't send any insights
if pctAggressive == 0:
if crashProtectionSymbol[0] in positiveMomentumSymbols:
algorithm.Log('(Alpha) pctAggressive is 0 but crashProtectionSymbol has positive momentum so we add it')
self.topMomentumSecurities = crashProtectionSecurity
else:
self.topMomentumSecurities = []
# if percentage aggressive is positive but less than 1,
# we need to make sure we are sending insights for the crash protection asset as well if it has positive absolute momentum
elif pctAggressive < 1:
if crashProtectionSymbol[0] in positiveMomentumSymbols and crashProtectionSymbol[0] not in topMomentumSymbols:
algorithm.Log('(Alpha) adding the crash protection asset to topMomentumSecurities')
self.topMomentumSecurities.append(crashProtectionSecurity[0])
# get top momentum tickers for logs
topMomentumTickers = [x.Symbol.Value for x in self.topMomentumSecurities]
algorithm.Log('(Alpha) top securities: ' + str(topMomentumTickers))
# update rebalancing time
self.rebalancingTime = self.rebalancingPeriod(algorithm.Time)
### end of rebalance calculations ---------------------------------------------------------------------------------------
### generate insights ---------------------------------------------------------------------------------------------------
insights = [] # list to store the new insights to be created
# loop through active securities and generate insights
for security in self.topMomentumSecurities:
# 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('(Portfolio) excluding 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)
def CalculatePctAggressive(self, calculations, canarySymbols):
'''
Description:
Calculate the percentage dedicated to risky assets
Args:
calculations: Dictionary with calculations for symbols
canarySymbols: Symbols for the canary assets
Returns:
Float with the percentage dedicated to risky assets
'''
# get a list with the canary assets that have positive absolute momentum
canaryPosMomList = list(filter(lambda x: x.momentumScore > 0 and x.Symbol in canarySymbols, calculations.values()))
# get the average positive (basically the options are 0, 0.5 or 1)
# this will be the allocation for risky assets
pctAggressive = len(canaryPosMomList) / len(canarySymbols)
return pctAggressive
class SymbolData:
''' Contain data specific to a symbol required by this model '''
def __init__(self, symbol):
self.Symbol = symbol
def CalculateMomentumScore(self, history):
''' Calculate the weighted average momentum value for each security '''
returnSeries = history.loc[str(self.Symbol)]['close'].pct_change(periods = 1).dropna() # 1-day returns for last year
cumRet1 = (returnSeries.tail(21).add(1).prod()) - 1 # 1-month momentum
cumRet3 = (returnSeries.tail(63).add(1).prod()) - 1 # 3-month momentum
cumRet6 = (returnSeries.tail(126).add(1).prod()) - 1 # 6-month momentum
cumRet12 = (returnSeries.tail(252).add(1).prod()) - 1 # 12-month momentum
self.momentumScore = (cumRet1 * 12 + cumRet3 * 4 + cumRet6 * 2 + cumRet12) # weighted average momentum