diff options
Diffstat (limited to 'python/analytics/utils.py')
| -rw-r--r-- | python/analytics/utils.py | 61 |
1 files changed, 40 insertions, 21 deletions
diff --git a/python/analytics/utils.py b/python/analytics/utils.py index 8a738a2b..8905000c 100644 --- a/python/analytics/utils.py +++ b/python/analytics/utils.py @@ -9,27 +9,42 @@ from pandas.api.types import CategoricalDtype from pandas.tseries.offsets import CustomBusinessDay, Day, QuarterBegin from pandas.tseries.holiday import get_calendar, HolidayCalendarFactory, GoodFriday -fed_cal = get_calendar('USFederalHolidayCalendar') -bond_cal = HolidayCalendarFactory('BondCalendar', fed_cal, GoodFriday) +fed_cal = get_calendar("USFederalHolidayCalendar") +bond_cal = HolidayCalendarFactory("BondCalendar", fed_cal, GoodFriday) bus_day = CustomBusinessDay(calendar=bond_cal()) from quantlib.time.date import nth_weekday, Wednesday, Date -tenor_t = CategoricalDtype(['1m', '3m', '6m', '1yr', '2yr', '3yr', '4yr', - '5yr', '7yr', '10yr', '15yr', '20yr', '25yr', - '30yr'], - ordered=True) +tenor_t = CategoricalDtype( + [ + "1m", + "3m", + "6m", + "1yr", + "2yr", + "3yr", + "4yr", + "5yr", + "7yr", + "10yr", + "15yr", + "20yr", + "25yr", + "30yr", + ], + ordered=True, +) def GHquad(n): """Gauss-Hermite quadrature weights""" Z, w = h_roots(n) - return Z*np.sqrt(2), w/np.sqrt(np.pi) + return Z * np.sqrt(2), w / np.sqrt(np.pi) def next_twentieth(d): r = d + relativedelta(day=20) - if(r < d): + if r < d: r += relativedelta(months=1) mod = r.month % 3 if mod != 0: @@ -54,13 +69,13 @@ def next_third_wed(d): def roll_date(d, tenor, nd_array=False): """ roll date d to the next CDS maturity""" - cutoff = pd.Timestamp('2015-09-20') + cutoff = pd.Timestamp("2015-09-20") def kwargs(t): if abs(t) == 0.5: - return {'months': int(12 * t)} + return {"months": int(12 * t)} else: - return {'years': int(t)} + return {"years": int(t)} if not isinstance(d, pd.Timestamp): cutoff = cutoff.date() @@ -68,24 +83,25 @@ def roll_date(d, tenor, nd_array=False): if isinstance(tenor, (int, float)): d_rolled = d + relativedelta(**kwargs(tenor), days=1) return next_twentieth(d_rolled) - elif hasattr(tenor, '__iter__'): + elif hasattr(tenor, "__iter__"): v = [next_twentieth(d + relativedelta(**kwargs(t), days=1)) for t in tenor] if nd_array: return np.array([pydate_to_TDate(d) for d in v]) else: return v else: - raise TypeError('tenor is not a number nor an iterable') + raise TypeError("tenor is not a number nor an iterable") else: # semi-annual rolling starting 2015-12-20 if isinstance(tenor, (int, float)): d_rolled = d + relativedelta(**kwargs(tenor)) - elif hasattr(tenor, '__iter__'): + elif hasattr(tenor, "__iter__"): d_rolled = d + relativedelta(years=1) else: - raise TypeError('tenor is not a number nor an iterable') + raise TypeError("tenor is not a number nor an iterable") - if((d >= d + relativedelta(month=9, day=20)) or - (d < d + relativedelta(month=3, day=20))): + if (d >= d + relativedelta(month=9, day=20)) or ( + d < d + relativedelta(month=3, day=20) + ): d_rolled += relativedelta(month=12, day=20) if d.month <= 3: d_rolled -= relativedelta(years=1) @@ -94,7 +110,7 @@ def roll_date(d, tenor, nd_array=False): if isinstance(tenor, (int, float)): return d_rolled else: - v = [d_rolled + relativedelta(**kwargs(t-1)) for t in tenor] + v = [d_rolled + relativedelta(**kwargs(t - 1)) for t in tenor] if nd_array: return np.array([pydate_to_TDate(d) for d in v]) else: @@ -102,7 +118,6 @@ def roll_date(d, tenor, nd_array=False): def build_table(rows, format_strings, row_format): - def apply_format(row, format_string): for r, f in zip(row, format_string): if f is None: @@ -116,13 +131,16 @@ def build_table(rows, format_strings, row_format): else: yield f.format(r) - return [row_format.format(*apply_format(row, format_string)) - for row, format_string in zip(rows, format_strings)] + return [ + row_format.format(*apply_format(row, format_string)) + for row, format_string in zip(rows, format_strings) + ] def memoize(f=None, *, hasher=lambda args: (hash(args),)): if f is None: return partial(memoize, hasher=hasher) + @wraps(f) def cached_f(*args, **kwargs): self = args[0] @@ -133,4 +151,5 @@ def memoize(f=None, *, hasher=lambda args: (hash(args),)): v = f(*args, **kwargs) self._cache[key] = v return v + return cached_f |