Overall Statistics Total Trades842Average Win0.29%Average Loss-0.19%Compounding Annual Return8.604%Drawdown12.600%Expectancy1.291Net Profit227.421%Sharpe Ratio1.235Probabilistic Sharpe Ratio78.823%Loss Rate9%Win Rate91%Profit-Loss Ratio1.53Alpha0.073Beta-0.012Annual Standard Deviation0.058Annual Variance0.003Information Ratio-0.069Tracking Error0.192Treynor Ratio-6.097Total Fees\$846.44
```import pandas as pd
import numpy as np
from scipy.optimize import minimize

class PortfolioOptimizer:

'''
Description:
Implementation of a custom optimizer that calculates the weights for each asset to optimize a given objective function
Details:
Optimization can be:
- Equal Weighting
- Maximize Portfolio Return
- Minimize Portfolio Standard Deviation
- Mean-Variance (minimize Standard Deviation given a target return)
- Maximize Portfolio Sharpe Ratio
- Maximize Portfolio Sortino Ratio
- Risk Parity Portfolio
Constraints:
- Weights must be between some given boundaries
- Weights must sum to 1
'''

def __init__(self,
minWeight = 0,
maxWeight = 1):

'''
Description:
Initialize the CustomPortfolioOptimizer
Args:
minWeight(float): The lower bound on portfolio weights
maxWeight(float): The upper bound on portfolio weights
'''

self.minWeight = minWeight
self.maxWeight = maxWeight

def Optimize(self, objFunction, dailyReturnsDf, targetReturn = None):

'''
Description:
Perform portfolio optimization given a series of returns
Args:
objFunction: The objective function to optimize (equalWeighting, maxReturn, minVariance, meanVariance, maxSharpe, maxSortino, riskParity)
dailyReturnsDf: DataFrame of historical daily arithmetic returns
Returns:
Array of double with the portfolio weights (size: K x 1)
'''

# initial weights: equally weighted
size = dailyReturnsDf.columns.size # K x 1
self.initWeights = np.array(size * [1. / size])

# get sample covariance matrix
covariance = dailyReturnsDf.cov()
# get the sample covariance matrix of only negative returns for sortino ratio
negativeReturnsDf = dailyReturnsDf[dailyReturnsDf < 0]
covarianceNegativeReturns = negativeReturnsDf.cov()

if objFunction == 'equalWeighting':
return self.initWeights

bounds = tuple((self.minWeight, self.maxWeight) for x in range(size))
constraints = [{'type': 'eq', 'fun': lambda x: np.sum(x) - 1.0}]

if objFunction == 'meanVariance':
# if no target return is provided, use the resulting from equal weighting
if targetReturn is None:
targetReturn = self.CalculateAnnualizedPortfolioReturn(dailyReturnsDf, self.initWeights)
constraints.append( {'type': 'eq', 'fun': lambda weights:
self.CalculateAnnualizedPortfolioReturn(dailyReturnsDf, weights) - targetReturn} )

opt = minimize(lambda weights: self.ObjectiveFunction(objFunction, dailyReturnsDf,
covariance, covarianceNegativeReturns,
weights),
x0 = self.initWeights,
bounds = bounds,
constraints = constraints,
method = 'SLSQP')

return opt['x']

def ObjectiveFunction(self, objFunction, dailyReturnsDf, covariance, covarianceNegativeReturns, weights):

'''
Description:
Compute the objective function
Args:
objFunction: The objective function to optimize (equalWeighting, maxReturn, minVariance, meanVariance,
maxSharpe, maxSortino, riskParity)
dailyReturnsDf: DataFrame of historical daily returns
covariance: Sample covariance
covarianceNegativeReturns: Sample covariance matrix of only negative returns
weights: Portfolio weights
'''

if objFunction == 'maxReturn':
f = self.CalculateAnnualizedPortfolioReturn(dailyReturnsDf, weights)
return -f # convert to negative to be minimized
elif objFunction == 'minVariance':
f = self.CalculateAnnualizedPortfolioStd(covariance, weights)
return f
elif objFunction == 'meanVariance':
f = self.CalculateAnnualizedPortfolioStd(covariance, weights)
return f
elif objFunction == 'maxSharpe':
f = self.CalculateAnnualizedPortfolioSharpeRatio(dailyReturnsDf, covariance, weights)
return -f # convert to negative to be minimized
elif objFunction == 'maxSortino':
f = self.CalculateAnnualizedPortfolioSortinoRatio(dailyReturnsDf, covarianceNegativeReturns, weights)
return -f # convert to negative to be minimized
elif objFunction == 'riskParity':
f = self.CalculateRiskParityFunction(covariance, weights)
return f
else:
raise ValueError(f'PortfolioOptimizer.ObjectiveFunction: objFunction input has to be one of equalWeighting,'
+ ' maxReturn, minVariance, meanVariance, maxSharpe, maxSortino or riskParity')

def CalculateAnnualizedPortfolioReturn(self, dailyReturnsDf, weights):

annualizedPortfolioReturns = np.sum( ((1 + dailyReturnsDf.mean())**252 - 1) * weights )

return annualizedPortfolioReturns

def CalculateAnnualizedPortfolioStd(self, covariance, weights):

annualizedPortfolioStd = np.sqrt( np.dot(weights.T, np.dot(covariance * 252, weights)) )

if annualizedPortfolioStd == 0:
raise ValueError(f'PortfolioOptimizer.CalculateAnnualizedPortfolioStd: annualizedPortfolioStd cannot be zero. Weights: {weights}')

return annualizedPortfolioStd

def CalculateAnnualizedPortfolioNegativeStd(self, covarianceNegativeReturns, weights):

annualizedPortfolioNegativeStd = np.sqrt( np.dot(weights.T, np.dot(covarianceNegativeReturns * 252, weights)) )

if annualizedPortfolioNegativeStd == 0:
raise ValueError(f'PortfolioOptimizer.CalculateAnnualizedPortfolioNegativeStd: annualizedPortfolioNegativeStd cannot be zero. Weights: {weights}')

return annualizedPortfolioNegativeStd

def CalculateAnnualizedPortfolioSharpeRatio(self, dailyReturnsDf, covariance, weights):

annualizedPortfolioReturn = self.CalculateAnnualizedPortfolioReturn(dailyReturnsDf, weights)
annualizedPortfolioStd = self.CalculateAnnualizedPortfolioStd(covariance, weights)
annualizedPortfolioSharpeRatio = annualizedPortfolioReturn / annualizedPortfolioStd

return annualizedPortfolioSharpeRatio

def CalculateAnnualizedPortfolioSortinoRatio(self, dailyReturnsDf, covarianceNegativeReturns, weights):

annualizedPortfolioReturn = self.CalculateAnnualizedPortfolioReturn(dailyReturnsDf, weights)
annualizedPortfolioNegativeStd = self.CalculateAnnualizedPortfolioNegativeStd(covarianceNegativeReturns, weights)
annualizedPortfolioSortinoRatio = annualizedPortfolioReturn / annualizedPortfolioNegativeStd

return annualizedPortfolioSortinoRatio

def CalculateRiskParityFunction(self, covariance, weights):

''' Spinu formulation for risk parity portfolio '''

assetsRiskBudget = self.initWeights
portfolioVolatility = self.CalculateAnnualizedPortfolioStd(covariance, weights)

x = weights / portfolioVolatility
riskParity = (np.dot(x.T, np.dot(covariance, x)) / 2) - np.dot(assetsRiskBudget.T, np.log(x))

return riskParity```
```from OptimalWeightsAlphaCreation import OptimalWeightsAlphaCreationModel

class PortfolioOptimizationSystem(QCAlgorithmFramework):

def Initialize(self):

### user-defined inputs ---------------------------------------------------------------------------------------------------

self.SetStartDate(2006, 1, 1)  # Set Start Date
#self.SetEndDate(2020, 5, 1)     # Set End Date
self.SetCash(100000)            # Set Strategy Cash

# UNIVERSE -------------------------------------------------------------

# select tickers
tickers = ['IEF', 'TLT', 'GLD', 'SPY', 'QQQ']

# ALPHA ----------------------------------------------------------------

# select the objective function to optimize the portfolio weights
# options are: equalWeighting, maxReturn, minVariance, meanVariance, maxSharpe, maxSortino, riskParity
objectiveFunction = 'riskParity'

# select number of lookback days for optimization
lookbackOptimization = 63

# select the logic for rebalancing period
# date rules (for the first trading day of period): Expiry.EndOfDay, Expiry.EndOfWeek, Expiry.EndOfMonth,
#                                                   Expiry.EndOfQuarter, Expiry.EndOfYear
rebalancingPeriod = Expiry.EndOfMonth

### plotting ----------------------------------------------------------------------------------------------------------------

# let's plot the series of daily total portfolio exposure %
portfolioExposurePlot = Chart('Chart Total Portfolio Exposure %')
portfolioExposurePlot.AddSeries(Series('Daily Portfolio Exposure %', SeriesType.Line, ''))

# let's plot the series of drawdown % from the most recent high
drawdownPlot = Chart('Chart Drawdown %')

# let's plot the series of optimal weights
optWeightsPlot = Chart('Chart Optimal Weights %')
for ticker in tickers:

### select modules ------------------------------------------------------------------------------------------------------------

# set the brokerage model for slippage and fees
self.SetBrokerageModel(AlphaStreamsBrokerageModel())

# set requested data resolution
self.UniverseSettings.Resolution = Resolution.Daily

# Universe -------------------------------------------------------------
symbols = []
# loop through the tickers list and create symbols for the Universe
for ticker in tickers:
symbols.append(Symbol.Create(ticker, SecurityType.Equity, Market.USA))
self.SetUniverseSelection(ManualUniverseSelectionModel(symbols))

# Alpha ---------------------------------------------------------------
self.SetAlpha(OptimalWeightsAlphaCreationModel(rebalancingPeriod = rebalancingPeriod,
objectiveFunction = objectiveFunction,
lookbackOptimization = lookbackOptimization))

# Portfolio ------------------------------------------------------------
pcm = InsightWeightingPortfolioConstructionModel(lambda time: rebalancingPeriod(time))
pcm.RebalanceOnInsightChanges = False # disable rebalancing on insights changes (new/expired insights)
pcm.RebalanceOnSecurityChanges = False # enable rebalancing on universe changes
self.SetPortfolioConstruction(pcm)

# Execution ------------------------------------------------------------
self.SetExecution(ImmediateExecutionModel())

# Risk -----------------------------------------------------------------
self.SetRiskManagement(NullRiskManagementModel())```