Coverage for rivapy / instruments / specification_from_csv.py: 99%
173 statements
« prev ^ index » next coverage.py v7.12.0, created at 2025-11-27 14:36 +0000
1import pandas as pd
2from datetime import datetime, timedelta
3from dateutil.relativedelta import relativedelta
4from typing import Union as _Union
5from rivapy.tools.holidays_compat import HolidayBase as _HolidayBase, EuropeanCentralBank as _ECB
6from rivapy.instruments._logger import logger
7from rivapy.instruments import (
8 DepositSpecification,
9 ForwardRateAgreementSpecification,
10 InterestRateSwapSpecification,
11 IrFixedLegSpecification,
12 IrFloatLegSpecification,
13 IrOISLegSpecification,
14 InterestRateBasisSwapSpecification,
15)
16from rivapy.instruments.components import ConstNotionalStructure
17from rivapy.tools.datetools import (
18 DayCounter,
19 Period,
20 Schedule,
21 _date_to_datetime,
22 _datetime_to_date_list,
23 _term_to_period,
24 roll_day,
25 calc_start_day,
26 calc_end_day,
27)
30def load_specifications_from_pd(df: pd.DataFrame, ref_date: datetime, calendar: _Union[_HolidayBase, str] = _ECB()):
31 """Takes in a pandas data frame which must have has the required columns.
33 Args:
34 df (pd.DataFrame): Contains the column information for the market quotes of a given instrument
36 Returns:
37 List[Specification]: List of Specification items used in Rivapy, e.g., in yield curve bootstrapping.
38 """
39 # df = pd.read_csv(file_path, parse_dates=True)
40 specs = []
41 for _, row in df.iterrows():
42 # if row["Maturity"] == "2M": # DEBUG TEST 2025
43 # logger.debug("debugging for spepcific instrument case")
44 spec = make_specification_from_row(row, ref_date, calendar)
45 specs.append(spec)
46 return specs
49def make_specification_from_row(row: pd.DataFrame, ref_date: datetime, calendar: _Union[_HolidayBase, str] = _ECB()):
50 inst_type = row["Instrument"].upper()
52 if inst_type == "DEPOSIT":
53 return make_deposit_spec(row, ref_date, calendar)
54 elif inst_type == "OIS":
55 return make_ois_spec(row, ref_date, calendar)
56 elif inst_type == "FRA":
57 return make_fra_spec(row, ref_date, calendar)
58 elif inst_type == "IRS":
59 return make_irswap_spec(row, ref_date, calendar)
60 elif inst_type == "TBS":
61 return make_basis_swap_spec(row, ref_date, calendar)
62 else:
63 raise ValueError(f"Unsupported instrument type {inst_type}")
66def make_deposit_spec(row: pd.DataFrame, ref_date: datetime, calendar: _Union[_HolidayBase, str] = _ECB()):
67 """Create a deposit specification object given the required information from an input data frame row.
69 Args:
70 row (pd.DataFrame): Row containing the required information for the deposit specification specified by header information
71 ref_date (datetime): The reference date for the deposit instrument
72 calendar (_Union[_HolidayBase, str], optional): calendar object from which date calculations are affected. Defaults to _ECB().
74 Returns:
75 _type_: DepositSpecification object
76 """
77 label = row["Instrument"] + "_" + row["Maturity"]
79 dep_spec = DepositSpecification(
80 obj_id=label,
81 issue_date=ref_date,
82 # end_date: _Optional[_Union[date, datetime]] = None,
83 # start_date: _Optional[_Union[date, datetime]] = None,
84 # maturity_date: _Optional[_Union[date, datetime]] = None,
85 currency=row["Currency"],
86 # notional: float = 100.0, # we let notional default to 100
87 rate=float(row["Quote"]),
88 term=row["Maturity"],
89 day_count_convention=row["DayCountFloat"],
90 business_day_convention=row["RollConventionFloat"],
91 # roll_convention: _Union[RollRule, str] = RollRule.EOM, # leave as default
92 spot_days=int(row["SpotLag"][:-1]), # make assumption it is always given in DAYS convert -> int
93 calendar=calendar,
94 issuer="dummy_issuer",
95 securitization_level="NONE",
96 # payment_days: int = 0,
97 # adjust_start_date: bool = True,
98 # adjust_end_date: bool = False,
99 )
101 return dep_spec
104def make_fra_spec(row: pd.DataFrame, ref_date: datetime, calendar: _Union[_HolidayBase, str] = _ECB()):
105 """Create a Forward rate agreement (FRA) specification object given the required information from an input data frame row.
106 In the Maturity column, the format is expected to be of the form "YYMxZZM" where YY is the number of months to the start date
107 and ZZ is the number of months to the end date. This is different compared to the other instruments in this column by design.
109 Args:
110 row (pd.DataFrame): Row containing the required information for the FRA specification specified by header information
111 ref_date (datetime): The reference date for the FRA instrument
112 calendar (_Union[_HolidayBase, str], optional): calendar object from which date calculations are affected. Defaults to _ECB().
114 Returns:
115 _type_: ForwardRateAgreement Specification object
116 """
117 label = row["Instrument"] + "_" + row["Maturity"]
118 # maturity must be in the form of YYMxZZM where YY and ZZ are integers
119 s = row["Maturity"].upper()
120 start_str, end_str = s.split("X")
121 start_period = int(start_str[:-1]) # remove last "M"
122 end_period = int(end_str[:-1])
124 spot_date = roll_day(
125 day=ref_date + timedelta(days=int(row["SpotLag"][:-1])),
126 calendar=calendar,
127 business_day_convention=row["RollConventionFloat"],
128 start_day=None,
129 )
131 start_date = roll_day(
132 day=spot_date + relativedelta(months=start_period),
133 calendar=calendar,
134 business_day_convention=row["RollConventionFloat"],
135 start_day=None,
136 ) # spot_date + start_period #need roll convention: ddc, bdc, holiday, date
137 end_date = roll_day(
138 day=start_date + relativedelta(months=end_period - start_period),
139 calendar=calendar,
140 business_day_convention=row["RollConventionFloat"],
141 start_day=None,
142 ) # start_date + end_period #need roll convention: ddc, bdc, holiday, date
144 fra_spec = ForwardRateAgreementSpecification(
145 obj_id=label,
146 trade_date=ref_date,
147 notional=1.0,
148 rate=float(row["Quote"]),
149 start_date=start_date,
150 end_date=end_date,
151 udlID=row["UnderlyingIndex"],
152 rate_start_date=start_date,
153 rate_end_date=end_date,
154 # maturity_date=,
155 day_count_convention=row["DayCountFixed"],
156 business_day_convention=row["RollConventionFixed"],
157 rate_day_count_convention=row["DayCountFloat"],
158 rate_business_day_convention=row["RollConventionFloat"],
159 calendar=calendar,
160 currency=row["Currency"],
161 # payment_days: int = 0,
162 spot_days=int(row["SpotLag"][:-1]),
163 # start_period: int = None,
164 # end_period: int = None,
165 # ir_index: str = None,
166 # issuer: str = None,
167 )
169 return fra_spec
172def make_irswap_spec(row: pd.DataFrame, ref_date: datetime, calendar: _Union[_HolidayBase, str] = _ECB()):
173 """Create a interest rate swap (IRS) specification object given the required information from an input data frame row.
174 Creates both fixed and floating legs.
176 Args:
177 row (pd.DataFrame): Row containing the required information for the IRS specification specified by header information
178 ref_date (datetime): The reference date for the IRS instrument
179 calendar (_Union[_HolidayBase, str], optional): calendar object from which date calculations are affected. Defaults to _ECB().
181 Returns:
182 _type_: IRS specification object
183 """
184 # the following information is expected:
185 instr = row["Instrument"]
186 fixDayCount = row["DayCountFixed"]
187 floatDayCount = row["DayCountFloat"]
188 basisDayCount = row["DayCountBasis"]
189 maturity = row["Maturity"]
190 underlyingIndex = row["UnderlyingIndex"]
191 tenor = row["UnderlyingTenor"]
192 underlyingPayFreq = row["UnderlyingPaymentFrequency"]
193 basisTenor = row["BasisTenor"]
194 basisPayFreq = row["BasisPaymentFrequency"]
195 fixPayFreq = row["PaymentFrequencyFixed"]
196 rollConvFloat = row["RollConventionFloat"]
197 rollConvFix = row["RollConventionFixed"]
198 rollConvBasis = row["RollConventionBasis"]
199 spotLag = row["SpotLag"] # expect form "1D", i.e 1 day
200 parRate = float(row["Quote"])
201 currency = row["Currency"]
202 label = instr + "_" + maturity
204 # we use the helper function with spotlag in place of maturity to effctively shift the date
205 spot_date = calc_end_day(start_day=ref_date, term=spotLag, business_day_convention=rollConvFix, calendar=calendar)
206 expiry = calc_end_day(spot_date, maturity, rollConvFix, calendar) # get expiry of swap (cannot be before last paydate of legs)
208 # FIXED LEG
209 fix_schedule = Schedule(
210 start_day=spot_date, end_day=expiry, time_period=fixPayFreq, business_day_convention=rollConvFix, calendar=calendar, ref_date=ref_date
211 ).generate_dates(False)
213 fix_start_dates = fix_schedule[:-1]
214 fix_end_dates = fix_schedule[1:]
215 fix_pay_dates = fix_end_dates
217 # # definition of the fixed leg
218 fixed_leg = IrFixedLegSpecification(
219 fixed_rate=parRate,
220 obj_id=label + "_fixed_leg3",
221 notional=1.0,
222 start_dates=fix_start_dates,
223 end_dates=fix_end_dates,
224 pay_dates=fix_pay_dates,
225 currency=currency,
226 day_count_convention=fixDayCount,
227 )
229 # FLOAT LEG
230 flt_schedule = Schedule(
231 start_day=spot_date,
232 end_day=expiry,
233 time_period=underlyingPayFreq,
234 business_day_convention=rollConvFloat,
235 calendar=calendar,
236 ref_date=ref_date,
237 ).generate_dates(False)
239 flt_start_dates = flt_schedule[:-1]
240 flt_end_dates = flt_schedule[1:]
241 flt_pay_dates = flt_end_dates
243 flt_reset_schedule = Schedule(
244 start_day=spot_date, end_day=expiry, time_period=tenor, business_day_convention=rollConvFloat, calendar=calendar, ref_date=ref_date
245 ).generate_dates(False)
247 flt_reset_dates = flt_reset_schedule[:-1]
249 ns = ConstNotionalStructure(1.0)
250 spread = 0.00
252 # # definition of the floating leg
253 float_leg = IrFloatLegSpecification(
254 obj_id=label + "_float_leg",
255 notional=ns,
256 reset_dates=flt_reset_dates,
257 start_dates=flt_start_dates,
258 end_dates=flt_end_dates,
259 rate_start_dates=flt_start_dates,
260 rate_end_dates=flt_end_dates,
261 pay_dates=flt_pay_dates,
262 currency=currency,
263 udl_id=underlyingIndex,
264 fixing_id="test_fixing_id",
265 day_count_convention=floatDayCount,
266 spread=spread,
267 )
269 # # definition of the IR swap - assume fixed leg is the pay leg
270 ir_swap = InterestRateSwapSpecification(
271 obj_id=label,
272 notional=ns,
273 issue_date=ref_date,
274 maturity_date=expiry,
275 pay_leg=fixed_leg,
276 receive_leg=float_leg,
277 currency=currency,
278 day_count_convention=floatDayCount,
279 issuer="dummy_issuer",
280 securitization_level="COLLATERALIZED",
281 )
283 return ir_swap
286def make_ois_spec(row: pd.DataFrame, ref_date: datetime, calendar: _Union[_HolidayBase, str] = _ECB()):
287 """Create an overnight index swap (OIS) specification object given the required information from an input data frame row.
288 Creates both fixed and floating legs.
290 Args:
291 row (pd.DataFrame): Row containing the required information for the OIS specification specified by header information
292 ref_date (datetime): The reference date for the OIS instrument
293 calendar (_Union[_HolidayBase, str], optional): calendar object from which date calculations are affected. Defaults to _ECB().
295 Returns:
296 _type_: OIS specification object
297 """
298 label = row["Instrument"] + "_" + row["Maturity"]
300 # the following information is expected:
301 instr = row["Instrument"]
302 fixDayCount = row["DayCountFixed"]
303 floatDayCount = row["DayCountFloat"]
304 basisDayCount = row["DayCountBasis"]
305 maturity = row["Maturity"]
306 underlyingIndex = row["UnderlyingIndex"]
307 tenor = row["UnderlyingTenor"]
308 underlyingPayFreq = row["UnderlyingPaymentFrequency"]
309 basisTenor = row["BasisTenor"]
310 basisPayFreq = row["BasisPaymentFrequency"]
311 fixPayFreq = row["PaymentFrequencyFixed"]
312 rollConvFloat = row["RollConventionFloat"]
313 rollConvFix = row["RollConventionFixed"]
314 rollConvBasis = row["RollConventionBasis"]
315 spotLag = row["SpotLag"] # expect form "1D", i.e 1 day
316 parRate = float(row["Quote"])
317 currency = row["Currency"]
318 label = instr + "_" + maturity
320 # we use the helper function with spotlag in place of maturity to effctively shift the date
321 spot_date = calc_end_day(start_day=ref_date, term=spotLag, business_day_convention=rollConvFix, calendar=calendar)
322 expiry = calc_end_day(spot_date, maturity, rollConvFix, calendar) # get expiry of swap (cannot be before last paydate of legs)
323 expiry_unadjusted = calc_end_day(start_day=spot_date, term=maturity, calendar=calendar)
325 # FIXED LEG
326 fix_schedule = Schedule(
327 start_day=spot_date,
328 end_day=expiry_unadjusted,
329 time_period=fixPayFreq,
330 business_day_convention=rollConvFix,
331 calendar=calendar,
332 ref_date=ref_date,
333 ).generate_dates(False)
335 if fix_schedule[-1] != expiry:
336 logger.error(
337 "Unexpected schedule generation for OIS fixed leg: last date in schedule {} does not match adjusted expiry {}".format(
338 fix_schedule[-1], expiry
339 )
340 )
342 fix_start_dates = fix_schedule[:-1]
343 fix_end_dates = fix_schedule[1:]
344 fix_pay_dates = fix_end_dates
346 # # definition of the fixed leg
347 fixed_leg = IrFixedLegSpecification(
348 fixed_rate=parRate,
349 obj_id=label + "_fixed_leg3",
350 notional=1.0,
351 start_dates=fix_start_dates,
352 end_dates=fix_end_dates,
353 pay_dates=fix_pay_dates,
354 currency=currency,
355 day_count_convention=fixDayCount,
356 )
358 # FLOAT LEG - OIS
359 flt_schedule = Schedule(
360 start_day=spot_date,
361 end_day=expiry_unadjusted,
362 time_period=underlyingPayFreq,
363 business_day_convention=rollConvFloat,
364 calendar=calendar,
365 ref_date=ref_date,
366 ).generate_dates(False)
368 flt_start_dates = flt_schedule[:-1]
369 flt_end_dates = flt_schedule[1:]
370 flt_pay_dates = flt_end_dates
372 flt_reset_schedule = Schedule(
373 start_day=spot_date, end_day=expiry, time_period=tenor, business_day_convention=rollConvFloat, calendar=calendar, ref_date=ref_date
374 ).generate_dates(False)
376 flt_reset_dates = flt_reset_schedule[:-1]
378 res = IrOISLegSpecification.ois_scheduler_2D(flt_start_dates, flt_end_dates)
380 daily_rate_start_dates = res[0] # 2D list: coupon i -> list of daily starts
381 daily_rate_end_dates = res[1] # 2D list: coupon i -> list of daily ends
382 daily_rate_reset_dates = res[2] # 2D list: coupon i -> list of reset dates
383 daily_rate_pay_dates = res[3]
385 ns = ConstNotionalStructure(1.0)
386 spread = 0.00
388 # # definition of the floating leg
390 float_leg = IrOISLegSpecification(
391 obj_id=label + "_float_leg",
392 notional=ns,
393 rate_reset_dates=daily_rate_reset_dates,
394 start_dates=flt_start_dates,
395 end_dates=flt_end_dates,
396 rate_start_dates=daily_rate_start_dates,
397 rate_end_dates=daily_rate_end_dates,
398 pay_dates=daily_rate_pay_dates,
399 currency=currency,
400 udl_id=underlyingIndex,
401 fixing_id="test_fixing_id",
402 day_count_convention=floatDayCount,
403 rate_day_count_convention=floatDayCount,
404 spread=spread,
405 )
407 # # definition of the IR swap - assume fixed leg is the pay leg
408 ois = InterestRateSwapSpecification(
409 obj_id=label,
410 notional=ns,
411 issue_date=ref_date,
412 maturity_date=expiry,
413 pay_leg=fixed_leg,
414 receive_leg=float_leg,
415 currency=currency,
416 day_count_convention=floatDayCount,
417 issuer="dummy_issuer",
418 securitization_level="COLLATERALIZED",
419 )
421 return ois
424def make_basis_swap_spec(row: pd.DataFrame, ref_date: datetime, calendar: _Union[_HolidayBase, str] = _ECB()):
425 """Create a basis swap (BS) specification object given the required information from an input data frame row.
426 Creates both fixed and floating legs.
428 Args:
429 row (pd.DataFrame): Row containing the required information for the IRS specification specified by header information
430 ref_date (datetime): The reference date for the IRS instrument
431 calendar (_Union[_HolidayBase, str], optional): calendar object from which date calculations are affected. Defaults to _ECB().
433 Returns:
434 _type_: IRS specification object
435 """
436 # TODO THIS NEEDS to be ammended to take in the new expected maturtiy or TBS instrument
437 # type and to correctls parse the information and generates dates for the
438 # pay leg, recieve leg, and the "spread leg"
439 # mainly basd on tenors. the spprad leg will by design have the frequency of the pay leg
440 # the following information is expected:
441 instr = row["Instrument"]
442 fixDayCount = row["DayCountFixed"]
443 floatDayCount = row["DayCountFloat"]
444 basisDayCount = row["DayCountBasis"]
445 maturity = row["Maturity"]
446 underlyingIndex = row["UnderlyingIndex"]
447 tenor = row["UnderlyingTenor"]
448 underlyingPayFreq = row["UnderlyingPaymentFrequency"]
449 basisTenor = row["BasisTenor"]
450 basisPayFreq = row["BasisPaymentFrequency"]
451 fixPayFreq = row["PaymentFrequencyFixed"]
452 rollConvFloat = row["RollConventionFloat"]
453 rollConvFix = row["RollConventionFixed"]
454 rollConvBasis = row["RollConventionBasis"]
455 spotLag = row["SpotLag"] # expect form "1D", i.e 1 day
457 # special for Basis swap, e.g. TBS
458 tenorShort = row["UnderlyingTenorShort"]
459 underlyingPayFreqShort = row["UnderlyingPaymentFrequencyShort"]
460 fixPayFreqShort = row["PaymentFrequencyFixedShort"]
462 # FOR A TBS, the quote given is the basis point spread, e.g. +8.5 bps
463 spreadRate = float(row["Quote"]) / 10000.0 # convert to decimal
464 currency = row["Currency"]
465 label = instr + "_" + maturity
467 # we use the helper function with spotlag in place of maturity to effctively shift the date
468 spot_date = calc_end_day(start_day=ref_date, term=spotLag, business_day_convention=rollConvFloat, calendar=calendar)
469 expiry = calc_end_day(spot_date, maturity, rollConvFloat, calendar) # get expiry of swap (cannot be before last paydate of legs)
470 ns = ConstNotionalStructure(1.0)
471 spread = float(row["Quote"])
473 # -------------------------------
474 # PAY LEG (float) - short tenor
475 pay_schedule = Schedule(
476 start_day=spot_date,
477 end_day=expiry,
478 time_period=underlyingPayFreqShort,
479 business_day_convention=rollConvFloat,
480 calendar=calendar,
481 ref_date=ref_date,
482 ).generate_dates(False)
484 pay_start_dates = pay_schedule[:-1]
485 pay_end_dates = pay_schedule[1:]
486 pay_pay_dates = pay_end_dates
488 pay_reset_schedule = Schedule(
489 start_day=spot_date, end_day=expiry, time_period=tenorShort, business_day_convention=rollConvFloat, calendar=calendar, ref_date=ref_date
490 ).generate_dates(False)
492 pay_reset_dates = pay_reset_schedule[:-1]
494 # # definition of the floating leg
495 pay_leg = IrFloatLegSpecification(
496 obj_id=label + "_pay_leg",
497 notional=ns,
498 reset_dates=pay_reset_dates,
499 start_dates=pay_start_dates,
500 end_dates=pay_end_dates,
501 rate_start_dates=pay_start_dates,
502 rate_end_dates=pay_end_dates,
503 pay_dates=pay_pay_dates,
504 currency=currency,
505 udl_id=underlyingIndex,
506 fixing_id="test_fixing_id",
507 day_count_convention=floatDayCount,
508 spread=0.0, # what should this be?
509 )
511 # -------------------------------
512 # RECIEVE LEG (float) - long tenor
513 rec_schedule = Schedule(
514 start_day=spot_date,
515 end_day=expiry,
516 time_period=underlyingPayFreq,
517 business_day_convention=rollConvFloat,
518 calendar=calendar,
519 ref_date=ref_date,
520 ).generate_dates(False)
522 rec_start_dates = rec_schedule[:-1]
523 rec_end_dates = rec_schedule[1:]
524 rec_pay_dates = rec_end_dates
526 rec_reset_schedule = Schedule(
527 start_day=spot_date, end_day=expiry, time_period=tenor, business_day_convention=rollConvFloat, calendar=calendar, ref_date=ref_date
528 ).generate_dates(
529 False
530 ) # TODO NEEDS CHANGE
532 rec_reset_dates = rec_reset_schedule[:-1]
534 # # definition of the floating leg
535 receive_leg = IrFloatLegSpecification(
536 obj_id=label + "_receive_leg",
537 notional=ns,
538 reset_dates=rec_reset_dates,
539 start_dates=rec_start_dates,
540 end_dates=rec_end_dates,
541 rate_start_dates=rec_start_dates,
542 rate_end_dates=rec_end_dates,
543 pay_dates=rec_pay_dates,
544 currency=currency,
545 udl_id=underlyingIndex,
546 fixing_id="test_fixing_id",
547 day_count_convention=floatDayCount,
548 spread=0.0,
549 )
551 # -------------------------------
552 # The spread leg represents the fixed +x bps cashflows applied to the pay leg
553 # same payment frerq as short leg
554 spread_schedule = Schedule(
555 start_day=spot_date, end_day=expiry, time_period=fixPayFreqShort, business_day_convention=rollConvFix, calendar=calendar, ref_date=ref_date
556 ).generate_dates(False)
558 spread_start_dates = spread_schedule[:-1]
559 spread_end_dates = spread_schedule[1:]
560 spread_pay_dates = spread_end_dates
562 # # definition of the SPREAD leg - which represents the
563 spread_leg = IrFixedLegSpecification(
564 fixed_rate=spreadRate,
565 obj_id=label + "_spread_leg",
566 notional=1.0,
567 start_dates=spread_start_dates,
568 end_dates=spread_end_dates,
569 pay_dates=spread_pay_dates,
570 currency=currency,
571 day_count_convention=fixDayCount,
572 )
574 # # definition of the IR swap - assume fixed leg is the pay leg
575 basis_swap = InterestRateBasisSwapSpecification(
576 obj_id=label,
577 notional=ns,
578 issue_date=ref_date,
579 maturity_date=expiry,
580 pay_leg=pay_leg,
581 receive_leg=receive_leg,
582 spread_leg=spread_leg,
583 currency=currency,
584 day_count_convention=floatDayCount,
585 issuer="dummy_issuer",
586 securitization_level="COLLATERALIZED",
587 )
589 return basis_swap