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
|
from . import dbconn
from quantlib.indexes.api import UsdLiborSwapIsdaFixAm
from quantlib.quotes import SimpleQuote
from quantlib.time.api import Date, Period, Years, pydate_from_qldate
from quantlib.instruments.api import MakeSwaption
from quantlib.instruments.swap import SwapType
from quantlib.pricingengines.api import BlackSwaptionEngine
from scipy.optimize import brentq
from yieldcurve import YC
class IRSwaption:
""" adapter class for the QuantLib code"""
def __init__(
self,
swap_index,
option_tenor,
strike,
option_type="payer",
direction="Long",
notional=10_000_000,
yc=None,
):
self._qloption = (
MakeSwaption(swap_index, option_tenor, strike)
.with_nominal(notional)
.with_underlying_type(SwapType[option_type.title()])()
)
if type(direction) is bool:
self._direction = 2 * direction - 1
else:
self.direction = direction
self._yc = yc or swap_index.forwarding_term_structure
self._sigma = SimpleQuote(0.218)
self._qloption.set_pricing_engine(BlackSwaptionEngine(self._yc, self._sigma))
@property
def direction(self):
if self._direction == 1.0:
return "Long"
else:
return "Short"
@direction.setter
def direction(self, d):
if d == "Long":
self._direction = 1.0
elif d == "Short":
self._direction = -1.0
else:
raise ValueError("Direction needs to be either 'Long' or 'Short'")
@property
def pv(self):
return self._direction * self._qloption.npv
@pv.setter
def pv(self, val):
def handle(x):
self.sigma = x
return self._direction * (self.pv - val)
eta = 1.1
a = 0.1
b = a * eta
while True:
if handle(b) > 0:
break
b *= eta
self.sigma = brentq(handle, a, b)
@property
def sigma(self):
return self._sigma.value
@sigma.setter
def sigma(self, s):
self._sigma.value = s
def from_tradeid(trade_id):
with dbconn("dawndb") as conn:
with conn.cursor() as c:
c.execute("SELECT * from swaptions " "WHERE id = %s", (trade_id,))
rec = c.fetchone()
yc = YC(evaluation_date=rec.trade_date, fixed=True, extrapolation=True)
p = Period(int(rec.security_id.replace("USISDA", "")), Years)
swap_index = UsdLiborSwapIsdaFixAm(p, yc)
instance = IRSwaption(
swap_index,
Date.from_datetime(rec.expiration_date),
rec.strike,
rec.option_type,
rec.buysell,
rec.notional,
)
try:
instance.pv = rec.price / 100 * rec.notional * instance._direction
except ValueError:
pass
return instance
@property
def value_date(self):
return pydate_from_qldate(self._qloption.valuation_date)
@value_date.setter
def value_date(self, d):
self.yc.link_to(YC(evaluation_date=d, fixed=True))
@property
def strike(self):
return self._qloption.underlying_swap().fixed_rate
|