Coverage for rivapy/pricing/bond_pricing.py: 100%

1from datetime import datetime 

2from scipy.optimize import brentq 

3from rivapy.tools.interfaces import BaseDatedCurve, HasExpectedCashflows 

4from rivapy.marketdata import DiscountCurveParametrized, ConstantRate 

5from rivapy.pricing.pricing_request import PricingRequest 

6from rivapy.pricing._logger import logger 

7 

8class SimpleCashflowPricer: 

9 @staticmethod 

10 def pv_cashflows(val_date: datetime, specification: HasExpectedCashflows, discount_curve: BaseDatedCurve)->float: 

11 logger.info('Start computing pv cashflows for bond ' + specification.obj_id) 

12 

13 cashflows = specification.expected_cashflows() 

14 pv_cashflows = 0.0 

15 for c in cashflows: 

16 if c[0]>= val_date: 

17 df = discount_curve.value(val_date, c[0]) 

18 logger.debug('Cashflow ' + str(c[1]) + ', date: ' + str(c[0]) + ', df: ' + str(df)) 

19 pv_cashflows += df*c[1] 

20 logger.info('Finished computing pv cashflows for bond ' + specification.obj_id + ', pv_cashflows: '+ str(pv_cashflows) ) 

21 return pv_cashflows 

22 

23 @staticmethod 

24 def compute_yield(target_dirty_price: float, val_date: datetime, 

25 specification: HasExpectedCashflows)->float: 

26 logger.info('Start computing bond yield for bond ' + specification.obj_id + ', dirty price: ' + str(target_dirty_price)) 

27 def target_function(r: float)->float: 

28 dc = DiscountCurveParametrized('', val_date, ConstantRate(r)) 

29 price = SimpleCashflowPricer.pv_cashflows(val_date, specification, dc) 

30 logger.debug('Target function called with r: ' + str(r) + ', price: ' + str(price) + ', target_dirty_price: ' + str(target_dirty_price)) 

31 return price - target_dirty_price 

32 result = brentq(target_function, -0.2, 1.5, full_output = False) 

33 logger.info('Finished computing bond yield') 

34 return result 

35