from blist import sorteddict from contextlib import closing from itertools import islice import datetime import lz4.block import os import pandas as pd from quantlib.settings import Settings from quantlib.time.api import (WeekendsOnly, Date, Period, Days, Schedule, Annual, Semiannual, today, Actual360, Months, Years, ModifiedFollowing, Thirty360, Actual365Fixed, calendar_from_name) from quantlib.currency.api import USDCurrency, EURCurrency from quantlib.indexes.ibor_index import IborIndex from quantlib.termstructures.yields.api import ( PiecewiseYieldCurve, DepositRateHelper, SwapRateHelper, BootstrapTrait, Interpolator) from quantlib.time.date import pydate_from_qldate import numpy as np from quantlib.quotes import SimpleQuote from db import dbconn, dbengine from pyisda.curve import YieldCurve from pyisda.date import BadDay import warnings def load_curves(currency="USD", date=None): """load the prebuilt curves from the database""" sql_str = f"SELECT * FROM {currency}_curves" if date: sql_str += " WHERE effective_date=%s" with closing(dbconn('serenitasdb')) as conn: with conn.cursor() as c: if date: c.execute(sql_str, (date,)) if c: curve, = c.fetchone() 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]) def get_curve(effective_date, currency="USD"): if f'_{currency}_curves' in globals(): curves = globals()[f'_{currency}_curves'] else: 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]: last_curve = curves[curves.keys()[-1]] return last_curve if effective_date in curves: return curves[effective_date] else: warnings.warn("cache miss for date: {}".format(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"): conn = dbconn("serenitasdb") 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]} 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)) with open(futures_file) as fh: quotes = [float(line.split(",")[1]) for line in fh] return quotes def rate_helpers(currency="USD", MarkitData=None, evaluation_date=None): """Utility function to build a list of RateHelpers Parameters ---------- currency : str, optional One of `USD`, `EUR` at the moment, defaults to `USD` MarkitData : dict, optional MarkitData for the current evaluation_date Returns ------- helpers : list List of QuantLib RateHelpers """ settings = Settings() if evaluation_date is None: evaluation_date = pydate_from_qldate(settings.evaluation_date) if isinstance(evaluation_date, pd.Timestamp): 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']) calendar = WeekendsOnly() if currency == "USD": isda_ibor = IborIndex("IsdaIbor", Period(3, Months), 2, USDCurrency(), calendar, ModifiedFollowing, False, Actual360()) fix_freq = Semiannual elif currency == "EUR": isda_ibor = IborIndex("IsdaIbor", Period(6, Months), 2, EURCurrency(), calendar, ModifiedFollowing, False, Actual360()) fix_freq = Annual # we use SimpleQuotes, rather than just float to make it updateable deps = [DepositRateHelper(SimpleQuote(q), Period(t), 2, calendar, ModifiedFollowing, False, Actual360()) 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), calendar, fix_freq, ModifiedFollowing, Thirty360(), isda_ibor) for t, q in MarkitData['swaps']] return deps + swaps def get_dates(date, currency="USD"): """computes the list of curve dates on a given date""" if currency == "USD": month_periods = [1, 2, 3, 6, 12] 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] dates = deposit_dates + swap_dates return [pydate_from_qldate(d) for d in dates] 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') 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 helpers = rate_helpers(currency, MarkitData, evaluation_date) if fixed: _yc = PiecewiseYieldCurve.from_reference_date( BootstrapTrait.Discount, Interpolator.LogLinear, settings.evaluation_date, helpers, Actual365Fixed()) else: _yc = PiecewiseYieldCurve(BootstrapTrait.Discount, Interpolator.LogLinear, 0, calendar, helpers, Actual365Fixed()) if extrapolation: _yc.extrapolation = True return _yc 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) 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) def ql_to_jp(ql_yc): """convert a QuantLib yield curve to a JP's one""" if ql_yc._trait == BootstrapTrait.Discount: 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') else: raise RuntimeError('QuantLib curve needs to use Discount trait') def build_curves(currency="USD"): settings = Settings() calendar = WeekendsOnly() if currency == "USD": isda_ibor = IborIndex("IsdaIbor", Period(3, Months), 2, USDCurrency(), calendar, ModifiedFollowing, False, Actual360()) fix_freq = Semiannual elif currency == "EUR": isda_ibor = IborIndex("IsdaIbor", Period(6, Months), 2, EURCurrency(), calendar, ModifiedFollowing, False, Actual360()) fix_freq = Annual engine = dbengine('serenitasdb') rates = pd.read_sql_table('{}_rates'.format(currency.lower()), engine, index_col='effective_date') quotes = [SimpleQuote() for c in rates.columns] gen = zip(quotes, rates.columns) deps = [DepositRateHelper(q, Period(t), 2, calendar, ModifiedFollowing, False, Actual360()) for q, t in islice(gen, 6)] swaps = [SwapRateHelper.from_tenor(q, Period(t), calendar, fix_freq, ModifiedFollowing, Thirty360(), isda_ibor) for q, t in gen] sql_str = f"INSERT INTO {currency}_curves VALUES(%s, %s) ON CONFLICT DO NOTHING" conn = dbconn('serenitasdb') for effective_date, curve_data in rates.iterrows(): print(effective_date) settings.evaluation_date = Date.from_datetime(effective_date) for q, val in zip(quotes, curve_data): 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()) jp_yc = ql_to_jp(ql_yc) with conn.cursor() as c: c.execute(sql_str, (effective_date, lz4.block.compress(jp_yc.__getstate__()))) conn.commit() if __name__ == "__main__": # evaluation_date = Date(29, 4, 2014) Settings.instance().evaluation_date = today() import matplotlib.pyplot as plt 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) 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] f2 = [ts.forward_rate(d, d + p2, Actual360(), 0).rate for d in sched] plt.plot(days, f2, days, f3, days, f6)