Risk Parity Portfolio Construction: Equal Risk Contribution
Risk parity challenges the fundamental assumption of traditional portfolio construction: that diversification should be measured in capital terms. A 60/40 stock/bond portfolio may appear diversified by weight, but equities contribute approximately 90% of the portfolio's total risk. Risk parity instead equalizes each asset's contribution to total portfolio volatility, producing portfolios that are truly diversified from a risk perspective. Pioneered by Bridgewater Associates' All Weather fund, risk parity has grown into a $200+ billion strategy category. This guide implements risk parity from the simple inverse-volatility approach through the full Equal Risk Contribution (ERC) optimization.Key Takeaways
- Equal weight is not equal risk. A 60/40 portfolio has 90% equity risk. Risk parity corrects this imbalance.
- Inverse-volatility weighting is the simplest risk parity approach but ignores correlations between assets.
- Equal Risk Contribution (ERC) accounts for both volatility and correlation, ensuring each asset contributes equally to total risk.
- Leverage is often required to achieve competitive returns, since risk parity naturally overweights low-volatility assets (bonds).
Why Equal Weight Fails as Diversification
``python
import numpy as np
import pandas as pd
from scipy.optimize import minimize
def demonstrate_risk_concentration(
weights: np.ndarray,
cov_matrix: np.ndarray,
asset_names: list[str],
) -> pd.DataFrame:
"""
Show how capital-weighted portfolios concentrate risk.
"""
port_vol = np.sqrt(weights @ cov_matrix @ weights)
# Marginal risk contribution: d(port_vol) / d(w_i)
marginal_risk = cov_matrix @ weights / port_vol
# Component risk contribution: w_i * MRC_i
risk_contribution = weights * marginal_risk
# Percentage of total risk
risk_pct = risk_contribution / risk_contribution.sum()
results = pd.DataFrame({
"asset": asset_names,
"weight": weights,
"volatility": np.sqrt(np.diag(cov_matrix)),
"marginal_risk": marginal_risk,
"risk_contribution": risk_contribution,
"risk_pct": risk_pct,
})
print("Portfolio Risk Decomposition:")
print(f"Portfolio Volatility: {port_vol:.2%}\n")
for _, row in results.iterrows():
bar = "#" int(row["risk_pct"] 50)
print(f" {row['asset']:>12s}: weight={row['weight']:.1%} "
f"risk={row['risk_pct']:.1%} {bar}")
return results
Example: 60/40 portfolio
cov = np.array([
[0.0256, 0.0020], # Stocks: 16% vol
[0.0020, 0.0016], # Bonds: 4% vol
])
weights_6040 = np.array([0.60, 0.40])
demonstrate_risk_concentration(weights_6040, cov, ["Stocks", "Bonds"])
`
Inverse-Volatility Weighting
The simplest form of risk parity weights each asset inversely proportional to its volatility. This is fast, intuitive, and works well when correlations are roughly uniform.
`python
class InverseVolatilityPortfolio:
"""
Inverse-volatility risk parity: w_i proportional to 1/sigma_i.
Ignores correlations but is robust and easy to implement.
"""
def __init__(self, vol_lookback: int = 63):
self.vol_lookback = vol_lookback
def compute_weights(
self,
returns: pd.DataFrame,
) -> pd.Series:
"""
Compute inverse-volatility weights from return history.
"""
# Annualized volatility
vols = returns.iloc[-self.vol_lookback:].std() * np.sqrt(252)
# Inverse volatility weights
inv_vol = 1.0 / vols
weights = inv_vol / inv_vol.sum()
return weights
def rolling_weights(
self, returns: pd.DataFrame
) -> pd.DataFrame:
"""Compute time series of rolling weights."""
all_weights = []
for i in range(self.vol_lookback, len(returns)):
window = returns.iloc[i - self.vol_lookback:i]
vols = window.std() * np.sqrt(252)
inv_vol = 1.0 / vols
w = inv_vol / inv_vol.sum()
all_weights.append(w)
return pd.DataFrame(
all_weights,
index=returns.index[self.vol_lookback:],
columns=returns.columns,
)
`
Equal Risk Contribution (ERC) Portfolio
The full ERC optimization finds weights where each asset contributes exactly the same amount to total portfolio volatility.
`python
class EqualRiskContribution:
"""
Equal Risk Contribution portfolio optimization.
Finds weights where RC_i = RC_j for all i, j.
RC_i = w_i * (Sigma @ w)_i / sqrt(w @ Sigma @ w)
"""
def __init__(
self,
cov_lookback: int = 126,
shrinkage: bool = True,
):
self.cov_lookback = cov_lookback
self.shrinkage = shrinkage
def _risk_contribution(
self, weights: np.ndarray, cov: np.ndarray
) -> np.ndarray:
"""Compute risk contribution of each asset."""
port_vol = np.sqrt(weights @ cov @ weights)
marginal = cov @ weights / port_vol
return weights * marginal
def _erc_objective(
self, weights: np.ndarray, cov: np.ndarray
) -> float:
"""
Objective: minimize sum of squared differences between
risk contributions and target (1/N).
"""
rc = self._risk_contribution(weights, cov)
target_rc = np.sum(rc) / len(weights)
return np.sum((rc - target_rc) ** 2)
def optimize(
self,
cov: np.ndarray,
long_only: bool = True,
) -> np.ndarray:
"""
Find ERC weights via numerical optimization.
"""
n = cov.shape[0]
# Initial guess: inverse volatility
vols = np.sqrt(np.diag(cov))
w0 = (1 / vols) / (1 / vols).sum()
# Constraints
constraints = [
{"type": "eq", "fun": lambda w: np.sum(w) - 1.0},
]
# Bounds
if long_only:
bounds = [(0.01, 0.99) for _ in range(n)]
else:
bounds = [(-0.99, 0.99) for _ in range(n)]
result = minimize(
self._erc_objective,
w0,
args=(cov,),
method="SLSQP",
bounds=bounds,
constraints=constraints,
options={"maxiter": 1000, "ftol": 1e-12},
)
if not result.success:
print(f"Warning: Optimization did not converge: {result.message}")
return result.x
def compute_weights(
self, returns: pd.DataFrame
) -> pd.Series:
"""Compute ERC weights from return data."""
recent = returns.iloc[-self.cov_lookback:]
if self.shrinkage:
cov = self._shrinkage_cov(recent)
else:
cov = recent.cov().values * 252 # Annualized
weights = self.optimize(cov)
return pd.Series(weights, index=returns.columns)
def _shrinkage_cov(self, returns: pd.DataFrame) -> np.ndarray:
"""Ledoit-Wolf shrinkage covariance estimator."""
from sklearn.covariance import LedoitWolf
lw = LedoitWolf()
lw.fit(returns.values)
return lw.covariance_ * 252 # Annualized
def verify_erc(
self,
weights: np.ndarray,
cov: np.ndarray,
asset_names: list[str] = None,
) -> pd.DataFrame:
"""Verify that weights achieve equal risk contribution."""
rc = self._risk_contribution(weights, cov)
total_risk = rc.sum()
rc_pct = rc / total_risk
if asset_names is None:
asset_names = [f"Asset_{i}" for i in range(len(weights))]
results = pd.DataFrame({
"asset": asset_names,
"weight": weights,
"risk_contribution": rc,
"risk_pct": rc_pct,
"target_pct": 1 / len(weights),
"deviation": rc_pct - 1 / len(weights),
})
max_deviation = results["deviation"].abs().max()
print(f"ERC Verification (max deviation: {max_deviation:.4%}):")
for _, row in results.iterrows():
print(f" {row['asset']:>12s}: w={row['weight']:.3f} "
f"risk={row['risk_pct']:.3f} "
f"target={row['target_pct']:.3f}")
return results
`
Leveraged Risk Parity
Risk parity portfolios typically have lower volatility than equity-heavy portfolios. Leverage is applied to scale the portfolio to a target volatility.
`python
class LeveragedRiskParity:
"""
Risk parity with leverage targeting a specific volatility level.
"""
def __init__(
self,
target_vol: float = 0.10, # 10% target volatility
max_leverage: float = 3.0,
rebalance_freq: str = "ME", # Monthly rebalancing
):
self.target_vol = target_vol
self.max_leverage = max_leverage
self.rebalance_freq = rebalance_freq
self.erc = EqualRiskContribution()
def backtest(
self, returns: pd.DataFrame, lookback: int = 126
) -> pd.DataFrame:
"""
Backtest leveraged risk parity strategy.
"""
results = []
rebalance_dates = returns.resample(self.rebalance_freq).last().index
current_weights = None
current_leverage = 1.0
for date in returns.index[lookback:]:
if current_weights is None or date in rebalance_dates:
# Recompute ERC weights
window = returns.loc[:date].iloc[-lookback:]
erc_weights = self.erc.compute_weights(window)
# Compute realized vol of ERC portfolio
erc_returns = (window * erc_weights).sum(axis=1)
realized_vol = erc_returns.std() * np.sqrt(252)
# Leverage to target volatility
leverage = min(
self.target_vol / max(realized_vol, 0.01),
self.max_leverage,
)
current_weights = erc_weights * leverage
current_leverage = leverage
# Portfolio return for this day
day_return = (returns.loc[date] * current_weights).sum()
# Borrowing cost for leverage (assume SOFR + 50bps)
borrowing_cost = max(0, current_leverage - 1) * 0.055 / 252
results.append({
"date": date,
"return": day_return - borrowing_cost,
"leverage": current_leverage,
"gross_return": day_return,
"borrowing_cost": borrowing_cost,
})
results_df = pd.DataFrame(results).set_index("date")
# Performance summary
ret = results_df["return"]
ann_ret = ret.mean() * 252
ann_vol = ret.std() * np.sqrt(252)
sharpe = ann_ret / ann_vol if ann_vol > 0 else 0
cumulative = (1 + ret).cumprod()
max_dd = ((cumulative.cummax() - cumulative) / cumulative.cummax()).max()
print(f"Leveraged Risk Parity Backtest:")
print(f" Annual Return: {ann_ret:.2%}")
print(f" Annual Vol: {ann_vol:.2%} (target: {self.target_vol:.2%})")
print(f" Sharpe: {sharpe:.2f}")
print(f" Max Drawdown: {max_dd:.2%}")
print(f" Avg Leverage: {results_df['leverage'].mean():.2f}x")
return results_df
`
Multi-Asset Risk Parity
Extend risk parity across diverse asset classes.
`python
def multi_asset_risk_parity(
asset_classes: dict[str, pd.Series],
target_vol: float = 0.10,
) -> dict:
"""
Build a multi-asset risk parity portfolio.
Args:
asset_classes: dict of {name: return_series} for each asset class
target_vol: target portfolio volatility
"""
returns_df = pd.DataFrame(asset_classes).dropna()
# ERC optimization
erc = EqualRiskContribution(cov_lookback=252)
weights = erc.compute_weights(returns_df)
# Scale to target vol
port_returns = (returns_df * weights).sum(axis=1)
realized_vol = port_returns.std() * np.sqrt(252)
leverage = target_vol / realized_vol
final_weights = weights * leverage
print(f"\nMulti-Asset Risk Parity Allocation:")
print(f"{'Asset':>20s} {'ERC Weight':>12s} {'Final Weight':>12s}")
for asset in returns_df.columns:
print(f"{asset:>20s} {weights[asset]:>12.1%} {final_weights[asset]:>12.1%}")
print(f"\n{'Total':>20s} {weights.sum():>12.1%} {final_weights.sum():>12.1%}")
print(f"{'Leverage':>20s} {'':>12s} {leverage:>12.2f}x")
return {
"erc_weights": weights,
"leveraged_weights": final_weights,
"leverage": leverage,
"target_vol": target_vol,
"realized_vol": realized_vol,
}
``