1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
|
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)
|
