Enhanced Portfolio Optimization - Replication

import pandas as pd
import numpy as np
from tqdm import tqdm

Code

df = pd.read_csv("../data/49_Industry_Portfolios_Daily.csv", skiprows=9, nrows=25400)
df = df.rename(columns={"Unnamed: 0": "Date"}).dropna()
df["Date"] = pd.to_datetime(df["Date"], format="%Y%m%d")

Code

ff = pd.readdf = pd.read_csv("../data/F-F_Research_Data_Factors_daily.csv", skiprows=4, nrows=25400)
ff = ff.rename(columns={"Unnamed: 0": "Date"}).dropna()
ff["Date"] = pd.to_datetime(ff["Date"], format="%Y%m%d")

Code

df.replace(-99.99, np.nan, inplace=True)
ff.replace(-99.99, np.nan, inplace=True)
df.set_index("Date", inplace=True)
ff.set_index("Date", inplace=True)
df = df / 100
ff = ff / 100

def calSharpeRatio(r, multiplier=12):
    """
    Args:
        r: return series
    """
    sharpe = r.mean() * multiplier / (np.std(r) * np.sqrt(multiplier))
    return sharpe

1 Data Preparation

1. Narrow the backtesting period to 1942 to 2018 to keep consistent with the paper.
2. Turn daily returns into monthly returns.

backtestingPeriods = pd.DatetimeIndex(np.arange("1942-01", "2019-01", dtype="datetime64")).to_period("M")

dailyRet = df.copy(deep=True)
rfRet = ff.copy(deep=True)["RF"]


def _dailyToMonthlyRetHelper(x):
    return (1+x).prod(min_count=1) - 1


monthlyRet = dailyRet.groupby(pd.Grouper(
    level="Date", freq="1M")).apply(_dailyToMonthlyRetHelper)
monthlyRet.index = monthlyRet.index.to_period("M")

rfMonthlyRet = rfRet.groupby(pd.Grouper(
    level="Date", freq="1M")).apply(_dailyToMonthlyRetHelper)
rfMonthlyRet.index = rfMonthlyRet.index.to_period("M")

dailyExcessRet = dailyRet.subtract(rfRet.values, axis=0)
monthlyExcessRet = dailyExcessRet.groupby(
    pd.Grouper(freq="1M")).apply(_dailyToMonthlyRetHelper)
monthlyExcessRet.index = monthlyExcessRet.index.to_period("M")

2 1/N portfolio

# equally weighted
portfolioExcessRet = monthlyExcessRet.mean(axis=1, skipna=True)

_dateMask = portfolioExcessRet.index.isin(backtestingPeriods)
calSharpeRatio(portfolioExcessRet[_dateMask])

0.5902477272794807

3 MVO and EPO Implementation

Here is the implementation of MVO and EPO. EPO adds the shrinkage in its covariance matrix estimation. To extend the testing scope, I also added a long-only constraint which utilizes the cvxpy package.

def calStandardMVO(s, V, gamma, **kwargs):
    """
    Args:
        s: signal
        V: covariance matrix
        gamma: risk aversion (usually between 1 to 10)
    Return:
        portfolio allocations
    """
    N = len(s)
    V_inv = np.linalg.inv(V)
    # gamma = np.ones(N)@V_inv@s
    x = 1 / gamma * V_inv@s
    return x

def calEPO(s, V, w, gamma, long_only=False):
    sigma = np.sqrt(np.diag(V))
    sigmaM = np.power(np.diag(sigma),2)
    shrunkV = (1-w) * V + w * sigmaM
    if not long_only:
        x = 1/gamma * np.linalg.inv(shrunkV) @ s
        return x
    else:
        import cvxpy as cp
        N = len(s)
        x = cp.Variable(N)
        constraints=[x >= 0, np.ones(N).T@x == 1]
        objecive = cp.Maximize(s.T@x - gamma/2*cp.quad_form(x, shrunkV))
        prob=cp.Problem(objecive, constraints)
        prob.solve()
        return x.value

4 Signal Implementation

The cross-sectional momentum signal is defined as follows. It looks at the past 12-month returns of each assets, and assign positive weights to assets that exceed the average.

\[ s_t^i=\operatorname{XSMOM}_t^i:=c_t\left(r_{t-12, t}^i-\frac{1}{n} \sum_{j=1, \ldots, n} r_{t-12, t}^j\right) \]

The weights are then normalized to sum to 1 respectively for long and short positions, scaled by \(c_t\)

\[ \sum_i s_t^i 1_{\left\{s_t^i>0\right\}}=\sum_i\left|s_t^i\right| 1_{\left\{s_t^i<0\right\}}=1 \]

def signalXSMOM(sectorRet):
    sectorCumRet = (1 + sectorRet).prod(axis=0) - 1
    # avg returns across sectors
    avg = sectorCumRet.mean()
    s = sectorCumRet - avg
    posSum = np.sum(s[s >= 0])
    negSum = np.sum(s[s < 0])
    s = np.where(s >= 0, s / posSum, -s/negSum)
    return s

5 Backtesting

To make the code concise, I will write a backtesting function that takes in an allocation function as an argument.

def backtest(allocation_func, **kwargs):
    portfolioRet = list()
    for valuationDate in backtestingPeriods:
        # risk model based on past 5 years
        monthlySlice = monthlyExcessRet.loc[valuationDate - 60: valuationDate - 1]
        monthlySlice = monthlySlice.dropna(axis=1)
        V = monthlySlice.cov()
        investables = V.columns
        
        # signal generated based on past 12 months
        s = signalXSMOM(monthlySlice.loc[valuationDate - 12: valuationDate - 1])
        
        # portfolio allocation
        x = allocation_func(s, V, gamma=len(investables)/0.4, **kwargs)
        x = pd.Series(index=investables, data=x)
        
        # append the return
        thisMonthRet = monthlyExcessRet.loc[valuationDate]
        thisMonthPortfolioRet = (thisMonthRet * x).dropna().sum()
        portfolioRet.append(thisMonthPortfolioRet)
    return calSharpeRatio(pd.Series(portfolioRet))

allocation_funcs = {
    "standard MVO": (calStandardMVO, {}),
    "EPO": (calEPO, {'w' : 0.5, 'long_only': False}),
    "EPO long only": (calEPO, {'w' : 0.5, 'long_only': True})
}

for name, func in allocation_funcs.items():
    print(name, backtest(func[0], **func[1]))

standard MVO 0.16825747296896637
EPO 0.7977512637644565
EPO long only 0.8336651855961855