diff options
Diffstat (limited to 'python/analytics/cms_spread.py')
| -rw-r--r-- | python/analytics/cms_spread.py | 58 |
1 files changed, 38 insertions, 20 deletions
diff --git a/python/analytics/cms_spread.py b/python/analytics/cms_spread.py index ff222268..49a59c71 100644 --- a/python/analytics/cms_spread.py +++ b/python/analytics/cms_spread.py @@ -7,9 +7,14 @@ from math import exp, sqrt, log from .black import bachelier from bbg_helpers import BBG_IP, retrieve_data, init_bbg_session from quantlib.time.api import ( - Date, Period, Days, Years, UnitedStates, Actual365Fixed, today) -from quantlib.time.calendars.united_states import GOVERNMENTBOND + Date, Period, Days, Months, Years, UnitedStates, Actual365Fixed, today, + Following) +from quantlib.time.calendars.united_states import GovernmentBond from quantlib.indexes.swap.usd_libor_swap import UsdLiborSwapIsdaFixAm +from quantlib.experimental.coupons.swap_spread_index import SwapSpreadIndex +from quantlib.termstructures.volatility.api import ( + VolatilityType, SwaptionVolatilityMatrix) +from quantlib.math.matrix import Matrix from scipy.interpolate import RectBivariateSpline from yieldcurve import YC from db import dbconn @@ -58,36 +63,49 @@ def get_forward_spread(tenor1, tenor2, maturity): yc.extrapolation = True conn = dbconn('serenitasdb') fixing_date, fixing1, fixing2 = get_fixings(conn, tenor1, tenor2) - USISDA1 = UsdLiborSwapIsdaFixAm(Period(tenor1, Years), - forwarding=yc, discounting=yc) + USISDA1 = UsdLiborSwapIsdaFixAm(Period(tenor1, Years), yc, yc) USISDA1.add_fixing(fixing_date, fixing1) - USISDA2 = UsdLiborSwapIsdaFixAm(Period(tenor2, Years), - forwarding=yc, discounting=yc) + USISDA2 = UsdLiborSwapIsdaFixAm(Period(tenor2, Years), yc, yc) USISDA2.add_fixing(fixing_date, fixing2) - expiration = UnitedStates(GOVERNMENTBOND).advance( + spread_index = SwapSpreadIndex(f"{tenor1}-{tenor2}", USISDA1, USISDA2) + expiration = UnitedStates(GovernmentBond).advance( Date.from_datetime(maturity), 0, Days) + return spread_index.fixing(expiration) - USFS1 = USISDA1.underlying_swap(expiration) - USFS2 = USISDA2.underlying_swap(expiration) - return USFS2.fair_rate - USFS1.fair_rate - -def get_swaption_vol_surface(source="ICPL", vol_type="N"): - if vol_type == "N": +def get_swaption_vol_data(source="ICPL", vol_type=VolatilityType.ShiftedLognormal): + if vol_type == VolatilityType.Normal: table_name = "swaption_normal_vol" else: table_name = "swaption_lognormal_vol" - sql_str = "SELECT * FROM {table_name} WHERE source = %s ORDER BY date DESC LIMIT 1" + sql_str = f"SELECT * FROM {table_name} WHERE source = %s ORDER BY date DESC LIMIT 1" conn = dbconn('serenitasdb') with conn.cursor() as c: c.execute(sql_str, (source,)) surf_data = next(c) - date, surf = surf_data[0], np.array(surf_data[1:-1]) - if source == "ICPL": - tenors = [1/12, 0.25, 0.5] + list(range(1, 11)) + [15., 20., 25., 30.] - else: - tenors = [1/12, 0.25, 0.5, 0.75] + list(range(1, 11)) + [15., 20., 25., 30.] - return RectBivariateSpline(tenors, tenors[-14:], surf.T) + return surf_data[0], np.array(surf_data[1:-1], order='F').T, vol_type + +def get_swaption_vol_surface(date, data, vol_type): + tenors = [1/12, 0.25, 0.5, 0.75] + list(range(1, 11)) + [15., 20., 25., 30.] + return RectBivariateSpline(tenors, tenors[-14:], data.T) + +def get_swaption_vol_matrix(date, data, vol_type): + # figure out what to do with nan + calendar = UnitedStates() + m = Matrix.from_ndarray(data) + option_tenors = [Period(i, Months) for i in [1, 3, 6, 9]] + \ + [Period(i, Years) for i in range(1, 11)] + \ + [Period(i, Years) for i in [15, 20, 25, 30]] + swap_tenors = option_tenors[-14:] + return (SwaptionVolatilityMatrix. + from_reference_date(Date.from_datetime(date), + calendar, + Following, + option_tenors, + swap_tenors, + m, + Actual365Fixed(), + vol_type=vol_type)) def plot_surf(surf, tenors): xx, yy = np.meshgrid(tenors, tenors[-14:]) |