diff options
Diffstat (limited to 'python')
| -rw-r--r-- | python/yieldcurve.py | 317 |
1 files changed, 217 insertions, 100 deletions
diff --git a/python/yieldcurve.py b/python/yieldcurve.py index aebb2c64..14faaa74 100644 --- a/python/yieldcurve.py +++ b/python/yieldcurve.py @@ -7,14 +7,32 @@ import lz4.block import os import pandas as pd from quantlib.settings import Settings -from quantlib.time.api import (WeekendsOnly, Japan, - Date, Period, Days, Schedule, Annual, - Semiannual, today, Actual360, Months, Years, - ModifiedFollowing, Thirty360, Actual365Fixed) +from quantlib.time.api import ( + WeekendsOnly, + Japan, + Date, + Period, + Days, + Schedule, + Annual, + Semiannual, + today, + Actual360, + Months, + Years, + ModifiedFollowing, + Thirty360, + Actual365Fixed, +) from quantlib.currency.api import USDCurrency, EURCurrency, JPYCurrency from quantlib.indexes.ibor_index import IborIndex from quantlib.termstructures.yields.api import ( - PiecewiseYieldCurve, DepositRateHelper, SwapRateHelper, BootstrapTrait, Interpolator) + PiecewiseYieldCurve, + DepositRateHelper, + SwapRateHelper, + BootstrapTrait, + Interpolator, +) from quantlib.time.date import pydate_from_qldate import numpy as np @@ -31,8 +49,7 @@ def load_curves(currency="USD", date=None): if date: sql_str += " WHERE effective_date=%s" - - with closing(dbconn('serenitasdb')) as conn: + with closing(dbconn("serenitasdb")) as conn: with conn.cursor() as c: if date: c.execute(sql_str, (date,)) @@ -41,16 +58,19 @@ def load_curves(currency="USD", date=None): return YieldCurve.from_bytes(lz4.block.decompress(curve)) else: c.execute(sql_str) - return sorteddict([ - (d, YieldCurve.from_bytes(lz4.block.decompress(curve))) - for d, curve in c]) + return sorteddict( + [ + (d, YieldCurve.from_bytes(lz4.block.decompress(curve))) + for d, curve in c + ] + ) def get_curve(effective_date, currency="USD"): - if f'_{currency}_curves' in globals(): - curves = globals()[f'_{currency}_curves'] + if f"_{currency}_curves" in globals(): + curves = globals()[f"_{currency}_curves"] else: - curves = globals()[f'_{currency}_curves'] = load_curves(currency) + curves = globals()[f"_{currency}_curves"] = load_curves(currency) if isinstance(effective_date, datetime.datetime): effective_date = effective_date.date() if effective_date > curves.keys()[-1]: @@ -59,38 +79,46 @@ def get_curve(effective_date, currency="USD"): if effective_date in curves: return curves[effective_date] else: - warnings.warn("cache miss for date: {}".format(effective_date), - RuntimeWarning) + warnings.warn( + f"cache miss for {currency} curve on {effective_date}", RuntimeWarning + ) ql_yc = YC(currency=currency, evaluation_date=effective_date) jp_yc = ql_to_jp(ql_yc) curves[effective_date] = jp_yc return jp_yc -def getMarkitIRData(effective_date=datetime.date.today(), - currency="USD"): +def getMarkitIRData(effective_date=datetime.date.today(), currency="USD"): conn = dbconn("serenitasdb") - sql_str = "SELECT * FROM {}_rates WHERE effective_date <= %s " \ - "ORDER BY effective_date DESC LIMIT 1".format(currency) + sql_str = ( + "SELECT * FROM {}_rates WHERE effective_date <= %s " + "ORDER BY effective_date DESC LIMIT 1".format(currency) + ) with conn.cursor() as c: c.execute(sql_str, (effective_date,)) col_names = [col[0] for col in c.description] r = c.fetchone() - MarkitData = {'effectiveasof': r[0], - 'deposits': [(t, rate) for t, rate in zip(col_names[1:7], r[1:7]) - if rate is not None], - 'swaps': [(t, rate) for t, rate in zip(col_names[7:], r[7:]) - if rate is not None]} + MarkitData = { + "effectiveasof": r[0], + "deposits": [ + (t, rate) for t, rate in zip(col_names[1:7], r[1:7]) if rate is not None + ], + "swaps": [ + (t, rate) for t, rate in zip(col_names[7:], r[7:]) if rate is not None + ], + } return MarkitData def get_futures_data(date=datetime.date.today()): - futures_file = os.path.join(os.environ['DATA_DIR'], "Yield Curves", - "futures-{0:%Y-%m-%d}.csv".format(date)) + futures_file = os.path.join( + os.environ["DATA_DIR"], "Yield Curves", "futures-{0:%Y-%m-%d}.csv".format(date) + ) with open(futures_file) as fh: quotes = [float(line.split(",")[1]) for line in fh] return quotes + def get_curve_params(currency): if currency == "USD": currency = USDCurrency() @@ -116,11 +144,15 @@ def get_curve_params(currency): mm_dc = Actual360() floating_freq = Period(6, Months) fixed_freq = Semiannual - CurveParams = namedtuple('CurveParam', - 'currency, calendar, fixed_dc, floating_dc, ' - 'mm_dc, floating_freq, fixed_freq') - return CurveParams(currency, calendar, fixed_dc, floating_dc, mm_dc, - floating_freq, fixed_freq) + CurveParams = namedtuple( + "CurveParam", + "currency, calendar, fixed_dc, floating_dc, " + "mm_dc, floating_freq, fixed_freq", + ) + return CurveParams( + currency, calendar, fixed_dc, floating_dc, mm_dc, floating_freq, fixed_freq + ) + def rate_helpers(currency="USD", MarkitData=None, evaluation_date=None): """Utility function to build a list of RateHelpers @@ -144,25 +176,52 @@ def rate_helpers(currency="USD", MarkitData=None, evaluation_date=None): evaluation_date = evaluation_date.date() if not MarkitData: MarkitData = getMarkitIRData(evaluation_date, currency) - if MarkitData['effectiveasof'] != evaluation_date: - warnings.warn("Yield curve effective date: {0} doesn't " - "match the evaluation date: {1}".format( - MarkitData['effectiveasof'], - evaluation_date), - RuntimeWarning) - settings.evaluation_date = Date.from_datetime(MarkitData['effectiveasof']) + if MarkitData["effectiveasof"] != evaluation_date: + warnings.warn( + "Yield curve effective date: {0} doesn't " + "match the evaluation date: {1}".format( + MarkitData["effectiveasof"], evaluation_date + ), + RuntimeWarning, + ) + settings.evaluation_date = Date.from_datetime(MarkitData["effectiveasof"]) params = get_curve_params(currency) - isda_ibor = IborIndex("IsdaIbor", params.floating_freq, 2, - params.currency, params.calendar, ModifiedFollowing, - False, params.floating_dc) + isda_ibor = IborIndex( + "IsdaIbor", + params.floating_freq, + 2, + params.currency, + params.calendar, + ModifiedFollowing, + False, + params.floating_dc, + ) # we use SimpleQuotes, rather than just float to make it updateable - deps = [DepositRateHelper(SimpleQuote(q), Period(t), 2, params.calendar, - ModifiedFollowing, False, params.mm_dc) - for t, q in MarkitData['deposits']] + deps = [ + DepositRateHelper( + SimpleQuote(q), + Period(t), + 2, + params.calendar, + ModifiedFollowing, + False, + params.mm_dc, + ) + for t, q in MarkitData["deposits"] + ] # this matches with bloomberg, but according to Markit, maturity should be unadjusted - swaps = [SwapRateHelper.from_tenor(SimpleQuote(q), Period(t), params.calendar, - params.fixed_freq, ModifiedFollowing, - params.fixed_dc, isda_ibor) for t, q in MarkitData['swaps']] + swaps = [ + SwapRateHelper.from_tenor( + SimpleQuote(q), + Period(t), + params.calendar, + params.fixed_freq, + ModifiedFollowing, + params.fixed_dc, + isda_ibor, + ) + for t, q in MarkitData["swaps"] + ] return deps + swaps @@ -173,12 +232,18 @@ def get_dates(date, currency="USD"): year_periods = [2, 3, 4, 5, 6, 7, 8, 9, 10, 12, 15, 20, 25, 30] calendar = WeekendsOnly() settle_date = calendar.advance(Date.from_datetime(date), 2, 0) - deposit_dates = [calendar.advance(settle_date, period=Period(m, Months), - convention=ModifiedFollowing) - for m in month_periods] - swap_dates = [calendar.advance(settle_date, period=Period(y, Years), - convention=ModifiedFollowing) - for y in year_periods] + deposit_dates = [ + calendar.advance( + settle_date, period=Period(m, Months), convention=ModifiedFollowing + ) + for m in month_periods + ] + swap_dates = [ + calendar.advance( + settle_date, period=Period(y, Years), convention=ModifiedFollowing + ) + for y in year_periods + ] dates = deposit_dates + swap_dates return [pydate_from_qldate(d) for d in dates] @@ -187,26 +252,41 @@ def roll_yc(yc, forward_date): """returns the expected forward yield cuve on a forward_date""" dates = [d for d in yc.dates if d >= forward_date] dfs = np.array([yc.discount_factor(d, forward_date) for d in dates]) - return YieldCurve.from_discount_factors(forward_date, dates, dfs, 'ACT/365F') + return YieldCurve.from_discount_factors(forward_date, dates, dfs, "ACT/365F") -def YC(helpers=None, currency="USD", MarkitData=None, evaluation_date=None, - fixed=False, extrapolation=False): +def YC( + helpers=None, + currency="USD", + MarkitData=None, + evaluation_date=None, + fixed=False, + extrapolation=False, +): calendar = WeekendsOnly() settings = Settings() if evaluation_date: settings.evaluation_date = Date.from_datetime(evaluation_date) - if helpers is None: # might roll back the evaluation date + if helpers is None: # might roll back the evaluation date helpers = rate_helpers(currency, MarkitData, evaluation_date) if fixed: _yc = PiecewiseYieldCurve.from_reference_date( - BootstrapTrait.Discount, Interpolator.LogLinear, - settings.evaluation_date, helpers, Actual365Fixed()) + BootstrapTrait.Discount, + Interpolator.LogLinear, + settings.evaluation_date, + helpers, + Actual365Fixed(), + ) else: - _yc = PiecewiseYieldCurve(BootstrapTrait.Discount, - Interpolator.LogLinear, - 0, calendar, helpers, Actual365Fixed()) + _yc = PiecewiseYieldCurve( + BootstrapTrait.Discount, + Interpolator.LogLinear, + 0, + calendar, + helpers, + Actual365Fixed(), + ) if extrapolation: _yc.extrapolation = True return _yc @@ -214,22 +294,30 @@ def YC(helpers=None, currency="USD", MarkitData=None, evaluation_date=None, def jpYC(effective_date, currency="USD", MarkitData=None): if MarkitData is None: - markit_data = getMarkitIRData(effective_date, - currency) - periods, rates = zip(*markit_data['deposits']) - periods_swaps, rates_swaps = zip(*markit_data['swaps']) - types = 'M'*len(periods) + 'S'*len(periods_swaps) + markit_data = getMarkitIRData(effective_date, currency) + periods, rates = zip(*markit_data["deposits"]) + periods_swaps, rates_swaps = zip(*markit_data["swaps"]) + types = "M" * len(periods) + "S" * len(periods_swaps) rates = np.array(rates + rates_swaps) periods = list(periods + periods_swaps) if currency == "USD": - fixed_period = '6M' - float_period = '3M' - elif currency == 'EUR': - fixed_period = '12M' - float_period = '6M' - return YieldCurve(effective_date, types, periods, rates, 'ACT/360', - fixed_period, float_period, '30/360', 'ACT/360', - BadDay.MODIFIED) + fixed_period = "6M" + float_period = "3M" + elif currency == "EUR": + fixed_period = "12M" + float_period = "6M" + return YieldCurve( + effective_date, + types, + periods, + rates, + "ACT/360", + fixed_period, + float_period, + "30/360", + "ACT/360", + BadDay.MODIFIED, + ) def ql_to_jp(ql_yc): @@ -238,27 +326,47 @@ def ql_to_jp(ql_yc): dfs = np.array(ql_yc.data[1:]) dates = [pydate_from_qldate(d) for d in ql_yc.dates[1:]] trade_date = pydate_from_qldate(ql_yc.dates[0]) - return YieldCurve.from_discount_factors(trade_date, dates, dfs, 'ACT/365F') + return YieldCurve.from_discount_factors(trade_date, dates, dfs, "ACT/365F") else: - raise RuntimeError('QuantLib curve needs to use Discount trait') + raise RuntimeError("QuantLib curve needs to use Discount trait") def build_curves(currency="USD"): settings = Settings() params = get_curve_params(currency) - isda_ibor = IborIndex("IsdaIbor", params.floating_freq, 2, - params.currency, params.calendar, ModifiedFollowing, - False, params.floating_dc) - rates = pd.read_sql_table(f"{currency.lower()}_rates", - serenitas_engine, - index_col='effective_date') + isda_ibor = IborIndex( + "IsdaIbor", + params.floating_freq, + 2, + params.currency, + params.calendar, + ModifiedFollowing, + False, + params.floating_dc, + ) + rates = pd.read_sql_table( + f"{currency.lower()}_rates", serenitas_engine, index_col="effective_date" + ) quotes = [SimpleQuote() for c in rates.columns] gen = zip(quotes, rates.columns) - deps = [DepositRateHelper(q, Period(t), 2, params.calendar, ModifiedFollowing, - False, params.mm_dc) for q, t in islice(gen, 6)] - swaps = [SwapRateHelper.from_tenor(q, Period(t), params.calendar, - params.fixed_freq, ModifiedFollowing, - params.fixed_dc, isda_ibor) for q, t in gen] + deps = [ + DepositRateHelper( + q, Period(t), 2, params.calendar, ModifiedFollowing, False, params.mm_dc + ) + for q, t in islice(gen, 6) + ] + swaps = [ + SwapRateHelper.from_tenor( + q, + Period(t), + params.calendar, + params.fixed_freq, + ModifiedFollowing, + params.fixed_dc, + isda_ibor, + ) + for q, t in gen + ] sql_str = f"INSERT INTO {currency}_curves VALUES(%s, %s) ON CONFLICT DO NOTHING" conn = serenitas_engine.raw_connection() for effective_date, curve_data in rates.iterrows(): @@ -268,13 +376,19 @@ def build_curves(currency="USD"): q.value = val valid_deps = [d for d in deps if not np.isnan(d.quote)] valid_swaps = [s for s in swaps if not np.isnan(s.quote)] - ql_yc = PiecewiseYieldCurve(BootstrapTrait.Discount, - Interpolator.LogLinear, - 0, calendar, valid_deps + valid_swaps, - Actual365Fixed()) + ql_yc = PiecewiseYieldCurve( + BootstrapTrait.Discount, + Interpolator.LogLinear, + 0, + calendar, + valid_deps + valid_swaps, + Actual365Fixed(), + ) jp_yc = ql_to_jp(ql_yc) with conn.cursor() as c: - c.execute(sql_str, (effective_date, lz4.block.compress(jp_yc.__getstate__()))) + c.execute( + sql_str, (effective_date, lz4.block.compress(jp_yc.__getstate__())) + ) conn.commit() conn.close() @@ -286,16 +400,19 @@ if __name__ == "__main__": import matplotlib.pyplot as plt from quantlib.time.api import calendar_from_name from pandas.plotting import register_matplotlib_converters + register_matplotlib_converters() helpers = rate_helpers("USD") ts = YC(helpers) - cal = calendar_from_name('USA') - p1 = Period('1M') - p2 = Period('2M') - p3 = Period('3M') - p6 = Period('6M') - p12 = Period('12M') - sched = Schedule.from_rule(ts.reference_date, ts.reference_date + Period('5Y'), Period('3M'), cal) + cal = calendar_from_name("USA") + p1 = Period("1M") + p2 = Period("2M") + p3 = Period("3M") + p6 = Period("6M") + p12 = Period("12M") + sched = Schedule.from_rule( + ts.reference_date, ts.reference_date + Period("5Y"), Period("3M"), cal + ) days = [pydate_from_qldate(d) for d in sched] f3 = [ts.forward_rate(d, d + p3, Actual360(), 0).rate for d in sched] f6 = [ts.forward_rate(d, d + p6, Actual360(), 0).rate for d in sched] |