Coverage for rivapy / pricing / bond_pricing.py: 88%
194 statements
« prev ^ index » next coverage.py v7.12.0, created at 2025-11-27 14:36 +0000
1from datetime import datetime, date
2from typing import List, Tuple, Union as _Union, Optional as _Optional
3from scipy.optimize import brentq
4from rivapy.tools.enums import DayCounterType, InterestRateIndex
5from rivapy.tools.interfaces import BaseDatedCurve
6from rivapy.instruments.bond_specifications import DeterministicCashflowBondSpecification, FloatingRateBondSpecification
7from rivapy.marketdata.curves import DiscountCurveComposition
8from rivapy.marketdata import DiscountCurveParametrized, ConstantRate
9from rivapy.pricing.pricing_request import PricingRequest
10from rivapy.instruments._logger import logger
11from rivapy.marketdata.curves import DiscountCurve
12from rivapy.tools.datetools import (
13 Period,
14 _date_to_datetime,
15 _term_to_period,
16 _string_to_calendar,
17 DayCounter,
18 Schedule,
19 roll_day,
20 calc_start_day,
21 _period_to_string,
22)
23from typing import Tuple, Union as _Union, List as _List
26class DeterministicCashflowPricer:
27 """Deterministic cashflow pricer utilities.
29 This class provides static and instance helpers for pricing deterministic cashflow
30 instruments (fixed- and floating-rate bonds, deposits, zero-coupon instruments).
31 It contains methods to generate expected cashflows, discount them, compute PVs,
32 yields, z-spreads and duration measures.
33 """
35 def __init__(
36 self,
37 val_date: datetime,
38 spec: DeterministicCashflowBondSpecification,
39 discount_curve: DiscountCurve,
40 fwd_curve: _Union[DiscountCurve, None] = None,
41 ):
42 """
43 Initialize the DeterministicCashflowPricer.
45 Args:
46 val_date (datetime): The valuation date.
47 spec (DeterministicCashflowBondSpecification): The specification of the cashflow instrument.
48 discount_curve (DiscountCurve): The discount curve to use for pricing.
49 fwd_curve (Union[DiscountCurve, None], optional): The forward curve to use for pricing. Defaults to None.
50 """
51 self._val_date = _date_to_datetime(val_date)
52 self._spec = spec
53 self._discount_curve = discount_curve
54 self._fwd_curve = fwd_curve
55 self._cashflows = None
57 @property
58 def cashflows(self) -> _List[Tuple[datetime, float]]:
59 """Get the cashflows of the instrument.
61 Returns:
62 Cashflow: The cashflows of the instrument.
63 """
64 if self._cashflows is None:
65 self._cashflows = DeterministicCashflowPricer.get_expected_cashflows(self._spec, self._val_date, self._fwd_curve)
66 return self._cashflows
68 @cashflows.setter
69 def cashflows(self, value: _List[Tuple[datetime, float]]):
70 """Set the cashflows of the instrument.
72 Returns:
73 Cashflow: The cashflows of the instrument.
74 """
75 self._cashflows = value
77 def expected_cashflows(self) -> _List[Tuple[datetime, float]]:
78 """Get the expected cashflows of the instrument.
80 Returns:
81 List[Tuple[datetime, float]]: The expected cashflows of the instrument.
82 """
83 return DeterministicCashflowPricer.get_expected_cashflows(self._spec, self._val_date, self._fwd_curve)
85 @staticmethod
86 def get_expected_cashflows(
87 spec: DeterministicCashflowBondSpecification,
88 val_date: _Union[datetime.date, datetime, None] = None,
89 fwd_curve: _Union[DiscountCurve, None] = None,
90 ) -> List[Tuple[datetime, float]]:
91 """
92 Calculate the expected cashflows for a deterministic cashflow instrument.
94 Args:
95 spec (DeterministicCashflowBondSpecification): The instrument specification containing schedule, notional, coupon type, etc.
96 val_date (datetime.date or datetime, optional): The valuation date, required for floating rate calculation. Defaults to None.
97 curve (DiscountCurve, optional): The forward curve used for floating rate calculation. Defaults to None.
99 Returns:
100 List[Tuple[datetime, float]]: A sorted list of tuples, each containing the payment date and cashflow amount.
101 """
102 cashflows = []
103 if spec._coupon_type != "zero":
104 # schedule = spec.get_schedule()
105 # # schedule for accrual periods rolled out
106 # dates = schedule._roll_out(
107 # from_=spec._start_date if not spec._backwards else spec._end_date,
108 # to_=spec._end_date if not spec._backwards else spec._start_date,
109 # term=_term_to_period(spec._frequency),
110 # long_stub=spec.stub_type_is_Long,
111 # backwards=spec.backwards,
112 # )
113 dates = spec.dates
114 # print(spec._notional.get_amortization_schedule())
115 dcc = DayCounter(spec.day_count_convention)
116 for d1, d2 in zip(dates[:-1], dates[1:]):
118 if spec.coupon_type == "float":
119 if val_date is None or fwd_curve is None:
120 raise ValueError("val_date and fwd_curve must be provided for floating rate cashflow calculation.")
121 rate = DeterministicCashflowPricer.get_float_rate(spec, val_date, d1, d2, fwd_curve)
122 else:
123 rate = spec.coupon
124 # normalize day count convention to canonical string before comparison
125 if spec._adjust_accruals is False:
126 d1_adj = d1
127 d2_adj = d2
128 else:
129 d1_adj = roll_day(d1, spec.calendar, spec.business_day_convention)
130 d2_adj = roll_day(d2, spec.calendar, spec.business_day_convention)
131 if DayCounterType.to_string(spec.day_count_convention) == DayCounterType.ActActICMA.value:
132 nr = spec.get_nr_annual_payments()
133 if nr == 0:
134 raise ValueError("Number of annual payments is zero. Please check the frequency setting in the bond specification.")
135 dcv = dcc.yf(d1_adj, d2_adj, dates, nr)
136 else:
137 dcv = dcc.yf(d1_adj, d2_adj)
138 amount = spec._notional.get_amount_per_date(d1_adj) * rate * dcv
139 payment_date = roll_day(d2_adj, spec.calendar, spec.business_day_convention, settle_days=spec.payment_days)
140 cashflows.append((payment_date, amount))
141 # add notional amortizations to cashflow list
142 cashflows.extend(spec._notional.get_amortization_schedule())
143 # add notional exchanges at start and end date if applicable (and only remaining notionals after amortizations)
144 if spec._notional_exchange:
145 # add notional exchange at start and end date
146 not_init = spec._issue_price if spec._issue_price is not None else spec._notional.get_amount(0)
147 cashflows.append((spec._start_date, not_init * (-1)))
148 rem_amount = spec._notional.get_amount(0) - spec._amortization_scheme.get_total_amortization()
149 if rem_amount > 0:
150 cashflows.append(
151 (
152 roll_day(spec._maturity_date, spec._calendar, spec._business_day_convention, settle_days=spec._payment_days),
153 rem_amount,
154 )
155 )
156 cashflows = sorted(cashflows)
157 return cashflows
159 # ToDo: consider rate/index period shorter than coupon period
160 @staticmethod
161 def get_float_rate(
162 specification: DeterministicCashflowBondSpecification,
163 val_date: _Union[datetime.date, datetime, None] = None,
164 d1: _Union[datetime.date, datetime, None] = None,
165 d2: _Union[datetime.date, datetime, None] = None,
166 curve: _Union[DiscountCurve, None] = None,
167 ) -> float:
168 """
169 Get the floating rate for a given period.
171 Args:
172 specification (DeterministicCashflowBondSpecification): The bond or instrument specification containing index, margin, calendar, and conventions.
173 val_date (datetime.date or datetime, optional): The valuation date as of which forward rates are calculated. Defaults to None.
174 d1 (datetime.date or datetime, optional): The start date of the interest period. Defaults to None.
175 d2 (datetime.date or datetime, optional): The end date of the interest period. Defaults to None.
176 curve (DiscountCurve, optional): The forward curve used for forward rate calculation. Defaults to None.
177 Returns:
178 float: The floating rate for the given period, including margin.
179 """
180 if specification._ir_index is not None: # For the first period, check if we have a fixing rate or if d1 is before curve date
181 spot_days = InterestRateIndex(specification._ir_index).value.spot_days
182 else:
183 spot_days = specification._spot_days
184 fixing_date = calc_start_day(
185 d1,
186 f"{spot_days}D",
187 business_day_convention=specification._business_day_convention,
188 calendar=specification._calendar,
189 )
190 refdate = getattr(curve, "refdate", None) if curve is not None else None
191 # guard against calc_start_day returning None (invalid inputs) or refdate being None
192 if fixing_date is not None and refdate is not None and fixing_date <= refdate:
193 try:
194 print(specification._ir_index)
195 fix_name = (
196 InterestRateIndex(specification._ir_index).value.name
197 if specification._ir_index is not None and isinstance(InterestRateIndex(specification._ir_index), InterestRateIndex)
198 else _period_to_string(specification._frequency)
199 )
200 rate = specification._fixings.get_fixing(fix_name, fixing_date)
201 except Exception as e:
202 logger.warning(f"No fixing found for {specification._index} on {fixing_date}. Using 0.0 as fixed rate. Error: {e}")
203 rate = 0.0
204 else:
205 # For other periods use forward rate from curve
206 rate = curve.value_fwd_rate(val_date, d1, d2) if curve is not None else 0.0
207 rate += specification._margin / 10000.0 # add margin
208 return rate
210 @staticmethod
211 def get_accrued_interest(
212 specification: DeterministicCashflowBondSpecification,
213 trade_date: _Union[date, datetime, None] = None,
214 fwd_curve: _Union[DiscountCurve, None] = None,
215 ) -> float:
216 """
217 Get the accrued interest for a given instrument specification.
219 Args:
220 specification (DeterministicCashflowBondSpecification): The bond specification.
221 val_date (datetime.date or datetime, optional): The valuation date. Defaults to None.
223 Returns:
224 float: The accrued interest.
225 """
226 if trade_date is None:
227 raise ValueError("trade_date must be provided.")
228 if specification._coupon_type == "zero":
229 return 0.0
230 else:
231 # schedule = specification.get_schedule()
232 # # schedule for payment periods rolled out
233 # dates = schedule._roll_out(from_=specification._start_date, to_=specification._end_date, term=_term_to_period(specification._frequency))
234 dates = specification.accrual_dates
235 dates = sorted(dates)
236 # find the last coupon date before or on trade_date
237 last_coupon_date = None
238 next_coupon_date = None
239 for d in dates:
240 if d <= trade_date:
241 last_coupon_date = d
242 elif d > trade_date and next_coupon_date is None:
243 next_coupon_date = d
244 break
245 if last_coupon_date is None or next_coupon_date is None:
246 return 0.0 # No accrued interest if trade_date is before first coupon or after last coupon
248 dcc = DayCounter(specification.day_count_convention)
249 # Calculate the fraction of the coupon period that has accrued
250 if isinstance(specification, FloatingRateBondSpecification):
251 accrual_fraction = dcc.yf(last_coupon_date, trade_date, specification.dates, specification.get_nr_annual_payments()) / dcc.yf(
252 last_coupon_date, next_coupon_date, specification.dates, specification.get_nr_annual_payments()
253 )
254 yf = dcc.yf(last_coupon_date, next_coupon_date, specification.dates, specification.get_nr_annual_payments())
255 else:
256 accrual_fraction = dcc.yf(last_coupon_date, trade_date) / dcc.yf(last_coupon_date, next_coupon_date)
257 yf = dcc.yf(last_coupon_date, next_coupon_date)
258 # Calculate the accrued interest
259 if specification._coupon_type == "float":
260 rate = DeterministicCashflowPricer.get_float_rate(specification, trade_date, last_coupon_date, next_coupon_date, fwd_curve)
261 else:
262 rate = specification._coupon
263 print(
264 "Fraction: ",
265 accrual_fraction,
266 " YF: ",
267 yf,
268 "Trade date: ",
269 trade_date,
270 " Last coupon: ",
271 last_coupon_date,
272 " Next coupon: ",
273 next_coupon_date,
274 )
275 accrued_interest = specification._notional.get_amount_per_date(trade_date) * rate * accrual_fraction * yf
276 return accrued_interest
278 def pv_cashflows(self) -> float:
279 """Get the present value of the cashflows.
281 Returns:
282 float: The present value of the cashflows.
283 """
284 return DeterministicCashflowPricer.get_pv_cashflows(self._val_date, self._spec, self._discount_curve, self._fwd_curve)
286 @staticmethod
287 def get_pv_cashflows(
288 val_date: datetime,
289 specification: DeterministicCashflowBondSpecification,
290 discount_curve: DiscountCurve,
291 fwd_curve: _Union[DiscountCurve, None] = None,
292 cashflows: _Union[List[Tuple[datetime, float]], None] = None,
293 ) -> float:
294 """Discount and sum cashflows to obtain present value.
296 Args:
297 val_date (date | datetime): Valuation date used for discounting.
298 specification (DeterministicCashflowBondSpecification): Instrument specification.
299 discount_curve (DiscountCurve): Curve used to obtain discount factors.
300 fwd_curve (DiscountCurve, optional): Forward curve used for floating-rate cashflows.
301 cashflows (List[(date, amount)], optional): Precomputed cashflows; if not
302 provided they will be generated from the specification.
304 Returns:
305 float: Present value obtained by discounting future cashflows occurring after val_date.
306 """
308 # logger.info('Start computing pv cashflows for bond ' + specification.obj_id)
310 if cashflows is None:
311 cashflows = DeterministicCashflowPricer.get_expected_cashflows(
312 specification, val_date=val_date, fwd_curve=fwd_curve
313 ) # get only cashflows
315 pv_cashflows = 0.0
316 for c in cashflows:
317 if c[0] > val_date:
318 df = discount_curve.value(val_date, c[0])
319 logger.debug("Cashflow " + str(c[1]) + ", date: " + str(c[0]) + ", df: " + str(df))
320 pv_cashflows += df * c[1]
321 # logger.info('Finished computing pv cashflows for bond ' + specification.obj_id + ', pv_cashflows: '+ str(pv_cashflows) )
322 return pv_cashflows
324 @staticmethod
325 def get_dirty_price(
326 val_date: datetime,
327 specification: DeterministicCashflowBondSpecification,
328 discount_curve: DiscountCurve,
329 fwd_curve: _Union[DiscountCurve, None] = None,
330 ) -> float:
331 """Compute the dirty price (present value including accrued interest) of a bond.
333 Args:
334 val_date (date | datetime): Valuation date.
335 specification (DeterministicCashflowBondSpecification): Instrument specification.
336 discount_curve (DiscountCurve): Curve used for discounting.
337 fwd_curve (DiscountCurve, optional): Forward curve for floating-rate cashflows.
339 Returns:
340 float: Dirty price (PV of future cashflows after val_date).
341 """
342 logger.info("Start computing dirty price for bond " + specification.obj_id)
343 pv_cashflows = DeterministicCashflowPricer.get_pv_cashflows(val_date, specification, discount_curve, fwd_curve)
344 logger.info("Finished computing dirty price for bond " + specification.obj_id + ", dirty_price: " + str(pv_cashflows))
345 return pv_cashflows
347 def dirty_price(self) -> float:
348 """Get the dirty price of the bond.
350 Returns:
351 float: The dirty price of the bond.
352 """
353 return DeterministicCashflowPricer.get_dirty_price(self._val_date, self._spec, self._discount_curve, self._fwd_curve)
355 @staticmethod
356 def clean_price(
357 val_date: datetime,
358 specification: DeterministicCashflowBondSpecification,
359 discount_curve: DiscountCurve,
360 fwd_curve: _Union[DiscountCurve, None] = None,
361 ) -> float:
362 """Compute the clean price of a bond (dirty price minus accrued interest).
364 Args:
365 val_date (date | datetime): Valuation date.
366 specification (DeterministicCashflowBondSpecification): Instrument specification.
367 discount_curve (DiscountCurve): Discount curve used for discounting.
368 fwd_curve (DiscountCurve, optional): Forward curve for floating-rate cashflows.
370 Returns:
371 float: Clean price (dirty price less accrued interest).
372 """
373 dirty_price = DeterministicCashflowPricer.get_dirty_price(val_date, specification, discount_curve, fwd_curve)
374 accrued_interest = DeterministicCashflowPricer.get_accrued_interest(specification, val_date)
375 return dirty_price - accrued_interest
377 def clean_price(self) -> float:
378 """Get the clean price of the bond.
380 Returns:
381 float: The clean price of the bond.
382 """
383 return self.dirty_price() - self.accrued_interest()
385 def compute_yield(self, target_dirty_price: float) -> float:
386 """Compute the yield of the bond.
388 Args:
389 target_dirty_price (float): The target dirty price.
391 Returns:
392 float: The computed yield.
393 """
394 return DeterministicCashflowPricer.get_compute_yield(target_dirty_price, self._val_date, self._spec, cashflows=self.cashflows)
396 # TODO: add accrued interest
397 @staticmethod
398 def get_compute_yield(
399 target_dirty_price: float,
400 val_date: datetime,
401 specification: DeterministicCashflowBondSpecification,
402 cashflows: _Union[List[Tuple[datetime, float]], None] = None,
403 fwd_curve: _Union[DiscountCurve, None] = None,
404 ) -> float:
405 logger.info("Start computing bond z-spread for bond " + specification.obj_id + ", dirty price: " + str(target_dirty_price))
406 if cashflows is None:
407 cashflows = DeterministicCashflowPricer.get_expected_cashflows(specification, val_date, fwd_curve=fwd_curve)
409 def target_function(r: float) -> float:
410 dc = ConstantRate(r)
411 price = DeterministicCashflowPricer.get_pv_cashflows(val_date, specification, dc, cashflows=cashflows)
412 logger.debug("Target function called with r: " + str(r) + ", price: " + str(price) + ", target_dirty_price: " + str(target_dirty_price))
413 return price - target_dirty_price
415 result = brentq(target_function, -0.2, 1.5, full_output=False)
416 logger.info("Finished computing bond z-spread")
417 return result
419 @staticmethod
420 def get_z_spread(
421 target_dirty_price: float,
422 val_date: datetime,
423 specification: DeterministicCashflowBondSpecification,
424 discount_curve: DiscountCurve,
425 cashflows: _Union[List[Tuple[datetime, float]], None] = None,
426 fwd_curve: _Union[DiscountCurve, None] = None,
427 ) -> float:
428 logger.info("Start computing z-spread for bond " + specification.obj_id + ", dirty price: " + str(target_dirty_price))
429 if cashflows is None and fwd_curve is not None:
430 cashflows = DeterministicCashflowPricer.get_expected_cashflows(specification, val_date, fwd_curve=fwd_curve)
431 else:
432 logger.error("To compute z-spread with floating rate bonds, cashflows, or fwd_curve must be provided to calculate cashflows.")
434 def target_function(r: float) -> float:
435 dc = DiscountCurveComposition(discount_curve, 1.0, r)
436 price = DeterministicCashflowPricer.get_pv_cashflows(val_date, specification, dc, cashflows=cashflows)
437 logger.debug("Target function called with r: " + str(r) + ", price: " + str(price) + ", target_dirty_price: " + str(target_dirty_price))
438 return price - target_dirty_price
440 result = brentq(target_function, -0.2, 1.5, full_output=False)
441 logger.info("Finished computing z-spread.")
442 return result
444 def z_spread(self, target_dirty_price: float) -> float:
445 """Compute the z-spread of the bond.
447 Args:
448 target_dirty_price (float): The target dirty price.
449 Returns:
450 float: The computed z-spread.
451 """
452 return DeterministicCashflowPricer.get_z_spread(target_dirty_price, self._val_date, self._spec, self._discount_curve, self._cashflows)
454 ############################# PRICER ONLY METHODS BELOW #####################################
456 def macaulay_duration(self) -> float:
457 """Compute the Macaulay duration for the instrument.
459 Macaulay duration is the weighted average time until cashflows are received,
460 weighted by the present value of the cashflows.
462 Returns:
463 float: Macaulay duration expressed in the same time units used by day count.
464 """
465 logger.info("Start computing macaulay duration for bond " + self._spec.obj_id)
466 cashflows = self.expected_cashflows()
467 pv_cashflows = DeterministicCashflowPricer.get_pv_cashflows(
468 self._val_date, self._spec, self._discount_curve, self._fwd_curve, cashflows=cashflows
469 )
470 # print(pv_cashflows)
471 macaulay_duration = 0.0
472 dcc = DayCounter(self._spec.day_count_convention) # , self._spec._get_coupon_frequency if self._spec._coupon_type != "zero" else None)
473 for c in cashflows:
474 if c[0] > self._val_date:
475 df = self._discount_curve.value(self._val_date, c[0])
476 t = dcc.yf(self._val_date, c[0], self._spec.dates, self._spec.nr_annual_payments)
477 macaulay_duration += t * df * c[1]
478 if pv_cashflows > 0:
479 macaulay_duration /= pv_cashflows
480 logger.info("Finished computing macaulay duration for bond " + self._spec.obj_id + ", macaulay_duration: " + str(macaulay_duration))
481 return macaulay_duration
483 def modified_duration(
484 self,
485 target_dirty_price: float = 100.0,
486 ) -> float:
487 """Compute the modified duration for the instrument.
489 Modified duration approximates the percentage price change for a unit change
490 in yield and is computed from the Macaulay duration and yield.
492 Args:
493 target_dirty_price (float, optional): Dirty price used for yield inversion.
494 Defaults to 100.0.
496 Returns:
497 float: Modified duration.
498 """
499 logger.info("Start computing modified duration for bond " + self._spec.obj_id)
500 macaulay_duration = self.macaulay_duration()
501 yld = self.compute_yield(target_dirty_price)
502 modified_duration = macaulay_duration / (1 + yld)
503 logger.info("Finished computing modified duration for bond " + self._spec.obj_id + ", modified_duration: " + str(modified_duration))
504 return modified_duration