Coverage for rivapy / pricing / pricing_data.py: 29%
424 statements
« prev ^ index » next coverage.py v7.12.0, created at 2025-11-27 14:36 +0000
1from typing import Tuple, Iterable
2from datetime import datetime
3from dateutil.relativedelta import relativedelta
4from enum import IntEnum as _IntEnum
6from rivapy import _pyvacon_available
8if _pyvacon_available:
9 import pyvacon as _pyvacon
12from rivapy.instruments import CDSSpecification
14from rivapy.marketdata import DiscountCurve, SurvivalCurve
15from rivapy.tools.interfaces import BaseDatedCurve
16from typing import Union as _Union
17from datetime import date, datetime
18from rivapy.instruments.bond_specifications import BondBaseSpecification
19from rivapy.instruments.deposit_specifications import DepositSpecification
20from rivapy.instruments.fra_specifications import ForwardRateAgreementSpecification
21from rivapy.instruments.ir_swap_specification import InterestRateSwapSpecification, IrFixedLegSpecification, IrFloatLegSpecification
22from rivapy.tools._converter import _add_converter
23from rivapy.tools.datetools import _date_to_datetime
24from rivapy.pricing.pricing_request import (
25 PricingRequest,
26 BondPricingRequest,
27 DepositPricingRequest,
28 ForwardRateAgreementPricingRequest,
29 InterestRateSwapPricingRequest,
30)
32from rivapy.pricing.factory import _factory
34# from rivapy.pricing.deposit_pricing import DepositPricer
35# from rivapy.pricing.fra_pricing import ForwardRateAgreementPricer
37# double declaration
38# class CDSPricingData:
39# def __init__(self, spec, val_date, discount_curve, survival_curve, recovery_curve=None):
40# self.spec = spec
41# self.val_date = val_date
42# self.discount_curve = discount_curve
43# self.survival_curve = survival_curve
44# self.recovery_curve = recovery_curve
45# self._pricer_type = 'ISDA'
46#
47# def price(self):
48# pass
50if _pyvacon_available:
51 import pyvacon.pyvacon_swig as _analytics
53 BondPricingParameter = _add_converter(_analytics.BondPricingParameter)
54 # getPricingData = _converter(_analytics.getPricingData)
55else:
57 class BondPricingParameter:
58 pass
61class BasePricingData:
62 def __init__(self, pricer: str, pricing_request: PricingRequest):
63 self.pricer = pricer
64 self.pricing_request = pricing_request
65 # TODO: analyse if simulationData is needed (here)
67 @property
68 def pricer(self) -> str:
69 """
70 Getter for configured pricer.
72 Returns:
73 str: Configured pricer.
74 """
75 return self.__pricer
77 @pricer.setter
78 def pricer(self, pricer: str):
79 """
80 Setter for pricer configuration.
82 Args:
83 pricer (str): Pricer to be applied.
84 """
85 self.__pricer = pricer
87 @property
88 def pricing_request(self):
89 """
90 Getter for configured pricing request.
92 Returns:
93 PricingRequest: Configured pricing request.
94 """
95 return self.__pricing_request
97 @pricing_request.setter
98 def pricing_request(self, pricing_request: PricingRequest):
99 """
100 Setter for pricing request configuration.
102 Args:
103 pricing_request (PricingRequest): Configured pricing request.
104 """
105 self.__pricing_request = pricing_request
108class BondPricingData(BasePricingData):
109 def __init__(
110 self,
111 bond: BondBaseSpecification,
112 valuation_date: _Union[date, datetime],
113 discount_curve: DiscountCurve,
114 fixing_curve: DiscountCurve,
115 parameters: BondPricingParameter,
116 pricing_request: BondPricingRequest,
117 pricer: str = "BondPricer",
118 past_fixing: float = None,
119 survival_curve: SurvivalCurve = None,
120 recovery_curve: BaseDatedCurve = None,
121 ):
122 super().__init__(pricer, pricing_request)
123 self.__bond = bond # spec
124 self.valuation_date = valuation_date # valDate
125 self.discount_curve = discount_curve # discountCurve
126 self.fixing_curve = fixing_curve # fixingCurve
127 self.parameters = parameters # param
128 self.past_fixing = past_fixing # pastFixing
129 self.survival_curve = survival_curve # sc
130 self.recovery_curve = recovery_curve # recoveryCurve
132 @property
133 def bond(self):
134 return self.__bond
136 @property
137 def valuation_date(self):
138 return self.__valuation_date
140 @valuation_date.setter
141 def valuation_date(self, valuation_date: _Union[date, datetime]):
142 self.__valuation_date = _date_to_datetime(valuation_date)
144 @property
145 def discount_curve(self):
146 return self.__discount_curve
148 @discount_curve.setter
149 def discount_curve(self, discount_curve: DiscountCurve):
150 self.__discount_curve = discount_curve
152 @property
153 def fixing_curve(self):
154 return self.__fixing_curve
156 @fixing_curve.setter
157 def fixing_curve(self, fixing_curve: DiscountCurve):
158 self.__fixing_curve = fixing_curve
160 @property
161 def parameters(self):
162 return self.__parameters
164 @parameters.setter
165 def parameters(self, parameters: BondPricingParameter):
166 self.__parameters = parameters
168 @property
169 def past_fixing(self):
170 return self.__past_fixing
172 @past_fixing.setter
173 def past_fixing(self, past_fixing):
174 self.__past_fixing = past_fixing
176 @property
177 def survival_curve(self):
178 return self.__survival_curve
180 @survival_curve.setter
181 def survival_curve(self, survival_curve: SurvivalCurve):
182 self.__survival_curve = survival_curve
184 @property
185 def recovery_curve(self):
186 return self.__recovery_curve
188 @recovery_curve.setter
189 def recovery_curve(self, recovery_curve: BaseDatedCurve):
190 self.__recovery_curve = recovery_curve
193class ResultType(_IntEnum):
194 PRICE = 0
195 DELTA = 1
196 GAMMA = 2
197 THETA = 3
198 RHO = 4
199 VEGA = 5
200 VANNA = 6
203class PricingResults:
204 def set_price(self, price: float):
205 self._price = price
207 def getPrice(self):
208 return self._price
211def _create_pricing_request(pr_dict: Iterable[ResultType]):
212 result = _pyvacon.finance.pricing.PricingRequest()
213 for d in pr_dict:
214 if d is ResultType.DELTA or d is ResultType.GAMMA:
215 result.setDeltaGamma(True)
216 elif d is ResultType.THETA:
217 result.setTheta(True)
218 elif d is ResultType.RHO:
219 result.setRho(True)
220 elif d is ResultType.VEGA:
221 result.setVega(True)
222 elif d is ResultType.VANNA:
223 result.setVanna(True)
224 return result
227class Black76PricingData:
228 def __init__(self, val_date: datetime, spec, discount_curve, vol_surface, pricing_request: Iterable[ResultType]):
229 """Constructor for Black76PricingDate
231 Args:
232 val_date ([datetime]): Valuation date.
233 spec ([type]): Specification.
234 discount_curve ([type]): Discount curve.
235 vol_surface ([type]): Volatility surface.
236 pricing_request (Iterable[ResultType]): Pricing request. Can be selected from rivapy.pricing.ResultType.
237 """
239 self.spec = spec
240 self.val_date = val_date
241 self.discount_curve = discount_curve
242 self.vol_surface = vol_surface
243 self.pricing_request = pricing_request
244 self._pyvacon_obj = None
246 def _get_pyvacon_obj(self):
247 if self._pyvacon_obj is None:
248 self._pyvacon_obj = _pyvacon.finance.pricing.Black76PricingData()
249 self._pyvacon_obj.valDate = self.val_date
250 self._pyvacon_obj.spec = self.spec._get_pyvacon_obj()
251 self._pyvacon_obj.dsc = self.discount_curve._get_pyvacon_obj()
252 self._pyvacon_obj.param = _pyvacon.finance.pricing.PricingParameter()
253 self._pyvacon_obj.vol = self.vol_surface._get_pyvacon_obj()
254 self._pyvacon_obj.pricingRequest = _create_pricing_request(self.pricing_request)
255 return self._pyvacon_obj
257 def price(self):
258 return _pyvacon.finance.pricing.BasePricer.price(self._get_pyvacon_obj())
261class AmericanPdePricingData:
262 def __init__(
263 self,
264 val_date: datetime,
265 spec,
266 discount_curve,
267 vol_surface,
268 pricing_request: Iterable[ResultType],
269 time_steps_year: int = 60,
270 spot_steps: int = 200,
271 ):
272 """Constructor for AmericanPdePricingDate
274 Args:
275 val_date ([datetime]): Valuation date.
276 spec ([type]): Specification
277 discount_curve ([type]): Discount curve.
278 vol_surface ([type]): Volatility surface.
279 pricing_request (Iterable[ResultType]): Pricing request. Can be selected from rivapy.pricing.ResultType.
280 time_steps_year (int, optional): [description]. Defaults to 60.
281 spot_steps (int, optional): [description]. Defaults to 200.
282 """
284 self.val_date = val_date
285 self.spec = spec
286 self.discount_curve = discount_curve
287 self.vol_surface = vol_surface
288 self.pricing_request = pricing_request
289 self.time_steps_year = time_steps_year
290 self.spot_steps = spot_steps
291 self._pyvacon_obj = None
293 def _get_pyvacon_obj(self):
294 if self._pyvacon_obj is None:
295 self._pyvacon_obj = _pyvacon.finance.pricing.LocalVolPdePricingData()
296 self._pyvacon_obj.valDate = self.val_date
297 self._pyvacon_obj.spec = self.spec._get_pyvacon_obj().convertIntoBarrierSpecification()
298 self._pyvacon_obj.dsc = self.discount_curve._get_pyvacon_obj()
299 self._pyvacon_obj.param = _pyvacon.finance.pricing.PdePricingParameter()
300 self._pyvacon_obj.param.nTimeStepsPerYear = self.time_steps_year
301 self._pyvacon_obj.param.nSpotSteps = self.spot_steps
302 self._pyvacon_obj.vol = self.vol_surface._get_pyvacon_obj()
303 self._pyvacon_obj.pricingRequest = _create_pricing_request(self.pricing_request)
304 return self._pyvacon_obj
306 def price(self):
307 return _pyvacon.finance.pricing.BasePricer.price(self._get_pyvacon_obj())
310class CDSPricingData:
311 def __init__(
312 self, spec: CDSSpecification, val_date, discount_curve, survival_curve, recovery_curve=None, integration_step=relativedelta(days=30)
313 ):
314 self.spec = spec
315 self.val_date = val_date
316 self.discount_curve = discount_curve
317 self.survival_curve = survival_curve
318 self.recovery_curve = recovery_curve
319 self._pricer_type = "ISDA"
320 self.integration_step = integration_step
322 def _pv_protection_leg(self, valuation_date: datetime, integration_stepsize: relativedelta) -> float:
323 prev_date = max(self.val_date, self.spec.protection_start)
324 current_date = min(prev_date + self.integration_step, self.spec.expiry)
325 pv_protection = 0.0
327 while current_date <= self.spec.expiry:
328 default_prob = self.survival_curve.value(valuation_date, prev_date) - self.survival_curve.value(valuation_date, current_date)
329 recovery = self.spec.recovery
330 if recovery is None and self.recovery_curve is not None:
331 recovery = self.recovery_curve.value(valuation_date, current_date)
332 pv_protection += self.discount_curve.value(valuation_date, current_date) * (1.0 - recovery) * default_prob
333 prev_date = current_date
334 current_date += self.integration_step
336 if prev_date < self.spec.expiry and current_date > self.spec.expiry:
337 default_prob = self.survival_curve.value(valuation_date, prev_date) - self.survival_curve.value(valuation_date, self.spec.expiry)
338 recovery = self.spec.recovery
339 if recovery is None and self.recovery_curve is not None:
340 recovery = self.recovery_curve.value(valuation_date, self.spec.expiry)
341 pv_protection += self.discount_curve.value(valuation_date, self.spec.expiry) * (1.0 - recovery) * default_prob
343 return pv_protection
345 def _pv_premium_leg(self, valuation_date: datetime) -> Tuple[float, float]:
346 premium_period_start = self.spec.protection_start
347 risk_adj_factor_premium = 0
348 accrued = 0
349 # TODO include daycounter into CDSSpecification
350 dc = _pyvacon.finance.definition.DayCounter(_pyvacon.finance.definition.DayCounter.Type.Act365Fixed)
351 for premium_payment in self.spec.premium_pay_dates:
352 if premium_payment >= valuation_date:
353 period_length = dc.yf(premium_period_start, premium_payment)
354 survival_prob = self.survival_curve.value(valuation_date, premium_payment)
355 df = self.discount_curve.value(valuation_date, premium_payment)
356 risk_adj_factor_premium += period_length * survival_prob * df
357 default_prob = self.survival_curve.value(valuation_date, premium_period_start) - self.survival_curve.value(
358 valuation_date, premium_payment
359 )
360 accrued += period_length * default_prob * df
361 premium_period_start = premium_payment
362 return risk_adj_factor_premium, accrued
364 def par_spread(self, valuation_date: datetime, integration_stepsize: relativedelta) -> float:
365 prev_date = max(self.val_date, self.spec.protection_start)
366 current_date = min(prev_date + self.integration_step, self.spec.expiry)
367 pv_protection = 0.0
368 premium_period_start = self.spec.protection_start
369 risk_adj_factor_premium = 0
370 accrued = 0
372 while current_date <= self.spec.expiry:
373 default_prob = self.survival_curve.value(valuation_date, prev_date) - self.survival_curve.value(valuation_date, current_date)
374 recovery = self.spec.recovery
375 if recovery is None and self.recovery_curve is not None:
376 recovery = self.recovery_curve.value(valuation_date, current_date)
377 pv_protection += self.discount_curve.value(valuation_date, current_date) * (1.0 - recovery) * default_prob
378 prev_date = current_date
379 current_date += self.integration_step
381 if prev_date < self.spec.expiry and current_date > self.spec.expiry:
382 default_prob = self.survival_curve.value(valuation_date, prev_date) - self.survival_curve.value(valuation_date, self.spec.expiry)
383 recovery = self.spec.recovery
384 if recovery is None and self.recovery_curve is not None:
385 recovery = self.recovery_curve.value(valuation_date, self.spec.expiry)
386 pv_protection += self.discount_curve.value(valuation_date, self.spec.expiry) * (1.0 - recovery) * default_prob
388 dc = _pyvacon.finance.definition.DayCounter(_pyvacon.finance.definition.DayCounter.Type.Act365Fixed)
389 for premium_payment in self.spec.premium_pay_dates:
390 if premium_payment >= valuation_date:
391 period_length = dc.yf(premium_period_start, premium_payment)
392 survival_prob = self.survival_curve.value(valuation_date, premium_payment)
393 df = self.discount_curve.value(valuation_date, premium_payment)
394 risk_adj_factor_premium += period_length * survival_prob * df
395 default_prob = self.survival_curve.value(valuation_date, premium_period_start) - self.survival_curve.value(
396 valuation_date, premium_payment
397 )
398 accrued += period_length * default_prob * df
399 premium_period_start = premium_payment
401 PV_accrued = (1 / 2) * accrued
402 PV_premium = (1) * risk_adj_factor_premium
403 PV_protection = ((1 - recovery)) * pv_protection
405 par_spread_i = (PV_protection) / ((PV_premium + PV_accrued))
406 return par_spread_i
408 def price(self):
409 pv_protection = self._pv_protection_leg(self.val_date, self.integration_step)
410 pr_results = PricingResults()
411 pr_results.pv_protection = self.spec.notional * pv_protection
412 premium_leg, accrued = self._pv_premium_leg(self.val_date)
413 pr_results.premium_leg = self.spec.premium * self.spec.notional * premium_leg
414 pr_results.accrued = 0.5 * self.spec.premium * self.spec.notional * accrued
415 pr_results.par_spread = self.par_spread(self.val_date, self.integration_step)
416 pr_results.set_price(pr_results.pv_protection - pr_results.premium_leg - pr_results.accrued)
417 return pr_results
420class AnalyticSwaptionPricingData:
421 def __init__(self, val_date: datetime, spec, discount_curve, vol_cube, pricing_request: Iterable[ResultType]):
422 """Constructor for AnalyticSwaptionPricingData
424 Args:
425 val_date ([datetime]): Valuation date.
426 spec: Swaptions specification
427 discount_curve: Discount curve.
428 vol_cube: Volatility cube.
429 pricing_request (Iterable[ResultType]): Pricing request. Can be selected from rivapy.pricing.ResultType.
430 """
432 self.val_date = val_date
433 self.spec = spec
434 self.discount_curve = discount_curve
435 self.vol_cube = vol_cube
436 self.pricing_request = pricing_request
437 self._pyvacon_obj = None
439 def _get_pyvacon_obj(self):
440 if self._pyvacon_obj is None:
441 self._pyvacon_obj = _pyvacon.finance.pricing.AnalyticSwaptionPricingData()
442 self._pyvacon_obj.valDate = self.val_date
443 self._pyvacon_obj.spec = self.spec
444 self._pyvacon_obj.dsc = self.discount_curve._get_pyvacon_obj()
445 self._pyvacon_obj.volCube = self.vol_cube
446 self._pyvacon_obj.pricingRequest = _create_pricing_request(self.pricing_request)
447 self._pyvacon_obj.pricer = "AnalyticSwaptionPricer"
448 return self._pyvacon_obj
450 def price(self):
451 return _pyvacon.finance.pricing.BasePricer.price(self._get_pyvacon_obj())
454class AnalyticCapPricingData:
455 def __init__(self, val_date: datetime, spec, discount_curve, vol_surface, pricing_request: Iterable[ResultType]):
456 """Constructor for AnalyticCapPricingData
458 Args:
459 val_date ([datetime]): Valuation date.
460 spec: Specification
461 discount_curve: Discount curve.
462 vol_surface ([type]): Volatility surface.
463 pricing_request (Iterable[ResultType]): Pricing request. Can be selected from rivapy.pricing.ResultType.
464 """
466 self.val_date = val_date
467 self.spec = spec
468 self.discount_curve = discount_curve
469 self.vol_surface = vol_surface
470 self.pricing_request = pricing_request
471 self._pyvacon_obj = None
473 def _get_pyvacon_obj(self):
474 if self._pyvacon_obj is None:
475 self._pyvacon_obj = _pyvacon.finance.pricing.AnalyticCapPricingData()
476 self._pyvacon_obj.valDate = self.val_date
477 self._pyvacon_obj.spec = self.spec
478 self._pyvacon_obj.dscCurve = self.discount_curve._get_pyvacon_obj()
479 self._pyvacon_obj.volSurface = self.vol_surface
480 self._pyvacon_obj.pricingRequest = _create_pricing_request(self.pricing_request)
481 self._pyvacon_obj.pricer = "AnalyticCapPricer"
482 return self._pyvacon_obj
484 def price(self):
485 return _pyvacon.finance.pricing.BasePricer.price(self._get_pyvacon_obj())
488class InterestRateSwapPricingData:
489 def __init__(self, val_date: datetime, spec, ccy, leg_pricing_data, pricing_request: Iterable[ResultType]):
490 """Constructor for AnalyticCapPricingData
492 Args:
493 val_date ([datetime]): Valuation date.
494 spec: Specification
495 discount_curve: Discount curve.
496 vol_surface ([type]): Volatility surface.
497 pricing_request (Iterable[ResultType]): Pricing request. Can be selected from rivapy.pricing.ResultType.
498 """
500 self.val_date = val_date
501 self.spec = spec
502 self.ccy = ccy
503 self.leg_pricing_data = leg_pricing_data
504 self.pricing_request = pricing_request
505 self._pyvacon_obj = None
507 def _get_pyvacon_obj(self):
508 if self._pyvacon_obj is None:
509 self._pyvacon_obj = _pyvacon.finance.pricing.InterestRateSwapPricingData()
510 self._pyvacon_obj.pricer = "InterestRateSwapPricer"
511 self._pyvacon_obj.pricingRequest = _create_pricing_request(self.pricing_request)
512 self._pyvacon_obj.valDate = self.val_date
513 self._pyvacon_obj.setCurr(self.ccy)
514 for leg_data in self.leg_pricing_data:
515 self._pyvacon_obj.addLegData(leg_data._get_pyvacon_obj())
516 return self._pyvacon_obj
518 def price(self):
519 return _pyvacon.finance.pricing.BasePricer.price(self._get_pyvacon_obj())
522class InterestRateSwapLegPricingData:
523 def __init__(self, spec, discount_curve, fx_rate: float, weight: float):
524 """Constructor for AnalyticCapPricingData
526 Args:
527 val_date ([datetime]): Valuation date.
528 spec: Specification
529 discount_curve: Discount curve.
530 vol_surface ([type]): Volatility surface.
531 pricing_request (Iterable[ResultType]): Pricing request. Can be selected from rivapy.pricing.ResultType.
532 """
534 self.discount_curve = discount_curve
535 self.spec = spec
536 self.fx_rate = fx_rate
537 self.weight = weight
538 self._pyvacon_obj = None
540 def _get_pyvacon_obj(self):
541 if self._pyvacon_obj is None:
542 self._pyvacon_obj = _pyvacon.finance.pricing.InterestRateSwapLegPricingData()
543 self._pyvacon_obj.discountCurve = self.discount_curve._get_pyvacon_obj()
544 self._pyvacon_obj.spec = self.spec
545 self._pyvacon_obj.fxRate = self.fx_rate
546 self._pyvacon_obj.weight = self.weight
547 return self._pyvacon_obj
550class InterestRateSwapFloatLegPricingData:
551 def __init__(self, spec, discount_curve, fx_rate: float, weight: float, fixing_curve=None):
552 """Constructor for AnalyticCapPricingData
554 Args:
555 val_date ([datetime]): Valuation date.
556 spec: Specification
557 discount_curve: Discount curve.
558 vol_surface ([type]): Volatility surface.
559 pricing_request (Iterable[ResultType]): Pricing request. Can be selected from rivapy.pricing.ResultType.
560 """
562 self.discount_curve = discount_curve
563 self.fixing_curve = fixing_curve
564 self.spec = spec
565 self.fx_rate = fx_rate
566 self.weight = weight
567 self._pyvacon_obj = None
569 def _get_pyvacon_obj(self):
570 if self._pyvacon_obj is None:
571 self._pyvacon_obj = _pyvacon.finance.pricing.InterestRateSwapFloatLegPricingData()
572 self._pyvacon_obj.discountCurve = self.discount_curve._get_pyvacon_obj()
573 self._pyvacon_obj.fixingCurve = self.fixing_curve._get_pyvacon_obj()
574 self._pyvacon_obj.spec = self.spec
575 self._pyvacon_obj.fxRate = self.fx_rate
576 self._pyvacon_obj.weight = self.weight
577 return self._pyvacon_obj
580class DepositPricingData(BasePricingData):
582 def __init__(
583 self,
584 deposit: DepositSpecification,
585 val_date: _Union[date, datetime],
586 pricing_request: DepositPricingRequest,
587 pricer: str,
588 discount_curve: DiscountCurve,
589 # fixing_curve: DiscountCurve,
590 parameters: dict = None,
591 # past_fixing: float = None
592 ):
593 """Constructor for DepositPricingData
595 Args:
596 deposit (DepositSpecification): Instrument specific specification class object
597 valuation_date (_Union[date, datetime]): valuatiton date
598 pricing_request (DepositPricingRequest): Instrument specific Pricing Request class with the desired output/calculation prameters
599 pricer (str): chosen pricing algorithm
600 discount_curve (DiscountCurve): discount curve (i.e. (dates, discountFactors))
601 parameters (dict): Extra parameters...
602 """
604 super().__init__(pricer, pricing_request)
605 self.spec = deposit # spec
606 self.val_date = val_date # valDate
607 self.discount_curve = discount_curve # discountCurve
608 if parameters == None:
609 parameters = {}
610 self.parameters = parameters # param
612 # in the case for floating rate deposits?
614 # self.fixing_curve = fixing_curve # fixingCurve
615 # self.past_fixing = past_fixing # pastFixing
617 def price(self):
618 # obtain correct pricer
619 pricer_obj = _factory()["DepositPricer"]
620 pricer = pricer_obj(self.val_date, self.spec, self.discount_curve) # TODO ignore spread curve for now
621 # pricer = DepositPricer(self.valuation_date, self.__spec, self.discount_curve)
623 # pass correct required pricer information and calculate
624 val = pricer.price()
626 return val
629class ForwardRateAgreementPricingData(BasePricingData):
631 def __init__(
632 self,
633 fra: ForwardRateAgreementSpecification,
634 val_date: _Union[date, datetime],
635 pricing_request: ForwardRateAgreementPricingRequest,
636 pricer: str,
637 discount_curve: DiscountCurve,
638 forward_curve: DiscountCurve,
639 parameters: dict = None,
640 ):
641 """Constructor for ForwardrateAgreementPricingData
643 Args:
644 fra (ForwardRateAgreementSpecification): Instrument specific specification class object
645 valuation_date (_Union[date, datetime]): valuatiton date
646 pricing_request (DepositPricingRequest): Instrument specific Pricing Request class with the desired output/calculation prameters
647 pricer (str): chosen pricing algorithm
648 discount_curve (DiscountCurve): discount curve (i.e. (dates, discountFactors))
649 forward_curve (DiscountCurve): Forward curve (i.e. (dates, forward rate)) #TODO do we implement a ForwardCruve class?
650 parameters (dict): Extra parameters...
651 """
652 super().__init__(pricer, pricing_request)
654 self.spec = fra
655 self.val_date = val_date # valDate
656 self.discount_curve = discount_curve # discountCurve
657 self.forward_curve = forward_curve # discountCurve
658 if parameters == None:
659 parameters = {}
660 self.parameters = parameters # param
662 def price(self):
663 """Calculate the price of the forward rate agreement.
665 Raises:
666 ValueError: _description_
667 RuntimeError: _description_
669 Returns:
670 _type_: _description_
671 """
672 try:
673 factory = _factory()
674 if "ForwardRateAgreementPricer" not in factory:
675 raise ValueError("ForwardRateAgreementPricer not found in factory")
677 pricer_obj = factory["ForwardRateAgreementPricer"]
678 pricer = pricer_obj(self.val_date, self.spec, self.discount_curve, self.forward_curve)
679 return pricer.price()
680 except Exception as e:
681 raise RuntimeError(f"Error pricing forward rate agreement: {str(e)}")
684class InterestRateSwapPricingData_rivapy(BasePricingData): # TODO!!!!
686 def __init__(
687 self,
688 spec: InterestRateSwapSpecification,
689 val_date: _Union[date, datetime],
690 pricing_request: InterestRateSwapPricingRequest,
691 pricer: str,
692 ccy,
693 leg_pricing_data,
694 ):
695 """Constructor for
697 Args:
698 val_date ([datetime]): Valuation date.
699 spec: Specification
700 discount_curve: Discount curve.
701 vol_surface ([type]): Volatility surface.
702 pricing_request (Iterable[ResultType]): Pricing request. Can be selected from rivapy.pricing.ResultType.
703 """
705 super().__init__(pricer, pricing_request)
707 self.val_date = val_date
708 self.spec = spec
709 self.ccy = ccy
710 self.leg_pricing_data = leg_pricing_data
712 # TODO right now leg_pricing_data is expected to have:
713 # discount_curve_pay_leg: DiscountCurve,
714 # discount_curve_receive_leg: DiscountCurve,
715 # fixing_curve_pay_leg: DiscountCurve,
716 # fixing_curve_receive_leg: DiscountCurve,
717 # fx_fwd_curve_pay_leg: DiscountCurve, # TODO FxForwardCurve ... do we need anotheer class
718 # fx_fwd_curve_receive_leg: DiscountCurve,
719 # pricing_param: Dict = {}, #
720 # fixing_map : FixingTable = None,
721 # fx_pay_leg : float = 1.0,
722 # fx_receive_leg : float = 1.0
724 # unpacking other_pricing_data ...
725 self.discount_curve_pay_leg = leg_pricing_data["discount_curve_pay_leg"]
726 self.discount_curve_receive_leg = leg_pricing_data["discount_curve_receive_leg"]
727 self.fixing_curve_pay_leg = leg_pricing_data["fixing_curve_pay_leg"]
728 self.fixing_curve_receive_leg = leg_pricing_data["fixing_curve_receive_leg"]
729 self.fx_fwd_curve_pay_leg = leg_pricing_data["fx_fwd_curve_pay_leg"]
730 self.fx_fwd_curve_receive_leg = leg_pricing_data["fx_fwd_curve_receive_leg"]
731 self.pricing_param = leg_pricing_data["pricing_param"]
732 self.fixing_map = leg_pricing_data["fixing_map"]
733 self.fx_pay_leg = leg_pricing_data["fx_pay_leg"]
734 self.fx_receive_leg = leg_pricing_data["fx_receive_leg"]
736 def price(self):
737 # Obtain correct pricer, right now it is hardcoded for simplicity # TODO
738 pricer_obj = _factory()["InterestRateSwapPricer"] # not working for some reason?
739 # from rivapy.pricing.interest_rate_swap_pricing import InterestRateSwapPricer
741 print(self.discount_curve_pay_leg)
742 # pricer = InterestRateSwapPricer(
743 pricer = pricer_obj(
744 self.val_date,
745 self.spec,
746 discount_curve_pay_leg=self.discount_curve_pay_leg,
747 discount_curve_receive_leg=self.discount_curve_receive_leg,
748 fixing_curve_pay_leg=self.fixing_curve_pay_leg,
749 fixing_curve_receive_leg=self.fixing_curve_receive_leg,
750 fx_fwd_curve_pay_leg=self.fx_fwd_curve_pay_leg,
751 fx_fwd_curve_receive_leg=self.fx_fwd_curve_receive_leg,
752 pricing_request=self.pricing_request,
753 pricing_param=self.pricing_param,
754 fixing_map=self.fixing_map,
755 fx_pay_leg=self.fx_pay_leg,
756 fx_receive_leg=self.fx_receive_leg,
757 )
759 # pass correct required pricer information and calculate
760 val = pricer.price()
762 return val
765class InterestRateSwapLegPricingData_rivapy:
766 def __init__(
767 self,
768 spec,
769 discount_curve: DiscountCurve,
770 forward_curve: DiscountCurve,
771 fixing_map,
772 desired_rate=None,
773 fx_rate: float = 1.0,
774 weight: float = None,
775 ):
776 """Constructor for
778 Args:
779 val_date ([datetime]): Valuation date.
780 spec: Specification
781 discount_curve: Discount curve.
782 vol_surface ([type]): Volatility surface.
783 pricing_request (Iterable[ResultType]): Pricing request. Can be selected from rivapy.pricing.ResultType.
784 """
786 self.discount_curve = discount_curve
787 self.forward_curve = forward_curve
788 self.fixing_map = fixing_map
789 self.spec = spec
790 if fx_rate is not None: # where to use?
791 self.fx_rate = fx_rate
792 if weight is not None: # where to use?
793 self.weight = weight
794 if desired_rate is not None:
795 self.desired_rate = desired_rate
798class InterestRateSwapFloatLegPricingData_rivapy(InterestRateSwapLegPricingData_rivapy):
799 def __init__(
800 self,
801 spec,
802 discount_curve,
803 forward_curve,
804 fixing_map,
805 fixing_grace_period: int,
806 spread: float = None,
807 fx_rate: float = 1.0,
808 weight: float = None,
809 fixing_curve: DiscountCurve = None,
810 ):
811 """Constructor for
813 Args:
814 val_date ([datetime]): Valuation date.
815 spec: Specification
816 discount_curve: Discount curve.
817 vol_surface ([type]): Volatility surface.
818 pricing_request (Iterable[ResultType]): Pricing request. Can be selected from rivapy.pricing.ResultType.
819 """
821 # HERE the FX_RATE is ACTUALLY THE SPREAD
822 super().__init__(spec, discount_curve, forward_curve, fixing_map, fx_rate=fx_rate, weight=weight)
824 self.fixing_curve = fixing_curve
825 self.fixing_grace_period = fixing_grace_period
826 self.spread = spread