from .index_data import get_index_quotes, index_returns from . import on_the_run from db import serenitas_engine, dawn_engine from analytics import CreditIndex, Portfolio from analytics.utils import roll_date from dateutil.relativedelta import relativedelta from analytics.basket_index import MarkitBasketIndex from statsmodels.sandbox.regression.predstd import wls_prediction_std from scipy.interpolate import interp1d from itertools import chain from copy import deepcopy from matplotlib import cm import datetime import pandas as pd import math import statsmodels.formula.api as smf import numpy as np import matplotlib.pyplot as plt def curve_spread_diff(index='IG', rolling=6, years=3, percentage=False, percentage_base='5yr'): otr = on_the_run(index) # look at spreads df = get_index_quotes(index, list(range(otr - rolling, otr + 1)), tenor=['3yr', '5yr', '7yr', '10yr'], years=years) spreads = df.groupby(level=['date', 'tenor']).nth(-1)['close_spread'].unstack(-1) spreads_diff = spreads.diff(axis=1) del spreads_diff['3yr'] spreads_diff.columns = ['3-5', '5-7', '7-10'] spreads_diff['5-10'] = spreads_diff['5-7'] + spreads_diff['7-10'] if percentage is True: spreads_diff = spreads.apply(lambda df: df/df[percentage_base], axis=1) return spreads_diff def spreads_diff_table(spreads_diff): def current(s): return s.iat[-1] def zscore(s): return (s.iat[-1] - s.mean()) / s.std() df = spreads_diff.agg(['min', 'max', 'mean', current, zscore]) ((spreads_diff - spreads_diff.mean())/spreads_diff.std()).plot() return df def theta_matrix_by_series(index='IG', rolling=6): otr = on_the_run(index) df = get_index_quotes(index, list(range(otr - rolling, otr + 1)), tenor=['3yr', '5yr', '7yr', '10yr']) #now get_index_quotes are all based on theta2/duration2 df['theta_per_dur'] = df.theta / df.duration theta_matrix = df.groupby(level=['date', 'tenor','series']).nth(-1)['theta_per_dur'] theta_matrix = theta_matrix.loc[theta_matrix.index[-1][0]].unstack(0) return theta_matrix[['3yr', '5yr', '7yr', '10yr']] def ratio_within_series(index='IG', rolling=6, param='duration'): otr = on_the_run(index) df = get_index_quotes(index, list(range(otr - rolling, otr + 1)), tenor=['3yr', '5yr', '7yr', '10yr']).unstack() ratio = (df[param]. apply(lambda s: s / df[param]['5yr'].values, raw=True)) ratio.columns = pd.MultiIndex.from_product([[f"{param}_ratio_to_5yr"], ratio.columns]) df = df.join(ratio).groupby(['date']).tail(1) df = df.reset_index(level=['index', 'version'], drop=True) return df def on_the_run_theta(index='IG', rolling=6): otr = on_the_run(index) df = get_index_quotes(index, list(range(otr - rolling, otr + 1)), tenor=['3yr', '5yr', '7yr', '10yr']) df['theta_per_dur'] = df.theta/df.duration theta_matrix = df.groupby(level=['date', 'tenor']).nth(-1)['theta_per_dur'] theta_matrix.unstack(-1).plot() def curve_returns(index='IG', rolling=6, years=3): # look at returns otr = on_the_run(index) df = index_returns(index=index, series=list(range(otr - rolling, otr + 1)), tenor=['3yr', '5yr', '7yr', '10yr'], years=years) # on-the-run returns df = df.reset_index('index', drop=True) returns = df.price_return.dropna().unstack('tenor').groupby(level='date').nth(-1) strategies_return = pd.DataFrame( {'5-10': 1.78 * returns['5yr'] - returns['10yr'], '7-10': 1.33 * returns['7yr'] - returns['10yr'], '3-5-10': -2 * returns['3yr'] + 3 * returns['5yr'] - returns['10yr'], '3-5': returns['5yr'] - 1.56 * returns['3yr'], '3-7': returns['7yr'] - 2.07 * returns['3yr'], '5yr long': returns['5yr']}) return strategies_return def curve_returns_stats(strategies_return): ''' Takes a curve_return df''' strategies_return_monthly = (strategies_return. groupby(pd.Grouper(freq='M')). agg(lambda df: (1 + df).prod() - 1)) def sharpe(df, period="daily"): if period == "daily": return df.mean() / df.std() * math.sqrt(252) else: return df.mean() / df.std() * math.sqrt(12) results = strategies_return.agg([sharpe, lambda df: df.nsmallest(10).mean(), lambda df: df.std()]) sharpe_monthly = strategies_return_monthly.agg(sharpe, period="monthly") sharpe_monthly.name = 'Monthly Sharpe' results.index = ['Sharpe', 'Mean Worst 10 Days DrawDown', 'Standard Deviation'] return results.append(sharpe_monthly) def cross_series_curve(index='IG', rolling=6): otr = on_the_run(index) df = index_returns(index= index, series=list(range(otr - rolling, otr + 1)), tenor=['3yr', '5yr', '7yr', '10yr']) # look cross series - 3y to 5y df = df.reset_index().set_index(['date', 'index', 'tenor', 'series']) returns1 = df.xs(['5yr', index], level=['tenor','index']).price_return.unstack(-1) price_diff = pd.DataFrame() for ind in list(range(otr - 2, otr + 1)): price_diff[ind] = returns1[ind] - 1.6 * returns1[ind - 4] price_diff = price_diff.stack().groupby(level='date').nth(-1) monthly_returns_cross_series = (price_diff. groupby(pd.Grouper(freq='M')). agg(lambda df: (1 + df).prod() - 1)) plt.plot(monthly_returns_cross_series) def forward_loss(index='IG'): start_date = (pd.Timestamp.now() - pd.DateOffset(years=3)).date() df = pd.read_sql_query("SELECT date, index, series, tenor, duration, close_spread, "\ "close_spread*duration / 100 AS indexel " \ "FROM index_quotes WHERE index=%s AND date >= %s " \ "ORDER BY date DESC, series ASC, duration ASC", serenitase_engine, parse_dates=['date'], params=[index, start_date]) df1 = pd.read_sql_query("SELECT index, series, tenor, maturity FROM index_maturity", serenitas_engine, parse_dates=['maturity']) df = df.merge(df1, on=['index','series','tenor']) df = df.set_index(['date','index', 'maturity']).dropna() df = df.groupby(level=['date','index', 'maturity']).nth(-1) # annual change, to take out some noise df['fwd_loss_rate'] = df.indexel.diff(2)/df.duration.diff(2) def curve_model(tenor_1='5yr', tenor_2='10yr'): #OLS model df = ratio_within_series(param='close_spread') df = pd.concat([df.duration[tenor_1], df.duration[tenor_2], df.close_spread[tenor_1], df.close_spread_ratio_to_5yr[tenor_2], df.theta[tenor_1], df.theta[tenor_2]], axis=1, keys=['duration1', 'duration2', 'close_spread', 'ratio', 'theta1', 'theta2']) df = np.log(df) ols_model = smf.ols('ratio ~ close_spread + duration1 + theta1 + theta2', data=df).fit() return df, ols_model def curve_model_results(df, model): df = df.dropna() a, b, c = wls_prediction_std(model) b.name = 'down_2_stdev' c.name = 'up_2_stdev' df = df.join(b) df = df.join(c) #dr/dspread = exp(k) + spread_coeff * duration ^ dur_coeff * spread ^ (spread_coeff-1) cols = ['ratio', 'close_spread', 'down_2_stdev', 'up_2_stdev'] df[cols] = np.exp(df[cols]) df['predicted'] = np.exp(model.predict()) df[['predicted', 'down_2_stdev', 'up_2_stdev']]=\ df[['predicted', 'down_2_stdev', 'up_2_stdev']].multiply(df['close_spread'].values, axis=0) ax = df[['predicted', 'down_2_stdev', 'up_2_stdev']].reset_index(level='series', drop=True).plot() df['dr_dspread'] = np.exp(model.params[0]) * model.params[2] * df.duration1 ** model.params[1] * df.close_spread ** (model.params[2] - 1) return df def spread_fin_crisis(index='IG'): otr = on_the_run(index) # look at spreads df = get_index_quotes(index, list(range(8, otr + 1)), tenor=['3yr', '5yr', '7yr', '10yr'], years=20) spreads = df.groupby(level=['date', 'tenor']).nth(-1)['close_spread'].unstack(-1) spreads_diff = spreads.diff(axis=1) to_plot = pd.DataFrame() to_plot['spread'] = spreads['5yr'] to_plot['3 - 5 diff'] = spreads_diff['5yr'] to_plot['5 - 10 diff'] = spreads_diff['7yr'] + spreads_diff['10yr'] fig = plt.figure() ax = fig.add_subplot(111) ax2 = ax.twinx() # Create another axes that shares the same x-axis as ax width = 0.4 to_plot['spread'].plot(color='red', ax=ax) to_plot['5 - 10 diff'].plot(color='blue', ax=ax2) to_plot['3 - 5 diff'].plot(color='green', ax=ax2) plt.legend(bbox_to_anchor=(.5, -.1), ncol = 2) plt.show() def forward_spread(report_date, index='IG', series=None, tenors=['3yr', '5yr', '7yr', '10yr']): if series is None: series = on_the_run(index = index) b_index = MarkitBasketIndex(index, series, tenors, value_date=report_date) b_index.tweak() f_spread = [] date_range = pd.bdate_range(pd.datetime.today(), max(b_index.maturities), freq='M') for d in date_range.date: b_index.value_date = d f_spread.append(b_index.spread()) return pd.concat(f_spread, keys=date_range).unstack(-1) def spot_forward(index='IG', series=None, tenors=['3yr', '5yr', '7yr', '10yr']): ''' Calculates the 1-year forward spot rate ''' if series is None: series = on_the_run(index) b_index = MarkitBasketIndex(index, series, tenors) b_index.tweak() spreads_current = b_index.spread() spreads_current.name = 'current' spreads_1yr = pd.Series([b_index.spread(m - relativedelta(years=1), b_index.coupon(m)) \ for m in b_index.maturities], index=tenors) spreads_1yr.name = '1yr' df = pd.concat([spreads_current, spreads_1yr], axis=1) maturity_1yr = roll_date(b_index.index_desc.issue_date[0], 1) df_0 = pd.DataFrame({'current':[0., b_index.spread(maturity_1yr, 0.01 if index == "IG" else 0.05)], '1yr': [0., 0.]}, index=['0yr', '1yr']) df_0.index.name = 'tenor' df = df_0.append(df) df['maturity'] = [b_index.value_date, maturity_1yr] + b_index.maturities return df.reset_index().set_index('maturity') def curve_pos(value_date, index_type='IG'): ''' value_date : :class:`datetime.date` index : string one of 'IG', 'HY' or 'EU' Returns a Portfolio of curve trades ''' if index_type == "EU": index_type = "ITRX" sql_string = "SELECT index, series, tenor, notional "\ "FROM list_cds_positions(%s, %s) " \ "JOIN index_desc " \ "ON security_id=redindexcode AND " \ "index_desc.maturity=list_cds_positions.maturity" df = pd.read_sql_query(sql_string, dawn_engine, params=[value_date, f'SER_{index_type}CURVE']) portf = Portfolio([CreditIndex(row.index, row.series, row.tenor, value_date, -row.notional) for row in df[['index', 'tenor', 'series', 'notional']]. itertuples(index=False)]) portf.mark() return portf def curve_shape(value_date, index='IG', percentile=.95, spread=None): ''' Returns a function to linearly interpolate between the curve based on maturity (in years)''' curve_shape = curve_spread_diff(index, 10, 5, True) steepness = (curve_shape['10yr']/curve_shape['3yr']) series = on_the_run(index) if spread is None: sql_string = "SELECT closespread FROM index_quotes where index = %s " \ "and series = %s and tenor = %s and date = %s" spread_df = pd.read_sql_query(sql_string, serenitas_engine, params=[index, series, '5yr', value_date]) spread = spread_df.iloc[0][0] sql_string = "SELECT tenor, maturity FROM index_maturity where index = %s and series = %s" lookup_table = pd.read_sql_query(sql_string, serenitas_engine, parse_dates=['maturity'], params=[index, series]) df = curve_shape[steepness == steepness.quantile(percentile, 'nearest')] df = df * spread/df['5yr'][0] df = df.stack().rename('spread') df = df.reset_index().merge(lookup_table, on=['tenor']) df['year_frac'] = (df.maturity - pd.to_datetime(value_date)).dt.days/365 return interp1d(np.hstack([0, df.year_frac]), np.hstack([0, df.spread])) def plot_curve_shape(date): ''' Plots the curve shape that's being used for the scenarios''' curve_per = np.arange(.01, .99, .1) time_per = np.arange(.1, 10.1, .5) r=[] for per in curve_per: shape = curve_shape(date, percentile = per) r.append(shape(time_per)) df = pd.DataFrame(r, index=curve_per, columns=time_per) fig = plt.figure() ax = fig.gca(projection='3d') xx, yy = np.meshgrid(curve_per, time_per) z = np.vstack(r).transpose() surf = ax.plot_surface(xx, yy, z, cmap=cm.viridis) ax.set_xlabel("steepness percentile") ax.set_ylabel("tenor") ax.set_zlabel("spread") def pos_pnl_abs(portf, value_date, index='IG', rolling=6, years=3): ''' Runs PNL analysis on portf using historical on-the-run spread levels - off-the-runs spreads are duration linearly interpolated''' series = on_the_run(index) df = get_index_quotes(index, list(range(series - rolling, series + 1)), tenor=['3yr', '5yr', '7yr', '10yr'], years=years) df = df.groupby(level=['date', 'tenor']).nth(-1)['close_spread'].unstack(-1) sql_string = "SELECT tenor, maturity FROM index_maturity where index = %s and series = %s" lookup_table = pd.read_sql_query(sql_string, serenitas_engine, parse_dates=['maturity'], params=[index, series]) lookup_table['year_frac'] = (lookup_table.maturity - pd.to_datetime(value_date)).dt.days/365 portf_copy = deepcopy(portf) portf_copy.reset_pv() r = [] for date, row in df.iterrows(): f = interp1d(np.hstack([0, lookup_table['year_frac']]), np.hstack([row[0]/2, row])) for ind in portf_copy.indices: ind.spread = f((ind.end_date - value_date).days/365) r.append([[date, f(5)] + [portf_copy.pnl]]) df = pd.DataFrame.from_records(chain(*r), columns=['date', 'five_yr_spread', 'pnl']) return df.set_index('date') def curve_scen_table(portf, shock=10): ''' Runs PNL scenario on portf by shocking different points on the curve. off-the-runs shocks are linearly interpolated''' otr_year_frac = np.array([(e - portf.value_date).days / 365 \ for e in roll_date(portf.value_date, [3, 5, 10])]) portf_year_frac = [(ind.end_date - ind.value_date).days / 365 for ind in portf.indices] r = [] for i, tenor1 in enumerate(['3yr', '5yr', '10yr']): for j, tenor2 in enumerate(['3yr', '5yr', '10yr']): shocks = np.full(4, 0) shocks[i+1] += shock shocks[j+1] -= shock # f is the shock amount interpolated based on tenor f = interp1d(np.hstack([0, otr_year_frac]), shocks) portf_copy = deepcopy(portf) portf_copy.reset_pv() for ind, yf in zip(portf_copy.indices, portf_year_frac): ind.spread += float(f(yf)) r.append((tenor1, tenor2, portf_copy.pnl)) return pd.DataFrame.from_records(r, columns=['tighter', 'wider', 'pnl'])