Skip to content

Instantly share code, notes, and snippets.

@robcarver17
Last active October 9, 2022 03:14
Show Gist options
  • Save robcarver17/fd921a6d0146090ec1a2335c3ec28395 to your computer and use it in GitHub Desktop.
Save robcarver17/fd921a6d0146090ec1a2335c3ec28395 to your computer and use it in GitHub Desktop.
import pandas as pd
import numpy as np
def get_expected_risk_for_system(system):
value_of_positions_proportion_capital = get_positions_as_proportion_of_capital(system)
instrument_returns = get_instrument_returns(system)
instrument_returns = instrument_returns.ffill().reindex(value_of_positions_proportion_capital.index)
rolling_std = instrument_returns.ewm(span=30).std()
rolling_corr = instrument_returns.ewm(span=120).corr()
list_of_instruments = system.get_instrument_list()
expected_risk = calc_expected_risk_over_time(rolling_corr, rolling_std, value_of_positions_proportion_capital,
list_of_instruments)
return expected_risk
def get_positions_as_proportion_of_capital(system):
list_of_instruments = system.get_instrument_list()
positions = [system.portfolio.get_actual_position(instrument_code) for instrument_code in list_of_instruments]
positions = pd.concat(positions, axis=1)
positions.columns = list_of_instruments
positions[positions.isna()] = 0.0
## Need the notional exposure of each position
block_sizes = [system.data.get_value_of_block_price_move(instrument_code) for instrument_code in
list_of_instruments]
block_sizes = np.array([block_sizes] * len(positions), ndmin=2)
block_sizes = pd.DataFrame(block_sizes, index=positions.index, columns=list_of_instruments)
fx_rates = [system.data.get_fx_for_instrument(instrument_code, system.config.base_currency) for instrument_code in
list_of_instruments]
fx_rates = pd.concat(fx_rates, axis=1)
fx_rates.columns = list_of_instruments
fx_rates = fx_rates.ffill().reindex(positions.index)
prices = [system.rawdata.daily_denominator_price(instrument_code) for instrument_code in list_of_instruments]
prices = pd.concat(prices, axis=1)
prices.columns = list_of_instruments
prices = prices.ffill().reindex(positions.index)
capital = system.config.notional_trading_capital
value_of_each_contract = prices * block_sizes * fx_rates
value_of_positions = value_of_each_contract * positions
value_of_positions_proportion_capital = value_of_positions / capital
return value_of_positions_proportion_capital
def get_instrument_returns(system):
list_of_instruments = system.get_instrument_list()
instrument_returns = [system.rawdata.get_percentage_returns(instrument_code) for instrument_code in
list_of_instruments]
instrument_returns = pd.concat(instrument_returns, axis=1)
instrument_returns.columns = list_of_instruments
return instrument_returns
def calc_expected_risk_over_time(rolling_corr, rolling_std, value_of_positions_proportion_capital, list_of_instruments):
positions_index = value_of_positions_proportion_capital.index
risk = [calc_risk_for_date(rolling_corr, rolling_std, index_date,
value_of_positions_proportion_capital, list_of_instruments)
for index_date in positions_index]
risk_series = pd.Series(risk, index=positions_index)
return risk_series
def calc_risk_for_date(rolling_corr, rolling_std, index_date, value_of_positions_proportion_capital,
list_of_instruments):
std_dev = rolling_std.loc[index_date].values
std_dev[np.isnan(std_dev)] = 0.0
weights = value_of_positions_proportion_capital.loc[index_date].values
weights[np.isnan(weights)] = 0.0
cmatrix = get_corr_matrix_for_date(rolling_corr, index_date, list_of_instruments)
sigma = sigma_from_corr_and_std(std_dev, cmatrix)
portfolio_variance = weights.dot(sigma).dot(weights.transpose())
portfolio_std = portfolio_variance ** .5
annualised_portfolio_std = portfolio_std * 16.0
return annualised_portfolio_std
def get_corr_matrix_for_date(rolling_corr, index_date, list_of_instruments):
cmatrix = [rolling_corr.loc[(str(index_date), instrument)].values for instrument in list_of_instruments]
cmatrix = pd.DataFrame(cmatrix)
cmatrix[cmatrix.isna()] = 0.0
return np.array(cmatrix)
from syscore.optimisation_utils import sigma_from_corr_and_std
## or use this function
def sigma_from_corr_and_std(stdev_list, corrmatrix):
sigma = np.diag(stdev_list).dot(corrmatrix).dot(np.diag(stdev_list))
return sigma
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment