from contextlib import closing from itertools import islice import datetime import lz4 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, BadDay import warnings from db import dbengine, dbconn def get_curves(currency="USD", date=None): """load the prebuilt curve from the database""" if date: sql_str = "SELECT curve FROM {}_curves WHERE effective_date=%s".format(currency) else: sql_str = "SELECT * FROM {}_curves".format(currency) 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 {d: YieldCurve.from_bytes(lz4.block.decompress(curve)) for d, curve in c} _USD_curves = get_curves("USD") _EUR_curves = get_curves("EUR") 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, r[i]) for i, t in \ enumerate(col_names[1:7], 1) if r[i] is not None], 'swaps': [(t, r[i]) for i, t in enumerate(col_names[7:], 7)]} 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 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 = get_dates(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): if helpers is None: helpers = rate_helpers(currency, MarkitData) calendar = WeekendsOnly() return PiecewiseYieldCurve(BootstrapTrait.Discount, Interpolator.LogLinear, 0, calendar, helpers, Actual365Fixed()) 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, rate_swaps = zip(*markit_data['swaps']) types = 'M'*len(periods) + 'S'*len(periods_swaps) rates = np.array(rates + rates_swaps) periods = list(period + 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 RuntimeErrror('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 = "INSERT INTO {}_curves VALUES(%s, %s) ON CONFLICT DO NOTHING".format(currency) 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 ts = YC() cal = calendar_from_name('USA') p1 = Period('1Mo') p2 = Period('2Mo') p3 = Period('3Mo') p6 = Period('6Mo') p12 = Period('12Mo') sched = Schedule(ts.reference_date, ts.reference_date+Period('5Yr'), Period('3Mo'), 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)