diff options
Diffstat (limited to 'python/analytics/curve_trades.py')
| -rw-r--r-- | python/analytics/curve_trades.py | 453 |
1 files changed, 273 insertions, 180 deletions
diff --git a/python/analytics/curve_trades.py b/python/analytics/curve_trades.py index ea3ee3b1..6eb13d7f 100644 --- a/python/analytics/curve_trades.py +++ b/python/analytics/curve_trades.py @@ -19,18 +19,24 @@ import numpy as np import matplotlib.pyplot as plt -def curve_spread_diff(index='IG', rolling=6, years=3, percentage=False, percentage_base='5yr'): +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) + 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'] + 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) + spreads_diff = spreads.apply(lambda df: df / df[percentage_base], axis=1) return spreads_diff @@ -40,71 +46,85 @@ def spreads_diff_table(spreads_diff): 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() + + 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): +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'] + 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']] + return theta_matrix[["3yr", "5yr", "7yr", "10yr"]] -def ratio_within_series(index='IG', rolling=6, param='duration'): +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) + 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): +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'] + 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): + +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) + 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) + 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']}) + { + "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''' + """ + Takes a curve_return df""" - strategies_return_monthly = (strategies_return. - groupby(pd.Grouper(freq='M')). - agg(lambda df: (1 + df).prod() - 1)) + 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": @@ -112,127 +132,158 @@ def curve_returns_stats(strategies_return): else: return df.mean() / df.std() * math.sqrt(12) - results = strategies_return.agg([sharpe, lambda df: df.nsmallest(10).mean(), lambda df: df.std()]) + 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'] + 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): +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']) + 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) + 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)) + 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'): +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 = 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) + 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) + 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']) +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() + 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' + 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'] + # 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) + 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'): +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) + 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'] + 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 + 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) + 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=(0.5, -0.1), ncol=2) plt.show() -def forward_spread(report_date, index='IG', series=None, tenors=['3yr', '5yr', '7yr', '10yr']): +def forward_spread( + report_date, index="IG", series=None, tenors=["3yr", "5yr", "7yr", "10yr"] +): if series is None: - series = on_the_run(index = index) + 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') + 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']): +def spot_forward(index="IG", series=None, tenors=["3yr", "5yr", "7yr", "10yr"]): - ''' - Calculates the 1-year forward spot rate ''' + """ + Calculates the 1-year forward spot rate """ if series is None: series = on_the_run(index) @@ -240,89 +291,114 @@ def spot_forward(index='IG', series=None, tenors=['3yr', '5yr', '7yr', '10yr']): 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' + 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_0 = pd.DataFrame( + { + "current": [ + 0.0, + b_index.spread(maturity_1yr, 0.01 if index == "IG" else 0.05), + ], + "1yr": [0.0, 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') + 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'): +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 ''' + 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']) + 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 = 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): +def curve_shape(value_date, index="IG", percentile=0.95, spread=None): - ''' + """ Returns a function to linearly interpolate between the curve - based on maturity (in years)''' + based on maturity (in years)""" curve_shape = curve_spread_diff(index, 10, 5, True) - steepness = (curve_shape['10yr']/curve_shape['3yr']) + 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 " \ + 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_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]) + 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 + 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''' + """ + 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=[] + curve_per = np.arange(0.01, 0.99, 0.1) + time_per = np.arange(0.1, 10.1, 0.5) + r = [] for per in curve_per: - shape = curve_shape(date, percentile = 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') + 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) @@ -331,48 +407,65 @@ def plot_curve_shape(date): ax.set_zlabel("spread") -def pos_pnl_abs(portf, value_date, index='IG', rolling=6, years=3): +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''' + 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) + 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 + 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])) + 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) + 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') + 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] + 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']): + 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 + 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) @@ -380,4 +473,4 @@ def curve_scen_table(portf, shock=10): 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']) + return pd.DataFrame.from_records(r, columns=["tighter", "wider", "pnl"]) |