from common import root import os import requests, zipfile from io import BytesIO import xml.etree.ElementTree as ET import datetime from quantlib.settings import Settings from quantlib.time.api import (WeekendsOnly, Date, Period, Days, Schedule, Annual, Semiannual, today, Actual360, Months, 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 import pandas as pd import matplotlib.pyplot as plt from quantlib.quotes import SimpleQuote from db import dbconn from pyisda.curve import YieldCurve import array def getMarkitIRData(effective_date = datetime.date.today(), currency = "USD"): download_date = effective_date - datetime.timedelta(days = 1) conn = dbconn("serenitasdb") sql_str = "SELECT * FROM {}_rates WHERE effective_date = %s".format(currency) with conn.cursor() as c: c.execute(sql_str, (effective_date,)) col_names = [c[0] for c in c.description] r = c.fetchone() if r: 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)]} else: basedir = os.path.join(root, "data", "Yield Curves") filename = "InterestRates_{0}_{1:%Y%m%d}".format(currency, download_date) if not os.path.exists(os.path.join(basedir, filename + '.xml')): r = requests.get('http://www.markit.com/news/{0}.zip'.format(filename)) if "zip" in r.headers['content-type']: with zipfile.ZipFile(BytesIO(r.content)) as z: z.extractall(path = os.path.join(root, "data", "Yield Curves")) else: return getMarkitIRData(download_date-datetime.timedelta(days=1)) tree = ET.parse(os.path.join(root, "data", "Yield Curves", filename + '.xml')) deposits = zip([e.text for e in tree.findall('./deposits/*/tenor')], [float(e.text) for e in tree.findall('./deposits/*/parrate')]) swaps = zip([e.text for e in tree.findall('./swaps/*/tenor')], [float(e.text) for e in tree.findall('./swaps/*/parrate')]) effectiveasof = tree.find('./effectiveasof').text MarkitData = {'deposits': list(deposits), 'swaps': list(swaps), 'effectiveasof': pd.Timestamp(effectiveasof).date()} instruments = MarkitData['deposits'] + MarkitData['swaps'] names = ",".join(['"{}"'.format(r[0]) for r in instruments]) values = ",".join(["%s"] * (len(instruments) + 1)) # +1 for effective_date insert_str = ("INSERT INTO {0}_rates(effective_date, {1}) VALUES({2})". format(currency, names, values)) with conn.cursor() as c: c.execute(insert_str, [MarkitData['effectiveasof']] +[r[1] for r in instruments]) conn.commit() conn.close() return MarkitData def get_futures_data(date = datetime.date.today()): futures_file = os.path.join(root, "data", "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): settings = Settings() if not MarkitData: MarkitData = getMarkitIRData(pydate_from_qldate(settings.evaluation_date), currency) if MarkitData['effectiveasof'] != pydate_from_qldate(settings.evaluation_date): raise RuntimeError("Yield curve effective date: {0} doesn't " \ "match the evaluation date: {1}".format( MarkitData['effectiveasof'], pydate_from_qldate(settings.evaluation_date))) 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 deps = [DepositRateHelper(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(q, Period(t), calendar, fix_freq, ModifiedFollowing, Thirty360(), isda_ibor) for t, q in MarkitData['swaps']] return deps + swaps def YC(currency="USD", helpers = None, MarkitData=None): if helpers is None: helpers = rate_helpers(currency, MarkitData) calendar = WeekendsOnly() return PiecewiseYieldCurve(BootstrapTrait.Discount, Interpolator.LogLinear, 0, calendar, helpers, Actual365Fixed()) def ql_to_jp(ql_yc): """ convert a QuantLib yield curve to a JP's one""" if ql_yc._trait == BootstrapTrait.Discount: dfs = array.array('d', 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') if __name__=="__main__": #evaluation_date = Date(29, 4, 2014) Settings.instance().evaluation_date = today() 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)