Coverage for rivapy / instruments / components.py: 66%

1# -*- coding: utf-8 -*- 

2from typing import Union as _Union, List, Tuple, Dict, Any, Optional 

3import numpy as np 

4from datetime import datetime, date 

5import rivapy.tools.interfaces as interfaces 

6from rivapy.tools.datetools import _date_to_datetime, Period 

7from rivapy.tools._validators import _check_positivity, _check_relation, _is_chronological 

8from rivapy.tools.enums import DayCounterType, Rating, Sector, Country, ESGRating 

9import abc 

10from rivapy.instruments._logger import logger 

11 

12 

13class Coupon: 

14 def __init__( 

15 self, 

16 accrual_start: _Union[date, datetime], 

17 accrual_end: _Union[date, datetime], 

18 payment_date: _Union[date, datetime], 

19 day_count_convention: _Union[DayCounterType, str], 

20 annualised_fixed_coupon: float, 

21 fixing_date: _Union[date, datetime], 

22 floating_period_start: _Union[date, datetime], 

23 floating_period_end: _Union[date, datetime], 

24 floating_spread: float = 0.0, 

25 floating_rate_cap: float = 1e10, 

26 floating_rate_floor: float = -1e10, 

27 floating_reference_index: str = "dummy_reference_index", 

28 amortisation_factor: float = 1.0, 

29 ): 

30 # accrual start and end date as well as payment date 

31 if _is_chronological(accrual_start, [accrual_end], payment_date): 

32 self.__accrual_start = accrual_start 

33 self.__accrual_end = accrual_end 

34 self.__payment_date = payment_date 

35 

36 self.__day_count_convention = DayCounterType.to_string(day_count_convention) 

37 

38 self.__annualised_fixed_coupon = _check_positivity(annualised_fixed_coupon) 

39 

40 self.__fixing_date = _date_to_datetime(fixing_date) 

41 

42 # spread on floating rate 

43 self.__spread = floating_spread 

44 

45 # cap/floor on floating rate 

46 self.__floating_rate_floor, self.__floating_rate_cap = _check_relation(floating_rate_floor, floating_rate_cap) 

47 

48 # reference index for fixing floating rates 

49 if floating_reference_index == "": 

50 # do not leave reference index empty as this causes pricer to ignore floating rate coupons! 

51 self.floating_reference_index = "dummy_reference_index" 

52 else: 

53 self.__floating_reference_index = floating_reference_index 

54 self.__amortisation_factor = _check_positivity(amortisation_factor) 

55 

56 

57class Issuer(interfaces.FactoryObject): 

58 def __init__( 

59 self, obj_id: str, name: str, rating: _Union[Rating, str], esg_rating: _Union[ESGRating, str], country: _Union[Country, str], sector: Sector 

60 ): 

61 self.__obj_id = obj_id 

62 self.__name = name 

63 self.__rating = Rating.to_string(rating) 

64 self.__esg_rating = ESGRating.to_string(esg_rating) 

65 self.__country = Country.to_string(country) 

66 self.__sector = Sector.to_string(sector) 

67 

68 @staticmethod 

69 def _create_sample( 

70 n_samples: int, 

71 seed: int = None, 

72 issuer: List[str] = None, 

73 rating_probs: np.ndarray = None, 

74 country_probs: np.ndarray = None, 

75 sector_probs: np.ndarray = None, 

76 esg_rating_probs: np.ndarray = None, 

77 ) -> List: 

78 """Just sample some test data 

79 

80 Args: 

81 n_samples (int): Number of samples. 

82 seed (int, optional): If set, the seed is set, if None, no seed is explicitely set. Defaults to None. 

83 issuer (List[str], optional): List of issuer names chosen from. If None, a unqiue name for each samples is generated. Defaults to None. 

84 rating_probs (np.ndarray): Numpy array defining the probability for each rating (ratings ordererd from AAA (first) to D (last array element)). If None, all ratings are chosen with equal probabilities. 

85 Raises: 

86 Exception: _description_ 

87 

88 Returns: 

89 List: List of sampled issuers. 

90 """ 

91 if seed is not None: 

92 np.random.seed(seed) 

93 result = [] 

94 ratings = list(Rating) 

95 if rating_probs is not None: 

96 if len(ratings) != rating_probs.shape[0]: 

97 raise Exception("Number of rating probabilities must equal number of ratings") 

98 else: 

99 rating_probs = np.ones( 

100 ( 

101 len( 

102 ratings, 

103 ) 

104 ) 

105 ) / len(ratings) 

106 

107 if country_probs is not None: 

108 if len(Country) != country_probs.shape[0]: 

109 raise Exception("Number of country probabilities must equal number of countries") 

110 else: 

111 country_probs = np.ones( 

112 ( 

113 len( 

114 Country, 

115 ) 

116 ) 

117 ) / len(Country) 

118 

119 if sector_probs is not None: 

120 if len(Sector) != sector_probs.shape[0]: 

121 raise Exception("Number of sector probabilities must equal number of sectors") 

122 else: 

123 sector_probs = np.ones( 

124 ( 

125 len( 

126 Sector, 

127 ) 

128 ) 

129 ) / len(Sector) 

130 

131 if esg_rating_probs is not None: 

132 if len(ESGRating) != esg_rating_probs.shape[0]: 

133 raise Exception("Number of ESG rating probabilities must equal number of ESG ratings") 

134 else: 

135 esg_rating_probs = np.ones( 

136 ( 

137 len( 

138 ESGRating, 

139 ) 

140 ) 

141 ) / len(ESGRating) 

142 

143 esg_ratings = list(ESGRating) 

144 sectors = list(Sector) 

145 country = list(Country) 

146 if issuer is None: 

147 issuer = ["Issuer_" + str(i) for i in range(n_samples)] 

148 elif (n_samples is not None) and (n_samples != len(issuer)): 

149 raise Exception("Cannot create data since length of issuer list does not equal number of samples. Set n_namples to None.") 

150 for i in range(n_samples): 

151 result.append( 

152 Issuer( 

153 "Issuer_" + str(i), 

154 issuer[i], 

155 np.random.choice(ratings, p=rating_probs), 

156 np.random.choice(esg_ratings, p=esg_rating_probs), 

157 np.random.choice(country, p=country_probs), 

158 np.random.choice(sectors, p=sector_probs), 

159 ) 

160 ) 

161 return result 

162 

163 def _to_dict(self) -> dict: 

164 return { 

165 "obj_id": self.obj_id, 

166 "name": self.name, 

167 "rating": self.rating, 

168 "esg_rating": self.esg_rating, 

169 "country": self.country, 

170 "sector": self.sector, 

171 } 

172 

173 @property 

174 def obj_id(self) -> str: 

175 """ 

176 Getter for issuer id. 

177 

178 Returns: 

179 str: Issuer id. 

180 """ 

181 return self.__obj_id 

182 

183 @property 

184 def name(self) -> str: 

185 """ 

186 Getter for issuer name. 

187 

188 Returns: 

189 str: Issuer name. 

190 """ 

191 return self.__name 

192 

193 @property 

194 def rating(self) -> str: 

195 """ 

196 Getter for issuer's rating. 

197 

198 Returns: 

199 Rating: Issuer's rating. 

200 """ 

201 return self.__rating 

202 

203 @rating.setter 

204 def rating(self, rating: _Union[Rating, str]): 

205 """ 

206 Setter for issuer's rating. 

207 

208 Args: 

209 rating: Rating of issuer. 

210 """ 

211 self.__rating = Rating.to_string(rating) 

212 

213 @property 

214 def esg_rating(self) -> str: 

215 """ 

216 Getter for issuer's rating. 

217 

218 Returns: 

219 Rating: Issuer's rating. 

220 """ 

221 return self.__esg_rating 

222 

223 @esg_rating.setter 

224 def esg_rating(self, esg_rating: _Union[ESGRating, str]): 

225 """ 

226 Setter for issuer's rating. 

227 

228 Args: 

229 rating: Rating of issuer. 

230 """ 

231 self.__esg_rating = ESGRating.to_string(esg_rating) 

232 

233 @property 

234 def country(self) -> str: 

235 """ 

236 Getter for issuer's country. 

237 

238 Returns: 

239 Country: Issuer's country. 

240 """ 

241 return self.__country 

242 

243 @property 

244 def sector(self) -> str: 

245 """ 

246 Getter for issuer's sector. 

247 

248 Returns: 

249 Sector: Issuer's sector. 

250 """ 

251 return self.__sector 

252 

253 @sector.setter 

254 def sector(self, sector: _Union[Sector, str]) -> str: 

255 """ 

256 Setter for issuer's sector. 

257 

258 Returns: 

259 Sector: Issuer's sector. 

260 """ 

261 self.__sector = Sector.to_string(sector) 

262 

263 

264class CashFlow: 

265 # goal is to define a dynamically growing class that is still able to use 

266 # type validation and dot-access e.g. class.variable 

267 # the point for dynamically growing is to allow for flexibility of future development and use cases 

268 # In the end, it might be better to just define clearly the CashFlow class with 

269 # strict attributes ... #TODO 

270 

271 # Define expected types here 

272 # Can be expanded when we know for sure which features we want to ensure typing for 

273 _schema = { 

274 "start_date": datetime, 

275 "end_date": datetime, 

276 "ccy": str, 

277 "amortization": bool, 

278 "prepayment_risk": bool, 

279 } 

280 

281 def __init__(self, val: float = None): 

282 self.val = val 

283 self._attributes = {} 

284 

285 def __getattr__(self, name: str) -> Any: 

286 """overwritting default getter for dynamically growing one 

287 

288 Args: 

289 name (str): name of the the desired attribute 

290 

291 Raises: 

292 AttributeError: attribute name not included 

293 

294 Returns: 

295 Any: value of the desired attribute 

296 """ 

297 try: 

298 return self._attributes[name] 

299 except KeyError: 

300 raise AttributeError(f"'{type(self).__name__}' object has no attribute '{name}'") 

301 

302 def __setattr__(self, name: str, value: Any): 

303 """overwriting default setter for dynamically growing one 

304 which also checks for expected type validation. 

305 

306 Args: 

307 name (str): new name for desired attribute 

308 value (Any): value to be stored in desired attribute 

309 

310 Raises: 

311 TypeError: For known attributes defined in schema, raise error if type mismatch for value 

312 """ 

313 if name in {"val", "_attributes"}: # avoid infinite recursion 

314 super().__setattr__(name, value) # use the the normal attribute storage from base class 

315 else: # logic for new attirbute storage 

316 expected_type = self._schema.get(name) # if it doesnt exist, can attempt to set new attribute 

317 if expected_type is not None and not isinstance(value, expected_type): 

318 raise TypeError(f"Attribute '{name}' must be of type {expected_type}, got {type(value)}") 

319 self._attributes[name] = value 

320 

321 def __delattr__(self, name: str): 

322 if name in self._attributes: 

323 del self._attributes[name] 

324 else: 

325 raise AttributeError(f"'{type(self).__name__}' object has no attribute '{name}'") 

326 

327 def keys(self): 

328 return list(self._attributes.keys()) 

329 

330 def items(self): 

331 return self._attributes.items() 

332 

333 def __dir__(self): 

334 """overwritten in order to show dynamically stored attributes as well. 

335 

336 Returns: 

337 _type_: _description_ 

338 """ 

339 return super().__dir__() + list(self._attributes.keys()) 

340 

341 @staticmethod 

342 def _create_sample(n_samples: int, seed: int = None): 

343 """Creates a sample of random ``CashFlow`` objects. 

344 

345 Returns: 

346 List[CashFlow]: List of sampled ``CashFlow`` objects 

347 """ 

348 result = [] 

349 if seed is not None: 

350 np.random.seed(seed) 

351 

352 for i in range(n_samples): 

353 cashflow_val = np.random.choice(np.arange(1000, 10000, 100), 1)[0] 

354 result.append( 

355 { 

356 "val": cashflow_val, 

357 } 

358 ) 

359 

360 def _to_dict(self) -> dict: 

361 result = { 

362 "val": self.val, 

363 "attributes": self._attributes, 

364 } 

365 return result 

366 

367 

368class NotionalStructure(interfaces.FactoryObject): 

369 """Abstract base class for notional structures.""" 

370 

371 @abc.abstractmethod 

372 def __init__(self): 

373 pass 

374 

375 @abc.abstractmethod 

376 def get_amount(self, period: int = None) -> float: 

377 """here, period is the INDEX that maps to the notional amount""" 

378 pass 

379 

380 def get_pay_date_start(self, period: int) -> Optional[datetime]: 

381 """get the notional exchange date at the beginning of the period 

382 

383 Args: 

384 period (int): is the index of the period 

385 

386 Returns: 

387 the date of notional exchange at the beginning of the period, 

388 or None if there is no notional exchange at the beginning 

389 of the period 

390 

391 """ 

392 

393 return None 

394 

395 def get_pay_date_end(self, period: int) -> Optional[datetime]: 

396 """get the notional exchange date at the end of the period 

397 

398 Args: 

399 period (int): is the index of the period 

400 

401 Returns: 

402 the date of notional exchange at the beginning of the period, 

403 or None if there is no notional exchange at the beginning 

404 of the period 

405 

406 """ 

407 return None 

408 

409 def get_amount_per_date(self, date: date) -> float: 

410 pass 

411 

412 def get_amortization_schedule(self) -> List[Tuple[date, float]]: 

413 pass 

414 

415 @abc.abstractmethod 

416 def get_size(self) -> int: 

417 pass 

418 

419 @abc.abstractmethod 

420 def _to_dict(self) -> Dict: 

421 return_dict = {} 

422 return return_dict 

423 

424 

425class ConstNotionalStructure(NotionalStructure): 

426 """Constant notional means that it does not change over the lifetime. 

427 Meaning that there are no notional cashflows as well, inflow or outflow. 

428 

429 Args: 

430 NotionalStructure (_type_): _description_ 

431 """ 

432 

433 def __init__(self, notional: float): 

434 """Constructor for a notional structure with a constant notional. 

435 

436 Args: 

437 notional (float): _description_ 

438 """ 

439 self._notional = [notional] 

440 self._start_date = None 

441 self._end_date = None 

442 

443 # region properties 

444 

445 @property 

446 def notional(self) -> float: 

447 return self._notional 

448 

449 @notional.setter 

450 def notional(self, notional: float): 

451 self._notional[0] = notional 

452 

453 @property 

454 def start_date(self) -> list[datetime]: 

455 if self._start_date is None: 

456 return None 

457 return self._start_date 

458 

459 @start_date.setter 

460 def start_date(self, start_date: list[datetime]): 

461 self._start_date = start_date 

462 

463 @property 

464 def end_date(self) -> list[datetime]: 

465 if self._end_date is None: 

466 return None 

467 else: 

468 return self._end_date 

469 

470 @end_date.setter 

471 def end_date(self, end_date: list[datetime]): 

472 self._end_date = end_date 

473 

474 # endregion 

475 

476 # region class methods 

477 def get_amount(self, period: int = None) -> float: 

478 """Get the value of the notional. 

479 

480 Note: Kept list structure to stay consistent with other notional structures. 

481 However, expectation is that of only one entry in this list. 

482 

483 Args: 

484 period (int): index rerferencing to a specific period of rolled out notional 

485 

486 Returns: 

487 float: notional value 

488 """ 

489 if period is not None and period > 1: 

490 logger.warning("ConstNotionalStructure only has one period with constant notional.") 

491 return self._notional[0] 

492 

493 def get_amount_per_date(self, date): 

494 return self._notional[0] 

495 

496 def get_size(self) -> int: 

497 """If the notional structure is constant, we expect the size to be 1. 

498 Otherwise, return the amount of notional time stamps used. 

499 

500 Returns: 

501 int: _description_ 

502 """ 

503 return len(self._notional) 

504 

505 def get_amortizations_by_index(self) -> List[Tuple[int, float]]: 

506 return [(1, self._notional[0])] 

507 

508 def get_amortization_schedule(self) -> Optional[List[Tuple[date, float]]]: 

509 """Return amortization schedule as list of (date, amount) or None if end dates are missing. 

510 

511 Returns: 

512 Optional[List[Tuple[date, float]]]: amortization schedule or None when end dates are not set 

513 """ 

514 if getattr(self, "_end_date", None) is None: 

515 # use plural message to be consistent with other notional structures 

516 logger.error("End dates of notional structure are not set.") 

517 return [] 

518 else: 

519 return [(self._end_date, self._notional[0])] 

520 

521 def _to_dict(self) -> Dict: 

522 return_dict = { 

523 "notional": self._notional, 

524 } 

525 return return_dict 

526 

527 # endregion 

528 

529 

530class LinearNotionalStructure(NotionalStructure): 

531 def __init__(self, start_notional: float, end_notional: float = 0.0, n_steps: int = 1): 

532 """Constructor for a linear notional structure 

533 

534 Args: 

535 start_notional (float): notional at the beginning of the structure 

536 end_notional (float): notional at the end of the structure, set to start_notional if not provided 

537 n_steps (int): number of steps to linearly interpolate between start and end notional, results in n_steps amortizations 

538 """ 

539 if n_steps < 1: 

540 raise ValueError("n_steps must be at least 1") 

541 self._notional = list(np.linspace(start_notional, end_notional, n_steps)) 

542 self._start_notional = start_notional 

543 self._end_notional = end_notional 

544 self._n_steps = n_steps 

545 self._start_date = None 

546 self._end_date = None 

547 self._dates = None 

548 

549 # region properties 

550 

551 @property 

552 def n_steps(self) -> int: 

553 return self._n_steps 

554 

555 @n_steps.setter 

556 def n_steps(self, n_steps: int): 

557 self._n_steps = n_steps 

558 self._notional = list(np.linspace(self._start_notional, self._end_notional, n_steps)) 

559 # print(self._notional) 

560 

561 @property 

562 def start_date(self) -> list[datetime]: 

563 return self._start_date 

564 

565 @start_date.setter 

566 def start_date(self, start_date: list[datetime]): 

567 self._start_date = start_date 

568 

569 @property 

570 def end_date(self) -> list[datetime]: 

571 return self._end_date 

572 

573 @end_date.setter 

574 def end_date(self, end_date: list[datetime]): 

575 self._end_date = end_date 

576 

577 @property 

578 def notional(self) -> list[float]: 

579 return self._notional 

580 

581 @notional.setter 

582 def notional(self, notional: list[float]): 

583 self._notional = notional 

584 self._start_notional = notional[0] 

585 self._end_notional = notional[-1] 

586 

587 @property 

588 def start_notional(self) -> float: 

589 return self._start_notional 

590 

591 @start_notional.setter 

592 def start_notional(self, start_notional: float): 

593 self._start_notional = start_notional 

594 self._notional = list(np.linspace(self._start_notional, self._end_notional, self._n_steps)) 

595 

596 @property 

597 def end_notional(self) -> float: 

598 return self._end_notional 

599 

600 @end_notional.setter 

601 def end_notional(self, end_notional: float): 

602 self._end_notional = end_notional 

603 self._notional = list(np.linspace(self._start_notional, self._end_notional, self._n_steps)) 

604 

605 # endregion 

606 

607 # region class methods 

608 

609 def get_amount(self, period: int = None) -> float: 

610 if period is None: 

611 return self._notional[0] 

612 return self._notional[period] 

613 

614 def get_amount_per_date(self, date): 

615 if self._end_date is None or self._start_date is None: 

616 raise Exception("Start or end dates of notional structure are not set.") 

617 if date > self._end_date[-1]: 

618 raise Exception("Date is after end date of notional structure") 

619 earlier_dates = [d for d in self._start_date if d <= date] 

620 if not earlier_dates: 

621 raise Exception("Date is before start date of notional structure") 

622 # Find the last one and return its index 

623 return self._notional[self._start_date.index(earlier_dates[-1])] 

624 

625 def get_size(self) -> int: 

626 """Returns the number of notionals 

627 

628 Returns: 

629 int: _description_ 

630 """ 

631 return len(self._notional) 

632 

633 def get_amortizations_by_index(self) -> List[Tuple[int, float]]: 

634 """Returns a list of tuples (index, notional) representing the amortizations by index.""" 

635 amortizations = [] 

636 n = len(self._notional) 

637 if n <= 1: 

638 return [(1, self._start_notional - self._end_notional)] 

639 # compute the per-step change between consecutive notionals 

640 per_step_change = float(self._notional[0] - self._notional[1]) 

641 # The tests expect an entry for each step index (1..n) repeating the per-step change 

642 for i in range(1, n): 

643 amortizations.append((i, per_step_change)) 

644 return amortizations 

645 

646 def get_amortization_schedule(self) -> List[Tuple[date, float]]: 

647 """Returns a list of tuples (date, notional) representing the amortization schedule.""" 

648 schedule = [] 

649 if self._end_date is None: 

650 logger.error("End dates of notional structure are not set.") 

651 else: 

652 laenge = len(self._notional) 

653 # print(laenge) 

654 if len(self._notional) == 1: 

655 return [(self._end_date[0], self._start_notional - self._end_notional)] 

656 else: 

657 for i in range(1, len(self._notional)): 

658 change = self._notional[i - 1] - self._notional[i] 

659 n1 = self._notional[i - 1] 

660 n2 = self._notional[i] 

661 schedule.append((self._end_date[i - 1], change)) 

662 return schedule 

663 

664 def _to_dict(self) -> Dict: 

665 # TODO fill out more 

666 return_dict = { 

667 "notional": self._notional, 

668 } 

669 return return_dict 

670 

671 # endregion 

672 

673 

674class VariableNotionalStructure(NotionalStructure): 

675 def __init__(self, notionals: list[float], pay_date_start: list[datetime], pay_date_end: list[datetime]): 

676 """Constructor for a variable notional structure 

677 

678 Args: 

679 notionals (list[float]): values for each period, referenced by index and matched to the pay_date_start/end 

680 pay_date_start (list[datetime]): start date of the payment period 

681 pay_date_end (list[datetime]): end date of the payment period 

682 """ 

683 self._notional = notionals 

684 self._pay_date_start = pay_date_start 

685 self._pay_date_end = pay_date_end 

686 

687 def get_amount(self, period: int = None) -> float: 

688 if period is None: 

689 return self._notional[0] 

690 return self._notional[period] 

691 

692 def get_pay_date_start(self, period: int) -> datetime: 

693 return self._pay_date_start[period] 

694 

695 def get_pay_date_end(self, period: int) -> datetime: 

696 return self._pay_date_end[period] 

697 

698 def get_size(self) -> int: 

699 """Returns the number of notionals 

700 

701 Returns: 

702 int: _description_ 

703 """ 

704 return len(self._notional) 

705 

706 def _to_dict(self) -> Dict: 

707 # TODO fill out more 

708 return_dict = { 

709 "notional": self._notional, 

710 "pay_date_start": self._pay_date_start, 

711 "pay_date_end": self._pay_date_end, 

712 } 

713 return return_dict 

714 

715 

716class ResettingNotionalStructure(NotionalStructure): 

717 def __init__( 

718 self, 

719 ref_currency: str, 

720 fx_fixing_id: str, 

721 notionals: list[float], 

722 pay_date_start: list[datetime], 

723 pay_date_end: list[datetime], 

724 fixing_dates: list[datetime], 

725 ): 

726 """Notional is recalculated/reset dynamically based on underlying referenced by fx_fixing_id at specific datets (fixing_dates) 

727 

728 Args: 

729 ref_currency (str): Currency of the reference 

730 fx_fixing_id (str): Id of the fixing 

731 notionals (list[float]): notional values 

732 pay_date_start (list[datetime]): start of accrual period for that notional 

733 pay_date_end (list[datetime]): end of accrual period for that notional 

734 fixing_dates (list[datetime]): date at which notional is reset 

735 """ 

736 

737 self._ref_currency = ref_currency 

738 self._fx_fixing_id = fx_fixing_id 

739 self._notional = notionals 

740 self._pay_date_start = pay_date_start 

741 self._pay_date_end = pay_date_end 

742 self._fixing_date = fixing_dates 

743 

744 def get_amount(self, period: int) -> float: 

745 return self._notional[period] 

746 

747 def get_pay_date_start(self, period: int) -> datetime: 

748 return self._pay_date_start[period] 

749 

750 def get_pay_date_end(self, period: int) -> datetime: 

751 return self._pay_date_end[period] 

752 

753 def get_fixing_date(self, period: int) -> datetime: 

754 return self._fixing_date[period] 

755 

756 def get_reference_currency(self) -> str: 

757 return self._ref_currency 

758 

759 def get_size(self) -> int: 

760 """Returns the number of notionals 

761 

762 Returns: 

763 int: _description_ 

764 """ 

765 return len(self._notional) 

766 

767 def _to_dict(self) -> Dict: 

768 # TODO fill out more 

769 return_dict = { 

770 "notional": self._notional, 

771 "pay_date_start": self._pay_date_start, 

772 "pay_date_end": self._pay_date_end, 

773 "ref_currency": self._ref_currency, 

774 "fx_fixing_id": self._fx_fixing_id, 

775 "fixing_date": self._fixing_date, 

776 } 

777 return return_dict 

778 

779 

780class AmortizationScheme(interfaces.FactoryObject): 

781 """ 

782 Abstract base class for amortization schemes. 

783 - none --> constant --> ConstNotionalStructure 

784 - linear --> linear amortization --> LinearNotionalStructure 

785 - variable --> variable amortization --> VariableNotionalStructure 

786 - requires list of percentages and periods/dates(!?) 

787 - requires consistency of dates to instrument dates at least regarding start and end date of the instrument 

788 - requires implementation of abstract methods 

789 - methods: get_amortization_periods, get_amortization_percentages_per_period, get_total_amortization_percentage, _to_dict, etc. 

790 - subclasses implement specific schemes 

791 """ 

792 

793 @abc.abstractmethod 

794 def __init__(self): 

795 pass 

796 pass 

797 

798 @abc.abstractmethod 

799 def get_total_amortization(self) -> float: 

800 pass 

801 

802 @abc.abstractmethod 

803 def _to_dict(self) -> Dict: 

804 pass 

805 

806 @classmethod 

807 def _from_string(cls, data: Optional[str] = None) -> "AmortizationScheme": 

808 """Create an AmortizationScheme object from a string representation. 

809 

810 Args: 

811 data (str): String representation of the AmortizationScheme. 

812 

813 Returns: 

814 AmortizationScheme: The created AmortizationScheme object. 

815 """ 

816 if data == "linear": 

817 return LinearAmortizationScheme() # default to single step 

818 elif data == "constant" or data is None: 

819 return ZeroAmortizationScheme() 

820 else: 

821 raise ValueError(f"Unknown AmortizationScheme type: {data}") 

822 

823 

824class LinearAmortizationScheme(AmortizationScheme): 

825 def __init__(self, total_amortization: float = 100.0, n_steps: int = 1): 

826 """Constructor for a linear amortization scheme 

827 

828 Args: 

829 n_steps (int): number of steps to linearly amortize the notional 

830 total_amortization (float): total amortization percentage (default is 100.0) 

831 """ 

832 if n_steps < 1: 

833 raise ValueError("n_steps must be at least 1") 

834 else: 

835 self._n_steps = n_steps 

836 if total_amortization < 0.0 or total_amortization > 100.0: 

837 raise ValueError("total_amortization must be between 0.0 and 100.0") 

838 else: 

839 self._total_amortization = total_amortization 

840 

841 @property 

842 def n_steps(self) -> int: 

843 return self._n_steps 

844 

845 @n_steps.setter 

846 def n_steps(self, n_steps: int): 

847 if n_steps < 1: 

848 raise ValueError("n_steps must be at least 1") 

849 else: 

850 self._n_steps = n_steps 

851 

852 @property 

853 def total_amortization(self) -> float: 

854 return self._total_amortization 

855 

856 @total_amortization.setter 

857 def total_amortization(self, total_amortization: float): 

858 if total_amortization < 0.0 or total_amortization > 100.0: 

859 raise ValueError("total_amortization must be between 0.0 and 100.0") 

860 else: 

861 self._total_amortization = total_amortization 

862 

863 def get_total_amortization(self) -> float: 

864 return self._total_amortization 

865 

866 def _to_dict(self) -> Dict: 

867 # TODO fill out more 

868 return_dict = { 

869 "n_steps": self._n_steps, 

870 "total_amortization": self._total_amortization, 

871 } 

872 return return_dict 

873 

874 

875class ZeroAmortizationScheme(AmortizationScheme): 

876 def __init__(self): 

877 """Constructor for a constant amortization scheme (no amortization)""" 

878 self._total_amortization = 0.0 

879 

880 # @property 

881 # def total_amortization(self) -> float: 

882 # return self._total_percentage 

883 

884 # @total_amortization.setter 

885 # def total_amortization(self, total_percentage: float):