Vijay Vaidyanathan who founded a custom-indexing firm that is recently being acquired has a solid track, he also wrote notebook for teaching portfolio management theory in EDHEC (France). In his notebook, I learn CPPI.
It’s basically to swap between risky assets and conservative assets dynamically to reach the highest possible upside potential while imposing stringent limits to portfolio drawdown.
We can break down into the math and codes to execute the math.

def run_cppi(risky_r, safe_r=None, m=3, start=1000, floor=0.8, riskfree_rate=0.03, drawdown=None):
"""
Run a backtest of the CPPI strategy, given a set of returns for the risky asset
Returns a dictionary containing: Asset Value History, Risk Budget History, Risky Weight History
"""
# set up the CPPI parameters
dates = risky_r.index
n_steps = len(dates)
account_value = start
floor_value = start*floor
peak = account_value
if isinstance(risky_r, pd.Series):
risky_r = pd.DataFrame(risky_r, columns=["R"])
if safe_r is None:
safe_r = pd.DataFrame().reindex_like(risky_r)
safe_r.values[:] = riskfree_rate/12 # fast way to set all values to a number
# set up some DataFrames for saving intermediate values
account_history = pd.DataFrame().reindex_like(risky_r)
risky_w_history = pd.DataFrame().reindex_like(risky_r)
cushion_history = pd.DataFrame().reindex_like(risky_r)
floorval_history = pd.DataFrame().reindex_like(risky_r)
peak_history = pd.DataFrame().reindex_like(risky_r)
for step in range(n_steps):
if drawdown is not None:
peak = np.maximum(peak, account_value)
floor_value = peak*(1-drawdown)
cushion = (account_value - floor_value)/account_value
risky_w = m*cushion
risky_w = np.minimum(risky_w, 1)
risky_w = np.maximum(risky_w, 0)
safe_w = 1-risky_w
risky_alloc = account_value*risky_w
safe_alloc = account_value*safe_w
# recompute the new account value at the end of this step
account_value = risky_alloc*(1+risky_r.iloc[step]) + safe_alloc*(1+safe_r.iloc[step])
# save the histories for analysis and plotting
cushion_history.iloc[step] = cushion
risky_w_history.iloc[step] = risky_w
account_history.iloc[step] = account_value
floorval_history.iloc[step] = floor_value
peak_history.iloc[step] = peak
risky_wealth = start*(1+risky_r).cumprod()
backtest_result = {
"Wealth": account_history,
"Risky Wealth": risky_wealth,
"Risk Budget": cushion_history,
"Risky Allocation": risky_w_history,
"m": m,
"start": start,
"floor": floor,
"risky_r":risky_r,
"safe_r": safe_r,
"drawdown": drawdown,
"peak": peak_history,
"floor": floorval_history
}
return backtest_result
# For loop over dates
for step in range( len(risky_rets.index) ):
# computing the cushion (as a percentage of the current account value)
cushion = (account_value - floor_value) / account_value
# compute the weight for the allocation on the risky asset
risky_w = m * cushion
risky_w = np.minimum(risky_w, 1)
risky_w = np.maximum(risky_w, 0)
# the last two conditions ensure that the risky weight is in [0,1]
# compute the weight for the allocation on the safe asset
safe_w = 1 - risky_w
# compute the value allocation
risky_allocation = risky_w * account_value
safe_allocation = safe_w * account_value
# compute the new account value: this is given by the new values from both the risky and the safe assets
account_value = risky_allocation * (1 + risky_rets.iloc[step] ) + safe_allocation * (1 + safe_rets.iloc[step] )
# save data: current account value, cushions, weights
account_history.iloc[step] = account_value
cushion_history.iloc[step] = cushion
risky_w_history.iloc[step] = risky_w
# given the CPPI wealth saved in the account_history, we can get back the CPPI returns
cppi_rets = ( account_history / account_history.shift(1) - 1 ).dropna()
Sometimes we would like to dynamically update the floor value based on the previous peak of the wealth growth, i.e. setting the drawdown constraint.
res = erk.run_cppi(risky_rets, m = 1/0.2, start=1000, floor=0.8, drawdown=0.2, riskfree_rate=0.03)
sector = "Fin"
fig, ax = plt.subplots(1,2,figsize=(18,4))
ax = ax.flatten()
res["Wealth"][sector].plot(ax=ax[0], grid=True, label="CPPI "+sector)
res["Risky Wealth"][sector].plot(ax=ax[0], grid=True, label=sector, style="k:")
res["floor"][sector].plot(ax=ax[0], grid=True, color="r", linestyle="--", label="Fixed floor value")
ax[0].legend(fontsize=11)
# Weights Fin
res["Risky Allocation"][sector].plot(ax=ax[1], grid=True, label="Risky weight in "+sector)
ax[1].legend(fontsize=11)
plt.show()