get system risk

scratch_16.py

import pandas as pd
import numpy as np

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_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)

## or use this function
def sigma_from_corr_and_std(stdev_list, corrmatrix):
    stdev = np.array(stdev_list, ndmin=2).transpose()
    sigma = stdev * corrmatrix * stdev

    return sigma



def correlation_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

    ## Use absolute
    weights = value_of_positions_proportion_capital.abs().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 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 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 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 get_expected_risk_for_system(system):

    value_of_positions_proportion_capital = get_positions_as_proportion_of_capital(system)

    instrument_returns = get_instrument_returns(system)

    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