Overview
Simple measures of volatility can be calculated using standard python scripts.Straight Vol
code that uses a straight calculation of variance as the measure of volatiity.
data = [{"return":0.01},{"return":-0.01},{"return":0.002},{"return":0.013},{"return":0.007},{"return":-0.017},{"return":-0.01},];
square_returns = [x['return']*x['return'] for x in data]
vol = sum(square_returns)/len(square_returns)
EWMA Vol
The following calculates an ewma of the volatility of a price series.
let data = [{"return":0.01},{"return":-0.01},{"return":0.002},{"return":0.013},{"return":0.007},{"return":-0.017},{"return":-0.01},];
def exponential_weights(data, alpha=0.5):
weighted_values = []
current_weight = 1.0
accumulated = 0.0
weight_sum = 0.0
for val in data:
accumulated = accumulated * (1 - alpha) + val * alpha
weighted_values.append(accumulated)
return weighted_values
square_returns = [x['return']*x['return'] for x in data]
vols = exponential_weights(square_returns)
Full Script
from arch import arch_model
import pandas as pd
import math
def log_diff(first, second):
return math.log(second) - math.log(first)
def arithmetic_return(first, second):
return second/first - 1
def returns(data,tickers = None, diff=None):
results = []
if diff == None:
diff = log_diff
for index, item in enumerate(data):
if index>0:
record = {}
results.append(record)
for ticker in tickers:
record[ticker] = diff(data[index-1][ticker], item[ticker])
pass
pass
pass
return results
def variance(data, ticker, prices=True):
total = 0
if prices == True:
data = returns([{ticker:x[ticker]} for x in data], [ticker])
for item in data:
total += item[ticker]*item[ticker]
return total/len(data)
def covariance(data, ticker1, ticker2):
pass
def ewma_correlation(data, span=2, tickers=None, prices=True):
cov = ewma_covariance(data, span, tickers, prices)
column_list = cov.columns.tolist()
cov2 = cov.copy()
for ticker in column_list:
for ticker2 in column_list:
covar = cov.loc[ticker,ticker2]
var1 = cov.loc[ticker,ticker]
var2 = cov.loc[ticker2, ticker2]
correlation = covar/(math.sqrt(var1)*math.sqrt(var2))
cov2.loc[ticker,ticker2] = correlation
pass
pass
return cov2
def ewma_covariance(data, span=2, tickers=None, prices=False):
ticks = None
if tickers == None:
ticks = data[0].keys()
pass
if prices == True:
data = returns(data, ticks)
if tickers != None:
mapped = []
for item in data:
record = {}
mapped.append(record)
for ticker in tickers:
record[ticker] = item[ticker]
pass
pass
data = mapped
df = pd.DataFrame(data)
# Calculate EWMA covariance using a decay factor (com = center of mass, or use alpha/halflife)
# span = 2 corresponds to alpha = 2 / (span + 1)
ewma_cov = df.ewm(span=span).cov()
# Get the covariance matrix for the very last date/row
last_cov = df.ewm(span=span).cov().loc[df.index[-1]]
return last_cov
#last_cov_matrix = ewma_cov.iloc[-len(ticks):]
#return last_cov_matrix
def univariate_garch(data, prices=True):
cov = ewma_correlation(data)
column_list = cov.columns.tolist()
if prices == True:
data = returns(data, column_list)
data = pd.DataFrame(data)
vols = {}
for ticker in column_list:
model = arch_model(data[ticker], vol="Garch", p=1, q=1, dist="Normal")
# 4. Fit the model
model_results = model.fit(update_freq=5)
vol = model_results.conditional_volatility[1]
vols[ticker] = vol
'''
volatility = model_results.conditional_volatility
Get the most recent single volatility number (the last data point):
latest_vol = model_results.conditional_volatility.iloc[-1]
Get conditional variance instead (squared volatility):
variance = model_results.conditional_variance [1]
(https://arch.readthedocs.io/en/stable/univariate/generated/generated/arch.univariate.base.ARCHModelResult.conditional_volatility.html)
'''
pass
for ticker in column_list:
for ticker2 in column_list:
value = cov.loc[ticker,ticker2]
cov.loc[ticker,ticker2] = value * vols[ticker] * vols[ticker]
pass
pass
return cov
'''
last_cov = ewma_cov.iloc[-num_assets:].copy()
# Remove the outer level of the index (the timestamp/step) to make it a clean N x N matrix
last_cov = last_cov.droplevel(0)
# 2. Modify an off-diagonal value (Asset_A vs Asset_B)
# To keep the matrix valid, update BOTH symmetric positions
last_cov.at['Asset_A', 'Asset_B'] = 0.0015
last_cov.at['Asset_B', 'Asset_A'] = 0.0015
# Modify a diagonal value (Variance of Asset_A)
last_cov.at['Asset_A', 'Asset_A'] = 0.0025
'''