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

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

2from typing import Union as _Union, List 

3import numpy as np 

4from datetime import datetime, date 

5import rivapy.tools.interfaces as interfaces 

6from rivapy.tools.datetools import _date_to_datetime 

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

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

9 

10 

11class Coupon: 

12 def __init__(self, 

13 accrual_start: _Union[date, datetime], 

14 accrual_end: _Union[date, datetime], 

15 payment_date: _Union[date, datetime], 

16 day_count_convention: _Union[DayCounterType, str], 

17 annualised_fixed_coupon: float, 

18 fixing_date: _Union[date, datetime], 

19 floating_period_start: _Union[date, datetime], 

20 floating_period_end: _Union[date, datetime], 

21 floating_spread: float = 0.0, 

22 floating_rate_cap: float = 1e10, 

23 floating_rate_floor: float = -1e10, 

24 floating_reference_index: str = 'dummy_reference_index', 

25 amortisation_factor: float = 1.0): 

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

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

28 self.__accrual_start = accrual_start 

29 self.__accrual_end = accrual_end 

30 self.__payment_date = payment_date 

31 

32 self.__day_count_convention = DayCounterType.to_string(day_count_convention) 

33 

34 self.__annualised_fixed_coupon = _check_positivity(annualised_fixed_coupon) 

35 

36 self.__fixing_date = _date_to_datetime(fixing_date) 

37 

38 # spread on floating rate 

39 self.__spread = floating_spread 

40 

41 # cap/floor on floating rate 

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

43 

44 # reference index for fixing floating rates 

45 if floating_reference_index == '': 

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

47 self.floating_reference_index = 'dummy_reference_index' 

48 else: 

49 self.__floating_reference_index = floating_reference_index 

50 self.__amortisation_factor = _check_positivity(amortisation_factor) 

51 

52 

53class Issuer(interfaces.FactoryObject): 

54 def __init__(self, 

55 obj_id: str, 

56 name: str, 

57 rating: _Union[Rating, str], 

58 esg_rating: _Union[ESGRating, str], 

59 country: _Union[Country, str], 

60 sector: Sector): 

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(n_samples: int, seed: int = None, issuer: List[str] = None, 

70 rating_probs: np.ndarray = None, 

71 country_probs:np.ndarray = None, 

72 sector_probs:np.ndarray = None, 

73 esg_rating_probs:np.ndarray=None )->List: 

74 """Just sample some test data 

75 

76 Args: 

77 n_samples (int): Number of samples. 

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

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

80 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. 

81 Raises: 

82 Exception: _description_ 

83 

84 Returns: 

85 List: List of sampled issuers. 

86 """ 

87 if seed is not None: 

88 np.random.seed(seed) 

89 result = [] 

90 ratings = list(Rating) 

91 if rating_probs is not None: 

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

93 raise Exception('Number of rating probabilities must equal number of ratings') 

94 else: 

95 rating_probs = np.ones((len(ratings,)))/len(ratings) 

96 

97 if country_probs is not None: 

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

99 raise Exception('Number of country probabilities must equal number of countries') 

100 else: 

101 country_probs = np.ones((len(Country,)))/len(Country) 

102 

103 if sector_probs is not None: 

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

105 raise Exception('Number of sector probabilities must equal number of sectors') 

106 else: 

107 sector_probs = np.ones((len(Sector,)))/len(Sector) 

108 

109 if esg_rating_probs is not None: 

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

111 raise Exception('Number of ESG rating probabilities must equal number of ESG ratings') 

112 else: 

113 esg_rating_probs = np.ones((len(ESGRating,)))/len(ESGRating) 

114 

115 

116 

117 esg_ratings = list(ESGRating) 

118 sectors = list(Sector) 

119 country = list(Country) 

120 if issuer is None: 

121 issuer = ['Issuer_'+str(i) for i in range(n_samples)] 

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

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

124 for i in range(n_samples): 

125 result.append(Issuer('Issuer_'+str(i), issuer[i], 

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

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

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

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

130 return result 

131 

132 def _to_dict(self) -> dict: 

133 return {'obj_id': self.obj_id, 

134 'name': self.name, 'rating': self.rating, 

135 'esg_rating': self.esg_rating, 

136 'country': self.country, 'sector': self.sector} 

137 

138 @property 

139 def obj_id(self) -> str: 

140 """ 

141 Getter for issuer id. 

142 

143 Returns: 

144 str: Issuer id. 

145 """ 

146 return self.__obj_id 

147 

148 @property 

149 def name(self) -> str: 

150 """ 

151 Getter for issuer name. 

152 

153 Returns: 

154 str: Issuer name. 

155 """ 

156 return self.__name 

157 

158 @property 

159 def rating(self) -> str: 

160 """ 

161 Getter for issuer's rating. 

162 

163 Returns: 

164 Rating: Issuer's rating. 

165 """ 

166 return self.__rating 

167 

168 @rating.setter 

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

170 """ 

171 Setter for issuer's rating. 

172 

173 Args: 

174 rating: Rating of issuer. 

175 """ 

176 self.__rating =Rating.to_string(rating) 

177 

178 @property 

179 def esg_rating(self) -> str: 

180 """ 

181 Getter for issuer's rating. 

182 

183 Returns: 

184 Rating: Issuer's rating. 

185 """ 

186 return self.__esg_rating 

187 

188 @esg_rating.setter 

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

190 """ 

191 Setter for issuer's rating. 

192 

193 Args: 

194 rating: Rating of issuer. 

195 """ 

196 self.__esg_rating = ESGRating.to_string(esg_rating) 

197 

198 @property 

199 def country(self) -> str: 

200 """ 

201 Getter for issuer's country. 

202 

203 Returns: 

204 Country: Issuer's country. 

205 """ 

206 return self.__country 

207 

208 @property 

209 def sector(self) -> str: 

210 """ 

211 Getter for issuer's sector. 

212 

213 Returns: 

214 Sector: Issuer's sector. 

215 """ 

216 return self.__sector 

217 

218 @sector.setter 

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

220 """ 

221 Setter for issuer's sector. 

222 

223 Returns: 

224 Sector: Issuer's sector. 

225 """ 

226 self.__sector = Sector.to_string(sector) 

227