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
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
|
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)
|
