-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathbondcalendar.py
91 lines (79 loc) · 3.34 KB
/
bondcalendar.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
import numpy as np
import pandas as pd
from collections import defaultdict
import datetime
import functools
BOND_TICKERS = [
'AO20',
'AA21',
'A2E2',
'AY24',
'AA25',
'AA26',
'A2E7',
'DICA',
'AA37',
'PARA',
'AA46',
]
@functools.total_ordering
class Calendar(object):
def __init__(self, df, bondName):
self.bondName = bondName
self.dates = self._tranformDates( df['Date'].values )
self.payments = df['Total'].values
self.paymentsDict = {d:p for d,p in zip(self.dates, self.payments)}
@property
def _maturity(self):
return max(self.dates)
def __lt__(self, other):
return self._maturity < other._maturity
def __eq__(self, other):
return self._maturity == other._maturity
def _tranformDates(self, Dates):
result = []
for date in Dates:
if isinstance( date, str ):
day,month,year = [ int(component) for component in date.split('/') ]
if year<100:
year+=2000
result.append( np.datetime64(datetime.date(year=year, month=month, day=day), 'D') )
else:
result.append( date.astype('datetime64[D]') )
return result
class ListOfCalendars(object):
def __init__(self, calendars):
self.calendars = calendars
self.payments = defaultdict(tuple)
self._populatePayments()
def _populatePayments(self):
for calendar in self.calendars:
for date in calendar.dates:
self.payments[date]+=calendar.paymentsDict[date]
def getPaymentMatrix(self):
minDate = min([min(calendar.dates) for calendar in self.calendars]+[np.datetime64(datetime.date.today(),'D')])
maxDate = max([max(calendar.dates) for calendar in self.calendars])
dates = pd.date_range(start=pd.to_datetime(minDate).replace(day=1), end=pd.to_datetime(maxDate).replace(day=1), freq='MS')
pmtMatrix = np.empty((len(self.calendars),len(dates)))
for i,calendar in enumerate(sorted(self.calendars)):
for j,date in enumerate(dates):
paymentsDict = { pd.to_datetime(k).replace(day=1) : v for k,v in calendar.paymentsDict.items() }
calendarDates = [ pd.to_datetime(d).replace(day=1) for d in calendar.dates ]
pmtMatrix[i,j] = paymentsDict[date] if date in calendarDates else 0
return pmtMatrix
class BondCalendarLoader(object):
def __init__(self, xlsPath):
self.xlsPath = xlsPath
self.calendarDFs = pd.read_excel(self.xlsPath, sheet_name=None)
self.cl = None
def getCalendarList(self):
if not self.cl:
self.calendarDict = { k:self.calendarDFs[k] for k in BOND_TICKERS }
self.cl = ListOfCalendars( [ Calendar(self.calendarDFs[bondName], bondName) for bondName in BOND_TICKERS ] )
return self.cl
def getBondPrices(self):
pricesDF = self.calendarDFs.get('Prices').loc[:,['Bond','Price']]
pricesDF = pricesDF.set_index('Bond')
pricesDF = pricesDF.reindex(BOND_TICKERS)
tickersOrderedByMaturity = [ cal.bondName for cal in sorted(self.cl.calendars) ]
return pricesDF.loc[tickersOrderedByMaturity]