Pipeline custom factor for downside volatilty

Hoping a python guru can post the code for downside volatility (measuring deviation of returns that are below zero.) as a custom factor.

Thanks!

6 responses

I believe the custom factor below should do what you want.

class Downside_Volatility(CustomFactor):
# Use returns as the input
inputs = [Returns(window_length=2)]
# The trading days to calculate volatility over (can override)
window_length = 10
def compute(self, today, assets, out, returns):
# set any returns greater than 0 to NaN so they are excluded from our calculations
returns[returns > 0] = np.nan
out[:] = np.nanstd(returns, axis=0)



It's a little more correct to use log returns in the calculation (https://en.wikipedia.org/wiki/Volatility_(finance)) so this could be used too.

class Downside_Log_Volatility(CustomFactor):
# Use returns as the input
inputs = [Returns(window_length=2)]
# The trading days to calculate volatility over (can override)
window_length = 10
def compute(self, today, assets, out, returns):
# set any returns greater than 0 to NaN so they are excluded from our calculations
returns[returns > 0] = np.nan
log_returns = np.log(returns+1)
out[:] = np.nanstd(log_returns, axis=0)



Note the above code was edited from the original.

      returns[returns < 0] = np.nan



was changed to

        returns[returns > 0] = np.nan


per the post below.

Fantastic, thanks for the help as always Dan!

Looking at the above code...

Shouldn't

# set any returns less than 0 to NaN so they are excluded from our calculations
returns[returns < 0] = np.nan

actually be:
returns[returns > 0] = np.nan

if you want downside volatility? Don't you want to set any returns ABOVE zero to NAN so you are only capturing the standard deviation of downside movements? I think your original code is actually calculating upside volatility.

Let me know if I'm missing something...

-Brian

Oops. Yes. The above would be upside volatility. My mistake. I edited the above so now it's correct. Thanks for pointing that out.

I'm wondering if I have the coding correct for the two custom factors below. Can someone lend a hand to let me know if the lines in BOLD are correctly written?

*# get index and calculate returns. SPY code is 8554
benchmark_index = np.where((assets == 8554) == True)[0][0]
*

# Index Beta
class Index_Beta(CustomFactor):
"""
Index Beta:
Slope coefficient of 1-year regression of price returns against index returns
Notes:
High value suggests high market risk
Slope calculated using regression MLE
"""
inputs = [USEquityPricing.close]
window_length = 252
def compute(self, today, assets, out, close):
**# get index and calculate returns. SPY code is 8554
benchmark_index = np.where((assets == 8554) == True)[0][0]**
benchmark_close = close[:, benchmark_index]
benchmark_returns = (
(benchmark_close - np.roll(benchmark_close, 1)) / np.roll(benchmark_close, 1))[1:]
betas = []
# get beta for individual securities using MLE
for col in close.T:
col_returns = ((col - np.roll(col, 1)) / np.roll(col, 1))[1:]
col_cov = np.cov(col_returns, benchmark_returns)
betas.append(col_cov[0, 1] / col_cov[1, 1])
out[:] = betas

# Downside Beta
class Downside_Beta(CustomFactor):
"""
Downside Beta:
Slope coefficient of 1-year regression of price returns against negative index returns
Notes:
High value suggests high exposure to the downmarket
Slope calculated using regression MLE
"""
inputs = [USEquityPricing.close]
window_length = 252
def compute(self, today, assets, out, close):
**# get index and calculate returns. SPY code is 8554
benchmark_index = np.where((assets == 8554) == True)[0][0]**
benchmark_close = close[:, benchmark_index]
benchmark_returns = (
(benchmark_close - np.roll(benchmark_close, 1)) / np.roll(benchmark_close, 1))[1:]
# days where benchmark is negative
negative_days = np.argwhere(benchmark_returns < 0).flatten()
# negative days for benchmark
bmark_neg_day_returns = [benchmark_returns[i]
for i in negative_days]
betas = []
# get beta for individual securities using MLE
for col in close.T:
col_returns = ((col - np.roll(col, 1)) / np.roll(col, 1))[1:]
col_neg_day_returns = [col_returns[i] for i in negative_days]
col_cov = np.cov(col_neg_day_returns, bmark_neg_day_returns)
betas.append(col_cov[0, 1] / col_cov[1, 1])
out[:] = betas


Hi Daniel,

Didn't test, but here is a working version using OLS if interested.

def REG(X,Y):
model = sm.OLS(Y, X).fit()

return model.params

class Downside_Beta(CustomFactor):
inputs = [USEquityPricing.close]
window_length = 252
def compute(self, today, assets, out, close):
bench = close[:, assets.searchsorted(8554)]
benchmark_returns = np.diff(np.log(bench))
negative_bench_returns = np.where(benchmark_returns < 0, benchmark_returns, 0)
betas=[]
for col_ix, values in enumerate(close.T):
returns = np.diff(np.log(values))
betas.append(REG(negative_bench_returns, returns))

out[:]=betas